threads: cleanup decode_pq handling

diff --git a/src/tm-threads.c b/src/tm-threads.c
index 99808ad2ca59257da3cf922cddf388d96d87e9a0..361bad6df0b633d9a7b360f3f54829de787ba6aa 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -109,6 +109,22 @@ void TmThreadsUnsetFlag(ThreadVars *tv, uint32_t flag)
     SC_ATOMIC_AND(tv->flags, ~flag);
 }
 
+TmEcode TmThreadsProcessDecodePseudoPackets(
+        ThreadVars *tv, PacketQueueNoLock *decode_pq, TmSlot *slot)
+{
+    while (decode_pq->top != NULL) {
+        Packet *extra_p = PacketDequeueNoLock(decode_pq);
+        if (unlikely(extra_p == NULL))
+            continue;
+        DEBUG_VALIDATE_BUG_ON(extra_p->flow != NULL);
+
+        if (TmThreadsSlotProcessPkt(tv, slot, extra_p) != TM_ECODE_OK) {
+            SCReturnInt(TM_ECODE_FAILED);
+        }
+    }
+    SCReturnInt(TM_ECODE_OK);
+}
+
 /**
  * \brief Separate run function so we can call it recursively.
  */
@@ -127,22 +143,8 @@ TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p, TmSlot *slot)
             return TM_ECODE_FAILED;
         }
 
-        /* handle new pseudo packets immediately */
-        while (tv->decode_pq.top != NULL) {
-            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) {
-                r = TmThreadsSlotVarRun(tv, extra_p, s->slot_next);
-                if (unlikely(r == TM_ECODE_FAILED)) {
-                    TmThreadsSlotProcessPktFail(tv, s, extra_p);
-                    return TM_ECODE_FAILED;
-                }
-            }
-            tv->tmqh_out(tv, extra_p);
+        if (TmThreadsProcessDecodePseudoPackets(tv, &tv->decode_pq, s->slot_next) != TM_ECODE_OK) {
+            return TM_ECODE_FAILED;
         }
     }
 

diff --git a/src/tm-threads.h b/src/tm-threads.h
index 240f7e3be0882fb29c995fed419a14563f3dffde..ec791bea47be8bc9ba478c271a8c218b3496ebbd 100644 (file)
--- a/src/tm-threads.h
+++ b/src/tm-threads.h
@@ -124,6 +124,9 @@ uint32_t TmThreadCountThreadsByTmmFlags(uint8_t flags);
 
 TmEcode TmThreadWaitOnThreadRunning(void);
 
+TmEcode TmThreadsProcessDecodePseudoPackets(
+        ThreadVars *tv, PacketQueueNoLock *decode_pq, TmSlot *slot);
+
 static inline void TmThreadsCleanDecodePQ(PacketQueueNoLock *pq)
 {
     while (1) {