dpdk: add interrupt (power-saving) mode

When packet load is low, Suricata can run in interrupt
mode. This more resembles the classic approach of processing
packets - CPU cores run low and only fetch packets
on interrupt.

Ticket: OISF#5839

src/runmode-dpdk.c

@@ -111,6 +111,7 @@ static void *ParseDpdkConfigAndConfigureDevice(const char *iface);
 static void DPDKDerefConfig(void *conf);
 
 #define DPDK_CONFIG_DEFAULT_THREADS                     "auto"
+#define DPDK_CONFIG_DEFAULT_POWER_SAVING_MODE           0
 #define DPDK_CONFIG_DEFAULT_MEMPOOL_SIZE                65535
 #define DPDK_CONFIG_DEFAULT_MEMPOOL_CACHE_SIZE          "auto"
 #define DPDK_CONFIG_DEFAULT_RX_DESCRIPTORS              1024
@@ -126,6 +127,7 @@ static void DPDKDerefConfig(void *conf);
 
 DPDKIfaceConfigAttributes dpdk_yaml = {
     .threads = "threads",
+    .power_saving = "power-saving",
     .promisc = "promisc",
     .multicast = "multicast",
     .checksum_checks = "checksum-checks",
@@ -434,6 +436,15 @@ static int ConfigSetThreads(DPDKIfaceConfig *iconf, const char *entry_str)
     SCReturnInt(0);
 }
 
+static bool ConfigSetPowerSavingMode(DPDKIfaceConfig *iconf, int entry_bool)
+{
+    SCEnter();
+    if (entry_bool)
+        iconf->flags |= DPDK_POWER_SAVE;
+
+    SCReturnBool(true);
+}
+
 static int ConfigSetRxQueues(DPDKIfaceConfig *iconf, uint16_t nb_queues)
 {
     SCEnter();
@@ -695,6 +706,13 @@ static int ConfigLoad(DPDKIfaceConfig *iconf, const char *iface)
     if (retval < 0)
         SCReturnInt(retval);
 
+    retval = ConfGetChildValueBoolWithDefault(
+                     if_root, if_default, dpdk_yaml.power_saving, &entry_bool) != 1
+                     ? ConfigSetPowerSavingMode(iconf, DPDK_CONFIG_DEFAULT_POWER_SAVING_MODE)
+                     : ConfigSetPowerSavingMode(iconf, entry_bool);
+    if (retval != true)
+        SCReturnInt(-EINVAL);
+
     // currently only mapping "1 thread == 1 RX (and 1 TX queue in IPS mode)" is supported
     retval = ConfigSetRxQueues(iconf, (uint16_t)iconf->threads);
     if (retval < 0)
@@ -1106,6 +1124,11 @@ static void DeviceInitPortConf(const DPDKIfaceConfig *iconf,
             },
     };
 
+    if (iconf->flags & DPDK_POWER_SAVE) {
+        SCLogConfig("Switching to interrupt (power-saving) mode");
+        port_conf->intr_conf.rxq = 1;
+    }
+
     // configure RX offloads
     if (dev_info->rx_offload_capa & RTE_ETH_RX_OFFLOAD_RSS_HASH) {
         if (iconf->nb_rx_queues > 1) {

src/runmode-dpdk.h

@@ -25,6 +25,7 @@
 
 typedef struct DPDKIfaceConfigAttributes_ {
     const char *threads;
+    const char *power_saving;
     const char *promisc;
     const char *multicast;
     const char *checksum_checks;

src/source-dpdk.c

@@ -91,6 +91,43 @@ TmEcode NoDPDKSupportExit(ThreadVars *tv, const void *initdata, void **data)
 #include "util-dpdk-bonding.h"
 #include <numa.h>
 
+#define MIN_ZERO_POLL_COUNT          10U
+#define MIN_ZERO_POLL_COUNT_TO_SLEEP 10U
+#define MINIMUM_SLEEP_TIME_US        1U
+#define STANDARD_SLEEP_TIME_US       100U
+#define MAX_EPOLL_TIMEOUT_S          5U
+
+static int InterruptsRXEnable(uint16_t portid, uint16_t queueid)
+{
+    uint32_t event_data;
+    event_data = portid << UINT16_WIDTH | queueid;
+    int32_t ret = rte_eth_dev_rx_intr_ctl_q(portid, queueid, RTE_EPOLL_PER_THREAD,
+            RTE_INTR_EVENT_ADD, (void *)((uintptr_t)event_data));
+    if (ret) {
+        SCLogError("%s-Q%d: rte_eth_dev_rx_intr_ctl_q failed: %s", DPDKGetPortNameByPortID(portid),
+                queueid, rte_strerror(-ret));
+        return ret;
+    }
+
+    return 0;
+}
+
+static inline uint32_t InterruptsSleepHeuristic(uint32_t no_pkt_polls_count)
+{
+    if (no_pkt_polls_count < MIN_ZERO_POLL_COUNT_TO_SLEEP)
+        return MINIMUM_SLEEP_TIME_US;
+
+    return STANDARD_SLEEP_TIME_US;
+}
+
+static void InterruptsTurnOnOff(uint16_t port_id, uint16_t queue_id, bool on)
+{
+    if (on)
+        rte_eth_dev_rx_intr_enable(port_id, queue_id);
+    else
+        rte_eth_dev_rx_intr_disable(port_id, queue_id);
+}
+
 #define BURST_SIZE 32
 static struct timeval machine_start_time = { 0, 0 };
 
@@ -104,6 +141,7 @@ typedef struct DPDKThreadVars_ {
     TmSlot *slot;
     LiveDevice *livedev;
     ChecksumValidationMode checksum_mode;
+    int16_t intr_en;
     /* references to packet and drop counters */
     uint16_t capture_dpdk_packets;
     uint16_t capture_dpdk_rx_errs;
@@ -375,8 +413,22 @@ static TmEcode ReceiveDPDKLoop(ThreadVars *tv, void *data, void *slot)
     TmThreadsSetFlag(tv, THV_RUNNING);
     PacketPoolWait();
 
+    if (ptv->intr_en == 1) {
+        if (InterruptsRXEnable(ptv->port_id, ptv->queue_id) == 0) {
+            SCLogDebug("Enabling interrupt for port %d queue %d", ptv->port_id, ptv->queue_id);
+            ptv->intr_en = 1;
+        } else {
+            SCLogConfig(
+                    "Failed to enable interrupt (power-saving) mode, falling back to polling mode");
+            ptv->intr_en = 0;
+        }
+    }
     rte_eth_stats_reset(ptv->port_id);
     rte_eth_xstats_reset(ptv->port_id);
+
+    uint32_t pwd_zero_rx_packet_polls_count = 0;
+    uint32_t pwd_idle_hint = 0;
+
     while (1) {
         if (unlikely(suricata_ctl_flags != 0)) {
             SCLogDebug("Stopping Suricata!");
@@ -391,14 +443,34 @@ static TmEcode ReceiveDPDKLoop(ThreadVars *tv, void *data, void *slot)
         }
 
         nb_rx = rte_eth_rx_burst(ptv->port_id, ptv->queue_id, ptv->received_mbufs, BURST_SIZE);
+
         if (unlikely(nb_rx == 0)) {
             t = DPDKSetTimevalReal(&machine_start_time);
             uint64_t msecs = SCTIME_MSECS(t);
             if (msecs > last_timeout_msec + 100) {
                 TmThreadsCaptureHandleTimeout(tv, NULL);
                 last_timeout_msec = msecs;
             }
-            continue;
+
+            if (ptv->intr_en) {
+                pwd_zero_rx_packet_polls_count++;
+                if (pwd_zero_rx_packet_polls_count <= MIN_ZERO_POLL_COUNT)
+                    continue;
+
+                pwd_idle_hint = InterruptsSleepHeuristic(pwd_zero_rx_packet_polls_count);
+
+                if (pwd_idle_hint < STANDARD_SLEEP_TIME_US) {
+                    rte_delay_us(pwd_idle_hint);
+                } else {
+                    InterruptsTurnOnOff(ptv->port_id, ptv->queue_id, 1);
+                    struct rte_epoll_event event;
+                    rte_epoll_wait(RTE_EPOLL_PER_THREAD, &event, 1, MAX_EPOLL_TIMEOUT_S);
+                    InterruptsTurnOnOff(ptv->port_id, ptv->queue_id, 0);
+                    continue;
+                }
+            }
+        } else if (pwd_zero_rx_packet_polls_count) {
+            pwd_zero_rx_packet_polls_count = 0;
         }
 
         ptv->pkts += (uint64_t)nb_rx;
@@ -522,6 +594,7 @@ static TmEcode ReceiveDPDKThreadInit(ThreadVars *tv, const void *initdata, void
     ptv->checksum_mode = dpdk_config->checksum_mode;
 
     ptv->threads = dpdk_config->threads;
+    ptv->intr_en = (dpdk_config->flags & DPDK_POWER_SAVE) != 0;
     ptv->port_id = dpdk_config->port_id;
     ptv->out_port_id = dpdk_config->out_port_id;
     ptv->port_socket_id = dpdk_config->socket_id;

src/source-dpdk.h

@@ -36,8 +36,9 @@ typedef enum { DPDK_COPY_MODE_NONE, DPDK_COPY_MODE_TAP, DPDK_COPY_MODE_IPS } Dpd
 
 /* DPDK Flags */
 // General flags
-#define DPDK_PROMISC   (1 << 0) /**< Promiscuous mode */
-#define DPDK_MULTICAST (1 << 1) /**< Enable multicast packets */
+#define DPDK_PROMISC    (1 << 0) /**< Promiscuous mode */
+#define DPDK_MULTICAST  (1 << 1) /**< Enable multicast packets */
+#define DPDK_POWER_SAVE (1 << 2) /**< Power-saving mode */
 // Offloads
 #define DPDK_RX_CHECKSUM_OFFLOAD (1 << 4) /**< Enable chsum offload */
 

suricata.yaml.in

@@ -753,6 +753,7 @@ dpdk:
       # - auto takes all cores
       # in IPS mode it is required to specify the number of cores and the numbers on both interfaces must match
       threads: auto
+      power-saving: no # yes to switch to interrupt mode 
       promisc: true # promiscuous mode - capture all packets
       multicast: true # enables also detection on multicast packets
       checksum-checks: true # if Suricata should validate checksums