summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--opentracker.c73
-rw-r--r--ot_accesslist.c12
-rw-r--r--ot_clean.c39
-rw-r--r--ot_event.c18
-rw-r--r--ot_fullscrape.c5
-rw-r--r--ot_http.c16
-rw-r--r--ot_livesync.c12
-rw-r--r--ot_mutex.c25
-rw-r--r--ot_mutex.h1
-rw-r--r--ot_stats.c227
-rw-r--r--ot_stats.h17
-rw-r--r--ot_sync.c2
-rw-r--r--ot_udp.c23
-rw-r--r--ot_vector.c3
-rw-r--r--proxy.c3
-rw-r--r--tests/integration/harness.py6
-rw-r--r--tests/integration/test_concurrency.py209
-rw-r--r--tests/stubs.c3
-rw-r--r--trackerlogic.c25
-rw-r--r--trackerlogic.h75
20 files changed, 602 insertions, 192 deletions
diff --git a/opentracker.c b/opentracker.c
index 11da486..a546fc5 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -39,20 +39,22 @@
39#include "trackerlogic.h" 39#include "trackerlogic.h"
40 40
41/* Globals */ 41/* Globals */
42time_t g_now_seconds;
43uint32_t g_tracker_id; 42uint32_t g_tracker_id;
44volatile int g_opentracker_running = 1; 43_Atomic int g_opentracker_running = 1;
45int g_self_pipe[2]; 44int g_self_pipe[2];
46 45
47
48static void panic(const char *routine) __attribute__((noreturn)); 46static void panic(const char *routine) __attribute__((noreturn));
49static void panic(const char *routine) { 47static void panic(const char *routine) {
50 fprintf(stderr, "%s: %s\n", routine, strerror(errno)); 48 fprintf(stderr, "%s: %s\n", routine, strerror(errno));
51 exit(111); 49 exit(111);
52} 50}
53 51
52/* Say we are stopping, wake the loop, and let main do the work on its own stack. */
54static void signal_handler(int s) { 53static void signal_handler(int s) {
55 if (s == SIGINT) { 54 if (s == SIGINT) {
55 const char byte = 0;
56 int saved = errno;
57
56 /* Any new interrupt signal quits the application */ 58 /* Any new interrupt signal quits the application */
57 signal(SIGINT, SIG_DFL); 59 signal(SIGINT, SIG_DFL);
58 60
@@ -60,13 +62,8 @@ static void signal_handler(int s) {
60 but cancel their operations and return */ 62 but cancel their operations and return */
61 g_opentracker_running = 0; 63 g_opentracker_running = 0;
62 64
63 trackerlogic_deinit(); 65 (void)!write(g_self_pipe[1], &byte, 1);
64 66 errno = saved;
65#ifdef WANT_SYSLOGS
66 closelog();
67#endif
68
69 exit(0);
70 } 67 }
71} 68}
72 69
@@ -87,7 +84,7 @@ static void install_signal_handlers(void) {
87 84
88 sa.sa_handler = signal_handler; 85 sa.sa_handler = signal_handler;
89 sigemptyset(&sa.sa_mask); 86 sigemptyset(&sa.sa_mask);
90 sa.sa_flags = SA_RESTART; 87 sa.sa_flags = 0;
91 if ((sigaction(SIGINT, &sa, NULL) == -1) || (sigaction(SIGALRM, &sa, NULL) == -1)) 88 if ((sigaction(SIGINT, &sa, NULL) == -1) || (sigaction(SIGALRM, &sa, NULL) == -1))
92 panic("install_signal_handlers"); 89 panic("install_signal_handlers");
93 90
@@ -203,12 +200,12 @@ static void handle_write(const int64_t sock) {
203 waiting to write and go back to waiting for the next request */ 200 waiting to write and go back to waiting for the next request */
204 http_interest_drop(sock, cookie, EVENT_OUT); 201 http_interest_drop(sock, cookie, EVENT_OUT);
205 http_interest_add(sock, cookie, EVENT_IN); 202 http_interest_add(sock, cookie, EVENT_IN);
206 fdstate_set_expiry(sock, g_now_seconds + OT_CLIENT_TIMEOUT); 203 fdstate_set_expiry(sock, time_future_seconds(OT_CLIENT_TIMEOUT));
207 } else 204 } else
208 handle_dead(sock); 205 handle_dead(sock);
209} 206}
210 207
211static void handle_accept(const int64_t serversocket) { 208static void handle_accept(const int64_t serversocket, struct ot_workstruct *ws) {
212 struct http_data *cookie; 209 struct http_data *cookie;
213 int64_t sock; 210 int64_t sock;
214 ot_ip6 ip; 211 ot_ip6 ip;
@@ -218,22 +215,21 @@ static void handle_accept(const int64_t serversocket) {
218 215
219 socket_set_blocking((int)sock, 0); 216 socket_set_blocking((int)sock, 0);
220 217
221 if (!(cookie = (struct http_data *)malloc(sizeof(struct http_data)))) { 218 if (!(cookie = (struct http_data *)calloc(1, sizeof(struct http_data)))) {
222 close((int)sock); 219 close((int)sock);
223 continue; 220 continue;
224 } 221 }
225 memset(cookie, 0, sizeof(struct http_data));
226 memcpy(cookie->ip, ip, sizeof(ot_ip6)); 222 memcpy(cookie->ip, ip, sizeof(ot_ip6));
227 cookie->interest = EVENT_IN; 223 cookie->interest = EVENT_IN;
228 224
229 if (event_add((int)sock, EVENT_IN, fdstate_open(sock, cookie, g_now_seconds + OT_CLIENT_TIMEOUT)) == -1) { 225 if (event_add((int)sock, EVENT_IN, fdstate_open(sock, cookie, time_future_seconds(OT_CLIENT_TIMEOUT))) == -1) {
230 fdstate_close(sock); 226 fdstate_close(sock);
231 free(cookie); 227 free(cookie);
232 close((int)sock); 228 close((int)sock);
233 continue; 229 continue;
234 } 230 }
235 231
236 stats_issue_event(EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip); 232 stats_issue_event(ws, EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip);
237 } 233 }
238} 234}
239 235
@@ -242,7 +238,7 @@ static void handle_accept(const int64_t serversocket) {
242 238
243static void *server_mainloop(void *args) { 239static void *server_mainloop(void *args) {
244 struct ot_workstruct ws; 240 struct ot_workstruct ws;
245 time_t next_timeout_check = g_now_seconds + OT_CLIENT_TIMEOUT_CHECKINTERVAL; 241 time_t next_timeout_check = time_future_seconds(OT_CLIENT_TIMEOUT_CHECKINTERVAL);
246 struct iovec *iovector; 242 struct iovec *iovector;
247 size_t iovec_entries; 243 size_t iovec_entries;
248 int is_partial; 244 int is_partial;
@@ -250,6 +246,8 @@ static void *server_mainloop(void *args) {
250 (void)args; 246 (void)args;
251 247
252 /* Initialize our "thread local storage" */ 248 /* Initialize our "thread local storage" */
249 memset(&ws, 0, sizeof(ws));
250 stats_counters_register(&ws);
253 ws.inbuf = malloc(G_INBUF_SIZE); 251 ws.inbuf = malloc(G_INBUF_SIZE);
254 ws.outbuf = malloc(G_OUTBUF_SIZE); 252 ws.outbuf = malloc(G_OUTBUF_SIZE);
255#ifdef _DEBUG_HTTPERROR 253#ifdef _DEBUG_HTTPERROR
@@ -261,7 +259,7 @@ static void *server_mainloop(void *args) {
261 259
262 random_bytes(ws.rand48_state, sizeof(ws.rand48_state)); 260 random_bytes(ws.rand48_state, sizeof(ws.rand48_state));
263 261
264 for (;;) { 262 while (g_opentracker_running) {
265 ot_event ready[OT_EVENT_BATCH]; 263 ot_event ready[OT_EVENT_BATCH];
266 int64_t sock; 264 int64_t sock;
267 int count, i; 265 int count, i;
@@ -280,7 +278,7 @@ static void *server_mainloop(void *args) {
280 cookie = fdstate_cookie((int)sock); 278 cookie = fdstate_cookie((int)sock);
281 279
282 if ((intptr_t)cookie == FLAG_TCP) 280 if ((intptr_t)cookie == FLAG_TCP)
283 handle_accept(sock); 281 handle_accept(sock, &ws);
284 else if ((intptr_t)cookie == FLAG_UDP) 282 else if ((intptr_t)cookie == FLAG_UDP)
285 handle_udp6(sock, &ws); 283 handle_udp6(sock, &ws);
286 else if ((intptr_t)cookie == FLAG_SELFPIPE) 284 else if ((intptr_t)cookie == FLAG_SELFPIPE)
@@ -298,16 +296,26 @@ static void *server_mainloop(void *args) {
298 while ((sock = mutex_workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1) 296 while ((sock = mutex_workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1)
299 http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial); 297 http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial);
300 298
301 if (g_now_seconds > next_timeout_check) { 299 if (time_now_seconds() > next_timeout_check) {
302 int cursor = 0; 300 int cursor = 0;
303 301
304 while ((sock = fdstate_next_expired(g_now_seconds, &cursor)) != -1) 302 while ((sock = fdstate_next_expired(time_now_seconds(), &cursor)) != -1)
305 handle_dead(sock); 303 handle_dead(sock);
306 next_timeout_check = g_now_seconds + OT_CLIENT_TIMEOUT_CHECKINTERVAL; 304 next_timeout_check = time_future_seconds(OT_CLIENT_TIMEOUT_CHECKINTERVAL);
307 } 305 }
308 306
309 livesync_ticker(); 307 livesync_ticker();
310 } 308 }
309
310 /* This work struct is about to go out of scope, so hand its tally over
311 before anything can read it there */
312 stats_counters_retire(&ws);
313
314 free(ws.inbuf);
315 free(ws.outbuf);
316#ifdef _DEBUG_HTTPERROR
317 free(ws.debugbuf);
318#endif
311 return 0; 319 return 0;
312} 320}
313 321
@@ -461,17 +469,7 @@ int drop_privileges(const char *const serveruser, const char *const serverdir) {
461} 469}
462 470
463/* Maintain our copy of the clock. time() on BSDs is very expensive. */ 471/* Maintain our copy of the clock. time() on BSDs is very expensive. */
464static void *time_caching_worker(void *args) {
465 (void)args;
466 while (1) {
467 g_now_seconds = time(NULL);
468 sleep(5);
469 }
470 return NULL;
471}
472
473int main(int argc, char **argv) { 472int main(int argc, char **argv) {
474 pthread_t thread_id; /* time cacher */
475 473
476 random_init(); 474 random_init();
477 if (fdstate_init() == -1) 475 if (fdstate_init() == -1)
@@ -492,8 +490,6 @@ int main(int argc, char **argv) {
492 if (drop_privileges(g_serveruser ? g_serveruser : "nobody", g_serverdir) == -1) 490 if (drop_privileges(g_serveruser ? g_serveruser : "nobody", g_serverdir) == -1)
493 panic("drop_privileges failed, exiting. Last error"); 491 panic("drop_privileges failed, exiting. Last error");
494 492
495 g_now_seconds = time(NULL);
496 pthread_create(&thread_id, NULL, time_caching_worker, NULL);
497 493
498 /* Create our self pipe which allows us to interrupt mainloops 494 /* Create our self pipe which allows us to interrupt mainloops
499 wait in case some data is available to send out */ 495 wait in case some data is available to send out */
@@ -524,6 +520,13 @@ int main(int argc, char **argv) {
524 520
525 server_mainloop(0); 521 server_mainloop(0);
526 522
523 /* The loop has left, so no thread is serving any more and the teardown
524 can run where a debugger and the sanitizers can follow it */
525 trackerlogic_deinit();
526#ifdef WANT_SYSLOGS
527 closelog();
528#endif
529
527 event_deinit(); 530 event_deinit();
528 fdstate_deinit(); 531 fdstate_deinit();
529 return 0; 532 return 0;
diff --git a/ot_accesslist.c b/ot_accesslist.c
index 4bf1701..e8e0476 100644
--- a/ot_accesslist.c
+++ b/ot_accesslist.c
@@ -93,7 +93,7 @@ static ot_accesslist *accesslist_make(ot_accesslist *next, size_t size) {
93 This will never delete head, because that might still be in use. */ 93 This will never delete head, because that might still be in use. */
94static void accesslist_clean(ot_accesslist *accesslist) { 94static void accesslist_clean(ot_accesslist *accesslist) {
95 while (accesslist && accesslist->next) { 95 while (accesslist && accesslist->next) {
96 if (accesslist->next->base + 5 < g_now_minutes) 96 if (accesslist->next->base + 5 < time_now_minutes())
97 accesslist->next = accesslist_free(accesslist->next); 97 accesslist->next = accesslist_free(accesslist->next);
98 accesslist = accesslist->next; 98 accesslist = accesslist->next;
99 } 99 }
@@ -197,7 +197,7 @@ static void accesslist_readfile(void) {
197 197
198 /* Mark old accesslist as "needs to live another five minutes" */ 198 /* Mark old accesslist as "needs to live another five minutes" */
199 if (g_accesslist) 199 if (g_accesslist)
200 g_accesslist->base = g_now_minutes; 200 g_accesslist->base = time_now_minutes();
201 accesslist_new->next = g_accesslist; 201 accesslist_new->next = g_accesslist;
202 g_accesslist = accesslist_new; /* Only now set a new list */ 202 g_accesslist = accesslist_new; /* Only now set a new list */
203 203
@@ -205,11 +205,11 @@ static void accesslist_readfile(void) {
205 /* If we have dynamic accesslists, reloading a new one will always void the add/delete lists. 205 /* If we have dynamic accesslists, reloading a new one will always void the add/delete lists.
206 Insert empty ones at the list head */ 206 Insert empty ones at the list head */
207 if (g_accesslist_add && (accesslist_new = accesslist_make(g_accesslist_add, 0)) != NULL) { 207 if (g_accesslist_add && (accesslist_new = accesslist_make(g_accesslist_add, 0)) != NULL) {
208 g_accesslist_add->base = g_now_minutes; 208 g_accesslist_add->base = time_now_minutes();
209 g_accesslist_add = accesslist_new; 209 g_accesslist_add = accesslist_new;
210 } 210 }
211 if (g_accesslist_delete && (accesslist_new = accesslist_make(g_accesslist_delete, 0)) != NULL) { 211 if (g_accesslist_delete && (accesslist_new = accesslist_make(g_accesslist_delete, 0)) != NULL) {
212 g_accesslist_delete->base = g_now_minutes; 212 g_accesslist_delete->base = time_now_minutes();
213 g_accesslist_delete = accesslist_new; 213 g_accesslist_delete = accesslist_new;
214 } 214 }
215#endif 215#endif
@@ -326,7 +326,7 @@ static void *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_
326 if (accesslist_new) { 326 if (accesslist_new) {
327 memcpy(accesslist_new->list, (*removing_from)->list, sizeof(ot_hash) * off); 327 memcpy(accesslist_new->list, (*removing_from)->list, sizeof(ot_hash) * off);
328 memcpy(accesslist_new->list + off, (*removing_from)->list + off + 1, sizeof(ot_hash) * ((*removing_from)->size - off - 1)); 328 memcpy(accesslist_new->list + off, (*removing_from)->list + off + 1, sizeof(ot_hash) * ((*removing_from)->size - off - 1));
329 (*removing_from)->base = g_now_minutes; 329 (*removing_from)->base = time_now_minutes();
330 *removing_from = accesslist_new; 330 *removing_from = accesslist_new;
331 } 331 }
332 } 332 }
@@ -351,7 +351,7 @@ static void *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_
351 memcpy(accesslist_new->list, (*adding_to)->list, sizeof(ot_hash) * off); 351 memcpy(accesslist_new->list, (*adding_to)->list, sizeof(ot_hash) * off);
352 memcpy(accesslist_new->list + off, info_hash, sizeof(info_hash)); 352 memcpy(accesslist_new->list + off, info_hash, sizeof(info_hash));
353 memcpy(accesslist_new->list + off + 1, (*adding_to)->list + off, sizeof(ot_hash) * ((*adding_to)->size - off)); 353 memcpy(accesslist_new->list + off + 1, (*adding_to)->list + off, sizeof(ot_hash) * ((*adding_to)->size - off));
354 (*adding_to)->base = g_now_minutes; 354 (*adding_to)->base = time_now_minutes();
355 *adding_to = accesslist_new; 355 *adding_to = accesslist_new;
356 } 356 }
357 } 357 }
diff --git a/ot_clean.c b/ot_clean.c
index bf0504c..77ea239 100644
--- a/ot_clean.c
+++ b/ot_clean.c
@@ -4,6 +4,7 @@
4 $id$ */ 4 $id$ */
5 5
6/* System */ 6/* System */
7#include <errno.h>
7#include <pthread.h> 8#include <pthread.h>
8#include <string.h> 9#include <string.h>
9#include <unistd.h> 10#include <unistd.h>
@@ -46,7 +47,7 @@ static ssize_t clean_single_bucket(ot_peer *peers, size_t peer_count, size_t pee
46 47
47int clean_single_peer_list(ot_peerlist *peer_list, size_t peer_size) { 48int clean_single_peer_list(ot_peerlist *peer_list, size_t peer_size) {
48 ot_vector *peer_vector = &peer_list->peers; 49 ot_vector *peer_vector = &peer_list->peers;
49 time_t timedout = (time_t)(g_now_minutes - peer_list->base); 50 time_t timedout = (time_t)(time_now_minutes() - peer_list->base);
50 size_t num_buckets = 1, removed_seeders = 0; 51 size_t num_buckets = 1, removed_seeders = 0;
51 52
52 /* No need to clean empty torrent */ 53 /* No need to clean empty torrent */
@@ -87,11 +88,11 @@ int clean_single_peer_list(ot_peerlist *peer_list, size_t peer_size) {
87 vector_redistribute_buckets(peer_list, peer_size); 88 vector_redistribute_buckets(peer_list, peer_size);
88 89
89 if (peer_list->peer_count) 90 if (peer_list->peer_count)
90 peer_list->base = g_now_minutes; 91 peer_list->base = time_now_minutes();
91 else { 92 else {
92 /* When we got here, the last time that torrent 93 /* When we got here, the last time that torrent
93 has been touched is OT_PEER_TIMEOUT Minutes before */ 94 has been touched is OT_PEER_TIMEOUT Minutes before */
94 peer_list->base = g_now_minutes - OT_PEER_TIMEOUT; 95 peer_list->base = time_now_minutes() - OT_PEER_TIMEOUT;
95 } 96 }
96 return 0; 97 return 0;
97} 98}
@@ -103,6 +104,29 @@ int clean_single_torrent(ot_torrent *torrent) {
103 return clean_single_peer_list(torrent->peer_list6, OT_PEER_SIZE6) * clean_single_peer_list(torrent->peer_list4, OT_PEER_SIZE4); 104 return clean_single_peer_list(torrent->peer_list6, OT_PEER_SIZE6) * clean_single_peer_list(torrent->peer_list4, OT_PEER_SIZE4);
104} 105}
105 106
107static pthread_mutex_t g_clean_mutex = PTHREAD_MUTEX_INITIALIZER;
108static pthread_cond_t g_clean_wakeup = PTHREAD_COND_INITIALIZER;
109
110static void clean_sleep(void) {
111 struct timespec deadline;
112
113 clock_gettime(CLOCK_REALTIME, &deadline);
114 deadline.tv_sec += OT_CLEAN_SLEEP / 1000000;
115 deadline.tv_nsec += (OT_CLEAN_SLEEP % 1000000) * 1000;
116 if (deadline.tv_nsec >= 1000000000) {
117 ++deadline.tv_sec;
118 deadline.tv_nsec -= 1000000000;
119 }
120
121 /* Both sides hold the mutex, so the flag cannot be cleared and signalled
122 in the window between testing it and waiting on it */
123 pthread_mutex_lock(&g_clean_mutex);
124 while (g_opentracker_running)
125 if (pthread_cond_timedwait(&g_clean_wakeup, &g_clean_mutex, &deadline) == ETIMEDOUT)
126 break;
127 pthread_mutex_unlock(&g_clean_mutex);
128}
129
106/* Clean up all peers in current bucket, remove timedout pools and 130/* Clean up all peers in current bucket, remove timedout pools and
107 torrents */ 131 torrents */
108static void *clean_worker(void *args) { 132static void *clean_worker(void *args) {
@@ -125,7 +149,7 @@ static void *clean_worker(void *args) {
125 mutex_bucket_unlock(bucket, delta_torrentcount); 149 mutex_bucket_unlock(bucket, delta_torrentcount);
126 if (!g_opentracker_running) 150 if (!g_opentracker_running)
127 return NULL; 151 return NULL;
128 usleep(OT_CLEAN_SLEEP); 152 clean_sleep();
129 } 153 }
130 stats_cleanup(); 154 stats_cleanup();
131#ifdef WANT_ACCESSLIST 155#ifdef WANT_ACCESSLIST
@@ -138,4 +162,9 @@ static void *clean_worker(void *args) {
138static pthread_t thread_id; 162static pthread_t thread_id;
139void clean_init(void) { pthread_create(&thread_id, NULL, clean_worker, NULL); } 163void clean_init(void) { pthread_create(&thread_id, NULL, clean_worker, NULL); }
140 164
141void clean_deinit(void) { pthread_cancel(thread_id); } 165void clean_deinit(void) {
166 pthread_mutex_lock(&g_clean_mutex);
167 pthread_cond_signal(&g_clean_wakeup);
168 pthread_mutex_unlock(&g_clean_mutex);
169 pthread_join(thread_id, NULL);
170}
diff --git a/ot_event.c b/ot_event.c
index ce0feef..1542bf5 100644
--- a/ot_event.c
+++ b/ot_event.c
@@ -83,8 +83,10 @@ int event_wait(ot_event *out, int max, int timeout_ms) {
83 g_ready_space = max; 83 g_ready_space = max;
84 } 84 }
85 85
86 while ((count = epoll_wait(g_epoll, g_ready, max, timeout_ms)) == -1 && errno == EINTR) 86 /* An interrupted wait is reported as no events rather than retried here,
87 ; 87 so that the caller gets to look at whatever the signal was about */
88 if ((count = epoll_wait(g_epoll, g_ready, max, timeout_ms)) == -1)
89 return errno == EINTR ? 0 : -1;
88 for (i = 0; i < count; ++i) { 90 for (i = 0; i < count; ++i) {
89 uint32_t events = g_ready[i].events; 91 uint32_t events = g_ready[i].events;
90 92
@@ -189,8 +191,10 @@ int event_wait(ot_event *out, int max, int timeout_ms) {
189 deadline = &timeout; 191 deadline = &timeout;
190 } 192 }
191 193
192 while ((count = kevent(g_kqueue, NULL, 0, g_ready, max, deadline)) == -1 && errno == EINTR) 194 /* An interrupted wait is reported as no events rather than retried here,
193 ; 195 so that the caller gets to look at whatever the signal was about */
196 if ((count = kevent(g_kqueue, NULL, 0, g_ready, max, deadline)) == -1)
197 return errno == EINTR ? 0 : -1;
194 for (i = 0; i < count; ++i) { 198 for (i = 0; i < count; ++i) {
195 uint64_t udata = (uint64_t)(uintptr_t)g_ready[i].udata; 199 uint64_t udata = (uint64_t)(uintptr_t)g_ready[i].udata;
196 int eof = (g_ready[i].flags & (EV_EOF | EV_ERROR)) != 0; 200 int eof = (g_ready[i].flags & (EV_EOF | EV_ERROR)) != 0;
@@ -310,8 +314,10 @@ int event_del(int fd) {
310int event_wait(ot_event *out, int max, int timeout_ms) { 314int event_wait(ot_event *out, int max, int timeout_ms) {
311 int ready, found = 0, i; 315 int ready, found = 0, i;
312 316
313 while ((ready = poll(g_pfds, (nfds_t)g_count, timeout_ms)) == -1 && errno == EINTR) 317 /* An interrupted wait is reported as no events rather than retried here,
314 ; 318 so that the caller gets to look at whatever the signal was about */
319 if ((ready = poll(g_pfds, (nfds_t)g_count, timeout_ms)) == -1)
320 return errno == EINTR ? 0 : -1;
315 if (ready <= 0) 321 if (ready <= 0)
316 return ready; 322 return ready;
317 323
diff --git a/ot_fullscrape.c b/ot_fullscrape.c
index 8518bd2..a86d6d7 100644
--- a/ot_fullscrape.c
+++ b/ot_fullscrape.c
@@ -55,6 +55,8 @@ static void *fullscrape_worker(void *args) {
55 while (g_opentracker_running) { 55 while (g_opentracker_running) {
56 ot_tasktype tasktype = TASK_FULLSCRAPE; 56 ot_tasktype tasktype = TASK_FULLSCRAPE;
57 ot_taskid taskid = mutex_workqueue_poptask(&tasktype); 57 ot_taskid taskid = mutex_workqueue_poptask(&tasktype);
58 if (!taskid)
59 break;
58#ifdef WANT_COMPRESSION_ZSTD 60#ifdef WANT_COMPRESSION_ZSTD
59 if (tasktype & TASK_FLAG_ZSTD) 61 if (tasktype & TASK_FLAG_ZSTD)
60 fullscrape_make_zstd(taskid, tasktype); 62 fullscrape_make_zstd(taskid, tasktype);
@@ -77,7 +79,8 @@ void fullscrape_init(void) {
77} 79}
78 80
79void fullscrape_deinit(void) { 81void fullscrape_deinit(void) {
80 pthread_cancel( thread_id ); 82 mutex_workqueue_wakeall();
83 pthread_join( thread_id, NULL );
81} 84}
82 85
83void fullscrape_deliver( int64_t sock, ot_tasktype tasktype ) { 86void fullscrape_deliver( int64_t sock, ot_tasktype tasktype ) {
diff --git a/ot_http.c b/ot_http.c
index 2216453..927d618 100644
--- a/ot_http.c
+++ b/ot_http.c
@@ -143,7 +143,7 @@ ssize_t http_issue_error(const int64_t sock, struct ot_workstruct *ws, int code)
143#ifdef _DEBUG_HTTPERROR 143#ifdef _DEBUG_HTTPERROR
144 fprintf(stderr, "DEBUG: invalid request was: %s\n", ws->debugbuf); 144 fprintf(stderr, "DEBUG: invalid request was: %s\n", ws->debugbuf);
145#endif 145#endif
146 stats_issue_event(EVENT_FAILED, FLAG_TCP, code); 146 stats_issue_event(ws, EVENT_FAILED, FLAG_TCP, code);
147 http_senddata(sock, ws); 147 http_senddata(sock, ws);
148 return ws->reply_size = -2; 148 return ws->reply_size = -2;
149} 149}
@@ -222,7 +222,7 @@ ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t
222 cookie->flag &= ~STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER; 222 cookie->flag &= ~STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER;
223 } 223 }
224 224
225 fdstate_set_expiry(sock, g_now_seconds + OT_CLIENT_TIMEOUT_SEND); 225 fdstate_set_expiry(sock, time_future_seconds(OT_CLIENT_TIMEOUT_SEND));
226 http_interest_add(sock, cookie, EVENT_OUT); 226 http_interest_add(sock, cookie, EVENT_OUT);
227 return 0; 227 return 0;
228} 228}
@@ -359,7 +359,7 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *
359 ot_scrape_log this_peer, *new_peer; 359 ot_scrape_log this_peer, *new_peer;
360 int exactmatch; 360 int exactmatch;
361 memcpy(this_peer.ip, cookie->ip, sizeof(ot_ip6)); 361 memcpy(this_peer.ip, cookie->ip, sizeof(ot_ip6));
362 this_peer.last_fullscrape = g_now_seconds; 362 this_peer.last_fullscrape = time_now_seconds();
363 pthread_mutex_lock(&g_modest_fullscrape_mutex); 363 pthread_mutex_lock(&g_modest_fullscrape_mutex);
364 new_peer = vector_find_or_insert(&g_modest_fullscrape_timeouts, &this_peer, sizeof(ot_scrape_log), sizeof(ot_ip6), &exactmatch); 364 new_peer = vector_find_or_insert(&g_modest_fullscrape_timeouts, &this_peer, sizeof(ot_scrape_log), sizeof(ot_ip6), &exactmatch);
365 if (!new_peer) { 365 if (!new_peer) {
@@ -402,7 +402,7 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *
402#endif 402#endif
403#endif 403#endif
404 404
405 stats_issue_event(EVENT_FULLSCRAPE_REQUEST, 0, (uintptr_t)cookie->ip); 405 stats_issue_event(ws, EVENT_FULLSCRAPE_REQUEST, 0, (uintptr_t)cookie->ip);
406 406
407#ifdef _DEBUG_HTTPERROR 407#ifdef _DEBUG_HTTPERROR
408 fprintf(stderr, "%s", ws->debugbuf); 408 fprintf(stderr, "%s", ws->debugbuf);
@@ -474,7 +474,7 @@ static ssize_t http_handle_scrape(const int64_t sock, struct ot_workstruct *ws,
474 /* Enough for http header + whole scrape string */ 474 /* Enough for http header + whole scrape string */
475 return_tcp_scrape_for_torrent((const ot_hash *)multiscrape_buf, numwant, &reply, ws->outbuf + G_OUTBUF_SIZE); 475 return_tcp_scrape_for_torrent((const ot_hash *)multiscrape_buf, numwant, &reply, ws->outbuf + G_OUTBUF_SIZE);
476 ws->reply_size = reply - ws->reply; 476 ws->reply_size = reply - ws->reply;
477 stats_issue_event(EVENT_SCRAPE, FLAG_TCP, ws->reply_size); 477 stats_issue_event(ws, EVENT_SCRAPE, FLAG_TCP, ws->reply_size);
478 return ws->reply_size; 478 return ws->reply_size;
479} 479}
480 480
@@ -682,7 +682,7 @@ static ssize_t http_handle_announce(const int64_t sock, struct ot_workstruct
682#endif 682#endif
683 683
684 /* XXX DEBUG 684 /* XXX DEBUG
685 stats_issue_event( EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ws->reply ); 685 stats_issue_event(ws, EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ws->reply );
686 */ 686 */
687 687
688 /* Scanned whole query string */ 688 /* Scanned whole query string */
@@ -694,7 +694,7 @@ static ssize_t http_handle_announce(const int64_t sock, struct ot_workstruct
694 else 694 else
695 ws->reply_size = add_peer_to_torrent_and_return_peers(FLAG_TCP, ws, numwant); 695 ws->reply_size = add_peer_to_torrent_and_return_peers(FLAG_TCP, ws, numwant);
696 696
697 stats_issue_event(EVENT_ANNOUNCE, FLAG_TCP, ws->reply_size); 697 stats_issue_event(ws, EVENT_ANNOUNCE, FLAG_TCP, ws->reply_size);
698 return ws->reply_size; 698 return ws->reply_size;
699} 699}
700 700
@@ -710,7 +710,7 @@ ssize_t http_handle_request(const int64_t sock, struct ot_workstruct *ws) {
710 log->size = ws->request_size; 710 log->size = ws->request_size;
711 log->data = malloc(ws->request_size); 711 log->data = malloc(ws->request_size);
712 log->next = 0; 712 log->next = 0;
713 log->time = g_now_seconds; 713 log->time = time_now_seconds();
714 memcpy(log->ip, cookie->ip, sizeof(ot_ip6)); 714 memcpy(log->ip, cookie->ip, sizeof(ot_ip6));
715 if (log->data) { 715 if (log->data) {
716 memcpy(log->data, ws->request, ws->request_size); 716 memcpy(log->data, ws->request, ws->request_size);
diff --git a/ot_livesync.c b/ot_livesync.c
index 5a63f3c..45271bc 100644
--- a/ot_livesync.c
+++ b/ot_livesync.c
@@ -70,8 +70,8 @@ void livesync_init(void) {
70 g_v6_buf.fill = sizeof(g_tracker_id) + sizeof(uint32_t); 70 g_v6_buf.fill = sizeof(g_tracker_id) + sizeof(uint32_t);
71 g_v4_buf.fill = sizeof(g_tracker_id) + sizeof(uint32_t); 71 g_v4_buf.fill = sizeof(g_tracker_id) + sizeof(uint32_t);
72 72
73 g_v6_buf.next_packet_time = g_now_seconds + LIVESYNC_MAXDELAY; 73 g_v6_buf.next_packet_time = time_future_seconds(LIVESYNC_MAXDELAY);
74 g_v4_buf.next_packet_time = g_now_seconds + LIVESYNC_MAXDELAY; 74 g_v4_buf.next_packet_time = time_future_seconds(LIVESYNC_MAXDELAY);
75 75
76 pthread_create(&thread_id, NULL, livesync_worker, NULL); 76 pthread_create(&thread_id, NULL, livesync_worker, NULL);
77} 77}
@@ -106,7 +106,7 @@ static void livesync_issue_peersync(sync_buffer *buf) {
106 106
107 memcpy(mycopy, buf->data, fill); 107 memcpy(mycopy, buf->data, fill);
108 buf->fill = sizeof(g_tracker_id) + sizeof(uint32_t); 108 buf->fill = sizeof(g_tracker_id) + sizeof(uint32_t);
109 buf->next_packet_time = g_now_seconds + LIVESYNC_MAXDELAY; 109 buf->next_packet_time = time_future_seconds(LIVESYNC_MAXDELAY);
110 110
111 /* From now this thread has a local copy of the buffer and 111 /* From now this thread has a local copy of the buffer and
112 has modified the protected element */ 112 has modified the protected element */
@@ -143,7 +143,7 @@ static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size)
143 off += sizeof(ot_hash) + peer_size; 143 off += sizeof(ot_hash) + peer_size;
144 } 144 }
145 145
146 stats_issue_event(EVENT_SYNC, 0, (ws->request_size - sizeof(g_tracker_id) - sizeof(uint32_t)) / ((ssize_t)sizeof(ot_hash) + peer_size)); 146 stats_issue_event(ws, EVENT_SYNC, 0, (ws->request_size - sizeof(g_tracker_id) - sizeof(uint32_t)) / ((ssize_t)sizeof(ot_hash) + peer_size));
147} 147}
148 148
149/* Tickle the live sync module from time to time, so no events get 149/* Tickle the live sync module from time to time, so no events get
@@ -152,13 +152,13 @@ static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size)
152void livesync_ticker(void) { 152void livesync_ticker(void) {
153 /* livesync_issue_peersync sets g_next_packet_time */ 153 /* livesync_issue_peersync sets g_next_packet_time */
154 pthread_mutex_lock(&g_outbuf_mutex); 154 pthread_mutex_lock(&g_outbuf_mutex);
155 if (g_now_seconds > g_v6_buf.next_packet_time && g_v6_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t)) 155 if (time_now_seconds() > g_v6_buf.next_packet_time && g_v6_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t))
156 livesync_issue_peersync(&g_v6_buf); 156 livesync_issue_peersync(&g_v6_buf);
157 else 157 else
158 pthread_mutex_unlock(&g_outbuf_mutex); 158 pthread_mutex_unlock(&g_outbuf_mutex);
159 159
160 pthread_mutex_lock(&g_outbuf_mutex); 160 pthread_mutex_lock(&g_outbuf_mutex);
161 if (g_now_seconds > g_v4_buf.next_packet_time && g_v4_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t)) 161 if (time_now_seconds() > g_v4_buf.next_packet_time && g_v4_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t))
162 livesync_issue_peersync(&g_v4_buf); 162 livesync_issue_peersync(&g_v4_buf);
163 else 163 else
164 pthread_mutex_unlock(&g_outbuf_mutex); 164 pthread_mutex_unlock(&g_outbuf_mutex);
diff --git a/ot_mutex.c b/ot_mutex.c
index 37d9962..5392264 100644
--- a/ot_mutex.c
+++ b/ot_mutex.c
@@ -6,6 +6,7 @@
6/* System */ 6/* System */
7#include <arpa/inet.h> 7#include <arpa/inet.h>
8#include <pthread.h> 8#include <pthread.h>
9#include <stdatomic.h>
9#include <stdio.h> 10#include <stdio.h>
10#include <stdlib.h> 11#include <stdlib.h>
11#include <unistd.h> 12#include <unistd.h>
@@ -25,7 +26,9 @@
25/* Our global all torrents list */ 26/* Our global all torrents list */
26static ot_vector all_torrents[OT_BUCKET_COUNT]; 27static ot_vector all_torrents[OT_BUCKET_COUNT];
27static pthread_mutex_t bucket_mutex[OT_BUCKET_COUNT]; 28static pthread_mutex_t bucket_mutex[OT_BUCKET_COUNT];
28static size_t g_torrent_count; 29/* Every bucket's unlock adjusts this, and a bucket lock only keeps other
30 threads out of that one bucket, so the counter needs its own protection */
31static _Atomic size_t g_torrent_count;
29 32
30/* Self pipe from opentracker.c */ 33/* Self pipe from opentracker.c */
31extern int g_self_pipe[2]; 34extern int g_self_pipe[2];
@@ -47,14 +50,15 @@ ot_vector *mutex_bucket_lock_by_hash(ot_hash const hash) { return mutex_bucket_l
47 50
48void mutex_bucket_unlock(int bucket, int delta_torrentcount) { 51void mutex_bucket_unlock(int bucket, int delta_torrentcount) {
49 pthread_mutex_unlock(bucket_mutex + bucket); 52 pthread_mutex_unlock(bucket_mutex + bucket);
50 g_torrent_count += delta_torrentcount; 53 if (delta_torrentcount)
54 atomic_fetch_add_explicit(&g_torrent_count, (size_t)delta_torrentcount, memory_order_relaxed);
51} 55}
52 56
53void mutex_bucket_unlock_by_hash(ot_hash const hash, int delta_torrentcount) { 57void mutex_bucket_unlock_by_hash(ot_hash const hash, int delta_torrentcount) {
54 mutex_bucket_unlock(bucket_from_hash(hash), delta_torrentcount); 58 mutex_bucket_unlock(bucket_from_hash(hash), delta_torrentcount);
55} 59}
56 60
57size_t mutex_get_torrent_count(void) { return g_torrent_count; } 61size_t mutex_get_torrent_count(void) { return atomic_load_explicit(&g_torrent_count, memory_order_relaxed); }
58 62
59/* TaskQueue Magic */ 63/* TaskQueue Magic */
60 64
@@ -145,9 +149,13 @@ ot_taskid mutex_workqueue_poptask(ot_tasktype *tasktype) {
145 break; 149 break;
146 } 150 }
147 151
148 /* Wait until the next task is being fed */ 152 /* Wait until the next task is being fed, unless the tracker is on its
149 if (!taskid) 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;
150 pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex); 157 pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex);
158 }
151 } 159 }
152 160
153 /* Release lock */ 161 /* Release lock */
@@ -271,6 +279,13 @@ void mutex_init(void) {
271 memset(all_torrents, 0, sizeof(all_torrents)); 279 memset(all_torrents, 0, sizeof(all_torrents));
272} 280}
273 281
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
274void mutex_deinit(void) { 289void mutex_deinit(void) {
275 size_t i; 290 size_t i;
276 for (i = 0; i < OT_BUCKET_COUNT; ++i) 291 for (i = 0; i < OT_BUCKET_COUNT; ++i)
diff --git a/ot_mutex.h b/ot_mutex.h
index 3bbbe9c..c048fdb 100644
--- a/ot_mutex.h
+++ b/ot_mutex.h
@@ -10,6 +10,7 @@
10#include "trackerlogic.h" 10#include "trackerlogic.h"
11 11
12void mutex_init(void); 12void mutex_init(void);
13void mutex_workqueue_wakeall(void);
13void mutex_deinit(void); 14void mutex_deinit(void);
14 15
15ot_vector *mutex_bucket_lock(int bucket); 16ot_vector *mutex_bucket_lock(int bucket);
diff --git a/ot_stats.c b/ot_stats.c
index 45dc8db..5f97a2f 100644
--- a/ot_stats.c
+++ b/ot_stats.c
@@ -6,6 +6,7 @@
6/* System */ 6/* System */
7#include <inttypes.h> 7#include <inttypes.h>
8#include <pthread.h> 8#include <pthread.h>
9#include <stdatomic.h>
9#include <stdio.h> 10#include <stdio.h>
10#include <stdlib.h> 11#include <stdlib.h>
11#include <string.h> 12#include <string.h>
@@ -36,27 +37,58 @@
36static void stats_make(size_t *iovec_entries, struct iovec **iovector, ot_tasktype mode); 37static void stats_make(size_t *iovec_entries, struct iovec **iovector, ot_tasktype mode);
37#define OT_STATS_TMPSIZE 8192 38#define OT_STATS_TMPSIZE 8192
38 39
39/* Clumsy counters... to be rethought */ 40/* Each serving thread keeps its own statistics tally. The list is linked
40static unsigned long long ot_overall_tcp_connections; 41 through intrusive next pointers within the workstruct.
41static unsigned long long ot_overall_udp_connections; 42
42static unsigned long long ot_overall_tcp_successfulannounces; 43 Registration happens once per thread and the adding up once per statistics
43static unsigned long long ot_overall_udp_successfulannounces; 44 request, so this lock is never on anybody's hot path. */
44static unsigned long long ot_overall_tcp_successfulscrapes; 45static struct ot_workstruct *g_counter_contexts;
45static unsigned long long ot_overall_udp_successfulscrapes; 46static ot_counter g_retired[OT_COUNTER_COUNT];
46static unsigned long long ot_overall_udp_connectionidmissmatches; 47static pthread_mutex_t g_counter_mutex = PTHREAD_MUTEX_INITIALIZER;
47static unsigned long long ot_overall_tcp_connects; 48
48static unsigned long long ot_overall_udp_connects; 49void stats_counters_register(struct ot_workstruct *ws) {
49static unsigned long long ot_overall_completed; 50 pthread_mutex_lock(&g_counter_mutex);
50static unsigned long long ot_full_scrape_count; 51 ws->counters_next = g_counter_contexts;
51static unsigned long long ot_full_scrape_request_count; 52 g_counter_contexts = ws;
52static unsigned long long ot_full_scrape_size; 53 pthread_mutex_unlock(&g_counter_mutex);
53static unsigned long long ot_failed_request_counts[CODE_HTTPERROR_COUNT]; 54}
55
56void stats_counters_retire(struct ot_workstruct *ws) {
57 struct ot_workstruct **link;
58 size_t i;
59
60 pthread_mutex_lock(&g_counter_mutex);
61 for (link = &g_counter_contexts; *link; link = &(*link)->counters_next)
62 if (*link == ws) {
63 *link = ws->counters_next;
64 break;
65 }
66
67 /* When retiring a worker thread, collect the counter tally in the global */
68 for (i = 0; i < OT_COUNTER_COUNT; ++i)
69 atomic_fetch_add_explicit(g_retired + i, atomic_load_explicit(ws->counters + i, memory_order_relaxed), memory_order_relaxed);
70 pthread_mutex_unlock(&g_counter_mutex);
71}
72
73/* What every thread has counted so far, added together */
74static void counters_total(unsigned long long *total) {
75 struct ot_workstruct *ws;
76 size_t i;
77
78 pthread_mutex_lock(&g_counter_mutex);
79 for (i = 0; i < OT_COUNTER_COUNT; ++i)
80 total[i] = atomic_load_explicit(g_retired + i, memory_order_relaxed);
81 for (ws = g_counter_contexts; ws; ws = ws->counters_next)
82 for (i = 0; i < OT_COUNTER_COUNT; ++i)
83 total[i] += atomic_load_explicit(ws->counters + i, memory_order_relaxed);
84 pthread_mutex_unlock(&g_counter_mutex);
85}
86
87#define COUNT(index, by) atomic_fetch_add_explicit(ws->counters + (index), (unsigned long long)(by), memory_order_relaxed)
88
54static const char *ot_failed_request_names[] = { 89static const char *ot_failed_request_names[] = {
55 "302 Redirect", "400 Parse Error", "400 Invalid Parameter", "400 Invalid Parameter (compact=0)", "400 Not Modest", 90 "302 Redirect", "400 Parse Error", "400 Invalid Parameter", "400 Invalid Parameter (compact=0)", "400 Not Modest",
56 "402 Payment Required", "403 Access Denied", "404 Not found", "500 Internal Server Error"}; 91 "402 Payment Required", "403 Access Denied", "404 Not found", "500 Internal Server Error"};
57static unsigned long long ot_renewed[OT_PEER_TIMEOUT];
58static unsigned long long ot_overall_sync_count;
59static unsigned long long ot_overall_stall_count;
60 92
61static time_t ot_start_time; 93static time_t ot_start_time;
62 94
@@ -347,39 +379,61 @@ static unsigned long events_per_time(unsigned long long events, time_t t) { retu
347 379
348static void stats_connections_mrtg(char **reply, const char *reply_end) { 380static void stats_connections_mrtg(char **reply, const char *reply_end) {
349 ot_time t = time(NULL) - ot_start_time; 381 ot_time t = time(NULL) - ot_start_time;
382 unsigned long long total[OT_COUNTER_COUNT];
383
384 counters_total(total);
385
350 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker connections, %lu conns/s :: %lu success/s.", 386 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker connections, %lu conns/s :: %lu success/s.",
351 ot_overall_tcp_connections + ot_overall_udp_connections, 387 total[OT_COUNTER_TCP_CONNECTIONS] + total[OT_COUNTER_UDP_CONNECTIONS],
352 ot_overall_tcp_successfulannounces + ot_overall_udp_successfulannounces + ot_overall_udp_connects, (int)t, (int)(t / 3600), 388 total[OT_COUNTER_TCP_ANNOUNCES] + total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], (int)t, (int)(t / 3600),
353 events_per_time(ot_overall_tcp_connections + ot_overall_udp_connections, t), 389 events_per_time(total[OT_COUNTER_TCP_CONNECTIONS] + total[OT_COUNTER_UDP_CONNECTIONS], t),
354 events_per_time(ot_overall_tcp_successfulannounces + ot_overall_udp_successfulannounces + ot_overall_udp_connects, t)); 390 events_per_time(total[OT_COUNTER_TCP_ANNOUNCES] + total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], t));
355} 391}
356 392
357static void stats_udpconnections_mrtg(char **reply, const char *reply_end) { 393static void stats_udpconnections_mrtg(char **reply, const char *reply_end) {
358 ot_time t = time(NULL) - ot_start_time; 394 ot_time t = time(NULL) - ot_start_time;
359 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker udp4 stats, %lu conns/s :: %lu success/s.", ot_overall_udp_connections, 395 unsigned long long total[OT_COUNTER_COUNT];
360 ot_overall_udp_successfulannounces + ot_overall_udp_connects, (int)t, (int)(t / 3600), events_per_time(ot_overall_udp_connections, t), 396
361 events_per_time(ot_overall_udp_successfulannounces + ot_overall_udp_connects, t)); 397 counters_total(total);
398
399 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker udp4 stats, %lu conns/s :: %lu success/s.", total[OT_COUNTER_UDP_CONNECTIONS],
400 total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], (int)t, (int)(t / 3600),
401 events_per_time(total[OT_COUNTER_UDP_CONNECTIONS], t),
402 events_per_time(total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], t));
362} 403}
363 404
364static void stats_tcpconnections_mrtg(char **reply, const char *reply_end) { 405static void stats_tcpconnections_mrtg(char **reply, const char *reply_end) {
365 time_t t = time(NULL) - ot_start_time; 406 time_t t = time(NULL) - ot_start_time;
366 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker tcp4 stats, %lu conns/s :: %lu success/s.", ot_overall_tcp_connections, 407 unsigned long long total[OT_COUNTER_COUNT];
367 ot_overall_tcp_successfulannounces, (int)t, (int)(t / 3600), events_per_time(ot_overall_tcp_connections, t), 408
368 events_per_time(ot_overall_tcp_successfulannounces, t)); 409 counters_total(total);
410
411 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker tcp4 stats, %lu conns/s :: %lu success/s.", total[OT_COUNTER_TCP_CONNECTIONS],
412 total[OT_COUNTER_TCP_ANNOUNCES], (int)t, (int)(t / 3600), events_per_time(total[OT_COUNTER_TCP_CONNECTIONS], t),
413 events_per_time(total[OT_COUNTER_TCP_ANNOUNCES], t));
369} 414}
370 415
371static void stats_scrape_mrtg(char **reply, const char *reply_end) { 416static void stats_scrape_mrtg(char **reply, const char *reply_end) {
372 time_t t = time(NULL) - ot_start_time; 417 time_t t = time(NULL) - ot_start_time;
418 unsigned long long total[OT_COUNTER_COUNT];
419
420 counters_total(total);
421
373 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker scrape stats, %lu scrape/s (tcp and udp)", 422 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker scrape stats, %lu scrape/s (tcp and udp)",
374 ot_overall_tcp_successfulscrapes, ot_overall_udp_successfulscrapes, (int)t, (int)(t / 3600), 423 total[OT_COUNTER_TCP_SCRAPES], total[OT_COUNTER_UDP_SCRAPES], (int)t, (int)(t / 3600),
375 events_per_time((ot_overall_tcp_successfulscrapes + ot_overall_udp_successfulscrapes), t)); 424 events_per_time((total[OT_COUNTER_TCP_SCRAPES] + total[OT_COUNTER_UDP_SCRAPES]), t));
376} 425}
377 426
378static void stats_fullscrapes_mrtg(char **reply, const char *reply_end) { 427static void stats_fullscrapes_mrtg(char **reply, const char *reply_end) {
379 ot_time t = time(NULL) - ot_start_time; 428 ot_time t = time(NULL) - ot_start_time;
429 unsigned long long total[OT_COUNTER_COUNT];
430
431 counters_total(total);
432
380 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker full scrape stats, %lu conns/s :: %lu bytes/s.", 433 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker full scrape stats, %lu conns/s :: %lu bytes/s.",
381 ot_full_scrape_count * 1000, ot_full_scrape_size, (int)t, (int)(t / 3600), events_per_time(ot_full_scrape_count, t), 434 total[OT_COUNTER_FULLSCRAPES] * 1000, total[OT_COUNTER_FULLSCRAPE_BYTES], (int)t, (int)(t / 3600),
382 events_per_time(ot_full_scrape_size, t)); 435 events_per_time(total[OT_COUNTER_FULLSCRAPES], t),
436 events_per_time(total[OT_COUNTER_FULLSCRAPE_BYTES], t));
383} 437}
384 438
385static void stats_peers_mrtg(char **reply, const char *reply_end) { 439static void stats_peers_mrtg(char **reply, const char *reply_end) {
@@ -398,30 +452,46 @@ static void stats_torrents_mrtg(char **reply, const char *reply_end) {
398} 452}
399 453
400static void stats_httperrors_txt(char **reply, const char *reply_end) { 454static void stats_httperrors_txt(char **reply, const char *reply_end) {
455 unsigned long long total[OT_COUNTER_COUNT];
456
457 counters_total(total);
458
401 fmt_append(reply, reply_end, "302 RED %llu\n400 ... %llu\n400 PAR %llu\n400 COM %llu\n403 IP %llu\n404 INV %llu\n500 SRV %llu\n", 459 fmt_append(reply, reply_end, "302 RED %llu\n400 ... %llu\n400 PAR %llu\n400 COM %llu\n403 IP %llu\n404 INV %llu\n500 SRV %llu\n",
402 ot_failed_request_counts[0], ot_failed_request_counts[1], ot_failed_request_counts[2], ot_failed_request_counts[3], 460 total[OT_COUNTER_FAILED + 0], total[OT_COUNTER_FAILED + 1], total[OT_COUNTER_FAILED + 2], total[OT_COUNTER_FAILED + 3],
403 ot_failed_request_counts[4], ot_failed_request_counts[5], ot_failed_request_counts[6]); 461 total[OT_COUNTER_FAILED + 4], total[OT_COUNTER_FAILED + 5], total[OT_COUNTER_FAILED + 6]);
404} 462}
405 463
406static void stats_return_renew_bucket(char **reply, const char *reply_end) { 464static void stats_return_renew_bucket(char **reply, const char *reply_end) {
465 unsigned long long total[OT_COUNTER_COUNT];
466
467 counters_total(total);
468
407 int i; 469 int i;
408 470
409 for (i = 0; i < OT_PEER_TIMEOUT; ++i) 471 for (i = 0; i < OT_PEER_TIMEOUT; ++i)
410 fmt_append(reply, reply_end, "%02i %llu\n", i, ot_renewed[i]); 472 fmt_append(reply, reply_end, "%02i %llu\n", i, total[OT_COUNTER_RENEWED + i]);
411} 473}
412 474
413static void stats_return_sync_mrtg(char **reply, const char *reply_end) { 475static void stats_return_sync_mrtg(char **reply, const char *reply_end) {
414 ot_time t = time(NULL) - ot_start_time; 476 ot_time t = time(NULL) - ot_start_time;
415 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker connections, %lu conns/s :: %lu success/s.", ot_overall_sync_count, 0LL, 477 unsigned long long total[OT_COUNTER_COUNT];
416 (int)t, (int)(t / 3600), events_per_time(ot_overall_tcp_connections + ot_overall_udp_connections, t), 478
417 events_per_time(ot_overall_tcp_successfulannounces + ot_overall_udp_successfulannounces + ot_overall_udp_connects, t)); 479 counters_total(total);
480
481 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker connections, %lu conns/s :: %lu success/s.", total[OT_COUNTER_SYNCS], 0LL,
482 (int)t, (int)(t / 3600), events_per_time(total[OT_COUNTER_TCP_CONNECTIONS] + total[OT_COUNTER_UDP_CONNECTIONS], t),
483 events_per_time(total[OT_COUNTER_TCP_ANNOUNCES] + total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], t));
418} 484}
419 485
420static void stats_return_completed_mrtg(char **reply, const char *reply_end) { 486static void stats_return_completed_mrtg(char **reply, const char *reply_end) {
421 ot_time t = time(NULL) - ot_start_time; 487 ot_time t = time(NULL) - ot_start_time;
488 unsigned long long total[OT_COUNTER_COUNT];
489
490 counters_total(total);
491
422 492
423 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker, %lu completed/h.", ot_overall_completed, 0LL, (int)t, (int)(t / 3600), 493 fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker, %lu completed/h.", total[OT_COUNTER_COMPLETED], 0LL, (int)t, (int)(t / 3600),
424 events_per_time(ot_overall_completed, t / 3600)); 494 events_per_time(total[OT_COUNTER_COMPLETED], t / 3600));
425} 495}
426 496
427#ifdef WANT_LOG_NUMWANT 497#ifdef WANT_LOG_NUMWANT
@@ -464,6 +534,10 @@ static void stats_return_fulllog(size_t *iovec_entries, struct iovec **iovector,
464#endif 534#endif
465 535
466static void stats_return_everything(char **reply, const char *reply_end) { 536static void stats_return_everything(char **reply, const char *reply_end) {
537 unsigned long long total[OT_COUNTER_COUNT];
538
539 counters_total(total);
540
467 torrent_stats stats = {0, 0, 0}; 541 torrent_stats stats = {0, 0, 0};
468 int i; 542 int i;
469 543
@@ -482,25 +556,25 @@ static void stats_return_everything(char **reply, const char *reply_end) {
482 fmt_append(reply, reply_end, " </torrents>\n"); 556 fmt_append(reply, reply_end, " </torrents>\n");
483 fmt_append(reply, reply_end, " <peers>\n <count>%llu</count>\n </peers>\n", stats.peer_count); 557 fmt_append(reply, reply_end, " <peers>\n <count>%llu</count>\n </peers>\n", stats.peer_count);
484 fmt_append(reply, reply_end, " <seeds>\n <count>%llu</count>\n </seeds>\n", stats.seed_count); 558 fmt_append(reply, reply_end, " <seeds>\n <count>%llu</count>\n </seeds>\n", stats.seed_count);
485 fmt_append(reply, reply_end, " <completed>\n <count>%llu</count>\n </completed>\n", ot_overall_completed); 559 fmt_append(reply, reply_end, " <completed>\n <count>%llu</count>\n </completed>\n", total[OT_COUNTER_COMPLETED]);
486 fmt_append(reply, reply_end, " <connections>\n"); 560 fmt_append(reply, reply_end, " <connections>\n");
487 fmt_append(reply, reply_end, " <tcp>\n <accept>%llu</accept>\n <announce>%llu</announce>\n <scrape>%llu</scrape>\n </tcp>\n", 561 fmt_append(reply, reply_end, " <tcp>\n <accept>%llu</accept>\n <announce>%llu</announce>\n <scrape>%llu</scrape>\n </tcp>\n",
488 ot_overall_tcp_connections, ot_overall_tcp_successfulannounces, ot_overall_tcp_successfulscrapes); 562 total[OT_COUNTER_TCP_CONNECTIONS], total[OT_COUNTER_TCP_ANNOUNCES], total[OT_COUNTER_TCP_SCRAPES]);
489 fmt_append(reply, reply_end, " <udp>\n <overall>%llu</overall>\n <connect>%llu</connect>\n <announce>%llu</announce>\n <scrape>%llu</scrape>\n <missmatch>%llu</missmatch>\n </udp>\n", 563 fmt_append(reply, reply_end, " <udp>\n <overall>%llu</overall>\n <connect>%llu</connect>\n <announce>%llu</announce>\n <scrape>%llu</scrape>\n <missmatch>%llu</missmatch>\n </udp>\n",
490 ot_overall_udp_connections, ot_overall_udp_connects, ot_overall_udp_successfulannounces, ot_overall_udp_successfulscrapes, 564 total[OT_COUNTER_UDP_CONNECTIONS], total[OT_COUNTER_UDP_CONNECTS], total[OT_COUNTER_UDP_ANNOUNCES], total[OT_COUNTER_UDP_SCRAPES],
491 ot_overall_udp_connectionidmissmatches); 565 total[OT_COUNTER_UDP_CONNID_MISSMATCH]);
492 fmt_append(reply, reply_end, " <livesync>\n <count>%llu</count>\n </livesync>\n", ot_overall_sync_count); 566 fmt_append(reply, reply_end, " <livesync>\n <count>%llu</count>\n </livesync>\n", total[OT_COUNTER_SYNCS]);
493 fmt_append(reply, reply_end, " </connections>\n"); 567 fmt_append(reply, reply_end, " </connections>\n");
494 fmt_append(reply, reply_end, " <debug>\n"); 568 fmt_append(reply, reply_end, " <debug>\n");
495 fmt_append(reply, reply_end, " <renew>\n"); 569 fmt_append(reply, reply_end, " <renew>\n");
496 for (i = 0; i < OT_PEER_TIMEOUT; ++i) 570 for (i = 0; i < OT_PEER_TIMEOUT; ++i)
497 fmt_append(reply, reply_end, " <count interval=\"%02i\">%llu</count>\n", i, ot_renewed[i]); 571 fmt_append(reply, reply_end, " <count interval=\"%02i\">%llu</count>\n", i, total[OT_COUNTER_RENEWED + i]);
498 fmt_append(reply, reply_end, " </renew>\n"); 572 fmt_append(reply, reply_end, " </renew>\n");
499 fmt_append(reply, reply_end, " <http_error>\n"); 573 fmt_append(reply, reply_end, " <http_error>\n");
500 for (i = 0; i < CODE_HTTPERROR_COUNT; ++i) 574 for (i = 0; i < CODE_HTTPERROR_COUNT; ++i)
501 fmt_append(reply, reply_end, " <count code=\"%s\">%llu</count>\n", ot_failed_request_names[i], ot_failed_request_counts[i]); 575 fmt_append(reply, reply_end, " <count code=\"%s\">%llu</count>\n", ot_failed_request_names[i], total[OT_COUNTER_FAILED + i]);
502 fmt_append(reply, reply_end, " </http_error>\n"); 576 fmt_append(reply, reply_end, " </http_error>\n");
503 fmt_append(reply, reply_end, " <mutex_stall>\n <count>%llu</count>\n </mutex_stall>\n", ot_overall_stall_count); 577 fmt_append(reply, reply_end, " <mutex_stall>\n <count>%llu</count>\n </mutex_stall>\n", total[OT_COUNTER_STALLS]);
504 fmt_append(reply, reply_end, " </debug>\n"); 578 fmt_append(reply, reply_end, " </debug>\n");
505 fmt_append(reply, reply_end, "</stats>"); 579 fmt_append(reply, reply_end, "</stats>");
506} 580}
@@ -608,33 +682,35 @@ static void stats_make(size_t *iovec_entries, struct iovec **iovector, ot_taskty
608 iovec_trim_last(iovec_entries, iovector, r); 682 iovec_trim_last(iovec_entries, iovector, r);
609} 683}
610 684
611void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) { 685void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) {
686 if (!ws)
687 return;
688
612 switch (event) { 689 switch (event) {
613 case EVENT_ACCEPT: 690 case EVENT_ACCEPT:
614 if (proto == FLAG_TCP) 691 if (proto == FLAG_TCP)
615 ot_overall_tcp_connections++; 692 COUNT(OT_COUNTER_TCP_CONNECTIONS, 1);
616 else 693 else
617 ot_overall_udp_connections++; 694 COUNT(OT_COUNTER_UDP_CONNECTIONS, 1);
618#ifdef WANT_LOG_NETWORKS 695#ifdef WANT_LOG_NETWORKS
619 stat_increase_network_count(&stats_network_counters_root, 0, event_data); 696 stat_increase_network_count(&stats_network_counters_root, 0, event_data);
620#endif 697#endif
621 break; 698 break;
622 case EVENT_ANNOUNCE: 699 case EVENT_ANNOUNCE:
623 if (proto == FLAG_TCP) 700 if (proto == FLAG_TCP)
624 ot_overall_tcp_successfulannounces++; 701 COUNT(OT_COUNTER_TCP_ANNOUNCES, 1);
625 else 702 else
626 ot_overall_udp_successfulannounces++; 703 COUNT(OT_COUNTER_UDP_ANNOUNCES, 1);
627 break; 704 break;
628 case EVENT_CONNECT: 705 case EVENT_CONNECT:
629 if (proto == FLAG_TCP) 706 if (proto == FLAG_TCP)
630 ot_overall_tcp_connects++; 707 COUNT(OT_COUNTER_TCP_CONNECTS, 1);
631 else 708 else
632 ot_overall_udp_connects++; 709 COUNT(OT_COUNTER_UDP_CONNECTS, 1);
633 break; 710 break;
634 case EVENT_COMPLETED: 711 case EVENT_COMPLETED:
635#ifdef WANT_SYSLOGS 712#ifdef WANT_SYSLOGS
636 if (event_data) { 713 {
637 struct ot_workstruct *ws = (struct ot_workstruct *)event_data;
638 char timestring[64]; 714 char timestring[64];
639 char hash_hex[42], peerid_hex[42], ip_readable[64]; 715 char hash_hex[42], peerid_hex[42], ip_readable[64];
640 struct tm time_now; 716 struct tm time_now;
@@ -655,47 +731,47 @@ void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_
655 syslog(LOG_INFO, "time=%s event=completed info_hash=%s peer_id=%s ip=%s", timestring, hash_hex, peerid_hex, ip_readable); 731 syslog(LOG_INFO, "time=%s event=completed info_hash=%s peer_id=%s ip=%s", timestring, hash_hex, peerid_hex, ip_readable);
656 } 732 }
657#endif 733#endif
658 ot_overall_completed++; 734 COUNT(OT_COUNTER_COMPLETED, 1);
659 break; 735 break;
660 case EVENT_SCRAPE: 736 case EVENT_SCRAPE:
661 if (proto == FLAG_TCP) 737 if (proto == FLAG_TCP)
662 ot_overall_tcp_successfulscrapes++; 738 COUNT(OT_COUNTER_TCP_SCRAPES, 1);
663 else 739 else
664 ot_overall_udp_successfulscrapes++; 740 COUNT(OT_COUNTER_UDP_SCRAPES, 1);
665 break; 741 break;
666 case EVENT_FULLSCRAPE: 742 case EVENT_FULLSCRAPE:
667 ot_full_scrape_count++; 743 COUNT(OT_COUNTER_FULLSCRAPES, 1);
668 ot_full_scrape_size += event_data; 744 COUNT(OT_COUNTER_FULLSCRAPE_BYTES, event_data);
669 break; 745 break;
670 case EVENT_FULLSCRAPE_REQUEST: { 746 case EVENT_FULLSCRAPE_REQUEST: {
671 ot_ip6 *ip = (ot_ip6 *)event_data; /* ugly hack to transfer ip to stats */ 747 ot_ip6 *ip = (ot_ip6 *)event_data; /* ugly hack to transfer ip to stats */
672 char _debug[512]; 748 char _debug[512];
673 int off = snprintf(_debug, sizeof(_debug), "[%08d] scrp: ", (unsigned int)(g_now_seconds - ot_start_time) / 60); 749 int off = snprintf(_debug, sizeof(_debug), "[%08d] scrp: ", (unsigned int)(time_now_seconds() - ot_start_time) / 60);
674 off += ip_format(_debug + off, *ip); 750 off += ip_format(_debug + off, *ip);
675 off += snprintf(_debug + off, sizeof(_debug) - off, " - FULL SCRAPE\n"); 751 off += snprintf(_debug + off, sizeof(_debug) - off, " - FULL SCRAPE\n");
676 (void)write(2, _debug, off); 752 (void)write(2, _debug, off);
677 ot_full_scrape_request_count++; 753 COUNT(OT_COUNTER_FULLSCRAPE_REQUESTS, 1);
678 } break; 754 } break;
679 case EVENT_FULLSCRAPE_REQUEST_GZIP: { 755 case EVENT_FULLSCRAPE_REQUEST_GZIP: {
680 ot_ip6 *ip = (ot_ip6 *)event_data; /* ugly hack to transfer ip to stats */ 756 ot_ip6 *ip = (ot_ip6 *)event_data; /* ugly hack to transfer ip to stats */
681 char _debug[512]; 757 char _debug[512];
682 int off = snprintf(_debug, sizeof(_debug), "[%08d] scrp: ", (unsigned int)(g_now_seconds - ot_start_time) / 60); 758 int off = snprintf(_debug, sizeof(_debug), "[%08d] scrp: ", (unsigned int)(time_now_seconds() - ot_start_time) / 60);
683 off += ip_format(_debug + off, *ip); 759 off += ip_format(_debug + off, *ip);
684 off += snprintf(_debug + off, sizeof(_debug) - off, " - FULL SCRAPE\n"); 760 off += snprintf(_debug + off, sizeof(_debug) - off, " - FULL SCRAPE\n");
685 (void)write(2, _debug, off); 761 (void)write(2, _debug, off);
686 ot_full_scrape_request_count++; 762 COUNT(OT_COUNTER_FULLSCRAPE_REQUESTS, 1);
687 } break; 763 } break;
688 case EVENT_FAILED: 764 case EVENT_FAILED:
689 ot_failed_request_counts[event_data]++; 765 COUNT(OT_COUNTER_FAILED + event_data, 1);
690 break; 766 break;
691 case EVENT_RENEW: 767 case EVENT_RENEW:
692 ot_renewed[event_data]++; 768 COUNT(OT_COUNTER_RENEWED + event_data, 1);
693 break; 769 break;
694 case EVENT_SYNC: 770 case EVENT_SYNC:
695 ot_overall_sync_count += event_data; 771 COUNT(OT_COUNTER_SYNCS, event_data);
696 break; 772 break;
697 case EVENT_BUCKET_LOCKED: 773 case EVENT_BUCKET_LOCKED:
698 ot_overall_stall_count++; 774 COUNT(OT_COUNTER_STALLS, 1);
699 break; 775 break;
700#ifdef WANT_SPOT_WOODPECKER 776#ifdef WANT_SPOT_WOODPECKER
701 case EVENT_WOODPECKER: 777 case EVENT_WOODPECKER:
@@ -705,7 +781,7 @@ void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_
705 break; 781 break;
706#endif 782#endif
707 case EVENT_CONNID_MISSMATCH: 783 case EVENT_CONNID_MISSMATCH:
708 ++ot_overall_udp_connectionidmissmatches; 784 COUNT(OT_COUNTER_UDP_CONNID_MISSMATCH, 1);
709 default: 785 default:
710 break; 786 break;
711 } 787 }
@@ -725,9 +801,11 @@ static void *stats_worker(void *args) {
725 801
726 (void)args; 802 (void)args;
727 803
728 while (1) { 804 while (g_opentracker_running) {
729 ot_tasktype tasktype = TASK_STATS; 805 ot_tasktype tasktype = TASK_STATS;
730 ot_taskid taskid = mutex_workqueue_poptask(&tasktype); 806 ot_taskid taskid = mutex_workqueue_poptask(&tasktype);
807 if (!taskid)
808 break;
731 stats_make(&iovec_entries, &iovector, tasktype); 809 stats_make(&iovec_entries, &iovector, tasktype);
732 if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector)) 810 if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector))
733 iovec_free(&iovec_entries, &iovector); 811 iovec_free(&iovec_entries, &iovector);
@@ -739,10 +817,11 @@ void stats_deliver(int64_t sock, int tasktype) { mutex_workqueue_pushtask(sock,
739 817
740static pthread_t thread_id; 818static pthread_t thread_id;
741void stats_init(void) { 819void stats_init(void) {
742 ot_start_time = g_now_seconds; 820 ot_start_time = time_now_seconds();
743 pthread_create(&thread_id, NULL, stats_worker, NULL); 821 pthread_create(&thread_id, NULL, stats_worker, NULL);
744} 822}
745 823
746void stats_deinit(void) { 824void stats_deinit(void) {
747 pthread_cancel(thread_id); 825 mutex_workqueue_wakeall();
826 pthread_join(thread_id, NULL);
748} 827}
diff --git a/ot_stats.h b/ot_stats.h
index fb6aad8..f484765 100644
--- a/ot_stats.h
+++ b/ot_stats.h
@@ -27,21 +27,10 @@ typedef enum {
27 EVENT_CONNID_MISSMATCH 27 EVENT_CONNID_MISSMATCH
28} ot_status_event; 28} ot_status_event;
29 29
30enum { 30void stats_counters_register(struct ot_workstruct *ws);
31 CODE_HTTPERROR_302, 31void stats_counters_retire(struct ot_workstruct *ws);
32 CODE_HTTPERROR_400,
33 CODE_HTTPERROR_400_PARAM,
34 CODE_HTTPERROR_400_COMPACT,
35 CODE_HTTPERROR_402_NOTMODEST,
36 CODE_HTTPERROR_402_PAYMENT_REQUIRED,
37 CODE_HTTPERROR_403_IP,
38 CODE_HTTPERROR_404,
39 CODE_HTTPERROR_500,
40 32
41 CODE_HTTPERROR_COUNT 33void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data);
42};
43
44void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_data);
45void stats_deliver(int64_t sock, int tasktype); 34void stats_deliver(int64_t sock, int tasktype);
46void stats_cleanup(void); 35void stats_cleanup(void);
47void return_stats_for_tracker(char **reply, const char *reply_end, int mode, int format); 36void return_stats_for_tracker(char **reply, const char *reply_end, int mode, int format);
diff --git a/ot_sync.c b/ot_sync.c
index f656bef..3668aa1 100644
--- a/ot_sync.c
+++ b/ot_sync.c
@@ -141,7 +141,7 @@ static void *sync_worker(void *args) {
141 ot_tasktype tasktype = TASK_SYNC_OUT; 141 ot_tasktype tasktype = TASK_SYNC_OUT;
142 ot_taskid taskid = mutex_workqueue_poptask(&tasktype); 142 ot_taskid taskid = mutex_workqueue_poptask(&tasktype);
143 sync_make(&iovec_entries, &iovector); 143 sync_make(&iovec_entries, &iovector);
144 stats_issue_event(EVENT_SYNC_OUT, FLAG_TCP, iovec_length(&iovec_entries, &iovector)); 144 stats_issue_event(ws, EVENT_SYNC_OUT, FLAG_TCP, iovec_length(&iovec_entries, &iovector));
145 if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector)) 145 if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector))
146 iovec_free(&iovec_entries, &iovector); 146 iovec_free(&iovec_entries, &iovector);
147 } 147 }
diff --git a/ot_udp.c b/ot_udp.c
index 4e760bb..14210c3 100644
--- a/ot_udp.c
+++ b/ot_udp.c
@@ -30,7 +30,7 @@ static void udp_generate_rijndael_round_key(void) {
30 rijndaelKeySetupEnc128(g_rijndael_round_key, (uint8_t *)key); 30 rijndaelKeySetupEnc128(g_rijndael_round_key, (uint8_t *)key);
31 31
32 g_key_of_the_hour[0] = random_uint32(); 32 g_key_of_the_hour[0] = random_uint32();
33 g_hour_of_the_key = g_now_minutes; 33 g_hour_of_the_key = time_now_minutes();
34} 34}
35 35
36/* Generate current and previous connection id for ip */ 36/* Generate current and previous connection id for ip */
@@ -39,14 +39,14 @@ static void udp_make_connectionid(uint32_t connid[2], const ot_ip6 remoteip, int
39 int i; 39 int i;
40 uint32_t current_key_of_the_hour; 40 uint32_t current_key_of_the_hour;
41 41
42 if (g_now_minutes - g_hour_of_the_key >= 60) { 42 if (time_now_minutes() - g_hour_of_the_key >= 60) {
43 uint32_t old_key_of_the_hour = g_key_of_the_hour[0]; 43 uint32_t old_key_of_the_hour = g_key_of_the_hour[0];
44 uint32_t new_key_of_the_hour = random_uint32(); 44 uint32_t new_key_of_the_hour = random_uint32();
45 /* If in the meantime another thread has performed 45 /* If in the meantime another thread has performed
46 key rotation, do not overwrite their results */ 46 key rotation, do not overwrite their results */
47 if (g_now_minutes - g_hour_of_the_key >= 60) { 47 if (time_now_minutes() - g_hour_of_the_key >= 60) {
48 /* Upgrade en bloc */ 48 /* Upgrade en bloc */
49 g_hour_of_the_key = g_now_minutes; 49 g_hour_of_the_key = time_now_minutes();
50 g_key_of_the_hour[0] = new_key_of_the_hour; 50 g_key_of_the_hour[0] = new_key_of_the_hour;
51 g_key_of_the_hour[1] = old_key_of_the_hour; 51 g_key_of_the_hour[1] = old_key_of_the_hour;
52 } 52 }
@@ -77,8 +77,8 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) {
77 return 0; 77 return 0;
78 byte_count = (size_t)result; 78 byte_count = (size_t)result;
79 79
80 stats_issue_event(EVENT_ACCEPT, FLAG_UDP, (uintptr_t)remoteip); 80 stats_issue_event(ws, EVENT_ACCEPT, FLAG_UDP, (uintptr_t)remoteip);
81 stats_issue_event(EVENT_READ, FLAG_UDP, byte_count); 81 stats_issue_event(ws, EVENT_READ, FLAG_UDP, byte_count);
82 82
83 /* Minimum udp tracker packet size, also catches error */ 83 /* Minimum udp tracker packet size, also catches error */
84 if (byte_count < 16) 84 if (byte_count < 16)
@@ -111,7 +111,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) {
111 outpacket[1] = inpacket[3]; 111 outpacket[1] = inpacket[3];
112 memcpy(&outpacket[2], &err_idmiss[0], s); 112 memcpy(&outpacket[2], &err_idmiss[0], s);
113 socket_send(serversocket, ws->outbuf, 8 + s, remoteip, remoteport); 113 socket_send(serversocket, ws->outbuf, 8 + s, remoteip, remoteport);
114 stats_issue_event(EVENT_CONNID_MISSMATCH, FLAG_UDP, 8 + s); 114 stats_issue_event(ws, EVENT_CONNID_MISSMATCH, FLAG_UDP, 8 + s);
115 return 1; 115 return 1;
116 } 116 }
117 } 117 }
@@ -128,7 +128,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) {
128 outpacket[3] = connid[1]; 128 outpacket[3] = connid[1];
129 129
130 socket_send(serversocket, ws->outbuf, 16, remoteip, remoteport); 130 socket_send(serversocket, ws->outbuf, 16, remoteip, remoteport);
131 stats_issue_event(EVENT_CONNECT, FLAG_UDP, 16); 131 stats_issue_event(ws, EVENT_CONNECT, FLAG_UDP, 16);
132 break; 132 break;
133 case 1: /* This is an announce action */ 133 case 1: /* This is an announce action */
134 /* Minimum udp announce packet size */ 134 /* Minimum udp announce packet size */
@@ -188,7 +188,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) {
188 } 188 }
189 189
190 socket_send(serversocket, ws->outbuf, ws->reply_size, remoteip, remoteport); 190 socket_send(serversocket, ws->outbuf, ws->reply_size, remoteip, remoteport);
191 stats_issue_event(EVENT_ANNOUNCE, FLAG_UDP, ws->reply_size); 191 stats_issue_event(ws, EVENT_ANNOUNCE, FLAG_UDP, ws->reply_size);
192 break; 192 break;
193 193
194 case 2: /* This is a scrape action */ 194 case 2: /* This is a scrape action */
@@ -199,7 +199,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) {
199 return_udp_scrape_for_torrent(*(ot_hash *)(((char *)inpacket) + 16 + 20 * scrape_count), ((char *)outpacket) + 8 + 12 * scrape_count); 199 return_udp_scrape_for_torrent(*(ot_hash *)(((char *)inpacket) + 16 + 20 * scrape_count), ((char *)outpacket) + 8 + 12 * scrape_count);
200 200
201 socket_send(serversocket, ws->outbuf, 8 + 12 * scrape_count, remoteip, remoteport); 201 socket_send(serversocket, ws->outbuf, 8 + 12 * scrape_count, remoteip, remoteport);
202 stats_issue_event(EVENT_SCRAPE, FLAG_UDP, scrape_count); 202 stats_issue_event(ws, EVENT_SCRAPE, FLAG_UDP, scrape_count);
203 break; 203 break;
204 } 204 }
205 return 1; 205 return 1;
@@ -210,6 +210,7 @@ static void *udp_worker(void *args) {
210 struct ot_workstruct ws; 210 struct ot_workstruct ws;
211 memset(&ws, 0, sizeof(ws)); 211 memset(&ws, 0, sizeof(ws));
212 212
213 stats_counters_register(&ws);
213 ws.inbuf = malloc(G_INBUF_SIZE); 214 ws.inbuf = malloc(G_INBUF_SIZE);
214 ws.outbuf = malloc(G_OUTBUF_SIZE); 215 ws.outbuf = malloc(G_OUTBUF_SIZE);
215#ifdef _DEBUG_HTTPERROR 216#ifdef _DEBUG_HTTPERROR
@@ -219,6 +220,8 @@ static void *udp_worker(void *args) {
219 while (g_opentracker_running) 220 while (g_opentracker_running)
220 handle_udp6(sock, &ws); 221 handle_udp6(sock, &ws);
221 222
223 stats_counters_retire(&ws);
224
222 free(ws.inbuf); 225 free(ws.inbuf);
223 free(ws.outbuf); 226 free(ws.outbuf);
224#ifdef _DEBUG_HTTPERROR 227#ifdef _DEBUG_HTTPERROR
diff --git a/ot_vector.c b/ot_vector.c
index a56d55b..8310634 100644
--- a/ot_vector.c
+++ b/ot_vector.c
@@ -208,10 +208,9 @@ void vector_redistribute_buckets(ot_peerlist *peer_list, size_t peer_size) {
208 return; 208 return;
209 209
210 /* Assume near perfect distribution */ 210 /* Assume near perfect distribution */
211 bucket_list_new = malloc(num_buckets_new * sizeof(ot_vector)); 211 bucket_list_new = calloc(1, num_buckets_new * sizeof(ot_vector));
212 if (!bucket_list_new) 212 if (!bucket_list_new)
213 return; 213 return;
214 memset(bucket_list_new, 0, num_buckets_new * sizeof(ot_vector));
215 214
216 bucket_size_new = OT_VECTOR_MIN_MEMBERS; 215 bucket_size_new = OT_VECTOR_MIN_MEMBERS;
217 while (bucket_size_new * num_buckets_new < peer_list->peer_count) 216 while (bucket_size_new * num_buckets_new < peer_list->peer_count)
diff --git a/proxy.c b/proxy.c
index a5c6bb7..b06e256 100644
--- a/proxy.c
+++ b/proxy.c
@@ -78,7 +78,8 @@ void exerr(const char *message) {
78 exit(111); 78 exit(111);
79} 79}
80 80
81void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) { 81void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) {
82 (void)ws;
82 (void)event; 83 (void)event;
83 (void)proto; 84 (void)proto;
84 (void)event_data; 85 (void)event_data;
diff --git a/tests/integration/harness.py b/tests/integration/harness.py
index 03a0561..122c36d 100644
--- a/tests/integration/harness.py
+++ b/tests/integration/harness.py
@@ -121,6 +121,7 @@ class Tracker:
121 all: LeakSanitizer reports from an exit handler, so a killed tracker 121 all: LeakSanitizer reports from an exit handler, so a killed tracker
122 reports nothing and the suite passes over leaks it never looked for. 122 reports nothing and the suite passes over leaks it never looked for.
123 """ 123 """
124 refused_to_exit = False
124 if self.process is not None and self.process.poll() is None: 125 if self.process is not None and self.process.poll() is None:
125 self.process.send_signal(signal.SIGINT) 126 self.process.send_signal(signal.SIGINT)
126 try: 127 try:
@@ -128,11 +129,16 @@ class Tracker:
128 except subprocess.TimeoutExpired: 129 except subprocess.TimeoutExpired:
129 self.process.kill() 130 self.process.kill()
130 self.process.wait() 131 self.process.wait()
132 refused_to_exit = True
131 complaints = [ 133 complaints = [
132 line 134 line
133 for line in self.output().splitlines() 135 for line in self.output().splitlines()
134 if "Sanitizer" in line or "runtime error" in line or line.startswith("ERROR") 136 if "Sanitizer" in line or "runtime error" in line or line.startswith("ERROR")
135 ] 137 ]
138 if refused_to_exit:
139 # Killing it throws away whatever the sanitizer had to say on the
140 # way out, so a test that ends this way has checked nothing.
141 complaints.append("the tracker did not exit within 5 s of SIGINT and had to be killed")
136 self._cleanup() 142 self._cleanup()
137 return complaints 143 return complaints
138 144
diff --git a/tests/integration/test_concurrency.py b/tests/integration/test_concurrency.py
new file mode 100644
index 0000000..1374852
--- /dev/null
+++ b/tests/integration/test_concurrency.py
@@ -0,0 +1,209 @@
1"""Several clients at once.
2
3The rest of the suite talks to the tracker one request at a time, which
4leaves the worker threads and the bucket locks almost untouched. These reach
5them.
6
7What they assert is weak on purpose. A lost update needs two threads inside
8the same few instructions, which is rare enough that these tests pass
9against code that has the race: measured, eight runs in a row. Their worth
10is that they make the racing code run at all, so that "make check-tsan"
11has something to report. Reaching the code and judging it are two jobs, and
12these only do the first.
13"""
14
15import struct
16import threading
17import unittest
18
19from harness import TrackerTestCase, bdecode
20
21THREADS = 8
22PER_THREAD = 40
23
24
25def run_threads(target, count=THREADS):
26 errors = []
27
28 def guarded(index):
29 try:
30 target(index)
31 except Exception as exception: # noqa: BLE001 - reported, not swallowed
32 errors.append("thread %d: %r" % (index, exception))
33
34 threads = [threading.Thread(target=guarded, args=(i,)) for i in range(count)]
35 for thread in threads:
36 thread.start()
37 for thread in threads:
38 thread.join()
39 return errors
40
41
42class ConcurrentAnnounce(TrackerTestCase):
43 def test_every_peer_of_one_swarm_arrives(self):
44 """One torrent, one bucket, every thread contending for its lock."""
45 info_hash = self.hash_for("concurrent-one")
46
47 def announce(index):
48 for i in range(PER_THREAD):
49 self.tracker.announce(info_hash, port=10000 + index * PER_THREAD + i, left=0)
50
51 self.assertEqual(run_threads(announce), [])
52 answer = self.tracker.scrape(info_hash)
53 self.assertEqual(answer[b"files"][info_hash][b"complete"], THREADS * PER_THREAD)
54
55 def test_inserts_spread_over_buckets(self):
56 """Every thread inserts into a different bucket, so the bucket locks
57 serialise nothing between them, and the counter they all adjust is
58 the one global the locks do not cover. Under check-tsan this is what
59 names mutex_bucket_unlock if that counter stops being atomic."""
60
61 def announce(index):
62 for i in range(PER_THREAD):
63 self.tracker.announce(struct.pack(">HH", index, i) + b"\0" * 16, port=11000 + i, left=0)
64
65 self.assertEqual(run_threads(announce), [])
66 self.assertEqual(self.tracker.torrent_count(), THREADS * PER_THREAD)
67
68 def test_a_swarm_that_fills_and_empties_at_once(self):
69 """Inserts and removals racing in the same buckets."""
70 info_hash = self.hash_for("concurrent-churn")
71 for port in range(12000, 12000 + THREADS * PER_THREAD):
72 self.tracker.announce(info_hash, port=port, left=0)
73
74 def churn(index):
75 base = 12000 + index * PER_THREAD
76 for i in range(PER_THREAD):
77 self.tracker.announce(info_hash, port=base + i, left=0, event="stopped")
78 self.tracker.announce(info_hash, port=base + i, left=100)
79
80 self.assertEqual(run_threads(churn), [])
81 answer = self.tracker.scrape(info_hash)
82 self.assertEqual(answer[b"files"][info_hash][b"incomplete"], THREADS * PER_THREAD)
83 self.assertEqual(answer[b"files"][info_hash][b"complete"], 0)
84
85
86class ConcurrentMixedBearers(TrackerTestCase):
87 def test_tcp_and_udp_reach_the_same_swarm(self):
88 info_hash = self.hash_for("concurrent-mixed")
89
90 def over_tcp(index):
91 for i in range(PER_THREAD):
92 self.tracker.announce(info_hash, port=13000 + index * PER_THREAD + i, left=0)
93
94 def over_udp(index):
95 sock, connection_id = self.tracker.udp_connect()
96 try:
97 for i in range(PER_THREAD):
98 self.tracker.udp_announce(
99 sock, connection_id, info_hash, left=0, port=14000 + index * PER_THREAD + i
100 )
101 finally:
102 sock.close()
103
104 errors = []
105 threads = [threading.Thread(target=lambda i=i: over_tcp(i)) for i in range(THREADS // 2)]
106 threads += [threading.Thread(target=lambda i=i: over_udp(i)) for i in range(THREADS // 2)]
107 for thread in threads:
108 thread.start()
109 for thread in threads:
110 thread.join()
111
112 self.assertEqual(errors, [])
113 answer = self.tracker.scrape(info_hash)
114 self.assertEqual(answer[b"files"][info_hash][b"complete"], THREADS * PER_THREAD)
115
116
117class ConcurrentWorkerTasks(TrackerTestCase):
118 def test_full_scrapes_while_the_swarm_moves(self):
119 """The full scrape runs on a worker thread, walking the same buckets
120 the announces are changing."""
121 for i in range(200):
122 self.tracker.announce(struct.pack(">I", i) + b"\0" * 16, port=15000 + i, left=0)
123
124 def announce(index):
125 for i in range(PER_THREAD):
126 self.tracker.announce(struct.pack(">I", i) + b"\0" * 16, port=16000 + index * PER_THREAD + i, left=0)
127
128 def scrape(index):
129 for _ in range(6):
130 answer = bdecode(self.tracker.get("/scrape"))
131 for counts in answer[b"files"].values():
132 if counts[b"complete"] < 0 or counts[b"incomplete"] < 0:
133 raise AssertionError("negative count in a full scrape: %r" % counts)
134
135 threads = [threading.Thread(target=announce, args=(i,)) for i in range(THREADS // 2)]
136 threads += [threading.Thread(target=scrape, args=(i,)) for i in range(2)]
137 for thread in threads:
138 thread.start()
139 for thread in threads:
140 thread.join()
141
142 self.assertEqual(self.tracker.torrent_count(), 200)
143
144 def test_stats_while_the_swarm_moves(self):
145 def announce(index):
146 for i in range(PER_THREAD):
147 self.tracker.announce(struct.pack(">HH", index, i) + b"\0" * 16, port=17000 + i, left=i % 2)
148
149 def poll(index):
150 for _ in range(10):
151 for mode in ("everything", "top100", "peer", "torr", "s24s"):
152 self.tracker.stats(mode)
153
154 threads = [threading.Thread(target=announce, args=(i,)) for i in range(THREADS // 2)]
155 threads += [threading.Thread(target=poll, args=(i,)) for i in range(2)]
156 for thread in threads:
157 thread.start()
158 for thread in threads:
159 thread.join()
160
161 self.assertEqual(self.tracker.torrent_count(), (THREADS // 2) * PER_THREAD)
162
163
164class CountersAcrossContexts(TrackerTestCase):
165 """With udp workers configured, every one of them counts into a block of
166 its own, and the figure reported is those blocks added up. A registry
167 that lost a block, or a worker that wrote into somebody else's, would
168 show here and nowhere else in the suite: without this directive udp is
169 served by the main loop, which is a single context."""
170
171 config = "listen.udp.workers 4\n"
172
173 def test_the_udp_count_is_the_sum_of_every_worker(self):
174 before = int(self.tracker.stats("udp4").splitlines()[0])
175
176 packets = 0
177 for round_ in range(6):
178 sock, connection_id = self.tracker.udp_connect() # one packet
179 packets += 1
180 try:
181 for i in range(10):
182 self.tracker.udp_announce(sock, connection_id, self.hash_for("ctx%d" % i), left=0)
183 packets += 1
184 finally:
185 sock.close()
186
187 after = int(self.tracker.stats("udp4").splitlines()[0])
188 self.assertEqual(after - before, packets, "every packet counted exactly once")
189
190 def test_announces_from_several_workers_reach_one_swarm(self):
191 info_hash = self.hash_for("ctx-swarm")
192
193 def announce(index):
194 sock, connection_id = self.tracker.udp_connect()
195 try:
196 for i in range(PER_THREAD):
197 self.tracker.udp_announce(
198 sock, connection_id, info_hash, left=0, port=18000 + index * PER_THREAD + i
199 )
200 finally:
201 sock.close()
202
203 self.assertEqual(run_threads(announce, 4), [])
204 answer = self.tracker.scrape(info_hash)
205 self.assertEqual(answer[b"files"][info_hash][b"complete"], 4 * PER_THREAD)
206
207
208if __name__ == "__main__":
209 unittest.main()
diff --git a/tests/stubs.c b/tests/stubs.c
index 09a7312..a8e6962 100644
--- a/tests/stubs.c
+++ b/tests/stubs.c
@@ -16,8 +16,7 @@ void free_peerlist(ot_peerlist *peer_list) { (void)peer_list; }
16/* ot_accesslist.c is built with WANT_ACCESSLIST_BLACK so that the hash parser 16/* 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, 17 is reachable. These two belong to accesslist_readfile and the reload worker,
18 neither of which any test drives. */ 18 neither of which any test drives. */
19time_t g_now_seconds; 19_Atomic int g_opentracker_running;
20volatile int g_opentracker_running;
21 20
22/* ot_mutex.c wakes the main loop through the self pipe when a result is 21/* ot_mutex.c wakes the main loop through the self pipe when a result is
23 ready. The queue tests never get that far. */ 22 ready. The queue tests never get that far. */
diff --git a/trackerlogic.c b/trackerlogic.c
index 1f38f07..477c4a6 100644
--- a/trackerlogic.c
+++ b/trackerlogic.c
@@ -117,8 +117,8 @@ size_t add_peer_to_torrent_and_return_peers(PROTO_FLAG proto, struct ot_workstru
117 } else 117 } else
118 clean_single_torrent(torrent); 118 clean_single_torrent(torrent);
119 119
120 torrent->peer_list6->base = g_now_minutes; 120 torrent->peer_list6->base = time_now_minutes();
121 torrent->peer_list4->base = g_now_minutes; 121 torrent->peer_list4->base = time_now_minutes();
122 122
123 peer_list = peer_size == OT_PEER_SIZE6 ? torrent->peer_list6 : torrent->peer_list4; 123 peer_list = peer_size == OT_PEER_SIZE6 ? torrent->peer_list6 : torrent->peer_list4;
124 124
@@ -149,16 +149,16 @@ size_t add_peer_to_torrent_and_return_peers(PROTO_FLAG proto, struct ot_workstru
149 peer_list->peer_count++; 149 peer_list->peer_count++;
150 if (OT_PEERFLAG(ws->peer) & PEER_FLAG_COMPLETED) { 150 if (OT_PEERFLAG(ws->peer) & PEER_FLAG_COMPLETED) {
151 peer_list->down_count++; 151 peer_list->down_count++;
152 stats_issue_event(EVENT_COMPLETED, 0, (uintptr_t)ws); 152 stats_issue_event(ws, EVENT_COMPLETED, 0, 0);
153 } 153 }
154 if (OT_PEERFLAG(ws->peer) & PEER_FLAG_SEEDING) 154 if (OT_PEERFLAG(ws->peer) & PEER_FLAG_SEEDING)
155 peer_list->seed_count++; 155 peer_list->seed_count++;
156 156
157 } else { 157 } else {
158 stats_issue_event(EVENT_RENEW, 0, OT_PEERTIME(peer_dest, peer_size)); 158 stats_issue_event(ws, EVENT_RENEW, 0, OT_PEERTIME(peer_dest, peer_size));
159#ifdef WANT_SPOT_WOODPECKER 159#ifdef WANT_SPOT_WOODPECKER
160 if ((OT_PEERTIME(peer_dest, peer_size) > 0) && (OT_PEERTIME(peer_dest, peer_size) < 20)) 160 if ((OT_PEERTIME(peer_dest, peer_size) > 0) && (OT_PEERTIME(peer_dest, peer_size) < 20))
161 stats_issue_event(EVENT_WOODPECKER, 0, (uintptr_t)&ws->peer); 161 stats_issue_event(ws, EVENT_WOODPECKER, 0, (uintptr_t)&ws->peer);
162#endif 162#endif
163#ifdef WANT_SYNC_LIVE 163#ifdef WANT_SYNC_LIVE
164 /* Won't live sync peers that come back too fast. Only exception: 164 /* Won't live sync peers that come back too fast. Only exception:
@@ -176,7 +176,7 @@ size_t add_peer_to_torrent_and_return_peers(PROTO_FLAG proto, struct ot_workstru
176 peer_list->seed_count++; 176 peer_list->seed_count++;
177 if (!(OT_PEERFLAG_D(peer_dest, peer_size) & PEER_FLAG_COMPLETED) && (OT_PEERFLAG(ws->peer) & PEER_FLAG_COMPLETED)) { 177 if (!(OT_PEERFLAG_D(peer_dest, peer_size) & PEER_FLAG_COMPLETED) && (OT_PEERFLAG(ws->peer) & PEER_FLAG_COMPLETED)) {
178 peer_list->down_count++; 178 peer_list->down_count++;
179 stats_issue_event(EVENT_COMPLETED, 0, (uintptr_t)ws); 179 stats_issue_event(ws, EVENT_COMPLETED, 0, 0);
180 } 180 }
181 if (OT_PEERFLAG_D(peer_dest, peer_size) & PEER_FLAG_COMPLETED) 181 if (OT_PEERFLAG_D(peer_dest, peer_size) & PEER_FLAG_COMPLETED)
182 OT_PEERFLAG(ws->peer) |= PEER_FLAG_COMPLETED; 182 OT_PEERFLAG(ws->peer) |= PEER_FLAG_COMPLETED;
@@ -571,6 +571,13 @@ void trackerlogic_deinit(void) {
571 int delta_torrentcount = 0; 571 int delta_torrentcount = 0;
572 size_t bucket, j; 572 size_t bucket, j;
573 573
574 /* Stop the background workers first */
575 stats_deinit();
576 livesync_deinit();
577 accesslist_deinit();
578 fullscrape_deinit();
579 clean_deinit();
580
574 /* Free all torrents... */ 581 /* Free all torrents... */
575 for (bucket = 0; bucket < OT_BUCKET_COUNT; ++bucket) { 582 for (bucket = 0; bucket < OT_BUCKET_COUNT; ++bucket) {
576 ot_vector *torrents_list = mutex_bucket_lock(bucket); 583 ot_vector *torrents_list = mutex_bucket_lock(bucket);
@@ -586,12 +593,6 @@ void trackerlogic_deinit(void) {
586 mutex_bucket_unlock(bucket, delta_torrentcount); 593 mutex_bucket_unlock(bucket, delta_torrentcount);
587 } 594 }
588 595
589 /* Deinitialise background worker threads */
590 stats_deinit();
591 livesync_deinit();
592 accesslist_deinit();
593 fullscrape_deinit();
594 clean_deinit();
595 /* Release mutexes */ 596 /* Release mutexes */
596 mutex_deinit(); 597 mutex_deinit();
597} 598}
diff --git a/trackerlogic.h b/trackerlogic.h
index 63abf78..ad6c911 100644
--- a/trackerlogic.h
+++ b/trackerlogic.h
@@ -6,6 +6,7 @@
6#ifndef OT_TRACKERLOGIC_H__ 6#ifndef OT_TRACKERLOGIC_H__
7#define OT_TRACKERLOGIC_H__ 7#define OT_TRACKERLOGIC_H__
8 8
9#include <stdatomic.h>
9#include <stdint.h> 10#include <stdint.h>
10#include <stdlib.h> 11#include <stdlib.h>
11#include <sys/time.h> 12#include <sys/time.h>
@@ -67,10 +68,28 @@ typedef struct {
67 on startup */ 68 on startup */
68#define RANDOMTORRENTS (1024 * 1024 * 1) 69#define RANDOMTORRENTS (1024 * 1024 * 1)
69 70
70/* From opentracker.c */ 71#if defined(CLOCK_REALTIME_COARSE)
71extern time_t g_now_seconds; 72#define OT_CLOCK_CHEAP CLOCK_REALTIME_COARSE /* Linux */
72extern volatile int g_opentracker_running; 73#elif defined(CLOCK_REALTIME_FAST)
73#define g_now_minutes (g_now_seconds / 60) 74#define OT_CLOCK_CHEAP CLOCK_REALTIME_FAST /* the BSDs */
75#else
76#define OT_CLOCK_CHEAP CLOCK_REALTIME /* macOS and whatever else */
77#endif
78
79static inline ot_time time_now_seconds(void) {
80 struct timespec now;
81 clock_gettime(OT_CLOCK_CHEAP, &now);
82 return (ot_time)now.tv_sec;
83}
84
85static inline ot_time time_now_minutes(void) { return time_now_seconds() / 60; }
86
87/* For the deadlines the tracker sets on itself, which read better as the
88 span they allow than as an addition to the current second */
89static inline ot_time time_future_seconds(int span) { return time_now_seconds() + span; }
90/* Cleared from the interrupt handler and read by every worker */
91extern _Atomic int g_opentracker_running;
92
74 93
75extern uint32_t g_tracker_id; 94extern uint32_t g_tracker_id;
76typedef enum { FLAG_TCP, FLAG_UDP, FLAG_MCA, FLAG_SELFPIPE } PROTO_FLAG; 95typedef enum { FLAG_TCP, FLAG_UDP, FLAG_MCA, FLAG_SELFPIPE } PROTO_FLAG;
@@ -129,8 +148,56 @@ struct ot_peerlist {
129}; 148};
130#define OT_PEERLIST_HASBUCKETS(peer_list) ((peer_list)->peers.size > (peer_list)->peers.space) 149#define OT_PEERLIST_HASBUCKETS(peer_list) ((peer_list)->peers.size > (peer_list)->peers.space)
131 150
151/* The http replies the tracker counts when it has to refuse a request. */
152enum {
153 CODE_HTTPERROR_302,
154 CODE_HTTPERROR_400,
155 CODE_HTTPERROR_400_PARAM,
156 CODE_HTTPERROR_400_COMPACT,
157 CODE_HTTPERROR_402_NOTMODEST,
158 CODE_HTTPERROR_402_PAYMENT_REQUIRED,
159 CODE_HTTPERROR_403_IP,
160 CODE_HTTPERROR_404,
161 CODE_HTTPERROR_500,
162
163 CODE_HTTPERROR_COUNT
164};
165
166/* One statistic. Used when workers report stats to their workstruct */
167typedef _Atomic unsigned long long ot_counter;
168
169/* What each thread counts for itself. The two runs at the end are indexed by
170 the http error code and by the age in minutes of a renewed peer */
171typedef enum {
172 OT_COUNTER_TCP_CONNECTIONS,
173 OT_COUNTER_UDP_CONNECTIONS,
174 OT_COUNTER_TCP_ANNOUNCES,
175 OT_COUNTER_UDP_ANNOUNCES,
176 OT_COUNTER_TCP_SCRAPES,
177 OT_COUNTER_UDP_SCRAPES,
178 OT_COUNTER_TCP_CONNECTS,
179 OT_COUNTER_UDP_CONNECTS,
180 OT_COUNTER_UDP_CONNID_MISSMATCH,
181 OT_COUNTER_COMPLETED,
182 OT_COUNTER_FULLSCRAPES,
183 OT_COUNTER_FULLSCRAPE_REQUESTS,
184 OT_COUNTER_FULLSCRAPE_BYTES,
185 OT_COUNTER_SYNCS,
186 OT_COUNTER_STALLS,
187
188 OT_COUNTER_FAILED, /* CODE_HTTPERROR_COUNT of them */
189 OT_COUNTER_RENEWED = OT_COUNTER_FAILED + CODE_HTTPERROR_COUNT, /* OT_PEER_TIMEOUT of them */
190
191 OT_COUNTER_COUNT = OT_COUNTER_RENEWED + OT_PEER_TIMEOUT
192} ot_counter_index;
193
132struct ot_workstruct { 194struct ot_workstruct {
133 /* Thread specific, static */ 195 /* Thread specific, static */
196
197 /* This thread's own statistics, and the link ot_stats keeps them on */
198 ot_counter counters[OT_COUNTER_COUNT];
199 struct ot_workstruct *counters_next;
200
134 char *inbuf; 201 char *inbuf;
135#define G_INBUF_SIZE 8192 202#define G_INBUF_SIZE 8192
136 char *outbuf; 203 char *outbuf;
generated by cgit v1.3 (git 2.53.0) at 2026-10-27 23:39:39 +0000