LU-2789 lod: initialize objects in a detached stripe array

Initializing the ldo_stripe member of lod_device in place exposes some races with attribute setting so instead we use a detached array and assign that to ldo_stripe once it's complete. Signed-off-by: John L. Hammond <[email protected]> Change-Id: I63d57a21ed7402829d4b3889c1f32d4ddec1f096 Reviewed-on: http://review.whamcloud.com/5422 Reviewed-by: Alex Zhuravlev <[email protected]> Tested-by: Hudson Tested-by: Maloo <[email protected]> Reviewed-by: Mike Pershin <[email protected]> Reviewed-by: Oleg Drokin <[email protected]>
jxiong · Feb 20, 2013 · 42472bb · 42472bb
1 parent 62f58ba
commit 42472bb
Show file tree

Hide file tree

Showing 3 changed files with 57 additions and 28 deletions.
diff --git a/lustre/lod/lod_lov.c b/lustre/lod/lod_lov.c
@@ -677,19 +677,20 @@ int lod_initialize_objects(const struct lu_env *env, struct lod_object *lo,
 	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;
+	struct dt_object       **stripe;
+	int			 stripe_len;
+	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;
+	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);
@@ -721,10 +722,21 @@ 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);
+	} else {
+		lo->ldo_stripe = stripe;
+		lo->ldo_stripes_allocated = stripe_len;
+	}
+
 	RETURN(rc);
 }
 

diff --git a/lustre/lod/lod_object.c b/lustre/lod/lod_object.c
@@ -945,7 +945,7 @@ int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
 	ENTRY;
 
 	LASSERT(lo->ldo_stripe);
-	LASSERT(lo->ldo_stripe > 0);
+	LASSERT(lo->ldo_stripenr > 0);
 	LASSERT(lo->ldo_striping_cached == 0);
 
 	/* create all underlying objects */

diff --git a/lustre/lod/lod_qos.c b/lustre/lod/lod_qos.c
@@ -657,7 +657,8 @@ static int lod_qos_is_ost_used(const struct lu_env *env, int ost, int stripes)
 
 /* Allocate objects on OSTs with round-robin algorithm */
 static int lod_alloc_rr(const struct lu_env *env, struct lod_object *lo,
-			int flags, struct thandle *th)
+			struct dt_object **stripe, int flags,
+			struct thandle *th)
 {
 	struct lod_device *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
 	struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
@@ -790,7 +791,7 @@ static int lod_alloc_rr(const struct lu_env *env, struct lod_object *lo,
 		 * We've successfuly declared (reserved) an object
 		 */
 		lod_qos_ost_in_use(env, stripe_idx, ost_idx);
-		lo->ldo_stripe[stripe_idx] = o;
+		stripe[stripe_idx] = o;
 		stripe_idx++;
 
 	}
@@ -824,7 +825,8 @@ static int lod_alloc_rr(const struct lu_env *env, struct lod_object *lo,
 
 /* alloc objects on osts with specific stripe offset */
 static int lod_alloc_specific(const struct lu_env *env, struct lod_object *lo,
-			      int flags, struct thandle *th)
+			      struct dt_object **stripe, int flags,
+			      struct thandle *th)
 {
 	struct lod_device *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
 	struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
@@ -916,7 +918,7 @@ static int lod_alloc_specific(const struct lu_env *env, struct lod_object *lo,
 		/*
 		 * We've successfuly declared (reserved) an object
 		 */
-		lo->ldo_stripe[stripe_num] = o;
+		stripe[stripe_num] = o;
 		stripe_num++;
 
 		/* We have enough stripes */
@@ -971,7 +973,8 @@ static inline int lod_qos_is_usable(struct lod_device *lod)
    - network resources (shared OSS's)
  */
 static int lod_alloc_qos(const struct lu_env *env, struct lod_object *lo,
-			 int flags, struct thandle *th)
+			 struct dt_object **stripe, int flags,
+			 struct thandle *th)
 {
 	struct lod_device   *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
 	struct obd_statfs   *sfs = &lod_env_info(env)->lti_osfs;
@@ -1132,7 +1135,7 @@ static int lod_alloc_qos(const struct lu_env *env, struct lod_object *lo,
 					  idx, (int) PTR_ERR(o));
 				continue;
 			}
-			lo->ldo_stripe[nfound++] = o;
+			stripe[nfound++] = o;
 			lod_qos_used(m, osts, idx, &total_weight);
 			rc = 0;
 			break;
@@ -1154,9 +1157,9 @@ static int lod_alloc_qos(const struct lu_env *env, struct lod_object *lo,
 		 */
 		LCONSOLE_INFO("wanted %d, found %d\n", stripe_cnt, nfound);
 		for (i = 0; i < nfound; i++) {
-			LASSERT(lo->ldo_stripe[i]);
-			lu_object_put(env, &lo->ldo_stripe[i]->do_lu);
-			lo->ldo_stripe[i] = NULL;
+			LASSERT(stripe[i] != NULL);
+			lu_object_put(env, &stripe[i]->do_lu);
+			stripe[i] = NULL;
 		}
 
 		/* makes sense to rebalance next time */
@@ -1349,9 +1352,11 @@ int lod_qos_prep_create(const struct lu_env *env, struct lod_object *lo,
 			struct lu_attr *attr, const struct lu_buf *buf,
 			struct thandle *th)
 {
-	struct lod_device *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
-	int flag = LOV_USES_ASSIGNED_STRIPE;
-	int i, rc = 0;
+	struct lod_device      *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
+	struct dt_object      **stripe;
+	int			stripe_len;
+	int			flag = LOV_USES_ASSIGNED_STRIPE;
+	int			i, rc;
 	ENTRY;
 
 	LASSERT(lo);
@@ -1381,22 +1386,34 @@ int lod_qos_prep_create(const struct lu_env *env, struct lod_object *lo,
 		LASSERT(lo->ldo_stripenr > 0);
 		lo->ldo_stripenr = lod_get_stripecnt(d, LOV_MAGIC,
 				lo->ldo_stripenr);
-		i = sizeof(struct dt_object *) * lo->ldo_stripenr;
-		OBD_ALLOC(lo->ldo_stripe, i);
-		if (lo->ldo_stripe == NULL)
+
+		stripe_len = lo->ldo_stripenr;
+		OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_len);
+		if (stripe == NULL)
 			GOTO(out, rc = -ENOMEM);
-		lo->ldo_stripes_allocated = lo->ldo_stripenr;
 
 		lod_getref(&d->lod_ost_descs);
 		/* XXX: support for non-0 files w/o objects */
 		if (lo->ldo_def_stripe_offset >= d->lod_desc.ld_tgt_count) {
 			lod_qos_statfs_update(env, d);
-			rc = lod_alloc_qos(env, lo, flag, th);
+			rc = lod_alloc_qos(env, lo, stripe, flag, th);
 			if (rc == -EAGAIN)
-				rc = lod_alloc_rr(env, lo, flag, th);
-		} else
-			rc = lod_alloc_specific(env, lo, flag, th);
+				rc = lod_alloc_rr(env, lo, stripe, flag, th);
+		} else {
+			rc = lod_alloc_specific(env, lo, stripe, flag, th);
+		}
 		lod_putref(d, &d->lod_ost_descs);
+
+		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);
+		} else {
+			lo->ldo_stripe = stripe;
+			lo->ldo_stripes_allocated = stripe_len;
+		}
 	} else {
 		/*
 		 * lod_qos_parse_config() found supplied buf as a predefined