LU-3536 osd: allocate it for each iteration.

Add osd iteration structure(osd_it_ea) to specific SLAB,
and allocate new osd_it_ea for each iteration, so iteration
can be nested, which will help DNE and LFSCK.

Since iteration for iam and quota are not so often,
we just allocate them with normal OBD_ALLOC_PTR.

Signed-off-by: wang di <di.wang@intel.com>
Change-Id: I6402259708264f9341f314e7a2f6afe16cc66481
Reviewed-on: http://review.whamcloud.com/13223
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-by: Fan Yong <fan.yong@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/osd-ldiskfs/osd_handler.c b/lustre/osd-ldiskfs/osd_handler.c
index 39a26cf..dddbba3 100644 (file)
--- a/lustre/osd-ldiskfs/osd_handler.c
+++ b/lustre/osd-ldiskfs/osd_handler.c
@@ -86,6 +86,9 @@ CFS_MODULE_PARM(ldiskfs_track_declares_assert, "i", int, 0644,
 /* 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,
@@ -93,6 +96,11 @@ static struct lu_kmem_descr ldiskfs_caches[] = {
                .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
        }
 };
@@ -4647,27 +4655,21 @@ static struct dt_it *osd_it_iam_init(const struct lu_env *env,
                                      __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;
+       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 (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)) {
@@ -4677,11 +4679,7 @@ static struct dt_it *osd_it_iam_init(const struct lu_env *env,
                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);
        }
 }
@@ -4692,17 +4690,13 @@ static struct dt_it *osd_it_iam_init(const struct lu_env *env,
 
 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);
 }
 
 /**
@@ -4946,38 +4940,39 @@ static struct dt_it *osd_it_ea_init(const struct lu_env *env,
 {
        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;
@@ -4989,7 +4984,7 @@ static struct dt_it *osd_it_ea_init(const struct lu_env *env,
        set_file_inode(file, obj->oo_inode);
 
        lu_object_get(lo);
-       RETURN((struct dt_it *) it);
+       RETURN((struct dt_it *) oie);
 }
 
 /**
@@ -4999,18 +4994,19 @@ static struct dt_it *osd_it_ea_init(const struct lu_env *env,
  */
 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;
 }
 

diff --git a/lustre/osd-ldiskfs/osd_internal.h b/lustre/osd-ldiskfs/osd_internal.h
index 0831074..97337ea 100644 (file)
--- a/lustre/osd-ldiskfs/osd_internal.h
+++ b/lustre/osd-ldiskfs/osd_internal.h
@@ -427,17 +427,18 @@ struct osd_it_ea_dirent {
  * mode (i.e. iterator over ldiskfs style directory)
  */
 struct osd_it_ea {
-        struct osd_object   *oie_obj;
-        /** used in ldiskfs iterator, to stored file pointer */
-        struct file          oie_file;
-        /** how many entries have been read-cached from storage */
-        int                  oie_rd_dirent;
-        /** current entry is being iterated by caller */
-        int                  oie_it_dirent;
-        /** current processing entry */
-        struct osd_it_ea_dirent *oie_dirent;
-        /** buffer to hold entries, size == OSD_IT_EA_BUFSIZE */
-        void                *oie_buf;
+       struct osd_object       *oie_obj;
+       /** used in ldiskfs iterator, to stored file pointer */
+       struct file             oie_file;
+       /** how many entries have been read-cached from storage */
+       int                     oie_rd_dirent;
+       /** current entry is being iterated by caller */
+       int                     oie_it_dirent;
+       /** current processing entry */
+       struct osd_it_ea_dirent *oie_dirent;
+       /** buffer to hold entries, size == OSD_IT_EA_BUFSIZE */
+       void                    *oie_buf;
+       struct dentry           oie_dentry;
 };
 
 /**
@@ -517,10 +518,10 @@ struct osd_thread_info {
          * XXX temporary: for ->i_op calls.
          */
         struct timespec        oti_time;
-        /*
-         * XXX temporary: fake struct file for osd_object_sync
-         */
-        struct file            oti_file;
+       /*
+        * XXX temporary: fake struct file for osd_object_sync
+        */
+       struct file            oti_file;
         /*
          * XXX temporary: for capa operations.
          */
@@ -537,18 +538,9 @@ struct osd_thread_info {
          * in open iterator session.
          */
 
-        /** osd iterator context used for iterator session */
-
-       union {
-               struct osd_it_iam       oti_it;
-               /* ldiskfs iterator data structure,
-                * see osd_it_ea_{init, fini} */
-               struct osd_it_ea        oti_it_ea;
-               struct osd_it_quota     oti_it_quota;
-       };
-
        /** pre-allocated buffer used by oti_it_ea, size OSD_IT_EA_BUFSIZE */
        void                    *oti_it_ea_buf;
+       unsigned int            oti_it_ea_buf_used:1;
 
        struct kstatfs          oti_ksfs;
 
@@ -568,7 +560,6 @@ struct osd_thread_info {
 
        struct osd_idmap_cache oti_cache;
 
-       unsigned int           oti_it_inline:1;
         int                    oti_r_locks;
         int                    oti_w_locks;
         int                    oti_txns;

diff --git a/lustre/osd-ldiskfs/osd_quota.c b/lustre/osd-ldiskfs/osd_quota.c
index 753562d..8b7281e 100644 (file)
--- a/lustre/osd-ldiskfs/osd_quota.c
+++ b/lustre/osd-ldiskfs/osd_quota.c
@@ -154,7 +154,6 @@ static struct dt_it *osd_it_acct_init(const struct lu_env *env,
                                      struct dt_object *dt,
                                      __u32 attr, struct lustre_capa *capa)
 {
-       struct osd_thread_info  *info = osd_oti_get(env);
        struct osd_it_quota     *it;
        struct lu_object        *lo = &dt->do_lu;
        struct osd_object       *obj = osd_dt_obj(dt);
@@ -163,19 +162,10 @@ static struct dt_it *osd_it_acct_init(const struct lu_env *env,
 
        LASSERT(lu_object_exists(lo));
 
-       if (info == NULL)
+       OBD_ALLOC_PTR(it);
+       if (it == NULL)
                RETURN(ERR_PTR(-ENOMEM));
 
-       if (info->oti_it_inline) {
-               OBD_ALLOC_PTR(it);
-               if (it == NULL)
-                       RETURN(ERR_PTR(-ENOMEM));
-       } else {
-               it = &info->oti_it_quota;
-               info->oti_it_inline = 1;
-       }
-
-       memset(it, 0, sizeof(*it));
        lu_object_get(lo);
        it->oiq_obj = obj;
        INIT_LIST_HEAD(&it->oiq_list);
@@ -196,7 +186,6 @@ static struct dt_it *osd_it_acct_init(const struct lu_env *env,
  */
 static void osd_it_acct_fini(const struct lu_env *env, struct dt_it *di)
 {
-       struct osd_thread_info *info = osd_oti_get(env);
        struct osd_it_quota *it = (struct osd_it_quota *)di;
        struct osd_quota_leaf *leaf, *tmp;
        ENTRY;
@@ -208,10 +197,7 @@ static void osd_it_acct_fini(const struct lu_env *env, struct dt_it *di)
                OBD_FREE_PTR(leaf);
        }
 
-       if (it != &info->oti_it_quota)
-               OBD_FREE_PTR(it);
-       else
-               info->oti_it_inline = 0;
+       OBD_FREE_PTR(it);
 
        EXIT;
 }

diff --git a/lustre/osd-ldiskfs/osd_scrub.c b/lustre/osd-ldiskfs/osd_scrub.c
index 4b3e4a1..91ed16a 100644 (file)
--- a/lustre/osd-ldiskfs/osd_scrub.c
+++ b/lustre/osd-ldiskfs/osd_scrub.c
@@ -1966,7 +1966,7 @@ osd_ios_general_scan(struct osd_thread_info *info, struct osd_device *dev,
                                                .oifb_info = info,
                                                .oifb_dev = dev,
                                                .oifb_dentry = dentry };
-       struct file                  *filp  = &info->oti_it_ea.oie_file;
+       struct file                  *filp  = &info->oti_file;
        struct inode                 *inode = dentry->d_inode;
        const struct file_operations *fops  = inode->i_fop;
        int                           rc;

diff --git a/lustre/osd-zfs/osd_handler.c b/lustre/osd-zfs/osd_handler.c
index 62965a7..2ea2f1f 100644 (file)
--- a/lustre/osd-zfs/osd_handler.c
+++ b/lustre/osd-zfs/osd_handler.c
@@ -74,6 +74,9 @@ struct lu_context_key osd_key;
 /* Slab for OSD object allocation */
 struct kmem_cache *osd_object_kmem;
 
+/* Slab to allocate osd_zap_it */
+struct kmem_cache *osd_zapit_cachep;
+
 static struct lu_kmem_descr osd_caches[] = {
        {
                .ckd_cache = &osd_object_kmem,
@@ -81,6 +84,11 @@ static struct lu_kmem_descr osd_caches[] = {
                .ckd_size  = sizeof(struct osd_object)
        },
        {
+               .ckd_cache = &osd_zapit_cachep,
+               .ckd_name  = "osd_zapit_cache",
+               .ckd_size  = sizeof(struct osd_zap_it)
+       },
+       {
                .ckd_cache = NULL
        }
 };

diff --git a/lustre/osd-zfs/osd_index.c b/lustre/osd-zfs/osd_index.c
index b498b23..b28f65f 100644 (file)
--- a/lustre/osd-zfs/osd_index.c
+++ b/lustre/osd-zfs/osd_index.c
@@ -168,22 +168,13 @@ static struct dt_it *osd_index_it_init(const struct lu_env *env,
        LASSERT(osd_object_is_zap(obj->oo_db));
        LASSERT(info);
 
-       if (info->oti_it_inline) {
-               OBD_ALLOC_PTR(it);
-               if (it == NULL)
-                       RETURN(ERR_PTR(-ENOMEM));
-       } else {
-               it = &info->oti_it_zap;
-               info->oti_it_inline = 1;
-       }
+       OBD_SLAB_ALLOC_PTR_GFP(it, osd_zapit_cachep, GFP_NOFS);
+       if (it == NULL)
+               RETURN(ERR_PTR(-ENOMEM));
 
        rc = osd_obj_cursor_init(&it->ozi_zc, obj, 0);
        if (rc != 0) {
-               if (it != &info->oti_it_zap)
-                       OBD_FREE_PTR(it);
-               else
-                       info->oti_it_inline = 0;
-
+               OBD_SLAB_FREE_PTR(it, osd_zapit_cachep);
                RETURN(ERR_PTR(rc));
        }
 
@@ -197,7 +188,6 @@ static struct dt_it *osd_index_it_init(const struct lu_env *env,
 
 static void osd_index_it_fini(const struct lu_env *env, struct dt_it *di)
 {
-       struct osd_thread_info  *info   = osd_oti_get(env);
        struct osd_zap_it       *it     = (struct osd_zap_it *)di;
        struct osd_object       *obj;
        ENTRY;
@@ -209,10 +199,7 @@ static void osd_index_it_fini(const struct lu_env *env, struct dt_it *di)
 
        osd_zap_cursor_fini(it->ozi_zc);
        lu_object_put(env, &obj->oo_dt.do_lu);
-       if (it != &info->oti_it_zap)
-               OBD_FREE_PTR(it);
-       else
-               info->oti_it_inline = 0;
+       OBD_SLAB_FREE_PTR(it, osd_zapit_cachep);
 
        EXIT;
 }

diff --git a/lustre/osd-zfs/osd_internal.h b/lustre/osd-zfs/osd_internal.h
index 3888c5b..c6166f2 100644 (file)
--- a/lustre/osd-zfs/osd_internal.h
+++ b/lustre/osd-zfs/osd_internal.h
@@ -170,12 +170,6 @@ struct osd_thread_info {
 
        char                     oti_buf[64];
 
-       /** osd iterator context used for iterator session */
-       union {
-               struct osd_zap_it       oti_it_zap;
-               struct osd_it_quota     oti_it_quota;
-       };
-
        char                     oti_str[64];
        union {
                char             oti_key[MAXNAMELEN + 1];
@@ -191,7 +185,6 @@ struct osd_thread_info {
 
        struct lquota_id_info    oti_qi;
        struct lu_seq_range      oti_seq_range;
-       unsigned int             oti_it_inline:1;
 };
 
 extern struct lu_context_key osd_key;
@@ -427,6 +420,7 @@ enum {
        LPROC_OSD_LAST,
 };
 
+extern struct kmem_cache *osd_zapit_cachep;
 /* osd_lproc.c */
 extern struct lprocfs_vars lprocfs_osd_obd_vars[];
 

diff --git a/lustre/osd-zfs/osd_quota.c b/lustre/osd-zfs/osd_quota.c
index 3247ddd..3a0fbda 100644 (file)
--- a/lustre/osd-zfs/osd_quota.c
+++ b/lustre/osd-zfs/osd_quota.c
@@ -167,14 +167,9 @@ static struct dt_it *osd_it_acct_init(const struct lu_env *env,
        if (info == NULL)
                RETURN(ERR_PTR(-ENOMEM));
 
-       if (info->oti_it_inline) {
-               OBD_ALLOC_PTR(it);
-               if (it == NULL)
-                       RETURN(ERR_PTR(-ENOMEM));
-       } else {
-               it = &info->oti_it_quota;
-               info->oti_it_inline = 1;
-       }
+       OBD_ALLOC_PTR(it);
+       if (it == NULL)
+               RETURN(ERR_PTR(-ENOMEM));
 
        memset(it, 0, sizeof(*it));
        it->oiq_oid = osd_quota_fid2dmu(lu_object_fid(lo));
@@ -182,11 +177,7 @@ static struct dt_it *osd_it_acct_init(const struct lu_env *env,
        /* initialize zap cursor */
        rc = osd_zap_cursor_init(&it->oiq_zc, osd->od_os, it->oiq_oid, 0);
        if (rc != 0) {
-               if (it != &info->oti_it_quota)
-                       OBD_FREE_PTR(it);
-               else
-                       info->oti_it_inline = 0;
-
+               OBD_FREE_PTR(it);
                RETURN(ERR_PTR(rc));
        }
 
@@ -205,16 +196,12 @@ static struct dt_it *osd_it_acct_init(const struct lu_env *env,
  */
 static void osd_it_acct_fini(const struct lu_env *env, struct dt_it *di)
 {
-       struct osd_thread_info  *info   = osd_oti_get(env);
        struct osd_it_quota     *it     = (struct osd_it_quota *)di;
        ENTRY;
 
        osd_zap_cursor_fini(it->oiq_zc);
        lu_object_put(env, &it->oiq_obj->oo_dt.do_lu);
-       if (it != &info->oti_it_quota)
-               OBD_FREE_PTR(it);
-       else
-               info->oti_it_inline = 0;
+       OBD_FREE_PTR(it);
 
        EXIT;
 }