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 <john.hammond@intel.com>
Change-Id: I63d57a21ed7402829d4b3889c1f32d4ddec1f096
Reviewed-on: http://review.whamcloud.com/5422
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Mike Pershin <mike.pershin@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/lod/lod_lov.c b/lustre/lod/lod_lov.c
index 6ec7823..65326fa 100644 (file)
--- a/lustre/lod/lod_lov.c
+++ b/lustre/lod/lod_lov.c
@@ -677,7 +677,9 @@ 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);
@@ -685,11 +687,10 @@ int lod_initialize_objects(const struct lu_env *env, struct lod_object *lo,
        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
index fd36da5..4bd2aa2 100644 (file)
--- 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
index b1aa0c1..2b81ad8 100644 (file)
--- 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 @@ repeat_find:
                 * 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 @@ out:
 
 /* 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 @@ repeat_find:
                /*
                 * 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