Whamcloud - gitweb
LU-9753 ofd: 64-bits diff variable to avoid overflow
[fs/lustre-release.git] / lustre / target / update_trans.c
index 28bf6be..36ab331 100644 (file)
@@ -20,7 +20,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2014, Intel Corporation.
+ * Copyright (c) 2015, 2016, Intel Corporation.
  */
 /*
  * lustre/target/update_trans.c
@@ -72,8 +72,7 @@ static void top_multiple_thandle_dump(struct top_multiple_thandle *tmt,
        struct sub_thandle      *st;
 
        LASSERT(tmt->tmt_magic == TOP_THANDLE_MAGIC);
-       CDEBUG(mask, "%s tmt %p refcount %d committed %d result %d"
-              "batchid "LPU64"\n",
+       CDEBUG(mask, "%s tmt %p refcount %d committed %d result %d batchid %llu\n",
               tmt->tmt_master_sub_dt ?
               tmt->tmt_master_sub_dt->dd_lu_dev.ld_obd->obd_name :
               "NULL",
@@ -81,12 +80,17 @@ static void top_multiple_thandle_dump(struct top_multiple_thandle *tmt,
               tmt->tmt_result, tmt->tmt_batchid);
 
        list_for_each_entry(st, &tmt->tmt_sub_thandle_list, st_sub_list) {
-               CDEBUG(mask, "st %p obd %s committed %d sub_th %p "
-                      " cookie "DOSTID": %u\n",
+               struct sub_thandle_cookie *stc;
+
+               CDEBUG(mask, "st %p obd %s committed %d stopped %d sub_th %p\n",
                       st, st->st_dt->dd_lu_dev.ld_obd->obd_name,
-                      st->st_committed, st->st_sub_th,
-                      POSTID(&st->st_cookie.lgc_lgl.lgl_oi),
-                      st->st_cookie.lgc_index);
+                      st->st_committed, st->st_stopped, st->st_sub_th);
+
+               list_for_each_entry(stc, &st->st_cookie_list, stc_list) {
+                       CDEBUG(mask, " cookie "DFID".%u\n",
+                              PFID(&stc->stc_cookie.lgc_lgl.lgl_oi.oi_fid),
+                              stc->stc_cookie.lgc_index);
+               }
        }
 }
 
@@ -99,33 +103,46 @@ static void top_multiple_thandle_dump(struct top_multiple_thandle *tmt,
  * \param[in] env      execution environment
  * \param[in] record   update records being written
  * \param[in] sub_th   sub transaction handle
+ * \param[in] record_size total update record size
  *
  * \retval             0 if writing succeeds
  * \retval             negative errno if writing fails
  */
 static int sub_declare_updates_write(const struct lu_env *env,
                                     struct llog_update_record *record,
-                                    struct thandle *sub_th)
+                                    struct thandle *sub_th, size_t record_size)
 {
        struct llog_ctxt        *ctxt;
        struct dt_device        *dt = sub_th->th_dev;
+       int                     left = record_size;
        int rc;
 
        /* If ctxt is NULL, it means not need to write update,
         * 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) {
+       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);
+       rc = llog_declare_add(env, ctxt->loc_handle,
+                             &record->lur_hdr, sub_th);
+       if (rc < 0)
+               GOTO(out_put, rc);
+
+       while (left > ctxt->loc_chunk_size) {
+               rc = llog_declare_add(env, ctxt->loc_handle,
+                                     &record->lur_hdr, sub_th);
+               if (rc < 0)
+                       GOTO(out_put, rc);
+
+               left -= ctxt->loc_chunk_size;
+       }
 
+out_put:
        llog_ctxt_put(ctxt);
 
        return rc;
@@ -148,21 +165,31 @@ static int sub_declare_updates_write(const struct lu_env *env,
  */
 static int sub_updates_write(const struct lu_env *env,
                             struct llog_update_record *record,
-                            struct thandle *sub_th,
-                            struct llog_cookie *cookie)
+                            struct sub_thandle *sub_th)
 {
-       struct dt_device        *dt = sub_th->th_dev;
-       struct llog_ctxt        *ctxt;
-       int                     rc;
+       struct dt_device *dt = sub_th->st_dt;
+       struct llog_ctxt *ctxt;
+       struct llog_update_record *lur = NULL;
+       __u32 update_count = 0;
+       __u32 param_count = 0;
+       __u32 last_update_count = 0;
+       __u32 last_param_count = 0;
+       char *start;
+       char *cur;
+       char *next;
+       struct sub_thandle_cookie *stc;
+       size_t reclen;
+       bool eof = false;
+       int rc;
        ENTRY;
 
        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) {
+       /* 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);
        }
@@ -174,16 +201,120 @@ static int sub_updates_write(const struct lu_env *env,
                 "lrh_len %u record_size %zu\n", record->lur_hdr.lrh_len,
                 llog_update_record_size(record));
 
-       rc = llog_add(env, ctxt->loc_handle, &record->lur_hdr,
-                     cookie, sub_th);
-       llog_ctxt_put(ctxt);
+       if (likely(record->lur_hdr.lrh_len <= ctxt->loc_chunk_size)) {
+               OBD_ALLOC_PTR(stc);
+               if (stc == NULL)
+                       GOTO(llog_put, rc = -ENOMEM);
+               INIT_LIST_HEAD(&stc->stc_list);
+
+               rc = llog_add(env, ctxt->loc_handle, &record->lur_hdr,
+                             &stc->stc_cookie, sub_th->st_sub_th);
+
+               CDEBUG(D_INFO, "%s: Add update log "DFID".%u: rc = %d\n",
+                      dt->dd_lu_dev.ld_obd->obd_name,
+                      PFID(&stc->stc_cookie.lgc_lgl.lgl_oi.oi_fid),
+                      stc->stc_cookie.lgc_index, rc);
+
+               if (rc > 0) {
+                       list_add(&stc->stc_list, &sub_th->st_cookie_list);
+                       rc = 0;
+               } else {
+                       OBD_FREE_PTR(stc);
+               }
 
-       CDEBUG(D_INFO, "%s: Add update log "DOSTID":%u.\n",
-              dt->dd_lu_dev.ld_obd->obd_name,
-              POSTID(&cookie->lgc_lgl.lgl_oi), cookie->lgc_index);
+               GOTO(llog_put, rc);
+       }
 
-       if (rc > 0)
-               rc = 0;
+       /* Split the records into chunk_size update record */
+       OBD_ALLOC_LARGE(lur, ctxt->loc_chunk_size);
+       if (lur == NULL)
+               GOTO(llog_put, rc = -ENOMEM);
+
+       memcpy(lur, &record->lur_hdr, sizeof(record->lur_hdr));
+       lur->lur_update_rec.ur_update_count = 0;
+       lur->lur_update_rec.ur_param_count = 0;
+       start = (char *)&record->lur_update_rec.ur_ops;
+       cur = next = start;
+       do {
+               if (update_count < record->lur_update_rec.ur_update_count)
+                       next = (char *)update_op_next_op(
+                                               (struct update_op *)cur);
+               else if (param_count < record->lur_update_rec.ur_param_count)
+                       next = (char *)update_param_next_param(
+                                               (struct update_param *)cur);
+               else
+                       eof = true;
+
+               /*
+                * If its size > llog chunk_size, then write current chunk to
+                * the update llog, NB the padding should >= LLOG_MIN_REC_SIZE.
+                *
+                * So check padding length is either >= LLOG_MIN_REC_SIZE or is
+                * 0 (record length just matches the chunk size).
+                */
+               reclen = __llog_update_record_size(
+                               __update_records_size(next - start));
+               if ((reclen + LLOG_MIN_REC_SIZE <= ctxt->loc_chunk_size ||
+                    reclen == ctxt->loc_chunk_size) &&
+                   !eof) {
+                       cur = next;
+
+                       if (update_count <
+                           record->lur_update_rec.ur_update_count)
+                               update_count++;
+                       else if (param_count <
+                                record->lur_update_rec.ur_param_count)
+                               param_count++;
+                       continue;
+               }
+
+               lur->lur_update_rec.ur_update_count = update_count -
+                                                     last_update_count;
+               lur->lur_update_rec.ur_param_count = param_count -
+                                                    last_param_count;
+               memcpy(&lur->lur_update_rec.ur_ops, start, cur - start);
+               lur->lur_hdr.lrh_len = llog_update_record_size(lur);
+
+               LASSERT(lur->lur_hdr.lrh_len ==
+                        __llog_update_record_size(
+                               __update_records_size(cur - start)));
+               LASSERT(lur->lur_hdr.lrh_len <= ctxt->loc_chunk_size);
+
+               update_records_dump(&lur->lur_update_rec, D_INFO, true);
+
+               OBD_ALLOC_PTR(stc);
+               if (stc == NULL)
+                       GOTO(llog_put, rc = -ENOMEM);
+               INIT_LIST_HEAD(&stc->stc_list);
+
+               rc = llog_add(env, ctxt->loc_handle, &lur->lur_hdr,
+                             &stc->stc_cookie, sub_th->st_sub_th);
+
+               CDEBUG(D_INFO, "%s: Add update log "DFID".%u: rc = %d\n",
+                       dt->dd_lu_dev.ld_obd->obd_name,
+                       PFID(&stc->stc_cookie.lgc_lgl.lgl_oi.oi_fid),
+                       stc->stc_cookie.lgc_index, rc);
+
+               if (rc > 0) {
+                       list_add(&stc->stc_list, &sub_th->st_cookie_list);
+                       rc = 0;
+               } else {
+                       OBD_FREE_PTR(stc);
+                       GOTO(llog_put, rc);
+               }
+
+               last_update_count = update_count;
+               last_param_count = param_count;
+               start = cur;
+               lur->lur_update_rec.ur_update_count = 0;
+               lur->lur_update_rec.ur_param_count = 0;
+               lur->lur_update_rec.ur_flags |= UPDATE_RECORD_CONTINUE;
+       } while (!eof);
+
+llog_put:
+       if (lur != NULL)
+               OBD_FREE_LARGE(lur, ctxt->loc_chunk_size);
+       llog_ctxt_put(ctxt);
 
        RETURN(rc);
 }
@@ -308,33 +439,21 @@ struct sub_thandle *create_sub_thandle(struct top_multiple_thandle *tmt,
                RETURN(ERR_PTR(-ENOMEM));
 
        INIT_LIST_HEAD(&st->st_sub_list);
+       INIT_LIST_HEAD(&st->st_cookie_list);
        st->st_dt = dt_dev;
 
        list_add(&st->st_sub_list, &tmt->tmt_sub_thandle_list);
        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);
        list_for_each_entry(st, &tmt->tmt_sub_thandle_list, st_sub_list) {
                if (st->st_sub_th == sub_th) {
                        st->st_committed = 1;
@@ -343,6 +462,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;
@@ -355,6 +475,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)
 {
@@ -434,6 +574,7 @@ int sub_thandle_trans_create(const struct lu_env *env,
        st->st_sub_th = sub_th;
 
        sub_th->th_wait_submit = 1;
+       sub_thandle_register_stop_cb(st, top_th->tt_multiple_thandle);
        return 0;
 }
 
@@ -503,7 +644,8 @@ static int declare_updates_write(const struct lu_env *env,
                if (st->st_sub_th == NULL)
                        continue;
 
-               rc = sub_declare_updates_write(env, record, st->st_sub_th);
+               rc = sub_declare_updates_write(env, record, st->st_sub_th,
+                                              tmt->tmt_record_size);
                if (rc < 0)
                        break;
        }
@@ -522,6 +664,7 @@ static void distribute_txn_assign_batchid(struct top_multiple_thandle *new)
 {
        struct target_distribute_txn_data *tdtd;
        struct dt_device *dt = new->tmt_master_sub_dt;
+       struct sub_thandle *st;
 
        LASSERT(dt != NULL);
        tdtd = dt2lu_dev(dt)->ld_site->ls_tgt->lut_tdtd;
@@ -529,6 +672,10 @@ static void distribute_txn_assign_batchid(struct top_multiple_thandle *new)
        new->tmt_batchid = tdtd->tdtd_batchid++;
        list_add_tail(&new->tmt_commit_list, &tdtd->tdtd_list);
        spin_unlock(&tdtd->tdtd_batchid_lock);
+       list_for_each_entry(st, &new->tmt_sub_thandle_list, st_sub_list) {
+               if (st->st_sub_th != NULL)
+                       sub_thandle_register_commit_cb(st, new);
+       }
        top_multiple_thandle_get(new);
        top_multiple_thandle_dump(new, D_INFO);
 }
@@ -545,6 +692,7 @@ void distribute_txn_insert_by_batchid(struct top_multiple_thandle *new)
        struct dt_device *dt = new->tmt_master_sub_dt;
        struct top_multiple_thandle *tmt;
        struct target_distribute_txn_data *tdtd;
+       struct sub_thandle *st;
        bool    at_head = false;
 
        LASSERT(dt != NULL);
@@ -562,6 +710,12 @@ void distribute_txn_insert_by_batchid(struct top_multiple_thandle *new)
                list_add(&new->tmt_commit_list, &tdtd->tdtd_list);
        }
        spin_unlock(&tdtd->tdtd_batchid_lock);
+
+       list_for_each_entry(st, &new->tmt_sub_thandle_list, st_sub_list) {
+               if (st->st_sub_th != NULL)
+                       sub_thandle_register_commit_cb(st, new);
+       }
+
        top_multiple_thandle_get(new);
        top_multiple_thandle_dump(new, D_INFO);
        if (new->tmt_committed && at_head)
@@ -634,6 +788,11 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
        ENTRY;
 
        if (tmt == NULL) {
+               if (th->th_sync)
+                       top_th->tt_master_sub_thandle->th_sync = th->th_sync;
+               if (th->th_local)
+                       top_th->tt_master_sub_thandle->th_local = th->th_local;
+               top_th->tt_master_sub_thandle->th_tags = th->th_tags;
                rc = dt_trans_start(env, top_th->tt_master_sub_thandle->th_dev,
                                    top_th->tt_master_sub_thandle);
                RETURN(rc);
@@ -647,16 +806,18 @@ 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;
-               st->st_sub_th->th_local = th->th_local;
+               if (th->th_sync)
+                       st->st_sub_th->th_sync = th->th_sync;
+               if (th->th_local)
+                       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,
                                    st->st_sub_th);
                if (rc != 0)
                        GOTO(out, rc);
 
-               sub_thandle_register_stop_cb(st, tmt);
-               sub_thandle_register_commit_cb(st, tmt);
+               LASSERT(st->st_started == 0);
+               st->st_started = 1;
        }
 out:
        th->th_result = rc;
@@ -782,6 +943,12 @@ int top_trans_stop(const struct lu_env *env, struct dt_device *master_dev,
 
        if (likely(top_th->tt_multiple_thandle == NULL)) {
                LASSERT(master_dev != NULL);
+
+               if (th->th_sync)
+                       top_th->tt_master_sub_thandle->th_sync = th->th_sync;
+               if (th->th_local)
+                       top_th->tt_master_sub_thandle->th_local = th->th_local;
+               top_th->tt_master_sub_thandle->th_tags = th->th_tags;
                rc = dt_trans_stop(env, master_dev,
                                   top_th->tt_master_sub_thandle);
                OBD_FREE_PTR(top_th);
@@ -809,13 +976,13 @@ 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);
                }
 
                lur = tur->tur_update_records;
                /* Write updates to the master MDT */
-               rc = sub_updates_write(env, lur, master_st->st_sub_th,
-                                      &master_st->st_cookie);
+               rc = sub_updates_write(env, lur, master_st);
 
                /* Cleanup the common parameters in the update records,
                 * master transno callback might add more parameters.
@@ -828,6 +995,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);
                }
        }
@@ -836,11 +1004,20 @@ stop_master_trans:
        /* Step 2: Stop the transaction on the master MDT, and fill the
         * 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_local)
+                       master_st->st_sub_th->th_local = th->th_local;
+               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);
+               /* 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);
@@ -877,8 +1054,7 @@ stop_master_trans:
                            st->st_sub_th->th_result < 0)
                                continue;
 
-                       rc = sub_updates_write(env, lur, st->st_sub_th,
-                                              &st->st_cookie);
+                       rc = sub_updates_write(env, lur, st);
                        if (rc < 0) {
                                th->th_result = rc;
                                break;
@@ -892,8 +1068,10 @@ stop_other_trans:
                if (st == master_st || st->st_sub_th == NULL)
                        continue;
 
-               st->st_sub_th->th_sync = th->th_sync;
-               st->st_sub_th->th_local = th->th_local;
+               if (th->th_sync)
+                       st->st_sub_th->th_sync = th->th_sync;
+               if (th->th_local)
+                       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;
                rc = dt_trans_stop(env, st->st_sub_th->th_dev,
@@ -940,8 +1118,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;
@@ -962,6 +1141,7 @@ create_sub_thandle_with_thandle(struct top_thandle *top_th,
        st->st_sub_th = sub_th;
 
        sub_th->th_top = &top_th->tt_super;
+       sub_thandle_register_stop_cb(st, top_th->tt_multiple_thandle);
        return st;
 }
 
@@ -982,6 +1162,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;
@@ -1015,19 +1196,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);
@@ -1052,7 +1243,15 @@ void top_multiple_thandle_destroy(struct top_multiple_thandle *tmt)
        LASSERT(tmt->tmt_magic == TOP_THANDLE_MAGIC);
        list_for_each_entry_safe(st, tmp, &tmt->tmt_sub_thandle_list,
                                 st_sub_list) {
+               struct sub_thandle_cookie *stc;
+               struct sub_thandle_cookie *tmp;
+
                list_del(&st->st_sub_list);
+               list_for_each_entry_safe(stc, tmp, &st->st_cookie_list,
+                                        stc_list) {
+                       list_del(&stc->stc_list);
+                       OBD_FREE_PTR(stc);
+               }
                OBD_FREE_PTR(st);
        }
        OBD_FREE_PTR(tmt);
@@ -1083,23 +1282,28 @@ static int distribute_txn_cancel_records(const struct lu_env *env,
                struct llog_ctxt        *ctxt;
                struct obd_device       *obd;
                struct llog_cookie      *cookie;
+               struct sub_thandle_cookie *stc;
                int rc;
 
-               cookie = &st->st_cookie;
-               if (fid_is_zero(&cookie->lgc_lgl.lgl_oi.oi_fid))
-                       continue;
-
                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))
+                               continue;
 
-               rc = llog_cat_cancel_records(env, ctxt->loc_handle, 1,
-                                            cookie);
+                       rc = llog_cat_cancel_records(env, ctxt->loc_handle, 1,
+                                                    cookie);
+                       CDEBUG(D_HA, "%s: batchid %llu cancel update log "
+                              DFID".%u: rc = %d\n", obd->obd_name,
+                              tmt->tmt_batchid,
+                              PFID(&cookie->lgc_lgl.lgl_oi.oi_fid),
+                              cookie->lgc_index, rc);
+               }
 
                llog_ctxt_put(ctxt);
-               CDEBUG(D_HA, "%s: batchid %llu cancel update log "DOSTID
-                      ".%u : rc = %d\n", obd->obd_name, tmt->tmt_batchid,
-                      POSTID(&cookie->lgc_lgl.lgl_oi), cookie->lgc_index, rc);
        }
 
        RETURN(0);
@@ -1216,6 +1420,7 @@ distribute_txn_commit_batchid_update(const struct lu_env *env,
 
        th = dt_trans_create(env, tdtd->tdtd_lut->lut_bottom);
        if (IS_ERR(th)) {
+               atomic_dec(&tdtd->tdtd_refcount);
                OBD_FREE_PTR(dtbd);
                RETURN(PTR_ERR(th));
        }
@@ -1241,13 +1446,15 @@ distribute_txn_commit_batchid_update(const struct lu_env *env,
        rc = dt_record_write(env, tdtd->tdtd_batchid_obj, &buf,
                             &off, th);
 
-       CDEBUG(D_INFO, "%s: update batchid "LPU64": rc = %d\n",
+       CDEBUG(D_INFO, "%s: update batchid %llu: rc = %d\n",
               tdtd->tdtd_lut->lut_obd->obd_name, batchid, rc);
 
 stop:
        dt_trans_stop(env, tdtd->tdtd_lut->lut_bottom, th);
-       if (rc < 0)
+       if (rc < 0) {
+               atomic_dec(&tdtd->tdtd_refcount);
                OBD_FREE_PTR(dtbd);
+       }
        RETURN(rc);
 }
 
@@ -1315,7 +1522,7 @@ distribute_txn_commit_batchid_init(const struct lu_env *env,
 
 out_put:
        if (rc < 0 && dt_obj != NULL) {
-               lu_object_put(env, &dt_obj->do_lu);
+               dt_object_put(env, dt_obj);
                tdtd->tdtd_batchid_obj = NULL;
        }
        return rc;
@@ -1358,7 +1565,7 @@ static int distribute_txn_commit_thread(void *_arg)
        wake_up(&thread->t_ctl_waitq);
        INIT_LIST_HEAD(&list);
 
-       CDEBUG(D_HA, "%s: start commit thread committed batchid "LPU64"\n",
+       CDEBUG(D_HA, "%s: start commit thread committed batchid %llu\n",
               tdtd->tdtd_lut->lut_obd->obd_name,
               tdtd->tdtd_committed_batchid);
 
@@ -1378,8 +1585,8 @@ static int distribute_txn_commit_thread(void *_arg)
                                list_move_tail(&tmt->tmt_commit_list, &list);
                        } else if (!tdtd->tdtd_lut->lut_obd->obd_recovering) {
                                LASSERTF(tmt->tmt_batchid >= batchid,
-                                        "tmt %p tmt_batchid: "LPU64", batchid "
-                                         LPU64"\n", tmt, tmt->tmt_batchid,
+                                        "tmt %p tmt_batchid: %llu, batchid "
+                                         "%llu\n", tmt, tmt->tmt_batchid,
                                         batchid);
                                /* There are three types of distribution
                                 * transaction result
@@ -1407,28 +1614,15 @@ static int distribute_txn_commit_thread(void *_arg)
                }
                spin_unlock(&tdtd->tdtd_batchid_lock);
 
-               CDEBUG(D_HA, "%s: batchid: "LPU64" committed batchid "
-                      LPU64"\n", tdtd->tdtd_lut->lut_obd->obd_name, batchid,
+               CDEBUG(D_HA, "%s: batchid: %llu committed batchid "
+                      "%llu\n", tdtd->tdtd_lut->lut_obd->obd_name, batchid,
                       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? */
@@ -1493,21 +1687,31 @@ int distribute_txn_init(const struct lu_env *env,
        int                     rc;
        ENTRY;
 
-       spin_lock_init(&tdtd->tdtd_batchid_lock);
        INIT_LIST_HEAD(&tdtd->tdtd_list);
+       INIT_LIST_HEAD(&tdtd->tdtd_replay_finish_list);
+       INIT_LIST_HEAD(&tdtd->tdtd_replay_list);
+       spin_lock_init(&tdtd->tdtd_batchid_lock);
+       spin_lock_init(&tdtd->tdtd_replay_list_lock);
+       tdtd->tdtd_replay_handler = distribute_txn_replay_handle;
+       tdtd->tdtd_replay_ready = 0;
 
        tdtd->tdtd_batchid = lut->lut_last_transno + 1;
 
        init_waitqueue_head(&lut->lut_tdtd_commit_thread.t_ctl_waitq);
        init_waitqueue_head(&tdtd->tdtd_commit_thread_waitq);
+       init_waitqueue_head(&tdtd->tdtd_recovery_threads_waitq);
        atomic_set(&tdtd->tdtd_refcount, 0);
+       atomic_set(&tdtd->tdtd_recovery_threads_count, 0);
 
        tdtd->tdtd_lut = lut;
+       if (lut->lut_bottom->dd_rdonly)
+               RETURN(0);
+
        rc = distribute_txn_commit_batchid_init(env, tdtd);
        if (rc != 0)
                RETURN(rc);
 
-       task = kthread_run(distribute_txn_commit_thread, tdtd, "tdtd-%u",
+       task = kthread_run(distribute_txn_commit_thread, tdtd, "dist_txn-%u",
                           index);
        if (IS_ERR(task))
                RETURN(PTR_ERR(task));
@@ -1545,7 +1749,7 @@ void distribute_txn_fini(const struct lu_env *env,
 
        dtrq_list_destroy(tdtd);
        if (tdtd->tdtd_batchid_obj != NULL) {
-               lu_object_put(env, &tdtd->tdtd_batchid_obj->do_lu);
+               dt_object_put(env, tdtd->tdtd_batchid_obj);
                tdtd->tdtd_batchid_obj = NULL;
        }
 }