ipcpd: Enable locking threads to a CPU core

This adds a function that locks a thread to a random core. This
greatly improves performance on multi-cpu systems. There is no
portable way to do this, this only implements it for GNU/Linux.

Signed-off-by: Dimitri Staessens <dimitri.staessens@ugent.be>
Signed-off-by: Sander Vrijders <sander.vrijders@ugent.be>

8 files changed, 49 insertions, 2 deletions

diff --git a/src/ipcpd/CMakeLists.txt b/src/ipcpd/CMakeLists.txt
index d717d14a..6679ab50 100644
--- a/src/ipcpd/CMakeLists.txt
+++ b/src/ipcpd/CMakeLists.txt
@@ -12,6 +12,8 @@ set(IPCP_ADD_THREADS 4 CACHE STRING
   "Number of extra threads to start when an IPCP faces thread starvation")
 set(IPCP_SCHED_THR_MUL 2 CACHE STRING
   "Number of scheduler threads per QoS cube")
+set(DISABLE_CORE_LOCK FALSE CACHE BOOL
+  "Disable locking performance threads to a core")
 
 if ((IPCP_QOS_CUBE_BE_PRIO LESS 0) OR (IPCP_QOS_CUBE_BE_PRIO GREATER 99))
   message(FATAL_ERROR "Invalid priority for best effort QoS cube")
diff --git a/src/ipcpd/config.h.in b/src/ipcpd/config.h.in
index a4893f50..de7d4498 100644
--- a/src/ipcpd/config.h.in
+++ b/src/ipcpd/config.h.in
@@ -35,7 +35,6 @@
 
 #define IPCP_MIN_THREADS    @IPCP_MIN_THREADS@
 #define IPCP_ADD_THREADS    @IPCP_ADD_THREADS@
-
 #cmakedefine HAVE_LIBGCRYPT
 
 /* normal IPCP */
@@ -45,6 +44,7 @@
 #define IPCP_SCHED_THR_MUL  @IPCP_SCHED_THR_MUL@
 #define PFT_SIZE            @PFT_SIZE@
 
+#cmakedefine DISABLE_CORE_LOCK
 #cmakedefine IPCP_FLOW_STATS
 
 /* udp */
diff --git a/src/ipcpd/eth/eth.c b/src/ipcpd/eth/eth.c
index b43caae1..d1ee088d 100644
--- a/src/ipcpd/eth/eth.c
+++ b/src/ipcpd/eth/eth.c
@@ -777,6 +777,8 @@ static void * eth_ipcp_sdu_reader(void * o)
 
         (void) o;
 
+        ipcp_lock_to_core();
+
         memset(br_addr, 0xff, MAC_SIZE * sizeof(uint8_t));
 
         while (true) {
@@ -893,6 +895,8 @@ static void * eth_ipcp_sdu_writer(void * o)
 
         (void) o;
 
+        ipcp_lock_to_core();
+
         while (true) {
                 fevent(eth_data.np1_flows, eth_data.fq, NULL);
 
diff --git a/src/ipcpd/ipcp.c b/src/ipcpd/ipcp.c
index d834997f..2321e2a8 100644
--- a/src/ipcpd/ipcp.c
+++ b/src/ipcpd/ipcp.c
@@ -20,6 +20,11 @@
  * Foundation, Inc., http://www.fsf.org/about/contact/.
  */
 
+#if defined(__linux__) && !defined(DISABLE_CORE_LOCK)
+#define _GNU_SOURCE
+#define NPROC (sysconf(_SC_NPROCESSORS_ONLN))
+#endif
+
 #define _POSIX_C_SOURCE 200112L
 #define __XSI_VISIBLE   500
 
@@ -42,6 +47,9 @@
 #include <string.h>
 #include <sys/socket.h>
 #include <stdlib.h>
+#if defined(__linux__) && !defined(DISABLE_CORE_LOCK)
+#include <unistd.h>
+#endif
 
 struct cmd {
         struct list_head next;
@@ -779,3 +787,24 @@ int ipcp_wait_state(enum ipcp_state         state,
 
         return ret;
 }
+
+void ipcp_lock_to_core(void)
+{
+#if defined(__linux__) && !defined(DISABLE_CORE_LOCK)
+        cpu_set_t           cpus;
+        size_t              cpu;
+
+        /* Choose a random core. */
+        cpu = rand() % NPROC;
+
+        CPU_ZERO(&cpus);
+        CPU_SET(cpu, &cpus);
+
+        if (pthread_setaffinity_np(pthread_self(), sizeof(cpus), &cpus))
+                log_warn("Failed to lock thread %lu to CPU %lu/%lu.",
+                         pthread_self(), cpu, NPROC);
+        else
+                log_dbg("Locked thread %lu to CPU %lu/%lu.",
+                        pthread_self(), cpu, NPROC);
+#endif
+}
diff --git a/src/ipcpd/ipcp.h b/src/ipcpd/ipcp.h
index c81ea1aa..3771c1e4 100644
--- a/src/ipcpd/ipcp.h
+++ b/src/ipcpd/ipcp.h
@@ -138,4 +138,7 @@ uint8_t *       ipcp_hash_dup(const uint8_t * hash);
 void            ipcp_hash_str(char            buf[],
                               const uint8_t * hash);
 
+/* Helper function to lock a thread to a core. */
+void            ipcp_lock_to_core(void);
+
 #endif /* OUROBOROS_IPCPD_IPCP_H */
diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c
index 787fc289..358f6388 100644
--- a/src/ipcpd/local/main.c
+++ b/src/ipcpd/local/main.c
@@ -97,6 +97,8 @@ static void * ipcp_local_sdu_loop(void * o)
 {
         (void) o;
 
+        ipcp_lock_to_core();
+
         while (true) {
                 int     fd;
                 ssize_t idx;
diff --git a/src/ipcpd/normal/sdu_sched.c b/src/ipcpd/normal/sdu_sched.c
index d4826ce5..56455721 100644
--- a/src/ipcpd/normal/sdu_sched.c
+++ b/src/ipcpd/normal/sdu_sched.c
@@ -20,12 +20,13 @@
  * Foundation, Inc., http://www.fsf.org/about/contact/.
  */
 
-#define _POSIX_C_SOURCE 199309L
+#define _POSIX_C_SOURCE 200112L
 
 #include "config.h"
 
 #include <ouroboros/errno.h>
 
+#include "ipcp.h"
 #include "sdu_sched.h"
 
 #include <assert.h>
@@ -67,6 +68,8 @@ static void * sdu_reader(void * o)
         sched = ((struct sched_info *) o)->sch;
         qc    = ((struct sched_info *) o)->qc;
 
+        ipcp_lock_to_core();
+
         free(o);
 
         fq = fqueue_create();
diff --git a/src/ipcpd/udp/main.c b/src/ipcpd/udp/main.c
index ba613150..be7491f4 100644
--- a/src/ipcpd/udp/main.c
+++ b/src/ipcpd/udp/main.c
@@ -470,6 +470,8 @@ static void * ipcp_udp_sdu_reader(void * o)
 
         (void) o;
 
+        ipcp_lock_to_core();
+
         while (true) {
                 pthread_rwlock_rdlock(&udp_data.flows_lock);
                 pthread_mutex_lock(&udp_data.fd_set_lock);
@@ -518,6 +520,8 @@ static void * ipcp_udp_sdu_loop(void * o)
 
         (void) o;
 
+        ipcp_lock_to_core();
+
         while (true) {
                 fevent(udp_data.np1_flows, udp_data.fq, NULL);
                 while ((fd = fqueue_next(udp_data.fq)) >= 0) {