* 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, 2014, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
* Pravin Shelar <pravin.shelar@sun.com> : Added fid in dirent
*/
-#define DEBUG_SUBSYSTEM S_MDS
+#define DEBUG_SUBSYSTEM S_OSD
#include <linux/module.h>
/* struct ptlrpc_thread */
#include <lustre_net.h>
#include <lustre_fid.h>
+/* process_config */
+#include <lustre_param.h>
#include "osd_internal.h"
#include "osd_dynlocks.h"
#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");
/* 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
}
};
/*
* Concurrency: doesn't matter
*/
-static int osd_read_locked(const struct lu_env *env, struct osd_object *o)
-{
- return osd_oti_get(env)->oti_r_locks > 0;
-}
/*
* Concurrency: doesn't matter
id->oii_ino, PTR_ERR(inode));
} else if (id->oii_gen != OSD_OII_NOGEN &&
inode->i_generation != id->oii_gen) {
- CDEBUG(D_INODE, "unmatched inode: ino = %u, gen0 = %u, "
- "gen1 = %u\n",
+ CDEBUG(D_INODE, "unmatched inode: ino = %u, oii_gen = %u, "
+ "i_generation = %u\n",
id->oii_ino, id->oii_gen, inode->i_generation);
iput(inode);
inode = ERR_PTR(-ESTALE);
/* due to parallel readdir and unlink,
* we can have dead inode here. */
CDEBUG(D_INODE, "stale inode: ino = %u\n", id->oii_ino);
- make_bad_inode(inode);
iput(inode);
inode = ERR_PTR(-ESTALE);
} else if (is_bad_inode(inode)) {
return inode;
}
+static struct inode *osd_iget_check(struct osd_thread_info *info,
+ struct osd_device *dev,
+ const struct lu_fid *fid,
+ struct osd_inode_id *id,
+ bool in_oi)
+{
+ struct inode *inode;
+ int rc = 0;
+ ENTRY;
+
+ inode = ldiskfs_iget(osd_sb(dev), id->oii_ino);
+ if (IS_ERR(inode)) {
+ rc = PTR_ERR(inode);
+ if (!in_oi || (rc != -ENOENT && rc != -ESTALE)) {
+ CDEBUG(D_INODE, "no inode: ino = %u, rc = %d\n",
+ id->oii_ino, rc);
+
+ GOTO(put, rc);
+ }
+
+ goto check_oi;
+ }
+
+ if (is_bad_inode(inode)) {
+ rc = -ENOENT;
+ if (!in_oi) {
+ CDEBUG(D_INODE, "bad inode: ino = %u\n", id->oii_ino);
+
+ GOTO(put, rc);
+ }
+
+ goto check_oi;
+ }
+
+ if (id->oii_gen != OSD_OII_NOGEN &&
+ inode->i_generation != id->oii_gen) {
+ rc = -ESTALE;
+ if (!in_oi) {
+ CDEBUG(D_INODE, "unmatched inode: ino = %u, "
+ "oii_gen = %u, i_generation = %u\n",
+ id->oii_ino, id->oii_gen, inode->i_generation);
+
+ GOTO(put, rc);
+ }
+
+ goto check_oi;
+ }
+
+ if (inode->i_nlink == 0) {
+ rc = -ENOENT;
+ if (!in_oi) {
+ CDEBUG(D_INODE, "stale inode: ino = %u\n", id->oii_ino);
+
+ GOTO(put, rc);
+ }
+
+ goto check_oi;
+ }
+
+check_oi:
+ if (rc != 0) {
+ struct osd_inode_id saved_id = *id;
+
+ 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:
+ * 1. rc = 0.
+ * Backup/restore caused the OI invalid.
+ * 2. rc = 0.
+ * Someone unlinked the object but NOT removed
+ * the OI mapping, such as mount target device
+ * as ldiskfs, and modify something directly.
+ * 3. rc = -ENOENT.
+ * Someone just removed the object between the
+ * former oi_lookup and the iget. It is normal.
+ * 4. Other failure cases.
+ *
+ * Generally, when the device is mounted, it will
+ * auto check whether the system is restored from
+ * file-level backup or not. We trust such detect
+ * to distinguish the 1st case from the 2nd case. */
+ if (rc == 0) {
+ if (!IS_ERR(inode) && inode->i_generation != 0 &&
+ inode->i_generation == id->oii_gen) {
+ rc = -ENOENT;
+ } else {
+ __u32 level = D_LFSCK;
+
+ rc = -EREMCHG;
+ if (!thread_is_running(&dev->od_scrub.os_thread))
+ level |= D_CONSOLE;
+
+ CDEBUG(level, "%s: the OI mapping for the FID "
+ DFID" become inconsistent, the given ID "
+ "%u/%u, the ID in OI mapping %u/%u\n",
+ osd_name(dev), PFID(fid),
+ saved_id.oii_ino, saved_id.oii_gen,
+ id->oii_ino, id->oii_ino);
+ }
+ }
+ } else {
+ if (id->oii_gen == OSD_OII_NOGEN)
+ osd_id_gen(id, inode->i_ino, inode->i_generation);
+
+ /* Do not update file c/mtime in ldiskfs.
+ * NB: we don't have any lock to protect this because we don't
+ * have reference on osd_object now, but contention with
+ * another lookup + attr_set can't happen in the tiny window
+ * between if (...) and set S_NOCMTIME. */
+ if (!(inode->i_flags & S_NOCMTIME))
+ inode->i_flags |= S_NOCMTIME;
+ }
+
+ GOTO(put, rc);
+
+put:
+ if (rc != 0) {
+ if (!IS_ERR(inode))
+ iput(inode);
+
+ inode = ERR_PTR(rc);
+ }
+
+ return inode;
+}
+
/**
* \retval +v: new filter_fid, does not contain self-fid
* \retval 0: filter_fid_old, contains self-fid
return rc;
}
+static int osd_lma_self_repair(struct osd_thread_info *info,
+ struct osd_device *osd, struct inode *inode,
+ const struct lu_fid *fid, __u32 compat)
+{
+ handle_t *jh;
+ int rc;
+
+ LASSERT(current->journal_info == NULL);
+
+ jh = osd_journal_start_sb(osd_sb(osd), LDISKFS_HT_MISC,
+ osd_dto_credits_noquota[DTO_XATTR_SET]);
+ if (IS_ERR(jh)) {
+ rc = PTR_ERR(jh);
+ CWARN("%s: cannot start journal for lma_self_repair: rc = %d\n",
+ osd_name(osd), rc);
+ return rc;
+ }
+
+ rc = osd_ea_fid_set(info, inode, fid, compat, 0);
+ if (rc != 0)
+ CWARN("%s: cannot self repair the LMA: rc = %d\n",
+ osd_name(osd), rc);
+ ldiskfs_journal_stop(jh);
+ return rc;
+}
+
static int osd_check_lma(const struct lu_env *env, struct osd_object *obj)
{
struct osd_thread_info *info = osd_oti_get(env);
+ struct osd_device *osd = osd_obj2dev(obj);
struct lustre_mdt_attrs *lma = &info->oti_mdt_attrs;
struct inode *inode = obj->oo_inode;
struct dentry *dentry = &info->oti_obj_dentry;
struct lu_fid *fid = NULL;
+ const struct lu_fid *rfid = lu_object_fid(&obj->oo_dt.do_lu);
int rc;
ENTRY;
- if (OBD_FAIL_CHECK(OBD_FAIL_OSD_COMPAT_INVALID_ENTRY))
- RETURN(0);
-
CLASSERT(LMA_OLD_SIZE >= sizeof(*lma));
rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LMA,
info->oti_mdt_attrs_old, LMA_OLD_SIZE);
- if (rc == -ENODATA && !fid_is_igif(lu_object_fid(&obj->oo_dt.do_lu)) &&
- osd_obj2dev(obj)->od_check_ff) {
+ 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)
+ 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,
+ LMAC_FID_ON_OST);
RETURN(0);
+ }
}
if (unlikely(rc == -ENODATA))
if (unlikely((lma->lma_incompat & ~LMA_INCOMPAT_SUPP) ||
CFS_FAIL_CHECK(OBD_FAIL_OSD_LMA_INCOMPAT))) {
CWARN("%s: unsupported incompat LMA feature(s) %#x for "
- "fid = "DFID", ino = %lu\n",
- osd_obj2dev(obj)->od_svname,
+ "fid = "DFID", ino = %lu\n", osd_name(osd),
lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
- PFID(lu_object_fid(&obj->oo_dt.do_lu)),
- inode->i_ino);
+ PFID(rfid), inode->i_ino);
rc = -EOPNOTSUPP;
} else if (!(lma->lma_compat & LMAC_NOT_IN_OI)) {
fid = &lma->lma_self_fid;
}
}
- if (fid != NULL &&
- unlikely(!lu_fid_eq(lu_object_fid(&obj->oo_dt.do_lu), fid))) {
- CDEBUG(D_INODE, "%s: FID "DFID" != self_fid "DFID"\n",
- osd_obj2dev(obj)->od_svname,
- PFID(lu_object_fid(&obj->oo_dt.do_lu)),
- PFID(&lma->lma_self_fid));
+ if (fid != NULL && unlikely(!lu_fid_eq(rfid, fid))) {
+ __u32 level = D_LFSCK;
+
+ if (fid_is_idif(rfid) && fid_is_idif(fid)) {
+ struct ost_id *oi = &info->oti_ostid;
+ struct lu_fid *fid1 = &info->oti_fid3;
+ __u32 idx = fid_idif_ost_idx(rfid);
+
+ /* For old IDIF, the OST index is not part of the IDIF,
+ * Means that different OSTs may have the same IDIFs.
+ * Under such case, we need to make some compatible
+ * check to make sure to trigger OI scrub properly. */
+ if (idx != 0 && fid_idif_ost_idx(fid) == 0) {
+ /* Given @rfid is new, LMA is old. */
+ fid_to_ostid(fid, oi);
+ ostid_to_fid(fid1, oi, idx);
+ if (lu_fid_eq(fid1, rfid)) {
+ if (osd->od_index_in_idif)
+ osd_lma_self_repair(info, osd,
+ inode, rfid,
+ LMAC_FID_ON_OST);
+ RETURN(0);
+ }
+ }
+ }
+
+
rc = -EREMCHG;
+ if (!thread_is_running(&osd->od_scrub.os_thread))
+ level |= D_CONSOLE;
+
+ CDEBUG(level, "%s: FID "DFID" != self_fid "DFID"\n",
+ osd_name(osd), PFID(rfid), PFID(fid));
}
RETURN(rc);
LINVRNT(osd_invariant(obj));
LASSERT(obj->oo_inode == NULL);
- LASSERTF(fid_is_sane(fid) || fid_is_idif(fid), DFID, PFID(fid));
+ LASSERTF(fid_is_sane(fid) || fid_is_idif(fid), DFID"\n", PFID(fid));
dev = osd_dev(ldev);
scrub = &dev->od_scrub;
if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT))
RETURN(-ENOENT);
+ /* For the object is created as locking anchor, or for the object to
+ * be created on disk. No need to osd_oi_lookup() at here because FID
+ * shouldn't never be re-used, if it's really a duplicate FID from
+ * unexpected reason, we should be able to detect it later by calling
+ * do_create->osd_oi_insert(). */
+ if (conf != NULL && conf->loc_flags & LOC_F_NEW)
+ GOTO(out, result = 0);
+
/* Search order: 1. per-thread cache. */
if (lu_fid_eq(fid, &oic->oic_fid) &&
likely(oic->oic_dev == dev)) {
}
id = &info->oti_id;
- if (!cfs_list_empty(&scrub->os_inconsistent_items)) {
+ if (!list_empty(&scrub->os_inconsistent_items)) {
/* Search order: 2. OI scrub pending list. */
result = osd_oii_lookup(dev, fid, id);
if (result == 0)
goto iget;
}
- /*
- * Objects are created as locking anchors or place holders for objects
- * yet to be created. No need to osd_oi_lookup() at here because FID
- * shouldn't never be re-used, if it's really a duplicate FID from
- * unexpected reason, we should be able to detect it later by calling
- * do_create->osd_oi_insert()
- */
- if (conf != NULL && conf->loc_flags & LOC_F_NEW)
- GOTO(out, result = 0);
-
/* Search order: 3. OI files. */
result = osd_oi_lookup(info, dev, fid, id, OI_CHECK_FLD);
if (result == -ENOENT) {
in_oi = true;
iget:
- inode = osd_iget(info, dev, id);
+ inode = osd_iget_check(info, dev, fid, id, in_oi);
if (IS_ERR(inode)) {
result = PTR_ERR(inode);
if (result == -ENOENT || result == -ESTALE) {
- if (!in_oi) {
+ if (!in_oi)
fid_zero(&oic->oic_fid);
- GOTO(out, result = 0);
- }
- /* XXX: There are three possible cases:
- * 1. Backup/restore caused the OI invalid.
- * 2. Someone unlinked the object but NOT removed
- * the OI mapping, such as mount target device
- * as ldiskfs, and modify something directly.
- * 3. Someone just removed the object between the
- * former oi_lookup and the iget. It is normal.
- *
- * It is diffcult to distinguish the 2nd from the
- * 1st case. Relatively speaking, the 1st case is
- * common than the 2nd case, trigger OI scrub. */
- result = osd_oi_lookup(info, dev, fid, id, true);
- if (result == 0)
- /* It is the case 1 or 2. */
- goto trigger;
-
- if (result == -ENOENT)
- /* It is the case 3. */
- result = 0;
+ GOTO(out, result = -ENOENT);
} else if (result == -EREMCHG) {
trigger:
+ if (!in_oi)
+ fid_zero(&oic->oic_fid);
+
if (unlikely(triggered))
GOTO(out, result = saved);
if (thread_is_running(&scrub->os_thread)) {
result = -EINPROGRESS;
} else if (!dev->od_noscrub) {
- result = osd_scrub_start(dev);
- LCONSOLE_ERROR("%.16s: trigger OI scrub by RPC "
- "for "DFID", rc = %d [1]\n",
- LDISKFS_SB(osd_sb(dev))->s_es->\
- s_volume_name,PFID(fid), result);
+ /* 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);
+ CDEBUG(D_LFSCK | D_CONSOLE, "%.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
if (result != 0) {
iput(inode);
obj->oo_inode = NULL;
- if (result == -EREMCHG)
+ if (result == -EREMCHG) {
+ if (!in_oi) {
+ result = osd_oi_lookup(info, dev, fid, id,
+ OI_CHECK_FLD);
+ if (result != 0) {
+ fid_zero(&oic->oic_fid);
+ GOTO(out, result);
+ }
+ }
+
goto trigger;
+ }
GOTO(out, result);
}
{
struct osd_thandle *oh = container_of(th, typeof(*oh), ot_super);
- return oh->ot_credits > osd_journal(dev)->j_max_transaction_buffers;
+ return oh->ot_credits > osd_transaction_size(dev);
}
/*
dt_txn_hook_commit(th);
/* call per-transaction callbacks if any */
- cfs_list_for_each_entry_safe(dcb, tmp, &oh->ot_dcb_list, dcb_linkage) {
+ list_for_each_entry_safe(dcb, tmp, &oh->ot_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);
- cfs_list_del_init(&dcb->dcb_linkage);
+ list_del_init(&dcb->dcb_linkage);
dcb->dcb_func(NULL, th, dcb, error);
}
lu_context_exit(&th->th_ctx);
lu_context_fini(&th->th_ctx);
- OBD_FREE_PTR(oh);
+ thandle_put(th);
}
static struct thandle *osd_trans_create(const struct lu_env *env,
- struct dt_device *d)
+ struct dt_device *d)
{
- struct osd_thread_info *oti = osd_oti_get(env);
- struct osd_iobuf *iobuf = &oti->oti_iobuf;
- struct osd_thandle *oh;
- struct thandle *th;
- ENTRY;
+ struct osd_thread_info *oti = osd_oti_get(env);
+ struct osd_iobuf *iobuf = &oti->oti_iobuf;
+ struct osd_thandle *oh;
+ struct thandle *th;
+ ENTRY;
- /* on pending IO in this thread should left from prev. request */
- LASSERT(cfs_atomic_read(&iobuf->dr_numreqs) == 0);
+ /* on pending IO in this thread should left from prev. request */
+ LASSERT(atomic_read(&iobuf->dr_numreqs) == 0);
- th = ERR_PTR(-ENOMEM);
- OBD_ALLOC_GFP(oh, sizeof *oh, __GFP_IO);
- if (oh != NULL) {
+ th = ERR_PTR(-ENOMEM);
+ OBD_ALLOC_GFP(oh, sizeof *oh, GFP_NOFS);
+ if (oh != NULL) {
oh->ot_quota_trans = &oti->oti_quota_trans;
memset(oh->ot_quota_trans, 0, sizeof(*oh->ot_quota_trans));
- th = &oh->ot_super;
- th->th_dev = d;
- th->th_result = 0;
- th->th_tags = LCT_TX_HANDLE;
- oh->ot_credits = 0;
- oti->oti_dev = osd_dt_dev(d);
- CFS_INIT_LIST_HEAD(&oh->ot_dcb_list);
- osd_th_alloced(oh);
+ th = &oh->ot_super;
+ th->th_dev = d;
+ 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);
+ osd_th_alloced(oh);
memset(oti->oti_declare_ops, 0,
sizeof(oti->oti_declare_ops));
memset(oti->oti_declare_ops_cred, 0,
sizeof(oti->oti_declare_ops_cred));
oti->oti_rollback = false;
- }
- RETURN(th);
+ }
+ RETURN(th);
}
/*
* Concurrency: shouldn't matter.
*/
-int osd_trans_start(const struct lu_env *env, struct dt_device *d,
- struct thandle *th)
+static int osd_trans_start(const struct lu_env *env, struct dt_device *d,
+ struct thandle *th)
{
struct osd_thread_info *oti = osd_oti_get(env);
struct osd_device *dev = osd_dt_dev(d);
LDISKFS_SB(osd_sb(dev))->s_es->s_volume_name,
oh->ot_credits,
osd_journal(dev)->j_max_transaction_buffers);
- CWARN(" create: %u/%u, delete: %u/%u, destroy: %u/%u\n",
+ 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_DELETE],
- oti->oti_declare_ops_cred[OSD_OT_DELETE],
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",
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_DESTROY],
- oti->oti_declare_ops_cred[OSD_OT_DESTROY]);
+ 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_cred[OSD_OT_REF_DEL]);
if (last_credits != oh->ot_credits &&
- time_after(jiffies, last_printed + 60 * HZ)) {
+ time_after(jiffies, last_printed +
+ msecs_to_jiffies(60 * MSEC_PER_SEC))) {
libcfs_debug_dumpstack(NULL);
last_credits = oh->ot_credits;
last_printed = jiffies;
*
* This should be removed when we can calculate the
* credits precisely. */
- oh->ot_credits = osd_journal(dev)->j_max_transaction_buffers;
+ oh->ot_credits = osd_transaction_size(dev);
}
/*
* XXX temporary stuff. Some abstraction layer should
* be used.
*/
- jh = ldiskfs_journal_start_sb(osd_sb(dev), oh->ot_credits);
+ jh = osd_journal_start_sb(osd_sb(dev), LDISKFS_HT_MISC, oh->ot_credits);
osd_th_started(oh);
if (!IS_ERR(jh)) {
oh->ot_handle = jh;
RETURN(rc);
}
+static int osd_seq_exists(const struct lu_env *env,
+ struct osd_device *osd, u64 seq)
+{
+ struct lu_seq_range *range = &osd_oti_get(env)->oti_seq_range;
+ struct seq_server_site *ss = osd_seq_site(osd);
+ int rc;
+ ENTRY;
+
+ LASSERT(ss != NULL);
+ LASSERT(ss->ss_server_fld != NULL);
+
+ rc = osd_fld_lookup(env, osd, seq, range);
+ if (rc != 0) {
+ if (rc != -ENOENT)
+ CERROR("%s: can't lookup FLD sequence "LPX64
+ ": rc = %d\n", osd_name(osd), seq, rc);
+ RETURN(0);
+ }
+
+ RETURN(ss->ss_node_id == range->lsr_index);
+}
+
/*
* Concurrency: shouldn't matter.
*/
-static int osd_trans_stop(const struct lu_env *env, struct thandle *th)
+static int osd_trans_stop(const struct lu_env *env, struct dt_device *dt,
+ struct thandle *th)
{
- int rc = 0;
- struct osd_thandle *oh;
- struct osd_thread_info *oti = osd_oti_get(env);
- struct osd_iobuf *iobuf = &oti->oti_iobuf;
+ int rc = 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;
- ENTRY;
+ struct lquota_trans *qtrans;
+ ENTRY;
- oh = container_of0(th, struct osd_thandle, ot_super);
+ oh = container_of0(th, struct osd_thandle, ot_super);
- if (qsd != NULL)
- /* inform the quota slave device that the transaction is
- * stopping */
- qsd_op_end(env, qsd, oh->ot_quota_trans);
+ qtrans = oh->ot_quota_trans;
oh->ot_quota_trans = NULL;
if (oh->ot_handle != NULL) {
if (rc != 0)
CERROR("Failure to stop transaction: %d\n", rc);
} else {
- OBD_FREE_PTR(oh);
+ thandle_put(&oh->ot_super);
}
+ /* inform the quota slave device that the transaction is stopping */
+ qsd_op_end(env, qsd, qtrans);
+
/* as we want IO to journal and data IO be concurrent, we don't block
* awaiting data IO completion in osd_do_bio(), instead we wait here
* once transaction is submitted to the journal. all reqular requests
* completed otherwise iobuf may be corrupted by different request
*/
wait_event(iobuf->dr_wait,
- cfs_atomic_read(&iobuf->dr_numreqs) == 0);
+ atomic_read(&iobuf->dr_numreqs) == 0);
+ osd_fini_iobuf(oti->oti_dev, iobuf);
if (!rc)
rc = iobuf->dr_error;
LASSERT(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC);
LASSERT(&dcb->dcb_func != NULL);
- cfs_list_add(&dcb->dcb_linkage, &oh->ot_dcb_list);
+ list_add(&dcb->dcb_linkage, &oh->ot_dcb_list);
return 0;
}
osd_index_fini(obj);
if (inode != NULL) {
struct qsd_instance *qsd = osd_obj2dev(obj)->od_quota_slave;
- qid_t uid = inode->i_uid;
- qid_t gid = inode->i_gid;
+ qid_t uid = i_uid_read(inode);
+ qid_t gid = i_gid_read(inode);
iput(inode);
obj->oo_inode = NULL;
d ? d->id_ops->id_name : "plain");
}
+#define GRANT_FOR_LOCAL_OIDS 32 /* 128kB for last_rcvd, quota files, ... */
+
/*
* Concurrency: shouldn't matter.
*/
}
spin_lock(&osd->od_osfs_lock);
- /* cache 1 second */
- if (cfs_time_before_64(osd->od_osfs_age, cfs_time_shift_64(-1))) {
- result = sb->s_op->statfs(sb->s_root, ksfs);
- if (likely(result == 0)) { /* N.B. statfs can't really fail */
- osd->od_osfs_age = cfs_time_current_64();
- statfs_pack(&osd->od_statfs, ksfs);
- if (sb->s_flags & MS_RDONLY)
- sfs->os_state = OS_STATE_READONLY;
- }
+ 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;
}
- if (likely(result == 0))
- *sfs = osd->od_statfs;
spin_unlock(&osd->od_osfs_lock);
- if (unlikely(env == NULL))
+ if (unlikely(env == NULL))
OBD_FREE_PTR(ksfs);
+ /* 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);
+ }
+
return result;
}
struct dt_device_param *param)
{
struct super_block *sb = osd_sb(osd_dt_dev(dev));
+ int ea_overhead;
/*
* XXX should be taken from not-yet-existing fs abstraction layer.
*/
- param->ddp_mnt = osd_dt_dev(dev)->od_mnt;
param->ddp_max_name_len = LDISKFS_NAME_LEN;
param->ddp_max_nlink = LDISKFS_LINK_MAX;
param->ddp_block_shift = sb->s_blocksize_bits;
if (test_opt(sb, POSIX_ACL))
param->ddp_mntopts |= MNTOPT_ACL;
+ /* LOD might calculate the max stripe count based on max_ea_size,
+ * so we need take account in the overhead as well,
+ * xattr_header + magic + xattr_entry_head */
+ ea_overhead = sizeof(struct ldiskfs_xattr_header) + sizeof(__u32) +
+ LDISKFS_XATTR_LEN(XATTR_NAME_MAX_LEN);
+
#if defined(LDISKFS_FEATURE_INCOMPAT_EA_INODE)
- if (LDISKFS_HAS_INCOMPAT_FEATURE(sb, LDISKFS_FEATURE_INCOMPAT_EA_INODE))
- param->ddp_max_ea_size = LDISKFS_XATTR_MAX_LARGE_EA_SIZE;
- else
+ if (LDISKFS_HAS_INCOMPAT_FEATURE(sb, LDISKFS_FEATURE_INCOMPAT_EA_INODE))
+ param->ddp_max_ea_size = LDISKFS_XATTR_MAX_LARGE_EA_SIZE -
+ ea_overhead;
+ else
#endif
- param->ddp_max_ea_size = sb->s_blocksize;
-
+ param->ddp_max_ea_size = sb->s_blocksize - ea_overhead;
}
/*
*/
static int osd_sync(const struct lu_env *env, struct dt_device *d)
{
- CDEBUG(D_HA, "syncing OSD %s\n", LUSTRE_OSD_LDISKFS_NAME);
- return ldiskfs_force_commit(osd_sb(osd_dt_dev(d)));
+ int rc;
+
+ CDEBUG(D_CACHE, "syncing OSD %s\n", LUSTRE_OSD_LDISKFS_NAME);
+
+ rc = ldiskfs_force_commit(osd_sb(osd_dt_dev(d)));
+
+ CDEBUG(D_CACHE, "synced OSD %s: rc = %d\n",
+ LUSTRE_OSD_LDISKFS_NAME, rc);
+
+ return rc;
}
/**
* If we mount with --data_journal we may need more.
*/
const int osd_dto_credits_noquota[DTO_NR] = {
- /**
- * Insert/Delete.
- * INDEX_EXTRA_TRANS_BLOCKS(8) +
- * SINGLEDATA_TRANS_BLOCKS(8)
- * XXX Note: maybe iam need more, since iam have more level than
- * EXT3 htree.
- */
- [DTO_INDEX_INSERT] = 16,
- [DTO_INDEX_DELETE] = 16,
- /**
+ /**
+ * Insert.
+ * INDEX_EXTRA_TRANS_BLOCKS(8) +
+ * SINGLEDATA_TRANS_BLOCKS(8)
+ * XXX Note: maybe iam need more, since iam have more level than
+ * EXT3 htree.
+ */
+ [DTO_INDEX_INSERT] = 16,
+ /**
+ * Delete
+ * just modify a single entry, probably merge few within a block
+ */
+ [DTO_INDEX_DELETE] = 1,
+ /**
* Used for OI scrub
- */
- [DTO_INDEX_UPDATE] = 16,
- /**
- * Create a object. The same as create object in EXT3.
- * DATA_TRANS_BLOCKS(14) +
- * INDEX_EXTRA_BLOCKS(8) +
- * 3(inode bits, groups, GDT)
- */
- [DTO_OBJECT_CREATE] = 25,
- /**
- * XXX: real credits to be fixed
- */
- [DTO_OBJECT_DELETE] = 25,
- /**
- * Attr set credits (inode)
- */
- [DTO_ATTR_SET_BASE] = 1,
- /**
- * Xattr set. The same as xattr of EXT3.
- * DATA_TRANS_BLOCKS(14)
- * XXX Note: in original MDS implmentation INDEX_EXTRA_TRANS_BLOCKS
- * are also counted in. Do not know why?
- */
- [DTO_XATTR_SET] = 14,
- [DTO_LOG_REC] = 14,
- /**
- * credits for inode change during write.
- */
- [DTO_WRITE_BASE] = 3,
- /**
- * credits for single block write.
- */
- [DTO_WRITE_BLOCK] = 14,
- /**
- * Attr set credits for chown.
- * This is extra credits for setattr, and it is null without quota
- */
- [DTO_ATTR_SET_CHOWN]= 0
+ */
+ [DTO_INDEX_UPDATE] = 16,
+ /**
+ * 4(inode, inode bits, groups, GDT)
+ * notice: OI updates are counted separately with DTO_INDEX_INSERT
+ */
+ [DTO_OBJECT_CREATE] = 4,
+ /**
+ * 4(inode, inode bits, groups, GDT)
+ * notice: OI updates are counted separately with DTO_INDEX_DELETE
+ */
+ [DTO_OBJECT_DELETE] = 4,
+ /**
+ * Attr set credits (inode)
+ */
+ [DTO_ATTR_SET_BASE] = 1,
+ /**
+ * Xattr set. The same as xattr of EXT3.
+ * DATA_TRANS_BLOCKS(14)
+ * XXX Note: in original MDS implmentation INDEX_EXTRA_TRANS_BLOCKS
+ * are also counted in. Do not know why?
+ */
+ [DTO_XATTR_SET] = 14,
+ /**
+ * credits for inode change during write.
+ */
+ [DTO_WRITE_BASE] = 3,
+ /**
+ * credits for single block write.
+ */
+ [DTO_WRITE_BLOCK] = 14,
+ /**
+ * Attr set credits for chown.
+ * This is extra credits for setattr, and it is null without quota
+ */
+ [DTO_ATTR_SET_CHOWN] = 0
};
static const struct dt_device_operations osd_dt_ops = {
static void osd_inode_getattr(const struct lu_env *env,
- struct inode *inode, struct lu_attr *attr)
-{
- attr->la_valid |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE |
- LA_SIZE | LA_BLOCKS | LA_UID | LA_GID |
- LA_FLAGS | LA_NLINK | LA_RDEV | LA_BLKSIZE |
- LA_TYPE;
-
- attr->la_atime = LTIME_S(inode->i_atime);
- attr->la_mtime = LTIME_S(inode->i_mtime);
- attr->la_ctime = LTIME_S(inode->i_ctime);
- attr->la_mode = inode->i_mode;
- attr->la_size = i_size_read(inode);
- attr->la_blocks = inode->i_blocks;
- attr->la_uid = inode->i_uid;
- attr->la_gid = inode->i_gid;
- attr->la_flags = LDISKFS_I(inode)->i_flags;
- attr->la_nlink = inode->i_nlink;
- attr->la_rdev = inode->i_rdev;
- attr->la_blksize = 1 << inode->i_blkbits;
- attr->la_blkbits = inode->i_blkbits;
+ struct inode *inode, struct lu_attr *attr)
+{
+ attr->la_valid |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE |
+ LA_SIZE | LA_BLOCKS | LA_UID | LA_GID |
+ LA_FLAGS | LA_NLINK | LA_RDEV | LA_BLKSIZE |
+ LA_TYPE;
+
+ attr->la_atime = LTIME_S(inode->i_atime);
+ attr->la_mtime = LTIME_S(inode->i_mtime);
+ attr->la_ctime = LTIME_S(inode->i_ctime);
+ attr->la_mode = inode->i_mode;
+ attr->la_size = i_size_read(inode);
+ 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_nlink = inode->i_nlink;
+ attr->la_rdev = inode->i_rdev;
+ attr->la_blksize = 1 << inode->i_blkbits;
+ attr->la_blkbits = 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 dt_object *dt,
+ struct lu_attr *attr,
+ struct lustre_capa *capa)
{
- struct osd_object *obj = osd_dt_obj(dt);
+ struct osd_object *obj = osd_dt_obj(dt);
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LINVRNT(osd_invariant(obj));
+ if (!dt_object_exists(dt))
+ return -ENOENT;
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
- return -EACCES;
+ 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);
struct osd_object *obj;
struct osd_thread_info *info = osd_oti_get(env);
struct lquota_id_info *qi = &info->oti_qi;
+ qid_t uid;
+ qid_t gid;
long long bspace;
int rc = 0;
- bool allocated;
+ bool enforce;
ENTRY;
LASSERT(dt != NULL);
* We still need to call the osd_declare_qid() to calculate the journal
* credits for updating quota accounting files and to trigger quota
* space adjustment once the operation is completed.*/
- if ((attr->la_valid & LA_UID) != 0 &&
- attr->la_uid != obj->oo_inode->i_uid) {
+ if (attr->la_valid & LA_UID || attr->la_valid & LA_GID) {
+ /* USERQUOTA */
+ uid = i_uid_read(obj->oo_inode);
qi->lqi_type = USRQUOTA;
-
+ enforce = (attr->la_valid & LA_UID) && (attr->la_uid != uid);
/* inode accounting */
qi->lqi_is_blk = false;
- /* one more inode for the new owner ... */
+ /* one more inode for the new uid ... */
qi->lqi_id.qid_uid = attr->la_uid;
qi->lqi_space = 1;
- allocated = (attr->la_uid == 0) ? true : false;
- rc = osd_declare_qid(env, oh, qi, allocated, NULL);
+ /* Reserve credits for the new uid */
+ rc = osd_declare_qid(env, oh, qi, NULL, enforce, NULL);
if (rc == -EDQUOT || rc == -EINPROGRESS)
rc = 0;
if (rc)
RETURN(rc);
/* and one less inode for the current uid */
- qi->lqi_id.qid_uid = obj->oo_inode->i_uid;
+ qi->lqi_id.qid_uid = uid;
qi->lqi_space = -1;
- rc = osd_declare_qid(env, oh, qi, true, NULL);
+ rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
if (rc == -EDQUOT || rc == -EINPROGRESS)
rc = 0;
if (rc)
/* block accounting */
qi->lqi_is_blk = true;
- /* more blocks for the new owner ... */
+ /* more blocks for the new uid ... */
qi->lqi_id.qid_uid = attr->la_uid;
qi->lqi_space = bspace;
- allocated = (attr->la_uid == 0) ? true : false;
- rc = osd_declare_qid(env, oh, qi, allocated, NULL);
+ /*
+ * Credits for the new uid has been reserved, re-use "obj"
+ * to save credit reservation.
+ */
+ rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
if (rc == -EDQUOT || rc == -EINPROGRESS)
rc = 0;
if (rc)
RETURN(rc);
- /* and finally less blocks for the current owner */
- qi->lqi_id.qid_uid = obj->oo_inode->i_uid;
+ /* and finally less blocks for the current uid */
+ qi->lqi_id.qid_uid = uid;
qi->lqi_space = -bspace;
- rc = osd_declare_qid(env, oh, qi, true, NULL);
+ rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
if (rc == -EDQUOT || rc == -EINPROGRESS)
rc = 0;
if (rc)
RETURN(rc);
- }
- if (attr->la_valid & LA_GID &&
- attr->la_gid != obj->oo_inode->i_gid) {
+ /* GROUP QUOTA */
+ gid = i_gid_read(obj->oo_inode);
qi->lqi_type = GRPQUOTA;
+ enforce = (attr->la_valid & LA_GID) && (attr->la_gid != gid);
/* inode accounting */
qi->lqi_is_blk = false;
- /* one more inode for the new group owner ... */
+ /* one more inode for the new gid ... */
qi->lqi_id.qid_gid = attr->la_gid;
qi->lqi_space = 1;
- allocated = (attr->la_gid == 0) ? true : false;
- rc = osd_declare_qid(env, oh, qi, allocated, NULL);
+ rc = osd_declare_qid(env, oh, qi, NULL, enforce, NULL);
if (rc == -EDQUOT || rc == -EINPROGRESS)
rc = 0;
if (rc)
RETURN(rc);
/* and one less inode for the current gid */
- qi->lqi_id.qid_gid = obj->oo_inode->i_gid;
+ qi->lqi_id.qid_gid = gid;
qi->lqi_space = -1;
- rc = osd_declare_qid(env, oh, qi, true, NULL);
+ rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
if (rc == -EDQUOT || rc == -EINPROGRESS)
rc = 0;
if (rc)
/* block accounting */
qi->lqi_is_blk = true;
- /* more blocks for the new owner ... */
+ /* more blocks for the new gid ... */
qi->lqi_id.qid_gid = attr->la_gid;
qi->lqi_space = bspace;
- allocated = (attr->la_gid == 0) ? true : false;
- rc = osd_declare_qid(env, oh, qi, allocated, NULL);
+ rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
if (rc == -EDQUOT || rc == -EINPROGRESS)
rc = 0;
if (rc)
RETURN(rc);
- /* and finally less blocks for the current owner */
- qi->lqi_id.qid_gid = obj->oo_inode->i_gid;
+ /* and finally less blocks for the current gid */
+ qi->lqi_id.qid_gid = gid;
qi->lqi_space = -bspace;
- rc = osd_declare_qid(env, oh, qi, true, NULL);
+ rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
if (rc == -EDQUOT || rc == -EINPROGRESS)
rc = 0;
if (rc)
}
static int osd_inode_setattr(const struct lu_env *env,
- struct inode *inode, const struct lu_attr *attr)
+ struct inode *inode, const struct lu_attr *attr)
{
- __u64 bits;
+ __u64 bits = attr->la_valid;
+
+ /* Only allow set size for regular file */
+ if (!S_ISREG(inode->i_mode))
+ bits &= ~(LA_SIZE | LA_BLOCKS);
- bits = attr->la_valid;
+ if (bits == 0)
+ return 0;
if (bits & LA_ATIME)
inode->i_atime = *osd_inode_time(env, inode, attr->la_atime);
if (bits & LA_BLOCKS)
inode->i_blocks = attr->la_blocks;
#endif
- if (bits & LA_MODE)
- inode->i_mode = (inode->i_mode & S_IFMT) |
- (attr->la_mode & ~S_IFMT);
- if (bits & LA_UID)
- inode->i_uid = attr->la_uid;
- if (bits & LA_GID)
- inode->i_gid = attr->la_gid;
- if (bits & LA_NLINK)
+ if (bits & LA_MODE)
+ inode->i_mode = (inode->i_mode & S_IFMT) |
+ (attr->la_mode & ~S_IFMT);
+ if (bits & LA_UID)
+ i_uid_write(inode, attr->la_uid);
+ if (bits & LA_GID)
+ i_gid_write(inode, attr->la_gid);
+ if (bits & LA_NLINK)
set_nlink(inode, attr->la_nlink);
- if (bits & LA_RDEV)
- inode->i_rdev = attr->la_rdev;
+ if (bits & LA_RDEV)
+ inode->i_rdev = attr->la_rdev;
if (bits & LA_FLAGS) {
/* always keep S_NOCMTIME */
static int osd_quota_transfer(struct inode *inode, const struct lu_attr *attr)
{
- if ((attr->la_valid & LA_UID && attr->la_uid != inode->i_uid) ||
- (attr->la_valid & LA_GID && attr->la_gid != inode->i_gid)) {
+ if ((attr->la_valid & LA_UID && attr->la_uid != i_uid_read(inode)) ||
+ (attr->la_valid & LA_GID && attr->la_gid != i_gid_read(inode))) {
struct iattr iattr;
int rc;
+ ll_vfs_dq_init(inode);
iattr.ia_valid = 0;
if (attr->la_valid & LA_UID)
iattr.ia_valid |= ATTR_UID;
if (attr->la_valid & LA_GID)
iattr.ia_valid |= ATTR_GID;
- iattr.ia_uid = attr->la_uid;
- iattr.ia_gid = attr->la_gid;
+ iattr.ia_uid = make_kuid(&init_user_ns, attr->la_uid);
+ iattr.ia_gid = make_kgid(&init_user_ns, attr->la_gid);
rc = ll_vfs_dq_transfer(inode, &iattr);
if (rc) {
CERROR("%s: quota transfer failed: rc = %d. Is quota "
- "enforcement enabled on the ldiskfs filesystem?",
- inode->i_sb->s_id, rc);
+ "enforcement enabled on the ldiskfs "
+ "filesystem?\n", inode->i_sb->s_id, rc);
return rc;
}
}
}
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 dt_object *dt,
+ const struct lu_attr *attr,
+ struct thandle *handle,
+ struct lustre_capa *capa)
{
- struct osd_object *obj = osd_dt_obj(dt);
- struct inode *inode;
- int rc;
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct inode *inode;
+ int rc;
- LASSERT(handle != NULL);
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(osd_invariant(obj));
+ if (!dt_object_exists(dt))
+ return -ENOENT;
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
- return -EACCES;
+ LASSERT(handle != NULL);
+ 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);
}
inode = obj->oo_inode;
- ll_vfs_dq_init(inode);
rc = osd_quota_transfer(inode, attr);
if (rc)
return rc;
}
-struct dentry *osd_child_dentry_get(const struct lu_env *env,
- struct osd_object *obj,
- const char *name, const int namelen)
+static struct dentry *osd_child_dentry_get(const struct lu_env *env,
+ struct osd_object *obj,
+ const char *name, const int namelen)
{
return osd_child_dentry_by_inode(env, obj->oo_inode, name, namelen);
}
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 {
{
LASSERT(ah);
- memset(ah, 0, sizeof(*ah));
ah->dah_parent = parent;
ah->dah_mode = child_mode;
}
* should not happen since quota enforcement is no longer
* enabled on ldiskfs (lquota takes care of it).
*/
- LASSERTF(result == 0, "%d", result);
+ LASSERTF(result == 0, "%d\n", result);
ll_dirty_inode(inode, I_DIRTY_DATASYNC);
}
result = osd_create_type_f(dof->dof_type)(info, obj, attr, hint, dof,
th);
- if (result == 0) {
+ if (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);
- }
+ }
+
+ if (obj->oo_inode != NULL) {
+ LASSERT(obj->oo_inode->i_state & I_NEW);
+
+ unlock_new_inode(obj->oo_inode);
+ }
/* restore previous umask value */
current->fs->umask = umask;
* \retval 0, on success
*/
static int __osd_oi_insert(const struct lu_env *env, struct osd_object *obj,
- const struct lu_fid *fid, struct thandle *th)
+ const struct lu_fid *fid, struct thandle *th)
{
- struct osd_thread_info *info = osd_oti_get(env);
- struct osd_inode_id *id = &info->oti_id;
- struct osd_device *osd = osd_obj2dev(obj);
+ struct osd_thread_info *info = osd_oti_get(env);
+ struct osd_inode_id *id = &info->oti_id;
+ struct osd_device *osd = osd_obj2dev(obj);
+ struct osd_thandle *oh;
- LASSERT(obj->oo_inode != NULL);
+ LASSERT(obj->oo_inode != NULL);
+
+ oh = container_of0(th, struct osd_thandle, ot_super);
+ LASSERT(oh->ot_handle);
osd_id_gen(id, obj->oo_inode->i_ino, obj->oo_inode->i_generation);
- return osd_oi_insert(info, osd, fid, id, th, OI_CHECK_FLD);
+ return osd_oi_insert(info, osd, fid, id, oh->ot_handle, OI_CHECK_FLD);
}
int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
- obd_seq seq, struct lu_seq_range *range)
+ u64 seq, struct lu_seq_range *range)
{
struct seq_server_site *ss = osd_seq_site(osd);
- int rc;
if (fid_seq_is_idif(seq)) {
fld_range_set_ost(range);
LASSERT(ss != NULL);
fld_range_set_any(range);
- rc = fld_server_lookup(env, ss->ss_server_fld, seq, range);
- if (rc != 0) {
- CERROR("%s: cannot find FLD range for "LPX64": rc = %d\n",
- osd_name(osd), seq, rc);
- }
- return rc;
+ /* OSD will only do local fld lookup */
+ return fld_local_lookup(env, ss->ss_server_fld, seq, range);
}
/*
struct dt_object_format *dof,
struct thandle *handle)
{
- struct lu_seq_range *range = &osd_oti_get(env)->oti_seq_range;
struct osd_thandle *oh;
int rc;
ENTRY;
osd_trans_declare_op(env, oh, OSD_OT_CREATE,
osd_dto_credits_noquota[DTO_OBJECT_CREATE]);
- if (!fid_is_on_ost(osd_oti_get(env), osd_dt_dev(handle->th_dev),
- lu_object_fid(&dt->do_lu), OI_CHECK_FLD))
- /* 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);
+ /* 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
RETURN(0);
rc = osd_declare_inode_qid(env, attr->la_uid, attr->la_gid, 1, oh,
- false, false, NULL, false);
+ osd_dt_obj(dt), false, NULL, false);
if (rc != 0)
RETURN(rc);
- /* It does fld look up inside declare, and the result will be
- * added to fld cache, so the following fld lookup inside insert
- * does not need send RPC anymore, so avoid send rpc with holding
- * transaction */
- if (fid_is_norm(lu_object_fid(&dt->do_lu)) &&
- !fid_is_last_id(lu_object_fid(&dt->do_lu)))
- osd_fld_lookup(env, osd_dt_dev(handle->th_dev),
- fid_seq(lu_object_fid(&dt->do_lu)), range);
-
-
RETURN(rc);
}
static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
- struct lu_attr *attr,
- struct dt_allocation_hint *hint,
- struct dt_object_format *dof,
- struct thandle *th)
+ struct lu_attr *attr,
+ struct dt_allocation_hint *hint,
+ struct dt_object_format *dof, struct thandle *th)
{
- 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;
+ 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;
+ ENTRY;
- ENTRY;
+ if (dt_object_exists(dt))
+ return -EEXIST;
- LINVRNT(osd_invariant(obj));
- LASSERT(!dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(osd_write_locked(env, obj));
- LASSERT(th != NULL);
+ LINVRNT(osd_invariant(obj));
+ LASSERT(!dt_object_remote(dt));
+ LASSERT(osd_write_locked(env, obj));
+ LASSERT(th != NULL);
if (unlikely(fid_is_acct(fid)))
/* Quota files can't be created from the kernel any more,
osd_trans_declare_op(env, oh, OSD_OT_DELETE,
osd_dto_credits_noquota[DTO_INDEX_DELETE] + 3);
/* one less inode */
- rc = osd_declare_inode_qid(env, inode->i_uid, inode->i_gid, -1, oh,
- false, true, NULL, false);
+ rc = osd_declare_inode_qid(env, i_uid_read(inode), i_gid_read(inode),
+ -1, oh, obj, false, NULL, false);
if (rc)
RETURN(rc);
/* data to be truncated */
- rc = osd_declare_inode_qid(env, inode->i_uid, inode->i_gid, 0, oh,
- true, true, NULL, false);
+ rc = osd_declare_inode_qid(env, i_uid_read(inode), i_gid_read(inode),
+ 0, oh, obj, true, NULL, false);
RETURN(rc);
}
RETURN(-EPERM);
if (S_ISDIR(inode->i_mode)) {
- LASSERT(osd_inode_unlinked(inode) || inode->i_nlink == 1);
+ LASSERT(osd_inode_unlinked(inode) || inode->i_nlink == 1 ||
+ inode->i_nlink == 2);
/* it will check/delete the inode from remote parent,
* how to optimize it? unlink performance impaction XXX */
result = osd_delete_from_remote_parent(env, osd, obj, oh);
osd_trans_exec_op(env, th, OSD_OT_DESTROY);
- result = osd_oi_delete(osd_oti_get(env), osd, fid, th, OI_CHECK_FLD);
+ 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) */
* \ldiskfs_dentry_param is used only to pass fid from osd to ldiskfs.
* its inmemory API.
*/
-void osd_get_ldiskfs_dirent_param(struct ldiskfs_dentry_param *param,
- const struct dt_rec *fid)
+static void osd_get_ldiskfs_dirent_param(struct ldiskfs_dentry_param *param,
+ const struct lu_fid *fid)
{
- if (!fid_is_namespace_visible((const struct lu_fid *)fid) ||
+ if (!fid_is_namespace_visible(fid) ||
OBD_FAIL_CHECK(OBD_FAIL_FID_IGIF)) {
param->edp_magic = 0;
return;
static int osd_add_dot_dotdot_internal(struct osd_thread_info *info,
struct inode *dir,
- struct inode *parent_dir,
- const struct dt_rec *dot_fid,
- const struct dt_rec *dot_dot_fid,
+ struct inode *parent_dir,
+ const struct lu_fid *dot_fid,
+ const struct lu_fid *dot_dot_fid,
struct osd_thandle *oth)
{
struct ldiskfs_dentry_param *dot_ldp;
struct osd_device *osd,
struct osd_object *pobj,
const struct lu_fid *fid,
+ __u32 type,
struct thandle *th)
{
struct osd_thread_info *info = osd_oti_get(env);
oh = container_of(th, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle->h_transaction != NULL);
- /* FIXME: Insert index api needs to know the mode of
- * the remote object. Just use S_IFDIR for now */
- local = ldiskfs_create_inode(oh->ot_handle, pobj->oo_inode, S_IFDIR);
+ local = ldiskfs_create_inode(oh->ot_handle, pobj->oo_inode, type);
if (IS_ERR(local)) {
CERROR("%s: create local error %d\n", osd_name(osd),
(int)PTR_ERR(local));
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) {
RETURN(ERR_PTR(rc));
}
+ if (!S_ISDIR(type))
+ RETURN(local);
+
rc = osd_add_dot_dotdot_internal(info, local, pobj->oo_inode,
- (const struct dt_rec *)lu_object_fid(&pobj->oo_dt.do_lu),
- (const struct dt_rec *)fid, oh);
+ lu_object_fid(&pobj->oo_dt.do_lu),
+ fid, oh);
if (rc != 0) {
CERROR("%s: "DFID" add dot dotdot error: rc = %d\n",
osd_name(osd), PFID(fid), rc);
* \retval -ve, on error
*/
static int osd_object_ea_create(const struct lu_env *env, struct dt_object *dt,
- struct lu_attr *attr,
- struct dt_allocation_hint *hint,
- struct dt_object_format *dof,
- struct thandle *th)
+ struct lu_attr *attr,
+ struct dt_allocation_hint *hint,
+ struct dt_object_format *dof,
+ struct thandle *th)
{
- 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;
+ 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;
- ENTRY;
+ ENTRY;
- LASSERT(osd_invariant(obj));
- LASSERT(!dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(osd_write_locked(env, obj));
- LASSERT(th != NULL);
+ if (dt_object_exists(dt))
+ RETURN(-EEXIST);
+
+ LASSERT(osd_invariant(obj));
+ LASSERT(!dt_object_remote(dt));
+ LASSERT(osd_write_locked(env, obj));
+ LASSERT(th != NULL);
if (unlikely(fid_is_acct(fid)))
/* Quota files can't be created from the kernel any more,
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),
+ result = __osd_object_create(info, obj, attr, hint, dof, th);
+ 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);
+ result = osd_ea_fid_set(info, obj->oo_inode, tfid,
+ LMAC_FID_ON_OST, 0);
+ } else {
+ 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)
result = __osd_oi_insert(env, obj, fid, th);
* Concurrency: @dt is write locked.
*/
static int osd_object_ref_add(const struct lu_env *env,
- struct dt_object *dt, struct thandle *th)
+ struct dt_object *dt, struct thandle *th)
{
- struct osd_object *obj = osd_dt_obj(dt);
- struct inode *inode = obj->oo_inode;
- bool need_dirty = false;
- int rc = 0;
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct inode *inode = obj->oo_inode;
+ struct osd_thandle *oh;
+ int rc = 0;
- LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(osd_write_locked(env, obj));
- LASSERT(th != NULL);
+ if (!dt_object_exists(dt))
+ return -ENOENT;
+
+ LINVRNT(osd_invariant(obj));
+ LASSERT(!dt_object_remote(dt));
+ LASSERT(osd_write_locked(env, obj));
+ LASSERT(th != NULL);
+
+ oh = container_of0(th, struct osd_thandle, ot_super);
+ LASSERT(oh->ot_handle != NULL);
osd_trans_exec_op(env, th, OSD_OT_REF_ADD);
- /* This based on ldiskfs_inc_count(), which is not exported.
- *
+ CDEBUG(D_INODE, DFID" increase nlink %d\n",
+ PFID(lu_object_fid(&dt->do_lu)), inode->i_nlink);
+ /*
* The DIR_NLINK feature allows directories to exceed LDISKFS_LINK_MAX
* (65000) subdirectories by storing "1" in i_nlink if the link count
* would otherwise overflow. Directory tranversal tools understand
* in case they are being linked into the PENDING directory
*/
spin_lock(&obj->oo_guard);
- if (unlikely(!S_ISDIR(inode->i_mode) &&
- inode->i_nlink >= LDISKFS_LINK_MAX)) {
- /* MDD should have checked this, but good to be safe */
- rc = -EMLINK;
- } else if (unlikely(inode->i_nlink == 0 ||
- (S_ISDIR(inode->i_mode) &&
- inode->i_nlink >= LDISKFS_LINK_MAX))) {
+ if (unlikely(inode->i_nlink == 0))
/* inc_nlink from 0 may cause WARN_ON */
set_nlink(inode, 1);
- need_dirty = true;
- } else if (!S_ISDIR(inode->i_mode) ||
- (S_ISDIR(inode->i_mode) && inode->i_nlink >= 2)) {
- inc_nlink(inode);
- need_dirty = true;
- } /* else (S_ISDIR(inode->i_mode) && inode->i_nlink == 1) { ; } */
-
- LASSERT(inode->i_nlink <= LDISKFS_LINK_MAX);
+ else {
+ ldiskfs_inc_count(oh->ot_handle, inode);
+ if (!S_ISDIR(inode->i_mode))
+ LASSERT(inode->i_nlink <= LDISKFS_LINK_MAX);
+ }
spin_unlock(&obj->oo_guard);
- if (need_dirty)
- ll_dirty_inode(inode, I_DIRTY_DATASYNC);
-
+ ll_dirty_inode(inode, I_DIRTY_DATASYNC);
LINVRNT(osd_invariant(obj));
return rc;
}
static int osd_declare_object_ref_del(const struct lu_env *env,
- struct dt_object *dt,
- struct thandle *handle)
+ struct dt_object *dt,
+ struct thandle *handle)
{
- struct osd_thandle *oh;
+ struct osd_thandle *oh;
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(handle != NULL);
+ LASSERT(!dt_object_remote(dt));
+ LASSERT(handle != NULL);
- oh = container_of0(handle, struct osd_thandle, ot_super);
- LASSERT(oh->ot_handle == NULL);
+ oh = container_of0(handle, struct osd_thandle, ot_super);
+ LASSERT(oh->ot_handle == NULL);
osd_trans_declare_op(env, oh, OSD_OT_REF_DEL,
osd_dto_credits_noquota[DTO_ATTR_SET_BASE]);
* Concurrency: @dt is write locked.
*/
static int osd_object_ref_del(const struct lu_env *env, struct dt_object *dt,
- struct thandle *th)
+ struct thandle *th)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
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) && !dt_object_remote(dt));
+ LASSERT(!dt_object_remote(dt));
LASSERT(osd_write_locked(env, obj));
LASSERT(th != NULL);
+ oh = container_of0(th, struct osd_thandle, ot_super);
+ LASSERT(oh->ot_handle != NULL);
+
osd_trans_exec_op(env, th, OSD_OT_REF_DEL);
spin_lock(&obj->oo_guard);
return 0;
}
- /* This based on ldiskfs_dec_count(), which is not exported.
- *
- * If a directory already has nlink == 1, then do not drop the nlink
- * count to 0, even temporarily, to avoid race conditions with other
- * threads not holding oo_guard seeing i_nlink == 0 in rare cases.
- *
- * nlink == 1 means the directory has/had > EXT4_LINK_MAX subdirs.
- */
- if (!S_ISDIR(inode->i_mode) || inode->i_nlink > 1) {
- drop_nlink(inode);
+ CDEBUG(D_INODE, DFID" decrease nlink %d\n",
+ PFID(lu_object_fid(&dt->do_lu)), inode->i_nlink);
- spin_unlock(&obj->oo_guard);
- ll_dirty_inode(inode, I_DIRTY_DATASYNC);
- LINVRNT(osd_invariant(obj));
- } else {
- spin_unlock(&obj->oo_guard);
- }
+ ldiskfs_dec_count(oh->ot_handle, inode);
+ spin_unlock(&obj->oo_guard);
+
+ ll_dirty_inode(inode, I_DIRTY_DATASYNC);
+ LINVRNT(osd_invariant(obj));
return 0;
}
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_get(env, dt, buf->lb_buf);
- return sizeof(dt_obj_version_t);
+ if (buf->lb_len == 0)
+ return sizeof(dt_obj_version_t);
+
+ if (buf->lb_len < sizeof(dt_obj_version_t))
+ return -ERANGE;
+
+ osd_object_version_get(env, dt, buf->lb_buf);
+
+ return sizeof(dt_obj_version_t);
}
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(inode->i_op != NULL && inode->i_op->getxattr != NULL);
+ if (!dt_object_exists(dt))
+ return -ENOENT;
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
- return -EACCES;
+ 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;
return __osd_xattr_get(inode, dentry, name, buf->lb_buf, buf->lb_len);
}
int fl, struct thandle *handle)
{
struct osd_thandle *oh;
+ int credits;
+ struct super_block *sb = osd_sb(osd_dev(dt->do_lu.lo_dev));
LASSERT(handle != NULL);
oh = container_of0(handle, struct osd_thandle, ot_super);
LASSERT(oh->ot_handle == NULL);
- osd_trans_declare_op(env, oh, OSD_OT_XATTR_SET,
- strcmp(name, XATTR_NAME_VERSION) == 0 ?
- osd_dto_credits_noquota[DTO_ATTR_SET_BASE] :
- osd_dto_credits_noquota[DTO_XATTR_SET]);
+ /* optimistic optimization: LMA is set first and usually fit inode */
+ if (strcmp(name, XATTR_NAME_LMA) == 0) {
+ if (dt_object_exists(dt))
+ credits = 0;
+ else
+ credits = 1;
+ } else if (strcmp(name, XATTR_NAME_VERSION) == 0) {
+ credits = 1;
+ } else {
+ 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;
+ }
+ /*
+ * xattr set may involve inode quota change, reserve credits for
+ * dquot_initialize()
+ */
+ oh->ot_credits += LDISKFS_MAXQUOTAS_INIT_BLOCKS(sb);
+ }
+
+ osd_trans_declare_op(env, oh, OSD_OT_XATTR_SET, credits);
return 0;
}
if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
return -EACCES;
+ CDEBUG(D_INODE, DFID" set xattr '%s' with size %zu\n",
+ PFID(lu_object_fid(&dt->do_lu)), name, buf->lb_len);
+
osd_trans_exec_op(env, handle, OSD_OT_XATTR_SET);
if (fl & LU_XATTR_REPLACE)
fs_flags |= XATTR_REPLACE;
if (fl & LU_XATTR_CREATE)
fs_flags |= XATTR_CREATE;
+ if (strcmp(name, XATTR_NAME_LMV) == 0) {
+ struct lustre_mdt_attrs *lma = &info->oti_mdt_attrs;
+ int rc;
+
+ rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
+ if (rc != 0)
+ RETURN(rc);
+
+ lma->lma_incompat |= LMAI_STRIPED;
+ lustre_lma_swab(lma);
+ rc = __osd_xattr_set(info, inode, XATTR_NAME_LMA, lma,
+ sizeof(*lma), XATTR_REPLACE);
+ if (rc != 0)
+ RETURN(rc);
+ }
+
+ if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_OVERFLOW) &&
+ strcmp(name, XATTR_NAME_LINK) == 0)
+ return -ENOSPC;
+
return __osd_xattr_set(info, inode, name, buf->lb_buf, buf->lb_len,
fs_flags);
}
* Concurrency: @dt is read locked.
*/
static int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
- struct lu_buf *buf, struct lustre_capa *capa)
+ const struct lu_buf *buf, struct lustre_capa *capa)
{
- 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;
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(inode->i_op != NULL && inode->i_op->listxattr != NULL);
- LASSERT(osd_read_locked(env, obj) || osd_write_locked(env, obj));
+ if (!dt_object_exists(dt))
+ return -ENOENT;
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
- return -EACCES;
+ 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_inode = inode;
dentry->d_sb = inode->i_sb;
- return inode->i_op->listxattr(dentry, buf->lb_buf, buf->lb_len);
+ return inode->i_op->listxattr(dentry, buf->lb_buf, buf->lb_len);
}
static int osd_declare_xattr_del(const struct lu_env *env,
- struct dt_object *dt, const char *name,
- struct thandle *handle)
+ struct dt_object *dt, const char *name,
+ struct thandle *handle)
{
- struct osd_thandle *oh;
+ struct osd_thandle *oh;
+ struct super_block *sb = osd_sb(osd_dev(dt->do_lu.lo_dev));
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(handle != NULL);
+ LASSERT(!dt_object_remote(dt));
+ LASSERT(handle != NULL);
- oh = container_of0(handle, struct osd_thandle, ot_super);
- LASSERT(oh->ot_handle == NULL);
+ oh = container_of0(handle, struct osd_thandle, ot_super);
+ LASSERT(oh->ot_handle == NULL);
osd_trans_declare_op(env, oh, OSD_OT_XATTR_SET,
osd_dto_credits_noquota[DTO_XATTR_SET]);
+ /*
+ * xattr del may involve inode quota change, reserve credits for
+ * dquot_initialize()
+ */
+ oh->ot_credits += LDISKFS_MAXQUOTAS_INIT_BLOCKS(sb);
return 0;
}
* 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 lustre_capa *capa)
{
- 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;
- int rc;
+ 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;
+ int rc;
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(inode->i_op != NULL && inode->i_op->removexattr != NULL);
- LASSERT(handle != NULL);
+ if (!dt_object_exists(dt))
+ return -ENOENT;
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
- return -EACCES;
+ 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_inode = inode;
dentry->d_sb = inode->i_sb;
- rc = inode->i_op->removexattr(dentry, name);
- return rc;
+ rc = inode->i_op->removexattr(dentry, name);
+ return rc;
}
static struct obd_capa *osd_capa_get(const struct lu_env *env,
if (!osd->od_fl_capa)
RETURN(ERR_PTR(-ENOENT));
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
+ if (!dt_object_exists(dt))
+ RETURN(ERR_PTR(-ENOENT));
+
+ LASSERT(!dt_object_remote(dt));
LINVRNT(osd_invariant(obj));
/* renewal sanity check */
case LC_ID_NONE:
RETURN(NULL);
case LC_ID_PLAIN:
- capa->lc_uid = obj->oo_inode->i_uid;
- capa->lc_gid = obj->oo_inode->i_gid;
+ 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] = obj->oo_inode->i_uid;
+ s[0] = i_uid_read(obj->oo_inode);
cfs_get_random_bytes(&(s[1]), sizeof(__u32));
- s[2] = obj->oo_inode->i_gid;
+ 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(oc);
}
-static int osd_object_sync(const struct lu_env *env, struct dt_object *dt)
+static int osd_object_sync(const struct lu_env *env, struct dt_object *dt,
+ __u64 start, __u64 end)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
file->f_dentry = dentry;
file->f_mapping = inode->i_mapping;
file->f_op = inode->i_fop;
-#ifndef HAVE_FILE_FSYNC_4ARGS
- mutex_lock(&inode->i_mutex);
-#endif
- rc = do_fsync(file, 0);
-#ifndef HAVE_FILE_FSYNC_4ARGS
- mutex_unlock(&inode->i_mutex);
-#endif
+ set_file_inode(file, inode);
+
+ rc = ll_vfs_fsync_range(file, start, end, 0);
+
RETURN(rc);
}
* recheck under lock.
*/
if (!osd_has_index(obj))
- result = osd_iam_container_init(env, obj, dir);
+ result = osd_iam_container_init(env, obj,
+ obj->oo_dir);
else
result = 0;
up_write(&obj->oo_ext_idx_sem);
* \retval 0 success
* \retval -ve failure
*/
-
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)
+ const struct dt_key *key,
+ struct thandle *handle,
+ struct lustre_capa *capa)
{
- struct osd_thread_info *oti = osd_oti_get(env);
- struct osd_object *obj = osd_dt_obj(dt);
- struct osd_thandle *oh;
- struct iam_path_descr *ipd;
- struct iam_container *bag = &obj->oo_dir->od_container;
- int rc;
+ struct osd_thread_info *oti = osd_oti_get(env);
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct osd_thandle *oh;
+ struct iam_path_descr *ipd;
+ struct iam_container *bag = &obj->oo_dir->od_container;
+ int rc;
+ ENTRY;
- ENTRY;
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
- LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(bag->ic_object == obj->oo_inode);
- LASSERT(handle != NULL);
+ LINVRNT(osd_invariant(obj));
+ 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);
+ if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
+ RETURN(-EACCES);
osd_trans_exec_op(env, handle, OSD_OT_DELETE);
int rc;
ENTRY;
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
+ LASSERT(!dt_object_remote(dt));
LASSERT(handle != NULL);
oh = container_of0(handle, struct osd_thandle, ot_super);
inode = osd_dt_obj(dt)->oo_inode;
LASSERT(inode);
- rc = osd_declare_inode_qid(env, inode->i_uid, inode->i_gid, 0, oh,
- true, true, NULL, false);
+ rc = osd_declare_inode_qid(env, i_uid_read(inode), i_gid_read(inode),
+ 0, oh, osd_dt_obj(dt), true, NULL, false);
RETURN(rc);
}
return rc;
}
-static int osd_mdt_seq_exists(const struct lu_env *env,
- struct osd_device *osd, obd_seq seq)
-{
- struct lu_seq_range *range = &osd_oti_get(env)->oti_seq_range;
- struct seq_server_site *ss = osd_seq_site(osd);
- int rc;
- ENTRY;
-
- if (ss == NULL)
- RETURN(1);
-
- /* XXX: currently, each MDT only store avaible sequence on disk, and no
- * allocated sequences information on disk, so we have to lookup FLDB,
- * but it probably makes more sense also store allocated sequence
- * locally, so we do not need do remote FLDB lookup in OSD */
- rc = osd_fld_lookup(env, osd, seq, range);
- if (rc != 0) {
- CERROR("%s: Can not lookup fld for "LPX64"\n",
- osd_name(osd), seq);
- RETURN(0);
- }
-
- RETURN(ss->ss_node_id == range->lsr_index);
-}
-
static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
- struct lu_fid *fid)
+ const struct lu_fid *fid)
{
+ struct seq_server_site *ss = osd_seq_site(osd);
ENTRY;
/* FID seqs not in FLDB, must be local seq */
if (unlikely(!fid_seq_in_fldb(fid_seq(fid))))
RETURN(0);
- /* Currently only check this for FID on MDT */
- if (osd_mdt_seq_exists(env, osd, fid_seq(fid)))
+ /* If FLD is not being initialized yet, it only happens during the
+ * initialization, likely during mgs initialization, and we assume
+ * this is local FID. */
+ if (ss == NULL || ss->ss_server_fld == NULL)
+ RETURN(0);
+
+ /* Only check the local FLDB here */
+ if (osd_seq_exists(env, osd, fid_seq(fid)))
RETURN(0);
RETURN(1);
* \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 lustre_capa *capa)
{
- struct osd_object *obj = osd_dt_obj(dt);
- struct inode *dir = obj->oo_inode;
- struct dentry *dentry;
- struct osd_thandle *oh;
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct inode *dir = obj->oo_inode;
+ struct dentry *dentry;
+ struct osd_thandle *oh;
struct ldiskfs_dir_entry_2 *de = NULL;
- struct buffer_head *bh;
- struct htree_lock *hlock = NULL;
+ struct buffer_head *bh;
+ struct htree_lock *hlock = NULL;
struct lu_fid *fid = &osd_oti_get(env)->oti_fid;
struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
int rc;
- ENTRY;
+ ENTRY;
- LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(handle != NULL);
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
+
+ LINVRNT(osd_invariant(obj));
+ LASSERT(!dt_object_remote(dt));
+ LASSERT(handle != NULL);
osd_trans_exec_op(env, handle, OSD_OT_DELETE);
down_write(&obj->oo_ext_idx_sem);
}
- bh = ldiskfs_find_entry(dir, &dentry->d_name, &de, hlock);
+ bh = osd_ldiskfs_find_entry(dir, &dentry->d_name, &de, NULL, hlock);
if (bh) {
__u32 ino = 0;
* \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 lustre_capa *capa)
{
- struct osd_object *obj = osd_dt_obj(dt);
- struct iam_path_descr *ipd;
- struct iam_container *bag = &obj->oo_dir->od_container;
- struct osd_thread_info *oti = osd_oti_get(env);
- struct iam_iterator *it = &oti->oti_idx_it;
- struct iam_rec *iam_rec;
- int rc;
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct iam_path_descr *ipd;
+ struct iam_container *bag = &obj->oo_dir->od_container;
+ struct osd_thread_info *oti = osd_oti_get(env);
+ struct iam_iterator *it = &oti->oti_idx_it;
+ struct iam_rec *iam_rec;
+ int rc;
+ ENTRY;
- ENTRY;
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
- LASSERT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(bag->ic_object == obj->oo_inode);
+ LASSERT(osd_invariant(obj));
+ 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);
* \retval -ve failure
*/
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)
-{
- 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 osd_thread_info *oti = osd_oti_get(env);
- struct iam_rec *iam_rec;
- int rc;
+ const struct dt_rec *rec,
+ const struct dt_key *key, struct thandle *th,
+ struct lustre_capa *capa, int ignore_quota)
+{
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct iam_path_descr *ipd;
+ struct osd_thandle *oh;
+ struct iam_container *bag;
+ struct osd_thread_info *oti = osd_oti_get(env);
+ struct iam_rec *iam_rec;
+ int rc;
+ ENTRY;
- ENTRY;
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
- LINVRNT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(bag->ic_object == obj->oo_inode);
- LASSERT(th != NULL);
+ LINVRNT(osd_invariant(obj));
+ LASSERT(!dt_object_remote(dt));
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
+ 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);
* \retval -ve, on error
*/
static int __osd_ea_add_rec(struct osd_thread_info *info,
- struct osd_object *pobj, struct inode *cinode,
- const char *name, const struct dt_rec *fid,
- struct htree_lock *hlock, struct thandle *th)
+ struct osd_object *pobj, struct inode *cinode,
+ const char *name, const struct lu_fid *fid,
+ struct htree_lock *hlock, struct thandle *th)
{
struct ldiskfs_dentry_param *ldp;
struct dentry *child;
child->d_fsdata = (void *)ldp;
ll_vfs_dq_init(pobj->oo_inode);
rc = osd_ldiskfs_add_entry(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;
+ int rc1;
+
+ bh = osd_ldiskfs_find_entry(pobj->oo_inode, &child->d_name, &de,
+ NULL, hlock);
+ if (bh != NULL) {
+ rc1 = ldiskfs_journal_get_write_access(oth->ot_handle,
+ bh);
+ if (rc1 == 0) {
+ if (S_ISDIR(cinode->i_mode))
+ de->file_type = LDISKFS_DIRENT_LUFID |
+ LDISKFS_FT_REG_FILE;
+ else
+ de->file_type = LDISKFS_DIRENT_LUFID |
+ LDISKFS_FT_DIR;
+ ldiskfs_handle_dirty_metadata(oth->ot_handle,
+ NULL, bh);
+ brelse(bh);
+ }
+ }
+ }
RETURN(rc);
}
* \retval -ve, on error
*/
static int osd_add_dot_dotdot(struct osd_thread_info *info,
- struct osd_object *dir,
- struct inode *parent_dir, const char *name,
- const struct dt_rec *dot_fid,
- const struct dt_rec *dot_dot_fid,
- struct thandle *th)
+ struct osd_object *dir,
+ struct inode *parent_dir, const char *name,
+ const struct lu_fid *dot_fid,
+ const struct lu_fid *dot_dot_fid,
+ struct thandle *th)
{
struct inode *inode = dir->oo_inode;
struct osd_thandle *oth;
dot_dot_fid, NULL, th);
}
- result = osd_add_dot_dotdot_internal(info, dir->oo_inode,
- parent_dir, dot_fid,
- dot_dot_fid, oth);
+ if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PARENT)) {
+ struct lu_fid tfid = *dot_dot_fid;
+
+ tfid.f_oid--;
+ result = osd_add_dot_dotdot_internal(info,
+ dir->oo_inode, parent_dir, dot_fid,
+ &tfid, oth);
+ } else {
+ result = osd_add_dot_dotdot_internal(info,
+ dir->oo_inode, parent_dir, dot_fid,
+ dot_dot_fid, oth);
+ }
+
if (result == 0)
dir->oo_compat_dotdot_created = 1;
}
* value, return by respective functions.
*/
static int osd_ea_add_rec(const struct lu_env *env, struct osd_object *pobj,
- struct inode *cinode, const char *name,
- const struct dt_rec *fid, struct thandle *th)
+ struct inode *cinode, const char *name,
+ const struct lu_fid *fid, struct thandle *th)
{
struct osd_thread_info *info = osd_oti_get(env);
struct htree_lock *hlock;
pobj->oo_inode, 0);
} else {
down_write(&pobj->oo_ext_idx_sem);
- }
- rc = osd_add_dot_dotdot(info, pobj, cinode, name,
- (struct dt_rec *)lu_object_fid(&pobj->oo_dt.do_lu),
+ }
+
+ rc = osd_add_dot_dotdot(info, pobj, cinode, name,
+ lu_object_fid(&pobj->oo_dt.do_lu),
fid, th);
} else {
if (hlock != NULL) {
if (OBD_FAIL_CHECK(OBD_FAIL_FID_INDIR)) {
struct lu_fid *tfid = &info->oti_fid;
- *tfid = *(const struct lu_fid *)fid;
+ *tfid = *fid;
tfid->f_ver = ~0;
rc = __osd_ea_add_rec(info, pobj, cinode, name,
- (const struct dt_rec *)tfid,
- hlock, th);
+ tfid, hlock, th);
} else {
rc = __osd_ea_add_rec(info, pobj, cinode, name, fid,
hlock, th);
if (!fid_is_norm(fid) && !fid_is_igif(fid))
RETURN_EXIT;
+ if (scrub->os_pos_current > id->oii_ino)
+ RETURN_EXIT;
+
again:
rc = osd_oi_lookup(oti, dev, fid, id, OI_CHECK_FLD);
if (rc != 0 && rc != -ENOENT)
}
if (!dev->od_noscrub && ++once == 1) {
- CDEBUG(D_LFSCK, "Trigger OI scrub by RPC for "DFID"\n",
- PFID(fid));
- rc = osd_scrub_start(dev);
- LCONSOLE_ERROR("%.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 = osd_scrub_start(dev, SS_AUTO_PARTIAL | SS_CLEAR_DRYRUN |
+ SS_CLEAR_FAILOUT);
+ 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;
}
return rc;
}
-int osd_add_oi_cache(struct osd_thread_info *info, struct osd_device *osd,
- struct osd_inode_id *id, const struct lu_fid *fid)
+void osd_add_oi_cache(struct osd_thread_info *info, struct osd_device *osd,
+ struct osd_inode_id *id, const struct lu_fid *fid)
{
CDEBUG(D_INODE, "add "DFID" %u:%u to info %p\n", PFID(fid),
id->oii_ino, id->oii_gen, info);
info->oti_cache.oic_lid = *id;
info->oti_cache.oic_fid = *fid;
info->oti_cache.oic_dev = osd;
+}
- return 0;
+/**
+ * Get parent FID from the linkEA.
+ *
+ * For a directory which parent resides on remote MDT, to satisfy the
+ * local e2fsck, we insert it into the /REMOTE_PARENT_DIR locally. On
+ * the other hand, to make the lookup(..) on the directory can return
+ * the real parent FID, we append the real parent FID after its ".."
+ * name entry in the /REMOTE_PARENT_DIR.
+ *
+ * Unfortunately, such PFID-in-dirent cannot be preserved via file-level
+ * backup. So after the restore, we cannot get the right parent FID from
+ * its ".." name entry in the /REMOTE_PARENT_DIR. Under such case, since
+ * we have stored the real parent FID in the directory object's linkEA,
+ * we can parse the linkEA for the real parent FID.
+ *
+ * \param[in] env pointer to the thread context
+ * \param[in] obj pointer to the object to be handled
+ * \param[out]fid pointer to the buffer to hold the parent FID
+ *
+ * \retval 0 for getting the real parent FID successfully
+ * \retval negative error number on failure
+ */
+static int osd_get_pfid_from_linkea(const struct lu_env *env,
+ struct osd_object *obj,
+ struct lu_fid *fid)
+{
+ struct osd_thread_info *oti = osd_oti_get(env);
+ 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 = { NULL };
+ int rc;
+ ENTRY;
+
+ fid_zero(fid);
+ if (!S_ISDIR(inode->i_mode))
+ RETURN(-EIO);
+
+again:
+ rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LINK,
+ buf->lb_buf, buf->lb_len);
+ if (rc == -ERANGE) {
+ rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LINK,
+ NULL, 0);
+ if (rc > 0) {
+ lu_buf_realloc(buf, rc);
+ if (buf->lb_buf == NULL)
+ RETURN(-ENOMEM);
+
+ goto again;
+ }
+ }
+
+ if (unlikely(rc == 0))
+ RETURN(-ENODATA);
+
+ if (rc < 0)
+ RETURN(rc);
+
+ if (unlikely(buf->lb_buf == NULL)) {
+ lu_buf_realloc(buf, rc);
+ if (buf->lb_buf == NULL)
+ RETURN(-ENOMEM);
+
+ goto again;
+ }
+
+ ldata.ld_buf = buf;
+ rc = linkea_init(&ldata);
+ if (rc == 0) {
+ linkea_first_entry(&ldata);
+ linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen, NULL, fid);
+ }
+
+ RETURN(rc);
}
/**
* \retval -ve, on error
*/
static int osd_ea_lookup_rec(const struct lu_env *env, struct osd_object *obj,
- struct dt_rec *rec, const struct dt_key *key)
-{
- struct inode *dir = obj->oo_inode;
- struct dentry *dentry;
- struct ldiskfs_dir_entry_2 *de;
- struct buffer_head *bh;
- struct lu_fid *fid = (struct lu_fid *) rec;
- struct htree_lock *hlock = NULL;
- int ino;
- int rc;
+ struct dt_rec *rec, const struct dt_key *key)
+{
+ struct inode *dir = obj->oo_inode;
+ struct dentry *dentry;
+ struct ldiskfs_dir_entry_2 *de;
+ struct buffer_head *bh;
+ struct lu_fid *fid = (struct lu_fid *) rec;
+ struct htree_lock *hlock = NULL;
+ int ino;
+ int rc;
ENTRY;
- LASSERT(dir->i_op != NULL && dir->i_op->lookup != NULL);
+ LASSERT(dir->i_op != NULL);
+ LASSERT(dir->i_op->lookup != NULL);
- dentry = osd_child_dentry_get(env, obj,
- (char *)key, strlen((char *)key));
+ dentry = osd_child_dentry_get(env, obj,
+ (char *)key, strlen((char *)key));
- if (obj->oo_hl_head != NULL) {
- hlock = osd_oti_get(env)->oti_hlock;
- ldiskfs_htree_lock(hlock, obj->oo_hl_head,
- dir, LDISKFS_HLOCK_LOOKUP);
- } else {
+ if (obj->oo_hl_head != NULL) {
+ hlock = osd_oti_get(env)->oti_hlock;
+ ldiskfs_htree_lock(hlock, obj->oo_hl_head,
+ dir, LDISKFS_HLOCK_LOOKUP);
+ } else {
down_read(&obj->oo_ext_idx_sem);
- }
+ }
- bh = osd_ldiskfs_find_entry(dir, dentry, &de, hlock);
- if (bh) {
+ bh = osd_ldiskfs_find_entry(dir, &dentry->d_name, &de, NULL, hlock);
+ if (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;
struct osd_device *dev = osd_obj2dev(obj);
- struct osd_scrub *scrub = &dev->od_scrub;
- struct scrub_file *sf = &scrub->os_file;
ino = le32_to_cpu(de->inode);
if (OBD_FAIL_CHECK(OBD_FAIL_FID_LOOKUP)) {
/* done with de, release bh */
brelse(bh);
- if (rc != 0)
- rc = osd_ea_fid_get(env, obj, ino, fid, id);
- else
+ if (rc != 0) {
+ 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. */
+ 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) {
fid_zero(&oic->oic_fid);
GOTO(out, rc);
}
- rc = osd_add_oi_cache(osd_oti_get(env), osd_obj2dev(obj), id,
- fid);
- if (rc != 0)
- GOTO(out, rc);
- if ((scrub->os_pos_current <= ino) &&
- ((sf->sf_flags & SF_INCONSISTENT) ||
- (sf->sf_flags & SF_UPGRADE && fid_is_igif(fid)) ||
- ldiskfs_test_bit(osd_oi_fid2idx(dev, fid),
- sf->sf_oi_bitmap)))
- osd_consistency_check(oti, dev, oic);
+ 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);
} else {
rc = -ENOENT;
}
* \retval osd_object on success
* \retval -ve on error
*/
-struct osd_object *osd_object_find(const struct lu_env *env,
- struct dt_object *dt,
- const struct lu_fid *fid)
+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;
* in the cache, otherwise lu_object_alloc() crashes
* -bzzz
*/
- luch = lu_object_find_at(env, ludev, 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"
+ 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);
child = ERR_PTR(-ENOENT);
}
- } else
- child = (void *)luch;
+ } else {
+ child = ERR_CAST(luch);
+ }
- return child;
+ return child;
}
/**
int rc;
ENTRY;
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
+ LASSERT(!dt_object_remote(dt));
LASSERT(handle != NULL);
oh = container_of0(handle, struct osd_thandle, ot_super);
osd_trans_declare_op(env, oh, OSD_OT_INSERT,
osd_dto_credits_noquota[DTO_INDEX_INSERT]);
- if (osd_dt_obj(dt)->oo_inode == NULL) {
- const char *name = (const char *)key;
- /* Object is not being created yet. Only happens when
- * 1. declare directory create
- * 2. declare insert .
- * 3. declare insert ..
- */
- LASSERT(strcmp(name, dotdot) == 0 || strcmp(name, dot) == 0);
- } else {
+ if (osd_dt_obj(dt)->oo_inode != NULL) {
struct inode *inode = osd_dt_obj(dt)->oo_inode;
/* We ignore block quota on meta pool (MDTs), so needn't
* calculate how many blocks will be consumed by this index
* insert */
- rc = osd_declare_inode_qid(env, inode->i_uid, inode->i_gid, 0,
- oh, true, true, NULL, false);
+ rc = osd_declare_inode_qid(env, i_uid_read(inode),
+ i_gid_read(inode), 0, oh,
+ osd_dt_obj(dt), true, NULL, false);
}
if (fid == NULL)
{
struct osd_object *obj = osd_dt_obj(dt);
struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
- struct lu_fid *fid = (struct lu_fid *) rec;
+ struct dt_insert_rec *rec1 = (struct dt_insert_rec *)rec;
+ 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;
int rc;
ENTRY;
- LASSERT(osd_invariant(obj));
- LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
- LASSERT(th != NULL);
+ if (!dt_object_exists(dt))
+ RETURN(-ENOENT);
+
+ LASSERT(osd_invariant(obj));
+ 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);
+ if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
+ RETURN(-EACCES);
- LASSERTF(fid_is_sane(fid), "fid"DFID" is insane!", PFID(fid));
+ LASSERTF(fid_is_sane(fid), "fid"DFID" is insane!\n", PFID(fid));
rc = osd_remote_fid(env, osd, fid);
if (rc < 0) {
child_inode = igrab(omm->omm_remote_parent->d_inode);
} else {
child_inode = osd_create_local_agent_inode(env, osd,
- obj, fid,
- th);
+ obj, fid, rec1->rec_type & S_IFMT, th);
if (IS_ERR(child_inode))
RETURN(PTR_ERR(child_inode));
}
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_inode));
- RETURN(PTR_ERR(child_inode));
+ (int)PTR_ERR(child));
+ RETURN(PTR_ERR(child));
}
child_inode = igrab(child->oo_inode);
}
- rc = osd_ea_add_rec(env, obj, child_inode, name, rec, th);
+ 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)
__u32 unused,
struct lustre_capa *capa)
{
- 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));
-
- if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_READ))
- return ERR_PTR(-EACCES);
-
- it = &oti->oti_it;
- ipd = osd_it_ipd_get(env, bag);
- if (likely(ipd != NULL)) {
- it->oi_obj = obj;
- it->oi_ipd = ipd;
- lu_object_get(lo);
- iam_it_init(&it->oi_it, bag, IAM_IT_MOVE, ipd);
- return (struct dt_it *)it;
- }
- return ERR_PTR(-ENOMEM);
+ 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 (!dt_object_exists(dt))
+ return ERR_PTR(-ENOENT);
+
+ if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_READ))
+ return ERR_PTR(-EACCES);
+
+ OBD_ALLOC_PTR(it);
+ if (it == NULL)
+ return ERR_PTR(-ENOMEM);
+
+ ipd = osd_it_ipd_get(env, bag);
+ if (likely(ipd != NULL)) {
+ it->oi_obj = obj;
+ it->oi_ipd = ipd;
+ lu_object_get(lo);
+ iam_it_init(&it->oi_it, bag, IAM_IT_MOVE, ipd);
+ return (struct dt_it *)it;
+ } else {
+ 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_it_iam *it = (struct osd_it_iam *)di;
- struct osd_object *obj = it->oi_obj;
+ 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);
+ 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);
+ 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)
-{
- struct osd_object *obj = osd_dt_obj(dt);
- struct osd_thread_info *info = osd_oti_get(env);
- struct osd_it_ea *it = &info->oti_it_ea;
- struct file *file = &it->oie_file;
- struct lu_object *lo = &dt->do_lu;
- struct dentry *obj_dentry = &info->oti_it_dentry;
- ENTRY;
- LASSERT(lu_object_exists(lo));
+ struct dt_object *dt,
+ __u32 attr,
+ struct lustre_capa *capa)
+{
+ struct osd_object *obj = osd_dt_obj(dt);
+ struct osd_thread_info *info = osd_oti_get(env);
+ struct osd_it_ea *oie;
+ struct file *file;
+ struct lu_object *lo = &dt->do_lu;
+ struct dentry *obj_dentry;
+ ENTRY;
- obj_dentry->d_inode = obj->oo_inode;
- obj_dentry->d_sb = osd_sb(osd_obj2dev(obj));
- obj_dentry->d_name.hash = 0;
+ if (!dt_object_exists(dt))
+ RETURN(ERR_PTR(-ENOENT));
- 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;
+ 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;
+
+ 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;
- /* 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;
file->f_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_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);
- EXIT;
+ 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_ea_buf_used = 0;
+ OBD_SLAB_FREE_PTR(oie, osd_itea_cachep);
+ EXIT;
}
/**
{
}
+struct osd_filldir_cbs {
+#ifdef HAVE_DIR_CONTEXT
+ struct dir_context ctx;
+#endif
+ struct osd_it_ea *it;
+};
/**
* It is called internally by ->readdir(). It fills the
* iterator's in-memory data structure with required
* \retval 0 on success
* \retval 1 on buffer full
*/
-static int osd_ldiskfs_filldir(char *buf, const char *name, int namelen,
+static int osd_ldiskfs_filldir(void *buf, const char *name, int namelen,
loff_t offset, __u64 ino,
unsigned d_type)
{
- struct osd_it_ea *it = (struct osd_it_ea *)buf;
+ 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;
*
* \retval 0 on success
* \retval -ve on error
+ * \retval +1 reach the end of entry
*/
static int osd_ldiskfs_it_fill(const struct lu_env *env,
const struct dt_it *di)
struct osd_object *obj = it->oie_obj;
struct inode *inode = obj->oo_inode;
struct htree_lock *hlock = NULL;
- int result = 0;
+ struct file *filp = &it->oie_file;
+ int rc = 0;
+ struct osd_filldir_cbs buf = {
+#ifdef HAVE_DIR_CONTEXT
+ .ctx.actor = osd_ldiskfs_filldir,
+#endif
+ .it = it
+ };
ENTRY;
it->oie_dirent = it->oie_buf;
down_read(&obj->oo_ext_idx_sem);
}
- result = inode->i_fop->readdir(&it->oie_file, it,
- (filldir_t) osd_ldiskfs_filldir);
+#ifdef HAVE_DIR_CONTEXT
+ buf.ctx.pos = filp->f_pos;
+ rc = inode->i_fop->iterate(filp, &buf.ctx);
+ filp->f_pos = buf.ctx.pos;
+#else
+ rc = inode->i_fop->readdir(filp, &buf, osd_ldiskfs_filldir);
+#endif
if (hlock != NULL)
ldiskfs_htree_unlock(hlock);
else
up_read(&obj->oo_ext_idx_sem);
- if (it->oie_rd_dirent == 0) {
- result = -EIO;
- } else {
- it->oie_dirent = it->oie_buf;
- it->oie_it_dirent = 1;
- }
+ if (it->oie_rd_dirent == 0) {
+ /*If it does not get any dirent, it means it has been reached
+ *to the end of the dir */
+ it->oie_file.f_pos = ldiskfs_get_htree_eof(&it->oie_file);
+ if (rc == 0)
+ rc = 1;
+ } else {
+ it->oie_dirent = it->oie_buf;
+ it->oie_it_dirent = 1;
+ }
- RETURN(result);
+ RETURN(rc);
}
/**
LASSERT(de->rec_len >= de->name_len + sizeof(struct osd_fid_pack));
rc = ldiskfs_journal_get_write_access(jh, bh);
- if (rc != 0) {
- CERROR("%.16s: fail to write access for update dirent: "
- "name = %.*s, rc = %d\n",
- LDISKFS_SB(sb)->s_es->s_volume_name,
- ent->oied_namelen, ent->oied_name, rc);
+ if (rc != 0)
RETURN(rc);
- }
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 (rc != 0)
- CERROR("%.16s: fail to dirty metadata for update dirent: "
- "name = %.*s, rc = %d\n",
- LDISKFS_SB(sb)->s_es->s_volume_name,
- ent->oied_namelen, ent->oied_name, rc);
+ rc = ldiskfs_handle_dirty_metadata(jh, NULL, bh);
RETURN(rc);
}
if (osd_dirent_has_space(de->rec_len, ent->oied_namelen,
dir->i_sb->s_blocksize)) {
rc = ldiskfs_journal_get_write_access(jh, bh);
- if (rc != 0) {
- CERROR("%.16s: fail to write access for reinsert "
- "dirent: name = %.*s, rc = %d\n",
- LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
- ent->oied_namelen, ent->oied_name, rc);
+ if (rc != 0)
RETURN(rc);
- }
de->name[de->name_len] = 0;
rec = (struct osd_fid_pack *)(de->name + de->name_len + 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);
- if (rc != 0)
- CERROR("%.16s: fail to dirty metadata for reinsert "
- "dirent: name = %.*s, rc = %d\n",
- LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
- ent->oied_namelen, ent->oied_name, rc);
+ rc = ldiskfs_handle_dirty_metadata(jh, NULL, bh);
RETURN(rc);
}
rc = ldiskfs_delete_entry(jh, dir, de, bh);
- if (rc != 0) {
- CERROR("%.16s: fail to delete entry for reinsert dirent: "
- "name = %.*s, rc = %d\n",
- LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
- ent->oied_namelen, ent->oied_name, rc);
+ 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, (const struct dt_rec *)fid);
+ 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);
/* It is too bad, we cannot reinsert the name entry back.
* That means we lose it! */
if (rc != 0)
- CERROR("%.16s: fail to insert entry for reinsert dirent: "
- "name = %.*s, rc = %d\n",
+ 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,
- ent->oied_namelen, ent->oied_name, rc);
+ dir->i_ino, dir->i_generation,
+ ent->oied_namelen, ent->oied_name, PFID(fid), rc);
RETURN(rc);
}
osd_dto_credits_noquota[DTO_INDEX_INSERT] + 1 + 1 + 2;
again:
- if (dev->od_dirent_journal) {
- jh = ldiskfs_journal_start_sb(sb, credits);
+ if (dev->od_dirent_journal != 0) {
+ jh = osd_journal_start_sb(sb, LDISKFS_HT_MISC, credits);
if (IS_ERR(jh)) {
rc = PTR_ERR(jh);
- CERROR("%.16s: fail to start trans for dirent "
- "check_repair: credits %d, name %.*s, rc %d\n",
- devname, credits, ent->oied_namelen,
- ent->oied_name, rc);
+ CDEBUG(D_LFSCK, "%.16s: fail to start trans for dirent "
+ "check_repair, dir = %lu/%u, credits = %d, "
+ "name = %.*s: rc = %d\n",
+ devname, dir->i_ino, dir->i_generation, credits,
+ ent->oied_namelen, ent->oied_name, rc);
RETURN(rc);
}
}
}
- bh = osd_ldiskfs_find_entry(dir, dentry, &de, hlock);
+ bh = osd_ldiskfs_find_entry(dir, &dentry->d_name, &de, NULL, hlock);
/* For dot/dotdot entry, if there is not enough space to hold the
* FID-in-dirent, just keep them there. It only happens when the
* device upgraded from 1.8 or restored from MDT file-level backup.
inode = osd_iget(info, dev, id);
if (IS_ERR(inode)) {
rc = PTR_ERR(inode);
- if (rc == -ENOENT || rc == -ESTALE) {
- *attr |= LUDA_IGNORE;
- rc = 0;
- }
+ 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);
}
- /* skip the REMOTE_PARENT_DIR. */
- if (inode == dev->od_mdt_map->omm_remote_parent->d_inode)
- GOTO(out_inode, rc = 0);
-
rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
if (rc == 0) {
LASSERT(!(lma->lma_compat & LMAC_NOT_IN_OI));
GOTO(out_inode, rc = 0);
}
- if (!dev->od_dirent_journal) {
+ if (dev->od_dirent_journal == 0) {
iput(inode);
brelse(bh);
if (hlock != NULL)
rc = osd_dirent_update(jh, sb, ent, fid, bh, de);
if (rc == 0)
*attr |= LUDA_REPAIR;
+ else
+ CDEBUG(D_LFSCK, "%.16s: fail to update FID "
+ "in the dirent, dir = %lu/%u, "
+ "name = %.*s, "DFID": rc = %d\n",
+ devname, dir->i_ino, dir->i_generation,
+ ent->oied_namelen, ent->oied_name,
+ PFID(fid), rc);
} else {
/* Do not repair under dryrun mode. */
if (*attr & LUDA_VERIFY_DRYRUN) {
GOTO(out_inode, rc = 0);
}
- if (!dev->od_dirent_journal) {
+ if (dev->od_dirent_journal == 0) {
iput(inode);
brelse(bh);
if (hlock != NULL)
fid, bh, de, hlock);
if (rc == 0)
*attr |= LUDA_REPAIR;
+ else
+ CDEBUG(D_LFSCK, "%.16s: fail to append FID "
+ "after the dirent, dir = %lu/%u, "
+ "name = %.*s, "DFID": rc = %d\n",
+ devname, dir->i_ino, dir->i_generation,
+ ent->oied_namelen, ent->oied_name,
+ PFID(fid), rc);
}
} else if (rc == -ENODATA) {
/* Do not repair under dryrun mode. */
GOTO(out_inode, rc = 0);
}
- if (!dev->od_dirent_journal) {
+ if (dev->od_dirent_journal == 0) {
iput(inode);
brelse(bh);
if (hlock != NULL)
rc = osd_ea_fid_set(info, inode, fid, 0, 0);
if (rc == 0)
*attr |= LUDA_REPAIR;
+ else
+ CDEBUG(D_LFSCK, "%.16s: fail to set LMA for "
+ "update dirent, dir = %lu/%u, "
+ "name = %.*s, "DFID": rc = %d\n",
+ devname, dir->i_ino, dir->i_generation,
+ ent->oied_namelen, ent->oied_name,
+ PFID(fid), rc);
} else {
lu_igif_build(fid, inode->i_ino, inode->i_generation);
/* It is probably IGIF object. Only aappend the
fid, bh, de, hlock);
if (rc == 0)
*attr |= LUDA_UPGRADE;
+ else
+ CDEBUG(D_LFSCK, "%.16s: fail to append IGIF "
+ "after the dirent, dir = %lu/%u, "
+ "name = %.*s, "DFID": rc = %d\n",
+ devname, dir->i_ino, dir->i_generation,
+ ent->oied_namelen, ent->oied_name,
+ PFID(fid), rc);
}
}
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);
struct osd_it_ea *it = (struct osd_it_ea *)di;
struct osd_object *obj = it->oie_obj;
struct osd_device *dev = osd_obj2dev(obj);
- struct osd_scrub *scrub = &dev->od_scrub;
- struct scrub_file *sf = &scrub->os_file;
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;
struct lu_fid *fid = &it->oie_dirent->oied_fid;
struct lu_dirent *lde = (struct lu_dirent *)dtrec;
__u32 ino = it->oie_dirent->oied_ino;
int rc = 0;
ENTRY;
+ LASSERT(obj->oo_inode != dev->od_mdt_map->omm_remote_parent->d_inode);
+
if (attr & LUDA_VERIFY) {
- attr |= LUDA_TYPE;
- if (unlikely(ino == osd_sb(dev)->s_root->d_inode->i_ino)) {
+ if (unlikely(ino == osd_remote_parent_ino(dev))) {
attr |= LUDA_IGNORE;
- rc = 0;
+ /* 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 (!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);
} else {
attr &= ~LU_DIRENT_ATTRS_MASK;
if (!fid_is_sane(fid)) {
- if (OBD_FAIL_CHECK(OBD_FAIL_FID_LOOKUP))
- RETURN(-ENOENT);
+ if (it->oie_dirent->oied_namelen == 2 &&
+ it->oie_dirent->oied_name[0] == '.' &&
+ it->oie_dirent->oied_name[1] == '.') {
+ /* 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))
+ rc = osd_get_pfid_from_linkea(env, obj,
+ fid);
+ } else {
+ if (OBD_FAIL_CHECK(OBD_FAIL_FID_LOOKUP))
+ RETURN(-ENOENT);
- rc = osd_ea_fid_get(env, obj, ino, fid, id);
+ 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 = osd_add_oi_cache(oti, dev, id, fid);
+ if (likely(!(attr & LUDA_IGNORE) && rc == 0))
+ osd_add_oi_cache(oti, dev, id, fid);
- if (!(attr & LUDA_VERIFY) &&
- (scrub->os_pos_current <= ino) &&
- ((sf->sf_flags & SF_INCONSISTENT) ||
- (sf->sf_flags & SF_UPGRADE && fid_is_igif(fid)) ||
- ldiskfs_test_bit(osd_oi_fid2idx(dev, fid), sf->sf_oi_bitmap)))
- osd_consistency_check(oti, dev, oic);
+ RETURN(rc > 0 ? 0 : rc);
+}
- RETURN(rc);
+/**
+ * Returns the record size size at current position.
+ *
+ * This function will return record(lu_dirent) size in bytes.
+ *
+ * \param[in] env execution environment
+ * \param[in] di iterator's in memory structure
+ * \param[in] attr attribute of the entry, only requires LUDA_TYPE to
+ * calculate the lu_dirent size.
+ *
+ * \retval record size(in bytes & in memory) of the current lu_dirent
+ * entry.
+ */
+static int osd_it_ea_rec_size(const struct lu_env *env, const struct dt_it *di,
+ __u32 attr)
+{
+ struct osd_it_ea *it = (struct osd_it_ea *)di;
+
+ return lu_dirent_calc_size(it->oie_dirent->oied_namelen, attr);
}
/**
it->oie_file.f_pos = hash;
rc = osd_ldiskfs_it_fill(env, di);
+ if (rc > 0)
+ rc = -ENODATA;
+
if (rc == 0)
rc = +1;
* mode (i.e. to run 2.0 mds on 1.8 disk) (b11826)
*/
static const struct dt_index_operations osd_index_ea_ops = {
- .dio_lookup = osd_index_ea_lookup,
- .dio_declare_insert = osd_index_declare_ea_insert,
- .dio_insert = osd_index_ea_insert,
- .dio_declare_delete = osd_index_declare_ea_delete,
- .dio_delete = osd_index_ea_delete,
- .dio_it = {
- .init = osd_it_ea_init,
- .fini = osd_it_ea_fini,
- .get = osd_it_ea_get,
- .put = osd_it_ea_put,
- .next = osd_it_ea_next,
- .key = osd_it_ea_key,
- .key_size = osd_it_ea_key_size,
- .rec = osd_it_ea_rec,
- .store = osd_it_ea_store,
- .load = osd_it_ea_load
- }
+ .dio_lookup = osd_index_ea_lookup,
+ .dio_declare_insert = osd_index_declare_ea_insert,
+ .dio_insert = osd_index_ea_insert,
+ .dio_declare_delete = osd_index_declare_ea_delete,
+ .dio_delete = osd_index_ea_delete,
+ .dio_it = {
+ .init = osd_it_ea_init,
+ .fini = osd_it_ea_fini,
+ .get = osd_it_ea_get,
+ .put = osd_it_ea_put,
+ .next = osd_it_ea_next,
+ .key = osd_it_ea_key,
+ .key_size = osd_it_ea_key_size,
+ .rec = osd_it_ea_rec,
+ .rec_size = osd_it_ea_rec_size,
+ .store = osd_it_ea_store,
+ .load = osd_it_ea_load
+ }
};
static void *osd_key_init(const struct lu_context *ctx,
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);
OBD_FREE_PTR(info);
}
return osd_procfs_init(osd, name);
}
+static int osd_fid_init(const struct lu_env *env, struct osd_device *osd)
+{
+ struct seq_server_site *ss = osd_seq_site(osd);
+ int rc;
+ ENTRY;
+
+ if (osd->od_is_ost || osd->od_cl_seq != NULL)
+ RETURN(0);
+
+ if (unlikely(ss == NULL))
+ RETURN(-ENODEV);
+
+ OBD_ALLOC_PTR(osd->od_cl_seq);
+ if (osd->od_cl_seq == NULL)
+ RETURN(-ENOMEM);
+
+ 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);
+}
+
+static void osd_fid_fini(const struct lu_env *env, struct osd_device *osd)
+{
+ if (osd->od_cl_seq == NULL)
+ return;
+
+ seq_client_fini(osd->od_cl_seq);
+ OBD_FREE_PTR(osd->od_cl_seq);
+ osd->od_cl_seq = NULL;
+}
+
static int osd_shutdown(const struct lu_env *env, struct osd_device *o)
{
ENTRY;
o->od_quota_slave = NULL;
}
+ osd_fid_fini(env, o);
+
RETURN(0);
}
{
ENTRY;
- if (o->od_fsops) {
- fsfilt_put_ops(o->od_fsops);
- o->od_fsops = NULL;
- }
-
if (o->od_mnt != NULL) {
shrink_dcache_sb(osd_sb(o));
osd_sync(env, &o->od_dt_dev);
RETURN(-E2BIG);
strcpy(o->od_mntdev, dev);
- o->od_fsops = fsfilt_get_ops(mt_str(LDD_MT_LDISKFS));
- if (IS_ERR(o->od_fsops)) {
- CERROR("%s: Can't find fsfilt_ldiskfs\n", name);
- o->od_fsops = NULL;
- RETURN(-ENOTSUPP);
- }
-
OBD_PAGE_ALLOC(__page, GFP_IOFS);
if (__page == NULL)
GOTO(out, rc = -ENOMEM);
#endif
if (!LDISKFS_HAS_COMPAT_FEATURE(o->od_mnt->mnt_sb,
- LDISKFS_FEATURE_COMPAT_HAS_JOURNAL)) {
+ LDISKFS_FEATURE_COMPAT_HAS_JOURNAL)) {
CERROR("%s: device %s is mounted w/o journal\n", name, dev);
GOTO(out_mnt, rc = -EINVAL);
}
+#ifdef LDISKFS_MOUNT_DIRDATA
+ if (LDISKFS_HAS_INCOMPAT_FEATURE(o->od_mnt->mnt_sb,
+ LDISKFS_FEATURE_INCOMPAT_DIRDATA))
+ LDISKFS_SB(osd_sb(o))->s_mount_opt |= LDISKFS_MOUNT_DIRDATA;
+ else if (!o->od_is_ost)
+ CWARN("%s: device %s was upgraded from Lustre-1.x without "
+ "enabling the dirdata feature. If you do not want to "
+ "downgrade to Lustre-1.x again, you can enable it via "
+ "'tune2fs -O dirdata device'\n", name, dev);
+#endif
inode = osd_sb(o)->s_root->d_inode;
- ldiskfs_set_inode_state(inode, LDISKFS_STATE_LUSTRE_NO_OI);
lu_local_obj_fid(fid, OSD_FS_ROOT_OID);
rc = osd_ea_fid_set(info, inode, fid, LMAC_NOT_IN_OI, 0);
if (rc != 0) {
out:
if (__page)
OBD_PAGE_FREE(__page);
- if (rc)
- fsfilt_put_ops(o->od_fsops);
return rc;
}
struct osd_device *o = osd_dev(d);
ENTRY;
- osd_procfs_fini(o);
osd_shutdown(env, o);
+ osd_procfs_fini(o);
osd_scrub_cleanup(env, o);
osd_obj_map_fini(o);
osd_umount(env, o);
spin_lock_init(&o->od_osfs_lock);
mutex_init(&o->od_otable_mutex);
- o->od_osfs_age = cfs_time_shift_64(-1000);
o->od_capa_hash = init_capa_hash();
if (o->od_capa_hash == NULL)
o->od_writethrough_cache = 1;
o->od_readcache_max_filesize = OSD_MAX_CACHE_SIZE;
- rc = osd_mount(env, o, cfg);
- if (rc)
- GOTO(out_capa, rc);
-
cplen = strlcpy(o->od_svname, lustre_cfg_string(cfg, 4),
sizeof(o->od_svname));
if (cplen >= sizeof(o->od_svname)) {
rc = -E2BIG;
- GOTO(out_mnt, rc);
+ GOTO(out_capa, rc);
}
if (server_name_is_ost(o->od_svname))
o->od_is_ost = 1;
+ o->od_full_scrub_ratio = OFSR_DEFAULT;
+ o->od_full_scrub_threshold_rate = FULL_SCRUB_THRESHOLD_RATE_DEFAULT;
+ rc = osd_mount(env, o, cfg);
+ if (rc != 0)
+ GOTO(out_capa, rc);
+
rc = osd_obj_map_init(env, o);
if (rc != 0)
GOTO(out_mnt, rc);
if (rc != 0)
GOTO(out_site, rc);
- CFS_INIT_LIST_HEAD(&o->od_ios_list);
+ INIT_LIST_HEAD(&o->od_ios_list);
/* setup scrub, including OI files initialization */
rc = osd_scrub_setup(env, o);
if (rc < 0)
GOTO(out_scrub, rc);
}
- LASSERT(l->ld_site->ls_linkage.next && l->ld_site->ls_linkage.prev);
+ LASSERT(l->ld_site->ls_linkage.next != NULL);
+ LASSERT(l->ld_site->ls_linkage.prev != NULL);
/* initialize quota slave instance */
o->od_quota_slave = qsd_init(env, o->od_svname, &o->od_dt_dev,
static int osd_process_config(const struct lu_env *env,
struct lu_device *d, struct lustre_cfg *cfg)
{
- struct osd_device *o = osd_dev(d);
- int err;
- ENTRY;
+ struct osd_device *o = osd_dev(d);
+ int rc;
+ ENTRY;
- switch(cfg->lcfg_command) {
- case LCFG_SETUP:
- err = osd_mount(env, o, cfg);
- break;
- case LCFG_CLEANUP:
+ switch (cfg->lcfg_command) {
+ case LCFG_SETUP:
+ rc = osd_mount(env, o, cfg);
+ break;
+ case LCFG_CLEANUP:
lu_dev_del_linkage(d->ld_site, d);
- err = osd_shutdown(env, o);
+ rc = osd_shutdown(env, o);
break;
- default:
- err = -ENOSYS;
- }
+ case LCFG_PARAM:
+ LASSERT(&o->od_dt_dev);
+ rc = class_process_proc_param(PARAM_OSD, lprocfs_osd_obd_vars,
+ cfg, &o->od_dt_dev);
+ if (rc > 0 || rc == -ENOSYS)
+ rc = class_process_proc_param(PARAM_OST,
+ lprocfs_osd_obd_vars,
+ cfg, &o->od_dt_dev);
+ break;
+ default:
+ rc = -ENOSYS;
+ }
- RETURN(err);
+ RETURN(rc);
}
static int osd_recovery_complete(const struct lu_env *env,
}
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)
+ if (osd->od_quota_slave != NULL) {
/* set up quota slave objects */
result = qsd_prepare(env, osd->od_quota_slave);
+ if (result != 0)
+ 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);
}
+static int osd_fid_alloc(const struct lu_env *env, struct obd_export *exp,
+ struct lu_fid *fid, struct md_op_data *op_data)
+{
+ struct osd_device *osd = osd_dev(exp->exp_obd->obd_lu_dev);
+
+ return seq_client_alloc_fid(env, osd->od_cl_seq, fid);
+}
+
static const struct lu_object_operations osd_lu_obj_ops = {
.loo_object_init = osd_object_init,
.loo_object_delete = osd_object_delete,
.ldto_device_fini = osd_device_fini
};
-struct lu_device_type osd_device_type = {
+static struct lu_device_type osd_device_type = {
.ldt_tags = LU_DEVICE_DT,
.ldt_name = LUSTRE_OSD_LDISKFS_NAME,
.ldt_ops = &osd_device_type_ops,
.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.
*/
static struct obd_ops osd_obd_device_ops = {
.o_owner = THIS_MODULE,
.o_connect = osd_obd_connect,
- .o_disconnect = osd_obd_disconnect
+ .o_disconnect = osd_obd_disconnect,
+ .o_fid_alloc = osd_fid_alloc,
+ .o_health_check = osd_health_check,
};
static int __init osd_mod_init(void)
{
- struct lprocfs_static_vars lvars;
int rc;
osd_oi_mod_init();
- lprocfs_osd_init_vars(&lvars);
rc = lu_kmem_init(ldiskfs_caches);
if (rc)
return rc;
- rc = class_register_type(&osd_obd_device_ops, NULL, lvars.module_vars,
+ rc = class_register_type(&osd_obd_device_ops, NULL, true,
+ lprocfs_osd_module_vars,
LUSTRE_OSD_LDISKFS_NAME, &osd_device_type);
if (rc)
lu_kmem_fini(ldiskfs_caches);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff