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