flow: enforce flow assumption

Enforce assumption that packets in ThreadVars::decode_pq have no flow
attached to it because this is only true for packets while they are
in the FlowWorker.

diff --git a/src/tm-threads.c b/src/tm-threads.c
index 5668dbc94a36f86601f52f382d6ea4bf3cf09591..69ca2e04029995a226c7c491c0c155e89364871d 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -43,6 +43,7 @@
 #include "util-profiling.h"
 #include "util-signal.h"
 #include "queue.h"
+#include "util-validate.h"
 
 #ifdef PROFILE_LOCKING
 thread_local uint64_t mutex_lock_contention;
@@ -117,6 +118,7 @@ TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p, TmSlot *slot)
         PACKET_PROFILING_TMM_START(p, s->tm_id);
         TmEcode r = s->SlotFunc(tv, p, SC_ATOMIC_GET(s->slot_data));
         PACKET_PROFILING_TMM_END(p, s->tm_id);
+        DEBUG_VALIDATE_BUG_ON(p->flow != NULL);
 
         /* handle error */
         if (unlikely(r == TM_ECODE_FAILED)) {
@@ -130,6 +132,7 @@ TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p, TmSlot *slot)
             Packet *extra_p = PacketDequeueNoLock(&tv->decode_pq);
             if (unlikely(extra_p == NULL))
                 continue;
+            DEBUG_VALIDATE_BUG_ON(extra_p->flow != NULL);
 
             /* see if we need to process the packet */
             if (s->slot_next != NULL) {
@@ -175,6 +178,7 @@ static int TmThreadTimeoutLoop(ThreadVars *tv, TmSlot *s)
                 Packet *p = PacketDequeue(tv->stream_pq);
                 SCMutexUnlock(&tv->stream_pq->mutex_q);
                 if (likely(p)) {
+                    DEBUG_VALIDATE_BUG_ON(p->flow != NULL);
                     r = TmThreadsSlotProcessPkt(tv, fw_slot, p);
                     if (r == TM_ECODE_FAILED) {
                         break;

diff --git a/src/tm-threads.h b/src/tm-threads.h
index 6507ce16f550c1cf903341dda0efdd97542f6572..6a5346be362993ef467666cf8911240fe306c805 100644 (file)
--- a/src/tm-threads.h
+++ b/src/tm-threads.h
@@ -165,6 +165,9 @@ static inline bool TmThreadsHandleInjectedPackets(ThreadVars *tv)
             SCMutexUnlock(&pq->mutex_q);
             if (extra_p == NULL)
                 break;
+#ifdef DEBUG_VALIDATION
+            BUG_ON(extra_p->flow != NULL);
+#endif
             TmEcode r = TmThreadsSlotVarRun(tv, extra_p, tv->tm_flowworker);
             if (r == TM_ECODE_FAILED) {
                 TmThreadsSlotProcessPktFail(tv, tv->tm_flowworker, extra_p);

diff --git a/src/tmqh-packetpool.c b/src/tmqh-packetpool.c
index bf69b9932442f4525f9333d7a52a2171b631d24c..431fa2956e72c7a529cc60b731decd2d181331c9 100644 (file)
--- a/src/tmqh-packetpool.c
+++ b/src/tmqh-packetpool.c
@@ -470,8 +470,10 @@ void TmqhReleasePacketsToPacketPool(PacketQueue *pq)
     if (pq == NULL)
         return;
 
-    while ( (p = PacketDequeue(pq)) != NULL)
+    while ((p = PacketDequeue(pq)) != NULL) {
+        DEBUG_VALIDATE_BUG_ON(p->flow != NULL);
         TmqhOutputPacketpool(NULL, p);
+    }
 
     return;
 }