Configurable number of server threads

doc/freediameter.conf.sample

@@ -144,6 +144,11 @@
 # Default: Relaying is enabled.
 #NoRelay;
 
+# Number of server threads that can handle incoming messages at the same time.
+#  TODO: implement dynamic # of threads depending on the length of the queue.
+# Default: 4
+#AppServThreads = 4;
+
 # Other applications are configured by loading appropriate extensions.
 
 ##############################################################

freeDiameter/config.c

@@ -58,6 +58,7 @@ int fd_conf_init()
 	fd_g_config->cnf_port     = 3868;
 	fd_g_config->cnf_port_tls = 3869;
 	fd_g_config->cnf_sctp_str = 30;
+	fd_g_config->cnf_dispthr  = 4;
 	fd_list_init(&fd_g_config->cnf_endpoints, NULL);
 	fd_list_init(&fd_g_config->cnf_apps, NULL);
 	#ifdef DISABLE_SCTP
@@ -91,6 +92,7 @@ void fd_conf_dump()
 	fd_log_debug("  Local port ............. : %hu\n", fd_g_config->cnf_port);
 	fd_log_debug("  Local secure port ...... : %hu\n", fd_g_config->cnf_port_tls);
 	fd_log_debug("  Number of SCTP streams . : %hu\n", fd_g_config->cnf_sctp_str);
+	fd_log_debug("  Number of server threads : %hu\n", fd_g_config->cnf_dispthr);
 	if (FD_IS_LIST_EMPTY(&fd_g_config->cnf_endpoints)) {
 		fd_log_debug("  Local endpoints ........ : Default (use all available)\n");
 	} else {

freeDiameter/fdd.l

@@ -122,6 +122,7 @@ qstring		\"[^\"\n]*\"
 (?i:"Prefer_TCP")	{ return PREFERTCP;	}
 (?i:"TLS_old_method")	{ return OLDTLS;	}
 (?i:"SCTP_streams")	{ return SCTPSTREAMS;	}
+(?i:"AppServThreads")	{ return APPSERVTHREADS;}
 (?i:"ListenOn")		{ return LISTENON;	}
 (?i:"TcTimer")		{ return TCTIMER;	}
 (?i:"TwTimer")		{ return TWTIMER;	}

freeDiameter/fdd.y

@@ -104,6 +104,7 @@ struct peer_info fddpi;
 %token		OLDTLS
 %token		NOTLS
 %token		SCTPSTREAMS
+%token		APPSERVTHREADS
 %token		LISTENON
 %token		TCTIMER
 %token		TWTIMER
@@ -132,6 +133,7 @@ conffile:		/* Empty is OK -- for simplicity here, we reject in daemon later */
 			| conffile sctpstreams
 			| conffile listenon
 			| conffile norelay
+			| conffile appservthreads
 			| conffile noip
 			| conffile noip6
 			| conffile notcp
@@ -230,6 +232,14 @@ norelay:		NORELAY ';'
 			}
 			;
 
+appservthreads:		APPSERVTHREADS '=' INTEGER ';'
+			{
+				CHECK_PARAMS_DO( ($3 > 0) && ($3 < 1024),
+					{ yyerror (&yylloc, conf, "Invalid value"); YYERROR; } );
+				conf->cnf_dispthr = (uint16_t)$3;
+			}
+			;
+
 noip:			NOIP ';'
 			{
 				if (got_peer_noipv6) { 

freeDiameter/main.c

@@ -95,11 +95,13 @@ int main(int argc, char * argv[])
 	CHECK_FCT(  fd_queues_init()  );
 	CHECK_FCT(  fd_msg_init()  );
 	CHECK_FCT(  fd_p_expi_init()  );
-	CHECK_FCT(  fd_rtdisp_init()  );
 	
 	/* Parse the configuration file */
 	CHECK_FCT( fd_conf_parse() );
 	
+	/* Create the daemon's threads */
+	CHECK_FCT(  fd_rtdisp_init()  );
+	
 	/* Load the dynamic extensions */
 	CHECK_FCT(  fd_ext_load()  );
 	

freeDiameter/routing_dispatch.c

@@ -1069,10 +1069,10 @@ static void * routing_out_thr(void * arg)
 /*                     The functions for the other files                        */
 /********************************************************************************/
 
-/* Later: make this more dynamic */
-static pthread_t dispatch = (pthread_t)NULL;
-static enum thread_state disp_state = INITIAL;
+static pthread_t * dispatch = NULL;
+static enum thread_state * disp_state = NULL;
 
+/* Later: make this more dynamic */
 static pthread_t rt_out = (pthread_t)NULL;
 static enum thread_state out_state = INITIAL;
 
@@ -1082,7 +1082,16 @@ static enum thread_state in_state = INITIAL;
 /* Initialize the routing and dispatch threads */
 int fd_rtdisp_init(void)
 {
-	CHECK_POSIX( pthread_create( &dispatch, NULL, dispatch_thr, &disp_state ) );
+	int i;
+	
+	/* Prepare the array for dispatch */
+	CHECK_MALLOC( dispatch = calloc(fd_g_config->cnf_dispthr, sizeof(pthread_t)) );
+	CHECK_MALLOC( disp_state = calloc(fd_g_config->cnf_dispthr, sizeof(enum thread_state)) );
+	
+	/* Create the threads */
+	for (i=0; i < fd_g_config->cnf_dispthr; i++) {
+		CHECK_POSIX( pthread_create( &dispatch[i], NULL, dispatch_thr, &disp_state[i] ) );
+	}
 	CHECK_POSIX( pthread_create( &rt_out, NULL, routing_out_thr, &out_state) );
 	CHECK_POSIX( pthread_create( &rt_in,  NULL, routing_in_thr,  &in_state) );
 	
@@ -1138,6 +1147,8 @@ static void stop_thread_delayed(enum thread_state *st, pthread_t * thr, char * t
 /* Stop the thread after up to one second of wait */
 int fd_rtdisp_fini(void)
 {
+	int i;
+	
 	/* Destroy the incoming queue */
 	CHECK_FCT_DO( fd_queues_fini(&fd_g_incoming), /* ignore */);
 	
@@ -1154,7 +1165,9 @@ int fd_rtdisp_fini(void)
 	CHECK_FCT_DO( fd_queues_fini(&fd_g_local), /* ignore */);
 	
 	/* Stop the Dispatch thread */
-	stop_thread_delayed(&disp_state, &dispatch, "Dispatching");
+	for (i=0; i < fd_g_config->cnf_dispthr; i++) {
+		stop_thread_delayed(&disp_state[i], &dispatch[i], "Dispatching");
+	}
 	
 	return 0;
 }

include/freeDiameter/freeDiameter.h

@@ -90,6 +90,7 @@ struct fd_config {
 	uint16_t	 cnf_sctp_str;	/* default max number of streams for SCTP associations (def: 30) */
 	struct fd_list	 cnf_endpoints;	/* the local endpoints to bind the server to. list of struct fd_endpoint. default is empty (bind all) */
 	struct fd_list	 cnf_apps;	/* Applications locally supported (except relay, see flags). Use fd_disp_app_support to add one. list of struct fd_app. */
+	uint16_t	 cnf_dispthr;	/* Number of dispatch threads to create */
 	struct {
 		unsigned no_fwd : 1;	/* the peer does not relay messages (0xffffff app id) */
 		unsigned no_ip4 : 1;	/* disable IP */