* Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2012, 2015, Intel Corporation.
+ * Copyright (c) 2012, 2017, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
!lu_object_is_dying(handle->lgh_obj->do_lu.lo_header);
}
+static void *rec_tail(struct llog_rec_hdr *rec)
+{
+ return (void *)((char *)rec + rec->lrh_len -
+ sizeof(struct llog_rec_tail));
+}
+
/**
* Write a padding record to the llog
*
* 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.
+ * \param[in,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
struct dt_object *o;
__u32 chunk_size;
size_t left;
-
+ __u32 orig_last_idx;
ENTRY;
- LASSERT(env);
llh = loghandle->lgh_hdr;
- LASSERT(llh);
o = loghandle->lgh_obj;
- LASSERT(o);
- LASSERT(th);
chunk_size = llh->llh_hdr.lrh_len;
CDEBUG(D_OTHER, "new record %x to "DFID"\n",
&lgi->lgi_off, th);
RETURN(rc);
- } else if (loghandle->lgh_cur_idx > 0) {
+ } else if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
+ lgi->lgi_off = llh->llh_hdr.lrh_len +
+ (idx - 1) * reclen;
+ } else if (reccookie != NULL && reccookie->lgc_index > 0) {
/**
- * The lgh_cur_offset can be used only if index is
+ * The lgc_offset can be used only if index is
* the same.
*/
- if (idx != loghandle->lgh_cur_idx) {
+ if (idx != reccookie->lgc_index) {
CERROR("%s: modify index mismatch %d %d\n",
o->do_lu.lo_dev->ld_obd->obd_name, idx,
- loghandle->lgh_cur_idx);
+ reccookie->lgc_index);
RETURN(-EFAULT);
}
- lgi->lgi_off = loghandle->lgh_cur_offset;
- CDEBUG(D_OTHER, "modify record "DOSTID": idx:%d, "
+ lgi->lgi_off = reccookie->lgc_offset;
+ CDEBUG(D_OTHER, "modify record "DFID": idx:%u, "
"len:%u offset %llu\n",
- POSTID(&loghandle->lgh_id.lgl_oi), idx,
+ PFID(&loghandle->lgh_id.lgl_oi.oi_fid), idx,
rec->lrh_len, (long long)lgi->lgi_off);
- } 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
RETURN(-ENOSPC);
LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
+ orig_last_idx = loghandle->lgh_last_idx;
lgi->lgi_off = lgi->lgi_attr.la_size;
+
+ if (loghandle->lgh_max_size > 0 &&
+ lgi->lgi_off >= loghandle->lgh_max_size) {
+ CDEBUG(D_OTHER, "llog is getting too large (%u > %u) at %u "
+ DFID"\n", (unsigned)lgi->lgi_off,
+ loghandle->lgh_max_size, (int)loghandle->lgh_last_idx,
+ PFID(&loghandle->lgh_id.lgl_oi.oi_fid));
+ /* this is to signal that this llog is full */
+ loghandle->lgh_last_idx = LLOG_HDR_BITMAP_SIZE(llh) - 1;
+ RETURN(-ENOSPC);
+ }
+
left = chunk_size - (lgi->lgi_off & (chunk_size - 1));
/* NOTE: padding is a record, but no bit is set */
if (left != 0 && left != reclen &&
rc = llog_osd_pad(env, o, &lgi->lgi_off, left, index, th);
if (rc)
RETURN(rc);
+
loghandle->lgh_last_idx++; /* for pad rec */
}
/* if it's the last idx in log file, then return -ENOSPC
if (rc < 0)
GOTO(out, rc);
- CDEBUG(D_OTHER, "added record "DOSTID": idx: %u, %u off"LPU64"\n",
- POSTID(&loghandle->lgh_id.lgl_oi), index, rec->lrh_len,
+ CDEBUG(D_HA, "added record "DFID".%u, %u off%llu\n",
+ PFID(lu_object_fid(&o->do_lu)), index, rec->lrh_len,
lgi->lgi_off);
if (reccookie != NULL) {
reccookie->lgc_lgl = loghandle->lgh_id;
mutex_unlock(&loghandle->lgh_hdr_mutex);
/* restore llog last_idx */
- if (--loghandle->lgh_last_idx == 0 &&
+ if (dt_object_remote(o)) {
+ loghandle->lgh_last_idx = orig_last_idx;
+ } else 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);
* 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.
+ * \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.
+ * \param[in] extra_flags Flags describing the extra 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)
+ struct llog_handle *loghandle)
{
+ enum changelog_rec_flags flags = CLF_SUPPORTED;
+ enum changelog_rec_extra_flags extra_flags = CLFE_SUPPORTED;
+
+ if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_XATTR))
+ extra_flags &= ~CLFE_XATTR;
+ if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_OMODE))
+ extra_flags &= ~CLFE_OPEN;
+ if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_NID))
+ extra_flags &= ~CLFE_NID;
+ if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_UIDGID))
+ extra_flags &= ~CLFE_UIDGID;
+ if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_EXTRA_FLAGS))
+ flags &= ~CLF_EXTRA_FLAGS;
+ if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_JOBID))
+ flags &= ~CLF_JOBID;
+
+ if (flags == CLF_SUPPORTED && extra_flags == CLFE_SUPPORTED)
+ return;
+
if (hdr->lrh_type != CHANGELOG_REC)
return;
do {
struct changelog_rec *rec = (struct changelog_rec *)(hdr + 1);
+ enum changelog_rec_extra_flags xflag = CLFE_INVALID;
+
+ if (flags & CLF_EXTRA_FLAGS &&
+ rec->cr_flags & CLF_EXTRA_FLAGS) {
+ xflag = changelog_rec_extra_flags(rec)->cr_extra_flags &
+ extra_flags;
+ }
+
+ if (unlikely(hdr->lrh_len == 0)) {
+ /* It is corruption case, we cannot know the next rec,
+ * jump to the last one directly to avoid dead loop. */
+ LCONSOLE(D_WARNING, "Hit invalid llog record: "
+ "idx %u, type %u, id %u\n",
+ hdr->lrh_index, hdr->lrh_type, hdr->lrh_id);
+ hdr = llog_rec_hdr_next(last_hdr);
+ if (unlikely(hdr == last_hdr))
+ LCONSOLE(D_WARNING, "The last record crashed: "
+ "idx %u, type %u, id %u\n",
+ hdr->lrh_index, hdr->lrh_type,
+ hdr->lrh_id);
+ break;
+ }
- changelog_remap_rec(rec, rec->cr_flags & flags);
+ changelog_remap_rec(rec, rec->cr_flags & flags, xflag);
hdr = llog_rec_hdr_next(hdr);
+ /* Yield CPU to avoid soft-lockup if there are too many records
+ * to be handled. */
+ cond_resched();
} while ((char *)hdr <= (char *)last_hdr);
}
if (len == 0 || len & (chunk_size - 1))
RETURN(-EINVAL);
- CDEBUG(D_OTHER, "looking for log index %u (cur idx %u off "LPU64")\n",
- next_idx, *cur_idx, *cur_offset);
-
LASSERT(loghandle);
LASSERT(loghandle->lgh_ctxt);
if (rc)
GOTO(out, rc);
+ CDEBUG(D_OTHER, "looking for log index %u (cur idx %u off"
+ "%llu), size %llu\n", next_idx, *cur_idx,
+ *cur_offset, lgi->lgi_attr.la_size);
+
while (*cur_offset < lgi->lgi_attr.la_size) {
struct llog_rec_hdr *rec, *last_rec;
struct llog_rec_tail *tail;
goto retry;
CERROR("%s: can't read llog block from log "DFID
- " offset "LPU64": rc = %d\n",
+ " offset %llu: rc = %d\n",
o->do_lu.lo_dev->ld_obd->obd_name,
PFID(lu_object_fid(&o->do_lu)), *cur_offset,
rc);
if (!force_mini_rec)
goto retry;
- CERROR("%s: invalid llog block at log id "DOSTID"/%u "
- "offset "LPU64"\n",
+ CERROR("%s: invalid llog block at log id "DFID":%x "
+ "offset %llu\n",
o->do_lu.lo_dev->ld_obd->obd_name,
- POSTID(&loghandle->lgh_id.lgl_oi),
+ PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
loghandle->lgh_id.lgl_ogen, *cur_offset);
GOTO(out, rc = -EINVAL);
}
if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
lustre_swab_llog_rec(last_rec);
- LASSERT(last_rec->lrh_index == tail->lrt_index);
+
+ if (last_rec->lrh_index != tail->lrt_index) {
+ CERROR("%s: invalid llog tail at log id "DFID":%x "
+ "offset %llu last_rec idx %u tail idx %u"
+ "lrt len %u read_size %d\n",
+ o->do_lu.lo_dev->ld_obd->obd_name,
+ PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
+ loghandle->lgh_id.lgl_ogen, *cur_offset,
+ last_rec->lrh_index, tail->lrt_index,
+ tail->lrt_len, rc);
+ GOTO(out, rc = -EINVAL);
+ }
*cur_idx = tail->lrt_index;
/* this shouldn't happen */
if (tail->lrt_index == 0) {
- CERROR("%s: invalid llog tail at log id "DOSTID"/%u "
- "offset "LPU64" bytes %d\n",
+ CERROR("%s: invalid llog tail at log id "DFID":%x "
+ "offset %llu bytes %d\n",
o->do_lu.lo_dev->ld_obd->obd_name,
- POSTID(&loghandle->lgh_id.lgl_oi),
+ PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
loghandle->lgh_id.lgl_ogen, *cur_offset, rc);
GOTO(out, rc = -EINVAL);
}
}
/* 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);
+ changelog_block_trim_ext(rec, last_rec, loghandle);
GOTO(out, rc = 0);
rc = dt_read(env, o, &lgi->lgi_buf, &cur_offset);
if (rc < 0) {
CERROR("%s: can't read llog block from log "DFID
- " offset "LPU64": rc = %d\n",
+ " offset %llu: rc = %d\n",
o->do_lu.lo_dev->ld_obd->obd_name,
PFID(lu_object_fid(&o->do_lu)), cur_offset, rc);
GOTO(out, rc);
GOTO(out, rc);
if (rc < sizeof(*tail)) {
- CERROR("%s: invalid llog block at log id "DOSTID"/%u "
- "offset "LPU64"\n",
+ CERROR("%s: invalid llog block at log id "DFID":%x "
+ "offset %llu\n",
o->do_lu.lo_dev->ld_obd->obd_name,
- POSTID(&loghandle->lgh_id.lgl_oi),
+ PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
loghandle->lgh_id.lgl_ogen, cur_offset);
GOTO(out, rc = -EINVAL);
}
/* this shouldn't happen */
if (tail->lrt_index == 0) {
- CERROR("%s: invalid llog tail at log id "DOSTID"/%u "
- "offset "LPU64"\n",
+ CERROR("%s: invalid llog tail at log id "DFID":%x "
+ "offset %llu\n",
o->do_lu.lo_dev->ld_obd->obd_name,
- POSTID(&loghandle->lgh_id.lgl_oi),
+ PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
loghandle->lgh_id.lgl_ogen, cur_offset);
GOTO(out, rc = -EINVAL);
}
}
/* 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);
+ changelog_block_trim_ext(rec, last_rec, loghandle);
GOTO(out, rc = 0);
}
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);
+ dt_object_put(env, dir);
return ERR_PTR(-ENOTDIR);
}
} else {
struct ls_device *ls;
struct local_oid_storage *los = NULL;
int rc = 0;
+ bool new_id = false;
ENTRY;
dt_read_lock(env, llog_dir, 0);
rc = dt_lookup_dir(env, llog_dir, name, &lgi->lgi_fid);
dt_read_unlock(env, llog_dir);
- lu_object_put(env, &llog_dir->do_lu);
+ dt_object_put(env, llog_dir);
if (rc == -ENOENT && open_param == LLOG_OPEN_NEW) {
/* generate fid for new llog */
rc = local_object_fid_generate(env, los,
&lgi->lgi_fid);
+ new_id = true;
}
if (rc < 0)
GOTO(out, rc);
} else {
LASSERTF(open_param & LLOG_OPEN_NEW, "%#x\n", open_param);
/* generate fid for new llog */
+generate:
rc = local_object_fid_generate(env, los, &lgi->lgi_fid);
if (rc < 0)
GOTO(out, rc);
+ new_id = true;
}
o = ls_locate(env, ls, &lgi->lgi_fid, NULL);
if (IS_ERR(o))
GOTO(out_name, rc = PTR_ERR(o));
+ if (dt_object_exists(o) && new_id) {
+ /* llog exists with just generated ID, e.g. some old llog file
+ * still is in use or is orphan, drop a warn and skip it. */
+ CDEBUG(D_INFO, "%s: llog exists with the same FID: "DFID
+ ", skipping\n",
+ o->do_lu.lo_dev->ld_obd->obd_name,
+ PFID(lu_object_fid(&o->do_lu)));
+ dt_object_put(env, o);
+ /* just skip this llog ID, we shouldn't delete it because we
+ * don't know exactly what is its purpose and state. */
+ goto generate;
+ }
+
after_open:
/* No new llog is expected but doesn't exist */
- if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o))
+ if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o)) {
+ CDEBUG(D_INFO, "%s: llog FID: "DFID" obj %p doesn`t exist\n",
+ o->do_lu.lo_dev->ld_obd->obd_name,
+ PFID(lu_object_fid(&o->do_lu)), o);
GOTO(out_put, rc = -ENOENT);
-
+ }
fid_to_logid(&lgi->lgi_fid, &handle->lgh_id);
handle->lgh_obj = o;
handle->private_data = los;
RETURN(rc);
out_put:
- lu_object_put(env, &o->do_lu);
+ dt_object_put(env, o);
out_name:
if (handle->lgh_name != NULL)
OBD_FREE(handle->lgh_name, strlen(name) + 1);
RETURN(dir);
if (!dt_try_as_dir(env, dir)) {
- lu_object_put(env, &dir->do_lu);
+ dt_object_put(env, dir);
RETURN(ERR_PTR(-ENOTDIR));
}
}
dt_write_unlock(env, dir);
- lu_object_put(env, &dir->do_lu);
+ dt_object_put(env, dir);
RETURN(rc);
}
rc = dt_declare_insert(env, llog_dir,
(struct dt_rec *)rec,
(struct dt_key *)res->lgh_name, th);
- lu_object_put(env, &llog_dir->do_lu);
+ dt_object_put(env, llog_dir);
if (rc)
CERROR("%s: can't declare named llog %s: rc = %d\n",
o->do_lu.lo_dev->ld_obd->obd_name,
(struct dt_key *)res->lgh_name,
th, 1);
dt_read_unlock(env, llog_dir);
- lu_object_put(env, &llog_dir->do_lu);
+ dt_object_put(env, llog_dir);
if (rc)
CERROR("%s: can't create named llog %s: rc = %d\n",
o->do_lu.lo_dev->ld_obd->obd_name,
if (handle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
/* Remove the object from the cache, otherwise it may
* hold LOD being released during cleanup process */
- lu_object_put_nocache(env, &handle->lgh_obj->do_lu);
+ dt_object_put_nocache(env, handle->lgh_obj);
LASSERT(handle->private_data == NULL);
RETURN(rc);
} else {
- lu_object_put(env, &handle->lgh_obj->do_lu);
+ dt_object_put(env, handle->lgh_obj);
}
los = handle->private_data;
LASSERT(los);
}
dt_write_unlock(env, dir);
- lu_object_put(env, &dir->do_lu);
+ dt_object_put(env, dir);
RETURN(rc);
}
out_put:
if (!(IS_ERR_OR_NULL(llog_dir)))
- lu_object_put(env, &llog_dir->do_lu);
+ dt_object_put(env, llog_dir);
RETURN(rc);
}
out_unlock:
dt_write_unlock(env, o);
if (!(IS_ERR_OR_NULL(llog_dir)))
- lu_object_put(env, &llog_dir->do_lu);
+ dt_object_put(env, llog_dir);
RETURN(rc);
}
EXIT;
out:
- lu_object_put(env, &o->do_lu);
+ dt_object_put(env, o);
RETURN(rc);
}
EXPORT_SYMBOL(llog_osd_get_cat_list);
out_trans:
dt_trans_stop(env, d, th);
out:
- lu_object_put(env, &o->do_lu);
+ dt_object_put(env, o);
RETURN(rc);
}
EXPORT_SYMBOL(llog_osd_put_cat_list);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff