threading: use tm_flowworker for pseudo packets

Pseudo packets don't need to be processed by the decoding layer.

diff --git a/src/source-af-packet.c b/src/source-af-packet.c
index ba2a2516041281ee4a1acaa76516db855fad4c8f..3bc03fe57c36a000d035254c211470dc7a4805ec 100644 (file)
--- a/src/source-af-packet.c
+++ b/src/source-af-packet.c
@@ -1617,7 +1617,7 @@ TmEcode ReceiveAFPLoop(ThreadVars *tv, void *data, void *slot)
                 last_dump = current_time;
             }
             /* poll timed out, lets see handle our timeout path */
-            TmThreadsCaptureHandleTimeout(tv, ptv->slot, NULL);
+            TmThreadsCaptureHandleTimeout(tv, NULL);
 
         } else if ((r < 0) && (errno != EINTR)) {
             SCLogError(SC_ERR_AFP_READ, "Error reading data from iface '%s': (%d) %s",

diff --git a/src/source-netmap.c b/src/source-netmap.c
index 579ad1295b373459f57b00a766958d13b43e0640..210c1ee81c7c0a6f032e1b0786d8f60d22b4c989 100644 (file)
--- a/src/source-netmap.c
+++ b/src/source-netmap.c
@@ -640,7 +640,7 @@ static TmEcode ReceiveNetmapLoop(ThreadVars *tv, void *data, void *slot)
             StatsSyncCountersIfSignalled(tv);
 
             /* poll timed out, lets handle the timeout */
-            TmThreadsCaptureHandleTimeout(tv, ntv->slot, NULL);
+            TmThreadsCaptureHandleTimeout(tv, NULL);
             continue;
         }
 

diff --git a/src/source-nfq.c b/src/source-nfq.c
index d5859be2f8279a5f087f2031e56c9098cac7ca53..b5a143decc734836a49314f6f1641a5b28eaa449 100644 (file)
--- a/src/source-nfq.c
+++ b/src/source-nfq.c
@@ -980,7 +980,7 @@ static void NFQRecvPkt(NFQQueueVars *t, NFQThreadVars *tv)
                 NFQVerdictCacheFlush(t);
 
             /* handle timeout */
-            TmThreadsCaptureHandleTimeout(tv->tv, tv->slot, NULL);
+            TmThreadsCaptureHandleTimeout(tv->tv, NULL);
         } else {
 #ifdef COUNTERS
             NFQMutexLock(t);

diff --git a/src/source-pcap.c b/src/source-pcap.c
index 9ec461508c94de418548fceba9e1f9dcdc2b65fc..a7f9d38ee5c081aa1369eefc6386096cd777106a 100644 (file)
--- a/src/source-pcap.c
+++ b/src/source-pcap.c
@@ -266,7 +266,7 @@ TmEcode ReceivePcapLoop(ThreadVars *tv, void *data, void *slot)
             if (r == PCAP_ERROR_BREAK && ptv->cb_result == TM_ECODE_FAILED) {
                 SCReturnInt(TM_ECODE_FAILED);
             }
-            TmThreadsCaptureHandleTimeout(tv, ptv->slot, NULL);
+            TmThreadsCaptureHandleTimeout(tv, NULL);
         } else if (unlikely(r < 0)) {
             int dbreak = 0;
             SCLogError(SC_ERR_PCAP_DISPATCH, "error code %" PRId32 " %s",

diff --git a/src/source-pfring.c b/src/source-pfring.c
index 82a0bb85eca9a3dbe573c7d0debf05e47f8aea54..2f5487b45fadb7e2845f3d71b1456b37eb386a8b 100644 (file)
--- a/src/source-pfring.c
+++ b/src/source-pfring.c
@@ -434,7 +434,7 @@ TmEcode ReceivePfringLoop(ThreadVars *tv, void *data, void *slot)
             }
 
             /* pfring didn't use the packet yet */
-            TmThreadsCaptureHandleTimeout(tv, ptv->slot, p);
+            TmThreadsCaptureHandleTimeout(tv, p);
 
         } else {
             SCLogError(SC_ERR_PF_RING_RECV,"pfring_recv error  %" PRId32 "", r);

diff --git a/src/tm-threads.c b/src/tm-threads.c
index b592950a9b73520918dac9b4ec37b07ed522dbc4..19016e3642b471910aec3f691a83042427d69465 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -588,7 +588,7 @@ static void *TmThreadsSlotVar(void *td)
             tv->tmqh_out(tv, p);
 
             /* now handle the stream pq packets */
-            TmThreadsHandleInjectedPackets(tv, s);
+            TmThreadsHandleInjectedPackets(tv);
         }
 
         if (TmThreadsCheckFlag(tv, THV_KILL)) {

diff --git a/src/tm-threads.h b/src/tm-threads.h
index 16130b1753583650b74570507c486de8c81a5bf2..6ce9c5e07be1256e4fde23eb0bbf21252a352599 100644 (file)
--- a/src/tm-threads.h
+++ b/src/tm-threads.h
@@ -162,7 +162,7 @@ static inline void TmThreadsSlotProcessPktFail(ThreadVars *tv, TmSlot *s, Packet
  *         manager.
  *  \param s pipeline to run on these packets.
  */
-static inline void TmThreadsHandleInjectedPackets(ThreadVars *tv, TmSlot *s)
+static inline void TmThreadsHandleInjectedPackets(ThreadVars *tv)
 {
     PacketQueue *pq = tv->stream_pq_local;
     if (pq && pq->len > 0) {
@@ -172,9 +172,9 @@ static inline void TmThreadsHandleInjectedPackets(ThreadVars *tv, TmSlot *s)
             SCMutexUnlock(&pq->mutex_q);
             if (extra_p == NULL)
                 break;
-            TmEcode r = TmThreadsSlotVarRun(tv, extra_p, s);
+            TmEcode r = TmThreadsSlotVarRun(tv, extra_p, tv->tm_flowworker);
             if (r == TM_ECODE_FAILED) {
-                TmThreadsSlotProcessPktFail(tv, s, extra_p);
+                TmThreadsSlotProcessPktFail(tv, tv->tm_flowworker, extra_p);
                 break;
             }
             tv->tmqh_out(tv, extra_p);
@@ -200,7 +200,7 @@ static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet
 
     tv->tmqh_out(tv, p);
 
-    TmThreadsHandleInjectedPackets(tv, s);
+    TmThreadsHandleInjectedPackets(tv);
 
     return TM_ECODE_OK;
 }
@@ -210,7 +210,7 @@ static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet
  *
  *  Meant for detect reload process that interupts an sleeping capture thread
  *  to force a packet through the engine to complete a reload */
-static inline void TmThreadsCaptureInjectPacket(ThreadVars *tv, TmSlot *slot, Packet *p)
+static inline void TmThreadsCaptureInjectPacket(ThreadVars *tv, Packet *p)
 {
     TmThreadsUnsetFlag(tv, THV_CAPTURE_INJECT_PKT);
     if (p == NULL)
@@ -218,18 +218,18 @@ static inline void TmThreadsCaptureInjectPacket(ThreadVars *tv, TmSlot *slot, Pa
     if (p != NULL) {
         p->flags |= PKT_PSEUDO_STREAM_END;
         PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT);
-        if (TmThreadsSlotProcessPkt(tv, slot, p) != TM_ECODE_OK) {
+        if (TmThreadsSlotProcessPkt(tv, tv->tm_flowworker, p) != TM_ECODE_OK) {
             TmqhOutputPacketpool(tv, p);
         }
     }
 }
 
-static inline void TmThreadsCaptureHandleTimeout(ThreadVars *tv, TmSlot *slot, Packet *p)
+static inline void TmThreadsCaptureHandleTimeout(ThreadVars *tv, Packet *p)
 {
     if (TmThreadsCheckFlag(tv, THV_CAPTURE_INJECT_PKT)) {
-        TmThreadsCaptureInjectPacket(tv, slot, p);
+        TmThreadsCaptureInjectPacket(tv, p);
     } else {
-        TmThreadsHandleInjectedPackets(tv, slot);
+        TmThreadsHandleInjectedPackets(tv);
 
         /* packet could have been passed to us that we won't use
          * return it to the pool. */