Accomodate pcap-file mode to signal flow mgr to wakeup when it exceeds a certain time interval. This let's the flow mgr keep in sync with pcap timestamp changes

src/flow-hash.c

@@ -325,7 +325,7 @@ static Flow *FlowGetNew(Packet *p) {
                             "(ts.tv_sec: %"PRIuMAX", ts.tv_usec:%"PRIuMAX")",
                             (uintmax_t)p->ts.tv_sec, (uintmax_t)p->ts.tv_usec);
                     flow_flags |= FLOW_EMERGENCY; /* XXX mutex this */
-                    SCCondSignal(&flow_manager_cond);
+                    FlowWakeupFlowManagerThread();
                 }
                 SCLogDebug("We need to prune some flows with emerg bit (2)");
 

src/flow-manager.h

@@ -27,6 +27,8 @@
 SCCondT flow_manager_cond;
 SCMutex flow_manager_mutex;
 
+#define FlowWakeupFlowManagerThread() SCCondSignal(&flow_manager_cond)
+
 void FlowManagerThreadSpawn(void);
 void FlowKillFlowManagerThread(void);
 

src/source-pcap-file.c

@@ -45,6 +45,7 @@
 #include "tmqh-packetpool.h"
 #include "tm-threads.h"
 #include "util-optimize.h"
+#include "flow-manager.h"
 
 extern uint8_t suricata_ctl_flags;
 extern int max_pending_packets;
@@ -283,6 +284,8 @@ TmEcode ReceivePcapFileThreadDeinit(ThreadVars *tv, void *data) {
     SCReturnInt(TM_ECODE_OK);
 }
 
+double prev_signaled_ts = 0;
+
 TmEcode DecodePcapFile(ThreadVars *tv, Packet *p, void *data, PacketQueue *pq, PacketQueue *postpq)
 {
     SCEnter();
@@ -301,6 +304,12 @@ TmEcode DecodePcapFile(ThreadVars *tv, Packet *p, void *data, PacketQueue *pq, P
     SCPerfCounterAddUI64(dtv->counter_avg_pkt_size, tv->sc_perf_pca, GET_PKT_LEN(p));
     SCPerfCounterSetUI64(dtv->counter_max_pkt_size, tv->sc_perf_pca, GET_PKT_LEN(p));
 
+    double curr_ts = p->ts.tv_sec + p->ts.tv_usec / 1000.0;
+    if (curr_ts < prev_signaled_ts || (curr_ts - prev_signaled_ts) > 2.0) {
+        prev_signaled_ts = curr_ts;
+        FlowWakeupFlowManagerThread();
+    }
+
     /* update the engine time representation based on the timestamp
      * of the packet. */
     TimeSet(&p->ts);