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

Merge thread kill functions. Merge slot's tm_id with the one used by packet profiling. Remove some junk unused code from ms sync pts. Timeout setup cleanup as well. packet q dbg_maxlen now u32 var.

Browse Source
remotes/origin/master-1.1.x
Anoop Saldanha 15 years ago committed by Victor Julien
parent e335bdbfbc
commit 54f6e4ff4d
6 changed files with 84 additions and 170 deletions
Show Stats Download Patch File Download Diff File

58
src/decode.c
Unescape Escape View File

@ -53,38 +53,6 @@ void DecodeTunnel(ThreadVars *tv, DecodeThreadVars *dtv, Packet *p,
}
}
/**
* \brief Get a packet. We try to get a packet from the packetpool first, but
* if that is empty we alloc a packet that is free'd again after
* processing.
*
* \retval p packet, NULL on error
*/
Packet *PacketGetFromQueueOrAlloc(void) {
Packet *p = NULL;
/* try the pool first */
if (PacketPoolSize() > 0) {
p = PacketPoolGetPacket();
}
if (p == NULL) {
/* non fatal, we're just not processing a packet then */
p = SCMalloc(SIZE_OF_PACKET);
if (p == NULL) {
return NULL;
}
PACKET_INITIALIZE(p);
p->flags |= PKT_ALLOC;
SCLogDebug("allocated a new packet...");
}
PACKET_PROFILING_START(p);
return p;
}
/**
* \brief Get a malloced packet.
*
@ -106,6 +74,32 @@ Packet *PacketGetFromAlloc(void)
return p;
}
/**
* \brief Get a packet. We try to get a packet from the packetpool first, but
* if that is empty we alloc a packet that is free'd again after
* processing.
*
* \retval p packet, NULL on error
*/
Packet *PacketGetFromQueueOrAlloc(void)
{
Packet *p = NULL;
/* try the pool first */
if (PacketPoolSize() > 0) {
p = PacketPoolGetPacket();
}
if (p == NULL) {
/* non fatal, we're just not processing a packet then */
p = PacketGetFromAlloc();
} else {
PACKET_PROFILING_START(p);
}
return p;
}
/**
* \brief Setup a pseudo packet (tunnel or reassembled frags)
*

2
src/decode.h
Unescape Escape View File

@ -470,7 +470,7 @@ typedef struct PacketQueue_ {
Packet *bot;
uint32_t len;
#ifdef DBG_PERF
uint16_t dbg_maxlen;
uint32_t dbg_maxlen;
#endif /* DBG_PERF */
SCMutex mutex_q;
SCCondT cond_q;

55
src/flow.c
Unescape Escape View File

@ -1533,31 +1533,8 @@ void FlowForceReassembly(void)
return;
}
/** \brief Thread that manages the various queue's and removes timed out flows.
* \param td ThreadVars casted to void ptr
*
* IDEAS/TODO
* Create a 'emergency mode' in which flow handling threads can indicate
* we are/seem to be under attack..... maybe this thread should check
* key indicators for that like:
* - number of flows created in the last x time
* - avg number of pkts per flow (how?)
* - avg flow age
*
* Keep an eye on the spare list, alloc flows if needed...
*/
void *FlowManagerThread(void *td)
void FlowForceReassemblySetup(void)
{
ThreadVars *th_v = (ThreadVars *)td;
struct timeval ts;
struct timeval tsdiff;
uint32_t established_cnt = 0, new_cnt = 0, closing_cnt = 0, nowcnt;
uint32_t sleeping = 0;
uint8_t emerg = FALSE;
uint32_t last_sec = 0;
memset(&ts, 0, sizeof(ts));
/* get StreamTCP TM's slot and TV containing this slot */
stream_pseudo_pkt_stream_tm_slot = TmSlotGetSlotForTM(TMM_STREAMTCP);
if (stream_pseudo_pkt_stream_tm_slot == NULL) {
@ -1614,6 +1591,36 @@ void *FlowManagerThread(void *td)
exit(EXIT_FAILURE);
}
return;
}
/** \brief Thread that manages the various queue's and removes timed out flows.
* \param td ThreadVars casted to void ptr
*
* IDEAS/TODO
* Create a 'emergency mode' in which flow handling threads can indicate
* we are/seem to be under attack..... maybe this thread should check
* key indicators for that like:
* - number of flows created in the last x time
* - avg number of pkts per flow (how?)
* - avg flow age
*
* Keep an eye on the spare list, alloc flows if needed...
*/
void *FlowManagerThread(void *td)
{
ThreadVars *th_v = (ThreadVars *)td;
struct timeval ts;
struct timeval tsdiff;
uint32_t established_cnt = 0, new_cnt = 0, closing_cnt = 0, nowcnt;
uint32_t sleeping = 0;
uint8_t emerg = FALSE;
uint32_t last_sec = 0;
memset(&ts, 0, sizeof(ts));
FlowForceReassemblySetup();
/* set the thread name */
SCSetThreadName(th_v->name);
SCLogDebug("%s started...", th_v->name);

21
src/threadvars.h
Unescape Escape View File

@ -51,25 +51,6 @@ struct TmSlot_;
/** Maximum no of times a thread can be restarted */
#define THV_MAX_RESTARTS 50
/**
* \brief Feature to enable ThreadVars implement a master-slave
* synchronization feature.
*/
typedef struct ThreadVarsMSSyncPt_ {
const char *name;
SCMutex m;
SCCondT cond;
int slave_hit;
int master_go;
/* indicates whether this syn point has been disabled or not. If disabled,
* the slave won't be able to use it anymore */
int disabled;
struct ThreadVarsMSSyncPt_ *next;
} ThreadVarsMSSyncPt;
/** \brief Per thread variable structure */
typedef struct ThreadVars_ {
pthread_t t;
@ -113,8 +94,6 @@ typedef struct ThreadVars_ {
SCMutex *m;
SCCondT *cond;
ThreadVarsMSSyncPt *ms_sync_pts;
uint8_t cap_flags; /**< Flags to indicate the capabilities of all the
TmModules resgitered under this thread */
struct ThreadVars_ *next;

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

@ -450,7 +450,7 @@ TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p,
Packet *extra_p;
for (s = slot; s != NULL; s = s->slot_next) {
PACKET_PROFILING_TMM_START(p, s->tm_module_id);
PACKET_PROFILING_TMM_START(p, s->tm_id);
if (unlikely(s->id == 0)) {
r = s->SlotFunc(tv, p, s->slot_data, &s->slot_pre_pq, &s->slot_post_pq);
@ -458,7 +458,7 @@ TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p,
r = s->SlotFunc(tv, p, s->slot_data, &s->slot_pre_pq, NULL);
}
PACKET_PROFILING_TMM_END(p, s->tm_module_id);
PACKET_PROFILING_TMM_END(p, s->tm_id);
/* handle error */
if (unlikely(r == TM_ECODE_FAILED)) {
@ -848,7 +848,7 @@ void TmSlotSetFuncAppend(ThreadVars *tv, TmModule *tm, void *data)
prof_tm = &tmm_modules[prof_tm_id];
if (prof_tm == tm) {
slot->tm_module_id = prof_tm_id;
slot->tm_id = prof_tm_id;
break;
}
}
@ -1372,38 +1372,54 @@ void TmThreadKillThread(ThreadVars *tv)
tv->InShutdownHandler(tv);
}
for (i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
SCCondSignal(&trans_q[tv->inq->id].cond_q);
if (tv->inq->q_type == 0)
SCCondSignal(&trans_q[tv->inq->id].cond_q);
else
SCCondSignal(&data_queues[tv->inq->id].cond_q);
}
/* to be sure, signal more */
int cnt = 0;
while (1) {
if (TmThreadsCheckFlag(tv, THV_CLOSED)) {
SCLogDebug("signalled the thread %" PRId32 " times", cnt);
break;
}
cnt++;
if (tv->InShutdownHandler != NULL) {
tv->InShutdownHandler(tv);
}
for (i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
SCCondSignal(&trans_q[tv->inq->id].cond_q);
if (tv->inq->q_type == 0)
SCCondSignal(&trans_q[tv->inq->id].cond_q);
else
SCCondSignal(&data_queues[tv->inq->id].cond_q);
}
usleep(100);
}
SCLogDebug("signalled tv->inq->id %" PRIu32 "", tv->inq->id);
}
if (tv->cond != NULL ) {
int cnt = 0;
while (1) {
if (TmThreadsCheckFlag(tv, THV_CLOSED)) {
SCLogDebug("signalled the thread %" PRId32 " times", cnt);
break;
}
cnt++;
pthread_cond_broadcast(tv->cond);
usleep(100);
}
}
/* join it */
pthread_join(tv->t, NULL);
SCLogDebug("thread %s stopped", tv->name);
return;
}
@ -1477,7 +1493,8 @@ TmSlot *TmThreadGetFirstTmSlotForPartialPattern(const char *tm_name)
return slots;
}
void TmThreadKillThreads(void) {
void TmThreadKillThreads(void)
{
ThreadVars *tv = NULL;
int i = 0;
@ -1485,86 +1502,7 @@ void TmThreadKillThreads(void) {
tv = tv_root[i];
while (tv) {
if (tv->inq != NULL) {
/* we wait till we dry out all the inq packets, before we
* kill this thread. Do note that you should have disabled
* packet acquire by now using TmThreadDisableReceiveThreads()*/
if (!(strlen(tv->inq->name) == strlen("packetpool") &&
strcasecmp(tv->inq->name, "packetpool") == 0)) {
PacketQueue *q = &trans_q[tv->inq->id];
while (q->len != 0) {
usleep(1000);
}
}
}
TmThreadsSetFlag(tv, THV_KILL);
TmThreadsSetFlag(tv, THV_DEINIT);
SCLogDebug("told thread %s to stop", tv->name);
if (tv->inq != NULL) {
int i;
/* make sure our packet pending counter doesn't block */
/* signal the queue for the number of users */
if (tv->InShutdownHandler != NULL) {
tv->InShutdownHandler(tv);
}
for (i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
if (tv->inq->q_type == 0)
SCCondSignal(&trans_q[tv->inq->id].cond_q);
else
SCCondSignal(&data_queues[tv->inq->id].cond_q);
}
/* to be sure, signal more */
int cnt = 0;
while (1) {
if (TmThreadsCheckFlag(tv, THV_CLOSED)) {
SCLogDebug("signalled the thread %" PRId32 " times", cnt);
break;
}
cnt++;
if (tv->InShutdownHandler != NULL) {
tv->InShutdownHandler(tv);
}
for (i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
if (tv->inq->q_type == 0)
SCCondSignal(&trans_q[tv->inq->id].cond_q);
else
SCCondSignal(&data_queues[tv->inq->id].cond_q);
}
usleep(100);
}
SCLogDebug("signalled tv->inq->id %" PRIu32 "", tv->inq->id);
}
if (tv->cond != NULL ) {
int cnt = 0;
while (1) {
if (TmThreadsCheckFlag(tv, THV_CLOSED)) {
SCLogDebug("signalled the thread %" PRId32 " times", cnt);
break;
}
cnt++;
pthread_cond_broadcast(tv->cond);
usleep(100);
}
}
/* join it */
pthread_join(tv->t, NULL);
SCLogDebug("thread %s stopped", tv->name);
TmThreadKillThread(tv);
tv = tv->next;
}

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

@ -65,10 +65,6 @@ typedef struct TmSlot_ {
/* linked list, only used when you have multiple slots(used by TmVarSlot) */
struct TmSlot_ *slot_next;
#ifdef PROFILING
int tm_module_id;
#endif
} TmSlot;
extern ThreadVars *tv_root[TVT_MAX];

Write Preview
Loading…
Cancel
Save