X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Ftarget%2Fupdate_trans.c;h=b8150fa5c694cd40cfce338429e79e68b7761c21;hp=e56c1ec7a101f8d5e7a574963bc08096d9483598;hb=f843facff59226d3788d855d1d6948523ab8d944;hpb=4f53536d002c13886210b672b657795baa067144 diff --git a/lustre/target/update_trans.c b/lustre/target/update_trans.c index e56c1ec..b8150fa 100644 --- a/lustre/target/update_trans.c +++ b/lustre/target/update_trans.c @@ -20,7 +20,7 @@ * GPL HEADER END */ /* - * Copyright (c) 2014, Intel Corporation. + * Copyright (c) 2015, 2017, 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,19 @@ 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 started %d stopped %d " + "result %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_started, st->st_stopped, + st->st_result, 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 +105,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 +167,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 +203,123 @@ 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 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 = record->lur_hdr.lrh_len; + if (reclen + LLOG_MIN_REC_SIZE <= ctxt->loc_chunk_size || + reclen == 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); + } + + GOTO(llog_put, rc); + } - 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); + /* 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; + + 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; + } - if (rc > 0) - rc = 0; + 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,30 +444,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] sub_th sub thandle being committed. - */ -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; @@ -340,6 +467,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; @@ -352,6 +480,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) { @@ -364,6 +512,50 @@ static void sub_thandle_register_commit_cb(struct sub_thandle *st, } /** + * Sub thandle stop call back + * + * After sub thandle is stopped, it will call this callback to notify + * the top thandle. + * + * \param[in] th sub thandle to be stopped + * \param[in] rc result of sub trans + */ +static void sub_trans_stop_cb(struct lu_env *env, + struct thandle *sub_th, + struct dt_txn_commit_cb *cb, int err) +{ + struct sub_thandle *st; + struct top_multiple_thandle *tmt = cb->dcb_data; + ENTRY; + + spin_lock(&tmt->tmt_sub_lock); + list_for_each_entry(st, &tmt->tmt_sub_thandle_list, st_sub_list) { + if (st->st_stopped) + continue; + + if (st->st_dt == sub_th->th_dev) { + st->st_stopped = 1; + st->st_result = err; + break; + } + } + spin_unlock(&tmt->tmt_sub_lock); + + wake_up(&tmt->tmt_stop_waitq); + RETURN_EXIT; +} + +static void sub_thandle_register_stop_cb(struct sub_thandle *st, + struct top_multiple_thandle *tmt) +{ + st->st_stop_dcb.dcb_func = sub_trans_stop_cb; + st->st_stop_dcb.dcb_data = tmt; + st->st_stop_dcb.dcb_flags = DCB_TRANS_STOP; + INIT_LIST_HEAD(&st->st_stop_dcb.dcb_linkage); + dt_trans_cb_add(st->st_sub_th, &st->st_stop_dcb); +} + +/** * Create sub thandle * * Create transaction handle for sub_thandle @@ -389,10 +581,10 @@ 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; } - /** * Create the top transaction. * @@ -427,8 +619,6 @@ top_trans_create(const struct lu_env *env, struct dt_device *master_dev) child_th->th_top = &top_th->tt_super; child_th->th_wait_submit = 1; top_th->tt_master_sub_thandle = child_th; - - top_th->tt_super.th_tags |= child_th->th_tags; } return &top_th->tt_super; } @@ -459,7 +649,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; } @@ -478,6 +669,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; @@ -485,6 +677,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); } @@ -501,6 +697,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); @@ -518,6 +715,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) @@ -590,6 +793,10 @@ 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; rc = dt_trans_start(env, top_th->tt_master_sub_thandle->th_dev, top_th->tt_master_sub_thandle); RETURN(rc); @@ -603,15 +810,17 @@ 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; - st->st_sub_th->th_tags = th->th_tags; + 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; rc = dt_trans_start(env, st->st_sub_th->th_dev, st->st_sub_th); if (rc != 0) GOTO(out, rc); - sub_thandle_register_commit_cb(st, tmt); + LASSERT(st->st_started == 0); + st->st_started = 1; } out: th->th_result = rc; @@ -659,6 +868,57 @@ static bool top_check_write_updates(struct top_thandle *top_th) } /** + * Check if top transaction is stopped + * + * Check if top transaction is stopped, only if all sub transaction + * is stopped, then the top transaction is stopped. + * + * \param [in] top_th top thandle + * + * \retval true if the top transaction is stopped. + * \retval false if the top transaction is not stopped. + */ +static bool top_trans_is_stopped(struct top_thandle *top_th) +{ + struct top_multiple_thandle *tmt; + struct sub_thandle *st; + bool all_stopped = true; + + tmt = top_th->tt_multiple_thandle; + list_for_each_entry(st, &tmt->tmt_sub_thandle_list, st_sub_list) { + if (!st->st_stopped && st->st_sub_th != NULL) { + all_stopped = false; + break; + } + + if (st->st_result != 0 && + top_th->tt_super.th_result == 0) + top_th->tt_super.th_result = st->st_result; + } + + return all_stopped; +} + +/** + * Wait result of top transaction + * + * Wait until all sub transaction get its result. + * + * \param [in] top_th top thandle. + * + * \retval the result of top thandle. + */ +static int top_trans_wait_result(struct top_thandle *top_th) +{ + struct l_wait_info lwi = {0}; + + l_wait_event(top_th->tt_multiple_thandle->tmt_stop_waitq, + top_trans_is_stopped(top_th), &lwi); + + RETURN(top_th->tt_super.th_result); +} + +/** * Stop the top transaction. * * Stop the transaction on the master device first, then stop transactions @@ -686,6 +946,11 @@ 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; rc = dt_trans_stop(env, master_dev, top_th->tt_master_sub_thandle); OBD_FREE_PTR(top_th); @@ -713,13 +978,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. @@ -732,6 +997,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); } } @@ -740,12 +1006,22 @@ 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; - master_st->st_sub_th->th_tags = th->th_tags; + 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_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) { + CERROR("%s: stop trans failed: rc = %d\n", + master_dev->dd_lu_dev.ld_obd->obd_name, rc); th->th_result = rc; GOTO(stop_other_trans, rc); } else if (tur != NULL && tur->tur_update_records != NULL) { @@ -781,9 +1057,11 @@ 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) { + CERROR("%s: write updates failed: rc = %d\n", + st->st_dt->dd_lu_dev.ld_obd->obd_name, + rc); th->th_result = rc; break; } @@ -796,17 +1074,25 @@ 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; - st->st_sub_th->th_tags = th->th_tags; + 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_result = th->th_result; rc = dt_trans_stop(env, st->st_sub_th->th_dev, st->st_sub_th); - if (unlikely(rc < 0 && th->th_result == 0)) - th->th_result = rc; + if (rc < 0) { + CERROR("%s: stop trans failed: rc = %d\n", + st->st_dt->dd_lu_dev.ld_obd->obd_name, rc); + if (th->th_result == 0) + th->th_result = rc; + } } + rc = top_trans_wait_result(top_th); + tmt->tmt_result = rc; + /* Balance for the refcount in top_trans_create, Note: if it is NOT * multiple node transaction, the top transaction will be destroyed. */ top_multiple_thandle_put(tmt); @@ -841,6 +1127,8 @@ 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; @@ -860,7 +1148,9 @@ create_sub_thandle_with_thandle(struct top_thandle *top_th, return st; 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; } @@ -881,6 +1171,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; @@ -914,21 +1205,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)); - top_th->tt_master_sub_thandle->th_sync = 1; - top_th->tt_super.th_sync = 1; + 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); @@ -953,7 +1252,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); @@ -984,23 +1291,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); @@ -1117,6 +1429,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)); } @@ -1135,20 +1448,22 @@ distribute_txn_commit_batchid_update(const struct lu_env *env, if (rc < 0) GOTO(stop, rc); - dt_trans_cb_add(th, &dtbd->dtbd_cb); + rc = dt_trans_cb_add(th, &dtbd->dtbd_cb); if (rc < 0) GOTO(stop, rc); 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); } @@ -1188,8 +1503,11 @@ distribute_txn_commit_batchid_init(const struct lu_env *env, dt_obj = dt_find_or_create(env, lut->lut_bottom, fid, dof, attr); - if (IS_ERR(dt_obj)) - GOTO(out_put, rc = PTR_ERR(dt_obj)); + if (IS_ERR(dt_obj)) { + rc = PTR_ERR(dt_obj); + dt_obj = NULL; + GOTO(out_put, rc); + } tdtd->tdtd_batchid_obj = dt_obj; @@ -1213,7 +1531,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; @@ -1256,7 +1574,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); @@ -1276,8 +1594,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 @@ -1305,28 +1623,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? */ @@ -1391,21 +1696,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)); @@ -1443,7 +1758,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; } }