[PATCH V2 net-next 0/3] RDS: optimized notification for zerocopy completion

[PATCH V2 net-next 0/3] RDS: optimized notification for zerocopy completion

[Sowmini Varadhan]

RDS applications use predominantly request-response, transacation 
based IPC, so that ingress and egress traffic are well-balanced,
and it is possible/desirable to reduce system-call overhead by
piggybacking the notifications for zerocopy completion response
with data. 

Moreover, it has been pointed out that socket functions block
if sk_err is non-zero, thus if the RDS code does not plan/need
to use sk_error_queue path for completion notification, it
is preferable to remove the sk_errror_queue related paths in
RDS.

Both of these goals are implemented in this series.

Sowmini Varadhan (3):
  selftests/net: revert the zerocopy Rx path for PF_RDS
  rds: deliver zerocopy completion notification with data
  selftests/net: reap zerocopy completions passed up as ancillary data.

 include/uapi/linux/errqueue.h              |    2 -
 include/uapi/linux/rds.h                   |    7 ++
 net/rds/af_rds.c                           |    7 ++-
 net/rds/message.c                          |   35 ++++------
 net/rds/rds.h                              |    2 +
 net/rds/recv.c                             |   34 +++++++++-
 tools/testing/selftests/net/msg_zerocopy.c |  109 +++++++++++-----------------
 7 files changed, 103 insertions(+), 93 deletions(-)

[PATCH V2 net-next 1/3] selftests/net: revert the zerocopy Rx path for PF_RDS

[Sowmini Varadhan]

In preparation for optimized reception of zerocopy completion,
revert the Rx side changes introduced by Commit dfb8434b0a94
("selftests/net: add zerocopy support for PF_RDS test case")

Signed-off-by: Sowmini Varadhan <sowmini.varadhan@oracle.com>
---
v2: prepare to remove sk_error_queue based path; remove recvmsg() as well,
    PF_RDS can also use recv() for the usage pattern in msg_zerocopy

 tools/testing/selftests/net/msg_zerocopy.c |   67 ----------------------------
 1 files changed, 0 insertions(+), 67 deletions(-)

diff --git a/tools/testing/selftests/net/msg_zerocopy.c b/tools/testing/selftests/net/msg_zerocopy.c
index 5cc2a53..eff9cf2 100644
--- a/tools/testing/selftests/net/msg_zerocopy.c
+++ b/tools/testing/selftests/net/msg_zerocopy.c
@@ -344,26 +344,6 @@ static int do_setup_tx(int domain, int type, int protocol)
 	return fd;
 }
 
-static int do_process_zerocopy_cookies(struct sock_extended_err *serr,
-				       uint32_t *ckbuf, size_t nbytes)
-{
-	int ncookies, i;
-
-	if (serr->ee_errno != 0)
-		error(1, 0, "serr: wrong error code: %u", serr->ee_errno);
-	ncookies = serr->ee_data;
-	if (ncookies > SO_EE_ORIGIN_MAX_ZCOOKIES)
-		error(1, 0, "Returned %d cookies, max expected %d\n",
-		      ncookies, SO_EE_ORIGIN_MAX_ZCOOKIES);
-	if (nbytes != ncookies * sizeof(uint32_t))
-		error(1, 0, "Expected %d cookies, got %ld\n",
-		      ncookies, nbytes/sizeof(uint32_t));
-	for (i = 0; i < ncookies; i++)
-		if (cfg_verbose >= 2)
-			fprintf(stderr, "%d\n", ckbuf[i]);
-	return ncookies;
-}
-
 static bool do_recv_completion(int fd)
 {
 	struct sock_extended_err *serr;
@@ -372,17 +352,10 @@ static bool do_recv_completion(int fd)
 	uint32_t hi, lo, range;
 	int ret, zerocopy;
 	char control[100];
-	uint32_t ckbuf[SO_EE_ORIGIN_MAX_ZCOOKIES];
-	struct iovec iov;
 
 	msg.msg_control = control;
 	msg.msg_controllen = sizeof(control);
 
-	iov.iov_base = ckbuf;
-	iov.iov_len = (SO_EE_ORIGIN_MAX_ZCOOKIES * sizeof(ckbuf[0]));
-	msg.msg_iov = &iov;
-	msg.msg_iovlen = 1;
-
 	ret = recvmsg(fd, &msg, MSG_ERRQUEUE);
 	if (ret == -1 && errno == EAGAIN)
 		return false;
@@ -402,10 +375,6 @@ static bool do_recv_completion(int fd)
 
 	serr = (void *) CMSG_DATA(cm);
 
-	if (serr->ee_origin == SO_EE_ORIGIN_ZCOOKIE) {
-		completions += do_process_zerocopy_cookies(serr, ckbuf, ret);
-		return true;
-	}
 	if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY)
 		error(1, 0, "serr: wrong origin: %u", serr->ee_origin);
 	if (serr->ee_errno != 0)
@@ -631,40 +600,6 @@ static void do_flush_datagram(int fd, int type)
 	bytes += cfg_payload_len;
 }
 
-
-static void do_recvmsg(int fd)
-{
-	int ret, off = 0;
-	char *buf;
-	struct iovec iov;
-	struct msghdr msg;
-	struct sockaddr_storage din;
-
-	buf = calloc(cfg_payload_len, sizeof(char));
-	iov.iov_base = buf;
-	iov.iov_len = cfg_payload_len;
-
-	memset(&msg, 0, sizeof(msg));
-	msg.msg_name = &din;
-	msg.msg_namelen = sizeof(din);
-	msg.msg_iov = &iov;
-	msg.msg_iovlen = 1;
-
-	ret = recvmsg(fd, &msg, MSG_TRUNC);
-
-	if (ret == -1)
-		error(1, errno, "recv");
-	if (ret != cfg_payload_len)
-		error(1, 0, "recv: ret=%u != %u", ret, cfg_payload_len);
-
-	if (memcmp(buf + off, payload, ret))
-		error(1, 0, "recv: data mismatch");
-
-	free(buf);
-	packets++;
-	bytes += cfg_payload_len;
-}
-
 static void do_rx(int domain, int type, int protocol)
 {
 	uint64_t tstop;
@@ -676,8 +611,6 @@ static void do_rx(int domain, int type, int protocol)
 	do {
 		if (type == SOCK_STREAM)
 			do_flush_tcp(fd);
-		else if (domain == PF_RDS)
-			do_recvmsg(fd);
 		else
 			do_flush_datagram(fd, type);
 
-- 
1.7.1

[PATCH V2 net-next 2/3] rds: deliver zerocopy completion notification with data

[Sowmini Varadhan]

This commit is an optimization of the commit 01883eda72bd
("rds: support for zcopy completion notification") for PF_RDS sockets.

RDS applications are predominantly request-response transactions, so
it is more efficient to reduce the number of system calls and have
zerocopy completion notification delivered as ancillary data on the
POLLIN channel.

Cookies are passed up as ancillary data (at level SOL_RDS) in a
struct rds_zcopy_cookies when the returned value of recvmsg() is
greater than, or equal to, 0. A max of RDS_MAX_ZCOOKIES may be passed
with each message.

This commit removes support for zerocopy completion notification on
MSG_ERRQUEUE for PF_RDS sockets.

Signed-off-by: Sowmini Varadhan <sowmini.varadhan@oracle.com>
---
v2: remove sk_error_queue path; lot of cautionary checks rds_recvmsg_zcookie()
    and callers to make sure we dont remove cookies from the queue and then
    fail to pass it up to caller

 include/uapi/linux/errqueue.h |    2 --
 include/uapi/linux/rds.h      |    7 +++++++
 net/rds/af_rds.c              |    7 +++++--
 net/rds/message.c             |   35 +++++++++++++----------------------
 net/rds/rds.h                 |    2 ++
 net/rds/recv.c                |   34 +++++++++++++++++++++++++++++++++-
 6 files changed, 60 insertions(+), 27 deletions(-)

diff --git a/include/uapi/linux/errqueue.h b/include/uapi/linux/errqueue.h
index 28812ed..dc64cfa 100644
--- a/include/uapi/linux/errqueue.h
+++ b/include/uapi/linux/errqueue.h
@@ -20,13 +20,11 @@ struct sock_extended_err {
 #define SO_EE_ORIGIN_ICMP6	3
 #define SO_EE_ORIGIN_TXSTATUS	4
 #define SO_EE_ORIGIN_ZEROCOPY	5
-#define SO_EE_ORIGIN_ZCOOKIE	6
 #define SO_EE_ORIGIN_TIMESTAMPING SO_EE_ORIGIN_TXSTATUS
 
 #define SO_EE_OFFENDER(ee)	((struct sockaddr*)((ee)+1))
 
 #define SO_EE_CODE_ZEROCOPY_COPIED	1
-#define	SO_EE_ORIGIN_MAX_ZCOOKIES	8
 
 /**
  *	struct scm_timestamping - timestamps exposed through cmsg
diff --git a/include/uapi/linux/rds.h b/include/uapi/linux/rds.h
index 12e3bca..a66b213 100644
--- a/include/uapi/linux/rds.h
+++ b/include/uapi/linux/rds.h
@@ -104,6 +104,7 @@
 #define RDS_CMSG_MASKED_ATOMIC_CSWP	9
 #define RDS_CMSG_RXPATH_LATENCY		11
 #define	RDS_CMSG_ZCOPY_COOKIE		12
+#define	RDS_CMSG_ZCOPY_COMPLETION	13
 
 #define RDS_INFO_FIRST			10000
 #define RDS_INFO_COUNTERS		10000
@@ -317,6 +318,12 @@ struct rds_rdma_notify {
 #define RDS_RDMA_DROPPED	3
 #define RDS_RDMA_OTHER_ERROR	4
 
+#define	RDS_MAX_ZCOOKIES	8
+struct rds_zcopy_cookies {
+	__u32 num;
+	__u32 cookies[RDS_MAX_ZCOOKIES];
+};
+
 /*
  * Common set of flags for all RDMA related structs
  */
diff --git a/net/rds/af_rds.c b/net/rds/af_rds.c
index a937f18..f712610 100644
--- a/net/rds/af_rds.c
+++ b/net/rds/af_rds.c
@@ -77,6 +77,7 @@ static int rds_release(struct socket *sock)
 	rds_send_drop_to(rs, NULL);
 	rds_rdma_drop_keys(rs);
 	rds_notify_queue_get(rs, NULL);
+	__skb_queue_purge(&rs->rs_zcookie_queue);
 
 	spin_lock_bh(&rds_sock_lock);
 	list_del_init(&rs->rs_item);
@@ -144,7 +145,7 @@ static int rds_getname(struct socket *sock, struct sockaddr *uaddr,
  *  -	to signal that a previously congested destination may have become
  *	uncongested
  *  -	A notification has been queued to the socket (this can be a congestion
- *	update, or a RDMA completion).
+ *	update, or a RDMA completion, or a MSG_ZEROCOPY completion).
  *
  * EPOLLOUT is asserted if there is room on the send queue. This does not mean
  * however, that the next sendmsg() call will succeed. If the application tries
@@ -178,7 +179,8 @@ static __poll_t rds_poll(struct file *file, struct socket *sock,
 		spin_unlock(&rs->rs_lock);
 	}
 	if (!list_empty(&rs->rs_recv_queue) ||
-	    !list_empty(&rs->rs_notify_queue))
+	    !list_empty(&rs->rs_notify_queue) ||
+	    !skb_queue_empty(&rs->rs_zcookie_queue))
 		mask |= (EPOLLIN | EPOLLRDNORM);
 	if (rs->rs_snd_bytes < rds_sk_sndbuf(rs))
 		mask |= (EPOLLOUT | EPOLLWRNORM);
@@ -513,6 +515,7 @@ static int __rds_create(struct socket *sock, struct sock *sk, int protocol)
 	INIT_LIST_HEAD(&rs->rs_recv_queue);
 	INIT_LIST_HEAD(&rs->rs_notify_queue);
 	INIT_LIST_HEAD(&rs->rs_cong_list);
+	skb_queue_head_init(&rs->rs_zcookie_queue);
 	spin_lock_init(&rs->rs_rdma_lock);
 	rs->rs_rdma_keys = RB_ROOT;
 	rs->rs_rx_traces = 0;
diff --git a/net/rds/message.c b/net/rds/message.c
index 6518345..2e8bdaf 100644
--- a/net/rds/message.c
+++ b/net/rds/message.c
@@ -58,32 +58,26 @@ void rds_message_addref(struct rds_message *rm)
 
 static inline bool skb_zcookie_add(struct sk_buff *skb, u32 cookie)
 {
-	struct sock_exterr_skb *serr = SKB_EXT_ERR(skb);
-	int ncookies;
-	u32 *ptr;
+	struct rds_zcopy_cookies *ck = (struct rds_zcopy_cookies *)skb->cb;
+	int ncookies = ck->num;
 
-	if (serr->ee.ee_origin != SO_EE_ORIGIN_ZCOOKIE)
+	if (ncookies == RDS_MAX_ZCOOKIES)
 		return false;
-	ncookies = serr->ee.ee_data;
-	if (ncookies == SO_EE_ORIGIN_MAX_ZCOOKIES)
-		return false;
-	ptr = skb_put(skb, sizeof(u32));
-	*ptr = cookie;
-	serr->ee.ee_data = ++ncookies;
+	ck->cookies[ncookies] = cookie;
+	ck->num =  ++ncookies;
 	return true;
 }
 
 static void rds_rm_zerocopy_callback(struct rds_sock *rs,
 				     struct rds_znotifier *znotif)
 {
-	struct sock *sk = rds_rs_to_sk(rs);
 	struct sk_buff *skb, *tail;
-	struct sock_exterr_skb *serr;
 	unsigned long flags;
 	struct sk_buff_head *q;
 	u32 cookie = znotif->z_cookie;
+	struct rds_zcopy_cookies *ck;
 
-	q = &sk->sk_error_queue;
+	q = &rs->rs_zcookie_queue;
 	spin_lock_irqsave(&q->lock, flags);
 	tail = skb_peek_tail(q);
 
@@ -91,22 +85,19 @@ static void rds_rm_zerocopy_callback(struct rds_sock *rs,
 		spin_unlock_irqrestore(&q->lock, flags);
 		mm_unaccount_pinned_pages(&znotif->z_mmp);
 		consume_skb(rds_skb_from_znotifier(znotif));
-		sk->sk_error_report(sk);
+		/* caller should wake up POLLIN */
 		return;
 	}
 
 	skb = rds_skb_from_znotifier(znotif);
-	serr = SKB_EXT_ERR(skb);
-	memset(&serr->ee, 0, sizeof(serr->ee));
-	serr->ee.ee_errno = 0;
-	serr->ee.ee_origin = SO_EE_ORIGIN_ZCOOKIE;
-	serr->ee.ee_info = 0;
+	ck = (struct rds_zcopy_cookies *)skb->cb;
+	memset(ck, 0, sizeof(*ck));
 	WARN_ON(!skb_zcookie_add(skb, cookie));
 
 	__skb_queue_tail(q, skb);
 
 	spin_unlock_irqrestore(&q->lock, flags);
-	sk->sk_error_report(sk);
+	/* caller should wake up POLLIN */
 
 	mm_unaccount_pinned_pages(&znotif->z_mmp);
 }
@@ -129,6 +120,7 @@ static void rds_message_purge(struct rds_message *rm)
 		if (rm->data.op_mmp_znotifier) {
 			zcopy = true;
 			rds_rm_zerocopy_callback(rs, rm->data.op_mmp_znotifier);
+			rds_wake_sk_sleep(rs);
 			rm->data.op_mmp_znotifier = NULL;
 		}
 		sock_put(rds_rs_to_sk(rs));
@@ -362,8 +354,7 @@ int rds_message_copy_from_user(struct rds_message *rm, struct iov_iter *from,
 		int total_copied = 0;
 		struct sk_buff *skb;
 
-		skb = alloc_skb(SO_EE_ORIGIN_MAX_ZCOOKIES * sizeof(u32),
-				GFP_KERNEL);
+		skb = alloc_skb(0, GFP_KERNEL);
 		if (!skb)
 			return -ENOMEM;
 		rm->data.op_mmp_znotifier = RDS_ZCOPY_SKB(skb);
diff --git a/net/rds/rds.h b/net/rds/rds.h
index 31cd388..33b1635 100644
--- a/net/rds/rds.h
+++ b/net/rds/rds.h
@@ -603,6 +603,8 @@ struct rds_sock {
 	/* Socket receive path trace points*/
 	u8			rs_rx_traces;
 	u8			rs_rx_trace[RDS_MSG_RX_DGRAM_TRACE_MAX];
+
+	struct sk_buff_head	rs_zcookie_queue;
 };
 
 static inline struct rds_sock *rds_sk_to_rs(const struct sock *sk)
diff --git a/net/rds/recv.c b/net/rds/recv.c
index b080961..9f1bf55 100644
--- a/net/rds/recv.c
+++ b/net/rds/recv.c
@@ -577,6 +577,35 @@ static int rds_cmsg_recv(struct rds_incoming *inc, struct msghdr *msg,
 	return ret;
 }
 
+static int rds_recvmsg_zcookie(struct rds_sock *rs, struct msghdr *msg)
+{
+	struct sk_buff *skb;
+	struct sk_buff_head *q = &rs->rs_zcookie_queue;
+	struct rds_zcopy_cookies *done;
+	int ret;
+
+	if (!sock_flag(rds_rs_to_sk(rs), SOCK_ZEROCOPY) || !skb_peek(q))
+		return 0;
+
+	if (!msg->msg_control ||
+	    msg->msg_controllen < CMSG_SPACE(sizeof(*done)))
+		return 0;
+
+	skb = skb_dequeue(q);
+	if (!skb)
+		return 0;
+	done = (struct rds_zcopy_cookies *)skb->cb;
+	ret = done->num;
+	if (put_cmsg(msg, SOL_RDS, RDS_CMSG_ZCOPY_COMPLETION, sizeof(*done),
+		     done)) {
+		skb_queue_head(q, skb);
+		ret = 0;
+	} else {
+		consume_skb(skb);
+	}
+	return ret;
+}
+
 int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
 		int msg_flags)
 {
@@ -586,6 +615,7 @@ int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
 	int ret = 0, nonblock = msg_flags & MSG_DONTWAIT;
 	DECLARE_SOCKADDR(struct sockaddr_in *, sin, msg->msg_name);
 	struct rds_incoming *inc = NULL;
+	int ncookies;
 
 	/* udp_recvmsg()->sock_recvtimeo() gets away without locking too.. */
 	timeo = sock_rcvtimeo(sk, nonblock);
@@ -611,7 +641,8 @@ int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
 
 		if (!rds_next_incoming(rs, &inc)) {
 			if (nonblock) {
-				ret = -EAGAIN;
+				ncookies = rds_recvmsg_zcookie(rs, msg);
+				ret = ncookies ? 0 : -EAGAIN;
 				break;
 			}
 
@@ -660,6 +691,7 @@ int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
 			ret = -EFAULT;
 			goto out;
 		}
+		ncookies = rds_recvmsg_zcookie(rs, msg);
 
 		rds_stats_inc(s_recv_delivered);
 
-- 
1.7.1

Re: [PATCH V2 net-next 2/3] rds: deliver zerocopy completion notification with data

[Willem de Bruijn]

On Fri, Feb 23, 2018 at 5:08 PM, Sowmini Varadhan
<sowmini.varadhan@oracle.com> wrote:
> This commit is an optimization of the commit 01883eda72bd
> ("rds: support for zcopy completion notification") for PF_RDS sockets.
>
> RDS applications are predominantly request-response transactions, so
> it is more efficient to reduce the number of system calls and have
> zerocopy completion notification delivered as ancillary data on the
> POLLIN channel.
>
> Cookies are passed up as ancillary data (at level SOL_RDS) in a
> struct rds_zcopy_cookies when the returned value of recvmsg() is
> greater than, or equal to, 0. A max of RDS_MAX_ZCOOKIES may be passed
> with each message.
>
> This commit removes support for zerocopy completion notification on
> MSG_ERRQUEUE for PF_RDS sockets.
>
> Signed-off-by: Sowmini Varadhan <sowmini.varadhan@oracle.com>
> ---

> diff --git a/net/rds/message.c b/net/rds/message.c
> index 6518345..2e8bdaf 100644
> --- a/net/rds/message.c
> +++ b/net/rds/message.c
> @@ -58,32 +58,26 @@ void rds_message_addref(struct rds_message *rm)
>
>  static inline bool skb_zcookie_add(struct sk_buff *skb, u32 cookie)
>  {
> -       struct sock_exterr_skb *serr = SKB_EXT_ERR(skb);
> -       int ncookies;
> -       u32 *ptr;
> +       struct rds_zcopy_cookies *ck = (struct rds_zcopy_cookies *)skb->cb;

please add a bounds check (not necessarily right here)

  BUILD_BUG_ON(sizeof(*ck) > sizeof(skb->cb));

>  static void rds_rm_zerocopy_callback(struct rds_sock *rs,
>                                      struct rds_znotifier *znotif)
>  {
> -       struct sock *sk = rds_rs_to_sk(rs);
>         struct sk_buff *skb, *tail;
> -       struct sock_exterr_skb *serr;
>         unsigned long flags;
>         struct sk_buff_head *q;
>         u32 cookie = znotif->z_cookie;
> +       struct rds_zcopy_cookies *ck;
>
> -       q = &sk->sk_error_queue;
> +       q = &rs->rs_zcookie_queue;
>         spin_lock_irqsave(&q->lock, flags);
>         tail = skb_peek_tail(q);
>
> @@ -91,22 +85,19 @@ static void rds_rm_zerocopy_callback(struct rds_sock *rs,
>                 spin_unlock_irqrestore(&q->lock, flags);
>                 mm_unaccount_pinned_pages(&znotif->z_mmp);
>                 consume_skb(rds_skb_from_znotifier(znotif));
> -               sk->sk_error_report(sk);
> +               /* caller should wake up POLLIN */

sk->sk_data_ready(sk);

> @@ -362,8 +354,7 @@ int rds_message_copy_from_user(struct rds_message *rm, struct iov_iter *from,
>                 int total_copied = 0;
>                 struct sk_buff *skb;
>
> -               skb = alloc_skb(SO_EE_ORIGIN_MAX_ZCOOKIES * sizeof(u32),
> -                               GFP_KERNEL);
> +               skb = alloc_skb(0, GFP_KERNEL);

Without the error queue, the struct no longer needs to be an skb,
per se. Converting to a different struct with list_head is definitely
a longer patch. But kmalloc will be cheaper than alloc_skb.
Perhaps something to try (as separate follow-on work).

> +static int rds_recvmsg_zcookie(struct rds_sock *rs, struct msghdr *msg)
> +{
> +       struct sk_buff *skb;
> +       struct sk_buff_head *q = &rs->rs_zcookie_queue;
> +       struct rds_zcopy_cookies *done;
> +       int ret;
> +
> +       if (!sock_flag(rds_rs_to_sk(rs), SOCK_ZEROCOPY) || !skb_peek(q))
> +               return 0;

Racy read?

> +
> +       if (!msg->msg_control ||

I'd move this first, so that the cookie queue need not even be probed
in the common case.

> +           msg->msg_controllen < CMSG_SPACE(sizeof(*done)))
> +               return 0;

if caller does not satisfy the contract on controllen size, can be
more explicit and return an error.

> +
> +       skb = skb_dequeue(q);
> +       if (!skb)
> +               return 0;
> +       done = (struct rds_zcopy_cookies *)skb->cb;
> +       ret = done->num;

done->num is guaranteed to be >= 1, so ret is not strictly needed.

> +       if (put_cmsg(msg, SOL_RDS, RDS_CMSG_ZCOPY_COMPLETION, sizeof(*done),
> +                    done)) {
> +               skb_queue_head(q, skb);
> +               ret = 0;
> +       } else {
> +               consume_skb(skb);
> +       }
> +       return ret;
> +}
> +
>  int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
>                 int msg_flags)
>  {
> @@ -586,6 +615,7 @@ int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
>         int ret = 0, nonblock = msg_flags & MSG_DONTWAIT;
>         DECLARE_SOCKADDR(struct sockaddr_in *, sin, msg->msg_name);
>         struct rds_incoming *inc = NULL;
> +       int ncookies;
>
>         /* udp_recvmsg()->sock_recvtimeo() gets away without locking too.. */
>         timeo = sock_rcvtimeo(sk, nonblock);
> @@ -611,7 +641,8 @@ int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
>
>                 if (!rds_next_incoming(rs, &inc)) {
>                         if (nonblock) {
> -                               ret = -EAGAIN;
> +                               ncookies = rds_recvmsg_zcookie(rs, msg);
> +                               ret = ncookies ? 0 : -EAGAIN;
>                                 break;
>                         }
>
> @@ -660,6 +691,7 @@ int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,
>                         ret = -EFAULT;
>                         goto out;
>                 }
> +               ncookies = rds_recvmsg_zcookie(rs, msg);

return value not used. can make rds_recvmsg_zcookie return a bool for
the non-blocking case.

>
>                 rds_stats_inc(s_recv_delivered);
>
> --
> 1.7.1

Re: [PATCH V2 net-next 2/3] rds: deliver zerocopy completion notification with data

[Sowmini Varadhan]

On (02/25/18 10:56), Willem de Bruijn wrote:
> > @@ -91,22 +85,19 @@ static void rds_rm_zerocopy_callback(struct rds_sock *rs,
> >                 spin_unlock_irqrestore(&q->lock, flags);
> >                 mm_unaccount_pinned_pages(&znotif->z_mmp);
> >                 consume_skb(rds_skb_from_znotifier(znotif));
> > -               sk->sk_error_report(sk);
> > +               /* caller should wake up POLLIN */
> 
> sk->sk_data_ready(sk);

yes, this was my first thought, but everything else in rds
is calling rds_wake_sk_sleep (this is even done in
rds_recv_incoming(), which actually queues up the data), 
so I chose to align with that model (and call this in the caller 
of rds_rm_zerocopy_callback()

> Without the error queue, the struct no longer needs to be an skb,
> per se. Converting to a different struct with list_head is definitely
> a longer patch. But kmalloc will be cheaper than alloc_skb.
> Perhaps something to try (as separate follow-on work).

right, I was thinking along these exact lines as well,
and was already planning a follow-up.

> > +       if (!sock_flag(rds_rs_to_sk(rs), SOCK_ZEROCOPY) || !skb_peek(q))
> > +               return 0;
> 
> Racy read?

Can you elaborate? I only put the skb_peek to quickly
bail for sockets that are not using zerocopy at all- 
if you race against something that's queuing data, and 
miss it on the peek, the next read/recv should find it.
Am I missing some race?


> 
> > +
> > +       if (!msg->msg_control ||
> 
> I'd move this first, so that the cookie queue need not even be probed
> in the common case.

you mean before the check for SOCK_ZEROCOPY?

> > +           msg->msg_controllen < CMSG_SPACE(sizeof(*done)))
> > +               return 0;
> 
> if caller does not satisfy the contract on controllen size, can be
> more explicit and return an error.

if SOCK_ZEROCOPY has been set, but the recv did not specify a cmsghdr,
you mean?

> > +               ncookies = rds_recvmsg_zcookie(rs, msg);

Will take care of the remaining comments in V3.

Re: [PATCH V2 net-next 2/3] rds: deliver zerocopy completion notification with data

[Willem de Bruijn]

On Sun, Feb 25, 2018 at 11:20 AM, Sowmini Varadhan
<sowmini.varadhan@oracle.com> wrote:
> On (02/25/18 10:56), Willem de Bruijn wrote:
>> > @@ -91,22 +85,19 @@ static void rds_rm_zerocopy_callback(struct rds_sock *rs,
>> >                 spin_unlock_irqrestore(&q->lock, flags);
>> >                 mm_unaccount_pinned_pages(&znotif->z_mmp);
>> >                 consume_skb(rds_skb_from_znotifier(znotif));
>> > -               sk->sk_error_report(sk);
>> > +               /* caller should wake up POLLIN */
>>
>> sk->sk_data_ready(sk);
>
> yes, this was my first thought, but everything else in rds
> is calling rds_wake_sk_sleep (this is even done in
> rds_recv_incoming(), which actually queues up the data),
> so I chose to align with that model (and call this in the caller
> of rds_rm_zerocopy_callback()

Ah, understood. Perhaps say "wakes" instead of "should wake".
I mistakenly read this as a todo.

>> Without the error queue, the struct no longer needs to be an skb,
>> per se. Converting to a different struct with list_head is definitely
>> a longer patch. But kmalloc will be cheaper than alloc_skb.
>> Perhaps something to try (as separate follow-on work).
>
> right, I was thinking along these exact lines as well,
> and was already planning a follow-up.
>
>> > +       if (!sock_flag(rds_rs_to_sk(rs), SOCK_ZEROCOPY) || !skb_peek(q))
>> > +               return 0;
>>
>> Racy read?
>
> Can you elaborate? I only put the skb_peek to quickly
> bail for sockets that are not using zerocopy at all-
> if you race against something that's queuing data, and
> miss it on the peek, the next read/recv should find it.
> Am I missing some race?

It''s a lockless access. But intentionally so, then. You're right, as long as
the subsequent skb_dequeue handles the case where the queue is
empty, it seems okay to optimistically probe lockless first.

>>
>> > +
>> > +       if (!msg->msg_control ||
>>
>> I'd move this first, so that the cookie queue need not even be probed
>> in the common case.
>
> you mean before the check for SOCK_ZEROCOPY?

Yes

>> > +           msg->msg_controllen < CMSG_SPACE(sizeof(*done)))
>> > +               return 0;
>>
>> if caller does not satisfy the contract on controllen size, can be
>> more explicit and return an error.
>
> if SOCK_ZEROCOPY has been set, but the recv did not specify a cmsghdr,
> you mean?

I mean if SOCK_ZEROCOPY has been set and the caller calls recvmsg
with a control buffer, but one that is too small to handle zerocopy cookie
notifications.

>> > +               ncookies = rds_recvmsg_zcookie(rs, msg);
>
> Will take care of the remaining comments in V3.

[PATCH V2 net-next 3/3] selftests/net: reap zerocopy completions passed up as ancillary data.

[Sowmini Varadhan]

PF_RDS sockets pass up cookies for zerocopy completion as ancillary
data. Update msg_zerocopy to reap this information.

Signed-off-by: Sowmini Varadhan <sowmini.varadhan@oracle.com>
---
v2: receive zerocopy completion notification as POLLIN

 tools/testing/selftests/net/msg_zerocopy.c |   60 ++++++++++++++++++++++++----
 1 files changed, 52 insertions(+), 8 deletions(-)

diff --git a/tools/testing/selftests/net/msg_zerocopy.c b/tools/testing/selftests/net/msg_zerocopy.c
index eff9cf2..8c466e8 100644
--- a/tools/testing/selftests/net/msg_zerocopy.c
+++ b/tools/testing/selftests/net/msg_zerocopy.c
@@ -344,7 +344,48 @@ static int do_setup_tx(int domain, int type, int protocol)
 	return fd;
 }
 
-static bool do_recv_completion(int fd)
+static int do_process_zerocopy_cookies(struct rds_zcopy_cookies *ck)
+{
+	int ncookies, i;
+
+	ncookies = ck->num;
+	if (ncookies > RDS_MAX_ZCOOKIES)
+		error(1, 0, "Returned %d cookies, max expected %d\n",
+		      ncookies, RDS_MAX_ZCOOKIES);
+	for (i = 0; i < ncookies; i++)
+		if (cfg_verbose >= 2)
+			fprintf(stderr, "%d\n", ck->cookies[i]);
+	return ncookies;
+}
+
+static int do_recvmsg_completion(int fd)
+{
+	struct msghdr msg;
+	char cmsgbuf[256];
+	struct cmsghdr *cmsg;
+	bool ret = false;
+	struct rds_zcopy_cookies *ck;
+
+	memset(&msg, 0, sizeof(msg));
+	msg.msg_control = cmsgbuf;
+	msg.msg_controllen = sizeof(cmsgbuf);
+
+	if (recvmsg(fd, &msg, MSG_DONTWAIT))
+		return ret;
+	for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
+		if (cmsg->cmsg_level == SOL_RDS &&
+		    cmsg->cmsg_type == RDS_CMSG_ZCOPY_COMPLETION) {
+
+			ck = (struct rds_zcopy_cookies *)CMSG_DATA(cmsg);
+			completions += do_process_zerocopy_cookies(ck);
+			ret = true;
+			break;
+		}
+	}
+	return ret;
+}
+
+static bool do_recv_completion(int fd, int domain)
 {
 	struct sock_extended_err *serr;
 	struct msghdr msg = {};
@@ -353,6 +394,9 @@ static bool do_recv_completion(int fd)
 	int ret, zerocopy;
 	char control[100];
 
+	if (domain == PF_RDS)
+		return do_recvmsg_completion(fd);
+
 	msg.msg_control = control;
 	msg.msg_controllen = sizeof(control);
 
@@ -409,20 +453,20 @@ static bool do_recv_completion(int fd)
 }
 
 /* Read all outstanding messages on the errqueue */
-static void do_recv_completions(int fd)
+static void do_recv_completions(int fd, int domain)
 {
-	while (do_recv_completion(fd)) {}
+	while (do_recv_completion(fd, domain)) {}
 }
 
 /* Wait for all remaining completions on the errqueue */
-static void do_recv_remaining_completions(int fd)
+static void do_recv_remaining_completions(int fd, int domain)
 {
 	int64_t tstop = gettimeofday_ms() + cfg_waittime_ms;
 
 	while (completions < expected_completions &&
 	       gettimeofday_ms() < tstop) {
-		if (do_poll(fd, POLLERR))
-			do_recv_completions(fd);
+		if (do_poll(fd, domain == PF_RDS ? POLLIN : POLLERR))
+			do_recv_completions(fd, domain);
 	}
 
 	if (completions < expected_completions)
@@ -503,13 +547,13 @@ static void do_tx(int domain, int type, int protocol)
 
 		while (!do_poll(fd, POLLOUT)) {
 			if (cfg_zerocopy)
-				do_recv_completions(fd);
+				do_recv_completions(fd, domain);
 		}
 
 	} while (gettimeofday_ms() < tstop);
 
 	if (cfg_zerocopy)
-		do_recv_remaining_completions(fd);
+		do_recv_remaining_completions(fd, domain);
 
 	if (close(fd))
 		error(1, errno, "close");
-- 
1.7.1

Re: [PATCH V2 net-next 3/3] selftests/net: reap zerocopy completions passed up as ancillary data.

[Willem de Bruijn]

On Fri, Feb 23, 2018 at 5:08 PM, Sowmini Varadhan
<sowmini.varadhan@oracle.com> wrote:
> PF_RDS sockets pass up cookies for zerocopy completion as ancillary
> data. Update msg_zerocopy to reap this information.
>
> Signed-off-by: Sowmini Varadhan <sowmini.varadhan@oracle.com>
> ---
> v2: receive zerocopy completion notification as POLLIN
>
>  tools/testing/selftests/net/msg_zerocopy.c |   60 ++++++++++++++++++++++++----
>  1 files changed, 52 insertions(+), 8 deletions(-)
>
> diff --git a/tools/testing/selftests/net/msg_zerocopy.c b/tools/testing/selftests/net/msg_zerocopy.c
> index eff9cf2..8c466e8 100644
> --- a/tools/testing/selftests/net/msg_zerocopy.c
> +++ b/tools/testing/selftests/net/msg_zerocopy.c
> @@ -344,7 +344,48 @@ static int do_setup_tx(int domain, int type, int protocol)
>         return fd;
>  }
>
> -static bool do_recv_completion(int fd)
> +static int do_process_zerocopy_cookies(struct rds_zcopy_cookies *ck)
> +{
> +       int ncookies, i;

Since this is example code, let's be pedantic about type: uint32_t.
Or don't use a local variable.

> +
> +       ncookies = ck->num;
> +       if (ncookies > RDS_MAX_ZCOOKIES)
> +               error(1, 0, "Returned %d cookies, max expected %d\n",
> +                     ncookies, RDS_MAX_ZCOOKIES);
> +       for (i = 0; i < ncookies; i++)
> +               if (cfg_verbose >= 2)
> +                       fprintf(stderr, "%d\n", ck->cookies[i]);
> +       return ncookies;
> +}
> +
> +static int do_recvmsg_completion(int fd)
> +{
> +       struct msghdr msg;
> +       char cmsgbuf[256];

more precise: CMSG_SPACE(sizeof(*ck));

> +       struct cmsghdr *cmsg;
> +       bool ret = false;
> +       struct rds_zcopy_cookies *ck;

then this must move above. Reverse christmas tree is preferred, anyway.

> +
> +       memset(&msg, 0, sizeof(msg));
> +       msg.msg_control = cmsgbuf;
> +       msg.msg_controllen = sizeof(cmsgbuf);
> +
> +       if (recvmsg(fd, &msg, MSG_DONTWAIT))
> +               return ret;

check msg_flags & MSG_CTRUNC

> +       for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
> +               if (cmsg->cmsg_level == SOL_RDS &&
> +                   cmsg->cmsg_type == RDS_CMSG_ZCOPY_COMPLETION) {
> +
> +                       ck = (struct rds_zcopy_cookies *)CMSG_DATA(cmsg);
> +                       completions += do_process_zerocopy_cookies(ck);
> +                       ret = true;
> +                       break;
> +               }

maybe warn on unexpected other type or level

> +       }
> +       return ret;
> +}
> +
> +static bool do_recv_completion(int fd, int domain)
>  {
>         struct sock_extended_err *serr;
>         struct msghdr msg = {};
> @@ -353,6 +394,9 @@ static bool do_recv_completion(int fd)
>         int ret, zerocopy;
>         char control[100];
>
> +       if (domain == PF_RDS)
> +               return do_recvmsg_completion(fd);
> +
>         msg.msg_control = control;
>         msg.msg_controllen = sizeof(control);
>
> @@ -409,20 +453,20 @@ static bool do_recv_completion(int fd)
>  }
>
>  /* Read all outstanding messages on the errqueue */
> -static void do_recv_completions(int fd)
> +static void do_recv_completions(int fd, int domain)
>  {
> -       while (do_recv_completion(fd)) {}
> +       while (do_recv_completion(fd, domain)) {}
>  }
>
>  /* Wait for all remaining completions on the errqueue */
> -static void do_recv_remaining_completions(int fd)
> +static void do_recv_remaining_completions(int fd, int domain)
>  {
>         int64_t tstop = gettimeofday_ms() + cfg_waittime_ms;
>
>         while (completions < expected_completions &&
>                gettimeofday_ms() < tstop) {
> -               if (do_poll(fd, POLLERR))
> -                       do_recv_completions(fd);
> +               if (do_poll(fd, domain == PF_RDS ? POLLIN : POLLERR))
> +                       do_recv_completions(fd, domain);
>         }
>
>         if (completions < expected_completions)
> @@ -503,13 +547,13 @@ static void do_tx(int domain, int type, int protocol)
>
>                 while (!do_poll(fd, POLLOUT)) {
>                         if (cfg_zerocopy)
> -                               do_recv_completions(fd);
> +                               do_recv_completions(fd, domain);
>                 }
>
>         } while (gettimeofday_ms() < tstop);
>
>         if (cfg_zerocopy)
> -               do_recv_remaining_completions(fd);
> +               do_recv_remaining_completions(fd, domain);
>
>         if (close(fd))
>                 error(1, errno, "close");
> --
> 1.7.1
>