SUNRPC: Support dynamic slot allocation for TCP connections

Allow the number of available slots to grow with the TCP window size.

Signed-off-by: Trond Myklebust <Trond.Myklebust@netapp.com>

3 files changed, 102 insertions, 18 deletions

diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index ea7b3c16cddd..be85cf04a479 100644
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -935,25 +935,66 @@ void xprt_transmit(struct rpc_task *task)
         spin_unlock_bh(&xprt->transport_lock);
 }
 
+static struct rpc_rqst *xprt_dynamic_alloc_slot(struct rpc_xprt *xprt, gfp_t gfp_flags)
+{
+        struct rpc_rqst *req = ERR_PTR(-EAGAIN);
+
+        if (!atomic_add_unless(&xprt->num_reqs, 1, xprt->max_reqs))
+                goto out;
+        req = kzalloc(sizeof(struct rpc_rqst), gfp_flags);
+        if (req != NULL)
+                goto out;
+        atomic_dec(&xprt->num_reqs);
+        req = ERR_PTR(-ENOMEM);
+out:
+        return req;
+}
+
+static bool xprt_dynamic_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
+{
+        if (atomic_add_unless(&xprt->num_reqs, -1, xprt->min_reqs)) {
+                kfree(req);
+                return true;
+        }
+        return false;
+}
+
 static void xprt_alloc_slot(struct rpc_task *task)
 {
         struct rpc_xprt *xprt = task->tk_xprt;
+        struct rpc_rqst *req;
 
-        task->tk_status = 0;
         if (!list_empty(&xprt->free)) {
-                struct rpc_rqst *req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
-                list_del_init(&req->rq_list);
-                task->tk_rqstp = req;
-                xprt_request_init(task, xprt);
-                return;
+                req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
+                list_del(&req->rq_list);
+                goto out_init_req;
+        }
+        req = xprt_dynamic_alloc_slot(xprt, GFP_NOWAIT);
+        if (!IS_ERR(req))
+                goto out_init_req;
+        switch (PTR_ERR(req)) {
+        case -ENOMEM:
+                rpc_delay(task, HZ >> 2);
+                dprintk("RPC:       dynamic allocation of request slot "
+                                "failed! Retrying\n");
+                break;
+        case -EAGAIN:
+                rpc_sleep_on(&xprt->backlog, task, NULL);
+                dprintk("RPC:       waiting for request slot\n");
         }
-        dprintk("RPC:       waiting for request slot\n");
         task->tk_status = -EAGAIN;
-        rpc_sleep_on(&xprt->backlog, task, NULL);
+        return;
+out_init_req:
+        task->tk_status = 0;
+        task->tk_rqstp = req;
+        xprt_request_init(task, xprt);
 }
 
 static void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
 {
+        if (xprt_dynamic_free_slot(xprt, req))
+                return;
+
         memset(req, 0, sizeof(*req));   /* mark unused */
 
         spin_lock(&xprt->reserve_lock);
@@ -972,7 +1013,9 @@ static void xprt_free_all_slots(struct rpc_xprt *xprt)
         }
 }
 
-struct rpc_xprt *xprt_alloc(struct net *net, int size, int num_prealloc)
+struct rpc_xprt *xprt_alloc(struct net *net, size_t size,
+                unsigned int num_prealloc,
+                unsigned int max_alloc)
 {
         struct rpc_xprt *xprt;
         struct rpc_rqst *req;
@@ -992,7 +1035,12 @@ struct rpc_xprt *xprt_alloc(struct net *net, int size, int num_prealloc)
         }
         if (i < num_prealloc)
                 goto out_free;
-        xprt->max_reqs = num_prealloc;
+        if (max_alloc > num_prealloc)
+                xprt->max_reqs = max_alloc;
+        else
+                xprt->max_reqs = num_prealloc;
+        xprt->min_reqs = num_prealloc;
+        atomic_set(&xprt->num_reqs, num_prealloc);
 
         return xprt;
 
@@ -1036,7 +1084,6 @@ void xprt_reserve(struct rpc_task *task)
         if (!xprt_lock_write(xprt, task))
                 return;
 
-        task->tk_status = -EIO;
         spin_lock(&xprt->reserve_lock);
         xprt_alloc_slot(task);
         spin_unlock(&xprt->reserve_lock);
@@ -1057,6 +1104,7 @@ static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
 {
         struct rpc_rqst *req = task->tk_rqstp;
 
+        INIT_LIST_HEAD(&req->rq_list);
         req->rq_timeout = task->tk_client->cl_timeout->to_initval;
         req->rq_task    = task;
         req->rq_xprt    = xprt;
diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
index 674a49224450..b446e100286f 100644
--- a/net/sunrpc/xprtrdma/transport.c
+++ b/net/sunrpc/xprtrdma/transport.c
@@ -283,6 +283,7 @@ xprt_setup_rdma(struct xprt_create *args)
         }
 
         xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt),
+                        xprt_rdma_slot_table_entries,
                         xprt_rdma_slot_table_entries);
         if (xprt == NULL) {
                 dprintk("RPC:       %s: couldn't allocate rpcrdma_xprt\n",
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index adaa54c6a09a..d7f97ef26590 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -54,7 +54,8 @@ static void xs_close(struct rpc_xprt *xprt);
  * xprtsock tunables
  */
 unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
-unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE;
+unsigned int xprt_tcp_slot_table_entries = RPC_MIN_SLOT_TABLE;
+unsigned int xprt_max_tcp_slot_table_entries = RPC_MAX_SLOT_TABLE;
 
 unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
 unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
@@ -75,6 +76,7 @@ static unsigned int xs_tcp_fin_timeout __read_mostly = XS_TCP_LINGER_TO;
 
 static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
 static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
+static unsigned int max_tcp_slot_table_limit = RPC_MAX_SLOT_TABLE_LIMIT;
 static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
 static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
 
@@ -104,6 +106,15 @@ static ctl_table xs_tunables_table[] = {
                 .extra2         = &max_slot_table_size
         },
         {
+                .procname       = "tcp_max_slot_table_entries",
+                .data           = &xprt_max_tcp_slot_table_entries,
+                .maxlen         = sizeof(unsigned int),
+                .mode           = 0644,
+                .proc_handler   = proc_dointvec_minmax,
+                .extra1         = &min_slot_table_size,
+                .extra2         = &max_tcp_slot_table_limit
+        },
+        {
                 .procname       = "min_resvport",
                 .data           = &xprt_min_resvport,
                 .maxlen         = sizeof(unsigned int),
@@ -2491,7 +2502,8 @@ static int xs_init_anyaddr(const int family, struct sockaddr *sap)
 }
 
 static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
-                                      unsigned int slot_table_size)
+                                      unsigned int slot_table_size,
+                                      unsigned int max_slot_table_size)
 {
         struct rpc_xprt *xprt;
         struct sock_xprt *new;
@@ -2501,7 +2513,8 @@ static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
                 return ERR_PTR(-EBADF);
         }
 
-        xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size);
+        xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size,
+                        max_slot_table_size);
         if (xprt == NULL) {
                 dprintk("RPC:       xs_setup_xprt: couldn't allocate "
                                 "rpc_xprt\n");
@@ -2543,7 +2556,8 @@ static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
         struct rpc_xprt *xprt;
         struct rpc_xprt *ret;
 
-        xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries);
+        xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
+                        xprt_max_tcp_slot_table_entries);
         if (IS_ERR(xprt))
                 return xprt;
         transport = container_of(xprt, struct sock_xprt, xprt);
@@ -2607,7 +2621,8 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
         struct sock_xprt *transport;
         struct rpc_xprt *ret;
 
-        xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries);
+        xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries,
+                        xprt_udp_slot_table_entries);
         if (IS_ERR(xprt))
                 return xprt;
         transport = container_of(xprt, struct sock_xprt, xprt);
@@ -2683,7 +2698,8 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
         struct sock_xprt *transport;
         struct rpc_xprt *ret;
 
-        xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries);
+        xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
+                        xprt_max_tcp_slot_table_entries);
         if (IS_ERR(xprt))
                 return xprt;
         transport = container_of(xprt, struct sock_xprt, xprt);
@@ -2762,7 +2778,8 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
                  */
                  return args->bc_xprt->xpt_bc_xprt;
         }
-        xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries);
+        xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
+                        xprt_tcp_slot_table_entries);
         if (IS_ERR(xprt))
                 return xprt;
         transport = container_of(xprt, struct sock_xprt, xprt);
@@ -2949,8 +2966,26 @@ static struct kernel_param_ops param_ops_slot_table_size = {
 #define param_check_slot_table_size(name, p) \
         __param_check(name, p, unsigned int);
 
+static int param_set_max_slot_table_size(const char *val,
+                                     const struct kernel_param *kp)
+{
+        return param_set_uint_minmax(val, kp,
+                        RPC_MIN_SLOT_TABLE,
+                        RPC_MAX_SLOT_TABLE_LIMIT);
+}
+
+static struct kernel_param_ops param_ops_max_slot_table_size = {
+        .set = param_set_max_slot_table_size,
+        .get = param_get_uint,
+};
+
+#define param_check_max_slot_table_size(name, p) \
+        __param_check(name, p, unsigned int);
+
 module_param_named(tcp_slot_table_entries, xprt_tcp_slot_table_entries,
                    slot_table_size, 0644);
+module_param_named(tcp_max_slot_table_entries, xprt_max_tcp_slot_table_entries,
+                   max_slot_table_size, 0644);
 module_param_named(udp_slot_table_entries, xprt_udp_slot_table_entries,
                    slot_table_size, 0644);