static int mdd_object_print(const struct lu_env *env, void *cookie,
lu_printer_t p, const struct lu_object *o)
{
- return (*p)(env, cookie, LUSTRE_MDD_NAME"-object@%p", o);
+ struct mdd_object *mdd = lu2mdd_obj((struct lu_object *)o);
+ return (*p)(env, cookie, LUSTRE_MDD_NAME"-object@%p(open_count=%d, "
+ "valid=%x, cltime=%llu, flags=%lx)",
+ mdd, mdd->mod_count, mdd->mod_valid,
+ mdd->mod_cltime, mdd->mod_flags);
}
static const struct lu_object_operations mdd_lu_obj_ops = {
/** mdd_path() lookup structure. */
struct path_lookup_info {
__u64 pli_recno; /**< history point */
+ __u64 pli_currec; /**< current record */
struct lu_fid pli_fid;
struct lu_fid pli_fids[MAX_PATH_DEPTH]; /**< path, in fids */
struct mdd_object *pli_mdd_obj;
pli->pli_fids[pli->pli_fidcount] = *tmpfid;
}
- /* Verify that our path hasn't changed since we started the lookup */
+ /* Verify that our path hasn't changed since we started the lookup.
+ Record the current index, and verify the path resolves to the
+ same fid. If it does, then the path is correct as of this index. */
+ spin_lock(&mdd->mdd_cl.mc_lock);
+ pli->pli_currec = mdd->mdd_cl.mc_index;
+ spin_unlock(&mdd->mdd_cl.mc_lock);
rc = mdd_path2fid(env, mdd, ptr, &pli->pli_fid);
if (rc) {
CDEBUG(D_INFO, "mdd_path2fid(%s) failed %d\n", ptr, rc);
return rc;
}
+static int mdd_path_historic(const struct lu_env *env,
+ struct path_lookup_info *pli)
+{
+ return 0;
+}
+
/* Returns the full path to this fid, as of changelog record recno. */
static int mdd_path(const struct lu_env *env, struct md_object *obj,
- char *path, int pathlen, __u64 recno, int *linkno)
+ char *path, int pathlen, __u64 *recno, int *linkno)
{
struct path_lookup_info *pli;
int tries = 3;
RETURN(-ENOMEM);
pli->pli_mdd_obj = md2mdd_obj(obj);
- pli->pli_recno = recno;
+ pli->pli_recno = *recno;
pli->pli_path = path;
pli->pli_pathlen = pathlen;
pli->pli_linkno = *linkno;
while (tries-- && rc == -EAGAIN)
rc = mdd_path_current(env, pli);
-#if 0 /* We need old path names only for replication */
/* For historical path lookup, the current links may not have existed
* at "recno" time. We must switch over to earlier links/parents
* by using the changelog records. If the earlier parent doesn't
* We may ignore this problem for the initial implementation and
* state that an "original" hardlink must still exist for us to find
* historic path name. */
- if (pli->pli_recno != -1)
+ if (pli->pli_recno != -1) {
rc = mdd_path_historic(env, pli);
-#endif
-
- /* return next link index to caller */
- *linkno = pli->pli_linkno;
+ } else {
+ *recno = pli->pli_currec;
+ /* Return next link index to caller */
+ *linkno = pli->pli_linkno;
+ }
OBD_FREE_PTR(pli);
unsigned int qnids[MAXQUOTAS] = { 0, 0 };
unsigned int qoids[MAXQUOTAS] = { 0, 0 };
int quota_opc = 0, block_count = 0;
- int inode_pending = 0, block_pending = 0;
+ int inode_pending[MAXQUOTAS] = { 0, 0 };
+ int block_pending[MAXQUOTAS] = { 0, 0 };
#endif
ENTRY;
mdd_quota_wrapper(la_copy, qnids);
mdd_quota_wrapper(la_tmp, qoids);
/* get file quota for new owner */
- lquota_chkquota(mds_quota_interface_ref, obd,
- qnids[USRQUOTA], qnids[GRPQUOTA], 1,
- &inode_pending, NULL, 0, NULL, 0);
+ lquota_chkquota(mds_quota_interface_ref, obd, qnids,
+ inode_pending, 1, NULL, 0, NULL, 0);
block_count = (la_tmp->la_blocks + 7) >> 3;
if (block_count) {
void *data = NULL;
mdd_data_get(env, mdd_obj, &data);
/* get block quota for new owner */
lquota_chkquota(mds_quota_interface_ref, obd,
- qnids[USRQUOTA],
- qnids[GRPQUOTA],
- block_count, &block_pending,
- NULL, LQUOTA_FLAGS_BLK,
- data, 1);
+ qnids, block_pending,
+ block_count, NULL,
+ LQUOTA_FLAGS_BLK, data, 1);
}
}
}
}
#ifdef HAVE_QUOTA_SUPPORT
if (quota_opc) {
- if (inode_pending)
- lquota_pending_commit(mds_quota_interface_ref, obd,
- qnids[USRQUOTA], qnids[GRPQUOTA],
- inode_pending, 0);
- if (block_pending)
- lquota_pending_commit(mds_quota_interface_ref, obd,
- qnids[USRQUOTA], qnids[GRPQUOTA],
- block_pending, 1);
+ lquota_pending_commit(mds_quota_interface_ref, obd, qnids,
+ inode_pending, 0);
+ lquota_pending_commit(mds_quota_interface_ref, obd, qnids,
+ block_pending, 1);
/* Trigger dqrel/dqacq for original owner and new owner.
* If failed, the next call for lquota_chkquota will
* process it. */
struct mds_obd *mds = &obd->u.mds;
unsigned int qids[MAXQUOTAS] = { 0, 0 };
int quota_opc = 0, block_count = 0;
- int inode_pending = 0, block_pending = 0;
+ int inode_pending[MAXQUOTAS] = { 0, 0 };
+ int block_pending[MAXQUOTAS] = { 0, 0 };
#endif
int rc = 0;
ENTRY;
quota_opc = FSFILT_OP_CREATE_PARTIAL_CHILD;
mdd_quota_wrapper(&ma->ma_attr, qids);
/* get file quota for child */
- lquota_chkquota(mds_quota_interface_ref, obd, qids[USRQUOTA],
- qids[GRPQUOTA], 1, &inode_pending, NULL, 0,
- NULL, 0);
+ lquota_chkquota(mds_quota_interface_ref, obd, qids,
+ inode_pending, 1, NULL, 0, NULL, 0);
switch (ma->ma_attr.la_mode & S_IFMT) {
case S_IFLNK:
case S_IFDIR:
}
/* get block quota for child */
if (block_count)
- lquota_chkquota(mds_quota_interface_ref, obd,
- qids[USRQUOTA], qids[GRPQUOTA],
- block_count, &block_pending, NULL,
+ lquota_chkquota(mds_quota_interface_ref, obd, qids,
+ block_pending, block_count, NULL,
LQUOTA_FLAGS_BLK, NULL, 0);
}
#endif
out_pending:
#ifdef HAVE_QUOTA_SUPPORT
if (quota_opc) {
- if (inode_pending)
- lquota_pending_commit(mds_quota_interface_ref, obd,
- qids[USRQUOTA], qids[GRPQUOTA],
- inode_pending, 0);
- if (block_pending)
- lquota_pending_commit(mds_quota_interface_ref, obd,
- qids[USRQUOTA], qids[GRPQUOTA],
- block_pending, 1);
+ lquota_pending_commit(mds_quota_interface_ref, obd, qids,
+ inode_pending, 0);
+ lquota_pending_commit(mds_quota_interface_ref, obd, qids,
+ block_pending, 1);
/* Trigger dqacq on the owner of child. If failed,
* the next call for lquota_chkquota will process it. */
lquota_adjust(mds_quota_interface_ref, obd, qids, 0, rc,
- FSFILT_OP_CREATE_PARTIAL_CHILD);
+ quota_opc);
}
#endif
return rc;
struct md_attr *ma)
{
struct mdd_object *mdd_obj = md2mdd_obj(obj);
+ struct mdd_device *mdd = mdo2mdd(obj);
struct thandle *handle;
int rc;
int reset = 1;
/* release open count */
mdd_obj->mod_count --;
- if (mdd_obj->mod_count == 0) {
+ if (mdd_obj->mod_count == 0 && mdd_obj->mod_flags & ORPHAN_OBJ) {
/* remove link to object from orphan index */
- if (mdd_obj->mod_flags & ORPHAN_OBJ)
- __mdd_orphan_del(env, mdd_obj, handle);
+ rc = __mdd_orphan_del(env, mdd_obj, handle);
+ if (rc == 0) {
+ CDEBUG(D_HA, "Object "DFID" is deleted from orphan "
+ "list, OSS objects to be destroyed.\n",
+ PFID(mdd_object_fid(mdd_obj)));
+ } else {
+ CERROR("Object "DFID" can not be deleted from orphan "
+ "list, maybe cause OST objects can not be "
+ "destroyed (err: %d).\n",
+ PFID(mdd_object_fid(mdd_obj)), rc);
+ /* If object was not deleted from orphan list, do not
+ * destroy OSS objects, which will be done when next
+ * recovery. */
+ GOTO(out, rc);
+ }
}
rc = mdd_iattr_get(env, mdd_obj, ma);
- if (rc == 0) {
- if (mdd_obj->mod_count == 0 && ma->ma_attr.la_nlink == 0) {
- rc = mdd_object_kill(env, mdd_obj, ma);
+ /* Object maybe not in orphan list originally, it is rare case for
+ * mdd_finish_unlink() failure. */
+ if (rc == 0 && ma->ma_attr.la_nlink == 0) {
#ifdef HAVE_QUOTA_SUPPORT
- if (mds->mds_quota) {
- quota_opc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
- mdd_quota_wrapper(&ma->ma_attr, qids);
- }
+ if (mds->mds_quota) {
+ quota_opc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
+ mdd_quota_wrapper(&ma->ma_attr, qids);
+ }
#endif
- if (rc == 0)
- reset = 0;
+ /* MDS_CLOSE_CLEANUP means destroy OSS objects by MDS. */
+ if (ma->ma_valid & MA_FLAGS &&
+ ma->ma_attr_flags & MDS_CLOSE_CLEANUP) {
+ rc = mdd_lov_destroy(env, mdd, mdd_obj, &ma->ma_attr);
+ } else {
+ rc = mdd_object_kill(env, mdd_obj, ma);
+ if (rc == 0)
+ reset = 0;
}
+
+ if (rc != 0)
+ CERROR("Error when prepare to delete Object "DFID" , "
+ "which will cause OST objects can not be "
+ "destroyed.\n", PFID(mdd_object_fid(mdd_obj)));
}
+ EXIT;
+out:
if (reset)
ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
lquota_adjust(mds_quota_interface_ref, obd, qids, 0, rc,
quota_opc);
#endif
- RETURN(rc);
+ return rc;
}
/*
memcpy(ent->lde_name, name, len);
result = mdd_append_attrs(env, mdd, attr, iops, it, ent);
+ if (result == -ESTALE)
+ goto next;
if (result != 0)
goto out;
} else {
next:
result = iops->next(env, it);
+ if (result == -ESTALE)
+ goto next;
} while (result == 0);
out:
return rc;
}
-static int mdd_readpage(const struct lu_env *env, struct md_object *obj,
- const struct lu_rdpg *rdpg)
+int mdd_readpage(const struct lu_env *env, struct md_object *obj,
+ const struct lu_rdpg *rdpg)
{
struct mdd_object *mdd_obj = md2mdd_obj(obj);
int rc;
return next->do_ops->do_object_sync(env, next);
}
+static dt_obj_version_t mdd_version_get(const struct lu_env *env,
+ struct md_object *obj)
+{
+ struct mdd_object *mdd_obj = md2mdd_obj(obj);
+
+ LASSERT(mdd_object_exists(mdd_obj));
+ return do_version_get(env, mdd_object_child(mdd_obj));
+}
+
+static void mdd_version_set(const struct lu_env *env, struct md_object *obj,
+ dt_obj_version_t version)
+{
+ struct mdd_object *mdd_obj = md2mdd_obj(obj);
+
+ LASSERT(mdd_object_exists(mdd_obj));
+ return do_version_set(env, mdd_object_child(mdd_obj), version);
+}
+
const struct md_object_operations mdd_obj_ops = {
.moo_permission = mdd_permission,
.moo_attr_get = mdd_attr_get,
.moo_readlink = mdd_readlink,
.moo_capa_get = mdd_capa_get,
.moo_object_sync = mdd_object_sync,
+ .moo_version_get = mdd_version_get,
+ .moo_version_set = mdd_version_set,
.moo_path = mdd_path,
};
git://git.whamcloud.com / fs / lustre-release.git / blobdiff