64 static uint32_t flowmgr_number = 1;
69 static uint32_t flowrec_number = 1;
75 static SCCtrlCondT flow_manager_ctrl_cond = PTHREAD_COND_INITIALIZER;
76 static SCCtrlMutex flow_manager_ctrl_mutex = PTHREAD_MUTEX_INITIALIZER;
77 static SCCtrlCondT flow_recycler_ctrl_cond = PTHREAD_COND_INITIALIZER;
78 static SCCtrlMutex flow_recycler_ctrl_mutex = PTHREAD_MUTEX_INITIALIZER;
105 #define FLOW_NORMAL_MODE_UPDATE_DELAY_SEC 1
106 #define FLOW_NORMAL_MODE_UPDATE_DELAY_NSEC 0
108 #define FLOW_EMERG_MODE_UPDATE_DELAY_SEC 0
109 #define FLOW_EMERG_MODE_UPDATE_DELAY_NSEC 300000
110 #define NEW_FLOW_COUNT_COND 10
149 struct timeval start_ts;
150 struct timeval cur_ts;
151 gettimeofday(&start_ts, NULL);
154 gettimeofday(&cur_ts, NULL);
155 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
157 "threads to shutdown in time");
189 static int FlowManagerFlowTimeout(
Flow *f,
SCTime_t ts, uint32_t *next_ts,
const bool emerg)
196 if (*next_ts == 0 || flow_times_out_at < *next_ts)
197 *next_ts = flow_times_out_at;
219 #ifdef CAPTURE_OFFLOAD
220 if (f->
flow_state != FLOW_STATE_CAPTURE_BYPASSED) {
233 SCLogDebug(
"Updated flow: %"PRId64
"", FlowGetId(f));
240 pkts_tosrc + pkts_todst);
242 counters->bypassed_pkts += pkts_tosrc + pkts_todst;
243 counters->bypassed_bytes += bytes_tosrc + bytes_todst;
246 SCLogDebug(
"No new packet, dead flow %"PRId64
"", FlowGetId(f));
254 counters->bypassed_count++;
278 if (f->
proto == IPPROTO_TCP &&
292 if (recycle.
len == 100) {
318 uint32_t checked = 0;
330 if (FlowManagerFlowTimeout(f,
ts, next_ts, emergency) == 0) {
332 counters->flows_notimeout++;
345 if (!FlowBypassedTimeout(f,
ts, counters)) {
354 counters->flows_timeout++;
356 RemoveFromHash(f, prev_f);
364 counters->flows_checked += checked;
365 if (checked > counters->rows_maxlen)
366 counters->rows_maxlen = checked;
369 static void FlowManagerHashRowClearEvictedList(
400 const uint32_t rows_checked = hash_max - hash_min;
401 uint32_t rows_skipped = 0;
402 uint32_t rows_empty = 0;
406 #define TYPE uint64_t
409 #define TYPE uint32_t
413 for (uint32_t idx = hash_min; idx < hash_max; idx+=
BITS) {
415 const uint32_t check =
MIN(
BITS, (hash_max - idx));
416 for (uint32_t i = 0; i < check; i++) {
425 for (uint32_t i = 0; i < check; i++) {
430 if (fb->evicted != NULL || fb->head != NULL) {
431 if (fb->evicted != NULL) {
437 if (fb->head != NULL) {
438 uint32_t next_ts = 0;
439 FlowManagerHashRowTimeout(td, fb->head,
ts, emergency, counters, &next_ts);
444 if (fb->evicted == NULL && fb->head == NULL) {
454 FlowManagerHashRowClearEvictedList(td,
evicted,
ts, counters);
461 cnt += ProcessAsideQueue(td, counters);
465 counters->rows_checked += rows_checked;
466 counters->rows_skipped += rows_skipped;
467 counters->rows_empty += rows_empty;
470 cnt += ProcessAsideQueue(td, counters);
472 counters->flows_removed +=
cnt;
482 const uint32_t rows, uint32_t *pos)
487 uint32_t rows_left = rows;
490 start = hash_min + (*pos);
491 if (start >= hash_max) {
494 end = start + rows_left;
495 if (end > hash_max) {
498 *pos = (end == hash_max) ? hash_min : end;
499 rows_left = rows_left - (end - start);
501 cnt += FlowTimeoutHash(td,
ts, start, end, counters);
528 RemoveFromHash(f, NULL);
530 FlowBucket *fb = f->
fb;
555 static uint32_t FlowCleanupHash(
void)
565 if (fb->head != NULL) {
567 cnt += FlowManagerHashRowCleanup(fb->head, &local_queue, 0);
569 if (fb->evicted != NULL) {
571 cnt += FlowManagerHashRowCleanup(fb->evicted, &local_queue, 1);
575 if (local_queue.
len >= 25) {
649 static void FlowCountersUpdate(
667 static TmEcode FlowManagerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
684 if ((ftd->
instance + 1) == flowmgr_number) {
694 FlowCountersInit(t, &ftd->
cnt);
722 static void GetWorkUnitSizing(
const uint32_t rows,
const uint32_t mp,
const bool emergency,
723 uint64_t *wu_sleep, uint32_t *wu_rows, uint32_t *rows_sec)
731 const uint32_t emp =
MAX(mp, 10);
732 const uint32_t rows_per_sec = (uint32_t)((
float)rows * (float)((
float)emp / (float)100));
735 const uint32_t work_per_unit =
MIN(rows_per_sec / 1000, 1000);
738 const uint32_t sleep_per_unit =
MAX(250, 1000 - work_per_unit);
739 SCLogDebug(
"mp %u emp %u rows %u rows_sec %u sleep %ums", mp, emp, rows, rows_per_sec,
742 *wu_sleep = sleep_per_unit;
743 *wu_rows = rows_per_sec;
744 *rows_sec = rows_per_sec;
756 const uint32_t rows = ftd->
max - ftd->
min;
759 uint32_t emerg_over_cnt = 0;
760 uint64_t next_run_ms = 0;
762 uint32_t rows_sec = 0;
763 uint32_t rows_per_wu = 0;
764 uint64_t sleep_per_wu = 0;
765 bool prev_emerg =
false;
766 uint32_t other_last_sec = 0;
781 GetWorkUnitSizing(rows, mp,
false, &sleep_per_wu, &rows_per_wu, &rows_sec);
800 const bool emerge_p = (emerg && !prev_emerg);
807 if (ts_ms >= next_run_ms) {
812 if (spare_perc < 90 || spare_perc > 110) {
819 FlowTimeoutCounters counters = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, };
827 SCLogDebug(
"hash %u:%u slice starting at %u with %u rows", ftd->
min, ftd->
max, pos,
830 const uint32_t ppos = pos;
831 FlowTimeoutHashInChunks(
841 FlowCountersUpdate(th_v, ftd, &counters);
844 SCLogDebug(
"flow_sparse_q.len = %" PRIu32
" prealloc: %" PRIu32
845 "flow_spare_q status: %" PRIu32
"%% flows at the queue",
858 if (emerg_over_cnt >= 30) {
865 SCLogNotice(
"Flow emergency mode over, back to normal... unsetting"
866 " FLOW_EMERGENCY bit (ts.tv_sec: %" PRIuMAX
", "
867 "ts.tv_usec:%" PRIuMAX
") flow_spare_q status(): %" PRIu32
868 "%% flows at the queue",
877 const uint32_t pmp = mp;
883 GetWorkUnitSizing(rows, mp, emerg, &sleep_per_wu, &rows_per_wu, &rows_sec);
888 next_run_ms = ts_ms + sleep_per_wu;
890 if (other_last_sec == 0 || other_last_sec < (uint32_t)
SCTIME_SECS(
ts)) {
905 if (emerg || !time_is_live) {
908 struct timeval cond_tv;
909 gettimeofday(&cond_tv, NULL);
910 struct timeval add_tv;
912 add_tv.tv_usec = (sleep_per_wu * 1000);
913 timeradd(&cond_tv, &add_tv, &cond_tv);
919 &flow_manager_ctrl_cond, &flow_manager_ctrl_mutex, &cond_time);
920 if (rc == ETIMEDOUT || rc < 0)
939 intmax_t setting = 1;
942 if (setting < 1 || setting > 1024) {
943 FatalError(
"invalid flow.managers setting %" PRIdMAX, setting);
945 flowmgr_number = (uint32_t)setting;
947 SCLogConfig(
"using %u flow manager threads", flowmgr_number);
950 for (uint32_t u = 0; u < flowmgr_number; u++) {
956 BUG_ON(tv_flowmgr == NULL);
958 if (tv_flowmgr == NULL) {
959 FatalError(
"flow manager thread creation failed");
962 FatalError(
"flow manager thread spawn failed");
980 static TmEcode FlowRecyclerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
986 SCLogError(
"initializing flow log API for thread failed");
1023 FlowEndCountersUpdate(
tv, &ftd->
fec, f);
1044 uint64_t recycled_cnt = 0;
1070 Recycler(th_v, ftd, f);
1079 if (ret_queue.
len > 0) {
1083 recycled_cnt +=
cnt;
1093 if (emerg || !time_is_live) {
1096 struct timeval cond_tv;
1097 gettimeofday(&cond_tv, NULL);
1098 cond_tv.tv_sec += 1;
1103 &flow_recycler_ctrl_cond, &flow_recycler_ctrl_mutex, &cond_time);
1104 if (rc == ETIMEDOUT || rc < 0) {
1122 SCLogPerf(
"%"PRIu64
" flows processed", recycled_cnt);
1126 static bool FlowRecyclerReadyToShutdown(
void)
1136 return ((
len == 0));
1142 intmax_t setting = 1;
1143 (void)
ConfGetInt(
"flow.recyclers", &setting);
1145 if (setting < 1 || setting > 1024) {
1146 FatalError(
"invalid flow.recyclers setting %" PRIdMAX, setting);
1148 flowrec_number = (uint32_t)setting;
1150 SCLogConfig(
"using %u flow recycler threads", flowrec_number);
1152 for (uint32_t u = 0; u < flowrec_number; u++) {
1159 if (tv_flowrec == NULL) {
1160 FatalError(
"flow recycler thread creation failed");
1163 FatalError(
"flow recycler thread spawn failed");
1181 (void)FlowCleanupHash();
1183 uint32_t flows = FlowCleanupHash();
1191 }
while (FlowRecyclerReadyToShutdown() ==
false);
1204 struct timeval start_ts;
1205 struct timeval cur_ts;
1206 gettimeofday(&start_ts, NULL);
1209 gettimeofday(&cur_ts, NULL);
1210 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1211 FatalError(
"unable to get all flow recycler "
1212 "threads to shutdown in time");