X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fobdclass%2Fllog_osd.c;h=a0843012c1d58a678f935f646bdcb38e0bafcac0;hp=b96964a1182f9ef640a9a8ebffcf01e67fb53e22;hb=907a321c9b9e2cd5f5ccf488cc516ba05dee0ad8;hpb=7b6aaef15a6547550c99cdaa538ea8b1d4d87c88 diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c index b96964a..a084301 100644 --- a/lustre/obdclass/llog_osd.c +++ b/lustre/obdclass/llog_osd.c @@ -15,25 +15,28 @@ * * You should have received a copy of the GNU General Public License * version 2 along with this program; If not, see - * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf - * - * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, - * CA 95054 USA or visit www.sun.com if you need additional information or - * have any questions. + * http://www.gnu.org/licenses/gpl-2.0.html * * GPL HEADER END */ /* - * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved. + * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2011, 2012, Intel, Inc. + * Copyright (c) 2012, 2014 Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ * Lustre is a trademark of Sun Microsystems, Inc. + */ +/* + * lustre/obdclass/llog_osd.c + * + * Low level llog routines on top of OSD API * - * lustre/obdclass/llog_osd.c - low level llog routines on top of OSD API + * This file provides set of methods for llog operations on top of + * dt_device. It contains all supported llog_operations interfaces and + * supplimental functions. * * Author: Alexey Zhuravlev * Author: Mikhail Pershin @@ -41,10 +44,6 @@ #define DEBUG_SUBSYSTEM S_LOG -#ifndef EXPORT_SYMTAB -#define EXPORT_SYMTAB -#endif - #include #include #include @@ -53,14 +52,20 @@ #include "llog_internal.h" #include "local_storage.h" -/* - * - multi-chunks or big-declaration approach - * - use unique sequence instead of llog sb tracking unique ids - * - re-use existing environment - * - named llog support (can be used for testing only at the present) - * - llog_origin_connect() work with OSD API +/** + * Implementation of the llog_operations::lop_declare_create + * + * This function is a wrapper over local_storage API function + * local_object_declare_create(). + * + * \param[in] env execution environment + * \param[in] los local_storage for bottom storage device + * \param[in] o dt_object to create + * \param[in] th current transaction handle + * + * \retval 0 on successful declaration of the new object + * \retval negative error if declaration was failed */ - static int llog_osd_declare_new_object(const struct lu_env *env, struct local_oid_storage *los, struct dt_object *o, @@ -76,6 +81,20 @@ static int llog_osd_declare_new_object(const struct lu_env *env, &lgi->lgi_dof, th); } +/** + * Implementation of the llog_operations::lop_create + * + * This function is a wrapper over local_storage API function + * local_object_create(). + * + * \param[in] env execution environment + * \param[in] los local_storage for bottom storage device + * \param[in] o dt_object to create + * \param[in] th current transaction handle + * + * \retval 0 on successful creation of the new object + * \retval negative error if creation was failed + */ static int llog_osd_create_new_object(const struct lu_env *env, struct local_oid_storage *los, struct dt_object *o, @@ -91,10 +110,34 @@ static int llog_osd_create_new_object(const struct lu_env *env, &lgi->lgi_dof, th); } +/** + * Write a padding record to the llog + * + * This function writes a padding record to the end of llog. That may + * be needed if llog contains records of variable size, e.g. config logs + * or changelogs. + * The padding record just aligns llog to the LLOG_CHUNK_SIZE boundary if + * the current record doesn't fit in the remaining space. + * + * It allocates full length to avoid two separate writes for header and tail. + * Such 2-steps scheme needs extra protection and complex error handling. + * + * \param[in] env execution environment + * \param[in] o dt_object to create + * \param[in,out] off pointer to the padding start offset + * \param[in] len padding length + * \param[in] index index of the padding record in a llog + * \param[in] th current transaction handle + * + * \retval 0 on successful padding write + * \retval negative error if write failed + */ static int llog_osd_pad(const struct lu_env *env, struct dt_object *o, loff_t *off, int len, int index, struct thandle *th) { struct llog_thread_info *lgi = llog_info(env); + struct llog_rec_hdr *rec; + struct llog_rec_tail *tail; int rc; ENTRY; @@ -103,97 +146,48 @@ static int llog_osd_pad(const struct lu_env *env, struct dt_object *o, LASSERT(off); LASSERT(len >= LLOG_MIN_REC_SIZE && (len & 0x7) == 0); - lgi->lgi_tail.lrt_len = lgi->lgi_lrh.lrh_len = len; - lgi->lgi_tail.lrt_index = lgi->lgi_lrh.lrh_index = index; - lgi->lgi_lrh.lrh_type = LLOG_PAD_MAGIC; - - lgi->lgi_buf.lb_buf = &lgi->lgi_lrh; - lgi->lgi_buf.lb_len = sizeof(lgi->lgi_lrh); - rc = dt_record_write(env, o, &lgi->lgi_buf, off, th); - if (rc) { - CERROR("%s: error writing padding record: rc = %d\n", - o->do_lu.lo_dev->ld_obd->obd_name, rc); - RETURN(rc); - } + OBD_ALLOC(rec, len); + if (rec == NULL) + RETURN(-ENOMEM); - lgi->lgi_buf.lb_buf = &lgi->lgi_tail; - lgi->lgi_buf.lb_len = sizeof(lgi->lgi_tail); - *off += len - sizeof(lgi->lgi_lrh) - sizeof(lgi->lgi_tail); - rc = dt_record_write(env, o, &lgi->lgi_buf, off, th); - if (rc) - CERROR("%s: error writing padding record: rc = %d\n", - o->do_lu.lo_dev->ld_obd->obd_name, rc); - - RETURN(rc); -} - -static int llog_osd_write_blob(const struct lu_env *env, struct dt_object *o, - struct llog_rec_hdr *rec, void *buf, - loff_t *off, struct thandle *th) -{ - struct llog_thread_info *lgi = llog_info(env); - int buflen = rec->lrh_len; - int rc; - - ENTRY; - - LASSERT(env); - LASSERT(o); - - if (buflen == 0) - CWARN("0-length record\n"); + rec->lrh_len = len; + rec->lrh_index = index; + rec->lrh_type = LLOG_PAD_MAGIC; - CDEBUG(D_OTHER, "write blob with type %x, buf %p/%u at off %llu\n", - rec->lrh_type, buf, buflen, *off); + tail = rec_tail(rec); + tail->lrt_len = len; + tail->lrt_index = index; - if (!buf) { - lgi->lgi_buf.lb_len = buflen; - lgi->lgi_buf.lb_buf = rec; - rc = dt_record_write(env, o, &lgi->lgi_buf, off, th); - if (rc) - CERROR("%s: error writing log record: rc = %d\n", - o->do_lu.lo_dev->ld_obd->obd_name, rc); - RETURN(rc); - } - - /* the buf case */ - rec->lrh_len = sizeof(*rec) + buflen + sizeof(lgi->lgi_tail); - lgi->lgi_buf.lb_len = sizeof(*rec); lgi->lgi_buf.lb_buf = rec; - rc = dt_record_write(env, o, &lgi->lgi_buf, off, th); - if (rc) { - CERROR("%s: error writing log hdr: rc = %d\n", - o->do_lu.lo_dev->ld_obd->obd_name, rc); - GOTO(out, rc); - } - - lgi->lgi_buf.lb_len = buflen; - lgi->lgi_buf.lb_buf = buf; - rc = dt_record_write(env, o, &lgi->lgi_buf, off, th); - if (rc) { - CERROR("%s: error writing log buffer: rc = %d\n", - o->do_lu.lo_dev->ld_obd->obd_name, rc); - GOTO(out, rc); - } - - lgi->lgi_tail.lrt_len = rec->lrh_len; - lgi->lgi_tail.lrt_index = rec->lrh_index; - lgi->lgi_buf.lb_len = sizeof(lgi->lgi_tail); - lgi->lgi_buf.lb_buf = &lgi->lgi_tail; + lgi->lgi_buf.lb_len = len; rc = dt_record_write(env, o, &lgi->lgi_buf, off, th); if (rc) - CERROR("%s: error writing log tail: rc = %d\n", + CERROR("%s: error writing padding record: rc = %d\n", o->do_lu.lo_dev->ld_obd->obd_name, rc); -out: + + OBD_FREE(rec, len); RETURN(rc); } +/** + * Implementation of the llog_operations::lop_read_header + * + * This function reads the current llog header from the bottom storage + * device. + * + * \param[in] env execution environment + * \param[in] handle llog handle of the current llog + * + * \retval 0 on successful header read + * \retval negative error if read failed + */ static int llog_osd_read_header(const struct lu_env *env, struct llog_handle *handle) { struct llog_rec_hdr *llh_hdr; struct dt_object *o; struct llog_thread_info *lgi; + enum llog_flag flags; int rc; ENTRY; @@ -216,6 +210,8 @@ static int llog_osd_read_header(const struct lu_env *env, RETURN(LLOG_EEMPTY); } + flags = handle->lgh_hdr->llh_flags; + lgi->lgi_off = 0; lgi->lgi_buf.lb_buf = handle->lgh_hdr; lgi->lgi_buf.lb_len = LLOG_CHUNK_SIZE; @@ -250,11 +246,31 @@ static int llog_osd_read_header(const struct lu_env *env, RETURN(-EIO); } + handle->lgh_hdr->llh_flags |= (flags & LLOG_F_EXT_MASK); handle->lgh_last_idx = handle->lgh_hdr->llh_tail.lrt_index; RETURN(0); } +/** + * Implementation of the llog_operations::lop_declare_write + * + * This function declares the new record write. + * + * \param[in] env execution environment + * \param[in] loghandle llog handle of the current llog + * \param[in] rec llog record header. This is a real header of the full + * llog record to write. This is the beginning of buffer + * to write, the length of buffer is stored in + * \a rec::lrh_len + * \param[in] idx index of the llog record. If \a idx == -1 then this is + * append case, otherwise \a idx is the index of record + * to modify + * \param[in] th current transaction handle + * + * \retval 0 on successful declaration + * \retval negative error if declaration failed + */ static int llog_osd_declare_write_rec(const struct lu_env *env, struct llog_handle *loghandle, struct llog_rec_hdr *rec, @@ -269,39 +285,60 @@ static int llog_osd_declare_write_rec(const struct lu_env *env, LASSERT(env); LASSERT(th); LASSERT(loghandle); + LASSERT(rec); + LASSERT(rec->lrh_len <= LLOG_CHUNK_SIZE); o = loghandle->lgh_obj; LASSERT(o); + lgi->lgi_buf.lb_len = sizeof(struct llog_log_hdr); + lgi->lgi_buf.lb_buf = NULL; /* each time we update header */ - rc = dt_declare_record_write(env, o, sizeof(struct llog_log_hdr), 0, + rc = dt_declare_record_write(env, o, &lgi->lgi_buf, 0, th); if (rc || idx == 0) /* if error or just header */ RETURN(rc); - if (dt_object_exists(o)) { - rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA); - lgi->lgi_off = lgi->lgi_attr.la_size; - LASSERT(ergo(rc == 0, lgi->lgi_attr.la_valid & LA_SIZE)); - if (rc) - RETURN(rc); - } else { - lgi->lgi_off = 0; - } - + /** + * the pad record can be inserted so take into account double + * record size + */ + lgi->lgi_buf.lb_len = rec->lrh_len * 2; + lgi->lgi_buf.lb_buf = NULL; /* XXX: implement declared window or multi-chunks approach */ - rc = dt_declare_record_write(env, o, 32 * 1024, lgi->lgi_off, th); + rc = dt_declare_record_write(env, o, &lgi->lgi_buf, -1, th); RETURN(rc); } -/* returns negative in on error; 0 if success && reccookie == 0; 1 otherwise */ -/* appends if idx == -1, otherwise overwrites record idx. */ +/** + * Implementation of the llog_operations::lop_write + * + * This function writes the new record in the llog or modify the existed one. + * + * \param[in] env execution environment + * \param[in] loghandle llog handle of the current llog + * \param[in] rec llog record header. This is a real header of + * the full llog record to write. This is + * the beginning of buffer to write, the length + * of buffer is stored in \a rec::lrh_len + * \param[out] reccookie pointer to the cookie to return back if needed. + * It is used for further cancel of this llog + * record. + * \param[in] idx index of the llog record. If \a idx == -1 then + * this is append case, otherwise \a idx is + * the index of record to modify + * \param[in] th current transaction handle + * + * \retval 0 on successful write && \a reccookie == NULL + * 1 on successful write && \a reccookie != NULL + * \retval negative error if write failed + */ static int llog_osd_write_rec(const struct lu_env *env, struct llog_handle *loghandle, struct llog_rec_hdr *rec, - struct llog_cookie *reccookie, int cookiecount, - void *buf, int idx, struct thandle *th) + struct llog_cookie *reccookie, + int idx, struct thandle *th) { struct llog_thread_info *lgi = llog_info(env); struct llog_log_hdr *llh; @@ -310,6 +347,7 @@ static int llog_osd_write_rec(const struct lu_env *env, struct llog_rec_tail *lrt; struct dt_object *o; size_t left; + bool header_is_updated = false; ENTRY; @@ -324,67 +362,100 @@ static int llog_osd_write_rec(const struct lu_env *env, rec->lrh_type, PFID(lu_object_fid(&o->do_lu))); /* record length should not bigger than LLOG_CHUNK_SIZE */ - if (buf) - rc = (reclen > LLOG_CHUNK_SIZE - sizeof(struct llog_rec_hdr) - - sizeof(struct llog_rec_tail)) ? -E2BIG : 0; - else - rc = (reclen > LLOG_CHUNK_SIZE) ? -E2BIG : 0; - if (rc) - RETURN(rc); + if (reclen > LLOG_CHUNK_SIZE) + RETURN(-E2BIG); rc = dt_attr_get(env, o, &lgi->lgi_attr, NULL); if (rc) RETURN(rc); - if (buf) - /* write_blob adds header and tail to lrh_len. */ - reclen = sizeof(*rec) + rec->lrh_len + - sizeof(struct llog_rec_tail); - - if (idx != -1) { - /* no header: only allowed to insert record 1 */ - if (idx != 1 && lgi->lgi_attr.la_size == 0) - LBUG(); - - if (idx && llh->llh_size && llh->llh_size != rec->lrh_len) - RETURN(-EINVAL); + /** + * The modification case. + * If idx set then the record with that index must be modified. + * There are three cases possible: + * 1) the common case is the llog header update (idx == 0) + * 2) the llog record modification during llog process. + * This is indicated by the \a loghandle::lgh_cur_idx > 0. + * In that case the \a loghandle::lgh_cur_offset + * 3) otherwise this is assumed that llog consist of records of + * fixed size, i.e. catalog. The llog header must has llh_size + * field equal to record size. The record offset is calculated + * just by /a idx value + * + * During modification we don't need extra header update because + * the bitmap and record count are not changed. The record header + * and tail remains the same too. + */ + if (idx != LLOG_NEXT_IDX) { + /* llog can be empty only when first record is being written */ + LASSERT(ergo(idx > 0, lgi->lgi_attr.la_size > 0)); - if (!ext2_test_bit(idx, llh->llh_bitmap)) + if (!ext2_test_bit(idx, llh->llh_bitmap)) { CERROR("%s: modify unset record %u\n", o->do_lu.lo_dev->ld_obd->obd_name, idx); - if (idx != rec->lrh_index) - CERROR("%s: index mismatch %d %u\n", + RETURN(-ENOENT); + } + + if (idx != rec->lrh_index) { + CERROR("%s: modify index mismatch %d %u\n", o->do_lu.lo_dev->ld_obd->obd_name, idx, rec->lrh_index); + RETURN(-EFAULT); + } - lgi->lgi_off = 0; - rc = llog_osd_write_blob(env, o, &llh->llh_hdr, NULL, - &lgi->lgi_off, th); - /* we are done if we only write the header or on error */ - if (rc || idx == 0) - RETURN(rc); + if (idx == LLOG_HEADER_IDX) { + /* llog header update */ + LASSERT(reclen == sizeof(struct llog_log_hdr)); + LASSERT(rec == &llh->llh_hdr); - if (buf) { - /* We assume that caller has set lgh_cur_* */ - lgi->lgi_off = loghandle->lgh_cur_offset; - CDEBUG(D_OTHER, - "modify record "LPX64": idx:%d/%u/%d, len:%u " - "offset %llu\n", - loghandle->lgh_id.lgl_oid, idx, rec->lrh_index, - loghandle->lgh_cur_idx, rec->lrh_len, - (long long)(lgi->lgi_off - sizeof(*llh))); - if (rec->lrh_index != loghandle->lgh_cur_idx) { - CERROR("%s: modify idx mismatch %u/%d\n", + lgi->lgi_off = 0; + 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); + } else if (loghandle->lgh_cur_idx > 0) { + /** + * The lgh_cur_offset can be used only if index is + * the same. + */ + if (idx != loghandle->lgh_cur_idx) { + CERROR("%s: modify index mismatch %d %d\n", o->do_lu.lo_dev->ld_obd->obd_name, idx, loghandle->lgh_cur_idx); RETURN(-EFAULT); } - } else { - /* Assumes constant lrh_len */ + + lgi->lgi_off = loghandle->lgh_cur_offset; + CDEBUG(D_OTHER, "modify record "DOSTID": idx:%d, " + "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 { + /* This can be result of lgh_cur_idx is not set during + * llog processing or llh_size is not set to proper + * record size for fixed records llog. Therefore it is + * impossible to get record offset. */ + CERROR("%s: can't get record offset, idx:%d, " + "len:%u.\n", o->do_lu.lo_dev->ld_obd->obd_name, + idx, rec->lrh_len); + RETURN(-EFAULT); } - rc = llog_osd_write_blob(env, o, rec, buf, &lgi->lgi_off, th); + /* update only data, header and tail remain the same */ + lgi->lgi_off += sizeof(struct llog_rec_hdr); + lgi->lgi_buf.lb_len = REC_DATA_LEN(rec); + lgi->lgi_buf.lb_buf = REC_DATA(rec); + rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); if (rc == 0 && reccookie) { reccookie->lgc_lgl = loghandle->lgh_id; reccookie->lgc_index = idx; @@ -393,12 +464,15 @@ static int llog_osd_write_rec(const struct lu_env *env, RETURN(rc); } - /* Make sure that records don't cross a chunk boundary, so we can + /** + * The append case. + * The most common case of using llog. The new index is assigned to + * the new record, new bit is set in llog bitmap and llog count is + * incremented. + * + * Make sure that records don't cross a chunk boundary, so we can * 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. - * - * We know that llog_current_log() will return a loghandle that is - * big enough to hold reclen, so all we care about is padding here. */ LASSERT(lgi->lgi_attr.la_valid & LA_SIZE); lgi->lgi_off = lgi->lgi_attr.la_size; @@ -410,55 +484,63 @@ static int llog_osd_write_rec(const struct lu_env *env, rc = llog_osd_pad(env, o, &lgi->lgi_off, left, index, th); if (rc) RETURN(rc); - loghandle->lgh_last_idx++; /*for pad rec*/ + 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_BITMAP_SIZE(llh) - 1) RETURN(-ENOSPC); + /* increment the last_idx along with llh_tail index, they should + * be equal for a llog lifetime */ loghandle->lgh_last_idx++; index = loghandle->lgh_last_idx; + llh->llh_tail.lrt_index = index; + /** + * NB: the caller should make sure only 1 process access + * the lgh_last_idx, e.g. append should be exclusive. + * Otherwise it might hit the assert. + */ LASSERT(index < LLOG_BITMAP_SIZE(llh)); rec->lrh_index = index; - if (buf == NULL) { - lrt = (struct llog_rec_tail *)((char *)rec + rec->lrh_len - - sizeof(*lrt)); - lrt->lrt_len = rec->lrh_len; - lrt->lrt_index = rec->lrh_index; - } - /* The caller should make sure only 1 process access the lgh_last_idx, - * Otherwise it might hit the assert.*/ - LASSERT(index < LLOG_BITMAP_SIZE(llh)); - cfs_spin_lock(&loghandle->lgh_hdr_lock); + lrt = rec_tail(rec); + lrt->lrt_len = rec->lrh_len; + lrt->lrt_index = rec->lrh_index; + + /* the lgh_hdr_lock protects llog header data from concurrent + * update/cancel, the llh_count and llh_bitmap are protected */ + spin_lock(&loghandle->lgh_hdr_lock); if (ext2_set_bit(index, llh->llh_bitmap)) { CERROR("%s: index %u already set in log bitmap\n", o->do_lu.lo_dev->ld_obd->obd_name, index); - cfs_spin_unlock(&loghandle->lgh_hdr_lock); + spin_unlock(&loghandle->lgh_hdr_lock); LBUG(); /* should never happen */ } llh->llh_count++; - cfs_spin_unlock(&loghandle->lgh_hdr_lock); - llh->llh_tail.lrt_index = index; + spin_unlock(&loghandle->lgh_hdr_lock); lgi->lgi_off = 0; - rc = llog_osd_write_blob(env, o, &llh->llh_hdr, NULL, &lgi->lgi_off, - th); + lgi->lgi_buf.lb_len = llh->llh_hdr.lrh_len; + lgi->lgi_buf.lb_buf = &llh->llh_hdr; + rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); if (rc) - RETURN(rc); + GOTO(out, rc); + header_is_updated = true; rc = dt_attr_get(env, o, &lgi->lgi_attr, NULL); if (rc) - RETURN(rc); + GOTO(out, rc); + LASSERT(lgi->lgi_attr.la_valid & LA_SIZE); lgi->lgi_off = lgi->lgi_attr.la_size; + 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); + if (rc < 0) + GOTO(out, rc); - rc = llog_osd_write_blob(env, o, rec, buf, &lgi->lgi_off, th); - if (rc) - RETURN(rc); - - CDEBUG(D_RPCTRACE, "added record "LPX64": idx: %u, %u\n", - loghandle->lgh_id.lgl_oid, index, rec->lrh_len); - if (rc == 0 && reccookie) { + CDEBUG(D_OTHER, "added record "DOSTID": idx: %u, %u\n", + POSTID(&loghandle->lgh_id.lgl_oi), index, rec->lrh_len); + if (reccookie != NULL) { reccookie->lgc_lgl = loghandle->lgh_id; reccookie->lgc_index = index; if ((rec->lrh_type == MDS_UNLINK_REC) || @@ -471,15 +553,36 @@ static int llog_osd_write_rec(const struct lu_env *env, rc = 1; } RETURN(rc); +out: + /* cleanup llog for error case */ + spin_lock(&loghandle->lgh_hdr_lock); + ext2_clear_bit(index, llh->llh_bitmap); + llh->llh_count--; + spin_unlock(&loghandle->lgh_hdr_lock); + + /* restore llog last_idx */ + loghandle->lgh_last_idx--; + llh->llh_tail.lrt_index = loghandle->lgh_last_idx; + + /* restore the header on disk if it was written */ + if (header_is_updated) { + lgi->lgi_off = 0; + lgi->lgi_buf.lb_len = llh->llh_hdr.lrh_len; + lgi->lgi_buf.lb_buf = &llh->llh_hdr; + dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); + } + + RETURN(rc); } -/* We can skip reading at least as many log blocks as the number of +/** + * We can skip reading at least as many log blocks as the number of * minimum sized log records we are skipping. If it turns 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. */ -static void llog_skip_over(__u64 *off, int curr, int goal) +static inline void llog_skip_over(__u64 *off, int curr, int goal) { if (goal <= curr) return; @@ -487,10 +590,49 @@ static void llog_skip_over(__u64 *off, int curr, int goal) ~(LLOG_CHUNK_SIZE - 1); } -/* sets: - * - cur_offset to the furthest point read in the log file - * - cur_idx to the log index preceeding cur_offset - * returns -EIO/-EINVAL on error +/** + * Remove optional fields that the client doesn't expect. + * This is typically in order to ensure compatibility with older clients. + * It is assumed that since we exclusively remove fields, the block will be + * big enough to handle the remapped records. It is also assumed that records + * of a block have the same format (i.e.: the same features enabled). + * + * \param[in,out] hdr Header of the block of records to remap. + * \param[in,out] last_hdr Last header, don't read past this point. + * \param[in] flags Flags describing the fields to keep. + */ +static void changelog_block_trim_ext(struct llog_rec_hdr *hdr, + struct llog_rec_hdr *last_hdr, + enum changelog_rec_flags flags) +{ + if (hdr->lrh_type != CHANGELOG_REC) + return; + + do { + struct changelog_rec *rec = (struct changelog_rec *)(hdr + 1); + + changelog_remap_rec(rec, rec->cr_flags & flags); + hdr = llog_rec_hdr_next(hdr); + } while ((char *)hdr <= (char *)last_hdr); +} + +/** + * Implementation of the llog_operations::lop_next_block + * + * This function finds the the next llog block to return which contains + * record with required index. It is main part of llog processing. + * + * \param[in] env execution environment + * \param[in] loghandle llog handle of the current llog + * \param[in,out] cur_idx index preceeding cur_offset + * \param[in] next_idx target index to find + * \param[in,out] cur_offset furtherst point read in the file + * \param[in] buf pointer to data buffer to fill + * \param[in] len required len to read, it is + * LLOG_CHUNK_SIZE usually. + * + * \retval 0 on successful buffer read + * \retval negative value on error */ static int llog_osd_next_block(const struct lu_env *env, struct llog_handle *loghandle, int *cur_idx, @@ -536,6 +678,7 @@ static int llog_osd_next_block(const struct lu_env *env, lgi->lgi_buf.lb_len = LLOG_CHUNK_SIZE - (*cur_offset & (LLOG_CHUNK_SIZE - 1)); lgi->lgi_buf.lb_buf = buf; + rc = dt_read(env, o, &lgi->lgi_buf, cur_offset); if (rc < 0) { CERROR("%s: can't read llog block from log "DFID @@ -556,10 +699,10 @@ static int llog_osd_next_block(const struct lu_env *env, GOTO(out, rc); if (rc < sizeof(*tail)) { - CERROR("%s: invalid llog block at log id "LPU64"/%u " + CERROR("%s: invalid llog block at log id "DOSTID"/%u " "offset "LPU64"\n", o->do_lu.lo_dev->ld_obd->obd_name, - loghandle->lgh_id.lgl_oid, + POSTID(&loghandle->lgh_id.lgl_oi), loghandle->lgh_id.lgl_ogen, *cur_offset); GOTO(out, rc = -EINVAL); } @@ -572,7 +715,7 @@ static int llog_osd_next_block(const struct lu_env *env, sizeof(struct llog_rec_tail)); /* get the last record in block */ last_rec = (struct llog_rec_hdr *)((char *)buf + rc - - le32_to_cpu(tail->lrt_len)); + tail->lrt_len); if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec)) lustre_swab_llog_rec(last_rec); @@ -582,10 +725,10 @@ 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 "LPU64"/%u " + CERROR("%s: invalid llog tail at log id "DOSTID"/%u " "offset "LPU64"\n", o->do_lu.lo_dev->ld_obd->obd_name, - loghandle->lgh_id.lgl_oid, + POSTID(&loghandle->lgh_id.lgl_oi), loghandle->lgh_id.lgl_ogen, *cur_offset); GOTO(out, rc = -EINVAL); } @@ -600,6 +743,12 @@ static int llog_osd_next_block(const struct lu_env *env, rec->lrh_index, next_idx); GOTO(out, rc = -ENOENT); } + + /* Trim unsupported extensions for compat w/ older clients */ + if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_JOBID)) + changelog_block_trim_ext(rec, last_rec, + CLF_VERSION | CLF_RENAME); + GOTO(out, rc = 0); } GOTO(out, rc = -EIO); @@ -607,6 +756,23 @@ out: return rc; } +/** + * Implementation of the llog_operations::lop_prev_block + * + * This function finds the llog block to return which contains + * record with required index but in reverse order - from end of llog + * to the beginning. + * It is main part of reverse llog processing. + * + * \param[in] env execution environment + * \param[in] loghandle llog handle of the current llog + * \param[in] prev_idx target index to find + * \param[in] buf pointer to data buffer to fill + * \param[in] len required len to read, it is LLOG_CHUNK_SIZE usually. + * + * \retval 0 on successful buffer read + * \retval negative value on error + */ static int llog_osd_prev_block(const struct lu_env *env, struct llog_handle *loghandle, int prev_idx, void *buf, int len) @@ -659,10 +825,10 @@ static int llog_osd_prev_block(const struct lu_env *env, GOTO(out, rc); if (rc < sizeof(*tail)) { - CERROR("%s: invalid llog block at log id "LPU64"/%u " + CERROR("%s: invalid llog block at log id "DOSTID"/%u " "offset "LPU64"\n", o->do_lu.lo_dev->ld_obd->obd_name, - loghandle->lgh_id.lgl_oid, + POSTID(&loghandle->lgh_id.lgl_oi), loghandle->lgh_id.lgl_ogen, cur_offset); GOTO(out, rc = -EINVAL); } @@ -683,10 +849,10 @@ static int llog_osd_prev_block(const struct lu_env *env, /* this shouldn't happen */ if (tail->lrt_index == 0) { - CERROR("%s: invalid llog tail at log id "LPU64"/%u " + CERROR("%s: invalid llog tail at log id "DOSTID"/%u " "offset "LPU64"\n", o->do_lu.lo_dev->ld_obd->obd_name, - loghandle->lgh_id.lgl_oid, + POSTID(&loghandle->lgh_id.lgl_oi), loghandle->lgh_id.lgl_ogen, cur_offset); GOTO(out, rc = -EINVAL); } @@ -701,6 +867,12 @@ static int llog_osd_prev_block(const struct lu_env *env, rec->lrh_index, prev_idx); GOTO(out, rc = -ENOENT); } + + /* Trim unsupported extensions for compat w/ older clients */ + if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_JOBID)) + changelog_block_trim_ext(rec, last_rec, + CLF_VERSION | CLF_RENAME); + GOTO(out, rc = 0); } GOTO(out, rc = -EIO); @@ -708,8 +880,18 @@ out: return rc; } -struct dt_object *llog_osd_dir_get(const struct lu_env *env, - struct llog_ctxt *ctxt) +/** + * This is helper function to get llog directory object. It is used by named + * llog operations to find/insert/delete llog entry from llog directory. + * + * \param[in] env execution environment + * \param[in] ctxt llog context + * + * \retval dt_object of llog directory + * \retval ERR_PTR of negative value on error + */ +static struct dt_object *llog_osd_dir_get(const struct lu_env *env, + struct llog_ctxt *ctxt) { struct dt_device *dt; struct dt_thread_info *dti = dt_info(env); @@ -722,6 +904,11 @@ struct dt_object *llog_osd_dir_get(const struct lu_env *env, if (rc) return ERR_PTR(rc); dir = dt_locate(env, dt, &dti->dti_fid); + + if (!IS_ERR(dir) && !dt_try_as_dir(env, dir)) { + lu_object_put(env, &dir->do_lu); + return ERR_PTR(-ENOTDIR); + } } else { lu_object_get(&ctxt->loc_dir->do_lu); dir = ctxt->loc_dir; @@ -730,6 +917,27 @@ struct dt_object *llog_osd_dir_get(const struct lu_env *env, return dir; } +/** + * Implementation of the llog_operations::lop_open + * + * This function opens the llog by its logid or by name, it may open also + * non existent llog and assing then new id to it. + * The llog_open/llog_close pair works similar to lu_object_find/put, + * the object may not exist prior open. The result of open is just dt_object + * in the llog header. + * + * \param[in] env execution environment + * \param[in] handle llog handle of the current llog + * \param[in] logid logid of llog to open (nameless llog) + * \param[in] name name of llog to open (named llog) + * \param[in] open_param + * LLOG_OPEN_NEW - new llog, may not exist + * LLOG_OPEN_EXIST - old llog, must exist + * + * \retval 0 on successful open, llog_handle::lgh_obj + * contains the dt_object of the llog. + * \retval negative value on error + */ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle, struct llog_logid *logid, char *name, enum llog_open_param open_param) @@ -755,9 +963,9 @@ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle, if (IS_ERR(ls)) RETURN(PTR_ERR(ls)); - cfs_mutex_lock(&ls->ls_los_mutex); - los = dt_los_find(ls, FID_SEQ_LLOG); - cfs_mutex_unlock(&ls->ls_los_mutex); + mutex_lock(&ls->ls_los_mutex); + los = dt_los_find(ls, name != NULL ? FID_SEQ_LLOG_NAME : FID_SEQ_LLOG); + mutex_unlock(&ls->ls_los_mutex); LASSERT(los); ls_device_put(env, ls); @@ -795,7 +1003,7 @@ static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle, GOTO(out, rc); } - o = ls_locate(env, ls, &lgi->lgi_fid); + o = ls_locate(env, ls, &lgi->lgi_fid, NULL); if (IS_ERR(o)) GOTO(out_name, rc = PTR_ERR(o)); @@ -820,6 +1028,16 @@ out: RETURN(rc); } +/** + * 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); @@ -827,10 +1045,24 @@ static int llog_osd_exist(struct llog_handle *handle) !lu_object_is_dying(handle->lgh_obj->do_lu.lo_header)); } +/** + * Implementation of the llog_operations::lop_declare_create + * + * This function declares the llog create. It declares also name insert + * into llog directory in case of named llog. + * + * \param[in] env execution environment + * \param[in] res llog handle of the current llog + * \param[in] th current transaction handle + * + * \retval 0 on successful create declaration + * \retval negative value on error + */ static int llog_osd_declare_create(const struct lu_env *env, struct llog_handle *res, struct thandle *th) { struct llog_thread_info *lgi = llog_info(env); + struct dt_insert_rec *rec = &lgi->lgi_dt_rec; struct local_oid_storage *los; struct dt_object *o; int rc; @@ -852,9 +1084,8 @@ static int llog_osd_declare_create(const struct lu_env *env, if (rc) RETURN(rc); - rc = dt_declare_record_write(env, o, LLOG_CHUNK_SIZE, 0, th); - if (rc) - RETURN(rc); + /* do not declare header initialization here as it's declared + * in llog_osd_declare_write_rec() which is always called */ if (res->lgh_name) { struct dt_object *llog_dir; @@ -862,10 +1093,11 @@ static int llog_osd_declare_create(const struct lu_env *env, llog_dir = llog_osd_dir_get(env, res->lgh_ctxt); if (IS_ERR(llog_dir)) RETURN(PTR_ERR(llog_dir)); - dt_declare_ref_add(env, o, th); logid_to_fid(&res->lgh_id, &lgi->lgi_fid); + rec->rec_fid = &lgi->lgi_fid; + rec->rec_type = S_IFREG; rc = dt_declare_insert(env, llog_dir, - (struct dt_rec *)&lgi->lgi_fid, + (struct dt_rec *)rec, (struct dt_key *)res->lgh_name, th); lu_object_put(env, &llog_dir->do_lu); if (rc) @@ -876,12 +1108,24 @@ static int llog_osd_declare_create(const struct lu_env *env, RETURN(rc); } -/* This is a callback from the llog_* functions. - * Assumes caller has already pushed us into the kernel context. */ +/** + * Implementation of the llog_operations::lop_create + * + * This function creates the llog according with llog_handle::lgh_obj + * and llog_handle::lgh_name. + * + * \param[in] env execution environment + * \param[in] res llog handle of the current llog + * \param[in] th current transaction handle + * + * \retval 0 on successful create + * \retval negative value on error + */ static int llog_osd_create(const struct lu_env *env, struct llog_handle *res, struct thandle *th) { struct llog_thread_info *lgi = llog_info(env); + struct dt_insert_rec *rec = &lgi->lgi_dt_rec; struct local_oid_storage *los; struct dt_object *o; int rc = 0; @@ -904,8 +1148,7 @@ static int llog_osd_create(const struct lu_env *env, struct llog_handle *res, rc = llog_osd_create_new_object(env, los, o, th); else rc = -EEXIST; - if (res->lgh_name) - dt_ref_add(env, o, th); + dt_write_unlock(env, o); if (rc) RETURN(rc); @@ -918,9 +1161,10 @@ static int llog_osd_create(const struct lu_env *env, struct llog_handle *res, RETURN(PTR_ERR(llog_dir)); logid_to_fid(&res->lgh_id, &lgi->lgi_fid); + rec->rec_fid = &lgi->lgi_fid; + rec->rec_type = S_IFREG; dt_read_lock(env, llog_dir, 0); - rc = dt_insert(env, llog_dir, - (struct dt_rec *)&lgi->lgi_fid, + rc = dt_insert(env, llog_dir, (struct dt_rec *)rec, (struct dt_key *)res->lgh_name, th, BYPASS_CAPA, 1); dt_read_unlock(env, llog_dir); @@ -933,6 +1177,18 @@ static int llog_osd_create(const struct lu_env *env, struct llog_handle *res, RETURN(rc); } +/** + * Implementation of the llog_operations::lop_close + * + * This function closes the llog. It just put llog object and referenced + * local storage. + * + * \param[in] env execution environment + * \param[in] handle llog handle of the current llog + * + * \retval 0 on successful llog close + * \retval negative value on error + */ static int llog_osd_close(const struct lu_env *env, struct llog_handle *handle) { struct local_oid_storage *los; @@ -954,6 +1210,20 @@ static int llog_osd_close(const struct lu_env *env, struct llog_handle *handle) 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) { @@ -985,7 +1255,6 @@ static int llog_osd_destroy(const struct lu_env *env, if (IS_ERR(llog_dir)) GOTO(out_trans, rc = PTR_ERR(llog_dir)); - dt_declare_ref_del(env, o, th); name = loghandle->lgh_name; rc = dt_declare_delete(env, llog_dir, (struct dt_key *)name, th); @@ -993,7 +1262,9 @@ static int llog_osd_destroy(const struct lu_env *env, GOTO(out_trans, rc); } - dt_declare_ref_del(env, o, th); + rc = dt_declare_ref_del(env, o, th); + if (rc < 0) + GOTO(out_trans, rc); rc = dt_declare_destroy(env, o, th); if (rc) @@ -1006,7 +1277,6 @@ static int llog_osd_destroy(const struct lu_env *env, dt_write_lock(env, o, 0); if (dt_object_exists(o)) { if (name) { - dt_ref_del(env, o, th); dt_read_lock(env, llog_dir, 0); rc = dt_delete(env, llog_dir, (struct dt_key *) name, @@ -1033,15 +1303,28 @@ out_trans: RETURN(rc); } +/** + * Implementation of the llog_operations::lop_setup + * + * This function setup the llog on local storage. + * + * \param[in] env execution environment + * \param[in] obd obd device the llog belongs to + * \param[in] olg the llog group, it is always zero group now. + * \param[in] ctxt_idx the llog index, it defines the purpose of this llog. + * Every new llog type have to use own index. + * \param[in] disk_obd the storage obd, where llog is stored. + * + * \retval 0 on successful llog setup + * \retval negative value on error + */ static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd, struct obd_llog_group *olg, int ctxt_idx, struct obd_device *disk_obd) { - struct local_oid_storage *los; struct llog_thread_info *lgi = llog_info(env); struct llog_ctxt *ctxt; int rc = 0; - ENTRY; LASSERT(obd); @@ -1056,31 +1339,51 @@ static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd, lgi->lgi_fid.f_oid = 1; lgi->lgi_fid.f_ver = 0; rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt, - &lgi->lgi_fid, &los); + &lgi->lgi_fid, + &ctxt->loc_los_nameless); + if (rc != 0) + GOTO(out, rc); + + lgi->lgi_fid.f_seq = FID_SEQ_LLOG_NAME; + lgi->lgi_fid.f_oid = 1; + lgi->lgi_fid.f_ver = 0; + rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt, + &lgi->lgi_fid, + &ctxt->loc_los_named); + if (rc != 0) { + local_oid_storage_fini(env, ctxt->loc_los_nameless); + ctxt->loc_los_nameless = NULL; + } + + GOTO(out, rc); + +out: llog_ctxt_put(ctxt); return rc; } +/** + * Implementation of the llog_operations::lop_cleanup + * + * This function cleanups the llog on local storage. + * + * \param[in] env execution environment + * \param[in] ctxt the llog context + * + * \retval 0 + */ static int llog_osd_cleanup(const struct lu_env *env, struct llog_ctxt *ctxt) { - struct dt_device *dt; - struct ls_device *ls; - struct local_oid_storage *los; - - LASSERT(ctxt->loc_exp->exp_obd); - dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt; - ls = ls_device_get(dt); - if (IS_ERR(ls)) - RETURN(PTR_ERR(ls)); + if (ctxt->loc_los_nameless != NULL) { + local_oid_storage_fini(env, ctxt->loc_los_nameless); + ctxt->loc_los_nameless = NULL; + } - cfs_mutex_lock(&ls->ls_los_mutex); - los = dt_los_find(ls, FID_SEQ_LLOG); - cfs_mutex_unlock(&ls->ls_los_mutex); - if (los != NULL) { - dt_los_put(los); - local_oid_storage_fini(env, los); + if (ctxt->loc_los_named != NULL) { + local_oid_storage_fini(env, ctxt->loc_los_named); + ctxt->loc_los_named = NULL; } - ls_device_put(env, ls); + return 0; } @@ -1101,9 +1404,30 @@ struct llog_operations llog_osd_ops = { }; EXPORT_SYMBOL(llog_osd_ops); -/* reads the catalog list */ +/** + * Read the special file which contains the list of llog catalogs IDs + * + * This function reads the CATALOGS file which contains the array of llog + * catalogs IDs. The main purpose of this file is to store OSP llogs indexed + * by OST/MDT number. + * + * \param[in] env execution environment + * \param[in] d corresponding storage device + * \param[in] idx position to start from, usually OST/MDT index + * \param[in] count how many catalog IDs to read + * \param[out] idarray the buffer for the data. If it is NULL then + * function returns just number of catalog IDs + * in the file. + * \param[in] fid LLOG_CATALOGS_OID for CATALOG object + * + * \retval 0 on successful read of catalog IDs + * \retval negative value on error + * \retval positive value which is number of records in + * the file if \a idarray is NULL + */ int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d, - int idx, int count, struct llog_catid *idarray) + int idx, int count, struct llog_catid *idarray, + const struct lu_fid *fid) { struct llog_thread_info *lgi = llog_info(env); struct dt_object *o = NULL; @@ -1117,7 +1441,7 @@ int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d, size = sizeof(*idarray) * count; lgi->lgi_off = idx * sizeof(*idarray); - lu_local_obj_fid(&lgi->lgi_fid, LLOG_CATALOGS_OID); + lgi->lgi_fid = *fid; o = dt_locate(env, d, &lgi->lgi_fid); if (IS_ERR(o)) RETURN(PTR_ERR(o)); @@ -1173,7 +1497,7 @@ out_trans: /* read for new ost index or for empty file */ memset(idarray, 0, size); - if (lgi->lgi_attr.la_size < lgi->lgi_off + size) + if (lgi->lgi_attr.la_size <= lgi->lgi_off) GOTO(out, rc = 0); if (lgi->lgi_attr.la_size < lgi->lgi_off + size) size = lgi->lgi_attr.la_size - lgi->lgi_off; @@ -1181,7 +1505,9 @@ out_trans: lgi->lgi_buf.lb_buf = idarray; lgi->lgi_buf.lb_len = size; rc = dt_record_read(env, o, &lgi->lgi_buf, &lgi->lgi_off); - if (rc) { + /* -EFAULT means the llog is a sparse file. This is not an error + * after arbitrary OST index is supported. */ + if (rc < 0 && rc != -EFAULT) { CERROR("%s: error reading CATALOGS: rc = %d\n", o->do_lu.lo_dev->ld_obd->obd_name, rc); GOTO(out, rc); @@ -1194,24 +1520,41 @@ out: } EXPORT_SYMBOL(llog_osd_get_cat_list); -/* writes the cat list */ +/** + * Write the special file which contains the list of llog catalogs IDs + * + * This function writes the CATALOG file which contains the array of llog + * catalogs IDs. It is used mostly to store OSP llogs indexed by OST/MDT + * number. + * + * \param[in] env execution environment + * \param[in] d corresponding storage device + * \param[in] idx position to start from, usually OST/MDT index + * \param[in] count how many catalog IDs to write + * \param[out] idarray the buffer with the data to write. + * \param[in] fid LLOG_CATALOGS_OID for CATALOG object + * + * \retval 0 on successful write of catalog IDs + * \retval negative value on error + */ int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d, - int idx, int count, struct llog_catid *idarray) + int idx, int count, struct llog_catid *idarray, + const struct lu_fid *fid) { struct llog_thread_info *lgi = llog_info(env); struct dt_object *o = NULL; struct thandle *th; int rc, size; - if (!count) + if (count == 0) RETURN(0); LASSERT(d); size = sizeof(*idarray) * count; lgi->lgi_off = idx * sizeof(*idarray); + lgi->lgi_fid = *fid; - lu_local_obj_fid(&lgi->lgi_fid, LLOG_CATALOGS_OID); o = dt_locate(env, d, &lgi->lgi_fid); if (IS_ERR(o)) RETURN(PTR_ERR(o)); @@ -1234,7 +1577,9 @@ int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d, if (IS_ERR(th)) GOTO(out, rc = PTR_ERR(th)); - rc = dt_declare_record_write(env, o, size, lgi->lgi_off, th); + lgi->lgi_buf.lb_len = size; + 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); @@ -1242,11 +1587,10 @@ int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d, if (rc) GOTO(out_trans, rc); - lgi->lgi_buf.lb_buf = idarray; - lgi->lgi_buf.lb_len = size; rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th); if (rc) - CDEBUG(D_INODE, "error writeing CATALOGS: rc = %d\n", rc); + CDEBUG(D_INODE, "can't write CATALOGS at index %d: rc = %d\n", + idx, rc); out_trans: dt_trans_stop(env, d, th); out: