- 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

@@ -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);

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);

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);

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);
 

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) {

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) {

src/suricata.c

@@ -1563,6 +1563,9 @@ int main(int argc, char **argv)
     SCCudaPBKillBatchingPackets();
 #endif
 
+    /* Disable packet acquire thread first */
+    TmThreadDisableReceiveThreads();
+
     TmThreadKillThreads();
     SCPerfReleaseResources();
     FlowShutdown();

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;

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
  *  */

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);

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);

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
 
 /**