LU-9359 pfl: instantiate enough component at mdd_create_data

mknod creates file without layout, then truncate trigger the MDT
create OST objects, while current implementation only instantiate the
1st component, and when the truncate size locates in the other
components, the lvb size info is lost.

This patch makes MDT creates enought OST objects to cover the file's
size.

This patch fixes the misunderstanding of ost_pool::op->size, it
indicates the buffer size allocated instead of the array count.

Another issue fixed is that in lod_alloc_qos(), only fill in the ost
inused array when the lod_qos_declare_object_on() succeeds.

Signed-off-by: Bobi Jam <bobijam.xu@intel.com>
Change-Id: Ie66950e9b3d8cc009cca58f63936b275759211f1
Reviewed-on: https://review.whamcloud.com/26706
Tested-by: Jenkins
Reviewed-by: Niu Yawei <yawei.niu@intel.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>

diff --git a/lustre/lod/lod_internal.h b/lustre/lod/lod_internal.h
index f92bb0e..c88439e 100644 (file)
--- a/lustre/lod/lod_internal.h
+++ b/lustre/lod/lod_internal.h
@@ -76,7 +76,6 @@ struct pool_desc {
        struct obd_device       *pool_lobd;     /* owner */
 };
 
-#define pool_tgt_size(p) ((p)->pool_obds.op_size)
 #define pool_tgt_count(p) ((p)->pool_obds.op_count)
 #define pool_tgt_array(p)  ((p)->pool_obds.op_array)
 #define pool_tgt_rw_sem(p) ((p)->pool_obds.op_rw_sem)
@@ -612,6 +611,7 @@ typedef int (*lod_obj_stripe_cb_t)(const struct lu_env *env,
                                   struct thandle *th, int stripe_idx,
                                   struct lod_obj_stripe_cb_data *data);
 /* lod_qos.c */
+int lod_prepare_inuse(const struct lu_env *env, struct lod_object *lo);
 int lod_prepare_create(const struct lu_env *env, struct lod_object *lo,
                       struct lu_attr *attr, const struct lu_buf *buf,
                       struct thandle *th);

diff --git a/lustre/lod/lod_object.c b/lustre/lod/lod_object.c
index 933b05d..d74cbbd 100644 (file)
--- a/lustre/lod/lod_object.c
+++ b/lustre/lod/lod_object.c
@@ -2169,13 +2169,16 @@ inline __u16 lod_comp_entry_stripecnt(struct lod_object *lo,
                                      struct lod_layout_component *entry,
                                      bool is_dir)
 {
+       struct lod_device *lod = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
+
        if (is_dir)
                return  0;
        else if (lod_comp_inited(entry))
                return entry->llc_stripenr;
+       else if ((__u16)-1 == entry->llc_stripenr)
+               return lod->lod_desc.ld_tgt_count;
        else
-               return lod_get_stripecnt(lu2lod_dev(lod2lu_obj(lo)->lo_dev), lo,
-                                        entry->llc_stripenr);
+               return lod_get_stripecnt(lod, lo, entry->llc_stripenr);
 }
 
 static int lod_comp_md_size(struct lod_object *lo, bool is_dir)
@@ -4069,6 +4072,8 @@ static int lod_declare_init_size(const struct lu_env *env,
                        continue;
 
                extent = &lod_comp->llc_extent;
+               CDEBUG(D_INFO, "%lld [%lld, %lld)\n",
+                      size, extent->e_start, extent->e_end);
                if (!lo->ldo_is_composite ||
                    (size >= extent->e_start && size < extent->e_end)) {
                        objects = lod_comp->llc_stripe;
@@ -4833,41 +4838,6 @@ static int lod_invalidate(const struct lu_env *env, struct dt_object *dt)
        return dt_invalidate(env, dt_object_child(dt));
 }
 
-/**
- * Resize per-thread ost list to hold OST target index list already used.
- *
- * \param[in,out] inuse                structure contains ost list array
- * \param[in] cnt              total stripe count of all components
- * \param[in] max              array's max size if @max > 0
- *
- * \retval 0           on success
- * \retval -ENOMEM     reallocation failed
- */
-int lod_inuse_resize(struct ost_pool *inuse, __u16 cnt, __u16 max)
-{
-       __u32 *array;
-       __u32 new = cnt * sizeof(__u32);
-
-       inuse->op_count = 0;
-
-       if (new <= inuse->op_size)
-               return 0;
-
-       if (max)
-               new = min_t(__u32, new, max);
-       OBD_ALLOC(array, new);
-       if (!array)
-               return -ENOMEM;
-
-       if (inuse->op_array)
-               OBD_FREE(inuse->op_array, inuse->op_size);
-
-       inuse->op_array = array;
-       inuse->op_size = new;
-
-       return 0;
-}
-
 static int lod_declare_layout_change(const struct lu_env *env,
                                     struct dt_object *dt,
                                     struct layout_intent *layout,
@@ -4878,14 +4848,12 @@ static int lod_declare_layout_change(const struct lu_env *env,
        struct lod_object *lo = lod_dt_obj(dt);
        struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
        struct dt_object *next = dt_object_child(dt);
-       struct lod_obj_stripe_cb_data data;
        struct ost_pool *inuse = &info->lti_inuse_osts;
        struct lod_layout_component *lod_comp;
        struct lov_comp_md_v1 *comp_v1 = NULL;
        bool replay = false;
        bool need_create = false;
        int i, rc;
-       __u32 stripe_cnt = 0;
        ENTRY;
 
        if (!S_ISREG(dt->do_lu.lo_header->loh_attr) || !dt_object_exists(dt) ||
@@ -4929,18 +4897,7 @@ static int lod_declare_layout_change(const struct lu_env *env,
                        GOTO(out, rc);
 
                /* Prepare inuse array for composite file */
-               for (i = 0; i < lo->ldo_comp_cnt; i++)
-                       stripe_cnt += lod_comp_entry_stripecnt(lo,
-                                               &lo->ldo_comp_entries[i],
-                                               false);
-               rc = lod_inuse_resize(inuse, stripe_cnt, d->lod_osd_max_easize);
-               if (rc)
-                       GOTO(out, rc);
-
-               data.locd_inuse = inuse;
-               rc = lod_obj_for_each_stripe(env, lo, NULL,
-                                            lod_obj_stripe_set_inuse_cb,
-                                            &data);
+               rc = lod_prepare_inuse(env, lo);
                if (rc)
                        GOTO(out, rc);
        }

diff --git a/lustre/lod/lod_pool.c b/lustre/lod/lod_pool.c
index 915ebb9..27b945e 100644 (file)
--- a/lustre/lod/lod_pool.c
+++ b/lustre/lod/lod_pool.c
@@ -474,8 +474,8 @@ int lod_ost_pool_init(struct ost_pool *op, unsigned int count)
        op->op_array = NULL;
        op->op_count = 0;
        init_rwsem(&op->op_rw_sem);
-       op->op_size = count;
-       OBD_ALLOC(op->op_array, op->op_size * sizeof(op->op_array[0]));
+       op->op_size = count * sizeof(op->op_array[0]);
+       OBD_ALLOC(op->op_array, op->op_size);
        if (op->op_array == NULL) {
                op->op_size = 0;
                RETURN(-ENOMEM);
@@ -500,21 +500,22 @@ int lod_ost_pool_init(struct ost_pool *op, unsigned int count)
 int lod_ost_pool_extend(struct ost_pool *op, unsigned int min_count)
 {
        __u32 *new;
-       int new_size;
+       __u32 new_size;
 
        LASSERT(min_count != 0);
 
-       if (op->op_count < op->op_size)
+       if (op->op_count * sizeof(op->op_array[0]) < op->op_size)
                return 0;
 
-       new_size = max(min_count, 2 * op->op_size);
-       OBD_ALLOC(new, new_size * sizeof(op->op_array[0]));
+       new_size = max_t(__u32, min_count * sizeof(op->op_array[0]),
+                        2 * op->op_size);
+       OBD_ALLOC(new, new_size);
        if (new == NULL)
                return -ENOMEM;
 
        /* copy old array to new one */
-       memcpy(new, op->op_array, op->op_size * sizeof(op->op_array[0]));
-       OBD_FREE(op->op_array, op->op_size * sizeof(op->op_array[0]));
+       memcpy(new, op->op_array, op->op_size);
+       OBD_FREE(op->op_array, op->op_size);
        op->op_array = new;
        op->op_size = new_size;
 
@@ -617,7 +618,7 @@ int lod_ost_pool_free(struct ost_pool *op)
 
        down_write(&op->op_rw_sem);
 
-       OBD_FREE(op->op_array, op->op_size * sizeof(op->op_array[0]));
+       OBD_FREE(op->op_array, op->op_size);
        op->op_array = NULL;
        op->op_count = 0;
        op->op_size = 0;

diff --git a/lustre/lod/lod_qos.c b/lustre/lod/lod_qos.c
index 5cd73db..81122e9 100644 (file)
--- a/lustre/lod/lod_qos.c
+++ b/lustre/lod/lod_qos.c
@@ -836,7 +836,7 @@ static inline int lod_comp_is_ost_used(struct ost_pool *inuse, int ost)
        if (inuse->op_size == 0)
                return 0;
 
-       LASSERT(inuse->op_count <= inuse->op_size);
+       LASSERT(inuse->op_count * sizeof(inuse->op_array[0]) <= inuse->op_size);
        for (j = 0; j < inuse->op_count; j++) {
                if (inuse->op_array[j] == ost)
                        return 1;
@@ -853,8 +853,9 @@ static inline int lod_comp_is_ost_used(struct ost_pool *inuse, int ost)
 static inline void lod_comp_ost_in_use(struct ost_pool *inuse, int ost)
 {
        LASSERT(inuse != NULL);
-       if (inuse->op_size && !lod_comp_is_ost_used(inuse,  ost)) {
-               LASSERT(inuse->op_count < inuse->op_size);
+       if (inuse->op_size && !lod_comp_is_ost_used(inuse, ost)) {
+               LASSERT(inuse->op_count * sizeof(inuse->op_array[0]) <
+                       inuse->op_size);
                inuse->op_array[inuse->op_count] = ost;
                inuse->op_count++;
        }
@@ -1576,15 +1577,12 @@ static int lod_alloc_qos(const struct lu_env *env, struct lod_object *lo,
                                continue;
 
                        QOS_DEBUG("stripe=%d to idx=%d\n", nfound, idx);
-
                        /*
                         * do not put >1 objects on a single OST
                         */
                        if (lod_qos_is_ost_used(env, idx, nfound) ||
                            lod_comp_is_ost_used(inuse, idx))
                                continue;
-                       lod_qos_ost_in_use(env, nfound, idx);
-                       lod_comp_ost_in_use(inuse, idx);
 
                        o = lod_qos_declare_object_on(env, lod, idx, th);
                        if (IS_ERR(o)) {
@@ -1592,6 +1590,9 @@ static int lod_alloc_qos(const struct lu_env *env, struct lod_object *lo,
                                          idx, (int) PTR_ERR(o));
                                continue;
                        }
+
+                       lod_qos_ost_in_use(env, nfound, idx);
+                       lod_comp_ost_in_use(inuse, idx);
                        stripe[nfound++] = o;
                        lod_qos_used(lod, osts, idx, &total_weight);
                        rc = 0;
@@ -2152,13 +2153,75 @@ int lod_obj_stripe_set_inuse_cb(const struct lu_env *env,
        return 0;
 }
 
+/**
+ * Resize per-thread ost list to hold OST target index list already used.
+ *
+ * \param[in,out] inuse                structure contains ost list array
+ * \param[in] cnt              total stripe count of all components
+ * \param[in] max              array's max size if @max > 0
+ *
+ * \retval 0           on success
+ * \retval -ENOMEM     reallocation failed
+ */
+static int lod_inuse_resize(struct ost_pool *inuse, __u16 cnt, __u16 max)
+{
+       __u32 *array;
+       __u32 new = cnt * sizeof(inuse->op_array[0]);
+
+       inuse->op_count = 0;
+
+       if (new <= inuse->op_size)
+               return 0;
+
+       if (max)
+               new = min_t(__u32, new, max);
+
+       OBD_ALLOC(array, new);
+       if (!array)
+               return -ENOMEM;
+
+       if (inuse->op_array)
+               OBD_FREE(inuse->op_array, inuse->op_size);
+
+       inuse->op_array = array;
+       inuse->op_size = new;
+
+       return 0;
+}
+
+int lod_prepare_inuse(const struct lu_env *env, struct lod_object *lo)
+{
+       struct lod_thread_info *info = lod_env_info(env);
+       struct lod_device *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
+       struct ost_pool *inuse = &info->lti_inuse_osts;
+       struct lod_obj_stripe_cb_data data;
+       __u32 stripe_cnt = 0;
+       int i;
+       int rc;
+
+       for (i = 0; i < lo->ldo_comp_cnt; i++)
+               stripe_cnt += lod_comp_entry_stripecnt(lo,
+                                       &lo->ldo_comp_entries[i], false);
+       rc = lod_inuse_resize(inuse, stripe_cnt, d->lod_osd_max_easize);
+       if (rc)
+               return rc;
+
+       data.locd_inuse = inuse;
+       return lod_obj_for_each_stripe(env, lo, NULL,
+                                      lod_obj_stripe_set_inuse_cb, &data);
+}
+
 int lod_prepare_create(const struct lu_env *env, struct lod_object *lo,
                       struct lu_attr *attr, const struct lu_buf *buf,
                       struct thandle *th)
 
 {
+       struct lod_thread_info *info = lod_env_info(env);
        struct lod_device *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
-       struct ost_pool inuse = { 0 };
+       struct ost_pool inuse_osts = { 0 };
+       struct ost_pool *inuse = &inuse_osts;
+       uint64_t size = 0;
+       int i;
        int rc;
        ENTRY;
 
@@ -2182,8 +2245,36 @@ int lod_prepare_create(const struct lu_env *env, struct lod_object *lo,
        if (rc)
                RETURN(rc);
 
-       /* prepare OST object creation for the 1st comp. */
-       rc = lod_qos_prep_create(env, lo, attr, th, 0, &inuse);
+       if (attr->la_valid & LA_SIZE)
+               size = attr->la_size;
+
+       /* only prepare inuse if multiple components to be created */
+       if (size && lo->ldo_is_composite) {
+               rc = lod_prepare_inuse(env, lo);
+               if (rc)
+                       RETURN(rc);
+               inuse = &info->lti_inuse_osts;
+       }
+
+       /**
+        * prepare OST object creation for the component covering file's
+        * size, the 1st component (including plain layout file) is always
+        * instantiated.
+        */
+       for (i = 0; i < lo->ldo_comp_cnt; i++) {
+               struct lod_layout_component *lod_comp;
+               struct lu_extent *extent;
+
+               lod_comp = &lo->ldo_comp_entries[i];
+               extent = &lod_comp->llc_extent;
+               CDEBUG(D_QOS, "%lld [%lld, %lld)\n",
+                      size, extent->e_start, extent->e_end);
+               if (!lo->ldo_is_composite || size >= extent->e_start) {
+                       rc = lod_qos_prep_create(env, lo, attr, th, i, inuse);
+                       if (rc)
+                               break;
+               }
+       }
 
        RETURN(rc);
 }

diff --git a/lustre/lov/lov_internal.h b/lustre/lov/lov_internal.h
index ae81569..b9743b6 100644 (file)
--- a/lustre/lov/lov_internal.h
+++ b/lustre/lov/lov_internal.h
@@ -182,7 +182,6 @@ void lsm_free(struct lov_stripe_md *lsm);
 })
 #endif
 
-#define pool_tgt_size(p) ((p)->pool_obds.op_size)
 #define pool_tgt_count(p) ((p)->pool_obds.op_count)
 #define pool_tgt_array(p) ((p)->pool_obds.op_array)
 #define pool_tgt_rw_sem(p) ((p)->pool_obds.op_rw_sem)

diff --git a/lustre/lov/lov_io.c b/lustre/lov/lov_io.c
index 1bf1bcb..865adf7 100644 (file)
--- a/lustre/lov/lov_io.c
+++ b/lustre/lov/lov_io.c
@@ -102,7 +102,8 @@ static int lov_io_sub_init(const struct lu_env *env, struct lov_io *lio,
        LASSERT(sub->sub_env == NULL);
        ENTRY;
 
-       if (unlikely(lov_r0(lov, index)->lo_sub[stripe] == NULL))
+       if (unlikely(!lov_r0(lov, index)->lo_sub ||
+                    !lov_r0(lov, index)->lo_sub[stripe]))
                RETURN(-EIO);
 
        /* obtain new environment */

diff --git a/lustre/lov/lov_pool.c b/lustre/lov/lov_pool.c
index a8d3f18..7a2b9ac 100644 (file)
--- a/lustre/lov/lov_pool.c
+++ b/lustre/lov/lov_pool.c
@@ -319,45 +319,46 @@ void lov_dump_pool(int level, struct pool_desc *pool)
 #define LOV_POOL_INIT_COUNT 2
 int lov_ost_pool_init(struct ost_pool *op, unsigned int count)
 {
-        ENTRY;
+       ENTRY;
 
-        if (count == 0)
-                count = LOV_POOL_INIT_COUNT;
-        op->op_array = NULL;
-        op->op_count = 0;
+       if (count == 0)
+               count = LOV_POOL_INIT_COUNT;
+       op->op_array = NULL;
+       op->op_count = 0;
        init_rwsem(&op->op_rw_sem);
-        op->op_size = count;
-        OBD_ALLOC(op->op_array, op->op_size * sizeof(op->op_array[0]));
-        if (op->op_array == NULL) {
-                op->op_size = 0;
-                RETURN(-ENOMEM);
-        }
-        EXIT;
-        return 0;
+       op->op_size = count * sizeof(op->op_array[0]);
+       OBD_ALLOC(op->op_array, op->op_size);
+       if (op->op_array == NULL) {
+               op->op_size = 0;
+               RETURN(-ENOMEM);
+       }
+       EXIT;
+       return 0;
 }
 
 /* Caller must hold write op_rwlock */
 int lov_ost_pool_extend(struct ost_pool *op, unsigned int min_count)
 {
-        __u32 *new;
-        int new_size;
-
-        LASSERT(min_count != 0);
-
-        if (op->op_count < op->op_size)
-                return 0;
-
-        new_size = max(min_count, 2 * op->op_size);
-        OBD_ALLOC(new, new_size * sizeof(op->op_array[0]));
-        if (new == NULL)
-                return -ENOMEM;
-
-        /* copy old array to new one */
-        memcpy(new, op->op_array, op->op_size * sizeof(op->op_array[0]));
-        OBD_FREE(op->op_array, op->op_size * sizeof(op->op_array[0]));
-        op->op_array = new;
-        op->op_size = new_size;
-        return 0;
+       __u32 *new;
+       __u32 new_size;
+
+       LASSERT(min_count != 0);
+
+       if (op->op_count * sizeof(op->op_array[0]) < op->op_size)
+               return 0;
+
+       new_size = max_t(__u32, min_count * sizeof(op->op_array[0]),
+                        2 * op->op_size);
+       OBD_ALLOC(new, new_size);
+       if (new == NULL)
+               return -ENOMEM;
+
+       /* copy old array to new one */
+       memcpy(new, op->op_array, op->op_size);
+       OBD_FREE(op->op_array, op->op_size);
+       op->op_array = new;
+       op->op_size = new_size;
+       return 0;
 }
 
 int lov_ost_pool_add(struct ost_pool *op, __u32 idx, unsigned int min_count)
@@ -409,20 +410,20 @@ int lov_ost_pool_remove(struct ost_pool *op, __u32 idx)
 
 int lov_ost_pool_free(struct ost_pool *op)
 {
-        ENTRY;
+       ENTRY;
 
-        if (op->op_size == 0)
-                RETURN(0);
+       if (op->op_size == 0)
+               RETURN(0);
 
        down_write(&op->op_rw_sem);
 
-        OBD_FREE(op->op_array, op->op_size * sizeof(op->op_array[0]));
-        op->op_array = NULL;
-        op->op_count = 0;
-        op->op_size = 0;
+       OBD_FREE(op->op_array, op->op_size);
+       op->op_array = NULL;
+       op->op_count = 0;
+       op->op_size = 0;
 
        up_write(&op->op_rw_sem);
-        RETURN(0);
+       RETURN(0);
 }
 
 

diff --git a/lustre/osc/osc_io.c b/lustre/osc/osc_io.c
index b7b10a0..9c3c8f1 100644 (file)
--- a/lustre/osc/osc_io.c
+++ b/lustre/osc/osc_io.c
@@ -499,6 +499,7 @@ static int osc_io_setattr_start(const struct lu_env *env,
        __u64                    size   = io->u.ci_setattr.sa_attr.lvb_size;
        unsigned int             ia_valid = io->u.ci_setattr.sa_valid;
        int                      result = 0;
+       ENTRY;
 
        /* truncate cache dirty pages first */
        if (cl_io_is_trunc(io))
@@ -585,7 +586,7 @@ static int osc_io_setattr_start(const struct lu_env *env,
                cbargs->opc_rpc_sent = result == 0;
        }
 
-       return result;
+       RETURN(result);
 }
 
 static void osc_io_setattr_end(const struct lu_env *env,