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=iS24gVNR; 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 DBEA95A026D for ; Fri, 31 Jul 2026 18:16:37 +0200 (CEST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=redhat.com; s=mimecast20190719; t=1785514596; 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=YlTRoc87/vxsRPRXFb2hpXD4gbhEmq+wk8PSAnGp310=; b=iS24gVNR/bFzJ//V7AstIPwBJCKmKjgfC22l+YIZC2gb0JrluP7lmDdnZpIhpNWoBiGqTt HtnZufP813T9ZzWkU1kbCotj3pV9nbQMlq+NE2RHH+k8imKGLit01oaDI/+M8m1HRvvhjW braJob8HHubL9KZZZJiQ35XgwA9G9ns= Received: from mx-prod-mc-03.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-470-D7Z7zRLnOyO_wRR78xHtoQ-1; Fri, 31 Jul 2026 12:16:35 -0400 X-MC-Unique: D7Z7zRLnOyO_wRR78xHtoQ-1 X-Mimecast-MFC-AGG-ID: D7Z7zRLnOyO_wRR78xHtoQ_1785514594 Received: from mx-prod-int-06.mail-002.prod.us-west-2.aws.redhat.com (mx-prod-int-06.mail-002.prod.us-west-2.aws.redhat.com [10.30.177.93]) (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-03.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTPS id 2ED9619560B7 for ; Fri, 31 Jul 2026 16:16:34 +0000 (UTC) Received: from lenovo-t14s.redhat.corp (headnet05.pony-001.prod.iad2.dc.redhat.com [10.2.32.117]) by mx-prod-int-06.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTP id 6B5F018004AD; Fri, 31 Jul 2026 16:16:33 +0000 (UTC) From: Laurent Vivier To: passt-dev@passt.top Subject: [PATCH v6 11/12] flow: Add queue pair tracking to flow management Date: Fri, 31 Jul 2026 18:16:16 +0200 Message-ID: <20260731161617.3550626-12-lvivier@redhat.com> In-Reply-To: <20260731161617.3550626-1-lvivier@redhat.com> References: <20260731161617.3550626-1-lvivier@redhat.com> MIME-Version: 1.0 X-Scanned-By: MIMEDefang 3.4.1 on 10.30.177.93 X-Mimecast-Spam-Score: 0 X-Mimecast-MFC-PROC-ID: 7bBQjDfhIucoyApNj8RydCcUq91QW7GRL18WzgXN5z4_1785514594 X-Mimecast-Originator: redhat.com Content-Transfer-Encoding: 8bit content-type: text/plain; charset="US-ASCII"; x-default=true Message-ID-Hash: YUK3NZ3FTIW2NXEBUEI2PXVCBUHIXOMC X-Message-ID-Hash: YUK3NZ3FTIW2NXEBUEI2PXVCBUHIXOMC 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: 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 + * 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) { 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 */ 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 * @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); 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 * * 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