X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Flov%2Flov_qos.c;h=c1f8c9674867891efa299545b7aad0ece428ffc5;hp=8913d90381e5e9a7f97a6a19e4b9e647833de1ec;hb=4f50bbfa20b8cfdf13c2647cfae057285df9e9e8;hpb=1f36caa3192328cf93f6422f58649eb81220cfbf

diff --git a/lustre/lov/lov_qos.c b/lustre/lov/lov_qos.c
index 8913d90..c1f8c96 100644
--- a/lustre/lov/lov_qos.c
+++ b/lustre/lov/lov_qos.c
@@ -26,7 +26,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright  2008 Sun Microsystems, Inc. All rights reserved
+ * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  */
 /*
@@ -47,6 +47,7 @@
 
 #include <obd_class.h>
 #include <obd_lov.h>
+#include <lustre/lustre_idl.h>
 #include "lov_internal.h"
 
 /* #define QOS_DEBUG 1 */
@@ -54,7 +55,6 @@
 
 #define TGT_BAVAIL(i) (lov->lov_tgts[i]->ltd_exp->exp_obd->obd_osfs.os_bavail *\
                        lov->lov_tgts[i]->ltd_exp->exp_obd->obd_osfs.os_bsize)
-#define TGT_FFREE(i)  (lov->lov_tgts[i]->ltd_exp->exp_obd->obd_osfs.os_ffree)
 
 
 int qos_add_tgt(struct obd_device *obd, __u32 index)
@@ -72,9 +72,9 @@ int qos_add_tgt(struct obd_device *obd, __u32 index)
                 RETURN(-ENOTCONN);
         }
 
-        down_write(&lov->lov_qos.lq_rw_sem);
-        mutex_down(&lov->lov_lock);
-        list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
+        cfs_down_write(&lov->lov_qos.lq_rw_sem);
+        cfs_mutex_down(&lov->lov_lock);
+        cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
                 if (obd_uuid_equals(&oss->lqo_uuid,
                                     &exp->exp_connection->c_remote_uuid)) {
                         found++;
@@ -91,7 +91,7 @@ int qos_add_tgt(struct obd_device *obd, __u32 index)
                        sizeof(oss->lqo_uuid));
         } else {
                 /* Assume we have to move this one */
-                list_del(&oss->lqo_oss_list);
+                cfs_list_del(&oss->lqo_oss_list);
         }
 
         oss->lqo_ost_count++;
@@ -99,13 +99,14 @@ int qos_add_tgt(struct obd_device *obd, __u32 index)
 
         /* Add sorted by # of OSTs.  Find the first entry that we're
            bigger than... */
-        list_for_each_entry(temposs, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
+        cfs_list_for_each_entry(temposs, &lov->lov_qos.lq_oss_list,
+                                lqo_oss_list) {
                 if (oss->lqo_ost_count > temposs->lqo_ost_count)
                         break;
         }
         /* ...and add before it.  If we're the first or smallest, temposs
            points to the list head, and we add to the end. */
-        list_add_tail(&oss->lqo_oss_list, &temposs->lqo_oss_list);
+        cfs_list_add_tail(&oss->lqo_oss_list, &temposs->lqo_oss_list);
 
         lov->lov_qos.lq_dirty = 1;
         lov->lov_qos.lq_rr.lqr_dirty = 1;
@@ -116,8 +117,8 @@ int qos_add_tgt(struct obd_device *obd, __u32 index)
                oss->lqo_ost_count);
 
 out:
-        mutex_up(&lov->lov_lock);
-        up_write(&lov->lov_qos.lq_rw_sem);
+        cfs_mutex_up(&lov->lov_lock);
+        cfs_up_write(&lov->lov_qos.lq_rw_sem);
         RETURN(rc);
 }
 
@@ -128,7 +129,7 @@ int qos_del_tgt(struct obd_device *obd, struct lov_tgt_desc *tgt)
         int rc = 0;
         ENTRY;
 
-        down_write(&lov->lov_qos.lq_rw_sem);
+        cfs_down_write(&lov->lov_qos.lq_rw_sem);
 
         oss = tgt->ltd_qos.ltq_oss;
         if (!oss)
@@ -138,14 +139,14 @@ int qos_del_tgt(struct obd_device *obd, struct lov_tgt_desc *tgt)
         if (oss->lqo_ost_count == 0) {
                 CDEBUG(D_QOS, "removing OSS %s\n",
                        obd_uuid2str(&oss->lqo_uuid));
-                list_del(&oss->lqo_oss_list);
+                cfs_list_del(&oss->lqo_oss_list);
                 OBD_FREE_PTR(oss);
         }
 
         lov->lov_qos.lq_dirty = 1;
         lov->lov_qos.lq_rr.lqr_dirty = 1;
 out:
-        up_write(&lov->lov_qos.lq_rw_sem);
+        cfs_up_write(&lov->lov_qos.lq_rw_sem);
         RETURN(rc);
 }
 
@@ -169,7 +170,7 @@ static int qos_calc_ppo(struct obd_device *obd)
                 GOTO(out, rc = -EAGAIN);
 
         /* find bavail on each OSS */
-        list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
+        cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
                 oss->lqo_bavail = 0;
         }
         lov->lov_qos.lq_active_oss_count = 0;
@@ -230,7 +231,7 @@ static int qos_calc_ppo(struct obd_device *obd)
         }
 
         /* Per-OSS penalty is prio * oss_avail / oss_osts / (num_oss - 1) / 2 */
-        list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
+        cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
                 temp = oss->lqo_bavail >> 1;
                 do_div(temp, oss->lqo_ost_count * num_active);
                 oss->lqo_penalty_per_obj = (temp * prio_wide) >> 8;
@@ -311,7 +312,7 @@ static int qos_used(struct lov_obd *lov, struct ost_pool *osts,
                 lov->lov_qos.lq_active_oss_count;
 
         /* Decrease all OSS penalties */
-        list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
+        cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
                 if (oss->lqo_penalty < oss->lqo_penalty_per_obj)
                         oss->lqo_penalty = 0;
                 else
@@ -372,7 +373,7 @@ static int qos_calc_rr(struct lov_obd *lov, struct ost_pool *src_pool,
         }
 
         /* Do actual allocation. */
-        down_write(&lov->lov_qos.lq_rw_sem);
+        cfs_down_write(&lov->lov_qos.lq_rw_sem);
 
         /*
          * Check again. While we were sleeping on @lq_rw_sem something could
@@ -380,7 +381,7 @@ static int qos_calc_rr(struct lov_obd *lov, struct ost_pool *src_pool,
          */
         if (!lqr->lqr_dirty) {
                 LASSERT(lqr->lqr_pool.op_size);
-                up_write(&lov->lov_qos.lq_rw_sem);
+                cfs_up_write(&lov->lov_qos.lq_rw_sem);
                 RETURN(0);
         }
 
@@ -393,7 +394,7 @@ static int qos_calc_rr(struct lov_obd *lov, struct ost_pool *src_pool,
         lqr->lqr_pool.op_count = real_count;
         rc = lov_ost_pool_extend(&lqr->lqr_pool, real_count);
         if (rc) {
-                up_write(&lov->lov_qos.lq_rw_sem);
+                cfs_up_write(&lov->lov_qos.lq_rw_sem);
                 RETURN(rc);
         }
         for (i = 0; i < lqr->lqr_pool.op_count; i++)
@@ -401,7 +402,7 @@ static int qos_calc_rr(struct lov_obd *lov, struct ost_pool *src_pool,
 
         /* Place all the OSTs from 1 OSS at the same time. */
         placed = 0;
-        list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
+        cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
                 int j = 0;
                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
                         if (lov->lov_tgts[src_pool->op_array[i]] &&
@@ -419,7 +420,7 @@ static int qos_calc_rr(struct lov_obd *lov, struct ost_pool *src_pool,
         }
 
         lqr->lqr_dirty = 0;
-        up_write(&lov->lov_qos.lq_rw_sem);
+        cfs_up_write(&lov->lov_qos.lq_rw_sem);
 
         if (placed != real_count) {
                 /* This should never happen */
@@ -460,7 +461,7 @@ void qos_shrink_lsm(struct lov_request_set *set)
                 oti_alloc_cookies(set->set_oti, set->set_count);
                 if (set->set_oti->oti_logcookies) {
                         memcpy(set->set_oti->oti_logcookies, cookies, newsize);
-                        OBD_FREE(cookies, oldsize);
+                        OBD_FREE_LARGE(cookies, oldsize);
                         set->set_cookies = set->set_oti->oti_logcookies;
                 } else {
                         CWARN("'leaking' %d bytes\n", oldsize - newsize);
@@ -472,7 +473,7 @@ void qos_shrink_lsm(struct lov_request_set *set)
         LASSERT(lsm->lsm_stripe_count >= set->set_count);
 
         newsize = lov_stripe_md_size(set->set_count);
-        OBD_ALLOC(lsm_new, newsize);
+        OBD_ALLOC_LARGE(lsm_new, newsize);
         if (lsm_new != NULL) {
                 int i;
                 memcpy(lsm_new, lsm, sizeof(*lsm));
@@ -485,44 +486,88 @@ void qos_shrink_lsm(struct lov_request_set *set)
                                       sizeof(struct lov_oinfo));
                 }
                 lsm_new->lsm_stripe_count = set->set_count;
-                OBD_FREE(lsm, sizeof(struct lov_stripe_md) +
-                         lsm->lsm_stripe_count * sizeof(struct lov_oinfo *));
+                OBD_FREE_LARGE(lsm, sizeof(struct lov_stripe_md) +
+                               lsm->lsm_stripe_count*sizeof(struct lov_oinfo*));
                 set->set_oi->oi_md = lsm_new;
         } else {
                 CWARN("'leaking' few bytes\n");
         }
 }
 
+/**
+ * Check whether we can create the object on the OST(refered by ost_idx)
+ * \retval:
+ *          0: create the object.
+ *          other value: did not create the object.
+ */
+static int lov_check_and_create_object(struct lov_obd *lov, int ost_idx,
+                                       struct lov_stripe_md *lsm,
+                                       struct lov_request *req,
+                                       struct obd_trans_info *oti)
+{
+        int stripe;
+        int rc = -EIO;
+        ENTRY;
+
+        CDEBUG(D_QOS, "Check and create on idx %d \n", ost_idx);
+        if (!lov->lov_tgts[ost_idx] ||
+            !lov->lov_tgts[ost_idx]->ltd_active)
+                RETURN(rc);
+
+        /* check if objects has been created on this ost */
+        for (stripe = 0; stripe < lsm->lsm_stripe_count; stripe++) {
+                /* already have object at this stripe */
+                if (ost_idx == lsm->lsm_oinfo[stripe]->loi_ost_idx)
+                        break;
+        }
+
+        if (stripe >= lsm->lsm_stripe_count) {
+                req->rq_idx = ost_idx;
+                rc = obd_create(lov->lov_tgts[ost_idx]->ltd_exp,
+                                req->rq_oi.oi_oa, &req->rq_oi.oi_md,
+                                oti);
+        }
+        RETURN(rc);
+}
+
 int qos_remedy_create(struct lov_request_set *set, struct lov_request *req)
 {
         struct lov_stripe_md *lsm = set->set_oi->oi_md;
         struct lov_obd *lov = &set->set_exp->exp_obd->u.lov;
-        unsigned ost_idx, ost_count = lov->desc.ld_tgt_count;
-        int stripe, i, rc = -EIO;
+        unsigned ost_idx = 0, ost_count;
+        struct pool_desc *pool;
+        struct ost_pool *osts = NULL;
+        int i, rc = -EIO;
         ENTRY;
 
-        ost_idx = (req->rq_idx + lsm->lsm_stripe_count) % ost_count;
-        for (i = 0; i < ost_count; i++, ost_idx = (ost_idx + 1) % ost_count) {
-                if (!lov->lov_tgts[ost_idx] ||
-                    !lov->lov_tgts[ost_idx]->ltd_active)
-                        continue;
-                /* check if objects has been created on this ost */
-                for (stripe = 0; stripe < lsm->lsm_stripe_count; stripe++) {
-                        if (stripe == req->rq_stripe)
-                                continue;
-                        if (ost_idx == lsm->lsm_oinfo[stripe]->loi_ost_idx)
+        /* First check whether we can create the objects on the pool */
+        pool = lov_find_pool(lov, lsm->lsm_pool_name);
+        if (pool != NULL) {
+                cfs_down_read(&pool_tgt_rw_sem(pool));
+                osts = &(pool->pool_obds);
+                ost_count = osts->op_count;
+                for (i = 0; i < ost_count; i++, ost_idx = osts->op_array[i]) {
+                        rc = lov_check_and_create_object(lov, ost_idx, lsm, req,
+                                                         set->set_oti);
+                        if (rc == 0)
                                 break;
                 }
+                cfs_up_read(&pool_tgt_rw_sem(pool));
+                lov_pool_putref(pool);
+                RETURN(rc);
+        }
 
-                if (stripe >= lsm->lsm_stripe_count) {
-                        req->rq_idx = ost_idx;
-                        rc = obd_create(lov->lov_tgts[ost_idx]->ltd_exp,
-                                        req->rq_oi.oi_oa, &req->rq_oi.oi_md,
-                                        set->set_oti);
-                        if (!rc)
-                                break;
-                }
+        ost_count = lov->desc.ld_tgt_count;
+        /* Then check whether we can create the objects on other OSTs */
+        ost_idx = (req->rq_idx + lsm->lsm_stripe_count) % ost_count;
+        for (i = 0; i < ost_count; i++, ost_idx = (ost_idx + 1) % ost_count) {
+                rc = lov_check_and_create_object(lov, ost_idx, lsm, req,
+                                                 set->set_oti);
+
+                if (rc == 0)
+                        break;
         }
+
         RETURN(rc);
 }
 
@@ -532,8 +577,8 @@ static int min_stripe_count(int stripe_cnt, int flags)
                 stripe_cnt - (stripe_cnt / 4) : stripe_cnt);
 }
 
-#define LOV_CREATE_RESEED_MULT 4
-#define LOV_CREATE_RESEED_MIN  1000
+#define LOV_CREATE_RESEED_MULT 30
+#define LOV_CREATE_RESEED_MIN  2000
 /* Allocate objects on osts with round-robin algorithm */
 static int alloc_rr(struct lov_obd *lov, int *idx_arr, int *stripe_cnt,
                     char *poolname, int flags)
@@ -554,7 +599,7 @@ static int alloc_rr(struct lov_obd *lov, int *idx_arr, int *stripe_cnt,
                 osts = &(lov->lov_packed);
                 lqr = &(lov->lov_qos.lq_rr);
         } else {
-                down_read(&pool_tgt_rw_sem(pool));
+                cfs_down_read(&pool_tgt_rw_sem(pool));
                 osts = &(pool->pool_obds);
                 lqr = &(pool->pool_rr);
         }
@@ -564,7 +609,7 @@ static int alloc_rr(struct lov_obd *lov, int *idx_arr, int *stripe_cnt,
                 GOTO(out, rc);
 
         if (--lqr->lqr_start_count <= 0) {
-                lqr->lqr_start_idx = ll_rand() % osts->op_count;
+                lqr->lqr_start_idx = cfs_rand() % osts->op_count;
                 lqr->lqr_start_count =
                         (LOV_CREATE_RESEED_MIN / max(osts->op_count, 1U) +
                          LOV_CREATE_RESEED_MULT) * max(osts->op_count, 1U);
@@ -577,7 +622,7 @@ static int alloc_rr(struct lov_obd *lov, int *idx_arr, int *stripe_cnt,
                 if (*stripe_cnt > 1 && (osts->op_count % (*stripe_cnt)) != 1)
                         ++lqr->lqr_offset_idx;
         }
-        down_read(&lov->lov_qos.lq_rw_sem);
+        cfs_down_read(&lov->lov_qos.lq_rw_sem);
         ost_start_idx_temp = lqr->lqr_start_idx;
 
 repeat_find:
@@ -627,12 +672,12 @@ repeat_find:
                 goto repeat_find;
         }
 
-        up_read(&lov->lov_qos.lq_rw_sem);
+        cfs_up_read(&lov->lov_qos.lq_rw_sem);
 
         *stripe_cnt = idx_pos - idx_arr;
 out:
         if (pool != NULL) {
-                up_read(&pool_tgt_rw_sem(pool));
+                cfs_up_read(&pool_tgt_rw_sem(pool));
                 /* put back ref got by lov_find_pool() */
                 lov_pool_putref(pool);
         }
@@ -655,7 +700,7 @@ static int alloc_specific(struct lov_obd *lov, struct lov_stripe_md *lsm,
         if (pool == NULL) {
                 osts = &(lov->lov_packed);
         } else {
-                down_read(&pool_tgt_rw_sem(pool));
+                cfs_down_read(&pool_tgt_rw_sem(pool));
                 osts = &(pool->pool_obds);
         }
 
@@ -724,7 +769,7 @@ repeat_find:
         rc = -EFBIG;
 out:
         if (pool != NULL) {
-                up_read(&pool_tgt_rw_sem(pool));
+                cfs_up_read(&pool_tgt_rw_sem(pool));
                 /* put back ref got by lov_find_pool() */
                 lov_pool_putref(pool);
         }
@@ -740,10 +785,8 @@ static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
                      char *poolname, int flags)
 {
         struct lov_obd *lov = &exp->exp_obd->u.lov;
-        static time_t last_warn = 0;
-        time_t now = cfs_time_current_sec();
-        __u64 total_bavail, total_weight = 0;
-        int nfound, good_osts, i, warn = 0, rc = 0;
+        __u64 total_weight = 0;
+        int nfound, good_osts, i, rc = 0;
         int stripe_cnt_min = min_stripe_count(*stripe_cnt, flags);
         struct pool_desc *pool;
         struct ost_pool *osts;
@@ -758,19 +801,24 @@ static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
                 osts = &(lov->lov_packed);
                 lqr = &(lov->lov_qos.lq_rr);
         } else {
-                down_read(&pool_tgt_rw_sem(pool));
+                cfs_down_read(&pool_tgt_rw_sem(pool));
                 osts = &(pool->pool_obds);
                 lqr = &(pool->pool_rr);
         }
 
-        lov_getref(exp->exp_obd);
+        obd_getref(exp->exp_obd);
+
+        /* wait for fresh statfs info if needed, the rpcs are sent in
+         * lov_create() */
+        qos_statfs_update(exp->exp_obd,
+                          cfs_time_shift_64(-2 * lov->desc.ld_qos_maxage), 1);
 
         /* Detect -EAGAIN early, before expensive lock is taken. */
         if (!lov->lov_qos.lq_dirty && lov->lov_qos.lq_same_space)
                 GOTO(out_nolock, rc = -EAGAIN);
 
         /* Do actual allocation, use write lock here. */
-        down_write(&lov->lov_qos.lq_rw_sem);
+        cfs_down_write(&lov->lov_qos.lq_rw_sem);
 
         /*
          * Check again, while we were sleeping on @lq_rw_sem things could
@@ -786,35 +834,12 @@ static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
         if (rc)
                 GOTO(out, rc);
 
-        total_bavail = 0;
         good_osts = 0;
-        /* Warn users about zero available space/inode every 30 min */
-        if (cfs_time_sub(now, last_warn) > 60 * 30)
-                warn = 1;
         /* Find all the OSTs that are valid stripe candidates */
         for (i = 0; i < osts->op_count; i++) {
-                __u64 bavail;
-
                 if (!lov->lov_tgts[osts->op_array[i]] ||
                     !lov->lov_tgts[osts->op_array[i]]->ltd_active)
                         continue;
-                bavail = TGT_BAVAIL(osts->op_array[i]);
-                if (!bavail) {
-                        if (warn) {
-                                CDEBUG(D_QOS, "no free space on %s\n",
-                                     obd_uuid2str(&lov->lov_tgts[osts->op_array[i]]->ltd_uuid));
-                                last_warn = now;
-                        }
-                        continue;
-                }
-                if (!TGT_FFREE(osts->op_array[i])) {
-                        if (warn) {
-                                CDEBUG(D_QOS, "no free inodes on %s\n",
-                                     obd_uuid2str(&lov->lov_tgts[osts->op_array[i]]->ltd_uuid));
-                                last_warn = now;
-                        }
-                        continue;
-                }
 
                 /* Fail Check before osc_precreate() is called
                    so we can only 'fail' single OSC. */
@@ -826,7 +851,6 @@ static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
 
                 lov->lov_tgts[osts->op_array[i]]->ltd_qos.ltq_usable = 1;
                 qos_calc_weight(lov, osts->op_array[i]);
-                total_bavail += bavail;
                 total_weight += lov->lov_tgts[osts->op_array[i]]->ltd_qos.ltq_weight;
 
                 good_osts++;
@@ -839,9 +863,6 @@ static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
         if (good_osts < stripe_cnt_min)
                 GOTO(out, rc = -EAGAIN);
 
-        if (!total_bavail)
-                GOTO(out, rc = -ENOSPC);
-
         /* We have enough osts */
         if (good_osts < *stripe_cnt)
                 *stripe_cnt = good_osts;
@@ -859,23 +880,23 @@ static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
 
                 if (total_weight) {
 #if BITS_PER_LONG == 32
-                        rand = ll_rand() % (unsigned)total_weight;
+                        rand = cfs_rand() % (unsigned)total_weight;
                         /* If total_weight > 32-bit, first generate the high
                          * 32 bits of the random number, then add in the low
                          * 32 bits (truncated to the upper limit, if needed) */
                         if (total_weight > 0xffffffffULL)
-                                rand = (__u64)(ll_rand() %
+                                rand = (__u64)(cfs_rand() %
                                           (unsigned)(total_weight >> 32)) << 32;
                         else
                                 rand = 0;
 
                         if (rand == (total_weight & 0xffffffff00000000ULL))
-                                rand |= ll_rand() % (unsigned)total_weight;
+                                rand |= cfs_rand() % (unsigned)total_weight;
                         else
-                                rand |= ll_rand();
+                                rand |= cfs_rand();
 
 #else
-                        rand = ((__u64)ll_rand() << 32 | ll_rand()) %
+                        rand = ((__u64)cfs_rand() << 32 | cfs_rand()) %
                                 total_weight;
 #endif
                 } else {
@@ -915,11 +936,11 @@ static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
         LASSERT(nfound == *stripe_cnt);
 
 out:
-        up_write(&lov->lov_qos.lq_rw_sem);
+        cfs_up_write(&lov->lov_qos.lq_rw_sem);
 
 out_nolock:
         if (pool != NULL) {
-                up_read(&pool_tgt_rw_sem(pool));
+                cfs_up_read(&pool_tgt_rw_sem(pool));
                 /* put back ref got by lov_find_pool() */
                 lov_pool_putref(pool);
         }
@@ -927,7 +948,7 @@ out_nolock:
         if (rc == -EAGAIN)
                 rc = alloc_rr(lov, idx_arr, stripe_cnt, poolname, flags);
 
-        lov_putref(exp->exp_obd);
+        obd_putref(exp->exp_obd);
         RETURN(rc);
 }
 
@@ -1027,7 +1048,7 @@ int qos_prep_create(struct obd_export *exp, struct lov_request_set *set)
 
         lsm = set->set_oi->oi_md;
         lsm->lsm_object_id = src_oa->o_id;
-        lsm->lsm_object_gr = src_oa->o_gr;
+        lsm->lsm_object_seq = src_oa->o_seq;
 
         if (!lsm->lsm_stripe_size)
                 lsm->lsm_stripe_size = lov->desc.ld_default_stripe_size;
@@ -1053,7 +1074,7 @@ int qos_prep_create(struct obd_export *exp, struct lov_request_set *set)
                 lov_set_add_req(req, set);
 
                 req->rq_buflen = sizeof(*req->rq_oi.oi_md);
-                OBD_ALLOC(req->rq_oi.oi_md, req->rq_buflen);
+                OBD_ALLOC_LARGE(req->rq_oi.oi_md, req->rq_buflen);
                 if (req->rq_oi.oi_md == NULL)
                         GOTO(out_err, rc = -ENOMEM);
 
@@ -1065,6 +1086,7 @@ int qos_prep_create(struct obd_export *exp, struct lov_request_set *set)
                 req->rq_stripe = i;
                 /* create data objects with "parent" OA */
                 memcpy(req->rq_oi.oi_oa, src_oa, sizeof(*req->rq_oi.oi_oa));
+                req->rq_oi.oi_cb_up = cb_create_update;
 
                 /* XXX When we start creating objects on demand, we need to
                  *     make sure that we always create the object on the
@@ -1082,6 +1104,10 @@ int qos_prep_create(struct obd_export *exp, struct lov_request_set *set)
 
         if (stripes < lsm->lsm_stripe_count)
                 qos_shrink_lsm(set);
+        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_LOV_PREP_CREATE)) {
+                qos_shrink_lsm(set);
+                rc = -EIO;
+        }
 
         if (oti && (src_oa->o_valid & OBD_MD_FLCOOKIE)) {
                 oti_alloc_cookies(oti, set->set_count);
@@ -1103,3 +1129,110 @@ void qos_update(struct lov_obd *lov)
         ENTRY;
         lov->lov_qos.lq_dirty = 1;
 }
+
+void qos_statfs_done(struct lov_obd *lov)
+{
+        LASSERT(lov->lov_qos.lq_statfs_in_progress);
+        cfs_down_write(&lov->lov_qos.lq_rw_sem);
+        lov->lov_qos.lq_statfs_in_progress = 0;
+        /* wake up any threads waiting for the statfs rpcs to complete */
+        cfs_waitq_signal(&lov->lov_qos.lq_statfs_waitq);
+        cfs_up_write(&lov->lov_qos.lq_rw_sem);
+}
+
+static int qos_statfs_ready(struct obd_device *obd, __u64 max_age)
+{
+        struct lov_obd         *lov = &obd->u.lov;
+        int rc;
+        ENTRY;
+        cfs_down_read(&lov->lov_qos.lq_rw_sem);
+        rc = lov->lov_qos.lq_statfs_in_progress == 0 ||
+             cfs_time_beforeq_64(max_age, obd->obd_osfs_age);
+        cfs_up_read(&lov->lov_qos.lq_rw_sem);
+        RETURN(rc);
+}
+
+/*
+ * Update statfs data if the current osfs age is older than max_age.
+ * If wait is not set, it means that we are called from lov_create()
+ * and we should just issue the rpcs without waiting for them to complete.
+ * If wait is set, we are called from alloc_qos() and we just have
+ * to wait for the request set to complete.
+ */
+void qos_statfs_update(struct obd_device *obd, __u64 max_age, int wait)
+{
+        struct lov_obd         *lov = &obd->u.lov;
+        struct obd_info        *oinfo;
+        int                     rc = 0;
+        struct ptlrpc_request_set *set = NULL;
+        ENTRY;
+
+        if (cfs_time_beforeq_64(max_age, obd->obd_osfs_age))
+                /* statfs data are quite recent, don't need to refresh it */
+                RETURN_EXIT;
+
+        if (!wait && lov->lov_qos.lq_statfs_in_progress)
+                /* statfs already in progress */
+                RETURN_EXIT;
+
+        cfs_down_write(&lov->lov_qos.lq_rw_sem);
+        if (lov->lov_qos.lq_statfs_in_progress) {
+                cfs_up_write(&lov->lov_qos.lq_rw_sem);
+                GOTO(out, rc = 0);
+        }
+        /* no statfs in flight, send rpcs */
+        lov->lov_qos.lq_statfs_in_progress = 1;
+        cfs_up_write(&lov->lov_qos.lq_rw_sem);
+
+        if (wait)
+                CDEBUG(D_QOS, "%s: did not manage to get fresh statfs data "
+                       "in a timely manner (osfs age "LPU64", max age "LPU64")"
+                       ", sending new statfs rpcs\n",
+                       obd_uuid2str(&lov->desc.ld_uuid), obd->obd_osfs_age,
+                       max_age);
+
+        /* need to send statfs rpcs */
+        CDEBUG(D_QOS, "sending new statfs requests\n");
+        memset(lov->lov_qos.lq_statfs_data, 0,
+               sizeof(*lov->lov_qos.lq_statfs_data));
+        oinfo = &lov->lov_qos.lq_statfs_data->lsd_oi;
+        oinfo->oi_osfs = &lov->lov_qos.lq_statfs_data->lsd_statfs;
+        oinfo->oi_flags = OBD_STATFS_NODELAY;
+        set = ptlrpc_prep_set();
+        if (!set)
+                GOTO(out_failed, rc = -ENOMEM);
+
+        rc = obd_statfs_async(obd, oinfo, max_age, set);
+        if (rc || cfs_list_empty(&set->set_requests)) {
+                if (rc)
+                        CWARN("statfs failed with %d\n", rc);
+                GOTO(out_failed, rc);
+        }
+        /* send requests via ptlrpcd */
+        oinfo->oi_flags |= OBD_STATFS_PTLRPCD;
+        ptlrpcd_add_rqset(set);
+        GOTO(out, rc);
+
+out_failed:
+        cfs_down_write(&lov->lov_qos.lq_rw_sem);
+        lov->lov_qos.lq_statfs_in_progress = 0;
+        /* wake up any threads waiting for the statfs rpcs to complete */
+        cfs_waitq_signal(&lov->lov_qos.lq_statfs_waitq);
+        cfs_up_write(&lov->lov_qos.lq_rw_sem);
+        wait = 0;
+out:
+        if (set)
+                ptlrpc_set_destroy(set);
+        if (wait) {
+                struct l_wait_info lwi = { 0 };
+                CDEBUG(D_QOS, "waiting for statfs requests to complete\n");
+                l_wait_event(lov->lov_qos.lq_statfs_waitq,
+                             qos_statfs_ready(obd, max_age), &lwi);
+                if (cfs_time_before_64(obd->obd_osfs_age, max_age))
+                        CDEBUG(D_QOS, "%s: still no fresh statfs data after "
+                                      "waiting (osfs age "LPU64", max age "
+                                      LPU64")\n",
+                                      obd_uuid2str(&lov->desc.ld_uuid),
+                                      obd->obd_osfs_age, max_age);
+        }
+}