X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Flod%2Flod_object.c;h=1fac13455c15df0ddb228e673c0bcf486dfce3dd;hb=37f6357a5c9f;hp=bf56b505b896174ad9edf309ed8d4aa635ce00a0;hpb=4d534365ab214e28452c54fd2e0d4781e2f290d6;p=fs%2Flustre-release.git

diff --git a/lustre/lod/lod_object.c b/lustre/lod/lod_object.c
index bf56b50..1fac134 100644
--- a/lustre/lod/lod_object.c
+++ b/lustre/lod/lod_object.c
@@ -23,7 +23,7 @@
  * Copyright  2009 Sun Microsystems, Inc. All rights reserved
  * Use is subject to license terms.
  *
- * Copyright (c) 2012, 2016, Intel Corporation.
+ * Copyright (c) 2012, 2017, Intel Corporation.
  */
 /*
  * lustre/lod/lod_object.c
@@ -40,6 +40,8 @@
 
 #define DEBUG_SUBSYSTEM S_MDS
 
+#include <linux/random.h>
+
 #include <obd.h>
 #include <obd_class.h>
 #include <obd_support.h>
@@ -1046,6 +1048,18 @@ static int lod_attr_get(const struct lu_env *env,
 	return dt_attr_get(env, dt_object_child(dt), attr);
 }
 
+static inline void lod_adjust_stripe_info(struct lod_layout_component *comp,
+					  struct lov_desc *desc)
+{
+	if (comp->llc_pattern != LOV_PATTERN_MDT) {
+		if (!comp->llc_stripe_count)
+			comp->llc_stripe_count =
+				desc->ld_default_stripe_count;
+	}
+	if (comp->llc_stripe_size <= 0)
+		comp->llc_stripe_size = desc->ld_default_stripe_size;
+}
+
 int lod_obj_for_each_stripe(const struct lu_env *env, struct lod_object *lo,
 			    struct thandle *th,
 			    struct lod_obj_stripe_cb_data *data)
@@ -1859,10 +1873,12 @@ static int lod_prep_md_striped_create(const struct lu_env *env,
 	int			rc = 0;
 	__u32			i;
 	__u32			j;
+	bool			is_specific = false;
 	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_magic) == LMV_USER_MAGIC ||
+		le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC_SPECIFIC);
 	LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
 
 	stripe_count = le32_to_cpu(lum->lum_stripe_count);
@@ -1878,6 +1894,12 @@ static int lod_prep_md_striped_create(const struct lu_env *env,
 	/* Start index must be the master MDT */
 	master_index = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
 	idx_array[0] = master_index;
+	if (le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC_SPECIFIC) {
+		is_specific = true;
+		for (i = 1; i < stripe_count; i++)
+			idx_array[i] = le32_to_cpu(lum->lum_objects[i].lum_mds);
+	}
+
 	for (i = 0; i < stripe_count; i++) {
 		struct lod_tgt_desc	*tgt = NULL;
 		struct dt_object	*dto;
@@ -1896,7 +1918,8 @@ static int lod_prep_md_striped_create(const struct lu_env *env,
 			CDEBUG(D_INFO, "try idx %d, mdt cnt %u, allocated %u\n",
 			       idx, lod->lod_remote_mdt_count + 1, i);
 
-			if (likely(!OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE))) {
+			if (likely(!is_specific &&
+				   !OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE))) {
 				/* check whether the idx already exists
 				 * in current allocated array */
 				for (k = 0; k < i; k++) {
@@ -1959,7 +1982,7 @@ static int lod_prep_md_striped_create(const struct lu_env *env,
 		       idx, i, PFID(&fid));
 		idx_array[i] = idx;
 		/* Set the start index for next stripe allocation */
-		if (i < stripe_count - 1)
+		if (!is_specific && i < stripe_count - 1)
 			idx_array[i + 1] = (idx + 1) %
 					   (lod->lod_remote_mdt_count + 1);
 		/* tgt_dt and fid must be ready after search avaible OSP
@@ -2042,7 +2065,7 @@ static int lod_declare_xattr_set_lmv(const struct lu_env *env,
 	       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) == 0)
+	if (lo->ldo_dir_stripe_count == 0)
 		GOTO(out, rc = 0);
 
 	/* prepare dir striped objects */
@@ -2316,7 +2339,7 @@ static int lod_declare_layout_add(const struct lu_env *env,
 
 	LASSERT(lo->ldo_is_composite);
 
-	if (lo->ldo_flr_state != LCM_FL_NOT_FLR)
+	if (lo->ldo_flr_state != LCM_FL_NONE)
 		RETURN(-EBUSY);
 
 	rc = lod_verify_striping(d, lo, buf, false);
@@ -2355,14 +2378,8 @@ static int lod_declare_layout_add(const struct lu_env *env,
 		lod_comp->llc_flags = comp_v1->lcm_entries[i].lcme_flags;
 
 		lod_comp->llc_stripe_count = v1->lmm_stripe_count;
-		if (!lod_comp->llc_stripe_count ||
-		    lod_comp->llc_stripe_count == (__u16)-1)
-			lod_comp->llc_stripe_count =
-				desc->ld_default_stripe_count;
 		lod_comp->llc_stripe_size = v1->lmm_stripe_size;
-		if (!lod_comp->llc_stripe_size)
-			lod_comp->llc_stripe_size =
-				desc->ld_default_stripe_size;
+		lod_adjust_stripe_info(lod_comp, desc);
 
 		if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
 			v3 = (struct lov_user_md_v3 *) v1;
@@ -2438,7 +2455,7 @@ static int lod_declare_layout_set(const struct lu_env *env,
 	struct lod_device	*d = lu2lod_dev(dt->do_lu.lo_dev);
 	struct lod_object	*lo = lod_dt_obj(dt);
 	struct lov_comp_md_v1	*comp_v1 = buf->lb_buf;
-	__u32	magic, id;
+	__u32	magic;
 	int	i, j, rc;
 	bool	changed = false;
 	ENTRY;
@@ -2465,15 +2482,27 @@ static int lod_declare_layout_set(const struct lu_env *env,
 	}
 
 	for (i = 0; i < comp_v1->lcm_entry_count; i++) {
-		id = comp_v1->lcm_entries[i].lcme_id;
+		__u32 id = comp_v1->lcm_entries[i].lcme_id;
+		__u32 flags = comp_v1->lcm_entries[i].lcme_flags;
+
+		if (flags & LCME_FL_INIT) {
+			if (changed)
+				lod_object_free_striping(env, lo);
+			RETURN(-EINVAL);
+		}
 
 		for (j = 0; j < lo->ldo_comp_cnt; j++) {
 			lod_comp = &lo->ldo_comp_entries[j];
-			if (id == lod_comp->llc_id || id == LCME_ID_ALL) {
-				lod_comp->llc_flags =
-					comp_v1->lcm_entries[i].lcme_flags;
-				changed = true;
+			if (id != lod_comp->llc_id)
+				continue;
+
+			if (flags & LCME_FL_NEG) {
+				flags &= ~LCME_FL_NEG;
+				lod_comp->llc_flags &= ~flags;
+			} else {
+				lod_comp->llc_flags |= flags;
 			}
+			changed = true;
 		}
 	}
 
@@ -2486,8 +2515,8 @@ static int lod_declare_layout_set(const struct lu_env *env,
 	lod_obj_inc_layout_gen(lo);
 
 	info->lti_buf.lb_len = lod_comp_md_size(lo, false);
-	rc = lod_sub_declare_xattr_set(env, dt, &info->lti_buf,
-				       XATTR_NAME_LOV, 0, th);
+	rc = lod_sub_declare_xattr_set(env, dt_object_child(dt), &info->lti_buf,
+				       XATTR_NAME_LOV, LU_XATTR_REPLACE, th);
 	RETURN(rc);
 }
 
@@ -2520,7 +2549,7 @@ static int lod_declare_layout_del(const struct lu_env *env,
 
 	LASSERT(lo->ldo_is_composite);
 
-	if (lo->ldo_flr_state != LCM_FL_NOT_FLR)
+	if (lo->ldo_flr_state != LCM_FL_NONE)
 		RETURN(-EBUSY);
 
 	magic = comp_v1->lcm_magic;
@@ -2547,6 +2576,12 @@ static int lod_declare_layout_del(const struct lu_env *env,
 		RETURN(-EINVAL);
 	}
 
+	if (id == LCME_ID_INVAL && !flags) {
+		CDEBUG(D_LAYOUT, "%s: no id or flags specified.\n",
+		       lod2obd(d)->obd_name);
+		RETURN(-EINVAL);
+	}
+
 	if (flags & LCME_FL_NEG) {
 		neg_flags = flags & ~LCME_FL_NEG;
 		flags = 0;
@@ -2729,7 +2764,7 @@ static int lod_layout_convert(struct lod_thread_info *info)
 	lcm->lcm_size = cpu_to_le32(size);
 	lcm->lcm_layout_gen = cpu_to_le32(le16_to_cpu(
 						lmm_save->lmm_layout_gen));
-	lcm->lcm_flags = cpu_to_le16(LCM_FL_NOT_FLR);
+	lcm->lcm_flags = cpu_to_le16(LCM_FL_NONE);
 	lcm->lcm_entry_count = cpu_to_le16(1);
 	lcm->lcm_mirror_count = 0;
 
@@ -2878,7 +2913,7 @@ static int lod_declare_layout_merge(const struct lu_env *env,
 	lcm->lcm_size = cpu_to_le32(size);
 	lcm->lcm_entry_count = cpu_to_le16(cur_entry_count + merge_entry_count);
 	lcm->lcm_mirror_count = cpu_to_le16(mirror_count);
-	if ((le16_to_cpu(lcm->lcm_flags) & LCM_FL_FLR_MASK) == LCM_FL_NOT_FLR)
+	if ((le16_to_cpu(lcm->lcm_flags) & LCM_FL_FLR_MASK) == LCM_FL_NONE)
 		lcm->lcm_flags = cpu_to_le32(LCM_FL_RDONLY);
 
 	LASSERT(dt_write_locked(env, dt_object_child(dt)));
@@ -2896,6 +2931,31 @@ out:
 }
 
 /**
+ * Split layouts, just set the LOVEA with the layout from mbuf.
+ */
+static int lod_declare_layout_split(const struct lu_env *env,
+		struct dt_object *dt, const struct lu_buf *mbuf,
+		struct thandle *th)
+{
+	struct lod_object *lo = lod_dt_obj(dt);
+	struct lov_comp_md_v1 *lcm = mbuf->lb_buf;
+	int rc;
+	ENTRY;
+
+	lod_obj_inc_layout_gen(lo);
+	lcm->lcm_layout_gen = cpu_to_le32(lo->ldo_layout_gen);
+
+	lod_object_free_striping(env, lo);
+	rc = lod_parse_striping(env, lo, mbuf);
+	if (rc)
+		RETURN(rc);
+
+	rc = lod_sub_declare_xattr_set(env, dt_object_child(dt), mbuf,
+				       XATTR_NAME_LOV, LU_XATTR_REPLACE, th);
+	RETURN(rc);
+}
+
+/**
  * Implementation of dt_object_operations::do_declare_xattr_set.
  *
  * \see dt_object_operations::do_declare_xattr_set() in the API description
@@ -2919,7 +2979,7 @@ static int lod_declare_xattr_set(const struct lu_env *env,
 
 	mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
 	if ((S_ISREG(mode) || mode == 0) &&
-	    !(fl & (LU_XATTR_REPLACE | LU_XATTR_MERGE)) &&
+	    !(fl & (LU_XATTR_REPLACE | LU_XATTR_MERGE | LU_XATTR_SPLIT)) &&
 	    (strcmp(name, XATTR_NAME_LOV) == 0 ||
 	     strcmp(name, XATTR_LUSTRE_LOV) == 0)) {
 		/*
@@ -2945,6 +3005,10 @@ static int lod_declare_xattr_set(const struct lu_env *env,
 		LASSERT(strcmp(name, XATTR_NAME_LOV) == 0 ||
 			strcmp(name, XATTR_LUSTRE_LOV) == 0);
 		rc = lod_declare_layout_merge(env, dt, buf, th);
+	} else if (fl & LU_XATTR_SPLIT) {
+		LASSERT(strcmp(name, XATTR_NAME_LOV) == 0 ||
+			strcmp(name, XATTR_LUSTRE_LOV) == 0);
+		rc = lod_declare_layout_split(env, dt, buf, th);
 	} else if (S_ISREG(mode) &&
 		   strlen(name) > strlen(XATTR_LUSTRE_LOV) + 1 &&
 		   strncmp(name, XATTR_LUSTRE_LOV,
@@ -3357,6 +3421,7 @@ out:
  * \param[in] env	execution environment
  * \param[in] dt	object
  * \param[in] attr	attributes the stripes will be created with
+ * \param[in] lmu	lmv_user_md if MDT indices are specified
  * \param[in] dof	format of stripes (see OSD API description)
  * \param[in] th	transaction handle
  * \param[in] declare	where to call "declare" or "execute" methods
@@ -3367,6 +3432,7 @@ out:
 static int lod_dir_striping_create_internal(const struct lu_env *env,
 					    struct dt_object *dt,
 					    struct lu_attr *attr,
+					    const struct lu_buf *lmu,
 					    struct dt_object_format *dof,
 					    struct thandle *th,
 					    bool declare)
@@ -3383,31 +3449,34 @@ static int lod_dir_striping_create_internal(const struct lu_env *env,
 
 	if (!LMVEA_DELETE_VALUES(lo->ldo_dir_stripe_count,
 				 lo->ldo_dir_stripe_offset)) {
-		struct lmv_user_md_v1 *v1 = info->lti_ea_store;
-		int stripe_count = lo->ldo_dir_stripe_count;
+		if (!lmu) {
+			struct lmv_user_md_v1 *v1 = info->lti_ea_store;
+			int stripe_count = lo->ldo_dir_stripe_count;
 
-		if (info->lti_ea_store_size < sizeof(*v1)) {
-			rc = lod_ea_store_resize(info, sizeof(*v1));
-			if (rc != 0)
-				RETURN(rc);
-			v1 = info->lti_ea_store;
-		}
+			if (info->lti_ea_store_size < sizeof(*v1)) {
+				rc = lod_ea_store_resize(info, sizeof(*v1));
+				if (rc != 0)
+					RETURN(rc);
+				v1 = info->lti_ea_store;
+			}
 
-		memset(v1, 0, sizeof(*v1));
-		v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
-		v1->lum_stripe_count = cpu_to_le32(stripe_count);
-		v1->lum_stripe_offset =
-				cpu_to_le32(lo->ldo_dir_stripe_offset);
+			memset(v1, 0, sizeof(*v1));
+			v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
+			v1->lum_stripe_count = cpu_to_le32(stripe_count);
+			v1->lum_stripe_offset =
+					cpu_to_le32(lo->ldo_dir_stripe_offset);
 
-		info->lti_buf.lb_buf = v1;
-		info->lti_buf.lb_len = sizeof(*v1);
+			info->lti_buf.lb_buf = v1;
+			info->lti_buf.lb_len = sizeof(*v1);
+			lmu = &info->lti_buf;
+		}
 
 		if (declare)
-			rc = lod_declare_xattr_set_lmv(env, dt, attr,
-						       &info->lti_buf, dof, th);
+			rc = lod_declare_xattr_set_lmv(env, dt, attr, lmu, dof,
+						       th);
 		else
-			rc = lod_xattr_set_lmv(env, dt, &info->lti_buf,
-					       XATTR_NAME_LMV, 0, th);
+			rc = lod_xattr_set_lmv(env, dt, lmu, XATTR_NAME_LMV, 0,
+					       th);
 		if (rc != 0)
 			RETURN(rc);
 	}
@@ -3485,10 +3554,12 @@ static int lod_dir_striping_create_internal(const struct lu_env *env,
 static int lod_declare_dir_striping_create(const struct lu_env *env,
 					   struct dt_object *dt,
 					   struct lu_attr *attr,
+					   struct lu_buf *lmu,
 					   struct dt_object_format *dof,
 					   struct thandle *th)
 {
-	return lod_dir_striping_create_internal(env, dt, attr, dof, th, true);
+	return lod_dir_striping_create_internal(env, dt, attr, lmu, dof, th,
+						true);
 }
 
 static int lod_dir_striping_create(const struct lu_env *env,
@@ -3497,7 +3568,8 @@ static int lod_dir_striping_create(const struct lu_env *env,
 				   struct dt_object_format *dof,
 				   struct thandle *th)
 {
-	return lod_dir_striping_create_internal(env, dt, attr, dof, th, false);
+	return lod_dir_striping_create_internal(env, dt, attr, NULL, dof, th,
+						false);
 }
 
 /**
@@ -3657,6 +3729,10 @@ out:
 	return rc;
 }
 
+
+static int lod_get_default_lov_striping(const struct lu_env *env,
+					struct lod_object *lo,
+					struct lod_default_striping *lds);
 /**
  * Implementation of dt_object_operations::do_xattr_set.
  *
@@ -3698,8 +3774,59 @@ static int lod_xattr_set(const struct lu_env *env,
 
 	if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
 	    strcmp(name, XATTR_NAME_LOV) == 0) {
-		/* default LOVEA */
-		rc = lod_xattr_set_lov_on_dir(env, dt, buf, name, fl, th);
+		struct lod_thread_info *info = lod_env_info(env);
+		struct lod_default_striping *lds = &info->lti_def_striping;
+		struct lov_user_md_v1 *v1 = buf->lb_buf;
+		char pool[LOV_MAXPOOLNAME + 1];
+		bool is_del;
+
+		/* get existing striping config */
+		rc = lod_get_default_lov_striping(env, lod_dt_obj(dt), lds);
+		if (rc)
+			RETURN(rc);
+
+		memset(pool, 0, sizeof(pool));
+		if (lds->lds_def_striping_set == 1)
+			lod_layout_get_pool(lds->lds_def_comp_entries,
+					    lds->lds_def_comp_cnt, pool,
+					    sizeof(pool));
+
+		is_del = LOVEA_DELETE_VALUES(v1->lmm_stripe_size,
+					     v1->lmm_stripe_count,
+					     v1->lmm_stripe_offset,
+					     NULL);
+
+		/* Retain the pool name if it is not given */
+		if (v1->lmm_magic == LOV_USER_MAGIC_V1 && pool[0] != '\0' &&
+			!is_del) {
+			struct lod_thread_info *info = lod_env_info(env);
+			struct lov_user_md_v3 *v3  = info->lti_ea_store;
+
+			memset(v3, 0, sizeof(*v3));
+			v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
+			v3->lmm_pattern = cpu_to_le32(v1->lmm_pattern);
+			v3->lmm_stripe_count =
+					cpu_to_le32(v1->lmm_stripe_count);
+			v3->lmm_stripe_offset =
+					cpu_to_le32(v1->lmm_stripe_offset);
+			v3->lmm_stripe_size = cpu_to_le32(v1->lmm_stripe_size);
+
+			strlcpy(v3->lmm_pool_name, pool,
+				sizeof(v3->lmm_pool_name));
+
+			info->lti_buf.lb_buf = v3;
+			info->lti_buf.lb_len = sizeof(*v3);
+			rc = lod_xattr_set_lov_on_dir(env, dt, &info->lti_buf,
+						      name, fl, th);
+		} else {
+			rc = lod_xattr_set_lov_on_dir(env, dt, buf, name,
+						      fl, th);
+		}
+
+		if (lds->lds_def_striping_set == 1 &&
+		    lds->lds_def_comp_entries != NULL)
+			lod_free_def_comp_entries(lds);
+
 		RETURN(rc);
 	} else if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
 		   strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
@@ -4104,13 +4231,7 @@ static void lod_striping_from_default(struct lod_object *lo,
 			if (!lo->ldo_is_composite)
 				continue;
 
-			if (obj_comp->llc_stripe_count <= 0 &&
-			    obj_comp->llc_pattern != LOV_PATTERN_MDT)
-				obj_comp->llc_stripe_count =
-					desc->ld_default_stripe_count;
-			if (obj_comp->llc_stripe_size <= 0)
-				obj_comp->llc_stripe_size =
-					desc->ld_default_stripe_size;
+			lod_adjust_stripe_info(obj_comp, desc);
 		}
 	} else if (lds->lds_dir_def_striping_set && S_ISDIR(mode)) {
 		if (lo->ldo_dir_stripe_count == 0)
@@ -4220,7 +4341,8 @@ static void lod_ah_init(const struct lu_env *env,
 		 * stripe count and try to create dir by default stripe.
 		 */
 		if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0 &&
-		    le32_to_cpu(lum1->lum_magic) == LMV_USER_MAGIC) {
+		    (le32_to_cpu(lum1->lum_magic) == LMV_USER_MAGIC ||
+		     le32_to_cpu(lum1->lum_magic) == LMV_USER_MAGIC_SPECIFIC)) {
 			lc->ldo_dir_stripe_count =
 				le32_to_cpu(lum1->lum_stripe_count);
 			lc->ldo_dir_stripe_offset =
@@ -4342,12 +4464,7 @@ out:
 		LASSERT(!lc->ldo_is_composite);
 		lod_comp = &lc->ldo_comp_entries[0];
 		desc = &d->lod_desc;
-		if (lod_comp->llc_stripe_count <= 0)
-			lod_comp->llc_stripe_count =
-				desc->ld_default_stripe_count;
-		if (lod_comp->llc_stripe_size <= 0)
-			lod_comp->llc_stripe_size =
-				desc->ld_default_stripe_size;
+		lod_adjust_stripe_info(lod_comp, desc);
 	}
 
 	EXIT;
@@ -4556,6 +4673,8 @@ static int lod_declare_create(const struct lu_env *env, struct dt_object *dt,
 							NULL, th);
 	} else if (dof->dof_type == DFT_DIR) {
 		struct seq_server_site *ss;
+		struct lu_buf buf = { NULL };
+		struct lu_buf *lmu = NULL;
 
 		ss = lu_site2seq(dt->do_lu.lo_dev->ld_site);
 
@@ -4605,9 +4724,14 @@ static int lod_declare_create(const struct lu_env *env, struct dt_object *dt,
 				else
 					GOTO(out, rc = -EINVAL);
 			}
+		} else if (hint && hint->dah_eadata) {
+			lmu = &buf;
+			lmu->lb_buf = (void *)hint->dah_eadata;
+			lmu->lb_len = hint->dah_eadata_len;
 		}
 
-		rc = lod_declare_dir_striping_create(env, dt, attr, dof, th);
+		rc = lod_declare_dir_striping_create(env, dt, attr, lmu, dof,
+						     th);
 	}
 out:
 	/* failed to create striping or to set initial size, let's reset
@@ -5298,7 +5422,7 @@ static int lod_declare_update_plain(const struct lu_env *env,
 	int i, rc;
 	ENTRY;
 
-	LASSERT(lo->ldo_flr_state == LCM_FL_NOT_FLR);
+	LASSERT(lo->ldo_flr_state == LCM_FL_NONE);
 
 	/*
 	 * In case the client is passing lovea, which only happens during
@@ -5336,15 +5460,6 @@ static int lod_declare_update_plain(const struct lu_env *env,
 			GOTO(out, rc);
 	}
 
-	if (layout->li_opc == LAYOUT_INTENT_TRUNC) {
-		/**
-		 * trunc transfers [size, eof) in the intent extent, while
-		 * we'd instantiated components covers [0, size).
-		 */
-		layout->li_extent.e_end = layout->li_extent.e_start;
-		layout->li_extent.e_start = 0;
-	}
-
 	/* Make sure defined layout covers the requested write range. */
 	lod_comp = &lo->ldo_comp_entries[lo->ldo_comp_cnt - 1];
 	if (lo->ldo_comp_cnt > 1 &&
@@ -5466,27 +5581,62 @@ static void lod_stale_components(struct lod_object *lo, int primary,
 	}
 }
 
-static int lod_declare_update_rdonly(const struct lu_env *env,
-		struct lod_object *lo, struct md_layout_change *mlc,
-		struct thandle *th)
+/**
+ * check an OST's availability
+ * \param[in] env	execution environment
+ * \param[in] lo	lod object
+ * \param[in] dt	dt object
+ * \param[in] index	mirror index
+ *
+ * \retval	negative if failed
+ * \retval	1 if \a dt is available
+ * \retval	0 if \a dt is not available
+ */
+static inline int lod_check_ost_avail(const struct lu_env *env,
+				      struct lod_object *lo,
+				      struct dt_object *dt, int index)
 {
-	struct lod_thread_info *info = lod_env_info(env);
-	struct lu_attr *layout_attr = &info->lti_layout_attr;
-	struct lod_layout_component *lod_comp;
-	struct layout_intent *layout = mlc->mlc_intent;
-	struct lu_extent extent = layout->li_extent;
-	unsigned int seq = 0;
-	int picked;
-	int i;
+	struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
+	struct lod_tgt_desc *ost;
+	__u32 idx;
+	int type = LU_SEQ_RANGE_OST;
 	int rc;
-	ENTRY;
 
-	LASSERT(mlc->mlc_opc == MD_LAYOUT_WRITE);
-	LASSERT(lo->ldo_flr_state == LCM_FL_RDONLY);
-	LASSERT(lo->ldo_mirror_count > 0);
+	rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu), &idx, &type);
+	if (rc < 0) {
+		CERROR("%s: can't locate "DFID":rc = %d\n",
+		       lod2obd(lod)->obd_name, PFID(lu_object_fid(&dt->do_lu)),
+		       rc);
+		return rc;
+	}
+
+	ost = OST_TGT(lod, idx);
+	if (ost->ltd_statfs.os_state &
+		(OS_STATE_READONLY | OS_STATE_ENOSPC | OS_STATE_ENOINO) ||
+	    ost->ltd_active == 0) {
+		CDEBUG(D_LAYOUT, DFID ": mirror %d OST%d unavail, rc = %d\n",
+		       PFID(lod_object_fid(lo)), index, idx, rc);
+		return 0;
+	}
 
-	CDEBUG(D_LAYOUT, DFID": trying to write :"DEXT"\n",
-	       PFID(lod_object_fid(lo)), PEXT(&extent));
+	return 1;
+}
+
+/**
+ * Pick primary mirror for write
+ * \param[in] env	execution environment
+ * \param[in] lo	object
+ * \param[in] extent	write range
+ */
+static int lod_primary_pick(const struct lu_env *env, struct lod_object *lo,
+			    struct lu_extent *extent)
+{
+	struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
+	unsigned int seq = 0;
+	struct lod_layout_component *lod_comp;
+	int i, j, rc;
+	int picked = -1, second_pick = -1, third_pick = -1;
+	ENTRY;
 
 	if (OBD_FAIL_CHECK(OBD_FAIL_FLR_RANDOM_PICK_MIRROR)) {
 		get_random_bytes(&seq, sizeof(seq));
@@ -5494,55 +5644,221 @@ static int lod_declare_update_rdonly(const struct lu_env *env,
 	}
 
 	/**
-	 * Pick a mirror as the primary.
-	 * Now it only picks the first mirror, this algo can be
-	 * revised later after knowing the topology of cluster or
-	 * the availability of OSTs.
+	 * Pick a mirror as the primary, and check the availability of OSTs.
+	 *
+	 * This algo can be revised later after knowing the topology of
+	 * cluster.
 	 */
-	for (picked = -1, i = 0; i < lo->ldo_mirror_count; i++) {
+	lod_qos_statfs_update(env, lod);
+	for (i = 0; i < lo->ldo_mirror_count; i++) {
+		bool ost_avail = true;
 		int index = (i + seq) % lo->ldo_mirror_count;
 
-		if (!lo->ldo_mirrors[index].lme_stale) {
-			picked = index;
-			break;
+		if (lo->ldo_mirrors[index].lme_stale) {
+			CDEBUG(D_LAYOUT, DFID": mirror %d stale\n",
+			       PFID(lod_object_fid(lo)), index);
+			continue;
 		}
-	}
-	if (picked < 0) /* failed to pick a primary */
-		RETURN(-ENODATA);
 
-	CDEBUG(D_LAYOUT, DFID": picked mirror %u as primary\n",
-	       PFID(lod_object_fid(lo)), lo->ldo_mirrors[picked].lme_id);
+		/* 2nd pick is for the primary mirror containing unavail OST */
+		if (lo->ldo_mirrors[index].lme_primary && second_pick < 0)
+			second_pick = index;
 
-	/* stale overlapping components from other mirrors */
-	lod_stale_components(lo, picked, &extent);
+		/* 3rd pick is for non-primary mirror containing unavail OST */
+		if (second_pick < 0 && third_pick < 0)
+			third_pick = index;
 
-	/* instantiate components for the picked mirror, start from 0 */
-	if (layout->li_opc == LAYOUT_INTENT_TRUNC) {
 		/**
-		 * trunc transfers [size, eof) in the intent extent, we'd
-		 * stale components overlapping [size, eof), while we'd
-		 * instantiated components covers [0, size).
+		 * we found a non-primary 1st pick, we'd like to find a
+		 * potential pirmary mirror.
 		 */
-		extent.e_end = extent.e_start;
-	}
-	extent.e_start = 0;
+		if (picked >= 0 && !lo->ldo_mirrors[index].lme_primary)
+			continue;
+
+		/* check the availability of OSTs */
+		lod_foreach_mirror_comp(lod_comp, lo, index) {
+			if (!lod_comp_inited(lod_comp) || !lod_comp->llc_stripe)
+				continue;
+
+			for (j = 0; j < lod_comp->llc_stripe_count; j++) {
+				struct dt_object *dt = lod_comp->llc_stripe[j];
+
+				rc = lod_check_ost_avail(env, lo, dt, index);
+				if (rc < 0)
+					RETURN(rc);
+
+				ost_avail = !!rc;
+				if (!ost_avail)
+					break;
+			} /* for all dt object in one component */
+			if (!ost_avail)
+				break;
+		} /* for all components in a mirror */
+
+		/**
+		 * the OSTs where allocated objects locates in the components
+		 * of the mirror are available.
+		 */
+		if (!ost_avail)
+			continue;
 
-	lod_foreach_mirror_comp(lod_comp, lo, picked) {
-		if (!lu_extent_is_overlapped(&extent,
-					     &lod_comp->llc_extent))
+		/* this mirror has all OSTs available */
+		picked = index;
+
+		/**
+		 * primary with all OSTs are available, this is the perfect
+		 * 1st pick.
+		 */
+		if (lo->ldo_mirrors[index].lme_primary)
 			break;
+	} /* for all mirrors */
+
+	/* failed to pick a sound mirror, lower our expectation */
+	if (picked < 0)
+		picked = second_pick;
+	if (picked < 0)
+		picked = third_pick;
+	if (picked < 0)
+		RETURN(-ENODATA);
 
-		if (lod_comp_inited(lod_comp))
+	RETURN(picked);
+}
+
+/**
+ * figure out the components should be instantiated for resync.
+ */
+static int lod_prepare_resync(const struct lu_env *env, struct lod_object *lo,
+			      struct lu_extent *extent)
+{
+	struct lod_thread_info *info = lod_env_info(env);
+	struct lod_layout_component *lod_comp;
+	unsigned int need_sync = 0;
+	int i;
+
+	CDEBUG(D_LAYOUT,
+	       DFID": instantiate all stale components in "DEXT"\n",
+	       PFID(lod_object_fid(lo)), PEXT(extent));
+
+	/**
+	 * instantiate all components within this extent, even non-stale
+	 * components.
+	 */
+	for (i = 0; i < lo->ldo_mirror_count; i++) {
+		if (!lo->ldo_mirrors[i].lme_stale)
 			continue;
 
-		CDEBUG(D_LAYOUT, "instantiate: %u / %u\n",
-		       i, lod_comp_index(lo, lod_comp));
+		lod_foreach_mirror_comp(lod_comp, lo, i) {
+			if (!lu_extent_is_overlapped(extent,
+						&lod_comp->llc_extent))
+				break;
 
-		info->lti_comp_idx[info->lti_count++] =
-						lod_comp_index(lo, lod_comp);
+			need_sync++;
+
+			if (lod_comp_inited(lod_comp))
+				continue;
+
+			CDEBUG(D_LAYOUT, "resync instantiate %d / %d\n",
+			       i, lod_comp_index(lo, lod_comp));
+			info->lti_comp_idx[info->lti_count++] =
+					lod_comp_index(lo, lod_comp);
+		}
 	}
 
-	lo->ldo_flr_state = LCM_FL_WRITE_PENDING;
+	return need_sync ? 0 : -EALREADY;
+}
+
+static int lod_declare_update_rdonly(const struct lu_env *env,
+		struct lod_object *lo, struct md_layout_change *mlc,
+		struct thandle *th)
+{
+	struct lod_thread_info *info = lod_env_info(env);
+	struct lu_attr *layout_attr = &info->lti_layout_attr;
+	struct lod_layout_component *lod_comp;
+	struct lu_extent extent = { 0 };
+	int rc;
+	ENTRY;
+
+	LASSERT(lo->ldo_flr_state == LCM_FL_RDONLY);
+	LASSERT(mlc->mlc_opc == MD_LAYOUT_WRITE ||
+		mlc->mlc_opc == MD_LAYOUT_RESYNC);
+	LASSERT(lo->ldo_mirror_count > 0);
+
+	if (mlc->mlc_opc == MD_LAYOUT_WRITE) {
+		struct layout_intent *layout = mlc->mlc_intent;
+		int picked;
+
+		extent = layout->li_extent;
+		CDEBUG(D_LAYOUT, DFID": trying to write :"DEXT"\n",
+		       PFID(lod_object_fid(lo)), PEXT(&extent));
+
+		picked = lod_primary_pick(env, lo, &extent);
+		if (picked < 0)
+			RETURN(picked);
+
+		CDEBUG(D_LAYOUT, DFID": picked mirror id %u as primary\n",
+		       PFID(lod_object_fid(lo)),
+		       lo->ldo_mirrors[picked].lme_id);
+
+		if (layout->li_opc == LAYOUT_INTENT_TRUNC) {
+			/**
+			 * trunc transfers [0, size) in the intent extent, we'd
+			 * stale components overlapping [size, eof).
+			 */
+			extent.e_start = extent.e_end;
+			extent.e_end = OBD_OBJECT_EOF;
+		}
+
+		/* stale overlapping components from other mirrors */
+		lod_stale_components(lo, picked, &extent);
+
+		/* restore truncate intent extent */
+		if (layout->li_opc == LAYOUT_INTENT_TRUNC)
+			extent.e_end = extent.e_start;
+
+		/* instantiate components for the picked mirror, start from 0 */
+		extent.e_start = 0;
+
+		lod_foreach_mirror_comp(lod_comp, lo, picked) {
+			if (!lu_extent_is_overlapped(&extent,
+						     &lod_comp->llc_extent))
+				break;
+
+			if (lod_comp_inited(lod_comp))
+				continue;
+
+			info->lti_comp_idx[info->lti_count++] =
+						lod_comp_index(lo, lod_comp);
+		}
+
+		lo->ldo_flr_state = LCM_FL_WRITE_PENDING;
+	} else { /* MD_LAYOUT_RESYNC */
+		int i;
+
+		/**
+		 * could contain multiple non-stale mirrors, so we need to
+		 * prep uninited all components assuming any non-stale mirror
+		 * could be picked as the primary mirror.
+		 */
+		for (i = 0; i < lo->ldo_mirror_count; i++) {
+			if (lo->ldo_mirrors[i].lme_stale)
+				continue;
+
+			lod_foreach_mirror_comp(lod_comp, lo, i) {
+				if (!lod_comp_inited(lod_comp))
+					break;
+
+				if (extent.e_end < lod_comp->llc_extent.e_end)
+					extent.e_end =
+						lod_comp->llc_extent.e_end;
+			}
+		}
+
+		rc = lod_prepare_resync(env, lo, &extent);
+		if (rc)
+			GOTO(out, rc);
+		/* change the file state to SYNC_PENDING */
+		lo->ldo_flr_state = LCM_FL_SYNC_PENDING;
+	}
 
 	/* Reset the layout version once it's becoming too large.
 	 * This way it can make sure that the layout version is
@@ -5561,6 +5877,8 @@ static int lod_declare_update_rdonly(const struct lu_env *env,
 
 	layout_attr->la_valid = LA_LAYOUT_VERSION;
 	layout_attr->la_layout_version = 0; /* set current version */
+	if (mlc->mlc_opc == MD_LAYOUT_RESYNC)
+		layout_attr->la_layout_version = LU_LAYOUT_RESYNC;
 	rc = lod_declare_attr_set(env, &lo->ldo_obj, layout_attr, th);
 	if (rc)
 		GOTO(out, rc);
@@ -5628,19 +5946,23 @@ static int lod_declare_update_write_pending(const struct lu_env *env,
 		CDEBUG(D_LAYOUT, DFID": intent to write: "DEXT"\n",
 		       PFID(lod_object_fid(lo)), PEXT(&extent));
 
+		if (mlc->mlc_intent->li_opc == LAYOUT_INTENT_TRUNC) {
+			/**
+			 * trunc transfers [0, size) in the intent extent, we'd
+			 * stale components overlapping [size, eof).
+			 */
+			extent.e_start = extent.e_end;
+			extent.e_end = OBD_OBJECT_EOF;
+		}
 		/* 1. stale overlapping components */
 		lod_stale_components(lo, primary, &extent);
 
 		/* 2. find out the components need instantiating.
 		 * instantiate [0, mlc->mlc_intent->e_end) */
-		if (mlc->mlc_intent->li_opc == LAYOUT_INTENT_TRUNC) {
-			/**
-			 * trunc transfers [size, eof) in the intent extent,
-			 * we'd stale components overlapping [size, eof),
-			 * while we'd instantiated components covers [0, size).
-			 */
+
+		/* restore truncate intent extent */
+		if (mlc->mlc_intent->li_opc == LAYOUT_INTENT_TRUNC)
 			extent.e_end = extent.e_start;
-		}
 		extent.e_start = 0;
 
 		lod_foreach_mirror_comp(lod_comp, lo, primary) {
@@ -5657,9 +5979,6 @@ static int lod_declare_update_write_pending(const struct lu_env *env,
 						lod_comp_index(lo, lod_comp);
 		}
 	} else { /* MD_LAYOUT_RESYNC */
-		/* figure out the components that have been instantiated in
-		 * in primary to decide what components should be instantiated
-		 * in stale mirrors */
 		lod_foreach_mirror_comp(lod_comp, lo, primary) {
 			if (!lod_comp_inited(lod_comp))
 				break;
@@ -5667,36 +5986,9 @@ static int lod_declare_update_write_pending(const struct lu_env *env,
 			extent.e_end = lod_comp->llc_extent.e_end;
 		}
 
-		CDEBUG(D_LAYOUT,
-		       DFID": instantiate all stale components in "DEXT"\n",
-		       PFID(lod_object_fid(lo)), PEXT(&extent));
-
-		/* 1. instantiate all components within this extent, even
-		 * non-stale components so that it won't need to instantiate
-		 * those components for mirror truncate later. */
-		for (i = 0; i < lo->ldo_mirror_count; i++) {
-			if (primary == i)
-				continue;
-
-			LASSERTF(lo->ldo_mirrors[i].lme_stale,
-				 "both %d and %d are primary\n", i, primary);
-
-			lod_foreach_mirror_comp(lod_comp, lo, i) {
-				if (!lu_extent_is_overlapped(&extent,
-							&lod_comp->llc_extent))
-					break;
-
-				if (lod_comp_inited(lod_comp))
-					continue;
-
-				CDEBUG(D_LAYOUT, "resync instantiate %d / %d\n",
-				       i, lod_comp_index(lo, lod_comp));
-
-				info->lti_comp_idx[info->lti_count++] =
-						lod_comp_index(lo, lod_comp);
-			}
-		}
-
+		rc = lod_prepare_resync(env, lo, &extent);
+		if (rc)
+			GOTO(out, rc);
 		/* change the file state to SYNC_PENDING */
 		lo->ldo_flr_state = LCM_FL_SYNC_PENDING;
 	}
@@ -5786,8 +6078,8 @@ static int lod_declare_update_sync_pending(const struct lu_env *env,
 		GOTO(out, rc = -EINVAL);
 	}
 
-	if (!sync_components || !resync_components) {
-		CDEBUG(D_LAYOUT, DFID": no mirror in sync or resync\n",
+	if (!sync_components || (mlc->mlc_resync_count && !resync_components)) {
+		CDEBUG(D_LAYOUT, DFID": no mirror in sync\n",
 		       PFID(lod_object_fid(lo)));
 
 		/* tend to return an error code here to prevent
@@ -5838,7 +6130,7 @@ static int lod_declare_layout_change(const struct lu_env *env,
 		GOTO(out, rc);
 
 	switch (lo->ldo_flr_state) {
-	case LCM_FL_NOT_FLR:
+	case LCM_FL_NONE:
 		rc = lod_declare_update_plain(env, lo, mlc->mlc_intent,
 					      &mlc->mlc_buf, th);
 		break;