[PATCH v2 0/7] UDP flow socket preliminaries

[PATCH v2 0/7] UDP flow socket preliminaries

[David Gibson]

As discussed on our recent call, I'm working towards using connected
sockets on both sides of UDP flows.  This series makes some
preliminary reworks that simplify things and make that easier.

v2:
 * Added patches 5..7, other patches unchanged.

David Gibson (7):
  udp: Common invocation of udp_sock_errs() for vhost-user and "buf"
    paths
  udp: Simplify checking of epoll event bits
  udp_vu: Factor things out of udp_vu_reply_sock_data() loop
  udp: Share more logic between vu and non-vu reply socket paths
  udp: Better handling of failure to forward from reply socket
  udp: Always hash socket facing flowsides
  udp: Add helper function for creating connected UDP socket

 udp.c          | 121 +++++++++++++++++++++--------------------
 udp_flow.c     | 143 ++++++++++++++++++++++++++-----------------------
 udp_internal.h |   2 +-
 udp_vu.c       |  68 +++++++----------------
 udp_vu.h       |   9 ++--
 5 files changed, 166 insertions(+), 177 deletions(-)

-- 
2.49.0

[PATCH v2 1/7] udp: Common invocation of udp_sock_errs() for vhost-user and "buf" paths

[David Gibson]

The vhost-user and non-vhost-user paths for both udp_listen_sock_handler()
and udp_reply_sock_handler() are more or less completely separate.  Both,
however, start with essentially the same invocation of udp_sock_errs(), so
that can be made common.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 udp.c          | 37 ++++++++++++++++++++-----------------
 udp_internal.h |  2 +-
 udp_vu.c       | 15 ---------------
 3 files changed, 21 insertions(+), 33 deletions(-)

diff --git a/udp.c b/udp.c
index 88128181..ca101fec 100644
--- a/udp.c
+++ b/udp.c
@@ -587,7 +587,8 @@ static int udp_sock_recverr(const struct ctx *c, union epoll_ref ref)
  *
  * Return: Number of errors handled, or < 0 if we have an unrecoverable error
  */
-int udp_sock_errs(const struct ctx *c, union epoll_ref ref, uint32_t events)
+static int udp_sock_errs(const struct ctx *c, union epoll_ref ref,
+			 uint32_t events)
 {
 	unsigned n_err = 0;
 	socklen_t errlen;
@@ -680,13 +681,6 @@ static void udp_buf_listen_sock_handler(const struct ctx *c,
 	const socklen_t sasize = sizeof(udp_meta[0].s_in);
 	int n, i;
 
-	if (udp_sock_errs(c, ref, events) < 0) {
-		err("UDP: Unrecoverable error on listening socket:"
-		    " (%s port %hu)", pif_name(ref.udp.pif), ref.udp.port);
-		/* FIXME: what now?  close/re-open socket? */
-		return;
-	}
-
 	if ((n = udp_sock_recv(c, ref.fd, events, udp_mh_recv)) <= 0)
 		return;
 
@@ -752,6 +746,13 @@ void udp_listen_sock_handler(const struct ctx *c,
 			     union epoll_ref ref, uint32_t events,
 			     const struct timespec *now)
 {
+	if (udp_sock_errs(c, ref, events) < 0) {
+		err("UDP: Unrecoverable error on listening socket:"
+		    " (%s port %hu)", pif_name(ref.udp.pif), ref.udp.port);
+		/* FIXME: what now?  close/re-open socket? */
+		return;
+	}
+
 	if (c->mode == MODE_VU) {
 		udp_vu_listen_sock_handler(c, ref, events, now);
 		return;
@@ -779,17 +780,8 @@ static void udp_buf_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 	uint8_t topif = pif_at_sidx(tosidx);
 	int n, i, from_s;
 
-	ASSERT(!c->no_udp && uflow);
-
 	from_s = uflow->s[ref.flowside.sidei];
 
-	if (udp_sock_errs(c, ref, events) < 0) {
-		flow_err(uflow, "Unrecoverable error on reply socket");
-		flow_err_details(uflow);
-		udp_flow_close(c, uflow);
-		return;
-	}
-
 	if ((n = udp_sock_recv(c, from_s, events, udp_mh_recv)) <= 0)
 		return;
 
@@ -827,6 +819,17 @@ static void udp_buf_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 			    uint32_t events, const struct timespec *now)
 {
+	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
+
+	ASSERT(!c->no_udp && uflow);
+
+	if (udp_sock_errs(c, ref, events) < 0) {
+		flow_err(uflow, "Unrecoverable error on reply socket");
+		flow_err_details(uflow);
+		udp_flow_close(c, uflow);
+		return;
+	}
+
 	if (c->mode == MODE_VU) {
 		udp_vu_reply_sock_handler(c, ref, events, now);
 		return;
diff --git a/udp_internal.h b/udp_internal.h
index 3b081f54..02724e59 100644
--- a/udp_internal.h
+++ b/udp_internal.h
@@ -30,5 +30,5 @@ size_t udp_update_hdr4(struct iphdr *ip4h, struct udp_payload_t *bp,
 size_t udp_update_hdr6(struct ipv6hdr *ip6h, struct udp_payload_t *bp,
                        const struct flowside *toside, size_t dlen,
 		       bool no_udp_csum);
-int udp_sock_errs(const struct ctx *c, union epoll_ref ref, uint32_t events);
+
 #endif /* UDP_INTERNAL_H */
diff --git a/udp_vu.c b/udp_vu.c
index c26a223d..84f52aff 100644
--- a/udp_vu.c
+++ b/udp_vu.c
@@ -227,12 +227,6 @@ void udp_vu_listen_sock_handler(const struct ctx *c, union epoll_ref ref,
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
 	int i;
 
-	if (udp_sock_errs(c, ref, events) < 0) {
-		err("UDP: Unrecoverable error on listening socket:"
-		    " (%s port %hu)", pif_name(ref.udp.pif), ref.udp.port);
-		return;
-	}
-
 	for (i = 0; i < UDP_MAX_FRAMES; i++) {
 		const struct flowside *toside;
 		union sockaddr_inany s_in;
@@ -300,15 +294,6 @@ void udp_vu_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
 	int i;
 
-	ASSERT(!c->no_udp);
-
-	if (udp_sock_errs(c, ref, events) < 0) {
-		flow_err(uflow, "Unrecoverable error on reply socket");
-		flow_err_details(uflow);
-		udp_flow_close(c, uflow);
-		return;
-	}
-
 	for (i = 0; i < UDP_MAX_FRAMES; i++) {
 		uint8_t topif = pif_at_sidx(tosidx);
 		ssize_t dlen;
-- 
@@ -227,12 +227,6 @@ void udp_vu_listen_sock_handler(const struct ctx *c, union epoll_ref ref,
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
 	int i;
 
-	if (udp_sock_errs(c, ref, events) < 0) {
-		err("UDP: Unrecoverable error on listening socket:"
-		    " (%s port %hu)", pif_name(ref.udp.pif), ref.udp.port);
-		return;
-	}
-
 	for (i = 0; i < UDP_MAX_FRAMES; i++) {
 		const struct flowside *toside;
 		union sockaddr_inany s_in;
@@ -300,15 +294,6 @@ void udp_vu_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
 	int i;
 
-	ASSERT(!c->no_udp);
-
-	if (udp_sock_errs(c, ref, events) < 0) {
-		flow_err(uflow, "Unrecoverable error on reply socket");
-		flow_err_details(uflow);
-		udp_flow_close(c, uflow);
-		return;
-	}
-
 	for (i = 0; i < UDP_MAX_FRAMES; i++) {
 		uint8_t topif = pif_at_sidx(tosidx);
 		ssize_t dlen;
-- 
2.49.0

[PATCH v2 2/7] udp: Simplify checking of epoll event bits

[David Gibson]

udp_{listen,reply}_sock_handler() can accept both EPOLLERR and EPOLLIN
events.  However, unlike most epoll event handlers we don't check the
event bits right there.  EPOLLERR is checked within udp_sock_errs() which
we call unconditionally.  Checking EPOLLIN is still more buried: it is
checked within both udp_sock_recv() and udp_vu_sock_recv().

We can simplify the logic and pass less extraneous parameters around by
moving the checking of the event bits to the top level event handlers.

This makes udp_{buf,vu}_{listen,reply}_sock_handler() no longer general
event handlers, but specific to EPOLLIN events, meaning new data.  So,
rename those functions to udp_{buf,vu}_{listen,reply}_sock_data() to better
reflect their function.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 udp.c    | 78 ++++++++++++++++++++++++--------------------------------
 udp_vu.c | 25 +++++++-----------
 udp_vu.h |  8 +++---
 3 files changed, 47 insertions(+), 64 deletions(-)

diff --git a/udp.c b/udp.c
index ca101fec..55021ac4 100644
--- a/udp.c
+++ b/udp.c
@@ -583,12 +583,10 @@ static int udp_sock_recverr(const struct ctx *c, union epoll_ref ref)
  * udp_sock_errs() - Process errors on a socket
  * @c:		Execution context
  * @ref:	epoll reference
- * @events:	epoll events bitmap
  *
  * Return: Number of errors handled, or < 0 if we have an unrecoverable error
  */
-static int udp_sock_errs(const struct ctx *c, union epoll_ref ref,
-			 uint32_t events)
+static int udp_sock_errs(const struct ctx *c, union epoll_ref ref)
 {
 	unsigned n_err = 0;
 	socklen_t errlen;
@@ -597,9 +595,6 @@ static int udp_sock_errs(const struct ctx *c, union epoll_ref ref,
 
 	ASSERT(!c->no_udp);
 
-	if (!(events & EPOLLERR))
-		return 0; /* Nothing to do */
-
 	/* Empty the error queue */
 	while ((rc = udp_sock_recverr(c, ref)) > 0)
 		n_err += rc;
@@ -632,15 +627,13 @@ static int udp_sock_errs(const struct ctx *c, union epoll_ref ref,
  * udp_sock_recv() - Receive datagrams from a socket
  * @c:		Execution context
  * @s:		Socket to receive from
- * @events:	epoll events bitmap
  * @mmh		mmsghdr array to receive into
  *
  * Return: Number of datagrams received
  *
  * #syscalls recvmmsg arm:recvmmsg_time64 i686:recvmmsg_time64
  */
-static int udp_sock_recv(const struct ctx *c, int s, uint32_t events,
-			 struct mmsghdr *mmh)
+static int udp_sock_recv(const struct ctx *c, int s, struct mmsghdr *mmh)
 {
 	/* For not entirely clear reasons (data locality?) pasta gets better
 	 * throughput if we receive tap datagrams one at a atime.  For small
@@ -653,9 +646,6 @@ static int udp_sock_recv(const struct ctx *c, int s, uint32_t events,
 
 	ASSERT(!c->no_udp);
 
-	if (!(events & EPOLLIN))
-		return 0;
-
 	n = recvmmsg(s, mmh, n, 0, NULL);
 	if (n < 0) {
 		err_perror("Error receiving datagrams");
@@ -666,22 +656,20 @@ static int udp_sock_recv(const struct ctx *c, int s, uint32_t events,
 }
 
 /**
- * udp_buf_listen_sock_handler() - Handle new data from socket
+ * udp_buf_listen_sock_data() - Handle new data from socket
  * @c:		Execution context
  * @ref:	epoll reference
- * @events:	epoll events bitmap
  * @now:	Current timestamp
  *
  * #syscalls recvmmsg
  */
-static void udp_buf_listen_sock_handler(const struct ctx *c,
-					union epoll_ref ref, uint32_t events,
-					const struct timespec *now)
+static void udp_buf_listen_sock_data(const struct ctx *c, union epoll_ref ref,
+				     const struct timespec *now)
 {
 	const socklen_t sasize = sizeof(udp_meta[0].s_in);
 	int n, i;
 
-	if ((n = udp_sock_recv(c, ref.fd, events, udp_mh_recv)) <= 0)
+	if ((n = udp_sock_recv(c, ref.fd, udp_mh_recv)) <= 0)
 		return;
 
 	/* We divide datagrams into batches based on how we need to send them,
@@ -746,33 +734,33 @@ void udp_listen_sock_handler(const struct ctx *c,
 			     union epoll_ref ref, uint32_t events,
 			     const struct timespec *now)
 {
-	if (udp_sock_errs(c, ref, events) < 0) {
-		err("UDP: Unrecoverable error on listening socket:"
-		    " (%s port %hu)", pif_name(ref.udp.pif), ref.udp.port);
-		/* FIXME: what now?  close/re-open socket? */
-		return;
+	if (events & EPOLLERR) {
+		if (udp_sock_errs(c, ref) < 0) {
+			err("UDP: Unrecoverable error on listening socket:"
+			    " (%s port %hu)", pif_name(ref.udp.pif), ref.udp.port);
+			/* FIXME: what now?  close/re-open socket? */
+			return;
+		}
 	}
 
-	if (c->mode == MODE_VU) {
-		udp_vu_listen_sock_handler(c, ref, events, now);
-		return;
+	if (events & EPOLLIN) {
+		if (c->mode == MODE_VU)
+			udp_vu_listen_sock_data(c, ref, now);
+		else
+			udp_buf_listen_sock_data(c, ref, now);
 	}
-
-	udp_buf_listen_sock_handler(c, ref, events, now);
 }
 
 /**
- * udp_buf_reply_sock_handler() - Handle new data from flow specific socket
+ * udp_buf_reply_sock_data() - Handle new data from flow specific socket
  * @c:		Execution context
  * @ref:	epoll reference
- * @events:	epoll events bitmap
  * @now:	Current timestamp
  *
  * #syscalls recvmmsg
  */
-static void udp_buf_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
-				       uint32_t events,
-				       const struct timespec *now)
+static void udp_buf_reply_sock_data(const struct ctx *c, union epoll_ref ref,
+				    const struct timespec *now)
 {
 	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
 	const struct flowside *toside = flowside_at_sidx(tosidx);
@@ -782,7 +770,7 @@ static void udp_buf_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 
 	from_s = uflow->s[ref.flowside.sidei];
 
-	if ((n = udp_sock_recv(c, from_s, events, udp_mh_recv)) <= 0)
+	if ((n = udp_sock_recv(c, from_s, udp_mh_recv)) <= 0)
 		return;
 
 	flow_trace(uflow, "Received %d datagrams on reply socket", n);
@@ -823,19 +811,21 @@ void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 
 	ASSERT(!c->no_udp && uflow);
 
-	if (udp_sock_errs(c, ref, events) < 0) {
-		flow_err(uflow, "Unrecoverable error on reply socket");
-		flow_err_details(uflow);
-		udp_flow_close(c, uflow);
-		return;
+	if (events & EPOLLERR) {
+		if (udp_sock_errs(c, ref) < 0) {
+			flow_err(uflow, "Unrecoverable error on reply socket");
+			flow_err_details(uflow);
+			udp_flow_close(c, uflow);
+			return;
+		}
 	}
 
-	if (c->mode == MODE_VU) {
-		udp_vu_reply_sock_handler(c, ref, events, now);
-		return;
+	if (events & EPOLLIN) {
+		if (c->mode == MODE_VU)
+			udp_vu_reply_sock_data(c, ref, now);
+		else
+			udp_buf_reply_sock_data(c, ref, now);
 	}
-
-	udp_buf_reply_sock_handler(c, ref, events, now);
 }
 
 /**
diff --git a/udp_vu.c b/udp_vu.c
index 84f52aff..698667f6 100644
--- a/udp_vu.c
+++ b/udp_vu.c
@@ -78,14 +78,12 @@ static int udp_vu_sock_info(int s, union sockaddr_inany *s_in)
  * udp_vu_sock_recv() - Receive datagrams from socket into vhost-user buffers
  * @c:		Execution context
  * @s:		Socket to receive from
- * @events:	epoll events bitmap
  * @v6:		Set for IPv6 connections
  * @dlen:	Size of received data (output)
  *
  * Return: Number of iov entries used to store the datagram
  */
-static int udp_vu_sock_recv(const struct ctx *c, int s, uint32_t events,
-			    bool v6, ssize_t *dlen)
+static int udp_vu_sock_recv(const struct ctx *c, int s, bool v6, ssize_t *dlen)
 {
 	struct vu_dev *vdev = c->vdev;
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
@@ -95,9 +93,6 @@ static int udp_vu_sock_recv(const struct ctx *c, int s, uint32_t events,
 
 	ASSERT(!c->no_udp);
 
-	if (!(events & EPOLLIN))
-		return 0;
-
 	/* compute L2 header length */
 	hdrlen = udp_vu_hdrlen(v6);
 
@@ -214,14 +209,13 @@ static void udp_vu_csum(const struct flowside *toside, int iov_used)
 }
 
 /**
- * udp_vu_listen_sock_handler() - Handle new data from socket
+ * udp_vu_listen_sock_data() - Handle new data from socket
  * @c:		Execution context
  * @ref:	epoll reference
- * @events:	epoll events bitmap
  * @now:	Current timestamp
  */
-void udp_vu_listen_sock_handler(const struct ctx *c, union epoll_ref ref,
-				uint32_t events, const struct timespec *now)
+void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
+			     const struct timespec *now)
 {
 	struct vu_dev *vdev = c->vdev;
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
@@ -262,7 +256,7 @@ void udp_vu_listen_sock_handler(const struct ctx *c, union epoll_ref ref,
 
 		v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
 
-		iov_used = udp_vu_sock_recv(c, ref.fd, events, v6, &dlen);
+		iov_used = udp_vu_sock_recv(c, ref.fd, v6, &dlen);
 		if (iov_used <= 0)
 			break;
 
@@ -277,14 +271,13 @@ void udp_vu_listen_sock_handler(const struct ctx *c, union epoll_ref ref,
 }
 
 /**
- * udp_vu_reply_sock_handler() - Handle new data from flow specific socket
+ * udp_vu_reply_sock_data() - Handle new data from flow specific socket
  * @c:		Execution context
  * @ref:	epoll reference
- * @events:	epoll events bitmap
  * @now:	Current timestamp
  */
-void udp_vu_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
-			        uint32_t events, const struct timespec *now)
+void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
+			    const struct timespec *now)
 {
 	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
 	const struct flowside *toside = flowside_at_sidx(tosidx);
@@ -313,7 +306,7 @@ void udp_vu_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 
 		v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
 
-		iov_used = udp_vu_sock_recv(c, from_s, events, v6, &dlen);
+		iov_used = udp_vu_sock_recv(c, from_s, v6, &dlen);
 		if (iov_used <= 0)
 			break;
 		flow_trace(uflow, "Received 1 datagram on reply socket");
diff --git a/udp_vu.h b/udp_vu.h
index ba7018d3..4f2262d0 100644
--- a/udp_vu.h
+++ b/udp_vu.h
@@ -6,8 +6,8 @@
 #ifndef UDP_VU_H
 #define UDP_VU_H
 
-void udp_vu_listen_sock_handler(const struct ctx *c, union epoll_ref ref,
-				uint32_t events, const struct timespec *now);
-void udp_vu_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
-			       uint32_t events, const struct timespec *now);
+void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
+			     const struct timespec *now);
+void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
+			    const struct timespec *now);
 #endif /* UDP_VU_H */
-- 
@@ -6,8 +6,8 @@
 #ifndef UDP_VU_H
 #define UDP_VU_H
 
-void udp_vu_listen_sock_handler(const struct ctx *c, union epoll_ref ref,
-				uint32_t events, const struct timespec *now);
-void udp_vu_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
-			       uint32_t events, const struct timespec *now);
+void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
+			     const struct timespec *now);
+void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
+			    const struct timespec *now);
 #endif /* UDP_VU_H */
-- 
2.49.0

[PATCH v2 3/7] udp_vu: Factor things out of udp_vu_reply_sock_data() loop

[David Gibson]

At the start of every cycle of the loop in udp_vu_reply_sock_data() we:
 - ASSERT that uflow is not NULL
 - Check if the target pif is PIF_TAP
 - Initialize the v6 boolean

However, all of these depend only on the flow, which doesn't change across
the loop.  This is probably a duplication from udp_vu_listen_sock_data(),
where the flow can be different for each packet.  For the reply socket
case, however, factor that logic out of the loop.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 udp_vu.c | 28 +++++++++++++---------------
 1 file changed, 13 insertions(+), 15 deletions(-)

diff --git a/udp_vu.c b/udp_vu.c
index 698667f6..6e1823a9 100644
--- a/udp_vu.c
+++ b/udp_vu.c
@@ -281,30 +281,28 @@ void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
 {
 	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
 	const struct flowside *toside = flowside_at_sidx(tosidx);
+	bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
 	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
 	int from_s = uflow->s[ref.flowside.sidei];
 	struct vu_dev *vdev = c->vdev;
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
+	uint8_t topif = pif_at_sidx(tosidx);
 	int i;
 
-	for (i = 0; i < UDP_MAX_FRAMES; i++) {
-		uint8_t topif = pif_at_sidx(tosidx);
-		ssize_t dlen;
-		int iov_used;
-		bool v6;
-
-		ASSERT(uflow);
+	ASSERT(uflow);
 
-		if (topif != PIF_TAP) {
-			uint8_t frompif = pif_at_sidx(ref.flowside);
+	if (topif != PIF_TAP) {
+		uint8_t frompif = pif_at_sidx(ref.flowside);
 
-			flow_err(uflow,
-				 "No support for forwarding UDP from %s to %s",
-				 pif_name(frompif), pif_name(topif));
-			continue;
-		}
+		flow_err(uflow,
+			 "No support for forwarding UDP from %s to %s",
+			 pif_name(frompif), pif_name(topif));
+		return;
+	}
 
-		v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
+	for (i = 0; i < UDP_MAX_FRAMES; i++) {
+		ssize_t dlen;
+		int iov_used;
 
 		iov_used = udp_vu_sock_recv(c, from_s, v6, &dlen);
 		if (iov_used <= 0)
-- 
@@ -281,30 +281,28 @@ void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
 {
 	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
 	const struct flowside *toside = flowside_at_sidx(tosidx);
+	bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
 	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
 	int from_s = uflow->s[ref.flowside.sidei];
 	struct vu_dev *vdev = c->vdev;
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
+	uint8_t topif = pif_at_sidx(tosidx);
 	int i;
 
-	for (i = 0; i < UDP_MAX_FRAMES; i++) {
-		uint8_t topif = pif_at_sidx(tosidx);
-		ssize_t dlen;
-		int iov_used;
-		bool v6;
-
-		ASSERT(uflow);
+	ASSERT(uflow);
 
-		if (topif != PIF_TAP) {
-			uint8_t frompif = pif_at_sidx(ref.flowside);
+	if (topif != PIF_TAP) {
+		uint8_t frompif = pif_at_sidx(ref.flowside);
 
-			flow_err(uflow,
-				 "No support for forwarding UDP from %s to %s",
-				 pif_name(frompif), pif_name(topif));
-			continue;
-		}
+		flow_err(uflow,
+			 "No support for forwarding UDP from %s to %s",
+			 pif_name(frompif), pif_name(topif));
+		return;
+	}
 
-		v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
+	for (i = 0; i < UDP_MAX_FRAMES; i++) {
+		ssize_t dlen;
+		int iov_used;
 
 		iov_used = udp_vu_sock_recv(c, from_s, v6, &dlen);
 		if (iov_used <= 0)
-- 
2.49.0

[PATCH v2 4/7] udp: Share more logic between vu and non-vu reply socket paths

[David Gibson]

Share some additional miscellaneous logic between the vhost-user and "buf"
paths for data on udp reply sockets.  The biggest piece is error handling
of cases where we can't forward between the two pifs of the flow.  We also
make common some more simple logic locating the correct flow and its
parameters.

This adds some lines of code due to extra comment lines, but nonetheless
reduces logic duplication.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 udp.c    | 41 ++++++++++++++++++++++++++---------------
 udp_vu.c | 26 +++++++++++---------------
 udp_vu.h |  3 ++-
 3 files changed, 39 insertions(+), 31 deletions(-)

diff --git a/udp.c b/udp.c
index 55021ac4..4258812e 100644
--- a/udp.c
+++ b/udp.c
@@ -754,24 +754,25 @@ void udp_listen_sock_handler(const struct ctx *c,
 /**
  * udp_buf_reply_sock_data() - Handle new data from flow specific socket
  * @c:		Execution context
- * @ref:	epoll reference
+ * @s:		Socket to read data from
+ * @tosidx:	Flow & side to forward data from @s to
  * @now:	Current timestamp
  *
+ * Return: true on success, false if can't forward from socket to flow's pif
+ *
  * #syscalls recvmmsg
  */
-static void udp_buf_reply_sock_data(const struct ctx *c, union epoll_ref ref,
+static bool udp_buf_reply_sock_data(const struct ctx *c,
+				    int s, flow_sidx_t tosidx,
 				    const struct timespec *now)
 {
-	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
 	const struct flowside *toside = flowside_at_sidx(tosidx);
-	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
+	struct udp_flow *uflow = udp_at_sidx(tosidx);
 	uint8_t topif = pif_at_sidx(tosidx);
-	int n, i, from_s;
-
-	from_s = uflow->s[ref.flowside.sidei];
+	int n, i;
 
-	if ((n = udp_sock_recv(c, from_s, udp_mh_recv)) <= 0)
-		return;
+	if ((n = udp_sock_recv(c, s, udp_mh_recv)) <= 0)
+		return true;
 
 	flow_trace(uflow, "Received %d datagrams on reply socket", n);
 	uflow->ts = now->tv_sec;
@@ -790,11 +791,10 @@ static void udp_buf_reply_sock_data(const struct ctx *c, union epoll_ref ref,
 	} else if (topif == PIF_TAP) {
 		tap_send_frames(c, &udp_l2_iov[0][0], UDP_NUM_IOVS, n);
 	} else {
-		uint8_t frompif = pif_at_sidx(ref.flowside);
-
-		flow_err(uflow, "No support for forwarding UDP from %s to %s",
-			 pif_name(frompif), pif_name(topif));
+		return false;
 	}
+
+	return true;
 }
 
 /**
@@ -821,10 +821,21 @@ void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 	}
 
 	if (events & EPOLLIN) {
+		flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
+		int s = ref.fd;
+		bool ret;
+
 		if (c->mode == MODE_VU)
-			udp_vu_reply_sock_data(c, ref, now);
+			ret = udp_vu_reply_sock_data(c, s, tosidx, now);
 		else
-			udp_buf_reply_sock_data(c, ref, now);
+			ret = udp_buf_reply_sock_data(c, s, tosidx, now);
+
+		if (!ret) {
+			flow_err(uflow,
+				 "No support for forwarding UDP from %s to %s",
+				 pif_name(pif_at_sidx(ref.flowside)),
+				 pif_name(pif_at_sidx(tosidx)));
+		}
 	}
 }
 
diff --git a/udp_vu.c b/udp_vu.c
index 6e1823a9..06bdeae6 100644
--- a/udp_vu.c
+++ b/udp_vu.c
@@ -273,38 +273,32 @@ void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
 /**
  * udp_vu_reply_sock_data() - Handle new data from flow specific socket
  * @c:		Execution context
- * @ref:	epoll reference
+ * @s:		Socket to read data from
+ * @tosidx:	Flow & side to forward data from @s to
  * @now:	Current timestamp
+ *
+ * Return: true on success, false if can't forward from socket to flow's pif
  */
-void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
+bool udp_vu_reply_sock_data(const struct ctx *c, int s, flow_sidx_t tosidx,
 			    const struct timespec *now)
 {
-	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
 	const struct flowside *toside = flowside_at_sidx(tosidx);
 	bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
-	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
-	int from_s = uflow->s[ref.flowside.sidei];
+	struct udp_flow *uflow = udp_at_sidx(tosidx);
 	struct vu_dev *vdev = c->vdev;
 	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
-	uint8_t topif = pif_at_sidx(tosidx);
 	int i;
 
 	ASSERT(uflow);
 
-	if (topif != PIF_TAP) {
-		uint8_t frompif = pif_at_sidx(ref.flowside);
-
-		flow_err(uflow,
-			 "No support for forwarding UDP from %s to %s",
-			 pif_name(frompif), pif_name(topif));
-		return;
-	}
+	if (pif_at_sidx(tosidx) != PIF_TAP)
+		return false;
 
 	for (i = 0; i < UDP_MAX_FRAMES; i++) {
 		ssize_t dlen;
 		int iov_used;
 
-		iov_used = udp_vu_sock_recv(c, from_s, v6, &dlen);
+		iov_used = udp_vu_sock_recv(c, s, v6, &dlen);
 		if (iov_used <= 0)
 			break;
 		flow_trace(uflow, "Received 1 datagram on reply socket");
@@ -318,4 +312,6 @@ void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
 		}
 		vu_flush(vdev, vq, elem, iov_used);
 	}
+
+	return true;
 }
diff --git a/udp_vu.h b/udp_vu.h
index 4f2262d0..2299b51f 100644
--- a/udp_vu.h
+++ b/udp_vu.h
@@ -8,6 +8,7 @@
 
 void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
 			     const struct timespec *now);
-void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
+bool udp_vu_reply_sock_data(const struct ctx *c, int s, flow_sidx_t tosidx,
 			    const struct timespec *now);
+
 #endif /* UDP_VU_H */
-- 
@@ -8,6 +8,7 @@
 
 void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
 			     const struct timespec *now);
-void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
+bool udp_vu_reply_sock_data(const struct ctx *c, int s, flow_sidx_t tosidx,
 			    const struct timespec *now);
+
 #endif /* UDP_VU_H */
-- 
2.49.0

[PATCH v2 5/7] udp: Better handling of failure to forward from reply socket

[David Gibson]

In udp_reply_sock_handler() if we're unable to forward the datagrams we
just print an error.  Generally this means we have an unsupported pair of
pifs in the flow table, though, and that hasn't change.  So, next time we
get a matching packet we'll just get the same failure.  In vhost-user mode
we don't even dequeue the incoming packets which triggered this so we're
likely to get the same failure immediately.

Instead, close the flow, in the same we we do for an unrecoverable error.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 udp.c | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/udp.c b/udp.c
index 4258812e..b9d98555 100644
--- a/udp.c
+++ b/udp.c
@@ -814,9 +814,7 @@ void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 	if (events & EPOLLERR) {
 		if (udp_sock_errs(c, ref) < 0) {
 			flow_err(uflow, "Unrecoverable error on reply socket");
-			flow_err_details(uflow);
-			udp_flow_close(c, uflow);
-			return;
+			goto fail;
 		}
 	}
 
@@ -831,12 +829,15 @@ void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 			ret = udp_buf_reply_sock_data(c, s, tosidx, now);
 
 		if (!ret) {
-			flow_err(uflow,
-				 "No support for forwarding UDP from %s to %s",
-				 pif_name(pif_at_sidx(ref.flowside)),
-				 pif_name(pif_at_sidx(tosidx)));
+			flow_err(uflow, "Unable to forward UDP");
+			goto fail;
 		}
 	}
+	return;
+
+fail:
+	flow_err_details(uflow);
+	udp_flow_close(c, uflow);
 }
 
 /**
-- 
@@ -814,9 +814,7 @@ void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 	if (events & EPOLLERR) {
 		if (udp_sock_errs(c, ref) < 0) {
 			flow_err(uflow, "Unrecoverable error on reply socket");
-			flow_err_details(uflow);
-			udp_flow_close(c, uflow);
-			return;
+			goto fail;
 		}
 	}
 
@@ -831,12 +829,15 @@ void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
 			ret = udp_buf_reply_sock_data(c, s, tosidx, now);
 
 		if (!ret) {
-			flow_err(uflow,
-				 "No support for forwarding UDP from %s to %s",
-				 pif_name(pif_at_sidx(ref.flowside)),
-				 pif_name(pif_at_sidx(tosidx)));
+			flow_err(uflow, "Unable to forward UDP");
+			goto fail;
 		}
 	}
+	return;
+
+fail:
+	flow_err_details(uflow);
+	udp_flow_close(c, uflow);
 }
 
 /**
-- 
2.49.0

[PATCH v2 6/7] udp: Always hash socket facing flowsides

[David Gibson]

For UDP packets from the tap interface (like TCP) we use a hash table to
look up which flow they belong to.  Unlike TCP, we sometimes also create a
hash table entry for the socket side of UDP flows.  We need that when we
receive a UDP packet from a "listening" socket which isn't specific to a
single flow.

At present we only do this for the initiating side of flows, which re-use
the listening socket.  For the target side we use a connected "reply"
socket specific to the single flow.

We have in mind changes that maye introduce some edge cases were we could
receive UDP packets on a non flow specific socket more often.  To allow for
those changes - and slightly simplifying things in the meantime - always
put both sides of a UDP flow - tap or socket - in the hash table.  It's
not that costly, and means we always have the option of falling back to a
hash lookup.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 udp_flow.c | 41 ++++++++++++++++++++---------------------
 1 file changed, 20 insertions(+), 21 deletions(-)

diff --git a/udp_flow.c b/udp_flow.c
index c6b8630a..7e809244 100644
--- a/udp_flow.c
+++ b/udp_flow.c
@@ -41,24 +41,22 @@ struct udp_flow *udp_at_sidx(flow_sidx_t sidx)
  */
 void udp_flow_close(const struct ctx *c, struct udp_flow *uflow)
 {
+	unsigned sidei;
+
 	if (uflow->closed)
 		return; /* Nothing to do */
 
-	if (uflow->s[INISIDE] >= 0) {
-		/* The listening socket needs to stay in epoll */
-		close(uflow->s[INISIDE]);
-		uflow->s[INISIDE] = -1;
-	}
-
-	if (uflow->s[TGTSIDE] >= 0) {
-		/* But the flow specific one needs to be removed */
-		epoll_del(c, uflow->s[TGTSIDE]);
-		close(uflow->s[TGTSIDE]);
-		uflow->s[TGTSIDE] = -1;
+	flow_foreach_sidei(sidei) {
+		flow_hash_remove(c, FLOW_SIDX(uflow, sidei));
+		if (uflow->s[sidei] >= 0) {
+			/* The listening socket needs to stay in epoll, but the
+			 * flow specific one needs to be removed */
+			if (sidei == TGTSIDE)
+				epoll_del(c, uflow->s[sidei]);
+			close(uflow->s[sidei]);
+			uflow->s[sidei] = -1;
+		}
 	}
-	flow_hash_remove(c, FLOW_SIDX(uflow, INISIDE));
-	if (!pif_is_socket(uflow->f.pif[TGTSIDE]))
-		flow_hash_remove(c, FLOW_SIDX(uflow, TGTSIDE));
 
 	uflow->closed = true;
 }
@@ -77,6 +75,7 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow,
 {
 	struct udp_flow *uflow = NULL;
 	const struct flowside *tgt;
+	unsigned sidei;
 	uint8_t tgtpif;
 
 	if (!(tgt = flow_target(c, flow, IPPROTO_UDP)))
@@ -143,14 +142,14 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow,
 		}
 	}
 
-	flow_hash_insert(c, FLOW_SIDX(uflow, INISIDE));
-
-	/* If the target side is a socket, it will be a reply socket that knows
-	 * its own flowside.  But if it's tap, then we need to look it up by
-	 * hash.
+	/* Tap sides always need to be looked up by hash.  Socket sides don't
+	 * always, but sometimes do (receiving packets on a socket not specific
+	 * to one flow).  Unconditionally hash both sides so all our bases are
+	 * covered
 	 */
-	if (!pif_is_socket(tgtpif))
-		flow_hash_insert(c, FLOW_SIDX(uflow, TGTSIDE));
+	flow_foreach_sidei(sidei)
+		flow_hash_insert(c, FLOW_SIDX(uflow, sidei));
+
 	FLOW_ACTIVATE(uflow);
 
 	return FLOW_SIDX(uflow, TGTSIDE);
-- 
@@ -41,24 +41,22 @@ struct udp_flow *udp_at_sidx(flow_sidx_t sidx)
  */
 void udp_flow_close(const struct ctx *c, struct udp_flow *uflow)
 {
+	unsigned sidei;
+
 	if (uflow->closed)
 		return; /* Nothing to do */
 
-	if (uflow->s[INISIDE] >= 0) {
-		/* The listening socket needs to stay in epoll */
-		close(uflow->s[INISIDE]);
-		uflow->s[INISIDE] = -1;
-	}
-
-	if (uflow->s[TGTSIDE] >= 0) {
-		/* But the flow specific one needs to be removed */
-		epoll_del(c, uflow->s[TGTSIDE]);
-		close(uflow->s[TGTSIDE]);
-		uflow->s[TGTSIDE] = -1;
+	flow_foreach_sidei(sidei) {
+		flow_hash_remove(c, FLOW_SIDX(uflow, sidei));
+		if (uflow->s[sidei] >= 0) {
+			/* The listening socket needs to stay in epoll, but the
+			 * flow specific one needs to be removed */
+			if (sidei == TGTSIDE)
+				epoll_del(c, uflow->s[sidei]);
+			close(uflow->s[sidei]);
+			uflow->s[sidei] = -1;
+		}
 	}
-	flow_hash_remove(c, FLOW_SIDX(uflow, INISIDE));
-	if (!pif_is_socket(uflow->f.pif[TGTSIDE]))
-		flow_hash_remove(c, FLOW_SIDX(uflow, TGTSIDE));
 
 	uflow->closed = true;
 }
@@ -77,6 +75,7 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow,
 {
 	struct udp_flow *uflow = NULL;
 	const struct flowside *tgt;
+	unsigned sidei;
 	uint8_t tgtpif;
 
 	if (!(tgt = flow_target(c, flow, IPPROTO_UDP)))
@@ -143,14 +142,14 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow,
 		}
 	}
 
-	flow_hash_insert(c, FLOW_SIDX(uflow, INISIDE));
-
-	/* If the target side is a socket, it will be a reply socket that knows
-	 * its own flowside.  But if it's tap, then we need to look it up by
-	 * hash.
+	/* Tap sides always need to be looked up by hash.  Socket sides don't
+	 * always, but sometimes do (receiving packets on a socket not specific
+	 * to one flow).  Unconditionally hash both sides so all our bases are
+	 * covered
 	 */
-	if (!pif_is_socket(tgtpif))
-		flow_hash_insert(c, FLOW_SIDX(uflow, TGTSIDE));
+	flow_foreach_sidei(sidei)
+		flow_hash_insert(c, FLOW_SIDX(uflow, sidei));
+
 	FLOW_ACTIVATE(uflow);
 
 	return FLOW_SIDX(uflow, TGTSIDE);
-- 
2.49.0

[PATCH v2 7/7] udp: Add helper function for creating connected UDP socket

[David Gibson]

Currently udp_flow_new() open codes creating and connecting a socket to use
for reply messages.  We have in mind some more places to use this logic,
plus it just makes for a rather large function.  Split this handling out
into a new udp_flow_sock() function.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 udp_flow.c | 104 +++++++++++++++++++++++++++++------------------------
 1 file changed, 58 insertions(+), 46 deletions(-)

diff --git a/udp_flow.c b/udp_flow.c
index 7e809244..bf4b8965 100644
--- a/udp_flow.c
+++ b/udp_flow.c
@@ -61,6 +61,61 @@ void udp_flow_close(const struct ctx *c, struct udp_flow *uflow)
 	uflow->closed = true;
 }
 
+/**
+ * udp_flow_sock() - Create, bind and connect a flow specific UDP socket
+ * @c:		Execution context
+ * @uflow:	UDP flow to open socket for
+ * @sidei:	Side of @uflow to open socket for
+ *
+ * Return: fd of new socket on success, -ve error code on failure
+ */
+static int udp_flow_sock(const struct ctx *c,
+			 const struct udp_flow *uflow, unsigned sidei)
+{
+	const struct flowside *side = &uflow->f.side[sidei];
+	struct mmsghdr discard[UIO_MAXIOV] = { 0 };
+	uint8_t pif = uflow->f.pif[sidei];
+	union {
+		flow_sidx_t sidx;
+		uint32_t data;
+	} fref = { .sidx = FLOW_SIDX(uflow, sidei) };
+	int rc, s;
+
+	s = flowside_sock_l4(c, EPOLL_TYPE_UDP_REPLY, pif, side, fref.data);
+	if (s < 0) {
+		flow_dbg_perror(uflow, "Couldn't open flow specific socket");
+		return s;
+	}
+
+	if (flowside_connect(c, s, pif, side) < 0) {
+		rc = -errno;
+		flow_dbg_perror(uflow, "Couldn't connect flow socket");
+		return rc;
+	}
+
+	/* It's possible, if unlikely, that we could receive some unrelated
+	 * packets in between the bind() and connect() of this socket.  For now
+	 * we just discard these.
+	 *
+	 * FIXME: Redirect these to an appropriate handler
+	 */
+	rc = recvmmsg(s, discard, ARRAY_SIZE(discard), MSG_DONTWAIT, NULL);
+	if (rc >= ARRAY_SIZE(discard)) {
+		flow_dbg(uflow, "Too many (%d) spurious reply datagrams", rc);
+		return -E2BIG;
+	}
+
+	if (rc > 0) {
+		flow_trace(uflow, "Discarded %d spurious reply datagrams", rc);
+	} else if (errno != EAGAIN) {
+		rc = -errno;
+		flow_perror(uflow, "Unexpected error discarding datagrams");
+		return rc;
+	}
+
+	return s;
+}
+
 /**
  * udp_flow_new() - Common setup for a new UDP flow
  * @c:		Execution context
@@ -74,13 +129,10 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow,
 				int s_ini, const struct timespec *now)
 {
 	struct udp_flow *uflow = NULL;
-	const struct flowside *tgt;
 	unsigned sidei;
-	uint8_t tgtpif;
 
-	if (!(tgt = flow_target(c, flow, IPPROTO_UDP)))
+	if (!flow_target(c, flow, IPPROTO_UDP))
 		goto cancel;
-	tgtpif = flow->f.pif[TGTSIDE];
 
 	uflow = FLOW_SET_TYPE(flow, FLOW_UDP, udp);
 	uflow->ts = now->tv_sec;
@@ -98,49 +150,9 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow,
 		}
 	}
 
-	if (pif_is_socket(tgtpif)) {
-		struct mmsghdr discard[UIO_MAXIOV] = { 0 };
-		union {
-			flow_sidx_t sidx;
-			uint32_t data;
-		} fref = {
-			.sidx = FLOW_SIDX(flow, TGTSIDE),
-		};
-		int rc;
-
-		uflow->s[TGTSIDE] = flowside_sock_l4(c, EPOLL_TYPE_UDP_REPLY,
-						     tgtpif, tgt, fref.data);
-		if (uflow->s[TGTSIDE] < 0) {
-			flow_dbg_perror(uflow,
-					"Couldn't open socket for spliced flow");
+	if (pif_is_socket(flow->f.pif[TGTSIDE]))
+		if ((uflow->s[TGTSIDE] = udp_flow_sock(c, uflow, TGTSIDE)) < 0)
 			goto cancel;
-		}
-
-		if (flowside_connect(c, uflow->s[TGTSIDE], tgtpif, tgt) < 0) {
-			flow_dbg_perror(uflow, "Couldn't connect flow socket");
-			goto cancel;
-		}
-
-		/* It's possible, if unlikely, that we could receive some
-		 * unrelated packets in between the bind() and connect() of this
-		 * socket.  For now we just discard these.  We could consider
-		 * trying to redirect these to an appropriate handler, if we
-		 * need to.
-		 */
-		rc = recvmmsg(uflow->s[TGTSIDE], discard, ARRAY_SIZE(discard),
-			      MSG_DONTWAIT, NULL);
-		if (rc >= ARRAY_SIZE(discard)) {
-			flow_dbg(uflow,
-				 "Too many (%d) spurious reply datagrams", rc);
-			goto cancel;
-		} else if (rc > 0) {
-			flow_trace(uflow,
-				   "Discarded %d spurious reply datagrams", rc);
-		} else if (errno != EAGAIN) {
-			flow_perror(uflow,
-				    "Unexpected error discarding datagrams");
-		}
-	}
 
 	/* Tap sides always need to be looked up by hash.  Socket sides don't
 	 * always, but sometimes do (receiving packets on a socket not specific
-- 
@@ -61,6 +61,61 @@ void udp_flow_close(const struct ctx *c, struct udp_flow *uflow)
 	uflow->closed = true;
 }
 
+/**
+ * udp_flow_sock() - Create, bind and connect a flow specific UDP socket
+ * @c:		Execution context
+ * @uflow:	UDP flow to open socket for
+ * @sidei:	Side of @uflow to open socket for
+ *
+ * Return: fd of new socket on success, -ve error code on failure
+ */
+static int udp_flow_sock(const struct ctx *c,
+			 const struct udp_flow *uflow, unsigned sidei)
+{
+	const struct flowside *side = &uflow->f.side[sidei];
+	struct mmsghdr discard[UIO_MAXIOV] = { 0 };
+	uint8_t pif = uflow->f.pif[sidei];
+	union {
+		flow_sidx_t sidx;
+		uint32_t data;
+	} fref = { .sidx = FLOW_SIDX(uflow, sidei) };
+	int rc, s;
+
+	s = flowside_sock_l4(c, EPOLL_TYPE_UDP_REPLY, pif, side, fref.data);
+	if (s < 0) {
+		flow_dbg_perror(uflow, "Couldn't open flow specific socket");
+		return s;
+	}
+
+	if (flowside_connect(c, s, pif, side) < 0) {
+		rc = -errno;
+		flow_dbg_perror(uflow, "Couldn't connect flow socket");
+		return rc;
+	}
+
+	/* It's possible, if unlikely, that we could receive some unrelated
+	 * packets in between the bind() and connect() of this socket.  For now
+	 * we just discard these.
+	 *
+	 * FIXME: Redirect these to an appropriate handler
+	 */
+	rc = recvmmsg(s, discard, ARRAY_SIZE(discard), MSG_DONTWAIT, NULL);
+	if (rc >= ARRAY_SIZE(discard)) {
+		flow_dbg(uflow, "Too many (%d) spurious reply datagrams", rc);
+		return -E2BIG;
+	}
+
+	if (rc > 0) {
+		flow_trace(uflow, "Discarded %d spurious reply datagrams", rc);
+	} else if (errno != EAGAIN) {
+		rc = -errno;
+		flow_perror(uflow, "Unexpected error discarding datagrams");
+		return rc;
+	}
+
+	return s;
+}
+
 /**
  * udp_flow_new() - Common setup for a new UDP flow
  * @c:		Execution context
@@ -74,13 +129,10 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow,
 				int s_ini, const struct timespec *now)
 {
 	struct udp_flow *uflow = NULL;
-	const struct flowside *tgt;
 	unsigned sidei;
-	uint8_t tgtpif;
 
-	if (!(tgt = flow_target(c, flow, IPPROTO_UDP)))
+	if (!flow_target(c, flow, IPPROTO_UDP))
 		goto cancel;
-	tgtpif = flow->f.pif[TGTSIDE];
 
 	uflow = FLOW_SET_TYPE(flow, FLOW_UDP, udp);
 	uflow->ts = now->tv_sec;
@@ -98,49 +150,9 @@ static flow_sidx_t udp_flow_new(const struct ctx *c, union flow *flow,
 		}
 	}
 
-	if (pif_is_socket(tgtpif)) {
-		struct mmsghdr discard[UIO_MAXIOV] = { 0 };
-		union {
-			flow_sidx_t sidx;
-			uint32_t data;
-		} fref = {
-			.sidx = FLOW_SIDX(flow, TGTSIDE),
-		};
-		int rc;
-
-		uflow->s[TGTSIDE] = flowside_sock_l4(c, EPOLL_TYPE_UDP_REPLY,
-						     tgtpif, tgt, fref.data);
-		if (uflow->s[TGTSIDE] < 0) {
-			flow_dbg_perror(uflow,
-					"Couldn't open socket for spliced flow");
+	if (pif_is_socket(flow->f.pif[TGTSIDE]))
+		if ((uflow->s[TGTSIDE] = udp_flow_sock(c, uflow, TGTSIDE)) < 0)
 			goto cancel;
-		}
-
-		if (flowside_connect(c, uflow->s[TGTSIDE], tgtpif, tgt) < 0) {
-			flow_dbg_perror(uflow, "Couldn't connect flow socket");
-			goto cancel;
-		}
-
-		/* It's possible, if unlikely, that we could receive some
-		 * unrelated packets in between the bind() and connect() of this
-		 * socket.  For now we just discard these.  We could consider
-		 * trying to redirect these to an appropriate handler, if we
-		 * need to.
-		 */
-		rc = recvmmsg(uflow->s[TGTSIDE], discard, ARRAY_SIZE(discard),
-			      MSG_DONTWAIT, NULL);
-		if (rc >= ARRAY_SIZE(discard)) {
-			flow_dbg(uflow,
-				 "Too many (%d) spurious reply datagrams", rc);
-			goto cancel;
-		} else if (rc > 0) {
-			flow_trace(uflow,
-				   "Discarded %d spurious reply datagrams", rc);
-		} else if (errno != EAGAIN) {
-			flow_perror(uflow,
-				    "Unexpected error discarding datagrams");
-		}
-	}
 
 	/* Tap sides always need to be looked up by hash.  Socket sides don't
 	 * always, but sometimes do (receiving packets on a socket not specific
-- 
2.49.0

Re: [PATCH v2 0/7] UDP flow socket preliminaries

[Stefano Brivio]

On Wed, 26 Mar 2025 14:44:00 +1100
David Gibson <david@gibson.dropbear.id.au> wrote:

> As discussed on our recent call, I'm working towards using connected
> sockets on both sides of UDP flows.  This series makes some
> preliminary reworks that simplify things and make that easier.
> 
> v2:
>  * Added patches 5..7, other patches unchanged.
> 
> David Gibson (7):
>   udp: Common invocation of udp_sock_errs() for vhost-user and "buf"
>     paths
>   udp: Simplify checking of epoll event bits
>   udp_vu: Factor things out of udp_vu_reply_sock_data() loop
>   udp: Share more logic between vu and non-vu reply socket paths
>   udp: Better handling of failure to forward from reply socket
>   udp: Always hash socket facing flowsides
>   udp: Add helper function for creating connected UDP socket

Applied.

-- 
Stefano

Re: [PATCH v2 4/7] udp: Share more logic between vu and non-vu reply socket paths

[Stefano Brivio]

On Wed, 26 Mar 2025 14:44:04 +1100
David Gibson <david@gibson.dropbear.id.au> wrote:

> Share some additional miscellaneous logic between the vhost-user and "buf"
> paths for data on udp reply sockets.  The biggest piece is error handling
> of cases where we can't forward between the two pifs of the flow.  We also
> make common some more simple logic locating the correct flow and its
> parameters.
> 
> This adds some lines of code due to extra comment lines, but nonetheless
> reduces logic duplication.
> 
> Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
> ---
>  udp.c    | 41 ++++++++++++++++++++++++++---------------
>  udp_vu.c | 26 +++++++++++---------------
>  udp_vu.h |  3 ++-
>  3 files changed, 39 insertions(+), 31 deletions(-)
> 
> diff --git a/udp.c b/udp.c
> index 55021ac4..4258812e 100644
> --- a/udp.c
> +++ b/udp.c
> @@ -754,24 +754,25 @@ void udp_listen_sock_handler(const struct ctx *c,
>  /**
>   * udp_buf_reply_sock_data() - Handle new data from flow specific socket
>   * @c:		Execution context
> - * @ref:	epoll reference
> + * @s:		Socket to read data from
> + * @tosidx:	Flow & side to forward data from @s to
>   * @now:	Current timestamp
>   *
> + * Return: true on success, false if can't forward from socket to flow's pif

"on success"...

> + *
>   * #syscalls recvmmsg
>   */
> -static void udp_buf_reply_sock_data(const struct ctx *c, union epoll_ref ref,
> +static bool udp_buf_reply_sock_data(const struct ctx *c,
> +				    int s, flow_sidx_t tosidx,
>  				    const struct timespec *now)
>  {
> -	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
>  	const struct flowside *toside = flowside_at_sidx(tosidx);
> -	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
> +	struct udp_flow *uflow = udp_at_sidx(tosidx);
>  	uint8_t topif = pif_at_sidx(tosidx);
> -	int n, i, from_s;
> -
> -	from_s = uflow->s[ref.flowside.sidei];
> +	int n, i;
>  
> -	if ((n = udp_sock_recv(c, from_s, udp_mh_recv)) <= 0)
> -		return;
> +	if ((n = udp_sock_recv(c, s, udp_mh_recv)) <= 0)
> +		return true;

is not exactly accurate: 0 means either no datagrams (should never
happen) or error on recvmmsg(). But I think it's clear enough, and
you'll have follow-ups anyway, so I went ahead and applied this.

Maybe:

 * Return: false if we can't forward from socket to flow's pif, true otherwise

? Actually, looking at the caller:

>  
>  	flow_trace(uflow, "Received %d datagrams on reply socket", n);
>  	uflow->ts = now->tv_sec;
> @@ -790,11 +791,10 @@ static void udp_buf_reply_sock_data(const struct ctx *c, union epoll_ref ref,
>  	} else if (topif == PIF_TAP) {
>  		tap_send_frames(c, &udp_l2_iov[0][0], UDP_NUM_IOVS, n);
>  	} else {
> -		uint8_t frompif = pif_at_sidx(ref.flowside);
> -
> -		flow_err(uflow, "No support for forwarding UDP from %s to %s",
> -			 pif_name(frompif), pif_name(topif));
> +		return false;
>  	}
> +
> +	return true;
>  }
>  
>  /**
> @@ -821,10 +821,21 @@ void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
>  	}
>  
>  	if (events & EPOLLIN) {
> +		flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
> +		int s = ref.fd;
> +		bool ret;
> +
>  		if (c->mode == MODE_VU)
> -			udp_vu_reply_sock_data(c, ref, now);
> +			ret = udp_vu_reply_sock_data(c, s, tosidx, now);
>  		else
> -			udp_buf_reply_sock_data(c, ref, now);
> +			ret = udp_buf_reply_sock_data(c, s, tosidx, now);
> +
> +		if (!ret) {

the opposite would probably be less surprising, if (ret) flow_err(...).

> +			flow_err(uflow,
> +				 "No support for forwarding UDP from %s to %s",
> +				 pif_name(pif_at_sidx(ref.flowside)),
> +				 pif_name(pif_at_sidx(tosidx)));
> +		}
>  	}
>  }
>  
> diff --git a/udp_vu.c b/udp_vu.c
> index 6e1823a9..06bdeae6 100644
> --- a/udp_vu.c
> +++ b/udp_vu.c
> @@ -273,38 +273,32 @@ void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
>  /**
>   * udp_vu_reply_sock_data() - Handle new data from flow specific socket
>   * @c:		Execution context
> - * @ref:	epoll reference
> + * @s:		Socket to read data from
> + * @tosidx:	Flow & side to forward data from @s to
>   * @now:	Current timestamp
> + *
> + * Return: true on success, false if can't forward from socket to flow's pif
>   */
> -void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
> +bool udp_vu_reply_sock_data(const struct ctx *c, int s, flow_sidx_t tosidx,
>  			    const struct timespec *now)
>  {
> -	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
>  	const struct flowside *toside = flowside_at_sidx(tosidx);
>  	bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
> -	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
> -	int from_s = uflow->s[ref.flowside.sidei];
> +	struct udp_flow *uflow = udp_at_sidx(tosidx);
>  	struct vu_dev *vdev = c->vdev;
>  	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
> -	uint8_t topif = pif_at_sidx(tosidx);
>  	int i;
>  
>  	ASSERT(uflow);
>  
> -	if (topif != PIF_TAP) {
> -		uint8_t frompif = pif_at_sidx(ref.flowside);
> -
> -		flow_err(uflow,
> -			 "No support for forwarding UDP from %s to %s",
> -			 pif_name(frompif), pif_name(topif));
> -		return;
> -	}
> +	if (pif_at_sidx(tosidx) != PIF_TAP)
> +		return false;
>  
>  	for (i = 0; i < UDP_MAX_FRAMES; i++) {
>  		ssize_t dlen;
>  		int iov_used;
>  
> -		iov_used = udp_vu_sock_recv(c, from_s, v6, &dlen);
> +		iov_used = udp_vu_sock_recv(c, s, v6, &dlen);
>  		if (iov_used <= 0)
>  			break;
>  		flow_trace(uflow, "Received 1 datagram on reply socket");
> @@ -318,4 +312,6 @@ void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
>  		}
>  		vu_flush(vdev, vq, elem, iov_used);
>  	}
> +
> +	return true;
>  }
> diff --git a/udp_vu.h b/udp_vu.h
> index 4f2262d0..2299b51f 100644
> --- a/udp_vu.h
> +++ b/udp_vu.h
> @@ -8,6 +8,7 @@
>  
>  void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
>  			     const struct timespec *now);
> -void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
> +bool udp_vu_reply_sock_data(const struct ctx *c, int s, flow_sidx_t tosidx,
>  			    const struct timespec *now);
> +
>  #endif /* UDP_VU_H */

-- 
Stefano

Re: [PATCH v2 4/7] udp: Share more logic between vu and non-vu reply socket paths

[David Gibson]

[-- Attachment #1: Type: text/plain, Size: 7078 bytes --]

On Wed, Mar 26, 2025 at 11:14:33PM +0100, Stefano Brivio wrote:
> On Wed, 26 Mar 2025 14:44:04 +1100
> David Gibson <david@gibson.dropbear.id.au> wrote:
> 
> > Share some additional miscellaneous logic between the vhost-user and "buf"
> > paths for data on udp reply sockets.  The biggest piece is error handling
> > of cases where we can't forward between the two pifs of the flow.  We also
> > make common some more simple logic locating the correct flow and its
> > parameters.
> > 
> > This adds some lines of code due to extra comment lines, but nonetheless
> > reduces logic duplication.
> > 
> > Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
> > ---
> >  udp.c    | 41 ++++++++++++++++++++++++++---------------
> >  udp_vu.c | 26 +++++++++++---------------
> >  udp_vu.h |  3 ++-
> >  3 files changed, 39 insertions(+), 31 deletions(-)
> > 
> > diff --git a/udp.c b/udp.c
> > index 55021ac4..4258812e 100644
> > --- a/udp.c
> > +++ b/udp.c
> > @@ -754,24 +754,25 @@ void udp_listen_sock_handler(const struct ctx *c,
> >  /**
> >   * udp_buf_reply_sock_data() - Handle new data from flow specific socket
> >   * @c:		Execution context
> > - * @ref:	epoll reference
> > + * @s:		Socket to read data from
> > + * @tosidx:	Flow & side to forward data from @s to
> >   * @now:	Current timestamp
> >   *
> > + * Return: true on success, false if can't forward from socket to flow's pif
> 
> "on success"...
> 
> > + *
> >   * #syscalls recvmmsg
> >   */
> > -static void udp_buf_reply_sock_data(const struct ctx *c, union epoll_ref ref,
> > +static bool udp_buf_reply_sock_data(const struct ctx *c,
> > +				    int s, flow_sidx_t tosidx,
> >  				    const struct timespec *now)
> >  {
> > -	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
> >  	const struct flowside *toside = flowside_at_sidx(tosidx);
> > -	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
> > +	struct udp_flow *uflow = udp_at_sidx(tosidx);
> >  	uint8_t topif = pif_at_sidx(tosidx);
> > -	int n, i, from_s;
> > -
> > -	from_s = uflow->s[ref.flowside.sidei];
> > +	int n, i;
> >  
> > -	if ((n = udp_sock_recv(c, from_s, udp_mh_recv)) <= 0)
> > -		return;
> > +	if ((n = udp_sock_recv(c, s, udp_mh_recv)) <= 0)
> > +		return true;
> 
> is not exactly accurate: 0 means either no datagrams (should never
> happen) or error on recvmmsg(). But I think it's clear enough, and
> you'll have follow-ups anyway, so I went ahead and applied this.
> 
> Maybe:
> 
>  * Return: false if we can't forward from socket to flow's pif, true otherwise
> 
> ? Actually, looking at the caller:

Yeah.  While writing subsequent patches, I think I spotted a better
way to structure this.  I'll put some fixups in the next series.

> >  	flow_trace(uflow, "Received %d datagrams on reply socket", n);
> >  	uflow->ts = now->tv_sec;
> > @@ -790,11 +791,10 @@ static void udp_buf_reply_sock_data(const struct ctx *c, union epoll_ref ref,
> >  	} else if (topif == PIF_TAP) {
> >  		tap_send_frames(c, &udp_l2_iov[0][0], UDP_NUM_IOVS, n);
> >  	} else {
> > -		uint8_t frompif = pif_at_sidx(ref.flowside);
> > -
> > -		flow_err(uflow, "No support for forwarding UDP from %s to %s",
> > -			 pif_name(frompif), pif_name(topif));
> > +		return false;
> >  	}
> > +
> > +	return true;
> >  }
> >  
> >  /**
> > @@ -821,10 +821,21 @@ void udp_reply_sock_handler(const struct ctx *c, union epoll_ref ref,
> >  	}
> >  
> >  	if (events & EPOLLIN) {
> > +		flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
> > +		int s = ref.fd;
> > +		bool ret;
> > +
> >  		if (c->mode == MODE_VU)
> > -			udp_vu_reply_sock_data(c, ref, now);
> > +			ret = udp_vu_reply_sock_data(c, s, tosidx, now);
> >  		else
> > -			udp_buf_reply_sock_data(c, ref, now);
> > +			ret = udp_buf_reply_sock_data(c, s, tosidx, now);
> > +
> > +		if (!ret) {
> 
> the opposite would probably be less surprising, if (ret) flow_err(...).
> 
> > +			flow_err(uflow,
> > +				 "No support for forwarding UDP from %s to %s",
> > +				 pif_name(pif_at_sidx(ref.flowside)),
> > +				 pif_name(pif_at_sidx(tosidx)));
> > +		}
> >  	}
> >  }
> >  
> > diff --git a/udp_vu.c b/udp_vu.c
> > index 6e1823a9..06bdeae6 100644
> > --- a/udp_vu.c
> > +++ b/udp_vu.c
> > @@ -273,38 +273,32 @@ void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
> >  /**
> >   * udp_vu_reply_sock_data() - Handle new data from flow specific socket
> >   * @c:		Execution context
> > - * @ref:	epoll reference
> > + * @s:		Socket to read data from
> > + * @tosidx:	Flow & side to forward data from @s to
> >   * @now:	Current timestamp
> > + *
> > + * Return: true on success, false if can't forward from socket to flow's pif
> >   */
> > -void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
> > +bool udp_vu_reply_sock_data(const struct ctx *c, int s, flow_sidx_t tosidx,
> >  			    const struct timespec *now)
> >  {
> > -	flow_sidx_t tosidx = flow_sidx_opposite(ref.flowside);
> >  	const struct flowside *toside = flowside_at_sidx(tosidx);
> >  	bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
> > -	struct udp_flow *uflow = udp_at_sidx(ref.flowside);
> > -	int from_s = uflow->s[ref.flowside.sidei];
> > +	struct udp_flow *uflow = udp_at_sidx(tosidx);
> >  	struct vu_dev *vdev = c->vdev;
> >  	struct vu_virtq *vq = &vdev->vq[VHOST_USER_RX_QUEUE];
> > -	uint8_t topif = pif_at_sidx(tosidx);
> >  	int i;
> >  
> >  	ASSERT(uflow);
> >  
> > -	if (topif != PIF_TAP) {
> > -		uint8_t frompif = pif_at_sidx(ref.flowside);
> > -
> > -		flow_err(uflow,
> > -			 "No support for forwarding UDP from %s to %s",
> > -			 pif_name(frompif), pif_name(topif));
> > -		return;
> > -	}
> > +	if (pif_at_sidx(tosidx) != PIF_TAP)
> > +		return false;
> >  
> >  	for (i = 0; i < UDP_MAX_FRAMES; i++) {
> >  		ssize_t dlen;
> >  		int iov_used;
> >  
> > -		iov_used = udp_vu_sock_recv(c, from_s, v6, &dlen);
> > +		iov_used = udp_vu_sock_recv(c, s, v6, &dlen);
> >  		if (iov_used <= 0)
> >  			break;
> >  		flow_trace(uflow, "Received 1 datagram on reply socket");
> > @@ -318,4 +312,6 @@ void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
> >  		}
> >  		vu_flush(vdev, vq, elem, iov_used);
> >  	}
> > +
> > +	return true;
> >  }
> > diff --git a/udp_vu.h b/udp_vu.h
> > index 4f2262d0..2299b51f 100644
> > --- a/udp_vu.h
> > +++ b/udp_vu.h
> > @@ -8,6 +8,7 @@
> >  
> >  void udp_vu_listen_sock_data(const struct ctx *c, union epoll_ref ref,
> >  			     const struct timespec *now);
> > -void udp_vu_reply_sock_data(const struct ctx *c, union epoll_ref ref,
> > +bool udp_vu_reply_sock_data(const struct ctx *c, int s, flow_sidx_t tosidx,
> >  			    const struct timespec *now);
> > +
> >  #endif /* UDP_VU_H */
> 

-- 
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

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]