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) ==
false) {
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;
843 GetWorkUnitSizing(rows, mp,
false, &sleep_per_wu, &rows_per_wu, &rows_sec);
858 const bool emerge_p = (emerg && !prev_emerg);
865 if (ts_ms >= next_run_ms) {
870 if (spare_perc < 90 || spare_perc > 110) {
877 FlowTimeoutCounters counters = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, };
885 SCLogDebug(
"hash %u:%u slice starting at %u with %u rows", ftd->
min, ftd->
max, pos,
888 const uint32_t ppos = pos;
889 FlowTimeoutHashInChunks(&ftd->
timeout,
ts, ftd->
min, ftd->
max, &counters,
899 FlowCountersUpdate(th_v, ftd, &counters);
902 SCLogDebug(
"flow_sparse_q.len = %" PRIu32
" prealloc: %" PRIu32
903 "flow_spare_q status: %" PRIu32
"%% flows at the queue",
916 if (emerg_over_cnt >= 30) {
923 SCLogNotice(
"Flow emergency mode over, back to normal... unsetting"
924 " FLOW_EMERGENCY bit (ts.tv_sec: %" PRIuMAX
", "
925 "ts.tv_usec:%" PRIuMAX
") flow_spare_q status(): %" PRIu32
926 "%% flows at the queue",
935 const uint32_t pmp = mp;
941 GetWorkUnitSizing(rows, mp, emerg, &sleep_per_wu, &rows_per_wu, &rows_sec);
946 next_run_ms = ts_ms + sleep_per_wu;
948 if (other_last_sec == 0 || other_last_sec < (uint32_t)
SCTIME_SECS(
ts)) {
968 if (emerg || !time_is_live) {
971 struct timeval cond_tv;
972 gettimeofday(&cond_tv, NULL);
973 struct timeval add_tv;
974 add_tv.tv_sec = sleep_per_wu / 1000;
975 add_tv.tv_usec = (sleep_per_wu % 1000) * 1000;
976 timeradd(&cond_tv, &add_tv, &cond_tv);
982 &flow_manager_ctrl_cond, &flow_manager_ctrl_mutex, &cond_time);
983 if (rc == ETIMEDOUT || rc < 0)
1002 intmax_t setting = 1;
1005 if (setting < 1 || setting > 1024) {
1006 FatalError(
"invalid flow.managers setting %" PRIdMAX, setting);
1008 flowmgr_number = (uint32_t)setting;
1010 SCLogConfig(
"using %u flow manager threads", flowmgr_number);
1013 for (uint32_t u = 0; u < flowmgr_number; u++) {
1019 BUG_ON(tv_flowmgr == NULL);
1021 if (tv_flowmgr == NULL) {
1022 FatalError(
"flow manager thread creation failed");
1025 FatalError(
"flow manager thread spawn failed");
1042 static TmEcode FlowRecyclerThreadInit(
ThreadVars *t,
const void *initdata,
void **data)
1048 SCLogError(
"initializing flow log API for thread failed");
1085 FlowEndCountersUpdate(
tv, &ftd->
fec, f);
1104 uint64_t recycled_cnt = 0;
1125 Recycler(th_v, ftd, f);
1134 if (ret_queue.
len > 0) {
1138 recycled_cnt +=
cnt;
1148 if (emerg || !time_is_live) {
1151 struct timeval cond_tv;
1152 gettimeofday(&cond_tv, NULL);
1153 cond_tv.tv_sec += 1;
1158 &flow_recycler_ctrl_cond, &flow_recycler_ctrl_mutex, &cond_time);
1159 if (rc == ETIMEDOUT || rc < 0) {
1177 SCLogPerf(
"%"PRIu64
" flows processed", recycled_cnt);
1181 static bool FlowRecyclerReadyToShutdown(
void)
1191 return ((
len == 0));
1197 intmax_t setting = 1;
1198 (void)
ConfGetInt(
"flow.recyclers", &setting);
1200 if (setting < 1 || setting > 1024) {
1201 FatalError(
"invalid flow.recyclers setting %" PRIdMAX, setting);
1203 flowrec_number = (uint32_t)setting;
1205 SCLogConfig(
"using %u flow recycler threads", flowrec_number);
1207 for (uint32_t u = 0; u < flowrec_number; u++) {
1214 if (tv_flowrec == NULL) {
1215 FatalError(
"flow recycler thread creation failed");
1218 FatalError(
"flow recycler thread spawn failed");
1235 (void)FlowCleanupHash();
1237 uint32_t flows = FlowCleanupHash();
1245 }
while (FlowRecyclerReadyToShutdown() ==
false);
1258 struct timeval start_ts;
1259 struct timeval cur_ts;
1260 gettimeofday(&start_ts, NULL);
1263 gettimeofday(&cur_ts, NULL);
1264 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1265 FatalError(
"unable to get all flow recycler "
1266 "threads to shutdown in time");