Protect flow table shared state (free list, hash table) with a
pthread_rwlock_t. Make flow_new_entry _Thread_local.
Filter flow_defer_handler() by qpair so each thread only processes
its own flows.
Hold the write lock during flow_defer_handler()'s second pass
(free-list rebuild), since the free list is shared with flow_alloc().
Since the lock is released between flow_alloc() and FLOW_ACTIVATE(),
other threads can observe intermediate flow states (NEW, INI, TGT,
TYPED) during traversal. Adapt flow_foreach() to skip them silently
instead of logging an error, and change flow_defer_handler()'s
free-list rebuild to break cluster merging across them instead of
asserting.
Signed-off-by: Laurent Vivier
---
flow.c | 100 ++++++++++++++++++++++++++++++++++++++++++++-------
flow_table.h | 2 +-
2 files changed, 88 insertions(+), 14 deletions(-)
diff --git a/flow.c b/flow.c
index 59963ea5b1c2..8deca3e3c7f1 100644
--- a/flow.c
+++ b/flow.c
@@ -12,6 +12,8 @@
#include
#include
+#include
+
#include "util.h"
#include "ip.h"
#include "passt.h"
@@ -77,7 +79,10 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES,
/* Global Flow Table */
/**
- * DOC: Theory of Operation - allocating and freeing flow entries
+ * DOC: Theory of Operation
+ *
+ * Allocating and freeing flow entries
+ * ===================================
*
* Flows are entries in flowtab[]. We need to routinely scan the whole table to
* perform deferred bookkeeping tasks on active entries, and sparse empty slots
@@ -125,11 +130,43 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES,
* when we encounter the start of a free cluster, we can immediately skip
* past it, meaning that in practice we only need (number of active
* connections) + (number of free clusters) iterations.
+ *
+ * Flow table locking
+ * ==================
+ *
+ * The flow table has three pieces of shared global state: the free cluster
+ * list (flow_first_free, flowtab[].free), the hash table (flow_hashtab[]),
+ * and the in-progress allocation pointer (flow_new_entry).
+ *
+ * A pthread_rwlock_t (flow_lock) protects the free list and hash table:
+ *
+ * - flow_alloc() and flow_alloc_cancel() take the write lock to modify
+ * the free list.
+ *
+ * - flow_hash_insert() and flow_hash_remove() take the write lock to
+ * modify the hash table.
+ *
+ * - flowside_lookup() takes the read lock to traverse the hash table.
+ *
+ * flow_new_entry is _Thread_local, so each worker thread independently
+ * tracks its own in-progress flow allocation.
+ *
+ * Between flow_alloc() and FLOW_ACTIVATE() (or flow_alloc_cancel()), the
+ * flow is in an intermediate state (NEW, INI, TGT, or TYPED). Other
+ * threads may observe these states during flow table traversal.
+ * flow_foreach() skips them silently, and the free-list rebuild in
+ * flow_defer_handler() breaks cluster merging across them.
+ *
+ * The write lock is held in flow_defer_handler() only for the second
+ * pass (free-list rebuild), which modifies the global free list shared
+ * with flow_alloc(). The first pass (deferred protocol handlers) runs
+ * without the lock and filters by qpair, so each thread only processes
+ * its own flows.
*/
unsigned flow_first_free;
union flow flowtab[FLOW_MAX];
-static const union flow *flow_new_entry; /* = NULL */
+static _Thread_local const union flow *flow_new_entry; /* = NULL */
int qpair_to_fd[FLOW_QPAIR_SIZE];
/* Hash table to index it */
@@ -142,6 +179,8 @@ static flow_sidx_t flow_hashtab[FLOW_HASH_SIZE];
static_assert(ARRAY_SIZE(flow_hashtab) >= 2 * FLOW_MAX,
"Safe linear probing requires hash table with more entries than the number of sides in the flow table");
+static pthread_rwlock_t flow_lock = PTHREAD_RWLOCK_INITIALIZER;
+
/** flowside_from_af() - Initialise flowside from addresses
* @side: flowside to initialise
* @af: Address family (AF_INET or AF_INET6)
@@ -592,12 +631,18 @@ void flow_activate(struct flow_common *f)
*/
union flow *flow_alloc(unsigned int qpair)
{
- union flow *flow = &flowtab[flow_first_free];
+ union flow *flow;
+
+ pthread_rwlock_wrlock(&flow_lock);
+
+ flow = &flowtab[flow_first_free];
assert(!flow_new_entry);
- if (flow_first_free >= FLOW_MAX)
+ if (flow_first_free >= FLOW_MAX) {
+ pthread_rwlock_unlock(&flow_lock);
return NULL;
+ }
assert(flow->f.state == FLOW_STATE_FREE);
assert(flow->f.type == FLOW_TYPE_NONE);
@@ -627,6 +672,8 @@ union flow *flow_alloc(unsigned int qpair)
flow_setqp(&flow->f, qpair);
flow_set_state(&flow->f, FLOW_STATE_NEW);
+ pthread_rwlock_unlock(&flow_lock);
+
return flow;
}
@@ -638,6 +685,8 @@ union flow *flow_alloc(unsigned int qpair)
*/
void flow_alloc_cancel(union flow *flow)
{
+ pthread_rwlock_wrlock(&flow_lock);
+
assert(flow_new_entry == flow);
assert(flow->f.state == FLOW_STATE_NEW ||
flow->f.state == FLOW_STATE_INI ||
@@ -655,6 +704,8 @@ void flow_alloc_cancel(union flow *flow)
flow->free.next = flow_first_free;
flow_first_free = FLOW_IDX(flow);
flow_new_entry = NULL;
+
+ pthread_rwlock_unlock(&flow_lock);
}
/**
@@ -739,10 +790,15 @@ static inline unsigned flow_hash_probe(const struct ctx *c, flow_sidx_t sidx)
*/
uint64_t flow_hash_insert(const struct ctx *c, flow_sidx_t sidx)
{
- uint64_t hash = flow_sidx_hash(c, sidx);
- unsigned b = flow_hash_probe_(hash, sidx);
+ uint64_t hash;
+ unsigned b;
+ pthread_rwlock_wrlock(&flow_lock);
+ hash = flow_sidx_hash(c, sidx);
+ b = flow_hash_probe_(hash, sidx);
flow_hashtab[b] = sidx;
+ pthread_rwlock_unlock(&flow_lock);
+
flow_dbg(flow_at_sidx(sidx), "Side %u hash table insert: bucket: %u",
sidx.sidei, b);
@@ -756,10 +812,15 @@ uint64_t flow_hash_insert(const struct ctx *c, flow_sidx_t sidx)
*/
void flow_hash_remove(const struct ctx *c, flow_sidx_t sidx)
{
- unsigned b = flow_hash_probe(c, sidx), s;
+ unsigned b, s;
+
+ pthread_rwlock_wrlock(&flow_lock);
+ b = flow_hash_probe(c, sidx);
- if (!flow_sidx_valid(flow_hashtab[b]))
+ if (!flow_sidx_valid(flow_hashtab[b])) {
+ pthread_rwlock_unlock(&flow_lock);
return; /* Redundant remove */
+ }
flow_dbg(flow_at_sidx(sidx), "Side %u hash table remove: bucket: %u",
sidx.sidei, b);
@@ -779,6 +840,7 @@ void flow_hash_remove(const struct ctx *c, flow_sidx_t sidx)
}
flow_hashtab[b] = FLOW_SIDX_NONE;
+ pthread_rwlock_unlock(&flow_lock);
}
/**
@@ -793,10 +855,12 @@ void flow_hash_remove(const struct ctx *c, flow_sidx_t sidx)
static flow_sidx_t flowside_lookup(const struct ctx *c, uint8_t proto,
uint8_t pif, const struct flowside *side)
{
- flow_sidx_t sidx;
+ flow_sidx_t sidx, ret;
union flow *flow;
unsigned b;
+ pthread_rwlock_rdlock(&flow_lock);
+
b = flow_hash(c, proto, pif, side) % FLOW_HASH_SIZE;
while ((sidx = flow_hashtab[b], flow = flow_at_sidx(sidx)) &&
!(FLOW_PROTO(&flow->f) == proto &&
@@ -804,7 +868,11 @@ static flow_sidx_t flowside_lookup(const struct ctx *c, uint8_t proto,
flowside_eq(&flow->f.side[sidx.sidei], side)))
b = mod_sub(b, 1, FLOW_HASH_SIZE);
- return flow_hashtab[b];
+ ret = flow_hashtab[b];
+
+ pthread_rwlock_unlock(&flow_lock);
+
+ return ret;
}
/**
@@ -879,8 +947,8 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
struct timespec *timer_run, unsigned int qpair)
{
struct flow_free_cluster *free_head = NULL;
- unsigned *last_next = &flow_first_free;
bool to_free[FLOW_MAX] = { 0 };
+ unsigned *last_next;
bool timer = false;
union flow *flow;
@@ -897,6 +965,9 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
flow_foreach(flow) {
bool closed = false;
+ if (flow->f.qpair != qpair)
+ continue;
+
switch (flow->f.type) {
case FLOW_TYPE_NONE:
assert(false);
@@ -928,6 +999,8 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
}
/* Second step: actually free the flows */
+ pthread_rwlock_wrlock(&flow_lock);
+ last_next = &flow_first_free;
flow_foreach_slot(flow) {
switch (flow->f.state) {
case FLOW_STATE_FREE: {
@@ -956,8 +1029,8 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
case FLOW_STATE_INI:
case FLOW_STATE_TGT:
case FLOW_STATE_TYPED:
- /* Incomplete flow at end of cycle */
- assert(false);
+ /* In-progress allocation on another thread */
+ free_head = NULL;
break;
case FLOW_STATE_ACTIVE:
@@ -989,6 +1062,7 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
}
*last_next = FLOW_MAX;
+ pthread_rwlock_unlock(&flow_lock);
}
/**
diff --git a/flow_table.h b/flow_table.h
index 3a33eef15f1e..57944b748920 100644
--- a/flow_table.h
+++ b/flow_table.h
@@ -72,7 +72,7 @@ extern union flow flowtab[];
(flow) += (flow)->free.n - 1; \
/* NOLINTNEXTLINE(readability-inconsistent-ifelse-braces) */\
else if ((flow)->f.state != FLOW_STATE_ACTIVE) { \
- flow_err((flow), "Bad flow state during traversal"); \
+ (void)0; /* Differs from bare continue */ \
continue; \
} else
--
2.54.0