WIP Introduce support for pinning specific analysis threads to a logical CPU core

docs/configuration.md

@@ -83,6 +83,7 @@ input_defaults = {
 
 analysis_defaults = {
   -- see: Default Sandbox Configuration Variables
+  -- cpu_affinity           = false
 }
 
 output_defaults = {
@@ -126,6 +127,8 @@ output_defaults = {
   `timer_event` function can inject in a single invocation (count,
   default 10).
 
+* **cpu_affinity** - if analysis threads should be pinned to a logical CPU core (bool, default false)
+
 #### Default Output Sandbox Configuration Variables
 
 * **remove_checkpoints_on_terminate** - removes the checkpoint entries when the

src/hs_analysis_plugins.c

@@ -6,6 +6,8 @@
 
 /** @brief Hindsight analysis plugin loader @file */
 
+#define _GNU_SOURCE
+
 #include "hs_analysis_plugins.h"
 
 #include <ctype.h>
@@ -16,6 +18,7 @@
 #include <luasandbox/util/protobuf.h>
 #include <luasandbox_output.h>
 #include <pthread.h>
+#include <sched.h>
 #include <signal.h>
 #include <stdlib.h>
 #include <string.h>
@@ -132,6 +135,7 @@ create_analysis_plugin(const hs_config *cfg, hs_sandbox_config *sbc)
   p->pm_im_limit = sbc->pm_im_limit;
   p->te_im_limit = sbc->te_im_limit;
   p->shutdown_terminate = sbc->shutdown_terminate;
+  p->cpu_affinity = sbc->cpu_affinity;
   p->ticker_interval = sbc->ticker_interval;
   int stagger = p->ticker_interval > 60 ? 60 : p->ticker_interval;
   // distribute when the timer_events will fire
@@ -258,6 +262,7 @@ static void add_to_analysis_plugins(const hs_sandbox_config *cfg,
   int idx = -1;
   int thread = cfg->thread % plugins->cfg->analysis_threads;
   hs_analysis_thread *at = &plugins->list[thread];
+  at->cpu_affinity = p->cpu_affinity;
   p->at = at;
 
   pthread_mutex_lock(&at->list_lock);
@@ -435,9 +440,21 @@ static void shutdown_timer_event(hs_analysis_thread *at)
 
 static void* input_thread(void *arg)
 {
+  cpu_set_t set;
   hs_analysis_thread *at = (hs_analysis_thread *)arg;
   hs_log(NULL, g_module, 6, "starting thread: %d", at->tid);
 
+  if (at->cpu_affinity) {
+    CPU_ZERO(&set);
+    CPU_SET(at->tid, &set);
+
+    if (sched_setaffinity(0, sizeof(set), &set) == -1) {
+      hs_log(NULL, g_module, 6, "failed to set CPU affinity for thead: %d", at->tid);
+    } else {
+      hs_log(NULL, g_module, 6, "set CPU affinity for thead: %d", at->tid);
+    }
+  }
+
   lsb_heka_message msg;
   lsb_init_heka_message(&msg, 8);
 

src/hs_analysis_plugins.h

@@ -39,6 +39,7 @@ struct hs_analysis_plugin {
   unsigned            im_limit;
   unsigned            pm_im_limit;
   unsigned            te_im_limit;
+  bool                cpu_affinity;
   time_t              ticker_expires;
 };
 
@@ -73,6 +74,7 @@ struct hs_analysis_thread {
   int       utilization;
   bool      stop;
   bool      sample;
+  bool      cpu_affinity;
 #ifdef HINDSIGHT_CLI
   bool      terminated;
 #endif

src/hs_config.c

@@ -58,6 +58,7 @@ static const char *cfg_sb_restricted_headers = "restricted_headers";
 static const char *cfg_sb_filename = "filename";
 static const char *cfg_sb_ticker_interval = "ticker_interval";
 static const char *cfg_sb_thread = "thread";
+static const char *cfg_sb_cpu_affinity = "cpu_affinity";
 static const char *cfg_sb_async_buffer = "async_buffer_size";
 static const char *cfg_sb_matcher = "message_matcher";
 static const char *cfg_sb_shutdown_terminate = "shutdown_on_terminate";
@@ -75,6 +76,7 @@ static void init_sandbox_config(hs_sandbox_config *cfg)
   cfg->message_matcher = NULL;
 
   cfg->thread = UINT_MAX;
+  cfg->cpu_affinity = false;
   cfg->async_buffer_size = 0;
   cfg->output_limit = 1024 * 64;
   cfg->memory_limit = 1024 * 1024 * 8;
@@ -519,6 +521,8 @@ bool hs_load_sandbox_config(const char *dir,
   if (type == 'a') {
     ret = get_unsigned_int(L, LUA_GLOBALSINDEX, cfg_sb_thread,
                            &cfg->thread);
+    ret = get_bool_item(L, LUA_GLOBALSINDEX, cfg_sb_cpu_affinity,
+                         &cfg->cpu_affinity);
     ret = get_unsigned_int(L, LUA_GLOBALSINDEX, cfg_sb_pm_im_limit,
                             &cfg->pm_im_limit);
     ret = get_unsigned_int(L, LUA_GLOBALSINDEX, cfg_sb_te_im_limit,
@@ -843,6 +847,8 @@ bool hs_output_runtime_cfg(lsb_output_buffer *ob, char type, const hs_config *cf
 
   if (type == 'a') {
     lsb_outputf(ob, "thread = %u\n", sbc->thread);
+    lsb_outputf(ob, "cpu_affinity = %s\n",
+              sbc->cpu_affinity ? "true" : "false");
     lsb_outputf(ob, "process_message_inject_limit = %u\n", sbc->pm_im_limit);
     lsb_outputf(ob, "timer_event_inject_limit = %u\n", sbc->te_im_limit);
   }

src/hs_config.h

@@ -36,6 +36,7 @@ typedef struct hs_sandbox_config
   char *message_matcher; // analysis/output sandbox only
 
   unsigned thread; // analysis sandbox only
+  bool cpu_affinity; // analysis sandbox only
   unsigned async_buffer_size; // output sandbox only
   unsigned output_limit;
   unsigned memory_limit;