tcp reuse: enable stream handling based on runmode

Add a way for runmodes to state that flow and stream run asynchorously.

In the stream engine, enable the TCP reuse handling only if that flag
is set.

src/flow-hash.c

@@ -410,7 +410,8 @@ static inline int FlowCompare(Flow *f, const Packet *p)
             /* okay, we need to setup a new flow for this packet.
              * Flag the flow that it's been replaced by a new one */
             f->flags |= FLOW_TCP_REUSED;
-            SCLogDebug("flow obsolete: TCP reuse will use a new flow");
+            SCLogDebug("flow obsolete: TCP reuse will use a new flow "
+                    "starting with packet %"PRIu64, p->pcap_cnt);
             return 0;
         }
         return 1;

src/runmode-erf-file.c

@@ -137,6 +137,7 @@ int RunModeErfFileAutoFp(DetectEngineCtx *de_ctx)
     int thread;
 
     RunModeInitialize();
+    RunmodeSetFlowStreamAsync();
 
     char *file = NULL;
     if (ConfGet("erf-file.file", &file) == 0) {

src/runmode-pcap-file.c

@@ -154,6 +154,7 @@ int RunModeFilePcapAutoFp(DetectEngineCtx *de_ctx)
     int thread;
 
     RunModeInitialize();
+    RunmodeSetFlowStreamAsync();
 
     char *file = NULL;
     if (ConfGet("pcap-file.file", &file) == 0) {

src/stream-tcp.c

@@ -71,6 +71,7 @@
 #include "util-profiling.h"
 #include "util-misc.h"
 #include "util-validate.h"
+#include "util-runmodes.h"
 
 #include "source-pcap-file.h"
 
@@ -4860,6 +4861,9 @@ static void TcpSessionReuseHandle(Packet *p) {
         return;
     }
 
+    SCLogDebug("steam starter packet %"PRIu64", and state "
+            "ready to be reused", p->pcap_cnt);
+
     /* ok, this packet needs a new flow */
 
     /* first, get a reference to the old flow */
@@ -4987,20 +4991,20 @@ TmEcode StreamTcp (ThreadVars *tv, Packet *p, void *data, PacketQueue *pq, Packe
         p->flags |= PKT_IGNORE_CHECKSUM;
     }
 
-// TODO autofp only somehow
-    /* "autofp" handling of TCP session/flow reuse */
-    if (!(p->flags & PKT_PSEUDO_STREAM_END)) {
-        /* apply previous reuses to this packet */
-        TcpSessionReuseHandleApplyToPacket(p);
-        if (p->flow == NULL)
-            return ret;
+    if (stt->runmode_flow_stream_async) {
+        /* "autofp" handling of TCP session/flow reuse */
+        if (!(p->flags & PKT_PSEUDO_STREAM_END)) {
+            /* apply previous reuses to this packet */
+            TcpSessionReuseHandleApplyToPacket(p);
+            if (p->flow == NULL)
+                return ret;
 
-        /* after that, check for 'new' reuse */
-        TcpSessionReuseHandle(p);
-        if (p->flow == NULL)
-            return ret;
+            /* after that, check for 'new' reuse */
+            TcpSessionReuseHandle(p);
+            if (p->flow == NULL)
+                return ret;
+        }
     }
-
     AppLayerProfilingReset(stt->ra_ctx->app_tctx);
 
     FLOWLOCK_WRLOCK(p->flow);
@@ -5110,6 +5114,11 @@ TmEcode StreamTcpThreadInit(ThreadVars *tv, void *initdata, void **data)
     if (stt->ssn_pool_id < 0 || ssn_pool == NULL)
         SCReturnInt(TM_ECODE_FAILED);
 
+    /* see if need to enable the TCP reuse handling in the stream engine */
+    stt->runmode_flow_stream_async = RunmodeGetFlowStreamAsync();
+    SCLogDebug("Flow and Stream engine run %s",
+            stt->runmode_flow_stream_async ? "asynchronous" : "synchronous");
+
     SCReturnInt(TM_ECODE_OK);
 }
 

src/stream-tcp.h

@@ -76,6 +76,10 @@ typedef struct TcpStreamCnf_ {
 typedef struct StreamTcpThread_ {
     int ssn_pool_id;
 
+    /** if set to true, we activate the TCP tuple reuse code in the
+     *  stream engine. */
+    int runmode_flow_stream_async;
+
     uint64_t pkts;
 
     /** queue for pseudo packet(s) that were created in the stream

src/util-runmodes.c

@@ -47,6 +47,20 @@
 
 #include "util-runmodes.h"
 
+/** set to true if flow engine and stream engine run in different
+ *  threads. */
+static int runmode_flow_stream_async = 0;
+
+void RunmodeSetFlowStreamAsync(void)
+{
+    runmode_flow_stream_async = 1;
+}
+
+int RunmodeGetFlowStreamAsync(void)
+{
+    return runmode_flow_stream_async;
+}
+
 /** \brief create a queue string for autofp to pass to
  *         the flow queue handler.
  *
@@ -103,6 +117,8 @@ int RunModeSetLiveCaptureAutoFp(DetectEngineCtx *de_ctx,
     if (thread_max < 1)
         thread_max = 1;
 
+    RunmodeSetFlowStreamAsync();
+
     queues = RunmodeAutoFpCreatePickupQueuesString(thread_max);
     if (queues == NULL) {
         SCLogError(SC_ERR_RUNMODE, "RunmodeAutoFpCreatePickupQueuesString failed");
@@ -493,6 +509,8 @@ int RunModeSetIPSAutoFp(DetectEngineCtx *de_ctx,
     if (thread_max < 1)
         thread_max = 1;
 
+    RunmodeSetFlowStreamAsync();
+
     queues = RunmodeAutoFpCreatePickupQueuesString(thread_max);
     if (queues == NULL) {
         SCLogError(SC_ERR_RUNMODE, "RunmodeAutoFpCreatePickupQueuesString failed");

src/util-runmodes.h

@@ -23,6 +23,8 @@
 #ifndef __UTIL_RUNMODES_H__
 #define __UTIL_RUNMODES_H__
 
+void RunmodeSetFlowStreamAsync(void);
+int RunmodeGetFlowStreamAsync(void);
 
 typedef void *(*ConfigIfaceParserFunc) (const char *);
 typedef void *(*ConfigIPSParserFunc) (int);