Merge git://git.kernel.org/pub/scm/linux/kernel/git/bpf/bpf

[linux.git] / net / sunrpc / clnt.c

diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index b03bfa055c0827972cfe497a2dbf98df42f28445..d8679b6027e933a2130fafcecd12e57b9975c8db 100644 (file)
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -528,6 +528,8 @@ struct rpc_clnt *rpc_create(struct rpc_create_args *args)
                .bc_xprt = args->bc_xprt,
        };
        char servername[48];
+       struct rpc_clnt *clnt;
+       int i;
 
        if (args->bc_xprt) {
                WARN_ON_ONCE(!(args->protocol & XPRT_TRANSPORT_BC));
@@ -590,7 +592,15 @@ struct rpc_clnt *rpc_create(struct rpc_create_args *args)
        if (args->flags & RPC_CLNT_CREATE_NONPRIVPORT)
                xprt->resvport = 0;
 
-       return rpc_create_xprt(args, xprt);
+       clnt = rpc_create_xprt(args, xprt);
+       if (IS_ERR(clnt) || args->nconnect <= 1)
+               return clnt;
+
+       for (i = 0; i < args->nconnect - 1; i++) {
+               if (rpc_clnt_add_xprt(clnt, &xprtargs, NULL, NULL) < 0)
+                       break;
+       }
+       return clnt;
 }
 EXPORT_SYMBOL_GPL(rpc_create);
 
@@ -968,13 +978,46 @@ struct rpc_clnt *rpc_bind_new_program(struct rpc_clnt *old,
 }
 EXPORT_SYMBOL_GPL(rpc_bind_new_program);
 
+struct rpc_xprt *
+rpc_task_get_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
+{
+       struct rpc_xprt_switch *xps;
+
+       if (!xprt)
+               return NULL;
+       rcu_read_lock();
+       xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
+       atomic_long_inc(&xps->xps_queuelen);
+       rcu_read_unlock();
+       atomic_long_inc(&xprt->queuelen);
+
+       return xprt;
+}
+
+static void
+rpc_task_release_xprt(struct rpc_clnt *clnt, struct rpc_xprt *xprt)
+{
+       struct rpc_xprt_switch *xps;
+
+       atomic_long_dec(&xprt->queuelen);
+       rcu_read_lock();
+       xps = rcu_dereference(clnt->cl_xpi.xpi_xpswitch);
+       atomic_long_dec(&xps->xps_queuelen);
+       rcu_read_unlock();
+
+       xprt_put(xprt);
+}
+
 void rpc_task_release_transport(struct rpc_task *task)
 {
        struct rpc_xprt *xprt = task->tk_xprt;
 
        if (xprt) {
                task->tk_xprt = NULL;
-               xprt_put(xprt);
+               if (task->tk_client)
+                       rpc_task_release_xprt(task->tk_client, xprt);
+               else
+                       xprt_put(xprt);
        }
 }
 EXPORT_SYMBOL_GPL(rpc_task_release_transport);
@@ -983,6 +1026,7 @@ void rpc_task_release_client(struct rpc_task *task)
 {
        struct rpc_clnt *clnt = task->tk_client;
 
+       rpc_task_release_transport(task);
        if (clnt != NULL) {
                /* Remove from client task list */
                spin_lock(&clnt->cl_lock);
@@ -992,14 +1036,34 @@ void rpc_task_release_client(struct rpc_task *task)
 
                rpc_release_client(clnt);
        }
-       rpc_task_release_transport(task);
+}
+
+static struct rpc_xprt *
+rpc_task_get_first_xprt(struct rpc_clnt *clnt)
+{
+       struct rpc_xprt *xprt;
+
+       rcu_read_lock();
+       xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
+       rcu_read_unlock();
+       return rpc_task_get_xprt(clnt, xprt);
+}
+
+static struct rpc_xprt *
+rpc_task_get_next_xprt(struct rpc_clnt *clnt)
+{
+       return rpc_task_get_xprt(clnt, xprt_iter_get_next(&clnt->cl_xpi));
 }
 
 static
 void rpc_task_set_transport(struct rpc_task *task, struct rpc_clnt *clnt)
 {
-       if (!task->tk_xprt)
-               task->tk_xprt = xprt_iter_get_next(&clnt->cl_xpi);
+       if (task->tk_xprt)
+               return;
+       if (task->tk_flags & RPC_TASK_NO_ROUND_ROBIN)
+               task->tk_xprt = rpc_task_get_first_xprt(clnt);
+       else
+               task->tk_xprt = rpc_task_get_next_xprt(clnt);
 }
 
 static
@@ -1462,6 +1526,19 @@ size_t rpc_max_bc_payload(struct rpc_clnt *clnt)
 }
 EXPORT_SYMBOL_GPL(rpc_max_bc_payload);
 
+unsigned int rpc_num_bc_slots(struct rpc_clnt *clnt)
+{
+       struct rpc_xprt *xprt;
+       unsigned int ret;
+
+       rcu_read_lock();
+       xprt = rcu_dereference(clnt->cl_xprt);
+       ret = xprt->ops->bc_num_slots(xprt);
+       rcu_read_unlock();
+       return ret;
+}
+EXPORT_SYMBOL_GPL(rpc_num_bc_slots);
+
 /**
  * rpc_force_rebind - force transport to check that remote port is unchanged
  * @clnt: client to rebind
@@ -1788,6 +1865,7 @@ rpc_xdr_encode(struct rpc_task *task)
        req->rq_snd_buf.head[0].iov_len = 0;
        xdr_init_encode(&xdr, &req->rq_snd_buf,
                        req->rq_snd_buf.head[0].iov_base, req);
+       xdr_free_bvec(&req->rq_snd_buf);
        if (rpc_encode_header(task, &xdr))
                return;
 
@@ -1827,8 +1905,6 @@ call_encode(struct rpc_task *task)
                        rpc_call_rpcerror(task, task->tk_status);
                }
                return;
-       } else {
-               xprt_request_prepare(task->tk_rqstp);
        }
 
        /* Add task to reply queue before transmission to avoid races */
@@ -2696,6 +2772,10 @@ int rpc_clnt_test_and_add_xprt(struct rpc_clnt *clnt,
                return -ENOMEM;
        data->xps = xprt_switch_get(xps);
        data->xprt = xprt_get(xprt);
+       if (rpc_xprt_switch_has_addr(data->xps, (struct sockaddr *)&xprt->addr)) {
+               rpc_cb_add_xprt_release(data);
+               goto success;
+       }
 
        task = rpc_call_null_helper(clnt, xprt, NULL,
                        RPC_TASK_SOFT|RPC_TASK_SOFTCONN|RPC_TASK_ASYNC|RPC_TASK_NULLCREDS,
@@ -2703,6 +2783,7 @@ int rpc_clnt_test_and_add_xprt(struct rpc_clnt *clnt,
        if (IS_ERR(task))
                return PTR_ERR(task);
        rpc_put_task(task);
+success:
        return 1;
 }
 EXPORT_SYMBOL_GPL(rpc_clnt_test_and_add_xprt);