[PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling

[Laurent Vivier <lvivier@redhat.com>]

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 <lvivier@redhat.com>
---
 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 <sched.h>
 #include <string.h>
 
+#include <pthread.h>
+
 #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