commit 6bcc6bf

New functions job_post() and job_wait() (untested).

  1@@ -13,6 +13,7 @@
  2 
  3 #include <setjmp.h>
  4 #include <pthread.h>
  5+#include <semaphore.h>
  6 
  7 
  8 /* nodeinfo 2.0 template */
  9@@ -319,6 +320,80 @@ static void *connection_thread(void *arg)
 10 }
 11 
 12 
 13+/** job control **/
 14+
 15+/* mutex to access the lists of jobs */
 16+static pthread_mutex_t job_mutex;
 17+
 18+/* semaphre to trigger job processing */
 19+static sem_t job_sem;
 20+
 21+/* list of input sockets */
 22+xs_list *job_sockets = NULL;
 23+
 24+/* list of queue items */
 25+xs_list *job_qitems = NULL;
 26+
 27+
 28+void job_post(FILE *socket, const xs_dict *q_item)
 29+/* posts a job, being an input connection or another queue item */
 30+{
 31+    /* lock the mutex */
 32+    pthread_mutex_lock(&job_mutex);
 33+
 34+    /* add to the appropriate fifo */
 35+    if (socket != NULL) {
 36+        xs *d = xs_data_new(&socket, sizeof(FILE *));
 37+        job_sockets = xs_list_append(job_sockets, d);
 38+    }
 39+    else
 40+    if (q_item != NULL)
 41+        job_qitems = xs_list_append(job_qitems, q_item);
 42+
 43+    /* unlock the mutex */
 44+    pthread_mutex_unlock(&job_mutex);
 45+
 46+    /* ask for someone to attend it */
 47+    sem_post(&job_sem);
 48+}
 49+
 50+
 51+int job_wait(FILE **socket, xs_dict **q_item)
 52+/* waits for an available job; returns 0 if nothing left to do */
 53+{
 54+    int done = 1;
 55+
 56+    *socket = NULL;
 57+    *q_item = NULL;
 58+
 59+    if (sem_wait(&job_sem) == 0) {
 60+        /* lock the mutex */
 61+        pthread_mutex_lock(&job_mutex);
 62+
 63+        /* try first to get a socket to process */
 64+        xs *job_socket = NULL;
 65+        job_sockets = xs_list_shift(job_sockets, &job_socket);
 66+
 67+        /* if empty, try a q_item */
 68+        if (job_socket == NULL)
 69+            job_qitems = xs_list_shift(job_qitems, q_item);
 70+
 71+        /* unlock the mutex */
 72+        pthread_mutex_unlock(&job_mutex);
 73+
 74+        if (job_socket != NULL) {
 75+            xs_data_get(job_socket, socket);
 76+            done = 0;
 77+        }
 78+        else
 79+        if (*q_item != NULL)
 80+            done = 0;
 81+    }
 82+
 83+    return done;
 84+}
 85+
 86+
 87 #ifndef MAX_THREADS
 88 #define MAX_THREADS 256
 89 #endif
 90@@ -348,8 +423,14 @@ void httpd(void)
 91 
 92     srv_log(xs_fmt("httpd start %s:%d %s", address, port, USER_AGENT));
 93 
 94-    /* thread creation */
 95+    /* initialize the job control engine */
 96+    pthread_mutex_init(&job_mutex, NULL);
 97+    sem_init(&job_sem, 0, 0);
 98+    job_sockets = xs_list_new();
 99+    job_qitems = xs_list_new();
100+
101 #ifdef _SC_NPROCESSORS_ONLN
102+    /* get number of CPUs on the machine */
103     n_threads = sysconf(_SC_NPROCESSORS_ONLN);
104 #endif
105 
106@@ -380,5 +461,8 @@ void httpd(void)
107     /* wait for the background thread to end */
108     pthread_join(threads[0], NULL);
109 
110+    job_sockets = xs_free(job_sockets);
111+    job_qitems  = xs_free(job_qitems);
112+
113     srv_log(xs_fmt("httpd stop %s:%d", address, port));
114 }