tmqh: return a queue from input queue handler

src/threadvars.h

@@ -88,7 +88,7 @@ typedef struct ThreadVars_ {
 
     /** incoming queue and handler */
     Tmq *inq;
-    struct Packet_ *(*TmqhInFn)(struct ThreadVars_ *);
+    struct PacketQueueNoLock_ (*TmqhInFn)(struct ThreadVars_ *);
 
     SC_ATOMIC_DECLARE(uint32_t, flags);
 

src/tm-queuehandlers.h

@@ -24,6 +24,8 @@
 #ifndef SURICATA_TM_QUEUEHANDLERS_H
 #define SURICATA_TM_QUEUEHANDLERS_H
 
+#include "packet-queue.h"
+
 enum {
     TMQH_NOT_SET,
     TMQH_SIMPLE,
@@ -35,7 +37,7 @@ enum {
 
 typedef struct Tmqh_ {
     const char *name;
-    Packet *(*InHandler)(ThreadVars *);
+    PacketQueueNoLock (*InHandler)(ThreadVars *);
     void (*InShutdownHandler)(ThreadVars *);
     void (*OutHandler)(ThreadVars *, Packet *);
     void *(*OutHandlerCtxSetup)(const char *);

src/tm-threads.c

@@ -489,39 +489,43 @@ static void *TmThreadsSlotVar(void *td)
     s = (TmSlot *)tv->tm_slots;
 
     while (run) {
-        /* input a packet */
+        /* get packet(s) from the queue. Can get more than one so should loop
-        p = tv->TmqhInFn(tv);
+         * over it. */
-
+        PacketQueueNoLock q = tv->TmqhInFn(tv);
-        /* if we didn't get a packet see if we need to do some housekeeping */
+        do {
-        if (unlikely(p == NULL)) {
+            p = PacketDequeueNoLock(&q);
-            if (tv->flow_queue && SC_ATOMIC_GET(tv->flow_queue->non_empty)) {
+
-                p = PacketGetFromQueueOrAlloc();
+            /* if we didn't get a packet see if we need to do some housekeeping */
-                if (p != NULL) {
+            if (unlikely(p == NULL)) {
-                    p->flags |= PKT_PSEUDO_STREAM_END;
+                if (tv->flow_queue && SC_ATOMIC_GET(tv->flow_queue->non_empty)) {
-                    PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT);
+                    p = PacketGetFromQueueOrAlloc();
+                    if (p != NULL) {
+                        p->flags |= PKT_PSEUDO_STREAM_END;
+                        PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT);
+                    }
                 }
             }
-        }
 
-        if (p != NULL) {
+            if (p != NULL) {
-            /* run the thread module(s) */
+                /* run the thread module(s) */
-            r = TmThreadsSlotVarRun(tv, p, s);
+                r = TmThreadsSlotVarRun(tv, p, s);
-            if (r == TM_ECODE_FAILED) {
+                if (r == TM_ECODE_FAILED) {
-                TmqhOutputPacketpool(tv, p);
+                    TmqhOutputPacketpool(tv, p);
-                TmThreadsSetFlag(tv, THV_FAILED);
+                    TmThreadsSetFlag(tv, THV_FAILED);
-                break;
+                    break;
-            }
+                }
 
-            /* output the packet */
+                /* output the packet */
-            tv->TmqhOutFn(tv, p);
+                tv->TmqhOutFn(tv, p);
 
-            /* now handle the stream pq packets */
+                /* now handle the stream pq packets */
-            TmThreadsHandleInjectedPackets(tv);
+                TmThreadsHandleInjectedPackets(tv);
-        }
+            }
 
-        if (TmThreadsCheckFlag(tv, (THV_KILL | THV_REQ_FLOW_LOOP))) {
+            if (TmThreadsCheckFlag(tv, (THV_KILL | THV_REQ_FLOW_LOOP))) {
-            run = false;
+                run = false;
-        }
+            }
+        } while (q.len != 0);
     }
     if (!SCTmThreadsSlotPacketLoopFinish(tv)) {
         goto error;

src/tmqh-flow.c

@@ -40,7 +40,7 @@
 #include "conf.h"
 #include "util-unittest.h"
 
-Packet *TmqhInputFlow(ThreadVars *t);
+PacketQueueNoLock TmqhInputFlow(ThreadVars *t);
 void TmqhOutputFlowHash(ThreadVars *t, Packet *p);
 void TmqhOutputFlowIPPair(ThreadVars *t, Packet *p);
 static void TmqhOutputFlowFTPHash(ThreadVars *t, Packet *p);
@@ -98,27 +98,21 @@ void TmqhFlowPrintAutofpHandler(void)
 }
 
 /* same as 'simple' */
-Packet *TmqhInputFlow(ThreadVars *tv)
+PacketQueueNoLock TmqhInputFlow(ThreadVars *tv)
 {
-    PacketQueue *q = tv->inq->pq;
-
     StatsSyncCountersIfSignalled(&tv->stats);
 
+    PacketQueue *q = tv->inq->pq;
     SCMutexLock(&q->mutex_q);
     if (q->len == 0) {
         /* if we have no packets in queue, wait... */
         SCCondWait(&q->cond_q, &q->mutex_q);
     }
-
+    PacketQueueNoLock pq = { .top = q->top, .bot = q->bot, q->len };
-    if (q->len > 0) {
+    q->bot = q->top = NULL;
-        Packet *p = PacketDequeue(q);
+    q->len = 0;
-        SCMutexUnlock(&q->mutex_q);
+    SCMutexUnlock(&q->mutex_q);
-        return p;
+    return pq;
-    } else {
-        /* return NULL if we have no pkt. Should only happen on signals. */
-        SCMutexUnlock(&q->mutex_q);
-        return NULL;
-    }
 }
 
 static int StoreQueueId(TmqhFlowCtx *ctx, char *name)

src/tmqh-packetpool.c

@@ -48,6 +48,8 @@ static inline PktPool *GetThreadPacketPool(void)
     return &thread_pkt_pool;
 }
 
+static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *);
+
 /**
  * \brief TmqhPacketpoolRegister
  * \initonly
@@ -297,9 +299,11 @@ void PacketPoolDestroy(void)
 #endif /* DEBUG_VALIDATION */
 }
 
-Packet *TmqhInputPacketpool(ThreadVars *tv)
+static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *tv)
 {
-    return PacketPoolGetPacket();
+    Packet *p = PacketPoolGetPacket();
+    PacketQueueNoLock pq = { .top = p, .bot = p, .len = (p != NULL) };
+    return pq;
 }
 
 void TmqhOutputPacketpool(ThreadVars *t, Packet *p)

src/tmqh-packetpool.h

@@ -72,7 +72,6 @@ typedef struct PktPool_ {
     PktPoolLockedStack return_stack;
 } PktPool;
 
-Packet *TmqhInputPacketpool(ThreadVars *);
 void TmqhOutputPacketpool(ThreadVars *, Packet *);
 void TmqhReleasePacketsToPacketPool(PacketQueue *);
 void TmqhPacketpoolRegister(void);

src/tmqh-simple.c

@@ -32,9 +32,9 @@
 #include "tm-queuehandlers.h"
 #include "tmqh-simple.h"
 
-Packet *TmqhInputSimple(ThreadVars *t);
+static PacketQueueNoLock TmqhInputSimple(ThreadVars *t);
-void TmqhOutputSimple(ThreadVars *t, Packet *p);
+static void TmqhOutputSimple(ThreadVars *t, Packet *p);
-void TmqhInputSimpleShutdownHandler(ThreadVars *);
+static void TmqhInputSimpleShutdownHandler(ThreadVars *);
 
 void TmqhSimpleRegister (void)
 {
@@ -44,31 +44,24 @@ void TmqhSimpleRegister (void)
     tmqh_table[TMQH_SIMPLE].OutHandler = TmqhOutputSimple;
 }
 
-Packet *TmqhInputSimple(ThreadVars *t)
+PacketQueueNoLock TmqhInputSimple(ThreadVars *tv)
 {
-    PacketQueue *q = t->inq->pq;
+    StatsSyncCountersIfSignalled(&tv->stats);
-
-    StatsSyncCountersIfSignalled(&t->stats);
 
+    PacketQueue *q = tv->inq->pq;
     SCMutexLock(&q->mutex_q);
-
     if (q->len == 0) {
         /* if we have no packets in queue, wait... */
         SCCondWait(&q->cond_q, &q->mutex_q);
     }
-
+    PacketQueueNoLock pq = { .top = q->top, .bot = q->bot, q->len };
-    if (q->len > 0) {
+    q->bot = q->top = NULL;
-        Packet *p = PacketDequeue(q);
+    q->len = 0;
-        SCMutexUnlock(&q->mutex_q);
+    SCMutexUnlock(&q->mutex_q);
-        return p;
+    return pq;
-    } else {
-        /* return NULL if we have no pkt. Should only happen on signals. */
-        SCMutexUnlock(&q->mutex_q);
-        return NULL;
-    }
 }
 
-void TmqhInputSimpleShutdownHandler(ThreadVars *tv)
+static void TmqhInputSimpleShutdownHandler(ThreadVars *tv)
 {
     int i;
 
@@ -83,7 +76,7 @@ void TmqhInputSimpleShutdownHandler(ThreadVars *tv)
     }
 }
 
-void TmqhOutputSimple(ThreadVars *t, Packet *p)
+static void TmqhOutputSimple(ThreadVars *t, Packet *p)
 {
     SCLogDebug("Packet %p, p->root %p, alloced %s", p, p->root, BOOL2STR(p->pool == NULL));