LU-12828 ldlm: FLOCK request can be processed twice

[fs/lustre-release.git] / lustre / mdc / mdc_locks.c

diff --git a/lustre/mdc/mdc_locks.c b/lustre/mdc/mdc_locks.c
index 7923428..7bceb53 100644 (file)
--- a/lustre/mdc/mdc_locks.c
+++ b/lustre/mdc/mdc_locks.c
@@ -860,6 +860,16 @@ out_lock:
        RETURN(rc);
 }
 
+static inline bool mdc_skip_mod_rpc_slot(const struct lookup_intent *it)
+{
+       if (it != NULL &&
+           (it->it_op == IT_GETATTR || it->it_op == IT_LOOKUP ||
+            it->it_op == IT_READDIR ||
+            (it->it_op == IT_LAYOUT && !(it->it_flags & MDS_FMODE_WRITE))))
+               return true;
+       return false;
+}
+
 /* We always reserve enough space in the reply packet for a stripe MD, because
  * we don't know in advance the file type. */
 static int mdc_enqueue_base(struct obd_export *exp,
@@ -871,7 +881,7 @@ static int mdc_enqueue_base(struct obd_export *exp,
                            __u64 extra_lock_flags)
 {
        struct obd_device *obddev = class_exp2obd(exp);
-       struct ptlrpc_request *req = NULL;
+       struct ptlrpc_request *req;
        __u64 flags, saved_flags = extra_lock_flags;
        struct ldlm_res_id res_id;
        static const union ldlm_policy_data lookup_policy = {
@@ -922,6 +932,7 @@ resend:
                LASSERTF(einfo->ei_type == LDLM_FLOCK, "lock type %d\n",
                         einfo->ei_type);
                res_id.name[3] = LDLM_FLOCK;
+               req = ldlm_enqueue_pack(exp, 0);
        } else if (it->it_op & IT_OPEN) {
                req = mdc_intent_open_pack(exp, it, op_data, acl_bufsize);
        } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
@@ -949,20 +960,7 @@ resend:
                req->rq_sent = ktime_get_real_seconds() + resends;
         }
 
-       /* It is important to obtain modify RPC slot first (if applicable), so
-        * that threads that are waiting for a modify RPC slot are not polluting
-        * our rpcs in flight counter.
-        * We do not do flock request limiting, though */
-       if (it) {
-               mdc_get_mod_rpc_slot(req, it);
-               rc = obd_get_request_slot(&obddev->u.cli);
-               if (rc != 0) {
-                       mdc_put_mod_rpc_slot(req, it);
-                       mdc_clear_replay_flag(req, 0);
-                       ptlrpc_req_finished(req);
-                       RETURN(rc);
-               }
-       }
+       einfo->ei_enq_slot = !mdc_skip_mod_rpc_slot(it);
 
        /* With Data-on-MDT the glimpse callback is needed too.
         * It is set here in advance but not in mdc_finish_enqueue()
@@ -973,6 +971,7 @@ resend:
 
        rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id, policy, &flags, NULL,
                              0, lvb_type, lockh, 0);
+
        if (!it) {
                /* For flock requests we immediatelly return without further
                   delay and let caller deal with the rest, since rest of
@@ -986,12 +985,10 @@ resend:
                    (einfo->ei_type == LDLM_FLOCK) &&
                    (einfo->ei_mode == LCK_NL))
                        goto resend;
+               ptlrpc_req_finished(req);
                RETURN(rc);
        }
 
-       obd_put_request_slot(&obddev->u.cli);
-       mdc_put_mod_rpc_slot(req, it);
-
        if (rc < 0) {
                CDEBUG(D_INFO,
                      "%s: ldlm_cli_enqueue "DFID":"DFID"=%s failed: rc = %d\n",
@@ -1335,19 +1332,15 @@ static int mdc_intent_getattr_async_interpret(const struct lu_env *env,
        struct obd_export *exp = ga->ga_exp;
        struct md_enqueue_info *minfo = ga->ga_minfo;
        struct ldlm_enqueue_info *einfo = &minfo->mi_einfo;
-       struct lookup_intent *it;
-       struct lustre_handle *lockh;
-       struct obd_device *obddev;
-       struct ldlm_reply *lockrep;
-       __u64 flags = LDLM_FL_HAS_INTENT;
+       struct lookup_intent     *it;
+       struct lustre_handle     *lockh;
+       struct ldlm_reply        *lockrep;
+       __u64                     flags = LDLM_FL_HAS_INTENT;
        ENTRY;
 
         it    = &minfo->mi_it;
         lockh = &minfo->mi_lockh;
 
-        obddev = class_exp2obd(exp);
-
-       obd_put_request_slot(&obddev->u.cli);
         if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GETATTR_ENQUEUE))
                 rc = -ETIMEDOUT;
 
@@ -1384,7 +1377,6 @@ int mdc_intent_getattr_async(struct obd_export *exp,
        struct lookup_intent    *it = &minfo->mi_it;
        struct ptlrpc_request   *req;
        struct mdc_getattr_args *ga;
-       struct obd_device       *obddev = class_exp2obd(exp);
        struct ldlm_res_id       res_id;
        union ldlm_policy_data policy = {
                                .l_inodebits = { MDS_INODELOCK_LOOKUP |
@@ -1405,12 +1397,6 @@ int mdc_intent_getattr_async(struct obd_export *exp,
        if (IS_ERR(req))
                RETURN(PTR_ERR(req));
 
-       rc = obd_get_request_slot(&obddev->u.cli);
-       if (rc != 0) {
-               ptlrpc_req_finished(req);
-               RETURN(rc);
-       }
-
        /* With Data-on-MDT the glimpse callback is needed too.
         * It is set here in advance but not in mdc_finish_enqueue()
         * to avoid possible races. It is safe to have glimpse handler
@@ -1421,7 +1407,6 @@ int mdc_intent_getattr_async(struct obd_export *exp,
        rc = ldlm_cli_enqueue(exp, &req, &minfo->mi_einfo, &res_id, &policy,
                              &flags, NULL, 0, LVB_T_NONE, &minfo->mi_lockh, 1);
        if (rc < 0) {
-               obd_put_request_slot(&obddev->u.cli);
                ptlrpc_req_finished(req);
                RETURN(rc);
        }