LU-1222 ldlm: Fix the race in AST sender vs multiple arriving RPCs

LU-1046 tried to close this race, but still left a bit of a window open.
Basically when two last RPCs arrive one of them is declared as "last"
and it is the one that would release "rpc" part of the arg refcount.
The problem is the "non last" rpc might actually finish processing later.

To combat this every RPC will now hold its own arg reference.

Change-Id: I0d1e69dc6ea37224422d4c7822bb1e6b9417585b
Signed-off-by: Oleg Drokin <green@whamcloud.com>
Reviewed-on: http://review.whamcloud.com/2350
Tested-by: Hudson
Reviewed-by: Bobi Jam <bobijam@whamcloud.com>
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Fan Yong <yong.fan@whamcloud.com>

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 19e5c12..558eee3 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -667,7 +667,6 @@ static int ldlm_cb_interpret(const struct lu_env *env,
         struct ldlm_cb_async_args *ca   = data;
         struct ldlm_lock          *lock = ca->ca_lock;
         struct ldlm_cb_set_arg    *arg  = ca->ca_set_arg;
         struct ldlm_cb_async_args *ca   = data;
         struct ldlm_lock          *lock = ca->ca_lock;
         struct ldlm_cb_set_arg    *arg  = ca->ca_set_arg;

-        int count;

         ENTRY;
 
         LASSERT(lock != NULL);
         ENTRY;
 
         LASSERT(lock != NULL);


@@ -680,11 +679,10 @@ static int ldlm_cb_interpret(const struct lu_env *env,
         }
         LDLM_LOCK_RELEASE(lock);
 
         }
         LDLM_LOCK_RELEASE(lock);
 

-        count = cfs_atomic_dec_return(&arg->rpcs);
-        if (count < arg->threshold)
+        if (cfs_atomic_dec_return(&arg->rpcs) < arg->threshold)

                 cfs_waitq_signal(&arg->waitq);
                 cfs_waitq_signal(&arg->waitq);

-        if (count == 0)
-                ldlm_csa_put(arg);
+
+        ldlm_csa_put(arg);

         RETURN(0);
 }
 
         RETURN(0);
 }
 


@@ -703,8 +701,8 @@ static inline int ldlm_bl_and_cp_ast_tail(struct ptlrpc_request *req,
                         cfs_atomic_inc(&arg->restart);
         } else {
                 LDLM_LOCK_GET(lock);
                         cfs_atomic_inc(&arg->restart);
         } else {
                 LDLM_LOCK_GET(lock);

-                if (cfs_atomic_inc_return(&arg->rpcs) == 1)
-                        cfs_atomic_inc(&arg->refcount);
+                cfs_atomic_inc(&arg->rpcs);
+                cfs_atomic_inc(&arg->refcount);

                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
         }
 
                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
         }