From a12b7fb24b637d430f173adf2caa8cb97eefdb11 Mon Sep 17 00:00:00 2001 From: Victor Julien Date: Wed, 18 Mar 2026 13:30:23 +0100 Subject: [PATCH] tmqh: return a queue from input queue handler --- src/threadvars.h | 2 +- src/tm-queuehandlers.h | 4 ++- src/tm-threads.c | 58 ++++++++++++++++++++++-------------------- src/tmqh-flow.c | 22 ++++++---------- src/tmqh-packetpool.c | 8 ++++-- src/tmqh-packetpool.h | 1 - src/tmqh-simple.c | 33 ++++++++++-------------- 7 files changed, 62 insertions(+), 66 deletions(-) diff --git a/src/threadvars.h b/src/threadvars.h index 09a41db711..d203d7031f 100644 --- a/src/threadvars.h +++ b/src/threadvars.h @@ -88,7 +88,7 @@ typedef struct ThreadVars_ { /** incoming queue and handler */ Tmq *inq; - struct Packet_ *(*TmqhInFn)(struct ThreadVars_ *); + struct PacketQueueNoLock_ (*TmqhInFn)(struct ThreadVars_ *); SC_ATOMIC_DECLARE(uint32_t, flags); diff --git a/src/tm-queuehandlers.h b/src/tm-queuehandlers.h index 5f2a2505ff..83151bbd43 100644 --- a/src/tm-queuehandlers.h +++ b/src/tm-queuehandlers.h @@ -24,6 +24,8 @@ #ifndef SURICATA_TM_QUEUEHANDLERS_H #define SURICATA_TM_QUEUEHANDLERS_H +#include "packet-queue.h" + enum { TMQH_NOT_SET, TMQH_SIMPLE, @@ -35,7 +37,7 @@ enum { typedef struct Tmqh_ { const char *name; - Packet *(*InHandler)(ThreadVars *); + PacketQueueNoLock (*InHandler)(ThreadVars *); void (*InShutdownHandler)(ThreadVars *); void (*OutHandler)(ThreadVars *, Packet *); void *(*OutHandlerCtxSetup)(const char *); diff --git a/src/tm-threads.c b/src/tm-threads.c index 539c484e12..3709ca8c82 100644 --- a/src/tm-threads.c +++ b/src/tm-threads.c @@ -489,39 +489,43 @@ static void *TmThreadsSlotVar(void *td) s = (TmSlot *)tv->tm_slots; while (run) { - /* input a packet */ - p = tv->TmqhInFn(tv); - - /* if we didn't get a packet see if we need to do some housekeeping */ - if (unlikely(p == NULL)) { - if (tv->flow_queue && SC_ATOMIC_GET(tv->flow_queue->non_empty)) { - p = PacketGetFromQueueOrAlloc(); - if (p != NULL) { - p->flags |= PKT_PSEUDO_STREAM_END; - PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT); + /* get packet(s) from the queue. Can get more than one so should loop + * over it. */ + PacketQueueNoLock q = tv->TmqhInFn(tv); + do { + p = PacketDequeueNoLock(&q); + + /* if we didn't get a packet see if we need to do some housekeeping */ + if (unlikely(p == NULL)) { + if (tv->flow_queue && SC_ATOMIC_GET(tv->flow_queue->non_empty)) { + p = PacketGetFromQueueOrAlloc(); + if (p != NULL) { + p->flags |= PKT_PSEUDO_STREAM_END; + PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT); + } } } - } - if (p != NULL) { - /* run the thread module(s) */ - r = TmThreadsSlotVarRun(tv, p, s); - if (r == TM_ECODE_FAILED) { - TmqhOutputPacketpool(tv, p); - TmThreadsSetFlag(tv, THV_FAILED); - break; - } + if (p != NULL) { + /* run the thread module(s) */ + r = TmThreadsSlotVarRun(tv, p, s); + if (r == TM_ECODE_FAILED) { + TmqhOutputPacketpool(tv, p); + TmThreadsSetFlag(tv, THV_FAILED); + break; + } - /* output the packet */ - tv->TmqhOutFn(tv, p); + /* output the packet */ + tv->TmqhOutFn(tv, p); - /* now handle the stream pq packets */ - TmThreadsHandleInjectedPackets(tv); - } + /* now handle the stream pq packets */ + TmThreadsHandleInjectedPackets(tv); + } - if (TmThreadsCheckFlag(tv, (THV_KILL | THV_REQ_FLOW_LOOP))) { - run = false; - } + if (TmThreadsCheckFlag(tv, (THV_KILL | THV_REQ_FLOW_LOOP))) { + run = false; + } + } while (q.len != 0); } if (!SCTmThreadsSlotPacketLoopFinish(tv)) { goto error; diff --git a/src/tmqh-flow.c b/src/tmqh-flow.c index 324c8b3c3b..a1da215a59 100644 --- a/src/tmqh-flow.c +++ b/src/tmqh-flow.c @@ -40,7 +40,7 @@ #include "conf.h" #include "util-unittest.h" -Packet *TmqhInputFlow(ThreadVars *t); +PacketQueueNoLock TmqhInputFlow(ThreadVars *t); void TmqhOutputFlowHash(ThreadVars *t, Packet *p); void TmqhOutputFlowIPPair(ThreadVars *t, Packet *p); static void TmqhOutputFlowFTPHash(ThreadVars *t, Packet *p); @@ -98,27 +98,21 @@ void TmqhFlowPrintAutofpHandler(void) } /* same as 'simple' */ -Packet *TmqhInputFlow(ThreadVars *tv) +PacketQueueNoLock TmqhInputFlow(ThreadVars *tv) { - PacketQueue *q = tv->inq->pq; - StatsSyncCountersIfSignalled(&tv->stats); + PacketQueue *q = tv->inq->pq; SCMutexLock(&q->mutex_q); if (q->len == 0) { /* if we have no packets in queue, wait... */ SCCondWait(&q->cond_q, &q->mutex_q); } - - if (q->len > 0) { - Packet *p = PacketDequeue(q); - SCMutexUnlock(&q->mutex_q); - return p; - } else { - /* return NULL if we have no pkt. Should only happen on signals. */ - SCMutexUnlock(&q->mutex_q); - return NULL; - } + PacketQueueNoLock pq = { .top = q->top, .bot = q->bot, q->len }; + q->bot = q->top = NULL; + q->len = 0; + SCMutexUnlock(&q->mutex_q); + return pq; } static int StoreQueueId(TmqhFlowCtx *ctx, char *name) diff --git a/src/tmqh-packetpool.c b/src/tmqh-packetpool.c index f3b387e12f..4113c20b4c 100644 --- a/src/tmqh-packetpool.c +++ b/src/tmqh-packetpool.c @@ -48,6 +48,8 @@ static inline PktPool *GetThreadPacketPool(void) return &thread_pkt_pool; } +static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *); + /** * \brief TmqhPacketpoolRegister * \initonly @@ -297,9 +299,11 @@ void PacketPoolDestroy(void) #endif /* DEBUG_VALIDATION */ } -Packet *TmqhInputPacketpool(ThreadVars *tv) +static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *tv) { - return PacketPoolGetPacket(); + Packet *p = PacketPoolGetPacket(); + PacketQueueNoLock pq = { .top = p, .bot = p, .len = (p != NULL) }; + return pq; } void TmqhOutputPacketpool(ThreadVars *t, Packet *p) diff --git a/src/tmqh-packetpool.h b/src/tmqh-packetpool.h index f2dedd93a6..c448bc98be 100644 --- a/src/tmqh-packetpool.h +++ b/src/tmqh-packetpool.h @@ -72,7 +72,6 @@ typedef struct PktPool_ { PktPoolLockedStack return_stack; } PktPool; -Packet *TmqhInputPacketpool(ThreadVars *); void TmqhOutputPacketpool(ThreadVars *, Packet *); void TmqhReleasePacketsToPacketPool(PacketQueue *); void TmqhPacketpoolRegister(void); diff --git a/src/tmqh-simple.c b/src/tmqh-simple.c index 553c1a43a3..36ec303176 100644 --- a/src/tmqh-simple.c +++ b/src/tmqh-simple.c @@ -32,9 +32,9 @@ #include "tm-queuehandlers.h" #include "tmqh-simple.h" -Packet *TmqhInputSimple(ThreadVars *t); -void TmqhOutputSimple(ThreadVars *t, Packet *p); -void TmqhInputSimpleShutdownHandler(ThreadVars *); +static PacketQueueNoLock TmqhInputSimple(ThreadVars *t); +static void TmqhOutputSimple(ThreadVars *t, Packet *p); +static void TmqhInputSimpleShutdownHandler(ThreadVars *); void TmqhSimpleRegister (void) { @@ -44,31 +44,24 @@ void TmqhSimpleRegister (void) tmqh_table[TMQH_SIMPLE].OutHandler = TmqhOutputSimple; } -Packet *TmqhInputSimple(ThreadVars *t) +PacketQueueNoLock TmqhInputSimple(ThreadVars *tv) { - PacketQueue *q = t->inq->pq; - - StatsSyncCountersIfSignalled(&t->stats); + StatsSyncCountersIfSignalled(&tv->stats); + PacketQueue *q = tv->inq->pq; SCMutexLock(&q->mutex_q); - if (q->len == 0) { /* if we have no packets in queue, wait... */ SCCondWait(&q->cond_q, &q->mutex_q); } - - if (q->len > 0) { - Packet *p = PacketDequeue(q); - SCMutexUnlock(&q->mutex_q); - return p; - } else { - /* return NULL if we have no pkt. Should only happen on signals. */ - SCMutexUnlock(&q->mutex_q); - return NULL; - } + PacketQueueNoLock pq = { .top = q->top, .bot = q->bot, q->len }; + q->bot = q->top = NULL; + q->len = 0; + SCMutexUnlock(&q->mutex_q); + return pq; } -void TmqhInputSimpleShutdownHandler(ThreadVars *tv) +static void TmqhInputSimpleShutdownHandler(ThreadVars *tv) { int i; @@ -83,7 +76,7 @@ void TmqhInputSimpleShutdownHandler(ThreadVars *tv) } } -void TmqhOutputSimple(ThreadVars *t, Packet *p) +static void TmqhOutputSimple(ThreadVars *t, Packet *p) { SCLogDebug("Packet %p, p->root %p, alloced %s", p, p->root, BOOL2STR(p->pool == NULL));