static int mdd_declare_links_add(const struct lu_env *env,
struct mdd_object *mdd_obj,
struct thandle *handle);
-static int mdd_links_add(const struct lu_env *env,
- struct mdd_object *mdd_obj,
- const struct lu_fid *pfid,
- const struct lu_name *lname,
- struct thandle *handle, int first);
+static inline int mdd_links_add(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ const struct lu_fid *pfid,
+ const struct lu_name *lname,
+ struct thandle *handle, int first);
+static inline int mdd_links_del(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ const struct lu_fid *pfid,
+ const struct lu_name *lname,
+ struct thandle *handle);
static int mdd_links_rename(const struct lu_env *env,
- struct mdd_object *mdd_obj,
- const struct lu_fid *oldpfid,
- const struct lu_name *oldlname,
- const struct lu_fid *newpfid,
- const struct lu_name *newlname,
- struct thandle *handle);
+ struct mdd_object *mdd_obj,
+ const struct lu_fid *oldpfid,
+ const struct lu_name *oldlname,
+ const struct lu_fid *newpfid,
+ const struct lu_name *newlname,
+ struct thandle *handle,
+ int first, int check);
static int
__mdd_lookup_locked(const struct lu_env *env, struct md_object *pobj,
* VTX feature has been checked already, no need check again.
*/
static inline int mdd_is_sticky(const struct lu_env *env,
- struct mdd_object *pobj,
- struct mdd_object *cobj)
+ struct mdd_object *pobj,
+ struct mdd_object *cobj)
{
- struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
- struct md_ucred *uc = md_ucred(env);
- int rc;
+ struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
+ struct lu_ucred *uc = lu_ucred_assert(env);
+ int rc;
- if (pobj) {
- rc = mdd_la_get(env, pobj, tmp_la, BYPASS_CAPA);
- if (rc)
- return rc;
+ if (pobj) {
+ rc = mdd_la_get(env, pobj, tmp_la, BYPASS_CAPA);
+ if (rc)
+ return rc;
- if (!(tmp_la->la_mode & S_ISVTX) ||
- (tmp_la->la_uid == uc->mu_fsuid))
- return 0;
- }
+ if (!(tmp_la->la_mode & S_ISVTX) ||
+ (tmp_la->la_uid == uc->uc_fsuid))
+ return 0;
+ }
- rc = mdd_la_get(env, cobj, tmp_la, BYPASS_CAPA);
- if (rc)
- return rc;
+ rc = mdd_la_get(env, cobj, tmp_la, BYPASS_CAPA);
+ if (rc)
+ return rc;
- if (tmp_la->la_uid == uc->mu_fsuid)
- return 0;
+ if (tmp_la->la_uid == uc->uc_fsuid)
+ return 0;
- return !mdd_capable(uc, CFS_CAP_FOWNER);
+ return !mdd_capable(uc, CFS_CAP_FOWNER);
}
/*
}
static int __mdd_index_insert_only(const struct lu_env *env,
- struct mdd_object *pobj,
- const struct lu_fid *lf, const char *name,
- struct thandle *handle,
- struct lustre_capa *capa)
+ struct mdd_object *pobj,
+ const struct lu_fid *lf, const char *name,
+ struct thandle *handle,
+ struct lustre_capa *capa)
{
- struct dt_object *next = mdd_object_child(pobj);
- int rc;
- ENTRY;
+ struct dt_object *next = mdd_object_child(pobj);
+ int rc;
+ ENTRY;
- if (dt_try_as_dir(env, next)) {
- struct md_ucred *uc = md_ucred(env);
+ if (dt_try_as_dir(env, next)) {
+ struct lu_ucred *uc = lu_ucred_check(env);
+ int ignore_quota;
- rc = next->do_index_ops->dio_insert(env, next,
- (struct dt_rec*)lf,
- (const struct dt_key *)name,
- handle, capa, uc->mu_cap &
- CFS_CAP_SYS_RESOURCE_MASK);
- } else {
- rc = -ENOTDIR;
- }
- RETURN(rc);
+ ignore_quota = uc ? uc->uc_cap & CFS_CAP_SYS_RESOURCE_MASK : 1;
+ rc = next->do_index_ops->dio_insert(env, next,
+ (struct dt_rec*)lf,
+ (const struct dt_key *)name,
+ handle, capa, ignore_quota);
+ } else {
+ rc = -ENOTDIR;
+ }
+ RETURN(rc);
}
/* insert named index, add reference if isdir */
const struct lu_name *fname,
struct thandle *handle)
{
- int reclen;
+ struct obd_device *obd = mdd2obd_dev(mdd);
+ struct llog_ctxt *ctxt;
+ struct llog_changelog_rec *rec;
+ struct lu_buf *buf;
+ int reclen;
+ int rc;
/* Not recording */
if (!(mdd->mdd_cl.mc_flags & CLM_ON))
return 0;
- /* we'll be writing payload + llog header */
- reclen = sizeof(struct llog_changelog_rec);
- if (fname)
- reclen += fname->ln_namelen;
- reclen = llog_data_len(reclen);
+ reclen = llog_data_len(sizeof(*rec) +
+ (fname != NULL ? fname->ln_namelen : 0));
+ buf = mdd_buf_alloc(env, reclen);
+ if (buf->lb_buf == NULL)
+ return -ENOMEM;
+
+ rec = buf->lb_buf;
+ rec->cr_hdr.lrh_len = reclen;
+ rec->cr_hdr.lrh_type = CHANGELOG_REC;
+
+ ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
+ if (ctxt == NULL)
+ return -ENXIO;
- return mdd_declare_llog_record(env, mdd, reclen, handle);
+ rc = llog_declare_add(env, ctxt->loc_handle, &rec->cr_hdr, handle);
+ llog_ctxt_put(ctxt);
+
+ return rc;
}
static int mdd_declare_changelog_ext_store(const struct lu_env *env,
const struct lu_name *sname,
struct thandle *handle)
{
- int reclen;
+ struct obd_device *obd = mdd2obd_dev(mdd);
+ struct llog_ctxt *ctxt;
+ struct llog_changelog_ext_rec *rec;
+ struct lu_buf *buf;
+ int reclen;
+ int rc;
/* Not recording */
if (!(mdd->mdd_cl.mc_flags & CLM_ON))
return 0;
- /* we'll be writing payload + llog header */
- reclen = sizeof(struct llog_changelog_ext_rec);
- if (tname)
- reclen += tname->ln_namelen;
- if (sname)
- reclen += 1 + sname->ln_namelen;
- reclen = llog_data_len(reclen);
+ reclen = llog_data_len(sizeof(*rec) +
+ (tname != NULL ? tname->ln_namelen : 0) +
+ (sname != NULL ? 1 + sname->ln_namelen : 0));
+ buf = mdd_buf_alloc(env, reclen);
+ if (buf->lb_buf == NULL)
+ return -ENOMEM;
+
+ rec = buf->lb_buf;
+ rec->cr_hdr.lrh_len = reclen;
+ rec->cr_hdr.lrh_type = CHANGELOG_REC;
+
+ ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
+ if (ctxt == NULL)
+ return -ENXIO;
+
+ rc = llog_declare_add(env, ctxt->loc_handle, &rec->cr_hdr, handle);
+ llog_ctxt_put(ctxt);
- return mdd_declare_llog_record(env, mdd, reclen, handle);
+ return rc;
+}
+
+/** Add a changelog entry \a rec to the changelog llog
+ * \param mdd
+ * \param rec
+ * \param handle - currently ignored since llogs start their own transaction;
+ * this will hopefully be fixed in llog rewrite
+ * \retval 0 ok
+ */
+int mdd_changelog_store(const struct lu_env *env, struct mdd_device *mdd,
+ struct llog_changelog_rec *rec, struct thandle *th)
+{
+ struct obd_device *obd = mdd2obd_dev(mdd);
+ struct llog_ctxt *ctxt;
+ int rc;
+
+ rec->cr_hdr.lrh_len = llog_data_len(sizeof(*rec) + rec->cr.cr_namelen);
+ rec->cr_hdr.lrh_type = CHANGELOG_REC;
+ rec->cr.cr_time = cl_time();
+
+ spin_lock(&mdd->mdd_cl.mc_lock);
+ /* NB: I suppose it's possible llog_add adds out of order wrt cr_index,
+ * but as long as the MDD transactions are ordered correctly for e.g.
+ * rename conflicts, I don't think this should matter. */
+ rec->cr.cr_index = ++mdd->mdd_cl.mc_index;
+ spin_unlock(&mdd->mdd_cl.mc_lock);
+
+ ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
+ if (ctxt == NULL)
+ return -ENXIO;
+
+ rc = llog_add(env, ctxt->loc_handle, &rec->cr_hdr, NULL, NULL, th);
+ llog_ctxt_put(ctxt);
+ if (rc > 0)
+ rc = 0;
+ return rc;
+}
+
+/** Add a changelog_ext entry \a rec to the changelog llog
+ * \param mdd
+ * \param rec
+ * \param handle - currently ignored since llogs start their own transaction;
+ * this will hopefully be fixed in llog rewrite
+ * \retval 0 ok
+ */
+int mdd_changelog_ext_store(const struct lu_env *env, struct mdd_device *mdd,
+ struct llog_changelog_ext_rec *rec,
+ struct thandle *th)
+{
+ struct obd_device *obd = mdd2obd_dev(mdd);
+ struct llog_ctxt *ctxt;
+ int rc;
+
+ rec->cr_hdr.lrh_len = llog_data_len(sizeof(*rec) + rec->cr.cr_namelen);
+ /* llog_lvfs_write_rec sets the llog tail len */
+ rec->cr_hdr.lrh_type = CHANGELOG_REC;
+ rec->cr.cr_time = cl_time();
+
+ spin_lock(&mdd->mdd_cl.mc_lock);
+ /* NB: I suppose it's possible llog_add adds out of order wrt cr_index,
+ * but as long as the MDD transactions are ordered correctly for e.g.
+ * rename conflicts, I don't think this should matter. */
+ rec->cr.cr_index = ++mdd->mdd_cl.mc_index;
+ spin_unlock(&mdd->mdd_cl.mc_lock);
+
+ ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
+ if (ctxt == NULL)
+ return -ENXIO;
+
+ /* nested journal transaction */
+ rc = llog_add(env, ctxt->loc_handle, &rec->cr_hdr, NULL, NULL, th);
+ llog_ctxt_put(ctxt);
+ if (rc > 0)
+ rc = 0;
+
+ return rc;
}
/** Store a namespace change changelog record
buf = mdd_buf_alloc(env, reclen);
if (buf->lb_buf == NULL)
RETURN(-ENOMEM);
- rec = (struct llog_changelog_rec *)buf->lb_buf;
+ rec = buf->lb_buf;
rec->cr.cr_flags = CLF_VERSION | (CLF_FLAGMASK & flags);
rec->cr.cr_type = (__u32)type;
target->mod_cltime = cfs_time_current_64();
- rc = mdd_changelog_llog_write(mdd, rec, handle);
+ rc = mdd_changelog_store(env, mdd, rec, handle);
if (rc < 0) {
CERROR("changelog failed: rc=%d, op%d %s c"DFID" p"DFID"\n",
rc, type, tname->ln_name, PFID(&rec->cr.cr_tfid),
LASSERT(tname != NULL);
LASSERT(handle != NULL);
- reclen = sizeof(*rec) + tname->ln_namelen;
- if (sname != NULL)
- reclen += 1 + sname->ln_namelen;
- reclen = llog_data_len(reclen);
+ reclen = llog_data_len(sizeof(*rec) +
+ sname != NULL ? 1 + sname->ln_namelen : 0);
buf = mdd_buf_alloc(env, reclen);
if (buf->lb_buf == NULL)
RETURN(-ENOMEM);
- rec = (struct llog_changelog_ext_rec *)buf->lb_buf;
+ rec = buf->lb_buf;
rec->cr.cr_flags = CLF_EXT_VERSION | (CLF_FLAGMASK & flags);
rec->cr.cr_type = (__u32)type;
fid_zero(&rec->cr.cr_tfid);
}
- rc = mdd_changelog_ext_llog_write(mdd, rec, handle);
+ rc = mdd_changelog_ext_store(env, mdd, rec, handle);
if (rc < 0) {
CERROR("changelog failed: rc=%d, op%d %s c"DFID" p"DFID"\n",
rc, type, tname->ln_name, PFID(sfid), PFID(tpfid));
struct mdd_device *mdd = mdo2mdd(src_obj);
struct dynlock_handle *dlh;
struct thandle *handle;
-#ifdef HAVE_QUOTA_SUPPORT
- struct obd_device *obd = mdd->mdd_obd_dev;
- struct obd_export *exp = md_quota(env)->mq_exp;
- struct mds_obd *mds = &obd->u.mds;
- unsigned int qids[MAXQUOTAS] = { 0, 0 };
- int quota_opc = 0, rec_pending[MAXQUOTAS] = { 0, 0 };
-#endif
int rc;
ENTRY;
-#ifdef HAVE_QUOTA_SUPPORT
- if (mds->mds_quota) {
- struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
-
- rc = mdd_la_get(env, mdd_tobj, la_tmp, BYPASS_CAPA);
- if (!rc) {
- void *data = NULL;
- mdd_data_get(env, mdd_tobj, &data);
- quota_opc = FSFILT_OP_LINK;
- mdd_quota_wrapper(la_tmp, qids);
- /* get block quota for parent */
- lquota_chkquota(mds_quota_interface_ref, obd, exp,
- qids, rec_pending, 1, NULL,
- LQUOTA_FLAGS_BLK, data, 1);
- }
- }
-#endif
-
handle = mdd_trans_create(env, mdd);
if (IS_ERR(handle))
GOTO(out_pending, rc = PTR_ERR(handle));
stop:
mdd_trans_stop(env, mdd, rc, handle);
out_pending:
-#ifdef HAVE_QUOTA_SUPPORT
- if (quota_opc) {
- lquota_pending_commit(mds_quota_interface_ref, obd,
- qids, rec_pending, 1);
- /* Trigger dqacq for the parent owner. If failed,
- * the next call for lquota_chkquota will process it. */
- lquota_adjust(mds_quota_interface_ref, obd, 0, qids, rc,
- quota_opc);
- }
-#endif
return rc;
}
if (rc)
return rc;
- return mdd_declare_object_kill(env, obj, ma, handle);
+ return mdo_declare_destroy(env, obj, handle);
}
/* caller should take a lock before calling */
struct thandle *th)
{
int rc = 0;
- int reset = 1;
int is_dir = S_ISDIR(ma->ma_attr.la_mode);
ENTRY;
PFID(mdd_object_fid(obj)),
obj->mod_count);
} else {
- rc = mdd_object_kill(env, obj, ma, th);
- if (rc == 0)
- reset = 0;
+ rc = mdo_destroy(env, obj, th);
}
}
- if (reset)
- ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
RETURN(rc);
}
struct mdd_device *mdd = mdo2mdd(pobj);
struct dynlock_handle *dlh;
struct thandle *handle;
-#ifdef HAVE_QUOTA_SUPPORT
- struct obd_device *obd = mdd->mdd_obd_dev;
- struct mds_obd *mds = &obd->u.mds;
- unsigned int qcids[MAXQUOTAS] = { 0, 0 };
- unsigned int qpids[MAXQUOTAS] = { 0, 0 };
- int quota_opc = 0;
-#endif
int rc, is_dir;
ENTRY;
rc = mdd_la_get(env, mdd_cobj, cattr,
mdd_object_capa(env, mdd_cobj));
-#ifdef HAVE_QUOTA_SUPPORT
- if (mds->mds_quota && ma->ma_valid & MA_INODE &&
- ma->ma_attr.la_nlink == 0) {
- struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
-
- rc = mdd_la_get(env, mdd_pobj, la_tmp, BYPASS_CAPA);
- if (!rc) {
- mdd_quota_wrapper(la_tmp, qpids);
- if (mdd_cobj->mod_count == 0) {
- quota_opc = FSFILT_OP_UNLINK;
- mdd_quota_wrapper(&ma->ma_attr, qcids);
- } else {
- quota_opc = FSFILT_OP_UNLINK_PARTIAL_PARENT;
- }
- }
- }
-#endif
if (!is_dir)
- /* old files may not have link ea; ignore errors */
- mdd_links_rename(env, mdd_cobj, mdo2fid(mdd_pobj),
- lname, NULL, NULL, handle);
+ /* old files may not have link ea; ignore errors */
+ mdd_links_del(env, mdd_cobj, mdo2fid(mdd_pobj), lname, handle);
/* if object is removed then we can't get its attrs, use last get */
if (cattr->la_nlink == 0) {
stop:
mdd_trans_stop(env, mdd, rc, handle);
-#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 3, 55, 0)
- if (rc == 0 && ma->ma_valid & MA_COOKIE && ma->ma_valid & MA_LOV &&
- ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_UNLINK_DESTROY)
- /* Since echo client is incapable of destorying ost object,
- * it will destory the object here. */
- rc = mdd_lovobj_unlink(env, mdd, mdd_cobj, la, ma, 1);
-#else
-#warning "please remove this after 2.4 (LOD/OSP)."
-#endif
-#ifdef HAVE_QUOTA_SUPPORT
- if (quota_opc)
- /* Trigger dqrel on the owner of child and parent. If failed,
- * the next call for lquota_chkquota will process it. */
- lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
- quota_opc);
-#endif
return rc;
}
}
-static int mdd_declare_create_data(const struct lu_env *env,
- struct mdd_device *mdd,
- struct mdd_object *obj,
- int lmm_size,
- struct thandle *handle)
-{
- struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
- int rc;
-
- buf->lb_buf = NULL;
- buf->lb_len = lmm_size;
- rc = mdo_declare_xattr_set(env, obj, buf, XATTR_NAME_LOV,
- 0, handle);
- if (rc)
- return rc;
-
- rc = mdd_declare_lov_objid_update(env, mdd, handle);
-
- return rc;
-}
-
static int mdd_create_data(const struct lu_env *env, struct md_object *pobj,
struct md_object *cobj, const struct md_op_spec *spec,
struct md_attr *ma)
struct mdd_device *mdd = mdo2mdd(cobj);
struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
struct mdd_object *son = md2mdd_obj(cobj);
- struct lov_mds_md *lmm = NULL;
- int lmm_size = 0;
struct thandle *handle;
- struct lu_attr *attr = &mdd_env_info(env)->mti_la_for_fix;
+ const struct lu_buf *buf;
+ struct lu_attr *attr = &mdd_env_info(env)->mti_cattr;
int rc;
ENTRY;
+ /* do not let users to create stripes via .lustre/
+ * mdd_obf_setup() sets IMMUTE_OBJ on this directory */
+ if (pobj && mdd_pobj->mod_flags & IMMUTE_OBJ)
+ RETURN(-ENOENT);
+
rc = mdd_cd_sanity_check(env, son);
if (rc)
RETURN(rc);
if (!md_should_create(spec->sp_cr_flags))
RETURN(0);
- lmm_size = ma->ma_lmm_size;
-
- rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size, spec, ma);
- if (rc)
- RETURN(rc);
+ /*
+ * there are following use cases for this function:
+ * 1) late striping - file was created with MDS_OPEN_DELAY_CREATE
+ * striping can be specified or not
+ * 2) CMD?
+ */
rc = mdd_la_get(env, son, attr, mdd_object_capa(env, son));
if (rc)
RETURN(rc);
if (IS_ERR(handle))
GOTO(out_free, rc = PTR_ERR(handle));
- rc = mdd_declare_create_data(env, mdd, son, lmm_size, handle);
- if (rc)
- GOTO(stop, rc);
-
- rc = mdd_trans_start(env, mdd, handle);
- if (rc)
- GOTO(stop, rc);
-
/*
* XXX: Setting the lov ea is not locked but setting the attr is locked?
* Should this be fixed?
*/
+ CDEBUG(D_OTHER, "ea %p/%u, cr_flags %Lo, no_create %u\n",
+ spec->u.sp_ea.eadata, spec->u.sp_ea.eadatalen,
+ spec->sp_cr_flags, spec->no_create);
+
+ if (spec->no_create) {
+ /* replay case */
+ buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
+ spec->u.sp_ea.eadatalen);
+ } else if (!(spec->sp_cr_flags & MDS_OPEN_HAS_OBJS)) {
+ if (spec->sp_cr_flags & MDS_OPEN_HAS_EA) {
+ /* lfs setstripe */
+ buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
+ spec->u.sp_ea.eadatalen);
+ } else {
+ buf = &LU_BUF_NULL;
+ }
+ } else {
+ /* MDS_OPEN_HAS_OBJS is not used anymore ? */
+ LBUG();
+ }
- /* Replay creates has objects already */
-#if 0
- if (spec->no_create) {
- CDEBUG(D_INFO, "we already have lov ea\n");
- rc = mdd_lov_set_md(env, mdd_pobj, son,
- (struct lov_mds_md *)spec->u.sp_ea.eadata,
- spec->u.sp_ea.eadatalen, handle, 0);
- } else
-#endif
- /* No need mdd_lsm_sanity_check here */
- rc = mdd_lov_set_md(env, mdd_pobj, son, lmm,
- lmm_size, handle, 0);
+ rc = dt_declare_xattr_set(env, mdd_object_child(son), buf,
+ XATTR_NAME_LOV, 0, handle);
+ if (rc)
+ GOTO(stop, rc);
- /* update lov_objid data, must be before transaction stop! */
- if (rc == 0)
- mdd_lov_objid_update(mdd, lmm);
+ rc = mdd_trans_start(env, mdd, handle);
+ if (rc)
+ GOTO(stop, rc);
+ rc = dt_xattr_set(env, mdd_object_child(son), buf, XATTR_NAME_LOV,
+ 0, handle, mdd_object_capa(env, son));
stop:
- mdd_trans_stop(env, mdd, rc, handle);
+ mdd_trans_stop(env, mdd, rc, handle);
out_free:
- /* Finish mdd_lov_create() stuff. */
- /* if no_create == 0 (not replay), we free lmm allocated by
- * mdd_lov_create() */
- mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
- RETURN(rc);
+ RETURN(rc);
}
/* Get fid from name and parent */
* (2) maybe, the child attributes should be set in OSD when creation.
*/
+ /*
+ * inode mode has been set in creation time, and it's based on umask,
+ * la_mode and acl, don't set here again! (which will go wrong
+ * because below function doesn't consider umask).
+ * I'd suggest set all object attributes in creation time, see above.
+ */
+ attr->la_valid &= ~LA_MODE;
rc = mdd_attr_set_internal(env, child, attr, handle, 0);
if (rc != 0)
RETURN(rc);
static int mdd_declare_create(const struct lu_env *env, struct mdd_device *mdd,
struct mdd_object *p, struct mdd_object *c,
const struct lu_name *name,
- struct lu_attr *attr, int lmm_size,
+ struct lu_attr *attr,
int got_def_acl,
struct thandle *handle,
const struct md_op_spec *spec)
{
struct mdd_thread_info *info = mdd_env_info(env);
- struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
int rc = 0;
rc = mdd_declare_object_create_internal(env, p, c, attr, handle, spec);
if (rc)
GOTO(out, rc);
- rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_LOV,
- 0, handle);
- if (rc)
- GOTO(out, rc);
+ /* replay case, create LOV EA from client data */
+ if (spec->no_create || (spec->sp_cr_flags & MDS_OPEN_HAS_EA)) {
+ const struct lu_buf *buf;
+
+ buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
+ spec->u.sp_ea.eadatalen);
+ rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_LOV,
+ 0, handle);
+ if (rc)
+ GOTO(out, rc);
+ }
if (S_ISLNK(attr->la_mode)) {
rc = dt_declare_record_write(env, mdd_object_child(c),
if (rc)
return rc;
- rc = mdd_declare_lov_objid_update(env, mdd, handle);
-
out:
return rc;
}
struct mdd_object *son = md2mdd_obj(child);
struct mdd_device *mdd = mdo2mdd(pobj);
struct lu_attr *attr = &ma->ma_attr;
- struct lov_mds_md *lmm = NULL;
struct thandle *handle;
struct lu_attr *pattr = &info->mti_pattr;
struct dynlock_handle *dlh;
const char *name = lname->ln_name;
- int rc, created = 0, initialized = 0, inserted = 0, lmm_size = 0;
+ int rc, created = 0, initialized = 0, inserted = 0;
int got_def_acl = 0;
-#ifdef HAVE_QUOTA_SUPPORT
- struct obd_device *obd = mdd->mdd_obd_dev;
- struct obd_export *exp = md_quota(env)->mq_exp;
- struct mds_obd *mds = &obd->u.mds;
- unsigned int qcids[MAXQUOTAS] = { 0, 0 };
- unsigned int qpids[MAXQUOTAS] = { 0, 0 };
- int quota_opc = 0, block_count = 0;
- int inode_pending[MAXQUOTAS] = { 0, 0 };
- int block_pending[MAXQUOTAS] = { 0, 0 };
- int parent_pending[MAXQUOTAS] = { 0, 0 };
-#endif
ENTRY;
/*
if (rc)
RETURN(rc);
-#ifdef HAVE_QUOTA_SUPPORT
- if (mds->mds_quota) {
- int same = 0;
-
- quota_opc = FSFILT_OP_CREATE;
- mdd_quota_wrapper(attr, qcids);
- mdd_quota_wrapper(pattr, qpids);
- /* get file quota for child */
- lquota_chkquota(mds_quota_interface_ref, obd, exp,
- qcids, inode_pending, 1, NULL, 0, NULL,
- 0);
- switch (attr->la_mode & S_IFMT) {
- case S_IFLNK:
- case S_IFDIR:
- block_count = 2;
- break;
- case S_IFREG:
- block_count = 1;
- break;
- }
- if (qcids[USRQUOTA] == qpids[USRQUOTA] &&
- qcids[GRPQUOTA] == qpids[GRPQUOTA]) {
- block_count += 1;
- same = 1;
- }
- /* get block quota for child and parent */
- if (block_count)
- lquota_chkquota(mds_quota_interface_ref, obd,
- exp, qcids, block_pending,
- block_count, NULL,
- LQUOTA_FLAGS_BLK, NULL, 0);
- if (!same)
- lquota_chkquota(mds_quota_interface_ref, obd,
- exp, qpids, parent_pending, 1,
- NULL, LQUOTA_FLAGS_BLK, NULL,
- 0);
- }
-#endif
-
if (OBD_FAIL_CHECK(OBD_FAIL_MDS_DQACQ_NET))
- GOTO(out_pending, rc = -EINPROGRESS);
-
- /*
- * No RPC inside the transaction, so OST objects should be created at
- * first.
- */
- if (S_ISREG(attr->la_mode)) {
- lmm_size = ma->ma_lmm_size;
- rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size,
- spec, ma);
- if (rc)
- GOTO(out_pending, rc);
- }
+ GOTO(out_free, rc = -EINPROGRESS);
if (!S_ISLNK(attr->la_mode)) {
struct lu_buf *acl_buf;
GOTO(out_free, rc = PTR_ERR(handle));
rc = mdd_declare_create(env, mdd, mdd_pobj, son, lname, attr,
- got_def_acl, lmm_size, handle, spec);
+ got_def_acl, handle, spec);
if (rc)
GOTO(out_stop, rc);
rc = mdd_object_initialize(env, mdo2fid(mdd_pobj), lname,
son, attr, handle, spec);
+
+ /*
+ * in case of replay we just set LOVEA provided by the client
+ * XXX: I think it would be interesting to try "old" way where
+ * MDT calls this xattr_set(LOV) in a different transaction.
+ * probably this way we code can be made better.
+ */
+ if (rc == 0 &&
+ (spec->no_create || (spec->sp_cr_flags & MDS_OPEN_HAS_EA))) {
+ const struct lu_buf *buf;
+
+ buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
+ spec->u.sp_ea.eadatalen);
+ rc = mdo_xattr_set(env, son, buf, XATTR_NAME_LOV, 0, handle,
+ BYPASS_CAPA);
+ }
mdd_write_unlock(env, son);
if (rc)
/*
inserted = 1;
- /* No need mdd_lsm_sanity_check here */
- rc = mdd_lov_set_md(env, mdd_pobj, son, lmm, lmm_size, handle, 0);
- if (rc) {
- CERROR("error on stripe info copy %d \n", rc);
- GOTO(cleanup, rc);
- }
-
if (S_ISLNK(attr->la_mode)) {
- struct md_ucred *uc = md_ucred(env);
+ struct lu_ucred *uc = lu_ucred_assert(env);
struct dt_object *dt = mdd_object_child(son);
const char *target_name = spec->u.sp_symname;
int sym_len = strlen(target_name);
loff_t pos = 0;
buf = mdd_buf_get_const(env, target_name, sym_len);
- rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle,
- mdd_object_capa(env, son),
- uc->mu_cap &
- CFS_CAP_SYS_RESOURCE_MASK);
+ rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle,
+ mdd_object_capa(env, son),
+ uc->uc_cap &
+ CFS_CAP_SYS_RESOURCE_MASK);
if (rc == sym_len)
rc = 0;
mdd_write_unlock(env, son);
}
- /* update lov_objid data, must be before transaction stop! */
- if (rc == 0)
- mdd_lov_objid_update(mdd, lmm);
-
mdd_pdo_write_unlock(env, mdd_pobj, dlh);
out_trans:
if (rc == 0)
out_stop:
mdd_trans_stop(env, mdd, rc, handle);
out_free:
- /* finish lov_create stuff, free all temporary data */
- mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
-out_pending:
-#ifdef HAVE_QUOTA_SUPPORT
- if (quota_opc) {
- lquota_pending_commit(mds_quota_interface_ref, obd, qcids,
- inode_pending, 0);
- lquota_pending_commit(mds_quota_interface_ref, obd, qcids,
- block_pending, 1);
- lquota_pending_commit(mds_quota_interface_ref, obd, qpids,
- parent_pending, 1);
- /* Trigger dqacq on the owner of child and parent. If failed,
- * the next call for lquota_chkquota will process it. */
- lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
- quota_opc);
- }
-#endif
/* The child object shouldn't be cached anymore */
if (rc)
- cfs_set_bit(LU_OBJECT_HEARD_BANSHEE,
+ set_bit(LU_OBJECT_HEARD_BANSHEE,
&child->mo_lu.lo_header->loh_flags);
return rc;
}
struct thandle *handle;
const struct lu_fid *tpobj_fid = mdo2fid(mdd_tpobj);
const struct lu_fid *spobj_fid = mdo2fid(mdd_spobj);
- int is_dir;
+ bool is_dir;
+ bool tobj_ref = 0;
+ bool tobj_locked = 0;
unsigned cl_flags = 0;
int rc, rc2;
-
-#ifdef HAVE_QUOTA_SUPPORT
- struct obd_device *obd = mdd->mdd_obd_dev;
- struct obd_export *exp = md_quota(env)->mq_exp;
- struct mds_obd *mds = &obd->u.mds;
- unsigned int qspids[MAXQUOTAS] = { 0, 0 };
- unsigned int qtcids[MAXQUOTAS] = { 0, 0 };
- unsigned int qtpids[MAXQUOTAS] = { 0, 0 };
- int quota_copc = 0, quota_popc = 0;
- int rec_pending[MAXQUOTAS] = { 0, 0 };
-#endif
ENTRY;
if (tobj)
mdd_tobj = md2mdd_obj(tobj);
-#ifdef HAVE_QUOTA_SUPPORT
- if (mds->mds_quota) {
- struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
-
- rc = mdd_la_get(env, mdd_spobj, la_tmp, BYPASS_CAPA);
- if (!rc) {
- mdd_quota_wrapper(la_tmp, qspids);
- if (!tobj) {
- rc = mdd_la_get(env, mdd_tpobj, la_tmp,
- BYPASS_CAPA);
- if (!rc) {
- void *data = NULL;
- mdd_data_get(env, mdd_tpobj, &data);
- quota_popc = FSFILT_OP_LINK;
- mdd_quota_wrapper(la_tmp, qtpids);
- /* get block quota for target parent */
- lquota_chkquota(mds_quota_interface_ref,
- obd, exp, qtpids,
- rec_pending, 1, NULL,
- LQUOTA_FLAGS_BLK,
- data, 1);
- }
- }
- }
- }
-#endif
mdd_sobj = mdd_object_find(env, mdd, lf);
handle = mdd_trans_create(env, mdd);
*/
if (tobj && mdd_object_exists(mdd_tobj)) {
mdd_write_lock(env, mdd_tobj, MOR_TGT_CHILD);
+ tobj_locked = 1;
if (mdd_is_dead_obj(mdd_tobj)) {
- mdd_write_unlock(env, mdd_tobj);
/* shld not be dead, something is wrong */
CERROR("tobj is dead, something is wrong\n");
rc = -EINVAL;
/* Remove dot reference. */
if (S_ISDIR(tg_attr->la_mode))
mdo_ref_del(env, mdd_tobj, handle);
+ tobj_ref = 1;
/* fetch updated nlink */
rc = mdd_la_get(env, mdd_tobj, tg_attr,
mdd_object_capa(env, mdd_tobj));
- if (rc)
+ if (rc != 0) {
+ CERROR("%s: Failed to get nlink for tobj "
+ DFID": rc = %d\n",
+ mdd2obd_dev(mdd)->obd_name,
+ PFID(tpobj_fid), rc);
GOTO(fixup_tpobj, rc);
+ }
- la->la_valid = LA_CTIME;
- rc = mdd_attr_check_set_internal(env, mdd_tobj, la, handle, 0);
- if (rc)
- GOTO(fixup_tpobj, rc);
+ la->la_valid = LA_CTIME;
+ rc = mdd_attr_check_set_internal(env, mdd_tobj, la, handle, 0);
+ if (rc != 0) {
+ CERROR("%s: Failed to set ctime for tobj "
+ DFID": rc = %d\n",
+ mdd2obd_dev(mdd)->obd_name,
+ PFID(tpobj_fid), rc);
+ GOTO(fixup_tpobj, rc);
+ }
/* XXX: this transfer to ma will be removed with LOD/OSP */
ma->ma_attr = *tg_attr;
ma->ma_valid |= MA_INODE;
- rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
- mdd_write_unlock(env, mdd_tobj);
- if (rc)
- GOTO(fixup_tpobj, rc);
+ rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
+ if (rc != 0) {
+ CERROR("%s: Failed to unlink tobj "
+ DFID": rc = %d\n",
+ mdd2obd_dev(mdd)->obd_name,
+ PFID(tpobj_fid), rc);
+ GOTO(fixup_tpobj, rc);
+ }
/* fetch updated nlink */
rc = mdd_la_get(env, mdd_tobj, tg_attr,
mdd_object_capa(env, mdd_tobj));
- if (rc)
+ if (rc != 0) {
+ CERROR("%s: Failed to get nlink for tobj "
+ DFID": rc = %d\n",
+ mdd2obd_dev(mdd)->obd_name,
+ PFID(tpobj_fid), rc);
GOTO(fixup_tpobj, rc);
+ }
/* XXX: this transfer to ma will be removed with LOD/OSP */
ma->ma_attr = *tg_attr;
ma->ma_valid |= MA_INODE;
- if (so_attr->la_nlink == 0) {
+ if (so_attr->la_nlink == 0)
cl_flags |= CLF_RENAME_LAST;
-#ifdef HAVE_QUOTA_SUPPORT
- if (mds->mds_quota && mdd_tobj->mod_count == 0) {
- quota_copc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
- mdd_quota_wrapper(&ma->ma_attr, qtcids);
- }
-#endif
- }
}
la->la_valid = LA_CTIME | LA_MTIME;
if (rc == 0 && mdd_sobj) {
mdd_write_lock(env, mdd_sobj, MOR_SRC_CHILD);
- rc = mdd_links_rename(env, mdd_sobj, mdo2fid(mdd_spobj), lsname,
- mdo2fid(mdd_tpobj), ltname, handle);
+ rc = mdd_links_rename(env, mdd_sobj, mdo2fid(mdd_spobj), lsname,
+ mdo2fid(mdd_tpobj), ltname, handle, 0, 0);
if (rc == -ENOENT)
/* Old files might not have EA entry */
mdd_links_add(env, mdd_sobj, mdo2fid(mdd_spobj),
if (mdd_tobj && mdd_object_exists(mdd_tobj) &&
!mdd_is_dead_obj(mdd_tobj)) {
+ if (tobj_ref) {
+ mdo_ref_add(env, mdd_tobj, handle);
+ if (is_dir)
+ mdo_ref_add(env, mdd_tobj, handle);
+ }
+
rc2 = __mdd_index_insert(env, mdd_tpobj,
mdo2fid(mdd_tobj), tname,
is_dir, handle,
CWARN("sp obj fix error %d\n",rc2);
}
cleanup:
+ if (tobj_locked)
+ mdd_write_unlock(env, mdd_tobj);
if (likely(tdlh) && sdlh != tdlh)
mdd_pdo_write_unlock(env, mdd_tpobj, tdlh);
if (likely(sdlh))
if (mdd_sobj)
mdd_object_put(env, mdd_sobj);
out_pending:
-#ifdef HAVE_QUOTA_SUPPORT
- if (mds->mds_quota) {
- if (quota_popc)
- lquota_pending_commit(mds_quota_interface_ref, obd,
- qtpids, rec_pending, 1);
-
- if (quota_copc) {
- /* Trigger dqrel on the source owner of parent.
- * If failed, the next call for lquota_chkquota will
- * process it. */
- lquota_adjust(mds_quota_interface_ref, obd, 0, qspids, rc,
- FSFILT_OP_UNLINK_PARTIAL_PARENT);
-
- /* Trigger dqrel on the target owner of child.
- * If failed, the next call for lquota_chkquota
- * will process it. */
- lquota_adjust(mds_quota_interface_ref, obd, qtcids,
- qtpids, rc, quota_copc);
- }
- }
-#endif
return rc;
}
-/** enable/disable storing of hardlink info */
-int mdd_linkea_enable = 1;
-CFS_MODULE_PARM(mdd_linkea_enable, "d", int, 0644,
- "record hardlink info in EAs");
+/**
+ * The data that link search is done on.
+ */
+struct mdd_link_data {
+ /**
+ * Buffer to keep link EA body.
+ */
+ struct lu_buf *ml_buf;
+ /**
+ * The matched header, entry and its lenght in the EA
+ */
+ struct link_ea_header *ml_leh;
+ struct link_ea_entry *ml_lee;
+ int ml_reclen;
+};
+
+static int mdd_links_new(const struct lu_env *env,
+ struct mdd_link_data *ldata)
+{
+ ldata->ml_buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
+ if (ldata->ml_buf->lb_buf == NULL)
+ return -ENOMEM;
+ ldata->ml_leh = ldata->ml_buf->lb_buf;
+ ldata->ml_leh->leh_magic = LINK_EA_MAGIC;
+ ldata->ml_leh->leh_len = sizeof(struct link_ea_header);
+ ldata->ml_leh->leh_reccount = 0;
+ return 0;
+}
+
+/** Read the link EA into a temp buffer.
+ * Uses the mdd_thread_info::mti_big_buf since it is generally large.
+ * A pointer to the buffer is stored in \a ldata::ml_buf.
+ *
+ * \retval 0 or error
+ */
+int mdd_links_read(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ struct mdd_link_data *ldata)
+{
+ struct lustre_capa *capa;
+ struct link_ea_header *leh;
+ int rc;
+
+ /* First try a small buf */
+ LASSERT(env != NULL);
+ ldata->ml_buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
+ if (ldata->ml_buf->lb_buf == NULL)
+ return -ENOMEM;
+
+ if (!mdd_object_exists(mdd_obj))
+ return -ENODATA;
+
+ capa = mdd_object_capa(env, mdd_obj);
+ rc = mdo_xattr_get(env, mdd_obj, ldata->ml_buf,
+ XATTR_NAME_LINK, capa);
+ if (rc == -ERANGE) {
+ /* Buf was too small, figure out what we need. */
+ mdd_buf_put(ldata->ml_buf);
+ rc = mdo_xattr_get(env, mdd_obj, ldata->ml_buf,
+ XATTR_NAME_LINK, capa);
+ if (rc < 0)
+ return rc;
+ ldata->ml_buf = mdd_buf_alloc(env, rc);
+ if (ldata->ml_buf->lb_buf == NULL)
+ return -ENOMEM;
+ rc = mdo_xattr_get(env, mdd_obj, ldata->ml_buf,
+ XATTR_NAME_LINK, capa);
+ }
+ if (rc < 0)
+ return rc;
+
+ leh = ldata->ml_buf->lb_buf;
+ if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
+ leh->leh_magic = LINK_EA_MAGIC;
+ leh->leh_reccount = __swab32(leh->leh_reccount);
+ leh->leh_len = __swab64(leh->leh_len);
+ /* entries are swabbed by mdd_lee_unpack */
+ }
+ if (leh->leh_magic != LINK_EA_MAGIC)
+ return -EINVAL;
+ if (leh->leh_reccount == 0)
+ return -ENODATA;
+
+ ldata->ml_leh = leh;
+ return 0;
+}
/** Read the link EA into a temp buffer.
* Uses the name_buf since it is generally large.
* \retval ptr to \a lu_buf (always \a mti_big_buf)
*/
struct lu_buf *mdd_links_get(const struct lu_env *env,
- struct mdd_object *mdd_obj)
+ struct mdd_object *mdd_obj)
{
- struct lu_buf *buf;
- struct lustre_capa *capa;
- struct link_ea_header *leh;
- int rc;
-
- /* First try a small buf */
- buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
- if (buf->lb_buf == NULL)
- return ERR_PTR(-ENOMEM);
+ struct mdd_link_data ldata = { 0 };
+ int rc;
- if (!mdd_object_exists(mdd_obj))
- return ERR_PTR(-ENODATA);
-
- capa = mdd_object_capa(env, mdd_obj);
- rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
- if (rc == -ERANGE) {
- /* Buf was too small, figure out what we need. */
- mdd_buf_put(buf);
- rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
- if (rc < 0)
- return ERR_PTR(rc);
- buf = mdd_buf_alloc(env, rc);
- if (buf->lb_buf == NULL)
- return ERR_PTR(-ENOMEM);
- rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
- }
- if (rc < 0)
- return ERR_PTR(rc);
-
- leh = buf->lb_buf;
- if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
- leh->leh_magic = LINK_EA_MAGIC;
- leh->leh_reccount = __swab32(leh->leh_reccount);
- leh->leh_len = __swab64(leh->leh_len);
- /* entries are swabbed by mdd_lee_unpack */
- }
- if (leh->leh_magic != LINK_EA_MAGIC)
- return ERR_PTR(-EINVAL);
- if (leh->leh_reccount == 0)
- return ERR_PTR(-ENODATA);
+ rc = mdd_links_read(env, mdd_obj, &ldata);
+ return rc ? ERR_PTR(rc) : ldata.ml_buf;
+}
- return buf;
+static int mdd_links_write(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ struct mdd_link_data *ldata,
+ struct thandle *handle)
+{
+ const struct lu_buf *buf = mdd_buf_get_const(env, ldata->ml_buf->lb_buf,
+ ldata->ml_leh->leh_len);
+ return mdo_xattr_set(env, mdd_obj, buf, XATTR_NAME_LINK, 0, handle,
+ mdd_object_capa(env, mdd_obj));
}
/** Pack a link_ea_entry.
lname->ln_namelen = *reclen - sizeof(struct link_ea_entry);
}
-/** Add a record to the end of link ea buf */
-static int __mdd_links_add(const struct lu_env *env, struct lu_buf *buf,
- const struct lu_fid *pfid,
- const struct lu_name *lname)
-{
- struct link_ea_header *leh;
- struct link_ea_entry *lee;
- int reclen;
-
- if (lname == NULL || pfid == NULL)
- return -EINVAL;
-
- /* Make sure our buf is big enough for the new one */
- leh = buf->lb_buf;
- reclen = lname->ln_namelen + sizeof(struct link_ea_entry);
- if (leh->leh_len + reclen > buf->lb_len) {
- if (mdd_buf_grow(env, leh->leh_len + reclen) < 0)
- return -ENOMEM;
- }
-
- leh = buf->lb_buf;
- lee = buf->lb_buf + leh->leh_len;
- reclen = mdd_lee_pack(lee, lname, pfid);
- leh->leh_len += reclen;
- leh->leh_reccount++;
- return 0;
-}
-
static int mdd_declare_links_add(const struct lu_env *env,
struct mdd_object *mdd_obj,
struct thandle *handle)
*/
#define LINKEA_MAX_COUNT 128
-static int mdd_links_add(const struct lu_env *env,
- struct mdd_object *mdd_obj,
- const struct lu_fid *pfid,
- const struct lu_name *lname,
- struct thandle *handle, int first)
+/** Add a record to the end of link ea buf */
+static int mdd_links_add_buf(const struct lu_env *env,
+ struct mdd_link_data *ldata,
+ const struct lu_name *lname,
+ const struct lu_fid *pfid)
{
- struct lu_buf *buf;
- struct link_ea_header *leh;
- int rc;
- ENTRY;
+ LASSERT(ldata->ml_leh != NULL);
- if (!mdd_linkea_enable)
- RETURN(0);
+ if (lname == NULL || pfid == NULL)
+ return -EINVAL;
- buf = first ? ERR_PTR(-ENODATA) : mdd_links_get(env, mdd_obj);
- if (IS_ERR(buf)) {
- rc = PTR_ERR(buf);
- if (rc != -ENODATA) {
- CERROR("link_ea read failed %d "DFID"\n", rc,
- PFID(mdd_object_fid(mdd_obj)));
- RETURN (rc);
- }
- /* empty EA; start one */
- buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
- if (buf->lb_buf == NULL)
- RETURN(-ENOMEM);
- leh = buf->lb_buf;
- leh->leh_magic = LINK_EA_MAGIC;
- leh->leh_len = sizeof(struct link_ea_header);
- leh->leh_reccount = 0;
- }
+ /* Make sure our buf is big enough for the new one */
+ if (ldata->ml_leh->leh_reccount > LINKEA_MAX_COUNT)
+ return -EOVERFLOW;
- leh = buf->lb_buf;
- if (leh->leh_reccount > LINKEA_MAX_COUNT)
- RETURN(-EOVERFLOW);
+ ldata->ml_reclen = lname->ln_namelen + sizeof(struct link_ea_entry);
+ if (ldata->ml_leh->leh_len + ldata->ml_reclen >
+ ldata->ml_buf->lb_len) {
+ if (mdd_buf_grow(env, ldata->ml_leh->leh_len +
+ ldata->ml_reclen) < 0)
+ return -ENOMEM;
+ }
- rc = __mdd_links_add(env, buf, pfid, lname);
- if (rc)
- RETURN(rc);
+ ldata->ml_leh = ldata->ml_buf->lb_buf;
+ ldata->ml_lee = ldata->ml_buf->lb_buf + ldata->ml_leh->leh_len;
+ ldata->ml_reclen = mdd_lee_pack(ldata->ml_lee, lname, pfid);
+ ldata->ml_leh->leh_len += ldata->ml_reclen;
+ ldata->ml_leh->leh_reccount++;
+ CDEBUG(D_INODE, "New link_ea name '%.*s' is added\n",
+ lname->ln_namelen, lname->ln_name);
+ return 0;
+}
- leh = buf->lb_buf;
- rc = mdo_xattr_set(env, mdd_obj,
- mdd_buf_get_const(env, buf->lb_buf, leh->leh_len),
- XATTR_NAME_LINK, 0, handle,
- mdd_object_capa(env, mdd_obj));
- if (rc) {
- if (rc == -ENOSPC)
- CDEBUG(D_INODE, "link_ea add failed %d "DFID"\n", rc,
- PFID(mdd_object_fid(mdd_obj)));
- else
- CERROR("link_ea add failed %d "DFID"\n", rc,
- PFID(mdd_object_fid(mdd_obj)));
- }
+/** Del the current record from the link ea buf */
+static void mdd_links_del_buf(const struct lu_env *env,
+ struct mdd_link_data *ldata,
+ const struct lu_name *lname)
+{
+ LASSERT(ldata->ml_leh != NULL);
- if (buf->lb_len > OBD_ALLOC_BIG)
- /* if we vmalloced a large buffer drop it */
- mdd_buf_put(buf);
+ ldata->ml_leh->leh_reccount--;
+ ldata->ml_leh->leh_len -= ldata->ml_reclen;
+ memmove(ldata->ml_lee, (char *)ldata->ml_lee + ldata->ml_reclen,
+ (char *)ldata->ml_leh + ldata->ml_leh->leh_len -
+ (char *)ldata->ml_lee);
+ CDEBUG(D_INODE, "Old link_ea name '%.*s' is removed\n",
+ lname->ln_namelen, lname->ln_name);
- RETURN (rc);
}
-static int mdd_links_rename(const struct lu_env *env,
- struct mdd_object *mdd_obj,
- const struct lu_fid *oldpfid,
- const struct lu_name *oldlname,
- const struct lu_fid *newpfid,
- const struct lu_name *newlname,
- struct thandle *handle)
+/**
+ * Check if such a link exists in linkEA.
+ *
+ * \param mdd_obj object being handled
+ * \param pfid parent fid the link to be found for
+ * \param lname name in the parent's directory entry pointing to this object
+ * \param ldata link data the search to be done on
+ *
+ * \retval 0 success
+ * \retval -ENOENT link does not exist
+ * \retval -ve on error
+ */
+static int mdd_links_find(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ struct mdd_link_data *ldata,
+ const struct lu_name *lname,
+ const struct lu_fid *pfid)
{
- struct lu_buf *buf;
- struct link_ea_header *leh;
- struct link_ea_entry *lee;
- struct lu_name *tmpname = &mdd_env_info(env)->mti_name;
- struct lu_fid *tmpfid = &mdd_env_info(env)->mti_fid;
- int reclen = 0;
- int count;
- int rc, rc2 = 0;
- ENTRY;
+ struct lu_name *tmpname = &mdd_env_info(env)->mti_name2;
+ struct lu_fid *tmpfid = &mdd_env_info(env)->mti_fid;
+ int count;
+
+ LASSERT(ldata->ml_leh != NULL);
+
+ /* link #0 */
+ ldata->ml_lee = (struct link_ea_entry *)(ldata->ml_leh + 1);
+
+ for (count = 0; count < ldata->ml_leh->leh_reccount; count++) {
+ mdd_lee_unpack(ldata->ml_lee, &ldata->ml_reclen,
+ tmpname, tmpfid);
+ if (tmpname->ln_namelen == lname->ln_namelen &&
+ lu_fid_eq(tmpfid, pfid) &&
+ (strncmp(tmpname->ln_name, lname->ln_name,
+ tmpname->ln_namelen) == 0))
+ break;
+ ldata->ml_lee = (struct link_ea_entry *)((char *)ldata->ml_lee +
+ ldata->ml_reclen);
+ }
- if (!mdd_linkea_enable)
- RETURN(0);
+ if (count == ldata->ml_leh->leh_reccount) {
+ CDEBUG(D_INODE, "Old link_ea name '%.*s' not found\n",
+ lname->ln_namelen, lname->ln_name);
+ return -ENOENT;
+ }
+ return 0;
+}
- if (mdd_obj->mod_flags & DEAD_OBJ)
- /* No more links, don't bother */
- RETURN(0);
+static int __mdd_links_add(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ struct mdd_link_data *ldata,
+ const struct lu_name *lname,
+ const struct lu_fid *pfid,
+ int first, int check)
+{
+ int rc;
- buf = mdd_links_get(env, mdd_obj);
- if (IS_ERR(buf)) {
- rc = PTR_ERR(buf);
- if (rc == -ENODATA)
- CDEBUG(D_INODE, "link_ea read failed %d "DFID"\n",
- rc, PFID(mdd_object_fid(mdd_obj)));
- else
- CERROR("link_ea read failed %d "DFID"\n",
- rc, PFID(mdd_object_fid(mdd_obj)));
- RETURN(rc);
- }
- leh = buf->lb_buf;
- lee = (struct link_ea_entry *)(leh + 1); /* link #0 */
-
- /* Find the old record */
- for(count = 0; count < leh->leh_reccount; count++) {
- mdd_lee_unpack(lee, &reclen, tmpname, tmpfid);
- if (tmpname->ln_namelen == oldlname->ln_namelen &&
- lu_fid_eq(tmpfid, oldpfid) &&
- (strncmp(tmpname->ln_name, oldlname->ln_name,
- tmpname->ln_namelen) == 0))
- break;
- lee = (struct link_ea_entry *)((char *)lee + reclen);
- }
- if ((count + 1) > leh->leh_reccount) {
- CDEBUG(D_INODE, "Old link_ea name '%.*s' not found\n",
- oldlname->ln_namelen, oldlname->ln_name);
- GOTO(out, rc = -ENOENT);
- }
+ if (ldata->ml_leh == NULL) {
+ rc = first ? -ENODATA : mdd_links_read(env, mdd_obj, ldata);
+ if (rc) {
+ if (rc != -ENODATA)
+ return rc;
+ rc = mdd_links_new(env, ldata);
+ if (rc)
+ return rc;
+ }
+ }
- /* Remove the old record */
- leh->leh_reccount--;
- leh->leh_len -= reclen;
- memmove(lee, (char *)lee + reclen, (char *)leh + leh->leh_len -
- (char *)lee);
-
- /* If renaming, add the new record */
- if (newpfid != NULL) {
- /* if the add fails, we still delete the out-of-date old link */
- rc2 = __mdd_links_add(env, buf, newpfid, newlname);
- leh = buf->lb_buf;
- }
+ if (check) {
+ rc = mdd_links_find(env, mdd_obj, ldata, lname, pfid);
+ if (rc && rc != -ENOENT)
+ return rc;
+ if (rc == 0)
+ return -EEXIST;
+ }
+
+ return mdd_links_add_buf(env, ldata, lname, pfid);
+}
- rc = mdo_xattr_set(env, mdd_obj,
- mdd_buf_get_const(env, buf->lb_buf, leh->leh_len),
- XATTR_NAME_LINK, 0, handle,
- mdd_object_capa(env, mdd_obj));
+static int __mdd_links_del(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ struct mdd_link_data *ldata,
+ const struct lu_name *lname,
+ const struct lu_fid *pfid)
+{
+ int rc;
+ if (ldata->ml_leh == NULL) {
+ rc = mdd_links_read(env, mdd_obj, ldata);
+ if (rc)
+ return rc;
+ }
+
+ rc = mdd_links_find(env, mdd_obj, ldata, lname, pfid);
+ if (rc)
+ return rc;
+
+ mdd_links_del_buf(env, ldata, lname);
+ return 0;
+}
+
+static int mdd_links_rename(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ const struct lu_fid *oldpfid,
+ const struct lu_name *oldlname,
+ const struct lu_fid *newpfid,
+ const struct lu_name *newlname,
+ struct thandle *handle,
+ int first, int check)
+{
+ struct mdd_link_data ldata = { 0 };
+ int updated = 0;
+ int rc2 = 0;
+ int rc = 0;
+ ENTRY;
+
+ LASSERT(oldpfid != NULL || newpfid != NULL);
+
+ if (mdd_obj->mod_flags & DEAD_OBJ)
+ /* No more links, don't bother */
+ RETURN(0);
+
+ if (oldpfid != NULL) {
+ rc = __mdd_links_del(env, mdd_obj, &ldata,
+ oldlname, oldpfid);
+ if (rc) {
+ if ((check == 0) ||
+ (rc != -ENODATA && rc != -ENOENT))
+ GOTO(out, rc);
+ /* No changes done. */
+ rc = 0;
+ } else {
+ updated = 1;
+ }
+ }
+
+ /* If renaming, add the new record */
+ if (newpfid != NULL) {
+ /* even if the add fails, we still delete the out-of-date
+ * old link */
+ rc2 = __mdd_links_add(env, mdd_obj, &ldata,
+ newlname, newpfid, first, check);
+ if (rc2 == -EEXIST)
+ rc2 = 0;
+ else if (rc2 == 0)
+ updated = 1;
+ }
+
+ if (updated)
+ rc = mdd_links_write(env, mdd_obj, &ldata, handle);
+ EXIT;
out:
- if (rc == 0)
- rc = rc2;
- if (rc)
- CDEBUG(D_INODE, "link_ea mv/unlink '%.*s' failed %d "DFID"\n",
- oldlname->ln_namelen, oldlname->ln_name, rc,
- PFID(mdd_object_fid(mdd_obj)));
+ if (rc == 0)
+ rc = rc2;
+ if (rc) {
+ if (oldpfid == NULL)
+ CERROR("link_ea add '%.*s' failed %d "DFID"\n",
+ newlname->ln_namelen, newlname->ln_name,
+ rc, PFID(mdd_object_fid(mdd_obj)));
+ else if (newpfid == NULL)
+ CERROR("link_ea del '%.*s' failed %d "DFID"\n",
+ oldlname->ln_namelen, oldlname->ln_name,
+ rc, PFID(mdd_object_fid(mdd_obj)));
+ else
+ CERROR("link_ea rename '%.*s'->'%.*s' failed %d "
+ DFID"\n",
+ oldlname->ln_namelen, oldlname->ln_name,
+ newlname->ln_namelen, newlname->ln_name,
+ rc, PFID(mdd_object_fid(mdd_obj)));
+ }
+
+ if (ldata.ml_buf && ldata.ml_buf->lb_len > OBD_ALLOC_BIG)
+ /* if we vmalloced a large buffer drop it */
+ mdd_buf_put(ldata.ml_buf);
+
+ return rc;
+}
- if (buf->lb_len > OBD_ALLOC_BIG)
- /* if we vmalloced a large buffer drop it */
- mdd_buf_put(buf);
+static inline int mdd_links_add(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ const struct lu_fid *pfid,
+ const struct lu_name *lname,
+ struct thandle *handle, int first)
+{
+ return mdd_links_rename(env, mdd_obj, NULL, NULL,
+ pfid, lname, handle, first, 0);
+}
- RETURN (rc);
+static inline int mdd_links_del(const struct lu_env *env,
+ struct mdd_object *mdd_obj,
+ const struct lu_fid *pfid,
+ const struct lu_name *lname,
+ struct thandle *handle)
+{
+ return mdd_links_rename(env, mdd_obj, pfid, lname,
+ NULL, NULL, handle, 0, 0);
}
const struct md_dir_operations mdd_dir_ops = {
git://git.whamcloud.com / fs / lustre-release.git / blobdiff