Decouple config parsing from binding and worker spawning

* Options are parsed now into a listener description list before
  binding them at once and starting the workers later
* UDP workers can now be configured per connection
* socket reuse is now done more sensibly, mcast sockets always
  reused
* Hold-off logic after spawning udp workers removed

1 files changed, 185 insertions, 57 deletions

diff --git a/opentracker.c b/opentracker.c
index 5b197d1..5f21ede 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -48,7 +48,22 @@ int                 g_self_pipe[2];
 
 static char        *g_serverdir;
 static char        *g_serveruser;
-static unsigned int g_udp_workers;
+
+typedef struct {
+  ot_ip6     ip;
+  uint16_t   port;
+  PROTO_FLAG proto;
+  int        shared;
+  int        workers;
+  int64_t    sock;
+} ot_listen;
+#define OT_LISTEN_INHERIT (-1)
+
+/* Config entries collected from command line and config files */
+static ot_listen *g_listen;
+static size_t     g_listen_count;
+static int        g_listen_shared;
+static int        g_listen_workers;
 
 static void         panic(const char *routine) __attribute__((noreturn));
 static void         panic(const char *routine) {
@@ -338,44 +353,156 @@ static void *server_mainloop(void *args) {
   return 0;
 }
 
-static int64_t ot_try_bind(ot_ip6 ip, uint16_t port, PROTO_FLAG proto) {
-  int64_t sock;
+static void listen_append(const ot_ip6 ip, uint16_t port, PROTO_FLAG proto, int shared, int workers) {
+  ot_listen *grown = realloc(g_listen, (g_listen_count + 1) * sizeof(ot_listen));
+  ot_listen *entry;
+
+  if (!grown)
+    panic("listen_append");
+  g_listen = grown;
+
+  entry = g_listen + g_listen_count++;
+  memcpy(entry->ip, ip, sizeof(ot_ip6));
+  entry->port      = port;
+  entry->proto     = proto;
+  entry->shared  = shared;
+  entry->workers = workers;
+  entry->sock      = -1;
+}
+
+typedef enum { LISTEN_OPTION_UNKNOWN, LISTEN_OPTION_SHARED, LISTEN_OPTION_WORKERS } listen_option_kind;
+
+typedef struct {
+  const char        *name;
+  size_t             len;
+  listen_option_kind kind;
+  int                takes_value;
+} listen_option_key;
+
+#define LISTEN_OPTION_KEY(s, k, v) {s, sizeof(s) - 1, k, v}
+
+static const listen_option_key listen_option_keys[] = {
+    LISTEN_OPTION_KEY("shared",  LISTEN_OPTION_SHARED,  0),
+    LISTEN_OPTION_KEY("workers", LISTEN_OPTION_WORKERS, 1),
+};
+static const size_t listen_option_count = sizeof(listen_option_keys) / sizeof(listen_option_keys[0]);
+
+static listen_option_kind parse_listen_option(char **p) {
+  size_t i;
+
+  for (i = 0; i < listen_option_count; ++i) {
+    const listen_option_key *key = &listen_option_keys[i];
+    char                     delimiter = (*p)[key->len];
+
+    if (strncmp(*p, key->name, key->len))
+      continue;
+    if (key->takes_value ? delimiter != '=' : (delimiter && !isspace((unsigned char)delimiter)))
+      continue;
+
+    *p += key->len + (key->takes_value ? 1 : 0);
+    return key->kind;
+  }
+  return LISTEN_OPTION_UNKNOWN;
+}
+
+/* Options trailing an address in a listen statement, in any order */
+static int listen_scan_options(char *p, int *shared, int *workers) {
+  uint64_t value;
+
+  for (;;) {
+    while (isspace((unsigned char)*p))
+      ++p;
+    if (!*p)
+      return 1;
+
+    switch (parse_listen_option(&p)) {
+    case LISTEN_OPTION_SHARED:
+      *shared = 1;
+      break;
+    case LISTEN_OPTION_WORKERS: {
+      size_t off = scan_udec(p, &value, UINT_MAX);
+
+      if (!off)
+        return 0;
+      *workers  = (int)value;
+      p        += off;
+      break;
+    }
+    default:
+      return 0;
+    }
+  }
+}
+
+static int listen_option(int value, int fallback) { return value == OT_LISTEN_INHERIT ? fallback : value; }
+
+static void listen_bind_one(ot_listen *entry) {
+  socket_sharing sharing = listen_option(entry->shared, g_listen_shared) ? SOCKET_SHARED : SOCKET_EXCLUSIVE;
 
 #ifdef _DEBUG
   {
     char *protos[] = {"TCP", "UDP", "UDP mcast"};
     char  _debug[512];
-    int   off  = snprintf(_debug, sizeof(_debug), "Binding socket type %s to address [", protos[proto]);
-    off       += ip_format(_debug + off, ip);
-    snprintf(_debug + off, sizeof(_debug) - off, "]:%d...", port);
+    int   off  = snprintf(_debug, sizeof(_debug), "Binding socket type %s to address [", protos[entry->proto]);
+    off       += ip_format(_debug + off, entry->ip);
+    snprintf(_debug + off, sizeof(_debug) - off, "]:%d...", entry->port);
     fputs(_debug, stderr);
   }
 #endif
 
-  if (proto == FLAG_TCP) {
-    if ((sock = socket_tcp_listen(ip, port)) == -1)
+#ifdef WANT_SYNC_LIVE
+  if (entry->proto == FLAG_MCA) {
+    livesync_bind_mcast(entry->ip, entry->port);
+#ifdef _DEBUG
+    fputs(" success.\n", stderr);
+#endif
+    return;
+  }
+#endif
+
+  if (entry->proto == FLAG_TCP) {
+    if ((entry->sock = socket_tcp_listen(entry->ip, entry->port, sharing)) == -1)
       panic("socket_tcp_listen");
   } else {
-    if ((sock = socket_udp_bind(ip, port)) == -1)
+    if ((entry->sock = socket_udp_bind(entry->ip, entry->port, sharing)) == -1)
       panic("socket_udp_bind");
   }
 
-  if (!io_fd(sock))
+  if (!io_fd(entry->sock))
     panic("io_fd");
 
-  io_setcookie(sock, (void *)proto);
-
-  if ((proto == FLAG_UDP) && g_udp_workers) {
-    io_block(sock);
-    udp_init(sock, g_udp_workers);
-  } else
-    io_wantread(sock);
+  io_setcookie(entry->sock, (void *)entry->proto);
+  io_wantread(entry->sock);
 
 #ifdef _DEBUG
   fputs(" success.\n", stderr);
 #endif
+}
+
+/* Bound in the order the entries were given.
+   Note: On macOS SO_REUSEPORT delivers everything to the socket that bound last. */
+static void listen_bind_all(void) {
+  size_t i;
+
+  for (i = 0; i < g_listen_count; ++i)
+    listen_bind_one(g_listen + i);
+}
+
+static void listen_start_udp_workers(void) {
+  size_t i;
 
-  return sock;
+  udp_init();
+
+  for (i = 0; i < g_listen_count; ++i) {
+    ot_listen *entry   = g_listen + i;
+    int        workers = listen_option(entry->workers, g_listen_workers);
+
+    if (entry->proto != FLAG_UDP || workers <= 0)
+      continue;
+
+    io_block(entry->sock);
+    udp_workers_install(entry->sock, (unsigned int)workers);
+  }
 }
 
 char *set_config_option(char **option, char *value) {
@@ -452,7 +579,7 @@ typedef enum {
     CONFIG_LISTEN_TCP,
     CONFIG_LISTEN_UDP,
     CONFIG_LISTEN_UDP_WORKERS,
-    CONFIG_LISTEN_REUSEPORT,
+    CONFIG_LISTEN_SHARED,
     CONFIG_ACCESS_WHITELIST,
     CONFIG_ACCESS_BLACKLIST,
     CONFIG_ACCESS_FIFO_ADD,
@@ -480,7 +607,7 @@ static const config_key config_keys[] = {
     CONFIG_KEY("listen.tcp",               CONFIG_LISTEN_TCP),
     CONFIG_KEY("listen.udp",               CONFIG_LISTEN_UDP),
     CONFIG_KEY("listen.udp.workers",       CONFIG_LISTEN_UDP_WORKERS),
-    CONFIG_KEY("listen.reuseport",         CONFIG_LISTEN_REUSEPORT),
+    CONFIG_KEY("listen.shared",            CONFIG_LISTEN_SHARED),
 #ifdef WANT_ACCESSLIST_WHITE
     CONFIG_KEY("access.whitelist",         CONFIG_ACCESS_WHITELIST),
 #elif defined(WANT_ACCESSLIST_BLACK)
@@ -537,9 +664,10 @@ int parse_configfile(char *config_filename) {
   }
 
   while (fgets(inbuf, sizeof(inbuf), config_file_handle)) {
-    char    *p = inbuf, *end;
-    uint64_t value;
-    uint16_t tmpport = 6969;
+    char       *p = inbuf, *end;
+    uint64_t    value;
+    uint16_t    tmpport = 6969;
+    config_kind kind;
 
     /* Skip white spaces */
     while (isspace((unsigned char)*p))
@@ -555,7 +683,8 @@ int parse_configfile(char *config_filename) {
       *--end = 0;
 
     /* Scan for commands */
-    switch (parse_config_key(&p)) {
+    kind = parse_config_key(&p);
+    switch (kind) {
       case CONFIG_TRACKER_ROOTDIR:
         set_config_option(&g_serverdir, p);
         break;
@@ -563,35 +692,29 @@ int parse_configfile(char *config_filename) {
         set_config_option(&g_serveruser, p);
         break;
       case CONFIG_LISTEN_TCP_UDP:
-        if (!scan_ip6_port(p, tmpip, &tmpport))
-          goto parse_error;
-        ot_try_bind(tmpip, tmpport, FLAG_TCP);
-        ++bound;
-        ot_try_bind(tmpip, tmpport, FLAG_UDP);
-        ++bound;
-        break;
       case CONFIG_LISTEN_TCP:
-        if (!scan_ip6_port(p, tmpip, &tmpport))
-          goto parse_error;
-        ot_try_bind(tmpip, tmpport, FLAG_TCP);
-        ++bound;
-        break;
-      case CONFIG_LISTEN_UDP:
-        if (!scan_ip6_port(p, tmpip, &tmpport))
+      case CONFIG_LISTEN_UDP: {
+        int shared = OT_LISTEN_INHERIT
+        int workers = OT_LISTEN_INHERIT;
+        int off    = scan_ip6_port(p, tmpip, &tmpport);
+
+        if (!off || !listen_scan_options(p + off, &shared, &workers))
           goto parse_error;
-        ot_try_bind(tmpip, tmpport, FLAG_UDP);
-        ++bound;
+        if (kind != CONFIG_LISTEN_UDP)
+          listen_append(tmpip, tmpport, FLAG_TCP, shared, OT_LISTEN_INHERIT), ++bound;
+        if (kind != CONFIG_LISTEN_TCP)
+          listen_append(tmpip, tmpport, FLAG_UDP, shared, workers), ++bound;
         break;
+      }
       case CONFIG_LISTEN_UDP_WORKERS:
-        if (scan_udec(p, &value, UINT_MAX))
-          g_udp_workers = (unsigned int)value;
+        if (!scan_udec(p, &value, UINT_MAX))
+          goto parse_error;
+        g_listen_workers = (int)value;
         break;
-      case CONFIG_LISTEN_REUSEPORT:
-        /* Like listen.udp.workers this is read when a socket is bound, so it
-           has to come before the listen lines it should apply to. */
+      case CONFIG_LISTEN_SHARED:
         if (!scan_udec(p, &value, 1))
           goto parse_error;
-        g_socket_reuseport = (int)value;
+        g_listen_shared = (int)value;
         break;
 #ifdef WANT_ACCESSLIST_WHITE
       case CONFIG_ACCESS_WHITELIST:
@@ -638,9 +761,14 @@ int parse_configfile(char *config_filename) {
         break;
       case CONFIG_LIVESYNC_CLUSTER_LISTEN:
         tmpport = LIVESYNC_PORT;
-        if (!scan_ip6_port(p, tmpip, &tmpport))
-          goto parse_error;
-        livesync_bind_mcast(tmpip, tmpport);
+        {
+          int shared = OT_LISTEN_INHERIT, dummy = OT_LISTEN_INHERIT;
+          int off       = scan_ip6_port(p, tmpip, &tmpport);
+
+          if (!off || !listen_scan_options(p + off, &shared, &dummy))
+            goto parse_error;
+          listen_append(tmpip, tmpport, FLAG_MCA, shared, OT_LISTEN_INHERIT);
+        }
       break;
 #endif
       default:
@@ -791,7 +919,7 @@ int main(int argc, char **argv) {
         usage(argv[0]);
         exit(1);
       }
-      ot_try_bind(serverip, (uint16_t)value, FLAG_TCP);
+      listen_append(serverip, (uint16_t)value, FLAG_TCP, OT_LISTEN_INHERIT, OT_LISTEN_INHERIT);
       bound++;
       break;
     case 'P':
@@ -799,7 +927,7 @@ int main(int argc, char **argv) {
         usage(argv[0]);
         exit(1);
       }
-      ot_try_bind(serverip, (uint16_t)value, FLAG_UDP);
+      listen_append(serverip, (uint16_t)value, FLAG_UDP, OT_LISTEN_INHERIT, OT_LISTEN_INHERIT);
       bound++;
       break;
 #ifdef WANT_SYNC_LIVE
@@ -808,7 +936,7 @@ int main(int argc, char **argv) {
         usage(argv[0]);
         exit(1);
       }
-      livesync_bind_mcast(serverip, (uint16_t)value);
+      listen_append(serverip, (uint16_t)value, FLAG_MCA, OT_LISTEN_INHERIT, OT_LISTEN_INHERIT);
       break;
 #endif
     case 'd':
@@ -849,12 +977,13 @@ int main(int argc, char **argv) {
     }
   }
 
-  /* Bind to our default tcp/udp ports */
   if (!bound) {
-    ot_try_bind(serverip, 6969, FLAG_TCP);
-    ot_try_bind(serverip, 6969, FLAG_UDP);
+    listen_append(serverip, 6969, FLAG_TCP, OT_LISTEN_INHERIT, OT_LISTEN_INHERIT);
+    listen_append(serverip, 6969, FLAG_UDP, OT_LISTEN_INHERIT, OT_LISTEN_INHERIT);
   }
 
+  listen_bind_all();
+
   defaul_signal_handlers();
 
 #ifdef WANT_SYSLOGS
@@ -893,8 +1022,7 @@ int main(int argc, char **argv) {
 
   install_signal_handlers();
 
-  if (!g_udp_workers)
-    udp_init(-1, 0);
+  listen_start_udp_workers();
 
   server_mainloop(0);