[PATCH v6 11/12] flow: Add queue pair tracking to flow management

[Laurent Vivier <lvivier@redhat.com>]

Add a qpair field to struct flow_common so each flow knows which
queue pair (and thus which worker thread) owns it.

flow_alloc() takes a qpair parameter, set at creation time.  All
protocol handlers pass their qpair when allocating flows from tap,
and use QPAIR_DEFAULT for flows initiated from the host side.

tcp_keepalive() and tcp_inactivity() filter by qpair so each
worker only processes its own flows.

Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
 flow.c       | 34 +++++++++++++++++++++++++++++++++-
 flow.h       | 13 ++++++++++++-
 flow_table.h |  2 +-
 icmp.c       |  3 +--
 tcp.c        | 20 +++++++++++++-------
 udp_flow.c   |  9 ++++-----
 6 files changed, 64 insertions(+), 17 deletions(-)

diff --git a/flow.c b/flow.c
index c63491804709..40b2e68be0ab 100644
--- a/flow.c
+++ b/flow.c
@@ -417,6 +417,37 @@ void flow_epollid_register(int epollid, int epollfd)
 	epoll_id_to_fd[epollid] = epollfd;
 }
 
+/**
+ * flow_qp() - Get the queue pair for a flow
+ * @f:		Flow to query (may be NULL)
+ *
+ * Return: queue pair number for the flow, or 0 if flow is NULL or has no
+ *         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