LU-3528 mdt: check object exists for remote directory

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

diff --git a/lustre/mdt/mdt_reint.c b/lustre/mdt/mdt_reint.c
index 617ac43..7e89ad6 100644 (file)
--- a/lustre/mdt/mdt_reint.c
+++ b/lustre/mdt/mdt_reint.c
@@ -349,7 +349,7 @@ static int mdt_md_create(struct mdt_thread_info *info)
                                OBD_FAIL_MDS_REINT_CREATE_WRITE);
 
                 /* Version of child will be updated on disk. */
-                info->mti_mos = child;
+               tgt_vbr_obj_set(info->mti_env, mdt_obj2dt(child));
                 rc = mdt_version_get_check_save(info, child, 2);
                 if (rc)
                         GOTO(out_put_child, rc);
@@ -428,7 +428,7 @@ int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo,
         /* VBR: update version if attr changed are important for recovery */
         if (do_vbr) {
                 /* update on-disk version of changed object */
-                info->mti_mos = mo;
+               tgt_vbr_obj_set(info->mti_env, mdt_obj2dt(mo));
                 rc = mdt_version_get_check_save(info, mo, 0);
                 if (rc)
                         GOTO(out_unlock, rc);
@@ -841,16 +841,21 @@ static int mdt_reint_unlink(struct mdt_thread_info *info,
                        OBD_FAIL_MDS_REINT_UNLINK_WRITE);
         /* save version when object is locked */
         mdt_version_get_save(info, mc, 1);
-        /*
-         * Now we can only make sure we need MA_INODE, in mdd layer, will check
-         * whether need MA_LOV and MA_COOKIE.
-         */
-        ma->ma_need = MA_INODE;
-        ma->ma_valid = 0;
-        mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
+       /*
+        * Now we can only make sure we need MA_INODE, in mdd layer, will check
+        * whether need MA_LOV and MA_COOKIE.
+        */
+       ma->ma_need = MA_INODE;
+       ma->ma_valid = 0;
+       mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
+
+       mutex_lock(&mc->mot_lov_mutex);
 
        rc = mdo_unlink(info->mti_env, mdt_object_child(mp),
                        mdt_object_child(mc), lname, ma, no_name);
+
+       mutex_unlock(&mc->mot_lov_mutex);
+
        if (rc == 0 && !lu_object_is_dying(&mc->mot_header))
                rc = mdt_attr_get_complex(info, mc, ma);
        if (rc == 0)
@@ -945,6 +950,13 @@ static int mdt_reint_link(struct mdt_thread_info *info,
         if (IS_ERR(ms))
                 GOTO(out_unlock_parent, rc = PTR_ERR(ms));
 
+       if (!mdt_object_exists(ms)) {
+               mdt_object_put(info->mti_env, ms);
+               CDEBUG(D_INFO, "%s: "DFID" does not exist.\n",
+                      mdt_obd_name(info->mti_mdt), PFID(rr->rr_fid1));
+               GOTO(out_unlock_parent, rc = -ENOENT);
+       }
+
        if (mdt_object_remote(ms)) {
                mdt_object_put(info->mti_env, ms);
                CERROR("%s: source inode "DFID" on remote MDT from "DFID"\n",
@@ -964,7 +976,7 @@ static int mdt_reint_link(struct mdt_thread_info *info,
         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
                        OBD_FAIL_MDS_REINT_LINK_WRITE);
 
-        info->mti_mos = ms;
+       tgt_vbr_obj_set(info->mti_env, mdt_obj2dt(ms));
         rc = mdt_version_get_check_save(info, ms, 1);
         if (rc)
                 GOTO(out_unlock_child, rc);
@@ -1252,7 +1264,7 @@ static int mdt_reint_rename(struct mdt_thread_info *info,
                 GOTO(out_unlock_target, rc);
         }
 
-        info->mti_mos = mold;
+       tgt_vbr_obj_set(info->mti_env, mdt_obj2dt(mold));
         /* save version after locking */
         mdt_version_get_save(info, mold, 2);
         mdt_set_capainfo(info, 2, old_fid, BYPASS_CAPA);
@@ -1327,22 +1339,27 @@ static int mdt_reint_rename(struct mdt_thread_info *info,
         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
                        OBD_FAIL_MDS_REINT_RENAME_WRITE);
 
+       /* Check if @dst is subdir of @src. */
+       rc = mdt_rename_sanity(info, old_fid);
+       if (rc)
+               GOTO(out_unlock_new, rc);
 
-        /* Check if @dst is subdir of @src. */
-        rc = mdt_rename_sanity(info, old_fid);
-        if (rc)
-                GOTO(out_unlock_new, rc);
+       if (mnew != NULL)
+               mutex_lock(&mnew->mot_lov_mutex);
 
-        rc = mdo_rename(info->mti_env, mdt_object_child(msrcdir),
-                        mdt_object_child(mtgtdir), old_fid, &slname,
-                        (mnew ? mdt_object_child(mnew) : NULL),
-                        lname, ma);
+       rc = mdo_rename(info->mti_env, mdt_object_child(msrcdir),
+                       mdt_object_child(mtgtdir), old_fid, &slname,
+                       (mnew ? mdt_object_child(mnew) : NULL),
+                       lname, ma);
 
-        /* handle last link of tgt object */
-        if (rc == 0) {
+       if (mnew != NULL)
+               mutex_unlock(&mnew->mot_lov_mutex);
+
+       /* handle last link of tgt object */
+       if (rc == 0) {
                mdt_counter_incr(req, LPROC_MDT_RENAME);
-                if (mnew)
-                        mdt_handle_last_unlink(info, mnew, ma);
+               if (mnew)
+                       mdt_handle_last_unlink(info, mnew, ma);
 
                mdt_rename_counter_tally(info, info->mti_mdt, req,
                                          msrcdir, mtgtdir);