Re: [PATCH v6 11/12] flow: Add queue pair tracking to flow management

On Fri, Jul 31, 2026 at 06:16:16PM +0200, Laurent Vivier wrote:

Add a qpair field to struct flow_common so each flow knows which queue pair (and thus which worker thread) owns it.

flow_alloc() takes a qpair parameter, set at creation time. All protocol handlers pass their qpair when allocating flows from tap, and use QPAIR_DEFAULT for flows initiated from the host side.

tcp_keepalive() and tcp_inactivity() filter by qpair so each worker only processes its own flows.

Signed-off-by: Laurent Vivier --- flow.c | 34 +++++++++++++++++++++++++++++++++- flow.h | 13 ++++++++++++- flow_table.h | 2 +- icmp.c | 3 +-- tcp.c | 20 +++++++++++++------- udp_flow.c | 9 ++++----- 6 files changed, 64 insertions(+), 17 deletions(-)

diff --git a/flow.c b/flow.c index c63491804709..40b2e68be0ab 100644 --- a/flow.c +++ b/flow.c @@ -417,6 +417,37 @@ void flow_epollid_register(int epollid, int epollfd) epoll_id_to_fd[epollid] = epollfd; }

+/** + * flow_qp() - Get the queue pair for a flow + * @f: Flow to query (may be NULL) + * + * Return: queue pair number for the flow, or 0 if flow is NULL or has no

This implies it's safe to call on NULL, which doesn't appear to be the case. 0 also seems a dubious value for "no valid queue pair assignment", since it is a valid qpair number.

+ * valid queue pair assignment + */ +/* cppcheck-suppress unusedFunction */ +unsigned int flow_qp(const struct flow_common *f) +{ + return f->qpair; +} + +/** + * flow_setqp() - Set queue pair assignment for a flow + * @f: Flow to update + * @qpair: Queue pair number to assign + */ +static void flow_setqp(struct flow_common *f, unsigned int qpair) +{ + assert(qpair < FLOW_QPAIR_MAX); + + if (f->qpair == qpair) + return; + + flow_trace((union flow *)f, "updating queue pair from %d to %d", + f->qpair, qpair); + + f->qpair = qpair; +} + /** * flow_initiate_() - Move flow to INI, setting pif[INISIDE] * @flow: Flow to change state @@ -603,7 +634,7 @@ void flow_activate(struct flow_common *f) * * Return: pointer to an unused flow entry, or NULL if the table is full */ -union flow *flow_alloc(void) +union flow *flow_alloc(unsigned int qpair)

Comment for new parameter?

{ union flow *flow = &flowtab[flow_first_free];

@@ -637,6 +668,7 @@ union flow *flow_alloc(void)

flow_new_entry = flow; memset(flow, 0, sizeof(*flow)); + flow_setqp(&flow->f, qpair); flow_set_state(&flow->f, FLOW_STATE_NEW);

return flow; diff --git a/flow.h b/flow.h index 8d85e27427db..b6f980b4f826 100644 --- a/flow.h +++ b/flow.h @@ -184,7 +184,7 @@ int flowside_connect(const struct ctx *c, int s, * @pif[]: Interface for each side of the flow * @side[]: Information for each side of the flow * @tap_omac: MAC address of remote endpoint as seen from the guest - * @epollid: epollfd identifier + * @qpair: Queue pair number assigned to this flow

In the comments, @epollfd is replaced with @qpair, but in the structure, @qpair is added as well as @epollfd. Did you mean to leave the comment, or remove the field?

*/ struct flow_common { #ifdef __GNUC__ @@ -205,11 +205,18 @@ struct flow_common {

#define EPOLLFD_ID_BITS 8 unsigned int epollid:EPOLLFD_ID_BITS; +#define FLOW_QPAIR_BITS 5 + unsigned int qpair:FLOW_QPAIR_BITS; };

#define EPOLLFD_ID_DEFAULT 0 #define EPOLLFD_ID_SIZE (1 << EPOLLFD_ID_BITS)

+#define FLOW_QPAIR_NUM (1 << FLOW_QPAIR_BITS) +#define FLOW_QPAIR_MAX (FLOW_QPAIR_NUM - 1) + +static_assert(VHOST_USER_MAX_VQS <= FLOW_QPAIR_MAX * 2); + #define FLOW_INDEX_BITS 17 /* 128k - 1 */ #define FLOW_MAX MAX_FROM_BITS(FLOW_INDEX_BITS)

@@ -270,6 +277,10 @@ void flow_epollid_set(struct flow_common *f, int epollid); int flow_epoll_set(const struct flow_common *f, int command, uint32_t events, int fd, unsigned int sidei); void flow_epollid_register(int epollid, int epollfd); +unsigned int flow_qp(const struct flow_common *f); +#define FLOW_QP(flow_) \ + (flow_qp(&(flow_)->f)) + void flow_defer_handler(const struct ctx *c, const struct timespec *now, unsigned int qpair); int flow_migrate_source_early(struct ctx *c, const struct migrate_stage *stage, diff --git a/flow_table.h b/flow_table.h index e4ff6f73c35c..3a33eef15f1e 100644 --- a/flow_table.h +++ b/flow_table.h @@ -196,7 +196,7 @@ static inline flow_sidx_t flow_sidx(const struct flow_common *f, */ #define FLOW_SIDX(f_, sidei) (flow_sidx(&(f_)->f, (sidei)))

-union flow *flow_alloc(void); +union flow *flow_alloc(unsigned int qpair); void flow_alloc_cancel(union flow *flow);

const struct flowside *flow_initiate_af(union flow *flow, uint8_t pif, diff --git a/icmp.c b/icmp.c index 92cb48dfaafe..e75d66d168be 100644 --- a/icmp.c +++ b/icmp.c @@ -183,11 +183,10 @@ static struct icmp_ping_flow *icmp_ping_new(const struct ctx *c, { uint8_t proto = af == AF_INET ? IPPROTO_ICMP : IPPROTO_ICMPV6; uint8_t flowtype = af == AF_INET ? FLOW_PING4 : FLOW_PING6; - union flow *flow = flow_alloc(); + union flow *flow = flow_alloc(qpair); struct icmp_ping_flow *pingf; const struct flowside *tgt;

- (void)qpair; if (!flow) return NULL;

diff --git a/tcp.c b/tcp.c index 982f8a4a845a..39f732f8bea3 100644 --- a/tcp.c +++ b/tcp.c @@ -1707,7 +1707,7 @@ static void tcp_conn_from_tap(const struct ctx *c, unsigned int qpair, int s = -1, mss; uint64_t hash;

- if (!(flow = flow_alloc())) + if (!(flow = flow_alloc(qpair))) return;

ini = flow_initiate_af(flow, PIF_TAP, @@ -2304,7 +2304,7 @@ static void tcp_rst_no_conn(const struct ctx *c, unsigned int qpair, int af, /** * tcp_tap_handler() - Handle packets from tap and state transitions * @c: Execution context - * @qpair: Queue pair on which to send packets + * @qpair: Queue pair to process

This looks like a correction to an earlier patch in the series, which ideally would be folded in there. Plus, "Queue pair to process" isn't terribly informative as noted elsewhere.

* @pif: pif on which the packet is arriving * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address @@ -2625,7 +2625,7 @@ void tcp_listen_handler(const struct ctx *c, union epoll_ref ref,

assert(!c->no_tcp);

- if (!(flow = flow_alloc())) + if (!(flow = flow_alloc(QPAIR_DEFAULT))) return;

s = accept4(ref.fd, &sa.sa, &sl, SOCK_NONBLOCK); @@ -3007,6 +3007,9 @@ static void tcp_keepalive(struct ctx *c, const struct timespec *now, flow_foreach_of_type(flow, FLOW_TCP) { struct tcp_tap_conn *conn = &flow->tcp;

+ if (conn->f.qpair != qpair) + continue; + if (conn->tap_inactive) { flow_dbg(conn, "No tap activity for least %us, send keepalive", KEEPALIVE_INTERVAL); @@ -3038,6 +3041,9 @@ static void tcp_inactivity(struct ctx *c, const struct timespec *now, flow_foreach_of_type(flow, FLOW_TCP) { struct tcp_tap_conn *conn = &flow->tcp;

+ if (conn->f.qpair != qpair) + continue; + if (conn->inactive) { /* No activity in this interval, reset */ flow_dbg(conn, "Inactive for at least %us, resetting", @@ -3828,7 +3834,7 @@ int tcp_flow_migrate_target(struct ctx *c, int fd) union flow *flow; int rc;

- if (!(flow = flow_alloc())) { + if (!(flow = flow_alloc(QPAIR_DEFAULT))) { err("Flow table full on migration target"); return 0; } @@ -4050,10 +4056,10 @@ int tcp_flow_migrate_target_ext(struct ctx *c, struct tcp_tap_conn *conn, if (tcp_set_peek_offset(conn, peek_offset, now)) goto fail;

- if (tcp_send_flag(c, conn, ACK, now, QPAIR_DEFAULT)) + if (tcp_send_flag(c, conn, ACK, now, conn->f.qpair)) goto fail;

- tcp_data_from_sock(c, conn, now, QPAIR_DEFAULT); + tcp_data_from_sock(c, conn, now, conn->f.qpair);

if ((rc = tcp_epoll_ctl(conn))) { flow_dbg(conn, @@ -4071,7 +4077,7 @@ fail: }

conn->flags = 0; /* Not waiting for ACK, don't schedule timer */ - tcp_rst(c, conn, now, QPAIR_DEFAULT); + tcp_rst(c, conn, now, conn->f.qpair);

return 0; } diff --git a/udp_flow.c b/udp_flow.c index ff6abcb8eba0..2578d640a3da 100644 --- a/udp_flow.c +++ b/udp_flow.c @@ -84,7 +84,6 @@ static int udp_flow_sock(const struct ctx *c, return s; }

- flow_epollid_set(&uflow->f, EPOLLFD_ID_DEFAULT); if (flow_epoll_set(&uflow->f, EPOLL_CTL_ADD, EPOLLIN, s, sidei) < 0) { rc = -errno; close(s); @@ -155,6 +154,7 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow, uflow->ttl[INISIDE] = uflow->ttl[TGTSIDE] = 0; uflow->activity[INISIDE] = 1; uflow->activity[TGTSIDE] = 0; + flow_epollid_set(&uflow->f, EPOLLFD_ID_DEFAULT);

Moving flow_epollid_set() looks correct (it was in a pretty odd place before). Arguably, not logically related to this patch, though.

flow_foreach_sidei(sidei) { if (pif_is_socket(uflow->f.pif[sidei])) @@ -236,8 +236,7 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, unsigned int qpair, return flow_sidx_opposite(sidx); }

- (void)qpair; - if (!(flow = flow_alloc())) { + if (!(flow = flow_alloc(qpair))) { char sastr[SOCKADDR_STRLEN];

err_ratelimit(now, "Couldn't allocate flow for UDP datagram from %s %s", @@ -273,6 +272,7 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, unsigned int qpair, * @daddr: Destination address guest side * @srcport: Source port on guest side * @dstport: Destination port on guest side + * @now: Current timestamp

This also looks like a (correct) fix that's more or less unrelated to the rest of this patch.

* * Return: sidx for the destination side of the flow for this packet, or * FLOW_SIDX_NONE if we couldn't find or create a flow. @@ -297,8 +297,7 @@ flow_sidx_t udp_flow_from_tap(const struct ctx *c, unsigned int qpair, return flow_sidx_opposite(sidx); }

- (void)qpair; - if (!(flow = flow_alloc())) { + if (!(flow = flow_alloc(qpair))) { char sstr[INET6_ADDRSTRLEN], dstr[INET6_ADDRSTRLEN];

err_ratelimit(now, "Couldn't allocate flow for UDP datagram from %s %s:%hu -> %s:%hu", -- 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