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 <lvivier@redhat.com>
> ---
>  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