X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fosd-ldiskfs%2Fosd_oi.c;h=118359a4e88441461b9fb2aad41820222457bf9c;hb=92e7ffbdb01f680ed9f299ed109f2189f144a31d;hp=bcad98e98c1e6579f63165406b7779c8a2bb3cac;hpb=4db03b974810789133e354b08029f0eb261fad5e;p=fs%2Flustre-release.git diff --git a/lustre/osd-ldiskfs/osd_oi.c b/lustre/osd-ldiskfs/osd_oi.c index bcad98e..118359a 100644 --- a/lustre/osd-ldiskfs/osd_oi.c +++ b/lustre/osd-ldiskfs/osd_oi.c @@ -1,6 +1,4 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: - * +/* * GPL HEADER START * * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. @@ -28,6 +26,8 @@ /* * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. + * + * Copyright (c) 2011, 2012, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -40,18 +40,6 @@ * Author: Nikita Danilov */ -/* - * oi uses two mechanisms to implement fid->cookie mapping: - * - * - persistent index, where cookie is a record and fid is a key, and - * - * - algorithmic mapping for "igif" fids. - * - */ - -#ifndef EXPORT_SYMTAB -# define EXPORT_SYMTAB -#endif #define DEBUG_SUBSYSTEM S_MDS #include @@ -67,23 +55,20 @@ /* fid_cpu_to_be() */ #include +#include #include "osd_oi.h" /* osd_lookup(), struct osd_thread_info */ #include "osd_internal.h" -#include "osd_igif.h" -#include "dt_object.h" +#include "osd_scrub.h" -#define OSD_OI_FID_NR (1UL << OSD_OI_FID_OID_BITS) -#define OSD_OI_FID_NR_MAX (1UL << OSD_OI_FID_OID_BITS_MAX) - -static unsigned int osd_oi_num = OSD_OI_FID_NR; -CFS_MODULE_PARM(osd_oi_num, "i", int, 0444, +static unsigned int osd_oi_count = OSD_OI_FID_NR; +CFS_MODULE_PARM(osd_oi_count, "i", int, 0444, "Number of Object Index containers to be created, " "it's only valid for new filesystem."); /** to serialize concurrent OI index initialization */ -static cfs_mutex_t oi_init_lock; +static struct mutex oi_init_lock; static struct dt_index_features oi_feat = { .dif_flags = DT_IND_UPDATE, @@ -94,6 +79,128 @@ static struct dt_index_features oi_feat = { #define OSD_OI_NAME_BASE "oi.16" +static void osd_oi_table_put(struct osd_thread_info *info, + struct osd_oi **oi_table, unsigned oi_count) +{ + struct iam_container *bag; + int i; + + for (i = 0; i < oi_count; i++) { + if (oi_table[i] == NULL) + continue; + + LASSERT(oi_table[i]->oi_inode != NULL); + + bag = &(oi_table[i]->oi_dir.od_container); + if (bag->ic_object == oi_table[i]->oi_inode) + iam_container_fini(bag); + iput(oi_table[i]->oi_inode); + oi_table[i]->oi_inode = NULL; + OBD_FREE_PTR(oi_table[i]); + oi_table[i] = NULL; + } +} + +static int osd_oi_index_create_one(struct osd_thread_info *info, + struct osd_device *osd, const char *name, + struct dt_index_features *feat) +{ + const struct lu_env *env = info->oti_env; + struct osd_inode_id *id = &info->oti_id; + struct buffer_head *bh; + struct inode *inode; + struct ldiskfs_dir_entry_2 *de; + struct dentry *dentry; + struct super_block *sb = osd_sb(osd); + struct inode *dir = sb->s_root->d_inode; + handle_t *jh; + int rc; + + dentry = osd_child_dentry_by_inode(env, dir, name, strlen(name)); + bh = osd_ldiskfs_find_entry(dir, dentry, &de, NULL); + if (bh) { + osd_id_gen(id, le32_to_cpu(de->inode), OSD_OII_NOGEN); + brelse(bh); + inode = osd_iget(info, osd, id); + if (!IS_ERR(inode)) { + iput(inode); + inode = ERR_PTR(-EEXIST); + } + return PTR_ERR(inode); + } + + jh = ldiskfs_journal_start_sb(sb, 100); + if (IS_ERR(jh)) + return PTR_ERR(jh); + + inode = ldiskfs_create_inode(jh, dir, (S_IFREG | S_IRUGO | S_IWUSR)); + if (IS_ERR(inode)) { + ldiskfs_journal_stop(jh); + return PTR_ERR(inode); + } + + if (feat->dif_flags & DT_IND_VARKEY) + rc = iam_lvar_create(inode, feat->dif_keysize_max, + feat->dif_ptrsize, feat->dif_recsize_max, + jh); + else + rc = iam_lfix_create(inode, feat->dif_keysize_max, + feat->dif_ptrsize, feat->dif_recsize_max, + jh); + dentry = osd_child_dentry_by_inode(env, dir, name, strlen(name)); + rc = osd_ldiskfs_add_entry(jh, dentry, inode, NULL); + ldiskfs_journal_stop(jh); + iput(inode); + return rc; +} + +static struct inode *osd_oi_index_open(struct osd_thread_info *info, + struct osd_device *osd, + const char *name, + struct dt_index_features *f, + bool create) +{ + struct dentry *dentry; + struct inode *inode; + int rc; + + dentry = ll_lookup_one_len(name, osd_sb(osd)->s_root, strlen(name)); + if (IS_ERR(dentry)) + return (void *) dentry; + + if (dentry->d_inode) { + LASSERT(!is_bad_inode(dentry->d_inode)); + inode = dentry->d_inode; + atomic_inc(&inode->i_count); + dput(dentry); + return inode; + } + + /* create */ + dput(dentry); + shrink_dcache_parent(osd_sb(osd)->s_root); + if (!create) + return ERR_PTR(-ENOENT); + + rc = osd_oi_index_create_one(info, osd, name, f); + if (rc) + return ERR_PTR(rc); + + dentry = ll_lookup_one_len(name, osd_sb(osd)->s_root, strlen(name)); + if (IS_ERR(dentry)) + return (void *) dentry; + + if (dentry->d_inode) { + LASSERT(!is_bad_inode(dentry->d_inode)); + inode = dentry->d_inode; + atomic_inc(&inode->i_count); + dput(dentry); + return inode; + } + + return ERR_PTR(-ENOENT); +} + /** * Open an OI(Ojbect Index) container. * @@ -103,46 +210,51 @@ static struct dt_index_features oi_feat = { * \retval 0 success * \retval -ve failure */ -static int -osd_oi_open(struct osd_thread_info *info, - struct dt_device *dev, char *name, struct dt_object **objp) +static int osd_oi_open(struct osd_thread_info *info, struct osd_device *osd, + char *name, struct osd_oi **oi_slot, bool create) { - const struct lu_env *env = info->oti_env; - struct dt_object *obj; - int rc; - - obj = dt_store_open(env, dev, "", name, &info->oti_fid); - if (IS_ERR(obj)) - return PTR_ERR(obj); - - oi_feat.dif_keysize_min = sizeof(info->oti_fid); - oi_feat.dif_keysize_max = sizeof(info->oti_fid); - - rc = obj->do_ops->do_index_try(env, obj, &oi_feat); - if (rc != 0) { - lu_object_put(info->oti_env, &obj->do_lu); - CERROR("%s: wrong index %s: rc = %d\n", - dev->dd_lu_dev.ld_obd->obd_name, name, rc); - return rc; - } + struct osd_directory *dir; + struct iam_container *bag; + struct inode *inode; + struct osd_oi *oi; + int rc; - *objp = obj; - return 0; -} + ENTRY; + oi_feat.dif_keysize_min = sizeof(struct lu_fid); + oi_feat.dif_keysize_max = sizeof(struct lu_fid); -static void -osd_oi_table_put(struct osd_thread_info *info, - struct osd_oi *oi_table, unsigned oi_count) -{ - int i; + inode = osd_oi_index_open(info, osd, name, &oi_feat, create); + if (IS_ERR(inode)) + RETURN(PTR_ERR(inode)); - for (i = 0; i < oi_count; i++) { - LASSERT(oi_table[i].oi_dir != NULL); + ldiskfs_set_inode_state(inode, LDISKFS_STATE_LUSTRE_NO_OI); + OBD_ALLOC_PTR(oi); + if (oi == NULL) + GOTO(out_inode, rc = -ENOMEM); - lu_object_put(info->oti_env, &oi_table[i].oi_dir->do_lu); - oi_table[i].oi_dir = NULL; - } + oi->oi_inode = inode; + dir = &oi->oi_dir; + + bag = &dir->od_container; + rc = iam_container_init(bag, &dir->od_descr, inode); + if (rc < 0) + GOTO(out_free, rc); + + rc = iam_container_setup(bag); + if (rc < 0) + GOTO(out_container, rc); + + *oi_slot = oi; + RETURN(0); + +out_container: + iam_container_fini(bag); +out_free: + OBD_FREE_PTR(oi); +out_inode: + iput(inode); + return rc; } /** @@ -156,233 +268,433 @@ osd_oi_table_put(struct osd_thread_info *info, * success, or error code in failure. * * \param oi_count Number of expected OI containers - * \param try_all Try to open all OIs even see failures + * \param create Create OIs if doesn't exist * * \retval +ve number of opened OI containers * \retval 0 no OI containers found * \retval -ve failure */ static int -osd_oi_table_open(struct osd_thread_info *info, struct dt_device *dev, - struct osd_oi *oi_table, unsigned oi_count, int try_all) +osd_oi_table_open(struct osd_thread_info *info, struct osd_device *osd, + struct osd_oi **oi_table, unsigned oi_count, bool create) { - int count = 0; - int rc = 0; - int i; + struct scrub_file *sf = &osd->od_scrub.os_file; + int count = 0; + int rc = 0; + int i; + ENTRY; + + /* NB: oi_count != 0 means that we have already created/known all OIs + * and have known exact number of OIs. */ + LASSERT(oi_count <= OSD_OI_FID_NR_MAX); + + for (i = 0; i < (oi_count != 0 ? oi_count : OSD_OI_FID_NR_MAX); i++) { + char name[12]; + + if (oi_table[i] != NULL) { + count++; + continue; + } + + sprintf(name, "%s.%d", OSD_OI_NAME_BASE, i); + rc = osd_oi_open(info, osd, name, &oi_table[i], create); + if (rc == 0) { + count++; + continue; + } + + if (rc == -ENOENT && create == false) { + if (oi_count == 0) + return count; + + rc = 0; + ldiskfs_set_bit(i, sf->sf_oi_bitmap); + continue; + } + + CERROR("%.16s: can't open %s: rc = %d\n", + LDISKFS_SB(osd_sb(osd))->s_es->s_volume_name, name, rc); + if (oi_count > 0) + CERROR("%.16s: expect to open total %d OI files.\n", + LDISKFS_SB(osd_sb(osd))->s_es->s_volume_name, + oi_count); + break; + } + + if (rc < 0) { + osd_oi_table_put(info, oi_table, oi_count > 0 ? oi_count : i); + count = rc; + } + + RETURN(count); +} - /* NB: oi_count != 0 means that we have already created/known all OIs - * and have known exact number of OIs. */ - LASSERT(oi_count <= OSD_OI_FID_NR_MAX); +int osd_oi_init(struct osd_thread_info *info, struct osd_device *osd) +{ + struct osd_scrub *scrub = &osd->od_scrub; + struct scrub_file *sf = &scrub->os_file; + struct osd_oi **oi; + int rc; + ENTRY; + + OBD_ALLOC(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX); + if (oi == NULL) + RETURN(-ENOMEM); + + mutex_lock(&oi_init_lock); + /* try to open existing multiple OIs first */ + rc = osd_oi_table_open(info, osd, oi, sf->sf_oi_count, false); + if (rc < 0) + GOTO(out, rc); + + if (rc > 0) { + if (rc == sf->sf_oi_count || sf->sf_oi_count == 0) + GOTO(out, rc); + + osd_scrub_file_reset(scrub, + LDISKFS_SB(osd_sb(osd))->s_es->s_uuid, + SF_RECREATED); + osd_oi_count = sf->sf_oi_count; + goto create; + } + + /* if previous failed then try found single OI from old filesystem */ + rc = osd_oi_open(info, osd, OSD_OI_NAME_BASE, &oi[0], false); + if (rc == 0) { /* found single OI from old filesystem */ + GOTO(out, rc = 1); + } else if (rc != -ENOENT) { + CERROR("%.16s: can't open %s: rc = %d\n", + LDISKFS_SB(osd_sb(osd))->s_es->s_volume_name, + OSD_OI_NAME_BASE, rc); + GOTO(out, rc); + } + + if (sf->sf_oi_count > 0) { + int i; + + memset(sf->sf_oi_bitmap, 0, SCRUB_OI_BITMAP_SIZE); + for (i = 0; i < osd_oi_count; i++) + ldiskfs_set_bit(i, sf->sf_oi_bitmap); + osd_scrub_file_reset(scrub, + LDISKFS_SB(osd_sb(osd))->s_es->s_uuid, + SF_RECREATED); + } + sf->sf_oi_count = osd_oi_count; + +create: + rc = osd_scrub_file_store(scrub); + if (rc < 0) { + osd_oi_table_put(info, oi, sf->sf_oi_count); + GOTO(out, rc); + } + + /* No OIs exist, new filesystem, create OI objects */ + rc = osd_oi_table_open(info, osd, oi, osd_oi_count, true); + LASSERT(ergo(rc >= 0, rc == osd_oi_count)); + + GOTO(out, rc); + +out: + if (rc < 0) { + OBD_FREE(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX); + } else { + LASSERT((rc & (rc - 1)) == 0); + osd->od_oi_table = oi; + osd->od_oi_count = rc; + rc = 0; + } + + mutex_unlock(&oi_init_lock); + return rc; +} - for (i = 0; i < (oi_count != 0 ? oi_count : OSD_OI_FID_NR_MAX); i++) { - char name[12]; +void osd_oi_fini(struct osd_thread_info *info, struct osd_device *osd) +{ + if (unlikely(osd->od_oi_table == NULL)) + return; - sprintf(name, "%s.%d", OSD_OI_NAME_BASE, i); - rc = osd_oi_open(info, dev, name, &oi_table[i].oi_dir); - if (rc == 0) { - count++; - continue; - } + osd_oi_table_put(info, osd->od_oi_table, osd->od_oi_count); - if (try_all) - continue; + OBD_FREE(osd->od_oi_table, + sizeof(*(osd->od_oi_table)) * OSD_OI_FID_NR_MAX); + osd->od_oi_table = NULL; +} - if (rc == -ENOENT && oi_count == 0) - return count; +static inline int fid_is_fs_root(const struct lu_fid *fid) +{ + /* Map root inode to special local object FID */ + return (unlikely(fid_seq(fid) == FID_SEQ_LOCAL_FILE && + fid_oid(fid) == OSD_FS_ROOT_OID)); +} - CERROR("%s: can't open %s: rc = %d\n", - dev->dd_lu_dev.ld_obd->obd_name, name, rc); +static int osd_oi_iam_lookup(struct osd_thread_info *oti, + struct osd_oi *oi, struct dt_rec *rec, + const struct dt_key *key) +{ + struct iam_container *bag; + struct iam_iterator *it = &oti->oti_idx_it; + struct iam_path_descr *ipd; + int rc; + ENTRY; - if (oi_count > 0) { - CERROR("%s: expect to open total %d OI files.\n", - dev->dd_lu_dev.ld_obd->obd_name, oi_count); - } + LASSERT(oi); + LASSERT(oi->oi_inode); - break; - } + bag = &oi->oi_dir.od_container; + ipd = osd_idx_ipd_get(oti->oti_env, bag); + if (IS_ERR(ipd)) + RETURN(-ENOMEM); - if (try_all) - return count; + /* got ipd now we can start iterator. */ + iam_it_init(it, bag, 0, ipd); - if (rc < 0) { - osd_oi_table_put(info, oi_table, count); - return rc; - } + rc = iam_it_get(it, (struct iam_key *)key); + if (rc > 0) + iam_reccpy(&it->ii_path.ip_leaf, (struct iam_rec *)rec); + iam_it_put(it); + iam_it_fini(it); + osd_ipd_put(oti->oti_env, bag, ipd); - return count; -} + LINVRNT(osd_invariant(obj)); -static int osd_oi_table_create(struct osd_thread_info *info, - struct dt_device *dev, - struct md_device *mdev, int oi_count) -{ - const struct lu_env *env; - struct md_object *mdo; - int i; - - env = info->oti_env; - for (i = 0; i < oi_count; ++i) { - char name[12]; - - sprintf(name, "%s.%d", OSD_OI_NAME_BASE, i); - - lu_local_obj_fid(&info->oti_fid, OSD_OI_FID_OID_FIRST + i); - oi_feat.dif_keysize_min = sizeof(info->oti_fid); - oi_feat.dif_keysize_max = sizeof(info->oti_fid); - - mdo = llo_store_create_index(env, mdev, dev, "", name, - &info->oti_fid, &oi_feat); - if (IS_ERR(mdo)) { - CERROR("Failed to create OI[%d] on %s: %d\n", - i, dev->dd_lu_dev.ld_obd->obd_name, - (int)PTR_ERR(mdo)); - RETURN(PTR_ERR(mdo)); - } - - lu_object_put(env, &mdo->mo_lu); - } - return 0; + RETURN(rc); } -int osd_oi_init(struct osd_thread_info *info, - struct osd_oi **oi_table, - struct dt_device *dev, - struct md_device *mdev) +int fid_is_on_ost(struct osd_thread_info *info, struct osd_device *osd, + const struct lu_fid *fid) { - struct osd_oi *oi; - int rc; - - OBD_ALLOC(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX); - if (oi == NULL) - return -ENOMEM; - - cfs_mutex_lock(&oi_init_lock); + struct lu_seq_range *range = &info->oti_seq_range; + int rc; + ENTRY; - rc = osd_oi_table_open(info, dev, oi, 0, 0); - if (rc != 0) - goto out; + if (unlikely(fid_is_local_file(fid) || fid_is_igif(fid))) + RETURN(0); - rc = osd_oi_open(info, dev, OSD_OI_NAME_BASE, &oi[0].oi_dir); - if (rc == 0) { /* found single OI from old filesystem */ - rc = 1; - goto out; - } + if (fid_is_idif(fid) || fid_is_last_id(fid)) + RETURN(1); - if (rc != -ENOENT) { - CERROR("%s: can't open %s: rc = %d\n", - dev->dd_lu_dev.ld_obd->obd_name, OSD_OI_NAME_BASE, rc); - goto out; - } + rc = osd_fld_lookup(info->oti_env, osd, fid, range); + if (rc != 0) { + CERROR("%s: Can not lookup fld for "DFID"\n", + osd2lu_dev(osd)->ld_obd->obd_name, PFID(fid)); + RETURN(rc); + } - /* create OI objects */ - rc = osd_oi_table_create(info, dev, mdev, osd_oi_num); - if (rc != 0) - goto out; + CDEBUG(D_INFO, "fid "DFID" range "DRANGE"\n", PFID(fid), + PRANGE(range)); - rc = osd_oi_table_open(info, dev, oi, osd_oi_num, 0); - LASSERT(rc == osd_oi_num || rc < 0); + if (range->lsr_flags == LU_SEQ_RANGE_OST) + RETURN(1); - out: - if (rc < 0) - OBD_FREE(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX); - else - *oi_table = oi; + RETURN(0); +} - cfs_mutex_unlock(&oi_init_lock); - return rc; +int __osd_oi_lookup(struct osd_thread_info *info, struct osd_device *osd, + const struct lu_fid *fid, struct osd_inode_id *id) +{ + struct lu_fid *oi_fid = &info->oti_fid2; + int rc; + + fid_cpu_to_be(oi_fid, fid); + rc = osd_oi_iam_lookup(info, osd_fid2oi(osd, fid), (struct dt_rec *)id, + (const struct dt_key *)oi_fid); + if (rc > 0) { + osd_id_unpack(id, id); + rc = 0; + } else if (rc == 0) { + rc = -ENOENT; + } + return rc; } -void osd_oi_fini(struct osd_thread_info *info, - struct osd_oi **oi_table, unsigned oi_count) +int osd_oi_lookup(struct osd_thread_info *info, struct osd_device *osd, + const struct lu_fid *fid, struct osd_inode_id *id, + bool check_fld) { - struct osd_oi *oi = *oi_table; + if (unlikely(fid_is_last_id(fid))) + return osd_obj_spec_lookup(info, osd, fid, id); + + if ((check_fld && fid_is_on_ost(info, osd, fid)) || fid_is_llog(fid)) + return osd_obj_map_lookup(info, osd, fid, id); - osd_oi_table_put(info, oi, oi_count); + if (fid_is_fs_root(fid)) { + osd_id_gen(id, osd_sb(osd)->s_root->d_inode->i_ino, + osd_sb(osd)->s_root->d_inode->i_generation); + return 0; + } - OBD_FREE(oi, sizeof(*oi) * OSD_OI_FID_NR_MAX); - *oi_table = NULL; + if (unlikely(fid_is_acct(fid))) + return osd_acct_obj_lookup(info, osd, fid, id); + + return __osd_oi_lookup(info, osd, fid, id); } -int osd_oi_lookup(struct osd_thread_info *info, struct osd_oi *oi, - const struct lu_fid *fid, struct osd_inode_id *id) +static int osd_oi_iam_refresh(struct osd_thread_info *oti, struct osd_oi *oi, + const struct dt_rec *rec, const struct dt_key *key, + struct thandle *th, bool insert) { - struct lu_fid *oi_fid = &info->oti_fid; - int rc; - - if (osd_fid_is_igif(fid)) { - lu_igif_to_id(fid, id); - rc = 0; - } else { - struct dt_object *idx; - const struct dt_key *key; - - if (!fid_is_norm(fid)) - return -ENOENT; - - idx = oi->oi_dir; - fid_cpu_to_be(oi_fid, fid); - key = (struct dt_key *) oi_fid; - rc = idx->do_index_ops->dio_lookup(info->oti_env, idx, - (struct dt_rec *)id, key, - BYPASS_CAPA); - if (rc > 0) { - id->oii_ino = be32_to_cpu(id->oii_ino); - id->oii_gen = be32_to_cpu(id->oii_gen); - rc = 0; - } else if (rc == 0) - rc = -ENOENT; - } - return rc; + struct iam_container *bag; + struct iam_path_descr *ipd; + struct osd_thandle *oh; + int rc; + ENTRY; + + LASSERT(oi); + LASSERT(oi->oi_inode); + ll_vfs_dq_init(oi->oi_inode); + + bag = &oi->oi_dir.od_container; + ipd = osd_idx_ipd_get(oti->oti_env, bag); + if (unlikely(ipd == NULL)) + RETURN(-ENOMEM); + + oh = container_of0(th, struct osd_thandle, ot_super); + LASSERT(oh->ot_handle != NULL); + LASSERT(oh->ot_handle->h_transaction != NULL); + if (insert) + rc = iam_insert(oh->ot_handle, bag, (const struct iam_key *)key, + (const struct iam_rec *)rec, ipd); + else + rc = iam_update(oh->ot_handle, bag, (const struct iam_key *)key, + (const struct iam_rec *)rec, ipd); + osd_ipd_put(oti->oti_env, bag, ipd); + LINVRNT(osd_invariant(obj)); + RETURN(rc); } -int osd_oi_insert(struct osd_thread_info *info, struct osd_oi *oi, - const struct lu_fid *fid, const struct osd_inode_id *id0, - struct thandle *th, int ignore_quota) +int osd_oi_insert(struct osd_thread_info *info, struct osd_device *osd, + const struct lu_fid *fid, const struct osd_inode_id *id, + struct thandle *th) { - struct lu_fid *oi_fid = &info->oti_fid; - struct dt_object *idx; - struct osd_inode_id *id; - const struct dt_key *key; - - if (!fid_is_norm(fid)) - return 0; - - idx = oi->oi_dir; - fid_cpu_to_be(oi_fid, fid); - key = (struct dt_key *) oi_fid; - - id = &info->oti_id; - id->oii_ino = cpu_to_be32(id0->oii_ino); - id->oii_gen = cpu_to_be32(id0->oii_gen); - return idx->do_index_ops->dio_insert(info->oti_env, idx, - (struct dt_rec *)id, - key, th, BYPASS_CAPA, - ignore_quota); + struct lu_fid *oi_fid = &info->oti_fid2; + struct osd_inode_id *oi_id = &info->oti_id2; + int rc = 0; + + if (unlikely(fid_is_last_id(fid))) + return osd_obj_spec_insert(info, osd, fid, id, th); + + if (fid_is_on_ost(info, osd, fid) || fid_is_llog(fid)) + return osd_obj_map_insert(info, osd, fid, id, th); + + fid_cpu_to_be(oi_fid, fid); + osd_id_pack(oi_id, id); + rc = osd_oi_iam_refresh(info, osd_fid2oi(osd, fid), + (const struct dt_rec *)oi_id, + (const struct dt_key *)oi_fid, th, true); + if (rc != 0) { + struct inode *inode; + struct lustre_mdt_attrs *lma = &info->oti_mdt_attrs; + + if (rc != -EEXIST) + return rc; + + rc = osd_oi_lookup(info, osd, fid, oi_id, false); + if (unlikely(rc != 0)) + return rc; + + if (osd_id_eq(id, oi_id)) { + CERROR("%.16s: the FID "DFID" is there already:%u/%u\n", + LDISKFS_SB(osd_sb(osd))->s_es->s_volume_name, + PFID(fid), id->oii_ino, id->oii_gen); + return -EEXIST; + } + + /* Check whether the mapping for oi_id is valid or not. */ + inode = osd_iget(info, osd, oi_id); + if (IS_ERR(inode)) { + rc = PTR_ERR(inode); + if (rc == -ENOENT || rc == -ESTALE) + goto update; + return rc; + } + + rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma); + iput(inode); + if (rc == -ENODATA) + goto update; + + if (rc != 0) + return rc; + + if (lu_fid_eq(fid, &lma->lma_self_fid)) { + CERROR("%.16s: the FID "DFID" is used by two objects: " + "%u/%u %u/%u\n", + LDISKFS_SB(osd_sb(osd))->s_es->s_volume_name, + PFID(fid), oi_id->oii_ino, oi_id->oii_gen, + id->oii_ino, id->oii_gen); + return -EEXIST; + } + +update: + osd_id_pack(oi_id, id); + rc = osd_oi_iam_refresh(info, osd_fid2oi(osd, fid), + (const struct dt_rec *)oi_id, + (const struct dt_key *)oi_fid, th, false); + if (rc != 0) + return rc; + } + + if (unlikely(fid_seq(fid) == FID_SEQ_LOCAL_FILE)) + rc = osd_obj_spec_insert(info, osd, fid, id, th); + return rc; +} + +static int osd_oi_iam_delete(struct osd_thread_info *oti, struct osd_oi *oi, + const struct dt_key *key, struct thandle *handle) +{ + struct iam_container *bag; + struct iam_path_descr *ipd; + struct osd_thandle *oh; + int rc; + ENTRY; + + LASSERT(oi); + LASSERT(oi->oi_inode); + ll_vfs_dq_init(oi->oi_inode); + + bag = &oi->oi_dir.od_container; + ipd = osd_idx_ipd_get(oti->oti_env, bag); + if (unlikely(ipd == NULL)) + RETURN(-ENOMEM); + + oh = container_of0(handle, struct osd_thandle, ot_super); + LASSERT(oh->ot_handle != NULL); + LASSERT(oh->ot_handle->h_transaction != NULL); + + rc = iam_delete(oh->ot_handle, bag, (const struct iam_key *)key, ipd); + osd_ipd_put(oti->oti_env, bag, ipd); + LINVRNT(osd_invariant(obj)); + RETURN(rc); } int osd_oi_delete(struct osd_thread_info *info, - struct osd_oi *oi, const struct lu_fid *fid, - struct thandle *th) + struct osd_device *osd, const struct lu_fid *fid, + struct thandle *th) { - struct lu_fid *oi_fid = &info->oti_fid; - struct dt_object *idx; - const struct dt_key *key; - - if (!fid_is_norm(fid)) - return 0; - - idx = oi->oi_dir; - fid_cpu_to_be(oi_fid, fid); - key = (struct dt_key *) oi_fid; - return idx->do_index_ops->dio_delete(info->oti_env, idx, - key, th, BYPASS_CAPA); + struct lu_fid *oi_fid = &info->oti_fid2; + + if (fid_is_last_id(fid)) + return 0; + + if (fid_is_on_ost(info, osd, fid) || fid_is_llog(fid)) + return osd_obj_map_delete(info, osd, fid, th); + + fid_cpu_to_be(oi_fid, fid); + return osd_oi_iam_delete(info, osd_fid2oi(osd, fid), + (const struct dt_key *)oi_fid, th); } -int osd_oi_mod_init() +int osd_oi_mod_init(void) { - if (osd_oi_num == 0 || osd_oi_num > OSD_OI_FID_NR_MAX) - osd_oi_num = OSD_OI_FID_NR; + if (osd_oi_count == 0 || osd_oi_count > OSD_OI_FID_NR_MAX) + osd_oi_count = OSD_OI_FID_NR; + + if ((osd_oi_count & (osd_oi_count - 1)) != 0) { + LCONSOLE_WARN("Round up oi_count %d to power2 %d\n", + osd_oi_count, size_roundup_power2(osd_oi_count)); + osd_oi_count = size_roundup_power2(osd_oi_count); + } - cfs_mutex_init(&oi_init_lock); + mutex_init(&oi_init_lock); return 0; }