* Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2011, 2013, Intel Corporation.
+ * Copyright (c) 2011, 2015, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
#define DEBUG_SUBSYSTEM S_OSD
#include <linux/module.h>
+#include <linux/user_namespace.h>
+#ifdef HAVE_UIDGID_HEADER
+# include <linux/uidgid.h>
+#endif
/* LUSTRE_VERSION_CODE */
#include <lustre_ver.h>
/* XATTR_{REPLACE,CREATE} */
#include <linux/xattr.h>
+#include <ldiskfs/ldiskfs.h>
+#include <ldiskfs/xattr.h>
+#include <ldiskfs/ldiskfs_extents.h>
+#undef ENTRY
/*
* struct OBD_{ALLOC,FREE}*()
* OBD_FAIL_CHECK
#include <md_object.h>
#include <lustre_quota.h>
-#include <ldiskfs/xattr.h>
#include <lustre_linkea.h>
int ldiskfs_pdo = 1;
-CFS_MODULE_PARM(ldiskfs_pdo, "i", int, 0644,
- "ldiskfs with parallel directory operations");
+module_param(ldiskfs_pdo, int, 0644);
+MODULE_PARM_DESC(ldiskfs_pdo, "ldiskfs with parallel directory operations");
int ldiskfs_track_declares_assert;
-CFS_MODULE_PARM(ldiskfs_track_declares_assert, "i", int, 0644,
- "LBUG during tracking of declares");
+module_param(ldiskfs_track_declares_assert, int, 0644);
+MODULE_PARM_DESC(ldiskfs_track_declares_assert, "LBUG during tracking of declares");
/* Slab to allocate dynlocks */
struct kmem_cache *dynlock_cachep;
+/* Slab to allocate osd_it_ea */
+struct kmem_cache *osd_itea_cachep;
+
static struct lu_kmem_descr ldiskfs_caches[] = {
{
.ckd_cache = &dynlock_cachep,
.ckd_size = sizeof(struct dynlock_handle)
},
{
+ .ckd_cache = &osd_itea_cachep,
+ .ckd_name = "osd_itea_cache",
+ .ckd_size = sizeof(struct osd_it_ea)
+ },
+ {
.ckd_cache = NULL
}
};
static const struct dt_index_operations osd_index_iam_ops;
static const struct dt_index_operations osd_index_ea_ops;
+static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
+ const struct lu_fid *fid);
+static int osd_process_scheduled_agent_removals(const struct lu_env *env,
+ struct osd_device *osd);
+
int osd_trans_declare_op2rb[] = {
[OSD_OT_ATTR_SET] = OSD_OT_ATTR_SET,
[OSD_OT_PUNCH] = OSD_OT_MAX,
[OSD_OT_WRITE] = OSD_OT_WRITE,
[OSD_OT_INSERT] = OSD_OT_DELETE,
[OSD_OT_DELETE] = OSD_OT_INSERT,
- [OSD_OT_UPDATE] = OSD_OT_MAX,
[OSD_OT_QUOTA] = OSD_OT_MAX,
};
}
/*
+ * the following set of functions are used to maintain per-thread
+ * cache of FID->ino mapping. this mechanism is needed to resolve
+ * FID to inode at dt_insert() which in turn stores ino in the
+ * directory entries to keep ldiskfs compatible with ext[34].
+ * due to locking-originated restrictions we can't lookup ino
+ * using LU cache (deadlock is possible). lookup using OI is quite
+ * expensive. so instead we maintain this cache and methods like
+ * dt_create() fill it. so in the majority of cases dt_insert() is
+ * able to find needed mapping in lockless manner.
+ */
+static struct osd_idmap_cache *
+osd_idc_find(const struct lu_env *env, struct osd_device *osd,
+ const struct lu_fid *fid)
+{
+ struct osd_thread_info *oti = osd_oti_get(env);
+ struct osd_idmap_cache *idc = oti->oti_ins_cache;
+ int i;
+ for (i = 0; i < oti->oti_ins_cache_used; i++) {
+ if (!lu_fid_eq(&idc[i].oic_fid, fid))
+ continue;
+ if (idc[i].oic_dev != osd)
+ continue;
+
+ return idc + i;
+ }
+
+ return NULL;
+}
+
+static struct osd_idmap_cache *
+osd_idc_add(const struct lu_env *env, struct osd_device *osd,
+ const struct lu_fid *fid)
+{
+ struct osd_thread_info *oti = osd_oti_get(env);
+ struct osd_idmap_cache *idc;
+ int i;
+
+ if (unlikely(oti->oti_ins_cache_used >= oti->oti_ins_cache_size)) {
+ i = oti->oti_ins_cache_size * 2;
+ if (i == 0)
+ i = OSD_INS_CACHE_SIZE;
+ OBD_ALLOC(idc, sizeof(*idc) * i);
+ if (idc == NULL)
+ return ERR_PTR(-ENOMEM);
+ if (oti->oti_ins_cache != NULL) {
+ memcpy(idc, oti->oti_ins_cache,
+ oti->oti_ins_cache_used * sizeof(*idc));
+ OBD_FREE(oti->oti_ins_cache,
+ oti->oti_ins_cache_used * sizeof(*idc));
+ }
+ oti->oti_ins_cache = idc;
+ oti->oti_ins_cache_size = i;
+ }
+
+ idc = oti->oti_ins_cache + oti->oti_ins_cache_used++;
+ idc->oic_fid = *fid;
+ idc->oic_dev = osd;
+ idc->oic_lid.oii_ino = 0;
+ idc->oic_lid.oii_gen = 0;
+ idc->oic_remote = 0;
+
+ return idc;
+}
+
+/*
+ * lookup mapping for the given fid in the cache, initialize a
+ * new one if not found. the initialization checks whether the
+ * object is local or remote. for local objects, OI is used to
+ * learn ino/generation. the function is used when the caller
+ * has no information about the object, e.g. at dt_insert().
+ */
+static struct osd_idmap_cache *
+osd_idc_find_or_init(const struct lu_env *env, struct osd_device *osd,
+ const struct lu_fid *fid)
+{
+ struct osd_idmap_cache *idc;
+ int rc;
+
+ idc = osd_idc_find(env, osd, fid);
+ LASSERT(!IS_ERR(idc));
+ if (idc != NULL)
+ return idc;
+
+ /* new mapping is needed */
+ idc = osd_idc_add(env, osd, fid);
+ if (IS_ERR(idc))
+ return idc;
+
+ /* initialize it */
+ rc = osd_remote_fid(env, osd, fid);
+ if (unlikely(rc < 0))
+ return ERR_PTR(rc);
+
+ if (rc == 0) {
+ /* the object is local, lookup in OI */
+ /* XXX: probably cheaper to lookup in LU first? */
+ rc = osd_oi_lookup(osd_oti_get(env), osd, fid,
+ &idc->oic_lid, 0);
+ if (unlikely(rc < 0)) {
+ CERROR("can't lookup: rc = %d\n", rc);
+ return ERR_PTR(rc);
+ }
+ } else {
+ /* the object is remote */
+ idc->oic_remote = 1;
+ }
+
+ return idc;
+}
+
+/*
+ * lookup mapping for given FID and fill it from the given object.
+ * the object is lolcal by definition.
+ */
+static int osd_idc_find_and_init(const struct lu_env *env,
+ struct osd_device *osd,
+ struct osd_object *obj)
+{
+ const struct lu_fid *fid = lu_object_fid(&obj->oo_dt.do_lu);
+ struct osd_idmap_cache *idc;
+
+ idc = osd_idc_find(env, osd, fid);
+ LASSERT(!IS_ERR(idc));
+ if (idc != NULL) {
+ if (obj->oo_inode == NULL)
+ return 0;
+ if (idc->oic_lid.oii_ino != obj->oo_inode->i_ino) {
+ LASSERT(idc->oic_lid.oii_ino == 0);
+ idc->oic_lid.oii_ino = obj->oo_inode->i_ino;
+ idc->oic_lid.oii_gen = obj->oo_inode->i_generation;
+ }
+ return 0;
+ }
+
+ /* new mapping is needed */
+ idc = osd_idc_add(env, osd, fid);
+ if (IS_ERR(idc))
+ return PTR_ERR(idc);
+
+ if (obj->oo_inode != NULL) {
+ idc->oic_lid.oii_ino = obj->oo_inode->i_ino;
+ idc->oic_lid.oii_gen = obj->oo_inode->i_generation;
+ }
+ return 0;
+}
+
+/*
* OSD object methods.
*/
init_rwsem(&mo->oo_sem);
init_rwsem(&mo->oo_ext_idx_sem);
spin_lock_init(&mo->oo_guard);
+ INIT_LIST_HEAD(&mo->oo_xattr_list);
return l;
} else {
return NULL;
{
struct inode *inode = NULL;
+ /* if we look for an inode withing a running
+ * transaction, then we risk to deadlock */
+ /* osd_dirent_check_repair() breaks this */
+ /*LASSERT(current->journal_info == NULL);*/
+
inode = ldiskfs_iget(osd_sb(dev), id->oii_ino);
if (IS_ERR(inode)) {
CDEBUG(D_INODE, "no inode: ino = %u, rc = %ld\n",
iput(inode);
inode = ERR_PTR(-ENOENT);
} else {
+ ldiskfs_clear_inode_state(inode, LDISKFS_STATE_LUSTRE_DESTROY);
if (id->oii_gen == OSD_OII_NOGEN)
osd_id_gen(id, inode->i_ino, inode->i_generation);
return inode;
}
+int osd_ldiskfs_add_entry(struct osd_thread_info *info,
+ handle_t *handle, struct dentry *child,
+ struct inode *inode, struct htree_lock *hlock)
+{
+ int rc, rc2;
+
+ rc = __ldiskfs_add_entry(handle, child, inode, hlock);
+ if (rc == -ENOBUFS || rc == -ENOSPC) {
+ char fidbuf[FID_LEN + 1];
+ struct lustre_mdt_attrs lma;
+ struct lu_fid fid = { };
+ char *errstr;
+ struct dentry *p_dentry = child->d_parent;
+
+ rc2 = osd_get_lma(info, p_dentry->d_inode, p_dentry,
+ &lma);
+ if (rc2 == 0) {
+ fid = lma.lma_self_fid;
+ snprintf(fidbuf, sizeof(fidbuf), DFID, PFID(&fid));
+ } else if (rc2 == -ENODATA) {
+ if (unlikely(p_dentry->d_inode ==
+ inode->i_sb->s_root->d_inode))
+ lu_local_obj_fid(&fid, OSD_FS_ROOT_OID);
+ else if (info->oti_dev && !info->oti_dev->od_is_ost &&
+ fid_seq_is_mdt0(fid_seq(&fid)))
+ lu_igif_build(&fid, p_dentry->d_inode->i_ino,
+ p_dentry->d_inode->i_generation);
+ snprintf(fidbuf, sizeof(fidbuf), DFID, PFID(&fid));
+ } else {
+ snprintf(fidbuf, FID_LEN, "%s", "unknown");
+ }
+
+ if (rc == -ENOSPC)
+ errstr = "has reached";
+ else
+ errstr = "is approaching";
+ CWARN("%.16s: directory (inode: %lu FID: %s) %s maximum entry limit\n",
+ LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
+ p_dentry->d_inode->i_ino, fidbuf, errstr);
+ /* ignore such error now */
+ if (rc == -ENOBUFS)
+ rc = 0;
+ }
+ return rc;
+}
+
+
static struct inode *
osd_iget_fid(struct osd_thread_info *info, struct osd_device *dev,
struct osd_inode_id *id, struct lu_fid *fid)
struct osd_device *dev,
const struct lu_fid *fid,
struct osd_inode_id *id,
- bool in_oi)
+ bool cached)
{
struct inode *inode;
int rc = 0;
ENTRY;
+ /* The cached OI mapping is trustable. If we cannot locate the inode
+ * via the cached OI mapping, then return the failure to the caller
+ * directly without further OI checking. */
+
inode = ldiskfs_iget(osd_sb(dev), id->oii_ino);
if (IS_ERR(inode)) {
rc = PTR_ERR(inode);
- if (!in_oi || (rc != -ENOENT && rc != -ESTALE)) {
+ if (cached || (rc != -ENOENT && rc != -ESTALE)) {
CDEBUG(D_INODE, "no inode: ino = %u, rc = %d\n",
id->oii_ino, rc);
if (is_bad_inode(inode)) {
rc = -ENOENT;
- if (!in_oi) {
+ if (cached) {
CDEBUG(D_INODE, "bad inode: ino = %u\n", id->oii_ino);
GOTO(put, rc);
if (id->oii_gen != OSD_OII_NOGEN &&
inode->i_generation != id->oii_gen) {
rc = -ESTALE;
- if (!in_oi) {
+ if (cached) {
CDEBUG(D_INODE, "unmatched inode: ino = %u, "
"oii_gen = %u, i_generation = %u\n",
id->oii_ino, id->oii_gen, inode->i_generation);
if (inode->i_nlink == 0) {
rc = -ENOENT;
- if (!in_oi) {
+ if (cached) {
CDEBUG(D_INODE, "stale inode: ino = %u\n", id->oii_ino);
GOTO(put, rc);
goto check_oi;
}
+ ldiskfs_clear_inode_state(inode, LDISKFS_STATE_LUSTRE_DESTROY);
+
check_oi:
if (rc != 0) {
LASSERTF(rc == -ESTALE || rc == -ENOENT, "rc = %d\n", rc);
rc = osd_oi_lookup(info, dev, fid, id, OI_CHECK_FLD);
- /* XXX: There are some possible cases:
+ /* XXX: There are four possible cases:
* 1. rc = 0.
* Backup/restore caused the OI invalid.
* 2. rc = 0.
if (rc == 0) {
if (!IS_ERR(inode) && inode->i_generation != 0 &&
inode->i_generation == id->oii_gen)
+ /* "id->oii_gen != OSD_OII_NOGEN" is for
+ * "@cached == false" case. */
rc = -ENOENT;
else
rc = -EREMCHG;
+ } else {
+ /* If the OI mapping was in OI file before the
+ * osd_iget_check(), but now, it is disappear,
+ * then it must be removed by race. That is a
+ * normal race case. */
}
} else {
if (id->oii_gen == OSD_OII_NOGEN)
rc = 0;
ostid_set_seq(ostid, le64_to_cpu(ff->ff_seq));
ostid_set_id(ostid, le64_to_cpu(ff->ff_objid));
- /* XXX: should use real OST index in the future. LU-3569 */
+ /* XXX: use 0 as the index for compatibility, the caller will
+ * handle index related issues when necessarry. */
ostid_to_fid(fid, ostid, 0);
} else if (rc == sizeof(struct filter_fid)) {
rc = 1;
if (rc == -ENODATA && !fid_is_igif(rfid) && osd->od_check_ff) {
fid = &lma->lma_self_fid;
rc = osd_get_idif(info, inode, dentry, fid);
- if ((rc > 0) || (rc == -ENODATA && osd->od_lma_self_repair)) {
+ if ((rc > 0) || (rc == -ENODATA && osd->od_index_in_idif)) {
/* For the given OST-object, if it has neither LMA nor
* FID in XATTR_NAME_FID, then the given FID (which is
* contained in the @obj, from client RPC for locating
* the OST-object) is trusted. We use it to generate
* the LMA. */
osd_lma_self_repair(info, osd, inode, rfid,
- fid_is_on_ost(info, osd, fid, OI_CHECK_FLD) ?
- LMAC_FID_ON_OST : 0);
+ LMAC_FID_ON_OST);
RETURN(0);
}
}
- if (unlikely(rc == -ENODATA))
- RETURN(0);
-
if (rc < 0)
RETURN(rc);
lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
PFID(rfid), inode->i_ino);
rc = -EOPNOTSUPP;
- } else if (!(lma->lma_compat & LMAC_NOT_IN_OI)) {
+ } else {
fid = &lma->lma_self_fid;
}
}
fid_to_ostid(fid, oi);
ostid_to_fid(fid1, oi, idx);
if (lu_fid_eq(fid1, rfid)) {
- if (osd->od_lma_self_repair)
+ if (osd->od_index_in_idif)
osd_lma_self_repair(info, osd,
inode, rfid,
LMAC_FID_ON_OST);
}
}
- CDEBUG(D_INODE, "%s: FID "DFID" != self_fid "DFID"\n",
- osd_name(osd), PFID(rfid), PFID(fid));
rc = -EREMCHG;
}
struct osd_device *dev;
struct osd_idmap_cache *oic;
struct osd_inode_id *id;
- struct inode *inode;
+ struct inode *inode = NULL;
struct osd_scrub *scrub;
struct scrub_file *sf;
- int result;
- int saved = 0;
- bool in_oi = false;
- bool triggered = false;
+ __u32 flags = SS_CLEAR_DRYRUN | SS_CLEAR_FAILOUT |
+ SS_AUTO_FULL;
+ int result = 0;
+ int rc1 = 0;
+ bool cached = true;
+ bool remote = false;
ENTRY;
LINVRNT(osd_invariant(obj));
LASSERT(info);
oic = &info->oti_cache;
- if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT))
+ if (OBD_FAIL_CHECK(OBD_FAIL_SRV_ENOENT))
RETURN(-ENOENT);
/* For the object is created as locking anchor, or for the object to
goto iget;
}
+ cached = false;
/* Search order: 3. OI files. */
result = osd_oi_lookup(info, dev, fid, id, OI_CHECK_FLD);
if (result == -ENOENT) {
- if (!fid_is_norm(fid) ||
+ if (!(fid_is_norm(fid) || fid_is_igif(fid)) ||
fid_is_on_ost(info, dev, fid, OI_CHECK_FLD) ||
!ldiskfs_test_bit(osd_oi_fid2idx(dev,fid),
sf->sf_oi_bitmap))
if (result != 0)
GOTO(out, result);
- in_oi = true;
-
iget:
- inode = osd_iget_check(info, dev, fid, id, in_oi);
+ inode = osd_iget_check(info, dev, fid, id, cached);
if (IS_ERR(inode)) {
result = PTR_ERR(inode);
- if (result == -ENOENT || result == -ESTALE) {
- if (!in_oi)
- fid_zero(&oic->oic_fid);
-
+ if (result == -ENOENT || result == -ESTALE)
GOTO(out, result = -ENOENT);
- } else if (result == -EREMCHG) {
-trigger:
- if (!in_oi)
- fid_zero(&oic->oic_fid);
-
- if (unlikely(triggered))
- GOTO(out, result = saved);
-
- triggered = true;
- if (thread_is_running(&scrub->os_thread)) {
- result = -EINPROGRESS;
- } else if (!dev->od_noscrub) {
- /* Since we do not know the right OI mapping,
- * we have to trigger OI scrub to scan the
- * whole device. */
- result = osd_scrub_start(dev, SS_AUTO_FULL |
- SS_CLEAR_DRYRUN | SS_CLEAR_FAILOUT);
- LCONSOLE_WARN("%.16s: trigger OI scrub by RPC "
- "for "DFID", rc = %d [1]\n",
- osd_name(dev), PFID(fid),result);
- if (result == 0 || result == -EALREADY)
- result = -EINPROGRESS;
- else
- result = -EREMCHG;
- }
+ if (result == -EREMCHG) {
+trigger:
/* We still have chance to get the valid inode: for the
* object which is referenced by remote name entry, the
* object on the local MDT will be linked under the dir
* only happened for the RPC from other MDT during the
* OI scrub, or for the client side RPC with FID only,
* such as FID to path, or from old connected client. */
- saved = result;
- result = osd_lookup_in_remote_parent(info, dev,
- fid, id);
- if (result == 0) {
- in_oi = false;
- goto iget;
+ if (!remote &&
+ !fid_is_on_ost(info, dev, fid, OI_CHECK_FLD)) {
+ rc1 = osd_lookup_in_remote_parent(info, dev,
+ fid, id);
+ if (rc1 == 0) {
+ remote = true;
+ cached = true;
+ flags |= SS_AUTO_PARTIAL;
+ flags &= ~SS_AUTO_FULL;
+ goto iget;
+ }
}
- result = saved;
+ if (thread_is_running(&scrub->os_thread)) {
+ if (scrub->os_partial_scan &&
+ !scrub->os_in_join) {
+ goto join;
+ } else {
+ if (inode != NULL && !IS_ERR(inode)) {
+ LASSERT(remote);
+
+ osd_add_oi_cache(info, dev, id,
+ fid);
+ osd_oii_insert(dev, oic, true);
+ } else {
+ result = -EINPROGRESS;
+ }
+ }
+ } else if (!dev->od_noscrub) {
+
+join:
+ rc1 = osd_scrub_start(dev, flags);
+ LCONSOLE_WARN("%.16s: trigger OI scrub by RPC "
+ "for the "DFID" with flags 0x%x,"
+ " rc = %d\n", osd_name(dev),
+ PFID(fid), flags, rc1);
+ if (rc1 == 0 || rc1 == -EALREADY) {
+ if (inode != NULL && !IS_ERR(inode)) {
+ LASSERT(remote);
+
+ osd_add_oi_cache(info, dev, id,
+ fid);
+ osd_oii_insert(dev, oic, true);
+ } else {
+ result = -EINPROGRESS;
+ }
+ } else {
+ result = -EREMCHG;
+ }
+ } else {
+ result = -EREMCHG;
+ }
}
- GOTO(out, result);
- }
+ if (inode == NULL || IS_ERR(inode))
+ GOTO(out, result);
+ } else if (remote) {
+ goto trigger;
+ }
- obj->oo_inode = inode;
- LASSERT(obj->oo_inode->i_sb == osd_sb(dev));
+ obj->oo_inode = inode;
+ LASSERT(obj->oo_inode->i_sb == osd_sb(dev));
result = osd_check_lma(env, obj);
if (result != 0) {
- iput(inode);
- obj->oo_inode = NULL;
- if (result == -EREMCHG) {
- if (!in_oi) {
+ if (result == -ENODATA) {
+ if (cached) {
result = osd_oi_lookup(info, dev, fid, id,
OI_CHECK_FLD);
if (result != 0) {
- fid_zero(&oic->oic_fid);
+ /* result == -ENOENT means that the OI
+ * mapping has been removed by race,
+ * the target inode belongs to other
+ * object.
+ *
+ * Others error also can be returned
+ * directly. */
+ iput(inode);
+ obj->oo_inode = NULL;
GOTO(out, result);
+ } else {
+ /* result == 0 means the cached OI
+ * mapping is still in the OI file,
+ * the target the inode is valid. */
}
+ } else {
+ /* The current OI mapping is from the OI file,
+ * since the inode has been found via
+ * osd_iget_check(), no need recheck OI. */
}
- goto trigger;
+ goto found;
}
- GOTO(out, result);
+ iput(inode);
+ inode = NULL;
+ obj->oo_inode = NULL;
+ if (result != -EREMCHG)
+ GOTO(out, result);
+
+ if (cached) {
+ result = osd_oi_lookup(info, dev, fid, id,
+ OI_CHECK_FLD);
+ /* result == -ENOENT means the cached OI mapping
+ * has been removed from the OI file by race,
+ * above target inode belongs to other object.
+ *
+ * Others error also can be returned directly. */
+ if (result != 0)
+ GOTO(out, result);
+
+ /* result == 0, goto trigger */
+ } else {
+ /* The current OI mapping is from the OI file,
+ * since the inode has been found via
+ * osd_iget_check(), no need recheck OI. */
+ }
+
+ goto trigger;
}
+found:
obj->oo_compat_dot_created = 1;
obj->oo_compat_dotdot_created = 1;
- if (!S_ISDIR(inode->i_mode) || !ldiskfs_pdo) /* done */
+ if (!S_ISDIR(inode->i_mode) || !ldiskfs_pdo) /* done */
GOTO(out, result = 0);
LASSERT(obj->oo_hl_head == NULL);
GOTO(out, result = 0);
out:
+ if (result != 0 && cached)
+ fid_zero(&oic->oic_fid);
+
LINVRNT(osd_invariant(obj));
return result;
}
result = osd_fid_lookup(env, obj, lu_object_fid(l), conf);
obj->oo_dt.do_body_ops = &osd_body_ops_new;
- if (result == 0 && obj->oo_inode != NULL)
+ if (result == 0 && obj->oo_inode != NULL) {
+ struct osd_thread_info *oti = osd_oti_get(env);
+ struct lustre_mdt_attrs *lma = &oti->oti_mdt_attrs;
+
osd_object_init0(obj);
+ result = osd_get_lma(oti, obj->oo_inode,
+ &oti->oti_obj_dentry, lma);
+ if (result == 0) {
+ /* Convert LMAI flags to lustre LMA flags
+ * and cache it to oo_lma_flags */
+ obj->oo_lma_flags =
+ lma_to_lustre_flags(lma->lma_incompat);
+ } else if (result == -ENODATA) {
+ result = 0;
+ }
+ }
LINVRNT(osd_invariant(obj));
return result;
}
+/* The first part of oxe_buf is xattr name, and is '\0' terminated.
+ * The left part is for value, binary mode. */
+struct osd_xattr_entry {
+ struct list_head oxe_list;
+ size_t oxe_len;
+ size_t oxe_namelen;
+ bool oxe_exist;
+ struct rcu_head oxe_rcu;
+ char oxe_buf[0];
+};
+
+static struct osd_xattr_entry *osd_oxc_lookup(struct osd_object *obj,
+ const char *name,
+ size_t namelen)
+{
+ struct osd_xattr_entry *oxe;
+
+ list_for_each_entry(oxe, &obj->oo_xattr_list, oxe_list) {
+ if (namelen == oxe->oxe_namelen &&
+ strncmp(name, oxe->oxe_buf, namelen) == 0)
+ return oxe;
+ }
+
+ return NULL;
+}
+
+static int osd_oxc_get(struct osd_object *obj, const char *name,
+ struct lu_buf *buf)
+{
+ struct osd_xattr_entry *oxe;
+ size_t vallen;
+ ENTRY;
+
+ rcu_read_lock();
+ oxe = osd_oxc_lookup(obj, name, strlen(name));
+ if (oxe == NULL) {
+ rcu_read_unlock();
+ RETURN(-ENOENT);
+ }
+
+ if (!oxe->oxe_exist) {
+ rcu_read_unlock();
+ RETURN(-ENODATA);
+ }
+
+ vallen = oxe->oxe_len - sizeof(*oxe) - oxe->oxe_namelen - 1;
+ LASSERT(vallen > 0);
+
+ if (buf->lb_buf == NULL) {
+ rcu_read_unlock();
+ RETURN(vallen);
+ }
+
+ if (buf->lb_len < vallen) {
+ rcu_read_unlock();
+ RETURN(-ERANGE);
+ }
+
+ memcpy(buf->lb_buf, oxe->oxe_buf + oxe->oxe_namelen + 1, vallen);
+ rcu_read_unlock();
+
+ RETURN(vallen);
+}
+
+static void osd_oxc_free(struct rcu_head *head)
+{
+ struct osd_xattr_entry *oxe;
+
+ oxe = container_of(head, struct osd_xattr_entry, oxe_rcu);
+ OBD_FREE(oxe, oxe->oxe_len);
+}
+
+static inline void __osd_oxc_del(struct osd_object *obj, const char *name)
+{
+ struct osd_xattr_entry *oxe;
+
+ oxe = osd_oxc_lookup(obj, name, strlen(name));
+ if (oxe != NULL) {
+ list_del(&oxe->oxe_list);
+ call_rcu(&oxe->oxe_rcu, osd_oxc_free);
+ }
+}
+
+static void osd_oxc_add(struct osd_object *obj, const char *name,
+ const char *buf, int buflen)
+{
+ struct osd_xattr_entry *oxe;
+ size_t namelen = strlen(name);
+ size_t len = sizeof(*oxe) + namelen + 1 + buflen;
+
+ OBD_ALLOC(oxe, len);
+ if (oxe == NULL)
+ return;
+
+ INIT_LIST_HEAD(&oxe->oxe_list);
+ oxe->oxe_len = len;
+ oxe->oxe_namelen = namelen;
+ memcpy(oxe->oxe_buf, name, namelen);
+ if (buflen > 0) {
+ LASSERT(buf != NULL);
+ memcpy(oxe->oxe_buf + namelen + 1, buf, buflen);
+ oxe->oxe_exist = true;
+ } else {
+ oxe->oxe_exist = false;
+ }
+
+ /* this should be rarely called, just remove old and add new */
+ spin_lock(&obj->oo_guard);
+ __osd_oxc_del(obj, name);
+ list_add_tail(&oxe->oxe_list, &obj->oo_xattr_list);
+ spin_unlock(&obj->oo_guard);
+}
+
+static void osd_oxc_del(struct osd_object *obj, const char *name)
+{
+ spin_lock(&obj->oo_guard);
+ __osd_oxc_del(obj, name);
+ spin_unlock(&obj->oo_guard);
+}
+
+static void osd_oxc_fini(struct osd_object *obj)
+{
+ struct osd_xattr_entry *oxe, *next;
+
+ list_for_each_entry_safe(oxe, next, &obj->oo_xattr_list, oxe_list) {
+ list_del(&oxe->oxe_list);
+ OBD_FREE(oxe, oxe->oxe_len);
+ }
+}
+
/*
* Concurrency: no concurrent access is possible that late in object
* life-cycle.
LINVRNT(osd_invariant(obj));
+ osd_oxc_fini(obj);
dt_object_fini(&obj->oo_dt);
if (obj->oo_hl_head != NULL)
ldiskfs_htree_lock_head_free(obj->oo_hl_head);
dt_txn_hook_commit(th);
/* call per-transaction callbacks if any */
- list_for_each_entry_safe(dcb, tmp, &oh->ot_dcb_list, dcb_linkage) {
+ list_for_each_entry_safe(dcb, tmp, &oh->ot_commit_dcb_list,
+ dcb_linkage) {
LASSERTF(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC,
"commit callback entry: magic=%x name='%s'\n",
dcb->dcb_magic, dcb->dcb_name);
lu_context_exit(&th->th_ctx);
lu_context_fini(&th->th_ctx);
- thandle_put(th);
+ OBD_FREE_PTR(oh);
}
+#ifndef HAVE_SB_START_WRITE
+# define sb_start_write(sb) do {} while (0)
+# define sb_end_write(sb) do {} while (0)
+#endif
+
static struct thandle *osd_trans_create(const struct lu_env *env,
struct dt_device *d)
{
/* on pending IO in this thread should left from prev. request */
LASSERT(atomic_read(&iobuf->dr_numreqs) == 0);
- th = ERR_PTR(-ENOMEM);
+ sb_start_write(osd_sb(osd_dt_dev(d)));
+
OBD_ALLOC_GFP(oh, sizeof *oh, GFP_NOFS);
if (oh != NULL) {
oh->ot_quota_trans = &oti->oti_quota_trans;
th->th_result = 0;
th->th_tags = LCT_TX_HANDLE;
oh->ot_credits = 0;
- atomic_set(&th->th_refc, 1);
- th->th_alloc_size = sizeof(*oh);
- oti->oti_dev = osd_dt_dev(d);
- INIT_LIST_HEAD(&oh->ot_dcb_list);
+ INIT_LIST_HEAD(&oh->ot_commit_dcb_list);
+ INIT_LIST_HEAD(&oh->ot_stop_dcb_list);
osd_th_alloced(oh);
memset(oti->oti_declare_ops, 0,
- sizeof(oti->oti_declare_ops));
- memset(oti->oti_declare_ops_rb, 0,
- sizeof(oti->oti_declare_ops_rb));
+ sizeof(oti->oti_declare_ops));
memset(oti->oti_declare_ops_cred, 0,
- sizeof(oti->oti_declare_ops_cred));
- oti->oti_rollback = false;
+ sizeof(oti->oti_declare_ops_cred));
+ memset(oti->oti_declare_ops_used, 0,
+ sizeof(oti->oti_declare_ops_used));
+ } else {
+ sb_end_write(osd_sb(osd_dt_dev(d)));
+ th = ERR_PTR(-ENOMEM);
}
RETURN(th);
}
+void osd_trans_dump_creds(const struct lu_env *env, struct thandle *th)
+{
+ struct osd_thread_info *oti = osd_oti_get(env);
+ struct osd_thandle *oh;
+
+ oh = container_of0(th, struct osd_thandle, ot_super);
+ LASSERT(oh != NULL);
+
+ CWARN(" create: %u/%u/%u, destroy: %u/%u/%u\n",
+ oti->oti_declare_ops[OSD_OT_CREATE],
+ oti->oti_declare_ops_cred[OSD_OT_CREATE],
+ oti->oti_declare_ops_used[OSD_OT_CREATE],
+ oti->oti_declare_ops[OSD_OT_DESTROY],
+ oti->oti_declare_ops_cred[OSD_OT_DESTROY],
+ oti->oti_declare_ops_used[OSD_OT_DESTROY]);
+ CWARN(" attr_set: %u/%u/%u, xattr_set: %u/%u/%u\n",
+ oti->oti_declare_ops[OSD_OT_ATTR_SET],
+ oti->oti_declare_ops_cred[OSD_OT_ATTR_SET],
+ oti->oti_declare_ops_used[OSD_OT_ATTR_SET],
+ oti->oti_declare_ops[OSD_OT_XATTR_SET],
+ oti->oti_declare_ops_cred[OSD_OT_XATTR_SET],
+ oti->oti_declare_ops_used[OSD_OT_XATTR_SET]);
+ CWARN(" write: %u/%u/%u, punch: %u/%u/%u, quota %u/%u/%u\n",
+ oti->oti_declare_ops[OSD_OT_WRITE],
+ oti->oti_declare_ops_cred[OSD_OT_WRITE],
+ oti->oti_declare_ops_used[OSD_OT_WRITE],
+ oti->oti_declare_ops[OSD_OT_PUNCH],
+ oti->oti_declare_ops_cred[OSD_OT_PUNCH],
+ oti->oti_declare_ops_used[OSD_OT_PUNCH],
+ oti->oti_declare_ops[OSD_OT_QUOTA],
+ oti->oti_declare_ops_cred[OSD_OT_QUOTA],
+ oti->oti_declare_ops_used[OSD_OT_QUOTA]);
+ CWARN(" insert: %u/%u/%u, delete: %u/%u/%u\n",
+ oti->oti_declare_ops[OSD_OT_INSERT],
+ oti->oti_declare_ops_cred[OSD_OT_INSERT],
+ oti->oti_declare_ops_used[OSD_OT_INSERT],
+ oti->oti_declare_ops[OSD_OT_DELETE],
+ oti->oti_declare_ops_cred[OSD_OT_DELETE],
+ oti->oti_declare_ops_used[OSD_OT_DELETE]);
+ CWARN(" ref_add: %u/%u/%u, ref_del: %u/%u/%u\n",
+ oti->oti_declare_ops[OSD_OT_REF_ADD],
+ oti->oti_declare_ops_cred[OSD_OT_REF_ADD],
+ oti->oti_declare_ops_used[OSD_OT_REF_ADD],
+ oti->oti_declare_ops[OSD_OT_REF_DEL],
+ oti->oti_declare_ops_cred[OSD_OT_REF_DEL],
+ oti->oti_declare_ops_used[OSD_OT_REF_DEL]);
+}
+
/*
* Concurrency: shouldn't matter.
*/
static unsigned long last_printed;
static int last_credits;
- CWARN("%.16s: too many transaction credits (%d > %d)\n",
- LDISKFS_SB(osd_sb(dev))->s_es->s_volume_name,
- oh->ot_credits,
- osd_journal(dev)->j_max_transaction_buffers);
- CWARN(" create: %u/%u, destroy: %u/%u\n",
- oti->oti_declare_ops[OSD_OT_CREATE],
- oti->oti_declare_ops_cred[OSD_OT_CREATE],
- oti->oti_declare_ops[OSD_OT_DESTROY],
- oti->oti_declare_ops_cred[OSD_OT_DESTROY]);
- CWARN(" attr_set: %u/%u, xattr_set: %u/%u\n",
- oti->oti_declare_ops[OSD_OT_ATTR_SET],
- oti->oti_declare_ops_cred[OSD_OT_ATTR_SET],
- oti->oti_declare_ops[OSD_OT_XATTR_SET],
- oti->oti_declare_ops_cred[OSD_OT_XATTR_SET]);
- CWARN(" write: %u/%u, punch: %u/%u, quota %u/%u\n",
- oti->oti_declare_ops[OSD_OT_WRITE],
- oti->oti_declare_ops_cred[OSD_OT_WRITE],
- oti->oti_declare_ops[OSD_OT_PUNCH],
- oti->oti_declare_ops_cred[OSD_OT_PUNCH],
- oti->oti_declare_ops[OSD_OT_QUOTA],
- oti->oti_declare_ops_cred[OSD_OT_QUOTA]);
- CWARN(" insert: %u/%u, delete: %u/%u\n",
- oti->oti_declare_ops[OSD_OT_INSERT],
- oti->oti_declare_ops_cred[OSD_OT_INSERT],
- oti->oti_declare_ops[OSD_OT_DELETE],
- oti->oti_declare_ops_cred[OSD_OT_DELETE]);
- CWARN(" ref_add: %u/%u, ref_del: %u/%u\n",
- oti->oti_declare_ops[OSD_OT_REF_ADD],
- oti->oti_declare_ops_cred[OSD_OT_REF_ADD],
- oti->oti_declare_ops[OSD_OT_REF_DEL],
- oti->oti_declare_ops_cred[OSD_OT_REF_DEL]);
-
+ /* don't make noise on a tiny testing systems
+ * actual credits misuse will be caught anyway */
if (last_credits != oh->ot_credits &&
time_after(jiffies, last_printed +
- msecs_to_jiffies(60 * MSEC_PER_SEC))) {
+ msecs_to_jiffies(60 * MSEC_PER_SEC)) &&
+ osd_transaction_size(dev) > 512) {
+ osd_trans_dump_creds(env, th);
libcfs_debug_dumpstack(NULL);
last_credits = oh->ot_credits;
last_printed = jiffies;
RETURN(ss->ss_node_id == range->lsr_index);
}
+static void osd_trans_stop_cb(struct osd_thandle *oth, int result)
+{
+ struct dt_txn_commit_cb *dcb;
+ struct dt_txn_commit_cb *tmp;
+
+ /* call per-transaction stop callbacks if any */
+ list_for_each_entry_safe(dcb, tmp, &oth->ot_stop_dcb_list,
+ dcb_linkage) {
+ LASSERTF(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC,
+ "commit callback entry: magic=%x name='%s'\n",
+ dcb->dcb_magic, dcb->dcb_name);
+ list_del_init(&dcb->dcb_linkage);
+ dcb->dcb_func(NULL, &oth->ot_super, dcb, result);
+ }
+}
+
/*
* Concurrency: shouldn't matter.
*/
static int osd_trans_stop(const struct lu_env *env, struct dt_device *dt,
struct thandle *th)
{
- int rc = 0;
+ int rc = 0, remove_agents = 0;
struct osd_thandle *oh;
struct osd_thread_info *oti = osd_oti_get(env);
struct osd_iobuf *iobuf = &oti->oti_iobuf;
- struct qsd_instance *qsd = oti->oti_dev->od_quota_slave;
+ struct osd_device *osd = osd_dt_dev(th->th_dev);
+ struct qsd_instance *qsd = osd->od_quota_slave;
struct lquota_trans *qtrans;
ENTRY;
oh = container_of0(th, struct osd_thandle, ot_super);
+ /* reset OI cache for safety */
+ oti->oti_ins_cache_used = 0;
+
+ remove_agents = oh->ot_remove_agents;
+
qtrans = oh->ot_quota_trans;
oh->ot_quota_trans = NULL;
- if (oh->ot_handle != NULL) {
+ if (oh->ot_handle != NULL) {
handle_t *hdl = oh->ot_handle;
/*
LASSERT(oti->oti_txns == 1);
oti->oti_txns--;
+
rc = dt_txn_hook_stop(env, th);
if (rc != 0)
- CERROR("Failure in transaction hook: %d\n", rc);
+ CERROR("%s: failed in transaction hook: rc = %d\n",
+ osd_name(osd), rc);
+ osd_trans_stop_cb(oh, rc);
/* hook functions might modify th_sync */
hdl->h_sync = th->th_sync;
- oh->ot_handle = NULL;
- OSD_CHECK_SLOW_TH(oh, oti->oti_dev,
- rc = ldiskfs_journal_stop(hdl));
- if (rc != 0)
- CERROR("Failure to stop transaction: %d\n", rc);
- } else {
- thandle_put(&oh->ot_super);
- }
+ oh->ot_handle = NULL;
+ OSD_CHECK_SLOW_TH(oh, osd, rc = ldiskfs_journal_stop(hdl));
+ if (rc != 0)
+ CERROR("%s: failed to stop transaction: rc = %d\n",
+ osd_name(osd), rc);
+ } else {
+ osd_trans_stop_cb(oh, th->th_result);
+ OBD_FREE_PTR(oh);
+ }
/* inform the quota slave device that the transaction is stopping */
qsd_op_end(env, qsd, qtrans);
*/
wait_event(iobuf->dr_wait,
atomic_read(&iobuf->dr_numreqs) == 0);
- osd_fini_iobuf(oti->oti_dev, iobuf);
+ osd_fini_iobuf(osd, iobuf);
if (!rc)
rc = iobuf->dr_error;
+ if (unlikely(remove_agents != 0))
+ osd_process_scheduled_agent_removals(env, osd);
+
+ sb_end_write(osd_sb(osd));
+
RETURN(rc);
}
LASSERT(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC);
LASSERT(&dcb->dcb_func != NULL);
- list_add(&dcb->dcb_linkage, &oh->ot_dcb_list);
+ if (dcb->dcb_flags & DCB_TRANS_STOP)
+ list_add(&dcb->dcb_linkage, &oh->ot_stop_dcb_list);
+ else
+ list_add(&dcb->dcb_linkage, &oh->ot_commit_dcb_list);
return 0;
}
static void osd_object_release(const struct lu_env *env,
struct lu_object *l)
{
+ struct osd_object *o = osd_obj(l);
+ /* nobody should be releasing a non-destroyed object with nlink=0
+ * the API allows this, but ldiskfs doesn't like and then report
+ * this inode as deleted */
+ if (unlikely(!o->oo_destroyed && o->oo_inode && o->oo_inode->i_nlink == 0))
+ LBUG();
}
/*
d ? d->id_ops->id_name : "plain");
}
-#define GRANT_FOR_LOCAL_OIDS 32 /* 128kB for last_rcvd, quota files, ... */
-
/*
* Concurrency: shouldn't matter.
*/
int osd_statfs(const struct lu_env *env, struct dt_device *d,
struct obd_statfs *sfs)
{
- struct osd_device *osd = osd_dt_dev(d);
- struct super_block *sb = osd_sb(osd);
- struct kstatfs *ksfs;
- int result = 0;
+ struct osd_device *osd = osd_dt_dev(d);
+ struct super_block *sb = osd_sb(osd);
+ struct kstatfs *ksfs;
+ __u64 reserved;
+ int result = 0;
if (unlikely(osd->od_mnt == NULL))
return -EINPROGRESS;
ksfs = &osd_oti_get(env)->oti_ksfs;
}
- spin_lock(&osd->od_osfs_lock);
result = sb->s_op->statfs(sb->s_root, ksfs);
- if (likely(result == 0)) { /* N.B. statfs can't really fail */
- statfs_pack(sfs, ksfs);
- if (sb->s_flags & MS_RDONLY)
- sfs->os_state = OS_STATE_READONLY;
- }
-
- spin_unlock(&osd->od_osfs_lock);
+ if (result)
+ goto out;
+
+ statfs_pack(sfs, ksfs);
+ if (unlikely(sb->s_flags & MS_RDONLY))
+ sfs->os_state = OS_STATE_READONLY;
+ if (LDISKFS_HAS_INCOMPAT_FEATURE(sb,
+ LDISKFS_FEATURE_INCOMPAT_EXTENTS))
+ sfs->os_maxbytes = sb->s_maxbytes;
+ else
+ sfs->os_maxbytes = LDISKFS_SB(sb)->s_bitmap_maxbytes;
- if (unlikely(env == NULL))
- OBD_FREE_PTR(ksfs);
+ /*
+ * Reserve some space so to avoid fragmenting the filesystem too much.
+ * Fragmentation not only impacts performance, but can also increase
+ * metadata overhead significantly, causing grant calculation to be
+ * wrong.
+ *
+ * Reserve 0.78% of total space, at least 8MB for small filesystems.
+ */
+ CLASSERT(OSD_STATFS_RESERVED > LDISKFS_MAX_BLOCK_SIZE);
+ reserved = OSD_STATFS_RESERVED >> sb->s_blocksize_bits;
+ if (likely(sfs->os_blocks >= reserved << OSD_STATFS_RESERVED_SHIFT))
+ reserved = sfs->os_blocks >> OSD_STATFS_RESERVED_SHIFT;
- /* Reserve a small amount of space for local objects like last_rcvd,
- * llog, quota files, ... */
- if (sfs->os_bavail <= GRANT_FOR_LOCAL_OIDS) {
- sfs->os_bavail = 0;
- } else {
- sfs->os_bavail -= GRANT_FOR_LOCAL_OIDS;
- /** Take out metadata overhead for indirect blocks */
- sfs->os_bavail -= sfs->os_bavail >> (sb->s_blocksize_bits - 3);
- }
+ sfs->os_blocks -= reserved;
+ sfs->os_bfree -= min(reserved, sfs->os_bfree);
+ sfs->os_bavail -= min(reserved, sfs->os_bavail);
- return result;
+out:
+ if (unlikely(env == NULL))
+ OBD_FREE_PTR(ksfs);
+ return result;
}
/**
*/
param->ddp_max_name_len = LDISKFS_NAME_LEN;
param->ddp_max_nlink = LDISKFS_LINK_MAX;
- param->ddp_block_shift = sb->s_blocksize_bits;
+ param->ddp_symlink_max = sb->s_blocksize;
param->ddp_mount_type = LDD_MT_LDISKFS;
- param->ddp_maxbytes = sb->s_maxbytes;
- /* Overhead estimate should be fairly accurate, so we really take a tiny
- * error margin which also avoids fragmenting the filesystem too much */
- param->ddp_grant_reserved = 2; /* end up to be 1.9% after conversion */
+ if (LDISKFS_HAS_INCOMPAT_FEATURE(sb, LDISKFS_FEATURE_INCOMPAT_EXTENTS))
+ param->ddp_maxbytes = sb->s_maxbytes;
+ else
+ param->ddp_maxbytes = LDISKFS_SB(sb)->s_bitmap_maxbytes;
/* inode are statically allocated, so per-inode space consumption
* is the space consumed by the directory entry */
param->ddp_inodespace = PER_OBJ_USAGE;
- /* per-fragment overhead to be used by the client code */
- param->ddp_grant_frag = 6 * LDISKFS_BLOCK_SIZE(sb);
- param->ddp_mntopts = 0;
+ /* EXT_INIT_MAX_LEN is the theoretical maximum extent size (32k blocks
+ * = 128MB) which is unlikely to be hit in real life. Report a smaller
+ * maximum length to not under count the actual number of extents
+ * needed for writing a file. */
+ param->ddp_max_extent_blks = EXT_INIT_MAX_LEN >> 2;
+ /* worst-case extent insertion metadata overhead */
+ param->ddp_extent_tax = 6 * LDISKFS_BLOCK_SIZE(sb);
+ param->ddp_mntopts = 0;
if (test_opt(sb, XATTR_USER))
param->ddp_mntopts |= MNTOPT_USERXATTR;
if (test_opt(sb, POSIX_ACL))
#ifdef HAVE_DEV_SET_RDONLY
CERROR("*** setting %s read-only ***\n", osd_dt_dev(d)->od_svname);
+ if (sb->s_op->freeze_fs) {
+ rc = sb->s_op->freeze_fs(sb);
+ if (rc)
+ goto out;
+ }
+
if (jdev && (jdev != dev)) {
CDEBUG(D_IOCTL | D_HA, "set journal dev %lx rdonly\n",
(long)jdev);
}
CDEBUG(D_IOCTL | D_HA, "set dev %lx rdonly\n", (long)dev);
dev_set_rdonly(dev);
-#else
- CERROR("%s: %lx CANNOT BE SET READONLY: rc = %d\n",
- osd_dt_dev(d)->od_svname, (long)dev, rc);
-#endif
- RETURN(rc);
-}
-/*
- * Concurrency: serialization provided by callers.
- */
-static int osd_init_capa_ctxt(const struct lu_env *env, struct dt_device *d,
- int mode, unsigned long timeout, __u32 alg,
- struct lustre_capa_key *keys)
-{
- struct osd_device *dev = osd_dt_dev(d);
- ENTRY;
+ if (sb->s_op->unfreeze_fs)
+ sb->s_op->unfreeze_fs(sb);
- dev->od_fl_capa = mode;
- dev->od_capa_timeout = timeout;
- dev->od_capa_alg = alg;
- dev->od_capa_keys = keys;
- RETURN(0);
+out:
+#endif
+ if (rc)
+ CERROR("%s: %lx CANNOT BE SET READONLY: rc = %d\n",
+ osd_dt_dev(d)->od_svname, (long)dev, rc);
+
+ RETURN(rc);
}
/**
.dt_sync = osd_sync,
.dt_ro = osd_ro,
.dt_commit_async = osd_commit_async,
- .dt_init_capa_ctxt = osd_init_capa_ctxt,
};
static void osd_object_read_lock(const struct lu_env *env,
return obj->oo_owner == env;
}
-static int capa_is_sane(const struct lu_env *env,
- struct osd_device *dev,
- struct lustre_capa *capa,
- struct lustre_capa_key *keys)
-{
- struct osd_thread_info *oti = osd_oti_get(env);
- struct lustre_capa *tcapa = &oti->oti_capa;
- struct obd_capa *oc;
- int i, rc = 0;
- ENTRY;
-
- oc = capa_lookup(dev->od_capa_hash, capa, 0);
- if (oc) {
- if (capa_is_expired(oc)) {
- DEBUG_CAPA(D_ERROR, capa, "expired");
- rc = -ESTALE;
- }
- capa_put(oc);
- RETURN(rc);
- }
-
- if (capa_is_expired_sec(capa)) {
- DEBUG_CAPA(D_ERROR, capa, "expired");
- RETURN(-ESTALE);
- }
-
- spin_lock(&capa_lock);
- for (i = 0; i < 2; i++) {
- if (keys[i].lk_keyid == capa->lc_keyid) {
- oti->oti_capa_key = keys[i];
- break;
- }
- }
- spin_unlock(&capa_lock);
-
- if (i == 2) {
- DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
- RETURN(-ESTALE);
- }
-
- rc = capa_hmac(tcapa->lc_hmac, capa, oti->oti_capa_key.lk_key);
- if (rc)
- RETURN(rc);
-
- if (memcmp(tcapa->lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac))) {
- DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
- RETURN(-EACCES);
- }
-
- oc = capa_add(dev->od_capa_hash, capa);
- capa_put(oc);
-
- RETURN(0);
-}
-
-int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
- struct lustre_capa *capa, __u64 opc)
-{
- const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
- struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
- struct lu_capainfo *lci;
- int rc;
-
- if (!osd->od_fl_capa)
- return 0;
-
- if (capa == BYPASS_CAPA)
- return 0;
-
- lci = lu_capainfo_get(env);
- if (unlikely(lci == NULL))
- return 0;
-
- if (lci->lci_auth == LC_ID_NONE)
- return 0;
-
- if (capa == NULL) {
- CERROR("%s: no capability provided for FID "DFID": rc = %d\n",
- osd_name(osd), PFID(fid), -EACCES);
- return -EACCES;
- }
-
- if (!lu_fid_eq(fid, &capa->lc_fid)) {
- DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",
- PFID(fid));
- return -EACCES;
- }
-
- if (!capa_opc_supported(capa, opc)) {
- DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
- return -EACCES;
- }
-
- rc = capa_is_sane(env, osd, capa, osd->od_capa_keys);
- if (rc != 0) {
- DEBUG_CAPA(D_ERROR, capa, "insane: rc = %d", rc);
- return -EACCES;
- }
-
- return 0;
-}
-
static struct timespec *osd_inode_time(const struct lu_env *env,
struct inode *inode, __u64 seconds)
{
return t;
}
-
static void osd_inode_getattr(const struct lu_env *env,
struct inode *inode, struct lu_attr *attr)
{
attr->la_blocks = inode->i_blocks;
attr->la_uid = i_uid_read(inode);
attr->la_gid = i_gid_read(inode);
- attr->la_flags = LDISKFS_I(inode)->i_flags;
+ attr->la_flags = ll_inode_to_ext_flags(inode->i_flags);
attr->la_nlink = inode->i_nlink;
attr->la_rdev = inode->i_rdev;
attr->la_blksize = 1 << inode->i_blkbits;
static int osd_attr_get(const struct lu_env *env,
struct dt_object *dt,
- struct lu_attr *attr,
- struct lustre_capa *capa)
+ struct lu_attr *attr)
{
struct osd_object *obj = osd_dt_obj(dt);
- LASSERT(dt_object_exists(dt));
+ if (unlikely(!dt_object_exists(dt)))
+ return -ENOENT;
+ if (unlikely(obj->oo_destroyed))
+ return -ENOENT;
+
LASSERT(!dt_object_remote(dt));
LINVRNT(osd_invariant(obj));
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
- return -EACCES;
-
spin_lock(&obj->oo_guard);
osd_inode_getattr(env, obj->oo_inode, attr);
+ if (obj->oo_lma_flags & LUSTRE_ORPHAN_FL)
+ attr->la_flags |= LUSTRE_ORPHAN_FL;
spin_unlock(&obj->oo_guard);
+
return 0;
}
osd_trans_declare_op(env, oh, OSD_OT_ATTR_SET,
osd_dto_credits_noquota[DTO_ATTR_SET_BASE]);
+ osd_trans_declare_op(env, oh, OSD_OT_XATTR_SET,
+ osd_dto_credits_noquota[DTO_XATTR_SET]);
+
if (attr == NULL || obj->oo_inode == NULL)
RETURN(rc);
if (bits == 0)
return 0;
- if (bits & LA_ATIME)
- inode->i_atime = *osd_inode_time(env, inode, attr->la_atime);
- if (bits & LA_CTIME)
- inode->i_ctime = *osd_inode_time(env, inode, attr->la_ctime);
- if (bits & LA_MTIME)
- inode->i_mtime = *osd_inode_time(env, inode, attr->la_mtime);
- if (bits & LA_SIZE) {
- LDISKFS_I(inode)->i_disksize = attr->la_size;
- i_size_write(inode, attr->la_size);
- }
+ if (bits & LA_ATIME)
+ inode->i_atime = *osd_inode_time(env, inode, attr->la_atime);
+ if (bits & LA_CTIME)
+ inode->i_ctime = *osd_inode_time(env, inode, attr->la_ctime);
+ if (bits & LA_MTIME)
+ inode->i_mtime = *osd_inode_time(env, inode, attr->la_mtime);
+ if (bits & LA_SIZE) {
+ LDISKFS_I(inode)->i_disksize = attr->la_size;
+ i_size_write(inode, attr->la_size);
+ }
-#if 0
- /* OSD should not change "i_blocks" which is used by quota.
- * "i_blocks" should be changed by ldiskfs only. */
- if (bits & LA_BLOCKS)
- inode->i_blocks = attr->la_blocks;
-#endif
+ /* OSD should not change "i_blocks" which is used by quota.
+ * "i_blocks" should be changed by ldiskfs only. */
if (bits & LA_MODE)
inode->i_mode = (inode->i_mode & S_IFMT) |
(attr->la_mode & ~S_IFMT);
if (bits & LA_RDEV)
inode->i_rdev = attr->la_rdev;
- if (bits & LA_FLAGS) {
- /* always keep S_NOCMTIME */
- inode->i_flags = ll_ext_to_inode_flags(attr->la_flags) |
- S_NOCMTIME;
- }
- return 0;
+ if (bits & LA_FLAGS) {
+ /* always keep S_NOCMTIME */
+ inode->i_flags = ll_ext_to_inode_flags(attr->la_flags) |
+ S_NOCMTIME;
+ }
+ return 0;
}
static int osd_quota_transfer(struct inode *inode, const struct lu_attr *attr)
static int osd_attr_set(const struct lu_env *env,
struct dt_object *dt,
const struct lu_attr *attr,
- struct thandle *handle,
- struct lustre_capa *capa)
+ struct thandle *handle)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode;
int rc;
+ if (!dt_object_exists(dt))
+ return -ENOENT;
+
LASSERT(handle != NULL);
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(osd_invariant(obj));
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
- return -EACCES;
-
osd_trans_exec_op(env, handle, OSD_OT_ATTR_SET);
if (OBD_FAIL_CHECK(OBD_FAIL_OSD_FID_MAPPING)) {
spin_lock(&obj->oo_guard);
rc = osd_inode_setattr(env, inode, attr);
spin_unlock(&obj->oo_guard);
+ if (rc != 0)
+ GOTO(out, rc);
+
+ ll_dirty_inode(inode, I_DIRTY_DATASYNC);
+
+ if (!(attr->la_valid & LA_FLAGS))
+ GOTO(out, rc);
+
+ /* Let's check if there are extra flags need to be set into LMA */
+ if (attr->la_flags & LUSTRE_LMA_FL_MASKS) {
+ struct osd_thread_info *info = osd_oti_get(env);
+ struct lustre_mdt_attrs *lma = &info->oti_mdt_attrs;
+
+ rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
+ if (rc != 0)
+ GOTO(out, rc);
+
+ lma->lma_incompat |=
+ lustre_to_lma_flags(attr->la_flags);
+ lustre_lma_swab(lma);
+ rc = __osd_xattr_set(info, inode, XATTR_NAME_LMA,
+ lma, sizeof(*lma), XATTR_REPLACE);
+ if (rc != 0) {
+ struct osd_device *osd = osd_obj2dev(obj);
+
+ CWARN("%s: set "DFID" lma flags %u failed: rc = %d\n",
+ osd_name(osd), PFID(lu_object_fid(&dt->do_lu)),
+ lma->lma_incompat, rc);
+ } else {
+ obj->oo_lma_flags =
+ attr->la_flags & LUSTRE_LMA_FL_MASKS;
+ }
+ osd_trans_exec_check(env, handle, OSD_OT_XATTR_SET);
+ }
+out:
+ osd_trans_exec_check(env, handle, OSD_OT_ATTR_SET);
- if (!rc)
- ll_dirty_inode(inode, I_DIRTY_DATASYNC);
return rc;
}
oth = container_of(th, struct osd_thandle, ot_super);
LASSERT(oth->ot_handle->h_transaction != NULL);
- if (hint && hint->dah_parent)
- parent = hint->dah_parent;
+ if (hint != NULL && hint->dah_parent != NULL &&
+ !dt_object_remote(hint->dah_parent))
+ parent = hint->dah_parent;
inode = ldiskfs_create_inode(oth->ot_handle,
parent ? osd_dt_obj(parent)->oo_inode :
osd_sb(osd)->s_root->d_inode,
mode);
if (!IS_ERR(inode)) {
- /* Do not update file c/mtime in ldiskfs.
- * NB: don't need any lock because no contention at this
- * early stage */
- inode->i_flags |= S_NOCMTIME;
+ /* Do not update file c/mtime in ldiskfs. */
+ inode->i_flags |= S_NOCMTIME;
/* For new created object, it must be consistent,
* and it is unnecessary to scrub against it. */
ldiskfs_set_inode_state(inode, LDISKFS_STATE_LUSTRE_NOSCRUB);
+
obj->oo_inode = inode;
result = 0;
} else {
ah->dah_parent = parent;
ah->dah_mode = child_mode;
+
+ if (parent != NULL && !dt_object_remote(parent)) {
+ /* will help to find FID->ino at dt_insert("..") */
+ struct osd_object *pobj = osd_dt_obj(parent);
+ osd_idc_find_and_init(env, osd_obj2dev(pobj), pobj);
+ }
}
static void osd_attr_init(struct osd_thread_info *info, struct osd_object *obj,
if (result)
return;
- if (attr->la_valid != 0) {
- result = osd_inode_setattr(info->oti_env, inode, attr);
- /*
- * The osd_inode_setattr() should always succeed here. The
- * only error that could be returned is EDQUOT when we are
- * trying to change the UID or GID of the inode. However, this
- * should not happen since quota enforcement is no longer
- * enabled on ldiskfs (lquota takes care of it).
- */
+ if (attr->la_valid != 0) {
+ result = osd_inode_setattr(info->oti_env, inode, attr);
+ /*
+ * The osd_inode_setattr() should always succeed here. The
+ * only error that could be returned is EDQUOT when we are
+ * trying to change the UID or GID of the inode. However, this
+ * should not happen since quota enforcement is no longer
+ * enabled on ldiskfs (lquota takes care of it).
+ */
LASSERTF(result == 0, "%d\n", result);
ll_dirty_inode(inode, I_DIRTY_DATASYNC);
- }
+ }
- attr->la_valid = valid;
+ attr->la_valid = valid;
}
/**
int result;
__u32 umask;
+ osd_trans_exec_op(info->oti_env, th, OSD_OT_CREATE);
+
/* we drop umask so that permissions we pass are not affected */
umask = current->fs->umask;
current->fs->umask = 0;
result = osd_create_type_f(dof->dof_type)(info, obj, attr, hint, dof,
th);
- if (result == 0) {
+ if (likely(obj->oo_inode != NULL)) {
+ LASSERT(obj->oo_inode->i_state & I_NEW);
+
+ /* Unlock the inode before attr initialization to avoid
+ * unnecessary dqget operations. LU-6378 */
+ unlock_new_inode(obj->oo_inode);
+ }
+
+ if (likely(result == 0)) {
osd_attr_init(info, obj, attr, dof);
osd_object_init0(obj);
- /* bz 24037 */
- if (obj->oo_inode && (obj->oo_inode->i_state & I_NEW))
- unlock_new_inode(obj->oo_inode);
- }
+ }
/* restore previous umask value */
current->fs->umask = umask;
+ osd_trans_exec_check(info->oti_env, th, OSD_OT_CREATE);
+
return result;
}
struct osd_inode_id *id = &info->oti_id;
struct osd_device *osd = osd_obj2dev(obj);
struct osd_thandle *oh;
+ int rc;
LASSERT(obj->oo_inode != NULL);
oh = container_of0(th, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle);
+ osd_trans_exec_op(env, th, OSD_OT_INSERT);
osd_id_gen(id, obj->oo_inode->i_ino, obj->oo_inode->i_generation);
- return osd_oi_insert(info, osd, fid, id, oh->ot_handle, OI_CHECK_FLD);
+ rc = osd_oi_insert(info, osd, fid, id, oh->ot_handle,
+ OI_CHECK_FLD, NULL);
+ osd_trans_exec_check(env, th, OSD_OT_INSERT);
+
+ return rc;
}
int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
oh = container_of0(handle, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle == NULL);
+ /* EA object consumes more credits than regular object: osd_mk_index
+ * vs. osd_mkreg: osd_mk_index will create 2 blocks for root_node and
+ * leaf_node, could involves the block, block bitmap, groups, GDT
+ * change for each block, so add 4 * 2 credits in that case. */
osd_trans_declare_op(env, oh, OSD_OT_CREATE,
- osd_dto_credits_noquota[DTO_OBJECT_CREATE]);
+ osd_dto_credits_noquota[DTO_OBJECT_CREATE] +
+ (dof->dof_type == DFT_INDEX) ? 4 * 2 : 0);
/* Reuse idle OI block may cause additional one OI block
* to be changed. */
osd_trans_declare_op(env, oh, OSD_OT_INSERT,
osd_dto_credits_noquota[DTO_INDEX_INSERT] + 1);
- /* If this is directory, then we expect . and .. to be inserted as
- * well. The one directory block always needs to be created for the
- * directory, so we could use DTO_WRITE_BASE here (GDT, block bitmap,
- * block), there is no danger of needing a tree for the first block.
- */
- if (attr && S_ISDIR(attr->la_mode)) {
- osd_trans_declare_op(env, oh, OSD_OT_INSERT,
- osd_dto_credits_noquota[DTO_WRITE_BASE]);
- osd_trans_declare_op(env, oh, OSD_OT_INSERT, 0);
- }
-
if (!attr)
RETURN(0);
if (rc != 0)
RETURN(rc);
+ /* will help to find FID->ino mapping at dt_insert() */
+ rc = osd_idc_find_and_init(env, osd_obj2dev(osd_dt_obj(dt)),
+ osd_dt_obj(dt));
+
RETURN(rc);
}
int result;
ENTRY;
+ if (dt_object_exists(dt))
+ return -EEXIST;
+
LINVRNT(osd_invariant(obj));
- LASSERT(!dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(osd_write_locked(env, obj));
LASSERT(th != NULL);
* 'tune2fs -O quota' will take care of creating them */
RETURN(-EPERM);
- osd_trans_exec_op(env, th, OSD_OT_CREATE);
- osd_trans_declare_rb(env, th, OSD_OT_REF_ADD);
-
- result = __osd_object_create(info, obj, attr, hint, dof, th);
- if (result == 0)
- result = __osd_oi_insert(env, obj, fid, th);
-
+ result = __osd_object_create(info, obj, attr, hint, dof, th);
+ if (result == 0) {
+ result = __osd_oi_insert(env, obj, fid, th);
+ if (obj->oo_dt.do_body_ops == &osd_body_ops_new)
+ obj->oo_dt.do_body_ops = &osd_body_ops;
+ }
LASSERT(ergo(result == 0,
- dt_object_exists(dt) && !dt_object_remote(dt)));
+ dt_object_exists(dt) && !dt_object_remote(dt)));
- LASSERT(osd_invariant(obj));
- RETURN(result);
+ LASSERT(osd_invariant(obj));
+ RETURN(result);
}
/**
int rc;
ENTRY;
+ if (inode == NULL)
+ RETURN(-ENOENT);
+
oh = container_of0(th, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle == NULL);
- LASSERT(inode);
osd_trans_declare_op(env, oh, OSD_OT_DESTROY,
osd_dto_credits_noquota[DTO_OBJECT_DELETE]);
/* data to be truncated */
rc = osd_declare_inode_qid(env, i_uid_read(inode), i_gid_read(inode),
0, oh, obj, true, NULL, false);
+ if (rc)
+ RETURN(rc);
+
+ /* will help to find FID->ino when this object is being
+ * added to PENDING/ */
+ rc = osd_idc_find_and_init(env, osd_obj2dev(obj), obj);
+
RETURN(rc);
}
osd_trans_exec_op(env, th, OSD_OT_DESTROY);
+ ldiskfs_set_inode_state(inode, LDISKFS_STATE_LUSTRE_DESTROY);
result = osd_oi_delete(osd_oti_get(env), osd, fid, oh->ot_handle,
OI_CHECK_FLD);
- /* XXX: add to ext3 orphan list */
- /* rc = ext3_orphan_add(handle_t *handle, struct inode *inode) */
+ osd_trans_exec_check(env, th, OSD_OT_DESTROY);
+ /* XXX: add to ext3 orphan list */
+ /* rc = ext3_orphan_add(handle_t *handle, struct inode *inode) */
- /* not needed in the cache anymore */
- set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
+ /* not needed in the cache anymore */
+ set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
+ obj->oo_destroyed = 1;
- RETURN(0);
+ RETURN(0);
}
/**
if (OBD_FAIL_CHECK(OBD_FAIL_FID_INLMA))
RETURN(0);
+ if (OBD_FAIL_CHECK(OBD_FAIL_OSD_OST_EA_FID_SET))
+ rc = -ENOMEM;
+
lustre_lma_init(lma, fid, compat, incompat);
lustre_lma_swab(lma);
{
struct ldiskfs_dentry_param *dot_ldp;
struct ldiskfs_dentry_param *dot_dot_ldp;
+ __u32 saved_nlink = dir->i_nlink;
+ int rc;
dot_dot_ldp = (struct ldiskfs_dentry_param *)info->oti_ldp2;
osd_get_ldiskfs_dirent_param(dot_dot_ldp, dot_dot_fid);
dot_ldp = (struct ldiskfs_dentry_param *)info->oti_ldp;
dot_ldp->edp_magic = 0;
- return ldiskfs_add_dot_dotdot(oth->ot_handle, parent_dir,
- dir, dot_ldp, dot_dot_ldp);
+
+ rc = ldiskfs_add_dot_dotdot(oth->ot_handle, parent_dir,
+ dir, dot_ldp, dot_dot_ldp);
+ /* The ldiskfs_add_dot_dotdot() may dir->i_nlink as 2, then
+ * the subseqent ref_add() will increase the dir->i_nlink
+ * as 3. That is incorrect for new created directory.
+ *
+ * It looks like hack, because we want to make the OSD API
+ * to be order-independent for new created directory object
+ * between dt_insert(..) and ref_add() operations.
+ *
+ * Here, we only restore the in-RAM dir-inode's nlink attr,
+ * becuase if the nlink attr is not 2, then there will be
+ * ref_add() called following the dt_insert(..), such call
+ * will make both the in-RAM and on-disk dir-inode's nlink
+ * attr to be set as 2. LU-7447 */
+ set_nlink(dir, saved_nlink);
+ return rc;
}
/**
RETURN(local);
}
+ ldiskfs_set_inode_state(local, LDISKFS_STATE_LUSTRE_NOSCRUB);
+ unlock_new_inode(local);
+
/* Set special LMA flag for local agent inode */
rc = osd_ea_fid_set(info, local, fid, 0, LMAI_AGENT);
if (rc != 0) {
}
/**
- * Delete local agent inode for remote entry
+ * when direntry is deleted, we have to take care of possible agent inode
+ * referenced by that. unfortunately we can't do this at that point:
+ * iget() within a running transaction leads to deadlock and we better do
+ * not call that every delete declaration to save performance. so we put
+ * a potention agent inode on a list and process that once the transaction
+ * is over. Notice it's not any worse in terms of real orphans as regular
+ * object destroy doesn't put inodes on the on-disk orphan list. this should
+ * be addressed separately
*/
-static int osd_delete_local_agent_inode(const struct lu_env *env,
- struct osd_device *osd,
- const struct lu_fid *fid,
- __u32 ino, struct osd_thandle *oh)
+static int osd_schedule_agent_inode_removal(const struct lu_env *env,
+ struct osd_thandle *oh,
+ __u32 ino)
{
- struct osd_thread_info *oti = osd_oti_get(env);
- struct osd_inode_id *id = &oti->oti_id;
- struct inode *inode;
- ENTRY;
+ struct osd_device *osd = osd_dt_dev(oh->ot_super.th_dev);
+ struct osd_obj_orphan *oor;
- id->oii_ino = le32_to_cpu(ino);
- id->oii_gen = OSD_OII_NOGEN;
- inode = osd_iget(oti, osd, id);
- if (IS_ERR(inode)) {
- CERROR("%s: iget error "DFID" id %u:%u\n", osd_name(osd),
- PFID(fid), id->oii_ino, id->oii_gen);
- RETURN(PTR_ERR(inode));
+ OBD_ALLOC_PTR(oor);
+ if (oor == NULL)
+ return -ENOMEM;
+
+ oor->oor_ino = ino;
+ oor->oor_env = (struct lu_env *)env;
+ spin_lock(&osd->od_osfs_lock);
+ list_add(&oor->oor_list, &osd->od_orphan_list);
+ spin_unlock(&osd->od_osfs_lock);
+
+ oh->ot_remove_agents = 1;
+
+ return 0;
+
+}
+
+static int osd_process_scheduled_agent_removals(const struct lu_env *env,
+ struct osd_device *osd)
+{
+ struct osd_thread_info *info = osd_oti_get(env);
+ struct osd_obj_orphan *oor, *tmp;
+ struct osd_inode_id id;
+ struct list_head list;
+ struct inode *inode;
+ struct lu_fid fid;
+ handle_t *jh;
+ __u32 ino;
+
+ INIT_LIST_HEAD(&list);
+
+ spin_lock(&osd->od_osfs_lock);
+ list_for_each_entry_safe(oor, tmp, &osd->od_orphan_list, oor_list) {
+ if (oor->oor_env == env) {
+ list_del(&oor->oor_list);
+ list_add(&oor->oor_list, &list);
+ }
}
+ spin_unlock(&osd->od_osfs_lock);
- clear_nlink(inode);
- mark_inode_dirty(inode);
- CDEBUG(D_INODE, "%s: delete remote inode "DFID" %lu\n",
- osd_name(osd), PFID(fid), inode->i_ino);
- iput(inode);
- RETURN(0);
+ list_for_each_entry_safe(oor, tmp, &list, oor_list) {
+
+ ino = oor->oor_ino;
+
+ list_del(&oor->oor_list);
+ OBD_FREE_PTR(oor);
+
+ osd_id_gen(&id, ino, OSD_OII_NOGEN);
+ inode = osd_iget_fid(info, osd, &id, &fid);
+ if (IS_ERR(inode))
+ continue;
+
+ if (!osd_remote_fid(env, osd, &fid)) {
+ iput(inode);
+ continue;
+ }
+
+ jh = osd_journal_start_sb(osd_sb(osd), LDISKFS_HT_MISC, 1);
+ clear_nlink(inode);
+ mark_inode_dirty(inode);
+ ldiskfs_journal_stop(jh);
+ iput(inode);
+ }
+
+ return 0;
}
/**
const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
struct osd_object *obj = osd_dt_obj(dt);
struct osd_thread_info *info = osd_oti_get(env);
- int result;
+ int result, on_ost = 0;
ENTRY;
+ if (dt_object_exists(dt))
+ RETURN(-EEXIST);
+
LASSERT(osd_invariant(obj));
- LASSERT(!dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(osd_write_locked(env, obj));
LASSERT(th != NULL);
* 'tune2fs -O quota' will take care of creating them */
RETURN(-EPERM);
- osd_trans_exec_op(env, th, OSD_OT_CREATE);
- osd_trans_declare_rb(env, th, OSD_OT_REF_ADD);
-
result = __osd_object_create(info, obj, attr, hint, dof, th);
- if (result == 0)
- result = osd_ea_fid_set(info, obj->oo_inode, fid,
- fid_is_on_ost(info, osd_obj2dev(obj),
- fid, OI_CHECK_FLD) ?
- LMAC_FID_ON_OST : 0, 0);
+ if (result == 0) {
+ if (fid_is_idif(fid) &&
+ !osd_dev(dt->do_lu.lo_dev)->od_index_in_idif) {
+ struct lu_fid *tfid = &info->oti_fid;
+ struct ost_id *oi = &info->oti_ostid;
+
+ fid_to_ostid(fid, oi);
+ ostid_to_fid(tfid, oi, 0);
+ on_ost = 1;
+ result = osd_ea_fid_set(info, obj->oo_inode, tfid,
+ LMAC_FID_ON_OST, 0);
+ } else {
+ on_ost = fid_is_on_ost(info, osd_obj2dev(obj),
+ fid, OI_CHECK_FLD);
+ result = osd_ea_fid_set(info, obj->oo_inode, fid,
+ on_ost ? LMAC_FID_ON_OST : 0,
+ 0);
+ }
+ if (obj->oo_dt.do_body_ops == &osd_body_ops_new)
+ obj->oo_dt.do_body_ops = &osd_body_ops;
+ }
if (result == 0)
result = __osd_oi_insert(env, obj, fid, th);
+ /* a small optimization - dt_insert() isn't usually applied
+ * to OST objects, so we don't need to cache OI mapping for
+ * OST objects */
+ if (result == 0 && on_ost == 0) {
+ struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
+ result = osd_idc_find_and_init(env, osd, obj);
+ LASSERT(result == 0);
+ }
+
LASSERT(ergo(result == 0,
dt_object_exists(dt) && !dt_object_remote(dt)));
LINVRNT(osd_invariant(obj));
struct osd_thandle *oh;
int rc = 0;
+ if (!dt_object_exists(dt) || obj->oo_destroyed)
+ return -ENOENT;
+
LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(osd_write_locked(env, obj));
LASSERT(th != NULL);
ll_dirty_inode(inode, I_DIRTY_DATASYNC);
LINVRNT(osd_invariant(obj));
+ osd_trans_exec_check(env, th, OSD_OT_REF_ADD);
+
return rc;
}
{
struct osd_thandle *oh;
- LASSERT(dt_object_exists(dt));
+ if (!dt_object_exists(dt))
+ return -ENOENT;
+
LASSERT(!dt_object_remote(dt));
LASSERT(handle != NULL);
struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
struct osd_thandle *oh;
+ if (!dt_object_exists(dt))
+ return -ENOENT;
+
LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(osd_write_locked(env, obj));
LASSERT(th != NULL);
ll_dirty_inode(inode, I_DIRTY_DATASYNC);
LINVRNT(osd_invariant(obj));
+ osd_trans_exec_check(env, th, OSD_OT_REF_DEL);
+
return 0;
}
* Concurrency: @dt is read locked.
*/
static int osd_xattr_get(const struct lu_env *env, struct dt_object *dt,
- struct lu_buf *buf, const char *name,
- struct lustre_capa *capa)
+ struct lu_buf *buf, const char *name)
{
- struct osd_object *obj = osd_dt_obj(dt);
- struct inode *inode = obj->oo_inode;
- struct osd_thread_info *info = osd_oti_get(env);
- struct dentry *dentry = &info->oti_obj_dentry;
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct inode *inode = obj->oo_inode;
+ struct osd_thread_info *info = osd_oti_get(env);
+ struct dentry *dentry = &info->oti_obj_dentry;
+ bool cache_xattr = false;
+ int rc;
- /* version get is not real XATTR but uses xattr API */
- if (strcmp(name, XATTR_NAME_VERSION) == 0) {
- /* for version we are just using xattr API but change inode
- * field instead */
+ /* version get is not real XATTR but uses xattr API */
+ if (strcmp(name, XATTR_NAME_VERSION) == 0) {
+ /* for version we are just using xattr API but change inode
+ * field instead */
if (buf->lb_len == 0)
return sizeof(dt_obj_version_t);
osd_object_version_get(env, dt, buf->lb_buf);
return sizeof(dt_obj_version_t);
- }
+ }
+
+ if (!dt_object_exists(dt))
+ return -ENOENT;
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(inode->i_op != NULL);
LASSERT(inode->i_op->getxattr != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
- return -EACCES;
+ if (strcmp(name, XATTR_NAME_LOV) == 0 ||
+ strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0)
+ cache_xattr = true;
- return __osd_xattr_get(inode, dentry, name, buf->lb_buf, buf->lb_len);
-}
+ if (cache_xattr) {
+ rc = osd_oxc_get(obj, name, buf);
+ if (rc != -ENOENT)
+ return rc;
+ }
+
+ rc = __osd_xattr_get(inode, dentry, name, buf->lb_buf, buf->lb_len);
+ if (cache_xattr) {
+ if (rc == -ENOENT || rc == -ENODATA)
+ osd_oxc_add(obj, name, NULL, 0);
+ else if (rc > 0 && buf->lb_buf != NULL)
+ osd_oxc_add(obj, name, buf->lb_buf, rc);
+ }
+ return rc;
+}
static int osd_declare_xattr_set(const struct lu_env *env,
struct dt_object *dt,
oh = container_of0(handle, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle == NULL);
- /* optimistic optimization: LMA is set first and usually fit inode */
if (strcmp(name, XATTR_NAME_LMA) == 0) {
- if (dt_object_exists(dt))
+ /* For non-upgrading case, the LMA is set first and
+ * usually fit inode. But for upgrade case, the LMA
+ * may be in another separated EA block. */
+ if (!dt_object_exists(dt))
credits = 0;
- else
+ else if (fl == LU_XATTR_REPLACE)
credits = 1;
+ else
+ goto upgrade;
} else if (strcmp(name, XATTR_NAME_VERSION) == 0) {
credits = 1;
} else {
+upgrade:
credits = osd_dto_credits_noquota[DTO_XATTR_SET];
- if (buf && buf->lb_len > sb->s_blocksize) {
- credits *= (buf->lb_len + sb->s_blocksize - 1) >>
- sb->s_blocksize_bits;
+
+ if (buf != NULL) {
+ ssize_t buflen;
+
+ if (buf->lb_buf == NULL && dt_object_exists(dt)) {
+ /* learn xattr size from osd_xattr_get if
+ attribute has not been read yet */
+ buflen = __osd_xattr_get(
+ osd_dt_obj(dt)->oo_inode,
+ &osd_oti_get(env)->oti_obj_dentry,
+ name, NULL, 0);
+ if (buflen < 0)
+ buflen = 0;
+ } else {
+ buflen = buf->lb_len;
+ }
+
+ if (buflen > sb->s_blocksize) {
+ credits += osd_calc_bkmap_credits(
+ sb, NULL, 0, -1,
+ (buflen + sb->s_blocksize - 1) >>
+ sb->s_blocksize_bits);
+ }
}
/*
* xattr set may involve inode quota change, reserve credits for
* dquot_initialize()
*/
- oh->ot_credits += LDISKFS_MAXQUOTAS_INIT_BLOCKS(sb);
+ credits += LDISKFS_MAXQUOTAS_INIT_BLOCKS(sb);
}
osd_trans_declare_op(env, oh, OSD_OT_XATTR_SET, credits);
* Concurrency: @dt is write locked.
*/
static int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
- const struct lu_buf *buf, const char *name, int fl,
- struct thandle *handle, struct lustre_capa *capa)
+ const struct lu_buf *buf, const char *name, int fl,
+ struct thandle *handle)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
struct osd_thread_info *info = osd_oti_get(env);
int fs_flags = 0;
+ int rc;
ENTRY;
- LASSERT(handle != NULL);
-
- /* version set is not real XATTR */
- if (strcmp(name, XATTR_NAME_VERSION) == 0) {
- /* for version we are just using xattr API but change inode
- * field instead */
- LASSERT(buf->lb_len == sizeof(dt_obj_version_t));
- osd_object_version_set(env, dt, buf->lb_buf);
- return sizeof(dt_obj_version_t);
- }
+ LASSERT(handle != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
- return -EACCES;
+ /* version set is not real XATTR */
+ if (strcmp(name, XATTR_NAME_VERSION) == 0) {
+ /* for version we are just using xattr API but change inode
+ * field instead */
+ LASSERT(buf->lb_len == sizeof(dt_obj_version_t));
+ osd_object_version_set(env, dt, buf->lb_buf);
+ return sizeof(dt_obj_version_t);
+ }
CDEBUG(D_INODE, DFID" set xattr '%s' with size %zu\n",
PFID(lu_object_fid(&dt->do_lu)), name, buf->lb_len);
strcmp(name, XATTR_NAME_LINK) == 0)
return -ENOSPC;
- return __osd_xattr_set(info, inode, name, buf->lb_buf, buf->lb_len,
+ rc = __osd_xattr_set(info, inode, name, buf->lb_buf, buf->lb_len,
fs_flags);
+ osd_trans_exec_check(env, handle, OSD_OT_XATTR_SET);
+
+ if (rc == 0 &&
+ (strcmp(name, XATTR_NAME_LOV) == 0 ||
+ strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0))
+ osd_oxc_add(obj, name, buf->lb_buf, buf->lb_len);
+
+ return rc;
}
/*
* Concurrency: @dt is read locked.
*/
static int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
- const struct lu_buf *buf, struct lustre_capa *capa)
+ const struct lu_buf *buf)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
struct osd_thread_info *info = osd_oti_get(env);
struct dentry *dentry = &info->oti_obj_dentry;
- LASSERT(dt_object_exists(dt));
+ if (!dt_object_exists(dt))
+ return -ENOENT;
+
LASSERT(!dt_object_remote(dt));
LASSERT(inode->i_op != NULL);
LASSERT(inode->i_op->listxattr != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
- return -EACCES;
-
dentry->d_inode = inode;
dentry->d_sb = inode->i_sb;
return inode->i_op->listxattr(dentry, buf->lb_buf, buf->lb_len);
struct osd_thandle *oh;
struct super_block *sb = osd_sb(osd_dev(dt->do_lu.lo_dev));
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(handle != NULL);
* Concurrency: @dt is write locked.
*/
static int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
- const char *name, struct thandle *handle,
- struct lustre_capa *capa)
+ const char *name, struct thandle *handle)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
struct dentry *dentry = &info->oti_obj_dentry;
int rc;
- LASSERT(dt_object_exists(dt));
+ if (!dt_object_exists(dt))
+ return -ENOENT;
+
LASSERT(!dt_object_remote(dt));
LASSERT(inode->i_op != NULL);
LASSERT(inode->i_op->removexattr != NULL);
LASSERT(handle != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
- return -EACCES;
-
osd_trans_exec_op(env, handle, OSD_OT_XATTR_SET);
ll_vfs_dq_init(inode);
dentry->d_inode = inode;
dentry->d_sb = inode->i_sb;
rc = inode->i_op->removexattr(dentry, name);
- return rc;
-}
-
-static struct obd_capa *osd_capa_get(const struct lu_env *env,
- struct dt_object *dt,
- struct lustre_capa *old, __u64 opc)
-{
- struct osd_thread_info *info = osd_oti_get(env);
- const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
- struct osd_object *obj = osd_dt_obj(dt);
- struct osd_device *osd = osd_obj2dev(obj);
- struct lustre_capa_key *key = &info->oti_capa_key;
- struct lustre_capa *capa = &info->oti_capa;
- struct obd_capa *oc;
- struct lu_capainfo *lci;
- int rc;
- ENTRY;
-
- if (!osd->od_fl_capa)
- RETURN(ERR_PTR(-ENOENT));
-
- LASSERT(dt_object_exists(dt));
- LASSERT(!dt_object_remote(dt));
- LINVRNT(osd_invariant(obj));
-
- /* renewal sanity check */
- if (old && osd_object_auth(env, dt, old, opc))
- RETURN(ERR_PTR(-EACCES));
-
- lci = lu_capainfo_get(env);
- if (unlikely(lci == NULL))
- RETURN(ERR_PTR(-ENOENT));
-
- switch (lci->lci_auth) {
- case LC_ID_NONE:
- RETURN(NULL);
- case LC_ID_PLAIN:
- capa->lc_uid = i_uid_read(obj->oo_inode);
- capa->lc_gid = i_gid_read(obj->oo_inode);
- capa->lc_flags = LC_ID_PLAIN;
- break;
- case LC_ID_CONVERT: {
- __u32 d[4], s[4];
-
- s[0] = i_uid_read(obj->oo_inode);
- cfs_get_random_bytes(&(s[1]), sizeof(__u32));
- s[2] = i_uid_read(obj->oo_inode);
- cfs_get_random_bytes(&(s[3]), sizeof(__u32));
- rc = capa_encrypt_id(d, s, key->lk_key, CAPA_HMAC_KEY_MAX_LEN);
- if (unlikely(rc))
- RETURN(ERR_PTR(rc));
-
- capa->lc_uid = ((__u64)d[1] << 32) | d[0];
- capa->lc_gid = ((__u64)d[3] << 32) | d[2];
- capa->lc_flags = LC_ID_CONVERT;
- break;
- }
- default:
- RETURN(ERR_PTR(-EINVAL));
- }
-
- capa->lc_fid = *fid;
- capa->lc_opc = opc;
- capa->lc_flags |= osd->od_capa_alg << 24;
- capa->lc_timeout = osd->od_capa_timeout;
- capa->lc_expiry = 0;
+ osd_trans_exec_check(env, handle, OSD_OT_XATTR_SET);
- oc = capa_lookup(osd->od_capa_hash, capa, 1);
- if (oc) {
- LASSERT(!capa_is_expired(oc));
- RETURN(oc);
- }
-
- spin_lock(&capa_lock);
- *key = osd->od_capa_keys[1];
- spin_unlock(&capa_lock);
-
- capa->lc_keyid = key->lk_keyid;
- capa->lc_expiry = cfs_time_current_sec() + osd->od_capa_timeout;
-
- rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
- if (rc) {
- DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
- RETURN(ERR_PTR(rc));
- }
+ if (rc == 0 &&
+ (strcmp(name, XATTR_NAME_LOV) == 0 ||
+ strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0))
+ osd_oxc_del(obj, name);
- oc = capa_add(osd->od_capa_hash, capa);
- RETURN(oc);
+ return rc;
}
static int osd_object_sync(const struct lu_env *env, struct dt_object *dt,
dentry->d_inode = inode;
dentry->d_sb = inode->i_sb;
- file->f_dentry = dentry;
+ file->f_path.dentry = dentry;
file->f_mapping = inode->i_mapping;
file->f_op = inode->i_fop;
set_file_inode(file, inode);
RETURN(rc);
}
-static int osd_data_get(const struct lu_env *env, struct dt_object *dt,
- void **data)
+static int osd_invalidate(const struct lu_env *env, struct dt_object *dt)
{
- struct osd_object *obj = osd_dt_obj(dt);
- ENTRY;
-
- *data = (void *)obj->oo_inode;
- RETURN(0);
+ return 0;
}
/*
result = 0;
} else if (feat == &dt_directory_features) {
dt->do_index_ops = &osd_index_ea_ops;
- if (obj->oo_inode != NULL && S_ISDIR(obj->oo_inode->i_mode))
+ if (obj->oo_inode == NULL || S_ISDIR(obj->oo_inode->i_mode))
result = 0;
else
result = -ENOTDIR;
}
LINVRNT(osd_invariant(obj));
- if (result == 0 && is_quota_glb_feat(feat) &&
+ if (result == 0 && feat == &dt_quota_glb_features &&
fid_seq(lu_object_fid(&dt->do_lu)) == FID_SEQ_QUOTA_GLB)
- result = osd_quota_migration(env, dt, feat);
+ result = osd_quota_migration(env, dt);
return result;
}
static int osd_otable_it_attr_get(const struct lu_env *env,
struct dt_object *dt,
- struct lu_attr *attr,
- struct lustre_capa *capa)
+ struct lu_attr *attr)
{
attr->la_valid = 0;
return 0;
.do_declare_xattr_del = osd_declare_xattr_del,
.do_xattr_del = osd_xattr_del,
.do_xattr_list = osd_xattr_list,
- .do_capa_get = osd_capa_get,
.do_object_sync = osd_object_sync,
- .do_data_get = osd_data_get,
+ .do_invalidate = osd_invalidate,
};
/**
.do_declare_xattr_del = osd_declare_xattr_del,
.do_xattr_del = osd_xattr_del,
.do_xattr_list = osd_xattr_list,
- .do_capa_get = osd_capa_get,
.do_object_sync = osd_object_sync,
- .do_data_get = osd_data_get,
+ .do_invalidate = osd_invalidate,
};
static const struct dt_object_operations osd_obj_otable_it_ops = {
oh = container_of0(handle, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle == NULL);
+ /* Recycle may cause additional three blocks to be changed. */
osd_trans_declare_op(env, oh, OSD_OT_DELETE,
- osd_dto_credits_noquota[DTO_INDEX_DELETE]);
+ osd_dto_credits_noquota[DTO_INDEX_DELETE] + 3);
return 0;
}
*/
static int osd_index_iam_delete(const struct lu_env *env, struct dt_object *dt,
const struct dt_key *key,
- struct thandle *handle,
- struct lustre_capa *capa)
+ struct thandle *handle)
{
struct osd_thread_info *oti = osd_oti_get(env);
struct osd_object *obj = osd_dt_obj(dt);
int rc;
ENTRY;
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
+
LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(bag->ic_object == obj->oo_inode);
LASSERT(handle != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
- RETURN(-EACCES);
-
osd_trans_exec_op(env, handle, OSD_OT_DELETE);
ipd = osd_idx_ipd_get(env, bag);
rc = iam_delete(oh->ot_handle, bag, (const struct iam_key *)key, ipd);
osd_ipd_put(env, bag, ipd);
LINVRNT(osd_invariant(obj));
+ osd_trans_exec_check(env, handle, OSD_OT_DELETE);
RETURN(rc);
}
int rc;
ENTRY;
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(handle != NULL);
oh = container_of0(handle, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle == NULL);
+ /* due to DNE we may need to remove an agent inode */
osd_trans_declare_op(env, oh, OSD_OT_DELETE,
- osd_dto_credits_noquota[DTO_INDEX_DELETE]);
+ osd_dto_credits_noquota[DTO_INDEX_DELETE] +
+ osd_dto_credits_noquota[DTO_OBJECT_DELETE]);
inode = osd_dt_obj(dt)->oo_inode;
- LASSERT(inode);
+ if (inode == NULL)
+ RETURN(-ENOENT);
rc = osd_declare_inode_qid(env, i_uid_read(inode), i_gid_read(inode),
0, oh, osd_dt_obj(dt), true, NULL, false);
if (de->file_type & LDISKFS_DIRENT_LUFID) {
rec = (struct osd_fid_pack *) (de->name + de->name_len + 1);
rc = osd_fid_unpack((struct lu_fid *)fid, rec);
+ if (rc == 0 && unlikely(!fid_is_sane((struct lu_fid *)fid)))
+ rc = -EINVAL;
}
return rc;
}
* \retval -ve, on error
*/
static int osd_index_ea_delete(const struct lu_env *env, struct dt_object *dt,
- const struct dt_key *key, struct thandle *handle,
- struct lustre_capa *capa)
+ const struct dt_key *key, struct thandle *handle)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *dir = obj->oo_inode;
int rc;
ENTRY;
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
+
LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(handle != NULL);
LASSERT(oh->ot_handle != NULL);
LASSERT(oh->ot_handle->h_transaction != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
- RETURN(-EACCES);
-
ll_vfs_dq_init(dir);
dentry = osd_child_dentry_get(env, obj,
(char *)key, strlen((char *)key));
}
bh = osd_ldiskfs_find_entry(dir, &dentry->d_name, &de, NULL, hlock);
- if (bh) {
- __u32 ino = 0;
-
+ if (!IS_ERR(bh)) {
/* If this is not the ".." entry, it might be a remote DNE
* entry and we need to check if the FID is for a remote
* MDT. If the FID is not in the directory entry (e.g.
if (strcmp((char *)key, dotdot) != 0) {
LASSERT(de != NULL);
rc = osd_get_fid_from_dentry(de, (struct dt_rec *)fid);
- /* If Fid is not in dentry, try to get it from LMA */
if (rc == -ENODATA) {
- struct osd_inode_id *id;
- struct inode *inode;
-
- /* Before trying to get fid from the inode,
- * check whether the inode is valid.
- *
- * If the inode has been deleted, do not go
- * ahead to do osd_ea_fid_get, which will set
- * the inode to bad inode, which might cause
- * the inode to be deleted uncorrectly */
- inode = ldiskfs_iget(osd_sb(osd),
- le32_to_cpu(de->inode));
- if (IS_ERR(inode)) {
- CDEBUG(D_INODE, "%s: "DFID"get inode"
- "error.\n", osd_name(osd),
- PFID(fid));
- rc = PTR_ERR(inode);
- } else {
- if (likely(inode->i_nlink != 0)) {
- id = &osd_oti_get(env)->oti_id;
- rc = osd_ea_fid_get(env, obj,
- le32_to_cpu(de->inode),
- fid, id);
- } else {
- CDEBUG(D_INFO, "%s: %u "DFID
- "deleted.\n",
- osd_name(osd),
- le32_to_cpu(de->inode),
- PFID(fid));
- rc = -ESTALE;
- }
- iput(inode);
- }
+ /* can't get FID, postpone to the end of the
+ * transaction when iget() is safe */
+ osd_schedule_agent_inode_removal(env, oh,
+ le32_to_cpu(de->inode));
+ } else if (rc == 0 &&
+ unlikely(osd_remote_fid(env, osd, fid))) {
+ osd_schedule_agent_inode_removal(env, oh,
+ le32_to_cpu(de->inode));
}
- if (rc == 0 &&
- unlikely(osd_remote_fid(env, osd, fid)))
- /* Need to delete agent inode */
- ino = le32_to_cpu(de->inode);
- }
- rc = ldiskfs_delete_entry(oh->ot_handle, dir, de, bh);
- brelse(bh);
- if (rc == 0 && unlikely(ino != 0)) {
- rc = osd_delete_local_agent_inode(env, osd, fid, ino,
- oh);
- if (rc != 0)
- CERROR("%s: del local inode "DFID": rc = %d\n",
- osd_name(osd), PFID(fid), rc);
}
- } else {
- rc = -ENOENT;
- }
+ rc = ldiskfs_delete_entry(oh->ot_handle, dir, de, bh);
+ brelse(bh);
+ } else {
+ rc = PTR_ERR(bh);
+ }
if (hlock != NULL)
ldiskfs_htree_unlock(hlock);
else
out:
LASSERT(osd_invariant(obj));
+ osd_trans_exec_check(env, handle, OSD_OT_DELETE);
RETURN(rc);
}
* \retval -ve failure
*/
static int osd_index_iam_lookup(const struct lu_env *env, struct dt_object *dt,
- struct dt_rec *rec, const struct dt_key *key,
- struct lustre_capa *capa)
+ struct dt_rec *rec, const struct dt_key *key)
{
struct osd_object *obj = osd_dt_obj(dt);
struct iam_path_descr *ipd;
int rc;
ENTRY;
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
+
LASSERT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(bag->ic_object == obj->oo_inode);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_LOOKUP))
- RETURN(-EACCES);
-
ipd = osd_idx_ipd_get(env, bag);
if (IS_ERR(ipd))
RETURN(-ENOMEM);
static int osd_index_iam_insert(const struct lu_env *env, struct dt_object *dt,
const struct dt_rec *rec,
const struct dt_key *key, struct thandle *th,
- struct lustre_capa *capa, int ignore_quota)
+ int ignore_quota)
{
struct osd_object *obj = osd_dt_obj(dt);
struct iam_path_descr *ipd;
struct osd_thandle *oh;
- struct iam_container *bag = &obj->oo_dir->od_container;
+ struct iam_container *bag;
struct osd_thread_info *oti = osd_oti_get(env);
struct iam_rec *iam_rec;
int rc;
ENTRY;
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
+
LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
+
+ bag = &obj->oo_dir->od_container;
LASSERT(bag->ic_object == obj->oo_inode);
LASSERT(th != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
- RETURN(-EACCES);
-
osd_trans_exec_op(env, th, OSD_OT_INSERT);
ipd = osd_idx_ipd_get(env, bag);
iam_rec, ipd);
osd_ipd_put(env, bag, ipd);
LINVRNT(osd_invariant(obj));
+ osd_trans_exec_check(env, th, OSD_OT_INSERT);
RETURN(rc);
}
child = osd_child_dentry_get(info->oti_env, pobj, name, strlen(name));
child->d_fsdata = (void *)ldp;
ll_vfs_dq_init(pobj->oo_inode);
- rc = osd_ldiskfs_add_entry(oth->ot_handle, child, cinode, hlock);
+ rc = osd_ldiskfs_add_entry(info, oth->ot_handle, child,
+ cinode, hlock);
if (rc == 0 && OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_TYPE)) {
struct ldiskfs_dir_entry_2 *de;
struct buffer_head *bh;
bh = osd_ldiskfs_find_entry(pobj->oo_inode, &child->d_name, &de,
NULL, hlock);
- if (bh != NULL) {
+ if (!IS_ERR(bh)) {
rc1 = ldiskfs_journal_get_write_access(oth->ot_handle,
bh);
if (rc1 == 0) {
else
de->file_type = LDISKFS_DIRENT_LUFID |
LDISKFS_FT_DIR;
- ldiskfs_journal_dirty_metadata(oth->ot_handle,
- bh);
- brelse(bh);
+ ldiskfs_handle_dirty_metadata(oth->ot_handle,
+ NULL, bh);
}
+ brelse(bh);
}
}
if (dir->oo_compat_dot_created) {
result = -EEXIST;
} else {
- LASSERT(inode == parent_dir);
+ LASSERT(inode->i_ino == parent_dir->i_ino);
dir->oo_compat_dot_created = 1;
result = 0;
}
return rc;
}
-static void
+static int
osd_consistency_check(struct osd_thread_info *oti, struct osd_device *dev,
struct osd_idmap_cache *oic)
{
ENTRY;
if (!fid_is_norm(fid) && !fid_is_igif(fid))
- RETURN_EXIT;
+ RETURN(0);
if (scrub->os_pos_current > id->oii_ino)
- RETURN_EXIT;
+ RETURN(0);
again:
- rc = osd_oi_lookup(oti, dev, fid, id, OI_CHECK_FLD);
- if (rc != 0 && rc != -ENOENT)
- RETURN_EXIT;
+ rc = osd_oi_lookup(oti, dev, fid, id, 0);
+ if (rc == -ENOENT) {
+ struct inode *inode;
+
+ *id = oic->oic_lid;
+ inode = osd_iget(oti, dev, &oic->oic_lid);
+
+ /* The inode has been removed (by race maybe). */
+ if (IS_ERR(inode)) {
+ rc = PTR_ERR(inode);
- if (rc == 0 && osd_id_eq(id, &oic->oic_lid))
- RETURN_EXIT;
+ RETURN(rc == -ESTALE ? -ENOENT : rc);
+ }
+
+ iput(inode);
+ /* The OI mapping is lost. */
+ if (id->oii_gen != OSD_OII_NOGEN)
+ goto trigger;
+
+ /* The inode may has been reused by others, we do not know,
+ * leave it to be handled by subsequent osd_fid_lookup(). */
+ RETURN(0);
+ } else if (rc != 0 || osd_id_eq(id, &oic->oic_lid)) {
+ RETURN(rc);
+ }
+trigger:
if (thread_is_running(&scrub->os_thread)) {
rc = osd_oii_insert(dev, oic, rc == -ENOENT);
/* There is race condition between osd_oi_lookup and OI scrub.
if (unlikely(rc == -EAGAIN))
goto again;
- RETURN_EXIT;
+ RETURN(0);
}
if (!dev->od_noscrub && ++once == 1) {
rc = osd_scrub_start(dev, SS_AUTO_PARTIAL | SS_CLEAR_DRYRUN |
SS_CLEAR_FAILOUT);
- LCONSOLE_WARN("%.16s: trigger OI scrub by RPC for "DFID
- ", rc = %d [2]\n",
- LDISKFS_SB(osd_sb(dev))->s_es->s_volume_name,
- PFID(fid), rc);
+ CDEBUG(D_LFSCK | D_CONSOLE, "%.16s: trigger OI scrub by RPC "
+ "for "DFID", rc = %d [2]\n",
+ LDISKFS_SB(osd_sb(dev))->s_es->s_volume_name,
+ PFID(fid), rc);
if (rc == 0 || rc == -EALREADY)
goto again;
}
- EXIT;
+ RETURN(0);
}
static int osd_fail_fid_lookup(struct osd_thread_info *oti,
struct lu_buf *buf = &oti->oti_big_buf;
struct dentry *dentry = &oti->oti_obj_dentry;
struct inode *inode = obj->oo_inode;
- struct linkea_data ldata = { 0 };
+ struct linkea_data ldata = { NULL };
int rc;
ENTRY;
}
bh = osd_ldiskfs_find_entry(dir, &dentry->d_name, &de, NULL, hlock);
- if (bh) {
+ if (!IS_ERR(bh)) {
struct osd_thread_info *oti = osd_oti_get(env);
struct osd_inode_id *id = &oti->oti_id;
struct osd_idmap_cache *oic = &oti->oti_cache;
/* done with de, release bh */
brelse(bh);
if (rc != 0) {
- if (unlikely(ino == osd_remote_parent_ino(dev)))
+ if (unlikely(ino == osd_remote_parent_ino(dev))) {
+ const char *name = (const char *)key;
+
/* If the parent is on remote MDT, and there
* is no FID-in-dirent, then we have to get
* the parent FID from the linkEA. */
- rc = osd_get_pfid_from_linkea(env, obj, fid);
- else
+ if (likely(strlen(name) == 2 &&
+ name[0] == '.' && name[1] == '.'))
+ rc = osd_get_pfid_from_linkea(env, obj,
+ fid);
+ } else {
rc = osd_ea_fid_get(env, obj, ino, fid, id);
+ }
} else {
osd_id_gen(id, ino, OSD_OII_NOGEN);
}
- if (rc != 0) {
+ if (rc != 0 || osd_remote_fid(env, dev, fid)) {
fid_zero(&oic->oic_fid);
+
GOTO(out, rc);
}
- if (osd_remote_fid(env, dev, fid))
- GOTO(out, rc = 0);
-
osd_add_oi_cache(osd_oti_get(env), osd_obj2dev(obj), id, fid);
- osd_consistency_check(oti, dev, oic);
+ rc = osd_consistency_check(oti, dev, oic);
+ if (rc != 0)
+ fid_zero(&oic->oic_fid);
} else {
- rc = -ENOENT;
+ rc = PTR_ERR(bh);
}
GOTO(out, rc);
}
/**
- * Find the osd object for given fid.
- *
- * \param fid need to find the osd object having this fid
- *
- * \retval osd_object on success
- * \retval -ve on error
- */
-static struct osd_object *osd_object_find(const struct lu_env *env,
- struct dt_object *dt,
- const struct lu_fid *fid)
-{
- struct lu_device *ludev = dt->do_lu.lo_dev;
- struct osd_object *child = NULL;
- struct lu_object *luch;
- struct lu_object *lo;
-
- /*
- * at this point topdev might not exist yet
- * (i.e. MGS is preparing profiles). so we can
- * not rely on topdev and instead lookup with
- * our device passed as topdev. this can't work
- * if the object isn't cached yet (as osd doesn't
- * allocate lu_header). IOW, the object must be
- * in the cache, otherwise lu_object_alloc() crashes
- * -bzzz
- */
- luch = lu_object_find_at(env, ludev->ld_site->ls_top_dev == NULL ?
- ludev : ludev->ld_site->ls_top_dev,
- fid, NULL);
- if (!IS_ERR(luch)) {
- if (lu_object_exists(luch)) {
- lo = lu_object_locate(luch->lo_header, ludev->ld_type);
- if (lo != NULL)
- child = osd_obj(lo);
- else
- LU_OBJECT_DEBUG(D_ERROR, env, luch,
- "lu_object can't be located"
- DFID"\n", PFID(fid));
-
- if (child == NULL) {
- lu_object_put(env, luch);
- CERROR("Unable to get osd_object\n");
- child = ERR_PTR(-ENOENT);
- }
- } else {
- LU_OBJECT_DEBUG(D_ERROR, env, luch,
- "lu_object does not exists "DFID"\n",
- PFID(fid));
- lu_object_put(env, luch);
- child = ERR_PTR(-ENOENT);
- }
- } else {
- child = ERR_CAST(luch);
- }
-
- return child;
-}
-
-/**
* Put the osd object once done with it.
*
* \param obj osd object that needs to be put
{
struct osd_thandle *oh;
struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
- struct lu_fid *fid = (struct lu_fid *)rec;
- int rc;
+ struct dt_insert_rec *rec1 = (struct dt_insert_rec *)rec;
+ const struct lu_fid *fid = rec1->rec_fid;
+ int credits, rc = 0;
+ struct osd_idmap_cache *idc;
ENTRY;
LASSERT(!dt_object_remote(dt));
LASSERT(handle != NULL);
+ LASSERT(fid != NULL);
+ LASSERT(rec1->rec_type != 0);
oh = container_of0(handle, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle == NULL);
- osd_trans_declare_op(env, oh, OSD_OT_INSERT,
- osd_dto_credits_noquota[DTO_INDEX_INSERT]);
+ credits = osd_dto_credits_noquota[DTO_INDEX_INSERT];
+
+ /* we can't call iget() while a transactions is running
+ * (this can lead to a deadlock), but we need to know
+ * inum and object type. so we find this information at
+ * declaration and cache in per-thread info */
+ idc = osd_idc_find_or_init(env, osd, fid);
+ if (IS_ERR(idc))
+ RETURN(PTR_ERR(idc));
+ if (idc->oic_remote) {
+ /* a reference to remote inode is represented by an
+ * agent inode which we have to create */
+ credits += osd_dto_credits_noquota[DTO_OBJECT_CREATE];
+ credits += osd_dto_credits_noquota[DTO_INDEX_INSERT];
+ }
+
+ osd_trans_declare_op(env, oh, OSD_OT_INSERT, credits);
if (osd_dt_obj(dt)->oo_inode != NULL) {
struct inode *inode = osd_dt_obj(dt)->oo_inode;
osd_dt_obj(dt), true, NULL, false);
}
- if (fid == NULL)
- RETURN(0);
-
- rc = osd_remote_fid(env, osd, fid);
- if (rc <= 0)
- RETURN(rc);
-
- rc = 0;
-
- osd_trans_declare_op(env, oh, OSD_OT_CREATE,
- osd_dto_credits_noquota[DTO_OBJECT_CREATE]);
- osd_trans_declare_op(env, oh, OSD_OT_INSERT,
- osd_dto_credits_noquota[DTO_INDEX_INSERT] + 1);
- osd_trans_declare_op(env, oh, OSD_OT_INSERT,
- osd_dto_credits_noquota[DTO_INDEX_INSERT] + 1);
-
RETURN(rc);
}
* \retval -ve, on error
*/
static int osd_index_ea_insert(const struct lu_env *env, struct dt_object *dt,
- const struct dt_rec *rec,
- const struct dt_key *key, struct thandle *th,
- struct lustre_capa *capa, int ignore_quota)
+ const struct dt_rec *rec,
+ const struct dt_key *key, struct thandle *th,
+ int ignore_quota)
{
struct osd_object *obj = osd_dt_obj(dt);
struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
const struct lu_fid *fid = rec1->rec_fid;
const char *name = (const char *)key;
struct osd_thread_info *oti = osd_oti_get(env);
- struct osd_inode_id *id = &oti->oti_id;
struct inode *child_inode = NULL;
- struct osd_object *child = NULL;
+ struct osd_idmap_cache *idc;
int rc;
ENTRY;
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
+
LASSERT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt));
LASSERT(!dt_object_remote(dt));
LASSERT(th != NULL);
osd_trans_exec_op(env, th, OSD_OT_INSERT);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
- RETURN(-EACCES);
-
LASSERTF(fid_is_sane(fid), "fid"DFID" is insane!\n", PFID(fid));
- rc = osd_remote_fid(env, osd, fid);
- if (rc < 0) {
- CERROR("%s: Can not find object "DFID" rc %d\n",
- osd_name(osd), PFID(fid), rc);
- RETURN(rc);
+ idc = osd_idc_find(env, osd, fid);
+ if (unlikely(idc == NULL)) {
+ /* this dt_insert() wasn't declared properly, so
+ * FID is missing in OI cache. we better do not
+ * lookup FID in FLDB/OI and don't risk to deadlock,
+ * but in some special cases (lfsck testing, etc)
+ * it's much simpler than fixing a caller */
+ CERROR("%s: "DFID" wasn't declared for insert\n",
+ osd_name(osd), PFID(fid));
+ dump_stack();
+ idc = osd_idc_find_or_init(env, osd, fid);
+ if (IS_ERR(idc))
+ RETURN(PTR_ERR(idc));
}
- if (rc == 1) {
+ if (idc->oic_remote) {
/* Insert remote entry */
if (strcmp(name, dotdot) == 0 && strlen(name) == 2) {
struct osd_mdobj_map *omm = osd->od_mdt_map;
}
} else {
/* Insert local entry */
- child = osd_object_find(env, dt, fid);
- if (IS_ERR(child)) {
- CERROR("%s: Can not find object "DFID"%u:%u: rc = %d\n",
- osd_name(osd), PFID(fid),
- id->oii_ino, id->oii_gen,
- (int)PTR_ERR(child));
- RETURN(PTR_ERR(child));
+ if (unlikely(idc->oic_lid.oii_ino == 0)) {
+ /* for a reason OI cache wasn't filled properly */
+ CERROR("%s: OIC for "DFID" isn't filled\n",
+ osd_name(osd), PFID(fid));
+ RETURN(-EINVAL);
+ }
+ child_inode = oti->oti_inode;
+ if (unlikely(child_inode == NULL)) {
+ struct ldiskfs_inode_info *lii;
+ OBD_ALLOC_PTR(lii);
+ if (lii == NULL)
+ RETURN(-ENOMEM);
+ child_inode = oti->oti_inode = &lii->vfs_inode;
}
- child_inode = igrab(child->oo_inode);
+ child_inode->i_sb = osd_sb(osd);
+ child_inode->i_ino = idc->oic_lid.oii_ino;
+ child_inode->i_mode = rec1->rec_type & S_IFMT;
}
rc = osd_ea_add_rec(env, obj, child_inode, name, fid, th);
CDEBUG(D_INODE, "parent %lu insert %s:%lu rc = %d\n",
obj->oo_inode->i_ino, name, child_inode->i_ino, rc);
- iput(child_inode);
- if (child != NULL)
- osd_object_put(env, child);
+ if (child_inode && child_inode != oti->oti_inode)
+ iput(child_inode);
LASSERT(osd_invariant(obj));
+ osd_trans_exec_check(env, th, OSD_OT_INSERT);
RETURN(rc);
}
*/
static struct dt_it *osd_it_iam_init(const struct lu_env *env,
- struct dt_object *dt,
- __u32 unused,
- struct lustre_capa *capa)
+ struct dt_object *dt,
+ __u32 unused)
{
- struct osd_it_iam *it;
- struct osd_thread_info *oti = osd_oti_get(env);
- struct osd_object *obj = osd_dt_obj(dt);
- struct lu_object *lo = &dt->do_lu;
- struct iam_path_descr *ipd;
- struct iam_container *bag = &obj->oo_dir->od_container;
-
- LASSERT(lu_object_exists(lo));
+ struct osd_it_iam *it;
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct lu_object *lo = &dt->do_lu;
+ struct iam_path_descr *ipd;
+ struct iam_container *bag = &obj->oo_dir->od_container;
- if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_READ))
- return ERR_PTR(-EACCES);
+ if (!dt_object_exists(dt))
+ return ERR_PTR(-ENOENT);
- if (oti->oti_it_inline) {
- OBD_ALLOC_PTR(it);
- if (it == NULL)
- return ERR_PTR(-ENOMEM);
- } else {
- it = &oti->oti_it;
- oti->oti_it_inline = 1;
- }
+ OBD_ALLOC_PTR(it);
+ if (it == NULL)
+ return ERR_PTR(-ENOMEM);
ipd = osd_it_ipd_get(env, bag);
if (likely(ipd != NULL)) {
iam_it_init(&it->oi_it, bag, IAM_IT_MOVE, ipd);
return (struct dt_it *)it;
} else {
- if (it != &oti->oti_it)
- OBD_FREE_PTR(it);
- else
- oti->oti_it_inline = 0;
-
+ OBD_FREE_PTR(it);
return ERR_PTR(-ENOMEM);
}
}
static void osd_it_iam_fini(const struct lu_env *env, struct dt_it *di)
{
- struct osd_thread_info *oti = osd_oti_get(env);
struct osd_it_iam *it = (struct osd_it_iam *)di;
struct osd_object *obj = it->oi_obj;
iam_it_fini(&it->oi_it);
osd_ipd_put(env, &obj->oo_dir->od_container, it->oi_ipd);
lu_object_put(env, &obj->oo_dt.do_lu);
- if (it != &oti->oti_it)
- OBD_FREE_PTR(it);
- else
- oti->oti_it_inline = 0;
+ OBD_FREE_PTR(it);
}
/**
*/
static struct dt_it *osd_it_ea_init(const struct lu_env *env,
struct dt_object *dt,
- __u32 attr,
- struct lustre_capa *capa)
+ __u32 attr)
{
struct osd_object *obj = osd_dt_obj(dt);
struct osd_thread_info *info = osd_oti_get(env);
- struct osd_it_ea *it;
+ struct osd_it_ea *oie;
struct file *file;
struct lu_object *lo = &dt->do_lu;
- struct dentry *obj_dentry = &info->oti_it_dentry;
+ struct dentry *obj_dentry;
ENTRY;
- LASSERT(lu_object_exists(lo));
+ if (!dt_object_exists(dt) || obj->oo_destroyed)
+ RETURN(ERR_PTR(-ENOENT));
- if (info->oti_it_inline) {
- OBD_ALLOC_PTR(it);
- if (it == NULL)
- RETURN(ERR_PTR(-ENOMEM));
- } else {
- it = &info->oti_it_ea;
- info->oti_it_inline = 1;
- }
+ OBD_SLAB_ALLOC_PTR_GFP(oie, osd_itea_cachep, GFP_NOFS);
+ if (oie == NULL)
+ RETURN(ERR_PTR(-ENOMEM));
+ obj_dentry = &oie->oie_dentry;
obj_dentry->d_inode = obj->oo_inode;
obj_dentry->d_sb = osd_sb(osd_obj2dev(obj));
obj_dentry->d_name.hash = 0;
- it->oie_rd_dirent = 0;
- it->oie_it_dirent = 0;
- it->oie_dirent = NULL;
- it->oie_buf = info->oti_it_ea_buf;
- it->oie_obj = obj;
+ oie->oie_rd_dirent = 0;
+ oie->oie_it_dirent = 0;
+ oie->oie_dirent = NULL;
+ if (unlikely(!info->oti_it_ea_buf_used)) {
+ oie->oie_buf = info->oti_it_ea_buf;
+ info->oti_it_ea_buf_used = 1;
+ } else {
+ OBD_ALLOC(oie->oie_buf, OSD_IT_EA_BUFSIZE);
+ if (oie->oie_buf == NULL)
+ RETURN(ERR_PTR(-ENOMEM));
+ }
+ oie->oie_obj = obj;
+
+ file = &oie->oie_file;
- file = &it->oie_file;
- /* Reset the "file" totally to avoid to reuse any old value from
- * former readdir handling, the "file->f_pos" should be zero. */
- memset(file, 0, sizeof(*file));
/* Only FMODE_64BITHASH or FMODE_32BITHASH should be set, NOT both. */
if (attr & LUDA_64BITHASH)
file->f_mode = FMODE_64BITHASH;
else
file->f_mode = FMODE_32BITHASH;
- file->f_dentry = obj_dentry;
- file->f_mapping = obj->oo_inode->i_mapping;
+ file->f_path.dentry = obj_dentry;
+ file->f_mapping = obj->oo_inode->i_mapping;
file->f_op = obj->oo_inode->i_fop;
set_file_inode(file, obj->oo_inode);
lu_object_get(lo);
- RETURN((struct dt_it *) it);
+ RETURN((struct dt_it *) oie);
}
/**
*/
static void osd_it_ea_fini(const struct lu_env *env, struct dt_it *di)
{
- struct osd_thread_info *info = osd_oti_get(env);
- struct osd_it_ea *it = (struct osd_it_ea *)di;
- struct osd_object *obj = it->oie_obj;
- struct inode *inode = obj->oo_inode;
+ struct osd_thread_info *info = osd_oti_get(env);
+ struct osd_it_ea *oie = (struct osd_it_ea *)di;
+ struct osd_object *obj = oie->oie_obj;
+ struct inode *inode = obj->oo_inode;
- ENTRY;
- it->oie_file.f_op->release(inode, &it->oie_file);
- lu_object_put(env, &obj->oo_dt.do_lu);
- if (it != &info->oti_it_ea)
- OBD_FREE_PTR(it);
+ ENTRY;
+ oie->oie_file.f_op->release(inode, &oie->oie_file);
+ lu_object_put(env, &obj->oo_dt.do_lu);
+ if (unlikely(oie->oie_buf != info->oti_it_ea_buf))
+ OBD_FREE(oie->oie_buf, OSD_IT_EA_BUFSIZE);
else
- info->oti_it_inline = 0;
+ info->oti_it_ea_buf_used = 0;
+ OBD_SLAB_FREE_PTR(oie, osd_itea_cachep);
EXIT;
}
* \retval 0 on success
* \retval 1 on buffer full
*/
+#ifdef HAVE_FILLDIR_USE_CTX
+static int osd_ldiskfs_filldir(struct dir_context *buf,
+ const char *name, int namelen,
+#else
static int osd_ldiskfs_filldir(void *buf, const char *name, int namelen,
+#endif
loff_t offset, __u64 ino,
unsigned d_type)
{
- struct osd_it_ea *it = ((struct osd_filldir_cbs *)buf)->it;
+ struct osd_it_ea *it =
+ ((struct osd_filldir_cbs *)buf)->it;
struct osd_object *obj = it->oie_obj;
struct osd_it_ea_dirent *ent = it->oie_dirent;
struct lu_fid *fid = &ent->oied_fid;
return it->oie_dirent->oied_namelen;
}
-static int
-osd_dirent_update(handle_t *jh, struct super_block *sb,
- struct osd_it_ea_dirent *ent, struct lu_fid *fid,
- struct buffer_head *bh, struct ldiskfs_dir_entry_2 *de)
+static inline bool
+osd_dot_dotdot_has_space(struct ldiskfs_dir_entry_2 *de, int dot_dotdot)
{
- struct osd_fid_pack *rec;
- int rc;
- ENTRY;
-
- LASSERT(de->file_type & LDISKFS_DIRENT_LUFID);
- LASSERT(de->rec_len >= de->name_len + sizeof(struct osd_fid_pack));
-
- rc = ldiskfs_journal_get_write_access(jh, bh);
- if (rc != 0)
- RETURN(rc);
+ LASSERTF(dot_dotdot == 1 || dot_dotdot == 2,
+ "dot_dotdot = %d\n", dot_dotdot);
- rec = (struct osd_fid_pack *)(de->name + de->name_len + 1);
- fid_cpu_to_be((struct lu_fid *)rec->fp_area, fid);
- rc = ldiskfs_journal_dirty_metadata(jh, bh);
+ if (LDISKFS_DIR_REC_LEN(de) >=
+ __LDISKFS_DIR_REC_LEN(dot_dotdot + 1 + sizeof(struct osd_fid_pack)))
+ return true;
- RETURN(rc);
+ return false;
}
-static inline int
-osd_dirent_has_space(__u16 reclen, __u16 namelen, unsigned blocksize)
+static inline bool
+osd_dirent_has_space(struct ldiskfs_dir_entry_2 *de, __u16 namelen,
+ unsigned blocksize, int dot_dotdot)
{
- if (ldiskfs_rec_len_from_disk(reclen, blocksize) >=
- __LDISKFS_DIR_REC_LEN(namelen + 1 + sizeof(struct osd_fid_pack)))
- return 1;
- else
- return 0;
-}
+ if (dot_dotdot > 0)
+ return osd_dot_dotdot_has_space(de, dot_dotdot);
-static inline int
-osd_dot_dotdot_has_space(struct ldiskfs_dir_entry_2 *de, int dot_dotdot)
-{
- LASSERTF(dot_dotdot == 1 || dot_dotdot == 2,
- "dot_dotdot = %d\n", dot_dotdot);
+ if (ldiskfs_rec_len_from_disk(de->rec_len, blocksize) >=
+ __LDISKFS_DIR_REC_LEN(namelen + 1 + sizeof(struct osd_fid_pack)))
+ return true;
- if (LDISKFS_DIR_REC_LEN(de) >=
- __LDISKFS_DIR_REC_LEN(dot_dotdot + 1 + sizeof(struct osd_fid_pack)))
- return 1;
- else
- return 0;
+ return false;
}
static int
osd_dirent_reinsert(const struct lu_env *env, handle_t *jh,
- struct inode *dir, struct inode *inode,
- struct osd_it_ea_dirent *ent, struct lu_fid *fid,
+ struct dentry *dentry, const struct lu_fid *fid,
struct buffer_head *bh, struct ldiskfs_dir_entry_2 *de,
- struct htree_lock *hlock)
+ struct htree_lock *hlock, int dot_dotdot)
{
- struct dentry *dentry;
+ struct inode *dir = dentry->d_parent->d_inode;
+ struct inode *inode = dentry->d_inode;
struct osd_fid_pack *rec;
struct ldiskfs_dentry_param *ldp;
+ int namelen = dentry->d_name.len;
int rc;
+ struct osd_thread_info *info = osd_oti_get(env);
ENTRY;
if (!LDISKFS_HAS_INCOMPAT_FEATURE(inode->i_sb,
RETURN(0);
/* There is enough space to hold the FID-in-dirent. */
- if (osd_dirent_has_space(de->rec_len, ent->oied_namelen,
- dir->i_sb->s_blocksize)) {
+ if (osd_dirent_has_space(de, namelen, dir->i_sb->s_blocksize,
+ dot_dotdot)) {
rc = ldiskfs_journal_get_write_access(jh, bh);
if (rc != 0)
RETURN(rc);
- de->name[de->name_len] = 0;
- rec = (struct osd_fid_pack *)(de->name + de->name_len + 1);
+ de->name[namelen] = 0;
+ rec = (struct osd_fid_pack *)(de->name + namelen + 1);
rec->fp_len = sizeof(struct lu_fid) + 1;
fid_cpu_to_be((struct lu_fid *)rec->fp_area, fid);
de->file_type |= LDISKFS_DIRENT_LUFID;
-
- rc = ldiskfs_journal_dirty_metadata(jh, bh);
+ rc = ldiskfs_handle_dirty_metadata(jh, NULL, bh);
RETURN(rc);
}
+ LASSERTF(dot_dotdot == 0, "dot_dotdot = %d\n", dot_dotdot);
+
rc = ldiskfs_delete_entry(jh, dir, de, bh);
if (rc != 0)
RETURN(rc);
- dentry = osd_child_dentry_by_inode(env, dir, ent->oied_name,
- ent->oied_namelen);
ldp = (struct ldiskfs_dentry_param *)osd_oti_get(env)->oti_ldp;
osd_get_ldiskfs_dirent_param(ldp, fid);
dentry->d_fsdata = (void *)ldp;
ll_vfs_dq_init(dir);
- rc = osd_ldiskfs_add_entry(jh, dentry, inode, hlock);
+ rc = osd_ldiskfs_add_entry(info, jh, dentry, inode, hlock);
/* It is too bad, we cannot reinsert the name entry back.
* That means we lose it! */
if (rc != 0)
CDEBUG(D_LFSCK, "%.16s: fail to reinsert the dirent, "
"dir = %lu/%u, name = %.*s, "DFID": rc = %d\n",
LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
- dir->i_ino, dir->i_generation,
- ent->oied_namelen, ent->oied_name, PFID(fid), rc);
+ dir->i_ino, dir->i_generation, namelen,
+ dentry->d_name.name, PFID(fid), rc);
RETURN(rc);
}
bool dirty = false;
ENTRY;
+ osd_id_gen(id, ent->oied_ino, OSD_OII_NOGEN);
+ inode = osd_iget(info, dev, id);
+ if (IS_ERR(inode)) {
+ rc = PTR_ERR(inode);
+ if (rc == -ENOENT || rc == -ESTALE) {
+ *attr |= LUDA_UNKNOWN;
+ rc = 0;
+ } else {
+ CDEBUG(D_LFSCK, "%.16s: fail to iget for dirent "
+ "check_repair, dir = %lu/%u, name = %.*s: "
+ "rc = %d\n",
+ devname, dir->i_ino, dir->i_generation,
+ ent->oied_namelen, ent->oied_name, rc);
+ }
+
+ RETURN(rc);
+ }
+
+ dentry = osd_child_dentry_by_inode(env, dir, ent->oied_name,
+ ent->oied_namelen);
+ rc = osd_get_lma(info, inode, dentry, lma);
+ if (rc == -ENODATA)
+ lma = NULL;
+ else if (rc != 0)
+ GOTO(out, rc);
+
if (ent->oied_name[0] == '.') {
if (ent->oied_namelen == 1)
dot_dotdot = 1;
dot_dotdot = 2;
}
- dentry = osd_child_dentry_get(env, obj, ent->oied_name,
- ent->oied_namelen);
-
/* We need to ensure that the name entry is still valid.
* Because it may be removed or renamed by other already.
*
credits = osd_dto_credits_noquota[DTO_INDEX_DELETE] +
osd_dto_credits_noquota[DTO_INDEX_INSERT] + 1 + 1 + 2;
+ if (dev->od_dirent_journal != 0) {
+
again:
- if (dev->od_dirent_journal) {
jh = osd_journal_start_sb(sb, LDISKFS_HT_MISC, credits);
if (IS_ERR(jh)) {
rc = PTR_ERR(jh);
"name = %.*s: rc = %d\n",
devname, dir->i_ino, dir->i_generation, credits,
ent->oied_namelen, ent->oied_name, rc);
- RETURN(rc);
+
+ GOTO(out_inode, rc);
}
if (obj->oo_hl_head != NULL) {
* For the whole directory, only dot/dotdot entry have no FID-in-dirent
* and needs to get FID from LMA when readdir, it will not affect the
* performance much. */
- if ((bh == NULL) || (le32_to_cpu(de->inode) != ent->oied_ino) ||
+ if (IS_ERR(bh) || (le32_to_cpu(de->inode) != inode->i_ino) ||
(dot_dotdot != 0 && !osd_dot_dotdot_has_space(de, dot_dotdot))) {
*attr |= LUDA_IGNORE;
- GOTO(out_journal, rc = 0);
- }
-
- osd_id_gen(id, ent->oied_ino, OSD_OII_NOGEN);
- inode = osd_iget(info, dev, id);
- if (IS_ERR(inode)) {
- rc = PTR_ERR(inode);
- if (rc == -ENOENT || rc == -ESTALE)
- rc = 1;
- else
- CDEBUG(D_LFSCK, "%.16s: fail to iget for dirent "
- "check_repair, dir = %lu/%u, name = %.*s: "
- "rc = %d\n",
- devname, dir->i_ino, dir->i_generation,
- ent->oied_namelen, ent->oied_name, rc);
- GOTO(out_journal, rc);
+ GOTO(out, rc = 0);
}
- rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
- if (rc == 0) {
- LASSERT(!(lma->lma_compat & LMAC_NOT_IN_OI));
+ if (lma != NULL) {
+ if (unlikely(lma->lma_compat & LMAC_NOT_IN_OI)) {
+ struct lu_fid *tfid = &lma->lma_self_fid;
+
+ *attr |= LUDA_IGNORE;
+ /* It must be REMOTE_PARENT_DIR and as the
+ * dotdot entry of remote directory */
+ if (unlikely(dot_dotdot != 2 ||
+ fid_seq(tfid) != FID_SEQ_LOCAL_FILE ||
+ fid_oid(tfid) != REMOTE_PARENT_DIR_OID)) {
+ CDEBUG(D_LFSCK, "%.16s: expect remote agent "
+ "parent directory, but got %.*s under "
+ "dir = %lu/%u with the FID "DFID"\n",
+ devname, ent->oied_namelen,
+ ent->oied_name, dir->i_ino,
+ dir->i_generation, PFID(tfid));
+
+ GOTO(out, rc = -EIO);
+ }
+
+ GOTO(out, rc = 0);
+ }
if (fid_is_sane(fid)) {
/* FID-in-dirent is valid. */
if (lu_fid_eq(fid, &lma->lma_self_fid))
- GOTO(out_inode, rc = 0);
+ GOTO(out, rc = 0);
/* Do not repair under dryrun mode. */
if (*attr & LUDA_VERIFY_DRYRUN) {
*attr |= LUDA_REPAIR;
- GOTO(out_inode, rc = 0);
+
+ GOTO(out, rc = 0);
}
- if (!dev->od_dirent_journal) {
- iput(inode);
+ if (jh == NULL) {
brelse(bh);
- if (hlock != NULL)
+ dev->od_dirent_journal = 1;
+ if (hlock != NULL) {
ldiskfs_htree_unlock(hlock);
- else
+ hlock = NULL;
+ } else {
up_read(&obj->oo_ext_idx_sem);
- dev->od_dirent_journal = 1;
+ }
+
goto again;
}
*fid = lma->lma_self_fid;
dirty = true;
/* Update the FID-in-dirent. */
- rc = osd_dirent_update(jh, sb, ent, fid, bh, de);
+ rc = osd_dirent_reinsert(env, jh, dentry, fid, bh, de,
+ hlock, dot_dotdot);
if (rc == 0)
*attr |= LUDA_REPAIR;
else
if (*attr & LUDA_VERIFY_DRYRUN) {
*fid = lma->lma_self_fid;
*attr |= LUDA_REPAIR;
- GOTO(out_inode, rc = 0);
+
+ GOTO(out, rc = 0);
}
- if (!dev->od_dirent_journal) {
- iput(inode);
+ if (jh == NULL) {
brelse(bh);
- if (hlock != NULL)
+ dev->od_dirent_journal = 1;
+ if (hlock != NULL) {
ldiskfs_htree_unlock(hlock);
- else
+ hlock = NULL;
+ } else {
up_read(&obj->oo_ext_idx_sem);
- dev->od_dirent_journal = 1;
+ }
+
goto again;
}
*fid = lma->lma_self_fid;
dirty = true;
/* Append the FID-in-dirent. */
- rc = osd_dirent_reinsert(env, jh, dir, inode, ent,
- fid, bh, de, hlock);
+ rc = osd_dirent_reinsert(env, jh, dentry, fid, bh, de,
+ hlock, dot_dotdot);
if (rc == 0)
*attr |= LUDA_REPAIR;
else
ent->oied_namelen, ent->oied_name,
PFID(fid), rc);
}
- } else if (rc == -ENODATA) {
+ } else {
/* Do not repair under dryrun mode. */
if (*attr & LUDA_VERIFY_DRYRUN) {
if (fid_is_sane(fid)) {
inode->i_generation);
*attr |= LUDA_UPGRADE;
}
- GOTO(out_inode, rc = 0);
+
+ GOTO(out, rc = 0);
}
- if (!dev->od_dirent_journal) {
- iput(inode);
+ if (jh == NULL) {
brelse(bh);
- if (hlock != NULL)
+ dev->od_dirent_journal = 1;
+ if (hlock != NULL) {
ldiskfs_htree_unlock(hlock);
- else
+ hlock = NULL;
+ } else {
up_read(&obj->oo_ext_idx_sem);
- dev->od_dirent_journal = 1;
+ }
+
goto again;
}
lu_igif_build(fid, inode->i_ino, inode->i_generation);
/* It is probably IGIF object. Only aappend the
* FID-in-dirent. OI scrub will process FID-in-LMA. */
- rc = osd_dirent_reinsert(env, jh, dir, inode, ent,
- fid, bh, de, hlock);
+ rc = osd_dirent_reinsert(env, jh, dentry, fid, bh, de,
+ hlock, dot_dotdot);
if (rc == 0)
*attr |= LUDA_UPGRADE;
else
}
}
- GOTO(out_inode, rc);
-
-out_inode:
- iput(inode);
+ GOTO(out, rc);
-out_journal:
- brelse(bh);
+out:
+ if (!IS_ERR(bh))
+ brelse(bh);
if (hlock != NULL) {
ldiskfs_htree_unlock(hlock);
} else {
- if (dev->od_dirent_journal)
+ if (dev->od_dirent_journal != 0)
up_write(&obj->oo_ext_idx_sem);
else
up_read(&obj->oo_ext_idx_sem);
}
+
if (jh != NULL)
ldiskfs_journal_stop(jh);
+
+out_inode:
+ iput(inode);
if (rc >= 0 && !dirty)
dev->od_dirent_journal = 0;
+
return rc;
}
/* If the parent is on remote MDT, and there
* is no FID-in-dirent, then we have to get
* the parent FID from the linkEA. */
- osd_get_pfid_from_linkea(env, obj, fid);
+ if (!fid_is_sane(fid) &&
+ it->oie_dirent->oied_namelen == 2 &&
+ it->oie_dirent->oied_name[0] == '.' &&
+ it->oie_dirent->oied_name[1] == '.')
+ osd_get_pfid_from_linkea(env, obj, fid);
} else {
rc = osd_dirent_check_repair(env, obj, it, fid, id,
&attr);
}
+
+ if (!fid_is_sane(fid)) {
+ attr &= ~LUDA_IGNORE;
+ attr |= LUDA_UNKNOWN;
+ }
} else {
attr &= ~LU_DIRENT_ATTRS_MASK;
if (!fid_is_sane(fid)) {
- if (OBD_FAIL_CHECK(OBD_FAIL_FID_LOOKUP) &&
- likely(it->oie_dirent->oied_namelen != 2 ||
- it->oie_dirent->oied_name[0] != '.' ||
- it->oie_dirent->oied_name[1] != '.'))
- RETURN(-ENOENT);
-
- if (unlikely(ino == osd_remote_parent_ino(dev)))
- /* If the parent is on remote MDT, and there
- * is no FID-in-dirent, then we have to get
- * the parent FID from the linkEA. */
+ bool is_dotdot = false;
+ if (it->oie_dirent->oied_namelen == 2 &&
+ it->oie_dirent->oied_name[0] == '.' &&
+ it->oie_dirent->oied_name[1] == '.')
+ is_dotdot = true;
+ /* If the parent is on remote MDT, and there
+ * is no FID-in-dirent, then we have to get
+ * the parent FID from the linkEA. */
+ if (ino == osd_remote_parent_ino(dev) && is_dotdot) {
rc = osd_get_pfid_from_linkea(env, obj, fid);
- else
+ } else {
+ if (is_dotdot == false &&
+ OBD_FAIL_CHECK(OBD_FAIL_FID_LOOKUP))
+ RETURN(-ENOENT);
+
rc = osd_ea_fid_get(env, obj, ino, fid, id);
+ }
} else {
osd_id_gen(id, ino, OSD_OII_NOGEN);
}
if (osd_remote_fid(env, dev, fid))
RETURN(0);
- if (likely(!(attr & LUDA_IGNORE) && rc == 0))
+ if (likely(!(attr & (LUDA_IGNORE | LUDA_UNKNOWN)) && rc == 0))
osd_add_oi_cache(oti, dev, id, fid);
RETURN(rc > 0 ? 0 : rc);
* \retval -ve, on error
*/
static int osd_index_ea_lookup(const struct lu_env *env, struct dt_object *dt,
- struct dt_rec *rec, const struct dt_key *key,
- struct lustre_capa *capa)
+ struct dt_rec *rec, const struct dt_key *key)
{
struct osd_object *obj = osd_dt_obj(dt);
int rc = 0;
LASSERT(S_ISDIR(obj->oo_inode->i_mode));
LINVRNT(osd_invariant(obj));
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_LOOKUP))
- return -EACCES;
-
rc = osd_ea_lookup_rec(env, obj, rec, key);
if (rc == 0)
rc = +1;
struct lu_context_key *key, void* data)
{
struct osd_thread_info *info = data;
+ struct ldiskfs_inode_info *lli = LDISKFS_I(info->oti_inode);
+ struct osd_idmap_cache *idc = info->oti_ins_cache;
+ if (info->oti_inode != NULL)
+ OBD_FREE_PTR(lli);
if (info->oti_hlock != NULL)
ldiskfs_htree_lock_free(info->oti_hlock);
OBD_FREE(info->oti_it_ea_buf, OSD_IT_EA_BUFSIZE);
lu_buf_free(&info->oti_iobuf.dr_pg_buf);
lu_buf_free(&info->oti_iobuf.dr_bl_buf);
lu_buf_free(&info->oti_big_buf);
+ if (idc != NULL) {
+ LASSERT(info->oti_ins_cache_size > 0);
+ OBD_FREE(idc, sizeof(*idc) * info->oti_ins_cache_size);
+ info->oti_ins_cache = NULL;
+ info->oti_ins_cache_size = 0;
+ }
OBD_FREE_PTR(info);
}
rc = seq_client_init(osd->od_cl_seq, NULL, LUSTRE_SEQ_METADATA,
osd->od_svname, ss->ss_server_seq);
-
if (rc != 0) {
OBD_FREE_PTR(osd->od_cl_seq);
osd->od_cl_seq = NULL;
+ RETURN(rc);
+ }
+
+ if (ss->ss_node_id == 0) {
+ /* If the OSD on the sequence controller(MDT0), then allocate
+ * sequence here, otherwise allocate sequence after connected
+ * to MDT0 (see mdt_register_lwp_callback()). */
+ rc = seq_server_alloc_meta(osd->od_cl_seq->lcs_srv,
+ &osd->od_cl_seq->lcs_space, env);
}
RETURN(rc);
struct osd_thread_info *info = osd_oti_get(env);
struct lu_fid *fid = &info->oti_fid;
struct inode *inode;
- int rc = 0;
+ int rc = 0, force_over_256tb = 0;
ENTRY;
if (o->od_mnt != NULL)
RETURN(-E2BIG);
strcpy(o->od_mntdev, dev);
- OBD_PAGE_ALLOC(__page, GFP_IOFS);
- if (__page == NULL)
- GOTO(out, rc = -ENOMEM);
-
str = lustre_cfg_string(cfg, 2);
s_flags = simple_strtoul(str, NULL, 0);
str = strstr(str, ":");
if (str)
lmd_flags = simple_strtoul(str + 1, NULL, 0);
opts = lustre_cfg_string(cfg, 3);
+#ifdef __BIG_ENDIAN
+ if (opts == NULL || strstr(opts, "bigendian_extents") == NULL) {
+ CERROR("%s: device %s extents feature is not guaranteed to "
+ "work on big-endian systems. Use \"bigendian_extents\" "
+ "mount option to override.\n", name, dev);
+ RETURN(-EINVAL);
+ }
+#endif
+ if (opts != NULL && strstr(opts, "force_over_128tb") != NULL) {
+ CWARN("force_over_128tb option is depricated."
+ "Filesystems less then 256TB can be created without any"
+ "force options. Use force_over_256tb option for"
+ "filesystems greather then 256TB.\n");
+ }
+
+ if (opts != NULL && strstr(opts, "force_over_256tb") != NULL)
+ force_over_256tb = 1;
+
+ __page = alloc_page(GFP_KERNEL);
+ if (__page == NULL)
+ GOTO(out, rc = -ENOMEM);
page = (unsigned long)page_address(__page);
options = (char *)page;
*options = '\0';
- if (opts == NULL)
- strcat(options, "user_xattr,acl");
- else
+ if (opts != NULL) {
+ /* strip out the options for back compatiblity */
+ static char *sout[] = {
+ "mballoc",
+ "iopen",
+ "noiopen",
+ "iopen_nopriv",
+ "extents",
+ "noextents",
+ /* strip out option we processed in osd */
+ "bigendian_extents",
+#if LUSTRE_VERSION_CODE >= OBD_OCD_VERSION(3,0,53,0)
+#warning "remove force_over_128 option"
+#else
+ "force_over_128tb (deprecated)",
+#endif
+ "force_over_256tb",
+ NULL
+ };
strcat(options, opts);
+ for (rc = 0, str = options; sout[rc]; ) {
+ char *op = strstr(str, sout[rc]);
+ if (op == NULL) {
+ rc++;
+ str = options;
+ continue;
+ }
+ if (op == options || *(op - 1) == ',') {
+ str = op + strlen(sout[rc]);
+ if (*str == ',' || *str == '\0') {
+ *str == ',' ? str++ : str;
+ memmove(op, str, strlen(str) + 1);
+ }
+ }
+ for (str = op; *str != ',' && *str != '\0'; str++)
+ ;
+ }
+ } else {
+ strncat(options, "user_xattr,acl", 14);
+ }
/* Glom up mount options */
if (*options != '\0')
strcat(options, ",");
- strlcat(options, "no_mbcache", PAGE_CACHE_SIZE);
+ strlcat(options, "no_mbcache", PAGE_SIZE);
type = get_fs_type("ldiskfs");
if (!type) {
GOTO(out, rc);
}
+ if (ldiskfs_blocks_count(LDISKFS_SB(osd_sb(o))->s_es) > (64ULL << 30) &&
+ force_over_256tb == 0) {
+ CERROR("%s: device %s LDISKFS does not support filesystems "
+ "greater than 256TB and can cause data corruption. "
+ "Use \"force_over_256tb\" mount option to override.\n",
+ name, dev);
+ GOTO(out, rc = -EINVAL);
+ }
+
#ifdef HAVE_DEV_SET_RDONLY
if (dev_check_rdonly(o->od_mnt->mnt_sb->s_bdev)) {
CERROR("%s: underlying device %s is marked as read-only. "
out:
if (__page)
- OBD_PAGE_FREE(__page);
+ __free_page(__page);
return rc;
}
spin_lock_init(&o->od_osfs_lock);
mutex_init(&o->od_otable_mutex);
-
- o->od_capa_hash = init_capa_hash();
- if (o->od_capa_hash == NULL)
- GOTO(out, rc = -ENOMEM);
+ INIT_LIST_HEAD(&o->od_orphan_list);
o->od_read_cache = 1;
o->od_writethrough_cache = 1;
sizeof(o->od_svname));
if (cplen >= sizeof(o->od_svname)) {
rc = -E2BIG;
- GOTO(out_capa, rc);
+ GOTO(out, rc);
}
- if (server_name_is_ost(o->od_svname))
+ o->od_index = -1; /* -1 means index is invalid */
+ rc = server_name2index(o->od_svname, &o->od_index, NULL);
+ if (rc == LDD_F_SV_TYPE_OST)
o->od_is_ost = 1;
o->od_full_scrub_ratio = OFSR_DEFAULT;
- o->od_full_scrub_speed = FULL_SCRUB_SPEED_DEFULT;
+ o->od_full_scrub_threshold_rate = FULL_SCRUB_THRESHOLD_RATE_DEFAULT;
rc = osd_mount(env, o, cfg);
if (rc != 0)
- GOTO(out_capa, rc);
+ GOTO(out, rc);
rc = osd_obj_map_init(env, o);
if (rc != 0)
if (rc != 0)
GOTO(out_site, rc);
- /* self-repair LMA by default */
- o->od_lma_self_repair = 1;
-
INIT_LIST_HEAD(&o->od_ios_list);
/* setup scrub, including OI files initialization */
rc = osd_scrub_setup(env, o);
osd_obj_map_fini(o);
out_mnt:
osd_umount(env, o);
-out_capa:
- cleanup_capa_hash(o->od_capa_hash);
out:
return rc;
}
struct osd_device *o = osd_dev(d);
ENTRY;
- cleanup_capa_hash(o->od_capa_hash);
/* XXX: make osd top device in order to release reference */
d->ld_site->ls_top_dev = d;
lu_site_purge(env, d->ld_site, -1);
}
static int osd_prepare(const struct lu_env *env, struct lu_device *pdev,
- struct lu_device *dev)
+ struct lu_device *dev)
{
- struct osd_device *osd = osd_dev(dev);
- int result = 0;
+ struct osd_device *osd = osd_dev(dev);
+ struct lr_server_data *lsd =
+ &osd->od_dt_dev.dd_lu_dev.ld_site->ls_tgt->lut_lsd;
+ int result = 0;
ENTRY;
if (osd->od_quota_slave != NULL) {
RETURN(result);
}
+ if (lsd->lsd_feature_incompat & OBD_COMPAT_OST) {
+#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(3, 0, 52, 0)
+ if (lsd->lsd_feature_rocompat & OBD_ROCOMPAT_IDX_IN_IDIF) {
+ osd->od_index_in_idif = 1;
+ } else {
+ osd->od_index_in_idif = 0;
+ result = osd_register_proc_index_in_idif(osd);
+ if (result != 0)
+ RETURN(result);
+ }
+#else
+ osd->od_index_in_idif = 1;
+#endif
+ }
+
result = osd_fid_init(env, osd);
RETURN(result);
.ldt_ctx_tags = LCT_LOCAL,
};
+static int osd_health_check(const struct lu_env *env, struct obd_device *obd)
+{
+ struct osd_device *osd = osd_dev(obd->obd_lu_dev);
+ struct super_block *sb = osd_sb(osd);
+
+ return (osd->od_mnt == NULL || sb->s_flags & MS_RDONLY);
+}
+
/*
* lprocfs legacy support.
*/
.o_connect = osd_obd_connect,
.o_disconnect = osd_obd_disconnect,
.o_fid_alloc = osd_fid_alloc,
+ .o_health_check = osd_health_check,
};
-static int __init osd_mod_init(void)
+static int __init osd_init(void)
{
int rc;
+ LASSERT(BH_DXLock < sizeof(((struct buffer_head *)0)->b_state) * 8);
+#if !defined(CONFIG_DEBUG_MUTEXES) && !defined(CONFIG_DEBUG_SPINLOCK)
+ /* please, try to keep osd_thread_info smaller than a page */
+ CLASSERT(sizeof(struct osd_thread_info) <= PAGE_SIZE);
+#endif
+
osd_oi_mod_init();
rc = lu_kmem_init(ldiskfs_caches);
return rc;
}
-static void __exit osd_mod_exit(void)
+static void __exit osd_exit(void)
{
class_unregister_type(LUSTRE_OSD_LDISKFS_NAME);
lu_kmem_fini(ldiskfs_caches);
}
-MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
+MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
MODULE_DESCRIPTION("Lustre Object Storage Device ("LUSTRE_OSD_LDISKFS_NAME")");
+MODULE_VERSION(LUSTRE_VERSION_STRING);
MODULE_LICENSE("GPL");
-cfs_module(osd, "0.1.0", osd_mod_init, osd_mod_exit);
+module_init(osd_init);
+module_exit(osd_exit);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff