LU-7450 osd: call commit_callback if no write updates

If it does not need write updates in some failure cases,
top_trans_stop should also call commit_callback to help
release the top_thandle in the commit list. Otherwise
it will stay in the commit list forever, as well as the
following top thandle, then update logs will be culmulated,
and cause long time recovery.

Signed-off-by: Di Wang <di.wang@intel.com>
Change-Id: I1feaf0bd6d20f14dfabb4572f49818083e697dbb
Reviewed-on: http://review.whamcloud.com/17268
Tested-by: Jenkins
Reviewed-by: James Simmons <uja.ornl@yahoo.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_update.h b/lustre/include/lustre_update.h
index 2ddd928..f6e1d2b 100644 (file)
--- a/lustre/include/lustre_update.h
+++ b/lustre/include/lustre_update.h
@@ -330,7 +330,8 @@ struct sub_thandle {
 
        /* If this sub thandle is committed */
        bool                    st_committed:1,
-                               st_stopped:1;
+                               st_stopped:1,
+                               st_started:1;
 };
 
 struct tx_arg;

diff --git a/lustre/target/update_trans.c b/lustre/target/update_trans.c
index 58bdfe5..a7b42d3 100644 (file)
--- a/lustre/target/update_trans.c
+++ b/lustre/target/update_trans.c
@@ -450,25 +450,11 @@ struct sub_thandle *create_sub_thandle(struct top_multiple_thandle *tmt,
        return st;
 }
 
-/**
- * sub thandle commit callback
- *
- * Mark the sub thandle to be committed and if all sub thandle are committed
- * notify the top thandle.
- *
- * \param[in] env      execution environment
- * \param[in] sub_th   sub thandle being committed
- * \param[in] cb       commit callback
- * \param[in] err      trans result
- */
-static void sub_trans_commit_cb(struct lu_env *env,
-                               struct thandle *sub_th,
-                               struct dt_txn_commit_cb *cb, int err)
+static void sub_trans_commit_cb_internal(struct top_multiple_thandle *tmt,
+                                        struct thandle *sub_th, int err)
 {
        struct sub_thandle      *st;
-       struct top_multiple_thandle *tmt = cb->dcb_data;
        bool                    all_committed = true;
-       ENTRY;
 
        /* Check if all sub thandles are committed */
        spin_lock(&tmt->tmt_sub_lock);
@@ -493,6 +479,26 @@ static void sub_trans_commit_cb(struct lu_env *env,
        RETURN_EXIT;
 }
 
+/**
+ * sub thandle commit callback
+ *
+ * Mark the sub thandle to be committed and if all sub thandle are committed
+ * notify the top thandle.
+ *
+ * \param[in] env      execution environment
+ * \param[in] sub_th   sub thandle being committed
+ * \param[in] cb       commit callback
+ * \param[in] err      trans result
+ */
+static void sub_trans_commit_cb(struct lu_env *env,
+                               struct thandle *sub_th,
+                               struct dt_txn_commit_cb *cb, int err)
+{
+       struct top_multiple_thandle *tmt = cb->dcb_data;
+
+       sub_trans_commit_cb_internal(tmt, sub_th, err);
+}
+
 static void sub_thandle_register_commit_cb(struct sub_thandle *st,
                                    struct top_multiple_thandle *tmt)
 {
@@ -813,6 +819,9 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
                                    st->st_sub_th);
                if (rc != 0)
                        GOTO(out, rc);
+
+               LASSERT(st->st_started == 0);
+               st->st_started = 1;
        }
 out:
        th->th_result = rc;
@@ -971,6 +980,7 @@ int top_trans_stop(const struct lu_env *env, struct dt_device *master_dev,
                        CERROR("%s: cannot prepare updates: rc = %d\n",
                               master_dev->dd_lu_dev.ld_obd->obd_name, rc);
                        th->th_result = rc;
+                       write_updates = false;
                        GOTO(stop_master_trans, rc);
                }
 
@@ -989,6 +999,7 @@ int top_trans_stop(const struct lu_env *env, struct dt_device *master_dev,
                        CERROR("%s: write updates failed: rc = %d\n",
                               master_dev->dd_lu_dev.ld_obd->obd_name, rc);
                        th->th_result = rc;
+                       write_updates = false;
                        GOTO(stop_master_trans, rc);
                }
        }
@@ -1004,6 +1015,13 @@ stop_master_trans:
                master_st->st_sub_th->th_tags = th->th_tags;
                master_st->st_sub_th->th_result = th->th_result;
                rc = dt_trans_stop(env, master_st->st_dt, master_st->st_sub_th);
+               /* If it does not write_updates, then we call submit callback
+                * here, otherwise callback is done through
+                * osd(osp)_trans_commit_cb() */
+               if (!master_st->st_started &&
+                   !list_empty(&tmt->tmt_commit_list))
+                       sub_trans_commit_cb_internal(tmt,
+                                               master_st->st_sub_th, rc);
                if (rc < 0) {
                        th->th_result = rc;
                        GOTO(stop_other_trans, rc);