mirror of https://github.com/OISF/suricata
flow: redesign of flow timeout handling
Goals: - reduce locking - take advantage of 'hot' caches - better locality Locking reduction New flow spare pool. The global pool is implmented as a list of blocks, where each block has a 100 spare flows. Worker threads fetch a block at a time, storing the block in the local thread storage. Flow Recycler now returns flows to the pool is blocks as well. Flow Recycler fetches all flows to be processed in one step instead of one at a time. Cache 'hot'ness Worker threads now check the timeout of flows they evaluate during lookup. The worker will have to read the flow into cache anyway, so the added overhead of checking the timeout value is minimal. When a flow is considered timed out, one of 2 things happens: - if the flow is 'owned' by the thread it is handled locally. Handling means checking if the flow needs 'timeout' work. - otherwise, the flow is added to a special 'evicted' list in the flow bucket where it will be picked up by the flow manager. Flow Manager timing By default the flow manager now tries to do passes of the flow hash in smaller steps, where the goal is to do full pass in 8 x the lowest timeout value it has to enforce. So if the lowest timeout value is 30s, a full pass will take 4 minutes. The goal here is to reduce locking overhead and not get in the way of the workers. In emergency mode each pass is full, and lower timeouts are used. Timing of the flow manager is also no longer relying on pthread condition variables, as these generally cause waking up much quicker than the desired timout. Instead a simple (u)sleep loop is used. Both changes reduce the number of hash passes a lot. Emergency behavior In emergency mode there a number of changes to the workers. In this scenario the flow memcap is fully used up and it is unavoidable that some flows won't be tracked. 1. flow spare pool fetches are reduced to once a second. This avoids locking overhead, while the chance of success was very low. 2. getting an active flow directly from the hash skips flows that had very recent activity to avoid the scenario where all flows get only into the NEW state before getting reused. Rather allow some to have a chance of completing. 3. TCP packets that are not SYN packets will not get a used flow, unless stream.midstream is enabled. The goal here is again to avoid evicting active flows unnecessarily. Better Localily Flow Manager injects flows into the worker threads now, instead of one or two packets. Advantage of this is that the worker threads can get packets from their local packet pools, avoiding constant overhead of packets returning to 'foreign' pools. Counters A lot of flow counters have been added and some have been renamed. Overall the worker threads increment 'flow.wrk.*' counters, while the flow manager increments 'flow.mgr.*'. Additionally, none of the counters are snapshots anymore, they all increment over time. The flow.memuse and flow.spare counters are exceptions. Misc FlowQueue has been split into a FlowQueuePrivate (unlocked) and FlowQueue. Flow no longer has 'prev' pointers and used a unified 'next' pointer for both hash and queue use.pull/5281/head
Victor Julien
7 years ago
parent
e0aa7c1dbc
commit
b3599507f4
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,260 @@
|
||||
/* Copyright (C) 2007-2020 Open Information Security Foundation
|
||||
*
|
||||
* You can copy, redistribute or modify this Program under the terms of
|
||||
* the GNU General Public License version 2 as published by the Free
|
||||
* Software Foundation.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License
|
||||
* version 2 along with this program; if not, write to the Free Software
|
||||
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
|
||||
* 02110-1301, USA.
|
||||
*/
|
||||
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \author Victor Julien <victor@inliniac.net>
|
||||
*
|
||||
* Flow queue handler functions
|
||||
*/
|
||||
|
||||
#include "suricata-common.h"
|
||||
#include "threads.h"
|
||||
#include "debug.h"
|
||||
#include "flow-private.h"
|
||||
#include "flow-queue.h"
|
||||
#include "flow-util.h"
|
||||
#include "flow-spare-pool.h"
|
||||
#include "util-error.h"
|
||||
#include "util-debug.h"
|
||||
#include "util-print.h"
|
||||
#include "util-validate.h"
|
||||
|
||||
typedef struct FlowSparePool {
|
||||
FlowQueuePrivate queue;
|
||||
struct FlowSparePool *next;
|
||||
} FlowSparePool;
|
||||
|
||||
static uint32_t flow_spare_pool_flow_cnt = 0;
|
||||
static uint32_t flow_spare_pool_block_size = 100;
|
||||
static FlowSparePool *flow_spare_pool = NULL;
|
||||
static SCMutex flow_spare_pool_m = SCMUTEX_INITIALIZER;
|
||||
|
||||
uint32_t FlowSpareGetPoolSize(void)
|
||||
{
|
||||
uint32_t size;
|
||||
SCMutexLock(&flow_spare_pool_m);
|
||||
size = flow_spare_pool_flow_cnt;
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
return size;
|
||||
}
|
||||
|
||||
static FlowSparePool *FlowSpareGetPool(void)
|
||||
{
|
||||
FlowSparePool *p = SCCalloc(1, sizeof(*p));
|
||||
if (p == NULL)
|
||||
return NULL;
|
||||
return p;
|
||||
}
|
||||
|
||||
static bool FlowSparePoolUpdateBlock(FlowSparePool *p)
|
||||
{
|
||||
DEBUG_VALIDATE_BUG_ON(p == NULL);
|
||||
|
||||
for (uint32_t i = p->queue.len; i < flow_spare_pool_block_size; i++)
|
||||
{
|
||||
Flow *f = FlowAlloc();
|
||||
if (f == NULL)
|
||||
return false;
|
||||
FlowQueuePrivateAppendFlow(&p->queue, f);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
#ifdef FSP_VALIDATE
|
||||
static void Validate(FlowSparePool *top, const uint32_t target)
|
||||
{
|
||||
if (top == NULL) {
|
||||
assert(target == 0);
|
||||
return;
|
||||
}
|
||||
|
||||
assert(top->queue.len >= 1);
|
||||
//if (top->next != NULL)
|
||||
// assert(top->next->queue.len == flow_spare_pool_block_size);
|
||||
|
||||
uint32_t cnt = 0;
|
||||
for (FlowSparePool *p = top; p != NULL; p = p->next)
|
||||
{
|
||||
assert(p->queue.len);
|
||||
cnt += p->queue.len;
|
||||
}
|
||||
assert(cnt == target);
|
||||
}
|
||||
#endif
|
||||
|
||||
void FlowSparePoolReturnFlow(Flow *f)
|
||||
{
|
||||
SCMutexLock(&flow_spare_pool_m);
|
||||
if (flow_spare_pool == NULL) {
|
||||
flow_spare_pool = FlowSpareGetPool();
|
||||
}
|
||||
DEBUG_VALIDATE_BUG_ON(flow_spare_pool == NULL);
|
||||
|
||||
/* if the top is full, get a new block */
|
||||
if (flow_spare_pool->queue.len >= flow_spare_pool_block_size) {
|
||||
FlowSparePool *p = FlowSpareGetPool();
|
||||
DEBUG_VALIDATE_BUG_ON(p == NULL);
|
||||
p->next = flow_spare_pool;
|
||||
flow_spare_pool = p;
|
||||
}
|
||||
/* add to the (possibly new) top */
|
||||
FlowQueuePrivateAppendFlow(&flow_spare_pool->queue, f);
|
||||
flow_spare_pool_flow_cnt++;
|
||||
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
}
|
||||
|
||||
void FlowSparePoolReturnFlows(FlowQueuePrivate *fqp)
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
FlowQueuePrivate FlowSpareGetFromPool(void)
|
||||
{
|
||||
SCMutexLock(&flow_spare_pool_m);
|
||||
if (flow_spare_pool == NULL || flow_spare_pool_flow_cnt == 0) {
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
FlowQueuePrivate empty = { NULL, NULL, 0 };
|
||||
return empty;
|
||||
}
|
||||
|
||||
/* top if full or its the only block we have */
|
||||
if (flow_spare_pool->queue.len >= flow_spare_pool_block_size || flow_spare_pool->next == NULL) {
|
||||
FlowSparePool *p = flow_spare_pool;
|
||||
flow_spare_pool = p->next;
|
||||
DEBUG_VALIDATE_BUG_ON(flow_spare_pool_flow_cnt < p->queue.len);
|
||||
flow_spare_pool_flow_cnt -= p->queue.len;
|
||||
#ifdef FSP_VALIDATE
|
||||
Validate(flow_spare_pool, flow_spare_pool_flow_cnt);
|
||||
#endif
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
|
||||
FlowQueuePrivate ret = p->queue;
|
||||
SCFree(p);
|
||||
return ret;
|
||||
/* next should always be full if it exists */
|
||||
} else if (flow_spare_pool->next != NULL) {
|
||||
FlowSparePool *p = flow_spare_pool->next;
|
||||
flow_spare_pool->next = p->next;
|
||||
DEBUG_VALIDATE_BUG_ON(flow_spare_pool_flow_cnt < p->queue.len);
|
||||
flow_spare_pool_flow_cnt -= p->queue.len;
|
||||
#ifdef FSP_VALIDATE
|
||||
Validate(flow_spare_pool, flow_spare_pool_flow_cnt);
|
||||
#endif
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
|
||||
FlowQueuePrivate ret = p->queue;
|
||||
SCFree(p);
|
||||
return ret;
|
||||
}
|
||||
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
FlowQueuePrivate empty = { NULL, NULL, 0 };
|
||||
return empty;
|
||||
}
|
||||
|
||||
void FlowSparePoolUpdate(uint32_t size)
|
||||
{
|
||||
const int64_t todo = (int64_t)flow_config.prealloc - (int64_t)size;
|
||||
if (todo < 0) {
|
||||
// remove
|
||||
} else if (todo > 0) {
|
||||
FlowSparePool *head = NULL, *tail = NULL;
|
||||
|
||||
uint32_t blocks = ((uint32_t)todo / flow_spare_pool_block_size) + 1;
|
||||
|
||||
uint32_t flow_cnt = 0;
|
||||
for (uint32_t cnt = 0; cnt < blocks; cnt++) {
|
||||
FlowSparePool *p = FlowSpareGetPool();
|
||||
if (p == NULL) {
|
||||
break;
|
||||
}
|
||||
const bool ok = FlowSparePoolUpdateBlock(p);
|
||||
if (p->queue.len == 0) {
|
||||
SCFree(p);
|
||||
break;
|
||||
}
|
||||
flow_cnt += p->queue.len;
|
||||
|
||||
/* prepend to list */
|
||||
p->next = head;
|
||||
head = p;
|
||||
if (tail == NULL)
|
||||
tail = p;
|
||||
if (!ok)
|
||||
break;
|
||||
}
|
||||
if (head) {
|
||||
SCMutexLock(&flow_spare_pool_m);
|
||||
if (flow_spare_pool == NULL) {
|
||||
flow_spare_pool = head;
|
||||
} else if (tail != NULL) {
|
||||
/* since these are 'full' buckets we don't put them
|
||||
* at the top but right after as the top is likely not
|
||||
* full. */
|
||||
tail->next = flow_spare_pool->next;
|
||||
flow_spare_pool->next = head;
|
||||
}
|
||||
|
||||
flow_spare_pool_flow_cnt += flow_cnt;
|
||||
#ifdef FSP_VALIDATE
|
||||
Validate(flow_spare_pool, flow_spare_pool_flow_cnt);
|
||||
#endif
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void FlowSparePoolInit(void)
|
||||
{
|
||||
SCMutexLock(&flow_spare_pool_m);
|
||||
for (uint32_t cnt = 0; cnt < flow_config.prealloc; ) {
|
||||
FlowSparePool *p = FlowSpareGetPool();
|
||||
if (p == NULL) {
|
||||
FatalError(SC_ERR_FLOW_INIT, "failed to initialize flow pool");
|
||||
}
|
||||
FlowSparePoolUpdateBlock(p);
|
||||
cnt += p->queue.len;
|
||||
|
||||
/* prepend to list */
|
||||
p->next = flow_spare_pool;
|
||||
flow_spare_pool = p;
|
||||
flow_spare_pool_flow_cnt = cnt;
|
||||
}
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
}
|
||||
|
||||
void FlowSparePoolDestroy(void)
|
||||
{
|
||||
SCMutexLock(&flow_spare_pool_m);
|
||||
for (FlowSparePool *p = flow_spare_pool; p != NULL; ) {
|
||||
uint32_t cnt = 0;
|
||||
Flow *f;
|
||||
while ((f = FlowQueuePrivateGetFromTop(&p->queue))) {
|
||||
FlowFree(f);
|
||||
cnt++;
|
||||
}
|
||||
flow_spare_pool_flow_cnt -= cnt;
|
||||
FlowSparePool *next = p->next;
|
||||
SCFree(p);
|
||||
p = next;
|
||||
}
|
||||
flow_spare_pool = NULL;
|
||||
SCMutexUnlock(&flow_spare_pool_m);
|
||||
}
|
||||
@ -0,0 +1,41 @@
|
||||
/* Copyright (C) 2007-2020 Open Information Security Foundation
|
||||
*
|
||||
* You can copy, redistribute or modify this Program under the terms of
|
||||
* the GNU General Public License version 2 as published by the Free
|
||||
* Software Foundation.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License
|
||||
* version 2 along with this program; if not, write to the Free Software
|
||||
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
|
||||
* 02110-1301, USA.
|
||||
*/
|
||||
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \author Victor Julien <victor@inliniac.net>
|
||||
*/
|
||||
|
||||
#ifndef __FLOW_SPARE_POOL_H__
|
||||
#define __FLOW_SPARE_POOL_H__
|
||||
|
||||
#include "suricata-common.h"
|
||||
#include "flow.h"
|
||||
|
||||
void FlowSparePoolInit(void);
|
||||
void FlowSparePoolDestroy(void);
|
||||
void FlowSparePoolUpdate(uint32_t size);
|
||||
|
||||
uint32_t FlowSpareGetPoolSize(void);
|
||||
|
||||
FlowQueuePrivate FlowSpareGetFromPool(void);
|
||||
|
||||
void FlowSparePoolReturnFlow(Flow *f);
|
||||
void FlowSparePoolReturnFlows(FlowQueuePrivate *fqp);
|
||||
|
||||
#endif /* __FLOW_SPARE_POOL_H__ */
|
||||
Loading…
Reference in New Issue