Re: [PATCH V2 net-next 2/3] rds: deliver zerocopy completion notification with data
From: Willem de Bruijn <willemdebruijn.kernel@gmail.com>
Date: 2018-02-25 15:56:49
On Fri, Feb 23, 2018 at 5:08 PM, Sowmini Varadhan [off-list ref] 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 <redacted>
---quoted hunk ↗ jump to hunk
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));
quoted hunk ↗ jump to hunk
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);
quoted hunk ↗ jump to hunk
@@ -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.
quoted hunk ↗ jump to hunk
+ 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