SUNRPC: Add a per-rpc_clnt spinlock

Use that to protect the rpc_clnt->cl_tasks list instead of using a global
lock.

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

3 files changed, 30 insertions, 20 deletions

diff --git a/include/linux/sunrpc/clnt.h b/include/linux/sunrpc/clnt.h
index 0801ab5407ce..2f4b520a7419 100644
--- a/include/linux/sunrpc/clnt.h
+++ b/include/linux/sunrpc/clnt.h
@@ -28,6 +28,7 @@ struct rpc_clnt {
         atomic_t                cl_users;       /* number of references */
         struct list_head        cl_clients;     /* Global list of clients */
         struct list_head        cl_tasks;       /* List of tasks */
+        spinlock_t              cl_lock;        /* spinlock */
         struct rpc_xprt *       cl_xprt;        /* transport */
         struct rpc_procinfo *   cl_procinfo;    /* procedure info */
         u32                     cl_prog,        /* RPC program number */
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index 6631ece14983..424dfdc6862c 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -149,6 +149,7 @@ static struct rpc_clnt * rpc_new_client(struct rpc_xprt *xprt, char *servname, s
                 goto out_no_stats;
         clnt->cl_program  = program;
         INIT_LIST_HEAD(&clnt->cl_tasks);
+        spin_lock_init(&clnt->cl_lock);
 
         if (!xprt_bound(clnt->cl_xprt))
                 clnt->cl_autobind = 1;
@@ -286,6 +287,7 @@ rpc_clone_client(struct rpc_clnt *clnt)
         new->cl_oneshot = 0;
         new->cl_dead = 0;
         INIT_LIST_HEAD(&new->cl_tasks);
+        spin_lock_init(&new->cl_lock);
         rpc_init_rtt(&new->cl_rtt_default, clnt->cl_xprt->timeout.to_initval);
         if (new->cl_auth)
                 atomic_inc(&new->cl_auth->au_count);
diff --git a/net/sunrpc/sched.c b/net/sunrpc/sched.c
index 6309f3b52c53..f56ebc5a08f7 100644
--- a/net/sunrpc/sched.c
+++ b/net/sunrpc/sched.c
@@ -270,17 +270,22 @@ static int rpc_wait_bit_interruptible(void *word)
 
 static void rpc_set_active(struct rpc_task *task)
 {
+        struct rpc_clnt *clnt;
         if (test_and_set_bit(RPC_TASK_ACTIVE, &task->tk_runstate) != 0)
                 return;
-        spin_lock(&rpc_sched_lock);
 #ifdef RPC_DEBUG
         task->tk_magic = RPC_TASK_MAGIC_ID;
+        spin_lock(&rpc_sched_lock);
         task->tk_pid = rpc_task_id++;
+        spin_unlock(&rpc_sched_lock);
 #endif
         /* Add to global list of all tasks */
-        if (task->tk_client)
-                list_add_tail(&task->tk_task, &task->tk_client->cl_tasks);
-        spin_unlock(&rpc_sched_lock);
+        clnt = task->tk_client;
+        if (clnt != NULL) {
+                spin_lock(&clnt->cl_lock);
+                list_add_tail(&task->tk_task, &clnt->cl_tasks);
+                spin_unlock(&clnt->cl_lock);
+        }
 }
 
 /*
@@ -924,10 +929,11 @@ static void rpc_release_task(struct rpc_task *task)
         dprintk("RPC: %5u release task\n", task->tk_pid);
 
         if (!list_empty(&task->tk_task)) {
+                struct rpc_clnt *clnt = task->tk_client;
                 /* Remove from client task list */
-                spin_lock(&rpc_sched_lock);
+                spin_lock(&clnt->cl_lock);
                 list_del(&task->tk_task);
-                spin_unlock(&rpc_sched_lock);
+                spin_unlock(&clnt->cl_lock);
         }
         BUG_ON (RPC_IS_QUEUED(task));
 
@@ -970,12 +976,19 @@ EXPORT_SYMBOL(rpc_run_task);
  * Kill all tasks for the given client.
  * XXX: kill their descendants as well?
  */
-static void rpc_killall_tasks_locked(struct list_head *head)
+void rpc_killall_tasks(struct rpc_clnt *clnt)
 {
         struct rpc_task *rovr;
 
 
-        list_for_each_entry(rovr, head, tk_task) {
+        if (list_empty(&clnt->cl_tasks))
+                return;
+        dprintk("RPC:       killing all tasks for client %p\n", clnt);
+        /*
+         * Spin lock all_tasks to prevent changes...
+         */
+        spin_lock(&clnt->cl_lock);
+        list_for_each_entry(rovr, &clnt->cl_tasks, tk_task) {
                 if (! RPC_IS_ACTIVATED(rovr))
                         continue;
                 if (!(rovr->tk_flags & RPC_TASK_KILLED)) {
@@ -984,17 +997,7 @@ static void rpc_killall_tasks_locked(struct list_head *head)
                         rpc_wake_up_task(rovr);
                 }
         }
-}
-
-void rpc_killall_tasks(struct rpc_clnt *clnt)
-{
-        dprintk("RPC:       killing all tasks for client %p\n", clnt);
-        /*
-         * Spin lock all_tasks to prevent changes...
-         */
-        spin_lock(&rpc_sched_lock);
-        rpc_killall_tasks_locked(&clnt->cl_tasks);
-        spin_unlock(&rpc_sched_lock);
+        spin_unlock(&clnt->cl_lock);
 }
 
 static void rpciod_killall(void)
@@ -1007,7 +1010,7 @@ static void rpciod_killall(void)
 
                 spin_lock(&rpc_sched_lock);
                 list_for_each_entry(clnt, &all_clients, cl_clients)
-                        rpc_killall_tasks_locked(&clnt->cl_tasks);
+                        rpc_killall_tasks(clnt);
                 spin_unlock(&rpc_sched_lock);
                 flush_workqueue(rpciod_workqueue);
                 if (!list_empty(&all_clients))
@@ -1110,6 +1113,9 @@ void rpc_show_tasks(void)
         printk("-pid- proc flgs status -client- -prog- --rqstp- -timeout "
                 "-rpcwait -action- ---ops--\n");
         list_for_each_entry(clnt, &all_clients, cl_clients) {
+                if (list_empty(&clnt->cl_tasks))
+                        continue;
+                spin_lock(&clnt->cl_lock);
                 list_for_each_entry(t, &clnt->cl_tasks, tk_task) {
                         const char *rpc_waitq = "none";
 
@@ -1126,6 +1132,7 @@ void rpc_show_tasks(void)
                                 rpc_waitq,
                                 t->tk_action, t->tk_ops);
                 }
+                spin_unlock(&clnt->cl_lock);
         }
 out:
         spin_unlock(&rpc_sched_lock);