summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--Makefile10
-rw-r--r--opentracker.c11
-rw-r--r--ot_accesslist.c46
-rw-r--r--ot_accesslist.h2
-rw-r--r--ot_fullscrape.c25
-rw-r--r--ot_fullscrape.h1
-rw-r--r--ot_http.c3
-rw-r--r--ot_mutex.c224
-rw-r--r--ot_mutex.h57
-rw-r--r--ot_stats.c9
-rw-r--r--ot_sync.c7
-rw-r--r--ot_tcp.c66
-rw-r--r--ot_tcp.h2
-rw-r--r--ot_udp.c69
-rw-r--r--ot_udp.h2
-rw-r--r--ot_worker.c75
-rw-r--r--ot_worker.h39
-rw-r--r--ot_workqueue.c243
-rw-r--r--ot_workqueue.h78
-rw-r--r--tests/integration/test_dynamic_accesslist.py50
-rw-r--r--tests/runner.c2
-rw-r--r--tests/stubs.c33
-rw-r--r--tests/t_mutex.c171
-rw-r--r--tests/t_trackerlogic.c243
-rw-r--r--tests/t_workqueue.c107
-rw-r--r--tests/test.h2
-rw-r--r--trackerlogic.c3
27 files changed, 1063 insertions, 517 deletions
diff --git a/Makefile b/Makefile
index 8f4dcf6..6a5a270 100644
--- a/Makefile
+++ b/Makefile
@@ -69,8 +69,8 @@ CFLAGS_libowfat=-I$(LIBOWFAT_HEADERS)
69LDFLAGS_libowfat=-L$(LIBOWFAT_LIBRARY) -lowfat 69LDFLAGS_libowfat=-L$(LIBOWFAT_LIBRARY) -lowfat
70 70
71BINARY =opentracker 71BINARY =opentracker
72HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h 72HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.h ot_worker.h ot_workqueue.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h
73SOURCES=opentracker.c trackerlogic.c ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c 73SOURCES=opentracker.c trackerlogic.c ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.c ot_worker.c ot_workqueue.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c
74SOURCES_proxy=proxy.c ot_batch.c ot_fdstate.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c 74SOURCES_proxy=proxy.c ot_batch.c ot_fdstate.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c
75 75
76all: $(BINARY) $(BINARY).debug 76all: $(BINARY) $(BINARY).debug
@@ -96,8 +96,8 @@ proxy.debug: $(SOURCES_proxy) $(HEADERS)
96# has to. tests/stubs.c supplies the few symbols those modules reference but 96# has to. tests/stubs.c supplies the few symbols those modules reference but
97# the tests never reach. 97# the tests never reach.
98TESTBIN=tests/runner 98TESTBIN=tests/runner
99TEST_SOURCES=tests/runner.c tests/test.c tests/stubs.c tests/t_urlencoded.c tests/t_iovec.c tests/t_vector.c tests/t_random.c tests/t_hex.c tests/t_accesslist.c tests/t_scan.c tests/t_buf.c tests/t_ip.c tests/t_socket.c tests/t_mutex.c tests/t_batch.c tests/t_fdstate.c tests/t_event.c tests/t_fmt.c 99TEST_SOURCES=tests/runner.c tests/test.c tests/stubs.c tests/t_urlencoded.c tests/t_iovec.c tests/t_vector.c tests/t_random.c tests/t_hex.c tests/t_accesslist.c tests/t_scan.c tests/t_buf.c tests/t_ip.c tests/t_socket.c tests/t_mutex.c tests/t_trackerlogic.c tests/t_workqueue.c tests/t_batch.c tests/t_fdstate.c tests/t_event.c tests/t_fmt.c
100TEST_UNITS=ot_urlencoded.c ot_fmt.c ot_event.c ot_batch.c ot_fdstate.c ot_buf.c ot_ip.c ot_socket.c ot_iovec.c ot_vector.c ot_random.c ot_hex.c ot_scan.c ot_accesslist.c ot_mutex.c 100TEST_UNITS=ot_urlencoded.c ot_fmt.c ot_event.c ot_batch.c ot_fdstate.c ot_buf.c ot_ip.c ot_socket.c ot_iovec.c ot_vector.c ot_random.c ot_hex.c ot_scan.c ot_accesslist.c ot_mutex.c ot_workqueue.c trackerlogic.c
101TEST_HEADERS=tests/test.h $(HEADERS) 101TEST_HEADERS=tests/test.h $(HEADERS)
102# Tests reach code that sits behind feature flags, so the units under test are 102# Tests reach code that sits behind feature flags, so the units under test are
103# built with those features enabled. A test binary is not a production binary. 103# built with those features enabled. A test binary is not a production binary.
@@ -136,7 +136,7 @@ check-sanitize: $(TESTBIN).sanitize
136PYTHON?=python3 136PYTHON?=python3
137# The client timeout is cut right down so that tests can reach the expiry 137# The client timeout is cut right down so that tests can reach the expiry
138# sweep. tests/integration/test_expiry.py mirrors these two numbers. 138# sweep. tests/integration/test_expiry.py mirrors these two numbers.
139FEATURES_integration=$(FEATURES) -DWANT_IP_FROM_QUERY_STRING -DWANT_ACCESSLIST_BLACK -DWANT_KEEPALIVE -DOT_CLIENT_TIMEOUT=5 -DOT_CLIENT_TIMEOUT_CHECKINTERVAL=1 139FEATURES_integration=$(FEATURES) -DWANT_IP_FROM_QUERY_STRING -DWANT_ACCESSLIST_BLACK -DWANT_DYNAMIC_ACCESSLIST -DWANT_KEEPALIVE -DOT_CLIENT_TIMEOUT=5 -DOT_CLIENT_TIMEOUT_CHECKINTERVAL=1
140CFLAGS_integration=$(CFLAGS) $(OPTS_debug) $(FEATURES_integration) -fsanitize=address,undefined -fno-omit-frame-pointer 140CFLAGS_integration=$(CFLAGS) $(OPTS_debug) $(FEATURES_integration) -fsanitize=address,undefined -fno-omit-frame-pointer
141LDFLAGS_integration=$(LDFLAGS) -fsanitize=address,undefined 141LDFLAGS_integration=$(LDFLAGS) -fsanitize=address,undefined
142 142
diff --git a/opentracker.c b/opentracker.c
index e5efff5..644e99c 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -29,6 +29,7 @@
29#include "ot_ip.h" 29#include "ot_ip.h"
30#include "ot_livesync.h" 30#include "ot_livesync.h"
31#include "ot_mutex.h" 31#include "ot_mutex.h"
32#include "ot_workqueue.h"
32#include "ot_random.h" 33#include "ot_random.h"
33#include "ot_scan.h" 34#include "ot_scan.h"
34#include "ot_event.h" 35#include "ot_event.h"
@@ -37,6 +38,7 @@
37#include "ot_stats.h" 38#include "ot_stats.h"
38#include "ot_tcp.h" 39#include "ot_tcp.h"
39#include "ot_udp.h" 40#include "ot_udp.h"
41#include "ot_worker.h"
40#include "trackerlogic.h" 42#include "trackerlogic.h"
41 43
42/* Globals */ 44/* Globals */
@@ -137,16 +139,18 @@ static void *server_mainloop(void *args) {
137 else if ((intptr_t)cookie == FLAG_SELFPIPE) 139 else if ((intptr_t)cookie == FLAG_SELFPIPE)
138 while (read((int)sock, ws.inbuf, G_INBUF_SIZE) > 0) 140 while (read((int)sock, ws.inbuf, G_INBUF_SIZE) > 0)
139 ; 141 ;
140 else { 142 else if (fdstate_claim((int)sock, ready[i].generation)) {
141 if (ready[i].what & EVENT_IN) 143 if (ready[i].what & EVENT_IN)
142 tcp_read(sock, &ws); 144 tcp_read(sock, &ws);
143 /* The read may have closed it, so look again before writing */ 145 /* The read may have closed it, so look again before writing */
144 if ((ready[i].what & EVENT_OUT) && fdstate_generation((int)sock) == ready[i].generation) 146 if ((ready[i].what & EVENT_OUT) && fdstate_generation((int)sock) == ready[i].generation)
145 tcp_write(sock); 147 tcp_write(sock);
148 if (fdstate_generation((int)sock) == ready[i].generation)
149 fdstate_release((int)sock);
146 } 150 }
147 } 151 }
148 152
149 while ((sock = mutex_workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1) 153 while ((sock = workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1)
150 http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial); 154 http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial);
151 155
152 if (time_now_seconds() > next_timeout_check) { 156 if (time_now_seconds() > next_timeout_check) {
@@ -426,8 +430,7 @@ int main(int argc, char **argv) {
426 listen_start_udp_workers(); 430 listen_start_udp_workers();
427 431
428 server_mainloop(0); 432 server_mainloop(0);
429 tcp_workers_deinit(); 433 worker_stop_all();
430 udp_workers_deinit();
431 434
432 /* No thread is serving any more, so the teardown can run where a debugger 435 /* No thread is serving any more, so the teardown can run where a debugger
433 and the sanitizers can follow it */ 436 and the sanitizers can follow it */
diff --git a/ot_accesslist.c b/ot_accesslist.c
index e8e0476..df084b8 100644
--- a/ot_accesslist.c
+++ b/ot_accesslist.c
@@ -273,7 +273,8 @@ static void *accesslist_worker(void *args) {
273} 273}
274 274
275#ifdef WANT_DYNAMIC_ACCESSLIST 275#ifdef WANT_DYNAMIC_ACCESSLIST
276static pthread_t thread_adder_id, thread_deleter_id; 276static pthread_t thread_adder_id, thread_deleter_id;
277static _Atomic int adder_left, deleter_left;
277static void *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_Atomic *adding_to, ot_accesslist *_Atomic *removing_from) { 278static void *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_Atomic *adding_to, ot_accesslist *_Atomic *removing_from) {
278 struct stat st; 279 struct stat st;
279 280
@@ -297,6 +298,9 @@ static void *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_
297 ssize_t linelen; 298 ssize_t linelen;
298 299
299 if (!fifo) { 300 if (!fifo) {
301 /* Woken on purpose: the loop condition above decides what happens next */
302 if (errno == EINTR)
303 continue;
300 fprintf(stderr, "Error when reading dynamic accesslists: Couldn't open FIFO at %s, error: %s\n", fifoname, strerror(errno)); 304 fprintf(stderr, "Error when reading dynamic accesslists: Couldn't open FIFO at %s, error: %s\n", fifoname, strerror(errno));
301 return NULL; 305 return NULL;
302 } 306 }
@@ -324,8 +328,14 @@ static void *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_
324 ptrdiff_t off = exactmatch - (*removing_from)->list; 328 ptrdiff_t off = exactmatch - (*removing_from)->list;
325 ot_accesslist *accesslist_new = accesslist_make(*removing_from, (*removing_from)->size - 1); 329 ot_accesslist *accesslist_new = accesslist_make(*removing_from, (*removing_from)->size - 1);
326 if (accesslist_new) { 330 if (accesslist_new) {
327 memcpy(accesslist_new->list, (*removing_from)->list, sizeof(ot_hash) * off); 331 size_t after = (*removing_from)->size - off - 1;
328 memcpy(accesslist_new->list + off, (*removing_from)->list + off + 1, sizeof(ot_hash) * ((*removing_from)->size - off - 1)); 332
333 /* Removing the only entry leaves a list with no room at all, and
334 memcpy may not be handed that even for nothing */
335 if (off)
336 memcpy(accesslist_new->list, (*removing_from)->list, sizeof(ot_hash) * off);
337 if (after)
338 memcpy(accesslist_new->list + off, (*removing_from)->list + off + 1, sizeof(ot_hash) * after);
329 (*removing_from)->base = time_now_minutes(); 339 (*removing_from)->base = time_now_minutes();
330 *removing_from = accesslist_new; 340 *removing_from = accesslist_new;
331 } 341 }
@@ -360,6 +370,7 @@ static void *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_
360 pthread_mutex_unlock(&g_accesslist_mutex); 370 pthread_mutex_unlock(&g_accesslist_mutex);
361 } 371 }
362 372
373 free(line);
363 fclose(fifo); 374 fclose(fifo);
364 } 375 }
365 return NULL; 376 return NULL;
@@ -367,11 +378,30 @@ static void *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_
367 378
368static void *accesslist_adder_worker(void *args) { 379static void *accesslist_adder_worker(void *args) {
369 (void)args; 380 (void)args;
370 return accesslist_adddel_worker(g_accesslist_pipe_add, &g_accesslist_add, &g_accesslist_delete); 381 accesslist_adddel_worker(g_accesslist_pipe_add, &g_accesslist_add, &g_accesslist_delete);
382 atomic_store(&adder_left, 1);
383 return NULL;
371} 384}
372static void *accesslist_deleter_worker(void *args) { 385static void *accesslist_deleter_worker(void *args) {
373 (void)args; 386 (void)args;
374 return accesslist_adddel_worker(g_accesslist_pipe_delete, &g_accesslist_delete, &g_accesslist_add); 387 accesslist_adddel_worker(g_accesslist_pipe_delete, &g_accesslist_delete, &g_accesslist_add);
388 atomic_store(&deleter_left, 1);
389 return NULL;
390}
391
392/* These sit in a blocking open or read on a FIFO, so they are woken the way
393 the other workers are, and the signal is repeated because it only
394 interrupts a call already under way. The pipe name is what says whether
395 the thread was ever started. */
396static void accesslist_stop_fifo_worker(pthread_t thread, _Atomic int *left, const char *pipe_name) {
397 if (!pipe_name)
398 return;
399
400 while (!atomic_load(left)) {
401 pthread_kill(thread, OT_SIGNAL_WAKE);
402 usleep(1000);
403 }
404 pthread_join(thread, NULL);
375} 405}
376#endif 406#endif
377 407
@@ -392,6 +422,12 @@ void accesslist_deinit(void) {
392 pthread_kill(thread_id, SIGHUP); 422 pthread_kill(thread_id, SIGHUP);
393 pthread_join(thread_id, NULL); 423 pthread_join(thread_id, NULL);
394 424
425#ifdef WANT_DYNAMIC_ACCESSLIST
426 /* Before the lists they are reading are freed below */
427 accesslist_stop_fifo_worker(thread_adder_id, &adder_left, g_accesslist_pipe_add);
428 accesslist_stop_fifo_worker(thread_deleter_id, &deleter_left, g_accesslist_pipe_delete);
429#endif
430
395 pthread_mutex_lock(&g_accesslist_mutex); 431 pthread_mutex_lock(&g_accesslist_mutex);
396 432
397 g_accesslist = accesslist_free(g_accesslist); 433 g_accesslist = accesslist_free(g_accesslist);
diff --git a/ot_accesslist.h b/ot_accesslist.h
index 0a8bf4c..9c39a7a 100644
--- a/ot_accesslist.h
+++ b/ot_accesslist.h
@@ -14,6 +14,8 @@
14 14
15#if defined(WANT_ACCESSLIST_BLACK) || defined(WANT_ACCESSLIST_WHITE) 15#if defined(WANT_ACCESSLIST_BLACK) || defined(WANT_ACCESSLIST_WHITE)
16#define WANT_ACCESSLIST 16#define WANT_ACCESSLIST
17/* Starts a thread that rereads the list when sent SIGHUP, and is stopped by
18 being sent one, so the caller blocks SIGHUP process wide beforehand */
17void accesslist_init(void); 19void accesslist_init(void);
18void accesslist_deinit(void); 20void accesslist_deinit(void);
19int accesslist_hashisvalid(const ot_hash hash); 21int accesslist_hashisvalid(const ot_hash hash);
diff --git a/ot_fullscrape.c b/ot_fullscrape.c
index a86d6d7..ae29308 100644
--- a/ot_fullscrape.c
+++ b/ot_fullscrape.c
@@ -23,6 +23,7 @@
23#include "ot_fullscrape.h" 23#include "ot_fullscrape.h"
24#include "ot_iovec.h" 24#include "ot_iovec.h"
25#include "ot_mutex.h" 25#include "ot_mutex.h"
26#include "ot_workqueue.h"
26#include "ot_urlencoded.h" 27#include "ot_urlencoded.h"
27#include "trackerlogic.h" 28#include "trackerlogic.h"
28 29
@@ -54,7 +55,7 @@ static void *fullscrape_worker(void *args) {
54 55
55 while (g_opentracker_running) { 56 while (g_opentracker_running) {
56 ot_tasktype tasktype = TASK_FULLSCRAPE; 57 ot_tasktype tasktype = TASK_FULLSCRAPE;
57 ot_taskid taskid = mutex_workqueue_poptask(&tasktype); 58 ot_taskid taskid = workqueue_poptask(&tasktype);
58 if (!taskid) 59 if (!taskid)
59 break; 60 break;
60#ifdef WANT_COMPRESSION_ZSTD 61#ifdef WANT_COMPRESSION_ZSTD
@@ -68,7 +69,7 @@ static void *fullscrape_worker(void *args) {
68 else 69 else
69#endif 70#endif
70 fullscrape_make(taskid, tasktype); 71 fullscrape_make(taskid, tasktype);
71 mutex_workqueue_pushchunked(taskid, NULL); 72 workqueue_pushchunked(taskid, NULL);
72 } 73 }
73 return NULL; 74 return NULL;
74} 75}
@@ -79,12 +80,12 @@ void fullscrape_init(void) {
79} 80}
80 81
81void fullscrape_deinit(void) { 82void fullscrape_deinit(void) {
82 mutex_workqueue_wakeall(); 83 workqueue_wakeall();
83 pthread_join( thread_id, NULL ); 84 pthread_join( thread_id, NULL );
84} 85}
85 86
86void fullscrape_deliver( int64_t sock, ot_tasktype tasktype ) { 87void fullscrape_deliver( int64_t sock, ot_tasktype tasktype ) {
87 mutex_workqueue_pushtask( sock, TASK_FULLSCRAPE | tasktype ); 88 workqueue_pushtask( sock, TASK_FULLSCRAPE | tasktype );
88} 89}
89 90
90static void fullscrape_write_one( ot_tasktype mode, char **cursor, ot_torrent *torrent, ot_hash *hash ) { 91static void fullscrape_write_one( ot_tasktype mode, char **cursor, ot_torrent *torrent, ot_hash *hash ) {
@@ -170,7 +171,7 @@ static void fullscrape_make(int taskid, ot_tasktype mode) {
170 if (r > re) { 171 if (r > re) {
171 iovector.iov_len = r - (char *)iovector.iov_base; 172 iovector.iov_len = r - (char *)iovector.iov_base;
172 173
173 if (mutex_workqueue_pushchunked(taskid, &iovector)) { 174 if (workqueue_pushchunked(taskid, &iovector)) {
174 free(iovector.iov_base); 175 free(iovector.iov_base);
175 mutex_bucket_unlock(bucket, 0); 176 mutex_bucket_unlock(bucket, 0);
176 return; 177 return;
@@ -200,7 +201,7 @@ static void fullscrape_make(int taskid, ot_tasktype mode) {
200 201
201 /* Send rest of data */ 202 /* Send rest of data */
202 iovector.iov_len = r - (char *)iovector.iov_base; 203 iovector.iov_len = r - (char *)iovector.iov_base;
203 if (mutex_workqueue_pushchunked(taskid, &iovector)) 204 if (workqueue_pushchunked(taskid, &iovector))
204 free(iovector.iov_base); 205 free(iovector.iov_base);
205} 206}
206 207
@@ -251,7 +252,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) {
251 while (!strm.avail_out) { 252 while (!strm.avail_out) {
252 iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base; 253 iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base;
253 254
254 if (mutex_workqueue_pushchunked(taskid, &iovector)) { 255 if (workqueue_pushchunked(taskid, &iovector)) {
255 free(iovector.iov_base); 256 free(iovector.iov_base);
256 mutex_bucket_unlock(bucket, 0); 257 mutex_bucket_unlock(bucket, 0);
257 return; 258 return;
@@ -291,7 +292,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) {
291 fprintf(stderr, "deflate() failed while in fullscrape_make()'s endgame.\n"); 292 fprintf(stderr, "deflate() failed while in fullscrape_make()'s endgame.\n");
292 293
293 iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base; 294 iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base;
294 if (mutex_workqueue_pushchunked(taskid, &iovector)) { 295 if (workqueue_pushchunked(taskid, &iovector)) {
295 free(iovector.iov_base); 296 free(iovector.iov_base);
296 deflateEnd(&strm); 297 deflateEnd(&strm);
297 return; 298 return;
@@ -314,7 +315,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) {
314 315
315 /* Only pass the new buffer if there actually was some data left in the buffer */ 316 /* Only pass the new buffer if there actually was some data left in the buffer */
316 iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base; 317 iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base;
317 if (!iovector.iov_len || mutex_workqueue_pushchunked(taskid, &iovector)) 318 if (!iovector.iov_len || workqueue_pushchunked(taskid, &iovector))
318 free(iovector.iov_base); 319 free(iovector.iov_base);
319 } 320 }
320 321
@@ -380,7 +381,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) {
380 while (outbuf.pos + OT_SCRAPE_MAXENTRYLEN > outbuf.size) { 381 while (outbuf.pos + OT_SCRAPE_MAXENTRYLEN > outbuf.size) {
381 iovector.iov_len = outbuf.size; 382 iovector.iov_len = outbuf.size;
382 383
383 if (mutex_workqueue_pushchunked(taskid, &iovector)) { 384 if (workqueue_pushchunked(taskid, &iovector)) {
384 free(iovector.iov_base); 385 free(iovector.iov_base);
385 ZSTD_freeCCtx(zstream); 386 ZSTD_freeCCtx(zstream);
386 return mutex_bucket_unlock(bucket, 0); 387 return mutex_bucket_unlock(bucket, 0);
@@ -418,7 +419,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) {
418 more_bytes = ZSTD_compressStream2(zstream, &outbuf, &inbuf, ZSTD_e_end); 419 more_bytes = ZSTD_compressStream2(zstream, &outbuf, &inbuf, ZSTD_e_end);
419 420
420 iovector.iov_len = outbuf.pos; 421 iovector.iov_len = outbuf.pos;
421 if (mutex_workqueue_pushchunked(taskid, &iovector)) { 422 if (workqueue_pushchunked(taskid, &iovector)) {
422 free(iovector.iov_base); 423 free(iovector.iov_base);
423 ZSTD_freeCCtx(zstream); 424 ZSTD_freeCCtx(zstream);
424 return; 425 return;
@@ -443,7 +444,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) {
443 444
444 /* Only pass the new buffer if there actually was some data left in the buffer */ 445 /* Only pass the new buffer if there actually was some data left in the buffer */
445 iovector.iov_len = outbuf.pos; 446 iovector.iov_len = outbuf.pos;
446 if (!iovector.iov_len || mutex_workqueue_pushchunked(taskid, &iovector)) 447 if (!iovector.iov_len || workqueue_pushchunked(taskid, &iovector))
447 free(iovector.iov_base); 448 free(iovector.iov_base);
448 } 449 }
449 450
diff --git a/ot_fullscrape.h b/ot_fullscrape.h
index 0e7690c..b759a27 100644
--- a/ot_fullscrape.h
+++ b/ot_fullscrape.h
@@ -9,6 +9,7 @@
9#ifdef WANT_FULLSCRAPE 9#ifdef WANT_FULLSCRAPE
10 10
11#include "ot_mutex.h" 11#include "ot_mutex.h"
12#include "ot_workqueue.h"
12 13
13void fullscrape_init(void); 14void fullscrape_init(void);
14void fullscrape_deinit(void); 15void fullscrape_deinit(void);
diff --git a/ot_http.c b/ot_http.c
index 602ff94..b0e1de3 100644
--- a/ot_http.c
+++ b/ot_http.c
@@ -22,6 +22,7 @@
22#include "ot_iovec.h" 22#include "ot_iovec.h"
23#include "ot_ip.h" 23#include "ot_ip.h"
24#include "ot_mutex.h" 24#include "ot_mutex.h"
25#include "ot_workqueue.h"
25#include "ot_scan.h" 26#include "ot_scan.h"
26#include "ot_event.h" 27#include "ot_event.h"
27#include "ot_fdstate.h" 28#include "ot_fdstate.h"
@@ -77,7 +78,7 @@ void http_close(const int64_t sock) {
77 free(cookie->batch); 78 free(cookie->batch);
78 buf_reset(&cookie->request); 79 buf_reset(&cookie->request);
79 if (cookie->flag & (STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER)) 80 if (cookie->flag & (STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER))
80 mutex_workqueue_canceltask(sock); 81 workqueue_canceltask(sock);
81 if (!(cookie->flag & STRUCT_HTTP_FLAG_ONSTACK)) 82 if (!(cookie->flag & STRUCT_HTTP_FLAG_ONSTACK))
82 free(cookie); 83 free(cookie);
83 } 84 }
diff --git a/ot_mutex.c b/ot_mutex.c
index 5392264..9b3fd24 100644
--- a/ot_mutex.c
+++ b/ot_mutex.c
@@ -30,9 +30,6 @@ static pthread_mutex_t bucket_mutex[OT_BUCKET_COUNT];
30 threads out of that one bucket, so the counter needs its own protection */ 30 threads out of that one bucket, so the counter needs its own protection */
31static _Atomic size_t g_torrent_count; 31static _Atomic size_t g_torrent_count;
32 32
33/* Self pipe from opentracker.c */
34extern int g_self_pipe[2];
35
36ot_vector *mutex_bucket_lock(int bucket) { 33ot_vector *mutex_bucket_lock(int bucket) {
37 pthread_mutex_lock(bucket_mutex + bucket); 34 pthread_mutex_lock(bucket_mutex + bucket);
38 return all_torrents + bucket; 35 return all_torrents + bucket;
@@ -60,237 +57,16 @@ void mutex_bucket_unlock_by_hash(ot_hash const hash, int delta_torrentcount) {
60 57
61size_t mutex_get_torrent_count(void) { return atomic_load_explicit(&g_torrent_count, memory_order_relaxed); } 58size_t mutex_get_torrent_count(void) { return atomic_load_explicit(&g_torrent_count, memory_order_relaxed); }
62 59
63/* TaskQueue Magic */
64
65struct ot_task {
66 ot_taskid taskid;
67 ot_tasktype tasktype;
68 int64_t sock;
69 size_t iovec_entries;
70 struct iovec *iovec;
71 struct ot_task *next;
72};
73
74static ot_taskid next_free_taskid = 1;
75static struct ot_task *tasklist;
76static pthread_mutex_t tasklist_mutex;
77static pthread_cond_t tasklist_being_filled;
78
79int mutex_workqueue_pushtask(int64_t sock, ot_tasktype tasktype) {
80 struct ot_task **tmptask, *task;
81
82 task = malloc(sizeof(struct ot_task));
83 if (!task)
84 return -1;
85
86 task->taskid = 0;
87 task->tasktype = tasktype;
88 task->sock = sock;
89 task->iovec_entries = 0;
90 task->iovec = NULL;
91 task->next = 0;
92
93 /* Want exclusive access to tasklist */
94 pthread_mutex_lock(&tasklist_mutex);
95
96 /* Skip to end of list */
97 tmptask = &tasklist;
98 while (*tmptask)
99 tmptask = &(*tmptask)->next;
100 *tmptask = task;
101
102 /* Inform waiting workers and release lock */
103 pthread_cond_broadcast(&tasklist_being_filled);
104 pthread_mutex_unlock(&tasklist_mutex);
105 return 0;
106}
107
108void mutex_workqueue_canceltask(int64_t sock) {
109 struct ot_task **task;
110
111 /* Want exclusive access to tasklist */
112 pthread_mutex_lock(&tasklist_mutex);
113
114 for (task = &tasklist; *task; task = &((*task)->next))
115 if ((*task)->sock == sock) {
116 struct iovec *iovec = (*task)->iovec;
117 struct ot_task *ptask = *task;
118 size_t i;
119
120 /* Free task's iovec, the buffers it points at and the vector itself */
121 for (i = 0; i < (*task)->iovec_entries; ++i)
122 free(iovec[i].iov_base);
123 free(iovec);
124
125 *task = (*task)->next;
126 free(ptask);
127 break;
128 }
129
130 /* Release lock */
131 pthread_mutex_unlock(&tasklist_mutex);
132}
133
134ot_taskid mutex_workqueue_poptask(ot_tasktype *tasktype) {
135 struct ot_task *task;
136 ot_taskid taskid = 0;
137
138 /* Want exclusive access to tasklist */
139 pthread_mutex_lock(&tasklist_mutex);
140
141 while (!taskid) {
142 /* Skip to the first unassigned task this worker wants to do */
143 for (task = tasklist; task; task = task->next)
144 if (!task->taskid && (TASK_CLASS_MASK & task->tasktype) == *tasktype) {
145 /* If we found an outstanding task, assign a taskid to it
146 and leave the loop */
147 task->taskid = taskid = ++next_free_taskid;
148 *tasktype = task->tasktype;
149 break;
150 }
151
152 /* Wait until the next task is being fed, unless the tracker is on its
153 way out, in which case the worker is meant to leave rather than wait */
154 if (!taskid) {
155 if (!g_opentracker_running)
156 break;
157 pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex);
158 }
159 }
160
161 /* Release lock */
162 pthread_mutex_unlock(&tasklist_mutex);
163
164 return taskid;
165}
166
167void mutex_workqueue_pushsuccess(ot_taskid taskid) {
168 struct ot_task **task;
169
170 /* Want exclusive access to tasklist */
171 pthread_mutex_lock(&tasklist_mutex);
172
173 for (task = &tasklist; *task; task = &((*task)->next))
174 if ((*task)->taskid == taskid) {
175 struct ot_task *ptask = *task;
176 *task = (*task)->next;
177 free(ptask);
178 break;
179 }
180
181 /* Release lock */
182 pthread_mutex_unlock(&tasklist_mutex);
183}
184
185int mutex_workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovec) {
186 struct ot_task *task;
187 const char byte = 'o';
188
189 /* Want exclusive access to tasklist */
190 pthread_mutex_lock(&tasklist_mutex);
191
192 for (task = tasklist; task; task = task->next)
193 if (task->taskid == taskid) {
194 task->iovec_entries = iovec_entries;
195 task->iovec = iovec;
196 task->tasktype = TASK_DONE;
197 break;
198 }
199
200 /* Release lock */
201 pthread_mutex_unlock(&tasklist_mutex);
202
203 /* A full pipe has already done this nudge's job, so a failure is fine */
204 (void)!write(g_self_pipe[1], &byte, 1);
205
206 /* Indicate whether the worker has to throw away results */
207 return task ? 0 : -1;
208}
209
210int mutex_workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec) {
211 struct ot_task *task;
212 const char byte = 'o';
213
214 /* Want exclusive access to tasklist */
215 pthread_mutex_lock(&tasklist_mutex);
216
217 for (task = tasklist; task; task = task->next)
218 if (task->taskid == taskid) {
219 if (iovec) {
220 if (iovec_append(&task->iovec_entries, &task->iovec, iovec))
221 task->tasktype = TASK_DONE_PARTIAL;
222 else
223 task = NULL;
224 } else
225 task->tasktype = TASK_DONE;
226 break;
227 }
228
229 /* Release lock */
230 pthread_mutex_unlock(&tasklist_mutex);
231
232 /* A full pipe has already done this nudge's job, so a failure is fine */
233 (void)!write(g_self_pipe[1], &byte, 1);
234
235 /* Indicate whether the worker has to throw away results */
236 return task ? 0 : -1;
237}
238
239int64_t mutex_workqueue_popresult(size_t *iovec_entries, struct iovec **iovec, int *is_partial) {
240 struct ot_task **task;
241 int64_t sock = -1;
242
243 *is_partial = 0;
244
245 /* Want exclusive access to tasklist */
246 pthread_mutex_lock(&tasklist_mutex);
247
248 for (task = &tasklist; *task; task = &((*task)->next))
249 if (((*task)->tasktype & TASK_CLASS_MASK) == TASK_DONE) {
250 struct ot_task *ptask = *task;
251 *iovec_entries = ptask->iovec_entries;
252 *iovec = ptask->iovec;
253 sock = ptask->sock;
254
255 if ((*task)->tasktype == TASK_DONE) {
256 *task = ptask->next;
257 free(ptask);
258 } else {
259 ptask->iovec_entries = 0;
260 ptask->iovec = NULL;
261 *is_partial = 1;
262 /* Prevent task from showing up immediately again unless new data was added */
263 (*task)->tasktype = TASK_FULLSCRAPE;
264 }
265 break;
266 }
267
268 /* Release lock */
269 pthread_mutex_unlock(&tasklist_mutex);
270 return sock;
271}
272
273void mutex_init(void) { 60void mutex_init(void) {
274 size_t i; 61 size_t i;
275 pthread_mutex_init(&tasklist_mutex, NULL);
276 pthread_cond_init(&tasklist_being_filled, NULL);
277 for (i = 0; i < OT_BUCKET_COUNT; ++i) 62 for (i = 0; i < OT_BUCKET_COUNT; ++i)
278 pthread_mutex_init(bucket_mutex + i, NULL); 63 pthread_mutex_init(bucket_mutex + i, NULL);
279 memset(all_torrents, 0, sizeof(all_torrents)); 64 memset(all_torrents, 0, sizeof(all_torrents));
280} 65}
281 66
282/* Let every worker parked on the queue see that the tracker is stopping */
283void mutex_workqueue_wakeall(void) {
284 pthread_mutex_lock(&tasklist_mutex);
285 pthread_cond_broadcast(&tasklist_being_filled);
286 pthread_mutex_unlock(&tasklist_mutex);
287}
288
289void mutex_deinit(void) { 67void mutex_deinit(void) {
290 size_t i; 68 size_t i;
291 for (i = 0; i < OT_BUCKET_COUNT; ++i) 69 for (i = 0; i < OT_BUCKET_COUNT; ++i)
292 pthread_mutex_destroy(bucket_mutex + i); 70 pthread_mutex_destroy(bucket_mutex + i);
293 pthread_mutex_destroy(&tasklist_mutex);
294 pthread_cond_destroy(&tasklist_being_filled);
295 memset(all_torrents, 0, sizeof(all_torrents)); 71 memset(all_torrents, 0, sizeof(all_torrents));
296} 72}
diff --git a/ot_mutex.h b/ot_mutex.h
index c048fdb..c1218d7 100644
--- a/ot_mutex.h
+++ b/ot_mutex.h
@@ -10,7 +10,6 @@
10#include "trackerlogic.h" 10#include "trackerlogic.h"
11 11
12void mutex_init(void); 12void mutex_init(void);
13void mutex_workqueue_wakeall(void);
14void mutex_deinit(void); 13void mutex_deinit(void);
15 14
16ot_vector *mutex_bucket_lock(int bucket); 15ot_vector *mutex_bucket_lock(int bucket);
@@ -21,60 +20,4 @@ void mutex_bucket_unlock_by_hash(ot_hash const hash, int delta_torrentcoun
21 20
22size_t mutex_get_torrent_count(void); 21size_t mutex_get_torrent_count(void);
23 22
24typedef enum {
25 TASK_STATS_CONNS = 0x0001,
26 TASK_STATS_TCP = 0x0002,
27 TASK_STATS_UDP = 0x0003,
28 TASK_STATS_SCRAPE = 0x0004,
29 TASK_STATS_FULLSCRAPE = 0x0005,
30 TASK_STATS_TPB = 0x0006,
31 TASK_STATS_HTTPERRORS = 0x0007,
32 TASK_STATS_VERSION = 0x0008,
33 TASK_STATS_RENEW = 0x000a,
34 TASK_STATS_SYNCS = 0x000b,
35 TASK_STATS_COMPLETED = 0x000c,
36 TASK_STATS_NUMWANTS = 0x000d,
37
38 TASK_STATS = 0x0100, /* Mask */
39 TASK_STATS_TORRENTS = 0x0101,
40 TASK_STATS_PEERS = 0x0102,
41 TASK_STATS_SLASH24S = 0x0103,
42 TASK_STATS_TOP10 = 0x0104,
43 TASK_STATS_TOP100 = 0x0105,
44 TASK_STATS_EVERYTHING = 0x0106,
45 TASK_STATS_FULLLOG = 0x0107,
46 TASK_STATS_WOODPECKERS = 0x0108,
47
48 TASK_FULLSCRAPE = 0x0200, /* Default mode */
49 TASK_FULLSCRAPE_TPB_BINARY = 0x0201,
50 TASK_FULLSCRAPE_TPB_ASCII = 0x0202,
51 TASK_FULLSCRAPE_TPB_ASCII_PLUS = 0x0203,
52 TASK_FULLSCRAPE_TPB_URLENCODED = 0x0204,
53 TASK_FULLSCRAPE_TRACKERSTATE = 0x0205,
54
55 TASK_DMEM = 0x0300,
56
57 TASK_DONE = 0x0f00,
58 TASK_DONE_PARTIAL = 0x0f01,
59
60 TASK_FLAG_GZIP = 0x1000,
61 TASK_FLAG_BZIP2 = 0x2000,
62 TASK_FLAG_ZSTD = 0x4000,
63 TASK_FLAG_CHUNKED = 0x8000,
64
65 TASK_TASK_MASK = 0x0fff,
66 TASK_CLASS_MASK = 0x0f00,
67 TASK_FLAGS_MASK = 0xf000
68} ot_tasktype;
69
70typedef unsigned long ot_taskid;
71
72int mutex_workqueue_pushtask(int64_t sock, ot_tasktype tasktype);
73void mutex_workqueue_canceltask(int64_t sock);
74void mutex_workqueue_pushsuccess(ot_taskid taskid);
75ot_taskid mutex_workqueue_poptask(ot_tasktype *tasktype);
76int mutex_workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovector);
77int mutex_workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec);
78int64_t mutex_workqueue_popresult(size_t *iovec_entries, struct iovec **iovector, int *is_partial);
79
80#endif 23#endif
diff --git a/ot_stats.c b/ot_stats.c
index 5f97a2f..d38355a 100644
--- a/ot_stats.c
+++ b/ot_stats.c
@@ -23,6 +23,7 @@
23#include "ot_iovec.h" 23#include "ot_iovec.h"
24#include "ot_ip.h" 24#include "ot_ip.h"
25#include "ot_mutex.h" 25#include "ot_mutex.h"
26#include "ot_workqueue.h"
26#include "ot_fmt.h" 27#include "ot_fmt.h"
27#include "ot_stats.h" 28#include "ot_stats.h"
28#include "trackerlogic.h" 29#include "trackerlogic.h"
@@ -803,17 +804,17 @@ static void *stats_worker(void *args) {
803 804
804 while (g_opentracker_running) { 805 while (g_opentracker_running) {
805 ot_tasktype tasktype = TASK_STATS; 806 ot_tasktype tasktype = TASK_STATS;
806 ot_taskid taskid = mutex_workqueue_poptask(&tasktype); 807 ot_taskid taskid = workqueue_poptask(&tasktype);
807 if (!taskid) 808 if (!taskid)
808 break; 809 break;
809 stats_make(&iovec_entries, &iovector, tasktype); 810 stats_make(&iovec_entries, &iovector, tasktype);
810 if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector)) 811 if (workqueue_pushresult(taskid, iovec_entries, iovector))
811 iovec_free(&iovec_entries, &iovector); 812 iovec_free(&iovec_entries, &iovector);
812 } 813 }
813 return NULL; 814 return NULL;
814} 815}
815 816
816void stats_deliver(int64_t sock, int tasktype) { mutex_workqueue_pushtask(sock, tasktype); } 817void stats_deliver(int64_t sock, int tasktype) { workqueue_pushtask(sock, tasktype); }
817 818
818static pthread_t thread_id; 819static pthread_t thread_id;
819void stats_init(void) { 820void stats_init(void) {
@@ -822,6 +823,6 @@ void stats_init(void) {
822} 823}
823 824
824void stats_deinit(void) { 825void stats_deinit(void) {
825 mutex_workqueue_wakeall(); 826 workqueue_wakeall();
826 pthread_join(thread_id, NULL); 827 pthread_join(thread_id, NULL);
827} 828}
diff --git a/ot_sync.c b/ot_sync.c
index 3668aa1..c19cf36 100644
--- a/ot_sync.c
+++ b/ot_sync.c
@@ -14,6 +14,7 @@
14#include "ot_fmt.h" 14#include "ot_fmt.h"
15#include "ot_iovec.h" 15#include "ot_iovec.h"
16#include "ot_mutex.h" 16#include "ot_mutex.h"
17#include "ot_workqueue.h"
17#include "ot_scan.h" 18#include "ot_scan.h"
18#include "ot_stats.h" 19#include "ot_stats.h"
19#include "ot_sync.h" 20#include "ot_sync.h"
@@ -139,10 +140,10 @@ static void *sync_worker(void *args) {
139 140
140 while (1) { 141 while (1) {
141 ot_tasktype tasktype = TASK_SYNC_OUT; 142 ot_tasktype tasktype = TASK_SYNC_OUT;
142 ot_taskid taskid = mutex_workqueue_poptask(&tasktype); 143 ot_taskid taskid = workqueue_poptask(&tasktype);
143 sync_make(&iovec_entries, &iovector); 144 sync_make(&iovec_entries, &iovector);
144 stats_issue_event(ws, EVENT_SYNC_OUT, FLAG_TCP, iovec_length(&iovec_entries, &iovector)); 145 stats_issue_event(ws, EVENT_SYNC_OUT, FLAG_TCP, iovec_length(&iovec_entries, &iovector));
145 if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector)) 146 if (workqueue_pushresult(taskid, iovec_entries, iovector))
146 iovec_free(&iovec_entries, &iovector); 147 iovec_free(&iovec_entries, &iovector);
147 } 148 }
148 return NULL; 149 return NULL;
@@ -158,7 +159,7 @@ void sync_deinit( ) {
158} 159}
159 160
160void sync_deliver( int64_t socket ) { 161void sync_deliver( int64_t socket ) {
161 mutex_workqueue_pushtask( socket, TASK_SYNC_OUT ); 162 workqueue_pushtask( socket, TASK_SYNC_OUT );
162} 163}
163 164
164#endif 165#endif
diff --git a/ot_tcp.c b/ot_tcp.c
index 47697ca..4a5008b 100644
--- a/ot_tcp.c
+++ b/ot_tcp.c
@@ -20,6 +20,7 @@
20#include "ot_http.h" 20#include "ot_http.h"
21#include "ot_mutex.h" 21#include "ot_mutex.h"
22#include "ot_socket.h" 22#include "ot_socket.h"
23#include "ot_worker.h"
23#include "ot_stats.h" 24#include "ot_stats.h"
24#include "ot_tcp.h" 25#include "ot_tcp.h"
25#include "trackerlogic.h" 26#include "trackerlogic.h"
@@ -164,75 +165,18 @@ void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws) {
164 } 165 }
165} 166}
166 167
167/* One of these per worker, each allocated on its own so that growing the
168 array below never moves the one a running worker holds. */
169struct tcp_worker {
170 pthread_t thread;
171 int64_t sock;
172 _Atomic int left;
173};
174
175static struct tcp_worker **g_workers;
176static size_t g_worker_count;
177
178/* A worker blocks in accept and nowhere else: the connection it is handed is 168/* A worker blocks in accept and nowhere else: the connection it is handed is
179 non blocking, so a client that stops talking is given to the event loop 169 non blocking, so a client that stops talking is given to the event loop
180 rather than keeping this thread. */ 170 rather than keeping this thread. */
181static void *tcp_worker(void *args) { 171static void tcp_run(struct ot_worker *me) {
182 struct tcp_worker *me = args;
183 struct ot_workstruct ws;
184
185 if (workstruct_init(&ws) == -1) {
186 fprintf(stderr, "Warning: a tcp worker could not get its buffers and will not run.\n");
187 atomic_store(&me->left, 1);
188 return NULL;
189 }
190
191 while (g_opentracker_running) 172 while (g_opentracker_running)
192 tcp_accept(me->sock, &ws); 173 tcp_accept(me->sock, &me->ws);
193
194 workstruct_deinit(&ws);
195 atomic_store(&me->left, 1);
196 return NULL;
197} 174}
198 175
199void tcp_worker_install(int64_t sock) { 176void tcp_worker_install(int64_t sock) {
200 struct tcp_worker **grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers));
201 struct tcp_worker *worker;
202
203 if (!grown || !(worker = calloc(1, sizeof(*worker)))) {
204 fprintf(stderr, "Warning: out of memory installing a tcp worker, this socket stays with the workers already running.\n");
205 g_workers = grown ? grown : g_workers;
206 return;
207 }
208 g_workers = grown;
209 worker->sock = sock;
210
211 socket_set_blocking((int)sock, 1); 177 socket_set_blocking((int)sock, 1);
212 socket_defer_accept((int)sock); 178 socket_defer_accept((int)sock);
213 179
214 if (pthread_create(&worker->thread, NULL, tcp_worker, worker)) 180 if (worker_start(tcp_run, sock) == -1)
215 free(worker); 181 fprintf(stderr, "Warning: could not install a tcp worker, this socket stays with the workers already running.\n");
216 else
217 g_workers[g_worker_count++] = worker;
218}
219
220void tcp_workers_deinit(void) {
221 size_t i;
222
223 for (i = 0; i < g_worker_count; ++i) {
224 struct tcp_worker *worker = g_workers[i];
225
226 /* Repeated because a signal only interrupts a call already under way */
227 while (!atomic_load(&worker->left)) {
228 pthread_kill(worker->thread, OT_SIGNAL_WAKE);
229 usleep(1000);
230 }
231 pthread_join(worker->thread, NULL);
232 free(worker);
233 }
234
235 free(g_workers);
236 g_workers = NULL;
237 g_worker_count = 0;
238} 182}
diff --git a/ot_tcp.h b/ot_tcp.h
index 96b6ec1..55a172c 100644
--- a/ot_tcp.h
+++ b/ot_tcp.h
@@ -22,7 +22,5 @@ void tcp_write(const int64_t sock);
22/* One thread accepting on one socket. Called once per worker. */ 22/* One thread accepting on one socket. Called once per worker. */
23void tcp_worker_install(int64_t sock); 23void tcp_worker_install(int64_t sock);
24 24
25/* Stop every worker installed above and wait for it to be gone */
26void tcp_workers_deinit(void);
27 25
28#endif 26#endif
diff --git a/ot_udp.c b/ot_udp.c
index 00ec566..309cbdb 100644
--- a/ot_udp.c
+++ b/ot_udp.c
@@ -16,21 +16,11 @@
16#include "ot_random.h" 16#include "ot_random.h"
17#include "ot_rijndael.h" 17#include "ot_rijndael.h"
18#include "ot_socket.h" 18#include "ot_socket.h"
19#include "ot_worker.h"
19#include "ot_stats.h" 20#include "ot_stats.h"
20#include "ot_udp.h" 21#include "ot_udp.h"
21#include "trackerlogic.h" 22#include "trackerlogic.h"
22 23
23/* One of these per worker, each allocated on its own so that growing the
24 array below never moves the one a running worker holds. */
25struct udp_worker {
26 pthread_t thread;
27 int64_t sock;
28 _Atomic int left;
29};
30
31static struct udp_worker **g_workers;
32static size_t g_worker_count;
33
34static uint32_t g_rijndael_round_key[44] = {0}; 24static uint32_t g_rijndael_round_key[44] = {0};
35static volatile uint32_t g_key_of_the_hour[2] = {0}; 25static volatile uint32_t g_key_of_the_hour[2] = {0};
36static volatile ot_time g_hour_of_the_key; 26static volatile ot_time g_hour_of_the_key;
@@ -217,22 +207,9 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) {
217 return 1; 207 return 1;
218} 208}
219 209
220static void *udp_worker(void *args) { 210static void udp_run(struct ot_worker *me) {
221 struct udp_worker *me = args;
222 struct ot_workstruct ws;
223
224 if (workstruct_init(&ws) == -1) {
225 fprintf(stderr, "Warning: a udp worker could not get its buffers and will not run.\n");
226 atomic_store(&me->left, 1);
227 return NULL;
228 }
229
230 while (g_opentracker_running) 211 while (g_opentracker_running)
231 handle_udp6(me->sock, &ws); 212 handle_udp6(me->sock, &me->ws);
232
233 workstruct_deinit(&ws);
234 atomic_store(&me->left, 1);
235 return NULL;
236} 213}
237 214
238void udp_init(void) { 215void udp_init(void) {
@@ -241,47 +218,11 @@ void udp_init(void) {
241} 218}
242 219
243void udp_worker_install(int64_t sock) { 220void udp_worker_install(int64_t sock) {
244 struct udp_worker **grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers));
245 struct udp_worker *worker;
246
247 if (!grown || !(worker = calloc(1, sizeof(*worker)))) {
248 fprintf(stderr, "Warning: out of memory installing a udp worker, this socket stays with the workers already running.\n");
249 g_workers = grown ? grown : g_workers;
250 return;
251 }
252 g_workers = grown;
253 worker->sock = sock;
254
255#ifdef _DEBUG 221#ifdef _DEBUG
256 fprintf(stderr, "Installing a worker on udp socket %ld\n", (unsigned long)sock); 222 fprintf(stderr, "Installing a worker on udp socket %ld\n", (unsigned long)sock);
257#endif 223#endif
258 socket_set_blocking((int)sock, 1); 224 socket_set_blocking((int)sock, 1);
259 225
260 if (pthread_create(&worker->thread, NULL, udp_worker, worker)) 226 if (worker_start(udp_run, sock) == -1)
261 free(worker); 227 fprintf(stderr, "Warning: could not install a udp worker, this socket stays with the workers already running.\n");
262 else
263 g_workers[g_worker_count++] = worker;
264} 228}
265
266void udp_workers_deinit(void) {
267 size_t i;
268
269 for (i = 0; i < g_worker_count; ++i) {
270 struct udp_worker *worker = g_workers[i];
271
272 /* Repeated because a signal only interrupts a read already under way: a
273 worker that was between two reads when the first one arrived is inside
274 the next one by now. */
275 while (!atomic_load(&worker->left)) {
276 pthread_kill(worker->thread, OT_SIGNAL_WAKE);
277 usleep(1000);
278 }
279 pthread_join(worker->thread, NULL);
280 free(worker);
281 }
282
283 free(g_workers);
284 g_workers = NULL;
285 g_worker_count = 0;
286}
287
diff --git a/ot_udp.h b/ot_udp.h
index b731aca..1cd363a 100644
--- a/ot_udp.h
+++ b/ot_udp.h
@@ -14,8 +14,6 @@ void udp_init(void);
14/* One thread reading one socket. Called once per worker. */ 14/* One thread reading one socket. Called once per worker. */
15void udp_worker_install(int64_t sock); 15void udp_worker_install(int64_t sock);
16 16
17/* Stop every worker installed above and wait for it to be gone */
18void udp_workers_deinit(void);
19int handle_udp6(int64_t serversocket, struct ot_workstruct *ws); 17int handle_udp6(int64_t serversocket, struct ot_workstruct *ws);
20 18
21#endif 19#endif
diff --git a/ot_worker.c b/ot_worker.c
new file mode 100644
index 0000000..97eaeb2
--- /dev/null
+++ b/ot_worker.c
@@ -0,0 +1,75 @@
1/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
2 It is considered beerware. Prost. Skol. Cheers or whatever.
3
4 $id$ */
5
6/* System */
7#include <pthread.h>
8#include <stdatomic.h>
9#include <stdio.h>
10#include <stdlib.h>
11#include <unistd.h>
12
13/* Opentracker */
14#include "ot_worker.h"
15#include "trackerlogic.h"
16
17static struct ot_worker **g_workers;
18static size_t g_worker_count;
19
20/* Everything a worker needs doing around its own loop, in one place, so that
21 a new kind of worker cannot miss a piece of it. */
22static void *worker_trampoline(void *args) {
23 struct ot_worker *me = args;
24
25 if (workstruct_init(&me->ws) == -1)
26 fprintf(stderr, "Warning: a worker could not get its buffers and will not run.\n");
27 else {
28 me->run(me);
29 workstruct_deinit(&me->ws);
30 }
31
32 atomic_store(&me->left, 1);
33 return NULL;
34}
35
36int worker_start(ot_worker_run run, int64_t sock) {
37 struct ot_worker **grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers));
38 struct ot_worker *worker;
39
40 if (!grown)
41 return -1;
42 g_workers = grown;
43
44 if (!(worker = calloc(1, sizeof(*worker))))
45 return -1;
46 worker->sock = sock;
47 worker->run = run;
48
49 if (pthread_create(&worker->thread, NULL, worker_trampoline, worker)) {
50 free(worker);
51 return -1;
52 }
53 g_workers[g_worker_count++] = worker;
54 return 0;
55}
56
57void worker_stop_all(void) {
58 size_t i;
59
60 for (i = 0; i < g_worker_count; ++i) {
61 struct ot_worker *worker = g_workers[i];
62
63 /* Repeated because a signal only interrupts a call already under way */
64 while (!atomic_load(&worker->left)) {
65 pthread_kill(worker->thread, OT_SIGNAL_WAKE);
66 usleep(1000);
67 }
68 pthread_join(worker->thread, NULL);
69 free(worker);
70 }
71
72 free(g_workers);
73 g_workers = NULL;
74 g_worker_count = 0;
75}
diff --git a/ot_worker.h b/ot_worker.h
new file mode 100644
index 0000000..ea4f379
--- /dev/null
+++ b/ot_worker.h
@@ -0,0 +1,39 @@
1/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
2 It is considered beerware. Prost. Skol. Cheers or whatever.
3
4 $id$ */
5
6#ifndef OT_WORKER_H__
7#define OT_WORKER_H__
8
9#include <pthread.h>
10#include <stdatomic.h>
11#include <stdint.h>
12
13#include "trackerlogic.h"
14
15struct ot_worker;
16
17/* What a worker does between being started and being told to stop. It reads
18 worker->sock and works out of worker->ws; everything around it, from the
19 buffers to the counters to saying goodbye, is handled here. */
20typedef void (*ot_worker_run)(struct ot_worker *worker);
21
22/* A thread that serves requests on a socket of its own, with the buffers and
23 counters it needs. Each is allocated on its own, so taking on another one
24 never moves the record a running worker holds. */
25struct ot_worker {
26 pthread_t thread;
27 _Atomic int left;
28 int64_t sock;
29 ot_worker_run run;
30 struct ot_workstruct ws;
31};
32
33/* Start one. Returns -1 if it could not be started. */
34int worker_start(ot_worker_run run, int64_t sock);
35
36/* Wake every worker until it has left, then join it */
37void worker_stop_all(void);
38
39#endif
diff --git a/ot_workqueue.c b/ot_workqueue.c
new file mode 100644
index 0000000..aac366c
--- /dev/null
+++ b/ot_workqueue.c
@@ -0,0 +1,243 @@
1/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
2 It is considered beerware. Prost. Skol. Cheers or whatever.
3
4 $id$ */
5
6/* System */
7#include <pthread.h>
8#include <stdlib.h>
9#include <string.h>
10#include <sys/uio.h>
11#include <unistd.h>
12
13/* Opentracker */
14#include "ot_iovec.h"
15#include "ot_workqueue.h"
16#include "trackerlogic.h"
17
18/* Pushing a result wakes the thread that will send it */
19extern int g_self_pipe[2];
20
21struct ot_task {
22 ot_taskid taskid;
23 ot_tasktype tasktype;
24 int64_t sock;
25 size_t iovec_entries;
26 struct iovec *iovec;
27 struct ot_task *next;
28};
29
30static ot_taskid next_free_taskid = 1;
31static struct ot_task *tasklist;
32static pthread_mutex_t tasklist_mutex;
33static pthread_cond_t tasklist_being_filled;
34
35int workqueue_pushtask(int64_t sock, ot_tasktype tasktype) {
36 struct ot_task **tmptask, *task;
37
38 task = malloc(sizeof(struct ot_task));
39 if (!task)
40 return -1;
41
42 task->taskid = 0;
43 task->tasktype = tasktype;
44 task->sock = sock;
45 task->iovec_entries = 0;
46 task->iovec = NULL;
47 task->next = 0;
48
49 /* Want exclusive access to tasklist */
50 pthread_mutex_lock(&tasklist_mutex);
51
52 /* Skip to end of list */
53 tmptask = &tasklist;
54 while (*tmptask)
55 tmptask = &(*tmptask)->next;
56 *tmptask = task;
57
58 /* Inform waiting workers and release lock */
59 pthread_cond_broadcast(&tasklist_being_filled);
60 pthread_mutex_unlock(&tasklist_mutex);
61 return 0;
62}
63
64void workqueue_canceltask(int64_t sock) {
65 struct ot_task **task;
66
67 /* Want exclusive access to tasklist */
68 pthread_mutex_lock(&tasklist_mutex);
69
70 for (task = &tasklist; *task; task = &((*task)->next))
71 if ((*task)->sock == sock) {
72 struct iovec *iovec = (*task)->iovec;
73 struct ot_task *ptask = *task;
74 size_t i;
75
76 /* Free task's iovec, the buffers it points at and the vector itself */
77 for (i = 0; i < (*task)->iovec_entries; ++i)
78 free(iovec[i].iov_base);
79 free(iovec);
80
81 *task = (*task)->next;
82 free(ptask);
83 break;
84 }
85
86 /* Release lock */
87 pthread_mutex_unlock(&tasklist_mutex);
88}
89
90ot_taskid workqueue_poptask(ot_tasktype *tasktype) {
91 struct ot_task *task;
92 ot_taskid taskid = 0;
93
94 /* Want exclusive access to tasklist */
95 pthread_mutex_lock(&tasklist_mutex);
96
97 while (!taskid) {
98 /* Skip to the first unassigned task this worker wants to do */
99 for (task = tasklist; task; task = task->next)
100 if (!task->taskid && (TASK_CLASS_MASK & task->tasktype) == *tasktype) {
101 /* If we found an outstanding task, assign a taskid to it
102 and leave the loop */
103 task->taskid = taskid = ++next_free_taskid;
104 *tasktype = task->tasktype;
105 break;
106 }
107
108 /* Wait until the next task is being fed, unless the tracker is on its
109 way out, in which case the worker is meant to leave rather than wait */
110 if (!taskid) {
111 if (!g_opentracker_running)
112 break;
113 pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex);
114 }
115 }
116
117 /* Release lock */
118 pthread_mutex_unlock(&tasklist_mutex);
119
120 return taskid;
121}
122
123void workqueue_pushsuccess(ot_taskid taskid) {
124 struct ot_task **task;
125
126 /* Want exclusive access to tasklist */
127 pthread_mutex_lock(&tasklist_mutex);
128
129 for (task = &tasklist; *task; task = &((*task)->next))
130 if ((*task)->taskid == taskid) {
131 struct ot_task *ptask = *task;
132 *task = (*task)->next;
133 free(ptask);
134 break;
135 }
136
137 /* Release lock */
138 pthread_mutex_unlock(&tasklist_mutex);
139}
140
141int workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovec) {
142 struct ot_task *task;
143 const char byte = 'o';
144
145 /* Want exclusive access to tasklist */
146 pthread_mutex_lock(&tasklist_mutex);
147
148 for (task = tasklist; task; task = task->next)
149 if (task->taskid == taskid) {
150 task->iovec_entries = iovec_entries;
151 task->iovec = iovec;
152 task->tasktype = TASK_DONE;
153 break;
154 }
155
156 /* Release lock */
157 pthread_mutex_unlock(&tasklist_mutex);
158
159 /* A full pipe has already done this nudge's job, so a failure is fine */
160 (void)!write(g_self_pipe[1], &byte, 1);
161
162 /* Indicate whether the worker has to throw away results */
163 return task ? 0 : -1;
164}
165
166int workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec) {
167 struct ot_task *task;
168 const char byte = 'o';
169
170 /* Want exclusive access to tasklist */
171 pthread_mutex_lock(&tasklist_mutex);
172
173 for (task = tasklist; task; task = task->next)
174 if (task->taskid == taskid) {
175 if (iovec) {
176 if (iovec_append(&task->iovec_entries, &task->iovec, iovec))
177 task->tasktype = TASK_DONE_PARTIAL;
178 else
179 task = NULL;
180 } else
181 task->tasktype = TASK_DONE;
182 break;
183 }
184
185 /* Release lock */
186 pthread_mutex_unlock(&tasklist_mutex);
187
188 /* A full pipe has already done this nudge's job, so a failure is fine */
189 (void)!write(g_self_pipe[1], &byte, 1);
190
191 /* Indicate whether the worker has to throw away results */
192 return task ? 0 : -1;
193}
194
195int64_t workqueue_popresult(size_t *iovec_entries, struct iovec **iovec, int *is_partial) {
196 struct ot_task **task;
197 int64_t sock = -1;
198
199 *is_partial = 0;
200
201 /* Want exclusive access to tasklist */
202 pthread_mutex_lock(&tasklist_mutex);
203
204 for (task = &tasklist; *task; task = &((*task)->next))
205 if (((*task)->tasktype & TASK_CLASS_MASK) == TASK_DONE) {
206 struct ot_task *ptask = *task;
207 *iovec_entries = ptask->iovec_entries;
208 *iovec = ptask->iovec;
209 sock = ptask->sock;
210
211 if ((*task)->tasktype == TASK_DONE) {
212 *task = ptask->next;
213 free(ptask);
214 } else {
215 ptask->iovec_entries = 0;
216 ptask->iovec = NULL;
217 *is_partial = 1;
218 /* Prevent task from showing up immediately again unless new data was added */
219 (*task)->tasktype = TASK_FULLSCRAPE;
220 }
221 break;
222 }
223
224 /* Release lock */
225 pthread_mutex_unlock(&tasklist_mutex);
226 return sock;
227}
228
229void workqueue_wakeall(void) {
230 pthread_mutex_lock(&tasklist_mutex);
231 pthread_cond_broadcast(&tasklist_being_filled);
232 pthread_mutex_unlock(&tasklist_mutex);
233}
234
235void workqueue_init(void) {
236 pthread_mutex_init(&tasklist_mutex, NULL);
237 pthread_cond_init(&tasklist_being_filled, NULL);
238}
239
240void workqueue_deinit(void) {
241 pthread_mutex_destroy(&tasklist_mutex);
242 pthread_cond_destroy(&tasklist_being_filled);
243}
diff --git a/ot_workqueue.h b/ot_workqueue.h
new file mode 100644
index 0000000..f99a7b5
--- /dev/null
+++ b/ot_workqueue.h
@@ -0,0 +1,78 @@
1/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
2 It is considered beerware. Prost. Skol. Cheers or whatever.
3
4 $id$ */
5
6#ifndef OT_WORKQUEUE_H__
7#define OT_WORKQUEUE_H__
8
9#include <stdint.h>
10#include <sys/uio.h>
11
12#include "trackerlogic.h"
13
14/* Work the tracker hands to a thread of its own: a full scrape or an
15 expensive stats page, which take too long to build while a client waits.
16 The task goes in here, the reply comes back out. */
17
18typedef enum {
19 TASK_STATS_CONNS = 0x0001,
20 TASK_STATS_TCP = 0x0002,
21 TASK_STATS_UDP = 0x0003,
22 TASK_STATS_SCRAPE = 0x0004,
23 TASK_STATS_FULLSCRAPE = 0x0005,
24 TASK_STATS_TPB = 0x0006,
25 TASK_STATS_HTTPERRORS = 0x0007,
26 TASK_STATS_VERSION = 0x0008,
27 TASK_STATS_RENEW = 0x000a,
28 TASK_STATS_SYNCS = 0x000b,
29 TASK_STATS_COMPLETED = 0x000c,
30 TASK_STATS_NUMWANTS = 0x000d,
31
32 TASK_STATS = 0x0100, /* Mask */
33 TASK_STATS_TORRENTS = 0x0101,
34 TASK_STATS_PEERS = 0x0102,
35 TASK_STATS_SLASH24S = 0x0103,
36 TASK_STATS_TOP10 = 0x0104,
37 TASK_STATS_TOP100 = 0x0105,
38 TASK_STATS_EVERYTHING = 0x0106,
39 TASK_STATS_FULLLOG = 0x0107,
40 TASK_STATS_WOODPECKERS = 0x0108,
41
42 TASK_FULLSCRAPE = 0x0200, /* Default mode */
43 TASK_FULLSCRAPE_TPB_BINARY = 0x0201,
44 TASK_FULLSCRAPE_TPB_ASCII = 0x0202,
45 TASK_FULLSCRAPE_TPB_ASCII_PLUS = 0x0203,
46 TASK_FULLSCRAPE_TPB_URLENCODED = 0x0204,
47 TASK_FULLSCRAPE_TRACKERSTATE = 0x0205,
48
49 TASK_DMEM = 0x0300,
50
51 TASK_DONE = 0x0f00,
52 TASK_DONE_PARTIAL = 0x0f01,
53
54 TASK_FLAG_GZIP = 0x1000,
55 TASK_FLAG_BZIP2 = 0x2000,
56 TASK_FLAG_ZSTD = 0x4000,
57 TASK_FLAG_CHUNKED = 0x8000,
58
59 TASK_TASK_MASK = 0x0fff,
60 TASK_CLASS_MASK = 0x0f00,
61 TASK_FLAGS_MASK = 0xf000
62} ot_tasktype;
63
64typedef unsigned long ot_taskid;
65
66int workqueue_pushtask(int64_t sock, ot_tasktype tasktype);
67void workqueue_canceltask(int64_t sock);
68void workqueue_pushsuccess(ot_taskid taskid);
69ot_taskid workqueue_poptask(ot_tasktype *tasktype);
70int workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovector);
71int workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec);
72int64_t workqueue_popresult(size_t *iovec_entries, struct iovec **iovector, int *is_partial);
73
74void workqueue_init(void);
75void workqueue_deinit(void);
76void workqueue_wakeall(void);
77
78#endif
diff --git a/tests/integration/test_dynamic_accesslist.py b/tests/integration/test_dynamic_accesslist.py
new file mode 100644
index 0000000..7fa5fde
--- /dev/null
+++ b/tests/integration/test_dynamic_accesslist.py
@@ -0,0 +1,50 @@
1"""Hashes added to and removed from the blacklist while the tracker runs.
2
3The two FIFOs are read by threads of their own, which is also what makes
4this the test that drives those threads through a shutdown.
5"""
6
7import os
8import time
9import unittest
10
11from harness import TrackerTestCase
12
13BLOCKED = bytes(range(1, 21))
14
15
16def as_hex(info_hash):
17 return "".join("%02x" % b for b in info_hash)
18
19
20class DynamicAccesslist(TrackerTestCase):
21 config = ("access.blacklist {dir}/blacklist\n"
22 "access.fifo_add {dir}/add.fifo\n"
23 "access.fifo_delete {dir}/delete.fifo\n")
24 files = {"blacklist": b""}
25
26 def feed(self, name, info_hash):
27 """Writing to the FIFO blocks until the reader has it, so this also
28 tells us the thread is alive."""
29 with open(os.path.join(self.tracker.dir, name), "w") as fifo:
30 fifo.write("%s added by a test\n" % as_hex(info_hash))
31
32 def wait_for(self, info_hash, blocked, patience=5.0):
33 """The reader thread picks the hash up on its own schedule, so this
34 waits for the answer to change rather than assuming it already has."""
35 deadline = time.time() + patience
36 while time.time() < deadline:
37 answer = self.tracker.announce(info_hash)
38 if (b"failure reason" in answer) == blocked:
39 return True
40 time.sleep(0.05)
41 return False
42
43 def test_a_hash_can_be_blocked_and_released_while_running(self):
44 self.assertNotIn(b"failure reason", self.tracker.announce(BLOCKED))
45
46 self.feed("add.fifo", BLOCKED)
47 self.assertTrue(self.wait_for(BLOCKED, blocked=True), "the hash was never blocked")
48
49 self.feed("delete.fifo", BLOCKED)
50 self.assertTrue(self.wait_for(BLOCKED, blocked=False), "the hash was never released")
diff --git a/tests/runner.c b/tests/runner.c
index 71a5e2a..97c6d51 100644
--- a/tests/runner.c
+++ b/tests/runner.c
@@ -28,6 +28,8 @@ static const struct module modules[] = {{"ot_urlencoded", test_urlencoded},
28 {"ot_ip", test_ip}, 28 {"ot_ip", test_ip},
29 {"ot_socket", test_socket}, 29 {"ot_socket", test_socket},
30 {"ot_mutex", test_mutex}, 30 {"ot_mutex", test_mutex},
31 {"trackerlogic", test_trackerlogic},
32 {"ot_workqueue", test_workqueue},
31 {"ot_batch", test_batch}, 33 {"ot_batch", test_batch},
32 {"ot_fdstate", test_fdstate}, 34 {"ot_fdstate", test_fdstate},
33 {"ot_event", test_event}, 35 {"ot_event", test_event},
diff --git a/tests/stubs.c b/tests/stubs.c
index a8e6962..5999e2d 100644
--- a/tests/stubs.c
+++ b/tests/stubs.c
@@ -8,11 +8,9 @@
8 behaviour should be supplied here rather than the test weakened. */ 8 behaviour should be supplied here rather than the test weakened. */
9 9
10/* Opentracker */ 10/* Opentracker */
11#include "ot_stats.h"
11#include "trackerlogic.h" 12#include "trackerlogic.h"
12 13
13/* ot_vector.c calls this from vector_clean_list, which no test drives yet. */
14void free_peerlist(ot_peerlist *peer_list) { (void)peer_list; }
15
16/* ot_accesslist.c is built with WANT_ACCESSLIST_BLACK so that the hash parser 14/* ot_accesslist.c is built with WANT_ACCESSLIST_BLACK so that the hash parser
17 is reachable. These two belong to accesslist_readfile and the reload worker, 15 is reachable. These two belong to accesslist_readfile and the reload worker,
18 neither of which any test drives. */ 16 neither of which any test drives. */
@@ -26,3 +24,32 @@ long io_trywrite(long d, const char *buf, long len) {
26 (void)buf; 24 (void)buf;
27 return len; 25 return len;
28} 26}
27
28/* trackerlogic.c is under test, and starting it up starts every module it
29 leans on. None of them is driven by these tests, so they are answered here
30 rather than linked in and left to run threads of their own. */
31void clean_init(void) {}
32void clean_deinit(void) {}
33int clean_single_torrent(ot_torrent *torrent) {
34 (void)torrent;
35 return 0;
36}
37void fullscrape_init(void) {}
38void fullscrape_deinit(void) {}
39void livesync_init(void) {}
40void livesync_deinit(void) {}
41void livesync_tell(struct ot_workstruct *ws) { (void)ws; }
42void stats_init(void) {}
43void stats_deinit(void) {}
44void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) {
45 (void)ws;
46 (void)event;
47 (void)proto;
48 (void)event_data;
49}
50void stats_counters_register(struct ot_workstruct *ws) { (void)ws; }
51void stats_counters_retire(struct ot_workstruct *ws) { (void)ws; }
52
53const char *g_stats_path;
54ssize_t g_stats_path_len;
55uint32_t g_tracker_id;
diff --git a/tests/t_mutex.c b/tests/t_mutex.c
index 3307d70..7c1463b 100644
--- a/tests/t_mutex.c
+++ b/tests/t_mutex.c
@@ -4,104 +4,135 @@
4 $id$ */ 4 $id$ */
5 5
6/* System */ 6/* System */
7#include <stdlib.h> 7#include <pthread.h>
8#include <stdatomic.h>
8#include <string.h> 9#include <string.h>
10#include <unistd.h>
9 11
10/* Opentracker */ 12/* Opentracker */
11#include "ot_mutex.h" 13#include "ot_mutex.h"
12#include "test.h" 14#include "test.h"
13 15
14/* A worker asks for a class of task and blocks until one turns up, so a task 16/* Every announce passes through here, and what it has to get right is which
15 whose class matches no worker is never claimed and whoever is waiting for 17 bucket a hash belongs to and that two threads never hold one bucket at
16 its answer waits forever. These cases pin the matching. */ 18 once. Both are checked against real mutexes rather than a stand-in. */
17 19
18static void case_task_goes_to_its_own_class(void) { 20/* A hash that lands in a chosen bucket: the bucket is the top
19 ot_tasktype asked; 21 OT_BUCKET_COUNT_BITS of the first four bytes, read in network order */
20 ot_taskid taskid; 22static void hash_for_bucket(ot_hash hash, int bucket) {
23 memset(hash, 0, sizeof(ot_hash));
24 hash[0] = (uint8_t)(bucket >> (OT_BUCKET_COUNT_BITS - 8));
25 hash[1] = (uint8_t)((bucket << (16 - OT_BUCKET_COUNT_BITS)) & 0xff);
26}
27
28static void case_a_hash_names_its_bucket(void) {
29 static const int buckets[] = {0, 1, 2, 17, 511, 512, OT_BUCKET_COUNT - 1};
30 size_t i;
31
32 test_case("a hash is served by the bucket its leading bits name");
33 for (i = 0; i < sizeof(buckets) / sizeof(buckets[0]); ++i) {
34 ot_hash hash;
35 ot_vector *by_hash, *by_number;
36
37 hash_for_bucket(hash, buckets[i]);
38 by_hash = mutex_bucket_lock_by_hash(hash);
39 mutex_bucket_unlock_by_hash(hash, 0);
21 40
22 test_case("a worker is handed the task of the class it asked for"); 41 by_number = mutex_bucket_lock(buckets[i]);
23 CHECK_INT(mutex_workqueue_pushtask(41, TASK_FULLSCRAPE_TPB_ASCII), 0); 42 mutex_bucket_unlock(buckets[i], 0);
24 asked = TASK_FULLSCRAPE;
25 taskid = mutex_workqueue_poptask(&asked);
26 CHECK(taskid != 0);
27 43
28 test_case("and the full task type, not just its class"); 44 CHECK(by_hash == by_number);
29 CHECK_INT(asked, TASK_FULLSCRAPE_TPB_ASCII); 45 }
30 mutex_workqueue_canceltask(41);
31} 46}
32 47
33static void case_flags_survive_the_queue(void) { 48static void case_every_bucket_is_its_own(void) {
34 ot_tasktype asked; 49 ot_vector *first, *last, *middle;
35 50
36 test_case("flags outside the class mask are carried through"); 51 test_case("no two buckets are the same one");
37 CHECK_INT(mutex_workqueue_pushtask(42, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP), 0); 52 first = mutex_bucket_lock(0);
38 asked = TASK_FULLSCRAPE; 53 mutex_bucket_unlock(0, 0);
39 CHECK(mutex_workqueue_poptask(&asked) != 0); 54 middle = mutex_bucket_lock(OT_BUCKET_COUNT / 2);
40 CHECK_INT(asked, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP); 55 mutex_bucket_unlock(OT_BUCKET_COUNT / 2, 0);
41 mutex_workqueue_canceltask(42); 56 last = mutex_bucket_lock(OT_BUCKET_COUNT - 1);
57 mutex_bucket_unlock(OT_BUCKET_COUNT - 1, 0);
58
59 CHECK(first != middle);
60 CHECK(middle != last);
61 CHECK(first != last);
42} 62}
43 63
44static void case_classes_do_not_cross(void) { 64static void case_the_count_follows_what_unlock_is_told(void) {
45 ot_tasktype asked; 65 size_t before = mutex_get_torrent_count();
46 ot_taskid taskid; 66
67 test_case("the torrent count moves by what unlocking reports, and not otherwise");
68 mutex_bucket_lock(3);
69 mutex_bucket_unlock(3, 0);
70 CHECK_INT((int)(mutex_get_torrent_count() - before), 0);
47 71
48 test_case("a stats worker passes over a fullscrape task"); 72 mutex_bucket_lock(3);
49 CHECK_INT(mutex_workqueue_pushtask(43, TASK_FULLSCRAPE_TPB_BINARY), 0); 73 mutex_bucket_unlock(3, 5);
50 CHECK_INT(mutex_workqueue_pushtask(44, TASK_STATS_PEERS), 0); 74 CHECK_INT((int)(mutex_get_torrent_count() - before), 5);
51 75
52 asked = TASK_STATS; 76 mutex_bucket_lock(3);
53 taskid = mutex_workqueue_poptask(&asked); 77 mutex_bucket_unlock(3, -2);
54 CHECK(taskid != 0); 78 CHECK_INT((int)(mutex_get_torrent_count() - before), 3);
55 CHECK_INT(asked, TASK_STATS_PEERS); 79}
56 80
57 test_case("and leaves it for the one that wants it"); 81static _Atomic int g_other_got_in;
58 asked = TASK_FULLSCRAPE;
59 taskid = mutex_workqueue_poptask(&asked);
60 CHECK(taskid != 0);
61 CHECK_INT(asked, TASK_FULLSCRAPE_TPB_BINARY);
62 82
63 mutex_workqueue_canceltask(43); 83static void *take_a_bucket(void *args) {
64 mutex_workqueue_canceltask(44); 84 int bucket = (int)(intptr_t)args;
85
86 mutex_bucket_lock(bucket);
87 atomic_store(&g_other_got_in, 1);
88 mutex_bucket_unlock(bucket, 0);
89 return NULL;
90}
91
92static int waited_for_entry(int patience_ms) {
93 int waited;
94
95 for (waited = 0; waited < patience_ms && !atomic_load(&g_other_got_in); ++waited)
96 usleep(1000);
97 return atomic_load(&g_other_got_in);
65} 98}
66 99
67static void case_every_task_gets_its_own_id(void) { 100static void case_a_held_bucket_keeps_others_out(void) {
68 ot_tasktype asked; 101 pthread_t other;
69 ot_taskid first, second;
70 102
71 test_case("two queued tasks are handed out under different ids"); 103 test_case("a second thread waits for the bucket this one holds");
72 CHECK_INT(mutex_workqueue_pushtask(45, TASK_FULLSCRAPE), 0); 104 atomic_store(&g_other_got_in, 0);
73 CHECK_INT(mutex_workqueue_pushtask(46, TASK_FULLSCRAPE), 0); 105 mutex_bucket_lock(9);
106 CHECK_INT(pthread_create(&other, NULL, take_a_bucket, (void *)(intptr_t)9), 0);
74 107
75 asked = TASK_FULLSCRAPE; 108 CHECK_INT(waited_for_entry(200), 0);
76 first = mutex_workqueue_poptask(&asked);
77 asked = TASK_FULLSCRAPE;
78 second = mutex_workqueue_poptask(&asked);
79 CHECK(first != 0);
80 CHECK(second != 0);
81 CHECK(first != second);
82 109
83 mutex_workqueue_canceltask(45); 110 mutex_bucket_unlock(9, 0);
84 mutex_workqueue_canceltask(46); 111 CHECK_INT(waited_for_entry(2000), 1);
112 pthread_join(other, NULL);
85} 113}
86 114
87static void case_cancelled_task_is_not_handed_out(void) { 115static void case_another_bucket_is_free_meanwhile(void) {
88 ot_tasktype asked; 116 pthread_t other;
89 117
90 test_case("a cancelled task is gone before any worker sees it"); 118 test_case("and a different bucket is not held up by it");
91 CHECK_INT(mutex_workqueue_pushtask(47, TASK_FULLSCRAPE_TPB_ASCII_PLUS), 0); 119 atomic_store(&g_other_got_in, 0);
92 CHECK_INT(mutex_workqueue_pushtask(48, TASK_FULLSCRAPE), 0); 120 mutex_bucket_lock(9);
93 mutex_workqueue_canceltask(47); 121 CHECK_INT(pthread_create(&other, NULL, take_a_bucket, (void *)(intptr_t)10), 0);
94 122
95 asked = TASK_FULLSCRAPE; 123 CHECK_INT(waited_for_entry(2000), 1);
96 CHECK(mutex_workqueue_poptask(&asked) != 0); 124 pthread_join(other, NULL);
97 CHECK_INT(asked, TASK_FULLSCRAPE); 125 mutex_bucket_unlock(9, 0);
98 mutex_workqueue_canceltask(48);
99} 126}
100 127
101void test_mutex(void) { 128void test_mutex(void) {
102 case_task_goes_to_its_own_class(); 129 mutex_init();
103 case_flags_survive_the_queue(); 130
104 case_classes_do_not_cross(); 131 case_a_hash_names_its_bucket();
105 case_every_task_gets_its_own_id(); 132 case_every_bucket_is_its_own();
106 case_cancelled_task_is_not_handed_out(); 133 case_the_count_follows_what_unlock_is_told();
134 case_a_held_bucket_keeps_others_out();
135 case_another_bucket_is_free_meanwhile();
136
137 mutex_deinit();
107} 138}
diff --git a/tests/t_trackerlogic.c b/tests/t_trackerlogic.c
new file mode 100644
index 0000000..d3d8755
--- /dev/null
+++ b/tests/t_trackerlogic.c
@@ -0,0 +1,243 @@
1/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
2 It is considered beerware. Prost. Skol. Cheers or whatever.
3
4 $id$ */
5
6/* System */
7#include <arpa/inet.h>
8#include <signal.h>
9#include <pthread.h>
10#include <string.h>
11
12/* Opentracker */
13#include "ot_mutex.h"
14#include "trackerlogic.h"
15#include "test.h"
16
17/* The swarm itself: what an announce does to a torrent, and what a scrape
18 says about it afterwards. The counts are read back through the udp scrape
19 because that is the one reply the tracker builds as plain numbers. */
20
21static struct ot_workstruct ws;
22
23static void hash_named(ot_hash hash, const char *name) {
24 memset(hash, 0, sizeof(ot_hash));
25 memcpy(hash, name, strlen(name) < sizeof(ot_hash) ? strlen(name) : sizeof(ot_hash));
26}
27
28/* An IPv4 peer arrives as the v4 mapped form the tracker stores */
29static void peer_v4(ot_peer6 peer, uint8_t last, uint16_t port) {
30 ot_ip6 ip = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 10, 0, 0, 0};
31 uint16_t net_port = htons(port);
32
33 ip[15] = last;
34 memset(peer, 0, sizeof(ot_peer6));
35 OT_SETIP(peer, ip);
36 OT_SETPORT(peer, &net_port);
37 OT_PEERFLAG(peer) = 0;
38}
39
40static void peer_v6(ot_peer6 peer, uint8_t last, uint16_t port) {
41 ot_ip6 ip = {0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
42 uint16_t net_port = htons(port);
43
44 ip[15] = last;
45 memset(peer, 0, sizeof(ot_peer6));
46 OT_SETIP(peer, ip);
47 OT_SETPORT(peer, &net_port);
48 OT_PEERFLAG(peer) = 0;
49}
50
51static void announce(ot_hash *hash, ot_peer6 peer) {
52 ws.hash = hash;
53 memcpy(ws.peer, peer, sizeof(ot_peer6));
54 ws.reply = ws.outbuf;
55 add_peer_to_torrent_and_return_peers(FLAG_TCP, &ws, 50);
56}
57
58/* seeders, downloads completed, leechers */
59static void scrape(ot_hash hash, uint32_t counts[3]) {
60 char reply[12];
61
62 return_udp_scrape_for_torrent(hash, reply);
63 counts[0] = ntohl(((uint32_t *)reply)[0]);
64 counts[1] = ntohl(((uint32_t *)reply)[1]);
65 counts[2] = ntohl(((uint32_t *)reply)[2]);
66}
67
68static void case_an_announce_makes_a_torrent(void) {
69 ot_hash hash;
70 ot_peer6 peer;
71 uint32_t counts[3];
72 size_t before = mutex_get_torrent_count();
73
74 test_case("the first announce for a hash creates its torrent");
75 hash_named(hash, "first-announce");
76 peer_v4(peer, 1, 6881);
77 announce(&hash, peer);
78
79 CHECK_INT((int)(mutex_get_torrent_count() - before), 1);
80 scrape(hash, counts);
81 CHECK_INT((int)counts[2], 1); /* one leecher */
82 CHECK_INT((int)counts[0], 0);
83}
84
85static void case_the_same_peer_twice_is_one_peer(void) {
86 ot_hash hash;
87 ot_peer6 peer;
88 uint32_t counts[3];
89
90 test_case("announcing twice renews a peer rather than adding another");
91 hash_named(hash, "renewing");
92 peer_v4(peer, 2, 6881);
93 announce(&hash, peer);
94 announce(&hash, peer);
95
96 scrape(hash, counts);
97 CHECK_INT((int)counts[2], 1);
98}
99
100static void case_a_port_makes_a_different_peer(void) {
101 ot_hash hash;
102 ot_peer6 peer;
103 uint32_t counts[3];
104
105 test_case("the same address on another port is another peer");
106 hash_named(hash, "two-ports");
107 peer_v4(peer, 3, 6881);
108 announce(&hash, peer);
109 peer_v4(peer, 3, 6882);
110 announce(&hash, peer);
111
112 scrape(hash, counts);
113 CHECK_INT((int)counts[2], 2);
114}
115
116static void case_seeding_is_counted_apart(void) {
117 ot_hash hash;
118 ot_peer6 peer;
119 uint32_t counts[3];
120
121 test_case("a seeder counts as complete, a leecher as incomplete");
122 hash_named(hash, "seed-and-leech");
123 peer_v4(peer, 4, 6881);
124 OT_PEERFLAG(peer) |= PEER_FLAG_SEEDING;
125 announce(&hash, peer);
126 peer_v4(peer, 5, 6881);
127 announce(&hash, peer);
128
129 scrape(hash, counts);
130 CHECK_INT((int)counts[0], 1);
131 CHECK_INT((int)counts[2], 1);
132}
133
134static void case_completion_without_seeding_is_not_believed(void) {
135 ot_hash hash;
136 ot_peer6 peer;
137 uint32_t counts[3];
138
139 test_case("a peer that claims to have completed but is still leeching is not counted");
140 hash_named(hash, "claiming");
141 peer_v4(peer, 6, 6881);
142 OT_PEERFLAG(peer) |= PEER_FLAG_COMPLETED;
143 announce(&hash, peer);
144
145 scrape(hash, counts);
146 CHECK_INT((int)counts[1], 0);
147 CHECK_INT((int)counts[2], 1); /* still a leecher */
148}
149
150static void case_completing_is_counted_once(void) {
151 ot_hash hash;
152 ot_peer6 peer;
153 uint32_t counts[3];
154
155 test_case("a seeder reporting completion is counted once, however often it says so");
156 hash_named(hash, "completing");
157 peer_v4(peer, 7, 6881);
158 OT_PEERFLAG(peer) |= PEER_FLAG_COMPLETED | PEER_FLAG_SEEDING;
159 announce(&hash, peer);
160 announce(&hash, peer);
161 announce(&hash, peer);
162
163 scrape(hash, counts);
164 CHECK_INT((int)counts[1], 1);
165 CHECK_INT((int)counts[0], 1);
166}
167
168static void case_a_peer_can_leave(void) {
169 ot_hash hash;
170 ot_peer6 peer;
171 uint32_t counts[3];
172
173 test_case("a peer that stops is gone from the swarm");
174 hash_named(hash, "leaving");
175 peer_v4(peer, 7, 6881);
176 announce(&hash, peer);
177 peer_v4(peer, 8, 6881);
178 announce(&hash, peer);
179 scrape(hash, counts);
180 CHECK_INT((int)counts[2], 2);
181
182 ws.hash = &hash;
183 memcpy(ws.peer, peer, sizeof(ot_peer6));
184 remove_peer_from_torrent(FLAG_TCP, &ws);
185
186 scrape(hash, counts);
187 CHECK_INT((int)counts[2], 1);
188}
189
190static void case_the_families_are_kept_apart(void) {
191 ot_hash hash;
192 ot_peer6 peer;
193 uint32_t counts[3];
194
195 test_case("a v4 and a v6 peer of one swarm are both counted");
196 hash_named(hash, "dual-stack");
197 peer_v4(peer, 9, 6881);
198 announce(&hash, peer);
199 peer_v6(peer, 9, 6881);
200 announce(&hash, peer);
201
202 scrape(hash, counts);
203 CHECK_INT((int)counts[2], 2);
204}
205
206static void case_a_hash_nobody_announced(void) {
207 ot_hash hash;
208 uint32_t counts[3];
209
210 test_case("a hash nobody has announced scrapes as nothing at all");
211 hash_named(hash, "never-seen");
212 scrape(hash, counts);
213 CHECK_INT((int)counts[0], 0);
214 CHECK_INT((int)counts[1], 0);
215 CHECK_INT((int)counts[2], 0);
216}
217
218void test_trackerlogic(void) {
219 sigset_t signal_mask;
220
221 /* accesslist_init starts a thread that waits for SIGHUP and is stopped by
222 being sent one, so the signal has to be blocked here as the tracker
223 blocks it, or stopping the tracker logic ends this process instead */
224 sigemptyset(&signal_mask);
225 sigaddset(&signal_mask, SIGHUP);
226 pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
227
228 trackerlogic_init();
229 CHECK_INT(workstruct_init(&ws), 0);
230
231 case_an_announce_makes_a_torrent();
232 case_the_same_peer_twice_is_one_peer();
233 case_a_port_makes_a_different_peer();
234 case_seeding_is_counted_apart();
235 case_completion_without_seeding_is_not_believed();
236 case_completing_is_counted_once();
237 case_a_peer_can_leave();
238 case_the_families_are_kept_apart();
239 case_a_hash_nobody_announced();
240
241 workstruct_deinit(&ws);
242 trackerlogic_deinit();
243}
diff --git a/tests/t_workqueue.c b/tests/t_workqueue.c
new file mode 100644
index 0000000..13e0b2e
--- /dev/null
+++ b/tests/t_workqueue.c
@@ -0,0 +1,107 @@
1/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
2 It is considered beerware. Prost. Skol. Cheers or whatever.
3
4 $id$ */
5
6/* System */
7#include <stdlib.h>
8#include <string.h>
9
10/* Opentracker */
11#include "ot_workqueue.h"
12#include "test.h"
13
14/* A worker asks for a class of task and blocks until one turns up, so a task
15 whose class matches no worker is never claimed and whoever is waiting for
16 its answer waits forever. These cases pin the matching. */
17
18static void case_task_goes_to_its_own_class(void) {
19 ot_tasktype asked;
20 ot_taskid taskid;
21
22 test_case("a worker is handed the task of the class it asked for");
23 CHECK_INT(workqueue_pushtask(41, TASK_FULLSCRAPE_TPB_ASCII), 0);
24 asked = TASK_FULLSCRAPE;
25 taskid = workqueue_poptask(&asked);
26 CHECK(taskid != 0);
27
28 test_case("and the full task type, not just its class");
29 CHECK_INT(asked, TASK_FULLSCRAPE_TPB_ASCII);
30 workqueue_canceltask(41);
31}
32
33static void case_flags_survive_the_queue(void) {
34 ot_tasktype asked;
35
36 test_case("flags outside the class mask are carried through");
37 CHECK_INT(workqueue_pushtask(42, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP), 0);
38 asked = TASK_FULLSCRAPE;
39 CHECK(workqueue_poptask(&asked) != 0);
40 CHECK_INT(asked, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP);
41 workqueue_canceltask(42);
42}
43
44static void case_classes_do_not_cross(void) {
45 ot_tasktype asked;
46 ot_taskid taskid;
47
48 test_case("a stats worker passes over a fullscrape task");
49 CHECK_INT(workqueue_pushtask(43, TASK_FULLSCRAPE_TPB_BINARY), 0);
50 CHECK_INT(workqueue_pushtask(44, TASK_STATS_PEERS), 0);
51
52 asked = TASK_STATS;
53 taskid = workqueue_poptask(&asked);
54 CHECK(taskid != 0);
55 CHECK_INT(asked, TASK_STATS_PEERS);
56
57 test_case("and leaves it for the one that wants it");
58 asked = TASK_FULLSCRAPE;
59 taskid = workqueue_poptask(&asked);
60 CHECK(taskid != 0);
61 CHECK_INT(asked, TASK_FULLSCRAPE_TPB_BINARY);
62
63 workqueue_canceltask(43);
64 workqueue_canceltask(44);
65}
66
67static void case_every_task_gets_its_own_id(void) {
68 ot_tasktype asked;
69 ot_taskid first, second;
70
71 test_case("two queued tasks are handed out under different ids");
72 CHECK_INT(workqueue_pushtask(45, TASK_FULLSCRAPE), 0);
73 CHECK_INT(workqueue_pushtask(46, TASK_FULLSCRAPE), 0);
74
75 asked = TASK_FULLSCRAPE;
76 first = workqueue_poptask(&asked);
77 asked = TASK_FULLSCRAPE;
78 second = workqueue_poptask(&asked);
79 CHECK(first != 0);
80 CHECK(second != 0);
81 CHECK(first != second);
82
83 workqueue_canceltask(45);
84 workqueue_canceltask(46);
85}
86
87static void case_cancelled_task_is_not_handed_out(void) {
88 ot_tasktype asked;
89
90 test_case("a cancelled task is gone before any worker sees it");
91 CHECK_INT(workqueue_pushtask(47, TASK_FULLSCRAPE_TPB_ASCII_PLUS), 0);
92 CHECK_INT(workqueue_pushtask(48, TASK_FULLSCRAPE), 0);
93 workqueue_canceltask(47);
94
95 asked = TASK_FULLSCRAPE;
96 CHECK(workqueue_poptask(&asked) != 0);
97 CHECK_INT(asked, TASK_FULLSCRAPE);
98 workqueue_canceltask(48);
99}
100
101void test_workqueue(void) {
102 case_task_goes_to_its_own_class();
103 case_flags_survive_the_queue();
104 case_classes_do_not_cross();
105 case_every_task_gets_its_own_id();
106 case_cancelled_task_is_not_handed_out();
107}
diff --git a/tests/test.h b/tests/test.h
index 0eddc82..d7296cc 100644
--- a/tests/test.h
+++ b/tests/test.h
@@ -87,6 +87,8 @@ void test_buf(void);
87void test_ip(void); 87void test_ip(void);
88void test_socket(void); 88void test_socket(void);
89void test_mutex(void); 89void test_mutex(void);
90void test_trackerlogic(void);
91void test_workqueue(void);
90void test_batch(void); 92void test_batch(void);
91void test_fdstate(void); 93void test_fdstate(void);
92void test_event(void); 94void test_event(void);
diff --git a/trackerlogic.c b/trackerlogic.c
index fc0fa56..b0b7f4e 100644
--- a/trackerlogic.c
+++ b/trackerlogic.c
@@ -20,6 +20,7 @@
20#include "ot_ip.h" 20#include "ot_ip.h"
21#include "ot_livesync.h" 21#include "ot_livesync.h"
22#include "ot_mutex.h" 22#include "ot_mutex.h"
23#include "ot_workqueue.h"
23#include "ot_random.h" 24#include "ot_random.h"
24#include "ot_stats.h" 25#include "ot_stats.h"
25#include "ot_udp.h" 26#include "ot_udp.h"
@@ -560,6 +561,7 @@ void trackerlogic_init(void) {
560 561
561 /* Initialise background worker threads */ 562 /* Initialise background worker threads */
562 mutex_init(); 563 mutex_init();
564 workqueue_init();
563 clean_init(); 565 clean_init();
564 fullscrape_init(); 566 fullscrape_init();
565 accesslist_init(); 567 accesslist_init();
@@ -623,5 +625,6 @@ void trackerlogic_deinit(void) {
623 } 625 }
624 626
625 /* Release mutexes */ 627 /* Release mutexes */
628 workqueue_deinit();
626 mutex_deinit(); 629 mutex_deinit();
627} 630}
generated by cgit v1.3 (git 2.53.0) at 2026-11-01 15:12:54 +0000