LU-3467 mdt: call MDT handlers via unified request handler

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

diff --git a/lustre/mdt/mdt_reint.c b/lustre/mdt/mdt_reint.c
index f8ab022..617ac43 100644 (file)
--- a/lustre/mdt/mdt_reint.c
+++ b/lustre/mdt/mdt_reint.c
@@ -66,7 +66,7 @@ static int mdt_create_pack_capa(struct mdt_thread_info *info, int rc,
         if (repbody->valid & OBD_MD_FLMDSCAPA)
                 RETURN(rc);
 
-       if (rc == 0 && info->mti_mdt->mdt_opts.mo_mds_capa &&
+       if (rc == 0 && info->mti_mdt->mdt_lut.lut_mds_capa &&
            exp_connect_flags(info->mti_exp) & OBD_CONNECT_MDS_CAPA) {
                 struct lustre_capa *capa;
 
@@ -469,14 +469,21 @@ int mdt_add_dirty_flag(struct mdt_thread_info *info, struct mdt_object *mo,
        /* If an up2date copy exists in the backend, add dirty flag */
        if ((ma->ma_valid & MA_HSM) && (ma->ma_hsm.mh_flags & HS_EXISTS)
            && !(ma->ma_hsm.mh_flags & (HS_DIRTY|HS_RELEASED))) {
+               struct mdt_lock_handle  *lh = &info->mti_lh[MDT_LH_CHILD];
 
                ma->ma_hsm.mh_flags |= HS_DIRTY;
+
+               mdt_lock_reg_init(lh, LCK_PW);
+               rc = mdt_object_lock(info, mo, lh, MDS_INODELOCK_XATTR,
+                                    MDT_LOCAL_LOCK);
+               if (rc != 0)
+                       RETURN(rc);
+
                rc = mdt_hsm_attr_set(info, mo, &ma->ma_hsm);
-               if (rc) {
+               if (rc)
                        CERROR("file attribute change error for "DFID": %d\n",
                                PFID(mdt_object_fid(mo)), rc);
-                       RETURN(rc);
-               }
+               mdt_object_unlock(info, mo, lh, rc);
        }
 
        RETURN(rc);
@@ -521,7 +528,7 @@ static int mdt_reint_setattr(struct mdt_thread_info *info,
                 if (rc)
                         GOTO(out_put, rc);
 
-                mfd = mdt_mfd_new();
+               mfd = mdt_mfd_new(med);
                 if (mfd == NULL) {
                         mdt_write_put(mo);
                         GOTO(out_put, rc = -ENOMEM);
@@ -549,7 +556,8 @@ static int mdt_reint_setattr(struct mdt_thread_info *info,
                 LASSERT(info->mti_ioepoch);
 
                spin_lock(&med->med_open_lock);
-               mfd = mdt_handle2mfd(info, &info->mti_ioepoch->handle);
+               mfd = mdt_handle2mfd(med, &info->mti_ioepoch->handle,
+                                    req_is_replay(req));
                if (mfd == NULL) {
                        spin_unlock(&med->med_open_lock);
                         CDEBUG(D_INODE, "no handle for file close: "
@@ -590,12 +598,12 @@ static int mdt_reint_setattr(struct mdt_thread_info *info,
         ma->ma_need = MA_INODE;
         ma->ma_valid = 0;
        rc = mdt_attr_get_complex(info, mo, ma);
-        if (rc != 0)
-                GOTO(out_put, rc);
+       if (rc != 0)
+               GOTO(out_put, rc);
 
-        mdt_pack_attr2body(info, repbody, &ma->ma_attr, mdt_object_fid(mo));
+       mdt_pack_attr2body(info, repbody, &ma->ma_attr, mdt_object_fid(mo));
 
-       if (info->mti_mdt->mdt_opts.mo_oss_capa &&
+       if (info->mti_mdt->mdt_lut.lut_oss_capa &&
            exp_connect_flags(info->mti_exp) & OBD_CONNECT_OSS_CAPA &&
            S_ISREG(lu_object_attr(&mo->mot_obj)) &&
            (ma->ma_attr.la_valid & LA_SIZE) && !som_au) {
@@ -820,11 +828,14 @@ static int mdt_reint_unlink(struct mdt_thread_info *info,
                GOTO(put_child, rc);
        }
 
-       rc = mdt_object_lock(info, mc, child_lh, MDS_INODELOCK_FULL,
-                             MDT_CROSS_LOCK);
-       if (rc != 0) {
+       /* We used to acquire MDS_INODELOCK_FULL here but we can't do
+        * this now because a running HSM restore on the child (unlink
+        * victim) will hold the layout lock. See LU-4002. */
+       rc = mdt_object_lock(info, mc, child_lh,
+                            MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE,
+                            MDT_CROSS_LOCK);
+       if (rc != 0)
                GOTO(put_child, rc);
-       }
 
         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
                        OBD_FAIL_MDS_REINT_UNLINK_WRITE);
@@ -942,8 +953,8 @@ static int mdt_reint_link(struct mdt_thread_info *info,
                GOTO(out_unlock_parent, rc = -EXDEV);
        }
 
-        rc = mdt_object_lock(info, ms, lhs, MDS_INODELOCK_UPDATE,
-                            MDT_CROSS_LOCK);
+       rc = mdt_object_lock(info, ms, lhs, MDS_INODELOCK_UPDATE |
+                            MDS_INODELOCK_XATTR, MDT_CROSS_LOCK);
         if (rc != 0) {
                 mdt_object_put(info->mti_env, ms);
                 GOTO(out_unlock_parent, rc);
@@ -1033,7 +1044,7 @@ static int mdt_rename_lock(struct mdt_thread_info *info,
 
        memset(policy, 0, sizeof *policy);
        policy->l_inodebits.bits = MDS_INODELOCK_UPDATE;
-#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 4, 53, 0)
+#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 5, 53, 0)
        /* In phase I, we will not do cross-rename, so local BFL lock would
         * be enough
         */
@@ -1234,8 +1245,8 @@ static int mdt_reint_rename(struct mdt_thread_info *info,
 
         lh_oldp = &info->mti_lh[MDT_LH_OLD];
         mdt_lock_reg_init(lh_oldp, LCK_EX);
-        rc = mdt_object_lock(info, mold, lh_oldp, MDS_INODELOCK_LOOKUP,
-                             MDT_CROSS_LOCK);
+       rc = mdt_object_lock(info, mold, lh_oldp, MDS_INODELOCK_LOOKUP |
+                            MDS_INODELOCK_XATTR, MDT_CROSS_LOCK);
         if (rc != 0) {
                 mdt_object_put(info->mti_env, mold);
                 GOTO(out_unlock_target, rc);
@@ -1282,8 +1293,14 @@ static int mdt_reint_rename(struct mdt_thread_info *info,
                        GOTO(out_unlock_old, rc = -EXDEV);
                }
 
-                rc = mdt_object_lock(info, mnew, lh_newp,
-                                     MDS_INODELOCK_FULL, MDT_CROSS_LOCK);
+               /* We used to acquire MDS_INODELOCK_FULL here but we
+                * can't do this now because a running HSM restore on
+                * the rename onto victim will hold the layout
+                * lock. See LU-4002. */
+               rc = mdt_object_lock(info, mnew, lh_newp,
+                                    MDS_INODELOCK_LOOKUP |
+                                    MDS_INODELOCK_UPDATE,
+                                    MDT_CROSS_LOCK);
                 if (rc != 0) {
                         mdt_object_put(info->mti_env, mnew);
                         GOTO(out_unlock_old, rc);
@@ -1294,7 +1311,14 @@ static int mdt_reint_rename(struct mdt_thread_info *info,
         } else if (rc != -EREMOTE && rc != -ENOENT) {
                 GOTO(out_unlock_old, rc);
         } else {
-                mdt_enoent_version_save(info, 3);
+               /* If mnew does not exist and mold are remote directory,
+                * it only allows rename if they are under same directory */
+               if (mtgtdir != msrcdir && mdt_object_remote(mold)) {
+                       CDEBUG(D_INFO, "Src child "DFID" is on another MDT\n",
+                              PFID(old_fid));
+                       GOTO(out_unlock_old, rc = -EXDEV);
+               }
+               mdt_enoent_version_save(info, 3);
         }
 
         /* step 5: rename it */