LU-6875 update: set st to NULL in error handler

[fs/lustre-release.git] / lustre / target / update_trans.c

diff --git a/lustre/target/update_trans.c b/lustre/target/update_trans.c
index 001f21d..00bc034 100644 (file)
--- a/lustre/target/update_trans.c
+++ b/lustre/target/update_trans.c
@@ -122,11 +122,12 @@ static int sub_declare_updates_write(const struct lu_env *env,
         * for example if the the OSP is used to connect to OST */
        ctxt = llog_get_context(dt->dd_lu_dev.ld_obd,
                                LLOG_UPDATELOG_ORIG_CTXT);
-       LASSERT(ctxt != NULL);
 
        /* Not ready to record updates yet. */
-       if (ctxt->loc_handle == NULL)
-               GOTO(out_put, rc = 0);
+       if (ctxt == NULL || ctxt->loc_handle == NULL) {
+               llog_ctxt_put(ctxt);
+               return 0;
+       }
 
        rc = llog_declare_add(env, ctxt->loc_handle,
                              &record->lur_hdr, sub_th);
@@ -184,12 +185,14 @@ static int sub_updates_write(const struct lu_env *env,
 
        ctxt = llog_get_context(dt->dd_lu_dev.ld_obd,
                                LLOG_UPDATELOG_ORIG_CTXT);
-       LASSERT(ctxt != NULL);
-
-       /* Not ready to record updates yet, usually happens
-        * in error handler path */
-       if (ctxt->loc_handle == NULL)
-               GOTO(llog_put, rc = 0);
+       /* If ctxt == NULL, then it means updates on OST (only happens
+        * during migration), and we do not track those updates for now */
+       /* If ctxt->loc_handle == NULL, then it does not need to record
+        * update, usually happens in error handler path */
+       if (ctxt == NULL || ctxt->loc_handle == NULL) {
+               llog_ctxt_put(ctxt);
+               RETURN(0);
+       }
 
        /* Since the cross-MDT updates will includes both local
         * and remote updates, the update ops count must > 1 */
@@ -468,6 +471,7 @@ static void sub_trans_commit_cb(struct lu_env *env,
        ENTRY;
 
        /* Check if all sub thandles are committed */
+       spin_lock(&tmt->tmt_sub_lock);
        list_for_each_entry(st, &tmt->tmt_sub_thandle_list, st_sub_list) {
                if (st->st_sub_th == sub_th) {
                        st->st_committed = 1;
@@ -476,6 +480,7 @@ static void sub_trans_commit_cb(struct lu_env *env,
                if (!st->st_committed)
                        all_committed = false;
        }
+       spin_unlock(&tmt->tmt_sub_lock);
 
        if (tmt->tmt_result == 0)
                tmt->tmt_result = err;
@@ -781,7 +786,8 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
        list_for_each_entry(st, &tmt->tmt_sub_thandle_list, st_sub_list) {
                if (st->st_sub_th == NULL)
                        continue;
-               st->st_sub_th->th_sync = th->th_sync;
+               if (th->th_sync)
+                       st->st_sub_th->th_sync = th->th_sync;
                st->st_sub_th->th_local = th->th_local;
                st->st_sub_th->th_tags = th->th_tags;
                rc = dt_trans_start(env, st->st_sub_th->th_dev,
@@ -970,7 +976,8 @@ stop_master_trans:
         * master transno in the update logs to other MDT. */
        if (master_st != NULL && master_st->st_sub_th != NULL) {
                master_st->st_sub_th->th_local = th->th_local;
-               master_st->st_sub_th->th_sync = th->th_sync;
+               if (th->th_sync)
+                       master_st->st_sub_th->th_sync = th->th_sync;
                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);
@@ -1024,7 +1031,8 @@ stop_other_trans:
                if (st == master_st || st->st_sub_th == NULL)
                        continue;
 
-               st->st_sub_th->th_sync = th->th_sync;
+               if (th->th_sync)
+                       st->st_sub_th->th_sync = th->th_sync;
                st->st_sub_th->th_local = th->th_local;
                st->st_sub_th->th_tags = th->th_tags;
                st->st_sub_th->th_result = th->th_result;
@@ -1072,8 +1080,9 @@ int top_trans_create_tmt(const struct lu_env *env,
        INIT_LIST_HEAD(&tmt->tmt_sub_thandle_list);
        INIT_LIST_HEAD(&tmt->tmt_commit_list);
        atomic_set(&tmt->tmt_refcount, 1);
-
+       spin_lock_init(&tmt->tmt_sub_lock);
        init_waitqueue_head(&tmt->tmt_stop_waitq);
+
        top_th->tt_multiple_thandle = tmt;
 
        return 0;
@@ -1114,6 +1123,7 @@ struct thandle *thandle_get_sub_by_dt(const struct lu_env *env,
                                      struct dt_device *sub_dt)
 {
        struct sub_thandle      *st = NULL;
+       struct sub_thandle      *master_st = NULL;
        struct top_thandle      *top_th;
        struct thandle          *sub_th = NULL;
        int                     rc = 0;
@@ -1147,19 +1157,29 @@ struct thandle *thandle_get_sub_by_dt(const struct lu_env *env,
                /* Add master sub th to the top trans list */
                tmt->tmt_master_sub_dt =
                        top_th->tt_master_sub_thandle->th_dev;
-               st = create_sub_thandle_with_thandle(top_th,
-                               top_th->tt_master_sub_thandle);
-               if (IS_ERR(st))
-                       GOTO(stop_trans, rc = PTR_ERR(st));
+               master_st = create_sub_thandle_with_thandle(top_th,
+                                       top_th->tt_master_sub_thandle);
+               if (IS_ERR(master_st)) {
+                       rc = PTR_ERR(master_st);
+                       master_st = NULL;
+                       GOTO(stop_trans, rc);
+               }
        }
 
        /* create and init sub th to the top trans list */
        st = create_sub_thandle_with_thandle(top_th, sub_th);
+       if (IS_ERR(st)) {
+               rc = PTR_ERR(st);
+               st = NULL;
+               GOTO(stop_trans, rc);
+       }
        st->st_sub_th->th_wait_submit = 1;
 stop_trans:
        if (rc < 0) {
-               if (st != NULL)
-                       OBD_FREE_PTR(st);
+               if (master_st != NULL) {
+                       list_del(&master_st->st_sub_list);
+                       OBD_FREE_PTR(master_st);
+               }
                sub_th->th_result = rc;
                dt_trans_stop(env, sub_dt, sub_th);
                sub_th = ERR_PTR(rc);
@@ -1228,7 +1248,8 @@ static int distribute_txn_cancel_records(const struct lu_env *env,
 
                obd = st->st_dt->dd_lu_dev.ld_obd;
                ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
-               LASSERT(ctxt);
+               if (ctxt == NULL)
+                       continue;
                list_for_each_entry(stc, &st->st_cookie_list, stc_list) {
                        cookie = &stc->stc_cookie;
                        if (fid_is_zero(&cookie->lgc_lgl.lgl_oi.oi_fid))
@@ -1556,23 +1577,10 @@ static int distribute_txn_commit_thread(void *_arg)
                       tdtd->tdtd_committed_batchid);
                /* update globally committed on a storage */
                if (batchid > tdtd->tdtd_committed_batchid) {
-                       distribute_txn_commit_batchid_update(&env, tdtd,
+                       rc = distribute_txn_commit_batchid_update(&env, tdtd,
                                                             batchid);
-                       spin_lock(&tdtd->tdtd_batchid_lock);
-                       if (batchid > tdtd->tdtd_batchid) {
-                               /* This might happen during recovery,
-                                * batchid is initialized as last transno,
-                                * and the batchid in the update records
-                                * on other MDTs might be bigger than
-                                * the batchid, so we need update it to
-                                * avoid duplicate batchid. */
-                               CDEBUG(D_HA, "%s update batchid from "LPU64
-                                      " to "LPU64"\n",
-                                      tdtd->tdtd_lut->lut_obd->obd_name,
-                                      tdtd->tdtd_batchid, batchid);
-                               tdtd->tdtd_batchid = batchid;
-                       }
-                       spin_unlock(&tdtd->tdtd_batchid_lock);
+                       if (rc == 0)
+                               batchid = 0;
                }
                /* cancel the records for committed batchid's */
                /* XXX: should we postpone cancel's till the end of recovery? */