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

[Laurent Vivier <lvivier@redhat.com>]

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 <lvivier@redhat.com>
---
 passt.c        |  12 +--
 tap.c          |   8 +-
 tcp.c          | 196 ++++++++++++++++++++++++++++---------------------
 tcp.h          |  15 ++--
 tcp_buf.c      |   8 +-
 tcp_internal.h |   6 +-
 tcp_vu.c       |  17 +++--
 tcp_vu.h       |   5 +-
 8 files changed, 149 insertions(+), 118 deletions(-)

diff --git a/passt.c b/passt.c
index 50545511c067..86ed339ff67f 100644
--- a/passt.c
+++ b/passt.c
@@ -108,11 +108,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);
@@ -261,7 +263,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, &now);
+			tcp_sock_handler(c, ref, eventmask, &now, QPAIR_DEFAULT);
 			break;
 		case EPOLL_TYPE_TCP_SPLICE:
 			tcp_splice_sock_handler(c, ref, eventmask, &now);
@@ -270,7 +272,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, &now);
+			tcp_timer_handler(c, ref, &now, QPAIR_DEFAULT);
 			break;
 		case EPOLL_TYPE_UDP_LISTEN:
 			udp_listen_sock_handler(c, ref, eventmask, &now);
@@ -310,7 +312,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, &now);
 }
diff --git a/tap.c b/tap.c
index 928cf5c086ee..9ce32d36f1d7 100644
--- a/tap.c
+++ b/tap.c
@@ -702,8 +702,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;
 
@@ -867,7 +865,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) {
@@ -952,8 +950,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;
 
@@ -1120,7 +1116,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 6b7612c27abf..982f8a4a845a 100644
--- a/tcp.c
+++ b/tcp.c
@@ -1273,17 +1273,18 @@ static void tcp_update_seqack_from_tap(const struct ctx *c,
  * @c:		Execution context
  * @conn:	Connection pointer
  * @now:	Current timestamp
+ * @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,
-			  const struct timespec *now)
+			  const struct timespec *now, unsigned int qpair)
 {
 	conn->seq_to_tap = conn->seq_ack_from_tap;
 	conn->events &= ~TAP_FIN_SENT;
 
 	if (tcp_set_peek_offset(conn, 0, now)) {
-		tcp_rst(c, conn, now);
+		tcp_rst(c, conn, now, qpair);
 		return -1;
 	}
 
@@ -1389,16 +1390,18 @@ int tcp_prepare_flags(const struct ctx *c, struct tcp_tap_conn *conn,
  * @conn:      Connection pointer
  * @flags:     TCP flags: if not set, send segment only if ACK is due
  * @now:	Current timestamp
+ * @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, const struct timespec *now)
+			 int flags, const struct timespec *now,
+			 unsigned int qpair)
 {
 	int ret;
 
 	if (c->mode == MODE_VU)
-		ret = tcp_vu_send_flag(c, conn, flags, now);
+		ret = tcp_vu_send_flag(c, conn, flags, now, qpair);
 	else
 		ret = tcp_buf_send_flag(c, conn, flags, now);
 
@@ -1446,15 +1449,16 @@ static void tcp_sock_rst(const struct ctx *c, struct tcp_tap_conn *conn,
  * @c:		Execution context
  * @conn:	Connection pointer
  * @now:	Current timestamp
+ * @qpair:	Queue pair to process
  */
 void tcp_rst_do(const struct ctx *c, struct tcp_tap_conn *conn,
-		const struct timespec *now)
+		const struct timespec *now, unsigned int qpair)
 {
 	if (conn->events == CLOSED)
 		return;
 
 	/* Send RST on tap */
-	tcp_send_flag(c, conn, RST, now);
+	tcp_send_flag(c, conn, RST, now, qpair);
 
 	tcp_sock_rst(c, conn, now);
 }
@@ -1482,12 +1486,13 @@ static void tcp_get_tap_ws(struct tcp_tap_conn *conn,
  * @conn:	Connection pointer
  * @wnd:	Window value, host order, unscaled
  * @now:	Current timestamp
+ * @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,
-				  const struct timespec *now)
+				  const struct timespec *now, unsigned int qpair)
 {
 	wnd = MIN(MAX_WINDOW, wnd << conn->ws_from_tap);
 
@@ -1498,7 +1503,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, now);
+		tcp_rewind_seq(c, conn, now, qpair);
 		return false;
 	}
 
@@ -1675,6 +1680,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
@@ -1685,10 +1691,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);
@@ -1791,7 +1798,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, now);
+			tcp_rst(c, conn, now, qpair);
 			goto cancel;
 		}
 
@@ -1799,7 +1806,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, now))
+		if (tcp_send_flag(c, conn, SYN | ACK, now, qpair))
 			goto cancel;
 
 		conn_event(c, conn, TAP_SYN_ACK_SENT, now);
@@ -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
  *
  * Return: negative on connection reset, 0 otherwise
  *
  * #syscalls recvmsg
  */
 static int tcp_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn,
-			      const struct timespec *now)
+			      const struct timespec *now, unsigned int qpair)
 {
 	uint32_t wnd_scaled = conn->wnd_from_tap << conn->ws_from_tap;
 	uint32_t already_sent;
@@ -1879,7 +1887,7 @@ static int tcp_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn,
 			   conn->seq_ack_from_tap, conn->seq_to_tap);
 		conn->seq_to_tap = conn->seq_ack_from_tap;
 		if (tcp_set_peek_offset(conn, 0, now)) {
-			tcp_rst(c, conn, now);
+			tcp_rst(c, conn, now, qpair);
 			return -1;
 		}
 	}
@@ -1895,7 +1903,7 @@ static int tcp_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn,
 	}
 
 	if (c->mode == MODE_VU)
-		return tcp_vu_data_from_sock(c, conn, already_sent, now);
+		return tcp_vu_data_from_sock(c, conn, already_sent, now, qpair);
 
 	return tcp_buf_data_from_sock(c, conn, already_sent, now);
 }
@@ -1924,6 +1932,7 @@ static ssize_t tcp_packet_data_len(const struct tcphdr *th, size_t l4len)
  * @p:		Pool of TCP packets, with TCP headers
  * @idx:	Index of first data packet in pool
  * @now:	Current timestamp
+ * @qpair:	Queue pair to process
  *
  * #syscalls sendmsg
  *
@@ -1931,7 +1940,7 @@ static ssize_t tcp_packet_data_len(const struct tcphdr *th, size_t l4len)
  */
 static int tcp_data_from_tap(const struct ctx *c, struct tcp_tap_conn *conn,
 			     const struct pool *p, int idx,
-			     const struct timespec *now)
+			     const struct timespec *now, 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;
@@ -1981,7 +1990,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, now))
+			if (tcp_send_flag(c, conn, ACK, now, qpair))
 				return -1;
 
 			tcp_timer_ctl(c, conn, now);
@@ -1992,7 +2001,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), now);
+						      ntohs(th->window), now,
+						      qpair);
 				return 1;
 			}
 
@@ -2018,7 +2028,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, now);
+					tcp_rewind_seq(c, conn, now, qpair);
 
 				max_ack_seq_wnd = ntohs(th->window);
 				max_ack_seq = ack_seq;
@@ -2080,17 +2090,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, now);
 
-	tcp_tap_window_update(c, conn, max_ack_seq_wnd, now);
+	tcp_tap_window_update(c, conn, max_ack_seq_wnd, now, 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, now))
+		if (tcp_rewind_seq(c, conn, now, qpair))
 			return -1;
 
-		tcp_data_from_sock(c, conn, now);
+		tcp_data_from_sock(c, conn, now, qpair);
 	}
 
 	if (!iov_i)
@@ -2106,7 +2116,7 @@ eintr:
 			 *   Then swiftly looked away and left.
 			 */
 			conn->seq_from_tap = seq_from_tap;
-			if (tcp_send_flag(c, conn, ACK, now))
+			if (tcp_send_flag(c, conn, ACK, now, qpair))
 				return -1;
 		}
 
@@ -2114,7 +2124,7 @@ eintr:
 			goto eintr;
 
 		if (errno == EAGAIN || errno == EWOULDBLOCK) {
-			if (tcp_send_flag(c, conn, ACK | DUP_ACK, now))
+			if (tcp_send_flag(c, conn, ACK | DUP_ACK, now, qpair))
 				return -1;
 
 			uint32_t events = tcp_conn_epoll_events(conn->events,
@@ -2150,7 +2160,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, now))
+			if (tcp_send_flag(c, conn, ACK | DUP_ACK, now, qpair))
 				return -1;
 		}
 		return p->count - idx;
@@ -2165,7 +2175,7 @@ out:
 
 		conn_event(c, conn, TAP_FIN_RCVD, now);
 	} else {
-		if (tcp_send_flag(c, conn, ACK_IF_NEEDED, now))
+		if (tcp_send_flag(c, conn, ACK_IF_NEEDED, now, qpair))
 			return -1;
 	}
 
@@ -2180,14 +2190,16 @@ out:
  * @opts:	Pointer to start of options
  * @optlen:	Bytes in options: caller MUST ensure available length
  * @now:	Current timestamp
+ * @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 struct timespec *now)
+				      const struct timespec *now,
+				      unsigned int qpair)
 {
-	tcp_tap_window_update(c, conn, ntohs(th->window), now);
+	tcp_tap_window_update(c, conn, ntohs(th->window), now, qpair);
 	tcp_get_tap_ws(conn, opts, optlen);
 
 	/* First value is not scaled */
@@ -2202,24 +2214,25 @@ static void tcp_conn_from_sock_finish(const struct ctx *c,
 
 	conn_event(c, conn, ESTABLISHED, now);
 	if (tcp_set_peek_offset(conn, 0, now)) {
-		tcp_rst(c, conn, now);
+		tcp_rst(c, conn, now, qpair);
 		return;
 	}
 
-	if (tcp_send_flag(c, conn, ACK, now)) {
-		tcp_rst(c, conn, now);
+	if (tcp_send_flag(c, conn, ACK, now, qpair)) {
+		tcp_rst(c, conn, now, 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, now);
+	tcp_data_from_sock(c, conn, now, 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
@@ -2227,7 +2240,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)
@@ -2285,12 +2298,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
@@ -2302,9 +2316,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;
@@ -2341,10 +2356,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;
 	}
 
@@ -2369,7 +2385,7 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af,
 	if (conn->events & SOCK_ACCEPTED) {
 		if (th->syn && th->ack && !th->fin) {
 			tcp_conn_from_sock_finish(c, conn, th, opts, optlen,
-						  now);
+						  now, qpair);
 			return 1;
 		}
 
@@ -2397,7 +2413,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, now))
+			if (tcp_send_flag(c, conn, ACK, now, qpair))
 				goto reset;
 
 			conn_event(c, conn, SOCK_FIN_SENT, now);
@@ -2408,8 +2424,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), now))
-			tcp_data_from_sock(c, conn, now);
+		if (tcp_tap_window_update(c, conn, ntohs(th->window), now, qpair))
+			tcp_data_from_sock(c, conn, now, qpair);
 
 		if (p->count - idx == 1)
 			return 1;
@@ -2441,13 +2457,13 @@ 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, now))
+			if (tcp_rewind_seq(c, conn, now, qpair))
 				return -1;
 		}
 
-		if (tcp_tap_window_update(c, conn, ntohs(th->window), now) ||
+		if (tcp_tap_window_update(c, conn, ntohs(th->window), now, qpair) ||
 		    retr)
-			tcp_data_from_sock(c, conn, now);
+			tcp_data_from_sock(c, conn, now, qpair);
 
 		if (conn->seq_ack_from_tap == conn->seq_to_tap) {
 			if (th->ack && conn->events & TAP_FIN_SENT)
@@ -2462,7 +2478,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, now);
+	count = tcp_data_from_tap(c, conn, p, idx, now, qpair);
 	if (count == -1)
 		goto reset;
 
@@ -2482,7 +2498,7 @@ int tcp_tap_handler(const struct ctx *c, uint8_t pif, sa_family_t af,
 		}
 
 		conn_event(c, conn, SOCK_FIN_SENT, now);
-		if (tcp_send_flag(c, conn, ACK, now))
+		if (tcp_send_flag(c, conn, ACK, now, qpair))
 			goto reset;
 
 		ack_due = 0;
@@ -2512,7 +2528,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, now);
+	tcp_rst(c, conn, now, qpair);
 	return p->count - idx;
 }
 
@@ -2521,21 +2537,22 @@ reset:
  * @c:		Execution context
  * @conn:	Connection pointer
  * @now:	Current timestamp
+ * @qpair:	Queue pair to process
  */
 static void tcp_connect_finish(const struct ctx *c, struct tcp_tap_conn *conn,
-			       const struct timespec *now)
+			       const struct timespec *now, 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, now);
+		tcp_rst(c, conn, now, qpair);
 		return;
 	}
 
-	if (tcp_send_flag(c, conn, SYN | ACK, now)) {
-		tcp_rst(c, conn, now);
+	if (tcp_send_flag(c, conn, SYN | ACK, now, qpair)) {
+		tcp_rst(c, conn, now, qpair);
 		return;
 	}
 
@@ -2578,7 +2595,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, now)) {
+	if (tcp_send_flag(c, conn, SYN, now, QPAIR_DEFAULT)) {
 		conn_flag(c, conn, CLOSING, now);
 		FLOW_ACTIVATE(conn);
 		return;
@@ -2672,13 +2689,14 @@ cancel:
  * @c:		Execution context
  * @ref:	epoll reference of timer (not connection)
  * @now:	Current timestamp
+ * @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,
-		       const struct timespec *now)
+		       const struct timespec *now, unsigned int qpair)
 {
 	struct itimerspec check_armed = { { 0 }, { 0 } };
 	struct tcp_tap_conn *conn = &FLOW(ref.flow)->tcp;
@@ -2697,8 +2715,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, now)) {
-			tcp_rst(c, conn, now);
+		if (tcp_send_flag(c, conn, ACK_IF_NEEDED, now, qpair)) {
+			tcp_rst(c, conn, now, qpair);
 			return;
 		}
 		tcp_timer_ctl(c, conn, now);
@@ -2710,11 +2728,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, now);
+				tcp_rst(c, conn, now, qpair);
 			} else {
 				flow_trace(conn, "SYN timeout, retry");
-				if (tcp_send_flag(c, conn, SYN, now)) {
-					tcp_rst(c, conn, now);
+				if (tcp_send_flag(c, conn, SYN, now, qpair)) {
+					tcp_rst(c, conn, now, qpair);
 					return;
 				}
 				conn->retries++;
@@ -2723,7 +2741,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, now);
+			tcp_rst(c, conn, now, qpair);
 		} else {
 			flow_dbg(conn, "ACK timeout, retry");
 
@@ -2731,10 +2749,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, now))
+			if (tcp_rewind_seq(c, conn, now, qpair))
 				return;
 
-			tcp_data_from_sock(c, conn, now);
+			tcp_data_from_sock(c, conn, now, qpair);
 			tcp_timer_ctl(c, conn, now);
 		}
 	}
@@ -2746,9 +2764,11 @@ void tcp_timer_handler(const struct ctx *c, union epoll_ref ref,
  * @ref:	epoll reference
  * @events:	epoll events bitmap
  * @now:	Current timestamp
+ * @qpair:	Queue pair to process
  */
 void tcp_sock_handler(const struct ctx *c, union epoll_ref ref,
-		      uint32_t events, const struct timespec *now)
+		      uint32_t events, const struct timespec *now,
+		      unsigned int qpair)
 {
 	struct tcp_tap_conn *conn = conn_at_sidx(ref.flowside);
 
@@ -2759,7 +2779,7 @@ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref,
 		return;
 
 	if (events & EPOLLERR) {
-		tcp_rst(c, conn, now);
+		tcp_rst(c, conn, now, qpair);
 		return;
 	}
 
@@ -2778,13 +2798,13 @@ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref,
 			conn_event(c, conn, SOCK_FIN_RCVD, now);
 
 		if (events & EPOLLIN)
-			tcp_data_from_sock(c, conn, now);
+			tcp_data_from_sock(c, conn, now, qpair);
 
 		if (events & EPOLLOUT) {
 			tcp_epoll_ctl(conn);
 			if (tcp_update_seqack_wnd(c, conn, false, NULL, now) &&
-			    tcp_send_flag(c, conn, ACK, now)) {
-				tcp_rst(c, conn, now);
+			    tcp_send_flag(c, conn, ACK, now, qpair)) {
+				tcp_rst(c, conn, now, qpair);
 				return;
 			}
 		}
@@ -2794,7 +2814,7 @@ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref,
 
 	/* EPOLLHUP during handshake: reset */
 	if (events & EPOLLHUP) {
-		tcp_rst(c, conn, now);
+		tcp_rst(c, conn, now, qpair);
 		return;
 	}
 
@@ -2804,7 +2824,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, now);
+			tcp_connect_finish(c, conn, now, qpair);
 		/* Data? Check later */
 	}
 }
@@ -2971,9 +2991,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)
 {
 	union flow *flow;
 
@@ -2988,8 +3010,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, now))
-				tcp_rst(c, conn, now);
+			if (tcp_send_flag(c, conn, KEEPALIVE, now, qpair))
+				tcp_rst(c, conn, now, qpair);
 		}
 
 		/* Ready to check fot next interval */
@@ -2999,9 +3021,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)
 {
 	union flow *flow;
 
@@ -3018,7 +3042,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, now);
+			tcp_rst(c, conn, now, qpair);
 		}
 
 		/* Ready to check fot next interval */
@@ -3030,9 +3054,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, now);
 
@@ -3045,8 +3071,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);
 }
 
 /**
@@ -4024,10 +4050,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))
+	if (tcp_send_flag(c, conn, ACK, now, QPAIR_DEFAULT))
 		goto fail;
 
-	tcp_data_from_sock(c, conn, now);
+	tcp_data_from_sock(c, conn, now, QPAIR_DEFAULT);
 
 	if ((rc = tcp_epoll_ctl(conn))) {
 		flow_dbg(conn,
@@ -4045,7 +4071,7 @@ fail:
 	}
 
 	conn->flags = 0; /* Not waiting for ACK, don't schedule timer */
-	tcp_rst(c, conn, now);
+	tcp_rst(c, conn, now, QPAIR_DEFAULT);
 
 	return 0;
 }
diff --git a/tcp.h b/tcp.h
index b68f882d6c40..eb4f8994f4b7 100644
--- a/tcp.h
+++ b/tcp.h
@@ -19,16 +19,19 @@
 struct ctx;
 
 void tcp_timer_handler(const struct ctx *c, union epoll_ref ref,
-		       const struct timespec *now);
+		       const struct timespec *now, 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, const struct timespec *now);
-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, const struct timespec *now,
+		      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_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 72c454122c37..adef31883b57 100644
--- a/tcp_buf.c
+++ b/tcp_buf.c
@@ -126,7 +126,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, now))
-			tcp_rst(c, conn, now);
+			tcp_rst(c, conn, now, QPAIR_DEFAULT);
 	}
 }
 
@@ -345,7 +345,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, now);
+		tcp_rst(c, conn, now, QPAIR_DEFAULT);
 		return -1;
 	}
 
@@ -370,7 +370,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, now);
+			tcp_rst(c, conn, now, QPAIR_DEFAULT);
 			return -errno;
 		}
 
@@ -399,7 +399,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, now);
 			if (ret) {
-				tcp_rst(c, conn, now);
+				tcp_rst(c, conn, now, QPAIR_DEFAULT);
 				return ret;
 			}
 
diff --git a/tcp_internal.h b/tcp_internal.h
index 1af251823b6f..c26133650388 100644
--- a/tcp_internal.h
+++ b/tcp_internal.h
@@ -175,11 +175,11 @@ void conn_event_do(const struct ctx *c, struct tcp_tap_conn *conn,
 	} while (0)
 
 void tcp_rst_do(const struct ctx *c, struct tcp_tap_conn *conn,
-		const struct timespec *now);
-#define tcp_rst(c, conn, now)						\
+		const struct timespec *now, unsigned int qpair);
+#define tcp_rst(c, conn, now, qpair)						\
 	do {								\
 		flow_dbg((conn), "TCP reset at %s:%i", __func__, __LINE__); \
-		tcp_rst_do(c, conn, now);				\
+		tcp_rst_do(c, conn, now, qpair);				\
 	} while (0)
 
 struct tcp_info_linux;
diff --git a/tcp_vu.c b/tcp_vu.c
index d2950dcc99e6..c9a1e08747b5 100644
--- a/tcp_vu.c
+++ b/tcp_vu.c
@@ -117,16 +117,17 @@ static int tcp_vu_send_dup(const struct ctx *c, struct vu_virtq *vq,
  * @conn:	Connection pointer
  * @flags:	TCP flags: if not set, send segment only if ACK is due
  * @now:	Current timestamp
+ * @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,
-		     const struct timespec *now)
+		     const struct timespec *now, unsigned int qpair)
 {
 	struct vu_dev *vdev = c->vdev;
-	int toguest = QPAIR_TOGUEST_QUEUE(QPAIR_DEFAULT);
+	int toguest = QPAIR_TOGUEST_QUEUE(qpair);
 	struct vu_virtq *vq = &vdev->vq[toguest];
 	size_t optlen, hdrlen, iov_cnt, iov_used;
 	struct vu_virtq_element flags_elem[2];
@@ -428,14 +429,16 @@ static void tcp_vu_prepare(const struct ctx *c, struct tcp_tap_conn *conn,
  * @conn:	Connection pointer
  * @already_sent:	Number of bytes already sent to tap, but not acked
  * @now:	Current timestamp
+ * @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,
-			  uint32_t already_sent, const struct timespec *now)
+			  uint32_t already_sent, const struct timespec *now,
+			  unsigned int qpair)
 {
 	uint32_t wnd_scaled = conn->wnd_from_tap << conn->ws_from_tap;
-	int toguest = QPAIR_TOGUEST_QUEUE(QPAIR_DEFAULT);
+	int toguest = QPAIR_TOGUEST_QUEUE(qpair);
 	struct vu_dev *vdev = c->vdev;
 	struct vu_virtq *vq = &vdev->vq[toguest];
 	ssize_t len, previous_dlen;
@@ -460,7 +463,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, now);
+			tcp_rst(c, conn, now, qpair);
 			return len;
 		}
 
@@ -481,9 +484,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, now);
+			ret = tcp_vu_send_flag(c, conn, FIN | ACK, now, qpair);
 			if (ret) {
-				tcp_rst(c, conn, now);
+				tcp_rst(c, conn, now, qpair);
 				return ret;
 			}
 
diff --git a/tcp_vu.h b/tcp_vu.h
index 6f2aa2e1a088..75df65d92152 100644
--- a/tcp_vu.h
+++ b/tcp_vu.h
@@ -7,8 +7,9 @@
 #define TCP_VU_H
 
 int tcp_vu_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags,
-		     const struct timespec *now);
+		     const struct timespec *now, unsigned int qpair);
 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 already_sent, const struct timespec *now,
+			  unsigned int qpair);
 
 #endif  /*TCP_VU_H */
-- 
2.54.0