added thread poll (minimal one);

include/sntl/pth_queue.h

@@ -18,6 +18,7 @@
 #define __PTH_QUEUE_H__
 
 #include <pthread.h>
+#include <tdata/idx_allocator.h>
 
 /* possible message types, ones with POLL_ prefix valid on for pth_dqtpoll_* */
 #define SYS_MSG        0x0f0affee
@@ -79,6 +80,7 @@ typedef struct pth_dqtpoll_s {
   pthread_t *poll; /** < Thread descriptors */
   int (*jobdata_callback)(void *); /** < Callback to have a deal with data */
   int flags; /** < Flags */
+  idx_allocator_t *idx; /** < index allocator for the poll threads */
   pthread_rwlock_t stats_lock; /** < rwlock for stats data */
   unsigned long spurious_wakeups; /** < amount of spurios wakeups */
   int poll_value; /** < value of the poll (totally) */

lib/queue.c

@@ -211,3 +211,119 @@ unsigned int pth_queue_length(pth_queue_t *queue)
 
   return c;
 }
+
+/* dynamic queue thread poll */
+
+static void *__poll_thread(void *poll)
+{
+  int r = 0;
+  pth_msg_t msgbuf;
+  pth_dqtpoll_t *p = (pth_dqtpoll_t *)poll;
+  pth_queue_t *q = p->queue;
+
+  while(1) {
+    r = pth_queue_get(q, NULL, &msgbuf);
+    if(r) continue;
+    switch(msgbuf.msgtype) {
+    case USR_MSG:
+      p->jobdata_callback(msgbuf.data);
+      break;
+    default:
+      /* TODO: do something ... */
+      break;
+    }
+  }
+
+  return NULL;
+}
+
+/* init poll, structure must be allocated */
+int pth_dqtpoll_init(pth_dqtpoll_t *tpoll, int (*jobdata_callback)(void *))
+{
+  int r = 0;
+  pth_queue_t *queue = malloc(sizeof(pth_queue_t));
+  pthread_t *poll = malloc(sizeof(pthread_t)*MAX_POLL_VALUE);
+  idx_allocator_t *idx = malloc(sizeof(idx_allocator_t));
+
+  /* check it for allocation */
+  if(!idx) goto __enomem;
+  if(!queue) goto __enomem;
+  if(!poll) goto __enomem;
+
+  /* init all the stuff */
+  if(idx_allocator_init(idx, MAX_POLL_VALUE, 0)) {
+  __enomem:
+    r = ENOMEM;
+    goto __finish;
+  }
+  if(pth_queue_init(queue)) goto __enomem; /* init queue */
+  if(pthread_rwlock_init(&(tpoll->stats_lock), NULL)) goto __enomem;
+
+  /* set parameters */
+  memset(poll, 0, sizeof(pthread_t)*MAX_POLL_VALUE);
+  tpoll->flags = 0;
+  tpoll->idx = idx;
+  tpoll->poll = poll;
+  tpoll->queue = queue;
+  tpoll->poll_value = 1;
+  tpoll->sleep_value = 0;
+  tpoll->spurious_wakeups = 0;
+  tpoll->jobdata_callback = jobdata_callback;
+
+  /* first thread initiation */
+  idx_reserve(idx, 0);
+  if(pthread_create(&(poll[0]), NULL, __poll_thread, tpoll)) {
+    pthread_rwlock_destroy(&(tpoll->stats_lock));
+    goto __enomem;
+  }
+
+ __finish:
+  if(r) {
+    if(idx) free(idx);
+    if(queue) {
+      pth_queue_destroy(queue, 0, NULL);
+      free(queue);
+    }
+    if(poll) free(poll);
+  }
+  return r;
+}
+
+/* run poll: poll */
+int pth_dqtpoll_run(pth_dqtpoll_t *tpoll)
+{
+  int r = 0;
+
+  pthread_rwlock_wrlock(&(tpoll->stats_lock));
+  if((tpoll->flags & DQTPOLL_RUNNING) || (tpoll->flags & DQTPOLL_DEADSTAGE)) r = EINVAL;
+  else {
+    tpoll->flags |= DQTPOLL_RUNNING;
+    tpoll->sleep_value = 1;
+  }
+  pthread_rwlock_unlock(&(tpoll->stats_lock));
+
+  return r;
+}
+
+/* add the job to the queue: poll, job data, message type  */
+int pth_dqtpoll_add(pth_dqtpoll_t *tpoll, void *job, unsigned int type)
+{
+  int r = 0;
+
+  r = pth_queue_add(tpoll->queue, job, type);
+
+  return r;
+}
+
+/* destroy the poll: poll, force flag
+ * if force flag is set (!= 0), give up
+ * about jobs, if no, do the job, but don't
+ * accept the new ones, and destroy all poll
+ * with last thread.
+ */
+int pth_dqtpoll_destroy(pth_dqtpoll_t *tpoll, int force)
+{
+  int r = 0;
+  return r;
+}
+