threading: improve thread queues checking by dumping more info

diff --git a/src/tm-threads.c b/src/tm-threads.c
index 7e3ea98aaa2abb48e4778eaf60274739241f50b1..9d5105a6ef5d6685f72d841a6afd0eb961f1d606 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -72,7 +72,6 @@ __thread uint64_t rwr_lock_cnt;
 
 /* prototypes */
 static int SetCPUAffinity(uint16_t cpu);
-
 static void TmThreadDeinitMC(ThreadVars *tv);
 
 /* root of the threadvars list */
@@ -230,6 +229,27 @@ static int TmThreadTimeoutLoop(ThreadVars *tv, TmSlot *s)
     return r;
 }
 
+/** \internal
+ *  \brief check 'slot' pre_pq and post_pq at thread cleanup
+ *         and dump detailed info about the state of the packets
+ *         and threads if in a unexpected state.
+ */
+static void CheckSlot(const TmSlot *slot)
+{
+    if (slot->slot_pre_pq.len || slot->slot_post_pq.len) {
+        for (Packet *xp = slot->slot_pre_pq.top; xp != NULL; xp = xp->next) {
+            SCLogNotice("pre_pq: slot id %u slot tm_id %u pre_pq.len %u packet src %s",
+                    slot->id, slot->tm_id, slot->slot_pre_pq.len, PktSrcToString(xp->pkt_src));
+        }
+        for (Packet *xp = slot->slot_post_pq.top; xp != NULL; xp = xp->next) {
+            SCLogNotice("post_pq: slot id %u slot tm_id %u post_pq.len %u packet src %s",
+                    slot->id, slot->tm_id, slot->slot_post_pq.len, PktSrcToString(xp->pkt_src));
+        }
+        TmThreadDumpThreads();
+        abort();
+    }
+}
+
 /*
 
     pcap/nfq
@@ -369,9 +389,7 @@ static void *TmThreadsSlotPktAcqLoop(void *td)
                 goto error;
             }
         }
-
-        BUG_ON(slot->slot_pre_pq.len);
-        BUG_ON(slot->slot_post_pq.len);
+        CheckSlot(slot);
     }
 
     tv->stream_pq = NULL;
@@ -488,8 +506,7 @@ static void *TmThreadsSlotPktAcqLoopAFL(void *td)
             }
         }
 
-        BUG_ON(slot->slot_pre_pq.len);
-        BUG_ON(slot->slot_post_pq.len);
+        CheckSlot(slot);
     }
 
     tv->stream_pq = NULL;
@@ -649,8 +666,7 @@ static void *TmThreadsSlotVar(void *td)
                 goto error;
             }
         }
-        BUG_ON(s->slot_pre_pq.len);
-        BUG_ON(s->slot_post_pq.len);
+        CheckSlot(s);
     }
 
     SCLogDebug("%s ending", tv->name);
@@ -2200,6 +2216,46 @@ uint32_t TmThreadCountThreadsByTmmFlags(uint8_t flags)
     return cnt;
 }
 
+static void TmThreadDoDumpSlots(const ThreadVars *tv)
+{
+    for (TmSlot *s = tv->tm_slots; s != NULL; s = s->slot_next) {
+        TmModule *m = TmModuleGetById(s->tm_id);
+        SCLogNotice("tv %p: -> slot %p id %d tm_id %d name %s %s",
+            tv, s, s->id, s->tm_id, m->name, (tv->type == 0 && tv->stream_pq == &s->slot_post_pq) ? "<==== stream_pq" : "");
+        if (tv->type == 0 && tv->stream_pq == &s->slot_pre_pq) {
+            SCLogNotice("tv %p: -> slot %p/%d holds stream_pq %p IN PRE_PQ SUPER WEIRD", tv, s, s->id, tv->stream_pq);
+        }
+        for (Packet *xp = s->slot_pre_pq.top; xp != NULL; xp = xp->next) {
+            SCLogNotice("tv %p: ==> pre_pq: slot id %u slot tm_id %u pre_pq.len %u packet src %s",
+                    tv, s->id, s->tm_id, s->slot_pre_pq.len, PktSrcToString(xp->pkt_src));
+        }
+        for (Packet *xp = s->slot_post_pq.top; xp != NULL; xp = xp->next) {
+            SCLogNotice("tv %p: ==> post_pq: slot id %u slot tm_id %u post_pq.len %u packet src %s",
+                    tv, s->id, s->tm_id, s->slot_post_pq.len, PktSrcToString(xp->pkt_src));
+        }
+    }
+}
+
+void TmThreadDumpThreads(void)
+{
+    SCMutexLock(&tv_root_lock);
+    for (int i = 0; i < TVT_MAX; i++) {
+        ThreadVars *tv = tv_root[i];
+        while (tv != NULL) {
+            const uint32_t flags = SC_ATOMIC_GET(tv->flags);
+            SCLogNotice("tv %p: type %u name %s tmm_flags %02X flags %X stream_pq %p",
+                    tv, tv->type, tv->name, tv->tmm_flags, flags, tv->stream_pq);
+            if (tv->inq && tv->stream_pq == &trans_q[tv->inq->id]) {
+                SCLogNotice("tv %p: stream_pq at tv->inq %u", tv, tv->inq->id);
+            }
+            TmThreadDoDumpSlots(tv);
+            tv = tv->next;
+        }
+    }
+    SCMutexUnlock(&tv_root_lock);
+    TmThreadsListThreads();
+}
+
 typedef struct Thread_ {
     ThreadVars *tv;     /**< threadvars structure */
     const char *name;
@@ -2229,7 +2285,15 @@ void TmThreadsListThreads(void)
         t = &thread_store.threads[s];
         if (t == NULL || t->in_use == 0)
             continue;
-        SCLogInfo("Thread %"PRIuMAX", %s type %d, tv %p", (uintmax_t)s+1, t->name, t->type, t->tv);
+
+        SCLogNotice("Thread %"PRIuMAX", %s type %d, tv %p in_use %d",
+                (uintmax_t)s+1, t->name, t->type, t->tv, t->in_use);
+        if (t->tv) {
+            ThreadVars *tv = t->tv;
+            const uint32_t flags = SC_ATOMIC_GET(tv->flags);
+            SCLogNotice("tv %p type %u name %s tmm_flags %02X flags %X",
+                    tv, tv->type, tv->name, tv->tmm_flags, flags);
+        }
     }
 
     SCMutexUnlock(&thread_store_lock);

diff --git a/src/tm-threads.h b/src/tm-threads.h
index eff914313c8fd605f18dceb84fd0e7a31483f992..5770f2c0265fef2166dee889e6ce2b4679727f49 100644 (file)
--- a/src/tm-threads.h
+++ b/src/tm-threads.h
@@ -115,6 +115,7 @@ void TmThreadClearThreadsFamily(int family);
 void TmThreadAppend(ThreadVars *, int);
 void TmThreadRemove(ThreadVars *, int);
 void TmThreadSetGroupName(ThreadVars *tv, const char *name);
+void TmThreadDumpThreads(void);
 
 TmEcode TmThreadSetCPUAffinity(ThreadVars *, uint16_t);
 TmEcode TmThreadSetThreadPriority(ThreadVars *, int);