76 static uint32_t flowmgr_number = 1;
81 static uint32_t flowrec_number = 1;
87 static SCCtrlCondT flow_manager_ctrl_cond = PTHREAD_COND_INITIALIZER;
88 static SCCtrlMutex flow_manager_ctrl_mutex = PTHREAD_MUTEX_INITIALIZER;
89 static SCCtrlCondT flow_recycler_ctrl_cond = PTHREAD_COND_INITIALIZER;
90 static SCCtrlMutex flow_recycler_ctrl_mutex = PTHREAD_MUTEX_INITIALIZER;
117 #define FLOW_NORMAL_MODE_UPDATE_DELAY_SEC 1
118 #define FLOW_NORMAL_MODE_UPDATE_DELAY_NSEC 0
120 #define FLOW_EMERG_MODE_UPDATE_DELAY_SEC 0
121 #define FLOW_EMERG_MODE_UPDATE_DELAY_NSEC 300000
122 #define NEW_FLOW_COUNT_COND 10
161 struct timeval start_ts;
162 struct timeval cur_ts;
163 gettimeofday(&start_ts, NULL);
166 gettimeofday(&cur_ts, NULL);
167 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
169 "threads to shutdown in time");
201 static int FlowManagerFlowTimeout(
Flow *f,
SCTime_t ts, uint32_t *next_ts,
const bool emerg)
208 if (*next_ts == 0 || flow_times_out_at < *next_ts)
209 *next_ts = flow_times_out_at;
231 #ifdef CAPTURE_OFFLOAD
232 if (f->
flow_state != FLOW_STATE_CAPTURE_BYPASSED) {
245 SCLogDebug(
"Updated flow: %"PRId64
"", FlowGetId(f));
252 pkts_tosrc + pkts_todst);
254 counters->bypassed_pkts += pkts_tosrc + pkts_todst;
255 counters->bypassed_bytes += bytes_tosrc + bytes_todst;
258 SCLogDebug(
"No new packet, dead flow %"PRId64
"", FlowGetId(f));
266 counters->bypassed_count++;
290 if (f->
proto == IPPROTO_TCP &&
304 if (recycle.
len == 100) {
330 uint32_t checked = 0;
342 if (FlowManagerFlowTimeout(f,
ts, next_ts, emergency) == 0) {
344 counters->flows_notimeout++;
357 if (!FlowBypassedTimeout(f,
ts, counters)) {
366 counters->flows_timeout++;
368 RemoveFromHash(f, prev_f);
376 counters->flows_checked += checked;
377 if (checked > counters->rows_maxlen)
378 counters->rows_maxlen = checked;
381 static void FlowManagerHashRowClearEvictedList(
412 const uint32_t rows_checked = hash_max - hash_min;
413 uint32_t rows_skipped = 0;
414 uint32_t rows_empty = 0;
418 #define TYPE uint64_t
421 #define TYPE uint32_t
425 for (uint32_t idx = hash_min; idx < hash_max; idx+=
BITS) {
427 const uint32_t check =
MIN(
BITS, (hash_max - idx));
428 for (uint32_t i = 0; i < check; i++) {
437 for (uint32_t i = 0; i < check; i++) {
442 if (fb->evicted != NULL || fb->head != NULL) {
443 if (fb->evicted != NULL) {
449 if (fb->head != NULL) {
450 uint32_t next_ts = 0;
451 FlowManagerHashRowTimeout(td, fb->head,
ts, emergency, counters, &next_ts);
456 if (fb->evicted == NULL && fb->head == NULL) {
466 FlowManagerHashRowClearEvictedList(td,
evicted,
ts, counters);
473 cnt += ProcessAsideQueue(td, counters);
477 counters->rows_checked += rows_checked;
478 counters->rows_skipped += rows_skipped;
479 counters->rows_empty += rows_empty;
482 cnt += ProcessAsideQueue(td, counters);
484 counters->flows_removed +=
cnt;
494 const uint32_t rows, uint32_t *pos)
499 uint32_t rows_left = rows;
502 start = hash_min + (*pos);
503 if (start >= hash_max) {
506 end = start + rows_left;
507 if (end > hash_max) {
510 *pos = (end == hash_max) ? hash_min : end;
511 rows_left = rows_left - (end - start);
513 cnt += FlowTimeoutHash(td,
ts, start, end, counters);
540 RemoveFromHash(f, NULL);
542 FlowBucket *fb = f->
fb;
567 static uint32_t FlowCleanupHash(
void)
577 if (fb->head != NULL) {
579 cnt += FlowManagerHashRowCleanup(fb->head, &local_queue, 0);
581 if (fb->evicted != NULL) {
583 cnt += FlowManagerHashRowCleanup(fb->evicted, &local_queue, 1);
587 if (local_queue.
len >= 25) {
661 static void FlowCountersUpdate(
679 static TmEcode FlowManagerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
696 if ((ftd->
instance + 1) == flowmgr_number) {
706 FlowCountersInit(t, &ftd->
cnt);
734 static void GetWorkUnitSizing(
const uint32_t rows,
const uint32_t mp,
const bool emergency,
735 uint64_t *wu_sleep, uint32_t *wu_rows, uint32_t *rows_sec)
743 const uint32_t emp =
MAX(mp, 10);
744 const uint32_t rows_per_sec = (uint32_t)((
float)rows * (float)((
float)emp / (float)100));
747 const uint32_t work_per_unit =
MIN(rows_per_sec / 1000, 1000);
750 const uint32_t sleep_per_unit =
MAX(250, 1000 - work_per_unit);
751 SCLogDebug(
"mp %u emp %u rows %u rows_sec %u sleep %ums", mp, emp, rows, rows_per_sec,
754 *wu_sleep = sleep_per_unit;
755 *wu_rows = rows_per_sec;
756 *rows_sec = rows_per_sec;
768 const uint32_t rows = ftd->
max - ftd->
min;
771 uint32_t emerg_over_cnt = 0;
772 uint64_t next_run_ms = 0;
774 uint32_t rows_sec = 0;
775 uint32_t rows_per_wu = 0;
776 uint64_t sleep_per_wu = 0;
777 bool prev_emerg =
false;
778 uint32_t other_last_sec = 0;
793 GetWorkUnitSizing(rows, mp,
false, &sleep_per_wu, &rows_per_wu, &rows_sec);
812 const bool emerge_p = (emerg && !prev_emerg);
819 if (ts_ms >= next_run_ms) {
824 if (spare_perc < 90 || spare_perc > 110) {
831 FlowTimeoutCounters counters = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, };
839 SCLogDebug(
"hash %u:%u slice starting at %u with %u rows", ftd->
min, ftd->
max, pos,
842 const uint32_t ppos = pos;
843 FlowTimeoutHashInChunks(
853 FlowCountersUpdate(th_v, ftd, &counters);
856 SCLogDebug(
"flow_sparse_q.len = %" PRIu32
" prealloc: %" PRIu32
857 "flow_spare_q status: %" PRIu32
"%% flows at the queue",
870 if (emerg_over_cnt >= 30) {
877 SCLogNotice(
"Flow emergency mode over, back to normal... unsetting"
878 " FLOW_EMERGENCY bit (ts.tv_sec: %" PRIuMAX
", "
879 "ts.tv_usec:%" PRIuMAX
") flow_spare_q status(): %" PRIu32
880 "%% flows at the queue",
889 const uint32_t pmp = mp;
895 GetWorkUnitSizing(rows, mp, emerg, &sleep_per_wu, &rows_per_wu, &rows_sec);
900 next_run_ms = ts_ms + sleep_per_wu;
902 if (other_last_sec == 0 || other_last_sec < (uint32_t)
SCTIME_SECS(
ts)) {
917 if (emerg || !time_is_live) {
920 struct timeval cond_tv;
921 gettimeofday(&cond_tv, NULL);
922 struct timeval add_tv;
924 add_tv.tv_usec = (sleep_per_wu * 1000);
925 timeradd(&cond_tv, &add_tv, &cond_tv);
931 &flow_manager_ctrl_cond, &flow_manager_ctrl_mutex, &cond_time);
932 if (rc == ETIMEDOUT || rc < 0)
951 intmax_t setting = 1;
954 if (setting < 1 || setting > 1024) {
955 FatalError(
"invalid flow.managers setting %" PRIdMAX, setting);
957 flowmgr_number = (uint32_t)setting;
959 SCLogConfig(
"using %u flow manager threads", flowmgr_number);
962 for (uint32_t u = 0; u < flowmgr_number; u++) {
968 BUG_ON(tv_flowmgr == NULL);
970 if (tv_flowmgr == NULL) {
971 FatalError(
"flow manager thread creation failed");
974 FatalError(
"flow manager thread spawn failed");
992 static TmEcode FlowRecyclerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
998 SCLogError(
"initializing flow log API for thread failed");
1035 FlowEndCountersUpdate(
tv, &ftd->
fec, f);
1056 uint64_t recycled_cnt = 0;
1082 Recycler(th_v, ftd, f);
1091 if (ret_queue.
len > 0) {
1095 recycled_cnt +=
cnt;
1105 if (emerg || !time_is_live) {
1108 struct timeval cond_tv;
1109 gettimeofday(&cond_tv, NULL);
1110 cond_tv.tv_sec += 1;
1115 &flow_recycler_ctrl_cond, &flow_recycler_ctrl_mutex, &cond_time);
1116 if (rc == ETIMEDOUT || rc < 0) {
1134 SCLogPerf(
"%"PRIu64
" flows processed", recycled_cnt);
1138 static bool FlowRecyclerReadyToShutdown(
void)
1148 return ((
len == 0));
1154 intmax_t setting = 1;
1155 (void)
ConfGetInt(
"flow.recyclers", &setting);
1157 if (setting < 1 || setting > 1024) {
1158 FatalError(
"invalid flow.recyclers setting %" PRIdMAX, setting);
1160 flowrec_number = (uint32_t)setting;
1162 SCLogConfig(
"using %u flow recycler threads", flowrec_number);
1164 for (uint32_t u = 0; u < flowrec_number; u++) {
1171 if (tv_flowrec == NULL) {
1172 FatalError(
"flow recycler thread creation failed");
1175 FatalError(
"flow recycler thread spawn failed");
1193 (void)FlowCleanupHash();
1195 uint32_t flows = FlowCleanupHash();
1203 }
while (FlowRecyclerReadyToShutdown() ==
false);
1216 struct timeval start_ts;
1217 struct timeval cur_ts;
1218 gettimeofday(&start_ts, NULL);
1221 gettimeofday(&cur_ts, NULL);
1222 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1223 FatalError(
"unable to get all flow recycler "
1224 "threads to shutdown in time");