check if all packets are processed before disabling detect threads + kill all threads <= detect after FFR + other minor fixes

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

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

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

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;

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;

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