From mboxrd@z Thu Jan 1 00:00:00 1970 Authentication-Results: passt.top; dmarc=pass (p=quarantine dis=none) header.from=redhat.com Authentication-Results: passt.top; dkim=pass (1024-bit key; unprotected) header.d=redhat.com header.i=@redhat.com header.a=rsa-sha256 header.s=mimecast20190719 header.b=TX287pof; dkim-atps=neutral Received: from us-smtp-delivery-124.mimecast.com (us-smtp-delivery-124.mimecast.com [170.10.133.124]) by passt.top (Postfix) with ESMTPS id 7918A5A026D for ; Fri, 31 Jul 2026 18:23:44 +0200 (CEST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=redhat.com; s=mimecast20190719; t=1785515023; h=from:from:reply-to:subject:subject:date:date:message-id:message-id: to:to:cc:cc:mime-version:mime-version:content-type:content-type: content-transfer-encoding:content-transfer-encoding: in-reply-to:in-reply-to:references:references; bh=XmIqdNKbmQTrJST8Tu2tpR/Z07PtIHmgFo9BQrrtkKM=; b=TX287pof+GyQ+nDKGEWVAomKynkjyJ36lzRxA8rwFjbhr3PklhilKYaR/2i6lot1vs3EEC WSuRH8nSEjM8xcttvO+feehu8NznGQcoELp6/SjtZ/fd/821wbqDLJzquy6cRoBkXGOH4C aiykxpqb2cVej/zGZEQjzy8bVF9FN2I= Received: from mx-prod-mc-05.mail-002.prod.us-west-2.aws.redhat.com (ec2-54-186-198-63.us-west-2.compute.amazonaws.com [54.186.198.63]) by relay.mimecast.com with ESMTP with STARTTLS (version=TLSv1.3, cipher=TLS_AES_256_GCM_SHA384) id us-mta-582-aJiksK3_M2u2rF-TSEjFmQ-1; Fri, 31 Jul 2026 12:23:42 -0400 X-MC-Unique: aJiksK3_M2u2rF-TSEjFmQ-1 X-Mimecast-MFC-AGG-ID: aJiksK3_M2u2rF-TSEjFmQ_1785515021 Received: from mx-prod-int-05.mail-002.prod.us-west-2.aws.redhat.com (mx-prod-int-05.mail-002.prod.us-west-2.aws.redhat.com [10.30.177.17]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange X25519 server-signature RSA-PSS (2048 bits) server-digest SHA256) (No client certificate requested) by mx-prod-mc-05.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTPS id 4C0891956088 for ; Fri, 31 Jul 2026 16:23:41 +0000 (UTC) Received: from lenovo-t14s.redhat.corp (headnet05.pony-001.prod.iad2.dc.redhat.com [10.2.32.117]) by mx-prod-int-05.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTP id 6F0D81956044; Fri, 31 Jul 2026 16:23:40 +0000 (UTC) From: Laurent Vivier To: passt-dev@passt.top Subject: [PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling Date: Fri, 31 Jul 2026 18:23:28 +0200 Message-ID: <20260731162329.3552800-10-lvivier@redhat.com> In-Reply-To: <20260731162329.3552800-1-lvivier@redhat.com> References: <20260731162329.3552800-1-lvivier@redhat.com> MIME-Version: 1.0 X-Scanned-By: MIMEDefang 3.0 on 10.30.177.17 X-Mimecast-Spam-Score: 0 X-Mimecast-MFC-PROC-ID: JFt-E8ZgZc6EWT8z4sWJGKDBOTFqcSGjUvDUVl_9ER8_1785515021 X-Mimecast-Originator: redhat.com Content-Transfer-Encoding: 8bit content-type: text/plain; charset="US-ASCII"; x-default=true Message-ID-Hash: QQHR3X4IBKQ7L5OL4I3Q5JR7TA7HE6RL X-Message-ID-Hash: QQHR3X4IBKQ7L5OL4I3Q5JR7TA7HE6RL X-MailFrom: lvivier@redhat.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; emergency; loop; banned-address; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header CC: Laurent Vivier X-Mailman-Version: 3.3.8 Precedence: list List-Id: Development discussion and patches for passt Archived-At: Archived-At: List-Archive: List-Archive: List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: 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