Re: [PATCH v5 06/12] tcp: Pass queue pair explicitly through TCP send path

On 6/19/26 08:00, David Gibson wrote:

On Tue, Jun 16, 2026 at 02:51:24PM +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().

For the to-guest functions which take a connection parameter, this seems odd to me. Can't they deduce the right queue from the connection?

The connection's qpair (conn->f.qpair) can change at any time when another thread processes a tap packet for the same flow and calls FLOW_MIGRATE(). If a socket event fires on qpair 0's epoll instance and we read conn->f.qpair during processing, another thread might change it concurrently via FLOW_MIGRATE(). The qpair from the epoll event is the stable, race-free reference for "which queue am I operating on." So the explicit parameter is intentional: it carries the qpair from the epoll event through the call chain, independent of the mutable flow state.

It makes sense, obviously, for anything without a connection parameter. And it makes sense to me for from-guest functions, since the guest might have migrated the connection to a different queue since last we heard about it.

...

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 --- passt.c | 12 +-- tap.c | 8 +- tcp.c | 199 ++++++++++++++++++++++++++++--------------------- tcp.h | 15 ++-- tcp_buf.c | 10 +-- tcp_internal.h | 7 +- tcp_vu.c | 19 +++-- tcp_vu.h | 6 +- 8 files changed, 157 insertions(+), 119 deletions(-)

diff --git a/passt.c b/passt.c index 71eb4f0192e2..9569f920ee28 100644 --- a/passt.c +++ b/passt.c @@ -98,11 +98,13 @@ struct passt_stats { * post_handler() - Run periodic and deferred tasks for L4 protocol handlers * @c: Execution context * @now: Current timestamp + * @qpair: Queue pair to process */ -static void post_handler(struct ctx *c, const struct timespec *now) +static void post_handler(struct ctx *c, const struct timespec *now, + unsigned int qpair) { if (!c->no_tcp) - tcp_defer_handler(c, now); + tcp_defer_handler(c, now, qpair);

flow_defer_handler(c, now); fwd_scan_ports_timer(c, now); @@ -251,7 +253,7 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events) pasta_netns_quit_timer_handler(c, ref); break; case EPOLL_TYPE_TCP: - tcp_sock_handler(c, ref, eventmask); + tcp_sock_handler(c, ref, eventmask, QPAIR_DEFAULT); break; case EPOLL_TYPE_TCP_SPLICE: tcp_splice_sock_handler(c, ref, eventmask, &now); @@ -260,7 +262,7 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events) tcp_listen_handler(c, ref, &now); break; case EPOLL_TYPE_TCP_TIMER: - tcp_timer_handler(c, ref); + tcp_timer_handler(c, ref, QPAIR_DEFAULT); break; case EPOLL_TYPE_UDP_LISTEN: udp_listen_sock_handler(c, ref, eventmask, &now); @@ -300,7 +302,7 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events) print_stats(c, &stats, &now); }

- post_handler(c, &now); + post_handler(c, &now, QPAIR_DEFAULT);

migrate_handler(c); } diff --git a/tap.c b/tap.c index 66dcb83665a7..ba2a573fa630 100644 --- a/tap.c +++ b/tap.c @@ -717,8 +717,6 @@ static int tap4_handler(struct ctx *c, unsigned int qpair, unsigned int i, j, seq_count; struct tap4_l4_t *seq;

- (void)qpair; - if (!c->ifi4 || !pool_tap4->count) return pool_tap4->count;

@@ -870,7 +868,7 @@ append: if (c->no_tcp) continue; for (k = 0; k < p->count; ) - k += tcp_tap_handler(c, PIF_TAP, AF_INET, + k += tcp_tap_handler(c, qpair, PIF_TAP, AF_INET, &seq->saddr, &seq->daddr, 0, p, k, now); } else if (seq->protocol == IPPROTO_UDP) { @@ -955,8 +953,6 @@ static int tap6_handler(struct ctx *c, unsigned int qpair, unsigned int i, j, seq_count = 0; struct tap6_l4_t *seq;

- (void)qpair; - if (!c->ifi6 || !pool_tap6->count) return pool_tap6->count;

@@ -1121,7 +1117,7 @@ append: if (c->no_tcp) continue; for (k = 0; k < p->count; ) - k += tcp_tap_handler(c, PIF_TAP, AF_INET6, + k += tcp_tap_handler(c, qpair, PIF_TAP, AF_INET6, &seq->saddr, &seq->daddr, seq->flow_lbl, p, k, now); } else if (seq->protocol == IPPROTO_UDP) { diff --git a/tcp.c b/tcp.c index c127b3132e5a..7f8e68a31994 100644 --- a/tcp.c +++ b/tcp.c @@ -1258,16 +1258,18 @@ static void tcp_update_seqack_from_tap(const struct ctx *c, * tcp_rewind_seq() - Rewind sequence to tap and socket offset to current ACK * @c: Execution context * @conn: Connection pointer + * @qpair: Queue pair to process * * Return: 0 on success, -1 on failure, with connection reset */ -static int tcp_rewind_seq(const struct ctx *c, struct tcp_tap_conn *conn) +static int tcp_rewind_seq(const struct ctx *c, struct tcp_tap_conn *conn, + unsigned int qpair) { conn->seq_to_tap = conn->seq_ack_from_tap; conn->events &= ~TAP_FIN_SENT;

if (tcp_set_peek_offset(conn, 0)) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return -1; }

@@ -1371,16 +1373,17 @@ int tcp_prepare_flags(const struct ctx *c, struct tcp_tap_conn *conn, * @c: Execution context * @conn: Connection pointer * @flags: TCP flags: if not set, send segment only if ACK is due + * @qpair: Queue pair to process * * Return: negative error code on fatal connection failure, 0 otherwise */ static int tcp_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, - int flags) + int flags, unsigned int qpair) { int ret;

if (c->mode == MODE_VU) - ret = tcp_vu_send_flag(c, conn, flags); + ret = tcp_vu_send_flag(c, conn, flags, qpair); else ret = tcp_buf_send_flag(c, conn, flags);

@@ -1425,14 +1428,16 @@ static void tcp_sock_rst(const struct ctx *c, struct tcp_tap_conn *conn) * tcp_rst_do() - Reset a tap connection: send RST segment on both sides, close * @c: Execution context * @conn: Connection pointer + * @qpair: Queue pair to process */ -void tcp_rst_do(const struct ctx *c, struct tcp_tap_conn *conn) +void tcp_rst_do(const struct ctx *c, struct tcp_tap_conn *conn, + unsigned int qpair) { if (conn->events == CLOSED) return;

/* Send RST on tap */ - tcp_send_flag(c, conn, RST); + tcp_send_flag(c, conn, RST, qpair);

tcp_sock_rst(c, conn); } @@ -1459,11 +1464,13 @@ static void tcp_get_tap_ws(struct tcp_tap_conn *conn, * @c: Execution context * @conn: Connection pointer * @wnd: Window value, host order, unscaled + * @qpair: Queue pair to process * * Return: false on zero window (not stored to wnd_from_tap), true otherwise */ static bool tcp_tap_window_update(const struct ctx *c, - struct tcp_tap_conn *conn, unsigned wnd) + struct tcp_tap_conn *conn, unsigned wnd, + unsigned int qpair)

In particular threading the qpair through this function, that's chiefly not about sending _or_ receiving seems odd..

...

{ wnd = MIN(MAX_WINDOW, wnd << conn->ws_from_tap);

@@ -1474,7 +1481,7 @@ static bool tcp_tap_window_update(const struct ctx *c, * that no data beyond the updated window will be acknowledged. */ if (!wnd && SEQ_LT(conn->seq_ack_from_tap, conn->seq_to_tap)) { - tcp_rewind_seq(c, conn); + tcp_rewind_seq(c, conn, qpair);

.. just to pass here, which only cares about it for an error case.

For tcp_tap_window_update(), the qpair is just threaded through to the error path (tcp_rewind_seq -> tcp_rst), which needs it for the same reason as above.

...

return false; }

@@ -1646,6 +1653,7 @@ static void tcp_bind_outbound(const struct ctx *c, /** * tcp_conn_from_tap() - Handle connection request (SYN segment) from tap * @c: Execution context + * @qpair: Queue pair for the flow * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address, pointer to in_addr or in6_addr * @daddr: Destination address, pointer to in_addr or in6_addr @@ -1656,10 +1664,11 @@ static void tcp_bind_outbound(const struct ctx *c, * * #syscalls:vu getsockname */ -static void tcp_conn_from_tap(const struct ctx *c, sa_family_t af, - const void *saddr, const void *daddr, - const struct tcphdr *th, const char *opts, - size_t optlen, const struct timespec *now) +static void tcp_conn_from_tap(const struct ctx *c, unsigned int qpair, + sa_family_t af, const void *saddr, + const void *daddr, const struct tcphdr *th, + const char *opts, size_t optlen, + const struct timespec *now) { in_port_t srcport = ntohs(th->source); in_port_t dstport = ntohs(th->dest); @@ -1760,7 +1769,7 @@ static void tcp_conn_from_tap(const struct ctx *c, sa_family_t af,

if (connect(s, &sa.sa, socklen_inany(&sa))) { if (errno != EINPROGRESS) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); goto cancel; }

@@ -1768,7 +1777,7 @@ static void tcp_conn_from_tap(const struct ctx *c, sa_family_t af, } else { tcp_get_sndbuf(conn);

- if (tcp_send_flag(c, conn, SYN | ACK)) + if (tcp_send_flag(c, conn, SYN | ACK, qpair)) goto cancel;

conn_event(c, conn, TAP_SYN_ACK_SENT); @@ -1830,15 +1839,17 @@ static int tcp_sock_consume(const struct tcp_tap_conn *conn, uint32_t ack_seq) * tcp_data_from_sock() - Handle new data from socket, queue to tap, in window * @c: Execution context * @conn: Connection pointer + * @qpair: Queue pair to process * * Return: negative on connection reset, 0 otherwise * * #syscalls recvmsg */ -static int tcp_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn) +static int tcp_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn, + unsigned int qpair) { if (c->mode == MODE_VU) - return tcp_vu_data_from_sock(c, conn); + return tcp_vu_data_from_sock(c, conn, qpair);

return tcp_buf_data_from_sock(c, conn); } @@ -1866,13 +1877,15 @@ static ssize_t tcp_packet_data_len(const struct tcphdr *th, size_t l4len) * @conn: Connection pointer * @p: Pool of TCP packets, with TCP headers * @idx: Index of first data packet in pool + * @qpair: Queue pair to process * * #syscalls sendmsg * * Return: count of consumed packets */ static int tcp_data_from_tap(const struct ctx *c, struct tcp_tap_conn *conn, - const struct pool *p, int idx) + const struct pool *p, int idx, + unsigned int qpair) { int i, iov_i, ack = 0, fin = 0, retr = 0, keep = -1, partial_send = 0; uint16_t max_ack_seq_wnd = conn->wnd_from_tap; @@ -1922,7 +1935,7 @@ static int tcp_data_from_tap(const struct ctx *c, struct tcp_tap_conn *conn, "keep-alive sequence: %u, previous: %u", seq, conn->seq_from_tap);

- if (tcp_send_flag(c, conn, ACK)) + if (tcp_send_flag(c, conn, ACK, qpair)) return -1;

tcp_timer_ctl(c, conn); @@ -1933,7 +1946,8 @@ static int tcp_data_from_tap(const struct ctx *c, struct tcp_tap_conn *conn,

if (p->count == 1) { tcp_tap_window_update(c, conn, - ntohs(th->window)); + ntohs(th->window), + qpair); return 1; }

@@ -1959,7 +1973,7 @@ static int tcp_data_from_tap(const struct ctx *c, struct tcp_tap_conn *conn, * well. */ if (!ntohs(th->window)) - tcp_rewind_seq(c, conn); + tcp_rewind_seq(c, conn, qpair);

max_ack_seq_wnd = ntohs(th->window); max_ack_seq = ack_seq; @@ -2024,17 +2038,17 @@ static int tcp_data_from_tap(const struct ctx *c, struct tcp_tap_conn *conn, if (ack && !tcp_sock_consume(conn, max_ack_seq)) tcp_update_seqack_from_tap(c, conn, max_ack_seq);

- tcp_tap_window_update(c, conn, max_ack_seq_wnd); + tcp_tap_window_update(c, conn, max_ack_seq_wnd, qpair);

if (retr) { flow_trace(conn, "fast re-transmit, ACK: %u, previous sequence: %u", conn->seq_ack_from_tap, conn->seq_to_tap);

- if (tcp_rewind_seq(c, conn)) + if (tcp_rewind_seq(c, conn, qpair)) return -1;

- tcp_data_from_sock(c, conn); + tcp_data_from_sock(c, conn, qpair); }

if (!iov_i) @@ -2050,7 +2064,7 @@ eintr: * Then swiftly looked away and left. */ conn->seq_from_tap = seq_from_tap; - if (tcp_send_flag(c, conn, ACK)) + if (tcp_send_flag(c, conn, ACK, qpair)) return -1; }

@@ -2058,7 +2072,7 @@ eintr: goto eintr;

if (errno == EAGAIN || errno == EWOULDBLOCK) { - if (tcp_send_flag(c, conn, ACK | DUP_ACK)) + if (tcp_send_flag(c, conn, ACK | DUP_ACK, qpair)) return -1;

uint32_t events = tcp_conn_epoll_events(conn->events, @@ -2094,7 +2108,7 @@ out: */ if (conn->seq_dup_ack_approx != (conn->seq_from_tap & 0xff)) { conn->seq_dup_ack_approx = conn->seq_from_tap & 0xff; - if (tcp_send_flag(c, conn, ACK | DUP_ACK)) + if (tcp_send_flag(c, conn, ACK | DUP_ACK, qpair)) return -1; } return p->count - idx; @@ -2109,7 +2123,7 @@ out:

conn_event(c, conn, TAP_FIN_RCVD); } else { - if (tcp_send_flag(c, conn, ACK_IF_NEEDED)) + if (tcp_send_flag(c, conn, ACK_IF_NEEDED, qpair)) return -1; }

@@ -2123,13 +2137,15 @@ out: * @th: TCP header of SYN, ACK segment: caller MUST ensure it's there * @opts: Pointer to start of options * @optlen: Bytes in options: caller MUST ensure available length + * @qpair: Queue pair to process */ static void tcp_conn_from_sock_finish(const struct ctx *c, struct tcp_tap_conn *conn, const struct tcphdr *th, - const char *opts, size_t optlen) + const char *opts, size_t optlen, + unsigned int qpair) { - tcp_tap_window_update(c, conn, ntohs(th->window)); + tcp_tap_window_update(c, conn, ntohs(th->window), qpair); tcp_get_tap_ws(conn, opts, optlen);

/* First value is not scaled */ @@ -2144,24 +2160,25 @@ static void tcp_conn_from_sock_finish(const struct ctx *c,

conn_event(c, conn, ESTABLISHED); if (tcp_set_peek_offset(conn, 0)) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return; }

- if (tcp_send_flag(c, conn, ACK)) { - tcp_rst(c, conn); + if (tcp_send_flag(c, conn, ACK, qpair)) { + tcp_rst(c, conn, qpair); return; }

/* The client might have sent data already, which we didn't * dequeue waiting for SYN,ACK from tap -- check now. */ - tcp_data_from_sock(c, conn); + tcp_data_from_sock(c, conn, qpair); }

/** * tcp_rst_no_conn() - Send RST in response to a packet with no connection * @c: Execution context + * @qpair: Queue pair on which to send the reply * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address of the packet we're responding to * @daddr: Destination address of the packet we're responding to @@ -2169,7 +2186,7 @@ static void tcp_conn_from_sock_finish(const struct ctx *c, * @th: TCP header of the packet we're responding to * @l4len: Packet length, including TCP header */ -static void tcp_rst_no_conn(const struct ctx *c, int af, +static void tcp_rst_no_conn(const struct ctx *c, unsigned int qpair, int af, const void *saddr, const void *daddr, uint32_t flow_lbl, const struct tcphdr *th, size_t l4len) @@ -2227,12 +2244,13 @@ 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, QPAIR_DEFAULT, buf, rst_l2len); + tap_send_single(c, qpair, buf, rst_l2len); }

/** * tcp_tap_handler() - Handle packets from tap and state transitions * @c: Execution context + * @qpair: Queue pair on which to send packets * @pif: pif on which the packet is arriving * @af: Address family, AF_INET or AF_INET6 * @saddr: Source address @@ -2244,9 +2262,10 @@ static void tcp_rst_no_conn(const struct ctx *c, int af, * * Return: count of consumed packets */ -int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, - const void *saddr, const void *daddr, uint32_t flow_lbl, - const struct pool *p, int idx, const struct timespec *now) +int tcp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif, + sa_family_t af, const void *saddr, const void *daddr, + uint32_t flow_lbl, const struct pool *p, int idx, + const struct timespec *now) { struct tcp_tap_conn *conn; struct tcphdr th_storage; @@ -2283,10 +2302,11 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, /* New connection from tap */ if (!flow) { if (opts && th->syn && !th->ack) - tcp_conn_from_tap(c, af, saddr, daddr, th, + tcp_conn_from_tap(c, qpair, af, saddr, daddr, th, opts, optlen, now); else - tcp_rst_no_conn(c, af, saddr, daddr, flow_lbl, th, l4len); + tcp_rst_no_conn(c, qpair, af, saddr, daddr, flow_lbl, th, + l4len); return 1; }

@@ -2310,7 +2330,7 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, /* Establishing connection from socket */ if (conn->events & SOCK_ACCEPTED) { if (th->syn && th->ack && !th->fin) { - tcp_conn_from_sock_finish(c, conn, th, opts, optlen); + tcp_conn_from_sock_finish(c, conn, th, opts, optlen, qpair); return 1; }

@@ -2337,7 +2357,7 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, goto reset; }

- if (tcp_send_flag(c, conn, ACK)) + if (tcp_send_flag(c, conn, ACK, qpair)) goto reset;

conn_event(c, conn, SOCK_FIN_SENT); @@ -2348,8 +2368,8 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, if (!th->ack) goto reset;

- if (tcp_tap_window_update(c, conn, ntohs(th->window))) - tcp_data_from_sock(c, conn); + if (tcp_tap_window_update(c, conn, ntohs(th->window), qpair)) + tcp_data_from_sock(c, conn, qpair);

if (p->count - idx == 1) return 1; @@ -2380,12 +2400,12 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, "fast re-transmit, ACK: %u, previous sequence: %u", ntohl(th->ack_seq), conn->seq_to_tap);

- if (tcp_rewind_seq(c, conn)) + if (tcp_rewind_seq(c, conn, qpair)) return -1; }

- if (tcp_tap_window_update(c, conn, ntohs(th->window)) || retr) - tcp_data_from_sock(c, conn); + if (tcp_tap_window_update(c, conn, ntohs(th->window), qpair) || retr) + tcp_data_from_sock(c, conn, qpair);

if (conn->seq_ack_from_tap == conn->seq_to_tap) { if (th->ack && conn->events & TAP_FIN_SENT) @@ -2400,7 +2420,7 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, }

/* Established connections accepting data from tap */ - count = tcp_data_from_tap(c, conn, p, idx); + count = tcp_data_from_tap(c, conn, p, idx, qpair); if (count == -1) goto reset;

@@ -2419,7 +2439,7 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, }

conn_event(c, conn, SOCK_FIN_SENT); - if (tcp_send_flag(c, conn, ACK)) + if (tcp_send_flag(c, conn, ACK, qpair)) goto reset;

ack_due = 0; @@ -2449,7 +2469,7 @@ reset: * remaining packets in the batch, since they'd be invalidated when our * RST is received, even if otherwise good. */ - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return p->count - idx; }

@@ -2457,20 +2477,22 @@ reset: * tcp_connect_finish() - Handle completion of connect() from EPOLLOUT event * @c: Execution context * @conn: Connection pointer + * @qpair: Queue pair to process */ -static void tcp_connect_finish(const struct ctx *c, struct tcp_tap_conn *conn) +static void tcp_connect_finish(const struct ctx *c, struct tcp_tap_conn *conn, + unsigned int qpair) { socklen_t sl; int so;

sl = sizeof(so); if (getsockopt(conn->sock, SOL_SOCKET, SO_ERROR, &so, &sl) || so) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return; }

- if (tcp_send_flag(c, conn, SYN | ACK)) { - tcp_rst(c, conn); + if (tcp_send_flag(c, conn, SYN | ACK, qpair)) { + tcp_rst(c, conn, qpair); return; }

@@ -2513,7 +2535,7 @@ static void tcp_tap_conn_from_sock(const struct ctx *c, union flow *flow,

conn->wnd_from_tap = WINDOW_DEFAULT;

- if (tcp_send_flag(c, conn, SYN)) { + if (tcp_send_flag(c, conn, SYN, QPAIR_DEFAULT)) { conn_flag(c, conn, CLOSING); FLOW_ACTIVATE(conn); return; @@ -2604,12 +2626,14 @@ cancel: * tcp_timer_handler() - timerfd events: close, send ACK, retransmit, or reset * @c: Execution context * @ref: epoll reference of timer (not connection) + * @qpair: Queue pair to process * * #syscalls timerfd_gettime|timerfd_gettime64 * #syscalls arm:timerfd_gettime64 i686:timerfd_gettime64 * #syscalls arm:timerfd_settime64 i686:timerfd_settime64 */ -void tcp_timer_handler(const struct ctx *c, union epoll_ref ref) +void tcp_timer_handler(const struct ctx *c, union epoll_ref ref, + unsigned int qpair)

This one seems even stranger to me. This is triggered off a timer, not a tap side event, how would we even know the relevant qpair except by looking at the connection.

...

{ struct itimerspec check_armed = { { 0 }, { 0 } }; struct tcp_tap_conn *conn = &FLOW(ref.flow)->tcp; @@ -2628,8 +2652,8 @@ void tcp_timer_handler(const struct ctx *c, union epoll_ref ref) return;

if (conn->flags & ACK_TO_TAP_DUE) { - if (tcp_send_flag(c, conn, ACK_IF_NEEDED)) { - tcp_rst(c, conn); + if (tcp_send_flag(c, conn, ACK_IF_NEEDED, qpair)) { + tcp_rst(c, conn, qpair); return; } tcp_timer_ctl(c, conn); @@ -2641,11 +2665,11 @@ void tcp_timer_handler(const struct ctx *c, union epoll_ref ref) max = MIN(TCP_MAX_RETRIES, max); if (conn->retries >= max) { flow_dbg(conn, "handshake timeout"); - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); } else { flow_trace(conn, "SYN timeout, retry"); - if (tcp_send_flag(c, conn, SYN)) { - tcp_rst(c, conn); + if (tcp_send_flag(c, conn, SYN, qpair)) { + tcp_rst(c, conn, qpair); return; } conn->retries++; @@ -2654,7 +2678,7 @@ void tcp_timer_handler(const struct ctx *c, union epoll_ref ref) } } else if (conn->retries == TCP_MAX_RETRIES) { flow_dbg(conn, "retransmissions count exceeded"); - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); } else { flow_dbg(conn, "ACK timeout, retry");

@@ -2662,10 +2686,10 @@ void tcp_timer_handler(const struct ctx *c, union epoll_ref ref) conn->wnd_from_tap = 1; /* Zero-window probe */

conn->retries++; - if (tcp_rewind_seq(c, conn)) + if (tcp_rewind_seq(c, conn, qpair)) return;

- tcp_data_from_sock(c, conn); + tcp_data_from_sock(c, conn, qpair); tcp_timer_ctl(c, conn); } } @@ -2676,9 +2700,10 @@ void tcp_timer_handler(const struct ctx *c, union epoll_ref ref) * @c: Execution context * @ref: epoll reference * @events: epoll events bitmap + * @qpair: Queue pair to process */ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref, - uint32_t events) + uint32_t events, unsigned int qpair)

Likewise here, how could the caller know a reasonable qpair based just on the socket side event?

For tcp_timer_handler() and tcp_sock_handler(), the caller knows the qpair because it is the worker thread's own qpair — each thread has its own epoll instance and passes its qpair to all event handlers. This also matters for timer migration: tcp_timer_handler() needs to compare the event's qpair against conn->f.qpair to detect that the flow has moved and the timerfd needs to be re-registered. If we read qpair from the connection, the comparison would always match.

...

{ struct tcp_tap_conn *conn = conn_at_sidx(ref.flowside);

@@ -2689,7 +2714,7 @@ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref, return;

if (events & EPOLLERR) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return; }

@@ -2708,13 +2733,13 @@ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref, conn_event(c, conn, SOCK_FIN_RCVD);

if (events & EPOLLIN) - tcp_data_from_sock(c, conn); + tcp_data_from_sock(c, conn, qpair);

if (events & EPOLLOUT) { tcp_epoll_ctl(conn); if (tcp_update_seqack_wnd(c, conn, false, NULL) && - tcp_send_flag(c, conn, ACK)) { - tcp_rst(c, conn); + tcp_send_flag(c, conn, ACK, qpair)) { + tcp_rst(c, conn, qpair); return; } } @@ -2724,7 +2749,7 @@ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref,

/* EPOLLHUP during handshake: reset */ if (events & EPOLLHUP) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return; }

@@ -2734,7 +2759,7 @@ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref,

if (conn->events == TAP_SYN_RCVD) { if (events & EPOLLOUT) - tcp_connect_finish(c, conn); + tcp_connect_finish(c, conn, qpair); /* Data? Check later */ } } @@ -2939,9 +2964,11 @@ int tcp_init(struct ctx *c)

/** * tcp_keepalive() - Send keepalives for connections which need it - * @: Execution context + * @c: Execution context + * @qpair: Queue pair to process */ -static void tcp_keepalive(struct ctx *c, const struct timespec *now) +static void tcp_keepalive(struct ctx *c, const struct timespec *now, + unsigned int qpair)

Here we're scanning the flow table, and each connection there could be associated with a different queue, so again, this parameter doesn't make sense to me.

For tcp_keepalive() and tcp_inactivity(), the qpair parameter is used to filter flows: each thread only processes connections belonging to its queue pair (added in "flow: Add queue pair tracking"). Thanks, Laurent

...

{ union flow *flow;

@@ -2956,8 +2983,8 @@ static void tcp_keepalive(struct ctx *c, const struct timespec *now) if (conn->tap_inactive) { flow_dbg(conn, "No tap activity for least %us, send keepalive", KEEPALIVE_INTERVAL); - if (tcp_send_flag(c, conn, KEEPALIVE)) - tcp_rst(c, conn); + if (tcp_send_flag(c, conn, KEEPALIVE, qpair)) + tcp_rst(c, conn, qpair); }

/* Ready to check fot next interval */ @@ -2967,9 +2994,11 @@ static void tcp_keepalive(struct ctx *c, const struct timespec *now)

/** * tcp_inactivity() - Scan for and close long-inactive connections - * @: Execution context + * @c: Execution context + * @qpair: Queue pair to process */ -static void tcp_inactivity(struct ctx *c, const struct timespec *now) +static void tcp_inactivity(struct ctx *c, const struct timespec *now, + unsigned int qpair)

Same here.

...

{ union flow *flow;

@@ -2986,7 +3015,7 @@ static void tcp_inactivity(struct ctx *c, const struct timespec *now) /* No activity in this interval, reset */ flow_dbg(conn, "Inactive for at least %us, resetting", INACTIVITY_INTERVAL); - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); }

/* Ready to check fot next interval */ @@ -2998,9 +3027,11 @@ static void tcp_inactivity(struct ctx *c, const struct timespec *now) * tcp_defer_handler() - Handler for TCP deferred tasks * @c: Execution context * @now: Current timestamp + * @qpair: Queue pair to process */ /* cppcheck-suppress [constParameterPointer, unmatchedSuppression] */ -void tcp_defer_handler(struct ctx *c, const struct timespec *now) +void tcp_defer_handler(struct ctx *c, const struct timespec *now, + unsigned int qpair) { tcp_payload_flush(c);

@@ -3013,8 +3044,8 @@ void tcp_defer_handler(struct ctx *c, const struct timespec *now) if (c->mode == MODE_PASTA) tcp_splice_refill(c);

- tcp_keepalive(c, now); - tcp_inactivity(c, now); + tcp_keepalive(c, now, qpair); + tcp_inactivity(c, now, qpair); }

/** @@ -3988,10 +4019,10 @@ int tcp_flow_migrate_target_ext(struct ctx *c, struct tcp_tap_conn *conn, int fd if (tcp_set_peek_offset(conn, peek_offset)) goto fail;

- if (tcp_send_flag(c, conn, ACK)) + if (tcp_send_flag(c, conn, ACK, QPAIR_DEFAULT)) goto fail;

- tcp_data_from_sock(c, conn); + tcp_data_from_sock(c, conn, QPAIR_DEFAULT);

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

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

return 0; } diff --git a/tcp.h b/tcp.h index 3262a807e5d4..490f1b140e44 100644 --- a/tcp.h +++ b/tcp.h @@ -18,18 +18,21 @@

struct ctx;

-void tcp_timer_handler(const struct ctx *c, union epoll_ref ref); +void tcp_timer_handler(const struct ctx *c, union epoll_ref ref, + unsigned int qpair); void tcp_listen_handler(const struct ctx *c, union epoll_ref ref, const struct timespec *now); void tcp_sock_handler(const struct ctx *c, union epoll_ref ref, - uint32_t events); -int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af, - const void *saddr, const void *daddr, uint32_t flow_lbl, - const struct pool *p, int idx, const struct timespec *now); + uint32_t events, unsigned int qpair); +int tcp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif, + sa_family_t af, const void *saddr, const void *daddr, + uint32_t flow_lbl, const struct pool *p, int idx, + const struct timespec *now); int tcp_listen(const struct ctx *c, uint8_t pif, unsigned rule, const union inany_addr *addr, const char *ifname, in_port_t port); int tcp_init(struct ctx *c); -void tcp_defer_handler(struct ctx *c, const struct timespec *now); +void tcp_defer_handler(struct ctx *c, const struct timespec *now, + unsigned int qpair);

void tcp_update_l2_buf(const unsigned char *eth_d);

diff --git a/tcp_buf.c b/tcp_buf.c index ca356089dc0b..ae8bebca5107 100644 --- a/tcp_buf.c +++ b/tcp_buf.c @@ -124,7 +124,7 @@ static void tcp_revert_seq(const struct ctx *c, struct tcp_tap_conn **conns, conn->seq_to_tap = seq; peek_offset = conn->seq_to_tap - conn->seq_ack_from_tap; if (tcp_set_peek_offset(conn, peek_offset)) - tcp_rst(c, conn); + tcp_rst(c, conn, QPAIR_DEFAULT); } }

@@ -334,7 +334,7 @@ int tcp_buf_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn) conn->seq_to_tap = conn->seq_ack_from_tap; already_sent = 0; if (tcp_set_peek_offset(conn, 0)) { - tcp_rst(c, conn); + tcp_rst(c, conn, QPAIR_DEFAULT); return -1; } } @@ -356,7 +356,7 @@ int tcp_buf_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn) }

if (tcp_prepare_iov(&mh_sock, iov_sock, already_sent, fill_bufs)) { - tcp_rst(c, conn); + tcp_rst(c, conn, QPAIR_DEFAULT); return -1; }

@@ -381,7 +381,7 @@ int tcp_buf_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn)

if (len < 0) { if (errno != EAGAIN && errno != EWOULDBLOCK) { - tcp_rst(c, conn); + tcp_rst(c, conn, QPAIR_DEFAULT); return -errno; }

@@ -410,7 +410,7 @@ int tcp_buf_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn)

ret = tcp_buf_send_flag(c, conn, FIN | ACK); if (ret) { - tcp_rst(c, conn); + tcp_rst(c, conn, QPAIR_DEFAULT); return ret; }

diff --git a/tcp_internal.h b/tcp_internal.h index 40472c9973c8..22f8825adccc 100644 --- a/tcp_internal.h +++ b/tcp_internal.h @@ -174,11 +174,12 @@ void conn_event_do(const struct ctx *c, struct tcp_tap_conn *conn, conn_event_do(c, conn, event); \ } while (0)

-void tcp_rst_do(const struct ctx *c, struct tcp_tap_conn *conn); -#define tcp_rst(c, conn) \ +void tcp_rst_do(const struct ctx *c, struct tcp_tap_conn *conn, + unsigned int qpair); +#define tcp_rst(c, conn, qpair) \ do { \ flow_dbg((conn), "TCP reset at %s:%i", __func__, __LINE__); \ - tcp_rst_do(c, conn); \ + tcp_rst_do(c, conn, qpair); \ } while (0)

struct tcp_info_linux; diff --git a/tcp_vu.c b/tcp_vu.c index 9ef6b5242c9c..4f76f599156f 100644 --- a/tcp_vu.c +++ b/tcp_vu.c @@ -116,15 +116,17 @@ static int tcp_vu_send_dup(const struct ctx *c, struct vu_virtq *vq, * @c: Execution context * @conn: Connection pointer * @flags: TCP flags: if not set, send segment only if ACK is due + * @qpair: Queue pair to process * * Return: -ECONNRESET on fatal connection error, * -EAGAIN if vhost-user buffers are unavailable, * 0 otherwise */ -int tcp_vu_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags) +int tcp_vu_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags, + unsigned int qpair) { struct vu_dev *vdev = c->vdev; - int rx_queue = QPAIR_TOGUEST_QUEUE(QPAIR_DEFAULT); + int rx_queue = QPAIR_TOGUEST_QUEUE(qpair); struct vu_virtq *vq = &vdev->vq[rx_queue]; size_t optlen, hdrlen, iov_cnt, iov_used; struct vu_virtq_element flags_elem[2]; @@ -424,13 +426,14 @@ static void tcp_vu_prepare(const struct ctx *c, struct tcp_tap_conn *conn, * in window * @c: Execution context * @conn: Connection pointer + * @qpair: Queue pair to process * * Return: negative on connection reset, 0 otherwise */ -int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn) +int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn, + unsigned int qpair) { uint32_t wnd_scaled = conn->wnd_from_tap << conn->ws_from_tap; - unsigned int qpair = QPAIR_DEFAULT; int rx_queue = QPAIR_TOGUEST_QUEUE(qpair); struct vu_dev *vdev = c->vdev; struct vu_virtq *vq = &vdev->vq[rx_queue]; @@ -454,7 +457,7 @@ int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn) conn->seq_to_tap = conn->seq_ack_from_tap; already_sent = 0; if (tcp_set_peek_offset(conn, 0)) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return -1; } } @@ -477,7 +480,7 @@ int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn) &elem_cnt, &frame_cnt); if (len < 0) { if (len != -EAGAIN && len != -EWOULDBLOCK) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return len; }

@@ -498,9 +501,9 @@ int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn) /* See tcp_buf_data_from_sock() */ conn->seq_ack_to_tap = conn->seq_from_tap;

- ret = tcp_vu_send_flag(c, conn, FIN | ACK); + ret = tcp_vu_send_flag(c, conn, FIN | ACK, qpair); if (ret) { - tcp_rst(c, conn); + tcp_rst(c, conn, qpair); return ret; }

diff --git a/tcp_vu.h b/tcp_vu.h index 6ab6057f352a..ae48420f4906 100644 --- a/tcp_vu.h +++ b/tcp_vu.h @@ -6,7 +6,9 @@ #ifndef TCP_VU_H #define TCP_VU_H

-int tcp_vu_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags); -int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn); +int tcp_vu_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags, + unsigned int qpair); +int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn, + unsigned int qpair);

#endif /*TCP_VU_H */ -- 2.54.0