* Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2012, 2016, Intel Corporation.
+ * Copyright (c) 2012, 2017, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
}
/**
+ * FLR: verify the layout version of object.
+ *
+ * \param[in] env execution environment
+ * \param[in] fo OFD object
+ * \param[in] oa OBDO structure with layout version
+ *
+ * \retval 0 on successful verification
+ * \retval -EINPROGRESS layout version is in transfer
+ * \retval -ESTALE the layout version on client is stale
+ */
+int ofd_verify_layout_version(const struct lu_env *env,
+ struct ofd_object *fo, const struct obdo *oa)
+{
+ __u32 layout_version;
+ int rc;
+ ENTRY;
+
+ if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_OST_SKIP_LV_CHECK)))
+ GOTO(out, rc = 0);
+
+ rc = ofd_object_ff_load(env, fo);
+ if (rc < 0) {
+ if (rc == -ENODATA)
+ rc = -EINPROGRESS;
+ GOTO(out, rc);
+ }
+
+ layout_version = fo->ofo_ff.ff_layout_version;
+ if (oa->o_layout_version >= layout_version &&
+ oa->o_layout_version <= layout_version + fo->ofo_ff.ff_range)
+ GOTO(out, rc = 0);
+
+ /* normal traffic, decide if to return ESTALE or EINPROGRESS */
+ layout_version &= ~LU_LAYOUT_RESYNC;
+
+ /* this update is not legitimate */
+ if ((oa->o_layout_version & ~LU_LAYOUT_RESYNC) <= layout_version)
+ GOTO(out, rc = -ESTALE);
+
+ /* layout version may not be transmitted yet */
+ if ((oa->o_layout_version & ~LU_LAYOUT_RESYNC) > layout_version)
+ GOTO(out, rc = -EINPROGRESS);
+
+ EXIT;
+
+out:
+ CDEBUG(D_INODE, DFID " verify layout version: %u vs. %u/%u, rc: %d\n",
+ PFID(lu_object_fid(&fo->ofo_obj.do_lu)),
+ oa->o_layout_version, fo->ofo_ff.ff_layout_version,
+ fo->ofo_ff.ff_range, rc);
+ return rc;
+
+}
+
+/**
* Prepare buffers for read request processing.
*
* This function converts remote buffers from client to local buffers
ofd_seq_put(env, oseq);
}
+ /* Process incoming grant info, set OBD_BRW_GRANTED flag and grant some
+ * space back if possible, we have to do this outside of the lock as
+ * grant preparation may need to sync whole fs thus wait for all the
+ * transactions to complete. */
+ tgt_grant_prepare_write(env, exp, oa, rnb, obj->ioo_bufcnt);
+
fo = ofd_object_find(env, ofd, fid);
if (IS_ERR(fo))
GOTO(out, rc = PTR_ERR(fo));
}
}
- /* Process incoming grant info, set OBD_BRW_GRANTED flag and grant some
- * space back if possible */
- tgt_grant_prepare_write(env, exp, oa, rnb, obj->ioo_bufcnt);
+ /* need to verify layout version */
+ if (oa->o_valid & OBD_MD_LAYOUT_VERSION) {
+ rc = ofd_verify_layout_version(env, fo, oa);
+ if (rc) {
+ ofd_read_unlock(env, fo);
+ ofd_object_put(env, fo);
+ GOTO(out, rc);
+ }
+
+ oa->o_valid &= ~OBD_MD_LAYOUT_VERSION;
+ }
if (ptlrpc_connection_is_local(exp->exp_connection))
dbt |= DT_BUFS_TYPE_LOCAL;
/* correct index for local buffers to continue with */
for (k = 0; k < rc; k++) {
lnb[j+k].lnb_flags = rnb[i].rnb_flags;
+ lnb[j+k].lnb_flags &= ~OBD_BRW_LOCALS;
if (!(rnb[i].rnb_flags & OBD_BRW_GRANTED))
lnb[j+k].lnb_rc = -ENOSPC;
}
if (unlikely(rc != 0))
GOTO(err, rc);
+ ofd_read_unlock(env, fo);
ofd_counter_incr(exp, LPROC_OFD_STATS_WRITE, jobid, tot_bytes);
RETURN(0);
err:
* \param[in] ofd OFD device
* \param[in] ofd_obj OFD object
* \param[in] la object attributes
- * \param[in] ff parent FID
+ * \param[in] oa obdo
*
* \retval 0 on successful attributes update
* \retval negative value on error
static int
ofd_write_attr_set(const struct lu_env *env, struct ofd_device *ofd,
struct ofd_object *ofd_obj, struct lu_attr *la,
- struct filter_fid *ff)
+ struct obdo *oa)
{
struct ofd_thread_info *info = ofd_info(env);
+ struct filter_fid *ff = &info->fti_mds_fid;
__u64 valid = la->la_valid;
- int rc;
struct thandle *th;
struct dt_object *dt_obj;
- int ff_needed = 0;
+ int fl = 0;
+ int rc;
ENTRY;
if (rc != 0)
GOTO(out, rc);
- if (ff != NULL) {
- rc = ofd_object_ff_load(env, ofd_obj);
- if (rc == -ENODATA)
- ff_needed = 1;
- else if (rc < 0)
- GOTO(out, rc);
- }
-
- if (!la->la_valid && !ff_needed)
+ if (!la->la_valid && !(oa->o_valid &
+ (OBD_MD_FLFID | OBD_MD_FLOSTLAYOUT | OBD_MD_LAYOUT_VERSION)))
/* no attributes to set */
GOTO(out, rc = 0);
GOTO(out_tx, rc);
}
- if (ff_needed) {
- if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
- ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
- else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
- le32_add_cpu(&ff->ff_parent.f_oid, -1);
-
- info->fti_buf.lb_buf = ff;
- info->fti_buf.lb_len = sizeof(*ff);
- rc = dt_declare_xattr_set(env, dt_obj, &info->fti_buf,
- XATTR_NAME_FID, 0, th);
- if (rc)
- GOTO(out_tx, rc);
- }
+ rc = dt_declare_xattr_set(env, dt_obj, &info->fti_buf,
+ XATTR_NAME_FID, 0, th);
+ if (rc)
+ GOTO(out_tx, rc);
/* We don't need a transno for this operation which will be re-executed
* anyway when the OST_WRITE (with a transno assigned) is replayed */
if (rc)
GOTO(out_tx, rc);
+ ofd_read_lock(env, ofd_obj);
+
/* set uid/gid/projid */
if (la->la_valid) {
rc = dt_attr_set(env, dt_obj, la, th);
if (rc)
- GOTO(out_tx, rc);
+ GOTO(out_unlock, rc);
}
- /* set filter fid EA */
- if (ff_needed) {
- if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
- GOTO(out_tx, rc);
-
- rc = dt_xattr_set(env, dt_obj, &info->fti_buf, XATTR_NAME_FID,
- 0, th);
- if (!rc)
- filter_fid_le_to_cpu(&ofd_obj->ofo_ff, ff, sizeof(*ff));
- }
-
- GOTO(out_tx, rc);
-
+ fl = ofd_object_ff_update(env, ofd_obj, oa, ff);
+ if (fl <= 0)
+ GOTO(out_unlock, rc = fl);
+
+ /* set filter fid EA.
+ * FIXME: it holds read lock of ofd object to modify the XATTR_NAME_FID
+ * while the write lock should be held. However, it should work because
+ * write RPCs only modify ff_{parent,layout} and those information will
+ * be the same from all the write RPCs. The reason that fl is not used
+ * in dt_xattr_set() is to allow this race. */
+ if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
+ GOTO(out_unlock, rc);
+ if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
+ ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
+ else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
+ le32_add_cpu(&ff->ff_parent.f_oid, -1);
+
+ info->fti_buf.lb_buf = ff;
+ info->fti_buf.lb_len = sizeof(*ff);
+ rc = dt_xattr_set(env, dt_obj, &info->fti_buf, XATTR_NAME_FID, 0, th);
+ if (rc == 0)
+ filter_fid_le_to_cpu(&ofd_obj->ofo_ff, ff, sizeof(*ff));
+
+ GOTO(out_unlock, rc);
+
+out_unlock:
+ ofd_read_unlock(env, ofd_obj);
out_tx:
dt_trans_stop(env, ofd->ofd_osd, th);
out:
static int
ofd_commitrw_write(const struct lu_env *env, struct obd_export *exp,
struct ofd_device *ofd, const struct lu_fid *fid,
- struct lu_attr *la, struct filter_fid *ff, int objcount,
+ struct lu_attr *la, struct obdo *oa, int objcount,
int niocount, struct niobuf_local *lnb,
unsigned long granted, int old_rc)
{
fo = ofd_object_find(env, ofd, fid);
LASSERT(fo != NULL);
- LASSERT(ofd_object_exists(fo));
o = ofd_object_child(fo);
LASSERT(o != NULL);
if (old_rc)
GOTO(out, rc = old_rc);
+ if (!ofd_object_exists(fo))
+ GOTO(out, rc = -ENOENT);
/*
* The first write to each object must set some attributes. It is
* dt_declare_write_commit() since quota enforcement is now handled in
* declare phases.
*/
- rc = ofd_write_attr_set(env, ofd, fo, la, ff);
+ rc = ofd_write_attr_set(env, ofd, fo, la, oa);
if (rc)
GOTO(out, rc);
if (IS_ERR(th))
GOTO(out, rc = PTR_ERR(th));
- th->th_sync |= ofd->ofd_syncjournal;
+ th->th_sync |= ofd->ofd_sync_journal;
if (th->th_sync == 0) {
for (i = 0; i < niocount; i++) {
if (!(lnb[i].lnb_flags & OBD_BRW_ASYNC)) {
if (rc)
GOTO(out_stop, rc);
+ ofd_read_lock(env, fo);
+ if (!ofd_object_exists(fo))
+ GOTO(out_unlock, rc = -ENOENT);
+
if (likely(!fake_write)) {
rc = dt_write_commit(env, o, lnb, niocount, th);
if (rc)
- GOTO(out_stop, rc);
+ GOTO(out_unlock, rc);
}
if (la->la_valid) {
rc = dt_attr_set(env, o, la, th);
if (rc)
- GOTO(out_stop, rc);
+ GOTO(out_unlock, rc);
}
/* get attr to return */
rc = dt_attr_get(env, o, la);
+out_unlock:
+ ofd_read_unlock(env, fo);
out_stop:
/* Force commit to make the just-deleted blocks
* reusable. LU-456 */
out:
dt_bufs_put(env, o, lnb, niocount);
- ofd_read_unlock(env, fo);
ofd_object_put(env, fo);
/* second put is pair to object_get in ofd_preprw_write */
ofd_object_put(env, fo);
struct niobuf_remote *rnb, int npages,
struct niobuf_local *lnb, int old_rc)
{
- struct ofd_thread_info *info = ofd_info(env);
- struct ofd_mod_data *fmd;
- __u64 valid;
- struct ofd_device *ofd = ofd_exp(exp);
- struct filter_fid *ff = NULL;
- const struct lu_fid *fid = &oa->o_oi.oi_fid;
- int rc = 0;
+ struct ofd_thread_info *info = ofd_info(env);
+ struct ofd_device *ofd = ofd_exp(exp);
+ const struct lu_fid *fid = &oa->o_oi.oi_fid;
+ __u64 valid;
+ int rc = 0;
LASSERT(npages > 0);
/* Don't update timestamps if this write is older than a
* setattr which modifies the timestamps. b=10150 */
-
- /* XXX when we start having persistent reservations this needs
- * to be changed to ofd_fmd_get() to create the fmd if it
- * doesn't already exist so we can store the reservation handle
- * there. */
valid = OBD_MD_FLUID | OBD_MD_FLGID | OBD_MD_FLPROJID;
- fmd = ofd_fmd_find(exp, fid);
- if (!fmd || fmd->fmd_mactime_xid < info->fti_xid)
+ if (tgt_fmd_check(exp, fid, info->fti_xid))
valid |= OBD_MD_FLATIME | OBD_MD_FLMTIME |
OBD_MD_FLCTIME;
- ofd_fmd_put(exp, fmd);
- la_from_obdo(&info->fti_attr, oa, valid);
- if (oa->o_valid & OBD_MD_FLFID) {
- ff = &info->fti_mds_fid;
- ofd_prepare_fidea(ff, oa);
- }
+ la_from_obdo(&info->fti_attr, oa, valid);
rc = ofd_commitrw_write(env, exp, ofd, fid, &info->fti_attr,
- ff, objcount, npages, lnb,
+ oa, objcount, npages, lnb,
oa->o_grant_used, old_rc);
if (rc == 0)
obdo_from_la(oa, &info->fti_attr,
git://git.whamcloud.com / fs / lustre-release.git / blobdiff