1 files changed, 40 insertions, 9 deletions

diff --git a/opentracker.c b/opentracker.c
index 34b8fd5..e0f16d7 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -49,7 +49,9 @@ static void         panic(const char *routine) {
   exit(111);
 }
 
-/* Say we are stopping, wake the loop, and let main do the work on its own stack. */
+/* Say we are stopping, wake the loop, and let main do the work on its own
+   stack. Everything other than SIGINT falls through and does nothing, which
+   is what OT_SIGNAL_WAKE wants: returning from the read is all it is for. */
 static void signal_handler(int s) {
   if (s == SIGINT) {
     const char byte  = 0;
@@ -84,8 +86,9 @@ static void install_signal_handlers(void) {
 
   sa.sa_handler = signal_handler;
   sigemptyset(&sa.sa_mask);
-  sa.sa_flags = SA_RESTART;
-  if ((sigaction(SIGINT, &sa, NULL) == -1) || (sigaction(SIGALRM, &sa, NULL) == -1))
+  sa.sa_flags = 0;
+  if ((sigaction(SIGINT, &sa, NULL) == -1) || (sigaction(SIGALRM, &sa, NULL) == -1) ||
+      (sigaction(OT_SIGNAL_WAKE, &sa, NULL) == -1))
     panic("install_signal_handlers");
 
   sigaddset(&signal_mask, SIGINT);
@@ -320,8 +323,17 @@ static void *server_mainloop(void *args) {
 }
 
 
+/* A group of sockets sharing one address only spreads datagrams if every
+   member asked for it before binding, the first one included, so whether
+   this entry will have workers has to be known here. */
+static socket_sharing listen_sharing_for(const ot_listen *entry) {
+  if (entry->proto == FLAG_UDP && entry->workers > 1 && socket_can_balance())
+    return SOCKET_BALANCED;
+  return entry->shared ? SOCKET_SHARED : SOCKET_EXCLUSIVE;
+}
+
 static void listen_bind_one(ot_listen *entry) {
-  socket_sharing sharing = entry->shared ? SOCKET_SHARED : SOCKET_EXCLUSIVE;
+  socket_sharing sharing = listen_sharing_for(entry);
 
 #ifdef _DEBUG
   {
@@ -370,6 +382,10 @@ static void listen_bind_all(void) {
     listen_bind_one(entries + i);
 }
 
+/* The socket the event loop was using becomes the first worker's. Where the
+   kernel spreads datagrams, every further worker binds one of its own, so
+   none of them share a receive queue.
+   All others read from the first */
 static void listen_start_udp_workers(void) {
   size_t     count, i;
   ot_listen *entries = config_listen_entries(&count);
@@ -377,14 +393,28 @@ static void listen_start_udp_workers(void) {
   udp_init();
 
   for (i = 0; i < count; ++i) {
-    ot_listen *entry = entries + i;
+    ot_listen *entry   = entries + i;
+    int        balance = listen_sharing_for(entry) == SOCKET_BALANCED;
+    int        worker;
 
     if (entry->proto != FLAG_UDP || entry->workers <= 0)
       continue;
 
-    socket_set_blocking((int)entry->sock, 1);
     event_del((int)entry->sock);
-    udp_workers_install(entry->sock, (unsigned int)entry->workers);
+    udp_worker_install(entry->sock);
+
+    for (worker = 1; worker < entry->workers; ++worker) {
+      int64_t sock = entry->sock;
+
+      if (balance) {
+        sock = socket_udp_bind(entry->ip, entry->port, SOCKET_BALANCED);
+        if (sock == -1) {
+          fprintf(stderr, "Warning: could not bind another socket for udp worker %d: %s. Sharing the first instead.\n", worker, strerror(errno));
+          sock = entry->sock;
+        }
+      }
+      udp_worker_install(sock);
+    }
   }
 }
 
@@ -519,9 +549,10 @@ int main(int argc, char **argv) {
   listen_start_udp_workers();
 
   server_mainloop(0);
+  udp_workers_deinit();
 
-  /* 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 */
+  /* No thread is serving any more, so the teardown can run where a debugger
+     and the sanitizers can follow it */
   trackerlogic_deinit();
 #ifdef WANT_SYSLOGS
   closelog();