Whamcloud - gitweb
LU-5223 lmv: build master LMV EA dynamically build via readdir
[fs/lustre-release.git] / lustre / lod / lod_lov.c
index b7460b4..80e96cd 100644 (file)
  * Copyright  2009 Sun Microsystems, Inc. All rights reserved
  * Use is subject to license terms.
  *
- * Copyright (c) 2012, Intel Corporation.
+ * Copyright (c) 2012, 2013, Intel Corporation.
  */
 /*
  * lustre/lod/lod_lov.c
  *
- * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com> 
+ * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  */
 
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
 #define DEBUG_SUBSYSTEM S_MDS
 
 #include <obd_class.h>
-#include <obd_lov.h>
+#include <lustre_lfsck.h>
+#include <lustre_lmv.h>
 
 #include "lod_internal.h"
 
@@ -187,6 +185,7 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
        struct lod_tgt_desc     *tgt_desc;
        struct lod_tgt_descs    *ltd;
        struct obd_uuid         obd_uuid;
+       bool                    for_ost;
        ENTRY;
 
        CDEBUG(D_CONFIG, "osp:%s idx:%d gen:%d\n", osp, index, gen);
@@ -215,6 +214,7 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
        data->ocd_index = index;
 
        if (strcmp(LUSTRE_OSC_NAME, type) == 0) {
+               for_ost = true;
                data->ocd_connect_flags |= OBD_CONNECT_AT |
                                           OBD_CONNECT_FULL20 |
                                           OBD_CONNECT_INDEX |
@@ -227,13 +227,16 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
                                           OBD_CONNECT_SKIP_ORPHAN |
                                           OBD_CONNECT_FID |
                                           OBD_CONNECT_LVB_TYPE |
-                                          OBD_CONNECT_VERSION;
+                                          OBD_CONNECT_VERSION |
+                                          OBD_CONNECT_PINGLESS |
+                                          OBD_CONNECT_LFSCK;
 
                data->ocd_group = tgt_index;
                ltd = &lod->lod_ost_descs;
        } else {
                struct obd_import *imp = obd->u.cli.cl_import;
 
+               for_ost = false;
                data->ocd_ibits_known = MDS_INODELOCK_UPDATE;
                data->ocd_connect_flags |= OBD_CONNECT_ACL |
                                           OBD_CONNECT_MDS_CAPA |
@@ -242,14 +245,15 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
                                           OBD_CONNECT_MDS_MDS |
                                           OBD_CONNECT_FID |
                                           OBD_CONNECT_AT |
-                                          OBD_CONNECT_FULL20;
+                                          OBD_CONNECT_FULL20 |
+                                          OBD_CONNECT_LFSCK;
                /* XXX set MDS-MDS flags, remove this when running this
                 * on client*/
                data->ocd_connect_flags |= OBD_CONNECT_MDS_MDS;
                spin_lock(&imp->imp_lock);
                imp->imp_server_timeout = 1;
                spin_unlock(&imp->imp_lock);
-               imp->imp_client->cli_request_portal = MDS_MDS_PORTAL;
+               imp->imp_client->cli_request_portal = OUT_PORTAL;
                CDEBUG(D_OTHER, "%s: Set 'mds' portal and timeout\n",
                      obd->obd_name);
                ltd = &lod->lod_mdt_descs;
@@ -264,7 +268,7 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
        }
 
        LASSERT(obd->obd_lu_dev);
-       LASSERT(obd->obd_lu_dev->ld_site = lod->lod_dt_dev.dd_lu_dev.ld_site);
+       LASSERT(obd->obd_lu_dev->ld_site == lod->lod_dt_dev.dd_lu_dev.ld_site);
 
        ldev = obd->obd_lu_dev;
        d = lu2dt_dev(ldev);
@@ -349,7 +353,12 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
        if (lod->lod_recovery_completed)
                ldev->ld_ops->ldo_recovery_complete(env, ldev);
 
-       RETURN(0);
+       rc = lfsck_add_target(env, lod->lod_child, d, exp, index, for_ost);
+       if (rc != 0)
+               CERROR("Fail to add LFSCK target: name = %s, type = %s, "
+                      "index = %u, rc = %d\n", osp, type, index, rc);
+
+       RETURN(rc);
 
 out_pool:
        lod_ost_pool_remove(&lod->lod_pool_info, index);
@@ -367,17 +376,23 @@ out_free:
 /*
  * helper function to schedule OST removal from the device table
  */
-static void __lod_del_device(struct lod_tgt_descs *ltd,
-                            unsigned idx)
+static void __lod_del_device(const struct lu_env *env, struct lod_device *lod,
+                            struct lod_tgt_descs *ltd, unsigned idx,
+                            bool for_ost)
 {
        LASSERT(LTD_TGT(ltd, idx));
+
+       lfsck_del_target(env, lod->lod_child, LTD_TGT(ltd, idx)->ltd_tgt,
+                        idx, for_ost);
+
        if (LTD_TGT(ltd, idx)->ltd_reap == 0) {
                LTD_TGT(ltd, idx)->ltd_reap = 1;
                ltd->ltd_death_row++;
        }
 }
 
-int lod_fini_tgt(struct lod_device *lod, struct lod_tgt_descs *ltd)
+int lod_fini_tgt(const struct lu_env *env, struct lod_device *lod,
+                struct lod_tgt_descs *ltd, bool for_ost)
 {
        int idx;
 
@@ -386,7 +401,7 @@ int lod_fini_tgt(struct lod_device *lod, struct lod_tgt_descs *ltd)
        lod_getref(ltd);
        mutex_lock(&ltd->ltd_mutex);
        cfs_foreach_bit(ltd->ltd_tgt_bitmap, idx)
-               __lod_del_device(ltd, idx);
+               __lod_del_device(env, lod, ltd, idx, for_ost);
        mutex_unlock(&ltd->ltd_mutex);
        lod_putref(lod, ltd);
        CFS_FREE_BITMAP(ltd->ltd_tgt_bitmap);
@@ -413,7 +428,7 @@ int lod_fini_tgt(struct lod_device *lod, struct lod_tgt_descs *ltd)
  */
 int lod_del_device(const struct lu_env *env, struct lod_device *lod,
                   struct lod_tgt_descs *ltd, char *osp, unsigned idx,
-                  unsigned gen)
+                  unsigned gen, bool for_ost)
 {
        struct obd_device *obd;
        int                rc = 0;
@@ -456,7 +471,7 @@ int lod_del_device(const struct lu_env *env, struct lod_device *lod,
                GOTO(out, rc = -EINVAL);
        }
 
-       __lod_del_device(ltd, idx);
+       __lod_del_device(env, lod, ltd, idx, for_ost);
        EXIT;
 out:
        mutex_unlock(&ltd->ltd_mutex);
@@ -468,7 +483,8 @@ int lod_ea_store_resize(struct lod_thread_info *info, int size)
 {
        int round = size_roundup_power2(size);
 
-       LASSERT(round <= lov_mds_md_size(LOV_MAX_STRIPE_COUNT, LOV_MAGIC_V3));
+       LASSERT(round <=
+               lov_mds_md_size(LOV_MAX_STRIPE_COUNT, LOV_MAGIC_V3));
        if (info->lti_ea_store) {
                LASSERT(info->lti_ea_store_size);
                LASSERT(info->lti_ea_store_size < round);
@@ -499,12 +515,12 @@ int lod_generate_and_set_lovea(const struct lu_env *env,
        struct lov_ost_data_v1  *objs;
        __u32                    magic;
        int                      i, rc, lmm_size;
+       int                      cplen = 0;
        ENTRY;
 
        LASSERT(lo);
-       LASSERT(lo->ldo_stripenr > 0);
 
-       magic = lo->ldo_pool ? LOV_MAGIC_V3 : LOV_MAGIC_V1;
+       magic = lo->ldo_pool != NULL ? LOV_MAGIC_V3 : LOV_MAGIC_V1;
        lmm_size = lov_mds_md_size(lo->ldo_stripenr, magic);
        if (info->lti_ea_store_size < lmm_size) {
                rc = lod_ea_store_resize(info, lmm_size);
@@ -512,42 +528,60 @@ int lod_generate_and_set_lovea(const struct lu_env *env,
                        RETURN(rc);
        }
 
+       if (lo->ldo_pattern == 0) /* default striping */
+               lo->ldo_pattern = LOV_PATTERN_RAID0;
+
        lmm = info->lti_ea_store;
 
        lmm->lmm_magic = cpu_to_le32(magic);
-       lmm->lmm_pattern = cpu_to_le32(LOV_PATTERN_RAID0);
-       lmm->lmm_object_id = cpu_to_le64(fid_ver_oid(fid));
-       lmm->lmm_object_seq = cpu_to_le64(fid_seq(fid));
+       lmm->lmm_pattern = cpu_to_le32(lo->ldo_pattern);
+       fid_to_lmm_oi(fid, &lmm->lmm_oi);
+       if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_LMMOI))
+               lmm->lmm_oi.oi.oi_id++;
+       lmm_oi_cpu_to_le(&lmm->lmm_oi, &lmm->lmm_oi);
        lmm->lmm_stripe_size = cpu_to_le32(lo->ldo_stripe_size);
        lmm->lmm_stripe_count = cpu_to_le16(lo->ldo_stripenr);
+       if (lo->ldo_pattern & LOV_PATTERN_F_RELEASED)
+               lmm->lmm_stripe_count = cpu_to_le16(lo->ldo_released_stripenr);
        lmm->lmm_layout_gen = 0;
        if (magic == LOV_MAGIC_V1) {
                objs = &lmm->lmm_objects[0];
        } else {
                struct lov_mds_md_v3 *v3 = (struct lov_mds_md_v3 *) lmm;
-               strncpy(v3->lmm_pool_name, lo->ldo_pool, LOV_MAXPOOLNAME);
+               cplen = strlcpy(v3->lmm_pool_name, lo->ldo_pool,
+                               sizeof(v3->lmm_pool_name));
+               if (cplen >= sizeof(v3->lmm_pool_name))
+                       RETURN(-E2BIG);
                objs = &v3->lmm_objects[0];
        }
 
        for (i = 0; i < lo->ldo_stripenr; i++) {
-               const struct lu_fid     *fid;
+               struct lu_fid           *fid    = &info->lti_fid;
                struct lod_device       *lod;
                __u32                   index;
+               int                     type    = LU_SEQ_RANGE_OST;
 
                lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
                LASSERT(lo->ldo_stripe[i]);
-               fid = lu_object_fid(&lo->ldo_stripe[i]->do_lu);
 
-               rc = fid_ostid_pack(fid, &info->lti_ostid);
+               *fid = *lu_object_fid(&lo->ldo_stripe[i]->do_lu);
+               if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MULTIPLE_REF)) {
+                       if (cfs_fail_val == 0)
+                               cfs_fail_val = fid->f_oid;
+                       else
+                               fid->f_oid = cfs_fail_val;
+               }
+
+               rc = fid_to_ostid(fid, &info->lti_ostid);
                LASSERT(rc == 0);
 
-               objs[i].l_object_id  = cpu_to_le64(info->lti_ostid.oi_id);
-               objs[i].l_object_seq = cpu_to_le64(info->lti_ostid.oi_seq);
+               ostid_cpu_to_le(&info->lti_ostid, &objs[i].l_ost_oi);
                objs[i].l_ost_gen    = cpu_to_le32(0);
-               rc = lod_fld_lookup(env, lod, fid, &index, LU_SEQ_RANGE_OST);
+               rc = lod_fld_lookup(env, lod, fid, &index, &type);
                if (rc < 0) {
                        CERROR("%s: Can not locate "DFID": rc = %d\n",
                               lod2obd(lod)->obd_name, PFID(fid), rc);
+                       lod_object_free_striping(env, lo);
                        RETURN(rc);
                }
                objs[i].l_ost_idx = cpu_to_le32(index);
@@ -557,38 +591,41 @@ int lod_generate_and_set_lovea(const struct lu_env *env,
        info->lti_buf.lb_len = lmm_size;
        rc = dt_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV, 0,
                          th, BYPASS_CAPA);
+       if (rc < 0)
+               lod_object_free_striping(env, lo);
 
        RETURN(rc);
 }
 
-int lod_get_lov_ea(const struct lu_env *env, struct lod_object *lo)
+int lod_get_ea(const struct lu_env *env, struct lod_object *lo,
+              const char *name)
 {
-       struct lod_thread_info *info = lod_env_info(env);
-       struct dt_object       *next = dt_object_child(&lo->ldo_obj);
+       struct lod_thread_info  *info = lod_env_info(env);
+       struct dt_object        *next = dt_object_child(&lo->ldo_obj);
        int                     rc;
        ENTRY;
 
        LASSERT(info);
 
-       if (unlikely(info->lti_ea_store_size == 0)) {
+       if (unlikely(info->lti_ea_store == NULL)) {
                /* just to enter in allocation block below */
                rc = -ERANGE;
        } else {
 repeat:
                info->lti_buf.lb_buf = info->lti_ea_store;
                info->lti_buf.lb_len = info->lti_ea_store_size;
-               rc = dt_xattr_get(env, next, &info->lti_buf, XATTR_NAME_LOV,
-                                 BYPASS_CAPA);
+               rc = dt_xattr_get(env, next, &info->lti_buf, name, BYPASS_CAPA);
        }
+
        /* if object is not striped or inaccessible */
-       if (rc == -ENODATA)
+       if (rc == -ENODATA || rc == -ENOENT)
                RETURN(0);
 
        if (rc == -ERANGE) {
                /* EA doesn't fit, reallocate new buffer */
-               rc = dt_xattr_get(env, next, &LU_BUF_NULL, XATTR_NAME_LOV,
+               rc = dt_xattr_get(env, next, &LU_BUF_NULL, name,
                                  BYPASS_CAPA);
-               if (rc == -ENODATA)
+               if (rc == -ENODATA || rc == -ENOENT)
                        RETURN(0);
                else if (rc < 0)
                        RETURN(rc);
@@ -613,8 +650,8 @@ int lod_store_def_striping(const struct lu_env *env, struct dt_object *dt,
        int                      rc;
        ENTRY;
 
-       LASSERT(S_ISDIR(dt->do_lu.lo_header->loh_attr));
-
+       if (S_ISDIR(dt->do_lu.lo_header->loh_attr))
+               RETURN(-ENOTDIR);
        /*
         * store striping defaults into new directory
         * used to implement defaults inheritance
@@ -628,31 +665,53 @@ int lod_store_def_striping(const struct lu_env *env, struct dt_object *dt,
                                lo->ldo_def_stripe_offset))
                RETURN(0);
 
-       /* XXX: use thread info */
-       OBD_ALLOC_PTR(v3);
-       if (v3 == NULL)
-               RETURN(-ENOMEM);
-
-       v3->lmm_magic = cpu_to_le32(LOV_MAGIC_V3);
-       v3->lmm_pattern = cpu_to_le32(LOV_PATTERN_RAID0);
-       v3->lmm_object_id = 0;
-       v3->lmm_object_seq = 0;
-       v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
+       v3 = info->lti_ea_store;
+       if (info->lti_ea_store_size < sizeof(*v3)) {
+               rc = lod_ea_store_resize(info, sizeof(*v3));
+               if (rc != 0)
+                       RETURN(rc);
+               v3 = info->lti_ea_store;
+       }
+       memset(v3, 0, sizeof(*v3));
+       v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
        v3->lmm_stripe_count = cpu_to_le16(lo->ldo_def_stripenr);
        v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset);
-       if (lo->ldo_pool)
-               strncpy(v3->lmm_pool_name, lo->ldo_pool, LOV_MAXPOOLNAME);
-
+       v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
+       if (lo->ldo_pool != NULL)
+               strlcpy(v3->lmm_pool_name, lo->ldo_pool,
+                       sizeof(v3->lmm_pool_name));
        info->lti_buf.lb_buf = v3;
        info->lti_buf.lb_len = sizeof(*v3);
        rc = dt_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV, 0, th,
                        BYPASS_CAPA);
 
-       OBD_FREE_PTR(v3);
-
        RETURN(rc);
 }
 
+static int validate_lod_and_idx(struct lod_device *md, int idx)
+{
+       if (unlikely(idx >= md->lod_ost_descs.ltd_tgts_size ||
+                    !cfs_bitmap_check(md->lod_ost_bitmap, idx))) {
+               CERROR("%s: bad idx: %d of %d\n", lod2obd(md)->obd_name, idx,
+                      md->lod_ost_descs.ltd_tgts_size);
+               return -EINVAL;
+       }
+
+       if (unlikely(OST_TGT(md, idx) == NULL)) {
+               CERROR("%s: bad lod_tgt_desc for idx: %d\n",
+                      lod2obd(md)->obd_name, idx);
+               return -EINVAL;
+       }
+
+       if (unlikely(OST_TGT(md, idx)->ltd_ost == NULL)) {
+               CERROR("%s: invalid lod device, for idx: %d\n",
+                      lod2obd(md)->obd_name , idx);
+               return -EINVAL;
+       }
+
+       return 0;
+}
+
 /*
  * allocate array of objects pointers, find/create objects
  * stripenr and other fields should be initialized by this moment
@@ -661,43 +720,50 @@ int lod_initialize_objects(const struct lu_env *env, struct lod_object *lo,
                           struct lov_ost_data_v1 *objs)
 {
        struct lod_thread_info  *info = lod_env_info(env);
-       struct lod_device       *md = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
+       struct lod_device       *md;
        struct lu_object        *o, *n;
        struct lu_device        *nd;
-       int                     i, idx, rc = 0;
+       struct dt_object       **stripe;
+       int                      stripe_len;
+       int                      i, idx, rc = 0;
        ENTRY;
 
-       LASSERT(lo);
+       LASSERT(lo != NULL);
+       md = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
        LASSERT(lo->ldo_stripe == NULL);
        LASSERT(lo->ldo_stripenr > 0);
        LASSERT(lo->ldo_stripe_size > 0);
 
-       i = sizeof(struct dt_object *) * lo->ldo_stripenr;
-       OBD_ALLOC(lo->ldo_stripe, i);
-       if (lo->ldo_stripe == NULL)
-               GOTO(out, rc = -ENOMEM);
-       lo->ldo_stripes_allocated = lo->ldo_stripenr;
+       stripe_len = lo->ldo_stripenr;
+       OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_len);
+       if (stripe == NULL)
+               RETURN(-ENOMEM);
 
        for (i = 0; i < lo->ldo_stripenr; i++) {
-               info->lti_ostid.oi_id = le64_to_cpu(objs[i].l_object_id);
-               info->lti_ostid.oi_seq = le64_to_cpu(objs[i].l_object_seq);
-               idx = le64_to_cpu(objs[i].l_ost_idx);
-               fid_ostid_unpack(&info->lti_fid, &info->lti_ostid, idx);
+               if (unlikely(lovea_slot_is_dummy(&objs[i])))
+                       continue;
+
+               ostid_le_to_cpu(&objs[i].l_ost_oi, &info->lti_ostid);
+               idx = le32_to_cpu(objs[i].l_ost_idx);
+               rc = ostid_to_fid(&info->lti_fid, &info->lti_ostid, idx);
+               if (rc != 0)
+                       GOTO(out, rc);
                LASSERTF(fid_is_sane(&info->lti_fid), ""DFID" insane!\n",
                         PFID(&info->lti_fid));
-               /*
-                * XXX: assertion is left for testing, to make
-                * sure we never process requests till configuration
-                * is completed. to be changed to -EINVAL
-                */
-
                lod_getref(&md->lod_ost_descs);
-               LASSERT(cfs_bitmap_check(md->lod_ost_bitmap, idx));
-               LASSERT(OST_TGT(md,idx));
-               LASSERTF(OST_TGT(md,idx)->ltd_ost, "idx %d\n", idx);
+
+               rc = validate_lod_and_idx(md, idx);
+               if (unlikely(rc != 0)) {
+                       lod_putref(md, &md->lod_ost_descs);
+                       GOTO(out, rc);
+               }
+
                nd = &OST_TGT(md,idx)->ltd_ost->dd_lu_dev;
                lod_putref(md, &md->lod_ost_descs);
 
+               /* In the function below, .hs_keycmp resolves to
+                * u_obj_hop_keycmp() */
+               /* coverity[overrun-buffer-val] */
                o = lu_object_find_at(env, nd, &info->lti_fid, NULL);
                if (IS_ERR(o))
                        GOTO(out, rc = PTR_ERR(o));
@@ -705,10 +771,22 @@ int lod_initialize_objects(const struct lu_env *env, struct lod_object *lo,
                n = lu_object_locate(o->lo_header, nd->ld_type);
                LASSERT(n);
 
-               lo->ldo_stripe[i] = container_of(n, struct dt_object, do_lu);
+               stripe[i] = container_of(n, struct dt_object, do_lu);
        }
 
 out:
+       if (rc != 0) {
+               for (i = 0; i < stripe_len; i++)
+                       if (stripe[i] != NULL)
+                               lu_object_put(env, &stripe[i]->do_lu);
+
+               OBD_FREE(stripe, sizeof(stripe[0]) * stripe_len);
+               lo->ldo_stripenr = 0;
+       } else {
+               lo->ldo_stripe = stripe;
+               lo->ldo_stripes_allocated = stripe_len;
+       }
+
        RETURN(rc);
 }
 
@@ -721,6 +799,7 @@ int lod_parse_striping(const struct lu_env *env, struct lod_object *lo,
        struct lov_mds_md_v1    *lmm;
        struct lov_ost_data_v1  *objs;
        __u32                    magic;
+       __u32                    pattern;
        int                      rc = 0;
        ENTRY;
 
@@ -730,15 +809,20 @@ int lod_parse_striping(const struct lu_env *env, struct lod_object *lo,
 
        lmm = (struct lov_mds_md_v1 *) buf->lb_buf;
        magic = le32_to_cpu(lmm->lmm_magic);
+       pattern = le32_to_cpu(lmm->lmm_pattern);
 
        if (magic != LOV_MAGIC_V1 && magic != LOV_MAGIC_V3)
                GOTO(out, rc = -EINVAL);
-       if (le32_to_cpu(lmm->lmm_pattern) != LOV_PATTERN_RAID0)
+       if (lov_pattern(pattern) != LOV_PATTERN_RAID0)
                GOTO(out, rc = -EINVAL);
 
+       lo->ldo_pattern = pattern;
        lo->ldo_stripe_size = le32_to_cpu(lmm->lmm_stripe_size);
-       lo->ldo_stripenr = le16_to_cpu(lmm->lmm_stripe_count);
        lo->ldo_layout_gen = le16_to_cpu(lmm->lmm_layout_gen);
+       lo->ldo_stripenr = le16_to_cpu(lmm->lmm_stripe_count);
+       /* released file stripenr fixup. */
+       if (pattern & LOV_PATTERN_F_RELEASED)
+               lo->ldo_stripenr = 0;
 
        LASSERT(buf->lb_len >= lov_mds_md_size(lo->ldo_stripenr, magic));
 
@@ -750,152 +834,223 @@ int lod_parse_striping(const struct lu_env *env, struct lod_object *lo,
                objs = &lmm->lmm_objects[0];
        }
 
-       rc = lod_initialize_objects(env, lo, objs);
+       if (lo->ldo_stripenr > 0)
+               rc = lod_initialize_objects(env, lo, objs);
 
 out:
        RETURN(rc);
 }
 
-/*
- * Load and parse striping information, create in-core representation for the
- * stripes
- */
-int lod_load_striping(const struct lu_env *env, struct lod_object *lo)
+int lod_load_striping_locked(const struct lu_env *env, struct lod_object *lo)
 {
        struct lod_thread_info  *info = lod_env_info(env);
+       struct lu_buf           *buf  = &info->lti_buf;
        struct dt_object        *next = dt_object_child(&lo->ldo_obj);
-       int                      rc;
+       int                      rc = 0;
        ENTRY;
 
-       /*
-        * currently this code is supposed to be called from declaration
-        * phase only, thus the object is not expected to be locked by caller
-        */
-       dt_write_lock(env, next, 0);
        /* already initialized? */
-       if (lo->ldo_stripe) {
-               int i;
-               /* check validity */
-               for (i = 0; i < lo->ldo_stripenr; i++)
-                       LASSERTF(lo->ldo_stripe[i], "stripe %d is NULL\n", i);
+       if (lo->ldo_stripe != NULL)
                GOTO(out, rc = 0);
-       }
 
        if (!dt_object_exists(next))
                GOTO(out, rc = 0);
 
-       /* only regular files can be striped */
-       if (!(lu_object_attr(lod2lu_obj(lo)) & S_IFREG))
+       /* Do not load stripe for slaves of striped dir */
+       if (lo->ldo_dir_slave_stripe)
                GOTO(out, rc = 0);
 
-       LASSERT(lo->ldo_stripenr == 0);
+       if (S_ISREG(lu_object_attr(lod2lu_obj(lo)))) {
+               rc = lod_get_lov_ea(env, lo);
+               if (rc <= 0)
+                       GOTO(out, rc);
+               /*
+                * there is LOV EA (striping information) in this object
+                * let's parse it and create in-core objects for the stripes
+                */
+               buf->lb_buf = info->lti_ea_store;
+               buf->lb_len = info->lti_ea_store_size;
+               rc = lod_parse_striping(env, lo, buf);
+       } else if (S_ISDIR(lu_object_attr(lod2lu_obj(lo)))) {
+               rc = lod_get_lmv_ea(env, lo);
+               if (rc < sizeof(struct lmv_mds_md_v1))
+                       GOTO(out, rc = rc > 0 ? -EINVAL : rc);
+
+               buf->lb_buf = info->lti_ea_store;
+               buf->lb_len = info->lti_ea_store_size;
+               if (rc == sizeof(struct lmv_mds_md_v1)) {
+                       rc = lod_load_lmv_shards(env, lo, buf, true);
+                       if (buf->lb_buf != info->lti_ea_store) {
+                               OBD_FREE_LARGE(info->lti_ea_store,
+                                              info->lti_ea_store_size);
+                               info->lti_ea_store = buf->lb_buf;
+                               info->lti_ea_store_size = buf->lb_len;
+                       }
 
-       rc = lod_get_lov_ea(env, lo);
-       if (rc <= 0)
-               GOTO(out, rc);
+                       if (rc < 0)
+                               GOTO(out, rc);
+               }
 
-       /*
-        * there is LOV EA (striping information) in this object
-        * let's parse it and create in-core objects for the stripes
-        */
-       info->lti_buf.lb_buf = info->lti_ea_store;
-       info->lti_buf.lb_len = info->lti_ea_store_size;
-       rc = lod_parse_striping(env, lo, &info->lti_buf);
+               /*
+                * there is LOV EA (striping information) in this object
+                * let's parse it and create in-core objects for the stripes
+                */
+               rc = lod_parse_dir_striping(env, lo, buf);
+       }
 out:
-       dt_write_unlock(env, next);
        RETURN(rc);
 }
 
+/**
+ * Load and parse striping information, create in-core representation for the
+ * stripes
+ **/
+int lod_load_striping(const struct lu_env *env, struct lod_object *lo)
+{
+       struct dt_object        *next = dt_object_child(&lo->ldo_obj);
+       int                     rc = 0;
+
+       /* currently this code is supposed to be called from declaration
+        * phase only, thus the object is not expected to be locked by caller */
+       dt_write_lock(env, next, 0);
+       rc = lod_load_striping_locked(env, lo);
+       dt_write_unlock(env, next);
+       return rc;
+}
+
+/* verify the striping information for directory */
 int lod_verify_striping(struct lod_device *d, const struct lu_buf *buf,
-                       int specific)
+                       bool is_from_disk)
 {
        struct lov_user_md_v1   *lum;
-       struct lov_user_md_v3   *v3 = NULL;
+       struct lov_user_md_v3   *lum3;
        struct pool_desc        *pool = NULL;
-       int                      rc;
+       __u32                    magic;
+       __u32                    stripe_size;
+       __u16                    stripe_count;
+       __u16                    stripe_offset;
+       size_t                   lum_size;
+       int                      rc = 0;
        ENTRY;
 
        lum = buf->lb_buf;
 
-       if (lum->lmm_magic != LOV_USER_MAGIC_V1 &&
-           lum->lmm_magic != LOV_USER_MAGIC_V3 &&
-           lum->lmm_magic != LOV_MAGIC_V1_DEF &&
-           lum->lmm_magic != LOV_MAGIC_V3_DEF) {
-               CDEBUG(D_IOCTL, "bad userland LOV MAGIC: %#x\n",
-                      lum->lmm_magic);
-               RETURN(-EINVAL);
+       LASSERT(sizeof(*lum) < sizeof(*lum3));
+
+       if (buf->lb_len < sizeof(*lum)) {
+               CDEBUG(D_IOCTL, "buf len %zd too small for lov_user_md\n",
+                      buf->lb_len);
+               GOTO(out, rc = -EINVAL);
+       }
+
+       magic = le32_to_cpu(lum->lmm_magic);
+       if (magic != LOV_USER_MAGIC_V1 &&
+           magic != LOV_USER_MAGIC_V3 &&
+           magic != LOV_MAGIC_V1_DEF &&
+           magic != LOV_MAGIC_V3_DEF) {
+               CDEBUG(D_IOCTL, "bad userland LOV MAGIC: %#x\n", magic);
+               GOTO(out, rc = -EINVAL);
        }
 
-       if ((specific && lum->lmm_pattern != LOV_PATTERN_RAID0) ||
-           (specific == 0 && lum->lmm_pattern != 0)) {
+       /* the user uses "0" for default stripe pattern normally. */
+       if (!is_from_disk && lum->lmm_pattern == 0)
+               lum->lmm_pattern = cpu_to_le32(LOV_PATTERN_RAID0);
+
+       if (le32_to_cpu(lum->lmm_pattern) != LOV_PATTERN_RAID0) {
                CDEBUG(D_IOCTL, "bad userland stripe pattern: %#x\n",
-                      lum->lmm_pattern);
-               RETURN(-EINVAL);
+                      le32_to_cpu(lum->lmm_pattern));
+               GOTO(out, rc = -EINVAL);
        }
 
        /* 64kB is the largest common page size we see (ia64), and matches the
         * check in lfs */
-       if (lum->lmm_stripe_size & (LOV_MIN_STRIPE_SIZE - 1)) {
-               CDEBUG(D_IOCTL, "stripe size %u not multiple of %u, fixing\n",
-                      lum->lmm_stripe_size, LOV_MIN_STRIPE_SIZE);
-               RETURN(-EINVAL);
+       stripe_size = le32_to_cpu(lum->lmm_stripe_size);
+       if (stripe_size & (LOV_MIN_STRIPE_SIZE - 1)) {
+               CDEBUG(D_IOCTL, "stripe size %u not a multiple of %u\n",
+                      stripe_size, LOV_MIN_STRIPE_SIZE);
+               GOTO(out, rc = -EINVAL);
        }
 
        /* an offset of -1 is treated as a "special" valid offset */
-       if (lum->lmm_stripe_offset != (typeof(lum->lmm_stripe_offset))(-1)) {
+       stripe_offset = le16_to_cpu(lum->lmm_stripe_offset);
+       if (stripe_offset != (typeof(stripe_offset))-1) {
                /* if offset is not within valid range [0, osts_size) */
-               if (lum->lmm_stripe_offset >= d->lod_osts_size) {
+               if (stripe_offset >= d->lod_osts_size) {
                        CDEBUG(D_IOCTL, "stripe offset %u >= bitmap size %u\n",
-                              lum->lmm_stripe_offset, d->lod_osts_size);
-                       RETURN(-EINVAL);
+                              stripe_offset, d->lod_osts_size);
+                       GOTO(out, rc = -EINVAL);
                }
 
                /* if lmm_stripe_offset is *not* in bitmap */
-               if (!cfs_bitmap_check(d->lod_ost_bitmap,
-                                     lum->lmm_stripe_offset)) {
+               if (!cfs_bitmap_check(d->lod_ost_bitmap, stripe_offset)) {
                        CDEBUG(D_IOCTL, "stripe offset %u not in bitmap\n",
-                              lum->lmm_stripe_offset);
-                       RETURN(-EINVAL);
+                              stripe_offset);
+                       GOTO(out, rc = -EINVAL);
                }
        }
 
-       if (lum->lmm_magic == LOV_USER_MAGIC_V3)
-               v3 = buf->lb_buf;
+       if (magic == LOV_USER_MAGIC_V1 || magic == LOV_MAGIC_V1_DEF)
+               lum_size = offsetof(struct lov_user_md_v1,
+                                   lmm_objects[0]);
+       else if (magic == LOV_USER_MAGIC_V3 || magic == LOV_MAGIC_V3_DEF)
+               lum_size = offsetof(struct lov_user_md_v3,
+                                   lmm_objects[0]);
+       else
+               GOTO(out, rc = -EINVAL);
 
-       if (v3)
-               pool = lod_find_pool(d, v3->lmm_pool_name);
+       stripe_count = le16_to_cpu(lum->lmm_stripe_count);
+       if (buf->lb_len != lum_size) {
+               CDEBUG(D_IOCTL, "invalid buf len %zd for lov_user_md with "
+                      "magic %#x and stripe_count %u\n",
+                      buf->lb_len, magic, stripe_count);
+               GOTO(out, rc = -EINVAL);
+       }
 
-       if (pool != NULL) {
-               __u16 offs = v3->lmm_stripe_offset;
+       if (!(magic == LOV_USER_MAGIC_V3 || magic == LOV_MAGIC_V3_DEF))
+               goto out;
 
-               if (offs != (typeof(v3->lmm_stripe_offset))(-1)) {
-                       rc = lod_check_index_in_pool(offs, pool);
-                       if (rc < 0) {
-                               lod_pool_putref(pool);
-                               RETURN(-EINVAL);
-                       }
-               }
+       lum3 = buf->lb_buf;
+       if (buf->lb_len < sizeof(*lum3)) {
+               CDEBUG(D_IOCTL, "buf len %zd too small for lov_user_md_v3\n",
+                      buf->lb_len);
+               GOTO(out, rc = -EINVAL);
+       }
 
-               if (specific && lum->lmm_stripe_count > pool_tgt_count(pool)) {
-                       CDEBUG(D_IOCTL,
-                              "stripe count %u > # OSTs %u in the pool\n",
-                              lum->lmm_stripe_count, pool_tgt_count(pool));
-                       lod_pool_putref(pool);
-                       RETURN(-EINVAL);
-               }
+       /* In the function below, .hs_keycmp resolves to
+        * pool_hashkey_keycmp() */
+       /* coverity[overrun-buffer-val] */
+       pool = lod_find_pool(d, lum3->lmm_pool_name);
+       if (pool == NULL)
+               goto out;
+
+       if (stripe_offset != (typeof(stripe_offset))-1) {
+               rc = lod_check_index_in_pool(stripe_offset, pool);
+               if (rc < 0)
+                       GOTO(out, rc = -EINVAL);
+       }
 
-               lod_pool_putref(pool);
+       if (is_from_disk && stripe_count > pool_tgt_count(pool)) {
+               CDEBUG(D_IOCTL,
+                      "stripe count %u > # OSTs %u in the pool\n",
+                      stripe_count, pool_tgt_count(pool));
+               GOTO(out, rc = -EINVAL);
        }
 
-       RETURN(0);
+out:
+       if (pool != NULL)
+               lod_pool_putref(pool);
+
+       RETURN(rc);
 }
 
 void lod_fix_desc_stripe_size(__u64 *val)
 {
-       if (*val < LOV_DEFAULT_STRIPE_SIZE) {
-               LCONSOLE_WARN("Increasing default stripe size to min %u\n",
-                             LOV_DEFAULT_STRIPE_SIZE);
-               *val = LOV_DEFAULT_STRIPE_SIZE;
+       if (*val < LOV_MIN_STRIPE_SIZE) {
+               if (*val != 0)
+                       LCONSOLE_INFO("Increasing default stripe size to "
+                                     "minimum value %u\n",
+                                     LOV_DESC_STRIPE_SIZE_DEFAULT);
+               *val = LOV_DESC_STRIPE_SIZE_DEFAULT;
        } else if (*val & (LOV_MIN_STRIPE_SIZE - 1)) {
                *val &= ~(LOV_MIN_STRIPE_SIZE - 1);
                LCONSOLE_WARN("Changing default stripe size to "LPU64" (a "
@@ -923,7 +1078,7 @@ void lod_fix_desc_qos_maxage(__u32 *val)
 {
        /* fix qos_maxage */
        if (*val == 0)
-               *val = QOS_DEFAULT_MAXAGE;
+               *val = LOV_DESC_QOS_MAXAGE_DEFAULT;
 }
 
 void lod_fix_desc(struct lov_desc *desc)
@@ -987,11 +1142,6 @@ int lod_pools_init(struct lod_device *lod, struct lustre_cfg *lcfg)
        lod->lod_qos.lq_prio_free = 232;
        /* Default threshold for rr (roughly 17%) */
        lod->lod_qos.lq_threshold_rr = 43;
-       /* Init statfs fields */
-       OBD_ALLOC_PTR(lod->lod_qos.lq_statfs_data);
-       if (NULL == lod->lod_qos.lq_statfs_data)
-               RETURN(-ENOMEM);
-       cfs_waitq_init(&lod->lod_qos.lq_statfs_waitq);
 
        /* Set up OST pool environment */
        lod->lod_pools_hash_body = cfs_hash_create("POOLS", HASH_POOLS_CUR_BITS,
@@ -1001,8 +1151,9 @@ int lod_pools_init(struct lod_device *lod, struct lustre_cfg *lcfg)
                                                   CFS_HASH_MAX_THETA,
                                                   &pool_hash_operations,
                                                   CFS_HASH_DEFAULT);
-       if (!lod->lod_pools_hash_body)
-               GOTO(out_statfs, rc = -ENOMEM);
+       if (lod->lod_pools_hash_body == NULL)
+               RETURN(-ENOMEM);
+
        CFS_INIT_LIST_HEAD(&lod->lod_pool_list);
        lod->lod_pool_count = 0;
        rc = lod_ost_pool_init(&lod->lod_pool_info, 0);
@@ -1018,8 +1169,7 @@ out_pool_info:
        lod_ost_pool_free(&lod->lod_pool_info);
 out_hash:
        cfs_hash_putref(lod->lod_pools_hash_body);
-out_statfs:
-       OBD_FREE_PTR(lod->lod_qos.lq_statfs_data);
+
        return rc;
 }
 
@@ -1034,13 +1184,16 @@ int lod_pools_fini(struct lod_device *lod)
                pool = cfs_list_entry(pos, struct pool_desc, pool_list);
                /* free pool structs */
                CDEBUG(D_INFO, "delete pool %p\n", pool);
+               /* In the function below, .hs_keycmp resolves to
+                * pool_hashkey_keycmp() */
+               /* coverity[overrun-buffer-val] */
                lod_pool_del(obd, pool->pool_name);
        }
 
        cfs_hash_putref(lod->lod_pools_hash_body);
        lod_ost_pool_free(&(lod->lod_qos.lq_rr.lqr_pool));
        lod_ost_pool_free(&lod->lod_pool_info);
-       OBD_FREE_PTR(lod->lod_qos.lq_statfs_data);
+
        RETURN(0);
 }