From 6cbd21542cf7a751e6f0f38864e7e9eae8fa3b14 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Tue, 27 Oct 2026 21:32:24 +0100 Subject: Plug in the new per-thread stats counter, also make sure that the ot_clean sleep does not stall tracker exit --- opentracker.c | 41 ++++++++--- ot_clean.c | 33 ++++++++- ot_fullscrape.c | 5 +- ot_http.c | 10 +-- ot_livesync.c | 2 +- ot_mutex.c | 16 +++- ot_mutex.h | 1 + ot_stats.c | 221 ++++++++++++++++++++++++++++++++++++++------------------ ot_stats.h | 3 +- ot_sync.c | 2 +- ot_udp.c | 15 ++-- proxy.c | 3 +- trackerlogic.c | 21 +++--- 13 files changed, 259 insertions(+), 114 deletions(-) diff --git a/opentracker.c b/opentracker.c index eb97b42..34b8fd5 100644 --- a/opentracker.c +++ b/opentracker.c @@ -43,15 +43,18 @@ uint32_t g_tracker_id; _Atomic int g_opentracker_running = 1; int g_self_pipe[2]; - static void panic(const char *routine) __attribute__((noreturn)); static void panic(const char *routine) { fprintf(stderr, "%s: %s\n", routine, strerror(errno)); exit(111); } +/* Say we are stopping, wake the loop, and let main do the work on its own stack. */ static void signal_handler(int s) { if (s == SIGINT) { + const char byte = 0; + int saved = errno; + /* Any new interrupt signal quits the application */ signal(SIGINT, SIG_DFL); @@ -59,13 +62,8 @@ static void signal_handler(int s) { but cancel their operations and return */ g_opentracker_running = 0; - trackerlogic_deinit(); - -#ifdef WANT_SYSLOGS - closelog(); -#endif - - exit(0); + (void)!write(g_self_pipe[1], &byte, 1); + errno = saved; } } @@ -207,7 +205,7 @@ static void handle_write(const int64_t sock) { handle_dead(sock); } -static void handle_accept(const int64_t serversocket) { +static void handle_accept(const int64_t serversocket, struct ot_workstruct *ws) { struct http_data *cookie; int64_t sock; ot_ip6 ip; @@ -231,7 +229,7 @@ static void handle_accept(const int64_t serversocket) { continue; } - stats_issue_event(EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip); + stats_issue_event(ws, EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip); } } @@ -248,6 +246,8 @@ static void *server_mainloop(void *args) { (void)args; /* Initialize our "thread local storage" */ + memset(&ws, 0, sizeof(ws)); + stats_counters_register(&ws); ws.inbuf = malloc(G_INBUF_SIZE); ws.outbuf = malloc(G_OUTBUF_SIZE); #ifdef _DEBUG_HTTPERROR @@ -259,7 +259,7 @@ static void *server_mainloop(void *args) { random_bytes(ws.rand48_state, sizeof(ws.rand48_state)); - for (;;) { + while (g_opentracker_running) { ot_event ready[OT_EVENT_BATCH]; int64_t sock; int count, i; @@ -278,7 +278,7 @@ static void *server_mainloop(void *args) { cookie = fdstate_cookie((int)sock); if ((intptr_t)cookie == FLAG_TCP) - handle_accept(sock); + handle_accept(sock, &ws); else if ((intptr_t)cookie == FLAG_UDP) handle_udp6(sock, &ws); else if ((intptr_t)cookie == FLAG_SELFPIPE) @@ -306,6 +306,16 @@ static void *server_mainloop(void *args) { livesync_ticker(); } + + /* This work struct is about to go out of scope, so hand its tally over + before anything can read it there */ + stats_counters_retire(&ws); + + free(ws.inbuf); + free(ws.outbuf); +#ifdef _DEBUG_HTTPERROR + free(ws.debugbuf); +#endif return 0; } @@ -510,6 +520,13 @@ int main(int argc, char **argv) { server_mainloop(0); + /* The loop has left, so no thread is serving any more and the teardown + can run where a debugger and the sanitizers can follow it */ + trackerlogic_deinit(); +#ifdef WANT_SYSLOGS + closelog(); +#endif + event_deinit(); fdstate_deinit(); return 0; diff --git a/ot_clean.c b/ot_clean.c index dbfa04e..77ea239 100644 --- a/ot_clean.c +++ b/ot_clean.c @@ -4,6 +4,7 @@ $id$ */ /* System */ +#include #include #include #include @@ -103,6 +104,29 @@ int clean_single_torrent(ot_torrent *torrent) { return clean_single_peer_list(torrent->peer_list6, OT_PEER_SIZE6) * clean_single_peer_list(torrent->peer_list4, OT_PEER_SIZE4); } +static pthread_mutex_t g_clean_mutex = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t g_clean_wakeup = PTHREAD_COND_INITIALIZER; + +static void clean_sleep(void) { + struct timespec deadline; + + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += OT_CLEAN_SLEEP / 1000000; + deadline.tv_nsec += (OT_CLEAN_SLEEP % 1000000) * 1000; + if (deadline.tv_nsec >= 1000000000) { + ++deadline.tv_sec; + deadline.tv_nsec -= 1000000000; + } + + /* Both sides hold the mutex, so the flag cannot be cleared and signalled + in the window between testing it and waiting on it */ + pthread_mutex_lock(&g_clean_mutex); + while (g_opentracker_running) + if (pthread_cond_timedwait(&g_clean_wakeup, &g_clean_mutex, &deadline) == ETIMEDOUT) + break; + pthread_mutex_unlock(&g_clean_mutex); +} + /* Clean up all peers in current bucket, remove timedout pools and torrents */ static void *clean_worker(void *args) { @@ -125,7 +149,7 @@ static void *clean_worker(void *args) { mutex_bucket_unlock(bucket, delta_torrentcount); if (!g_opentracker_running) return NULL; - usleep(OT_CLEAN_SLEEP); + clean_sleep(); } stats_cleanup(); #ifdef WANT_ACCESSLIST @@ -138,4 +162,9 @@ static void *clean_worker(void *args) { static pthread_t thread_id; void clean_init(void) { pthread_create(&thread_id, NULL, clean_worker, NULL); } -void clean_deinit(void) { pthread_cancel(thread_id); } +void clean_deinit(void) { + pthread_mutex_lock(&g_clean_mutex); + pthread_cond_signal(&g_clean_wakeup); + pthread_mutex_unlock(&g_clean_mutex); + pthread_join(thread_id, NULL); +} 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) { while (g_opentracker_running) { ot_tasktype tasktype = TASK_FULLSCRAPE; ot_taskid taskid = mutex_workqueue_poptask(&tasktype); + if (!taskid) + break; #ifdef WANT_COMPRESSION_ZSTD if (tasktype & TASK_FLAG_ZSTD) fullscrape_make_zstd(taskid, tasktype); @@ -77,7 +79,8 @@ void fullscrape_init(void) { } void fullscrape_deinit(void) { - pthread_cancel( thread_id ); + mutex_workqueue_wakeall(); + pthread_join( thread_id, NULL ); } void fullscrape_deliver( int64_t sock, ot_tasktype tasktype ) { diff --git a/ot_http.c b/ot_http.c index c4f74db..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) #ifdef _DEBUG_HTTPERROR fprintf(stderr, "DEBUG: invalid request was: %s\n", ws->debugbuf); #endif - stats_issue_event(EVENT_FAILED, FLAG_TCP, code); + stats_issue_event(ws, EVENT_FAILED, FLAG_TCP, code); http_senddata(sock, ws); return ws->reply_size = -2; } @@ -402,7 +402,7 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct * #endif #endif - stats_issue_event(EVENT_FULLSCRAPE_REQUEST, 0, (uintptr_t)cookie->ip); + stats_issue_event(ws, EVENT_FULLSCRAPE_REQUEST, 0, (uintptr_t)cookie->ip); #ifdef _DEBUG_HTTPERROR fprintf(stderr, "%s", ws->debugbuf); @@ -474,7 +474,7 @@ static ssize_t http_handle_scrape(const int64_t sock, struct ot_workstruct *ws, /* Enough for http header + whole scrape string */ return_tcp_scrape_for_torrent((const ot_hash *)multiscrape_buf, numwant, &reply, ws->outbuf + G_OUTBUF_SIZE); ws->reply_size = reply - ws->reply; - stats_issue_event(EVENT_SCRAPE, FLAG_TCP, ws->reply_size); + stats_issue_event(ws, EVENT_SCRAPE, FLAG_TCP, ws->reply_size); return ws->reply_size; } @@ -682,7 +682,7 @@ static ssize_t http_handle_announce(const int64_t sock, struct ot_workstruct #endif /* XXX DEBUG - stats_issue_event( EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ws->reply ); + stats_issue_event(ws, EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ws->reply ); */ /* Scanned whole query string */ @@ -694,7 +694,7 @@ static ssize_t http_handle_announce(const int64_t sock, struct ot_workstruct else ws->reply_size = add_peer_to_torrent_and_return_peers(FLAG_TCP, ws, numwant); - stats_issue_event(EVENT_ANNOUNCE, FLAG_TCP, ws->reply_size); + stats_issue_event(ws, EVENT_ANNOUNCE, FLAG_TCP, ws->reply_size); return ws->reply_size; } diff --git a/ot_livesync.c b/ot_livesync.c index fb0bef2..45271bc 100644 --- a/ot_livesync.c +++ b/ot_livesync.c @@ -143,7 +143,7 @@ static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size) off += sizeof(ot_hash) + peer_size; } - stats_issue_event(EVENT_SYNC, 0, (ws->request_size - sizeof(g_tracker_id) - sizeof(uint32_t)) / ((ssize_t)sizeof(ot_hash) + peer_size)); + stats_issue_event(ws, EVENT_SYNC, 0, (ws->request_size - sizeof(g_tracker_id) - sizeof(uint32_t)) / ((ssize_t)sizeof(ot_hash) + peer_size)); } /* Tickle the live sync module from time to time, so no events get diff --git a/ot_mutex.c b/ot_mutex.c index 37d9962..c44efed 100644 --- a/ot_mutex.c +++ b/ot_mutex.c @@ -6,6 +6,7 @@ /* System */ #include #include +#include #include #include #include @@ -145,9 +146,13 @@ ot_taskid mutex_workqueue_poptask(ot_tasktype *tasktype) { break; } - /* Wait until the next task is being fed */ - if (!taskid) + /* Wait until the next task is being fed, unless the tracker is on its + way out, in which case the worker is meant to leave rather than wait */ + if (!taskid) { + if (!g_opentracker_running) + break; pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex); + } } /* Release lock */ @@ -271,6 +276,13 @@ void mutex_init(void) { memset(all_torrents, 0, sizeof(all_torrents)); } +/* Let every worker parked on the queue see that the tracker is stopping */ +void mutex_workqueue_wakeall(void) { + pthread_mutex_lock(&tasklist_mutex); + pthread_cond_broadcast(&tasklist_being_filled); + pthread_mutex_unlock(&tasklist_mutex); +} + void mutex_deinit(void) { size_t i; 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 @@ #include "trackerlogic.h" void mutex_init(void); +void mutex_workqueue_wakeall(void); void mutex_deinit(void); ot_vector *mutex_bucket_lock(int bucket); diff --git a/ot_stats.c b/ot_stats.c index 63d9eda..5f97a2f 100644 --- a/ot_stats.c +++ b/ot_stats.c @@ -6,6 +6,7 @@ /* System */ #include #include +#include #include #include #include @@ -36,27 +37,58 @@ static void stats_make(size_t *iovec_entries, struct iovec **iovector, ot_tasktype mode); #define OT_STATS_TMPSIZE 8192 -/* Clumsy counters... to be rethought */ -static unsigned long long ot_overall_tcp_connections; -static unsigned long long ot_overall_udp_connections; -static unsigned long long ot_overall_tcp_successfulannounces; -static unsigned long long ot_overall_udp_successfulannounces; -static unsigned long long ot_overall_tcp_successfulscrapes; -static unsigned long long ot_overall_udp_successfulscrapes; -static unsigned long long ot_overall_udp_connectionidmissmatches; -static unsigned long long ot_overall_tcp_connects; -static unsigned long long ot_overall_udp_connects; -static unsigned long long ot_overall_completed; -static unsigned long long ot_full_scrape_count; -static unsigned long long ot_full_scrape_request_count; -static unsigned long long ot_full_scrape_size; -static unsigned long long ot_failed_request_counts[CODE_HTTPERROR_COUNT]; +/* Each serving thread keeps its own statistics tally. The list is linked + through intrusive next pointers within the workstruct. + + Registration happens once per thread and the adding up once per statistics + request, so this lock is never on anybody's hot path. */ +static struct ot_workstruct *g_counter_contexts; +static ot_counter g_retired[OT_COUNTER_COUNT]; +static pthread_mutex_t g_counter_mutex = PTHREAD_MUTEX_INITIALIZER; + +void stats_counters_register(struct ot_workstruct *ws) { + pthread_mutex_lock(&g_counter_mutex); + ws->counters_next = g_counter_contexts; + g_counter_contexts = ws; + pthread_mutex_unlock(&g_counter_mutex); +} + +void stats_counters_retire(struct ot_workstruct *ws) { + struct ot_workstruct **link; + size_t i; + + pthread_mutex_lock(&g_counter_mutex); + for (link = &g_counter_contexts; *link; link = &(*link)->counters_next) + if (*link == ws) { + *link = ws->counters_next; + break; + } + + /* When retiring a worker thread, collect the counter tally in the global */ + for (i = 0; i < OT_COUNTER_COUNT; ++i) + atomic_fetch_add_explicit(g_retired + i, atomic_load_explicit(ws->counters + i, memory_order_relaxed), memory_order_relaxed); + pthread_mutex_unlock(&g_counter_mutex); +} + +/* What every thread has counted so far, added together */ +static void counters_total(unsigned long long *total) { + struct ot_workstruct *ws; + size_t i; + + pthread_mutex_lock(&g_counter_mutex); + for (i = 0; i < OT_COUNTER_COUNT; ++i) + total[i] = atomic_load_explicit(g_retired + i, memory_order_relaxed); + for (ws = g_counter_contexts; ws; ws = ws->counters_next) + for (i = 0; i < OT_COUNTER_COUNT; ++i) + total[i] += atomic_load_explicit(ws->counters + i, memory_order_relaxed); + pthread_mutex_unlock(&g_counter_mutex); +} + +#define COUNT(index, by) atomic_fetch_add_explicit(ws->counters + (index), (unsigned long long)(by), memory_order_relaxed) + static const char *ot_failed_request_names[] = { "302 Redirect", "400 Parse Error", "400 Invalid Parameter", "400 Invalid Parameter (compact=0)", "400 Not Modest", "402 Payment Required", "403 Access Denied", "404 Not found", "500 Internal Server Error"}; -static unsigned long long ot_renewed[OT_PEER_TIMEOUT]; -static unsigned long long ot_overall_sync_count; -static unsigned long long ot_overall_stall_count; static time_t ot_start_time; @@ -347,39 +379,61 @@ static unsigned long events_per_time(unsigned long long events, time_t t) { retu static void stats_connections_mrtg(char **reply, const char *reply_end) { ot_time t = time(NULL) - ot_start_time; + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker connections, %lu conns/s :: %lu success/s.", - ot_overall_tcp_connections + ot_overall_udp_connections, - ot_overall_tcp_successfulannounces + ot_overall_udp_successfulannounces + ot_overall_udp_connects, (int)t, (int)(t / 3600), - events_per_time(ot_overall_tcp_connections + ot_overall_udp_connections, t), - events_per_time(ot_overall_tcp_successfulannounces + ot_overall_udp_successfulannounces + ot_overall_udp_connects, t)); + total[OT_COUNTER_TCP_CONNECTIONS] + total[OT_COUNTER_UDP_CONNECTIONS], + total[OT_COUNTER_TCP_ANNOUNCES] + total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], (int)t, (int)(t / 3600), + events_per_time(total[OT_COUNTER_TCP_CONNECTIONS] + total[OT_COUNTER_UDP_CONNECTIONS], t), + events_per_time(total[OT_COUNTER_TCP_ANNOUNCES] + total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], t)); } static void stats_udpconnections_mrtg(char **reply, const char *reply_end) { ot_time t = time(NULL) - ot_start_time; - 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, - ot_overall_udp_successfulannounces + ot_overall_udp_connects, (int)t, (int)(t / 3600), events_per_time(ot_overall_udp_connections, t), - events_per_time(ot_overall_udp_successfulannounces + ot_overall_udp_connects, t)); + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + + 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], + total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], (int)t, (int)(t / 3600), + events_per_time(total[OT_COUNTER_UDP_CONNECTIONS], t), + events_per_time(total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], t)); } static void stats_tcpconnections_mrtg(char **reply, const char *reply_end) { time_t t = time(NULL) - ot_start_time; - 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, - ot_overall_tcp_successfulannounces, (int)t, (int)(t / 3600), events_per_time(ot_overall_tcp_connections, t), - events_per_time(ot_overall_tcp_successfulannounces, t)); + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + + 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], + total[OT_COUNTER_TCP_ANNOUNCES], (int)t, (int)(t / 3600), events_per_time(total[OT_COUNTER_TCP_CONNECTIONS], t), + events_per_time(total[OT_COUNTER_TCP_ANNOUNCES], t)); } static void stats_scrape_mrtg(char **reply, const char *reply_end) { time_t t = time(NULL) - ot_start_time; + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker scrape stats, %lu scrape/s (tcp and udp)", - ot_overall_tcp_successfulscrapes, ot_overall_udp_successfulscrapes, (int)t, (int)(t / 3600), - events_per_time((ot_overall_tcp_successfulscrapes + ot_overall_udp_successfulscrapes), t)); + total[OT_COUNTER_TCP_SCRAPES], total[OT_COUNTER_UDP_SCRAPES], (int)t, (int)(t / 3600), + events_per_time((total[OT_COUNTER_TCP_SCRAPES] + total[OT_COUNTER_UDP_SCRAPES]), t)); } static void stats_fullscrapes_mrtg(char **reply, const char *reply_end) { ot_time t = time(NULL) - ot_start_time; + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + fmt_append(reply, reply_end, "%llu\n%llu\n%i seconds (%i hours)\nopentracker full scrape stats, %lu conns/s :: %lu bytes/s.", - ot_full_scrape_count * 1000, ot_full_scrape_size, (int)t, (int)(t / 3600), events_per_time(ot_full_scrape_count, t), - events_per_time(ot_full_scrape_size, t)); + total[OT_COUNTER_FULLSCRAPES] * 1000, total[OT_COUNTER_FULLSCRAPE_BYTES], (int)t, (int)(t / 3600), + events_per_time(total[OT_COUNTER_FULLSCRAPES], t), + events_per_time(total[OT_COUNTER_FULLSCRAPE_BYTES], t)); } static 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) { } static void stats_httperrors_txt(char **reply, const char *reply_end) { + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + 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", - ot_failed_request_counts[0], ot_failed_request_counts[1], ot_failed_request_counts[2], ot_failed_request_counts[3], - ot_failed_request_counts[4], ot_failed_request_counts[5], ot_failed_request_counts[6]); + total[OT_COUNTER_FAILED + 0], total[OT_COUNTER_FAILED + 1], total[OT_COUNTER_FAILED + 2], total[OT_COUNTER_FAILED + 3], + total[OT_COUNTER_FAILED + 4], total[OT_COUNTER_FAILED + 5], total[OT_COUNTER_FAILED + 6]); } static void stats_return_renew_bucket(char **reply, const char *reply_end) { + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + int i; for (i = 0; i < OT_PEER_TIMEOUT; ++i) - fmt_append(reply, reply_end, "%02i %llu\n", i, ot_renewed[i]); + fmt_append(reply, reply_end, "%02i %llu\n", i, total[OT_COUNTER_RENEWED + i]); } static void stats_return_sync_mrtg(char **reply, const char *reply_end) { ot_time t = time(NULL) - ot_start_time; - 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, - (int)t, (int)(t / 3600), events_per_time(ot_overall_tcp_connections + ot_overall_udp_connections, t), - events_per_time(ot_overall_tcp_successfulannounces + ot_overall_udp_successfulannounces + ot_overall_udp_connects, t)); + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + + 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, + (int)t, (int)(t / 3600), events_per_time(total[OT_COUNTER_TCP_CONNECTIONS] + total[OT_COUNTER_UDP_CONNECTIONS], t), + events_per_time(total[OT_COUNTER_TCP_ANNOUNCES] + total[OT_COUNTER_UDP_ANNOUNCES] + total[OT_COUNTER_UDP_CONNECTS], t)); } static void stats_return_completed_mrtg(char **reply, const char *reply_end) { ot_time t = time(NULL) - ot_start_time; + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + - 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), - events_per_time(ot_overall_completed, t / 3600)); + 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), + events_per_time(total[OT_COUNTER_COMPLETED], t / 3600)); } #ifdef WANT_LOG_NUMWANT @@ -464,6 +534,10 @@ static void stats_return_fulllog(size_t *iovec_entries, struct iovec **iovector, #endif static void stats_return_everything(char **reply, const char *reply_end) { + unsigned long long total[OT_COUNTER_COUNT]; + + counters_total(total); + torrent_stats stats = {0, 0, 0}; int i; @@ -482,25 +556,25 @@ static void stats_return_everything(char **reply, const char *reply_end) { fmt_append(reply, reply_end, " \n"); fmt_append(reply, reply_end, " \n %llu\n \n", stats.peer_count); fmt_append(reply, reply_end, " \n %llu\n \n", stats.seed_count); - fmt_append(reply, reply_end, " \n %llu\n \n", ot_overall_completed); + fmt_append(reply, reply_end, " \n %llu\n \n", total[OT_COUNTER_COMPLETED]); fmt_append(reply, reply_end, " \n"); fmt_append(reply, reply_end, " \n %llu\n %llu\n %llu\n \n", - ot_overall_tcp_connections, ot_overall_tcp_successfulannounces, ot_overall_tcp_successfulscrapes); + total[OT_COUNTER_TCP_CONNECTIONS], total[OT_COUNTER_TCP_ANNOUNCES], total[OT_COUNTER_TCP_SCRAPES]); fmt_append(reply, reply_end, " \n %llu\n %llu\n %llu\n %llu\n %llu\n \n", - ot_overall_udp_connections, ot_overall_udp_connects, ot_overall_udp_successfulannounces, ot_overall_udp_successfulscrapes, - ot_overall_udp_connectionidmissmatches); - fmt_append(reply, reply_end, " \n %llu\n \n", ot_overall_sync_count); + total[OT_COUNTER_UDP_CONNECTIONS], total[OT_COUNTER_UDP_CONNECTS], total[OT_COUNTER_UDP_ANNOUNCES], total[OT_COUNTER_UDP_SCRAPES], + total[OT_COUNTER_UDP_CONNID_MISSMATCH]); + fmt_append(reply, reply_end, " \n %llu\n \n", total[OT_COUNTER_SYNCS]); fmt_append(reply, reply_end, " \n"); fmt_append(reply, reply_end, " \n"); fmt_append(reply, reply_end, " \n"); for (i = 0; i < OT_PEER_TIMEOUT; ++i) - fmt_append(reply, reply_end, " %llu\n", i, ot_renewed[i]); + fmt_append(reply, reply_end, " %llu\n", i, total[OT_COUNTER_RENEWED + i]); fmt_append(reply, reply_end, " \n"); fmt_append(reply, reply_end, " \n"); for (i = 0; i < CODE_HTTPERROR_COUNT; ++i) - fmt_append(reply, reply_end, " %llu\n", ot_failed_request_names[i], ot_failed_request_counts[i]); + fmt_append(reply, reply_end, " %llu\n", ot_failed_request_names[i], total[OT_COUNTER_FAILED + i]); fmt_append(reply, reply_end, " \n"); - fmt_append(reply, reply_end, " \n %llu\n \n", ot_overall_stall_count); + fmt_append(reply, reply_end, " \n %llu\n \n", total[OT_COUNTER_STALLS]); fmt_append(reply, reply_end, " \n"); fmt_append(reply, reply_end, ""); } @@ -608,33 +682,35 @@ static void stats_make(size_t *iovec_entries, struct iovec **iovector, ot_taskty iovec_trim_last(iovec_entries, iovector, r); } -void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) { +void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) { + if (!ws) + return; + switch (event) { case EVENT_ACCEPT: if (proto == FLAG_TCP) - ot_overall_tcp_connections++; + COUNT(OT_COUNTER_TCP_CONNECTIONS, 1); else - ot_overall_udp_connections++; + COUNT(OT_COUNTER_UDP_CONNECTIONS, 1); #ifdef WANT_LOG_NETWORKS stat_increase_network_count(&stats_network_counters_root, 0, event_data); #endif break; case EVENT_ANNOUNCE: if (proto == FLAG_TCP) - ot_overall_tcp_successfulannounces++; + COUNT(OT_COUNTER_TCP_ANNOUNCES, 1); else - ot_overall_udp_successfulannounces++; + COUNT(OT_COUNTER_UDP_ANNOUNCES, 1); break; case EVENT_CONNECT: if (proto == FLAG_TCP) - ot_overall_tcp_connects++; + COUNT(OT_COUNTER_TCP_CONNECTS, 1); else - ot_overall_udp_connects++; + COUNT(OT_COUNTER_UDP_CONNECTS, 1); break; case EVENT_COMPLETED: #ifdef WANT_SYSLOGS - if (event_data) { - struct ot_workstruct *ws = (struct ot_workstruct *)event_data; + { char timestring[64]; char hash_hex[42], peerid_hex[42], ip_readable[64]; struct tm time_now; @@ -655,17 +731,17 @@ void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_ syslog(LOG_INFO, "time=%s event=completed info_hash=%s peer_id=%s ip=%s", timestring, hash_hex, peerid_hex, ip_readable); } #endif - ot_overall_completed++; + COUNT(OT_COUNTER_COMPLETED, 1); break; case EVENT_SCRAPE: if (proto == FLAG_TCP) - ot_overall_tcp_successfulscrapes++; + COUNT(OT_COUNTER_TCP_SCRAPES, 1); else - ot_overall_udp_successfulscrapes++; + COUNT(OT_COUNTER_UDP_SCRAPES, 1); break; case EVENT_FULLSCRAPE: - ot_full_scrape_count++; - ot_full_scrape_size += event_data; + COUNT(OT_COUNTER_FULLSCRAPES, 1); + COUNT(OT_COUNTER_FULLSCRAPE_BYTES, event_data); break; case EVENT_FULLSCRAPE_REQUEST: { ot_ip6 *ip = (ot_ip6 *)event_data; /* ugly hack to transfer ip to stats */ @@ -674,7 +750,7 @@ void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_ off += ip_format(_debug + off, *ip); off += snprintf(_debug + off, sizeof(_debug) - off, " - FULL SCRAPE\n"); (void)write(2, _debug, off); - ot_full_scrape_request_count++; + COUNT(OT_COUNTER_FULLSCRAPE_REQUESTS, 1); } break; case EVENT_FULLSCRAPE_REQUEST_GZIP: { ot_ip6 *ip = (ot_ip6 *)event_data; /* ugly hack to transfer ip to stats */ @@ -683,19 +759,19 @@ void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_ off += ip_format(_debug + off, *ip); off += snprintf(_debug + off, sizeof(_debug) - off, " - FULL SCRAPE\n"); (void)write(2, _debug, off); - ot_full_scrape_request_count++; + COUNT(OT_COUNTER_FULLSCRAPE_REQUESTS, 1); } break; case EVENT_FAILED: - ot_failed_request_counts[event_data]++; + COUNT(OT_COUNTER_FAILED + event_data, 1); break; case EVENT_RENEW: - ot_renewed[event_data]++; + COUNT(OT_COUNTER_RENEWED + event_data, 1); break; case EVENT_SYNC: - ot_overall_sync_count += event_data; + COUNT(OT_COUNTER_SYNCS, event_data); break; case EVENT_BUCKET_LOCKED: - ot_overall_stall_count++; + COUNT(OT_COUNTER_STALLS, 1); break; #ifdef WANT_SPOT_WOODPECKER case EVENT_WOODPECKER: @@ -705,7 +781,7 @@ void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_ break; #endif case EVENT_CONNID_MISSMATCH: - ++ot_overall_udp_connectionidmissmatches; + COUNT(OT_COUNTER_UDP_CONNID_MISSMATCH, 1); default: break; } @@ -725,9 +801,11 @@ static void *stats_worker(void *args) { (void)args; - while (1) { + while (g_opentracker_running) { ot_tasktype tasktype = TASK_STATS; ot_taskid taskid = mutex_workqueue_poptask(&tasktype); + if (!taskid) + break; stats_make(&iovec_entries, &iovector, tasktype); if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector)) iovec_free(&iovec_entries, &iovector); @@ -744,5 +822,6 @@ void stats_init(void) { } void stats_deinit(void) { - pthread_cancel(thread_id); + mutex_workqueue_wakeall(); + pthread_join(thread_id, NULL); } diff --git a/ot_stats.h b/ot_stats.h index a118b4f..f484765 100644 --- a/ot_stats.h +++ b/ot_stats.h @@ -30,8 +30,7 @@ typedef enum { void stats_counters_register(struct ot_workstruct *ws); void stats_counters_retire(struct ot_workstruct *ws); - -void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_data); +void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data); void stats_deliver(int64_t sock, int tasktype); void stats_cleanup(void); void 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) { ot_tasktype tasktype = TASK_SYNC_OUT; ot_taskid taskid = mutex_workqueue_poptask(&tasktype); sync_make(&iovec_entries, &iovector); - stats_issue_event(EVENT_SYNC_OUT, FLAG_TCP, iovec_length(&iovec_entries, &iovector)); + stats_issue_event(ws, EVENT_SYNC_OUT, FLAG_TCP, iovec_length(&iovec_entries, &iovector)); if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector)) iovec_free(&iovec_entries, &iovector); } diff --git a/ot_udp.c b/ot_udp.c index 19a6c7c..14210c3 100644 --- a/ot_udp.c +++ b/ot_udp.c @@ -77,8 +77,8 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) { return 0; byte_count = (size_t)result; - stats_issue_event(EVENT_ACCEPT, FLAG_UDP, (uintptr_t)remoteip); - stats_issue_event(EVENT_READ, FLAG_UDP, byte_count); + stats_issue_event(ws, EVENT_ACCEPT, FLAG_UDP, (uintptr_t)remoteip); + stats_issue_event(ws, EVENT_READ, FLAG_UDP, byte_count); /* Minimum udp tracker packet size, also catches error */ if (byte_count < 16) @@ -111,7 +111,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) { outpacket[1] = inpacket[3]; memcpy(&outpacket[2], &err_idmiss[0], s); socket_send(serversocket, ws->outbuf, 8 + s, remoteip, remoteport); - stats_issue_event(EVENT_CONNID_MISSMATCH, FLAG_UDP, 8 + s); + stats_issue_event(ws, EVENT_CONNID_MISSMATCH, FLAG_UDP, 8 + s); return 1; } } @@ -128,7 +128,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) { outpacket[3] = connid[1]; socket_send(serversocket, ws->outbuf, 16, remoteip, remoteport); - stats_issue_event(EVENT_CONNECT, FLAG_UDP, 16); + stats_issue_event(ws, EVENT_CONNECT, FLAG_UDP, 16); break; case 1: /* This is an announce action */ /* Minimum udp announce packet size */ @@ -188,7 +188,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) { } socket_send(serversocket, ws->outbuf, ws->reply_size, remoteip, remoteport); - stats_issue_event(EVENT_ANNOUNCE, FLAG_UDP, ws->reply_size); + stats_issue_event(ws, EVENT_ANNOUNCE, FLAG_UDP, ws->reply_size); break; case 2: /* This is a scrape action */ @@ -199,7 +199,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) { return_udp_scrape_for_torrent(*(ot_hash *)(((char *)inpacket) + 16 + 20 * scrape_count), ((char *)outpacket) + 8 + 12 * scrape_count); socket_send(serversocket, ws->outbuf, 8 + 12 * scrape_count, remoteip, remoteport); - stats_issue_event(EVENT_SCRAPE, FLAG_UDP, scrape_count); + stats_issue_event(ws, EVENT_SCRAPE, FLAG_UDP, scrape_count); break; } return 1; @@ -210,6 +210,7 @@ static void *udp_worker(void *args) { struct ot_workstruct ws; memset(&ws, 0, sizeof(ws)); + stats_counters_register(&ws); ws.inbuf = malloc(G_INBUF_SIZE); ws.outbuf = malloc(G_OUTBUF_SIZE); #ifdef _DEBUG_HTTPERROR @@ -219,6 +220,8 @@ static void *udp_worker(void *args) { while (g_opentracker_running) handle_udp6(sock, &ws); + stats_counters_retire(&ws); + free(ws.inbuf); free(ws.outbuf); #ifdef _DEBUG_HTTPERROR 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) { exit(111); } -void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) { +void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) { + (void)ws; (void)event; (void)proto; (void)event_data; diff --git a/trackerlogic.c b/trackerlogic.c index aac4448..477c4a6 100644 --- a/trackerlogic.c +++ b/trackerlogic.c @@ -149,16 +149,16 @@ size_t add_peer_to_torrent_and_return_peers(PROTO_FLAG proto, struct ot_workstru peer_list->peer_count++; if (OT_PEERFLAG(ws->peer) & PEER_FLAG_COMPLETED) { peer_list->down_count++; - stats_issue_event(EVENT_COMPLETED, 0, (uintptr_t)ws); + stats_issue_event(ws, EVENT_COMPLETED, 0, 0); } if (OT_PEERFLAG(ws->peer) & PEER_FLAG_SEEDING) peer_list->seed_count++; } else { - stats_issue_event(EVENT_RENEW, 0, OT_PEERTIME(peer_dest, peer_size)); + stats_issue_event(ws, EVENT_RENEW, 0, OT_PEERTIME(peer_dest, peer_size)); #ifdef WANT_SPOT_WOODPECKER if ((OT_PEERTIME(peer_dest, peer_size) > 0) && (OT_PEERTIME(peer_dest, peer_size) < 20)) - stats_issue_event(EVENT_WOODPECKER, 0, (uintptr_t)&ws->peer); + stats_issue_event(ws, EVENT_WOODPECKER, 0, (uintptr_t)&ws->peer); #endif #ifdef WANT_SYNC_LIVE /* 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 peer_list->seed_count++; if (!(OT_PEERFLAG_D(peer_dest, peer_size) & PEER_FLAG_COMPLETED) && (OT_PEERFLAG(ws->peer) & PEER_FLAG_COMPLETED)) { peer_list->down_count++; - stats_issue_event(EVENT_COMPLETED, 0, (uintptr_t)ws); + stats_issue_event(ws, EVENT_COMPLETED, 0, 0); } if (OT_PEERFLAG_D(peer_dest, peer_size) & PEER_FLAG_COMPLETED) OT_PEERFLAG(ws->peer) |= PEER_FLAG_COMPLETED; @@ -571,6 +571,13 @@ void trackerlogic_deinit(void) { int delta_torrentcount = 0; size_t bucket, j; + /* Stop the background workers first */ + stats_deinit(); + livesync_deinit(); + accesslist_deinit(); + fullscrape_deinit(); + clean_deinit(); + /* Free all torrents... */ for (bucket = 0; bucket < OT_BUCKET_COUNT; ++bucket) { ot_vector *torrents_list = mutex_bucket_lock(bucket); @@ -586,12 +593,6 @@ void trackerlogic_deinit(void) { mutex_bucket_unlock(bucket, delta_torrentcount); } - /* Deinitialise background worker threads */ - stats_deinit(); - livesync_deinit(); - accesslist_deinit(); - fullscrape_deinit(); - clean_deinit(); /* Release mutexes */ mutex_deinit(); } -- cgit v1.3