Whamcloud - gitweb
LU-1304 osd: ->do_ah_init() to accept new object
[fs/lustre-release.git] / lustre / lov / lov_qos.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2012, Whamcloud, Inc.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  */
36
37 #define DEBUG_SUBSYSTEM S_LOV
38
39 #ifdef __KERNEL__
40 #include <libcfs/libcfs.h>
41 #else
42 #include <liblustre.h>
43 #endif
44
45 #include <obd_class.h>
46 #include <obd_lov.h>
47 #include <lustre/lustre_idl.h>
48 #include "lov_internal.h"
49
50 /* #define QOS_DEBUG 1 */
51 #define D_QOS D_OTHER
52
53 #define TGT_BAVAIL(i) (lov->lov_tgts[i]->ltd_exp->exp_obd->obd_osfs.os_bavail *\
54                        lov->lov_tgts[i]->ltd_exp->exp_obd->obd_osfs.os_bsize)
55
56
57 int qos_add_tgt(struct obd_device *obd, __u32 index)
58 {
59         struct lov_obd *lov = &obd->u.lov;
60         struct lov_qos_oss *oss, *temposs;
61         struct obd_export *exp = lov->lov_tgts[index]->ltd_exp;
62         int rc = 0, found = 0;
63         ENTRY;
64
65         /* We only need this QOS struct on MDT, not clients - but we may not
66          * have registered the LOV's observer yet, so there's no way to know */
67         if (!exp || !exp->exp_connection) {
68                 CERROR("Missing connection\n");
69                 RETURN(-ENOTCONN);
70         }
71
72         cfs_down_write(&lov->lov_qos.lq_rw_sem);
73         cfs_mutex_lock(&lov->lov_lock);
74         cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
75                 if (obd_uuid_equals(&oss->lqo_uuid,
76                                     &exp->exp_connection->c_remote_uuid)) {
77                         found++;
78                         break;
79                 }
80         }
81
82         if (!found) {
83                 OBD_ALLOC_PTR(oss);
84                 if (!oss)
85                         GOTO(out, rc = -ENOMEM);
86                 memcpy(&oss->lqo_uuid,
87                        &exp->exp_connection->c_remote_uuid,
88                        sizeof(oss->lqo_uuid));
89         } else {
90                 /* Assume we have to move this one */
91                 cfs_list_del(&oss->lqo_oss_list);
92         }
93
94         oss->lqo_ost_count++;
95         lov->lov_tgts[index]->ltd_qos.ltq_oss = oss;
96
97         /* Add sorted by # of OSTs.  Find the first entry that we're
98            bigger than... */
99         cfs_list_for_each_entry(temposs, &lov->lov_qos.lq_oss_list,
100                                 lqo_oss_list) {
101                 if (oss->lqo_ost_count > temposs->lqo_ost_count)
102                         break;
103         }
104         /* ...and add before it.  If we're the first or smallest, temposs
105            points to the list head, and we add to the end. */
106         cfs_list_add_tail(&oss->lqo_oss_list, &temposs->lqo_oss_list);
107
108         lov->lov_qos.lq_dirty = 1;
109         lov->lov_qos.lq_rr.lqr_dirty = 1;
110
111         CDEBUG(D_QOS, "add tgt %s to OSS %s (%d OSTs)\n",
112                obd_uuid2str(&lov->lov_tgts[index]->ltd_uuid),
113                obd_uuid2str(&oss->lqo_uuid),
114                oss->lqo_ost_count);
115
116 out:
117         cfs_mutex_unlock(&lov->lov_lock);
118         cfs_up_write(&lov->lov_qos.lq_rw_sem);
119         RETURN(rc);
120 }
121
122 int qos_del_tgt(struct obd_device *obd, struct lov_tgt_desc *tgt)
123 {
124         struct lov_obd *lov = &obd->u.lov;
125         struct lov_qos_oss *oss;
126         int rc = 0;
127         ENTRY;
128
129         cfs_down_write(&lov->lov_qos.lq_rw_sem);
130
131         oss = tgt->ltd_qos.ltq_oss;
132         if (!oss)
133                 GOTO(out, rc = -ENOENT);
134
135         oss->lqo_ost_count--;
136         if (oss->lqo_ost_count == 0) {
137                 CDEBUG(D_QOS, "removing OSS %s\n",
138                        obd_uuid2str(&oss->lqo_uuid));
139                 cfs_list_del(&oss->lqo_oss_list);
140                 OBD_FREE_PTR(oss);
141         }
142
143         lov->lov_qos.lq_dirty = 1;
144         lov->lov_qos.lq_rr.lqr_dirty = 1;
145 out:
146         cfs_up_write(&lov->lov_qos.lq_rw_sem);
147         RETURN(rc);
148 }
149
150 /* Recalculate per-object penalties for OSSs and OSTs,
151    depends on size of each ost in an oss */
152 static int qos_calc_ppo(struct obd_device *obd)
153 {
154         struct lov_obd *lov = &obd->u.lov;
155         struct lov_qos_oss *oss;
156         __u64 ba_max, ba_min, temp;
157         __u32 num_active;
158         int rc, i, prio_wide;
159         time_t now, age;
160         ENTRY;
161
162         if (!lov->lov_qos.lq_dirty)
163                 GOTO(out, rc = 0);
164
165         num_active = lov->desc.ld_active_tgt_count - 1;
166         if (num_active < 1)
167                 GOTO(out, rc = -EAGAIN);
168
169         /* find bavail on each OSS */
170         cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
171                 oss->lqo_bavail = 0;
172         }
173         lov->lov_qos.lq_active_oss_count = 0;
174
175         /* How badly user wants to select osts "widely" (not recently chosen
176            and not on recent oss's).  As opposed to "freely" (free space
177            avail.) 0-256. */
178         prio_wide = 256 - lov->lov_qos.lq_prio_free;
179
180         ba_min = (__u64)(-1);
181         ba_max = 0;
182         now = cfs_time_current_sec();
183         /* Calculate OST penalty per object */
184         /* (lov ref taken in alloc_qos) */
185         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
186                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active)
187                         continue;
188                 temp = TGT_BAVAIL(i);
189                 if (!temp)
190                         continue;
191                 ba_min = min(temp, ba_min);
192                 ba_max = max(temp, ba_max);
193
194                 /* Count the number of usable OSS's */
195                 if (lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_bavail == 0)
196                         lov->lov_qos.lq_active_oss_count++;
197                 lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_bavail += temp;
198
199                 /* per-OST penalty is prio * TGT_bavail / (num_ost - 1) / 2 */
200                 temp >>= 1;
201                 lov_do_div64(temp, num_active);
202                 lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj =
203                         (temp * prio_wide) >> 8;
204
205                 age = (now - lov->lov_tgts[i]->ltd_qos.ltq_used) >> 3;
206                 if (lov->lov_qos.lq_reset || age > 32 * lov->desc.ld_qos_maxage)
207                         lov->lov_tgts[i]->ltd_qos.ltq_penalty = 0;
208                 else if (age > lov->desc.ld_qos_maxage)
209                         /* Decay the penalty by half for every 8x the update
210                          * interval that the device has been idle.  That gives
211                          * lots of time for the statfs information to be
212                          * updated (which the penalty is only a proxy for),
213                          * and avoids penalizing OSS/OSTs under light load. */
214                         lov->lov_tgts[i]->ltd_qos.ltq_penalty >>=
215                                 (age / lov->desc.ld_qos_maxage);
216         }
217
218         num_active = lov->lov_qos.lq_active_oss_count - 1;
219         if (num_active < 1) {
220                 /* If there's only 1 OSS, we can't penalize it, so instead
221                    we have to double the OST penalty */
222                 num_active = 1;
223                 for (i = 0; i < lov->desc.ld_tgt_count; i++) {
224                         if (lov->lov_tgts[i] == NULL)
225                                 continue;
226                         lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj <<= 1;
227                 }
228         }
229
230         /* Per-OSS penalty is prio * oss_avail / oss_osts / (num_oss - 1) / 2 */
231         cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
232                 temp = oss->lqo_bavail >> 1;
233                 lov_do_div64(temp, oss->lqo_ost_count * num_active);
234                 oss->lqo_penalty_per_obj = (temp * prio_wide) >> 8;
235
236                 age = (now - oss->lqo_used) >> 3;
237                 if (lov->lov_qos.lq_reset || age > 32 * lov->desc.ld_qos_maxage)
238                         oss->lqo_penalty = 0;
239                 else if (age > lov->desc.ld_qos_maxage)
240                         /* Decay the penalty by half for every 8x the update
241                          * interval that the device has been idle.  That gives
242                          * lots of time for the statfs information to be
243                          * updated (which the penalty is only a proxy for),
244                          * and avoids penalizing OSS/OSTs under light load. */
245                         oss->lqo_penalty >>= (age / lov->desc.ld_qos_maxage);
246         }
247
248         lov->lov_qos.lq_dirty = 0;
249         lov->lov_qos.lq_reset = 0;
250
251         /* If each ost has almost same free space,
252          * do rr allocation for better creation performance */
253         lov->lov_qos.lq_same_space = 0;
254         if ((ba_max * (256 - lov->lov_qos.lq_threshold_rr)) >> 8 < ba_min) {
255                 lov->lov_qos.lq_same_space = 1;
256                 /* Reset weights for the next time we enter qos mode */
257                 lov->lov_qos.lq_reset = 1;
258         }
259         rc = 0;
260
261 out:
262         if (!rc && lov->lov_qos.lq_same_space)
263                 RETURN(-EAGAIN);
264         RETURN(rc);
265 }
266
267 static int qos_calc_weight(struct lov_obd *lov, int i)
268 {
269         __u64 temp, temp2;
270
271         /* Final ost weight = TGT_BAVAIL - ost_penalty - oss_penalty */
272         temp = TGT_BAVAIL(i);
273         temp2 = lov->lov_tgts[i]->ltd_qos.ltq_penalty +
274                 lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_penalty;
275         if (temp < temp2)
276                 lov->lov_tgts[i]->ltd_qos.ltq_weight = 0;
277         else
278                 lov->lov_tgts[i]->ltd_qos.ltq_weight = temp - temp2;
279         return 0;
280 }
281
282 /* We just used this index for a stripe; adjust everyone's weights */
283 static int qos_used(struct lov_obd *lov, struct ost_pool *osts,
284                     __u32 index, __u64 *total_wt)
285 {
286         struct lov_qos_oss *oss;
287         int j;
288         ENTRY;
289
290         /* Don't allocate from this stripe anymore, until the next alloc_qos */
291         lov->lov_tgts[index]->ltd_qos.ltq_usable = 0;
292
293         oss = lov->lov_tgts[index]->ltd_qos.ltq_oss;
294
295         /* Decay old penalty by half (we're adding max penalty, and don't
296            want it to run away.) */
297         lov->lov_tgts[index]->ltd_qos.ltq_penalty >>= 1;
298         oss->lqo_penalty >>= 1;
299
300         /* mark the OSS and OST as recently used */
301         lov->lov_tgts[index]->ltd_qos.ltq_used =
302                 oss->lqo_used = cfs_time_current_sec();
303
304         /* Set max penalties for this OST and OSS */
305         lov->lov_tgts[index]->ltd_qos.ltq_penalty +=
306                 lov->lov_tgts[index]->ltd_qos.ltq_penalty_per_obj *
307                 lov->desc.ld_active_tgt_count;
308         oss->lqo_penalty += oss->lqo_penalty_per_obj *
309                 lov->lov_qos.lq_active_oss_count;
310
311         /* Decrease all OSS penalties */
312         cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
313                 if (oss->lqo_penalty < oss->lqo_penalty_per_obj)
314                         oss->lqo_penalty = 0;
315                 else
316                         oss->lqo_penalty -= oss->lqo_penalty_per_obj;
317         }
318
319         *total_wt = 0;
320         /* Decrease all OST penalties */
321         for (j = 0; j < osts->op_count; j++) {
322                 int i;
323
324                 i = osts->op_array[j];
325                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active)
326                         continue;
327                 if (lov->lov_tgts[i]->ltd_qos.ltq_penalty <
328                     lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj)
329                         lov->lov_tgts[i]->ltd_qos.ltq_penalty = 0;
330                 else
331                         lov->lov_tgts[i]->ltd_qos.ltq_penalty -=
332                         lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj;
333
334                 qos_calc_weight(lov, i);
335
336                 /* Recalc the total weight of usable osts */
337                 if (lov->lov_tgts[i]->ltd_qos.ltq_usable)
338                         *total_wt += lov->lov_tgts[i]->ltd_qos.ltq_weight;
339
340 #ifdef QOS_DEBUG
341                 CDEBUG(D_QOS, "recalc tgt %d usable=%d avail="LPU64
342                        " ostppo="LPU64" ostp="LPU64" ossppo="LPU64
343                        " ossp="LPU64" wt="LPU64"\n",
344                        i, lov->lov_tgts[i]->ltd_qos.ltq_usable,
345                        TGT_BAVAIL(i) >> 10,
346                        lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj >> 10,
347                        lov->lov_tgts[i]->ltd_qos.ltq_penalty >> 10,
348                        lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_penalty_per_obj>>10,
349                        lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_penalty >> 10,
350                        lov->lov_tgts[i]->ltd_qos.ltq_weight >> 10);
351 #endif
352         }
353
354         RETURN(0);
355 }
356
357 #define LOV_QOS_EMPTY ((__u32)-1)
358 /* compute optimal round-robin order, based on OSTs per OSS */
359 static int qos_calc_rr(struct lov_obd *lov, struct ost_pool *src_pool,
360                        struct lov_qos_rr *lqr)
361 {
362         struct lov_qos_oss *oss;
363         unsigned placed, real_count;
364         int i, rc;
365         ENTRY;
366
367         if (!lqr->lqr_dirty) {
368                 LASSERT(lqr->lqr_pool.op_size);
369                 RETURN(0);
370         }
371
372         /* Do actual allocation. */
373         cfs_down_write(&lov->lov_qos.lq_rw_sem);
374
375         /*
376          * Check again. While we were sleeping on @lq_rw_sem something could
377          * change.
378          */
379         if (!lqr->lqr_dirty) {
380                 LASSERT(lqr->lqr_pool.op_size);
381                 cfs_up_write(&lov->lov_qos.lq_rw_sem);
382                 RETURN(0);
383         }
384
385         real_count = src_pool->op_count;
386
387         /* Zero the pool array */
388         /* alloc_rr is holding a read lock on the pool, so nobody is adding/
389            deleting from the pool. The lq_rw_sem insures that nobody else
390            is reading. */
391         lqr->lqr_pool.op_count = real_count;
392         rc = lov_ost_pool_extend(&lqr->lqr_pool, real_count);
393         if (rc) {
394                 cfs_up_write(&lov->lov_qos.lq_rw_sem);
395                 RETURN(rc);
396         }
397         for (i = 0; i < lqr->lqr_pool.op_count; i++)
398                 lqr->lqr_pool.op_array[i] = LOV_QOS_EMPTY;
399
400         /* Place all the OSTs from 1 OSS at the same time. */
401         placed = 0;
402         cfs_list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
403                 int j = 0;
404                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
405                         if (lov->lov_tgts[src_pool->op_array[i]] &&
406                             (lov->lov_tgts[src_pool->op_array[i]]->ltd_qos.ltq_oss == oss)) {
407                               /* Evenly space these OSTs across arrayspace */
408                               int next = j * lqr->lqr_pool.op_count / oss->lqo_ost_count;
409                               while (lqr->lqr_pool.op_array[next] !=
410                                      LOV_QOS_EMPTY)
411                                         next = (next + 1) % lqr->lqr_pool.op_count;
412                               lqr->lqr_pool.op_array[next] = src_pool->op_array[i];
413                               j++;
414                               placed++;
415                         }
416                 }
417         }
418
419         lqr->lqr_dirty = 0;
420         cfs_up_write(&lov->lov_qos.lq_rw_sem);
421
422         if (placed != real_count) {
423                 /* This should never happen */
424                 LCONSOLE_ERROR_MSG(0x14e, "Failed to place all OSTs in the "
425                                    "round-robin list (%d of %d).\n",
426                                    placed, real_count);
427                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
428                         LCONSOLE(D_WARNING, "rr #%d ost idx=%d\n", i,
429                                  lqr->lqr_pool.op_array[i]);
430                 }
431                 lqr->lqr_dirty = 1;
432                 RETURN(-EAGAIN);
433         }
434
435 #ifdef QOS_DEBUG
436         for (i = 0; i < lqr->lqr_pool.op_count; i++) {
437                 LCONSOLE(D_QOS, "rr #%d ost idx=%d\n", i,
438                          lqr->lqr_pool.op_array[i]);
439         }
440 #endif
441
442         RETURN(0);
443 }
444
445
446 void qos_shrink_lsm(struct lov_request_set *set)
447 {
448         struct lov_stripe_md *lsm = set->set_oi->oi_md, *lsm_new;
449         /* XXX LOV STACKING call into osc for sizes */
450         unsigned oldsize, newsize;
451
452         if (set->set_oti && set->set_cookies && set->set_cookie_sent) {
453                 struct llog_cookie *cookies;
454                 oldsize = lsm->lsm_stripe_count * sizeof(*cookies);
455                 newsize = set->set_count * sizeof(*cookies);
456
457                 cookies = set->set_cookies;
458                 oti_alloc_cookies(set->set_oti, set->set_count);
459                 if (set->set_oti->oti_logcookies) {
460                         memcpy(set->set_oti->oti_logcookies, cookies, newsize);
461                         OBD_FREE_LARGE(cookies, oldsize);
462                         set->set_cookies = set->set_oti->oti_logcookies;
463                 } else {
464                         CWARN("'leaking' %d bytes\n", oldsize - newsize);
465                 }
466         }
467
468         CWARN("using fewer stripes for object "LPU64": old %u new %u\n",
469               lsm->lsm_object_id, lsm->lsm_stripe_count, set->set_count);
470         LASSERT(lsm->lsm_stripe_count >= set->set_count);
471
472         newsize = lov_stripe_md_size(set->set_count);
473         OBD_ALLOC_LARGE(lsm_new, newsize);
474         if (lsm_new != NULL) {
475                 int i;
476                 memcpy(lsm_new, lsm, sizeof(*lsm));
477                 for (i = 0; i < lsm->lsm_stripe_count; i++) {
478                         if (i < set->set_count) {
479                                 lsm_new->lsm_oinfo[i] = lsm->lsm_oinfo[i];
480                                 continue;
481                         }
482                         OBD_SLAB_FREE(lsm->lsm_oinfo[i], lov_oinfo_slab,
483                                       sizeof(struct lov_oinfo));
484                 }
485                 lsm_new->lsm_stripe_count = set->set_count;
486                 OBD_FREE_LARGE(lsm, sizeof(struct lov_stripe_md) +
487                                lsm->lsm_stripe_count*sizeof(struct lov_oinfo*));
488                 set->set_oi->oi_md = lsm_new;
489         } else {
490                 CWARN("'leaking' few bytes\n");
491         }
492 }
493
494 /**
495  * Check whether we can create the object on the OST(refered by ost_idx)
496  * \retval:
497  *          0: create the object.
498  *          other value: did not create the object.
499  */
500 static int lov_check_and_create_object(struct lov_obd *lov, int ost_idx,
501                                        struct lov_stripe_md *lsm,
502                                        struct lov_request *req,
503                                        struct obd_trans_info *oti)
504 {
505         __u16 stripe;
506         int rc = -EIO;
507         ENTRY;
508
509         CDEBUG(D_QOS, "Check and create on idx %d \n", ost_idx);
510         if (!lov->lov_tgts[ost_idx] ||
511             !lov->lov_tgts[ost_idx]->ltd_active)
512                 RETURN(rc);
513
514         /* check if objects has been created on this ost */
515         for (stripe = 0; stripe < lsm->lsm_stripe_count; stripe++) {
516                 /* already have object at this stripe */
517                 if (ost_idx == lsm->lsm_oinfo[stripe]->loi_ost_idx)
518                         break;
519         }
520
521         if (stripe >= lsm->lsm_stripe_count) {
522                 req->rq_idx = ost_idx;
523                 rc = obd_create(NULL, lov->lov_tgts[ost_idx]->ltd_exp,
524                                 req->rq_oi.oi_oa, &req->rq_oi.oi_md,
525                                 oti);
526         }
527         RETURN(rc);
528 }
529
530 int qos_remedy_create(struct lov_request_set *set, struct lov_request *req)
531 {
532         struct lov_stripe_md    *lsm = set->set_oi->oi_md;
533         struct lov_obd          *lov = &set->set_exp->exp_obd->u.lov;
534         unsigned                ost_idx;
535         unsigned                ost_count;
536         struct pool_desc        *pool;
537         struct ost_pool         *osts = NULL;
538         int                     i;
539         int                     rc = -EIO;
540         ENTRY;
541
542         /* First check whether we can create the objects on the pool */
543         pool = lov_find_pool(lov, lsm->lsm_pool_name);
544         if (pool != NULL) {
545                 cfs_down_read(&pool_tgt_rw_sem(pool));
546                 osts = &(pool->pool_obds);
547                 ost_count = osts->op_count;
548                 for (i = 0, ost_idx = osts->op_array[0]; i < ost_count;
549                      i++, ost_idx = osts->op_array[i]) {
550                         rc = lov_check_and_create_object(lov, ost_idx, lsm, req,
551                                                          set->set_oti);
552                         if (rc == 0)
553                                 break;
554                 }
555                 cfs_up_read(&pool_tgt_rw_sem(pool));
556                 lov_pool_putref(pool);
557                 RETURN(rc);
558         }
559
560         ost_count = lov->desc.ld_tgt_count;
561         /* Then check whether we can create the objects on other OSTs */
562         ost_idx = (req->rq_idx + lsm->lsm_stripe_count) % ost_count;
563         for (i = 0; i < ost_count; i++, ost_idx = (ost_idx + 1) % ost_count) {
564                 rc = lov_check_and_create_object(lov, ost_idx, lsm, req,
565                                                  set->set_oti);
566
567                 if (rc == 0)
568                         break;
569         }
570
571         RETURN(rc);
572 }
573
574 static int min_stripe_count(int stripe_cnt, int flags)
575 {
576         return (flags & LOV_USES_DEFAULT_STRIPE ?
577                 stripe_cnt - (stripe_cnt / 4) : stripe_cnt);
578 }
579
580 #define LOV_CREATE_RESEED_MULT 30
581 #define LOV_CREATE_RESEED_MIN  2000
582 /* Allocate objects on osts with round-robin algorithm */
583 static int alloc_rr(struct lov_obd *lov, int *idx_arr, int *stripe_cnt,
584                     char *poolname, int flags)
585 {
586         unsigned array_idx;
587         int i, rc, *idx_pos;
588         __u32 ost_idx;
589         int ost_start_idx_temp;
590         int speed = 0;
591         int stripe_cnt_min = min_stripe_count(*stripe_cnt, flags);
592         struct pool_desc *pool;
593         struct ost_pool *osts;
594         struct lov_qos_rr *lqr;
595         ENTRY;
596
597         pool = lov_find_pool(lov, poolname);
598         if (pool == NULL) {
599                 osts = &(lov->lov_packed);
600                 lqr = &(lov->lov_qos.lq_rr);
601         } else {
602                 cfs_down_read(&pool_tgt_rw_sem(pool));
603                 osts = &(pool->pool_obds);
604                 lqr = &(pool->pool_rr);
605         }
606
607         rc = qos_calc_rr(lov, osts, lqr);
608         if (rc)
609                 GOTO(out, rc);
610
611         if (--lqr->lqr_start_count <= 0) {
612                 lqr->lqr_start_idx = cfs_rand() % osts->op_count;
613                 lqr->lqr_start_count =
614                         (LOV_CREATE_RESEED_MIN / max(osts->op_count, 1U) +
615                          LOV_CREATE_RESEED_MULT) * max(osts->op_count, 1U);
616         } else if (stripe_cnt_min >= osts->op_count ||
617                    lqr->lqr_start_idx > osts->op_count) {
618                 /* If we have allocated from all of the OSTs, slowly
619                  * precess the next start if the OST/stripe count isn't
620                  * already doing this for us. */
621                 lqr->lqr_start_idx %= osts->op_count;
622                 if (*stripe_cnt > 1 && (osts->op_count % (*stripe_cnt)) != 1)
623                         ++lqr->lqr_offset_idx;
624         }
625         cfs_down_read(&lov->lov_qos.lq_rw_sem);
626         ost_start_idx_temp = lqr->lqr_start_idx;
627
628 repeat_find:
629         array_idx = (lqr->lqr_start_idx + lqr->lqr_offset_idx) % osts->op_count;
630         idx_pos = idx_arr;
631 #ifdef QOS_DEBUG
632         CDEBUG(D_QOS, "pool '%s' want %d startidx %d startcnt %d offset %d "
633                "active %d count %d arrayidx %d\n", poolname,
634                *stripe_cnt, lqr->lqr_start_idx, lqr->lqr_start_count,
635                lqr->lqr_offset_idx, osts->op_count, osts->op_count, array_idx);
636 #endif
637
638         for (i = 0; i < osts->op_count;
639                     i++, array_idx=(array_idx + 1) % osts->op_count) {
640                 ++lqr->lqr_start_idx;
641                 ost_idx = lqr->lqr_pool.op_array[array_idx];
642 #ifdef QOS_DEBUG
643                 CDEBUG(D_QOS, "#%d strt %d act %d strp %d ary %d idx %d\n",
644                        i, lqr->lqr_start_idx,
645                        ((ost_idx != LOV_QOS_EMPTY) && lov->lov_tgts[ost_idx]) ?
646                        lov->lov_tgts[ost_idx]->ltd_active : 0,
647                        idx_pos - idx_arr, array_idx, ost_idx);
648 #endif
649                 if ((ost_idx == LOV_QOS_EMPTY) || !lov->lov_tgts[ost_idx] ||
650                     !lov->lov_tgts[ost_idx]->ltd_active)
651                         continue;
652
653                 /* Fail Check before osc_precreate() is called
654                    so we can only 'fail' single OSC. */
655                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
656                         continue;
657
658                 /* Drop slow OSCs if we can */
659                 if (obd_precreate(lov->lov_tgts[ost_idx]->ltd_exp) > speed)
660                         continue;
661
662                 *idx_pos = ost_idx;
663                 idx_pos++;
664                 /* We have enough stripes */
665                 if (idx_pos - idx_arr == *stripe_cnt)
666                         break;
667         }
668         if ((speed < 2) && (idx_pos - idx_arr < stripe_cnt_min)) {
669                 /* Try again, allowing slower OSCs */
670                 speed++;
671                 lqr->lqr_start_idx = ost_start_idx_temp;
672                 goto repeat_find;
673         }
674
675         cfs_up_read(&lov->lov_qos.lq_rw_sem);
676
677         *stripe_cnt = idx_pos - idx_arr;
678 out:
679         if (pool != NULL) {
680                 cfs_up_read(&pool_tgt_rw_sem(pool));
681                 /* put back ref got by lov_find_pool() */
682                 lov_pool_putref(pool);
683         }
684
685         RETURN(rc);
686 }
687
688 /* alloc objects on osts with specific stripe offset */
689 static int alloc_specific(struct lov_obd *lov, struct lov_stripe_md *lsm,
690                           int *idx_arr)
691 {
692         unsigned ost_idx, array_idx, ost_count;
693         int i, rc, *idx_pos;
694         int speed = 0;
695         struct pool_desc *pool;
696         struct ost_pool *osts;
697         ENTRY;
698
699         pool = lov_find_pool(lov, lsm->lsm_pool_name);
700         if (pool == NULL) {
701                 osts = &(lov->lov_packed);
702         } else {
703                 cfs_down_read(&pool_tgt_rw_sem(pool));
704                 osts = &(pool->pool_obds);
705         }
706
707         ost_count = osts->op_count;
708
709 repeat_find:
710         /* search loi_ost_idx in ost array */
711         array_idx = 0;
712         for (i = 0; i < ost_count; i++) {
713                 if (osts->op_array[i] == lsm->lsm_oinfo[0]->loi_ost_idx) {
714                         array_idx = i;
715                         break;
716                 }
717         }
718         if (i == ost_count) {
719                 CERROR("Start index %d not found in pool '%s'\n",
720                        lsm->lsm_oinfo[0]->loi_ost_idx, lsm->lsm_pool_name);
721                 GOTO(out, rc = -EINVAL);
722         }
723
724         idx_pos = idx_arr;
725         for (i = 0; i < ost_count;
726              i++, array_idx = (array_idx + 1) % ost_count) {
727                 ost_idx = osts->op_array[array_idx];
728
729                 if (!lov->lov_tgts[ost_idx] ||
730                     !lov->lov_tgts[ost_idx]->ltd_active) {
731                         continue;
732                 }
733
734                 /* Fail Check before osc_precreate() is called
735                    so we can only 'fail' single OSC. */
736                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
737                         continue;
738
739                 /* Drop slow OSCs if we can, but not for requested start idx.
740                  *
741                  * This means "if OSC is slow and it is not the requested
742                  * start OST, then it can be skipped, otherwise skip it only
743                  * if it is inactive/recovering/out-of-space." */
744                 if ((obd_precreate(lov->lov_tgts[ost_idx]->ltd_exp) > speed) &&
745                     (i != 0 || speed >= 2))
746                         continue;
747
748                 *idx_pos = ost_idx;
749                 idx_pos++;
750                 /* We have enough stripes */
751                 if (idx_pos - idx_arr == lsm->lsm_stripe_count)
752                         GOTO(out, rc = 0);
753         }
754         if (speed < 2) {
755                 /* Try again, allowing slower OSCs */
756                 speed++;
757                 goto repeat_find;
758         }
759
760         /* If we were passed specific striping params, then a failure to
761          * meet those requirements is an error, since we can't reallocate
762          * that memory (it might be part of a larger array or something).
763          *
764          * We can only get here if lsm_stripe_count was originally > 1.
765          */
766         CERROR("can't lstripe objid "LPX64": have %d want %u\n",
767                lsm->lsm_object_id, (int)(idx_pos - idx_arr),
768                lsm->lsm_stripe_count);
769         rc = -EFBIG;
770 out:
771         if (pool != NULL) {
772                 cfs_up_read(&pool_tgt_rw_sem(pool));
773                 /* put back ref got by lov_find_pool() */
774                 lov_pool_putref(pool);
775         }
776
777         RETURN(rc);
778 }
779
780 /* Alloc objects on osts with optimization based on:
781    - free space
782    - network resources (shared OSS's)
783 */
784 static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
785                      char *poolname, int flags)
786 {
787         struct lov_obd *lov = &exp->exp_obd->u.lov;
788         __u64 total_weight = 0;
789         int nfound, good_osts, i, rc = 0;
790         int stripe_cnt_min = min_stripe_count(*stripe_cnt, flags);
791         struct pool_desc *pool;
792         struct ost_pool *osts;
793         ENTRY;
794
795         if (stripe_cnt_min < 1)
796                 RETURN(-EINVAL);
797
798         pool = lov_find_pool(lov, poolname);
799         if (pool == NULL) {
800                 osts = &(lov->lov_packed);
801         } else {
802                 cfs_down_read(&pool_tgt_rw_sem(pool));
803                 osts = &(pool->pool_obds);
804         }
805
806         obd_getref(exp->exp_obd);
807
808         /* wait for fresh statfs info if needed, the rpcs are sent in
809          * lov_create() */
810         qos_statfs_update(exp->exp_obd,
811                           cfs_time_shift_64(-2 * lov->desc.ld_qos_maxage), 1);
812
813         /* Detect -EAGAIN early, before expensive lock is taken. */
814         if (!lov->lov_qos.lq_dirty && lov->lov_qos.lq_same_space)
815                 GOTO(out_nolock, rc = -EAGAIN);
816
817         /* Do actual allocation, use write lock here. */
818         cfs_down_write(&lov->lov_qos.lq_rw_sem);
819
820         /*
821          * Check again, while we were sleeping on @lq_rw_sem things could
822          * change.
823          */
824         if (!lov->lov_qos.lq_dirty && lov->lov_qos.lq_same_space)
825                 GOTO(out, rc = -EAGAIN);
826
827         if (lov->desc.ld_active_tgt_count < 2)
828                 GOTO(out, rc = -EAGAIN);
829
830         rc = qos_calc_ppo(exp->exp_obd);
831         if (rc)
832                 GOTO(out, rc);
833
834         good_osts = 0;
835         /* Find all the OSTs that are valid stripe candidates */
836         for (i = 0; i < osts->op_count; i++) {
837                 if (!lov->lov_tgts[osts->op_array[i]] ||
838                     !lov->lov_tgts[osts->op_array[i]]->ltd_active)
839                         continue;
840
841                 /* Fail Check before osc_precreate() is called
842                    so we can only 'fail' single OSC. */
843                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && osts->op_array[i] == 0)
844                         continue;
845
846                 if (obd_precreate(lov->lov_tgts[osts->op_array[i]]->ltd_exp) > 2)
847                         continue;
848
849                 lov->lov_tgts[osts->op_array[i]]->ltd_qos.ltq_usable = 1;
850                 qos_calc_weight(lov, osts->op_array[i]);
851                 total_weight += lov->lov_tgts[osts->op_array[i]]->ltd_qos.ltq_weight;
852
853                 good_osts++;
854         }
855
856 #ifdef QOS_DEBUG
857         CDEBUG(D_QOS, "found %d good osts\n", good_osts);
858 #endif
859
860         if (good_osts < stripe_cnt_min)
861                 GOTO(out, rc = -EAGAIN);
862
863         /* We have enough osts */
864         if (good_osts < *stripe_cnt)
865                 *stripe_cnt = good_osts;
866
867         if (!*stripe_cnt)
868                 GOTO(out, rc = -EAGAIN);
869
870         /* Find enough OSTs with weighted random allocation. */
871         nfound = 0;
872         while (nfound < *stripe_cnt) {
873                 __u64 rand, cur_weight;
874
875                 cur_weight = 0;
876                 rc = -ENODEV;
877
878                 if (total_weight) {
879 #if BITS_PER_LONG == 32
880                         rand = cfs_rand() % (unsigned)total_weight;
881                         /* If total_weight > 32-bit, first generate the high
882                          * 32 bits of the random number, then add in the low
883                          * 32 bits (truncated to the upper limit, if needed) */
884                         if (total_weight > 0xffffffffULL)
885                                 rand = (__u64)(cfs_rand() %
886                                           (unsigned)(total_weight >> 32)) << 32;
887                         else
888                                 rand = 0;
889
890                         if (rand == (total_weight & 0xffffffff00000000ULL))
891                                 rand |= cfs_rand() % (unsigned)total_weight;
892                         else
893                                 rand |= cfs_rand();
894
895 #else
896                         rand = ((__u64)cfs_rand() << 32 | cfs_rand()) %
897                                 total_weight;
898 #endif
899                 } else {
900                         rand = 0;
901                 }
902
903                 /* On average, this will hit larger-weighted osts more often.
904                    0-weight osts will always get used last (only when rand=0).*/
905                 for (i = 0; i < osts->op_count; i++) {
906                         if (!lov->lov_tgts[osts->op_array[i]] ||
907                             !lov->lov_tgts[osts->op_array[i]]->ltd_qos.ltq_usable)
908                                 continue;
909
910                         cur_weight += lov->lov_tgts[osts->op_array[i]]->ltd_qos.ltq_weight;
911 #ifdef QOS_DEBUG
912                         CDEBUG(D_QOS, "stripe_cnt=%d nfound=%d cur_weight="LPU64
913                                       " rand="LPU64" total_weight="LPU64"\n",
914                                *stripe_cnt, nfound, cur_weight, rand, total_weight);
915 #endif
916                         if (cur_weight >= rand) {
917 #ifdef QOS_DEBUG
918                                 CDEBUG(D_QOS, "assigned stripe=%d to idx=%d\n",
919                                        nfound, osts->op_array[i]);
920 #endif
921                                 idx_arr[nfound++] = osts->op_array[i];
922                                 qos_used(lov, osts, osts->op_array[i], &total_weight);
923                                 rc = 0;
924                                 break;
925                         }
926                 }
927                 /* should never satisfy below condition */
928                 if (rc) {
929                         CERROR("Didn't find any OSTs?\n");
930                         break;
931                 }
932         }
933         LASSERT(nfound == *stripe_cnt);
934
935 out:
936         cfs_up_write(&lov->lov_qos.lq_rw_sem);
937
938 out_nolock:
939         if (pool != NULL) {
940                 cfs_up_read(&pool_tgt_rw_sem(pool));
941                 /* put back ref got by lov_find_pool() */
942                 lov_pool_putref(pool);
943         }
944
945         if (rc == -EAGAIN)
946                 rc = alloc_rr(lov, idx_arr, stripe_cnt, poolname, flags);
947
948         obd_putref(exp->exp_obd);
949         RETURN(rc);
950 }
951
952 /* return new alloced stripe count on success */
953 static int alloc_idx_array(struct obd_export *exp, struct lov_stripe_md *lsm,
954                            int newea, int **idx_arr, int *arr_cnt, int flags)
955 {
956         struct lov_obd *lov = &exp->exp_obd->u.lov;
957         int stripe_cnt = lsm->lsm_stripe_count;
958         int i, rc = 0;
959         int *tmp_arr = NULL;
960         ENTRY;
961
962         *arr_cnt = stripe_cnt;
963         OBD_ALLOC(tmp_arr, *arr_cnt * sizeof(int));
964         if (tmp_arr == NULL)
965                 RETURN(-ENOMEM);
966         for (i = 0; i < *arr_cnt; i++)
967                 tmp_arr[i] = -1;
968
969         if (newea ||
970             lsm->lsm_oinfo[0]->loi_ost_idx >= lov->desc.ld_tgt_count)
971                 rc = alloc_qos(exp, tmp_arr, &stripe_cnt,
972                                lsm->lsm_pool_name, flags);
973         else
974                 rc = alloc_specific(lov, lsm, tmp_arr);
975
976         if (rc)
977                 GOTO(out_arr, rc);
978
979         *idx_arr = tmp_arr;
980         RETURN(stripe_cnt);
981 out_arr:
982         OBD_FREE(tmp_arr, *arr_cnt * sizeof(int));
983         *arr_cnt = 0;
984         RETURN(rc);
985 }
986
987 static void free_idx_array(int *idx_arr, int arr_cnt)
988 {
989         if (arr_cnt)
990                 OBD_FREE(idx_arr, arr_cnt * sizeof(int));
991 }
992
993 int qos_prep_create(struct obd_export *exp, struct lov_request_set *set)
994 {
995         struct lov_obd *lov = &exp->exp_obd->u.lov;
996         struct lov_stripe_md *lsm;
997         struct obdo *src_oa = set->set_oi->oi_oa;
998         struct obd_trans_info *oti = set->set_oti;
999         int i, stripes, rc = 0, newea = 0;
1000         int flag = LOV_USES_ASSIGNED_STRIPE;
1001         int *idx_arr = NULL, idx_cnt = 0;
1002         ENTRY;
1003
1004         LASSERT(src_oa->o_valid & OBD_MD_FLID);
1005         LASSERT(src_oa->o_valid & OBD_MD_FLGROUP);
1006
1007         if (set->set_oi->oi_md == NULL) {
1008                 __u16 stripes_def = lov_get_stripecnt(lov, LOV_MAGIC, 0);
1009
1010                 /* If the MDS file was truncated up to some size, stripe over
1011                  * enough OSTs to allow the file to be created at that size.
1012                  * This may mean we use more than the default # of stripes. */
1013                 if (src_oa->o_valid & OBD_MD_FLSIZE) {
1014                         obd_size min_bavail = LUSTRE_STRIPE_MAXBYTES;
1015
1016                         /* Find a small number of stripes we can use
1017                            (up to # of active osts). */
1018                         stripes = 1;
1019                         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
1020                                 if (!lov->lov_tgts[i] ||
1021                                     !lov->lov_tgts[i]->ltd_active)
1022                                         continue;
1023                                 min_bavail = min(min_bavail, TGT_BAVAIL(i));
1024                                 if (min_bavail * stripes > src_oa->o_size)
1025                                         break;
1026                                 stripes++;
1027                         }
1028
1029                         if (stripes < stripes_def)
1030                                 stripes = stripes_def;
1031                 } else {
1032                         flag = LOV_USES_DEFAULT_STRIPE;
1033                         stripes = stripes_def;
1034                 }
1035
1036                 rc = lov_alloc_memmd(&set->set_oi->oi_md, stripes,
1037                                      lov->desc.ld_pattern ?
1038                                      lov->desc.ld_pattern : LOV_PATTERN_RAID0,
1039                                      LOV_MAGIC);
1040                 if (rc < 0)
1041                         GOTO(out_err, rc);
1042                 newea = 1;
1043                 rc = 0;
1044         }
1045
1046         lsm = set->set_oi->oi_md;
1047         lsm->lsm_object_id = src_oa->o_id;
1048         lsm->lsm_object_seq = src_oa->o_seq;
1049         lsm->lsm_layout_gen = 0; /* actual generation set in mdd_lov_create() */
1050
1051         if (!lsm->lsm_stripe_size)
1052                 lsm->lsm_stripe_size = lov->desc.ld_default_stripe_size;
1053         if (!lsm->lsm_pattern) {
1054                 LASSERT(lov->desc.ld_pattern);
1055                 lsm->lsm_pattern = lov->desc.ld_pattern;
1056         }
1057
1058         stripes = alloc_idx_array(exp, lsm, newea, &idx_arr, &idx_cnt, flag);
1059         if (stripes <= 0)
1060                 GOTO(out_err, rc = stripes ? stripes : -EIO);
1061         LASSERTF(stripes <= lsm->lsm_stripe_count,"requested %d allocated %d\n",
1062                  lsm->lsm_stripe_count, stripes);
1063
1064         for (i = 0; i < stripes; i++) {
1065                 struct lov_request *req;
1066                 int ost_idx = idx_arr[i];
1067                 LASSERT(ost_idx >= 0);
1068
1069                 OBD_ALLOC(req, sizeof(*req));
1070                 if (req == NULL)
1071                         GOTO(out_err, rc = -ENOMEM);
1072                 lov_set_add_req(req, set);
1073
1074                 req->rq_buflen = sizeof(*req->rq_oi.oi_md);
1075                 OBD_ALLOC_LARGE(req->rq_oi.oi_md, req->rq_buflen);
1076                 if (req->rq_oi.oi_md == NULL)
1077                         GOTO(out_err, rc = -ENOMEM);
1078
1079                 OBDO_ALLOC(req->rq_oi.oi_oa);
1080                 if (req->rq_oi.oi_oa == NULL)
1081                         GOTO(out_err, rc = -ENOMEM);
1082
1083                 req->rq_idx = ost_idx;
1084                 req->rq_stripe = i;
1085                 /* create data objects with "parent" OA */
1086                 memcpy(req->rq_oi.oi_oa, src_oa, sizeof(*req->rq_oi.oi_oa));
1087                 req->rq_oi.oi_cb_up = cb_create_update;
1088
1089                 /* XXX When we start creating objects on demand, we need to
1090                  *     make sure that we always create the object on the
1091                  *     stripe which holds the existing file size.
1092                  */
1093                 if (src_oa->o_valid & OBD_MD_FLSIZE) {
1094                         req->rq_oi.oi_oa->o_size =
1095                                 lov_size_to_stripe(lsm, src_oa->o_size, i);
1096
1097                         CDEBUG(D_INODE, "stripe %d has size "LPU64"/"LPU64"\n",
1098                                i, req->rq_oi.oi_oa->o_size, src_oa->o_size);
1099                 }
1100         }
1101         LASSERT(set->set_count == stripes);
1102
1103         if (stripes < lsm->lsm_stripe_count)
1104                 qos_shrink_lsm(set);
1105         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_LOV_PREP_CREATE)) {
1106                 qos_shrink_lsm(set);
1107                 rc = -EIO;
1108         }
1109
1110         if (oti && (src_oa->o_valid & OBD_MD_FLCOOKIE)) {
1111                 oti_alloc_cookies(oti, set->set_count);
1112                 if (!oti->oti_logcookies)
1113                         GOTO(out_err, rc = -ENOMEM);
1114                 set->set_cookies = oti->oti_logcookies;
1115         }
1116 out_err:
1117         if (newea && rc)
1118                 obd_free_memmd(exp, &set->set_oi->oi_md);
1119         if (idx_arr)
1120                 free_idx_array(idx_arr, idx_cnt);
1121         EXIT;
1122         return rc;
1123 }
1124
1125 void qos_update(struct lov_obd *lov)
1126 {
1127         ENTRY;
1128         lov->lov_qos.lq_dirty = 1;
1129 }
1130
1131 void qos_statfs_done(struct lov_obd *lov)
1132 {
1133         cfs_down_write(&lov->lov_qos.lq_rw_sem);
1134         if (lov->lov_qos.lq_statfs_in_progress) {
1135                 lov->lov_qos.lq_statfs_in_progress = 0;
1136                 /* wake up any threads waiting for the statfs rpcs to complete*/
1137                 cfs_waitq_signal(&lov->lov_qos.lq_statfs_waitq);
1138         }
1139         cfs_up_write(&lov->lov_qos.lq_rw_sem);
1140 }
1141
1142 static int qos_statfs_ready(struct obd_device *obd, __u64 max_age)
1143 {
1144         struct lov_obd         *lov = &obd->u.lov;
1145         int rc;
1146         ENTRY;
1147         cfs_down_read(&lov->lov_qos.lq_rw_sem);
1148         rc = lov->lov_qos.lq_statfs_in_progress == 0 ||
1149              cfs_time_beforeq_64(max_age, obd->obd_osfs_age);
1150         cfs_up_read(&lov->lov_qos.lq_rw_sem);
1151         RETURN(rc);
1152 }
1153
1154 /*
1155  * Update statfs data if the current osfs age is older than max_age.
1156  * If wait is not set, it means that we are called from lov_create()
1157  * and we should just issue the rpcs without waiting for them to complete.
1158  * If wait is set, we are called from alloc_qos() and we just have
1159  * to wait for the request set to complete.
1160  */
1161 void qos_statfs_update(struct obd_device *obd, __u64 max_age, int wait)
1162 {
1163         struct lov_obd         *lov = &obd->u.lov;
1164         struct obd_info        *oinfo;
1165         int                     rc = 0;
1166         struct ptlrpc_request_set *set = NULL;
1167         ENTRY;
1168
1169         if (cfs_time_beforeq_64(max_age, obd->obd_osfs_age))
1170                 /* statfs data are quite recent, don't need to refresh it */
1171                 RETURN_EXIT;
1172
1173         if (!wait && lov->lov_qos.lq_statfs_in_progress)
1174                 /* statfs already in progress */
1175                 RETURN_EXIT;
1176
1177         cfs_down_write(&lov->lov_qos.lq_rw_sem);
1178         if (lov->lov_qos.lq_statfs_in_progress) {
1179                 cfs_up_write(&lov->lov_qos.lq_rw_sem);
1180                 GOTO(out, rc = 0);
1181         }
1182         /* no statfs in flight, send rpcs */
1183         lov->lov_qos.lq_statfs_in_progress = 1;
1184         cfs_up_write(&lov->lov_qos.lq_rw_sem);
1185
1186         if (wait)
1187                 CDEBUG(D_QOS, "%s: did not manage to get fresh statfs data "
1188                        "in a timely manner (osfs age "LPU64", max age "LPU64")"
1189                        ", sending new statfs rpcs\n",
1190                        obd_uuid2str(&lov->desc.ld_uuid), obd->obd_osfs_age,
1191                        max_age);
1192
1193         /* need to send statfs rpcs */
1194         CDEBUG(D_QOS, "sending new statfs requests\n");
1195         memset(lov->lov_qos.lq_statfs_data, 0,
1196                sizeof(*lov->lov_qos.lq_statfs_data));
1197         oinfo = &lov->lov_qos.lq_statfs_data->lsd_oi;
1198         oinfo->oi_osfs = &lov->lov_qos.lq_statfs_data->lsd_statfs;
1199         oinfo->oi_flags = OBD_STATFS_NODELAY;
1200         set = ptlrpc_prep_set();
1201         if (!set)
1202                 GOTO(out_failed, rc = -ENOMEM);
1203
1204         rc = obd_statfs_async(obd->obd_self_export, oinfo, max_age, set);
1205         if (rc || cfs_list_empty(&set->set_requests)) {
1206                 if (rc)
1207                         CWARN("statfs failed with %d\n", rc);
1208                 GOTO(out_failed, rc);
1209         }
1210         /* send requests via ptlrpcd */
1211         oinfo->oi_flags |= OBD_STATFS_PTLRPCD;
1212         ptlrpcd_add_rqset(set);
1213         GOTO(out, rc);
1214
1215 out_failed:
1216         cfs_down_write(&lov->lov_qos.lq_rw_sem);
1217         lov->lov_qos.lq_statfs_in_progress = 0;
1218         /* wake up any threads waiting for the statfs rpcs to complete */
1219         cfs_waitq_signal(&lov->lov_qos.lq_statfs_waitq);
1220         cfs_up_write(&lov->lov_qos.lq_rw_sem);
1221         wait = 0;
1222 out:
1223         if (set)
1224                 ptlrpc_set_destroy(set);
1225         if (wait) {
1226                 struct l_wait_info lwi = { 0 };
1227                 CDEBUG(D_QOS, "waiting for statfs requests to complete\n");
1228                 l_wait_event(lov->lov_qos.lq_statfs_waitq,
1229                              qos_statfs_ready(obd, max_age), &lwi);
1230                 if (cfs_time_before_64(obd->obd_osfs_age, max_age))
1231                         CDEBUG(D_QOS, "%s: still no fresh statfs data after "
1232                                       "waiting (osfs age "LPU64", max age "
1233                                       LPU64")\n",
1234                                       obd_uuid2str(&lov->desc.ld_uuid),
1235                                       obd->obd_osfs_age, max_age);
1236         }
1237 }