Performance improvement, where it works: Let different udp worker threads have their own socket

5 files changed, 74 insertions, 11 deletions

diff --git a/opentracker.c b/opentracker.c
index a546fc5..f579b92 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -320,8 +320,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 +379,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 +390,30 @@ 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;
+        } else
+          socket_set_blocking((int)sock, 1);
+      }
+      udp_worker_install(sock);
+    }
   }
 }
 
diff --git a/ot_socket.c b/ot_socket.c
index 576087f..158459f 100644
--- a/ot_socket.c
+++ b/ot_socket.c
@@ -75,13 +75,37 @@ static int socket_v6(int type) {
   return s;
 }
 
+/* SO_REUSEPORT means three different things. Linux spreads datagrams over
+   the group. FreeBSD permits the bind and spreads nothing, which is what
+   SO_REUSEPORT_LB was added for. Darwin hands every datagram to whichever
+   socket bound last, so a group there starves all but one reader. */
+#if defined(SO_REUSEPORT_LB)
+#define OT_BALANCE_OPTION SO_REUSEPORT_LB
+#elif defined(SO_REUSEPORT) && !defined(__APPLE__)
+#define OT_BALANCE_OPTION SO_REUSEPORT
+#endif
+
+int socket_can_balance(void) {
+#ifdef OT_BALANCE_OPTION
+  return 1;
+#else
+  return 0;
+#endif
+}
+
 /* The only place in the tree that needs to know what a kernel calls this */
 static void socket_set_sharing(int s, socket_sharing sharing) {
   int one = 1;
 
   setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
+#ifdef OT_BALANCE_OPTION
+  if (sharing == SOCKET_BALANCED) {
+    setsockopt(s, SOL_SOCKET, OT_BALANCE_OPTION, &one, sizeof(one));
+    return;
+  }
+#endif
 #ifdef SO_REUSEPORT
-  if (sharing == SOCKET_SHARED)
+  if (sharing != SOCKET_EXCLUSIVE)
     setsockopt(s, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(one));
 #endif
 }
diff --git a/ot_socket.h b/ot_socket.h
index aa4e2d2..f73a5c9 100644
--- a/ot_socket.h
+++ b/ot_socket.h
@@ -16,8 +16,18 @@
    as the v4 mapped address */
 
 /* How an address may be shared with other sockets. Which socket option says
-   so differs between kernels, which is socket_set_sharing's business */
-typedef enum { SOCKET_EXCLUSIVE, SOCKET_SHARED } socket_sharing;
+   so differs between kernels, which is socket_set_sharing's business.
+
+   SHARED lets another socket bind the same address. BALANCED asks the kernel
+   to spread arriving datagrams over the sockets that did, which is a
+   different thing and not everywhere available: ask socket_can_balance
+   before binding a group that depends on it. */
+typedef enum { SOCKET_EXCLUSIVE, SOCKET_SHARED, SOCKET_BALANCED } socket_sharing;
+
+/* Whether this kernel spreads datagrams over a group of sockets bound to one
+   address. Where it does not, binding a group is worse than one socket with
+   several readers, so the caller wants to know first. */
+int     socket_can_balance(void);
 
 /* Create, bind and, for TCP, listen. The returned socket is non blocking
    Returns -1 on failure with errno set */
diff --git a/ot_udp.c b/ot_udp.c
index 14210c3..87741da 100644
--- a/ot_udp.c
+++ b/ot_udp.c
@@ -235,13 +235,12 @@ void udp_init(void) {
     udp_generate_rijndael_round_key();
 }
 
-void udp_workers_install(int64_t sock, unsigned int worker_count) {
+void udp_worker_install(int64_t sock) {
   pthread_t thread_id;
 
 #ifdef _DEBUG
-  fprintf(stderr, "Installing %d workers on udp socket %ld\n", worker_count, (unsigned long)sock);
+  fprintf(stderr, "Installing a worker on udp socket %ld\n", (unsigned long)sock);
 #endif
-  while (worker_count--)
-    pthread_create(&thread_id, NULL, udp_worker, (void *)sock);
+  pthread_create(&thread_id, NULL, udp_worker, (void *)sock);
 }
 
diff --git a/ot_udp.h b/ot_udp.h
index a8ab9fc..7d4432f 100644
--- a/ot_udp.h
+++ b/ot_udp.h
@@ -11,7 +11,8 @@
 #include "trackerlogic.h"
 
 void udp_init(void);
-void udp_workers_install(int64_t sock, unsigned int worker_count);
+/* One thread reading one socket. Called once per worker. */
+void udp_worker_install(int64_t sock);
 int  handle_udp6(int64_t serversocket, struct ot_workstruct *ws);
 
 #endif