Add pcap workers mode.

Some cards like Napatech or Myricom support libpcap wrappers that allow for
multiple streams, queues, ringbuffers. The workers mode can be of use in
those cases.

src/runmode-pcap.c

@@ -1,4 +1,4 @@
-/* Copyright (C) 2007-2010 Open Information Security Foundation
+/* Copyright (C) 2007-2012 Open Information Security Foundation
  *
  * You can copy, redistribute or modify this Program under the terms of
  * the GNU General Public License version 2 as published by the Free
@@ -38,6 +38,7 @@
 #include "util-device.h"
 #include "util-runmodes.h"
 #include "util-atomic.h"
+#include "util-misc.h"
 
 static const char *default_mode = NULL;
 
@@ -46,6 +47,8 @@ const char *RunModeIdsGetDefaultMode(void)
     return default_mode;
 }
 
+int RunModeIdsPcapWorkers(DetectEngineCtx *de_ctx);
+
 void RunModeIdsPcapRegister(void)
 {
     RunModeRegisterNewRunMode(RUNMODE_PCAP_DEV, "single",
@@ -62,6 +65,10 @@ void RunModeIdsPcapRegister(void)
                               "the same flow can be processed by any detect "
                               "thread",
                               RunModeIdsPcapAutoFp);
+    RunModeRegisterNewRunMode(RUNMODE_PCAP_DEV, "workers",
+                              "Workers pcap live mode, each thread does all"
+                              " tasks from acquisition to logging",
+                              RunModeIdsPcapWorkers);
 
     return;
 }
@@ -78,6 +85,7 @@ void PcapDerefConfig(void *conf)
 
 void *ParsePcapConfig(const char *iface)
 {
+    char *threadsstr = NULL;
     ConfNode *if_root;
     ConfNode *pcap_node;
     PcapIfaceConfig *aconf = SCMalloc(sizeof(*aconf));
@@ -128,11 +136,38 @@ void *ParsePcapConfig(const char *iface)
         return aconf;
     }
 
+    if (ConfGetChildValue(if_root, "threads", &threadsstr) != 1) {
+        aconf->threads = 1;
+    } else {
+        if (threadsstr != NULL) {
+            aconf->threads = (uint8_t)atoi(threadsstr);
+        }
+    }
+    if (aconf->threads == 0) {
+        aconf->threads = 1;
+    }
     if (aconf->buffer_size == 0) {
-        if ((ConfGetChildValueInt(if_root, "buffer-size", &value)) == 1) {
-            aconf->buffer_size = value;
-            SCLogInfo("Pcap will use %d buffer size (config file provided "
-                    "value)", (int)value);
+        const char *s_limit = ConfNodeLookupChildValue(if_root, "buffer-size");
+        if (s_limit != NULL) {
+            uint64_t bsize = 0;
+
+            if (ParseSizeStringU64(s_limit, &bsize) < 0) {
+                SCLogError(SC_ERR_INVALID_ARGUMENT,
+                    "Failed to parse pcap buffer size: %s",
+                    s_limit);
+            } else {
+                /* the string 2gb returns 2147483648 which is 1 to high
+                 * for a int. */
+                if (bsize == (uint64_t)((uint64_t)INT_MAX + (uint64_t)1))
+                    bsize = (uint64_t)INT_MAX;
+
+                if (bsize > INT_MAX) {
+                    SCLogError(SC_ERR_INVALID_ARGUMENT,
+                            "Failed to set pcap buffer size: 2gb max. %"PRIu64" > %d", bsize, INT_MAX);
+                } else {
+                    aconf->buffer_size = (int)bsize;
+                }
+            }
         }
     }
 
@@ -164,7 +199,8 @@ void *ParsePcapConfig(const char *iface)
 
 int PcapConfigGeThreadsCount(void *conf)
 {
-    return 1;
+    PcapIfaceConfig *pfp = (PcapIfaceConfig *)conf;
+    return pfp->threads;
 }
 
 /**
@@ -289,3 +325,36 @@ int RunModeIdsPcapAutoFp(DetectEngineCtx *de_ctx)
 
     SCReturnInt(0);
 }
+
+/**
+ * \brief Workers version of the PCAP LIVE processing.
+ *
+ * Start N threads with each thread doing all the work.
+ *
+ */
+int RunModeIdsPcapWorkers(DetectEngineCtx *de_ctx)
+{
+    int ret;
+    char *live_dev = NULL;
+    SCEnter();
+
+    RunModeInitialize();
+    TimeModeSetLive();
+
+    ConfGet("pcap.single-pcap-dev", &live_dev);
+
+    ret = RunModeSetLiveCaptureWorkers(de_ctx,
+                                    ParsePcapConfig,
+                                    PcapConfigGeThreadsCount,
+                                    "ReceivePcap",
+                                    "DecodePcap", "RxPcap",
+                                    live_dev);
+    if (ret != 0) {
+        SCLogError(SC_ERR_RUNMODE, "Unable to start runmode");
+        exit(EXIT_FAILURE);
+    }
+
+    SCLogInfo("RunModeIdsPcapWorkers initialised");
+
+    SCReturnInt(0);
+}

src/source-pcap.h

@@ -49,6 +49,8 @@ typedef struct PcapPacketVars_
 typedef struct PcapIfaceConfig_
 {
     char iface[PCAP_IFACE_NAME_LENGTH];
+    /* number of threads */
+    int threads;
     /* socket buffer size */
     int buffer_size;
     /* BPF filter */