threading: optimize and unify post_pq checks

TmThreadsSlotProcessPkt did not need to look all 'slots' as only the first
slots post_pq can have been used.

Unify post_pq cleanup handling.

src/tm-threads.c

@@ -110,7 +110,7 @@ void TmThreadsUnsetFlag(ThreadVars *tv, uint32_t flag)
 /**
  * \brief Separate run function so we can call it recursively.
  *
- * \todo Deal with post_pq for slots beyond the first.
+ * \note post_pq if only used for first slot
  */
 TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p, TmSlot *slot)
 {
@@ -602,37 +602,9 @@ static void *TmThreadsSlotVar(void *td)
             /* output the packet */
             tv->tmqh_out(tv, p);
 
-        } /* if (p != NULL) */
-
             /* now handle the post_pq packets */
-        TmSlot *slot;
+            TmThreadsSlotHandlePostPQs(tv, s);
-        for (slot = s; slot != NULL; slot = slot->slot_next) {
-            if (slot->slot_post_pq.top != NULL) {
-                while (1) {
-                    SCMutexLock(&slot->slot_post_pq.mutex_q);
-                    Packet *extra_p = PacketDequeue(&slot->slot_post_pq);
-                    SCMutexUnlock(&slot->slot_post_pq.mutex_q);
-
-                    if (extra_p == NULL)
-                        break;
-
-                    if (slot->slot_next != NULL) {
-                        r = TmThreadsSlotVarRun(tv, extra_p, slot->slot_next);
-                        if (r == TM_ECODE_FAILED) {
-                            SCMutexLock(&slot->slot_post_pq.mutex_q);
-                            TmqhReleasePacketsToPacketPool(&slot->slot_post_pq);
-                            SCMutexUnlock(&slot->slot_post_pq.mutex_q);
-
-                            TmqhOutputPacketpool(tv, extra_p);
-                            TmThreadsSetFlag(tv, THV_FAILED);
-                            break;
         }
-                    }
-                    /* output the packet */
-                    tv->tmqh_out(tv, extra_p);
-                } /* while */
-            } /* if */
-        } /* for */
 
         if (TmThreadsCheckFlag(tv, THV_KILL)) {
             run = 0;

src/tm-threads.h

@@ -160,82 +160,53 @@ static inline void TmThreadsSlotProcessPktFail(ThreadVars *tv, TmSlot *s, Packet
 }
 
 /**
- *  \brief Process the rest of the functions (if any) and queue.
+ *  \brief Handle timeout from the capture layer. Checks
+ *         post-pq which may have been filled by the flow
+ *         manager.
  */
-static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet *p)
+static inline void TmThreadsSlotHandlePostPQs(ThreadVars *tv, TmSlot *s)
 {
-    if (s == NULL) {
+    /* post process pq: only the first slot will possible have used it */
-        tv->tmqh_out(tv, p);
+    if (s->slot_post_pq.top != NULL) {
-        return TM_ECODE_OK;
-    }
-
-    TmEcode r = TmThreadsSlotVarRun(tv, p, s);
-    if (unlikely(r == TM_ECODE_FAILED)) {
-        TmThreadsSlotProcessPktFail(tv, s, p);
-        return TM_ECODE_FAILED;
-    }
-
-    tv->tmqh_out(tv, p);
-
-    /* post process pq */
-    for (TmSlot *slot = s; slot != NULL; slot = slot->slot_next) {
-        if (slot->slot_post_pq.top != NULL) {
         while (1) {
-                SCMutexLock(&slot->slot_post_pq.mutex_q);
+            SCMutexLock(&s->slot_post_pq.mutex_q);
-                Packet *extra_p = PacketDequeue(&slot->slot_post_pq);
+            Packet *extra_p = PacketDequeue(&s->slot_post_pq);
-                SCMutexUnlock(&slot->slot_post_pq.mutex_q);
+            SCMutexUnlock(&s->slot_post_pq.mutex_q);
-
             if (extra_p == NULL)
                 break;
 
-                if (slot->slot_next != NULL) {
+            if (s->slot_next != NULL) {
-                    r = TmThreadsSlotVarRun(tv, extra_p, slot->slot_next);
+                TmEcode r = TmThreadsSlotVarRun(tv, extra_p, s->slot_next);
                 if (r == TM_ECODE_FAILED) {
-                        TmThreadsSlotProcessPktFail(tv, slot, extra_p);
+                    TmThreadsSlotProcessPktFail(tv, s, extra_p);
                     break;
                 }
             }
             tv->tmqh_out(tv, extra_p);
         }
-        } /* if (slot->slot_post_pq.top != NULL) */
     }
-    return TM_ECODE_OK;
 }
 
 /**
- *  \brief Handle timeout from the capture layer. Checks
+ *  \brief Process the rest of the functions (if any) and queue.
- *         post-pq which may have been filled by the flow
- *         manager.
  */
-static inline TmEcode TmThreadsSlotHandlePostPQs(ThreadVars *tv, TmSlot *s)
+static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet *p)
 {
-    /* post process pq */
+    if (s == NULL) {
-    for (TmSlot *slot = s; slot != NULL; slot = slot->slot_next) {
+        tv->tmqh_out(tv, p);
-        if (slot->slot_post_pq.top != NULL) {
+        return TM_ECODE_OK;
-            while (1) {
+    }
-                SCMutexLock(&slot->slot_post_pq.mutex_q);
-                Packet *extra_p = PacketDequeue(&slot->slot_post_pq);
-                SCMutexUnlock(&slot->slot_post_pq.mutex_q);
-
-                if (extra_p == NULL)
-                    break;
-
-                if (slot->slot_next != NULL) {
-                    TmEcode r = TmThreadsSlotVarRun(tv, extra_p, slot->slot_next);
-                    if (r == TM_ECODE_FAILED) {
-                        SCMutexLock(&slot->slot_post_pq.mutex_q);
-                        TmqhReleasePacketsToPacketPool(&slot->slot_post_pq);
-                        SCMutexUnlock(&slot->slot_post_pq.mutex_q);
 
-                        TmqhOutputPacketpool(tv, extra_p);
+    TmEcode r = TmThreadsSlotVarRun(tv, p, s);
-                        TmThreadsSetFlag(tv, THV_FAILED);
+    if (unlikely(r == TM_ECODE_FAILED)) {
+        TmThreadsSlotProcessPktFail(tv, s, p);
         return TM_ECODE_FAILED;
     }
-                }
+
-                tv->tmqh_out(tv, extra_p);
+    tv->tmqh_out(tv, p);
-            }
+
-        }
+    TmThreadsSlotHandlePostPQs(tv, s);
-    }
+
     return TM_ECODE_OK;
 }