Go to the documentation of this file.
84 static uint32_t flowmgr_number = 1;
89 static uint32_t flowrec_number = 1;
111 #define FLOW_NORMAL_MODE_UPDATE_DELAY_SEC 1
112 #define FLOW_NORMAL_MODE_UPDATE_DELAY_NSEC 0
114 #define FLOW_EMERG_MODE_UPDATE_DELAY_SEC 0
115 #define FLOW_EMERG_MODE_UPDATE_DELAY_NSEC 300000
116 #define NEW_FLOW_COUNT_COND 10
155 struct timeval start_ts;
156 struct timeval cur_ts;
157 gettimeofday(&start_ts, NULL);
160 gettimeofday(&cur_ts, NULL);
161 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
163 "threads to shutdown in time");
195 static int FlowManagerFlowTimeout(
Flow *f,
SCTime_t ts, uint32_t *next_ts,
const bool emerg)
202 if (*next_ts == 0 || flow_times_out_at < *next_ts)
203 *next_ts = flow_times_out_at;
225 #ifdef CAPTURE_OFFLOAD
226 if (f->
flow_state != FLOW_STATE_CAPTURE_BYPASSED) {
239 SCLogDebug(
"Updated flow: %"PRId64
"", FlowGetId(f));
246 pkts_tosrc + pkts_todst);
248 counters->bypassed_pkts += pkts_tosrc + pkts_todst;
249 counters->bypassed_bytes += bytes_tosrc + bytes_todst;
252 SCLogDebug(
"No new packet, dead flow %"PRId64
"", FlowGetId(f));
260 counters->bypassed_count++;
297 if (recycle.
len == 100) {
323 uint32_t checked = 0;
335 if (FlowManagerFlowTimeout(f,
ts, next_ts, emergency) == 0) {
337 counters->flows_notimeout++;
350 if (!FlowBypassedTimeout(f,
ts, counters)) {
359 counters->flows_timeout++;
361 RemoveFromHash(f, prev_f);
369 counters->flows_checked += checked;
370 if (checked > counters->rows_maxlen)
371 counters->rows_maxlen = checked;
374 static void FlowManagerHashRowClearEvictedList(
405 const uint32_t rows_checked = hash_max - hash_min;
406 uint32_t rows_skipped = 0;
407 uint32_t rows_empty = 0;
411 #define TYPE uint64_t
414 #define TYPE uint32_t
418 for (uint32_t idx = hash_min; idx < hash_max; idx+=
BITS) {
420 const uint32_t check =
MIN(
BITS, (hash_max - idx));
421 for (uint32_t i = 0; i < check; i++) {
430 for (uint32_t i = 0; i < check; i++) {
435 if (fb->evicted != NULL || fb->head != NULL) {
436 if (fb->evicted != NULL) {
442 if (fb->head != NULL) {
443 uint32_t next_ts = 0;
444 FlowManagerHashRowTimeout(td, fb->head,
ts, emergency, counters, &next_ts);
449 if (fb->evicted == NULL && fb->head == NULL) {
459 FlowManagerHashRowClearEvictedList(td,
evicted,
ts, counters);
466 cnt += ProcessAsideQueue(td, counters);
470 counters->rows_checked += rows_checked;
471 counters->rows_skipped += rows_skipped;
472 counters->rows_empty += rows_empty;
475 cnt += ProcessAsideQueue(td, counters);
477 counters->flows_removed += cnt;
487 const uint32_t rows, uint32_t *pos)
492 uint32_t rows_left = rows;
495 start = hash_min + (*pos);
496 if (start >= hash_max) {
499 end = start + rows_left;
500 if (end > hash_max) {
503 *pos = (end == hash_max) ? hash_min : end;
504 rows_left = rows_left - (end - start);
506 cnt += FlowTimeoutHash(td,
ts, start, end, counters);
533 RemoveFromHash(f, NULL);
535 FlowBucket *fb = f->
fb;
560 static uint32_t FlowCleanupHash(
void)
570 if (fb->head != NULL) {
572 cnt += FlowManagerHashRowCleanup(fb->head, &local_queue, 0);
574 if (fb->evicted != NULL) {
576 cnt += FlowManagerHashRowCleanup(fb->evicted, &local_queue, 1);
580 if (local_queue.
len >= 25) {
654 static void FlowCountersUpdate(
672 static TmEcode FlowManagerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
689 if ((ftd->
instance + 1) == flowmgr_number) {
699 FlowCountersInit(t, &ftd->
cnt);
727 static void GetWorkUnitSizing(
const uint32_t rows,
const uint32_t mp,
const bool emergency,
728 uint64_t *wu_sleep, uint32_t *wu_rows, uint32_t *rows_sec)
736 const uint32_t emp =
MAX(mp, 10);
737 const uint32_t rows_per_sec = (uint32_t)((
float)rows * (float)((
float)emp / (float)100));
740 const uint32_t work_per_unit =
MIN(rows_per_sec / 1000, 1000);
743 const uint32_t sleep_per_unit =
MAX(250, 1000 - work_per_unit);
744 SCLogDebug(
"mp %u emp %u rows %u rows_sec %u sleep %ums", mp, emp, rows, rows_per_sec,
747 *wu_sleep = sleep_per_unit;
748 *wu_rows = rows_per_sec;
749 *rows_sec = rows_per_sec;
761 const uint32_t rows = ftd->
max - ftd->
min;
764 uint32_t emerg_over_cnt = 0;
765 uint64_t next_run_ms = 0;
767 uint32_t rows_sec = 0;
768 uint32_t rows_per_wu = 0;
769 uint64_t sleep_per_wu = 0;
771 bool prev_emerg =
false;
772 uint32_t other_last_sec = 0;
787 GetWorkUnitSizing(rows, mp, emerg, &sleep_per_wu, &rows_per_wu, &rows_sec);
807 const bool emerge_p = (emerg && !prev_emerg);
814 if (ts_ms >= next_run_ms) {
819 if (spare_perc < 90 || spare_perc > 110) {
826 FlowTimeoutCounters counters = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, };
834 SCLogDebug(
"hash %u:%u slice starting at %u with %u rows", ftd->
min, ftd->
max, pos,
837 const uint32_t ppos = pos;
838 FlowTimeoutHashInChunks(
848 FlowCountersUpdate(th_v, ftd, &counters);
851 SCLogDebug(
"flow_sparse_q.len = %" PRIu32
" prealloc: %" PRIu32
852 "flow_spare_q status: %" PRIu32
"%% flows at the queue",
864 if (emerg_over_cnt >= 30) {
871 SCLogNotice(
"Flow emergency mode over, back to normal... unsetting"
872 " FLOW_EMERGENCY bit (ts.tv_sec: %" PRIuMAX
", "
873 "ts.tv_usec:%" PRIuMAX
") flow_spare_q status(): %" PRIu32
874 "%% flows at the queue",
883 const uint32_t pmp = mp;
889 GetWorkUnitSizing(rows, mp, emerg, &sleep_per_wu, &rows_per_wu, &rows_sec);
894 next_run_ms = ts_ms + sleep_per_wu;
896 if (other_last_sec == 0 || other_last_sec < (uint32_t)
SCTIME_SECS(
ts)) {
911 if (emerg || !time_is_live) {
914 struct timeval cond_tv;
915 gettimeofday(&cond_tv, NULL);
916 struct timeval add_tv;
918 add_tv.tv_usec = (sleep_per_wu * 1000);
919 timeradd(&cond_tv, &add_tv, &cond_tv);
926 if (rc == ETIMEDOUT || rc < 0)
945 intmax_t setting = 1;
948 if (setting < 1 || setting > 1024) {
949 FatalError(
"invalid flow.managers setting %" PRIdMAX, setting);
951 flowmgr_number = (uint32_t)setting;
956 SCLogConfig(
"using %u flow manager threads", flowmgr_number);
959 for (uint32_t u = 0; u < flowmgr_number; u++) {
965 BUG_ON(tv_flowmgr == NULL);
967 if (tv_flowmgr == NULL) {
968 FatalError(
"flow manager thread creation failed");
971 FatalError(
"flow manager thread spawn failed");
989 static TmEcode FlowRecyclerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
995 SCLogError(
"initializing flow log API for thread failed");
1032 FlowEndCountersUpdate(
tv, &ftd->
fec, f);
1052 uint64_t recycled_cnt = 0;
1076 Recycler(th_v, ftd, f);
1087 if (emerg || !time_is_live) {
1090 struct timeval cond_tv;
1091 gettimeofday(&cond_tv, NULL);
1092 cond_tv.tv_sec += 1;
1098 if (rc == ETIMEDOUT || rc < 0) {
1116 SCLogPerf(
"%"PRIu64
" flows processed", recycled_cnt);
1120 static bool FlowRecyclerReadyToShutdown(
void)
1130 return ((
len == 0));
1136 intmax_t setting = 1;
1137 (void)
ConfGetInt(
"flow.recyclers", &setting);
1139 if (setting < 1 || setting > 1024) {
1140 FatalError(
"invalid flow.recyclers setting %" PRIdMAX, setting);
1142 flowrec_number = (uint32_t)setting;
1147 SCLogConfig(
"using %u flow recycler threads", flowrec_number);
1149 for (uint32_t u = 0; u < flowrec_number; u++) {
1156 if (tv_flowrec == NULL) {
1157 FatalError(
"flow recycler thread creation failed");
1160 FatalError(
"flow recycler thread spawn failed");
1178 (void)FlowCleanupHash();
1180 uint32_t flows = FlowCleanupHash();
1188 }
while (FlowRecyclerReadyToShutdown() ==
false);
1201 struct timeval start_ts;
1202 struct timeval cur_ts;
1203 gettimeofday(&start_ts, NULL);
1206 gettimeofday(&cur_ts, NULL);
1207 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1208 FatalError(
"unable to get all flow recycler "
1209 "threads to shutdown in time");
void FlowSparePoolUpdate(uint32_t size)
void FlowManagerThreadSpawn(void)
spawn the flow manager thread
int ConfGetInt(const char *name, intmax_t *val)
Retrieve a configuration value as an integer.
#define FROM_TIMEVAL(timev)
initialize a 'struct timespec' from a 'struct timeval'.
TmEcode OutputFlowLog(ThreadVars *tv, void *thread_data, Flow *f)
Run flow logger(s)
SCCtrlMutex flow_manager_ctrl_mutex
TmEcode TmThreadSpawn(ThreadVars *tv)
Spawns a thread associated with the ThreadVars instance tv.
void FlowForceReassemblyForFlow(Flow *f)
ThreadVars * TmThreadCreateMgmtThreadByName(const char *name, const char *module, int mucond)
Creates and returns the TV instance for a Management thread(MGMT). This function supports only custom...
void StatsIncr(ThreadVars *tv, uint16_t id)
Increments the local counter.
void FlowSparePoolReturnFlow(Flow *f)
const char * thread_name_flow_mgr
uint32_t FlowSpareGetPoolSize(void)
#define SC_ATOMIC_INIT(name)
wrapper for initializing an atomic variable.
#define FlowWakeupFlowRecyclerThread()
FlowStorageId GetFlowBypassInfoID(void)
uint32_t emergency_recovery
#define SC_ATOMIC_SET(name, val)
Set the value for the atomic variable.
uint32_t IPPairTimeoutHash(SCTime_t ts)
time out ippairs from the hash
void TmThreadsSetFlag(ThreadVars *tv, uint32_t flag)
Set a thread flag.
uint16_t StatsRegisterGlobalCounter(const char *name, uint64_t(*Func)(void))
Registers a counter, which represents a global value.
SC_ATOMIC_DECLARE(uint32_t, flowmgr_cnt)
SCCtrlCondT flow_manager_ctrl_cond
uint16_t flow_mgr_flows_notimeout
SCCtrlCondT flow_recycler_ctrl_cond
SC_ATOMIC_EXTERN(unsigned int, flow_flags)
#define SC_ATOMIC_ADD(name, val)
add a value to our atomic variable
const char * thread_name_flow_rec
void StatsSetUI64(ThreadVars *tv, uint16_t id, uint64_t x)
Sets a value of type double to the local counter.
void TmModuleFlowRecyclerRegister(void)
uint64_t FlowGetMemuse(void)
ThreadVars * tv_root[TVT_MAX]
#define StatsSyncCountersIfSignalled(tv)
void LiveDevSubBypassStats(LiveDevice *dev, uint64_t cnt, int family)
struct FlowManagerTimeoutThread FlowManagerTimeoutThread
uint16_t flow_mgr_flows_aside
#define FLOW_TIMEOUT_REASSEMBLY_DONE
#define TM_THREAD_NAME_MAX
#define FLOWLOCK_UNLOCK(fb)
void PacketPoolInit(void)
void FlowDisableFlowRecyclerThread(void)
Used to disable flow recycler thread(s).
TmEcode(* ThreadDeinit)(ThreadVars *, void *)
uint16_t flow_bypassed_bytes
void TmThreadsUnsetFlag(ThreadVars *tv, uint32_t flag)
Unset a thread flag.
void StatsDecr(ThreadVars *tv, uint16_t id)
Decrements the local counter.
uint16_t flow_mgr_full_pass
uint16_t StatsRegisterMaxCounter(const char *name, struct ThreadVars_ *tv)
Registers a counter, whose value holds the maximum of all the values assigned to it.
#define SC_ATOMIC_MEMORY_ORDER_RELAXED
bool(* BypassUpdate)(Flow *f, void *data, time_t tsec)
uint16_t counter_queue_avg
TmEcode OutputFlowLogThreadInit(ThreadVars *tv, void *initdata, void **data)
thread init for the flow logger This will run the thread init functions for the individual registered...
#define SCMutexUnlock(mut)
uint16_t flow_mgr_rows_maxlen
#define FLOWLOCK_WRLOCK(fb)
#define FlowTimeoutsReset()
int FlowForceReassemblyNeedReassembly(Flow *f)
Check if a flow needs forced reassembly, or any other processing.
void FlowDisableFlowManagerThread(void)
Used to disable flow manager thread(s).
Per thread variable structure.
void TmThreadTestThreadUnPaused(ThreadVars *tv)
Tests if the thread represented in the arg has been unpaused or not.
TmEcode(* Management)(ThreadVars *, void *)
bool TimeModeIsReady(void)
float MemcapsGetPressure(void)
Flow * FlowQueuePrivateGetFromTop(FlowQueuePrivate *fqc)
void FlowQueueAppendPrivate(FlowQueue *fq, FlowQueuePrivate *fqc)
struct FlowQueueTimeoutCounters FlowQueueTimeoutCounters
uint16_t flow_mgr_flows_aside_needs_work
struct ThreadVars_ * next
uint16_t counter_tcp_active_sessions
uint16_t flow_bypassed_cnt_clo
#define SC_ATOMIC_SUB(name, val)
sub a value from our atomic variable
FlowProtoTimeout flow_timeouts_emerg[FLOW_PROTO_MAX]
bool TimeModeIsLive(void)
#define SCCtrlMutexLock(mut)
void FlowTimeoutsEmergency(void)
struct FlowTimeoutCounters_ FlowTimeoutCounters
TmModule tmm_modules[TMM_SIZE]
void * output_thread_data
#define FBLOCK_UNLOCK(fb)
int FlowClearMemory(Flow *f, uint8_t proto_map)
Function clear the flow memory before queueing it to spare flow queue.
FlowProtoTimeout flow_timeouts_delta[FLOW_PROTO_MAX]
void FlowRecyclerThreadSpawn(void)
spawn the flow recycler thread
Data structures and function prototypes for keeping state for the detection engine.
struct FlowRecyclerThreadData_ FlowRecyclerThreadData
#define SCCtrlCondTimedwait
void StreamTcpThreadCacheCleanup(void)
uint32_t flows_aside_needs_work
void * FlowGetStorageById(const Flow *f, FlowStorageId id)
uint16_t flow_bypassed_pkts
#define SCCtrlMutexUnlock(mut)
FlowQueuePrivate FlowQueueExtractPrivate(FlowQueue *fq)
uint32_t DefragTimeoutHash(SCTime_t ts)
time out tracker from the hash
struct FlowManagerThreadData_ FlowManagerThreadData
void TmModuleFlowManagerRegister(void)
void FlowTimeoutsInit(void)
#define SC_ATOMIC_LOAD_EXPLICIT(name, order)
TmEcode(* ThreadInit)(ThreadVars *, const void *, void **)
struct LiveDevice_ * livedev
void StatsAddUI64(ThreadVars *tv, uint16_t id, uint64_t x)
Adds a value of type uint64_t to the local counter.
struct SCLogConfig_ SCLogConfig
Holds the config state used by the logging api.
uint16_t counter_queue_max
#define SCLogError(...)
Macro used to log ERROR messages.
TmEcode OutputFlowLogThreadDeinit(ThreadVars *tv, void *thread_data)
uint16_t memcap_pressure_max
uint32_t HostTimeoutHash(SCTime_t ts)
time out hosts from the hash
#define SC_ATOMIC_INITPTR(name)
FlowManagerTimeoutThread timeout
struct FlowCounters_ FlowCounters
uint16_t flow_emerg_mode_enter
#define SCCtrlMutexInit(mut, mutattr)
#define FLOW_END_FLAG_SHUTDOWN
uint16_t counter_flow_active
uint16_t flow_mgr_flows_timeout
FlowQueuePrivate aside_queue
#define timeradd(a, b, r)
#define FLOW_END_FLAG_TIMEOUT
#define StatsSyncCounters(tv)
FlowProtoTimeout flow_timeouts_normal[FLOW_PROTO_MAX]
uint16_t StatsRegisterAvgCounter(const char *name, struct ThreadVars_ *tv)
Registers a counter, whose value holds the average of all the values assigned to it.
#define SC_ATOMIC_GET(name)
Get the value from the atomic variable.
void PacketPoolDestroy(void)
void FlowQueuePrivateAppendFlow(FlowQueuePrivate *fqc, Flow *f)
void FlowEndCountersRegister(ThreadVars *t, FlowEndCounters *fec)
int TmThreadsCheckFlag(ThreadVars *tv, uint32_t flag)
Check if a thread flag is set.
#define SCLogNotice(...)
Macro used to log NOTICE messages.
uint16_t StatsRegisterCounter(const char *name, struct ThreadVars_ *tv)
Registers a normal, unqualified counter.
SCCtrlMutex flow_recycler_ctrl_mutex
#define SC_ATOMIC_AND(name, val)
Bitwise AND a value to our atomic variable.
uint16_t flow_emerg_mode_over
#define TM_FLAG_MANAGEMENT_TM
volatile uint8_t suricata_ctl_flags
uint16_t flow_mgr_flows_checked
uint16_t flow_mgr_rows_sec
uint32_t flows_aside_needs_work
uint32_t HttpRangeContainersTimeoutHash(const SCTime_t ts)