commit b2d186c

The pool of threads now process q_items.

Also, the purge is commanded as a q_item.

 1@@ -1188,6 +1188,14 @@ void process_queue_item(xs_dict *q_item)
 2             }
 3         }
 4     }
 5+    else
 6+    if (strcmp(type, "purge") == 0) {
 7+        srv_log(xs_dup("purge start"));
 8+
 9+        purge_all();
10+
11+        srv_log(xs_dup("purge end"));
12+    }
13 }
14 
15 

  1@@ -237,81 +237,6 @@ void term_handler(int s)
  2 }
  3 
  4 
  5-static void *purge_thread(void *arg)
  6-/* spawned purge */
  7-{
  8-    srv_log(xs_dup("purge start"));
  9-
 10-    purge_all();
 11-
 12-    srv_log(xs_dup("purge end"));
 13-
 14-    return NULL;
 15-}
 16-
 17-
 18-static void *background_thread(void *arg)
 19-/* background thread (queue management and other things) */
 20-{
 21-    time_t purge_time;
 22-
 23-    /* first purge time */
 24-    purge_time = time(NULL) + 10 * 60;
 25-
 26-    srv_log(xs_fmt("background thread started"));
 27-
 28-    while (srv_running) {
 29-        time_t t;
 30-
 31-        {
 32-            xs *list = user_list();
 33-            char *p, *uid;
 34-
 35-            /* process queues for all users */
 36-            p = list;
 37-            while (xs_list_iter(&p, &uid)) {
 38-                snac snac;
 39-
 40-                if (user_open(&snac, uid)) {
 41-                    process_user_queue(&snac);
 42-                    user_free(&snac);
 43-                }
 44-            }
 45-        }
 46-
 47-        /* global queue */
 48-        process_queue();
 49-
 50-        /* time to purge? */
 51-        if ((t = time(NULL)) > purge_time) {
 52-            pthread_t pth;
 53-
 54-            pthread_create(&pth, NULL, purge_thread, NULL);
 55-            pthread_detach(pth);
 56-
 57-            /* next purge time is tomorrow */
 58-            purge_time = t + 24 * 60 * 60;
 59-        }
 60-
 61-        /* sleep 3 seconds */
 62-        pthread_mutex_t dummy_mutex = PTHREAD_MUTEX_INITIALIZER;
 63-        pthread_cond_t  dummy_cond  = PTHREAD_COND_INITIALIZER;
 64-        struct timespec ts;
 65-
 66-        clock_gettime(CLOCK_REALTIME, &ts);
 67-        ts.tv_sec += 3;
 68-
 69-        pthread_mutex_lock(&dummy_mutex);
 70-        while (pthread_cond_timedwait(&dummy_cond, &dummy_mutex, &ts) == 0);
 71-        pthread_mutex_unlock(&dummy_mutex);
 72-    }
 73-
 74-    srv_log(xs_fmt("background thread stopped"));
 75-
 76-    return NULL;
 77-}
 78-
 79-
 80 /** job control **/
 81 
 82 /* mutex to access the lists of jobs */
 83@@ -391,6 +316,10 @@ static void *job_thread(void *arg)
 84             if (f != NULL)
 85                 httpd_connection(f);
 86         }
 87+        else {
 88+            /* it's a q_item */
 89+            process_queue_item(job);
 90+        }
 91     }
 92 
 93     srv_debug(0, xs_fmt("job thread %ld stopped", pid));
 94@@ -399,6 +328,67 @@ static void *job_thread(void *arg)
 95 }
 96 
 97 
 98+static void *background_thread(void *arg)
 99+/* background thread (queue management and other things) */
100+{
101+    time_t purge_time;
102+
103+    /* first purge time */
104+    purge_time = time(NULL) + 10 * 60;
105+
106+    srv_log(xs_fmt("background thread started"));
107+
108+    while (srv_running) {
109+        time_t t;
110+
111+        {
112+            xs *list = user_list();
113+            char *p, *uid;
114+
115+            /* process queues for all users */
116+            p = list;
117+            while (xs_list_iter(&p, &uid)) {
118+                snac snac;
119+
120+                if (user_open(&snac, uid)) {
121+                    process_user_queue(&snac);
122+                    user_free(&snac);
123+                }
124+            }
125+        }
126+
127+        /* global queue */
128+        process_queue();
129+
130+        /* time to purge? */
131+        if ((t = time(NULL)) > purge_time) {
132+            /* next purge time is tomorrow */
133+            purge_time = t + 24 * 60 * 60;
134+
135+            xs *q_item = xs_dict_new();
136+            q_item = xs_dict_append(q_item, "type", "purge");
137+            job_post(q_item);
138+        }
139+
140+        /* sleep 3 seconds */
141+        pthread_mutex_t dummy_mutex = PTHREAD_MUTEX_INITIALIZER;
142+        pthread_cond_t  dummy_cond  = PTHREAD_COND_INITIALIZER;
143+        struct timespec ts;
144+
145+        clock_gettime(CLOCK_REALTIME, &ts);
146+        ts.tv_sec += 3;
147+
148+        pthread_mutex_lock(&dummy_mutex);
149+        while (pthread_cond_timedwait(&dummy_cond, &dummy_mutex, &ts) == 0);
150+        pthread_mutex_unlock(&dummy_mutex);
151+    }
152+
153+    srv_log(xs_fmt("background thread stopped"));
154+
155+    return NULL;
156+}
157+
158+
159 void httpd(void)
160 /* starts the server */
161 {

1@@ -169,7 +169,7 @@ int send_to_actor(snac *snac, char *actor, char *msg, d_char **payload, int *p_s
2 int is_msg_public(snac *snac, char *msg);
3 
4 void process_user_queue(snac *snac);
5-
6+void process_queue_item(xs_dict *q_item);
7 void process_queue(void);
8 
9 void post(snac *snac, char *msg);