commit 2bfebba

Don't wait for 3 seconds if there were some q_items processed.

 1@@ -1160,9 +1160,10 @@ void process_user_queue_item(snac *snac, xs_dict *q_item)
 2 }
 3 
 4 
 5-void process_user_queue(snac *snac)
 6+int process_user_queue(snac *snac)
 7 /* processes a user's queue */
 8 {
 9+    int cnt = 0;
10     xs *list = user_queue(snac);
11 
12     xs_list *p = list;
13@@ -1177,7 +1178,10 @@ void process_user_queue(snac *snac)
14         }
15 
16         process_user_queue_item(snac, q_item);
17+        cnt++;
18     }
19+
20+    return cnt;
21 }
22 
23 
24@@ -1299,9 +1303,10 @@ void process_queue_item(xs_dict *q_item)
25 }
26 
27 
28-void process_queue(void)
29+int process_queue(void)
30 /* processes the global queue */
31 {
32+    int cnt = 0;
33     xs *list = queue();
34 
35     xs_list *p = list;
36@@ -1310,9 +1315,13 @@ void process_queue(void)
37     while (xs_list_iter(&p, &fn)) {
38         xs *q_item = dequeue(fn);
39 
40-        if (q_item != NULL)
41+        if (q_item != NULL) {
42             job_post(q_item);
43+            cnt++;
44+        }
45     }
46+
47+    return cnt;
48 }
49 
50 

 1@@ -17,6 +17,10 @@
 2 
 3 #include <sys/resource.h> // for getrlimit()
 4 
 5+#ifdef USE_POLL_FOR_SLEEP
 6+#include <poll.h>
 7+#endif
 8+
 9 
10 /* nodeinfo 2.0 template */
11 const char *nodeinfo_2_0_template = ""
12@@ -338,6 +342,7 @@ static void *job_thread(void *arg)
13     return NULL;
14 }
15 
16+#include <poll.h>
17 
18 static void *background_thread(void *arg)
19 /* background thread (queue management and other things) */
20@@ -351,6 +356,7 @@ static void *background_thread(void *arg)
21 
22     while (srv_running) {
23         time_t t;
24+        int cnt = 0;
25 
26         {
27             xs *list = user_list();
28@@ -362,14 +368,14 @@ static void *background_thread(void *arg)
29                 snac snac;
30 
31                 if (user_open(&snac, uid)) {
32-                    process_user_queue(&snac);
33+                    cnt += process_user_queue(&snac);
34                     user_free(&snac);
35                 }
36             }
37         }
38 
39         /* global queue */
40-        process_queue();
41+        cnt += process_queue();
42 
43         /* time to purge? */
44         if ((t = time(NULL)) > purge_time) {
45@@ -381,17 +387,24 @@ static void *background_thread(void *arg)
46             job_post(q_item);
47         }
48 
49-        /* sleep 3 seconds */
50-        pthread_mutex_t dummy_mutex = PTHREAD_MUTEX_INITIALIZER;
51-        pthread_cond_t  dummy_cond  = PTHREAD_COND_INITIALIZER;
52-        struct timespec ts;
53+        if (cnt == 0) {
54+            /* sleep 3 seconds */
55+
56+#ifdef USE_POLL_FOR_SLEEP
57+            poll(NULL, 0, 3 * 1000);
58+#else
59+            pthread_mutex_t dummy_mutex = PTHREAD_MUTEX_INITIALIZER;
60+            pthread_cond_t  dummy_cond  = PTHREAD_COND_INITIALIZER;
61+            struct timespec ts;
62 
63-        clock_gettime(CLOCK_REALTIME, &ts);
64-        ts.tv_sec += 3;
65+            clock_gettime(CLOCK_REALTIME, &ts);
66+            ts.tv_sec += 3;
67 
68-        pthread_mutex_lock(&dummy_mutex);
69-        while (pthread_cond_timedwait(&dummy_cond, &dummy_mutex, &ts) == 0);
70-        pthread_mutex_unlock(&dummy_mutex);
71+            pthread_mutex_lock(&dummy_mutex);
72+            while (pthread_cond_timedwait(&dummy_cond, &dummy_mutex, &ts) == 0);
73+            pthread_mutex_unlock(&dummy_mutex);
74+#endif
75+        }
76     }
77 
78     srv_log(xs_fmt("background thread stopped"));

 1@@ -190,9 +190,9 @@ d_char *get_actor_inbox(snac *snac, char *actor);
 2 int send_to_actor(snac *snac, char *actor, char *msg, d_char **payload, int *p_size, int timeout);
 3 int is_msg_public(snac *snac, char *msg);
 4 
 5-void process_user_queue(snac *snac);
 6+int process_user_queue(snac *snac);
 7 void process_queue_item(xs_dict *q_item);
 8-void process_queue(void);
 9+int process_queue(void);
10 
11 int activitypub_get_handler(d_char *req, char *q_path,
12                             char **body, int *b_size, char **ctype);