summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--Makefile10
-rw-r--r--man4/opentracker.conf.416
-rw-r--r--opentracker.c73
-rw-r--r--opentracker.conf.sample7
-rw-r--r--ot_accesslist.c46
-rw-r--r--ot_accesslist.h2
-rw-r--r--ot_fdstate.c9
-rw-r--r--ot_fullscrape.c25
-rw-r--r--ot_fullscrape.h1
-rw-r--r--ot_http.c41
-rw-r--r--ot_http.h7
-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.c27
-rw-r--r--ot_tcp.h4
-rw-r--r--ot_udp.c78
-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/integration/test_tcpworkers.py67
-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.c32
-rw-r--r--trackerlogic.h6
33 files changed, 1287 insertions, 506 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/man4/opentracker.conf.4 b/man4/opentracker.conf.4
index aa82e1d..1bd1c7a 100644
--- a/man4/opentracker.conf.4
+++ b/man4/opentracker.conf.4
@@ -42,6 +42,14 @@ in the thread that also serves TCP. See
42.Sx TUNING 42.Sx TUNING
43before raising it. 43before raising it.
44.Pp 44.Pp
45.It listen.tcp.workers Ar threads
46Specifies how many threads accept and answer HTTP connections, for every
47listen statement that does not carry a
48.Cm workers
49option of its own. Defaults to 0, which serves them from the event loop. See
50.Sx TUNING
51before raising it.
52.Pp
45.It listen.shared Ar boolean 53.It listen.shared Ar boolean
46Permits another process to bind the same address, for every listen statement 54Permits another process to bind the same address, for every listen statement
47that does not carry a 55that does not carry a
@@ -101,6 +109,14 @@ group only where the answer is yes.
101in the source distribution sweeps the setting against a running tracker. 109in the source distribution sweeps the setting against a running tracker.
102Measure on the machine that will carry the traffic. 110Measure on the machine that will carry the traffic.
103.Pp 111.Pp
112A TCP worker accepts a connection and answers it on its own stack, without
113the event loop seeing it at all; only a connection it cannot finish is handed
114over. One worker serves less than none, because accepting leaves the event
115loop's thread without anything gained in return, so the setting is worth
116either 0 or at least 2.
117.Pa tests/http_load.c
118in the source distribution drives the tracker over HTTP to measure it.
119.Pp
104On FreeBSD the 120On FreeBSD the
105.Xr accf_data 9 121.Xr accf_data 9
106filter has the kernel hold a new connection until data arrives on it, so the 122filter has the kernel hold a new connection until data arrives on it, so the
diff --git a/opentracker.c b/opentracker.c
index 2eaf988..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 */
@@ -109,20 +111,9 @@ static void *server_mainloop(void *args) {
109 111
110 (void)args; 112 (void)args;
111 113
112 /* Initialize our "thread local storage" */ 114 if (workstruct_init(&ws) == -1)
113 memset(&ws, 0, sizeof(ws));
114 stats_counters_register(&ws);
115 ws.inbuf = malloc(G_INBUF_SIZE);
116 ws.outbuf = malloc(G_OUTBUF_SIZE);
117#ifdef _DEBUG_HTTPERROR
118 ws.debugbuf = malloc(G_DEBUGBUF_SIZE);
119#endif
120
121 if (!ws.inbuf || !ws.outbuf)
122 panic("Initializing worker failed"); 115 panic("Initializing worker failed");
123 116
124 random_bytes(ws.rand48_state, sizeof(ws.rand48_state));
125
126 while (g_opentracker_running) { 117 while (g_opentracker_running) {
127 ot_event ready[OT_EVENT_BATCH]; 118 ot_event ready[OT_EVENT_BATCH];
128 int64_t sock; 119 int64_t sock;
@@ -148,16 +139,18 @@ static void *server_mainloop(void *args) {
148 else if ((intptr_t)cookie == FLAG_SELFPIPE) 139 else if ((intptr_t)cookie == FLAG_SELFPIPE)
149 while (read((int)sock, ws.inbuf, G_INBUF_SIZE) > 0) 140 while (read((int)sock, ws.inbuf, G_INBUF_SIZE) > 0)
150 ; 141 ;
151 else { 142 else if (fdstate_claim((int)sock, ready[i].generation)) {
152 if (ready[i].what & EVENT_IN) 143 if (ready[i].what & EVENT_IN)
153 tcp_read(sock, &ws); 144 tcp_read(sock, &ws);
154 /* The read may have closed it, so look again before writing */ 145 /* The read may have closed it, so look again before writing */
155 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)
156 tcp_write(sock); 147 tcp_write(sock);
148 if (fdstate_generation((int)sock) == ready[i].generation)
149 fdstate_release((int)sock);
157 } 150 }
158 } 151 }
159 152
160 while ((sock = mutex_workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1) 153 while ((sock = workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1)
161 http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial); 154 http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial);
162 155
163 if (time_now_seconds() > next_timeout_check) { 156 if (time_now_seconds() > next_timeout_check) {
@@ -173,13 +166,7 @@ static void *server_mainloop(void *args) {
173 166
174 /* This work struct is about to go out of scope, so hand its tally over 167 /* This work struct is about to go out of scope, so hand its tally over
175 before anything can read it there */ 168 before anything can read it there */
176 stats_counters_retire(&ws); 169 workstruct_deinit(&ws);
177
178 free(ws.inbuf);
179 free(ws.outbuf);
180#ifdef _DEBUG_HTTPERROR
181 free(ws.debugbuf);
182#endif
183 return 0; 170 return 0;
184} 171}
185 172
@@ -188,7 +175,7 @@ static void *server_mainloop(void *args) {
188 member asked for it before binding, the first one included, so whether 175 member asked for it before binding, the first one included, so whether
189 this entry will have workers has to be known here. */ 176 this entry will have workers has to be known here. */
190static socket_sharing listen_sharing_for(const ot_listen *entry) { 177static socket_sharing listen_sharing_for(const ot_listen *entry) {
191 if (entry->proto == FLAG_UDP && entry->workers > 1 && socket_can_balance()) 178 if (entry->proto != FLAG_MCA && entry->workers > 1 && socket_can_balance())
192 return SOCKET_BALANCED; 179 return SOCKET_BALANCED;
193 return entry->shared ? SOCKET_SHARED : SOCKET_EXCLUSIVE; 180 return entry->shared ? SOCKET_SHARED : SOCKET_EXCLUSIVE;
194} 181}
@@ -244,10 +231,41 @@ static void listen_bind_all(void) {
244 listen_bind_one(entries + i); 231 listen_bind_one(entries + i);
245} 232}
246 233
247/* The socket the event loop was using becomes the first worker's. Where the 234/* The socket the event loop was using is passed to the first worker.
248 kernel spreads datagrams, every further worker binds one of its own, so 235 If start up tests show that the kernel can spread connections over
249 none of them share a receive queue. 236 a group bound to one address, every worker binds one of its own.
250 All others read from the first */ 237 If not, all workers call accept on the first. */
238static void listen_start_tcp_workers(void) {
239 size_t count, i;
240 ot_listen *entries = config_listen_entries(&count);
241
242 for (i = 0; i < count; ++i) {
243 ot_listen *entry = entries + i;
244 int balance = listen_sharing_for(entry) == SOCKET_BALANCED;
245 int worker;
246
247 if (entry->proto != FLAG_TCP || entry->workers <= 0)
248 continue;
249
250 event_del((int)entry->sock);
251 tcp_worker_install(entry->sock);
252
253 for (worker = 1; worker < entry->workers; ++worker) {
254 int64_t sock = entry->sock;
255
256 if (balance) {
257 sock = socket_tcp_listen(entry->ip, entry->port, SOCKET_BALANCED);
258 if (sock == -1) {
259 fprintf(stderr, "Warning: could not bind another socket for tcp worker %d: %s. Sharing the first instead.\n", worker,
260 strerror(errno));
261 sock = entry->sock;
262 }
263 }
264 tcp_worker_install(sock);
265 }
266 }
267}
268
251static void listen_start_udp_workers(void) { 269static void listen_start_udp_workers(void) {
252 size_t count, i; 270 size_t count, i;
253 ot_listen *entries = config_listen_entries(&count); 271 ot_listen *entries = config_listen_entries(&count);
@@ -408,10 +426,11 @@ int main(int argc, char **argv) {
408 426
409 install_signal_handlers(); 427 install_signal_handlers();
410 428
429 listen_start_tcp_workers();
411 listen_start_udp_workers(); 430 listen_start_udp_workers();
412 431
413 server_mainloop(0); 432 server_mainloop(0);
414 udp_workers_deinit(); 433 worker_stop_all();
415 434
416 /* 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
417 and the sanitizers can follow it */ 436 and the sanitizers can follow it */
diff --git a/opentracker.conf.sample b/opentracker.conf.sample
index 1d535e8..f075703 100644
--- a/opentracker.conf.sample
+++ b/opentracker.conf.sample
@@ -21,9 +21,9 @@
21# 21#
22# A listen statement takes options after the address: 22# A listen statement takes options after the address:
23# 23#
24# workers=<count> handle this udp socket in blocking reads in that 24# workers=<count> handle this socket in that many dedicated worker
25# many dedicated worker threads instead of in the 25# threads instead of in the event loop. Defaults to 1
26# event loop. Defaults to 1. Ignored for tcp. 26# for udp and 0 for tcp.
27# shared allow another process to bind the same address and 27# shared allow another process to bind the same address and
28# let the kernel share the traffic. Only useful if you 28# let the kernel share the traffic. Only useful if you
29# deliberately run a second tracker there. 29# deliberately run a second tracker there.
@@ -35,6 +35,7 @@
35# anywhere. 35# anywhere.
36# 36#
37# listen.udp.workers 4 37# listen.udp.workers 4
38# listen.tcp.workers 4
38# listen.shared 1 39# listen.shared 1
39# 40#
40# Raising the worker count pays off where the kernel spreads arriving 41# Raising the worker count pays off where the kernel spreads arriving
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_fdstate.c b/ot_fdstate.c
index 7d742da..09daf50 100644
--- a/ot_fdstate.c
+++ b/ot_fdstate.c
@@ -91,7 +91,7 @@ uint32_t fdstate_open(int fd, void *cookie, ot_time expiry) {
91 } 91 }
92 92
93 word = atomic_load_explicit(&slot->state, memory_order_relaxed); 93 word = atomic_load_explicit(&slot->state, memory_order_relaxed);
94 atomic_store_explicit(&slot->cookie, cookie, memory_order_relaxed); 94 atomic_store_explicit(&slot->cookie, cookie, memory_order_release);
95 atomic_store_explicit(&slot->expiry, expiry, memory_order_relaxed); 95 atomic_store_explicit(&slot->expiry, expiry, memory_order_relaxed);
96 /* Publishes the two above to whoever claims the slot next */ 96 /* Publishes the two above to whoever claims the slot next */
97 atomic_store_explicit(&slot->state, FD_WORD(FD_GEN(word), FD_IDLE), memory_order_release); 97 atomic_store_explicit(&slot->state, FD_WORD(FD_GEN(word), FD_IDLE), memory_order_release);
@@ -183,14 +183,17 @@ int fdstate_next_expired(ot_time now, int *cursor) {
183void *fdstate_cookie(int fd) { 183void *fdstate_cookie(int fd) {
184 ot_fdslot *slot = fdstate_slot(fd); 184 ot_fdslot *slot = fdstate_slot(fd);
185 185
186 return slot ? atomic_load_explicit(&slot->cookie, memory_order_relaxed) : NULL; 186 /* Pairs with the release in fdstate_set_cookie and fdstate_open */
187 return slot ? atomic_load_explicit(&slot->cookie, memory_order_acquire) : NULL;
187} 188}
188 189
189void fdstate_set_cookie(int fd, void *cookie) { 190void fdstate_set_cookie(int fd, void *cookie) {
190 ot_fdslot *slot = fdstate_slot(fd); 191 ot_fdslot *slot = fdstate_slot(fd);
191 192
193 /* Publishes what the cookie points at to whoever reads it next, which may
194 be another thread entirely */
192 if (slot) 195 if (slot)
193 atomic_store_explicit(&slot->cookie, cookie, memory_order_relaxed); 196 atomic_store_explicit(&slot->cookie, cookie, memory_order_release);
194} 197}
195 198
196void fdstate_set_expiry(int fd, ot_time expiry) { 199void fdstate_set_expiry(int fd, ot_time expiry) {
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 bb6ff24..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"
@@ -39,22 +40,30 @@ ssize_t g_stats_path_len;
39 40
40enum { SUCCESS_HTTP_HEADER_LENGTH = 80, SUCCESS_HTTP_SIZE_OFF = 17 }; 41enum { SUCCESS_HTTP_HEADER_LENGTH = 80, SUCCESS_HTTP_SIZE_OFF = 17 };
41 42
42int http_adopt(const int64_t sock, struct http_data *cookie) { 43struct http_data *http_adopt(const int64_t sock) {
43 struct http_data *heap = malloc(sizeof(*heap)); 44 struct http_data *cookie = fdstate_cookie(sock), *heap;
45 int interest;
44 46
45 if (!heap) 47 if (!cookie || !(cookie->flag & STRUCT_HTTP_FLAG_ONSTACK))
46 return -1; 48 return cookie;
49
50 if (!(heap = malloc(sizeof(*heap))))
51 return NULL;
47 52
48 *heap = *cookie; 53 *heap = *cookie;
49 heap->flag &= ~STRUCT_HTTP_FLAG_ONSTACK; 54 heap->flag &= ~STRUCT_HTTP_FLAG_ONSTACK;
55
56 /* Publishing the cookie is what lets another thread reach this connection,
57 so nothing may touch it afterwards, not even to read what it asked for */
58 interest = heap->interest;
50 fdstate_set_cookie((int)sock, heap); 59 fdstate_set_cookie((int)sock, heap);
51 60
52 if (event_add((int)sock, heap->interest, fdstate_generation((int)sock)) == -1) { 61 if (event_add((int)sock, interest, fdstate_generation((int)sock)) == -1) {
53 fdstate_set_cookie((int)sock, cookie); 62 fdstate_set_cookie((int)sock, cookie);
54 free(heap); 63 free(heap);
55 return -1; 64 return NULL;
56 } 65 }
57 return 0; 66 return heap;
58} 67}
59 68
60/* Give up a connection: the one place that knows everything hanging off it. */ 69/* Give up a connection: the one place that knows everything hanging off it. */
@@ -69,7 +78,7 @@ void http_close(const int64_t sock) {
69 free(cookie->batch); 78 free(cookie->batch);
70 buf_reset(&cookie->request); 79 buf_reset(&cookie->request);
71 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))
72 mutex_workqueue_canceltask(sock); 81 workqueue_canceltask(sock);
73 if (!(cookie->flag & STRUCT_HTTP_FLAG_ONSTACK)) 82 if (!(cookie->flag & STRUCT_HTTP_FLAG_ONSTACK))
74 free(cookie); 83 free(cookie);
75 } 84 }
@@ -348,13 +357,17 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
348 } 357 }
349#endif 358#endif
350#endif 359#endif
351 /* Pass this task to the worker thread */
352 cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED; 360 cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED;
353 361
354 /* Clients waiting for us should not easily timeout */ 362 /* Clients waiting for us should not easily timeout */
355 fdstate_set_expiry(sock, 0); 363 fdstate_set_expiry(sock, 0);
356 fullscrape_deliver(sock, format);
357 http_interest_set(sock, cookie, 0); 364 http_interest_set(sock, cookie, 0);
365
366 /* Everything this connection needs has been written, so it can be given
367 a home of its own and only then be made reachable by a worker thread */
368 if (!http_adopt(sock))
369 HTTPERROR_500;
370 fullscrape_deliver(sock, format);
358 return ws->reply_size = -2; 371 return ws->reply_size = -2;
359 } 372 }
360#endif 373#endif
@@ -366,6 +379,9 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
366 379
367 /* Complex stats also include expensive memory debugging tools */ 380 /* Complex stats also include expensive memory debugging tools */
368 fdstate_set_expiry(sock, 0); 381 fdstate_set_expiry(sock, 0);
382
383 if (!http_adopt(sock))
384 HTTPERROR_500;
369 stats_deliver(sock, mode); 385 stats_deliver(sock, mode);
370 return ws->reply_size = -2; 386 return ws->reply_size = -2;
371 } 387 }
@@ -448,8 +464,11 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *
448 cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED; 464 cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED;
449 /* Clients waiting for us should not easily timeout */ 465 /* Clients waiting for us should not easily timeout */
450 fdstate_set_expiry(sock, 0); 466 fdstate_set_expiry(sock, 0);
451 fullscrape_deliver(sock, format);
452 http_interest_set(sock, cookie, 0); 467 http_interest_set(sock, cookie, 0);
468
469 if (!http_adopt(sock))
470 HTTPERROR_500;
471 fullscrape_deliver(sock, format);
453 return ws->reply_size = -2; 472 return ws->reply_size = -2;
454} 473}
455#endif 474#endif
diff --git a/ot_http.h b/ot_http.h
index 82d645b..9cacd21 100644
--- a/ot_http.h
+++ b/ot_http.h
@@ -40,9 +40,10 @@ void http_interest_set(const int64_t s, struct http_data *cookie, int
40/* Close a connection and release everything that belongs to it */ 40/* Close a connection and release everything that belongs to it */
41void http_close(const int64_t s); 41void http_close(const int64_t s);
42 42
43/* Promote a stack cookie to a heap one for persistence and handing it over 43/* Promote a stack cookie to a heap one, for persistence and for handing it
44 to the event loop. */ 44 over to the event loop or to another thread. Does nothing to one that is
45int http_adopt(const int64_t s, struct http_data *cookie); 45 already on the heap. Returns the connection's state, or NULL. */
46struct http_data *http_adopt(const int64_t s);
46 47
47ssize_t http_handle_request(const int64_t s, struct ot_workstruct *ws); 48ssize_t http_handle_request(const int64_t s, struct ot_workstruct *ws);
48ssize_t http_sendiovecdata(const int64_t s, struct ot_workstruct *ws, size_t iovec_entries, struct iovec *iovector, int is_partial); 49ssize_t http_sendiovecdata(const int64_t s, struct ot_workstruct *ws, size_t iovec_entries, struct iovec *iovector, int is_partial);
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 8647a47..4a5008b 100644
--- a/ot_tcp.c
+++ b/ot_tcp.c
@@ -5,6 +5,10 @@
5 5
6/* System */ 6/* System */
7#include <errno.h> 7#include <errno.h>
8#include <pthread.h>
9#include <stdatomic.h>
10#include <stdio.h>
11#include <stdlib.h>
8#include <string.h> 12#include <string.h>
9#include <unistd.h> 13#include <unistd.h>
10 14
@@ -16,6 +20,7 @@
16#include "ot_http.h" 20#include "ot_http.h"
17#include "ot_mutex.h" 21#include "ot_mutex.h"
18#include "ot_socket.h" 22#include "ot_socket.h"
23#include "ot_worker.h"
19#include "ot_stats.h" 24#include "ot_stats.h"
20#include "ot_tcp.h" 25#include "ot_tcp.h"
21#include "trackerlogic.h" 26#include "trackerlogic.h"
@@ -149,9 +154,29 @@ void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws) {
149 if (fdstate_generation((int)sock) != generation) 154 if (fdstate_generation((int)sock) != generation)
150 continue; 155 continue;
151 156
152 if (http_adopt(sock, &cookie) == -1) 157 /* Handling it may already have given it a heap home and handed it to
158 another thread, and then it is no longer this frame's to touch */
159 if (fdstate_cookie((int)sock) != &cookie)
160 fdstate_release((int)sock);
161 else if (!http_adopt(sock))
153 http_close(sock); 162 http_close(sock);
154 else 163 else
155 fdstate_release((int)sock); 164 fdstate_release((int)sock);
156 } 165 }
157} 166}
167
168/* A worker blocks in accept and nowhere else: the connection it is handed is
169 non blocking, so a client that stops talking is given to the event loop
170 rather than keeping this thread. */
171static void tcp_run(struct ot_worker *me) {
172 while (g_opentracker_running)
173 tcp_accept(me->sock, &me->ws);
174}
175
176void tcp_worker_install(int64_t sock) {
177 socket_set_blocking((int)sock, 1);
178 socket_defer_accept((int)sock);
179
180 if (worker_start(tcp_run, sock) == -1)
181 fprintf(stderr, "Warning: could not install a tcp worker, this socket stays with the workers already running.\n");
182}
diff --git a/ot_tcp.h b/ot_tcp.h
index b0a93f4..55a172c 100644
--- a/ot_tcp.h
+++ b/ot_tcp.h
@@ -19,4 +19,8 @@ void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws);
19void tcp_read(const int64_t sock, struct ot_workstruct *ws); 19void tcp_read(const int64_t sock, struct ot_workstruct *ws);
20void tcp_write(const int64_t sock); 20void tcp_write(const int64_t sock);
21 21
22/* One thread accepting on one socket. Called once per worker. */
23void tcp_worker_install(int64_t sock);
24
25
22#endif 26#endif
diff --git a/ot_udp.c b/ot_udp.c
index 1ec5148..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,31 +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 memset(&ws, 0, sizeof(ws));
224
225 stats_counters_register(&ws);
226 ws.inbuf = malloc(G_INBUF_SIZE);
227 ws.outbuf = malloc(G_OUTBUF_SIZE);
228#ifdef _DEBUG_HTTPERROR
229 ws.debugbuf = malloc(G_DEBUGBUF_SIZE);
230#endif
231
232 while (g_opentracker_running) 211 while (g_opentracker_running)
233 handle_udp6(me->sock, &ws); 212 handle_udp6(me->sock, &me->ws);
234
235 stats_counters_retire(&ws);
236
237 free(ws.inbuf);
238 free(ws.outbuf);
239#ifdef _DEBUG_HTTPERROR
240 free(ws.debugbuf);
241#endif
242
243 atomic_store(&me->left, 1);
244 return NULL;
245} 213}
246 214
247void udp_init(void) { 215void udp_init(void) {
@@ -250,47 +218,11 @@ void udp_init(void) {
250} 218}
251 219
252void udp_worker_install(int64_t sock) { 220void udp_worker_install(int64_t sock) {
253 struct udp_worker **grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers));
254 struct udp_worker *worker;
255
256 if (!grown || !(worker = calloc(1, sizeof(*worker)))) {
257 fprintf(stderr, "Warning: out of memory installing a udp worker, this socket stays with the workers already running.\n");
258 g_workers = grown ? grown : g_workers;
259 return;
260 }
261 g_workers = grown;
262 worker->sock = sock;
263
264#ifdef _DEBUG 221#ifdef _DEBUG
265 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);
266#endif 223#endif
267 socket_set_blocking((int)sock, 1); 224 socket_set_blocking((int)sock, 1);
268 225
269 if (pthread_create(&worker->thread, NULL, udp_worker, worker)) 226 if (worker_start(udp_run, sock) == -1)
270 free(worker); 227 fprintf(stderr, "Warning: could not install a udp worker, this socket stays with the workers already running.\n");
271 else
272 g_workers[g_worker_count++] = worker;
273}
274
275void udp_workers_deinit(void) {
276 size_t i;
277
278 for (i = 0; i < g_worker_count; ++i) {
279 struct udp_worker *worker = g_workers[i];
280
281 /* Repeated because a signal only interrupts a read already under way: a
282 worker that was between two reads when the first one arrived is inside
283 the next one by now. */
284 while (!atomic_load(&worker->left)) {
285 pthread_kill(worker->thread, OT_SIGNAL_WAKE);
286 usleep(1000);
287 }
288 pthread_join(worker->thread, NULL);
289 free(worker);
290 }
291
292 free(g_workers);
293 g_workers = NULL;
294 g_worker_count = 0;
295} 228}
296
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/integration/test_tcpworkers.py b/tests/integration/test_tcpworkers.py
new file mode 100644
index 0000000..e43612a
--- /dev/null
+++ b/tests/integration/test_tcpworkers.py
@@ -0,0 +1,67 @@
1"""The tracker with its tcp side handed to worker threads.
2
3With workers configured, a connection is accepted and answered on a worker's
4own stack. Anything it cannot finish there is copied to the heap and handed
5to the event loop, which is the one place where a worker and the main thread
6touch the same structures, so these drive all three outcomes: answered
7inline, held open, and parked waiting for another thread entirely.
8"""
9
10import threading
11import unittest
12import urllib.parse
13
14from harness import TrackerTestCase, bdecode
15
16WORKERS = 4
17
18
19class TcpWorkers(TrackerTestCase):
20 config = "listen.tcp.workers %d\n" % WORKERS
21
22 def announce_path(self, name, port=6881):
23 return "/announce?info_hash=%s&peer_id=%s&port=%d&left=0" % (
24 urllib.parse.quote_from_bytes(self.hash_for(name), safe=""),
25 "x" * 20,
26 port,
27 )
28
29 def test_many_at_once_are_all_answered(self):
30 """Every worker gets something to do, and the swarm adds up."""
31 info_hash = self.hash_for("workers-swarm")
32 errors = []
33
34 def announce(index):
35 try:
36 self.tracker.announce(info_hash, peer_id=b"-OT0001-%012d" % index, port=7000 + index, left=0)
37 except Exception as problem: # noqa: BLE001 - reported, not swallowed
38 errors.append("%d: %r" % (index, problem))
39
40 threads = [threading.Thread(target=announce, args=(i,)) for i in range(32)]
41 for thread in threads:
42 thread.start()
43 for thread in threads:
44 thread.join()
45
46 self.assertEqual(errors, [], "\n".join(errors))
47 self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"complete"], 32)
48
49 def test_a_kept_connection_outlives_the_worker_that_took_it(self):
50 """Keep-alive is the plain case of a connection a worker hands on."""
51 connection = self.tracker.connect()
52 try:
53 for i in range(4):
54 connection.send(self.announce_path("workers-kept"))
55 status, _, body = connection.reply()
56 self.assertIn(b"200", status)
57 self.assertIn(b"complete", bdecode(body))
58 finally:
59 connection.close()
60
61 def test_a_full_scrape_asked_for_by_a_worker_still_arrives(self):
62 """The reply is built by another thread, so the worker parks the
63 connection waiting for nothing at all before handing it over."""
64 for i in range(20):
65 self.tracker.announce(bytes([i]) + b"\0" * 19, port=7100 + i, left=0)
66 body = self.tracker.get("/scrape")
67 self.assertEqual(len(bdecode(body)[b"files"]), 20)
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 477c4a6..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();
@@ -567,6 +569,35 @@ void trackerlogic_init(void) {
567 stats_init(); 569 stats_init();
568} 570}
569 571
572int workstruct_init(struct ot_workstruct *ws) {
573 memset(ws, 0, sizeof(*ws));
574 ws->inbuf = malloc(G_INBUF_SIZE);
575 ws->outbuf = malloc(G_OUTBUF_SIZE);
576#ifdef _DEBUG_HTTPERROR
577 ws->debugbuf = malloc(G_DEBUGBUF_SIZE);
578#endif
579 if (!ws->inbuf || !ws->outbuf)
580 return -1;
581
582 /* Picking which peers to hand out starts here, so two workers seeded alike
583 would hand out alike */
584 random_bytes(ws->rand48_state, sizeof(ws->rand48_state));
585
586 stats_counters_register(ws);
587 return 0;
588}
589
590void workstruct_deinit(struct ot_workstruct *ws) {
591 /* The tally goes to the global before this struct goes out of scope */
592 stats_counters_retire(ws);
593
594 free(ws->inbuf);
595 free(ws->outbuf);
596#ifdef _DEBUG_HTTPERROR
597 free(ws->debugbuf);
598#endif
599}
600
570void trackerlogic_deinit(void) { 601void trackerlogic_deinit(void) {
571 int delta_torrentcount = 0; 602 int delta_torrentcount = 0;
572 size_t bucket, j; 603 size_t bucket, j;
@@ -594,5 +625,6 @@ void trackerlogic_deinit(void) {
594 } 625 }
595 626
596 /* Release mutexes */ 627 /* Release mutexes */
628 workqueue_deinit();
597 mutex_deinit(); 629 mutex_deinit();
598} 630}
diff --git a/trackerlogic.h b/trackerlogic.h
index d269267..53edabd 100644
--- a/trackerlogic.h
+++ b/trackerlogic.h
@@ -236,6 +236,12 @@ struct ot_workstruct {
236 int keep_alive; 236 int keep_alive;
237}; 237};
238 238
239/* Everything a thread that serves requests needs for itself. Set up and torn
240 down in one place so that a new kind of worker cannot miss a piece.
241 Returns -1 if the buffers could not be had. */
242int workstruct_init(struct ot_workstruct *ws);
243void workstruct_deinit(struct ot_workstruct *ws);
244
239/* 245/*
240 Exported functions 246 Exported functions
241*/ 247*/
generated by cgit v1.3 (git 2.53.0) at 2026-11-01 05:43:19 +0000