add extra config options, add offset to know when to call HS immediately

src/app-layer-detect-proto.c

@@ -276,7 +276,7 @@ static inline int PMGetProtoInspect(AppLayerProtoDetectThreadCtx *tctx,
     /* do the mpm search */
     uint32_t search_cnt = mpm_table[pm_ctx->mpm_ctx.mpm_type].Search(
             &pm_ctx->mpm_ctx, mpm_tctx, &tctx->pmq,
-            buf, searchlen);
+            buf, MPM_RXP_REQUIRED_HS_OFFSET + searchlen);
     if (search_cnt == 0) {
         if (buflen >= pm_ctx->mpm_ctx.maxdepth)
             return -1;

src/app-layer-ftp.c

@@ -415,7 +415,7 @@ static int FTPParseRequestCommand(
      * should make the use of the mpm very efficient */
     PmqReset(td->pmq);
     int mpm_cnt = mpm_table[FTP_MPM].Search(
-            ftp_mpm_ctx, td->ftp_mpm_thread_ctx, td->pmq, line->buf, line->len);
+            ftp_mpm_ctx, td->ftp_mpm_thread_ctx, td->pmq, line->buf, MPM_RXP_REQUIRED_HS_OFFSET + line->len);
     if (mpm_cnt) {
         *cmd_descriptor = &FtpCommands[td->pmq->rule_id_array[0]];
         SCReturnInt(1);

src/app-layer-smtp.c

@@ -943,7 +943,7 @@ static int SMTPProcessReply(SMTPState *state, Flow *f, AppLayerParserState *psta
      * should make the use of the mpm very efficient */
     PmqReset(td->pmq);
     int mpm_cnt = mpm_table[SMTP_MPM].Search(
-            smtp_mpm_ctx, td->smtp_mpm_thread_ctx, td->pmq, line->buf, 3);
+            smtp_mpm_ctx, td->smtp_mpm_thread_ctx, td->pmq, line->buf, MPM_RXP_REQUIRED_HS_OFFSET + 3);
     if (mpm_cnt == 0) {
         /* set decoder event - reply code invalid */
         SMTPSetEvent(state, SMTP_DECODER_EVENT_INVALID_REPLY);

src/suricata.c

@@ -217,7 +217,8 @@ bool g_disable_hashing = false;
 /* snapshot of the system's hugepages before system intitialization. */
 SystemHugepageSnapshot *prerun_snap = NULL;
 
-
+uint16_t g_rxp_deq_retry_after_us = 0;
+uint16_t g_rxp_minlength = 0;
 uint16_t g_worker_threads_cnt = 0;
 uint32_t g_mpm_groups_cnt = 0;
 

src/suricata.h

@@ -104,6 +104,9 @@ enum EngineMode {
     ENGINE_MODE_IPS,
 };
 
+#define MPM_RXP_REQUIRED_HS_OFFSET 100000000
+extern uint16_t g_rxp_deq_retry_after_us;
+extern uint16_t g_rxp_minlength;
 extern uint16_t g_worker_threads_cnt;
 extern uint32_t g_mpm_groups_cnt;
 

src/util-mpm-rxp.c

@@ -43,7 +43,7 @@
 #include <hs.h>
 #include <openssl/evp.h>
 
-#define MPM_RXP_DESCRIPTORS 2048 // descriptors used in the rxp queues
+#define MPM_RXP_DESCRIPTORS 1024 // descriptors used in the rxp queues
 #define MPM_RXP_OPERATIONS 32 // number of operations when de/enqueueing to RXP queues
 #define MPM_RXP_MAX_WORKERS 8 // should correspond to the number of workers
 #define MPM_RXP_PATTERNS_PATH "/tmp/suricata-mpm.patterns" // path to where extracted contents from Suricata rules will be temporarily stored
@@ -944,8 +944,8 @@ uint32_t SCRXPFinalizeSearchAndBulkDequeueAll(PrefilterRuleStore *pmq)
                                             g_rxp_ops, 
                                             MPM_RXP_OPERATIONS);
         if (deqed == 0) {
-            rte_delay_us_sleep(1);
-            pthread_yield();
+            rte_delay_us_sleep(g_rxp_deq_retry_after_us);
+            sched_yield();
         }
 
         // convert RXP results to PMQ
@@ -1017,7 +1017,7 @@ uint32_t SCRXPFinalizeSearchAndBulkDequeueAll(PrefilterRuleStore *pmq)
  * \retval matches Match count.
  */
 static uint32_t SCRXPSearchBulk(const MpmCtx *mpm_ctx, MpmThreadCtx *mpm_thread_ctx,
-                    PrefilterRuleStore *pmq, const uint8_t *buf, const uint32_t buflen)
+                    PrefilterRuleStore *pmq, const uint8_t *buf, uint32_t buflen)
 {
     uint32_t ret = 0;
     SCRXPCtx *rxp_ctx = (SCRXPCtx *)mpm_ctx->ctx;
@@ -1026,16 +1026,15 @@ static uint32_t SCRXPSearchBulk(const MpmCtx *mpm_ctx, MpmThreadCtx *mpm_thread_
     uint16_t enqed = 0;
     uint16_t prep_i = 0;
 
-    if (unlikely(buflen == 0)) {
+    if (unlikely(buflen == 0 || buflen == MPM_RXP_REQUIRED_HS_OFFSET)) {
         return 0;
     }
-    // TODO:
-    // try to think of traffic and rules that would be more suitable
-    // continue with stats - measure the time
-
+    if (buflen > MPM_RXP_REQUIRED_HS_OFFSET) {
+        buflen -= MPM_RXP_REQUIRED_HS_OFFSET;
+        goto hs_fallback;
+    }
     uint32_t seg_len = rxp_ctx->max_payload_size; // rxp_ctx->max_payload_size
-
-    if (buflen < 1) {
+    if (buflen < g_rxp_minlength) {
         goto hs_fallback;
     }
 
@@ -1539,14 +1538,46 @@ void RXPInit()
         FatalError("Unable to get %s configuration node", dpdk_node_query);
     }
     const char iface_name[] = "0000:51:00.0";
+    int32_t entry_int;
     ConfNode *if_node = ConfNodeLookupKeyValue(dpdk_node, "interface", iface_name);
     const char *entry_str = NULL;
     int retval = ConfGetChildValue(if_node, "threads", &entry_str);
     if (retval < 0)
         FatalError("Unable to get threads configuration node");
-    if (StringParseInt32(&g_worker_threads_cnt, 10, 0, entry_str) < 0) {
+    if (StringParseInt32(&entry_int, 10, 0, entry_str) < 0) {
         FatalError("Unable to parse threads configuration node");
     }
+    if (entry_int < 1) {
+        FatalError("Invalid threads configuration node");
+    }
+    g_worker_threads_cnt = (uint16_t)entry_int;
+    if (run_mode == RUNMODE_PCAP_FILE) {
+        // this mode runs in autofp mode and the 1 RX thread does the applayer 
+        // protocol detection, so we need to add 1 extra queue
+        g_worker_threads_cnt += 1;
+    }
+
+    retval = ConfGetChildValue(if_node, "rxp-min-buflen", &entry_str);
+    if (retval < 0)
+        FatalError("Unable to get rxp-min-buflen configuration node");
+    if (StringParseInt32(&entry_int, 10, 0, entry_str) < 0) {
+        FatalError("Unable to parse rxp-min-buflen configuration node");
+    }
+    if (entry_int < 1) {
+        FatalError("Invalid rxp-min-buflen configuration node");
+    }
+    g_rxp_minlength = (uint16_t)entry_int;
+
+    retval = ConfGetChildValue(if_node, "rxp-dequeue-retry-after-us", &entry_str);
+    if (retval < 0)
+        FatalError("Unable to get rxp-dequeue-retry-after-us configuration node");
+    if (StringParseInt32(&entry_int, 10, 0, entry_str) < 0) {
+        FatalError("Unable to parse rxp-dequeue-retry-after-us configuration node");
+    }
+    if (entry_int < 1) {
+        FatalError("Invalid rxp-dequeue-retry-after-us configuration node");
+    }
+    g_rxp_deq_retry_after_us = (uint16_t)entry_int;
 
     SCLogNotice("nb queues %u nb groups %u", g_worker_threads_cnt, g_mpm_groups_cnt);
     dev_conf.nb_queue_pairs = g_worker_threads_cnt;