From a844eecb0e115758eb87d515d37b9fcd877d2fec Mon Sep 17 00:00:00 2001 From: Anoop Saldanha Date: Fri, 9 Sep 2011 11:52:09 +0530 Subject: [PATCH] - Updated all runmodes to use synchronization points, right before each thread(slot function) tries to de-init the thread. - Main thread now first disables receive thread(s) before it kills receive and rest of the threads. --- src/runmode-erf-dag.c | 1 + src/runmode-erf-file.c | 1 + src/runmode-ipfw.c | 1 + src/runmode-nfq.c | 2 +- src/runmode-pcap-file.c | 4 ++ src/runmode-pfring.c | 2 + src/suricata.c | 3 + src/threadvars.h | 3 + src/tm-modules.c | 42 ++++++++++++++ src/tm-modules.h | 2 + src/tm-threads.c | 122 +++++++++++++++++++++++++++++++++++++--- src/tm-threads.h | 6 ++ 12 files changed, 180 insertions(+), 9 deletions(-) diff --git a/src/runmode-erf-dag.c b/src/runmode-erf-dag.c index 0b97517fc9..6b45beab70 100644 --- a/src/runmode-erf-dag.c +++ b/src/runmode-erf-dag.c @@ -101,6 +101,7 @@ int RunModeErfDagAuto(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv_receiveerf, tm_module, iface); + TmThreadMSRegisterSyncPt(tv_receiveerf, "ReceiveTMBeforeDeInit"); if (threading_set_cpu_affinity) { TmThreadSetCPUAffinity(tv_receiveerf, 0); diff --git a/src/runmode-erf-file.c b/src/runmode-erf-file.c index 7bfeb52ccf..68b1037a7b 100644 --- a/src/runmode-erf-file.c +++ b/src/runmode-erf-file.c @@ -90,6 +90,7 @@ int RunModeErfFileAuto(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv_receiveerf, tm_module, file); + TmThreadMSRegisterSyncPt(tv_receiveerf, "ReceiveTMBeforeDeInit"); if (threading_set_cpu_affinity) { TmThreadSetCPUAffinity(tv_receiveerf, 0); diff --git a/src/runmode-ipfw.c b/src/runmode-ipfw.c index a63b362938..6fdc681ad7 100644 --- a/src/runmode-ipfw.c +++ b/src/runmode-ipfw.c @@ -102,6 +102,7 @@ int RunModeIpsIPFWAuto(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv_receiveipfw, tm_module, NULL); + TmThreadMSRegisterSyncPt(tv_receiveipfw, "ReceiveTMBeforeDeInit"); if (threading_set_cpu_affinity) { TmThreadSetCPUAffinity(tv_receiveipfw, 0); diff --git a/src/runmode-nfq.c b/src/runmode-nfq.c index aa01a86cb1..5d10003860 100644 --- a/src/runmode-nfq.c +++ b/src/runmode-nfq.c @@ -118,8 +118,8 @@ int RunModeIpsNFQAuto(DetectEngineCtx *de_ctx) printf("ERROR: TmModuleGetByName failed for ReceiveNFQ\n"); exit(EXIT_FAILURE); } - TmSlotSetFuncAppend(tv_receivenfq, tm_module, (void *) NFQGetThread(i)); + TmThreadMSRegisterSyncPt(tv_receivenfq, "ReceiveTMBeforeDeInit"); TmThreadSetCPU(tv_receivenfq, RECEIVE_CPU_SET); diff --git a/src/runmode-pcap-file.c b/src/runmode-pcap-file.c index a1188c2291..25c8087abe 100644 --- a/src/runmode-pcap-file.c +++ b/src/runmode-pcap-file.c @@ -94,6 +94,7 @@ int RunModeFilePcapSingle(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv, tm_module, file); + TmThreadMSRegisterSyncPt(tv, "ReceiveTMBeforeDeInit"); tm_module = TmModuleGetByName("DecodePcapFile"); if (tm_module == NULL) { @@ -187,6 +188,7 @@ int RunModeFilePcapAuto(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv_receivepcap, tm_module, file); + TmThreadMSRegisterSyncPt(tv_receivepcap, "ReceiveTMBeforeDeInit"); TmThreadSetCPU(tv_receivepcap, RECEIVE_CPU_SET); @@ -228,6 +230,7 @@ int RunModeFilePcapAuto(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv_receivepcap, tm_module, file); + TmThreadMSRegisterSyncPt(tv_receivepcap, "ReceiveTMBeforeDeInit"); TmThreadSetCPU(tv_receivepcap, RECEIVE_CPU_SET); @@ -433,6 +436,7 @@ int RunModeFilePcapAutoFp(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv_receivepcap, tm_module, file); + TmThreadMSRegisterSyncPt(tv_receivepcap, "ReceiveTMBeforeDeInit"); tm_module = TmModuleGetByName("DecodePcapFile"); if (tm_module == NULL) { diff --git a/src/runmode-pfring.c b/src/runmode-pfring.c index db53b291e9..2f6bcd92c1 100644 --- a/src/runmode-pfring.c +++ b/src/runmode-pfring.c @@ -121,6 +121,7 @@ int RunModeIdsPfringAuto(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv_receivepfring, tm_module, NULL); + TmThreadMSRegisterSyncPt(tv_receivepfring, "ReceiveTMBeforeDeInit"); if (threading_set_cpu_affinity) { TmThreadSetCPUAffinity(tv_receivepfring, 0); @@ -371,6 +372,7 @@ int RunModeIdsPfringAutoFp(DetectEngineCtx *de_ctx) exit(EXIT_FAILURE); } TmSlotSetFuncAppend(tv_receive, tm_module, NULL); + TmThreadMSRegisterSyncPt(tv_receive, "ReceiveTMBeforeDeInit"); tm_module = TmModuleGetByName("DecodePfring"); if (tm_module == NULL) { diff --git a/src/suricata.c b/src/suricata.c index ced71286d8..7f8d9fdb6b 100644 --- a/src/suricata.c +++ b/src/suricata.c @@ -1563,6 +1563,9 @@ int main(int argc, char **argv) SCCudaPBKillBatchingPackets(); #endif + /* Disable packet acquire thread first */ + TmThreadDisableReceiveThreads(); + TmThreadKillThreads(); SCPerfReleaseResources(); FlowShutdown(); diff --git a/src/threadvars.h b/src/threadvars.h index 6bdd6bfe94..87da5ccfeb 100644 --- a/src/threadvars.h +++ b/src/threadvars.h @@ -60,6 +60,9 @@ typedef struct ThreadVarsMSSyncPt_ { 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; diff --git a/src/tm-modules.c b/src/tm-modules.c index 794c26622e..4b73eec661 100644 --- a/src/tm-modules.c +++ b/src/tm-modules.c @@ -63,6 +63,48 @@ TmModule *TmModuleGetByName(char *name) { return NULL; } +/** + * \brief Returns a TM Module by its id. + * + * \param id Id of the TM Module to return. + * + * \retval Pointer of the module to be returned if available; + * NULL if unavailable. + */ +TmModule *TmModuleGetById(int id) +{ + + if (id < 0 || id >= TMM_SIZE) { + SCLogError(SC_ERR_TM_MODULES_ERROR, "Threading module with the id " + "\"%d\" doesn't exist", id); + return NULL; + } + + return &tmm_modules[id]; +} + +/** + * \brief Given a TM Module, returns its id. + * + * \param tm Pointer to the TM Module. + * + * \retval id of the TM Module if available; -1 if unavailable. + */ +int TmModuleGetIDForTM(TmModule *tm) +{ + TmModule *t; + int i; + + for (i = 0; i < TMM_SIZE; i++) { + t = &tmm_modules[i]; + + if (strcmp(t->name, tm->name) == 0) + return i; + } + + return -1; +} + /** \brief LogFileNewCtx() Get a new LogFileCtx * \retval LogFileCtx * pointer if succesful, NULL if error * */ diff --git a/src/tm-modules.h b/src/tm-modules.h index a054a24c30..8c09a36e5d 100644 --- a/src/tm-modules.h +++ b/src/tm-modules.h @@ -93,6 +93,8 @@ LogFileCtx *LogFileNewCtx(); int LogFileFreeCtx(LogFileCtx *); TmModule *TmModuleGetByName(char *name); +TmModule *TmModuleGetById(int id); +int TmModuleGetIDForTM(TmModule *tm); TmEcode TmModuleRegister(char *name, int (*module_func)(ThreadVars *, Packet *, void *)); void TmModuleDebugList(void); void TmModuleRegisterTests(void); diff --git a/src/tm-threads.c b/src/tm-threads.c index 43676ebbc0..2878401a19 100644 --- a/src/tm-threads.c +++ b/src/tm-threads.c @@ -173,6 +173,10 @@ void *TmThreadsSlot1NoIn(void *td) } } /* while (run) */ + /* wait for synchronization from master, if this TV has a synchronization + * point set by this name */ + TmThreadsMSSlaveHitSyncPt(tv, "ReceiveTMBeforeDeInit"); + if (s->SlotThreadExitPrintStats != NULL) { s->SlotThreadExitPrintStats(tv, s->slot_data); } @@ -243,6 +247,10 @@ void *TmThreadsSlot1NoOut(void *td) } } /* while (run) */ + /* wait for synchronization from master, if this TV has a synchronization + * point set by this name */ + TmThreadsMSSlaveHitSyncPt(tv, "ReceiveTMBeforeDeInit"); + if (s->SlotThreadExitPrintStats != NULL) { s->SlotThreadExitPrintStats(tv, s->slot_data); } @@ -308,6 +316,10 @@ void *TmThreadsSlot1NoInOut(void *td) } } /* while (run) */ + /* wait for synchronization from master, if this TV has a synchronization + * point set by this name */ + TmThreadsMSSlaveHitSyncPt(tv, "ReceiveTMBeforeDeInit"); + if (s->SlotThreadExitPrintStats != NULL) { s->SlotThreadExitPrintStats(tv, s->slot_data); } @@ -403,6 +415,10 @@ void *TmThreadsSlot1(void *td) } } /* while (run) */ + /* wait for synchronization from master, if this TV has a synchronization + * point set by this name */ + TmThreadsMSSlaveHitSyncPt(tv, "ReceiveTMBeforeDeInit"); + if (s->SlotThreadExitPrintStats != NULL) { s->SlotThreadExitPrintStats(tv, s->slot_data); } @@ -554,6 +570,10 @@ void *TmThreadsSlotPktAcqLoop(void *td) { } SCPerfUpdateCounterArray(tv->sc_perf_pca, &tv->sc_perf_pctx, 0); + /* wait for synchronization from master, if this TV has a synchronization + * point set by this name */ + TmThreadsMSSlaveHitSyncPt(tv, "ReceiveTMBeforeDeInit"); + for (slot = s; slot != NULL; slot = slot->slot_next) { if (slot->SlotThreadExitPrintStats != NULL) { slot->SlotThreadExitPrintStats(tv, slot->slot_data); @@ -667,6 +687,10 @@ void *TmThreadsSlotVar(void *td) } /* while (run) */ SCPerfUpdateCounterArray(tv->sc_perf_pca, &tv->sc_perf_pctx, 0); + /* wait for synchronization from master, if this TV has a synchronization + * point set by this name */ + TmThreadsMSSlaveHitSyncPt(tv, "ReceiveTMBeforeDeInit"); + s = (TmSlot *)tv->tm_slots; for ( ; s != NULL; s = s->slot_next) { @@ -758,6 +782,9 @@ void TmSlotSetFuncAppend(ThreadVars *tv, TmModule *tm, void *data) slot->PktAcqLoop = tm->PktAcqLoop; slot->SlotThreadExitPrintStats = tm->ThreadExitPrintStats; slot->SlotThreadDeinit = tm->ThreadDeinit; + /* we don't have to check for the return value "-1". We wouldn't have + * received a TM as arg, if it didn't exist */ + slot->tm_id = TmModuleGetIDForTM(tm); tv->cap_flags |= tm->cap_flags; @@ -794,6 +821,41 @@ void TmSlotSetFuncAppend(ThreadVars *tv, TmModule *tm, void *data) return; } +/** + * \brief Returns the slot holding a TM with the particular tm_id. + * + * \param tm_id TM id of the TM whose slot has to be returned. + * + * \retval slots Pointer to the slot. + */ +TmSlot *TmSlotGetSlotForTM(int tm_id) +{ + ThreadVars *tv = NULL; + TmSlot *slots; + int i; + + SCMutexLock(&tv_root_lock); + + for (i = 0; i < TVT_MAX; i++) { + tv = tv_root[i]; + while (tv) { + slots = tv->tm_slots; + while (slots != NULL) { + if (slots->tm_id == tm_id) { + SCMutexUnlock(&tv_root_lock); + return slots; + } + slots = slots->slot_next; + } + tv = tv->next; + } + } + + SCMutexUnlock(&tv_root_lock); + + return NULL; +} + #if !defined OS_WIN32 && !defined __OpenBSD__ static int SetCPUAffinitySet(cpu_set_t *cs) { #if defined OS_FREEBSD @@ -1296,6 +1358,45 @@ void TmThreadKillThread(ThreadVars *tv) return; } +/** + * \brief Disable receive threads. + */ +void TmThreadDisableReceiveThreads(void) +{ + ThreadVars *tv = NULL; + + SCMutexLock(&tv_root_lock); + + /* all receive threads are part of packet processing threads */ + tv = tv_root[TVT_PPT]; + + /* we do have to keep in mind that TVs are arranged in the order + * right from receive to log. The moment we fail to find a + * receive TM amongst the slots in a tv, it indicates we are done + * with all receive threads */ + while (tv) { + /* obtain the slots for this TV */ + TmSlot *slots = tv->tm_slots; + TmModule *tm = TmModuleGetById(slots->tm_id); + + /* Kind of a hack. All packet_acquire/receive TMs in our engine + * have the string "receive" as a substring in their name */ + char *found = strcasestr(tm->name, "receive"); + if (found == NULL) + break; + + /* we found our receive TV. Send it a KILL signal. This is all + * we need to do to kill receive threads */ + TmThreadsSetFlag(tv, THV_KILL); + + tv = tv->next; + } + + SCMutexUnlock(&tv_root_lock); + + return; +} + void TmThreadKillThreads(void) { ThreadVars *tv = NULL; int i = 0; @@ -1307,6 +1408,8 @@ void TmThreadKillThreads(void) { TmThreadsSetFlag(tv, THV_KILL); SCLogDebug("told thread %s to stop", tv->name); + TmThreadsMSMasterDisableSlaveAllSyncPts(tv); + if (tv->inq != NULL) { int i; @@ -1452,16 +1555,19 @@ void TmThreadsMSSlaveHitSyncPt(ThreadVars *tv, const char *sync_pt_name) SCMutexLock(&ms_sync_pts->m); { - ms_sync_pts->slave_hit = 1; - while (1) { - SCCondWait(&ms_sync_pts->cond, &ms_sync_pts->m); + if (!ms_sync_pts->disabled) { + ms_sync_pts->slave_hit = 1; + while (1) { + SCCondWait(&ms_sync_pts->cond, &ms_sync_pts->m); - if (!ms_sync_pts->master_go) - continue; + if (!ms_sync_pts->master_go) + continue; - /* reset them */ - ms_sync_pts->slave_hit = 0; - ms_sync_pts->master_go = 0; + /* reset them */ + ms_sync_pts->slave_hit = 0; + ms_sync_pts->master_go = 0; + break; + } } } SCMutexUnlock(&ms_sync_pts->m); diff --git a/src/tm-threads.h b/src/tm-threads.h index acb46142bf..7fcbceff10 100644 --- a/src/tm-threads.h +++ b/src/tm-threads.h @@ -54,6 +54,9 @@ typedef struct TmSlot_ { * locks in the queue are NOT used */ PacketQueue slot_post_pq; + /* store the thread module id */ + int tm_id; + /* slot id, only used my TmVarSlot to know what the first slot is */ int id; @@ -70,6 +73,7 @@ extern ThreadVars *tv_root[TVT_MAX]; extern SCMutex tv_root_lock; void TmSlotSetFuncAppend(ThreadVars *, TmModule *, void *); +TmSlot *TmSlotGetSlotForTM(int); ThreadVars *TmThreadCreate(char *, char *, char *, char *, char *, char *, void *(fn_p)(void *), int); @@ -114,6 +118,8 @@ void TmThreadsMSMasterDisableSlaveSyncPt(ThreadVars *, const char *); void TmThreadsMSMasterReleaseSlaveAllSyncPts(ThreadVars *); void TmThreadsMSMasterDisableSlaveAllSyncPts(ThreadVars *tv); +void TmThreadDisableReceiveThreads(void); + #if 0 /**