X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Fllog_osd.c;h=e88e30f75e9bb8fa3404c31dc449fcb859926b47;hb=3450f960afa5f20b154c2ece27cd6010c12ecae9;hp=0a09fb530f7cc7b6e13c6ee5bb47450a879f8ecc;hpb=fb80ae7c7601a03c1181de381f067f553e7b8c6f;p=fs%2Flustre-release.git diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c index 0a09fb5..e88e30f 100644 --- a/lustre/obdclass/llog_osd.c +++ b/lustre/obdclass/llog_osd.c @@ -23,7 +23,7 @@ * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2012, 2014 Intel Corporation. + * Copyright (c) 2012, 2015, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -111,6 +111,23 @@ static int llog_osd_create_new_object(const struct lu_env *env, } /** + * Implementation of the llog_operations::lop_exist + * + * This function checks that llog exists on storage. + * + * \param[in] handle llog handle of the current llog + * + * \retval true if llog object exists and is not just destroyed + * \retval false if llog doesn't exist or just destroyed + */ +static int llog_osd_exist(struct llog_handle *handle) +{ + LASSERT(handle->lgh_obj); + return dt_object_exists(handle->lgh_obj) && + !lu_object_is_dying(handle->lgh_obj->do_lu.lo_header); +} + +/** * Write a padding record to the llog * * This function writes a padding record to the end of llog. That may @@ -378,10 +395,19 @@ static int llog_osd_write_rec(const struct lu_env *env, CDEBUG(D_OTHER, "new record %x to "DFID"\n", rec->lrh_type, PFID(lu_object_fid(&o->do_lu))); + if (!llog_osd_exist(loghandle)) + RETURN(-ENOENT); + /* record length should not bigger than */ if (reclen > loghandle->lgh_hdr->llh_hdr.lrh_len) RETURN(-E2BIG); + /* sanity check for fixed-records llog */ + if (idx != LLOG_HEADER_IDX && (llh->llh_flags & LLOG_F_IS_FIXSIZE)) { + LASSERT(llh->llh_size != 0); + LASSERT(llh->llh_size == reclen); + } + rc = dt_attr_get(env, o, &lgi->lgi_attr); if (rc) RETURN(rc); @@ -422,14 +448,44 @@ static int llog_osd_write_rec(const struct lu_env *env, if (idx == LLOG_HEADER_IDX) { /* llog header update */ - LASSERT(reclen >= sizeof(struct llog_log_hdr)); - LASSERT(rec == &llh->llh_hdr); + __u32 *bitmap = LLOG_HDR_BITMAP(llh); lgi->lgi_off = 0; - lgi->lgi_buf.lb_len = reclen; - lgi->lgi_buf.lb_buf = rec; + + /* If it does not indicate the bitmap index + * (reccookie == NULL), then it means update + * the whole update header. Otherwise only + * update header and bits needs to be updated, + * and in DNE cases, it will signaficantly + * shrink the RPC size. + * see distribute_txn_cancel_records()*/ + if (reccookie == NULL) { + lgi->lgi_buf.lb_len = reclen; + lgi->lgi_buf.lb_buf = rec; + rc = dt_record_write(env, o, &lgi->lgi_buf, + &lgi->lgi_off, th); + RETURN(rc); + } + + /* update the header */ + lgi->lgi_buf.lb_len = llh->llh_bitmap_offset; + lgi->lgi_buf.lb_buf = llh; rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); + if (rc != 0) + RETURN(rc); + + /* update the bitmap */ + index = reccookie->lgc_index; + lgi->lgi_off = llh->llh_bitmap_offset + + (index / (sizeof(*bitmap) * 8)) * + sizeof(*bitmap); + lgi->lgi_buf.lb_len = sizeof(*bitmap); + lgi->lgi_buf.lb_buf = + &bitmap[index/(sizeof(*bitmap)*8)]; + rc = dt_record_write(env, o, &lgi->lgi_buf, + &lgi->lgi_off, th); + RETURN(rc); } else if (loghandle->lgh_cur_idx > 0) { /** @@ -448,15 +504,9 @@ static int llog_osd_write_rec(const struct lu_env *env, "len:%u offset %llu\n", POSTID(&loghandle->lgh_id.lgl_oi), idx, rec->lrh_len, (long long)lgi->lgi_off); - } else if (llh->llh_size > 0) { - if (llh->llh_size != rec->lrh_len) { - CERROR("%s: wrong record size, llh_size is %u" - " but record size is %u\n", - o->do_lu.lo_dev->ld_obd->obd_name, - llh->llh_size, rec->lrh_len); - RETURN(-EINVAL); - } - lgi->lgi_off = sizeof(*llh) + (idx - 1) * reclen; + } else if (llh->llh_flags & LLOG_F_IS_FIXSIZE) { + lgi->lgi_off = llh->llh_hdr.lrh_len + + (idx - 1) * reclen; } else { /* This can be result of lgh_cur_idx is not set during * llog processing or llh_size is not set to proper @@ -491,6 +541,14 @@ static int llog_osd_write_rec(const struct lu_env *env, * process them page-at-a-time if needed. If it will cross a chunk * boundary, write in a fake (but referenced) entry to pad the chunk. */ + + + /* simulate ENOSPC when new plain llog is being added to the + * catalog */ + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_LLOG_CREATE_FAILED2) && + llh->llh_flags & LLOG_F_IS_CAT) + RETURN(-ENOSPC); + LASSERT(lgi->lgi_attr.la_valid & LA_SIZE); lgi->lgi_off = lgi->lgi_attr.la_size; left = chunk_size - (lgi->lgi_off & (chunk_size - 1)); @@ -503,9 +561,17 @@ static int llog_osd_write_rec(const struct lu_env *env, RETURN(rc); loghandle->lgh_last_idx++; /* for pad rec */ } - /* if it's the last idx in log file, then return -ENOSPC */ - if (loghandle->lgh_last_idx >= LLOG_HDR_BITMAP_SIZE(llh) - 1) - RETURN(-ENOSPC); + /* if it's the last idx in log file, then return -ENOSPC + * or wrap around if a catalog */ + if (llog_is_full(loghandle) || + unlikely(llh->llh_flags & LLOG_F_IS_CAT && + OBD_FAIL_PRECHECK(OBD_FAIL_CAT_RECORDS) && + loghandle->lgh_last_idx >= cfs_fail_val)) { + if (llh->llh_flags & LLOG_F_IS_CAT) + loghandle->lgh_last_idx = 0; + else + RETURN(-ENOSPC); + } /* increment the last_idx along with llh_tail index, they should * be equal for a llog lifetime */ @@ -523,26 +589,24 @@ static int llog_osd_write_rec(const struct lu_env *env, lrt->lrt_len = rec->lrh_len; lrt->lrt_index = rec->lrh_index; - /* the lgh_hdr_lock protects llog header data from concurrent + /* the lgh_hdr_mutex protects llog header data from concurrent * update/cancel, the llh_count and llh_bitmap are protected */ - down_write(&loghandle->lgh_hdr_lock); + mutex_lock(&loghandle->lgh_hdr_mutex); if (ext2_set_bit(index, LLOG_HDR_BITMAP(llh))) { CERROR("%s: index %u already set in log bitmap\n", o->do_lu.lo_dev->ld_obd->obd_name, index); - up_write(&loghandle->lgh_hdr_lock); + mutex_unlock(&loghandle->lgh_hdr_mutex); LBUG(); /* should never happen */ } llh->llh_count++; - /* XXX It is a bit tricky here, if the log object is local, - * we do not need lock during write here, because if there is - * race, the transaction(jbd2, what about ZFS?) will make sure the - * conflicts will all committed in the same transaction group. - * But for remote object, we need lock the whole process, so to - * set the version of the remote transaction to make sure they - * are being sent in order. (see osp_md_write()) */ - if (!dt_object_remote(o)) - up_write(&loghandle->lgh_hdr_lock); + if (!(llh->llh_flags & LLOG_F_IS_FIXSIZE)) { + /* Update the minimum size of the llog record */ + if (llh->llh_size == 0) + llh->llh_size = reclen; + else if (reclen < llh->llh_size) + llh->llh_size = reclen; + } if (lgi->lgi_attr.la_size == 0) { lgi->lgi_off = 0; @@ -550,8 +614,10 @@ static int llog_osd_write_rec(const struct lu_env *env, lgi->lgi_buf.lb_buf = &llh->llh_hdr; rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); if (rc != 0) - GOTO(out_remote_unlock, rc); + GOTO(out_unlock, rc); } else { + __u32 *bitmap = LLOG_HDR_BITMAP(llh); + /* Note: If this is not initialization (size == 0), then do not * write the whole header (8k bytes), only update header/tail * and bits needs to be updated. Because this update might be @@ -560,21 +626,20 @@ static int llog_osd_write_rec(const struct lu_env *env, * the RPC (1MB limit), if we write 8K for each operation, which * will cost a lot space, and keep us adding more updates to one * update log.*/ - lgi->lgi_off = offsetof(typeof(*llh), llh_count); - lgi->lgi_buf.lb_len = sizeof(llh->llh_count); - lgi->lgi_buf.lb_buf = &llh->llh_count; + lgi->lgi_off = 0; + lgi->lgi_buf.lb_len = llh->llh_bitmap_offset; + lgi->lgi_buf.lb_buf = &llh->llh_hdr; rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); if (rc != 0) - GOTO(out_remote_unlock, rc); + GOTO(out_unlock, rc); - lgi->lgi_off = offsetof(typeof(*llh), - llh_bitmap[index / (sizeof(*llh->llh_bitmap) * 8)]); - lgi->lgi_buf.lb_len = sizeof(*llh->llh_bitmap); - lgi->lgi_buf.lb_buf = - &llh->llh_bitmap[index/(sizeof(*llh->llh_bitmap)*8)]; + lgi->lgi_off = llh->llh_bitmap_offset + + (index / (sizeof(*bitmap) * 8)) * sizeof(*bitmap); + lgi->lgi_buf.lb_len = sizeof(*bitmap); + lgi->lgi_buf.lb_buf = &bitmap[index/(sizeof(*bitmap)*8)]; rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); if (rc != 0) - GOTO(out_remote_unlock, rc); + GOTO(out_unlock, rc); lgi->lgi_off = (unsigned long)LLOG_HDR_TAIL(llh) - (unsigned long)llh; @@ -582,22 +647,29 @@ static int llog_osd_write_rec(const struct lu_env *env, lgi->lgi_buf.lb_buf = LLOG_HDR_TAIL(llh); rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); if (rc != 0) - GOTO(out_remote_unlock, rc); + GOTO(out_unlock, rc); } -out_remote_unlock: +out_unlock: /* unlock here for remote object */ - if (dt_object_remote(o)) - up_write(&loghandle->lgh_hdr_lock); + mutex_unlock(&loghandle->lgh_hdr_mutex); if (rc) GOTO(out, rc); - rc = dt_attr_get(env, o, &lgi->lgi_attr); - if (rc) - GOTO(out, rc); + /* computed index can be used to determine offset for fixed-size + * records. This also allows to handle Catalog wrap around case */ + if (llh->llh_flags & LLOG_F_IS_FIXSIZE) { + lgi->lgi_off = llh->llh_hdr.lrh_len + (index - 1) * reclen; + } else { + rc = dt_attr_get(env, o, &lgi->lgi_attr); + if (rc) + GOTO(out, rc); + + LASSERT(lgi->lgi_attr.la_valid & LA_SIZE); + lgi->lgi_off = max_t(__u64, lgi->lgi_attr.la_size, + lgi->lgi_off); + } - LASSERT(lgi->lgi_attr.la_valid & LA_SIZE); - lgi->lgi_off = max_t(__u64, lgi->lgi_attr.la_size, lgi->lgi_off); lgi->lgi_buf.lb_len = reclen; lgi->lgi_buf.lb_buf = rec; rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); @@ -622,13 +694,17 @@ out_remote_unlock: RETURN(rc); out: /* cleanup llog for error case */ - down_write(&loghandle->lgh_hdr_lock); + mutex_lock(&loghandle->lgh_hdr_mutex); ext2_clear_bit(index, LLOG_HDR_BITMAP(llh)); llh->llh_count--; - up_write(&loghandle->lgh_hdr_lock); + mutex_unlock(&loghandle->lgh_hdr_mutex); /* restore llog last_idx */ - loghandle->lgh_last_idx--; + if (--loghandle->lgh_last_idx == 0 && + (llh->llh_flags & LLOG_F_IS_CAT) && llh->llh_cat_idx != 0) { + /* catalog had just wrap-around case */ + loghandle->lgh_last_idx = LLOG_HDR_BITMAP_SIZE(llh) - 1; + } LLOG_HDR_TAIL(llh)->lrt_index = loghandle->lgh_last_idx; RETURN(rc); @@ -640,14 +716,42 @@ out: * that we are not far enough along the log (because the * actual records are larger than minimum size) we just skip * some more records. + * + * Note: in llog_process_thread, it will use bitmap offset as + * the index to locate the record, which also includs some pad + * records, whose record size is very small, and it also does not + * consider pad record when recording minimum record size (otherwise + * min_record size might be too small), so in some rare cases, + * it might skip too much record for @goal, see llog_osd_next_block(). + * + * When force_mini_rec is true, it means we have to use LLOG_MIN_REC_SIZE + * as the min record size to skip over, usually because in the previous + * try, it skip too much record, see loog_osd_next(prev)_block(). */ -static inline void llog_skip_over(__u64 *off, int curr, int goal, - __u32 chunk_size) +static inline void llog_skip_over(struct llog_handle *lgh, __u64 *off, + int curr, int goal, __u32 chunk_size, + bool force_mini_rec) { - if (goal <= curr) - return; - *off = (*off + (goal - curr - 1) * LLOG_MIN_REC_SIZE) & - ~(chunk_size - 1); + struct llog_log_hdr *llh = lgh->lgh_hdr; + + /* Goal should not bigger than the record count */ + if (goal > lgh->lgh_last_idx) + goal = lgh->lgh_last_idx; + + if (goal > curr) { + if (llh->llh_flags & LLOG_F_IS_FIXSIZE) { + *off = chunk_size + (goal - 1) * llh->llh_size; + } else { + __u64 min_rec_size = LLOG_MIN_REC_SIZE; + + if (llh->llh_size > 0 && !force_mini_rec) + min_rec_size = llh->llh_size; + + *off = *off + (goal - curr - 1) * min_rec_size; + } + } + /* always align with lower chunk boundary*/ + *off &= ~(chunk_size - 1); } /** @@ -704,6 +808,9 @@ static int llog_osd_next_block(const struct lu_env *env, struct dt_device *dt; int rc; __u32 chunk_size; + int last_idx = *cur_idx; + __u64 last_offset = *cur_offset; + bool force_mini_rec = false; ENTRY; @@ -734,7 +841,8 @@ static int llog_osd_next_block(const struct lu_env *env, struct llog_rec_hdr *rec, *last_rec; struct llog_rec_tail *tail; - llog_skip_over(cur_offset, *cur_idx, next_idx, chunk_size); + llog_skip_over(loghandle, cur_offset, *cur_idx, + next_idx, chunk_size, force_mini_rec); /* read up to next llog chunk_size block */ lgi->lgi_buf.lb_len = chunk_size - @@ -743,6 +851,9 @@ static int llog_osd_next_block(const struct lu_env *env, rc = dt_read(env, o, &lgi->lgi_buf, cur_offset); if (rc < 0) { + if (rc == -EBADR && !force_mini_rec) + goto retry; + CERROR("%s: can't read llog block from log "DFID " offset "LPU64": rc = %d\n", o->do_lu.lo_dev->ld_obd->obd_name, @@ -757,10 +868,16 @@ static int llog_osd_next_block(const struct lu_env *env, memset(buf + rc, 0, len - rc); } - if (rc == 0) /* end of file, nothing to do */ + if (rc == 0) { /* end of file, nothing to do */ + if (!force_mini_rec) + goto retry; GOTO(out, rc); + } if (rc < sizeof(*tail)) { + if (!force_mini_rec) + goto retry; + CERROR("%s: invalid llog block at log id "DOSTID"/%u " "offset "LPU64"\n", o->do_lu.lo_dev->ld_obd->obd_name, @@ -788,18 +905,24 @@ static int llog_osd_next_block(const struct lu_env *env, /* this shouldn't happen */ if (tail->lrt_index == 0) { CERROR("%s: invalid llog tail at log id "DOSTID"/%u " - "offset "LPU64"\n", + "offset "LPU64" bytes %d\n", o->do_lu.lo_dev->ld_obd->obd_name, POSTID(&loghandle->lgh_id.lgl_oi), - loghandle->lgh_id.lgl_ogen, *cur_offset); + loghandle->lgh_id.lgl_ogen, *cur_offset, rc); GOTO(out, rc = -EINVAL); } - if (tail->lrt_index < next_idx) + if (tail->lrt_index < next_idx) { + last_idx = *cur_idx; + last_offset = *cur_offset; continue; + } /* sanity check that the start of the new buffer is no farther * than the record that we wanted. This shouldn't happen. */ if (rec->lrh_index > next_idx) { + if (!force_mini_rec && next_idx > last_idx) + goto retry; + CERROR("%s: missed desired record? %u > %u\n", o->do_lu.lo_dev->ld_obd->obd_name, rec->lrh_index, next_idx); @@ -812,6 +935,14 @@ static int llog_osd_next_block(const struct lu_env *env, CLF_VERSION | CLF_RENAME); GOTO(out, rc = 0); + +retry: + /* Note: because there are some pad records in the + * llog, so llog_skip_over() might skip too much + * records, let's try skip again with minimum record */ + force_mini_rec = true; + *cur_offset = last_offset; + *cur_idx = last_idx; } GOTO(out, rc = -EIO); out: @@ -863,8 +994,11 @@ static int llog_osd_prev_block(const struct lu_env *env, dt = lu2dt_dev(o->do_lu.lo_dev); LASSERT(dt); + /* Let's only use mini record size for previous block read + * for now XXX */ cur_offset = chunk_size; - llog_skip_over(&cur_offset, 0, prev_idx, chunk_size); + llog_skip_over(loghandle, &cur_offset, 0, prev_idx, + chunk_size, true); rc = dt_attr_get(env, o, &lgi->lgi_attr); if (rc) @@ -1023,6 +1157,7 @@ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle, dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt; LASSERT(dt); if (ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) { + struct lu_object_conf conf = { 0 }; if (logid != NULL) { logid_to_fid(logid, &lgi->lgi_fid); } else { @@ -1033,9 +1168,11 @@ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle, if (rc < 0) RETURN(rc); rc = 0; + conf.loc_flags = LOC_F_NEW; } - o = dt_locate(env, dt, &lgi->lgi_fid); + o = dt_locate_at(env, dt, &lgi->lgi_fid, + dt->dd_lu_dev.ld_site->ls_top_dev, &conf); if (IS_ERR(o)) RETURN(PTR_ERR(o)); @@ -1114,23 +1251,6 @@ out: } /** - * Implementation of the llog_operations::lop_exist - * - * This function checks that llog exists on storage. - * - * \param[in] handle llog handle of the current llog - * - * \retval true if llog object exists and is not just destroyed - * \retval false if llog doesn't exist or just destroyed - */ -static int llog_osd_exist(struct llog_handle *handle) -{ - LASSERT(handle->lgh_obj); - return (dt_object_exists(handle->lgh_obj) && - !lu_object_is_dying(handle->lgh_obj->do_lu.lo_header)); -} - -/** * Get dir for regular fid log object * * Get directory for regular fid log object, and these regular fid log @@ -1484,14 +1604,11 @@ llog_osd_regular_fid_del_name_entry(const struct lu_env *env, RETURN(rc); } - /** - * Implementation of the llog_operations::lop_destroy + * Implementation of the llog_operations::lop_declare_destroy * - * This function destroys the llog and deletes also entry in the + * This function declare destroys the llog and deletes also entry in the * llog directory in case of named llog. Llog should be opened prior that. - * Destroy method is not part of external transaction and does everything - * inside. * * \param[in] env execution environment * \param[in] loghandle llog handle of the current llog @@ -1499,14 +1616,12 @@ llog_osd_regular_fid_del_name_entry(const struct lu_env *env, * \retval 0 on successful destroy * \retval negative value on error */ -static int llog_osd_destroy(const struct lu_env *env, - struct llog_handle *loghandle) +static int llog_osd_declare_destroy(const struct lu_env *env, + struct llog_handle *loghandle, + struct thandle *th) { struct llog_ctxt *ctxt; struct dt_object *o, *llog_dir = NULL; - struct dt_device *d; - struct thandle *th; - char *name = NULL; int rc; ENTRY; @@ -1517,78 +1632,104 @@ static int llog_osd_destroy(const struct lu_env *env, o = loghandle->lgh_obj; LASSERT(o); - d = lu2dt_dev(o->do_lu.lo_dev); - LASSERT(d); - LASSERT(d == ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt); - - th = dt_trans_create(env, d); - if (IS_ERR(th)) - RETURN(PTR_ERR(th)); - if (loghandle->lgh_name) { llog_dir = llog_osd_dir_get(env, ctxt); if (IS_ERR(llog_dir)) - GOTO(out_trans, rc = PTR_ERR(llog_dir)); + RETURN(PTR_ERR(llog_dir)); - name = loghandle->lgh_name; rc = dt_declare_delete(env, llog_dir, - (struct dt_key *)name, th); - if (rc) - GOTO(out_trans, rc); + (struct dt_key *)loghandle->lgh_name, + th); + if (rc < 0) + GOTO(out_put, rc); } rc = dt_declare_ref_del(env, o, th); if (rc < 0) - GOTO(out_trans, rc); + GOTO(out_put, rc); rc = dt_declare_destroy(env, o, th); - if (rc) - GOTO(out_trans, rc); + if (rc < 0) + GOTO(out_put, rc); if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) { rc = llog_osd_regular_fid_del_name_entry(env, o, th, true); if (rc < 0) - GOTO(out_trans, rc); + GOTO(out_put, rc); } - rc = dt_trans_start_local(env, d, th); - if (rc) - GOTO(out_trans, rc); +out_put: + if (!(IS_ERR_OR_NULL(llog_dir))) + lu_object_put(env, &llog_dir->do_lu); - th->th_wait_submit = 1; + RETURN(rc); +} + + +/** + * Implementation of the llog_operations::lop_destroy + * + * This function destroys the llog and deletes also entry in the + * llog directory in case of named llog. Llog should be opened prior that. + * Destroy method is not part of external transaction and does everything + * inside. + * + * \param[in] env execution environment + * \param[in] loghandle llog handle of the current llog + * + * \retval 0 on successful destroy + * \retval negative value on error + */ +static int llog_osd_destroy(const struct lu_env *env, + struct llog_handle *loghandle, struct thandle *th) +{ + struct llog_ctxt *ctxt; + struct dt_object *o, *llog_dir = NULL; + int rc; + + ENTRY; + + ctxt = loghandle->lgh_ctxt; + LASSERT(ctxt != NULL); + + o = loghandle->lgh_obj; + LASSERT(o != NULL); dt_write_lock(env, o, 0); - if (dt_object_exists(o)) { - if (name) { - dt_read_lock(env, llog_dir, 0); - rc = dt_delete(env, llog_dir, - (struct dt_key *) name, - th); - dt_read_unlock(env, llog_dir); - if (rc) { - CERROR("%s: can't remove llog %s: rc = %d\n", - o->do_lu.lo_dev->ld_obd->obd_name, - name, rc); - GOTO(out_unlock, rc); - } - } - dt_ref_del(env, o, th); - rc = dt_destroy(env, o, th); - if (rc) - GOTO(out_unlock, rc); + if (!dt_object_exists(o)) + GOTO(out_unlock, rc = 0); - if (loghandle->lgh_ctxt->loc_flags & - LLOG_CTXT_FLAG_NORMAL_FID) { - rc = llog_osd_regular_fid_del_name_entry(env, o, th, - false); - if (rc < 0) - GOTO(out_unlock, rc); + if (loghandle->lgh_name) { + llog_dir = llog_osd_dir_get(env, ctxt); + if (IS_ERR(llog_dir)) + GOTO(out_unlock, rc = PTR_ERR(llog_dir)); + + dt_read_lock(env, llog_dir, 0); + rc = dt_delete(env, llog_dir, + (struct dt_key *)loghandle->lgh_name, + th); + dt_read_unlock(env, llog_dir); + if (rc) { + CERROR("%s: can't remove llog %s: rc = %d\n", + o->do_lu.lo_dev->ld_obd->obd_name, + loghandle->lgh_name, rc); + GOTO(out_unlock, rc); } } + + dt_ref_del(env, o, th); + rc = dt_destroy(env, o, th); + if (rc < 0) + GOTO(out_unlock, rc); + + if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) { + rc = llog_osd_regular_fid_del_name_entry(env, o, th, false); + if (rc < 0) + GOTO(out_unlock, rc); + } + out_unlock: dt_write_unlock(env, o); -out_trans: - dt_trans_stop(env, d, th); if (!(IS_ERR_OR_NULL(llog_dir))) lu_object_put(env, &llog_dir->do_lu); RETURN(rc); @@ -1685,6 +1826,7 @@ struct llog_operations llog_osd_ops = { .lop_next_block = llog_osd_next_block, .lop_prev_block = llog_osd_prev_block, .lop_read_header = llog_osd_read_header, + .lop_declare_destroy = llog_osd_declare_destroy, .lop_destroy = llog_osd_destroy, .lop_setup = llog_osd_setup, .lop_cleanup = llog_osd_cleanup, @@ -1702,6 +1844,7 @@ struct llog_operations llog_common_cat_ops = { .lop_next_block = llog_osd_next_block, .lop_prev_block = llog_osd_prev_block, .lop_read_header = llog_osd_read_header, + .lop_declare_destroy = llog_osd_declare_destroy, .lop_destroy = llog_osd_destroy, .lop_setup = llog_osd_setup, .lop_cleanup = llog_osd_cleanup, @@ -1900,7 +2043,16 @@ int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d, lgi->lgi_buf.lb_buf = idarray; rc = dt_declare_record_write(env, o, &lgi->lgi_buf, lgi->lgi_off, th); if (rc) - GOTO(out, rc); + GOTO(out_trans, rc); + + /* For update log, this happens during initialization, + * see lod_sub_prep_llog(), and we need make sure catlog + * file ID is written to catlist file(committed) before + * cross-MDT operation write update records to catlog FILE, + * otherwise, during failover these update records might + * missing */ + if (fid_is_update_log(fid)) + th->th_sync = 1; rc = dt_trans_start_local(env, d, th); if (rc)