Debug-validation: track thread id in flow

In workers mode, all packets of a single flow should be handled by
a single thread. In autofp mode, all packets in a flow should be
handled by the same stream/detect thread.

This patch stores the id in the flow and aborts if more than one
thread handling a flow.

The check and store are done from the stream engine, so the autofp
case works as well.

src/flow-util.h

@@ -39,6 +39,14 @@
 #define RESET_COUNTERS(f)
 #endif
 
+#ifdef DEBUG_VALIDATION
+#define RESET_THREAD_ID(f) do { \
+        (f)->thread_id = 0; \
+    } while (0)
+#else
+#define RESET_THREAD_ID(f)
+#endif
+
 #define FLOW_INITIALIZE(f) do { \
         (f)->sp = 0; \
         (f)->dp = 0; \
@@ -70,6 +78,7 @@
         SC_ATOMIC_INIT((f)->autofp_tmqh_flow_qid);  \
         (void) SC_ATOMIC_SET((f)->autofp_tmqh_flow_qid, -1);  \
         RESET_COUNTERS((f)); \
+        RESET_THREAD_ID((f)); \
     } while (0)
 
 /** \brief macro to recycle a flow before it goes into the spare queue for reuse.
@@ -107,6 +116,7 @@
             (void) SC_ATOMIC_SET((f)->autofp_tmqh_flow_qid, -1);   \
         }                                       \
         RESET_COUNTERS((f)); \
+        RESET_THREAD_ID((f)); \
     } while(0)
 
 #define FLOW_DESTROY(f) do { \

src/flow.h

@@ -382,6 +382,9 @@ typedef struct Flow_
     uint32_t tosrcpktcnt;
     uint64_t bytecnt;
 #endif
+#ifdef DEBUG_VALIDATION
+    u_long thread_id;
+#endif
 } Flow;
 
 enum {

src/stream-tcp.c

@@ -4171,6 +4171,17 @@ int StreamTcpPacket (ThreadVars *tv, Packet *p, StreamTcpThread *stt,
 
     DEBUG_ASSERT_FLOW_LOCKED(p->flow);
 
+#ifdef DEBUG_VALIDATION
+    if (p->pkt_src == PKT_SRC_WIRE) {
+        if (p->flow->thread_id == 0)
+            p->flow->thread_id = tv->thread_id;
+        else {
+            SCLogDebug("tv->id %d", (int)tv->thread_id);
+            BUG_ON(p->flow->thread_id != tv->thread_id);
+        }
+    }
+#endif
+
     SCLogDebug("p->pcap_cnt %"PRIu64, p->pcap_cnt);
 
     TcpSession *ssn = (TcpSession *)p->flow->protoctx;

src/threadvars.h

@@ -69,8 +69,7 @@ typedef struct ThreadVars_ {
     /** no of times the thread has been restarted on failure */
     uint8_t restarted;
 
-    /** local id */
-    int id;
+    u_long thread_id;
 
     /** queue's */
     Tmq *inq;

src/tm-threads.c

@@ -646,7 +646,9 @@ void *TmThreadsSlotPktAcqLoop(void *td) {
 
     if (tv->thread_setup_flags != 0)
         TmThreadSetupOptions(tv);
-
+#ifdef DEBUG_VALIDATION
+    tv->thread_id = SCGetThreadIdLong();
+#endif
     /* Drop the capabilities for this thread */
     SCDropCaps(tv);
 
@@ -759,7 +761,9 @@ void *TmThreadsSlotVar(void *td)
 
     if (tv->thread_setup_flags != 0)
         TmThreadSetupOptions(tv);
-
+#ifdef DEBUG_VALIDATION
+    tv->thread_id = SCGetThreadIdLong();
+#endif
     /* Drop the capabilities for this thread */
     SCDropCaps(tv);