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.

diff --git a/src/detect-engine.c b/src/detect-engine.c
index 753593826020d4ccea6bf2a3530ebb9d578c942d..70afeda38c16b868dfdff5bc74ed65e10527d7b1 100644 (file)
--- a/src/detect-engine.c
+++ b/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;

diff --git a/src/source-af-packet.c b/src/source-af-packet.c
index 3267b6c17248eae3c58a7ccf913afad379a8cf7f..942391dd4f0209060e30882221ef4eda80dc8de7 100644 (file)
--- a/src/source-af-packet.c
+++ b/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,

diff --git a/src/source-pcap.c b/src/source-pcap.c
index 7c7a53e8c9a5442e617d9e0b6def97bfa1c82990..7b9a6161d325912efaaac45806de3ea2ec84db0c 100644 (file)
--- a/src/source-pcap.c
+++ b/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);

diff --git a/src/source-pfring.c b/src/source-pfring.c
index 9b2db3841fd95cb5e89f19d18edb40b3fe91dffb..655031fb71f867801c590a81beeb610521fb2f2b 100644 (file)
--- a/src/source-pfring.c
+++ b/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);

diff --git a/src/threadvars.h b/src/threadvars.h
index 79a2b34dc4fd31fc6c2fa09c96e1b3b2b06dcde4..578770c1aaac4693af76002534f5dac6f22c0bec 100644 (file)
--- a/src/threadvars.h
+++ b/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 */

diff --git a/src/tm-threads.h b/src/tm-threads.h
index 2127a73abdb89811979033aba1d9c1a972383057..b39bf02f432c112889a5198093967aeff3e869c1 100644 (file)
--- a/src/tm-threads.h
+++ b/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);