#define DEBUG_SUBSYSTEM S_OSD
#include <linux/module.h>
+#include <linux/user_namespace.h>
+#ifdef HAVE_UIDGID_HEADER
+# include <linux/uidgid.h>
+#endif
/* LUSTRE_VERSION_CODE */
#include <lustre_ver.h>
/* XATTR_{REPLACE,CREATE} */
#include <linux/xattr.h>
+#include <ldiskfs/ldiskfs.h>
+#include <ldiskfs/xattr.h>
+#undef ENTRY
/*
* struct OBD_{ALLOC,FREE}*()
* OBD_FAIL_CHECK
#include <md_object.h>
#include <lustre_quota.h>
-#include <ldiskfs/xattr.h>
#include <lustre_linkea.h>
int ldiskfs_pdo = 1;
/* 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
}
};
[OSD_OT_WRITE] = OSD_OT_WRITE,
[OSD_OT_INSERT] = OSD_OT_DELETE,
[OSD_OT_DELETE] = OSD_OT_INSERT,
- [OSD_OT_UPDATE] = OSD_OT_MAX,
[OSD_OT_QUOTA] = OSD_OT_MAX,
};
if (rc == -ENODATA && !fid_is_igif(rfid) && osd->od_check_ff) {
fid = &lma->lma_self_fid;
rc = osd_get_idif(info, inode, dentry, fid);
- if ((rc > 0) || (rc == -ENODATA && osd->od_lma_self_repair)) {
+ if ((rc > 0) || (rc == -ENODATA && osd->od_index_in_idif)) {
/* For the given OST-object, if it has neither LMA nor
* FID in XATTR_NAME_FID, then the given FID (which is
* contained in the @obj, from client RPC for locating
* the OST-object) is trusted. We use it to generate
* the LMA. */
osd_lma_self_repair(info, osd, inode, rfid,
- fid_is_on_ost(info, osd, fid, OI_CHECK_FLD) ?
- LMAC_FID_ON_OST : 0);
+ LMAC_FID_ON_OST);
RETURN(0);
}
}
lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
PFID(rfid), inode->i_ino);
rc = -EOPNOTSUPP;
- } else if (!(lma->lma_compat & LMAC_NOT_IN_OI)) {
+ } else {
fid = &lma->lma_self_fid;
}
}
fid_to_ostid(fid, oi);
ostid_to_fid(fid1, oi, idx);
if (lu_fid_eq(fid1, rfid)) {
- if (osd->od_lma_self_repair)
+ if (osd->od_index_in_idif)
osd_lma_self_repair(info, osd,
inode, rfid,
LMAC_FID_ON_OST);
lu_context_exit(&th->th_ctx);
lu_context_fini(&th->th_ctx);
- thandle_put(th);
+ OBD_FREE_PTR(oh);
}
static struct thandle *osd_trans_create(const struct lu_env *env,
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);
struct osd_thandle *oh;
struct osd_thread_info *oti = osd_oti_get(env);
struct osd_iobuf *iobuf = &oti->oti_iobuf;
- struct qsd_instance *qsd = oti->oti_dev->od_quota_slave;
+ struct osd_device *osd = osd_dt_dev(th->th_dev);
+ struct qsd_instance *qsd = osd->od_quota_slave;
struct lquota_trans *qtrans;
ENTRY;
oti->oti_txns--;
rc = dt_txn_hook_stop(env, th);
if (rc != 0)
- CERROR("Failure in transaction hook: %d\n", rc);
+ CERROR("%s: failed in transaction hook: rc = %d\n",
+ osd_name(osd), rc);
/* hook functions might modify th_sync */
hdl->h_sync = th->th_sync;
- oh->ot_handle = NULL;
- OSD_CHECK_SLOW_TH(oh, oti->oti_dev,
- rc = ldiskfs_journal_stop(hdl));
- if (rc != 0)
- CERROR("Failure to stop transaction: %d\n", rc);
- } else {
- thandle_put(&oh->ot_super);
- }
+ oh->ot_handle = NULL;
+ OSD_CHECK_SLOW_TH(oh, osd, rc = ldiskfs_journal_stop(hdl));
+ if (rc != 0)
+ CERROR("%s: failed to stop transaction: rc = %d\n",
+ osd_name(osd), rc);
+ } else {
+ OBD_FREE_PTR(oh);
+ }
/* inform the quota slave device that the transaction is stopping */
qsd_op_end(env, qsd, qtrans);
*/
wait_event(iobuf->dr_wait,
atomic_read(&iobuf->dr_numreqs) == 0);
- osd_fini_iobuf(oti->oti_dev, iobuf);
+ osd_fini_iobuf(osd, iobuf);
if (!rc)
rc = iobuf->dr_error;
RETURN(rc);
}
-/*
- * Concurrency: serialization provided by callers.
- */
-static int osd_init_capa_ctxt(const struct lu_env *env, struct dt_device *d,
- int mode, unsigned long timeout, __u32 alg,
- struct lustre_capa_key *keys)
-{
- struct osd_device *dev = osd_dt_dev(d);
- ENTRY;
-
- dev->od_fl_capa = mode;
- dev->od_capa_timeout = timeout;
- dev->od_capa_alg = alg;
- dev->od_capa_keys = keys;
- RETURN(0);
-}
-
/**
* Note: we do not count into QUOTA here.
* If we mount with --data_journal we may need more.
.dt_sync = osd_sync,
.dt_ro = osd_ro,
.dt_commit_async = osd_commit_async,
- .dt_init_capa_ctxt = osd_init_capa_ctxt,
};
static void osd_object_read_lock(const struct lu_env *env,
return obj->oo_owner == env;
}
-static int capa_is_sane(const struct lu_env *env,
- struct osd_device *dev,
- struct lustre_capa *capa,
- struct lustre_capa_key *keys)
-{
- struct osd_thread_info *oti = osd_oti_get(env);
- struct lustre_capa *tcapa = &oti->oti_capa;
- struct obd_capa *oc;
- int i, rc = 0;
- ENTRY;
-
- oc = capa_lookup(dev->od_capa_hash, capa, 0);
- if (oc) {
- if (capa_is_expired(oc)) {
- DEBUG_CAPA(D_ERROR, capa, "expired");
- rc = -ESTALE;
- }
- capa_put(oc);
- RETURN(rc);
- }
-
- if (capa_is_expired_sec(capa)) {
- DEBUG_CAPA(D_ERROR, capa, "expired");
- RETURN(-ESTALE);
- }
-
- spin_lock(&capa_lock);
- for (i = 0; i < 2; i++) {
- if (keys[i].lk_keyid == capa->lc_keyid) {
- oti->oti_capa_key = keys[i];
- break;
- }
- }
- spin_unlock(&capa_lock);
-
- if (i == 2) {
- DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
- RETURN(-ESTALE);
- }
-
- rc = capa_hmac(tcapa->lc_hmac, capa, oti->oti_capa_key.lk_key);
- if (rc)
- RETURN(rc);
-
- if (memcmp(tcapa->lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac))) {
- DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
- RETURN(-EACCES);
- }
-
- oc = capa_add(dev->od_capa_hash, capa);
- capa_put(oc);
-
- RETURN(0);
-}
-
-int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
- struct lustre_capa *capa, __u64 opc)
-{
- const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
- struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
- struct lu_capainfo *lci;
- int rc;
-
- if (!osd->od_fl_capa)
- return 0;
-
- if (capa == BYPASS_CAPA)
- return 0;
-
- lci = lu_capainfo_get(env);
- if (unlikely(lci == NULL))
- return 0;
-
- if (lci->lci_auth == LC_ID_NONE)
- return 0;
-
- if (capa == NULL) {
- CERROR("%s: no capability provided for FID "DFID": rc = %d\n",
- osd_name(osd), PFID(fid), -EACCES);
- return -EACCES;
- }
-
- if (!lu_fid_eq(fid, &capa->lc_fid)) {
- DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",
- PFID(fid));
- return -EACCES;
- }
-
- if (!capa_opc_supported(capa, opc)) {
- DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
- return -EACCES;
- }
-
- rc = capa_is_sane(env, osd, capa, osd->od_capa_keys);
- if (rc != 0) {
- DEBUG_CAPA(D_ERROR, capa, "insane: rc = %d", rc);
- return -EACCES;
- }
-
- return 0;
-}
-
static struct timespec *osd_inode_time(const struct lu_env *env,
struct inode *inode, __u64 seconds)
{
static int osd_attr_get(const struct lu_env *env,
struct dt_object *dt,
- struct lu_attr *attr,
- struct lustre_capa *capa)
+ struct lu_attr *attr)
{
struct osd_object *obj = osd_dt_obj(dt);
LASSERT(!dt_object_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);
spin_unlock(&obj->oo_guard);
static int osd_attr_set(const struct lu_env *env,
struct dt_object *dt,
const struct lu_attr *attr,
- struct thandle *handle,
- struct lustre_capa *capa)
+ struct thandle *handle)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode;
LASSERT(!dt_object_remote(dt));
LASSERT(osd_invariant(obj));
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
- return -EACCES;
-
osd_trans_exec_op(env, handle, OSD_OT_ATTR_SET);
if (OBD_FAIL_CHECK(OBD_FAIL_OSD_FID_MAPPING)) {
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 {
result = osd_create_type_f(dof->dof_type)(info, obj, attr, hint, dof,
th);
- if (result == 0) {
+ if (likely(obj->oo_inode != NULL)) {
+ LASSERT(obj->oo_inode->i_state & I_NEW);
+
+ /* Unlock the inode before attr initialization to avoid
+ * unnecessary dqget operations. LU-6378 */
+ unlock_new_inode(obj->oo_inode);
+ }
+
+ if (likely(result == 0)) {
osd_attr_init(info, obj, attr, dof);
osd_object_init0(obj);
- /* bz 24037 */
- if (obj->oo_inode && (obj->oo_inode->i_state & I_NEW))
- unlock_new_inode(obj->oo_inode);
- }
+ }
/* restore previous umask value */
current->fs->umask = umask;
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) {
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,
+ 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 read locked.
*/
static int osd_xattr_get(const struct lu_env *env, struct dt_object *dt,
- struct lu_buf *buf, const char *name,
- struct lustre_capa *capa)
+ struct lu_buf *buf, const char *name)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
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);
}
* Concurrency: @dt is write locked.
*/
static int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
- const struct lu_buf *buf, const char *name, int fl,
- struct thandle *handle, struct lustre_capa *capa)
+ const struct lu_buf *buf, const char *name, int fl,
+ struct thandle *handle)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
return sizeof(dt_obj_version_t);
}
- 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);
* Concurrency: @dt is read locked.
*/
static int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
- const struct lu_buf *buf, struct lustre_capa *capa)
+ const struct lu_buf *buf)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
LASSERT(inode->i_op != NULL);
LASSERT(inode->i_op->listxattr != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
- return -EACCES;
-
dentry->d_inode = inode;
dentry->d_sb = inode->i_sb;
return inode->i_op->listxattr(dentry, buf->lb_buf, buf->lb_len);
* Concurrency: @dt is write locked.
*/
static int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
- const char *name, struct thandle *handle,
- struct lustre_capa *capa)
+ const char *name, struct thandle *handle)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *inode = obj->oo_inode;
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);
return rc;
}
-static struct obd_capa *osd_capa_get(const struct lu_env *env,
- struct dt_object *dt,
- struct lustre_capa *old, __u64 opc)
-{
- struct osd_thread_info *info = osd_oti_get(env);
- const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
- struct osd_object *obj = osd_dt_obj(dt);
- struct osd_device *osd = osd_obj2dev(obj);
- struct lustre_capa_key *key = &info->oti_capa_key;
- struct lustre_capa *capa = &info->oti_capa;
- struct obd_capa *oc;
- struct lu_capainfo *lci;
- int rc;
- ENTRY;
-
- if (!osd->od_fl_capa)
- RETURN(ERR_PTR(-ENOENT));
-
- if (!dt_object_exists(dt))
- RETURN(ERR_PTR(-ENOENT));
-
- LASSERT(!dt_object_remote(dt));
- LINVRNT(osd_invariant(obj));
-
- /* renewal sanity check */
- if (old && osd_object_auth(env, dt, old, opc))
- RETURN(ERR_PTR(-EACCES));
-
- lci = lu_capainfo_get(env);
- if (unlikely(lci == NULL))
- RETURN(ERR_PTR(-ENOENT));
-
- switch (lci->lci_auth) {
- case LC_ID_NONE:
- RETURN(NULL);
- case LC_ID_PLAIN:
- capa->lc_uid = i_uid_read(obj->oo_inode);
- capa->lc_gid = i_gid_read(obj->oo_inode);
- capa->lc_flags = LC_ID_PLAIN;
- break;
- case LC_ID_CONVERT: {
- __u32 d[4], s[4];
-
- s[0] = i_uid_read(obj->oo_inode);
- cfs_get_random_bytes(&(s[1]), sizeof(__u32));
- s[2] = i_uid_read(obj->oo_inode);
- cfs_get_random_bytes(&(s[3]), sizeof(__u32));
- rc = capa_encrypt_id(d, s, key->lk_key, CAPA_HMAC_KEY_MAX_LEN);
- if (unlikely(rc))
- RETURN(ERR_PTR(rc));
-
- capa->lc_uid = ((__u64)d[1] << 32) | d[0];
- capa->lc_gid = ((__u64)d[3] << 32) | d[2];
- capa->lc_flags = LC_ID_CONVERT;
- break;
- }
- default:
- RETURN(ERR_PTR(-EINVAL));
- }
-
- capa->lc_fid = *fid;
- capa->lc_opc = opc;
- capa->lc_flags |= osd->od_capa_alg << 24;
- capa->lc_timeout = osd->od_capa_timeout;
- capa->lc_expiry = 0;
-
- oc = capa_lookup(osd->od_capa_hash, capa, 1);
- if (oc) {
- LASSERT(!capa_is_expired(oc));
- RETURN(oc);
- }
-
- spin_lock(&capa_lock);
- *key = osd->od_capa_keys[1];
- spin_unlock(&capa_lock);
-
- capa->lc_keyid = key->lk_keyid;
- capa->lc_expiry = cfs_time_current_sec() + osd->od_capa_timeout;
-
- rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
- if (rc) {
- DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
- RETURN(ERR_PTR(rc));
- }
-
- oc = capa_add(osd->od_capa_hash, capa);
- RETURN(oc);
-}
-
static int osd_object_sync(const struct lu_env *env, struct dt_object *dt,
__u64 start, __u64 end)
{
RETURN(rc);
}
-static int osd_data_get(const struct lu_env *env, struct dt_object *dt,
- void **data)
-{
- struct osd_object *obj = osd_dt_obj(dt);
- ENTRY;
-
- *data = (void *)obj->oo_inode;
- RETURN(0);
-}
-
/*
* Index operations.
*/
static int osd_otable_it_attr_get(const struct lu_env *env,
struct dt_object *dt,
- struct lu_attr *attr,
- struct lustre_capa *capa)
+ struct lu_attr *attr)
{
attr->la_valid = 0;
return 0;
.do_declare_xattr_del = osd_declare_xattr_del,
.do_xattr_del = osd_xattr_del,
.do_xattr_list = osd_xattr_list,
- .do_capa_get = osd_capa_get,
.do_object_sync = osd_object_sync,
- .do_data_get = osd_data_get,
};
/**
.do_declare_xattr_del = osd_declare_xattr_del,
.do_xattr_del = osd_xattr_del,
.do_xattr_list = osd_xattr_list,
- .do_capa_get = osd_capa_get,
.do_object_sync = osd_object_sync,
- .do_data_get = osd_data_get,
};
static const struct dt_object_operations osd_obj_otable_it_ops = {
*/
static int osd_index_iam_delete(const struct lu_env *env, struct dt_object *dt,
const struct dt_key *key,
- struct thandle *handle,
- struct lustre_capa *capa)
+ struct thandle *handle)
{
struct osd_thread_info *oti = osd_oti_get(env);
struct osd_object *obj = osd_dt_obj(dt);
LASSERT(bag->ic_object == obj->oo_inode);
LASSERT(handle != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
- RETURN(-EACCES);
-
osd_trans_exec_op(env, handle, OSD_OT_DELETE);
ipd = osd_idx_ipd_get(env, bag);
* \retval -ve, on error
*/
static int osd_index_ea_delete(const struct lu_env *env, struct dt_object *dt,
- const struct dt_key *key, struct thandle *handle,
- struct lustre_capa *capa)
+ const struct dt_key *key, struct thandle *handle)
{
struct osd_object *obj = osd_dt_obj(dt);
struct inode *dir = obj->oo_inode;
LASSERT(oh->ot_handle != NULL);
LASSERT(oh->ot_handle->h_transaction != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
- RETURN(-EACCES);
-
ll_vfs_dq_init(dir);
dentry = osd_child_dentry_get(env, obj,
(char *)key, strlen((char *)key));
* \retval -ve failure
*/
static int osd_index_iam_lookup(const struct lu_env *env, struct dt_object *dt,
- struct dt_rec *rec, const struct dt_key *key,
- struct lustre_capa *capa)
+ struct dt_rec *rec, const struct dt_key *key)
{
struct osd_object *obj = osd_dt_obj(dt);
struct iam_path_descr *ipd;
LASSERT(!dt_object_remote(dt));
LASSERT(bag->ic_object == obj->oo_inode);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_LOOKUP))
- RETURN(-EACCES);
-
ipd = osd_idx_ipd_get(env, bag);
if (IS_ERR(ipd))
RETURN(-ENOMEM);
static int osd_index_iam_insert(const struct lu_env *env, struct dt_object *dt,
const struct dt_rec *rec,
const struct dt_key *key, struct thandle *th,
- struct lustre_capa *capa, int ignore_quota)
+ int ignore_quota)
{
struct osd_object *obj = osd_dt_obj(dt);
struct iam_path_descr *ipd;
LASSERT(bag->ic_object == obj->oo_inode);
LASSERT(th != NULL);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
- RETURN(-EACCES);
-
osd_trans_exec_op(env, th, OSD_OT_INSERT);
ipd = osd_idx_ipd_get(env, bag);
/* done with de, release bh */
brelse(bh);
if (rc != 0) {
- if (unlikely(ino == osd_remote_parent_ino(dev)))
+ if (unlikely(ino == osd_remote_parent_ino(dev))) {
+ const char *name = (const char *)key;
+
/* If the parent is on remote MDT, and there
* is no FID-in-dirent, then we have to get
* the parent FID from the linkEA. */
- rc = osd_get_pfid_from_linkea(env, obj, fid);
- else
+ if (likely(strlen(name) == 2 &&
+ name[0] == '.' && name[1] == '.'))
+ rc = osd_get_pfid_from_linkea(env, obj,
+ fid);
+ } else {
rc = osd_ea_fid_get(env, obj, ino, fid, id);
+ }
} else {
osd_id_gen(id, ino, OSD_OII_NOGEN);
}
* \retval -ve, on error
*/
static int osd_index_ea_insert(const struct lu_env *env, struct dt_object *dt,
- const struct dt_rec *rec,
- const struct dt_key *key, struct thandle *th,
- struct lustre_capa *capa, int ignore_quota)
+ const struct dt_rec *rec,
+ const struct dt_key *key, struct thandle *th,
+ int ignore_quota)
{
struct osd_object *obj = osd_dt_obj(dt);
struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
osd_trans_exec_op(env, th, OSD_OT_INSERT);
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
- RETURN(-EACCES);
-
LASSERTF(fid_is_sane(fid), "fid"DFID" is insane!\n", PFID(fid));
rc = osd_remote_fid(env, osd, fid);
*/
static struct dt_it *osd_it_iam_init(const struct lu_env *env,
- struct dt_object *dt,
- __u32 unused,
- struct lustre_capa *capa)
+ struct dt_object *dt,
+ __u32 unused)
{
- struct osd_it_iam *it;
- struct osd_thread_info *oti = osd_oti_get(env);
- struct osd_object *obj = osd_dt_obj(dt);
- struct lu_object *lo = &dt->do_lu;
- struct iam_path_descr *ipd;
- struct iam_container *bag = &obj->oo_dir->od_container;
+ 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);
-
- if (oti->oti_it_inline) {
- OBD_ALLOC_PTR(it);
- if (it == NULL)
- return ERR_PTR(-ENOMEM);
- } else {
- it = &oti->oti_it;
- oti->oti_it_inline = 1;
- }
+ OBD_ALLOC_PTR(it);
+ if (it == NULL)
+ return ERR_PTR(-ENOMEM);
ipd = osd_it_ipd_get(env, bag);
if (likely(ipd != NULL)) {
iam_it_init(&it->oi_it, bag, IAM_IT_MOVE, ipd);
return (struct dt_it *)it;
} else {
- if (it != &oti->oti_it)
- OBD_FREE_PTR(it);
- else
- oti->oti_it_inline = 0;
-
+ OBD_FREE_PTR(it);
return ERR_PTR(-ENOMEM);
}
}
static void osd_it_iam_fini(const struct lu_env *env, struct dt_it *di)
{
- struct osd_thread_info *oti = osd_oti_get(env);
struct osd_it_iam *it = (struct osd_it_iam *)di;
struct osd_object *obj = it->oi_obj;
iam_it_fini(&it->oi_it);
osd_ipd_put(env, &obj->oo_dir->od_container, it->oi_ipd);
lu_object_put(env, &obj->oo_dt.do_lu);
- if (it != &oti->oti_it)
- OBD_FREE_PTR(it);
- else
- oti->oti_it_inline = 0;
+ OBD_FREE_PTR(it);
}
/**
*/
static struct dt_it *osd_it_ea_init(const struct lu_env *env,
struct dt_object *dt,
- __u32 attr,
- struct lustre_capa *capa)
+ __u32 attr)
{
struct osd_object *obj = osd_dt_obj(dt);
struct osd_thread_info *info = osd_oti_get(env);
- struct osd_it_ea *it;
+ struct osd_it_ea *oie;
struct file *file;
struct lu_object *lo = &dt->do_lu;
- struct dentry *obj_dentry = &info->oti_it_dentry;
+ struct dentry *obj_dentry;
ENTRY;
if (!dt_object_exists(dt))
RETURN(ERR_PTR(-ENOENT));
- if (info->oti_it_inline) {
- OBD_ALLOC_PTR(it);
- if (it == NULL)
- RETURN(ERR_PTR(-ENOMEM));
- } else {
- it = &info->oti_it_ea;
- info->oti_it_inline = 1;
- }
+ OBD_SLAB_ALLOC_PTR_GFP(oie, osd_itea_cachep, GFP_NOFS);
+ if (oie == NULL)
+ RETURN(ERR_PTR(-ENOMEM));
+ obj_dentry = &oie->oie_dentry;
obj_dentry->d_inode = obj->oo_inode;
obj_dentry->d_sb = osd_sb(osd_obj2dev(obj));
obj_dentry->d_name.hash = 0;
- it->oie_rd_dirent = 0;
- it->oie_it_dirent = 0;
- it->oie_dirent = NULL;
- it->oie_buf = info->oti_it_ea_buf;
- it->oie_obj = obj;
+ oie->oie_rd_dirent = 0;
+ oie->oie_it_dirent = 0;
+ oie->oie_dirent = NULL;
+ if (unlikely(!info->oti_it_ea_buf_used)) {
+ oie->oie_buf = info->oti_it_ea_buf;
+ info->oti_it_ea_buf_used = 1;
+ } else {
+ OBD_ALLOC(oie->oie_buf, OSD_IT_EA_BUFSIZE);
+ if (oie->oie_buf == NULL)
+ RETURN(ERR_PTR(-ENOMEM));
+ }
+ oie->oie_obj = obj;
+
+ file = &oie->oie_file;
- file = &it->oie_file;
- /* Reset the "file" totally to avoid to reuse any old value from
- * former readdir handling, the "file->f_pos" should be zero. */
- memset(file, 0, sizeof(*file));
/* Only FMODE_64BITHASH or FMODE_32BITHASH should be set, NOT both. */
if (attr & LUDA_64BITHASH)
file->f_mode = FMODE_64BITHASH;
set_file_inode(file, obj->oo_inode);
lu_object_get(lo);
- RETURN((struct dt_it *) it);
+ RETURN((struct dt_it *) oie);
}
/**
*/
static void osd_it_ea_fini(const struct lu_env *env, struct dt_it *di)
{
- struct osd_thread_info *info = osd_oti_get(env);
- struct osd_it_ea *it = (struct osd_it_ea *)di;
- struct osd_object *obj = it->oie_obj;
- struct inode *inode = obj->oo_inode;
+ struct osd_thread_info *info = osd_oti_get(env);
+ struct osd_it_ea *oie = (struct osd_it_ea *)di;
+ struct osd_object *obj = oie->oie_obj;
+ struct inode *inode = obj->oo_inode;
- ENTRY;
- it->oie_file.f_op->release(inode, &it->oie_file);
- lu_object_put(env, &obj->oo_dt.do_lu);
- if (it != &info->oti_it_ea)
- OBD_FREE_PTR(it);
+ ENTRY;
+ oie->oie_file.f_op->release(inode, &oie->oie_file);
+ lu_object_put(env, &obj->oo_dt.do_lu);
+ if (unlikely(oie->oie_buf != info->oti_it_ea_buf))
+ OBD_FREE(oie->oie_buf, OSD_IT_EA_BUFSIZE);
else
- info->oti_it_inline = 0;
+ info->oti_it_ea_buf_used = 0;
+ OBD_SLAB_FREE_PTR(oie, osd_itea_cachep);
EXIT;
}
/* If the parent is on remote MDT, and there
* is no FID-in-dirent, then we have to get
* the parent FID from the linkEA. */
- osd_get_pfid_from_linkea(env, obj, fid);
+ if (!fid_is_sane(fid) &&
+ it->oie_dirent->oied_namelen == 2 &&
+ it->oie_dirent->oied_name[0] == '.' &&
+ it->oie_dirent->oied_name[1] == '.')
+ osd_get_pfid_from_linkea(env, obj, fid);
} else {
rc = osd_dirent_check_repair(env, obj, it, fid, id,
&attr);
} else {
attr &= ~LU_DIRENT_ATTRS_MASK;
if (!fid_is_sane(fid)) {
- if (OBD_FAIL_CHECK(OBD_FAIL_FID_LOOKUP) &&
- likely(it->oie_dirent->oied_namelen != 2 ||
- it->oie_dirent->oied_name[0] != '.' ||
- it->oie_dirent->oied_name[1] != '.'))
- RETURN(-ENOENT);
-
- if (unlikely(ino == osd_remote_parent_ino(dev)))
+ if (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. */
- rc = osd_get_pfid_from_linkea(env, obj, fid);
- else
+ 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);
+ }
} else {
osd_id_gen(id, ino, OSD_OII_NOGEN);
}
* \retval -ve, on error
*/
static int osd_index_ea_lookup(const struct lu_env *env, struct dt_object *dt,
- struct dt_rec *rec, const struct dt_key *key,
- struct lustre_capa *capa)
+ struct dt_rec *rec, const struct dt_key *key)
{
struct osd_object *obj = osd_dt_obj(dt);
int rc = 0;
LASSERT(S_ISDIR(obj->oo_inode->i_mode));
LINVRNT(osd_invariant(obj));
- if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_LOOKUP))
- return -EACCES;
-
rc = osd_ea_lookup_rec(env, obj, rec, key);
if (rc == 0)
rc = +1;
{
struct osd_thread_info *info = data;
+ if (info->oti_inode != NULL)
+ OBD_FREE_PTR(info->oti_inode);
if (info->oti_hlock != NULL)
ldiskfs_htree_lock_free(info->oti_hlock);
OBD_FREE(info->oti_it_ea_buf, OSD_IT_EA_BUFSIZE);
struct osd_thread_info *info = osd_oti_get(env);
struct lu_fid *fid = &info->oti_fid;
struct inode *inode;
- int rc = 0;
+ int rc = 0, force_over_128tb = 0;
ENTRY;
if (o->od_mnt != NULL)
RETURN(-E2BIG);
strcpy(o->od_mntdev, dev);
- OBD_PAGE_ALLOC(__page, GFP_IOFS);
- if (__page == NULL)
- GOTO(out, rc = -ENOMEM);
-
str = lustre_cfg_string(cfg, 2);
s_flags = simple_strtoul(str, NULL, 0);
str = strstr(str, ":");
if (str)
lmd_flags = simple_strtoul(str + 1, NULL, 0);
opts = lustre_cfg_string(cfg, 3);
+#ifdef __BIG_ENDIAN
+ if (opts == NULL || strstr(opts, "bigendian_extents") == NULL) {
+ CERROR("%s: device %s extents feature is not guaranteed to "
+ "work on big-endian systems. Use \"bigendian_extents\" "
+ "mount option to override.\n", name, dev);
+ RETURN(-EINVAL);
+ }
+#endif
+ if (opts != NULL && strstr(opts, "force_over_128tb") != NULL)
+ force_over_128tb = 1;
+
+ OBD_PAGE_ALLOC(__page, GFP_IOFS);
+ if (__page == NULL)
+ GOTO(out, rc = -ENOMEM);
page = (unsigned long)page_address(__page);
options = (char *)page;
*options = '\0';
- if (opts == NULL)
- strcat(options, "user_xattr,acl");
- else
+ if (opts != NULL) {
+ /* strip out the options for back compatiblity */
+ static char *sout[] = {
+ "mballoc",
+ "iopen",
+ "noiopen",
+ "iopen_nopriv",
+ "extents",
+ "noextents",
+ /* strip out option we processed in osd */
+ "bigendian_extents",
+ "force_over_128tb",
+ NULL
+ };
strcat(options, opts);
+ for (rc = 0, str = options; sout[rc]; ) {
+ char *op = strstr(str, sout[rc]);
+ if (op == NULL) {
+ rc++;
+ str = options;
+ continue;
+ }
+ if (op == options || *(op - 1) == ',') {
+ str = op + strlen(sout[rc]);
+ if (*str == ',' || *str == '\0') {
+ *str == ',' ? str++ : str;
+ memmove(op, str, strlen(str) + 1);
+ }
+ }
+ for (str = op; *str != ',' && *str != '\0'; str++)
+ ;
+ }
+ } else {
+ strncat(options, "user_xattr,acl", 14);
+ }
/* Glom up mount options */
if (*options != '\0')
GOTO(out, rc);
}
+ if (ldiskfs_blocks_count(LDISKFS_SB(osd_sb(o))->s_es) > (8ULL << 32) &&
+ force_over_128tb == 0) {
+ CERROR("%s: device %s LDISKFS does not support filesystems "
+ "greater than 128TB and can cause data corruption. "
+ "Use \"force_over_128tb\" mount option to override.\n",
+ name, dev);
+ GOTO(out, rc = -EINVAL);
+ }
+
#ifdef HAVE_DEV_SET_RDONLY
if (dev_check_rdonly(o->od_mnt->mnt_sb->s_bdev)) {
CERROR("%s: underlying device %s is marked as read-only. "
spin_lock_init(&o->od_osfs_lock);
mutex_init(&o->od_otable_mutex);
- o->od_capa_hash = init_capa_hash();
- if (o->od_capa_hash == NULL)
- GOTO(out, rc = -ENOMEM);
-
o->od_read_cache = 1;
o->od_writethrough_cache = 1;
o->od_readcache_max_filesize = OSD_MAX_CACHE_SIZE;
sizeof(o->od_svname));
if (cplen >= sizeof(o->od_svname)) {
rc = -E2BIG;
- GOTO(out_capa, rc);
+ GOTO(out, rc);
}
if (server_name_is_ost(o->od_svname))
o->od_full_scrub_threshold_rate = FULL_SCRUB_THRESHOLD_RATE_DEFAULT;
rc = osd_mount(env, o, cfg);
if (rc != 0)
- GOTO(out_capa, rc);
+ GOTO(out, rc);
rc = osd_obj_map_init(env, o);
if (rc != 0)
if (rc != 0)
GOTO(out_site, rc);
- /* self-repair LMA by default */
- o->od_lma_self_repair = 1;
-
INIT_LIST_HEAD(&o->od_ios_list);
/* setup scrub, including OI files initialization */
rc = osd_scrub_setup(env, o);
osd_obj_map_fini(o);
out_mnt:
osd_umount(env, o);
-out_capa:
- cleanup_capa_hash(o->od_capa_hash);
out:
return rc;
}
struct osd_device *o = osd_dev(d);
ENTRY;
- cleanup_capa_hash(o->od_capa_hash);
/* XXX: make osd top device in order to release reference */
d->ld_site->ls_top_dev = d;
lu_site_purge(env, d->ld_site, -1);
}
static int osd_prepare(const struct lu_env *env, struct lu_device *pdev,
- struct lu_device *dev)
+ struct lu_device *dev)
{
- struct osd_device *osd = osd_dev(dev);
- int result = 0;
+ struct osd_device *osd = osd_dev(dev);
+ struct lr_server_data *lsd =
+ &osd->od_dt_dev.dd_lu_dev.ld_site->ls_tgt->lut_lsd;
+ int result = 0;
ENTRY;
if (osd->od_quota_slave != NULL) {
RETURN(result);
}
+ if (lsd->lsd_feature_incompat & OBD_COMPAT_OST) {
+#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(3, 0, 52, 0)
+ if (lsd->lsd_feature_rocompat & OBD_ROCOMPAT_IDX_IN_IDIF) {
+ osd->od_index_in_idif = 1;
+ } else {
+ osd->od_index_in_idif = 0;
+ result = osd_register_proc_index_in_idif(osd);
+ if (result != 0)
+ RETURN(result);
+ }
+#else
+ osd->od_index_in_idif = 1;
+#endif
+ }
+
result = osd_fid_init(env, osd);
RETURN(result);
.ldt_ctx_tags = LCT_LOCAL,
};
+static int osd_health_check(const struct lu_env *env, struct obd_device *obd)
+{
+ struct osd_device *osd = osd_dev(obd->obd_lu_dev);
+ struct super_block *sb = osd_sb(osd);
+
+ return (osd->od_mnt == NULL || sb->s_flags & MS_RDONLY);
+}
+
/*
* lprocfs legacy support.
*/
.o_connect = osd_obd_connect,
.o_disconnect = osd_obd_disconnect,
.o_fid_alloc = osd_fid_alloc,
+ .o_health_check = osd_health_check,
};
static int __init osd_mod_init(void)
{
int rc;
+#if !defined(CONFIG_DEBUG_MUTEXES) && !defined(CONFIG_DEBUG_SPINLOCK)
+ /* please, try to keep osd_thread_info smaller than a page */
+ CLASSERT(sizeof(struct osd_thread_info) <= PAGE_SIZE);
+#endif
+
osd_oi_mod_init();
rc = lu_kmem_init(ldiskfs_caches);
MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
MODULE_DESCRIPTION("Lustre Object Storage Device ("LUSTRE_OSD_LDISKFS_NAME")");
+MODULE_VERSION(LUSTRE_VERSION_STRING);
MODULE_LICENSE("GPL");
-cfs_module(osd, "0.1.0", osd_mod_init, osd_mod_exit);
+module_init(osd_mod_init);
+module_exit(osd_mod_exit);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff