Add functionality to distribute udp to several workers

opentracker.c

@@ -46,6 +46,7 @@ int          g_self_pipe[2];
 
 static char * g_serverdir;
 static char * g_serveruser;
+static unsigned int g_udp_workers;
 
 static void panic( const char *routine ) {
   fprintf( stderr, "%s: %s\n", routine, strerror(errno) );
@@ -328,6 +329,10 @@ static int64_t ot_try_bind( ot_ip6 ip, uint16_t port, PROTO_FLAG proto ) {
 
   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 );
 
 #ifdef _DEBUG
@@ -416,6 +421,10 @@ int parse_configfile( char * config_filename ) {
       if( !scan_ip6_port( p+11, tmpip, &tmpport )) goto parse_error;
       ot_try_bind( tmpip, tmpport, FLAG_UDP );
       ++bound;
+    } else if(!byte_diff(p,18,"listen.udp.workers" ) && isspace(p[18])) {
+      char *value = p + 18;
+      while( isspace(*value) ) ++value;
+      scan_uint( value, &g_udp_workers );
 #ifdef WANT_ACCESSLIST_WHITE
     } else if(!byte_diff(p, 16, "access.whitelist" ) && isspace(p[16])) {
       set_config_option( &g_accesslist_filename, p+17 );

opentracker.conf.sample

@@ -7,6 +7,15 @@
 #      If no listen option is given (here or on the command line), opentracker
 #      listens on 0.0.0.0:6969 tcp and udp.
 #
+#      The next variable determines if udp sockets are handled in the event
+#      loop (set it to 0, the default) or are handled in blocking reads in
+#      dedicated worker threads. You have to set this value before the
+#      listen.tcp_udp or listen.udp statements before it takes effect, but you
+#      can re-set it for each listen statement. Normally you should keep it at
+#      the top of the config file.
+#
+# listen.udp.workers 4
+#
 # listen.tcp_udp 0.0.0.0
 # listen.tcp_udp 192.168.0.1:80
 # listen.tcp_udp 10.0.0.5:6969

ot_udp.c

@@ -4,6 +4,8 @@
    $id$ */
 
 /* System */
+#include <stdlib.h>
+#include <pthread.h>
 #include <string.h>
 #include <arpa/inet.h>
 #include <stdio.h>
@@ -120,8 +122,35 @@ int handle_udp6( int64 serversocket, struct ot_workstruct *ws ) {
   return 1;
 }
 
-void udp_init( ) {
+static void* udp_worker( void * args ) {
+  int64 sock = (int64)args;
+  struct ot_workstruct ws;
+  memset( &ws, 0, sizeof(ws) );
+
+  ws.inbuf=malloc(G_INBUF_SIZE);
+  ws.outbuf=malloc(G_OUTBUF_SIZE);
+#ifdef    _DEBUG_HTTPERROR
+  ws.debugbuf=malloc(G_DEBUGBUF_SIZE);
+#endif
+
+  while( g_opentracker_running )
+    handle_udp6( sock, &ws );
+
+  free( ws.inbuf );
+  free( ws.outbuf );
+#ifdef    _DEBUG_HTTPERROR
+  free( ws.debugbuf );
+#endif
+  return NULL;
+}
 
+void udp_init( int64 sock, unsigned int worker_count ) {
+  pthread_t thread_id;
+#ifdef _DEBUG
+  fprintf( stderr, " installing %d workers on udp socket %ld", worker_count, (unsigned long)sock );
+#endif
+  while( worker_count-- )
+    pthread_create( &thread_id, NULL, udp_worker, (void *)sock );
 }
 
 const char *g_version_udp_c = "$Source$: $Revision$\n";

ot_udp.h

@@ -6,6 +6,7 @@
 #ifndef __OT_UDP_H__
 #define __OT_UDP_H__
 
+void udp_init( int64 sock, unsigned int worker_count );
 int  handle_udp6( int64 serversocket, struct ot_workstruct *ws );
 
 #endif