66 static uint32_t flowmgr_number = 1;
71 static uint32_t flowrec_number = 1;
77 static SCCtrlCondT flow_manager_ctrl_cond = PTHREAD_COND_INITIALIZER;
78 static SCCtrlMutex flow_manager_ctrl_mutex = PTHREAD_MUTEX_INITIALIZER;
79 static SCCtrlCondT flow_recycler_ctrl_cond = PTHREAD_COND_INITIALIZER;
80 static SCCtrlMutex flow_recycler_ctrl_mutex = PTHREAD_MUTEX_INITIALIZER;
143 struct timeval start_ts;
144 struct timeval cur_ts;
145 gettimeofday(&start_ts, NULL);
148 gettimeofday(&cur_ts, NULL);
149 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
151 "threads to shutdown in time");
182 static bool FlowManagerFlowTimeout(
Flow *f,
SCTime_t ts, uint32_t *next_ts,
const bool emerg)
189 if (*next_ts == 0 || flow_times_out_at < *next_ts)
190 *next_ts = flow_times_out_at;
200 #ifdef CAPTURE_OFFLOAD
213 if (f->
flow_state != FLOW_STATE_CAPTURE_BYPASSED) {
226 SCLogDebug(
"Updated flow: %"PRId64
"", FlowGetId(f));
233 pkts_tosrc + pkts_todst);
235 counters->bypassed_pkts += pkts_tosrc + pkts_todst;
236 counters->bypassed_bytes += bytes_tosrc + bytes_todst;
239 SCLogDebug(
"No new packet, dead flow %" PRId64
"", FlowGetId(f));
247 counters->bypassed_count++;
282 if (f->
proto == IPPROTO_TCP &&
296 if (recycle.
len == 100) {
322 uint32_t checked = 0;
334 if (FlowManagerFlowTimeout(f,
ts, next_ts, emergency) ==
false) {
335 counters->flows_notimeout++;
346 #ifdef CAPTURE_OFFLOAD
349 if (!FlowBypassedTimeout(f,
ts, counters)) {
358 counters->flows_timeout++;
360 RemoveFromHash(f, prev_f);
368 counters->flows_checked += checked;
369 if (checked > counters->rows_maxlen)
370 counters->rows_maxlen = checked;
413 const uint32_t rows_checked = hash_max - hash_min;
414 uint32_t rows_skipped = 0;
415 uint32_t rows_empty = 0;
419 #define TYPE uint64_t
422 #define TYPE uint32_t
426 for (uint32_t idx = hash_min; idx < hash_max; idx+=
BITS) {
428 const uint32_t check =
MIN(
BITS, (hash_max - idx));
429 for (uint32_t i = 0; i < check; i++) {
438 for (uint32_t i = 0; i < check; i++) {
443 if (fb->evicted != NULL || fb->head != NULL) {
444 if (fb->evicted != NULL) {
450 if (fb->head != NULL) {
451 uint32_t next_ts = 0;
452 FlowManagerHashRowTimeout(td, fb->head,
ts, emergency, counters, &next_ts);
457 if (fb->evicted == NULL && fb->head == NULL) {
468 FlowManagerHashRowClearEvictedList(td,
evicted);
475 cnt += ProcessAsideQueue(td, counters);
479 counters->rows_checked += rows_checked;
480 counters->rows_skipped += rows_skipped;
481 counters->rows_empty += rows_empty;
484 cnt += ProcessAsideQueue(td, counters);
486 counters->flows_removed +=
cnt;
507 const uint32_t rows, uint32_t *pos)
512 uint32_t rows_left = rows;
515 start = hash_min + (*pos);
516 if (start >= hash_max) {
519 end = start + rows_left;
520 if (end > hash_max) {
523 *pos = (end == hash_max) ? hash_min : end;
524 rows_left = rows_left - (end - start);
526 cnt += FlowTimeoutHash(td,
ts, start, end, counters);
555 RemoveFromHash(f, NULL);
557 FlowBucket *fb = f->
fb;
577 #define RECYCLE_MAX_QUEUE_ITEMS 25
583 static uint32_t FlowCleanupHash(
void)
593 if (fb->head != NULL) {
595 cnt += FlowManagerHashRowCleanup(fb->head, &local_queue, 0);
597 if (fb->evicted != NULL) {
599 cnt += FlowManagerHashRowCleanup(fb->evicted, &local_queue, 1);
675 static void FlowCountersUpdate(
693 static TmEcode FlowManagerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
710 if ((ftd->
instance + 1) == flowmgr_number) {
720 FlowCountersInit(t, &ftd->
cnt);
757 static void GetWorkUnitSizing(
const uint32_t rows,
const uint32_t mp,
const bool emergency,
758 uint64_t *wu_sleep, uint32_t *wu_rows, uint32_t *rows_sec)
766 const uint32_t emp =
MAX(mp, 10);
767 const uint32_t rows_per_sec = (uint32_t)((
float)rows * (float)((
float)emp / (float)100));
770 const uint32_t work_per_unit =
MIN(rows_per_sec / 1000, 1000);
773 const uint32_t sleep_per_unit =
MAX(250, 1000 - work_per_unit);
774 SCLogDebug(
"mp %u emp %u rows %u rows_sec %u sleep %ums", mp, emp, rows, rows_per_sec,
777 *wu_sleep = sleep_per_unit;
778 *wu_rows = rows_per_sec;
779 *rows_sec = rows_per_sec;
791 const uint32_t rows = ftd->
max - ftd->
min;
794 uint32_t emerg_over_cnt = 0;
795 uint64_t next_run_ms = 0;
797 uint32_t rows_sec = 0;
798 uint32_t rows_per_wu = 0;
799 uint64_t sleep_per_wu = 0;
800 bool prev_emerg =
false;
801 uint32_t other_last_sec = 0;
816 GetWorkUnitSizing(rows, mp,
false, &sleep_per_wu, &rows_per_wu, &rows_sec);
835 const bool emerge_p = (emerg && !prev_emerg);
842 if (ts_ms >= next_run_ms) {
847 if (spare_perc < 90 || spare_perc > 110) {
854 FlowTimeoutCounters counters = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, };
862 SCLogDebug(
"hash %u:%u slice starting at %u with %u rows", ftd->
min, ftd->
max, pos,
865 const uint32_t ppos = pos;
866 FlowTimeoutHashInChunks(
876 FlowCountersUpdate(th_v, ftd, &counters);
879 SCLogDebug(
"flow_sparse_q.len = %" PRIu32
" prealloc: %" PRIu32
880 "flow_spare_q status: %" PRIu32
"%% flows at the queue",
893 if (emerg_over_cnt >= 30) {
900 SCLogNotice(
"Flow emergency mode over, back to normal... unsetting"
901 " FLOW_EMERGENCY bit (ts.tv_sec: %" PRIuMAX
", "
902 "ts.tv_usec:%" PRIuMAX
") flow_spare_q status(): %" PRIu32
903 "%% flows at the queue",
912 const uint32_t pmp = mp;
918 GetWorkUnitSizing(rows, mp, emerg, &sleep_per_wu, &rows_per_wu, &rows_sec);
923 next_run_ms = ts_ms + sleep_per_wu;
925 if (other_last_sec == 0 || other_last_sec < (uint32_t)
SCTIME_SECS(
ts)) {
945 if (emerg || !time_is_live) {
948 struct timeval cond_tv;
949 gettimeofday(&cond_tv, NULL);
950 struct timeval add_tv;
951 add_tv.tv_sec = sleep_per_wu / 1000;
952 add_tv.tv_usec = (sleep_per_wu % 1000) * 1000;
953 timeradd(&cond_tv, &add_tv, &cond_tv);
959 &flow_manager_ctrl_cond, &flow_manager_ctrl_mutex, &cond_time);
960 if (rc == ETIMEDOUT || rc < 0)
979 intmax_t setting = 1;
982 if (setting < 1 || setting > 1024) {
983 FatalError(
"invalid flow.managers setting %" PRIdMAX, setting);
985 flowmgr_number = (uint32_t)setting;
987 SCLogConfig(
"using %u flow manager threads", flowmgr_number);
990 for (uint32_t u = 0; u < flowmgr_number; u++) {
996 BUG_ON(tv_flowmgr == NULL);
998 if (tv_flowmgr == NULL) {
999 FatalError(
"flow manager thread creation failed");
1002 FatalError(
"flow manager thread spawn failed");
1019 static TmEcode FlowRecyclerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
1025 SCLogError(
"initializing flow log API for thread failed");
1062 FlowEndCountersUpdate(
tv, &ftd->
fec, f);
1081 uint64_t recycled_cnt = 0;
1107 Recycler(th_v, ftd, f);
1116 if (ret_queue.
len > 0) {
1120 recycled_cnt +=
cnt;
1130 if (emerg || !time_is_live) {
1133 struct timeval cond_tv;
1134 gettimeofday(&cond_tv, NULL);
1135 cond_tv.tv_sec += 1;
1140 &flow_recycler_ctrl_cond, &flow_recycler_ctrl_mutex, &cond_time);
1141 if (rc == ETIMEDOUT || rc < 0) {
1159 SCLogPerf(
"%"PRIu64
" flows processed", recycled_cnt);
1163 static bool FlowRecyclerReadyToShutdown(
void)
1173 return ((
len == 0));
1179 intmax_t setting = 1;
1180 (void)
ConfGetInt(
"flow.recyclers", &setting);
1182 if (setting < 1 || setting > 1024) {
1183 FatalError(
"invalid flow.recyclers setting %" PRIdMAX, setting);
1185 flowrec_number = (uint32_t)setting;
1187 SCLogConfig(
"using %u flow recycler threads", flowrec_number);
1189 for (uint32_t u = 0; u < flowrec_number; u++) {
1196 if (tv_flowrec == NULL) {
1197 FatalError(
"flow recycler thread creation failed");
1200 FatalError(
"flow recycler thread spawn failed");
1217 (void)FlowCleanupHash();
1219 uint32_t flows = FlowCleanupHash();
1227 }
while (FlowRecyclerReadyToShutdown() ==
false);
1240 struct timeval start_ts;
1241 struct timeval cur_ts;
1242 gettimeofday(&start_ts, NULL);
1245 gettimeofday(&cur_ts, NULL);
1246 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1247 FatalError(
"unable to get all flow recycler "
1248 "threads to shutdown in time");