1 files changed, 150 insertions, 71 deletions

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 <inttypes.h>
 #include <pthread.h>
+#include <stdatomic.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -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, "  </torrents>\n");
   fmt_append(reply, reply_end, "  <peers>\n    <count>%llu</count>\n  </peers>\n", stats.peer_count);
   fmt_append(reply, reply_end, "  <seeds>\n    <count>%llu</count>\n  </seeds>\n", stats.seed_count);
-  fmt_append(reply, reply_end, "  <completed>\n    <count>%llu</count>\n  </completed>\n", ot_overall_completed);
+  fmt_append(reply, reply_end, "  <completed>\n    <count>%llu</count>\n  </completed>\n", total[OT_COUNTER_COMPLETED]);
   fmt_append(reply, reply_end, "  <connections>\n");
   fmt_append(reply, reply_end, "    <tcp>\n      <accept>%llu</accept>\n      <announce>%llu</announce>\n      <scrape>%llu</scrape>\n    </tcp>\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, "    <udp>\n      <overall>%llu</overall>\n      <connect>%llu</connect>\n      <announce>%llu</announce>\n      <scrape>%llu</scrape>\n      <missmatch>%llu</missmatch>\n    </udp>\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, "    <livesync>\n      <count>%llu</count>\n    </livesync>\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, "    <livesync>\n      <count>%llu</count>\n    </livesync>\n", total[OT_COUNTER_SYNCS]);
   fmt_append(reply, reply_end, "  </connections>\n");
   fmt_append(reply, reply_end, "  <debug>\n");
   fmt_append(reply, reply_end, "    <renew>\n");
   for (i = 0; i < OT_PEER_TIMEOUT; ++i)
-    fmt_append(reply, reply_end, "      <count interval=\"%02i\">%llu</count>\n", i, ot_renewed[i]);
+    fmt_append(reply, reply_end, "      <count interval=\"%02i\">%llu</count>\n", i, total[OT_COUNTER_RENEWED + i]);
   fmt_append(reply, reply_end, "    </renew>\n");
   fmt_append(reply, reply_end, "    <http_error>\n");
   for (i = 0; i < CODE_HTTPERROR_COUNT; ++i)
-    fmt_append(reply, reply_end, "      <count code=\"%s\">%llu</count>\n", ot_failed_request_names[i], ot_failed_request_counts[i]);
+    fmt_append(reply, reply_end, "      <count code=\"%s\">%llu</count>\n", ot_failed_request_names[i], total[OT_COUNTER_FAILED + i]);
   fmt_append(reply, reply_end, "    </http_error>\n");
-  fmt_append(reply, reply_end, "    <mutex_stall>\n      <count>%llu</count>\n    </mutex_stall>\n", ot_overall_stall_count);
+  fmt_append(reply, reply_end, "    <mutex_stall>\n      <count>%llu</count>\n    </mutex_stall>\n", total[OT_COUNTER_STALLS]);
   fmt_append(reply, reply_end, "  </debug>\n");
   fmt_append(reply, reply_end, "</stats>");
 }
@@ -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);
 }