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

[David Gibson <david@gibson.dropbear.id.au>]

[-- Attachment #1: Type: text/plain, Size: 12356 bytes --]

On Fri, Jul 31, 2026 at 06:23:28PM +0200, Laurent Vivier wrote:
> 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.

Oof, unfortunately, I'm not sure this is quite enough.  The free list
management was bespoke designed for our single threaded use case, and
I'm not sure the current approach quite works any more with
multithreading.  More below.

> 
> 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.

The way free cluster consolidation is done is because it was near-free
to do so, because we were already doing a regular linear scan through
the entire table.  Now that each defer handler is only handling a
subset of the entries it's not so clear that this approach makes sense
any more.  At the very least with free consolidation passes from every
thread, the effective cost might be significantly higher.

>   */
>  
>  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);

Alas, this is not safe.  Restoring flow_first_free to the flow index
relied on the fact that when single threaded, there could be no other
entries allocated between flow_alloc() and FLOW_ACTIVATE().  Because
we always allocated the first free entry, if we cancel it will become
the first free entry again.

But with the lock dropped between flow_alloc() and
flow_alloc_cancel(), this could be restoring an out of date snapshot
of flow_first_free - it might no longer be the actual first free entry
any more if another thread freed something earlier in the table.

Honestly, I'm not sure if flow_alloc_cancel() makes sense any more in
a multi-threaded context.  The rules about when it was and wasn't ok
were already pretty confusing, so maybe we should go through the full
flow close path even for flows cancelled very early.

>  	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);

It might not be necessary to move this under the lock.  This reads the
contents of the individual flow (which IIUC is not covered by the
flow_lock), but doesn't depend on any of the shared state data.

> -	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;

As with insert it might be worth using flow_sidx_hash() and
flow_hash_probe_() separately here, so that the flow_sidx_hash() can
move out of the critical section.

> +	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;

As above, I think the hash can go outside the lock.

>  	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;

I think these can all now be additional labels on the
FLOW_STATE_ACTIVE case.  At which point, this no longer really wants
to be a switch, but rather just an if (state == FLOW_STATE_FREE) else.

>  			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
> 

-- 
David Gibson (he or they)	| I'll have my music baroque, and my code
david AT gibson.dropbear.id.au	| minimalist, thank you, not the other way
				| around.
http://www.ozlabs.org/~dgibson

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]