check if all packets are processed before disabling detect threads + kill all threads...

diff --git a/src/flow-timeout.c b/src/flow-timeout.c
index c481146f968c4c5305ad2277620cdfc869de7ff6..9b86b51f6c9b6908c30ee6edc408620018dd8133 100644 (file)
--- a/src/flow-timeout.c
+++ b/src/flow-timeout.c
@@ -605,10 +605,6 @@ void FlowForceReassembly(void)
     /* Do remember.  We need to have packet acquire disabled by now */
 
     /** ----- Part 1 ----- **/
-    /* First we need to kill the flow manager thread */
-    FlowKillFlowManagerThread();
-
-    /** ----- Part 2 ----- **/
     /* Check if all threads are idle.  We need this so that we have all
      * packets freeds.  As a consequence, no flows are in use */
 
@@ -636,7 +632,7 @@ void FlowForceReassembly(void)
 
     SCMutexUnlock(&tv_root_lock);
 
-    /** ----- Part 3 ----- **/
+    /** ----- Part 2 ----- **/
     /* Carry out flow reassembly for unattended flows */
     FlowForceReassemblyForHash();
 

diff --git a/src/stream-tcp.c b/src/stream-tcp.c
index e65d45a7bfe1d6929f3d89a64cfe774f39c785f1..2e6f219455cbf4630ce115c96f576dd4cdbe18f3 100644 (file)
--- a/src/stream-tcp.c
+++ b/src/stream-tcp.c
@@ -125,6 +125,7 @@ void TmModuleStreamTcpRegister (void)
     tmm_modules[TMM_STREAMTCP].ThreadDeinit = StreamTcpThreadDeinit;
     tmm_modules[TMM_STREAMTCP].RegisterTests = StreamTcpRegisterTests;
     tmm_modules[TMM_STREAMTCP].cap_flags = 0;
+    tmm_modules[TMM_STREAMTCP].flags = TM_FLAG_STREAM_TM;
 }
 
 void StreamTcpIncrMemuse(uint64_t size) {

diff --git a/src/suricata.c b/src/suricata.c
index 8ffb5b4e4ada38ce9d15279a32bbe85315d788d3..f3207c896bf244400b80f8c45cc5ae9bd3184095 100644 (file)
--- a/src/suricata.c
+++ b/src/suricata.c
@@ -1888,6 +1888,9 @@ int main(int argc, char **argv)
     SCCudaPBKillBatchingPackets();
 #endif
 
+    /* First we need to kill the flow manager thread */
+    FlowKillFlowManagerThread();
+
     /* Disable packet acquire thread first */
     TmThreadDisableReceiveThreads();
 
@@ -1902,7 +1905,7 @@ int main(int argc, char **argv)
 
     if (rule_reload == 1) {
         /* Disable detect threads first.  This is required by live rule swap */
-        TmThreadDisableDetectThreads();
+        TmThreadDisableUptoDetectThreads();
 
         /* wait if live rule swap is in progress */
         if (UtilSignalIsHandler(SIGUSR2, SignalHandlerSigusr2Idle)) {

diff --git a/src/tm-modules.h b/src/tm-modules.h
index 3e35f708628983d915ca65fc14f0b738c4cfaafa..6789888fa37a620dea58525722d4183d7288ab2e 100644 (file)
--- a/src/tm-modules.h
+++ b/src/tm-modules.h
@@ -30,7 +30,8 @@
 /* thread flags */
 #define TM_FLAG_RECEIVE_TM      0x01
 #define TM_FLAG_DECODE_TM       0x02
-#define TM_FLAG_DETECT_TM       0x04
+#define TM_FLAG_STREAM_TM       0x04
+#define TM_FLAG_DETECT_TM       0x08
 
 typedef struct TmModule_ {
     char *name;

diff --git a/src/tm-threads.c b/src/tm-threads.c
index 23b9bb9561af7c0d37016d8fd59840d48bb61f0e..196d7dac0d3eb56c2cde34ac467ae491c9073661 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -1549,9 +1549,9 @@ void TmThreadDisableReceiveThreads(void)
 }
 
 /**
- * \brief Disable all detect threads.
+ * \brief Disable all threads <= detect.
  */
-void TmThreadDisableDetectThreads(void)
+void TmThreadDisableUptoDetectThreads(void)
 {
     /* value in seconds */
 #define THREAD_KILL_MAX_WAIT_TIME 60
@@ -1572,14 +1572,34 @@ void TmThreadDisableDetectThreads(void)
      * receive TM amongst the slots in a tv, it indicates we are done
      * with all receive threads */
     while (tv) {
+        int disable = 0;
         /* obtain the slots for this TV */
         TmSlot *slots = tv->tm_slots;
         while (slots != NULL) {
             TmModule *tm = TmModuleGetById(slots->tm_id);
 
-            if (!(tm->flags & TM_FLAG_DETECT_TM)) {
-                slots = slots->slot_next;
-                continue;
+            if (tm->flags & (TM_FLAG_RECEIVE_TM | TM_FLAG_DECODE_TM |
+                             TM_FLAG_STREAM_TM | TM_FLAG_DETECT_TM)) {
+                disable = 1;
+                break;
+            }
+
+            slots = slots->slot_next;
+            continue;
+        }
+
+        if (disable) {
+            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);
+                    }
+                }
             }
 
             /* we found our receive TV.  Send it a KILL signal.  This is all
@@ -1607,8 +1627,6 @@ void TmThreadDisableDetectThreads(void)
                     exit(EXIT_FAILURE);
                 }
             }
-
-            break;
         }
 
         tv = tv->next;

diff --git a/src/tm-threads.h b/src/tm-threads.h
index 9715b1972eaaa43e4742458f889d40b51a5f6606..7f16384be80488592aa29dafc07644fa828728cc 100644 (file)
--- a/src/tm-threads.h
+++ b/src/tm-threads.h
@@ -113,7 +113,7 @@ TmEcode TmThreadsSlotVarRun (ThreadVars *tv, Packet *p, TmSlot *slot);
 
 ThreadVars *TmThreadsGetTVContainingSlot(TmSlot *);
 void TmThreadDisableReceiveThreads(void);
-void TmThreadDisableDetectThreads(void);
+void TmThreadDisableUptoDetectThreads(void);
 TmSlot *TmThreadGetFirstTmSlotForPartialPattern(const char *);
 
 /**