Changeset 1397:239ba25870d8 in freeDiameter for libfdcore/routing_dispatch.c

Timestamp:

Nov 15, 2019, 7:40:37 PM (4 years ago)

Author:

Thomas Klausner <tk@giga.or.at>

Branch:

default

Phase:

public

Message:

Allow parametrizing the number of threads for routing in/out.

This is for high-load situations where freeDiameter was limited
by the corresponding queues.

File:

• libfdcore/routing_dispatch.c (modified) (5 diffs)

libfdcore/routing_dispatch.c

@@ -35 +35 @@
 
 #include "fdcore-internal.h"
+
+#ifdef linux
+/* This needs -D_USE_GNU, and since I have no idea what else that does, let's simply copy the declaration. */
+
+/* Set thread name visible in the kernel and its interfaces.  */
+extern int pthread_setname_np (pthread_t __target_thread, const char *__name);
+#endif
 
 /********************************************************************************/
@@ -1153 +1160 @@
 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 = NOTRUNNING;
-
-static pthread_t rt_in  = (pthread_t)NULL;
-static enum thread_state in_state = NOTRUNNING;
+static pthread_t * rt_out = NULL;
+static enum thread_state * out_state = NULL;
+
+static pthread_t * rt_in  = NULL;
+static enum thread_state * in_state = NULL;
 
 /* Initialize the routing and dispatch threads */
@@ -1165 +1171 @@
         int i;
         
-        /* Prepare the array for dispatch */
+        /* Prepare the array for threads */
         CHECK_MALLOC( disp_state = calloc(fd_g_config->cnf_dispthr, sizeof(enum thread_state)) );
         CHECK_MALLOC( dispatch = calloc(fd_g_config->cnf_dispthr, sizeof(pthread_t)) );
+        CHECK_MALLOC( out_state = calloc(fd_g_config->cnf_rtoutthr, sizeof(enum thread_state)) );
+        CHECK_MALLOC( rt_out = calloc(fd_g_config->cnf_rtoutthr, sizeof(pthread_t)) );
+        CHECK_MALLOC( in_state = calloc(fd_g_config->cnf_rtinthr, sizeof(enum thread_state)) );
+        CHECK_MALLOC( rt_in = calloc(fd_g_config->cnf_rtinthr, sizeof(pthread_t)) );
         
         /* 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) );
+#ifdef linux
+                pthread_setname_np(dispatch[i], "fd-dispatch");
+#endif
+        }
+        for (i=0; i < fd_g_config->cnf_rtoutthr; i++) {
+                CHECK_POSIX( pthread_create( &rt_out[i], NULL, routing_out_thr, &out_state[i] ) );
+#ifdef linux
+                pthread_setname_np(rt_out[i], "fd-routing-out");
+#endif
+        }
+        for (i=0; i < fd_g_config->cnf_rtinthr; i++) {
+                CHECK_POSIX( pthread_create( &rt_in[i], NULL, routing_in_thr, &in_state[i] ) );
+#ifdef linux
+                pthread_setname_np(rt_in[i], "fd-routing-in");
+#endif
+        }
         
         /* Later: TODO("Set the thresholds for the queues to create more threads as needed"); */
@@ -1245 +1268 @@
         
         /* Stop the routing IN thread */
-        stop_thread_delayed(&in_state, &rt_in, "IN routing");
+        if (rt_in != NULL) {
+                for (i=0; i < fd_g_config->cnf_rtinthr; i++) {
+                        stop_thread_delayed(&in_state[i], &rt_in[i], "IN routing");
+                }
+                free(rt_in);
+                rt_in = NULL;
+        }
+        if (in_state != NULL) {
+                free(in_state);
+                in_state = NULL;
+        }
         
         /* Destroy the outgoing queue */
@@ -1251 +1284 @@
         
         /* Stop the routing OUT thread */
-        stop_thread_delayed(&out_state, &rt_out, "OUT routing");
+        if (rt_out != NULL) {
+                for (i=0; i < fd_g_config->cnf_rtinthr; i++) {
+                        stop_thread_delayed(&out_state[i], &rt_out[i], "OUT routing");
+                }
+                free(rt_out);
+                rt_out = NULL;
+        }
+        if (out_state != NULL) {
+                free(out_state);
+                out_state = NULL;
+        }
         
         /* Destroy the local queue */