68 static uint32_t flowmgr_number = 1;
73 static uint32_t flowrec_number = 1;
79 static SCCtrlCondT flow_manager_ctrl_cond = PTHREAD_COND_INITIALIZER;
80 static SCCtrlMutex flow_manager_ctrl_mutex = PTHREAD_MUTEX_INITIALIZER;
81 static SCCtrlCondT flow_recycler_ctrl_cond = PTHREAD_COND_INITIALIZER;
82 static SCCtrlMutex flow_recycler_ctrl_mutex = PTHREAD_MUTEX_INITIALIZER;
145 struct timeval start_ts;
146 struct timeval cur_ts;
147 gettimeofday(&start_ts, NULL);
150 gettimeofday(&cur_ts, NULL);
151 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
153 "threads to shutdown in time");
190 static bool FlowManagerFlowTimeout(
Flow *f,
SCTime_t ts, uint32_t *next_ts,
const bool emerg)
202 if (*next_ts == 0 || (uint32_t)
SCTIME_SECS(timesout_at) < *next_ts)
223 #ifdef CAPTURE_OFFLOAD
236 if (f->
flow_state != FLOW_STATE_CAPTURE_BYPASSED) {
249 SCLogDebug(
"Updated flow: %"PRId64
"", FlowGetId(f));
256 pkts_tosrc + pkts_todst);
258 counters->bypassed_pkts += pkts_tosrc + pkts_todst;
259 counters->bypassed_bytes += bytes_tosrc + bytes_todst;
262 SCLogDebug(
"No new packet, dead flow %" PRId64
"", FlowGetId(f));
270 counters->bypassed_count++;
305 if (f->
proto == IPPROTO_TCP &&
319 if (recycle.
len == 100) {
345 uint32_t checked = 0;
359 if (!FlowManagerFlowTimeout(f,
ts, next_ts, emergency)) {
361 counters->flows_notimeout++;
370 #ifdef CAPTURE_OFFLOAD
373 if (!FlowBypassedTimeout(f,
ts, counters)) {
382 counters->flows_timeout++;
384 RemoveFromHash(f, prev_f);
392 counters->flows_checked += checked;
393 if (checked > counters->rows_maxlen)
394 counters->rows_maxlen = checked;
437 const uint32_t rows_checked = hash_max - hash_min;
438 uint32_t rows_skipped = 0;
439 uint32_t rows_empty = 0;
443 #define TYPE uint64_t
446 #define TYPE uint32_t
450 for (uint32_t idx = hash_min; idx < hash_max; idx+=
BITS) {
452 const uint32_t check =
MIN(
BITS, (hash_max - idx));
453 for (uint32_t i = 0; i < check; i++) {
462 for (uint32_t i = 0; i < check; i++) {
467 if (fb->evicted != NULL || fb->head != NULL) {
468 if (fb->evicted != NULL) {
474 if (fb->head != NULL) {
475 uint32_t next_ts = 0;
476 FlowManagerHashRowTimeout(td, fb->head,
ts, emergency, counters, &next_ts);
481 if (fb->evicted == NULL && fb->head == NULL) {
492 FlowManagerHashRowClearEvictedList(td,
evicted);
499 cnt += ProcessAsideQueue(td, counters);
503 counters->rows_checked += rows_checked;
504 counters->rows_skipped += rows_skipped;
505 counters->rows_empty += rows_empty;
508 cnt += ProcessAsideQueue(td, counters);
510 counters->flows_removed +=
cnt;
532 const uint32_t rows, uint32_t *pos,
const uint32_t instance)
537 uint32_t rows_left = rows;
541 if (start >= hash_max) {
544 end = start + rows_left;
545 if (end > hash_max) {
548 *pos = (end == hash_max) ? hash_min : end;
549 rows_left = rows_left - (end - start);
551 SCLogDebug(
"instance %u: %u:%u (hash_min %u, hash_max %u *pos %u)", instance, start, end,
552 hash_min, hash_max, *pos);
554 cnt += FlowTimeoutHash(td,
ts, start, end, counters);
583 RemoveFromHash(f, NULL);
585 FlowBucket *fb = f->
fb;
605 #define RECYCLE_MAX_QUEUE_ITEMS 25
611 static uint32_t FlowCleanupHash(
void)
621 if (fb->head != NULL) {
623 cnt += FlowManagerHashRowCleanup(fb->head, &local_queue, 0);
625 if (fb->evicted != NULL) {
627 cnt += FlowManagerHashRowCleanup(fb->evicted, &local_queue, 1);
703 static void FlowCountersUpdate(
721 static TmEcode FlowManagerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
738 if ((ftd->
instance + 1) == flowmgr_number) {
748 FlowCountersInit(t, &ftd->
cnt);
785 static void GetWorkUnitSizing(
const uint32_t rows,
const uint32_t mp,
const bool emergency,
786 uint64_t *wu_sleep, uint32_t *wu_rows, uint32_t *rows_sec)
794 const uint32_t emp =
MAX(mp, 10);
795 const uint32_t rows_per_sec = (uint32_t)((
float)rows * (float)((
float)emp / (float)100));
798 const uint32_t work_per_unit =
MIN(rows_per_sec / 1000, 1000);
801 const uint32_t sleep_per_unit =
MAX(250, 1000 - work_per_unit);
802 SCLogDebug(
"mp %u emp %u rows %u rows_sec %u sleep %ums", mp, emp, rows, rows_per_sec,
805 *wu_sleep = sleep_per_unit;
806 *wu_rows = rows_per_sec;
807 *rows_sec = rows_per_sec;
819 const uint32_t rows = ftd->
max - ftd->
min;
822 uint32_t emerg_over_cnt = 0;
823 uint64_t next_run_ms = 0;
824 uint32_t pos = ftd->
min;
825 uint32_t rows_sec = 0;
826 uint32_t rows_per_wu = 0;
827 uint64_t sleep_per_wu = 0;
828 bool prev_emerg =
false;
829 uint32_t other_last_sec = 0;
836 GetWorkUnitSizing(rows, mp,
false, &sleep_per_wu, &rows_per_wu, &rows_sec);
857 const bool emerge_p = (emerg && !prev_emerg);
864 if (ts_ms >= next_run_ms) {
869 if (spare_perc < 90 || spare_perc > 110) {
876 FlowTimeoutCounters counters = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, };
884 SCLogDebug(
"hash %u:%u slice starting at %u with %u rows", ftd->
min, ftd->
max, pos,
887 const uint32_t ppos = pos;
888 FlowTimeoutHashInChunks(&ftd->
timeout,
ts, ftd->
min, ftd->
max, &counters,
898 FlowCountersUpdate(th_v, ftd, &counters);
901 SCLogDebug(
"flow_sparse_q.len = %" PRIu32
" prealloc: %" PRIu32
902 "flow_spare_q status: %" PRIu32
"%% flows at the queue",
915 if (emerg_over_cnt >= 30) {
922 SCLogNotice(
"Flow emergency mode over, back to normal... unsetting"
923 " FLOW_EMERGENCY bit (ts.tv_sec: %" PRIuMAX
", "
924 "ts.tv_usec:%" PRIuMAX
") flow_spare_q status(): %" PRIu32
925 "%% flows at the queue",
934 const uint32_t pmp = mp;
940 GetWorkUnitSizing(rows, mp, emerg, &sleep_per_wu, &rows_per_wu, &rows_sec);
945 next_run_ms = ts_ms + sleep_per_wu;
947 if (other_last_sec == 0 || other_last_sec < (uint32_t)
SCTIME_SECS(
ts)) {
967 if (emerg || !time_is_live) {
970 struct timeval cond_tv;
971 gettimeofday(&cond_tv, NULL);
972 struct timeval add_tv;
973 add_tv.tv_sec = sleep_per_wu / 1000;
974 add_tv.tv_usec = (sleep_per_wu % 1000) * 1000;
975 timeradd(&cond_tv, &add_tv, &cond_tv);
981 &flow_manager_ctrl_cond, &flow_manager_ctrl_mutex, &cond_time);
982 if (rc == ETIMEDOUT || rc < 0)
1001 intmax_t setting = 1;
1004 if (setting < 1 || setting > 1024) {
1005 FatalError(
"invalid flow.managers setting %" PRIdMAX, setting);
1007 flowmgr_number = (uint32_t)setting;
1009 SCLogConfig(
"using %u flow manager threads", flowmgr_number);
1012 for (uint32_t u = 0; u < flowmgr_number; u++) {
1018 BUG_ON(tv_flowmgr == NULL);
1020 if (tv_flowmgr == NULL) {
1021 FatalError(
"flow manager thread creation failed");
1024 FatalError(
"flow manager thread spawn failed");
1041 static TmEcode FlowRecyclerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
1047 SCLogError(
"initializing flow log API for thread failed");
1084 FlowEndCountersUpdate(
tv, &ftd->
fec, f);
1103 uint64_t recycled_cnt = 0;
1124 Recycler(th_v, ftd, f);
1133 if (ret_queue.
len > 0) {
1137 recycled_cnt +=
cnt;
1147 if (emerg || !time_is_live) {
1150 struct timeval cond_tv;
1151 gettimeofday(&cond_tv, NULL);
1152 cond_tv.tv_sec += 1;
1157 &flow_recycler_ctrl_cond, &flow_recycler_ctrl_mutex, &cond_time);
1158 if (rc == ETIMEDOUT || rc < 0) {
1176 SCLogPerf(
"%"PRIu64
" flows processed", recycled_cnt);
1180 static bool FlowRecyclerReadyToShutdown(
void)
1190 return ((
len == 0));
1196 intmax_t setting = 1;
1199 if (setting < 1 || setting > 1024) {
1200 FatalError(
"invalid flow.recyclers setting %" PRIdMAX, setting);
1202 flowrec_number = (uint32_t)setting;
1204 SCLogConfig(
"using %u flow recycler threads", flowrec_number);
1206 for (uint32_t u = 0; u < flowrec_number; u++) {
1213 if (tv_flowrec == NULL) {
1214 FatalError(
"flow recycler thread creation failed");
1217 FatalError(
"flow recycler thread spawn failed");
1234 (void)FlowCleanupHash();
1236 uint32_t flows = FlowCleanupHash();
1244 }
while (!FlowRecyclerReadyToShutdown());
1257 struct timeval start_ts;
1258 struct timeval cur_ts;
1259 gettimeofday(&start_ts, NULL);
1262 gettimeofday(&cur_ts, NULL);
1263 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1264 FatalError(
"unable to get all flow recycler "
1265 "threads to shutdown in time");