Allow the guest to negotiate multiple virtqueue pairs by advertising VIRTIO_NET_F_MQ and VHOST_USER_PROTOCOL_F_MQ feature flags, and increase VHOST_USER_MAX_VQS from 2 to 32, supporting up to 16 queue pairs. Replace the VHOST_USER_RX_QUEUE, VHOST_USER_TX_QUEUE, VHOST_USER_IS_QUEUE_TX(), and VHOST_USER_IS_QUEUE_RX() macros with a general set of QPAIR_* macros in passt.h that translate between queue pair numbers and virtqueue indices. These are needed now that queue indices are no longer limited to 0 and 1. Add a queue pair parameter to vu_send_single(), propagating it to the virtqueue selection. All callers currently pass QPAIR_DEFAULT (0): only the first RX queue is used for receiving. The guest kernel selects which TX queue to use for transmission. Full multi-RX-queue load balancing will be implemented separately. Signed-off-by: Laurent Vivier --- passt.h | 10 ++++++++++ tap.c | 9 ++++++--- tcp_vu.c | 6 ++++-- udp_vu.c | 3 ++- vhost_user.c | 12 +++++++----- vhost_user.h | 9 --------- virtio.h | 2 +- vu_common.c | 15 +++++++++------ vu_common.h | 3 ++- 9 files changed, 41 insertions(+), 28 deletions(-) diff --git a/passt.h b/passt.h index 51ccd4ff46a4..ec936c3d3f32 100644 --- a/passt.h +++ b/passt.h @@ -28,6 +28,16 @@ union epoll_ref; #include "udp.h" #include "vhost_user.h" +/* Queue pairs consist of one RX queue (even index) and one TX queue (odd index). + * Example: pair 0 has RX queue 0 and TX queue 1; pair 1 has RX queue 2 and TX queue 3. + */ +#define QPAIR_DEFAULT 0 /* Default queue pair */ +#define QPAIR_FROMGUEST_QUEUE(qpair) ((size_t)(qpair) * 2 + 1) /* TX queue index from pair */ +#define QPAIR_TOGUEST_QUEUE(qpair) ((size_t)(qpair) * 2) /* RX queue index from pair */ +#define QPAIR_FROM_QUEUE(queue) ((queue) / 2) /* Extract pair from queue */ +#define QPAIR_IS_FROMGUEST(queue) (!!(queue % 2)) /* Can receive data from guest */ +#define QPAIR_IS_TOGUEST(queue) (!(queue % 2)) /* Can send data to guest */ + /* Default address for our end on the tap interface. Bit 0 of byte 0 must be 0 * (unicast) and bit 1 of byte 1 must be 1 (locally administered). Otherwise * it's arbitrary. diff --git a/tap.c b/tap.c index dca53b9ec511..4956e4dc2589 100644 --- a/tap.c +++ b/tap.c @@ -155,7 +155,7 @@ void tap_send_single(const struct ctx *c, const void *data, size_t l2len) tap_send_frames(c, iov, iovcnt, 1); break; case MODE_VU: - vu_send_single(c, data, l2len); + vu_send_single(c, QPAIR_DEFAULT, data, l2len); break; } } @@ -1376,8 +1376,11 @@ static void tap_backend_show_hints(const struct ctx *c) break; case MODE_VU: info("You can start qemu with:"); - info(" kvm ... -chardev socket,id=chr0,path=%s -netdev vhost-user,id=netdev0,chardev=chr0 -device virtio-net,netdev=netdev0 -object memory-backend-memfd,id=memfd0,share=on,size=$RAMSIZE -numa node,memdev=memfd0\n", - c->sock_path); + info(" kvm ... -chardev socket,id=chr0,path=%s " + "-netdev vhost-user,id=netdev0,chardev=chr0,queues=$QUEUES " + "-device virtio-net,netdev=netdev0,mq=true " + "-object memory-backend-memfd,id=memfd0,share=on,size=$RAMSIZE " + "-numa node,memdev=memfd0\n", c->sock_path); break; } } diff --git a/tcp_vu.c b/tcp_vu.c index 868a35b968e1..d2950dcc99e6 100644 --- a/tcp_vu.c +++ b/tcp_vu.c @@ -126,7 +126,8 @@ int tcp_vu_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags, const struct timespec *now) { struct vu_dev *vdev = c->vdev; - struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE]; + int toguest = QPAIR_TOGUEST_QUEUE(QPAIR_DEFAULT); + struct vu_virtq *vq = &vdev->vq[toguest]; size_t optlen, hdrlen, iov_cnt, iov_used; struct vu_virtq_element flags_elem[2]; struct iov_tail payload, l2frame; @@ -434,8 +435,9 @@ int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn, uint32_t already_sent, const struct timespec *now) { uint32_t wnd_scaled = conn->wnd_from_tap << conn->ws_from_tap; + int toguest = QPAIR_TOGUEST_QUEUE(QPAIR_DEFAULT); struct vu_dev *vdev = c->vdev; - struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE]; + struct vu_virtq *vq = &vdev->vq[toguest]; ssize_t len, previous_dlen; int i, elem_cnt, frame_cnt; size_t hdrlen, fillsize; diff --git a/udp_vu.c b/udp_vu.c index edd9cae734a8..4003f5b42b56 100644 --- a/udp_vu.c +++ b/udp_vu.c @@ -150,9 +150,10 @@ void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx) const struct flowside *toside = flowside_at_sidx(tosidx); bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr)); static struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE]; + int toguest = QPAIR_TOGUEST_QUEUE(QPAIR_DEFAULT); static struct iovec iov_vu[VIRTQUEUE_MAX_SIZE]; struct vu_dev *vdev = c->vdev; - struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE]; + struct vu_virtq *vq = &vdev->vq[toguest]; size_t hdrlen = udp_vu_hdrlen(v6); int i; diff --git a/vhost_user.c b/vhost_user.c index a1259c2624c0..9e382c21f963 100644 --- a/vhost_user.c +++ b/vhost_user.c @@ -324,6 +324,7 @@ static bool vu_get_features_exec(struct vu_dev *vdev, 1ULL << VIRTIO_F_VERSION_1 | 1ULL << VIRTIO_NET_F_GUEST_CSUM | 1ULL << VIRTIO_NET_F_MRG_RXBUF | + 1ULL << VIRTIO_NET_F_MQ | 1ULL << VHOST_F_LOG_ALL | 1ULL << VHOST_USER_F_PROTOCOL_FEATURES; @@ -770,7 +771,8 @@ static void vu_check_queue_msg_file(struct vhost_user_msg *vmsg) bool nofd = vmsg->payload.u64 & VHOST_USER_VRING_NOFD_MASK; if (idx >= VHOST_USER_MAX_VQS) - die("Invalid vhost-user queue index: %u", idx); + die("Invalid vhost-user queue index: %u (maximum %u)", idx, + VHOST_USER_MAX_VQS - 1); if (nofd) { vmsg_close_fds(vmsg); @@ -812,7 +814,7 @@ static bool vu_set_vring_kick_exec(struct vu_dev *vdev, vdev->vq[idx].started = true; - if (vdev->vq[idx].kick_fd != -1 && VHOST_USER_IS_QUEUE_TX(idx)) { + if (vdev->vq[idx].kick_fd != -1 && QPAIR_IS_FROMGUEST(idx)) { vu_set_watch(vdev, idx); debug("Waiting for kicks on fd: %d for vq: %d", vdev->vq[idx].kick_fd, idx); @@ -899,7 +901,8 @@ static bool vu_get_protocol_features_exec(struct vu_dev *vdev, uint64_t features = 1ULL << VHOST_USER_PROTOCOL_F_REPLY_ACK | 1ULL << VHOST_USER_PROTOCOL_F_LOG_SHMFD | 1ULL << VHOST_USER_PROTOCOL_F_DEVICE_STATE | - 1ULL << VHOST_USER_PROTOCOL_F_RARP; + 1ULL << VHOST_USER_PROTOCOL_F_RARP | + 1ULL << VHOST_USER_PROTOCOL_F_MQ; (void)vdev; vmsg_set_reply_u64(vmsg, features); @@ -938,10 +941,9 @@ static bool vu_get_queue_num_exec(struct vu_dev *vdev, { (void)vdev; - /* NOLINTNEXTLINE(misc-redundant-expression) */ vmsg_set_reply_u64(vmsg, VHOST_USER_MAX_VQS / 2); - debug("VHOST_USER_MAX_VQS %u", VHOST_USER_MAX_VQS / 2); + debug("Using up to %u vhost-user queue pairs", VHOST_USER_MAX_VQS / 2); return true; } diff --git a/vhost_user.h b/vhost_user.h index e806a9e54e00..d2e51d3e86c3 100644 --- a/vhost_user.h +++ b/vhost_user.h @@ -201,15 +201,6 @@ struct vhost_user_msg { } __attribute__ ((__packed__)); #define VHOST_USER_HDR_SIZE sizeof(struct vhost_user_header) -/* index of the RX virtqueue */ -#define VHOST_USER_RX_QUEUE 0 -/* index of the TX virtqueue */ -#define VHOST_USER_TX_QUEUE 1 - -/* in case of multiqueue, the RX and TX queues are interleaved */ -#define VHOST_USER_IS_QUEUE_TX(n) (n % 2) -#define VHOST_USER_IS_QUEUE_RX(n) (!(n % 2)) - /* Default virtio-net header for passt */ #define VU_HEADER ((struct virtio_net_hdr){ \ .flags = VIRTIO_NET_HDR_F_DATA_VALID, \ diff --git a/virtio.h b/virtio.h index 8f2ae068d5ba..eb7570e63cd7 100644 --- a/virtio.h +++ b/virtio.h @@ -90,7 +90,7 @@ struct vu_dev_region { uint64_t mmap_addr; }; -#define VHOST_USER_MAX_VQS 2U +#define VHOST_USER_MAX_VQS 32U /* * Set a reasonable maximum number of ram slots, which will be supported by diff --git a/vu_common.c b/vu_common.c index d07f584f228a..8adcf42b954b 100644 --- a/vu_common.c +++ b/vu_common.c @@ -175,7 +175,7 @@ static void vu_handle_tx(struct vu_dev *vdev, int index, int out_sg_count; int count; - assert(VHOST_USER_IS_QUEUE_TX(index)); + assert(QPAIR_IS_FROMGUEST(index)); tap_flush_pools(); @@ -233,28 +233,31 @@ void vu_kick_cb(struct vu_dev *vdev, union epoll_ref ref, trace("vhost-user: got kick_data: %016"PRIx64" idx: %d", kick_data, ref.queue); - if (VHOST_USER_IS_QUEUE_TX(ref.queue)) + if (QPAIR_IS_FROMGUEST(ref.queue)) vu_handle_tx(vdev, ref.queue, now); } /** - * vu_send_single() - Send a buffer to the front-end using the RX virtqueue - * @c: execution context + * vu_send_single() - Send a buffer to the front-end using a specified virtqueue + * @c: Execution context + * @qpair: Queue pair on which to send the buffer * @buf: address of the buffer * @size: size of the buffer * * Return: number of bytes sent, -1 if there is an error */ -int vu_send_single(const struct ctx *c, const void *buf, size_t size) +int vu_send_single(const struct ctx *c, unsigned int qpair, const void *buf, size_t size) { struct vu_dev *vdev = c->vdev; - struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE]; struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE]; struct iovec in_sg[VIRTQUEUE_MAX_SIZE]; + struct vu_virtq *vq; size_t total, in_total; int elem_cnt; int i; + vq = &vdev->vq[QPAIR_TOGUEST_QUEUE(qpair)]; + trace("vu_send_single size %zu", size); if (!vu_queue_enabled(vq) || !vu_queue_started(vq)) { diff --git a/vu_common.h b/vu_common.h index 817384175a1d..f5603d9ddeb6 100644 --- a/vu_common.h +++ b/vu_common.h @@ -23,7 +23,8 @@ void vu_flush(const struct vu_dev *vdev, struct vu_virtq *vq, struct vu_virtq_element *elem, int elem_cnt, size_t frame_len); void vu_kick_cb(struct vu_dev *vdev, union epoll_ref ref, const struct timespec *now); -int vu_send_single(const struct ctx *c, const void *buf, size_t size); +int vu_send_single(const struct ctx *c, unsigned int qpair, const void *buf, + size_t size); void vu_pad(const struct iovec *iov, size_t cnt, size_t frame_len); #endif /* VU_COMMON_H */ -- 2.54.0

The previous commit threaded queue pair through vu_send_single(). Extend this to every other tap entry point: tap_send_single(), receive handlers (tap_add_packet, tap_handler, tap4/6_handler), and send helpers (tap_udp4/6_send, tap_icmp4/6_send). All callers pass QPAIR_DEFAULT except vu_handle_tx(), which derives the queue pair from the virtqueue index with QPAIR_FROM_QUEUE(). The parameter is plumbed but not yet consumed. Subsequent patches will use it to direct traffic to the correct queue pair. No functional change. Signed-off-by: Laurent Vivier --- arp.c | 6 ++--- dhcp.c | 3 ++- dhcpv6.c | 6 +++-- icmp.c | 6 +++-- ndp.c | 2 +- tap.c | 65 +++++++++++++++++++++++++++++++++-------------------- tap.h | 19 ++++++++-------- tcp.c | 2 +- udp.c | 4 ++-- vu_common.c | 7 +++--- 10 files changed, 72 insertions(+), 48 deletions(-) diff --git a/arp.c b/arp.c index bb042e9585a3..e97c4de86a99 100644 --- a/arp.c +++ b/arp.c @@ -112,7 +112,7 @@ int arp(const struct ctx *c, struct iov_tail *data) memcpy(resp.am.tha, am->sha, sizeof(resp.am.tha)); memcpy(resp.am.tip, am->sip, sizeof(resp.am.tip)); - tap_send_single(c, &resp, sizeof(resp)); + tap_send_single(c, QPAIR_DEFAULT, &resp, sizeof(resp)); return 1; } @@ -148,7 +148,7 @@ void arp_send_init_req(const struct ctx *c) memcpy(req.am.tip, &c->ip4.addr, sizeof(req.am.tip)); debug("Sending initial ARP request for guest MAC address"); - tap_send_single(c, &req, sizeof(req)); + tap_send_single(c, QPAIR_DEFAULT, &req, sizeof(req)); } /** @@ -202,5 +202,5 @@ void arp_announce(const struct ctx *c, struct in_addr *ip, eth_ntop(mac, mac_str, sizeof(mac_str)); debug("ARP announcement for %s / %s", ip_str, mac_str); - tap_send_single(c, &msg, sizeof(msg)); + tap_send_single(c, QPAIR_DEFAULT, &msg, sizeof(msg)); } diff --git a/dhcp.c b/dhcp.c index c3c742245ccf..1bd68ed5b87b 100644 --- a/dhcp.c +++ b/dhcp.c @@ -477,7 +477,8 @@ int dhcp(const struct ctx *c, struct iov_tail *data) else dst = c->ip4.addr; - tap_udp4_send(c, c->ip4.our_tap_addr, 67, dst, 68, &reply, dlen); + tap_udp4_send(c, QPAIR_DEFAULT, c->ip4.our_tap_addr, 67, dst, 68, + &reply, dlen); return 1; } diff --git a/dhcpv6.c b/dhcpv6.c index 29c7e320be8f..0986d6149515 100644 --- a/dhcpv6.c +++ b/dhcpv6.c @@ -407,7 +407,8 @@ static void dhcpv6_send_ia_notonlink(struct ctx *c, resp_not_on_link.hdr.xid = xid; - tap_udp6_send(c, src, 547, caddr, 546, xid, &resp_not_on_link, n); + tap_udp6_send(c, QPAIR_DEFAULT, src, 547, caddr, 546, xid, + &resp_not_on_link, n); } /** @@ -681,7 +682,8 @@ int dhcpv6(struct ctx *c, struct iov_tail *data, resp.hdr.xid = mh->xid; - tap_udp6_send(c, src, 547, saddr, 546, mh->xid, &resp, n); + tap_udp6_send(c, QPAIR_DEFAULT, src, 547, saddr, 546, mh->xid, &resp, + n); c->ip6.addr_seen = c->ip6.addr; return 1; diff --git a/icmp.c b/icmp.c index 0fe23667f517..a1015fe9f5c2 100644 --- a/icmp.c +++ b/icmp.c @@ -134,12 +134,14 @@ void icmp_sock_handler(const struct ctx *c, union epoll_ref ref, const struct in_addr *daddr = inany_v4(&ini->eaddr); assert(saddr && daddr); /* Must have IPv4 addresses */ - tap_icmp4_send(c, *saddr, *daddr, buf, pingf->f.tap_omac, n); + tap_icmp4_send(c, QPAIR_DEFAULT, *saddr, *daddr, buf, + pingf->f.tap_omac, n); } else if (pingf->f.type == FLOW_PING6) { const struct in6_addr *saddr = &ini->oaddr.a6; const struct in6_addr *daddr = &ini->eaddr.a6; - tap_icmp6_send(c, saddr, daddr, buf, pingf->f.tap_omac, n); + tap_icmp6_send(c, QPAIR_DEFAULT, saddr, daddr, buf, + pingf->f.tap_omac, n); } return; diff --git a/ndp.c b/ndp.c index 1f2bcb0cc7ea..6269cb38d93f 100644 --- a/ndp.c +++ b/ndp.c @@ -184,7 +184,7 @@ static void ndp_send(const struct ctx *c, const struct in6_addr *dst, { const struct in6_addr *src = &c->ip6.our_tap_ll; - tap_icmp6_send(c, src, dst, buf, c->our_tap_mac, l4len); + tap_icmp6_send(c, QPAIR_DEFAULT, src, dst, buf, c->our_tap_mac, l4len); } /** diff --git a/tap.c b/tap.c index 4956e4dc2589..206a1c36b779 100644 --- a/tap.c +++ b/tap.c @@ -124,10 +124,12 @@ unsigned long tap_l2_max_len(const struct ctx *c) /** * tap_send_single() - Send a single frame * @c: Execution context + * @qpair: Queue pair on which to send the frame * @data: Packet buffer * @l2len: Total L2 packet length */ -void tap_send_single(const struct ctx *c, const void *data, size_t l2len) +void tap_send_single(const struct ctx *c, unsigned int qpair, const void *data, + size_t l2len) { uint8_t padded[ETH_ZLEN] = { 0 }; struct iovec iov[2]; @@ -155,7 +157,7 @@ void tap_send_single(const struct ctx *c, const void *data, size_t l2len) tap_send_frames(c, iov, iovcnt, 1); break; case MODE_VU: - vu_send_single(c, QPAIR_DEFAULT, data, l2len); + vu_send_single(c, qpair, data, l2len); break; } } @@ -243,6 +245,7 @@ void *tap_push_uh4(struct udphdr *uh, struct in_addr src, in_port_t sport, /** * tap_udp4_send() - Send UDP over IPv4 packet * @c: Execution context + * @qpair: Queue pair on which to send packet * @src: IPv4 source address * @sport: UDP source port * @dst: IPv4 destination address @@ -250,8 +253,8 @@ void *tap_push_uh4(struct udphdr *uh, struct in_addr src, in_port_t sport, * @in: UDP payload contents (not including UDP header) * @dlen: UDP payload length (not including UDP header) */ -void tap_udp4_send(const struct ctx *c, struct in_addr src, in_port_t sport, - struct in_addr dst, in_port_t dport, +void tap_udp4_send(const struct ctx *c, unsigned int qpair, struct in_addr src, + in_port_t sport, struct in_addr dst, in_port_t dport, const void *in, size_t dlen) { size_t l4len = dlen + sizeof(struct udphdr); @@ -261,20 +264,22 @@ void tap_udp4_send(const struct ctx *c, struct in_addr src, in_port_t sport, char *data = tap_push_uh4(uh, src, sport, dst, dport, in, dlen); memcpy(data, in, dlen); - tap_send_single(c, buf, dlen + (data - buf)); + tap_send_single(c, qpair, buf, dlen + (data - buf)); } /** * tap_icmp4_send() - Send ICMPv4 packet * @c: Execution context + * @qpair: Queue pair on which to send packet * @src: IPv4 source address * @dst: IPv4 destination address * @in: ICMP packet, including ICMP header * @src_mac: MAC address to be used as source for message * @l4len: ICMP packet length, including ICMP header */ -void tap_icmp4_send(const struct ctx *c, struct in_addr src, struct in_addr dst, - const void *in, const void *src_mac, size_t l4len) +void tap_icmp4_send(const struct ctx *c, unsigned int qpair, struct in_addr src, + struct in_addr dst, const void *in, const void *src_mac, + size_t l4len) { char buf[USHRT_MAX]; struct iphdr *ip4h = tap_push_l2h(c, buf, src_mac, ETH_P_IP); @@ -284,7 +289,7 @@ void tap_icmp4_send(const struct ctx *c, struct in_addr src, struct in_addr dst, memcpy(icmp4h, in, l4len); csum_icmp4(icmp4h, icmp4h + 1, l4len - sizeof(*icmp4h)); - tap_send_single(c, buf, l4len + ((char *)icmp4h - buf)); + tap_send_single(c, qpair, buf, l4len + ((char *)icmp4h - buf)); } /** @@ -348,6 +353,7 @@ void *tap_push_uh6(struct udphdr *uh, /** * tap_udp6_send() - Send UDP over IPv6 packet * @c: Execution context + * @qpair: Queue pair on which to send packet * @src: IPv6 source address * @sport: UDP source port * @dst: IPv6 destination address @@ -356,7 +362,7 @@ void *tap_push_uh6(struct udphdr *uh, * @in: UDP payload contents (not including UDP header) * @dlen: UDP payload length (not including UDP header) */ -void tap_udp6_send(const struct ctx *c, +void tap_udp6_send(const struct ctx *c, unsigned int qpair, const struct in6_addr *src, in_port_t sport, const struct in6_addr *dst, in_port_t dport, uint32_t flow, void *in, size_t dlen) @@ -369,19 +375,20 @@ void tap_udp6_send(const struct ctx *c, char *data = tap_push_uh6(uh, src, sport, dst, dport, in, dlen); memcpy(data, in, dlen); - tap_send_single(c, buf, dlen + (data - buf)); + tap_send_single(c, qpair, buf, dlen + (data - buf)); } /** * tap_icmp6_send() - Send ICMPv6 packet * @c: Execution context + * @qpair: Queue pair on which to send packet * @src: IPv6 source address * @dst: IPv6 destination address * @in: ICMP packet, including ICMP header * @src_mac: MAC address to be used as source for message * @l4len: ICMP packet length, including ICMP header */ -void tap_icmp6_send(const struct ctx *c, +void tap_icmp6_send(const struct ctx *c, unsigned int qpair, const struct in6_addr *src, const struct in6_addr *dst, const void *in, const void *src_mac, size_t l4len) { @@ -393,7 +400,7 @@ void tap_icmp6_send(const struct ctx *c, memcpy(icmp6h, in, l4len); csum_icmp6(icmp6h, src, dst, icmp6h + 1, l4len - sizeof(*icmp6h)); - tap_send_single(c, buf, l4len + ((char *)icmp6h - buf)); + tap_send_single(c, qpair, buf, l4len + ((char *)icmp6h - buf)); } /** @@ -684,15 +691,19 @@ static bool tap4_is_fragment(const struct iphdr *iph, /** * tap4_handler() - IPv4 and ARP packet handler for tap file descriptor * @c: Execution context + * @qpair: Queue pair on which to send packets * @now: Current timestamp * * Return: count of packets consumed by handlers */ -static int tap4_handler(struct ctx *c, const struct timespec *now) +static int tap4_handler(struct ctx *c, unsigned int qpair, + const struct timespec *now) { unsigned int i, j, seq_count; struct tap4_l4_t *seq; + (void)qpair; + if (!c->ifi4 || !pool_tap4->count) return pool_tap4->count; @@ -930,15 +941,19 @@ found: /** * tap6_handler() - IPv6 packet handler for tap file descriptor * @c: Execution context + * @qpair: Queue pair on which to send packets * @now: Current timestamp * * Return: count of packets consumed by handlers */ -static int tap6_handler(struct ctx *c, const struct timespec *now) +static int tap6_handler(struct ctx *c, unsigned int qpair, + const struct timespec *now) { unsigned int i, j, seq_count = 0; struct tap6_l4_t *seq; + (void)qpair; + if (!c->ifi6 || !pool_tap6->count) return pool_tap6->count; @@ -1136,21 +1151,23 @@ void tap_flush_pools(void) /** * tap_handler() - IPv4/IPv6 and ARP packet handler for tap file descriptor * @c: Execution context + * @qpair: Queue pair on which to send packets * @now: Current timestamp */ -void tap_handler(struct ctx *c, const struct timespec *now) +void tap_handler(struct ctx *c, unsigned int qpair, const struct timespec *now) { - tap4_handler(c, now); - tap6_handler(c, now); + tap4_handler(c, qpair, now); + tap6_handler(c, qpair, now); } /** * tap_add_packet() - Queue/capture packet, update notion of guest MAC address * @c: Execution context + * @qpair: Queue pair associated with the packet * @data: Packet to add to the pool * @now: Current timestamp */ -void tap_add_packet(struct ctx *c, struct iov_tail *data, +void tap_add_packet(struct ctx *c, unsigned int qpair, struct iov_tail *data, const struct timespec *now) { size_t l2len = iov_tail_size(data); @@ -1177,14 +1194,14 @@ void tap_add_packet(struct ctx *c, struct iov_tail *data, case ETH_P_ARP: case ETH_P_IP: if (!pool_can_fit(pool_tap4, data)) { - tap4_handler(c, now); + tap4_handler(c, qpair, now); pool_flush(pool_tap4); } packet_add(pool_tap4, data); break; case ETH_P_IPV6: if (!pool_can_fit(pool_tap6, data)) { - tap6_handler(c, now); + tap6_handler(c, qpair, now); pool_flush(pool_tap6); } packet_add(pool_tap6, data); @@ -1269,7 +1286,7 @@ static void tap_passt_input(struct ctx *c, const struct timespec *now) n -= sizeof(uint32_t); data = IOV_TAIL_FROM_BUF(p, l2len, 0); - tap_add_packet(c, &data, now); + tap_add_packet(c, QPAIR_DEFAULT, &data, now); p += l2len; n -= l2len; @@ -1278,7 +1295,7 @@ static void tap_passt_input(struct ctx *c, const struct timespec *now) partial_len = n; partial_frame = p; - tap_handler(c, now); + tap_handler(c, QPAIR_DEFAULT, now); } /** @@ -1337,10 +1354,10 @@ static void tap_pasta_input(struct ctx *c, const struct timespec *now) continue; data = IOV_TAIL_FROM_BUF(pkt_buf + n, len, 0); - tap_add_packet(c, &data, now); + tap_add_packet(c, QPAIR_DEFAULT, &data, now); } - tap_handler(c, now); + tap_handler(c, QPAIR_DEFAULT, now); } /** diff --git a/tap.h b/tap.h index b335933fdbc9..0f8e98fc5f84 100644 --- a/tap.h +++ b/tap.h @@ -91,22 +91,23 @@ void *tap_push_ip6h(struct ipv6hdr *ip6h, const struct in6_addr *src, const struct in6_addr *dst, size_t l4len, uint8_t proto, uint32_t flow); -void tap_udp4_send(const struct ctx *c, struct in_addr src, in_port_t sport, - struct in_addr dst, in_port_t dport, +void tap_udp4_send(const struct ctx *c, unsigned int qpair, struct in_addr src, + in_port_t sport, struct in_addr dst, in_port_t dport, const void *in, size_t dlen); -void tap_icmp4_send(const struct ctx *c, struct in_addr src, struct in_addr dst, - const void *in, const void *src_mac, size_t l4len); +void tap_icmp4_send(const struct ctx *c, unsigned int qpair, struct in_addr src, + struct in_addr dst, const void *in, const void *src_mac, + size_t l4len); void *tap_push_ip6h(struct ipv6hdr *ip6h, const struct in6_addr *src, const struct in6_addr *dst, size_t l4len, uint8_t proto, uint32_t flow); -void tap_udp6_send(const struct ctx *c, +void tap_udp6_send(const struct ctx *c, unsigned int qpair, const struct in6_addr *src, in_port_t sport, const struct in6_addr *dst, in_port_t dport, uint32_t flow, void *in, size_t dlen); -void tap_icmp6_send(const struct ctx *c, +void tap_icmp6_send(const struct ctx *c, unsigned int qpair, const struct in6_addr *src, const struct in6_addr *dst, const void *in, const void *src_mac, size_t l4len); -void tap_send_single(const struct ctx *c, const void *data, size_t l2len); +void tap_send_single(const struct ctx *c, unsigned int qpair, const void *data, size_t l2len); size_t tap_send_frames(const struct ctx *c, const struct iovec *iov, size_t bufs_per_frame, size_t nframes); void eth_update_mac(struct ethhdr *eh, @@ -121,7 +122,7 @@ int tap_sock_unix_open(char *sock_path); void tap_sock_reset(struct ctx *c); void tap_backend_init(struct ctx *c); void tap_flush_pools(void); -void tap_handler(struct ctx *c, const struct timespec *now); -void tap_add_packet(struct ctx *c, struct iov_tail *data, +void tap_handler(struct ctx *c, unsigned int qpair, const struct timespec *now); +void tap_add_packet(struct ctx *c, unsigned int qpair, struct iov_tail *data, const struct timespec *now); #endif /* TAP_H */ diff --git a/tcp.c b/tcp.c index 3b78d2edd03e..6b7612c27abf 100644 --- a/tcp.c +++ b/tcp.c @@ -2285,7 +2285,7 @@ static void tcp_rst_no_conn(const struct ctx *c, int af, tcp_update_csum(psum, rsth, &payload, 0); rst_l2len = ((char *)rsth - buf) + sizeof(*rsth); - tap_send_single(c, buf, rst_l2len); + tap_send_single(c, QPAIR_DEFAULT, buf, rst_l2len); } /** diff --git a/udp.c b/udp.c index 505e55407fb4..589d48a17613 100644 --- a/udp.c +++ b/udp.c @@ -445,7 +445,7 @@ static void udp_send_tap_icmp4(const struct ctx *c, /* Try to obtain the MAC address of the generating node */ saddr_any = inany_from_v4(saddr); fwd_neigh_mac_get(c, &saddr_any, tap_omac); - tap_icmp4_send(c, saddr, eaddr, &msg, tap_omac, msglen); + tap_icmp4_send(c, QPAIR_DEFAULT, saddr, eaddr, &msg, tap_omac, msglen); } @@ -493,7 +493,7 @@ static void udp_send_tap_icmp6(const struct ctx *c, /* Try to obtain the MAC address of the generating node */ fwd_neigh_mac_get(c, (union inany_addr *) saddr, tap_omac); - tap_icmp6_send(c, saddr, eaddr, &msg, tap_omac, msglen); + tap_icmp6_send(c, QPAIR_DEFAULT, saddr, eaddr, &msg, tap_omac, msglen); } /** diff --git a/vu_common.c b/vu_common.c index 8adcf42b954b..7239cb42f38b 100644 --- a/vu_common.c +++ b/vu_common.c @@ -198,12 +198,13 @@ static void vu_handle_tx(struct vu_dev *vdev, int index, } data = IOV_TAIL(elem[count].out_sg, elem[count].out_num, 0); - if (IOV_DROP_HEADER(&data, struct virtio_net_hdr_mrg_rxbuf)) - tap_add_packet(vdev->context, &data, now); + if (IOV_DROP_HEADER(&data, struct virtio_net_hdr_mrg_rxbuf)) { + tap_add_packet(vdev->context, QPAIR_DEFAULT, &data, now); + } count++; } - tap_handler(vdev->context, now); + tap_handler(vdev->context, QPAIR_DEFAULT, now); if (count) { int i; -- 2.54.0

Add a qpair parameter to arp(), forwarding it to tap_send_single() instead of hardcoding QPAIR_DEFAULT, so replies go out on the same queue pair the request arrived on. arp_send_init_req() and arp_announce() are unsolicited, low-volume messages with no meaningful queue pair context, so they hardcode QPAIR_DEFAULT internally. No functional change. Signed-off-by: Laurent Vivier --- arp.c | 5 +++-- arp.h | 2 +- tap.c | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/arp.c b/arp.c index e97c4de86a99..a12b50ff8439 100644 --- a/arp.c +++ b/arp.c @@ -63,11 +63,12 @@ static bool ignore_arp(const struct ctx *c, /** * arp() - Check if this is a supported ARP message, reply as needed * @c: Execution context + * @qpair: Queue pair on which to send the reply * @data: Single packet with Ethernet buffer * * Return: 1 if handled, -1 on failure */ -int arp(const struct ctx *c, struct iov_tail *data) +int arp(const struct ctx *c, unsigned int qpair, struct iov_tail *data) { union inany_addr tgt; struct { @@ -112,7 +113,7 @@ int arp(const struct ctx *c, struct iov_tail *data) memcpy(resp.am.tha, am->sha, sizeof(resp.am.tha)); memcpy(resp.am.tip, am->sip, sizeof(resp.am.tip)); - tap_send_single(c, QPAIR_DEFAULT, &resp, sizeof(resp)); + tap_send_single(c, qpair, &resp, sizeof(resp)); return 1; } diff --git a/arp.h b/arp.h index 4b1f38bcec9b..5872b535635a 100644 --- a/arp.h +++ b/arp.h @@ -22,7 +22,7 @@ struct arpmsg { unsigned char tip[4]; } __attribute__((__packed__)); -int arp(const struct ctx *c, struct iov_tail *data); +int arp(const struct ctx *c, unsigned int qpair, struct iov_tail *data); void arp_send_init_req(const struct ctx *c); void arp_announce(const struct ctx *c, struct in_addr *ip, const unsigned char *mac); diff --git a/tap.c b/tap.c index 206a1c36b779..928cf5c086ee 100644 --- a/tap.c +++ b/tap.c @@ -727,7 +727,7 @@ resume: if (!eh) continue; if (ntohs(eh->h_proto) == ETH_P_ARP) { - arp(c, &data); + arp(c, qpair, &data); continue; } -- 2.54.0

Thread the queue pair parameter through the UDP socket handler and send path, replacing hardcoded QPAIR_DEFAULT values. This is the UDP counterpart to the equivalent TCP and ARP changes. Add a qpair parameter to udp_listen_sock_handler(), udp_sock_handler(), udp_sock_fwd(), udp_sock_recverr(), udp_sock_errs(), udp_tap_handler(), udp_send_tap_icmp4(), and udp_send_tap_icmp6(). The queue pair is passed from passt_worker() through udp_listen_sock_handler() and udp_sock_handler(), into udp_sock_fwd() and down into udp_sock_errs()/udp_sock_recverr() for ICMP error generation, as well as udp_vu_sock_to_tap() for vhost-user delivery. On the tap side, udp_tap_handler() receives the queue pair and passes it to udp_flow_from_tap(), which takes the parameter for completeness but does not yet use it: new flows are not associated with a specific queue pair at creation time. On the flow handling side, flow_defer_handler() receives the queue pair and passes it to udp_flow_defer() and udp_flush_flow(), so that deferred UDP datagrams are forwarded on the correct queue. udp_vu_sock_to_tap() now uses the passed qpair to select the RX virtqueue instead of always using QPAIR_DEFAULT. No functional change. Signed-off-by: Laurent Vivier --- flow.c | 6 ++++-- flow.h | 3 ++- passt.c | 8 +++++--- tap.c | 4 ++-- udp.c | 48 +++++++++++++++++++++++++++--------------------- udp.h | 9 +++++---- udp_flow.c | 23 ++++++++++++++--------- udp_flow.h | 13 +++++++------ udp_internal.h | 3 ++- udp_vu.c | 6 ++++-- udp_vu.h | 3 ++- 11 files changed, 74 insertions(+), 52 deletions(-) diff --git a/flow.c b/flow.c index 71918b773738..c63491804709 100644 --- a/flow.c +++ b/flow.c @@ -884,8 +884,10 @@ flow_sidx_t flow_lookup_sa(const struct ctx *c, uint8_t proto, uint8_t pif, * flow_defer_handler() - Handler for per-flow deferred and timed tasks * @c: Execution context * @now: Current timestamp + * @qpair: Queue pair to process */ -void flow_defer_handler(const struct ctx *c, const struct timespec *now) +void flow_defer_handler(const struct ctx *c, const struct timespec *now, + unsigned int qpair) { struct flow_free_cluster *free_head = NULL; unsigned *last_next = &flow_first_free; @@ -924,7 +926,7 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now) closed = icmp_ping_timer(c, &flow->ping, now); break; case FLOW_UDP: - closed = udp_flow_defer(c, &flow->udp, now); + closed = udp_flow_defer(c, &flow->udp, now, qpair); if (!closed && timer) closed = udp_flow_timer(c, &flow->udp, now); break; diff --git a/flow.h b/flow.h index db028455a7a4..8d85e27427db 100644 --- a/flow.h +++ b/flow.h @@ -270,7 +270,8 @@ 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); -void flow_defer_handler(const struct ctx *c, const struct timespec *now); +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, int fd); int flow_migrate_source_pre(struct ctx *c, const struct migrate_stage *stage, diff --git a/passt.c b/passt.c index 86ed339ff67f..a26653db7763 100644 --- a/passt.c +++ b/passt.c @@ -116,7 +116,7 @@ static void post_handler(struct ctx *c, const struct timespec *now, if (!c->no_tcp) tcp_defer_handler(c, now, qpair); - flow_defer_handler(c, now); + flow_defer_handler(c, now, qpair); fwd_scan_ports_timer(c, now); if (!c->no_ndp) @@ -275,10 +275,12 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events) tcp_timer_handler(c, ref, &now, QPAIR_DEFAULT); break; case EPOLL_TYPE_UDP_LISTEN: - udp_listen_sock_handler(c, ref, eventmask, &now); + udp_listen_sock_handler(c, ref, eventmask, &now, + QPAIR_DEFAULT); break; case EPOLL_TYPE_UDP: - udp_sock_handler(c, ref, eventmask, &now); + udp_sock_handler(c, ref, eventmask, &now, + QPAIR_DEFAULT); break; case EPOLL_TYPE_PING: icmp_sock_handler(c, ref, &now); diff --git a/tap.c b/tap.c index 9ce32d36f1d7..61d18ee161b1 100644 --- a/tap.c +++ b/tap.c @@ -872,7 +872,7 @@ append: if (c->no_udp) continue; for (k = 0; k < p->count; ) - k += udp_tap_handler(c, PIF_TAP, AF_INET, + k += udp_tap_handler(c, qpair, PIF_TAP, AF_INET, &seq->saddr, &seq->daddr, seq->ttl, p, k, now); } @@ -1123,7 +1123,7 @@ append: if (c->no_udp) continue; for (k = 0; k < p->count; ) - k += udp_tap_handler(c, PIF_TAP, AF_INET6, + k += udp_tap_handler(c, qpair, PIF_TAP, AF_INET6, &seq->saddr, &seq->daddr, seq->hop_limit, p, k, now); } diff --git a/udp.c b/udp.c index 589d48a17613..a3321289bb53 100644 --- a/udp.c +++ b/udp.c @@ -403,13 +403,14 @@ static void udp_tap_prepare(const struct mmsghdr *mmh, /** * udp_send_tap_icmp4() - Construct and send ICMPv4 to local peer * @c: Execution context + * @qpair: Queue pair on which to send the ICMPv4 packet * @ee: Extended error descriptor * @toside: Destination side of flow * @saddr: Address of ICMP generating node * @in: First bytes (max 8) of original UDP message body * @dlen: Length of the read part of original UDP message body */ -static void udp_send_tap_icmp4(const struct ctx *c, +static void udp_send_tap_icmp4(const struct ctx *c, unsigned int qpair, const struct sock_extended_err *ee, const struct flowside *toside, struct in_addr saddr, @@ -445,13 +446,14 @@ static void udp_send_tap_icmp4(const struct ctx *c, /* Try to obtain the MAC address of the generating node */ saddr_any = inany_from_v4(saddr); fwd_neigh_mac_get(c, &saddr_any, tap_omac); - tap_icmp4_send(c, QPAIR_DEFAULT, saddr, eaddr, &msg, tap_omac, msglen); + tap_icmp4_send(c, qpair, saddr, eaddr, &msg, tap_omac, msglen); } /** * udp_send_tap_icmp6() - Construct and send ICMPv6 to local peer * @c: Execution context + * @qpair: Queue pair on which to send the ICMPv6 packet * @ee: Extended error descriptor * @toside: Destination side of flow * @saddr: Address of ICMP generating node @@ -459,7 +461,7 @@ static void udp_send_tap_icmp4(const struct ctx *c, * @dlen: Length of the read part of original UDP message body * @flow: IPv6 flow identifier */ -static void udp_send_tap_icmp6(const struct ctx *c, +static void udp_send_tap_icmp6(const struct ctx *c, unsigned int qpair, const struct sock_extended_err *ee, const struct flowside *toside, const struct in6_addr *saddr, @@ -493,7 +495,7 @@ static void udp_send_tap_icmp6(const struct ctx *c, /* Try to obtain the MAC address of the generating node */ fwd_neigh_mac_get(c, (union inany_addr *) saddr, tap_omac); - tap_icmp6_send(c, QPAIR_DEFAULT, saddr, eaddr, &msg, tap_omac, msglen); + tap_icmp6_send(c, qpair, saddr, eaddr, &msg, tap_omac, msglen); } /** @@ -548,7 +550,7 @@ static int udp_pktinfo(struct msghdr *msg, union inany_addr *dst) */ static int udp_sock_recverr(const struct ctx *c, int s, flow_sidx_t sidx, uint8_t pif, in_port_t port, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { char buf[PKTINFO_SPACE + RECVERR_SPACE]; const struct sock_extended_err *ee; @@ -655,12 +657,12 @@ static int udp_sock_recverr(const struct ctx *c, int s, flow_sidx_t sidx, if (hdr->cmsg_level == IPPROTO_IP && (o4 = inany_v4(&otap)) && inany_v4(&toside->eaddr)) { dlen = MIN(dlen, ICMP4_MAX_DLEN); - udp_send_tap_icmp4(c, ee, toside, *o4, data, dlen); + udp_send_tap_icmp4(c, qpair, ee, toside, *o4, data, dlen); return 1; } if (hdr->cmsg_level == IPPROTO_IPV6 && !inany_v4(&toside->eaddr)) { - udp_send_tap_icmp6(c, ee, toside, &otap.a6, data, dlen, + udp_send_tap_icmp6(c, qpair, ee, toside, &otap.a6, data, dlen, FLOW_IDX(uflow)); return 1; } @@ -691,7 +693,7 @@ fail: */ static int udp_sock_errs(const struct ctx *c, int s, flow_sidx_t sidx, uint8_t pif, in_port_t port, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { struct udp_flow *uflow = udp_at_sidx(sidx); unsigned n_err = 0; @@ -701,7 +703,7 @@ static int udp_sock_errs(const struct ctx *c, int s, flow_sidx_t sidx, assert(!c->no_udp); /* Empty the error queue */ - while ((rc = udp_sock_recverr(c, s, sidx, pif, port, now)) > 0) + while ((rc = udp_sock_recverr(c, s, sidx, pif, port, now, qpair)) > 0) n_err += rc; if (rc < 0) @@ -862,7 +864,8 @@ static void udp_buf_sock_to_tap(const struct ctx *c, int s, int n, * @now: Current timestamp */ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, - uint8_t frompif, in_port_t port, const struct timespec *now) + uint8_t frompif, in_port_t port, const struct timespec *now, + unsigned int qpair) { union sockaddr_inany src; union inany_addr dst; @@ -878,7 +881,7 @@ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, strerror_(-rc)); /* Clear errors & carry on */ if (udp_sock_errs(c, s, FLOW_SIDX_NONE, - frompif, port, now) < 0) { + frompif, port, now, qpair) < 0) { err_ratelimit(now, "UDP: Unrecoverable error on listening socket: (%s port %hu)", pif_name(frompif), port); @@ -887,7 +890,7 @@ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, continue; } - tosidx = udp_flow_from_sock(c, frompif, &dst, port, &src, + tosidx = udp_flow_from_sock(c, qpair, frompif, &dst, port, &src, rule_hint, now); topif = pif_at_sidx(tosidx); @@ -895,7 +898,7 @@ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, udp_sock_to_sock(c, s, 1, tosidx); } else if (topif == PIF_TAP) { if (c->mode == MODE_VU) - udp_vu_sock_to_tap(c, s, 1, tosidx); + udp_vu_sock_to_tap(c, s, 1, tosidx, qpair); else udp_buf_sock_to_tap(c, s, 1, tosidx); } else if (flow_sidx_valid(tosidx)) { @@ -926,14 +929,15 @@ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, * @ref: epoll reference * @events: epoll events bitmap * @now: Current timestamp + * @qpair: Queue pair to process */ void udp_listen_sock_handler(const struct ctx *c, union epoll_ref ref, uint32_t events, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { if (events & (EPOLLERR | EPOLLIN)) { udp_sock_fwd(c, ref.fd, ref.listen.rule, - ref.listen.pif, ref.listen.port, now); + ref.listen.pif, ref.listen.port, now, qpair); } } @@ -943,9 +947,10 @@ void udp_listen_sock_handler(const struct ctx *c, * @ref: epoll reference * @events: epoll events bitmap * @now: Current timestamp + * @qpair: Queue pair to process */ -void udp_sock_handler(const struct ctx *c, union epoll_ref ref, - uint32_t events, const struct timespec *now) +void udp_sock_handler(const struct ctx *c, union epoll_ref ref, uint32_t events, + const struct timespec *now, unsigned int qpair) { struct udp_flow *uflow = udp_at_sidx(ref.flowside); @@ -953,7 +958,7 @@ void udp_sock_handler(const struct ctx *c, union epoll_ref ref, if (events & EPOLLERR) { if (udp_sock_errs(c, ref.fd, ref.flowside, - PIF_NONE, 0, now) < 0) { + PIF_NONE, 0, now, qpair) < 0) { flow_err_ratelimit( uflow, now, "Unrecoverable error on flow socket"); @@ -982,7 +987,7 @@ void udp_sock_handler(const struct ctx *c, union epoll_ref ref, } else if (topif == PIF_TAP) { if (c->mode == MODE_VU) { udp_vu_sock_to_tap(c, s, UDP_MAX_FRAMES, - tosidx); + tosidx, qpair); } else { udp_buf_sock_to_tap(c, s, n, tosidx); } @@ -1003,6 +1008,7 @@ fail: /** * udp_tap_handler() - Handle packets from tap * @c: Execution context + * @qpair: Queue pair to process * @pif: pif on which the packet is arriving * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address @@ -1016,7 +1022,7 @@ fail: * * #syscalls sendmmsg */ -int udp_tap_handler(const struct ctx *c, uint8_t pif, +int udp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif, sa_family_t af, const void *saddr, const void *daddr, uint8_t ttl, const struct pool *p, int idx, const struct timespec *now) @@ -1049,7 +1055,7 @@ int udp_tap_handler(const struct ctx *c, uint8_t pif, src = ntohs(uh->source); dst = ntohs(uh->dest); - tosidx = udp_flow_from_tap(c, pif, af, saddr, daddr, src, dst, now); + tosidx = udp_flow_from_tap(c, qpair, pif, af, saddr, daddr, src, dst, now); if (!(uflow = udp_at_sidx(tosidx))) { char sstr[INET6_ADDRSTRLEN], dstr[INET6_ADDRSTRLEN]; diff --git a/udp.h b/udp.h index b50283eab198..abd651f2fab6 100644 --- a/udp.h +++ b/udp.h @@ -12,10 +12,11 @@ #include "fwd.h" void udp_listen_sock_handler(const struct ctx *c, union epoll_ref ref, - uint32_t events, const struct timespec *now); -void udp_sock_handler(const struct ctx *c, union epoll_ref ref, - uint32_t events, const struct timespec *now); -int udp_tap_handler(const struct ctx *c, uint8_t pif, + uint32_t events, const struct timespec *now, + unsigned int qpair); +void udp_sock_handler(const struct ctx *c, union epoll_ref ref, uint32_t events, + const struct timespec *now, unsigned int qpair); +int udp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif, sa_family_t af, const void *saddr, const void *daddr, uint8_t ttl, const struct pool *p, int idx, const struct timespec *now); diff --git a/udp_flow.c b/udp_flow.c index f59649f6f2fc..ff6abcb8eba0 100644 --- a/udp_flow.c +++ b/udp_flow.c @@ -207,6 +207,7 @@ cancel: /** * udp_flow_from_sock() - Find or create UDP flow for incoming datagram * @c: Execution context + * @qpair: Queue pair for the flow * @pif: Interface the datagram is arriving from * @dst: Our (local) address to which the datagram is arriving * @port: Our (local) port number to which the datagram is arriving @@ -219,9 +220,9 @@ cancel: * 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. */ -flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, - const union inany_addr *dst, in_port_t port, - const union sockaddr_inany *s_in, +flow_sidx_t udp_flow_from_sock(const struct ctx *c, unsigned int qpair, + uint8_t pif, const union inany_addr *dst, + in_port_t port, const union sockaddr_inany *s_in, int rule_hint, const struct timespec *now) { const struct flowside *ini; @@ -235,6 +236,7 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, return flow_sidx_opposite(sidx); } + (void)qpair; if (!(flow = flow_alloc())) { char sastr[SOCKADDR_STRLEN]; @@ -264,6 +266,7 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, /** * udp_flow_from_tap() - Find or create UDP flow for tap packets * @c: Execution context + * @qpair: Queue pair for the flow * @pif: pif on which the packet is arriving * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address on guest side @@ -274,7 +277,7 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, * 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. */ -flow_sidx_t udp_flow_from_tap(const struct ctx *c, +flow_sidx_t udp_flow_from_tap(const struct ctx *c, unsigned int qpair, uint8_t pif, sa_family_t af, const void *saddr, const void *daddr, in_port_t srcport, in_port_t dstport, @@ -294,6 +297,7 @@ flow_sidx_t udp_flow_from_tap(const struct ctx *c, return flow_sidx_opposite(sidx); } + (void)qpair; if (!(flow = flow_alloc())) { char sstr[INET6_ADDRSTRLEN], dstr[INET6_ADDRSTRLEN]; @@ -323,15 +327,16 @@ flow_sidx_t udp_flow_from_tap(const struct ctx *c, * @uflow: Flow to handle * @sidei: Side of the flow to flush * @now: Current timestamp + * @qpair: Queue pair for the flow */ static void udp_flush_flow(const struct ctx *c, const struct udp_flow *uflow, unsigned sidei, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { /* We don't know exactly where the datagrams will come from, but we know * they'll have an interface and oport matching this flow */ udp_sock_fwd(c, uflow->s[sidei], -1, uflow->f.pif[sidei], - uflow->f.side[sidei].oport, now); + uflow->f.side[sidei].oport, now, qpair); } /** @@ -343,14 +348,14 @@ static void udp_flush_flow(const struct ctx *c, * Return: true if the connection is ready to free, false otherwise */ bool udp_flow_defer(const struct ctx *c, struct udp_flow *uflow, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { if (uflow->flush0) { - udp_flush_flow(c, uflow, INISIDE, now); + udp_flush_flow(c, uflow, INISIDE, now, qpair); uflow->flush0 = false; } if (uflow->flush1) { - udp_flush_flow(c, uflow, TGTSIDE, now); + udp_flush_flow(c, uflow, TGTSIDE, now, qpair); uflow->flush1 = false; } return uflow->closed; diff --git a/udp_flow.h b/udp_flow.h index 62cc9b3aae1f..f82a9a7c8923 100644 --- a/udp_flow.h +++ b/udp_flow.h @@ -40,18 +40,19 @@ struct udp_flow { }; struct udp_flow *udp_at_sidx(flow_sidx_t sidx); -flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, - const union inany_addr *dst, in_port_t port, - const union sockaddr_inany *s_in, - int rule_hint, const struct timespec *now); -flow_sidx_t udp_flow_from_tap(const struct ctx *c, +flow_sidx_t udp_flow_from_sock(const struct ctx *c, unsigned int qpair, + uint8_t pif, const union inany_addr *dst, + in_port_t port, + const union sockaddr_inany *s_in, int rule_hint, + const struct timespec *now); +flow_sidx_t udp_flow_from_tap(const struct ctx *c, unsigned int qpair, uint8_t pif, sa_family_t af, const void *saddr, const void *daddr, in_port_t srcport, in_port_t dstport, const struct timespec *now); void udp_flow_close(const struct ctx *c, struct udp_flow *uflow); bool udp_flow_defer(const struct ctx *c, struct udp_flow *uflow, - const struct timespec *now); + const struct timespec *now, unsigned int qpair); bool udp_flow_timer(const struct ctx *c, struct udp_flow *uflow, const struct timespec *now); void udp_flow_activity(struct udp_flow *uflow, unsigned int sidei, diff --git a/udp_internal.h b/udp_internal.h index 361cc7495a01..0cd6da49fc05 100644 --- a/udp_internal.h +++ b/udp_internal.h @@ -34,6 +34,7 @@ size_t udp_update_hdr6(struct ipv6hdr *ip6h, struct udphdr *uh, const struct flowside *toside, size_t dlen, bool no_udp_csum); void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, - uint8_t frompif, in_port_t port, const struct timespec *now); + uint8_t frompif, in_port_t port, const struct timespec *now, + unsigned int qpair); #endif /* UDP_INTERNAL_H */ diff --git a/udp_vu.c b/udp_vu.c index 4003f5b42b56..21fa891b3b51 100644 --- a/udp_vu.c +++ b/udp_vu.c @@ -144,13 +144,15 @@ static void udp_vu_prepare(const struct ctx *c, struct iov_tail *data, * @s: Socket to read data from * @n: Maximum number of datagrams to forward * @tosidx: Flow & side to forward data from @s to + * @qpair: Queue pair to process */ -void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx) +void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx, + unsigned int qpair) { const struct flowside *toside = flowside_at_sidx(tosidx); bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr)); static struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE]; - int toguest = QPAIR_TOGUEST_QUEUE(QPAIR_DEFAULT); + int toguest = QPAIR_TOGUEST_QUEUE(qpair); static struct iovec iov_vu[VIRTQUEUE_MAX_SIZE]; struct vu_dev *vdev = c->vdev; struct vu_virtq *vq = &vdev->vq[toguest]; diff --git a/udp_vu.h b/udp_vu.h index 1e38af35ad4e..40ab28119b10 100644 --- a/udp_vu.h +++ b/udp_vu.h @@ -10,6 +10,7 @@ void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref, const struct timespec *now); -void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx); +void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx, + unsigned int qpair); #endif /* UDP_VU_H */ -- 2.54.0

icmp_sock_handler() uses QPAIR_DEFAULT when calling tap_icmp4_send() and tap_icmp6_send(). Thread a qpair parameter from the caller in passt_worker() so the queue pair is selected at the entry point rather than hard-coded inside the handler. passt_worker() currently passes QPAIR_DEFAULT, preserving existing single-queue behavior. This is a preparatory step for per-queue worker threads in vhost-user mode. No functional change. Signed-off-by: Laurent Vivier --- icmp.c | 20 ++++++++++++-------- icmp.h | 7 ++++--- passt.c | 2 +- tap.c | 4 ++-- 4 files changed, 19 insertions(+), 14 deletions(-) diff --git a/icmp.c b/icmp.c index a1015fe9f5c2..92cb48dfaafe 100644 --- a/icmp.c +++ b/icmp.c @@ -67,9 +67,10 @@ static struct icmp_ping_flow *ping_at_sidx(flow_sidx_t sidx) * @c: Execution context * @ref: epoll reference * @now: Current timestamp + * @qpair: Queue pair to process */ void icmp_sock_handler(const struct ctx *c, union epoll_ref ref, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { struct icmp_ping_flow *pingf = ping_at_sidx(ref.flowside); const struct flowside *ini = &pingf->f.side[INISIDE]; @@ -134,13 +135,13 @@ void icmp_sock_handler(const struct ctx *c, union epoll_ref ref, const struct in_addr *daddr = inany_v4(&ini->eaddr); assert(saddr && daddr); /* Must have IPv4 addresses */ - tap_icmp4_send(c, QPAIR_DEFAULT, *saddr, *daddr, buf, + tap_icmp4_send(c, qpair, *saddr, *daddr, buf, pingf->f.tap_omac, n); } else if (pingf->f.type == FLOW_PING6) { const struct in6_addr *saddr = &ini->oaddr.a6; const struct in6_addr *daddr = &ini->eaddr.a6; - tap_icmp6_send(c, QPAIR_DEFAULT, saddr, daddr, buf, + tap_icmp6_send(c, qpair, saddr, daddr, buf, pingf->f.tap_omac, n); } return; @@ -165,6 +166,7 @@ static void icmp_ping_close(const struct ctx *c, /** * icmp_ping_new() - Prepare a new ping socket for a new id * @c: Execution context + * @qpair: Queue pair of the flow * @af: Address family, AF_INET or AF_INET6 * @id: ICMP id for the new socket * @saddr: Source address @@ -174,8 +176,8 @@ static void icmp_ping_close(const struct ctx *c, * Return: newly opened ping flow, or NULL on failure */ static struct icmp_ping_flow *icmp_ping_new(const struct ctx *c, - sa_family_t af, uint16_t id, - const void *saddr, + unsigned int qpair, sa_family_t af, + uint16_t id, const void *saddr, const void *daddr, const struct timespec *now) { @@ -185,6 +187,7 @@ static struct icmp_ping_flow *icmp_ping_new(const struct ctx *c, struct icmp_ping_flow *pingf; const struct flowside *tgt; + (void)qpair; if (!flow) return NULL; @@ -240,6 +243,7 @@ cancel: /** * icmp_tap_handler() - Handle packets from tap * @c: Execution context + * @qpair: Queue pair to process * @pif: pif on which the packet is arriving * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address @@ -249,8 +253,8 @@ cancel: * * Return: count of consumed packets (always 1, even if malformed) */ -int icmp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, - const void *saddr, const void *daddr, +int icmp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif, + sa_family_t af, const void *saddr, const void *daddr, struct iov_tail *data, const struct timespec *now) { struct iovec iov[MAX_IOV_ICMP]; @@ -307,7 +311,7 @@ int icmp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, if (flow) pingf = &flow->ping; - else if (!(pingf = icmp_ping_new(c, af, id, saddr, daddr, now))) + else if (!(pingf = icmp_ping_new(c, qpair, af, id, saddr, daddr, now))) return 1; tgt = &pingf->f.side[TGTSIDE]; diff --git a/icmp.h b/icmp.h index 8b8fa403cc91..0dd69328591a 100644 --- a/icmp.h +++ b/icmp.h @@ -14,9 +14,10 @@ struct ctx; struct icmp_ping_flow; void icmp_sock_handler(const struct ctx *c, union epoll_ref ref, - const struct timespec *now); -int icmp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, - const void *saddr, const void *daddr, + const struct timespec *now, + unsigned int qpair); +int icmp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif, + sa_family_t af, const void *saddr, const void *daddr, struct iov_tail *data, const struct timespec *now); void icmp_init(void); diff --git a/passt.c b/passt.c index a26653db7763..55708d96fb06 100644 --- a/passt.c +++ b/passt.c @@ -283,7 +283,7 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events) QPAIR_DEFAULT); break; case EPOLL_TYPE_PING: - icmp_sock_handler(c, ref, &now); + icmp_sock_handler(c, ref, &now, QPAIR_DEFAULT); break; case EPOLL_TYPE_VHOST_CMD: vu_control_handler(c->vdev, c->fd_tap, eventmask); diff --git a/tap.c b/tap.c index 84ac111d7a11..8bd1db427721 100644 --- a/tap.c +++ b/tap.c @@ -788,7 +788,7 @@ resume: tap_packet_debug(iph, NULL, NULL, 0, NULL, 1); - icmp_tap_handler(c, PIF_TAP, AF_INET, + icmp_tap_handler(c, qpair, PIF_TAP, AF_INET, &iph->saddr, &iph->daddr, &data, now); continue; @@ -1034,7 +1034,7 @@ resume: tap_packet_debug(NULL, ip6h, NULL, proto, NULL, 1); - icmp_tap_handler(c, PIF_TAP, AF_INET6, + icmp_tap_handler(c, qpair, PIF_TAP, AF_INET6, saddr, daddr, &data, now); continue; } -- 2.54.0

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

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

Thread the queue pair parameter through the UDP socket handler and send path, replacing hardcoded QPAIR_DEFAULT values. This is the UDP counterpart to the equivalent TCP and ARP changes.

Add a qpair parameter to udp_listen_sock_handler(), udp_sock_handler(), udp_sock_fwd(), udp_sock_recverr(), udp_sock_errs(), udp_tap_handler(), udp_send_tap_icmp4(), and udp_send_tap_icmp6().

The queue pair is passed from passt_worker() through udp_listen_sock_handler() and udp_sock_handler(), into udp_sock_fwd() and down into udp_sock_errs()/udp_sock_recverr() for ICMP error generation, as well as udp_vu_sock_to_tap() for vhost-user delivery.

On the tap side, udp_tap_handler() receives the queue pair and passes it to udp_flow_from_tap(), which takes the parameter for completeness but does not yet use it: new flows are not associated with a specific queue pair at creation time.

On the flow handling side, flow_defer_handler() receives the queue pair and passes it to udp_flow_defer() and udp_flush_flow(), so that deferred UDP datagrams are forwarded on the correct queue.

udp_vu_sock_to_tap() now uses the passed qpair to select the RX virtqueue instead of always using QPAIR_DEFAULT.

No functional change.

Signed-off-by: Laurent Vivier

Reviewed-by: David Gibson With a couple of nits about the description of the @qpair parameter in some places.

--- flow.c | 6 ++++-- flow.h | 3 ++- passt.c | 8 +++++--- tap.c | 4 ++-- udp.c | 48 +++++++++++++++++++++++++++--------------------- udp.h | 9 +++++---- udp_flow.c | 23 ++++++++++++++--------- udp_flow.h | 13 +++++++------ udp_internal.h | 3 ++- udp_vu.c | 6 ++++-- udp_vu.h | 3 ++- 11 files changed, 74 insertions(+), 52 deletions(-)

diff --git a/flow.c b/flow.c index 71918b773738..c63491804709 100644 --- a/flow.c +++ b/flow.c @@ -884,8 +884,10 @@ flow_sidx_t flow_lookup_sa(const struct ctx *c, uint8_t proto, uint8_t pif, * flow_defer_handler() - Handler for per-flow deferred and timed tasks * @c: Execution context * @now: Current timestamp + * @qpair: Queue pair to process */ -void flow_defer_handler(const struct ctx *c, const struct timespec *now) +void flow_defer_handler(const struct ctx *c, const struct timespec *now, + unsigned int qpair) { struct flow_free_cluster *free_head = NULL; unsigned *last_next = &flow_first_free; @@ -924,7 +926,7 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now) closed = icmp_ping_timer(c, &flow->ping, now); break; case FLOW_UDP: - closed = udp_flow_defer(c, &flow->udp, now); + closed = udp_flow_defer(c, &flow->udp, now, qpair); if (!closed && timer) closed = udp_flow_timer(c, &flow->udp, now); break; diff --git a/flow.h b/flow.h index db028455a7a4..8d85e27427db 100644 --- a/flow.h +++ b/flow.h @@ -270,7 +270,8 @@ 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); -void flow_defer_handler(const struct ctx *c, const struct timespec *now); +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, int fd); int flow_migrate_source_pre(struct ctx *c, const struct migrate_stage *stage, diff --git a/passt.c b/passt.c index 86ed339ff67f..a26653db7763 100644 --- a/passt.c +++ b/passt.c @@ -116,7 +116,7 @@ static void post_handler(struct ctx *c, const struct timespec *now, if (!c->no_tcp) tcp_defer_handler(c, now, qpair);

- flow_defer_handler(c, now); + flow_defer_handler(c, now, qpair); fwd_scan_ports_timer(c, now);

if (!c->no_ndp) @@ -275,10 +275,12 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events) tcp_timer_handler(c, ref, &now, QPAIR_DEFAULT); break; case EPOLL_TYPE_UDP_LISTEN: - udp_listen_sock_handler(c, ref, eventmask, &now); + udp_listen_sock_handler(c, ref, eventmask, &now, + QPAIR_DEFAULT); break; case EPOLL_TYPE_UDP: - udp_sock_handler(c, ref, eventmask, &now); + udp_sock_handler(c, ref, eventmask, &now, + QPAIR_DEFAULT); break; case EPOLL_TYPE_PING: icmp_sock_handler(c, ref, &now); diff --git a/tap.c b/tap.c index 9ce32d36f1d7..61d18ee161b1 100644 --- a/tap.c +++ b/tap.c @@ -872,7 +872,7 @@ append: if (c->no_udp) continue; for (k = 0; k < p->count; ) - k += udp_tap_handler(c, PIF_TAP, AF_INET, + k += udp_tap_handler(c, qpair, PIF_TAP, AF_INET, &seq->saddr, &seq->daddr, seq->ttl, p, k, now); } @@ -1123,7 +1123,7 @@ append: if (c->no_udp) continue; for (k = 0; k < p->count; ) - k += udp_tap_handler(c, PIF_TAP, AF_INET6, + k += udp_tap_handler(c, qpair, PIF_TAP, AF_INET6, &seq->saddr, &seq->daddr, seq->hop_limit, p, k, now); } diff --git a/udp.c b/udp.c index 589d48a17613..a3321289bb53 100644 --- a/udp.c +++ b/udp.c @@ -403,13 +403,14 @@ static void udp_tap_prepare(const struct mmsghdr *mmh, /** * udp_send_tap_icmp4() - Construct and send ICMPv4 to local peer * @c: Execution context + * @qpair: Queue pair on which to send the ICMPv4 packet * @ee: Extended error descriptor * @toside: Destination side of flow * @saddr: Address of ICMP generating node * @in: First bytes (max 8) of original UDP message body * @dlen: Length of the read part of original UDP message body */ -static void udp_send_tap_icmp4(const struct ctx *c, +static void udp_send_tap_icmp4(const struct ctx *c, unsigned int qpair, const struct sock_extended_err *ee, const struct flowside *toside, struct in_addr saddr, @@ -445,13 +446,14 @@ static void udp_send_tap_icmp4(const struct ctx *c, /* Try to obtain the MAC address of the generating node */ saddr_any = inany_from_v4(saddr); fwd_neigh_mac_get(c, &saddr_any, tap_omac); - tap_icmp4_send(c, QPAIR_DEFAULT, saddr, eaddr, &msg, tap_omac, msglen); + tap_icmp4_send(c, qpair, saddr, eaddr, &msg, tap_omac, msglen); }

/** * udp_send_tap_icmp6() - Construct and send ICMPv6 to local peer * @c: Execution context + * @qpair: Queue pair on which to send the ICMPv6 packet * @ee: Extended error descriptor * @toside: Destination side of flow * @saddr: Address of ICMP generating node @@ -459,7 +461,7 @@ static void udp_send_tap_icmp4(const struct ctx *c, * @dlen: Length of the read part of original UDP message body * @flow: IPv6 flow identifier */ -static void udp_send_tap_icmp6(const struct ctx *c, +static void udp_send_tap_icmp6(const struct ctx *c, unsigned int qpair, const struct sock_extended_err *ee, const struct flowside *toside, const struct in6_addr *saddr, @@ -493,7 +495,7 @@ static void udp_send_tap_icmp6(const struct ctx *c,

/* Try to obtain the MAC address of the generating node */ fwd_neigh_mac_get(c, (union inany_addr *) saddr, tap_omac); - tap_icmp6_send(c, QPAIR_DEFAULT, saddr, eaddr, &msg, tap_omac, msglen); + tap_icmp6_send(c, qpair, saddr, eaddr, &msg, tap_omac, msglen); }

/** @@ -548,7 +550,7 @@ static int udp_pktinfo(struct msghdr *msg, union inany_addr *dst) */ static int udp_sock_recverr(const struct ctx *c, int s, flow_sidx_t sidx, uint8_t pif, in_port_t port, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { char buf[PKTINFO_SPACE + RECVERR_SPACE]; const struct sock_extended_err *ee; @@ -655,12 +657,12 @@ static int udp_sock_recverr(const struct ctx *c, int s, flow_sidx_t sidx, if (hdr->cmsg_level == IPPROTO_IP && (o4 = inany_v4(&otap)) && inany_v4(&toside->eaddr)) { dlen = MIN(dlen, ICMP4_MAX_DLEN); - udp_send_tap_icmp4(c, ee, toside, *o4, data, dlen); + udp_send_tap_icmp4(c, qpair, ee, toside, *o4, data, dlen); return 1; }

if (hdr->cmsg_level == IPPROTO_IPV6 && !inany_v4(&toside->eaddr)) { - udp_send_tap_icmp6(c, ee, toside, &otap.a6, data, dlen, + udp_send_tap_icmp6(c, qpair, ee, toside, &otap.a6, data, dlen, FLOW_IDX(uflow)); return 1; } @@ -691,7 +693,7 @@ fail: */ static int udp_sock_errs(const struct ctx *c, int s, flow_sidx_t sidx, uint8_t pif, in_port_t port, - const struct timespec *now) + const struct timespec *now, unsigned int qpair)

Does this need to be added to the function comments as well?

{ struct udp_flow *uflow = udp_at_sidx(sidx); unsigned n_err = 0; @@ -701,7 +703,7 @@ static int udp_sock_errs(const struct ctx *c, int s, flow_sidx_t sidx, assert(!c->no_udp);

/* Empty the error queue */ - while ((rc = udp_sock_recverr(c, s, sidx, pif, port, now)) > 0) + while ((rc = udp_sock_recverr(c, s, sidx, pif, port, now, qpair)) > 0) n_err += rc;

if (rc < 0) @@ -862,7 +864,8 @@ static void udp_buf_sock_to_tap(const struct ctx *c, int s, int n, * @now: Current timestamp */ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, - uint8_t frompif, in_port_t port, const struct timespec *now) + uint8_t frompif, in_port_t port, const struct timespec *now, + unsigned int qpair) { union sockaddr_inany src; union inany_addr dst; @@ -878,7 +881,7 @@ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, strerror_(-rc)); /* Clear errors & carry on */ if (udp_sock_errs(c, s, FLOW_SIDX_NONE, - frompif, port, now) < 0) { + frompif, port, now, qpair) < 0) { err_ratelimit(now, "UDP: Unrecoverable error on listening socket: (%s port %hu)", pif_name(frompif), port); @@ -887,7 +890,7 @@ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, continue; }

- tosidx = udp_flow_from_sock(c, frompif, &dst, port, &src, + tosidx = udp_flow_from_sock(c, qpair, frompif, &dst, port, &src, rule_hint, now); topif = pif_at_sidx(tosidx);

@@ -895,7 +898,7 @@ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, udp_sock_to_sock(c, s, 1, tosidx); } else if (topif == PIF_TAP) { if (c->mode == MODE_VU) - udp_vu_sock_to_tap(c, s, 1, tosidx); + udp_vu_sock_to_tap(c, s, 1, tosidx, qpair); else udp_buf_sock_to_tap(c, s, 1, tosidx); } else if (flow_sidx_valid(tosidx)) { @@ -926,14 +929,15 @@ void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, * @ref: epoll reference * @events: epoll events bitmap * @now: Current timestamp + * @qpair: Queue pair to process

As with TCP, I'm not sure that "Queue pair to process" really conveys what this means. AIUI, this will be the qpair associated with the epoll set on which we received the event.

*/ void udp_listen_sock_handler(const struct ctx *c, union epoll_ref ref, uint32_t events, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { if (events & (EPOLLERR | EPOLLIN)) { udp_sock_fwd(c, ref.fd, ref.listen.rule, - ref.listen.pif, ref.listen.port, now); + ref.listen.pif, ref.listen.port, now, qpair); } }

@@ -943,9 +947,10 @@ void udp_listen_sock_handler(const struct ctx *c, * @ref: epoll reference * @events: epoll events bitmap * @now: Current timestamp + * @qpair: Queue pair to process */ -void udp_sock_handler(const struct ctx *c, union epoll_ref ref, - uint32_t events, const struct timespec *now) +void udp_sock_handler(const struct ctx *c, union epoll_ref ref, uint32_t events, + const struct timespec *now, unsigned int qpair) { struct udp_flow *uflow = udp_at_sidx(ref.flowside);

@@ -953,7 +958,7 @@ void udp_sock_handler(const struct ctx *c, union epoll_ref ref,

if (events & EPOLLERR) { if (udp_sock_errs(c, ref.fd, ref.flowside, - PIF_NONE, 0, now) < 0) { + PIF_NONE, 0, now, qpair) < 0) { flow_err_ratelimit( uflow, now, "Unrecoverable error on flow socket"); @@ -982,7 +987,7 @@ void udp_sock_handler(const struct ctx *c, union epoll_ref ref, } else if (topif == PIF_TAP) { if (c->mode == MODE_VU) { udp_vu_sock_to_tap(c, s, UDP_MAX_FRAMES, - tosidx); + tosidx, qpair); } else { udp_buf_sock_to_tap(c, s, n, tosidx); } @@ -1003,6 +1008,7 @@ fail: /** * udp_tap_handler() - Handle packets from tap * @c: Execution context + * @qpair: Queue pair to process * @pif: pif on which the packet is arriving * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address @@ -1016,7 +1022,7 @@ fail: * * #syscalls sendmmsg */ -int udp_tap_handler(const struct ctx *c, uint8_t pif, +int udp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif, sa_family_t af, const void *saddr, const void *daddr, uint8_t ttl, const struct pool *p, int idx, const struct timespec *now) @@ -1049,7 +1055,7 @@ int udp_tap_handler(const struct ctx *c, uint8_t pif, src = ntohs(uh->source); dst = ntohs(uh->dest);

- tosidx = udp_flow_from_tap(c, pif, af, saddr, daddr, src, dst, now); + tosidx = udp_flow_from_tap(c, qpair, pif, af, saddr, daddr, src, dst, now); if (!(uflow = udp_at_sidx(tosidx))) { char sstr[INET6_ADDRSTRLEN], dstr[INET6_ADDRSTRLEN];

diff --git a/udp.h b/udp.h index b50283eab198..abd651f2fab6 100644 --- a/udp.h +++ b/udp.h @@ -12,10 +12,11 @@ #include "fwd.h"

void udp_listen_sock_handler(const struct ctx *c, union epoll_ref ref, - uint32_t events, const struct timespec *now); -void udp_sock_handler(const struct ctx *c, union epoll_ref ref, - uint32_t events, const struct timespec *now); -int udp_tap_handler(const struct ctx *c, uint8_t pif, + uint32_t events, const struct timespec *now, + unsigned int qpair); +void udp_sock_handler(const struct ctx *c, union epoll_ref ref, uint32_t events, + const struct timespec *now, unsigned int qpair); +int udp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif, sa_family_t af, const void *saddr, const void *daddr, uint8_t ttl, const struct pool *p, int idx, const struct timespec *now); diff --git a/udp_flow.c b/udp_flow.c index f59649f6f2fc..ff6abcb8eba0 100644 --- a/udp_flow.c +++ b/udp_flow.c @@ -207,6 +207,7 @@ cancel: /** * udp_flow_from_sock() - Find or create UDP flow for incoming datagram * @c: Execution context + * @qpair: Queue pair for the flow * @pif: Interface the datagram is arriving from * @dst: Our (local) address to which the datagram is arriving * @port: Our (local) port number to which the datagram is arriving @@ -219,9 +220,9 @@ cancel: * 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. */ -flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, - const union inany_addr *dst, in_port_t port, - const union sockaddr_inany *s_in, +flow_sidx_t udp_flow_from_sock(const struct ctx *c, unsigned int qpair, + uint8_t pif, const union inany_addr *dst, + in_port_t port, const union sockaddr_inany *s_in, int rule_hint, const struct timespec *now) { const struct flowside *ini; @@ -235,6 +236,7 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, return flow_sidx_opposite(sidx); }

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

@@ -264,6 +266,7 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, /** * udp_flow_from_tap() - Find or create UDP flow for tap packets * @c: Execution context + * @qpair: Queue pair for the flow * @pif: pif on which the packet is arriving * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address on guest side @@ -274,7 +277,7 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, * 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. */ -flow_sidx_t udp_flow_from_tap(const struct ctx *c, +flow_sidx_t udp_flow_from_tap(const struct ctx *c, unsigned int qpair, uint8_t pif, sa_family_t af, const void *saddr, const void *daddr, in_port_t srcport, in_port_t dstport, @@ -294,6 +297,7 @@ flow_sidx_t udp_flow_from_tap(const struct ctx *c, return flow_sidx_opposite(sidx); }

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

@@ -323,15 +327,16 @@ flow_sidx_t udp_flow_from_tap(const struct ctx *c, * @uflow: Flow to handle * @sidei: Side of the flow to flush * @now: Current timestamp + * @qpair: Queue pair for the flow */ static void udp_flush_flow(const struct ctx *c, const struct udp_flow *uflow, unsigned sidei, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { /* We don't know exactly where the datagrams will come from, but we know * they'll have an interface and oport matching this flow */ udp_sock_fwd(c, uflow->s[sidei], -1, uflow->f.pif[sidei], - uflow->f.side[sidei].oport, now); + uflow->f.side[sidei].oport, now, qpair); }

/** @@ -343,14 +348,14 @@ static void udp_flush_flow(const struct ctx *c, * Return: true if the connection is ready to free, false otherwise */ bool udp_flow_defer(const struct ctx *c, struct udp_flow *uflow, - const struct timespec *now) + const struct timespec *now, unsigned int qpair) { if (uflow->flush0) { - udp_flush_flow(c, uflow, INISIDE, now); + udp_flush_flow(c, uflow, INISIDE, now, qpair); uflow->flush0 = false; } if (uflow->flush1) { - udp_flush_flow(c, uflow, TGTSIDE, now); + udp_flush_flow(c, uflow, TGTSIDE, now, qpair); uflow->flush1 = false; } return uflow->closed; diff --git a/udp_flow.h b/udp_flow.h index 62cc9b3aae1f..f82a9a7c8923 100644 --- a/udp_flow.h +++ b/udp_flow.h @@ -40,18 +40,19 @@ struct udp_flow { };

struct udp_flow *udp_at_sidx(flow_sidx_t sidx); -flow_sidx_t udp_flow_from_sock(const struct ctx *c, uint8_t pif, - const union inany_addr *dst, in_port_t port, - const union sockaddr_inany *s_in, - int rule_hint, const struct timespec *now); -flow_sidx_t udp_flow_from_tap(const struct ctx *c, +flow_sidx_t udp_flow_from_sock(const struct ctx *c, unsigned int qpair, + uint8_t pif, const union inany_addr *dst, + in_port_t port, + const union sockaddr_inany *s_in, int rule_hint, + const struct timespec *now); +flow_sidx_t udp_flow_from_tap(const struct ctx *c, unsigned int qpair, uint8_t pif, sa_family_t af, const void *saddr, const void *daddr, in_port_t srcport, in_port_t dstport, const struct timespec *now); void udp_flow_close(const struct ctx *c, struct udp_flow *uflow); bool udp_flow_defer(const struct ctx *c, struct udp_flow *uflow, - const struct timespec *now); + const struct timespec *now, unsigned int qpair); bool udp_flow_timer(const struct ctx *c, struct udp_flow *uflow, const struct timespec *now); void udp_flow_activity(struct udp_flow *uflow, unsigned int sidei, diff --git a/udp_internal.h b/udp_internal.h index 361cc7495a01..0cd6da49fc05 100644 --- a/udp_internal.h +++ b/udp_internal.h @@ -34,6 +34,7 @@ size_t udp_update_hdr6(struct ipv6hdr *ip6h, struct udphdr *uh, const struct flowside *toside, size_t dlen, <> bool no_udp_csum); void udp_sock_fwd(const struct ctx *c, int s, int rule_hint, - uint8_t frompif, in_port_t port, const struct timespec *now); + uint8_t frompif, in_port_t port, const struct timespec *now, + unsigned int qpair);

#endif /* UDP_INTERNAL_H */ diff --git a/udp_vu.c b/udp_vu.c index 4003f5b42b56..21fa891b3b51 100644 --- a/udp_vu.c +++ b/udp_vu.c @@ -144,13 +144,15 @@ static void udp_vu_prepare(const struct ctx *c, struct iov_tail *data, * @s: Socket to read data from * @n: Maximum number of datagrams to forward * @tosidx: Flow & side to forward data from @s to + * @qpair: Queue pair to process */ -void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx) +void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx, + unsigned int qpair) { const struct flowside *toside = flowside_at_sidx(tosidx); bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr)); static struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE]; - int toguest = QPAIR_TOGUEST_QUEUE(QPAIR_DEFAULT); + int toguest = QPAIR_TOGUEST_QUEUE(qpair); static struct iovec iov_vu[VIRTQUEUE_MAX_SIZE]; struct vu_dev *vdev = c->vdev; struct vu_virtq *vq = &vdev->vq[toguest]; diff --git a/udp_vu.h b/udp_vu.h index 1e38af35ad4e..40ab28119b10 100644 --- a/udp_vu.h +++ b/udp_vu.h @@ -10,6 +10,7 @@

void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref, const struct timespec *now); -void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx); +void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx, + unsigned int qpair);

#endif /* UDP_VU_H */ -- 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

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

Allow the guest to negotiate multiple virtqueue pairs by advertising VIRTIO_NET_F_MQ and VHOST_USER_PROTOCOL_F_MQ feature flags, and increase VHOST_USER_MAX_VQS from 2 to 32, supporting up to 16 queue pairs.

Replace the VHOST_USER_RX_QUEUE, VHOST_USER_TX_QUEUE, VHOST_USER_IS_QUEUE_TX(), and VHOST_USER_IS_QUEUE_RX() macros with a general set of QPAIR_* macros in passt.h that translate between queue pair numbers and virtqueue indices. These are needed now that queue indices are no longer limited to 0 and 1.

Add a queue pair parameter to vu_send_single(), propagating it to the virtqueue selection. All callers currently pass QPAIR_DEFAULT (0): only the first RX queue is used for receiving. The guest kernel selects which TX queue to use for transmission. Full multi-RX-queue load balancing will be implemented separately.

Signed-off-by: Laurent Vivier

Reviewed-by: David Gibson [snip]

/** - * vu_send_single() - Send a buffer to the front-end using the RX virtqueue - * @c: execution context + * vu_send_single() - Send a buffer to the front-end using a specified virtqueue + * @c: Execution context + * @qpair: Queue pair on which to send the buffer * @buf: address of the buffer * @size: size of the buffer * * Return: number of bytes sent, -1 if there is an error */ -int vu_send_single(const struct ctx *c, const void *buf, size_t size) +int vu_send_single(const struct ctx *c, unsigned int qpair, const void *buf, size_t size) { struct vu_dev *vdev = c->vdev; - struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE]; struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE]; struct iovec in_sg[VIRTQUEUE_MAX_SIZE]; + struct vu_virtq *vq;

Nit: belongs one line lower down...

size_t total, in_total; int elem_cnt; int i;

+ vq = &vdev->vq[QPAIR_TOGUEST_QUEUE(qpair)]; +

... or you could put this expression in the initialiser and it wouldn't move at all.

trace("vu_send_single size %zu", size);

if (!vu_queue_enabled(vq) || !vu_queue_started(vq)) { diff --git a/vu_common.h b/vu_common.h index 817384175a1d..f5603d9ddeb6 100644 --- a/vu_common.h +++ b/vu_common.h @@ -23,7 +23,8 @@ void vu_flush(const struct vu_dev *vdev, struct vu_virtq *vq, struct vu_virtq_element *elem, int elem_cnt, size_t frame_len); void vu_kick_cb(struct vu_dev *vdev, union epoll_ref ref, const struct timespec *now); -int vu_send_single(const struct ctx *c, const void *buf, size_t size); +int vu_send_single(const struct ctx *c, unsigned int qpair, const void *buf, + size_t size); void vu_pad(const struct iovec *iov, size_t cnt, size_t frame_len);

#endif /* VU_COMMON_H */ -- 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

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

Thread a qpair parameter from the entry points (tcp_sock_handler, tcp_timer_handler, tcp_tap_handler, tcp_defer_handler) through every intermediate function down to the vhost-user send functions, so callers explicitly select the target RX virtqueue instead of hardcoding QPAIR_DEFAULT.

Add a qpair parameter to tcp_send_flag(), tcp_data_from_sock(), tcp_rst_do() and its tcp_rst() macro, tcp_rewind_seq(), tcp_data_from_tap(), tcp_conn_from_sock_finish(), tcp_connect_finish(), tcp_tap_window_update(), tcp_conn_from_tap(), tcp_rst_no_conn(), tcp_keepalive(), and tcp_inactivity().

tcp_vu_send_flag() and tcp_vu_data_from_sock() now use the passed qpair to select the RX virtqueue instead of always using QPAIR_DEFAULT.

The buffer-based path (tcp_buf.c) does not thread qpair since it is only used in non-vhost-user mode.

No functional change.

Signed-off-by: Laurent Vivier

Reviewed-by: David Gibson .. with one minor concern about comment clarity: [snip]

@@ -1862,13 +1869,14 @@ static int tcp_sock_consume(const struct tcp_tap_conn *conn, uint32_t ack_seq) * @c: Execution context * @conn: Connection pointer * @now: Current timestamp + * @qpair: Queue pair to process

IIUC, for tcp_data_from_sock() and other socket initiated entry points, the qpair number will eventually be derived from which epoll set the event arrived on, yes? I'm not sure "Queue pair to process" really conveys that. -- 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

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

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

icmp_sock_handler() uses QPAIR_DEFAULT when calling tap_icmp4_send() and tap_icmp6_send(). Thread a qpair parameter from the caller in passt_worker() so the queue pair is selected at the entry point rather than hard-coded inside the handler.

passt_worker() currently passes QPAIR_DEFAULT, preserving existing single-queue behavior. This is a preparatory step for per-queue worker threads in vhost-user mode.

No functional change.

Signed-off-by: Laurent Vivier

Reviewed-by: David Gibson

@@ -67,9 +67,10 @@ static struct icmp_ping_flow *ping_at_sidx(flow_sidx_t sidx) * @c: Execution context * @ref: epoll reference * @now: Current timestamp + * @qpair: Queue pair to process

As with TCP and UDP, some more precision in this description would help. -- 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