detect reload: generic packet injection for capture

Capture methods that are non blocking will still not generate packets
that go through the system if there is no traffic. Some maintenance
tasks, like rule reloads rely on packets to complete.

This patch introduces a new thread flag, THV_CAPTURE_INJECT_PKT, that
instructs the capture thread to create a fake packet.

The capture implementations can call the TmThreadsCaptureInjectPacket
utility function either with the packet they already got from the pool
or without a packet. In this case the util func will get it's own
packet.

Implementations for pcap, AF_PACKET and PF_RING.

src/detect-engine.c

@@ -560,9 +560,14 @@ static void BreakCapture(void)
                 continue;
             }
 
+            /* signal capture method that we need a packet. */
+            TmThreadsSetFlag(tv, THV_CAPTURE_INJECT_PKT);
+            /* if the method supports it, BreakLoop. Otherwise we rely on
+             * the capture method's recv timeout */
             if (tm->PktAcqLoop && tm->PktAcqBreakLoop) {
                 tm->PktAcqBreakLoop(tv, SC_ATOMIC_GET(slots->slot_data));
             }
+
             break;
         }
         tv = tv->next;

src/source-af-packet.c

@@ -1241,6 +1241,10 @@ TmEcode ReceiveAFPLoop(ThreadVars *tv, void *data, void *slot)
                     AFPDumpCounters(ptv);
                     break;
             }
+        } else if (unlikely(r == 0)) {
+            /* poll timed out, lets see if we need to inject a fake packet  */
+            TmThreadsCaptureInjectPacket(tv, ptv->slot, NULL);
+
         } else if ((r < 0) && (errno != EINTR)) {
             SCLogError(SC_ERR_AFP_READ, "Error reading data from iface '%s': (%d" PRIu32 ") %s",
                        ptv->iface,

src/source-pcap.c

@@ -339,6 +339,8 @@ TmEcode ReceivePcapLoop(ThreadVars *tv, void *data, void *slot)
         } else if (ptv->cb_result == TM_ECODE_FAILED) {
             SCLogError(SC_ERR_PCAP_DISPATCH, "Pcap callback PcapCallbackLoop failed");
             SCReturnInt(TM_ECODE_FAILED);
+        } else if (unlikely(r == 0)) {
+            TmThreadsCaptureInjectPacket(tv, ptv->slot, NULL);
         }
 
         StatsSyncCountersIfSignalled(tv);

src/source-pfring.c

@@ -370,6 +370,14 @@ TmEcode ReceivePfringLoop(ThreadVars *tv, void *data, void *slot)
                 PfringDumpCounters(ptv);
                 last_dump = p->ts.tv_sec;
             }
+        } else if (unlikely(r == 0)) {
+            if (suricata_ctl_flags & (SURICATA_STOP | SURICATA_KILL)) {
+                SCReturnInt(TM_ECODE_OK);
+            }
+
+            /* pfring didn't use the packet yet */
+            TmThreadsCaptureInjectPacket(tv, ptv->slot, p);
+
         } else {
             SCLogError(SC_ERR_PF_RING_RECV,"pfring_recv error  %" PRId32 "", r);
             TmqhOutputPacketpool(ptv->tv, p);

src/threadvars.h

@@ -48,6 +48,11 @@ struct TmSlot_;
 #define THV_KILL_PKTACQ (1 << 9)    /**< flag thread to stop packet acq */
 #define THV_FLOW_LOOP (1 << 10)   /**< thread is in flow shutdown loop */
 
+/** signal thread's capture method to create a fake packet to force through
+ *  the engine. This is to force timely handling of maintenance taks like
+ *  rule reloads even if no packets are read by the capture method. */
+#define THV_CAPTURE_INJECT_PKT (1<<11)
+
 /** Thread flags set and read by threads, to control the threads, when they
  *  encounter certain conditions like failure */
 #define THV_RESTART_THREAD 0x01 /** restart the thread */

src/tm-threads.h

@@ -197,6 +197,25 @@ static inline TmEcode TmThreadsSlotProcessPkt(ThreadVars *tv, TmSlot *s, Packet
     return r;
 }
 
+/** \brief inject packet if THV_CAPTURE_INJECT_PKT is set
+ *  Allow caller to supply their own 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)
+{
+    if (TmThreadsCheckFlag(tv, THV_CAPTURE_INJECT_PKT)) {
+        TmThreadsUnsetFlag(tv, THV_CAPTURE_INJECT_PKT);
+        if (p == NULL)
+            p = PacketGetFromQueueOrAlloc();
+        if (p != NULL) {
+            p->flags |= PKT_PSEUDO_STREAM_END;
+            if (TmThreadsSlotProcessPkt(tv, slot, p) != TM_ECODE_OK) {
+                TmqhOutputPacketpool(tv, p);
+            }
+        }
+    }
+}
 
 void TmThreadsListThreads(void);
 int TmThreadsRegisterThread(ThreadVars *tv, const int type);