X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Flod%2Flod_lov.c;h=03388799f42f6887eee002f03f35bed9765bdc35;hp=5763ef4bbb280d6427d5185661a6ff7c58eeb301;hb=abe2d7d54a94f63374557343015f2080e750b9db;hpb=66523bca8a821ebba16f25c3b3a1ce64b906e443

diff --git a/lustre/lod/lod_lov.c b/lustre/lod/lod_lov.c
index 5763ef4..0338879 100644
--- a/lustre/lod/lod_lov.c
+++ b/lustre/lod/lod_lov.c
@@ -96,11 +96,15 @@ void lod_putref(struct lod_device *lod)
 		cfs_list_for_each_entry_safe(ost_desc, tmp, &kill, ltd_kill) {
 			int rc;
 			cfs_list_del(&ost_desc->ltd_kill);
-			/* XXX: remove from QoS structures */
-			/* disconnect from OSP */
+			/* remove from QoS structures */
+			rc = qos_del_tgt(lod, ost_desc);
+			if (rc)
+				CERROR("%s: qos_del_tgt(%s) failed: rc = %d\n",
+				       lod2obd(lod)->obd_name,
+				       obd_uuid2str(&ost_desc->ltd_uuid), rc);
 			rc = obd_disconnect(ost_desc->ltd_exp);
 			if (rc)
-				CERROR("%s: failed to disconnect %s (%d)\n",
+				CERROR("%s: failed to disconnect %s: rc = %d\n",
 				       lod2obd(lod)->obd_name,
 				       obd_uuid2str(&ost_desc->ltd_uuid), rc);
 			OBD_FREE_PTR(ost_desc);
@@ -268,6 +272,13 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
 		GOTO(out_mutex, rc);
 	}
 
+	rc = qos_add_tgt(lod, ost_desc);
+	if (rc) {
+		CERROR("%s: qos_add_tgt(%s) failed: rc = %d\n", obd->obd_name,
+		       obd_uuid2str(&ost_desc->ltd_uuid), rc);
+		GOTO(out_pool, rc);
+	}
+
 	/* The new OST is now a full citizen */
 	if (index >= lod->lod_desc.ld_tgt_count)
 		lod->lod_desc.ld_tgt_count = index + 1;
@@ -284,6 +295,7 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
 
 	RETURN(0);
 
+out_pool:
 	lod_ost_pool_remove(&lod->lod_pool_info, index);
 out_mutex:
 	cfs_mutex_unlock(&lod->lod_mutex);
@@ -369,6 +381,425 @@ out:
 	return(rc);
 }
 
+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));
+	if (info->lti_ea_store) {
+		LASSERT(info->lti_ea_store_size);
+		LASSERT(info->lti_ea_store_size < round);
+		CDEBUG(D_INFO, "EA store size %d is not enough, need %d\n",
+		       info->lti_ea_store_size, round);
+		OBD_FREE_LARGE(info->lti_ea_store, info->lti_ea_store_size);
+		info->lti_ea_store = NULL;
+		info->lti_ea_store_size = 0;
+	}
+
+	OBD_ALLOC_LARGE(info->lti_ea_store, round);
+	if (info->lti_ea_store == NULL)
+		RETURN(-ENOMEM);
+	info->lti_ea_store_size = round;
+	RETURN(0);
+}
+
+/*
+ * generate and write LOV EA for given striped object
+ */
+int lod_generate_and_set_lovea(const struct lu_env *env,
+			       struct lod_object *lo, struct thandle *th)
+{
+	struct lod_thread_info	*info = lod_env_info(env);
+	struct dt_object	*next = dt_object_child(&lo->ldo_obj);
+	const struct lu_fid	*fid  = lu_object_fid(&lo->ldo_obj.do_lu);
+	struct lov_mds_md_v1	*lmm;
+	struct lov_ost_data_v1	*objs;
+	__u32			 magic;
+	int			 i, rc, lmm_size;
+	ENTRY;
+
+	LASSERT(lo);
+	LASSERT(lo->ldo_stripenr > 0);
+
+	magic = lo->ldo_pool ? 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);
+		if (rc)
+			RETURN(rc);
+	}
+
+	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_stripe_size = cpu_to_le32(lo->ldo_stripe_size);
+	lmm->lmm_stripe_count = cpu_to_le16(lo->ldo_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);
+		objs = &v3->lmm_objects[0];
+	}
+
+	for (i = 0; i < lo->ldo_stripenr; i++) {
+		const struct lu_fid *fid;
+
+		LASSERT(lo->ldo_stripe[i]);
+		fid = lu_object_fid(&lo->ldo_stripe[i]->do_lu);
+
+		rc = fid_ostid_pack(fid, &info->lti_ostid);
+		LASSERT(rc == 0);
+		LASSERT(info->lti_ostid.oi_seq == FID_SEQ_OST_MDT0);
+
+		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);
+		objs[i].l_ost_gen    = cpu_to_le32(0);
+		objs[i].l_ost_idx    = cpu_to_le32(fid_idif_ost_idx(fid));
+	}
+
+	info->lti_buf.lb_buf = lmm;
+	info->lti_buf.lb_len = lmm_size;
+	rc = dt_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV, 0,
+			  th, BYPASS_CAPA);
+
+	RETURN(rc);
+}
+
+int lod_get_lov_ea(const struct lu_env *env, struct lod_object *lo)
+{
+	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)) {
+		/* 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);
+	}
+	/* if object is not striped or inaccessible */
+	if (rc == -ENODATA)
+		RETURN(0);
+
+	if (rc == -ERANGE) {
+		/* EA doesn't fit, reallocate new buffer */
+		rc = dt_xattr_get(env, next, &LU_BUF_NULL, XATTR_NAME_LOV,
+				  BYPASS_CAPA);
+		if (rc == -ENODATA)
+			RETURN(0);
+		else if (rc < 0)
+			RETURN(rc);
+
+		LASSERT(rc > 0);
+		rc = lod_ea_store_resize(info, rc);
+		if (rc)
+			RETURN(rc);
+		goto repeat;
+	}
+
+	RETURN(rc);
+}
+
+int lod_store_def_striping(const struct lu_env *env, struct dt_object *dt,
+			   struct thandle *th)
+{
+	struct lod_thread_info	*info = lod_env_info(env);
+	struct lod_object	*lo = lod_dt_obj(dt);
+	struct dt_object	*next = dt_object_child(dt);
+	struct lov_user_md_v3	*v3;
+	int			 rc;
+	ENTRY;
+
+	LASSERT(S_ISDIR(dt->do_lu.lo_header->loh_attr));
+
+	/*
+	 * store striping defaults into new directory
+	 * used to implement defaults inheritance
+	 */
+
+	/* probably nothing to inherite */
+	if (lo->ldo_striping_cached == 0)
+		RETURN(0);
+
+	if (LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size, lo->ldo_def_stripenr,
+				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->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);
+
+	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);
+}
+
+/*
+ * allocate array of objects pointers, find/create objects
+ * stripenr and other fields should be initialized by this moment
+ */
+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 lu_object	*o, *n;
+	struct lu_device	*nd;
+	int			i, idx, rc = 0;
+	ENTRY;
+
+	LASSERT(lo);
+	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;
+
+	for (i = 0; i < lo->ldo_stripenr; i++) {
+
+		info->lti_ostid.oi_id = le64_to_cpu(objs[i].l_object_id);
+		/* XXX: support for DNE? */
+		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);
+
+		/*
+		 * 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);
+		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);
+		nd = &OST_TGT(md,idx)->ltd_ost->dd_lu_dev;
+		lod_putref(md);
+
+		o = lu_object_find_at(env, nd, &info->lti_fid, NULL);
+		if (IS_ERR(o))
+			GOTO(out, rc = PTR_ERR(o));
+
+		n = lu_object_locate(o->lo_header, nd->ld_type);
+		LASSERT(n);
+
+		lo->ldo_stripe[i] = container_of(n, struct dt_object, do_lu);
+	}
+
+out:
+	RETURN(rc);
+}
+
+/*
+ * Parse striping information stored in lti_ea_store
+ */
+int lod_parse_striping(const struct lu_env *env, struct lod_object *lo,
+		       const struct lu_buf *buf)
+{
+	struct lov_mds_md_v1	*lmm;
+	struct lov_ost_data_v1	*objs;
+	__u32			 magic;
+	int			 rc = 0;
+	ENTRY;
+
+	LASSERT(buf);
+	LASSERT(buf->lb_buf);
+	LASSERT(buf->lb_len);
+
+	lmm = (struct lov_mds_md_v1 *) buf->lb_buf;
+	magic = le32_to_cpu(lmm->lmm_magic);
+
+	if (magic != LOV_MAGIC_V1 && magic != LOV_MAGIC_V3)
+		GOTO(out, rc = -EINVAL);
+	if (le32_to_cpu(lmm->lmm_pattern) != LOV_PATTERN_RAID0)
+		GOTO(out, rc = -EINVAL);
+
+	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);
+
+	LASSERT(buf->lb_len >= lov_mds_md_size(lo->ldo_stripenr, magic));
+
+	if (magic == LOV_MAGIC_V3) {
+		struct lov_mds_md_v3 *v3 = (struct lov_mds_md_v3 *) lmm;
+		objs = &v3->lmm_objects[0];
+		lod_object_set_pool(lo, v3->lmm_pool_name);
+	} else {
+		objs = &lmm->lmm_objects[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)
+{
+	struct lod_thread_info	*info = lod_env_info(env);
+	struct dt_object	*next = dt_object_child(&lo->ldo_obj);
+	int			 rc;
+	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);
+		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))
+		GOTO(out, rc = 0);
+
+	LASSERT(lo->ldo_stripenr == 0);
+
+	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
+	 */
+	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);
+out:
+	dt_write_unlock(env, next);
+	RETURN(rc);
+}
+
+int lod_verify_striping(struct lod_device *d, const struct lu_buf *buf,
+			int specific)
+{
+	struct lov_user_md_v1	*lum;
+	struct lov_user_md_v3	*v3 = NULL;
+	struct pool_desc	*pool = NULL;
+	int			 rc;
+	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);
+	}
+
+	if ((specific && lum->lmm_pattern != LOV_PATTERN_RAID0) ||
+	    (specific == 0 && lum->lmm_pattern != 0)) {
+		CDEBUG(D_IOCTL, "bad userland stripe pattern: %#x\n",
+		       lum->lmm_pattern);
+		RETURN(-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);
+	}
+
+	/* an offset of -1 is treated as a "special" valid offset */
+	if (lum->lmm_stripe_offset != (typeof(lum->lmm_stripe_offset))(-1)) {
+		/* if offset is not within valid range [0, osts_size) */
+		if (lum->lmm_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);
+		}
+
+		/* if lmm_stripe_offset is *not* in bitmap */
+		if (!cfs_bitmap_check(d->lod_ost_bitmap,
+				      lum->lmm_stripe_offset)) {
+			CDEBUG(D_IOCTL, "stripe offset %u not in bitmap\n",
+			       lum->lmm_stripe_offset);
+			RETURN(-EINVAL);
+		}
+	}
+
+	if (lum->lmm_magic == LOV_USER_MAGIC_V3)
+		v3 = buf->lb_buf;
+
+	if (v3)
+		pool = lod_find_pool(d, v3->lmm_pool_name);
+
+	if (pool != NULL) {
+		__u16 offs = v3->lmm_stripe_offset;
+
+		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);
+			}
+		}
+
+		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);
+		}
+
+		lod_pool_putref(pool);
+	}
+
+	RETURN(0);
+}
+
 void lod_fix_desc_stripe_size(__u64 *val)
 {
 	if (*val < PTLRPC_MAX_BRW_SIZE) {