added pulse messages, added thread poll with dynamic thread creation/destroy, fixed bugs and todos with fixme;

include/sntl/connection.h

@@ -26,6 +26,9 @@
 #define SXOREPLYREQ  44  /* protocol require reply with expression,
                           * or expression return for the request */
 #define SXOTIMEDOUT  45  /* timedout */
+#define ESXRCBADPROT 46	 /* invalid protocol */
+#define ESXNOCONNECT 47  /* connection is lost */
+#define ESXNOCHANSUP 48
 
 /* sexp helpers */
 #define SEXP_IS_LIST(sx) \
@@ -85,6 +88,7 @@ typedef struct __connection_t {
   pthread_t msgthread; /** < thread for message queue (2) */
   pth_queue_t *mqueue; /** < message queue (2) */
   pth_queue_t *rqueue; /** < message queue (1) */
+  pth_dqtpoll_t *tpoll; /** < thread poll for rpc requests */
   pthread_mutex_t oplock; /** < mutex used to sync operations on connection */
   pthread_rwlock_t chnl_lock; /** < rwlock used to sync ops with channels */
   int flags; /** < flags of the connection */

lib/queue.c

@@ -214,20 +214,81 @@ unsigned int pth_queue_length(pth_queue_t *queue)
 
 /* dynamic queue thread poll */
 
+struct __pthrd_data {
+  pth_dqtpoll_t *pthrd;
+  int myid;
+};
+
 static void *__poll_thread(void *poll)
 {
   int r = 0;
-  pth_msg_t msgbuf;
+  struct __pthrd_data *thrdata = (struct __pthrd_data *)poll;
-  pth_dqtpoll_t *p = (pth_dqtpoll_t *)poll;
+  struct __pthrd_data *npoll = NULL;
+  pth_msg_t msgbuf, sysbuf;
+  pth_dqtpoll_t *p = thrdata->pthrd;
   pth_queue_t *q = p->queue;
+  ulong_t myid = thrdata->myid;
 
   while(1) {
     r = pth_queue_get(q, NULL, &msgbuf);
-    if(r) continue;
+    pthread_rwlock_wrlock(&(p->stats_lock));
+    if(r) {
+      p->sleep_value++;
+      pthread_rwlock_unlock(&(p->stats_lock));
+      continue;
+    } else p->sleep_value--;
+    pthread_rwlock_unlock(&(p->stats_lock));
+
     switch(msgbuf.msgtype) {
     case USR_MSG:
+      /* schedule new task for poll - increase or decrease */
+      pthread_rwlock_rdlock(&(p->stats_lock));
+      if(pth_queue_length(p->queue) > p->sleep_value) {
+        memset(&sysbuf, 0, sizeof(pth_msg_t));
+        pth_queue_add(p->queue, &sysbuf, POLL_INCREASE);
+      } else if((pth_queue_length(p->queue) > p->sleep_value) && (p->sleep_value > 1)) {
+        memset(&sysbuf, 0, sizeof(pth_msg_t));
+        pth_queue_add(p->queue, &sysbuf, POLL_DECREASE);
+      }
+      pthread_rwlock_unlock(&(p->stats_lock));
+      /* do the job */
       p->jobdata_callback(msgbuf.data);
       break;
+    case POLL_DECREASE:
+      pthread_rwlock_rdlock(&(p->stats_lock));
+      if(p->poll_value > 1) r = 1; /* exit now */
+      pthread_rwlock_unlock(&(p->stats_lock));
+      if(r) {
+        pthread_rwlock_wrlock(&(p->stats_lock));
+        idx_free(p->idx, myid);
+        p->poll_value--;
+        pthread_rwlock_unlock(&(p->stats_lock));
+        free(poll);
+        return NULL;
+      }
+      break;
+    case POLL_INCREASE:
+      if((npoll = malloc(sizeof(struct __pthrd_data)))) {
+        pthread_rwlock_rdlock(&(p->stats_lock));
+        if(p->poll_value < MAX_POLL_VALUE) r = 1;
+        pthread_rwlock_unlock(&(p->stats_lock));
+        if(r) {
+          pthread_rwlock_wrlock(&(p->stats_lock));
+          if(p->poll_value < MAX_POLL_VALUE) { /* check it again */
+            npoll->myid = idx_allocate(p->idx);
+            npoll->pthrd = p;
+            p->poll_value++;
+            /* create thread here */
+            if(pthread_create(&(p->poll[npoll->myid]), NULL, __poll_thread, npoll)) {
+              idx_free(p->idx, npoll->myid);
+              p->poll_value--;
+              free(npoll);
+            }
+          }
+          pthread_rwlock_unlock(&(p->stats_lock));
+        }
+      }
+      break;
     default:
       /* TODO: do something ... */
       break;
@@ -244,8 +305,10 @@ int pth_dqtpoll_init(pth_dqtpoll_t *tpoll, int (*jobdata_callback)(void *))
   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));
+  struct __pthrd_data *ndata = malloc(sizeof(struct __pthrd_data));
 
   /* check it for allocation */
+  if(!ndata) goto __enomem;
   if(!idx) goto __enomem;
   if(!queue) goto __enomem;
   if(!poll) goto __enomem;
@@ -272,13 +335,16 @@ int pth_dqtpoll_init(pth_dqtpoll_t *tpoll, int (*jobdata_callback)(void *))
 
   /* first thread initiation */
   idx_reserve(idx, 0);
-  if(pthread_create(&(poll[0]), NULL, __poll_thread, tpoll)) {
+  ndata->myid = 0;
+  ndata->pthrd = tpoll;
+  if(pthread_create(&(poll[0]), NULL, __poll_thread, ndata)) {
     pthread_rwlock_destroy(&(tpoll->stats_lock));
     goto __enomem;
   }
 
  __finish:
   if(r) {
+    if(ndata) free(ndata);
     if(idx) free(idx);
     if(queue) {
       pth_queue_destroy(queue, 0, NULL);