From f5bf69f971cfc35b5faf51d007d6704cf3d42b7b Mon Sep 17 00:00:00 2001 From: Victor Julien Date: Thu, 25 Dec 2025 14:26:02 +0000 Subject: [PATCH] tmqh/flow: transfer more than one packet per fetch To avoid per packet locking overhead, transfer all available packets. --- src/runmode-pcap-file.c | 14 ++----- src/source-af-packet.c | 3 ++ src/source-nfq.c | 1 + src/source-pcap-file-helper.c | 1 + src/tm-queuehandlers.h | 1 + src/tm-threads.h | 13 +++++++ src/tmqh-flow.c | 73 ++++++++++++++++++++++++++++++----- src/tmqh-flow.h | 10 ++++- 8 files changed, 94 insertions(+), 22 deletions(-) diff --git a/src/runmode-pcap-file.c b/src/runmode-pcap-file.c index b5574d3004..93ef634835 100644 --- a/src/runmode-pcap-file.c +++ b/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 = - TmThreadCreatePacketHandler(tname, - "packetpool", "packetpool", - queues, "flow", - "pktacqloop"); + ThreadVars *tv_receivepcap = TmThreadCreatePacketHandler( + tname, "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 = - TmThreadCreatePacketHandler(tname, - qname, "flow", - "packetpool", "packetpool", - "varslot"); + ThreadVars *tv_detect_ncpu = TmThreadCreatePacketHandler( + tname, qname, "flow", "packetpool", "packetpool", "varslot"); if (tv_detect_ncpu == NULL) { FatalError("TmThreadsCreate failed"); } diff --git a/src/source-af-packet.c b/src/source-af-packet.c index ffb1d624e1..1d1e684d7d 100644 --- a/src/source-af-packet.c +++ b/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); } diff --git a/src/source-nfq.c b/src/source-nfq.c index 7f75cd171c..8e1cc09c38 100644 --- a/src/source-nfq.c +++ b/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); diff --git a/src/source-pcap-file-helper.c b/src/source-pcap-file-helper.c index e88db6d515..a1abed9de8 100644 --- a/src/source-pcap-file-helper.c +++ b/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); } diff --git a/src/tm-queuehandlers.h b/src/tm-queuehandlers.h index 83151bbd43..f5fbf84901 100644 --- a/src/tm-queuehandlers.h +++ b/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 diff --git a/src/tm-threads.h b/src/tm-threads.h index b92fc4b017..58792469db 100644 --- a/src/tm-threads.h +++ b/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); diff --git a/src/tmqh-flow.c b/src/tmqh-flow.c index a1da215a59..3bb99c2956 100644 --- a/src/tmqh-flow.c +++ b/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) \ - if (tmqh_table[TMQH_FLOW].OutHandler == (f)) \ - SCLogConfig("AutoFP mode using \"%s\" flow load balancer", (msg)) +#define PRINT_IF_FUNC(f, msg) \ + if (tmqh_table[TMQH_FLOW].OutHandler == (f)) \ + 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, - fctx->size); + SCLogPerf("AutoFP - Total flow handler queues - %" PRIu16, 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++; diff --git a/src/tmqh-flow.h b/src/tmqh-flow.h index e3513d0998..70502a697d 100644 --- a/src/tmqh-flow.h +++ b/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);