diff options
| -rw-r--r-- | activitypub.c | 15 | ||||
| -rw-r--r-- | httpd.c | 35 | ||||
| -rw-r--r-- | snac.h | 4 |
3 files changed, 38 insertions, 16 deletions
diff --git a/activitypub.c b/activitypub.c index 1d191b8..c05c859 100644 --- a/activitypub.c +++ b/activitypub.c @@ -1160,9 +1160,10 @@ void process_user_queue_item(snac *snac, xs_dict *q_item) } -void process_user_queue(snac *snac) +int process_user_queue(snac *snac) /* processes a user's queue */ { + int cnt = 0; xs *list = user_queue(snac); xs_list *p = list; @@ -1177,7 +1178,10 @@ void process_user_queue(snac *snac) } process_user_queue_item(snac, q_item); + cnt++; } + + return cnt; } @@ -1299,9 +1303,10 @@ void process_queue_item(xs_dict *q_item) } -void process_queue(void) +int process_queue(void) /* processes the global queue */ { + int cnt = 0; xs *list = queue(); xs_list *p = list; @@ -1310,9 +1315,13 @@ void process_queue(void) while (xs_list_iter(&p, &fn)) { xs *q_item = dequeue(fn); - if (q_item != NULL) + if (q_item != NULL) { job_post(q_item); + cnt++; + } } + + return cnt; } @@ -17,6 +17,10 @@ #include <sys/resource.h> // for getrlimit() +#ifdef USE_POLL_FOR_SLEEP +#include <poll.h> +#endif + /* nodeinfo 2.0 template */ const char *nodeinfo_2_0_template = "" @@ -338,6 +342,7 @@ static void *job_thread(void *arg) return NULL; } +#include <poll.h> static void *background_thread(void *arg) /* background thread (queue management and other things) */ @@ -351,6 +356,7 @@ static void *background_thread(void *arg) while (srv_running) { time_t t; + int cnt = 0; { xs *list = user_list(); @@ -362,14 +368,14 @@ static void *background_thread(void *arg) snac snac; if (user_open(&snac, uid)) { - process_user_queue(&snac); + cnt += process_user_queue(&snac); user_free(&snac); } } } /* global queue */ - process_queue(); + cnt += process_queue(); /* time to purge? */ if ((t = time(NULL)) > purge_time) { @@ -381,17 +387,24 @@ static void *background_thread(void *arg) job_post(q_item); } - /* sleep 3 seconds */ - pthread_mutex_t dummy_mutex = PTHREAD_MUTEX_INITIALIZER; - pthread_cond_t dummy_cond = PTHREAD_COND_INITIALIZER; - struct timespec ts; + if (cnt == 0) { + /* sleep 3 seconds */ + +#ifdef USE_POLL_FOR_SLEEP + poll(NULL, 0, 3 * 1000); +#else + pthread_mutex_t dummy_mutex = PTHREAD_MUTEX_INITIALIZER; + pthread_cond_t dummy_cond = PTHREAD_COND_INITIALIZER; + struct timespec ts; - clock_gettime(CLOCK_REALTIME, &ts); - ts.tv_sec += 3; + clock_gettime(CLOCK_REALTIME, &ts); + ts.tv_sec += 3; - pthread_mutex_lock(&dummy_mutex); - while (pthread_cond_timedwait(&dummy_cond, &dummy_mutex, &ts) == 0); - pthread_mutex_unlock(&dummy_mutex); + pthread_mutex_lock(&dummy_mutex); + while (pthread_cond_timedwait(&dummy_cond, &dummy_mutex, &ts) == 0); + pthread_mutex_unlock(&dummy_mutex); +#endif + } } srv_log(xs_fmt("background thread stopped")); @@ -190,9 +190,9 @@ d_char *get_actor_inbox(snac *snac, char *actor); int send_to_actor(snac *snac, char *actor, char *msg, d_char **payload, int *p_size, int timeout); int is_msg_public(snac *snac, char *msg); -void process_user_queue(snac *snac); +int process_user_queue(snac *snac); void process_queue_item(xs_dict *q_item); -void process_queue(void); +int process_queue(void); int activitypub_get_handler(d_char *req, char *q_path, char **body, int *b_size, char **ctype); |