tmqh/flow: transfer more than one packet per fetch

To avoid per packet locking overhead, transfer all available packets.

src/runmode-pcap-file.c

@@ -159,11 +159,8 @@ int RunModeFilePcapAutoFp(void)
     snprintf(tname, sizeof(tname), "%s#01", thread_name_autofp);
 
     /* create the threads */
-    ThreadVars *tv_receivepcap =
+    ThreadVars *tv_receivepcap = TmThreadCreatePacketHandler(
-        TmThreadCreatePacketHandler(tname,
+            tname, "packetpool", "packetpool", queues, "flow", "pktacqloop");
-                                    "packetpool", "packetpool",
-                                    queues, "flow",
-                                    "pktacqloop");
     SCFree(queues);
 
     if (tv_receivepcap == NULL) {
@@ -194,11 +191,8 @@ int RunModeFilePcapAutoFp(void)
         SCLogDebug("tname %s, qname %s", tname, qname);
         SCLogDebug("Assigning %s affinity to cpu %u", tname, cpu);
 
-        ThreadVars *tv_detect_ncpu =
+        ThreadVars *tv_detect_ncpu = TmThreadCreatePacketHandler(
-            TmThreadCreatePacketHandler(tname,
+                tname, qname, "flow", "packetpool", "packetpool", "varslot");
-                                        qname, "flow",
-                                        "packetpool", "packetpool",
-                                        "varslot");
         if (tv_detect_ncpu == NULL) {
             FatalError("TmThreadsCreate failed");
         }

src/source-af-packet.c

@@ -944,6 +944,7 @@ next_frame:
                 break;
         }
     }
+    TmThreadFlushOutQueue(ptv->tv);
     if (emergency_flush) {
         AFPDumpCounters(ptv);
     }
@@ -1084,9 +1085,11 @@ static int AFPReadFromRingV3(AFPThreadVars *ptv)
         ptv->frame_offset = (ptv->frame_offset + 1) % ptv->req.v3.tp_block_nr;
         /* return to maintenance task after one loop on the ring */
         if (ptv->frame_offset == 0) {
+            TmThreadFlushOutQueue(ptv->tv);
             SCReturnInt(AFP_READ_OK);
         }
     }
+    TmThreadFlushOutQueue(ptv->tv);
     SCReturnInt(AFP_READ_OK);
 }
 

src/source-nfq.c

@@ -1037,6 +1037,7 @@ TmEcode ReceiveNFQLoop(ThreadVars *tv, void *data, void *slot)
         }
         NFQRecvPkt(nq, ntv);
 
+        TmThreadFlushOutQueue(tv);
         StatsSyncCountersIfSignalled(&tv->stats);
     }
     SCReturnInt(TM_ECODE_OK);

src/source-pcap-file-helper.c

@@ -229,6 +229,7 @@ TmEcode PcapFileDispatch(PcapFileFileVars *ptv)
             SCLogError("Pcap callback PcapFileCallbackLoop failed for %s", ptv->filename);
             loop_result = TM_ECODE_FAILED;
         }
+        TmThreadFlushOutQueue(ptv->shared->tv);
         StatsSyncCountersIfSignalled(&ptv->shared->tv->stats);
     }
 

src/tm-queuehandlers.h

@@ -40,6 +40,7 @@ typedef struct Tmqh_ {
     PacketQueueNoLock (*InHandler)(ThreadVars *);
     void (*InShutdownHandler)(ThreadVars *);
     void (*OutHandler)(ThreadVars *, Packet *);
+    void (*OutFlush)(ThreadVars *);
     void *(*OutHandlerCtxSetup)(const char *);
     void (*OutHandlerCtxFree)(void *);
 #ifdef UNITTESTS

src/tm-threads.h

@@ -27,6 +27,7 @@
 
 #include "tmqh-packetpool.h"
 #include "tm-threads-common.h"
+#include "tm-queuehandlers.h"
 #include "tm-modules.h"
 #include "flow.h" // for the FlowQueue
 
@@ -278,6 +279,18 @@ static inline void TmThreadsCaptureBreakLoop(ThreadVars *tv)
     }
 }
 
+static inline void TmThreadFlushOutQueue(ThreadVars *tv)
+{
+    if (tv->outctx != NULL && tv->outq_id != TMQH_NOT_SET) {
+        Tmqh *qh = TmqhGetQueueHandlerByID(tv->outq_id);
+        if (qh != NULL) {
+            if (qh->OutFlush != NULL) {
+                qh->OutFlush(tv);
+            }
+        }
+    }
+}
+
 void TmThreadsSealThreads(void);
 void TmThreadsUnsealThreads(void);
 void TmThreadsListThreads(void);

src/tmqh-flow.c

@@ -1,4 +1,4 @@
-/* Copyright (C) 2007-2020 Open Information Security Foundation
+/* Copyright (C) 2007-2025 Open Information Security Foundation
  *
  * You can copy, redistribute or modify this Program under the terms of
  * the GNU General Public License version 2 as published by the Free
@@ -44,12 +44,15 @@ PacketQueueNoLock TmqhInputFlow(ThreadVars *t);
 void TmqhOutputFlowHash(ThreadVars *t, Packet *p);
 void TmqhOutputFlowIPPair(ThreadVars *t, Packet *p);
 static void TmqhOutputFlowFTPHash(ThreadVars *t, Packet *p);
+static void TmqhOutputFlush(ThreadVars *tv);
 void *TmqhOutputFlowSetupCtx(const char *queue_str);
 void TmqhOutputFlowFreeCtx(void *ctx);
 #ifdef UNITTESTS
 void TmqhFlowRegisterTests(void);
 #endif
 
+static uint16_t g_sync_pkts = 8;
+
 void TmqhFlowRegister(void)
 {
     tmqh_table[TMQH_FLOW].name = "flow";
@@ -82,13 +85,19 @@ void TmqhFlowRegister(void)
     } else {
         tmqh_table[TMQH_FLOW].OutHandler = TmqhOutputFlowHash;
     }
+    intmax_t sync_pkts;
+    if (SCConfGetInt("autofp-batch-size", &sync_pkts) != 1)
+        sync_pkts = 8;
+    g_sync_pkts = (uint16_t)sync_pkts;
+    SCLogNotice("autofp-block-size: %u", g_sync_pkts);
+    tmqh_table[TMQH_FLOW].OutFlush = TmqhOutputFlush;
 }
 
 void TmqhFlowPrintAutofpHandler(void)
 {
-#define PRINT_IF_FUNC(f, msg)                       \
+#define PRINT_IF_FUNC(f, msg)                                                                      \
-    if (tmqh_table[TMQH_FLOW].OutHandler == (f))    \
+    if (tmqh_table[TMQH_FLOW].OutHandler == (f))                                                   \
-        SCLogConfig("AutoFP mode using \"%s\" flow load balancer", (msg))
+    SCLogConfig("AutoFP mode using \"%s\" flow load balancer", (msg))
 
     PRINT_IF_FUNC(TmqhOutputFlowHash, "Hash");
     PRINT_IF_FUNC(TmqhOutputFlowIPPair, "IPPair");
@@ -145,6 +154,7 @@ static int StoreQueueId(TmqhFlowCtx *ctx, char *name)
         memset(ctx->queues + (ctx->size - 1), 0, sizeof(TmqhFlowMode));
     }
     ctx->queues[ctx->size - 1].q = tmq->pq;
+    SCLogNotice("size %d, mem %u", ctx->size, (uint32_t)(ctx->size * sizeof(TmqhFlowMode)));
 
     return 0;
 }
@@ -169,6 +179,8 @@ void *TmqhOutputFlowSetupCtx(const char *queue_str)
     TmqhFlowCtx *ctx = SCCalloc(1, sizeof(TmqhFlowCtx));
     if (unlikely(ctx == NULL))
         return NULL;
+    SCLogNotice("ctx %p", ctx);
+    ctx->last_flow_qid = 0xffff;
 
     char *str = SCStrdup(queue_str);
     if (unlikely(str == NULL)) {
@@ -178,16 +190,16 @@ void *TmqhOutputFlowSetupCtx(const char *queue_str)
 
     /* parse the comma separated string */
     do {
-        char *comma = strchr(tstr,',');
+        char *comma = strchr(tstr, ',');
         if (comma != NULL) {
             *comma = '\0';
             char *qname = tstr;
-            int r = StoreQueueId(ctx,qname);
+            int r = StoreQueueId(ctx, qname);
             if (r < 0)
                 goto error;
         } else {
             char *qname = tstr;
-            int r = StoreQueueId(ctx,qname);
+            int r = StoreQueueId(ctx, qname);
             if (r < 0)
                 goto error;
         }
@@ -208,20 +220,61 @@ void TmqhOutputFlowFreeCtx(void *ctx)
 {
     TmqhFlowCtx *fctx = (TmqhFlowCtx *)ctx;
 
-    SCLogPerf("AutoFP - Total flow handler queues - %" PRIu16,
+    SCLogPerf("AutoFP - Total flow handler queues - %" PRIu16, fctx->size);
-              fctx->size);
     SCFree(fctx->queues);
     SCFree(fctx);
 }
 
+static void Flush(ThreadVars *tv, TmqhFlowCtx *ctx, const uint32_t new_qid)
+{
+    SCLogDebug("flushing %p with new_qid %x last_flow_hash_cnt %u last_flow_qid %x", ctx, new_qid,
+            ctx->last_flow_hash_cnt, ctx->last_flow_qid);
+    const uint32_t old_qid = ctx->last_flow_qid;
+    ctx->last_flow_qid = new_qid;
+    ctx->last_flow_hash_cnt = 1;
+    const uint32_t qid = old_qid;
+    if (qid < ctx->size) {
+        SCLogDebug("flushing %p qid %x", ctx, qid);
+
+        PacketQueue *q = ctx->queues[qid].q;
+        SCMutexLock(&q->mutex_q);
+        do {
+            Packet *p = PacketDequeueNoLock(&ctx->last_flow_hash_queue);
+            PacketEnqueue(q, p);
+        } while (ctx->last_flow_hash_queue.len > 0);
+        SCCondSignal(&q->cond_q);
+        SCMutexUnlock(&q->mutex_q);
+    }
+}
+
+static void TmqhOutputFlush(ThreadVars *tv)
+{
+    TmqhFlowCtx *ctx = (TmqhFlowCtx *)tv->outctx;
+
+    Flush(tv, ctx, 0xffff);
+}
+
 void TmqhOutputFlowHash(ThreadVars *tv, Packet *p)
 {
     uint32_t qid;
     TmqhFlowCtx *ctx = (TmqhFlowCtx *)tv->outctx;
 
     if (p->flags & PKT_WANTS_FLOW) {
-        uint32_t hash = p->flow_hash;
+        const uint32_t hash = p->flow_hash;
         qid = hash % ctx->size;
+
+        if (qid != ctx->last_flow_qid || ctx->last_flow_hash_cnt == g_sync_pkts) {
+            Flush(tv, ctx, qid);
+        } else {
+            ctx->last_flow_hash_cnt++;
+            if (ctx->last_flow_hash_cnt_max < ctx->last_flow_hash_cnt) {
+                ctx->last_flow_hash_cnt_max = ctx->last_flow_hash_cnt;
+                SCLogNotice("max same hash cnt %u", ctx->last_flow_hash_cnt_max);
+            }
+            PacketEnqueueNoLock(&ctx->last_flow_hash_queue, p);
+            return;
+        }
+
     } else {
         qid = ctx->last++;
 

src/tmqh-flow.h

@@ -1,4 +1,4 @@
-/* Copyright (C) 2007-2010 Open Information Security Foundation
+/* Copyright (C) 2007-2026 Open Information Security Foundation
  *
  * You can copy, redistribute or modify this Program under the terms of
  * the GNU General Public License version 2 as published by the Free
@@ -33,12 +33,18 @@ typedef struct TmqhFlowMode_ {
  *  \param queues array of queue id's this flow handler outputs to */
 typedef struct TmqhFlowCtx_ {
     uint16_t size;
+
+    /* last for round robin non-flow distributions */
     uint16_t last;
 
+    uint16_t last_flow_qid;
+    uint32_t last_flow_hash_cnt;
+    uint32_t last_flow_hash_cnt_max;
+    PacketQueueNoLock last_flow_hash_queue;
     TmqhFlowMode *queues;
 } TmqhFlowCtx;
 
-void TmqhFlowRegister (void);
+void TmqhFlowRegister(void);
 void TmqhFlowRegisterTests(void);
 
 void TmqhFlowPrintAutofpHandler(void);