LU-8514 mdd: transaction failure should be checked

Transaction failure should not be silently ignored, otherwise
MDT doesn't know whether current operation have transaction, therefore
save lock upon transaction failure.

Add sanity.sh 407 for this.

Signed-off-by: Lai Siyao <lai.siyao@intel.com>
Change-Id: Ie133a77c7f1bf890319dbd3cc2b03412a23f5c82
Reviewed-on: http://review.whamcloud.com/22071
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-by: Mike Pershin <mike.pershin@intel.com>
Reviewed-by: wangdi <di.wang@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/obd_support.h b/lustre/include/obd_support.h
index db88c9f..b064f91 100644 (file)
--- a/lustre/include/obd_support.h
+++ b/lustre/include/obd_support.h
@@ -623,6 +623,7 @@ extern char obd_jobid_var[];
 #define OBD_FAIL_DT_DECLARE_DELETE             0x2016
 #define OBD_FAIL_DT_DELETE                     0x2017
 #define OBD_FAIL_DT_LOOKUP                     0x2018
+#define OBD_FAIL_DT_TXN_STOP                   0x2019
 
 #define OBD_FAIL_OSP_CHECK_INVALID_REC         0x2100
 #define OBD_FAIL_OSP_CHECK_ENOMEM              0x2101

diff --git a/lustre/mdd/mdd_dir.c b/lustre/mdd/mdd_dir.c
index 0e4d287..8417a75 100644 (file)
--- a/lustre/mdd/mdd_dir.c
+++ b/lustre/mdd/mdd_dir.c
@@ -1385,21 +1385,20 @@ static int mdd_link(const struct lu_env *env, struct md_object *tgt_obj,
                 * failure, reset rc here */
                rc = 0;
        }
-        EXIT;
+       EXIT;
 out_unlock:
-        mdd_write_unlock(env, mdd_sobj);
-        if (rc == 0)
+       mdd_write_unlock(env, mdd_sobj);
+       if (rc == 0)
                rc = mdd_changelog_ns_store(env, mdd, CL_HARDLINK, 0, mdd_sobj,
                                            mdo2fid(mdd_tobj), NULL, NULL,
                                            lname, NULL, handle);
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
-
+       rc = mdd_trans_stop(env, mdd, rc, handle);
        if (is_vmalloc_addr(ldata->ld_buf))
                /* if we vmalloced a large buffer drop it */
                lu_buf_free(ldata->ld_buf);
 out_pending:
-        return rc;
+       return rc;
 }
 
 static int mdd_mark_orphan_object(const struct lu_env *env,
@@ -1757,7 +1756,7 @@ cleanup:
        }
 
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
 
        return rc;
 }
@@ -1777,9 +1776,11 @@ static int mdd_cd_sanity_check(const struct lu_env *env,
         RETURN(0);
 }
 
-static int mdd_create_data(const struct lu_env *env, struct md_object *pobj,
-                           struct md_object *cobj, const struct md_op_spec *spec,
-                           struct md_attr *ma)
+static int mdd_create_data(const struct lu_env *env,
+                          struct md_object *pobj,
+                          struct md_object *cobj,
+                          const struct md_op_spec *spec,
+                          struct md_attr *ma)
 {
        struct mdd_device *mdd = mdo2mdd(cobj);
        struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
@@ -1811,14 +1812,14 @@ static int mdd_create_data(const struct lu_env *env, struct md_object *pobj,
        /* calling ->ah_make_hint() is used to transfer information from parent */
        mdd_object_make_hint(env, mdd_pobj, son, attr, spec, hint);
 
-        handle = mdd_trans_create(env, mdd);
-        if (IS_ERR(handle))
-                GOTO(out_free, rc = PTR_ERR(handle));
+       handle = mdd_trans_create(env, mdd);
+       if (IS_ERR(handle))
+               GOTO(out_free, rc = PTR_ERR(handle));
 
-        /*
-         * XXX: Setting the lov ea is not locked but setting the attr is locked?
-         * Should this be fixed?
-         */
+       /*
+        * XXX: Setting the lov ea is not locked but setting the attr is locked?
+        * Should this be fixed?
+        */
        CDEBUG(D_OTHER, "ea %p/%u, cr_flags "LPO64", no_create %u\n",
               spec->u.sp_ea.eadata, spec->u.sp_ea.eadatalen,
               spec->sp_cr_flags, spec->no_create);
@@ -1852,7 +1853,8 @@ static int mdd_create_data(const struct lu_env *env, struct md_object *pobj,
        rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, son, handle);
 
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
+
 out_free:
        RETURN(rc);
 }
@@ -2343,7 +2345,8 @@ static int mdd_index_delete(const struct lu_env *env,
        if (rc)
                GOTO(stop, rc);
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
+
        RETURN(rc);
 }
 
@@ -2485,8 +2488,6 @@ static int mdd_create(const struct lu_env *env, struct md_object *pobj,
        EXIT;
 err_insert:
        if (rc != 0) {
-               int rc2;
-
                if (spec->sp_cr_flags & MDS_OPEN_VOLATILE)
                        rc2 = __mdd_orphan_del(env, son, handle);
                else
@@ -3055,7 +3056,7 @@ cleanup:
                                            ltname, lsname, handle);
 
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
 
 out_pending:
        mdd_object_put(env, mdd_sobj);
@@ -3317,7 +3318,6 @@ static int mdd_migrate_xattrs(const struct lu_env *env,
        int                     list_xsize;
        struct lu_buf           list_xbuf;
        int                     rc;
-       int                     rc1;
 
        /* retrieve xattr list from the old object */
        list_xsize = mdo_xattr_list(env, mdd_sobj, &LU_BUF_NULL);
@@ -3392,9 +3392,7 @@ static int mdd_migrate_xattrs(const struct lu_env *env,
                if (rc != 0)
                        GOTO(stop_trans, rc);
 stop_trans:
-               rc1 = mdd_trans_stop(env, mdd, rc, handle);
-               if (rc == 0)
-                       rc = rc1;
+               rc = mdd_trans_stop(env, mdd, rc, handle);
                if (rc != 0)
                        GOTO(out, rc);
 next:
@@ -3596,13 +3594,8 @@ static int mdd_migrate_create(const struct lu_env *env,
        la_flag->la_flags = la->la_flags | LUSTRE_IMMUTABLE_FL;
        rc = mdo_attr_set(env, mdd_sobj, la_flag, handle);
 stop_trans:
-       if (handle != NULL) {
-               int rc1;
-
-               rc1 = mdd_trans_stop(env, mdd, rc, handle);
-               if (rc == 0)
-                       rc = rc1;
-       }
+       if (handle != NULL)
+               rc = mdd_trans_stop(env, mdd, rc, handle);
 out_free:
        if (lmm_buf.lb_buf != NULL)
                OBD_FREE(lmm_buf.lb_buf, lmm_buf.lb_len);
@@ -3619,9 +3612,9 @@ static int mdd_migrate_entries(const struct lu_env *env,
        struct thandle          *handle;
        struct dt_it            *it;
        const struct dt_it_ops  *iops;
-       int                      rc;
        int                      result;
        struct lu_dirent        *ent;
+       int                      rc;
        ENTRY;
 
        OBD_ALLOC(ent, NAME_MAX + sizeof(*ent) + 1);
@@ -3657,7 +3650,6 @@ static int mdd_migrate_entries(const struct lu_env *env,
                int                     recsize;
                int                     is_dir;
                bool                    target_exist = false;
-               int                     rc1;
 
                len = iops->key_size(env, it);
                if (len == 0)
@@ -3791,10 +3783,7 @@ static int mdd_migrate_entries(const struct lu_env *env,
 out_put:
                mdd_write_unlock(env, child);
                mdd_object_put(env, child);
-               rc1 = mdd_trans_stop(env, mdd, rc, handle);
-               if (rc == 0)
-                       rc = rc1;
-
+               rc = mdd_trans_stop(env, mdd, rc, handle);
                if (rc != 0)
                        GOTO(out, rc);
 next:
@@ -4089,7 +4078,7 @@ out_unlock:
        mdd_write_unlock(env, mdd_sobj);
 
 stop_trans:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
 
        RETURN(rc);
 }

diff --git a/lustre/mdd/mdd_object.c b/lustre/mdd/mdd_object.c
index 4163c85..8743d96 100644 (file)
--- a/lustre/mdd/mdd_object.c
+++ b/lustre/mdd/mdd_object.c
@@ -699,31 +699,31 @@ int mdd_changelog_data_store(const struct lu_env *env, struct mdd_device *mdd,
 static int mdd_changelog(const struct lu_env *env, enum changelog_rec_type type,
                         int flags, struct md_object *obj)
 {
-        struct thandle *handle;
-        struct mdd_object *mdd_obj = md2mdd_obj(obj);
-        struct mdd_device *mdd = mdo2mdd(obj);
-        int rc;
-        ENTRY;
+       struct thandle *handle;
+       struct mdd_object *mdd_obj = md2mdd_obj(obj);
+       struct mdd_device *mdd = mdo2mdd(obj);
+       int rc;
+       ENTRY;
 
-        handle = mdd_trans_create(env, mdd);
-        if (IS_ERR(handle))
+       handle = mdd_trans_create(env, mdd);
+       if (IS_ERR(handle))
                RETURN(PTR_ERR(handle));
 
        rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
        if (rc)
                GOTO(stop, rc);
 
-        rc = mdd_trans_start(env, mdd, handle);
-        if (rc)
-                GOTO(stop, rc);
+       rc = mdd_trans_start(env, mdd, handle);
+       if (rc)
+               GOTO(stop, rc);
 
-        rc = mdd_changelog_data_store(env, mdd, type, flags, mdd_obj,
-                                      handle);
+       rc = mdd_changelog_data_store(env, mdd, type, flags, mdd_obj,
+                                     handle);
 
 stop:
-        mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
 
-        RETURN(rc);
+       RETURN(rc);
 }
 
 /**
@@ -900,7 +900,8 @@ int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
        GOTO(stop, rc);
 
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
+
        return rc;
 }
 
@@ -1103,7 +1104,7 @@ static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
                                              handle);
 
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
 
        RETURN(rc);
 }
@@ -1143,7 +1144,7 @@ static int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
        struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
        struct mdd_device *mdd = mdo2mdd(obj);
        struct thandle *handle;
-       int  rc;
+       int rc;
        ENTRY;
 
        rc = mdd_la_get(env, mdd_obj, attr);
@@ -1183,9 +1184,9 @@ static int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
                                               handle);
 
 stop:
-        mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
 
-        RETURN(rc);
+       RETURN(rc);
 }
 
 /*
@@ -1559,7 +1560,8 @@ static int mdd_swap_layouts(const struct lu_env *env, struct md_object *obj1,
        EXIT;
 
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
+
        mdd_write_unlock(env, snd_o);
        mdd_write_unlock(env, fst_o);
 
@@ -1707,24 +1709,25 @@ static int mdd_declare_close(const struct lu_env *env,
  * No permission check is needed.
  */
 static int mdd_close(const struct lu_env *env, struct md_object *obj,
-                     struct md_attr *ma, int mode)
+                    struct md_attr *ma, int mode)
 {
-        struct mdd_object *mdd_obj = md2mdd_obj(obj);
-        struct mdd_device *mdd = mdo2mdd(obj);
-        struct thandle    *handle = NULL;
-       int rc, is_orphan = 0;
-        ENTRY;
+       struct mdd_object *mdd_obj = md2mdd_obj(obj);
+       struct mdd_device *mdd = mdo2mdd(obj);
+       struct thandle *handle = NULL;
+       int is_orphan = 0;
+       int rc;
+       ENTRY;
 
-        if (ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_KEEP_ORPHAN) {
+       if (ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_KEEP_ORPHAN) {
                mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
                mdd_obj->mod_count--;
                mdd_write_unlock(env, mdd_obj);
 
-                if (mdd_obj->mod_flags & ORPHAN_OBJ && !mdd_obj->mod_count)
-                        CDEBUG(D_HA, "Object "DFID" is retained in orphan "
-                               "list\n", PFID(mdd_object_fid(mdd_obj)));
-                RETURN(0);
-        }
+               if (mdd_obj->mod_flags & ORPHAN_OBJ && !mdd_obj->mod_count)
+                       CDEBUG(D_HA, "Object "DFID" is retained in orphan "
+                               "list\n", PFID(mdd_object_fid(mdd_obj)));
+               RETURN(0);
+       }
 
        /* mdd_finish_unlink() will always set orphan object as DEAD_OBJ, but
         * it might fail to add the object to orphan list (w/o ORPHAN_OBJ). */
@@ -1833,7 +1836,7 @@ out:
 
 stop:
        if (handle != NULL && !IS_ERR(handle))
-               mdd_trans_stop(env, mdd, rc, handle);
+               rc = mdd_trans_stop(env, mdd, rc, handle);
 
        return rc;
 }

diff --git a/lustre/mdd/mdd_orphans.c b/lustre/mdd/mdd_orphans.c
index e7ce0fd..594b3ae 100644 (file)
--- a/lustre/mdd/mdd_orphans.c
+++ b/lustre/mdd/mdd_orphans.c
@@ -318,13 +318,13 @@ static int orph_index_delete(const struct lu_env *env,
 
 
 static int orphan_object_destroy(const struct lu_env *env,
-                                 struct mdd_object *obj,
-                                 struct dt_key *key)
+                                struct mdd_object *obj,
+                                struct dt_key *key)
 {
-        struct thandle *th = NULL;
-        struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
-        int rc = 0;
-        ENTRY;
+       struct thandle *th = NULL;
+       struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
+       int rc = 0;
+       ENTRY;
 
        th = mdd_trans_create(env, mdd);
        if (IS_ERR(th)) {
@@ -332,22 +332,22 @@ static int orphan_object_destroy(const struct lu_env *env,
                RETURN(PTR_ERR(th));
        }
 
-        rc = orph_declare_index_delete(env, obj, th);
-        if (rc)
-                GOTO(stop, rc);
+       rc = orph_declare_index_delete(env, obj, th);
+       if (rc)
+               GOTO(stop, rc);
 
        rc = mdo_declare_destroy(env, obj, th);
-        if (rc)
-                GOTO(stop, rc);
+       if (rc)
+               GOTO(stop, rc);
 
-        rc = mdd_trans_start(env, mdd, th);
-        if (rc)
-                GOTO(stop, rc);
+       rc = mdd_trans_start(env, mdd, th);
+       if (rc)
+               GOTO(stop, rc);
 
-        mdd_write_lock(env, obj, MOR_TGT_CHILD);
-        if (likely(obj->mod_count == 0)) {
-                mdd_orphan_write_lock(env, mdd);
-                rc = mdd_orphan_delete_obj(env, mdd, key, th);
+       mdd_write_lock(env, obj, MOR_TGT_CHILD);
+       if (likely(obj->mod_count == 0)) {
+               mdd_orphan_write_lock(env, mdd);
+               rc = mdd_orphan_delete_obj(env, mdd, key, th);
                if (rc == 0) {
                        mdo_ref_del(env, obj, th);
                        if (S_ISDIR(mdd_object_type(obj))) {
@@ -356,15 +356,15 @@ static int orphan_object_destroy(const struct lu_env *env,
                        }
                        rc = mdo_destroy(env, obj, th);
                } else
-                        CERROR("could not delete object: rc = %d\n",rc);
-                mdd_orphan_write_unlock(env, mdd);
-        }
-        mdd_write_unlock(env, obj);
+                       CERROR("could not delete object: rc = %d\n", rc);
+               mdd_orphan_write_unlock(env, mdd);
+       }
+       mdd_write_unlock(env, obj);
 
 stop:
-        mdd_trans_stop(env, mdd, 0, th);
+       rc = mdd_trans_stop(env, mdd, 0, th);
 
-        RETURN(rc);
+       RETURN(rc);
 }
 
 /**

diff --git a/lustre/mdd/mdd_permission.c b/lustre/mdd/mdd_permission.c
index 00f14bc..68ebb7a 100644 (file)
--- a/lustre/mdd/mdd_permission.c
+++ b/lustre/mdd/mdd_permission.c
@@ -99,9 +99,10 @@ int mdd_acl_set(const struct lu_env *env, struct mdd_object *obj,
        struct thandle          *handle;
        posix_acl_xattr_header  *head;
        posix_acl_xattr_entry   *entry;
-       int                      rc, entry_count;
+       int                      entry_count;
        bool                     not_equiv, mode_change;
        mode_t                   mode;
+       int                      rc;
        ENTRY;
 
        head = (posix_acl_xattr_header *)(buf->lb_buf);
@@ -163,7 +164,7 @@ int mdd_acl_set(const struct lu_env *env, struct mdd_object *obj,
 unlock:
        mdd_write_unlock(env, obj);
 stop:
-       mdd_trans_stop(env, mdd, rc, handle);
+       rc = mdd_trans_stop(env, mdd, rc, handle);
 
        RETURN(rc);
 }

diff --git a/lustre/mdd/mdd_trans.c b/lustre/mdd/mdd_trans.c
index b5b32f6..23b40d4 100644 (file)
--- a/lustre/mdd/mdd_trans.c
+++ b/lustre/mdd/mdd_trans.c
@@ -63,6 +63,12 @@ int mdd_trans_start(const struct lu_env *env, struct mdd_device *mdd,
 int mdd_trans_stop(const struct lu_env *env, struct mdd_device *mdd,
                   int result, struct thandle *handle)
 {
+       int rc;
+
        handle->th_result = result;
-       return mdd_child_ops(mdd)->dt_trans_stop(env, mdd->mdd_child, handle);
+       rc = mdd_child_ops(mdd)->dt_trans_stop(env, mdd->mdd_child, handle);
+
+       /* if operation failed, return \a result, otherwise return status of
+        * dt_trans_stop */
+       return result ?: rc;
 }

diff --git a/lustre/obdclass/dt_object.c b/lustre/obdclass/dt_object.c
index c955f8e..f44f879 100644 (file)
--- a/lustre/obdclass/dt_object.c
+++ b/lustre/obdclass/dt_object.c
@@ -115,6 +115,9 @@ int dt_txn_hook_stop(const struct lu_env *env, struct thandle *th)
        if (th->th_local)
                return 0;
 
+       if (OBD_FAIL_CHECK(OBD_FAIL_DT_TXN_STOP))
+               return -EIO;
+
        list_for_each_entry(cb, &dev->dd_txn_callbacks, dtc_linkage) {
                struct thandle *dtc_th = th;
 

diff --git a/lustre/tests/sanity.sh b/lustre/tests/sanity.sh
index c2d49ac..f4cbf84 100755 (executable)
--- a/lustre/tests/sanity.sh
+++ b/lustre/tests/sanity.sh
@@ -15402,6 +15402,29 @@ test_406() {
 }
 run_test 406 "DNE support fs default striping"
 
+test_407() {
+       [ $MDSCOUNT -lt 2 ] && skip "needs >= 2 MDTs" && return
+
+       [[ $(lustre_version_code $SINGLEMDS) -lt $(version_code 2.8.55) ]] &&
+               skip "Need MDS version at least 2.8.55" && return
+
+       $LFS mkdir -i 0 -c 1 $DIR/$tdir.0 ||
+               error "$LFS mkdir -i 0 -c 1 $tdir.0 failed"
+       $LFS mkdir -i 1 -c 1 $DIR/$tdir.1 ||
+               error "$LFS mkdir -i 1 -c 1 $tdir.1 failed"
+       touch $DIR/$tdir.0/$tfile.0 || error "touch $tdir.0/$tfile.0 failed"
+
+       #define OBD_FAIL_DT_TXN_STOP    0x2019
+       for idx in $(seq $MDSCOUNT); do
+               do_facet mds$idx "lctl set_param fail_loc=0x2019"
+       done
+       $LFS mkdir -c 2 $DIR/$tdir && error "$LFS mkdir -c 2 $tdir should fail"
+       mv $DIR/$tdir.0/$tfile.0 $DIR/$tdir.1/$tfile.1 &&
+               error "mv $tdir.0/$tfile.0 $tdir.1/$tfile.1 should fail"
+       true
+}
+run_test 407 "transaction fail should cause operation fail"
+
 #
 # tests that do cleanup/setup should be run at the end
 #