rds: Add per peer RDS socket send buffer

Currently, an RDS socket's send buffer is shared by all the peers this
socket communicates with.  When one or more peers are slow to receive,
their unacknowledged data can consume a large portion of the socket's
buffer.  This will affect the communication with other peers of the
socket.

To resolve this issue, an RDS socket's send buffer is now per peer.
This works like opening multiple TCP sockets to different peers, each
peer has its own send buffer (in each socket).  In RDS, each peer has
its own buffer space in the socket.  With this per peer send buffer,
when one or more peers are slow, their data will not interfere with
data sent to other peers.

A sysctl parameter, sock_max_peers, is added to limit the number of
peers a socket can communicate with.  The default is 8192.  Its range
of valid value is 128 to 65536.

Orabug: 29492596

Tested-by: RDS CI <[email protected]>
Tested-by: Rose Wang <[email protected]>
Tested-by: Rosa Lopez <[email protected]>
Tested-by: Shih-Yu Huang <[email protected]>
Signed-off-by: Ka-Cheong Poon <[email protected]>
Reviewed-by: Venkat Venkatsubra <[email protected]>
Acked-by: Santosh Shilimkar <[email protected]>

Author: kcp-git

net/rds/af_rds.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006, 2018 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2006, 2019 Oracle and/or its affiliates. All rights reserved.
  *
  * This software is available to you under a choice of one of two
  * licenses.  You may choose to be licensed under the terms of the GNU
@@ -80,6 +80,17 @@ static unsigned long rds_sock_count;
 static LIST_HEAD(rds_sock_list);
 DECLARE_WAIT_QUEUE_HEAD(rds_poll_waitq);
 
+/* kmem cache slab for struct rds_buf_info */
+static struct kmem_cache *rds_rs_buf_info_slab;
+
+/* Helper function to be passed to rhashtable_free_and_destroy() to free a
+ * struct rs_buf_info.
+ */
+static void rds_buf_info_free(void *rsbi, void *arg __attribute__((unused)))
+{
+	kmem_cache_free(rds_rs_buf_info_slab, rsbi);
+}
+
 /*
  * This is called as the final descriptor referencing this socket is closed.
  * We have to unbind the socket so that another socket can be bound to the
@@ -112,6 +123,9 @@ static int rds_release(struct socket *sock)
 	rds_rdma_drop_keys(rs);
 	rds_notify_queue_get(rs, NULL);
 
+	rhashtable_free_and_destroy(&rs->rs_buf_info_tbl, rds_buf_info_free,
+				    NULL);
+
 	spin_lock_bh(&rds_sock_lock);
 	list_del_init(&rs->rs_item);
 	rds_sock_count--;
@@ -272,10 +286,18 @@ static unsigned int rds_poll(struct file *file, struct socket *sock,
 	if (!list_empty(&rs->rs_recv_queue)
 	 || !list_empty(&rs->rs_notify_queue))
 		mask |= (POLLIN | POLLRDNORM);
-	if (rs->rs_snd_bytes < rds_sk_sndbuf(rs))
-		mask |= (POLLOUT | POLLWRNORM);
 	read_unlock_irqrestore(&rs->rs_recv_lock, flags);
 
+	/* Use the number of destination this socket has to estimate the
+	 * send buffer size.  When there is no peer yet, return the default
+	 * send buffer size.
+	 */
+	spin_lock_irqsave(&rs->rs_snd_lock, flags);
+	if (rs->rs_snd_bytes < max_t(u32, rs->rs_buf_info_dest_cnt, 1) *
+	    rds_sk_sndbuf(rs))
+		mask |= (POLLOUT | POLLWRNORM);
+	spin_unlock_irqrestore(&rs->rs_snd_lock, flags);
+
 	/* clear state any time we wake a seen-congested socket */
 	if (mask)
 		rs->rs_seen_congestion = 0;
@@ -712,6 +734,77 @@ static int rds_getsockopt(struct socket *sock, int level, int optname,
 
 }
 
+/* Check if there is a rs_buf_info associated with the given address.  If not,
+ * add one to the rds_sock.  The found or added rs_buf_info is returned.  If
+ * there is no rs_buf_info found and a new rs_buf_info cannot be allocated,
+ * NULL is returned and ret is set to the error.  Once an address' rs_buf_info
+ * is added, it will not be removed until the rs_sock is closed.
+ */
+struct rs_buf_info *rds_add_buf_info(struct rds_sock *rs, struct in6_addr *addr,
+				     int *ret, gfp_t gfp)
+{
+	struct rs_buf_info *info, *tmp_info;
+	unsigned long flags;
+
+	/* Normal path, peer is expected to be found most of the time. */
+	info = rhashtable_lookup_fast(&rs->rs_buf_info_tbl, addr,
+				      rs_buf_info_params);
+	if (info) {
+		*ret = 0;
+		return info;
+	}
+
+	/* Allocate the buffer outside of lock first. */
+	tmp_info = kmem_cache_alloc(rds_rs_buf_info_slab, gfp);
+	if (!tmp_info) {
+		*ret = -ENOMEM;
+		return NULL;
+	}
+
+	spin_lock_irqsave(&rs->rs_snd_lock, flags);
+
+	/* Cannot add more peer. */
+	if (rs->rs_buf_info_dest_cnt + 1 > rds_sock_max_peers) {
+		spin_unlock_irqrestore(&rs->rs_snd_lock, flags);
+		kmem_cache_free(rds_rs_buf_info_slab, tmp_info);
+		*ret = -ENFILE;
+		return NULL;
+	}
+
+	tmp_info->rsbi_key = *addr;
+	tmp_info->rsbi_snd_bytes = 0;
+	*ret = rhashtable_insert_fast(&rs->rs_buf_info_tbl,
+				      &tmp_info->rsbi_link, rs_buf_info_params);
+	if (!*ret) {
+		rs->rs_buf_info_dest_cnt++;
+		spin_unlock_irqrestore(&rs->rs_snd_lock, flags);
+		return tmp_info;
+	} else if (*ret != -EEXIST) {
+		spin_unlock_irqrestore(&rs->rs_snd_lock, flags);
+		kmem_cache_free(rds_rs_buf_info_slab, tmp_info);
+		/* Very unlikely to happen... */
+		pr_err("%s: cannot add rs_buf_info for %pI6c: %d\n", __func__,
+		       addr, *ret);
+		return NULL;
+	}
+
+	/* Another thread beats us in adding the rs_buf_info.... */
+	info = rhashtable_lookup_fast(&rs->rs_buf_info_tbl, addr,
+				      rs_buf_info_params);
+	spin_unlock_irqrestore(&rs->rs_snd_lock, flags);
+	kmem_cache_free(rds_rs_buf_info_slab, tmp_info);
+
+	if (info) {
+		*ret = 0;
+		return info;
+	}
+
+	/* Should not happen... */
+	pr_err("%s: cannot find rs_buf_info for %pI6c\n", __func__, addr);
+	*ret = -EINVAL;
+	return NULL;
+}
+
 static int rds_connect(struct socket *sock, struct sockaddr *uaddr,
 		       int addr_len, int flags)
 {
@@ -800,6 +893,12 @@ static int rds_connect(struct socket *sock, struct sockaddr *uaddr,
 		break;
 	}
 
+	if (!ret &&
+	    !rds_add_buf_info(rs, &rs->rs_conn_addr, &ret, GFP_KERNEL)) {
+		/* Need to clear the connected info in case of error. */
+		rs->rs_conn_addr = in6addr_any;
+		rs->rs_conn_port = 0;
+	}
 	release_sock(sk);
 	return ret;
 }
@@ -842,6 +941,7 @@ static void rds_sock_destruct(struct sock *sk)
 static int __rds_create(struct socket *sock, struct sock *sk, int protocol)
 {
 	struct rds_sock *rs;
+	int ret;
 
 	sock_init_data(sock, sk);
 	sock->ops		= &rds_proto_ops;
@@ -863,6 +963,11 @@ static int __rds_create(struct socket *sock, struct sock *sk, int protocol)
 	rs->rs_netfilter_enabled = 0;
 	rs->rs_rx_traces = 0;
 
+	spin_lock_init(&rs->rs_snd_lock);
+	ret = rhashtable_init(&rs->rs_buf_info_tbl, &rs_buf_info_params);
+	if (ret)
+		return ret;
+
 	if (!ipv6_addr_any(&rs->rs_bound_addr)) {
 		printk(KERN_CRIT "bound addr %pI6c at create\n",
 		       &rs->rs_bound_addr);
@@ -879,6 +984,7 @@ static int __rds_create(struct socket *sock, struct sock *sk, int protocol)
 static int rds_create(struct net *net, struct socket *sock, int protocol, int kern)
 {
 	struct sock *sk;
+	int ret;
 
 	if (sock->type != SOCK_SEQPACKET ||
 	    (protocol && IPPROTO_OKA != protocol))
@@ -888,7 +994,10 @@ static int rds_create(struct net *net, struct socket *sock, int protocol, int ke
 	if (!sk)
 		return -ENOMEM;
 
-	return __rds_create(sock, sk, protocol);
+	ret = __rds_create(sock, sk, protocol);
+	if (ret)
+		sk_free(sk);
+	return ret;
 }
 
 void debug_sock_hold(struct sock *sk)
@@ -1194,6 +1303,7 @@ static void __exit rds_exit(void)
 	rds_info_deregister_func(RDS6_INFO_SOCKETS, rds6_sock_info);
 	rds_info_deregister_func(RDS6_INFO_RECV_MESSAGES, rds6_sock_inc_info);
 #endif
+	kmem_cache_destroy(rds_rs_buf_info_slab);
 }
 
 module_exit(rds_exit);
@@ -1204,6 +1314,14 @@ static int __init rds_init(void)
 {
 	int ret;
 
+	rds_rs_buf_info_slab = kmem_cache_create("rds_rs_buf_info",
+						 sizeof(struct rs_buf_info),
+						 0, SLAB_HWCACHE_ALIGN, NULL);
+	if (!rds_rs_buf_info_slab) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
 	net_get_random_once(&rds_gen_num, sizeof(rds_gen_num));
 
 	rds_bind_lock_init();

net/rds/rds.h

@@ -12,6 +12,7 @@
 #include <uapi/linux/rds.h>
 #include <linux/in6.h>
 #include <linux/sizes.h>
+#include <linux/rhashtable.h>
 
 #include "info.h"
 
@@ -730,6 +731,13 @@ struct rds_transport {
 				struct rdma_cm_event *event);
 };
 
+/* Used to store per peer socket buffer info. */
+struct rs_buf_info {
+	struct in6_addr		rsbi_key;
+	struct rhash_head	rsbi_link;
+	u32			rsbi_snd_bytes;
+};
+
 struct rds_sock {
 	struct sock		rs_sk;
 
@@ -762,28 +770,36 @@ struct rds_sock {
 	/* seen congestion (ENOBUFS) when sending? */
 	int			rs_seen_congestion;
 
-	/* rs_lock protects all these adjacent members before the newline */
-	spinlock_t		rs_lock;
-	struct list_head	rs_send_queue;
-	u32			rs_snd_bytes;
-	int			rs_rcv_bytes;
-	struct list_head	rs_notify_queue;	/* currently used for failed RDMAs */
-
-	/* Congestion wake_up. If rs_cong_monitor is set, we use cong_mask
+	/* rs_lock protects all these adjacent members before the newline.
+	 *
+	 * Congestion wake_up. If rs_cong_monitor is set, we use cong_mask
 	 * to decide whether the application should be woken up.
 	 * If not set, we use rs_cong_track to find out whether a cong map
 	 * update arrived.
 	 */
+	spinlock_t		rs_lock;
 	uint64_t		rs_cong_mask;
 	uint64_t		rs_cong_notify;
 	struct list_head	rs_cong_list;
 	unsigned long		rs_cong_track;
+	/* currently used for failed RDMAs */
+	struct list_head	rs_notify_queue;
+
+	/* rs_snd_lock protects all these adjacent members before the
+	 * newline
+	 */
+	spinlock_t		rs_snd_lock;
+	struct list_head	rs_send_queue;
+	u32			rs_snd_bytes; /* Total bytes to all peers */
+	u32			rs_buf_info_dest_cnt;
+	struct rhashtable	rs_buf_info_tbl;
 
 	/*
 	 * rs_recv_lock protects the receive queue, and is
 	 * used to serialize with rds_release.
 	 */
 	rwlock_t		rs_recv_lock;
+	int			rs_rcv_bytes;
 	struct list_head	rs_recv_queue;
 
 	/* just for stats reporting */
@@ -872,6 +888,25 @@ struct rds_statistics {
 };
 
 /* af_rds.c */
+#define	RDS_SOCK_BUF_INFO_HTBL_SIZE	512
+static const struct rhashtable_params rs_buf_info_params = {
+	.nelem_hint = RDS_SOCK_BUF_INFO_HTBL_SIZE,
+	.key_len = sizeof(struct in6_addr),
+	.key_offset = offsetof(struct rs_buf_info, rsbi_key),
+	.head_offset = offsetof(struct rs_buf_info, rsbi_link),
+};
+
+/* Maximum number of peers a socket can communicate with */
+extern unsigned int rds_sock_max_peers;
+
+struct rs_buf_info *rds_add_buf_info(struct rds_sock *rs, struct in6_addr *addr,
+				     int *ret, gfp_t gfp);
+static inline struct rs_buf_info *rds_get_buf_info(struct rds_sock *rs,
+						   struct in6_addr *addr)
+{
+	return rhashtable_lookup_fast(&rs->rs_buf_info_tbl, addr,
+				      rs_buf_info_params);
+}
 char *rds_str_array(char **array, size_t elements, size_t index);
 void rds_sock_addref(struct rds_sock *rs);
 void rds_sock_put(struct rds_sock *rs);

net/rds/recv.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006, 2018 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2006, 2019 Oracle and/or its affiliates. All rights reserved.
  *
  * This software is available to you under a choice of one of two
  * licenses.  You may choose to be licensed under the terms of the GNU
@@ -913,9 +913,9 @@ static int rds_notify_cong(struct rds_sock *rs, struct msghdr *msghdr)
 	if (err)
 		return err;
 
-	spin_lock_irqsave(&rs->rs_lock, flags);
+	spin_lock_irqsave(&rs->rs_snd_lock, flags);
 	rs->rs_cong_notify &= ~notify;
-	spin_unlock_irqrestore(&rs->rs_lock, flags);
+	spin_unlock_irqrestore(&rs->rs_snd_lock, flags);
 
 	return 0;
 }