LU-5287 export: hold exp_lock when modify exp_flags

[fs/lustre-release.git] / lustre / mdt / mdt_handler.c

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 59adef4..f244e61 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -1195,7 +1195,7 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi)
        info = tsi2mdt_info(tsi);
 
        if (info->mti_dlm_req != NULL)
-               ldlm_request_cancel(req, info->mti_dlm_req, 0);
+               ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP);
 
        if (req_capsule_get_size(info->mti_pill, &RMF_CAPA1, RCL_CLIENT))
                mdt_set_capainfo(info, 0, &info->mti_body->mbo_fid1,
@@ -1323,7 +1323,6 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
         struct mdt_body        *reqbody   = NULL;
         struct mdt_object      *parent    = info->mti_object;
         struct mdt_object      *child;
-        struct md_object       *next      = mdt_object_child(parent);
         struct lu_fid          *child_fid = &info->mti_tmp_fid1;
         struct lu_name         *lname     = NULL;
         struct mdt_lock_handle *lhp       = NULL;
@@ -1338,7 +1337,8 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
        LASSERT(ergo(is_resent,
                     lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT));
 
-       LASSERT(parent != NULL);
+       if (parent == NULL)
+               RETURN(-ENOENT);
 
        if (info->mti_cross_ref) {
                /* Only getattr on the child. Parent is on another node. */
@@ -1351,7 +1351,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 
                rc = mdt_check_resent_lock(info, child, lhc);
                if (rc < 0) {
-                       RETURN(-EPROTO);
+                       RETURN(rc);
                } else if (rc > 0) {
                        mdt_lock_handle_init(lhc);
                        mdt_lock_reg_init(lhc, LCK_PR);
@@ -1447,8 +1447,8 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 
                 /* step 2: lookup child's fid by name */
                 fid_zero(child_fid);
-                rc = mdo_lookup(info->mti_env, next, lname, child_fid,
-                                &info->mti_spec);
+               rc = mdo_lookup(info->mti_env, mdt_object_child(parent), lname,
+                               child_fid, &info->mti_spec);
                if (rc == -ENOENT)
                        mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG);
 
@@ -2329,8 +2329,14 @@ int mdt_check_resent_lock(struct mdt_thread_info *info,
 
                lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
                LASSERT(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT);
-               LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n",
-                        lhc->mlh_reg_lh.cookie);
+               if (lock == NULL) {
+                       /* Lock is pinned by ldlm_handle_enqueue0() as it is
+                        * a resend case, however, it could be already destroyed
+                        * due to client eviction or a raced cancel RPC. */
+                       LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64"\n",
+                                         lhc->mlh_reg_lh.cookie);
+                       RETURN(-ESTALE);
+               }
 
                if (!fid_res_name_eq(mdt_object_fid(mo),
                                     &lock->l_resource->lr_name)) {
@@ -3443,17 +3449,16 @@ static int mdt_intent_code(long itcode)
 static int mdt_intent_opc(long itopc, struct mdt_thread_info *info,
                          struct ldlm_lock **lockp, __u64 flags)
 {
-        struct req_capsule   *pill;
-        struct mdt_it_flavor *flv;
-        int opc;
-        int rc;
-        ENTRY;
-
-        opc = mdt_intent_code(itopc);
-        if (opc < 0)
-                RETURN(-EINVAL);
+       struct req_capsule      *pill = info->mti_pill;
+       struct ptlrpc_request   *req = mdt_info_req(info);
+       struct mdt_it_flavor    *flv;
+       int opc;
+       int rc;
+       ENTRY;
 
-        pill = info->mti_pill;
+       opc = mdt_intent_code(itopc);
+       if (opc < 0)
+               RETURN(-EINVAL);
 
        if (opc == MDT_IT_QUOTA) {
                struct lu_device *qmt = info->mti_mdt->mdt_qmt_dev;
@@ -3469,33 +3474,33 @@ static int mdt_intent_opc(long itopc, struct mdt_thread_info *info,
                RETURN(rc);
        }
 
-       flv  = &mdt_it_flavor[opc];
-        if (flv->it_fmt != NULL)
-                req_capsule_extend(pill, flv->it_fmt);
+       flv = &mdt_it_flavor[opc];
+       if (flv->it_fmt != NULL)
+               req_capsule_extend(pill, flv->it_fmt);
 
-        rc = mdt_unpack_req_pack_rep(info, flv->it_flags);
-        if (rc == 0) {
-                struct ptlrpc_request *req = mdt_info_req(info);
-               if (flv->it_flags & MUTABOR &&
-                   exp_connect_flags(req->rq_export) & OBD_CONNECT_RDONLY)
-                       RETURN(-EROFS);
-        }
-        if (rc == 0 && flv->it_act != NULL) {
+       rc = mdt_unpack_req_pack_rep(info, flv->it_flags);
+       if (rc < 0)
+               RETURN(rc);
+
+       if (flv->it_flags & MUTABOR &&
+           exp_connect_flags(req->rq_export) & OBD_CONNECT_RDONLY)
+               RETURN(-EROFS);
+
+       if (flv->it_act != NULL) {
                struct ldlm_reply *rep;
 
                /* execute policy */
                rc = flv->it_act(opc, info, lockp, flags);
 
                /* Check whether the reply has been packed successfully. */
-               if (mdt_info_req(info)->rq_repmsg != NULL) {
+               if (req->rq_repmsg != NULL) {
                        rep = req_capsule_server_get(info->mti_pill,
                                                     &RMF_DLM_REP);
                        rep->lock_policy_res2 =
                                ptlrpc_status_hton(rep->lock_policy_res2);
                }
-       } else {
-               rc = -EPROTO;
        }
+
        RETURN(rc);
 }
 
@@ -3992,7 +3997,7 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt,
        if (!obd) {
                CERROR("Can not find obd %s (%s in config)\n",
                       MDD_OBD_NAME, lustre_cfg_string(cfg, 0));
-               GOTO(class_detach, rc = -EINVAL);
+               GOTO(lcfg_cleanup, rc = -EINVAL);
        }
 
        lustre_cfg_free(lcfg);
@@ -4120,7 +4125,7 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt,
        if (!obd) {
                CERROR("Can not find obd %s (%s in config)\n", qmtname,
                       lustre_cfg_string(cfg, 0));
-               GOTO(class_detach, rc = -EINVAL);
+               GOTO(lcfg_cleanup, rc = -EINVAL);
        }
 
        lustre_cfg_free(lcfg);
@@ -5041,7 +5046,9 @@ static int mdt_obd_connect(const struct lu_env *env,
                         mdt_export_stats_init(obd, lexp, localdata);
 
                /* For phase I, sync for cross-ref operation. */
+               spin_lock(&lexp->exp_lock);
                lexp->exp_keep_sync = 1;
+               spin_unlock(&lexp->exp_lock);
         }
 
         if (rc != 0) {
@@ -5942,9 +5949,6 @@ static int __init mdt_mod_init(void)
                GOTO(lu_fini, rc);
 
        rc = class_register_type(&mdt_obd_device_ops, NULL, true, NULL,
-#ifndef HAVE_ONLY_PROCFS_SEQ
-                                NULL,
-#endif
                                 LUSTRE_MDT_NAME, &mdt_device_type);
        if (rc)
                GOTO(mds_fini, rc);