+static int lod_verify_md_striping(struct lod_device *lod,
+ const struct lmv_user_md_v1 *lum)
+{
+ int rc = 0;
+ ENTRY;
+
+ if (unlikely(le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC))
+ GOTO(out, rc = -EINVAL);
+
+ if (unlikely(le32_to_cpu(lum->lum_stripe_count) == 0))
+ GOTO(out, rc = -EINVAL);
+
+ if (unlikely(le32_to_cpu(lum->lum_stripe_count) >
+ lod->lod_remote_mdt_count + 1))
+ GOTO(out, rc = -EINVAL);
+out:
+ if (rc != 0)
+ CERROR("%s: invalid lmv_user_md: magic = %x, "
+ "stripe_offset = %d, stripe_count = %u: rc = %d\n",
+ lod2obd(lod)->obd_name, le32_to_cpu(lum->lum_magic),
+ (int)le32_to_cpu(lum->lum_stripe_offset),
+ le32_to_cpu(lum->lum_stripe_count), rc);
+ return rc;
+}
+
+int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
+ struct lu_buf *lmv_buf)
+{
+ struct lod_thread_info *info = lod_env_info(env);
+ struct lod_device *lod = lu2lod_dev(dt->do_lu.lo_dev);
+ struct lod_object *lo = lod_dt_obj(dt);
+ struct lmv_mds_md_v1 *lmm1;
+ int stripe_count;
+ int lmm_size;
+ int type = LU_SEQ_RANGE_ANY;
+ int i;
+ int rc;
+ __u32 mdtidx;
+ ENTRY;
+
+ LASSERT(lo->ldo_dir_striped != 0);
+ LASSERT(lo->ldo_stripenr > 0);
+ stripe_count = lo->ldo_stripenr + 1;
+ lmm_size = lmv_mds_md_size(stripe_count, LMV_MAGIC);
+ if (info->lti_ea_store_size < lmm_size) {
+ rc = lod_ea_store_resize(info, lmm_size);
+ if (rc != 0)
+ RETURN(rc);
+ }
+
+ lmm1 = (struct lmv_mds_md_v1 *)info->lti_ea_store;
+ lmm1->lmv_magic = cpu_to_le32(LMV_MAGIC);
+ lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
+ lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
+ rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
+ &mdtidx, &type);
+ if (rc != 0)
+ RETURN(rc);
+
+ lmm1->lmv_master_mdt_index = cpu_to_le32(mdtidx);
+ fid_cpu_to_le(&lmm1->lmv_stripe_fids[0], lu_object_fid(&dt->do_lu));
+ for (i = 0; i < lo->ldo_stripenr; i++) {
+ struct dt_object *dto;
+
+ dto = lo->ldo_stripe[i];
+ LASSERT(dto != NULL);
+ fid_cpu_to_le(&lmm1->lmv_stripe_fids[i + 1],
+ lu_object_fid(&dto->do_lu));
+ }
+
+ lmv_buf->lb_buf = info->lti_ea_store;
+ lmv_buf->lb_len = lmm_size;
+ lo->ldo_dir_striping_cached = 1;
+
+ RETURN(rc);
+}
+
+int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
+ const struct lu_buf *buf)
+{
+ struct lod_thread_info *info = lod_env_info(env);
+ struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
+ struct lod_tgt_descs *ltd = &lod->lod_mdt_descs;
+ struct dt_object **stripe;
+ union lmv_mds_md *lmm = buf->lb_buf;
+ struct lmv_mds_md_v1 *lmv1 = &lmm->lmv_md_v1;
+ struct lu_fid *fid = &info->lti_fid;
+ int i;
+ int rc = 0;
+ ENTRY;
+
+ if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_MIGRATE)
+ RETURN(0);
+
+ if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
+ RETURN(-EINVAL);
+
+ if (le32_to_cpu(lmv1->lmv_stripe_count) <= 1)
+ RETURN(0);
+
+ fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[0]);
+ /* Do not load striping information for slave inode */
+ if (!lu_fid_eq(fid, lu_object_fid(&lo->ldo_obj.do_lu))) {
+ lo->ldo_dir_slave_stripe = 1;
+ RETURN(0);
+ }
+
+ LASSERT(lo->ldo_stripe == NULL);
+ OBD_ALLOC(stripe, sizeof(stripe[0]) *
+ (le32_to_cpu(lmv1->lmv_stripe_count) - 1));
+ if (stripe == NULL)
+ RETURN(-ENOMEM);
+
+ /* skip master stripe */
+ for (i = 1; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
+ struct lod_tgt_desc *tgt;
+ int idx;
+ int type = LU_SEQ_RANGE_ANY;
+ struct dt_object *dto;
+
+ fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
+ rc = lod_fld_lookup(env, lod, fid, &idx, &type);
+ if (rc != 0)
+ GOTO(out, rc);
+
+ tgt = LTD_TGT(ltd, idx);
+ if (tgt == NULL)
+ GOTO(out, rc = -ESTALE);
+
+ dto = dt_locate_at(env, tgt->ltd_tgt, fid,
+ lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
+ NULL);
+ if (IS_ERR(dto))
+ GOTO(out, rc = PTR_ERR(dto));
+
+ stripe[i - 1] = dto;
+ }
+out:
+ lo->ldo_stripe = stripe;
+ lo->ldo_stripenr = le32_to_cpu(lmv1->lmv_stripe_count) - 1;
+ lo->ldo_stripes_allocated = le32_to_cpu(lmv1->lmv_stripe_count) - 1;
+ if (rc != 0)
+ lod_object_free_striping(env, lo);
+
+ RETURN(rc);
+}
+
+static int lod_prep_md_striped_create(const struct lu_env *env,
+ struct dt_object *dt,
+ struct lu_attr *attr,
+ const struct lmv_user_md_v1 *lum,
+ struct thandle *th)
+{
+ struct lod_device *lod = lu2lod_dev(dt->do_lu.lo_dev);
+ struct lod_tgt_descs *ltd = &lod->lod_mdt_descs;
+ struct lod_object *lo = lod_dt_obj(dt);
+ struct dt_object **stripe;
+ struct lu_buf lmv_buf;
+ int stripe_count;
+ int *idx_array;
+ int rc = 0;
+ int i;
+ int j;
+ ENTRY;
+
+ /* The lum has been verifed in lod_verify_md_striping */
+ LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC);
+ LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
+
+ /* Do not need allocated master stripe */
+ stripe_count = le32_to_cpu(lum->lum_stripe_count);
+ OBD_ALLOC(stripe, sizeof(stripe[0]) * (stripe_count - 1));
+ if (stripe == NULL)
+ RETURN(-ENOMEM);
+
+ OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
+ if (idx_array == NULL)
+ GOTO(out_free, rc = -ENOMEM);
+
+ idx_array[0] = le32_to_cpu(lum->lum_stripe_offset);
+ for (i = 1; i < stripe_count; i++) {
+ struct lod_tgt_desc *tgt;
+ struct dt_object *dto;
+ struct lu_fid fid;
+ int idx;
+ struct lu_object_conf conf = { 0 };
+
+ idx = (idx_array[i - 1] + 1) % (lod->lod_remote_mdt_count + 1);
+
+ for (j = 0; j < lod->lod_remote_mdt_count;
+ j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
+ bool already_allocated = false;
+ int k;
+
+ CDEBUG(D_INFO, "try idx %d, mdt cnt %d,"
+ " allocated %d, last allocated %d\n", idx,
+ lod->lod_remote_mdt_count, i, idx_array[i - 1]);
+
+ /* Find next avaible target */
+ if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx))
+ continue;
+
+ /* check whether the idx already exists
+ * in current allocated array */
+ for (k = 0; k < i; k++) {
+ if (idx_array[k] == idx) {
+ already_allocated = true;
+ break;
+ }
+ }
+
+ if (already_allocated)
+ continue;
+
+ break;
+ }
+
+ /* Can not allocate more stripes */
+ if (j == lod->lod_remote_mdt_count) {
+ CDEBUG(D_INFO, "%s: require stripes %d only get %d\n",
+ lod2obd(lod)->obd_name, stripe_count, i - 1);
+ break;
+ }
+
+ CDEBUG(D_INFO, "idx %d, mdt cnt %d,"
+ " allocated %d, last allocated %d\n", idx,
+ lod->lod_remote_mdt_count, i, idx_array[i - 1]);
+
+ tgt = LTD_TGT(ltd, idx);
+ LASSERT(tgt != NULL);
+
+ rc = obd_fid_alloc(tgt->ltd_exp, &fid, NULL);
+ if (rc < 0)
+ GOTO(out_put, rc);
+ rc = 0;
+
+ conf.loc_flags = LOC_F_NEW;
+ dto = dt_locate_at(env, tgt->ltd_tgt, &fid,
+ dt->do_lu.lo_dev->ld_site->ls_top_dev, &conf);
+ if (IS_ERR(dto))
+ GOTO(out_put, rc = PTR_ERR(dto));
+ stripe[i - 1] = dto;
+ idx_array[i] = idx;
+ }
+
+ lo->ldo_dir_striped = 1;
+ lo->ldo_stripe = stripe;
+ lo->ldo_stripenr = i - 1;
+ lo->ldo_stripes_allocated = stripe_count - 1;
+
+ if (lo->ldo_stripenr == 0)
+ GOTO(out_put, rc = -ENOSPC);
+
+ rc = lod_prep_lmv_md(env, dt, &lmv_buf);
+ if (rc != 0)
+ GOTO(out_put, rc);
+
+ for (i = 0; i < lo->ldo_stripenr; i++) {
+ struct dt_object *dto;
+
+ dto = stripe[i];
+ /* only create slave striped object */
+ rc = dt_declare_create(env, dto, attr, NULL, NULL, th);
+ if (rc != 0)
+ GOTO(out_put, rc);
+
+ if (!dt_try_as_dir(env, dto))
+ GOTO(out_put, rc = -EINVAL);
+
+ rc = dt_declare_insert(env, dto,
+ (const struct dt_rec *)lu_object_fid(&dto->do_lu),
+ (const struct dt_key *)dot, th);
+ if (rc != 0)
+ GOTO(out_put, rc);
+
+ /* master stripe FID will be put to .. */
+ rc = dt_declare_insert(env, dto,
+ (const struct dt_rec *)lu_object_fid(&dt->do_lu),
+ (const struct dt_key *)dotdot, th);
+ if (rc != 0)
+ GOTO(out_put, rc);
+
+ /* probably nothing to inherite */
+ if (lo->ldo_striping_cached &&
+ !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
+ lo->ldo_def_stripenr,
+ lo->ldo_def_stripe_offset)) {
+ struct lod_thread_info *info;
+ struct lov_user_md_v3 *v3;
+
+ /* sigh, lti_ea_store has been used for lmv_buf,
+ * so we have to allocate buffer for default
+ * stripe EA */
+ OBD_ALLOC_PTR(v3);
+ if (v3 == NULL)
+ GOTO(out_put, rc = -ENOMEM);
+
+ memset(v3, 0, sizeof(*v3));
+ v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
+ v3->lmm_stripe_count =
+ cpu_to_le32(lo->ldo_def_stripenr);
+ v3->lmm_stripe_offset =
+ cpu_to_le32(lo->ldo_def_stripe_offset);
+ v3->lmm_stripe_size =
+ cpu_to_le32(lo->ldo_def_stripe_size);
+ if (lo->ldo_pool)
+ strncpy(v3->lmm_pool_name, lo->ldo_pool,
+ LOV_MAXPOOLNAME);
+
+ info = lod_env_info(env);
+ info->lti_buf.lb_buf = v3;
+ info->lti_buf.lb_len = sizeof(*v3);
+ rc = dt_declare_xattr_set(env, dto,
+ &info->lti_buf,
+ XATTR_NAME_LOV,
+ 0, th);
+ OBD_FREE_PTR(v3);
+ if (rc != 0)
+ GOTO(out_put, rc);
+ }
+ rc = dt_declare_xattr_set(env, dto, &lmv_buf, XATTR_NAME_LMV, 0,
+ th);
+ if (rc != 0)
+ GOTO(out_put, rc);
+ }
+
+ rc = dt_declare_xattr_set(env, dt_object_child(dt), &lmv_buf,
+ XATTR_NAME_LMV, 0, th);
+ if (rc != 0)
+ GOTO(out_put, rc);
+
+out_put:
+ if (rc < 0) {
+ for (i = 0; i < stripe_count - 1; i++)
+ if (stripe[i] != NULL)
+ lu_object_put(env, &stripe[i]->do_lu);
+ OBD_FREE(stripe, sizeof(stripe[0]) * (stripe_count - 1));
+ }
+
+out_free:
+ if (idx_array != NULL)
+ OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
+
+ RETURN(rc);
+}
+
+/**
+ * Declare create striped md object.
+ */
+static int lod_declare_xattr_set_lmv(const struct lu_env *env,
+ struct dt_object *dt,
+ struct lu_attr *attr,
+ const struct lu_buf *lum_buf,
+ struct thandle *th)
+{
+ struct lod_object *lo = lod_dt_obj(dt);
+ struct lod_device *lod = lu2lod_dev(dt->do_lu.lo_dev);
+ struct lmv_user_md_v1 *lum;
+ int rc;
+ ENTRY;
+
+ lum = lum_buf->lb_buf;
+ LASSERT(lum != NULL);
+
+ CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
+ le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
+ (int)le32_to_cpu(lum->lum_stripe_offset));
+
+ if (le32_to_cpu(lum->lum_stripe_count) <= 1)
+ GOTO(out, rc = 0);
+
+ rc = lod_verify_md_striping(lod, lum);
+ if (rc != 0)
+ GOTO(out, rc);
+
+ /* prepare dir striped objects */
+ rc = lod_prep_md_striped_create(env, dt, attr, lum, th);
+ if (rc != 0) {
+ /* failed to create striping, let's reset
+ * config so that others don't get confused */
+ lod_object_free_striping(env, lo);
+ GOTO(out, rc);
+ }
+out:
+ RETURN(rc);
+}
+