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

1 files changed, 32 insertions, 3 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);
+    }
   }
 }