diff --git a/src/runmode-pcap-file.c b/src/runmode-pcap-file.c index 751f9163f1..af68330f17 100644 --- a/src/runmode-pcap-file.c +++ b/src/runmode-pcap-file.c @@ -158,11 +158,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) { @@ -193,11 +190,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 eb87c750e8..2385189d5f 100644 --- a/src/source-af-packet.c +++ b/src/source-af-packet.c @@ -951,6 +951,7 @@ next_frame: break; } } + TmThreadFlushOutQueue(ptv->tv); if (emergency_flush) { AFPDumpCounters(ptv); } @@ -1091,9 +1092,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 c5dbe86291..c50878a901 100644 --- a/src/source-nfq.c +++ b/src/source-nfq.c @@ -1039,6 +1039,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 50bbf7cc00..bcca1c35f7 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/threadvars.h b/src/threadvars.h index 2322d39f71..9018df19d1 100644 --- a/src/threadvars.h +++ b/src/threadvars.h @@ -87,7 +87,7 @@ typedef struct ThreadVars_ { /** incoming queue and handler */ Tmq *inq; - struct Packet_ * (*tmqh_in)(struct ThreadVars_ *); + struct PacketQueueNoLock_ (*TmqhInFn)(struct ThreadVars_ *); SC_ATOMIC_DECLARE(uint32_t, flags); @@ -102,7 +102,7 @@ typedef struct ThreadVars_ { /** outgoing queue and handler */ Tmq *outq; void *outctx; - void (*tmqh_out)(struct ThreadVars_ *, struct Packet_ *); + void (*TmqhOutFn)(struct ThreadVars_ *, struct Packet_ *); /** Queue for decoders to temporarily store extra packets they * generate. These packets are generated as part of the tunnel diff --git a/src/tm-queuehandlers.h b/src/tm-queuehandlers.h index 6820d8d666..f5fbf84901 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,12 +37,15 @@ enum { typedef struct Tmqh_ { const char *name; - Packet *(*InHandler)(ThreadVars *); + PacketQueueNoLock (*InHandler)(ThreadVars *); void (*InShutdownHandler)(ThreadVars *); void (*OutHandler)(ThreadVars *, Packet *); + void (*OutFlush)(ThreadVars *); void *(*OutHandlerCtxSetup)(const char *); void (*OutHandlerCtxFree)(void *); +#ifdef UNITTESTS void (*RegisterTests)(void); +#endif } Tmqh; extern Tmqh tmqh_table[TMQH_SIZE]; diff --git a/src/tm-threads.c b/src/tm-threads.c index 8897d07601..3de8c0cb32 100644 --- a/src/tm-threads.c +++ b/src/tm-threads.c @@ -315,11 +315,11 @@ static void *TmThreadsSlotPktAcqLoop(void *td) TmEcode r = TM_ECODE_OK; /* check if we are setup properly */ - if (s == NULL || s->PktAcqLoop == NULL || tv->tmqh_in == NULL || tv->tmqh_out == NULL) { + if (s == NULL || s->PktAcqLoop == NULL || tv->TmqhInFn == NULL || tv->TmqhOutFn == NULL) { SCLogError("TmSlot or ThreadVars badly setup: s=%p," - " PktAcqLoop=%p, tmqh_in=%p," - " tmqh_out=%p", - s, s ? s->PktAcqLoop : NULL, tv->tmqh_in, tv->tmqh_out); + " PktAcqLoop=%p, TmqhInFn=%p," + " TmqhOutFn=%p", + s, s ? s->PktAcqLoop : NULL, tv->TmqhInFn, tv->TmqhOutFn); TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); pthread_exit(NULL); return NULL; @@ -394,10 +394,10 @@ static void *TmThreadsLib(void *td) TmSlot *s = tv->tm_slots; /* check if we are setup properly */ - if (s == NULL || tv->tmqh_in == NULL || tv->tmqh_out == NULL) { - SCLogError("TmSlot or ThreadVars badly setup: s=%p, tmqh_in=%p," - " tmqh_out=%p", - s, tv->tmqh_in, tv->tmqh_out); + if (s == NULL || tv->TmqhInFn == NULL || tv->TmqhOutFn == NULL) { + SCLogError("TmSlot or ThreadVars badly setup: s=%p, TmqhInFn=%p," + " TmqhOutFn=%p", + s, tv->TmqhInFn, tv->TmqhOutFn); TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); return NULL; } @@ -436,7 +436,7 @@ static void *TmThreadsSlotVar(void *td) SCDropCaps(tv); /* check if we are setup properly */ - if (s == NULL || tv->tmqh_in == NULL || tv->tmqh_out == NULL) { + if (s == NULL || tv->TmqhInFn == NULL || tv->TmqhOutFn == NULL) { TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); pthread_exit(NULL); return NULL; @@ -497,39 +497,43 @@ static void *TmThreadsSlotVar(void *td) s = (TmSlot *)tv->tm_slots; while (run) { - /* input a packet */ - p = tv->tmqh_in(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->tmqh_out(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; @@ -992,9 +996,9 @@ ThreadVars *TmThreadCreate(const char *name, const char *inq_name, const char *i if (tmqh == NULL) goto error; - tv->tmqh_in = tmqh->InHandler; + tv->TmqhInFn = tmqh->InHandler; tv->inq_id = (uint8_t)id; - SCLogDebug("tv->tmqh_in %p", tv->tmqh_in); + SCLogDebug("tv->TmqhInFn %p", tv->TmqhInFn); } /* set the outgoing queue */ @@ -1010,7 +1014,7 @@ ThreadVars *TmThreadCreate(const char *name, const char *inq_name, const char *i if (tmqh == NULL) goto error; - tv->tmqh_out = tmqh->OutHandler; + tv->TmqhOutFn = tmqh->OutHandler; tv->outq_id = (uint8_t)id; if (outq_name != NULL && strcmp(outq_name, "packetpool") != 0) { diff --git a/src/tm-threads.h b/src/tm-threads.h index 161548e82a..d38bec8689 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 @@ -180,7 +181,7 @@ static inline bool TmThreadsHandleInjectedPackets(ThreadVars *tv) TmThreadsSlotProcessPktFail(tv, extra_p); break; } - tv->tmqh_out(tv, extra_p); + tv->TmqhOutFn(tv, extra_p); } return true; } else { @@ -194,7 +195,7 @@ static inline bool TmThreadsHandleInjectedPackets(ThreadVars *tv) static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet *p) { if (s == NULL) { - tv->tmqh_out(tv, p); + tv->TmqhOutFn(tv, p); return TM_ECODE_OK; } @@ -204,7 +205,7 @@ static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet return TM_ECODE_FAILED; } - tv->tmqh_out(tv, p); + tv->TmqhOutFn(tv, p); TmThreadsHandleInjectedPackets(tv); @@ -257,7 +258,7 @@ static inline void TmThreadsCaptureHandleTimeout(ThreadVars *tv, Packet *p) /* packet could have been passed to us that we won't use * return it to the pool. */ if (p != NULL) - tv->tmqh_out(tv, p); + tv->TmqhOutFn(tv, p); } static inline void TmThreadsCaptureBreakLoop(ThreadVars *tv) @@ -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 e6551219a1..6f2e052150 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 @@ -39,13 +39,18 @@ #include "conf.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); +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) { @@ -53,8 +58,9 @@ void TmqhFlowRegister(void) tmqh_table[TMQH_FLOW].InHandler = TmqhInputFlow; tmqh_table[TMQH_FLOW].OutHandlerCtxSetup = TmqhOutputFlowSetupCtx; tmqh_table[TMQH_FLOW].OutHandlerCtxFree = TmqhOutputFlowFreeCtx; +#ifdef UNITTESTS tmqh_table[TMQH_FLOW].RegisterTests = TmqhFlowRegisterTests; - +#endif const char *scheduler = NULL; if (SCConfGetNonNull("autofp-scheduler", &scheduler) == 1) { if (strcasecmp(scheduler, "round-robin") == 0) { @@ -78,13 +84,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"); @@ -94,27 +106,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) @@ -147,6 +153,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; } @@ -171,6 +178,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)) { @@ -180,16 +189,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; } @@ -210,20 +219,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++; @@ -390,16 +440,12 @@ static int TmqhOutputFlowSetupCtxTest03(void) PASS; } -#endif /* UNITTESTS */ - void TmqhFlowRegisterTests(void) { -#ifdef UNITTESTS UtRegisterTest("TmqhOutputFlowSetupCtxTest01", TmqhOutputFlowSetupCtxTest01); UtRegisterTest("TmqhOutputFlowSetupCtxTest02", TmqhOutputFlowSetupCtxTest02); - UtRegisterTest("TmqhOutputFlowSetupCtxTest03", - TmqhOutputFlowSetupCtxTest03); -#endif + UtRegisterTest("TmqhOutputFlowSetupCtxTest03", TmqhOutputFlowSetupCtxTest03); } +#endif /* UNITTESTS */ 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); 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));