dpdk: use DPDK EAL threads for workers v5.1

doc/userguide/upgrade.rst

@@ -77,6 +77,11 @@ Other Changes
   really enforced and there will be no hassh computation
   even if rules try to use it.
 
+- In DPDK capture mode, management threads need to be explicitly defined and
+  they cannot overlap with worker CPU affinity. Previously, Suricata only
+  warned if overlap was present.
+- DPDK EAL lcore arguments cannot be used. Suricata now manages the
+  lcore assignments itself based on the CPU affinity configuration.
 
 Upgrading to 8.0.1
 ------------------

src/runmode-dpdk.c

@@ -74,8 +74,15 @@ static void ArgumentsInit(struct Arguments *args, uint16_t capacity);
 static void ArgumentsCleanup(struct Arguments *args);
 static void ArgumentsAdd(struct Arguments *args, char *value);
 static void ArgumentsAddOptionAndArgument(struct Arguments *args, const char *opt, const char *arg);
+static void ArgumentsAddLcoreArguments(struct Arguments *args);
+static void ArgumentsEnsureEalOptionAllowed(const char *opt);
 static void InitEal(void);
 
+static uint16_t CountDigits(uint32_t value);
+static char *ConfigLcoreArgValGet(void);
+static int ConfigLcoreWorkersSet(uint32_t *cpus, size_t cap);
+static uint32_t ConfigLcoreMainGet(void);
+
 static void ConfigSetIface(DPDKIfaceConfig *iconf, const char *entry_str);
 static int ConfigSetThreads(DPDKIfaceConfig *iconf, const char *entry_str);
 static int ConfigSetRxQueues(DPDKIfaceConfig *iconf, uint16_t nb_queues, uint16_t max_queues);
@@ -180,6 +187,21 @@ static uint64_t GreatestPowOf2UpTo(uint64_t num)
     return num;
 }
 
+/**
+ * \brief Counts number of decimal digits in a given value
+ * \param value Value to count digits for
+ * \return Number of decimal digits in value (e.g. 123 -> 3, 0 -> 1)
+ */
+static uint16_t CountDigits(uint32_t value)
+{
+    uint16_t digits = 1;
+    while (value >= 10U) {
+        value /= 10U;
+        digits++;
+    }
+    return digits;
+}
+
 static char *AllocArgument(size_t arg_len)
 {
     SCEnter();
@@ -286,6 +308,181 @@ static void ArgumentsAddOptionAndArgument(struct Arguments *args, const char *op
     SCReturn;
 }
 
+static void ConfigLcoreManagementCpuSetGet(cpu_set_t *management_set)
+{
+    if (management_set == NULL) {
+        FatalError("Invalid buffer passed for management CPU set");
+    }
+
+    CPU_ZERO(management_set);
+
+    ThreadsAffinityType *taf = GetAffinityTypeForNameAndIface("management-cpu-set", NULL);
+    if (taf == NULL || CPU_COUNT(&taf->cpu_set) == 0) {
+        FatalError("Unable to obtain CPU affinity for \"management-cpu-set\"");
+    }
+
+    CPU_OR(management_set, management_set, &taf->cpu_set);
+}
+
+static uint32_t ConfigLcoreMainGet(void)
+{
+    cpu_set_t management_set;
+    ConfigLcoreManagementCpuSetGet(&management_set);
+
+    const uint32_t max_cpu = MIN((uint32_t)RTE_MAX_LCORE, (uint32_t)CPU_SETSIZE);
+    for (uint32_t cpu = 0; cpu < max_cpu; cpu++) {
+        if (CPU_ISSET(cpu, &management_set)) {
+            return cpu;
+        }
+    }
+
+    FatalError("Unable to obtain CPU affinity for \"management-cpu-set\"");
+}
+
+static void ConfigLcoreWorkersCpuSetMerge(cpu_set_t *worker_set)
+{
+    if (worker_set == NULL) {
+        FatalError("Invalid buffer passed for worker CPU set");
+    }
+
+    CPU_ZERO(worker_set);
+
+    bool fallback_to_worker_affinity = false;
+    LiveDevice *ldev = NULL, *ndev = NULL;
+    while (LiveDeviceForEach(&ldev, &ndev)) {
+        ThreadsAffinityType *taf = GetAffinityTypeForNameAndIface("worker-cpu-set", ldev->dev);
+        if (taf == NULL) {
+            fallback_to_worker_affinity = true;
+            continue;
+        }
+
+        CPU_OR(worker_set, worker_set, &taf->cpu_set);
+    }
+
+    if (fallback_to_worker_affinity) {
+        ThreadsAffinityType *taf = GetAffinityTypeForNameAndIface("worker-cpu-set", NULL);
+        if (taf == NULL) {
+            FatalError("Unable to obtain CPU affinity for \"worker-cpu-set\"");
+        }
+
+        CPU_OR(worker_set, worker_set, &taf->cpu_set);
+    }
+}
+
+static int ConfigLcoreWorkersSet(uint32_t *cpus, size_t cap)
+{
+    SCEnter();
+    if (cpus == NULL || cap == 0) {
+        FatalError("Invalid buffer passed for worker CPU set");
+    }
+
+    cpu_set_t worker_set;
+    ConfigLcoreWorkersCpuSetMerge(&worker_set);
+
+    size_t wrk_cpus = CPU_COUNT(&worker_set);
+    if (wrk_cpus > cap) {
+        FatalError(
+                "Too many worker CPU cores configured (%zu), max supported is %zu", wrk_cpus, cap);
+    }
+
+    size_t count = 0;
+    const uint32_t max_cpu = MIN((uint32_t)RTE_MAX_LCORE, (uint32_t)CPU_SETSIZE);
+    for (uint32_t cpu = 0; cpu < max_cpu; cpu++) {
+        if (!CPU_ISSET(cpu, &worker_set)) {
+            continue;
+        }
+        cpus[count++] = cpu;
+    }
+
+    if (count == 0) {
+        FatalError("No worker CPUs detected");
+    }
+
+    SCReturnInt((int)count);
+}
+
+static char *ConfigLcoreArgValGet(void)
+{
+    SCEnter();
+    uint32_t lcore_list[RTE_MAX_LCORE + 1];
+    size_t lcore_cnt = ConfigLcoreWorkersSet(lcore_list, ARRAY_SIZE(lcore_list) - 1);
+    uint32_t main_lcore = ConfigLcoreMainGet();
+    lcore_list[lcore_cnt++] = main_lcore;
+
+    size_t required_len = 0;
+    for (size_t i = 0; i < lcore_cnt; i++) {
+        required_len += CountDigits(lcore_list[i]);
+    }
+    required_len += (lcore_cnt - 1); // commas
+
+    char *lcore_arg = AllocArgument(required_len);
+    size_t offset = 0;
+    for (size_t i = 0; i < lcore_cnt; i++) {
+        size_t remaining = required_len + 1 - offset;
+        int ret = snprintf(&lcore_arg[offset], remaining, "%u", lcore_list[i]);
+        if (ret < 0 || (size_t)ret >= remaining) {
+            FatalError("Conversion of CPU affinity to lcore argument failed");
+        }
+        offset += (size_t)ret;
+        if (i + 1 < lcore_cnt) {
+            lcore_arg[offset++] = ',';
+        }
+    }
+
+    lcore_arg[offset] = '\0';
+
+    SCReturnCharPtr(lcore_arg);
+}
+
+static void ArgumentsAddLcoreArguments(struct Arguments *args)
+{
+    SCEnter();
+    ArgumentsAdd(args, AllocAndSetArgument("-l"));
+    char *lcore_arg = ConfigLcoreArgValGet();
+    ArgumentsAdd(args, lcore_arg);
+
+    ArgumentsAdd(args, AllocAndSetArgument("--main-lcore"));
+    uint32_t main_lcore = ConfigLcoreMainGet();
+    uint16_t digits = CountDigits(main_lcore);
+    char *main_lcore_arg = AllocArgument(digits);
+    int ret = snprintf(main_lcore_arg, digits + 1, "%u", main_lcore);
+    if (ret < 0 || ret > digits) {
+        FatalError("Conversion of management affinity to main lcore argument failed");
+    }
+    ArgumentsAdd(args, main_lcore_arg);
+
+    SCReturn;
+}
+
+static void ArgumentsEnsureEalOptionAllowed(const char *opt)
+{
+    if (opt == NULL) {
+        return;
+    }
+
+    const char *sanitized = opt;
+    while (*sanitized == '-') {
+        sanitized++;
+    }
+    if (*sanitized == '\0') {
+        return;
+    }
+
+    static const char *const forbidden_opts[] = {
+        "l",
+        "lcores",
+        "c",
+        "main-lcore",
+        "master-lcore",
+    };
+
+    for (size_t i = 0; i < ARRAY_SIZE(forbidden_opts); i++) {
+        if (strcasecmp(sanitized, forbidden_opts[i]) == 0) {
+            FatalError("DPDK EAL option \"%s\" conflicts with Suricata CPU affinity settings", opt);
+        }
+    }
+}
+
 static void InitEal(void)
 {
     SCEnter();
@@ -301,8 +498,10 @@ static void InitEal(void)
 
     ArgumentsInit(&args, EAL_ARGS);
     ArgumentsAdd(&args, AllocAndSetArgument("suricata"));
+    ArgumentsAddLcoreArguments(&args);
 
     TAILQ_FOREACH (param, &eal_params->head, next) {
+        ArgumentsEnsureEalOptionAllowed(param->name);
         if (SCConfNodeIsSequence(param)) {
             const char *key = param->name;
             SCConfNode *val;
@@ -414,8 +613,9 @@ static int ConfigSetThreads(DPDKIfaceConfig *iconf, const char *entry_str)
         SCLogError("No worker CPU cores with configured affinity were configured");
         SCReturnInt(-EINVAL);
     } else if (UtilAffinityCpusOverlap(wtaf, mtaf) != 0) {
-        SCLogWarning("Worker threads should not overlap with management threads in the CPU core "
-                     "affinity configuration");
+        SCLogError("Worker threads cannot overlap with management threads in the CPU core "
+                   "affinity configuration");
+        SCReturnInt(-EINVAL);
     }
 
     const char *active_runmode = RunmodeGetActive();

src/threadvars.h

@@ -57,11 +57,13 @@ struct TmSlot_;
 
 /** \brief Per thread variable structure */
 typedef struct ThreadVars_ {
-    pthread_t t;
+    uint64_t thread_id;
     /** function pointer to the function that runs the packet pipeline for
      *  this thread. It is passed directly to pthread_create(), hence the
      *  void pointers in and out. */
     void *(*tm_func)(void *);
+    void (*tm_spawn)(struct ThreadVars_ *);
+    void (*tm_join)(struct ThreadVars_ *);
 
     char name[16];
     char *printable_name;

src/tm-modules.h

@@ -47,6 +47,10 @@ typedef void (*ThreadExitPrintStatsFunc)(ThreadVars *, void *);
 typedef struct TmModule_ {
     const char *name;
 
+    /** thread management, if unspecified, falls back to pthreads */
+    void (*ThreadSpawn)(ThreadVars *);
+    void (*ThreadJoin)(ThreadVars *);
+
     /** thread handling */
     TmEcode (*ThreadInit)(ThreadVars *, const void *, void **);
     void (*ThreadExitPrintStats)(ThreadVars *, void *);