aiden
/
suricata
mirror of https://github.com/OISF/suricata
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge f5bf69f971 into 9a37bb659b

Browse Source
pull/15053/merge
Victor Julien 2 days ago committed by GitHub
parent 9a37bb659b f5bf69f971
commit 61f3c961c0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 181 additions and 112 deletions
Show Stats Download Patch File Download Diff File

14
src/runmode-pcap-file.c
Unescape Escape View File

@ -158,11 +158,8 @@ int RunModeFilePcapAutoFp(void)
snprintf(tname, sizeof(tname), "%s#01", thread_name_autofp); snprintf(tname, sizeof(tname), "%s#01", thread_name_autofp);
/* create the threads */ /* 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); SCFree(queues);
if (tv_receivepcap == NULL) { if (tv_receivepcap == NULL) {
@ -193,11 +190,8 @@ int RunModeFilePcapAutoFp(void)
SCLogDebug("tname %s, qname %s", tname, qname); SCLogDebug("tname %s, qname %s", tname, qname);
SCLogDebug("Assigning %s affinity to cpu %u", tname, cpu); 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) { if (tv_detect_ncpu == NULL) {
FatalError("TmThreadsCreate failed"); FatalError("TmThreadsCreate failed");
} }

3
src/source-af-packet.c
Unescape Escape View File

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

1
src/source-nfq.c
Unescape Escape View File

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

1
src/source-pcap-file-helper.c
Unescape Escape View File

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

4
src/threadvars.h
Unescape Escape View File

@ -87,7 +87,7 @@ typedef struct ThreadVars_ {
/** incoming queue and handler */ /** incoming queue and handler */
Tmq *inq; Tmq *inq;
struct Packet_ * (*tmqh_in)(struct ThreadVars_ *); struct PacketQueueNoLock_ (*TmqhInFn)(struct ThreadVars_ *);
SC_ATOMIC_DECLARE(uint32_t, flags); SC_ATOMIC_DECLARE(uint32_t, flags);
@ -102,7 +102,7 @@ typedef struct ThreadVars_ {
/** outgoing queue and handler */ /** outgoing queue and handler */
Tmq *outq; Tmq *outq;
void *outctx; void *outctx;
void (*tmqh_out)(struct ThreadVars_ *, struct Packet_ *); void (*TmqhOutFn)(struct ThreadVars_ *, struct Packet_ *);
/** Queue for decoders to temporarily store extra packets they /** Queue for decoders to temporarily store extra packets they
* generate. These packets are generated as part of the tunnel * generate. These packets are generated as part of the tunnel

7
src/tm-queuehandlers.h
Unescape Escape View File

@ -24,6 +24,8 @@
#ifndef SURICATA_TM_QUEUEHANDLERS_H #ifndef SURICATA_TM_QUEUEHANDLERS_H
#define SURICATA_TM_QUEUEHANDLERS_H #define SURICATA_TM_QUEUEHANDLERS_H
#include "packet-queue.h"
enum { enum {
TMQH_NOT_SET, TMQH_NOT_SET,
TMQH_SIMPLE, TMQH_SIMPLE,
@ -35,12 +37,15 @@ enum {
typedef struct Tmqh_ { typedef struct Tmqh_ {
const char *name; const char *name;
Packet *(*InHandler)(ThreadVars *); PacketQueueNoLock (*InHandler)(ThreadVars *);
void (*InShutdownHandler)(ThreadVars *); void (*InShutdownHandler)(ThreadVars *);
void (*OutHandler)(ThreadVars *, Packet *); void (*OutHandler)(ThreadVars *, Packet *);
void (*OutFlush)(ThreadVars *);
void *(*OutHandlerCtxSetup)(const char *); void *(*OutHandlerCtxSetup)(const char *);
void (*OutHandlerCtxFree)(void *); void (*OutHandlerCtxFree)(void *);
#ifdef UNITTESTS
void (*RegisterTests)(void); void (*RegisterTests)(void);
#endif
} Tmqh; } Tmqh;
extern Tmqh tmqh_table[TMQH_SIZE]; extern Tmqh tmqh_table[TMQH_SIZE];

82
src/tm-threads.c
Unescape Escape View File

@ -315,11 +315,11 @@ static void *TmThreadsSlotPktAcqLoop(void *td)
TmEcode r = TM_ECODE_OK; TmEcode r = TM_ECODE_OK;
/* check if we are setup properly */ /* 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," SCLogError("TmSlot or ThreadVars badly setup: s=%p,"
" PktAcqLoop=%p, tmqh_in=%p," " PktAcqLoop=%p, TmqhInFn=%p,"
" tmqh_out=%p", " TmqhOutFn=%p",
s, s ? s->PktAcqLoop : NULL, tv->tmqh_in, tv->tmqh_out); s, s ? s->PktAcqLoop : NULL, tv->TmqhInFn, tv->TmqhOutFn);
TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
pthread_exit(NULL); pthread_exit(NULL);
return NULL; return NULL;
@ -394,10 +394,10 @@ static void *TmThreadsLib(void *td)
TmSlot *s = tv->tm_slots; TmSlot *s = tv->tm_slots;
/* check if we are setup properly */ /* 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) {
SCLogError("TmSlot or ThreadVars badly setup: s=%p, tmqh_in=%p," SCLogError("TmSlot or ThreadVars badly setup: s=%p, TmqhInFn=%p,"
" tmqh_out=%p", " TmqhOutFn=%p",
s, tv->tmqh_in, tv->tmqh_out); s, tv->TmqhInFn, tv->TmqhOutFn);
TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE); TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
return NULL; return NULL;
} }
@ -436,7 +436,7 @@ static void *TmThreadsSlotVar(void *td)
SCDropCaps(tv); SCDropCaps(tv);
/* check if we are setup properly */ /* 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); TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
pthread_exit(NULL); pthread_exit(NULL);
return NULL; return NULL;
@ -497,39 +497,43 @@ static void *TmThreadsSlotVar(void *td)
s = (TmSlot *)tv->tm_slots; s = (TmSlot *)tv->tm_slots;
while (run) { while (run) {
/* input a packet */ /* get packet(s) from the queue. Can get more than one so should loop
p = tv->tmqh_in(tv); * over it. */
PacketQueueNoLock q = tv->TmqhInFn(tv);
/* if we didn't get a packet see if we need to do some housekeeping */ do {
if (unlikely(p == NULL)) { p = PacketDequeueNoLock(&q);
if (tv->flow_queue && SC_ATOMIC_GET(tv->flow_queue->non_empty)) {
p = PacketGetFromQueueOrAlloc(); /* if we didn't get a packet see if we need to do some housekeeping */
if (p != NULL) { if (unlikely(p == NULL)) {
p->flags |= PKT_PSEUDO_STREAM_END; if (tv->flow_queue && SC_ATOMIC_GET(tv->flow_queue->non_empty)) {
PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT); p = PacketGetFromQueueOrAlloc();
if (p != NULL) {
p->flags |= PKT_PSEUDO_STREAM_END;
PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT);
}
} }
} }
}
if (p != NULL) { if (p != NULL) {
/* run the thread module(s) */ /* run the thread module(s) */
r = TmThreadsSlotVarRun(tv, p, s); r = TmThreadsSlotVarRun(tv, p, s);
if (r == TM_ECODE_FAILED) { if (r == TM_ECODE_FAILED) {
TmqhOutputPacketpool(tv, p); TmqhOutputPacketpool(tv, p);
TmThreadsSetFlag(tv, THV_FAILED); TmThreadsSetFlag(tv, THV_FAILED);
break; break;
} }
/* output the packet */ /* output the packet */
tv->tmqh_out(tv, p); tv->TmqhOutFn(tv, p);
/* now handle the stream pq packets */ /* now handle the stream pq packets */
TmThreadsHandleInjectedPackets(tv); TmThreadsHandleInjectedPackets(tv);
} }
if (TmThreadsCheckFlag(tv, (THV_KILL | THV_REQ_FLOW_LOOP))) { if (TmThreadsCheckFlag(tv, (THV_KILL | THV_REQ_FLOW_LOOP))) {
run = false; run = false;
} }
} while (q.len != 0);
} }
if (!SCTmThreadsSlotPacketLoopFinish(tv)) { if (!SCTmThreadsSlotPacketLoopFinish(tv)) {
goto error; goto error;
@ -992,9 +996,9 @@ ThreadVars *TmThreadCreate(const char *name, const char *inq_name, const char *i
if (tmqh == NULL) if (tmqh == NULL)
goto error; goto error;
tv->tmqh_in = tmqh->InHandler; tv->TmqhInFn = tmqh->InHandler;
tv->inq_id = (uint8_t)id; tv->inq_id = (uint8_t)id;
SCLogDebug("tv->tmqh_in %p", tv->tmqh_in); SCLogDebug("tv->TmqhInFn %p", tv->TmqhInFn);
} }
/* set the outgoing queue */ /* set the outgoing queue */
@ -1010,7 +1014,7 @@ ThreadVars *TmThreadCreate(const char *name, const char *inq_name, const char *i
if (tmqh == NULL) if (tmqh == NULL)
goto error; goto error;
tv->tmqh_out = tmqh->OutHandler; tv->TmqhOutFn = tmqh->OutHandler;
tv->outq_id = (uint8_t)id; tv->outq_id = (uint8_t)id;
if (outq_name != NULL && strcmp(outq_name, "packetpool") != 0) { if (outq_name != NULL && strcmp(outq_name, "packetpool") != 0) {

21
src/tm-threads.h
Unescape Escape View File

@ -27,6 +27,7 @@
#include "tmqh-packetpool.h" #include "tmqh-packetpool.h"
#include "tm-threads-common.h" #include "tm-threads-common.h"
#include "tm-queuehandlers.h"
#include "tm-modules.h" #include "tm-modules.h"
#include "flow.h" // for the FlowQueue #include "flow.h" // for the FlowQueue
@ -180,7 +181,7 @@ static inline bool TmThreadsHandleInjectedPackets(ThreadVars *tv)
TmThreadsSlotProcessPktFail(tv, extra_p); TmThreadsSlotProcessPktFail(tv, extra_p);
break; break;
} }
tv->tmqh_out(tv, extra_p); tv->TmqhOutFn(tv, extra_p);
} }
return true; return true;
} else { } else {
@ -194,7 +195,7 @@ static inline bool TmThreadsHandleInjectedPackets(ThreadVars *tv)
static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet *p) static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet *p)
{ {
if (s == NULL) { if (s == NULL) {
tv->tmqh_out(tv, p); tv->TmqhOutFn(tv, p);
return TM_ECODE_OK; return TM_ECODE_OK;
} }
@ -204,7 +205,7 @@ static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet
return TM_ECODE_FAILED; return TM_ECODE_FAILED;
} }
tv->tmqh_out(tv, p); tv->TmqhOutFn(tv, p);
TmThreadsHandleInjectedPackets(tv); 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 /* packet could have been passed to us that we won't use
* return it to the pool. */ * return it to the pool. */
if (p != NULL) if (p != NULL)
tv->tmqh_out(tv, p); tv->TmqhOutFn(tv, p);
} }
static inline void TmThreadsCaptureBreakLoop(ThreadVars *tv) 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 TmThreadsSealThreads(void);
void TmThreadsUnsealThreads(void); void TmThreadsUnsealThreads(void);
void TmThreadsListThreads(void); void TmThreadsListThreads(void);

108
src/tmqh-flow.c
Unescape Escape View File

@ -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 * You can copy, redistribute or modify this Program under the terms of
* the GNU General Public License version 2 as published by the Free * the GNU General Public License version 2 as published by the Free
@ -39,13 +39,18 @@
#include "conf.h" #include "conf.h"
Packet *TmqhInputFlow(ThreadVars *t); PacketQueueNoLock TmqhInputFlow(ThreadVars *t);
void TmqhOutputFlowHash(ThreadVars *t, Packet *p); void TmqhOutputFlowHash(ThreadVars *t, Packet *p);
void TmqhOutputFlowIPPair(ThreadVars *t, Packet *p); void TmqhOutputFlowIPPair(ThreadVars *t, Packet *p);
static void TmqhOutputFlowFTPHash(ThreadVars *t, Packet *p); static void TmqhOutputFlowFTPHash(ThreadVars *t, Packet *p);
static void TmqhOutputFlush(ThreadVars *tv);
void *TmqhOutputFlowSetupCtx(const char *queue_str); void *TmqhOutputFlowSetupCtx(const char *queue_str);
void TmqhOutputFlowFreeCtx(void *ctx); void TmqhOutputFlowFreeCtx(void *ctx);
#ifdef UNITTESTS
void TmqhFlowRegisterTests(void); void TmqhFlowRegisterTests(void);
#endif
static uint16_t g_sync_pkts = 8;
void TmqhFlowRegister(void) void TmqhFlowRegister(void)
{ {
@ -53,8 +58,9 @@ void TmqhFlowRegister(void)
tmqh_table[TMQH_FLOW].InHandler = TmqhInputFlow; tmqh_table[TMQH_FLOW].InHandler = TmqhInputFlow;
tmqh_table[TMQH_FLOW].OutHandlerCtxSetup = TmqhOutputFlowSetupCtx; tmqh_table[TMQH_FLOW].OutHandlerCtxSetup = TmqhOutputFlowSetupCtx;
tmqh_table[TMQH_FLOW].OutHandlerCtxFree = TmqhOutputFlowFreeCtx; tmqh_table[TMQH_FLOW].OutHandlerCtxFree = TmqhOutputFlowFreeCtx;
#ifdef UNITTESTS
tmqh_table[TMQH_FLOW].RegisterTests = TmqhFlowRegisterTests; tmqh_table[TMQH_FLOW].RegisterTests = TmqhFlowRegisterTests;
#endif
const char *scheduler = NULL; const char *scheduler = NULL;
if (SCConfGetNonNull("autofp-scheduler", &scheduler) == 1) { if (SCConfGetNonNull("autofp-scheduler", &scheduler) == 1) {
if (strcasecmp(scheduler, "round-robin") == 0) { if (strcasecmp(scheduler, "round-robin") == 0) {
@ -78,13 +84,19 @@ void TmqhFlowRegister(void)
} else { } else {
tmqh_table[TMQH_FLOW].OutHandler = TmqhOutputFlowHash; 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) 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(TmqhOutputFlowHash, "Hash");
PRINT_IF_FUNC(TmqhOutputFlowIPPair, "IPPair"); PRINT_IF_FUNC(TmqhOutputFlowIPPair, "IPPair");
@ -94,27 +106,21 @@ void TmqhFlowPrintAutofpHandler(void)
} }
/* same as 'simple' */ /* same as 'simple' */
Packet *TmqhInputFlow(ThreadVars *tv) PacketQueueNoLock TmqhInputFlow(ThreadVars *tv)
{ {
PacketQueue *q = tv->inq->pq;
StatsSyncCountersIfSignalled(&tv->stats); StatsSyncCountersIfSignalled(&tv->stats);
PacketQueue *q = tv->inq->pq;
SCMutexLock(&q->mutex_q); SCMutexLock(&q->mutex_q);
if (q->len == 0) { if (q->len == 0) {
/* if we have no packets in queue, wait... */ /* if we have no packets in queue, wait... */
SCCondWait(&q->cond_q, &q->mutex_q); SCCondWait(&q->cond_q, &q->mutex_q);
} }
PacketQueueNoLock pq = { .top = q->top, .bot = q->bot, q->len };
if (q->len > 0) { q->bot = q->top = NULL;
Packet *p = PacketDequeue(q); q->len = 0;
SCMutexUnlock(&q->mutex_q); SCMutexUnlock(&q->mutex_q);
return p; return pq;
} else {
/* return NULL if we have no pkt. Should only happen on signals. */
SCMutexUnlock(&q->mutex_q);
return NULL;
}
} }
static int StoreQueueId(TmqhFlowCtx *ctx, char *name) 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)); memset(ctx->queues + (ctx->size - 1), 0, sizeof(TmqhFlowMode));
} }
ctx->queues[ctx->size - 1].q = tmq->pq; ctx->queues[ctx->size - 1].q = tmq->pq;
SCLogNotice("size %d, mem %u", ctx->size, (uint32_t)(ctx->size * sizeof(TmqhFlowMode)));
return 0; return 0;
} }
@ -171,6 +178,8 @@ void *TmqhOutputFlowSetupCtx(const char *queue_str)
TmqhFlowCtx *ctx = SCCalloc(1, sizeof(TmqhFlowCtx)); TmqhFlowCtx *ctx = SCCalloc(1, sizeof(TmqhFlowCtx));
if (unlikely(ctx == NULL)) if (unlikely(ctx == NULL))
return NULL; return NULL;
SCLogNotice("ctx %p", ctx);
ctx->last_flow_qid = 0xffff;
char *str = SCStrdup(queue_str); char *str = SCStrdup(queue_str);
if (unlikely(str == NULL)) { if (unlikely(str == NULL)) {
@ -180,16 +189,16 @@ void *TmqhOutputFlowSetupCtx(const char *queue_str)
/* parse the comma separated string */ /* parse the comma separated string */
do { do {
char *comma = strchr(tstr,','); char *comma = strchr(tstr, ',');
if (comma != NULL) { if (comma != NULL) {
*comma = '\0'; *comma = '\0';
char *qname = tstr; char *qname = tstr;
int r = StoreQueueId(ctx,qname); int r = StoreQueueId(ctx, qname);
if (r < 0) if (r < 0)
goto error; goto error;
} else { } else {
char *qname = tstr; char *qname = tstr;
int r = StoreQueueId(ctx,qname); int r = StoreQueueId(ctx, qname);
if (r < 0) if (r < 0)
goto error; goto error;
} }
@ -210,20 +219,61 @@ void TmqhOutputFlowFreeCtx(void *ctx)
{ {
TmqhFlowCtx *fctx = (TmqhFlowCtx *)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->queues);
SCFree(fctx); 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) void TmqhOutputFlowHash(ThreadVars *tv, Packet *p)
{ {
uint32_t qid; uint32_t qid;
TmqhFlowCtx *ctx = (TmqhFlowCtx *)tv->outctx; TmqhFlowCtx *ctx = (TmqhFlowCtx *)tv->outctx;
if (p->flags & PKT_WANTS_FLOW) { if (p->flags & PKT_WANTS_FLOW) {
uint32_t hash = p->flow_hash; const uint32_t hash = p->flow_hash;
qid = hash % ctx->size; 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 { } else {
qid = ctx->last++; qid = ctx->last++;
@ -390,16 +440,12 @@ static int TmqhOutputFlowSetupCtxTest03(void)
PASS; PASS;
} }
#endif /* UNITTESTS */
void TmqhFlowRegisterTests(void) void TmqhFlowRegisterTests(void)
{ {
#ifdef UNITTESTS
UtRegisterTest("TmqhOutputFlowSetupCtxTest01", UtRegisterTest("TmqhOutputFlowSetupCtxTest01",
TmqhOutputFlowSetupCtxTest01); TmqhOutputFlowSetupCtxTest01);
UtRegisterTest("TmqhOutputFlowSetupCtxTest02", UtRegisterTest("TmqhOutputFlowSetupCtxTest02",
TmqhOutputFlowSetupCtxTest02); TmqhOutputFlowSetupCtxTest02);
UtRegisterTest("TmqhOutputFlowSetupCtxTest03", UtRegisterTest("TmqhOutputFlowSetupCtxTest03", TmqhOutputFlowSetupCtxTest03);
TmqhOutputFlowSetupCtxTest03);
#endif
} }
#endif /* UNITTESTS */

10
src/tmqh-flow.h
Unescape Escape View File

@ -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 * You can copy, redistribute or modify this Program under the terms of
* the GNU General Public License version 2 as published by the Free * 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 */ * \param queues array of queue id's this flow handler outputs to */
typedef struct TmqhFlowCtx_ { typedef struct TmqhFlowCtx_ {
uint16_t size; uint16_t size;
/* last for round robin non-flow distributions */
uint16_t last; 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; TmqhFlowMode *queues;
} TmqhFlowCtx; } TmqhFlowCtx;
void TmqhFlowRegister (void); void TmqhFlowRegister(void);
void TmqhFlowRegisterTests(void); void TmqhFlowRegisterTests(void);
void TmqhFlowPrintAutofpHandler(void); void TmqhFlowPrintAutofpHandler(void);

8
src/tmqh-packetpool.c
Unescape Escape View File

@ -48,6 +48,8 @@ static inline PktPool *GetThreadPacketPool(void)
return &thread_pkt_pool; return &thread_pkt_pool;
} }
static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *);
/** /**
* \brief TmqhPacketpoolRegister * \brief TmqhPacketpoolRegister
* \initonly * \initonly
@ -297,9 +299,11 @@ void PacketPoolDestroy(void)
#endif /* DEBUG_VALIDATION */ #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) void TmqhOutputPacketpool(ThreadVars *t, Packet *p)

1
src/tmqh-packetpool.h
Unescape Escape View File

@ -72,7 +72,6 @@ typedef struct PktPool_ {
PktPoolLockedStack return_stack; PktPoolLockedStack return_stack;
} PktPool; } PktPool;
Packet *TmqhInputPacketpool(ThreadVars *);
void TmqhOutputPacketpool(ThreadVars *, Packet *); void TmqhOutputPacketpool(ThreadVars *, Packet *);
void TmqhReleasePacketsToPacketPool(PacketQueue *); void TmqhReleasePacketsToPacketPool(PacketQueue *);
void TmqhPacketpoolRegister(void); void TmqhPacketpoolRegister(void);

33
src/tmqh-simple.c
Unescape Escape View File

@ -32,9 +32,9 @@
#include "tm-queuehandlers.h" #include "tm-queuehandlers.h"
#include "tmqh-simple.h" #include "tmqh-simple.h"
Packet *TmqhInputSimple(ThreadVars *t); static PacketQueueNoLock TmqhInputSimple(ThreadVars *t);
void TmqhOutputSimple(ThreadVars *t, Packet *p); static void TmqhOutputSimple(ThreadVars *t, Packet *p);
void TmqhInputSimpleShutdownHandler(ThreadVars *); static void TmqhInputSimpleShutdownHandler(ThreadVars *);
void TmqhSimpleRegister (void) void TmqhSimpleRegister (void)
{ {
@ -44,31 +44,24 @@ void TmqhSimpleRegister (void)
tmqh_table[TMQH_SIMPLE].OutHandler = TmqhOutputSimple; tmqh_table[TMQH_SIMPLE].OutHandler = TmqhOutputSimple;
} }
Packet *TmqhInputSimple(ThreadVars *t) PacketQueueNoLock TmqhInputSimple(ThreadVars *tv)
{ {
PacketQueue *q = t->inq->pq; StatsSyncCountersIfSignalled(&tv->stats);
StatsSyncCountersIfSignalled(&t->stats);
PacketQueue *q = tv->inq->pq;
SCMutexLock(&q->mutex_q); SCMutexLock(&q->mutex_q);
if (q->len == 0) { if (q->len == 0) {
/* if we have no packets in queue, wait... */ /* if we have no packets in queue, wait... */
SCCondWait(&q->cond_q, &q->mutex_q); SCCondWait(&q->cond_q, &q->mutex_q);
} }
PacketQueueNoLock pq = { .top = q->top, .bot = q->bot, q->len };
if (q->len > 0) { q->bot = q->top = NULL;
Packet *p = PacketDequeue(q); q->len = 0;
SCMutexUnlock(&q->mutex_q); SCMutexUnlock(&q->mutex_q);
return p; return pq;
} else {
/* return NULL if we have no pkt. Should only happen on signals. */
SCMutexUnlock(&q->mutex_q);
return NULL;
}
} }
void TmqhInputSimpleShutdownHandler(ThreadVars *tv) static void TmqhInputSimpleShutdownHandler(ThreadVars *tv)
{ {
int i; 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)); SCLogDebug("Packet %p, p->root %p, alloced %s", p, p->root, BOOL2STR(p->pool == NULL));

Write Preview
Loading…
Cancel
Save