Whamcloud - gitweb
LU-7124 o2iblnd: limit cap.max_send_wr for MLX5
[fs/lustre-release.git] / lustre / lod / lod_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,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU General Public License version 2 for more details.  A copy is
14  * included in the COPYING file that accompanied this code.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright  2009 Sun Microsystems, Inc. All rights reserved
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2012, 2015, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * lustre/lod/lod_qos.c
33  *
34  * Implementation of different allocation algorithm used
35  * to distribute objects and data among OSTs.
36  */
37
38 #define DEBUG_SUBSYSTEM S_LOV
39
40 #include <asm/div64.h>
41 #include <libcfs/libcfs.h>
42 #include <lustre/lustre_idl.h>
43 #include <lustre_swab.h>
44 #include <obd_class.h>
45
46 #include "lod_internal.h"
47
48 /*
49  * force QoS policy (not RR) to be used for testing purposes
50  */
51 #define FORCE_QOS_
52
53 #define D_QOS   D_OTHER
54
55 #define QOS_DEBUG(fmt, ...)     CDEBUG(D_QOS, fmt, ## __VA_ARGS__)
56 #define QOS_CONSOLE(fmt, ...)   LCONSOLE(D_QOS, fmt, ## __VA_ARGS__)
57
58 #define TGT_BAVAIL(i) (OST_TGT(lod,i)->ltd_statfs.os_bavail * \
59                        OST_TGT(lod,i)->ltd_statfs.os_bsize)
60
61 /**
62  * Add a new target to Quality of Service (QoS) target table.
63  *
64  * Add a new OST target to the structure representing an OSS. Resort the list
65  * of known OSSs by the number of OSTs attached to each OSS. The OSS list is
66  * protected internally and no external locking is required.
67  *
68  * \param[in] lod               LOD device
69  * \param[in] ost_desc          OST description
70  *
71  * \retval 0                    on success
72  * \retval -ENOMEM              on error
73  */
74 int qos_add_tgt(struct lod_device *lod, struct lod_tgt_desc *ost_desc)
75 {
76         struct lod_qos_oss *oss = NULL, *temposs;
77         struct obd_export  *exp = ost_desc->ltd_exp;
78         int                 rc = 0, found = 0;
79         struct list_head   *list;
80         ENTRY;
81
82         down_write(&lod->lod_qos.lq_rw_sem);
83         /*
84          * a bit hacky approach to learn NID of corresponding connection
85          * but there is no official API to access information like this
86          * with OSD API.
87          */
88         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
89                 if (obd_uuid_equals(&oss->lqo_uuid,
90                                     &exp->exp_connection->c_remote_uuid)) {
91                         found++;
92                         break;
93                 }
94         }
95
96         if (!found) {
97                 OBD_ALLOC_PTR(oss);
98                 if (!oss)
99                         GOTO(out, rc = -ENOMEM);
100                 memcpy(&oss->lqo_uuid, &exp->exp_connection->c_remote_uuid,
101                        sizeof(oss->lqo_uuid));
102         } else {
103                 /* Assume we have to move this one */
104                 list_del(&oss->lqo_oss_list);
105         }
106
107         oss->lqo_ost_count++;
108         ost_desc->ltd_qos.ltq_oss = oss;
109
110         CDEBUG(D_QOS, "add tgt %s to OSS %s (%d OSTs)\n",
111                obd_uuid2str(&ost_desc->ltd_uuid), obd_uuid2str(&oss->lqo_uuid),
112                oss->lqo_ost_count);
113
114         /* Add sorted by # of OSTs.  Find the first entry that we're
115            bigger than... */
116         list = &lod->lod_qos.lq_oss_list;
117         list_for_each_entry(temposs, list, lqo_oss_list) {
118                 if (oss->lqo_ost_count > temposs->lqo_ost_count)
119                         break;
120         }
121         /* ...and add before it.  If we're the first or smallest, temposs
122            points to the list head, and we add to the end. */
123         list_add_tail(&oss->lqo_oss_list, &temposs->lqo_oss_list);
124
125         lod->lod_qos.lq_dirty = 1;
126         lod->lod_qos.lq_rr.lqr_dirty = 1;
127
128 out:
129         up_write(&lod->lod_qos.lq_rw_sem);
130         RETURN(rc);
131 }
132
133 /**
134  * Remove OST target from QoS table.
135  *
136  * Removes given OST target from QoS table and releases related OSS structure
137  * if no OSTs remain on the OSS.
138  *
139  * \param[in] lod               LOD device
140  * \param[in] ost_desc          OST description
141  *
142  * \retval 0                    on success
143  * \retval -ENOENT              if no OSS was found
144  */
145 int qos_del_tgt(struct lod_device *lod, struct lod_tgt_desc *ost_desc)
146 {
147         struct lod_qos_oss *oss;
148         int                 rc = 0;
149         ENTRY;
150
151         down_write(&lod->lod_qos.lq_rw_sem);
152         oss = ost_desc->ltd_qos.ltq_oss;
153         if (!oss)
154                 GOTO(out, rc = -ENOENT);
155
156         oss->lqo_ost_count--;
157         if (oss->lqo_ost_count == 0) {
158                 CDEBUG(D_QOS, "removing OSS %s\n",
159                        obd_uuid2str(&oss->lqo_uuid));
160                 list_del(&oss->lqo_oss_list);
161                 ost_desc->ltd_qos.ltq_oss = NULL;
162                 OBD_FREE_PTR(oss);
163         }
164
165         lod->lod_qos.lq_dirty = 1;
166         lod->lod_qos.lq_rr.lqr_dirty = 1;
167 out:
168         up_write(&lod->lod_qos.lq_rw_sem);
169         RETURN(rc);
170 }
171
172 /**
173  * Check whether the target is available for new OST objects.
174  *
175  * Request statfs data from the given target and verify it's active and not
176  * read-only. If so, then it can be used to place new OST objects. This
177  * function also maintains the number of active/inactive targets and sets
178  * dirty flags if those numbers change so others can run re-balance procedures.
179  * No external locking is required.
180  *
181  * \param[in] env       execution environment for this thread
182  * \param[in] d         LOD device
183  * \param[in] index     index of OST target to check
184  * \param[out] sfs      buffer for statfs data
185  *
186  * \retval 0            if the target is good
187  * \retval negative     negated errno on error
188
189  */
190 static int lod_statfs_and_check(const struct lu_env *env, struct lod_device *d,
191                                 int index, struct obd_statfs *sfs)
192 {
193         struct lod_tgt_desc *ost;
194         int                  rc;
195         ENTRY;
196
197         LASSERT(d);
198         ost = OST_TGT(d,index);
199         LASSERT(ost);
200
201         rc = dt_statfs(env, ost->ltd_ost, sfs);
202         if (rc && rc != -ENOTCONN)
203                 CERROR("%s: statfs: rc = %d\n", lod2obd(d)->obd_name, rc);
204
205         /* If the OST is readonly then we can't allocate objects there */
206         if (sfs->os_state & OS_STATE_READONLY)
207                 rc = -EROFS;
208
209         /* check whether device has changed state (active, inactive) */
210         if (rc != 0 && ost->ltd_active) {
211                 /* turned inactive? */
212                 spin_lock(&d->lod_desc_lock);
213                 if (ost->ltd_active) {
214                         ost->ltd_active = 0;
215                         if (rc == -ENOTCONN)
216                                 ost->ltd_connecting = 1;
217
218                         LASSERT(d->lod_desc.ld_active_tgt_count > 0);
219                         d->lod_desc.ld_active_tgt_count--;
220                         d->lod_qos.lq_dirty = 1;
221                         d->lod_qos.lq_rr.lqr_dirty = 1;
222                         CDEBUG(D_CONFIG, "%s: turns inactive\n",
223                                ost->ltd_exp->exp_obd->obd_name);
224                 }
225                 spin_unlock(&d->lod_desc_lock);
226         } else if (rc == 0 && ost->ltd_active == 0) {
227                 /* turned active? */
228                 LASSERTF(d->lod_desc.ld_active_tgt_count < d->lod_ostnr,
229                          "active tgt count %d, ost nr %d\n",
230                          d->lod_desc.ld_active_tgt_count, d->lod_ostnr);
231                 spin_lock(&d->lod_desc_lock);
232                 if (ost->ltd_active == 0) {
233                         ost->ltd_active = 1;
234                         ost->ltd_connecting = 0;
235                         d->lod_desc.ld_active_tgt_count++;
236                         d->lod_qos.lq_dirty = 1;
237                         d->lod_qos.lq_rr.lqr_dirty = 1;
238                         CDEBUG(D_CONFIG, "%s: turns active\n",
239                                ost->ltd_exp->exp_obd->obd_name);
240                 }
241                 spin_unlock(&d->lod_desc_lock);
242         }
243
244         RETURN(rc);
245 }
246
247 /**
248  * Maintain per-target statfs data.
249  *
250  * The function refreshes statfs data for all the targets every N seconds.
251  * The actual N is controlled via procfs and set to LOV_DESC_QOS_MAXAGE_DEFAULT
252  * initially.
253  *
254  * \param[in] env       execution environment for this thread
255  * \param[in] lod       LOD device
256  */
257 static void lod_qos_statfs_update(const struct lu_env *env,
258                                   struct lod_device *lod)
259 {
260         struct obd_device *obd = lod2obd(lod);
261         struct ost_pool   *osts = &(lod->lod_pool_info);
262         unsigned int       i;
263         int                idx;
264         __u64              max_age, avail;
265         ENTRY;
266
267         max_age = cfs_time_shift_64(-2 * lod->lod_desc.ld_qos_maxage);
268
269         if (cfs_time_beforeq_64(max_age, obd->obd_osfs_age))
270                 /* statfs data are quite recent, don't need to refresh it */
271                 RETURN_EXIT;
272
273         down_write(&lod->lod_qos.lq_rw_sem);
274         if (cfs_time_beforeq_64(max_age, obd->obd_osfs_age))
275                 goto out;
276
277         for (i = 0; i < osts->op_count; i++) {
278                 idx = osts->op_array[i];
279                 avail = OST_TGT(lod,idx)->ltd_statfs.os_bavail;
280                 if (lod_statfs_and_check(env, lod, idx,
281                                          &OST_TGT(lod, idx)->ltd_statfs))
282                         continue;
283                 if (OST_TGT(lod,idx)->ltd_statfs.os_bavail != avail)
284                         /* recalculate weigths */
285                         lod->lod_qos.lq_dirty = 1;
286         }
287         obd->obd_osfs_age = cfs_time_current_64();
288
289 out:
290         up_write(&lod->lod_qos.lq_rw_sem);
291         EXIT;
292 }
293
294 /**
295  * Calculate per-OST and per-OSS penalties
296  *
297  * Re-calculate penalties when the configuration changes, active targets
298  * change and after statfs refresh (all these are reflected by lq_dirty flag).
299  * On every OST and OSS: decay the penalty by half for every 8x the update
300  * interval that the device has been idle. That gives lots of time for the
301  * statfs information to be updated (which the penalty is only a proxy for),
302  * and avoids penalizing OSS/OSTs under light load.
303  * See lod_qos_calc_weight() for how penalties are factored into the weight.
304  *
305  * \param[in] lod       LOD device
306  *
307  * \retval 0            on success
308  * \retval -EAGAIN      the number of OSTs isn't enough
309  */
310 static int lod_qos_calc_ppo(struct lod_device *lod)
311 {
312         struct lod_qos_oss *oss;
313         __u64               ba_max, ba_min, temp;
314         __u32               num_active;
315         unsigned int        i;
316         int                 rc, prio_wide;
317         time_t              now, age;
318         ENTRY;
319
320         if (!lod->lod_qos.lq_dirty)
321                 GOTO(out, rc = 0);
322
323         num_active = lod->lod_desc.ld_active_tgt_count - 1;
324         if (num_active < 1)
325                 GOTO(out, rc = -EAGAIN);
326
327         /* find bavail on each OSS */
328         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list)
329                             oss->lqo_bavail = 0;
330         lod->lod_qos.lq_active_oss_count = 0;
331
332         /*
333          * How badly user wants to select OSTs "widely" (not recently chosen
334          * and not on recent OSS's).  As opposed to "freely" (free space
335          * avail.) 0-256
336          */
337         prio_wide = 256 - lod->lod_qos.lq_prio_free;
338
339         ba_min = (__u64)(-1);
340         ba_max = 0;
341         now = cfs_time_current_sec();
342         /* Calculate OST penalty per object
343          * (lod ref taken in lod_qos_prep_create()) */
344         cfs_foreach_bit(lod->lod_ost_bitmap, i) {
345                 LASSERT(OST_TGT(lod,i));
346                 temp = TGT_BAVAIL(i);
347                 if (!temp)
348                         continue;
349                 ba_min = min(temp, ba_min);
350                 ba_max = max(temp, ba_max);
351
352                 /* Count the number of usable OSS's */
353                 if (OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_bavail == 0)
354                         lod->lod_qos.lq_active_oss_count++;
355                 OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_bavail += temp;
356
357                 /* per-OST penalty is prio * TGT_bavail / (num_ost - 1) / 2 */
358                 temp >>= 1;
359                 do_div(temp, num_active);
360                 OST_TGT(lod,i)->ltd_qos.ltq_penalty_per_obj =
361                         (temp * prio_wide) >> 8;
362
363                 age = (now - OST_TGT(lod,i)->ltd_qos.ltq_used) >> 3;
364                 if (lod->lod_qos.lq_reset ||
365                     age > 32 * lod->lod_desc.ld_qos_maxage)
366                         OST_TGT(lod,i)->ltd_qos.ltq_penalty = 0;
367                 else if (age > lod->lod_desc.ld_qos_maxage)
368                         /* Decay OST penalty. */
369                         OST_TGT(lod,i)->ltd_qos.ltq_penalty >>=
370                                 (age / lod->lod_desc.ld_qos_maxage);
371         }
372
373         num_active = lod->lod_qos.lq_active_oss_count - 1;
374         if (num_active < 1) {
375                 /* If there's only 1 OSS, we can't penalize it, so instead
376                    we have to double the OST penalty */
377                 num_active = 1;
378                 cfs_foreach_bit(lod->lod_ost_bitmap, i)
379                         OST_TGT(lod,i)->ltd_qos.ltq_penalty_per_obj <<= 1;
380         }
381
382         /* Per-OSS penalty is prio * oss_avail / oss_osts / (num_oss - 1) / 2 */
383         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
384                 temp = oss->lqo_bavail >> 1;
385                 do_div(temp, oss->lqo_ost_count * num_active);
386                 oss->lqo_penalty_per_obj = (temp * prio_wide) >> 8;
387
388                 age = (now - oss->lqo_used) >> 3;
389                 if (lod->lod_qos.lq_reset ||
390                     age > 32 * lod->lod_desc.ld_qos_maxage)
391                         oss->lqo_penalty = 0;
392                 else if (age > lod->lod_desc.ld_qos_maxage)
393                         /* Decay OSS penalty. */
394                         oss->lqo_penalty >>= age / lod->lod_desc.ld_qos_maxage;
395         }
396
397         lod->lod_qos.lq_dirty = 0;
398         lod->lod_qos.lq_reset = 0;
399
400         /* If each ost has almost same free space,
401          * do rr allocation for better creation performance */
402         lod->lod_qos.lq_same_space = 0;
403         if ((ba_max * (256 - lod->lod_qos.lq_threshold_rr)) >> 8 < ba_min) {
404                 lod->lod_qos.lq_same_space = 1;
405                 /* Reset weights for the next time we enter qos mode */
406                 lod->lod_qos.lq_reset = 1;
407         }
408         rc = 0;
409
410 out:
411 #ifndef FORCE_QOS
412         if (!rc && lod->lod_qos.lq_same_space)
413                 RETURN(-EAGAIN);
414 #endif
415         RETURN(rc);
416 }
417
418 /**
419  * Calculate weight for a given OST target.
420  *
421  * The final OST weight is the number of bytes available minus the OST and
422  * OSS penalties.  See lod_qos_calc_ppo() for how penalties are calculated.
423  *
424  * \param[in] lod       LOD device, where OST targets are listed
425  * \param[in] i         OST target index
426  *
427  * \retval              0
428  */
429 static int lod_qos_calc_weight(struct lod_device *lod, int i)
430 {
431         __u64 temp, temp2;
432
433         temp = TGT_BAVAIL(i);
434         temp2 = OST_TGT(lod,i)->ltd_qos.ltq_penalty +
435                 OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_penalty;
436         if (temp < temp2)
437                 OST_TGT(lod,i)->ltd_qos.ltq_weight = 0;
438         else
439                 OST_TGT(lod,i)->ltd_qos.ltq_weight = temp - temp2;
440         return 0;
441 }
442
443 /**
444  * Re-calculate weights.
445  *
446  * The function is called when some OST target was used for a new object. In
447  * this case we should re-calculate all the weights to keep new allocations
448  * balanced well.
449  *
450  * \param[in] lod       LOD device
451  * \param[in] osts      OST pool where a new object was placed
452  * \param[in] index     OST target where a new object was placed
453  * \param[out] total_wt new total weight for the pool
454  *
455  * \retval              0
456  */
457 static int lod_qos_used(struct lod_device *lod, struct ost_pool *osts,
458                         __u32 index, __u64 *total_wt)
459 {
460         struct lod_tgt_desc *ost;
461         struct lod_qos_oss  *oss;
462         unsigned int j;
463         ENTRY;
464
465         ost = OST_TGT(lod,index);
466         LASSERT(ost);
467
468         /* Don't allocate on this devuce anymore, until the next alloc_qos */
469         ost->ltd_qos.ltq_usable = 0;
470
471         oss = ost->ltd_qos.ltq_oss;
472
473         /* Decay old penalty by half (we're adding max penalty, and don't
474            want it to run away.) */
475         ost->ltd_qos.ltq_penalty >>= 1;
476         oss->lqo_penalty >>= 1;
477
478         /* mark the OSS and OST as recently used */
479         ost->ltd_qos.ltq_used = oss->lqo_used = cfs_time_current_sec();
480
481         /* Set max penalties for this OST and OSS */
482         ost->ltd_qos.ltq_penalty +=
483                 ost->ltd_qos.ltq_penalty_per_obj * lod->lod_ostnr;
484         oss->lqo_penalty += oss->lqo_penalty_per_obj *
485                 lod->lod_qos.lq_active_oss_count;
486
487         /* Decrease all OSS penalties */
488         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
489                 if (oss->lqo_penalty < oss->lqo_penalty_per_obj)
490                         oss->lqo_penalty = 0;
491                 else
492                         oss->lqo_penalty -= oss->lqo_penalty_per_obj;
493         }
494
495         *total_wt = 0;
496         /* Decrease all OST penalties */
497         for (j = 0; j < osts->op_count; j++) {
498                 int i;
499
500                 i = osts->op_array[j];
501                 if (!cfs_bitmap_check(lod->lod_ost_bitmap, i))
502                         continue;
503
504                 ost = OST_TGT(lod,i);
505                 LASSERT(ost);
506
507                 if (ost->ltd_qos.ltq_penalty <
508                                 ost->ltd_qos.ltq_penalty_per_obj)
509                         ost->ltd_qos.ltq_penalty = 0;
510                 else
511                         ost->ltd_qos.ltq_penalty -=
512                                 ost->ltd_qos.ltq_penalty_per_obj;
513
514                 lod_qos_calc_weight(lod, i);
515
516                 /* Recalc the total weight of usable osts */
517                 if (ost->ltd_qos.ltq_usable)
518                         *total_wt += ost->ltd_qos.ltq_weight;
519
520                 QOS_DEBUG("recalc tgt %d usable=%d avail="LPU64
521                           " ostppo="LPU64" ostp="LPU64" ossppo="LPU64
522                           " ossp="LPU64" wt="LPU64"\n",
523                           i, ost->ltd_qos.ltq_usable, TGT_BAVAIL(i) >> 10,
524                           ost->ltd_qos.ltq_penalty_per_obj >> 10,
525                           ost->ltd_qos.ltq_penalty >> 10,
526                           ost->ltd_qos.ltq_oss->lqo_penalty_per_obj >> 10,
527                           ost->ltd_qos.ltq_oss->lqo_penalty >> 10,
528                           ost->ltd_qos.ltq_weight >> 10);
529         }
530
531         RETURN(0);
532 }
533
534 void lod_qos_rr_init(struct lod_qos_rr *lqr)
535 {
536         spin_lock_init(&lqr->lqr_alloc);
537         lqr->lqr_dirty = 1;
538 }
539
540
541 #define LOV_QOS_EMPTY ((__u32)-1)
542
543 /**
544  * Calculate optimal round-robin order with regard to OSSes.
545  *
546  * Place all the OSTs from pool \a src_pool in a special array to be used for
547  * round-robin (RR) stripe allocation.  The placement algorithm interleaves
548  * OSTs from the different OSSs so that RR allocation can balance OSSs evenly.
549  * Resorts the targets when the number of active targets changes (because of
550  * a new target or activation/deactivation).
551  *
552  * \param[in] lod       LOD device
553  * \param[in] src_pool  OST pool
554  * \param[in] lqr       round-robin list
555  *
556  * \retval 0            on success
557  * \retval -ENOMEM      fails to allocate the array
558  */
559 static int lod_qos_calc_rr(struct lod_device *lod, struct ost_pool *src_pool,
560                            struct lod_qos_rr *lqr)
561 {
562         struct lod_qos_oss  *oss;
563         struct lod_tgt_desc *ost;
564         unsigned placed, real_count;
565         unsigned int i;
566         int rc;
567         ENTRY;
568
569         if (!lqr->lqr_dirty) {
570                 LASSERT(lqr->lqr_pool.op_size);
571                 RETURN(0);
572         }
573
574         /* Do actual allocation. */
575         down_write(&lod->lod_qos.lq_rw_sem);
576
577         /*
578          * Check again. While we were sleeping on @lq_rw_sem something could
579          * change.
580          */
581         if (!lqr->lqr_dirty) {
582                 LASSERT(lqr->lqr_pool.op_size);
583                 up_write(&lod->lod_qos.lq_rw_sem);
584                 RETURN(0);
585         }
586
587         real_count = src_pool->op_count;
588
589         /* Zero the pool array */
590         /* alloc_rr is holding a read lock on the pool, so nobody is adding/
591            deleting from the pool. The lq_rw_sem insures that nobody else
592            is reading. */
593         lqr->lqr_pool.op_count = real_count;
594         rc = lod_ost_pool_extend(&lqr->lqr_pool, real_count);
595         if (rc) {
596                 up_write(&lod->lod_qos.lq_rw_sem);
597                 RETURN(rc);
598         }
599         for (i = 0; i < lqr->lqr_pool.op_count; i++)
600                 lqr->lqr_pool.op_array[i] = LOV_QOS_EMPTY;
601
602         /* Place all the OSTs from 1 OSS at the same time. */
603         placed = 0;
604         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
605                 int j = 0;
606
607                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
608                         int next;
609
610                         if (!cfs_bitmap_check(lod->lod_ost_bitmap,
611                                                 src_pool->op_array[i]))
612                                 continue;
613
614                         ost = OST_TGT(lod,src_pool->op_array[i]);
615                         LASSERT(ost && ost->ltd_ost);
616                         if (ost->ltd_qos.ltq_oss != oss)
617                                 continue;
618
619                         /* Evenly space these OSTs across arrayspace */
620                         next = j * lqr->lqr_pool.op_count / oss->lqo_ost_count;
621                         while (lqr->lqr_pool.op_array[next] != LOV_QOS_EMPTY)
622                                 next = (next + 1) % lqr->lqr_pool.op_count;
623
624                         lqr->lqr_pool.op_array[next] = src_pool->op_array[i];
625                         j++;
626                         placed++;
627                 }
628         }
629
630         lqr->lqr_dirty = 0;
631         up_write(&lod->lod_qos.lq_rw_sem);
632
633         if (placed != real_count) {
634                 /* This should never happen */
635                 LCONSOLE_ERROR_MSG(0x14e, "Failed to place all OSTs in the "
636                                    "round-robin list (%d of %d).\n",
637                                    placed, real_count);
638                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
639                         LCONSOLE(D_WARNING, "rr #%d ost idx=%d\n", i,
640                                  lqr->lqr_pool.op_array[i]);
641                 }
642                 lqr->lqr_dirty = 1;
643                 RETURN(-EAGAIN);
644         }
645
646 #if 0
647         for (i = 0; i < lqr->lqr_pool.op_count; i++)
648                 QOS_CONSOLE("rr #%d ost idx=%d\n", i, lqr->lqr_pool.op_array[i]);
649 #endif
650
651         RETURN(0);
652 }
653
654 /**
655  * Instantiate and declare creation of a new object.
656  *
657  * The function instantiates LU representation for a new object on the
658  * specified device. Also it declares an intention to create that
659  * object on the storage target.
660  *
661  * Note lu_object_anon() is used which is a trick with regard to LU/OSD
662  * infrastructure - in the existing precreation framework we can't assign FID
663  * at this moment, we do this later once a transaction is started. So the
664  * special method instantiates FID-less object in the cache and later it
665  * will get a FID and proper placement in LU cache.
666  *
667  * \param[in] env       execution environment for this thread
668  * \param[in] d         LOD device
669  * \param[in] ost_idx   OST target index where the object is being created
670  * \param[in] th        transaction handle
671  *
672  * \retval              object ptr on success, ERR_PTR() otherwise
673  */
674 static struct dt_object *lod_qos_declare_object_on(const struct lu_env *env,
675                                                    struct lod_device *d,
676                                                    __u32 ost_idx,
677                                                    struct thandle *th)
678 {
679         struct lod_tgt_desc *ost;
680         struct lu_object *o, *n;
681         struct lu_device *nd;
682         struct dt_object *dt;
683         int               rc;
684         ENTRY;
685
686         LASSERT(d);
687         LASSERT(ost_idx < d->lod_osts_size);
688         ost = OST_TGT(d,ost_idx);
689         LASSERT(ost);
690         LASSERT(ost->ltd_ost);
691
692         nd = &ost->ltd_ost->dd_lu_dev;
693
694         /*
695          * allocate anonymous object with zero fid, real fid
696          * will be assigned by OSP within transaction
697          * XXX: to be fixed with fully-functional OST fids
698          */
699         o = lu_object_anon(env, nd, NULL);
700         if (IS_ERR(o))
701                 GOTO(out, dt = ERR_PTR(PTR_ERR(o)));
702
703         n = lu_object_locate(o->lo_header, nd->ld_type);
704         if (unlikely(n == NULL)) {
705                 CERROR("can't find slice\n");
706                 lu_object_put(env, o);
707                 GOTO(out, dt = ERR_PTR(-EINVAL));
708         }
709
710         dt = container_of(n, struct dt_object, do_lu);
711
712         rc = lod_sub_object_declare_create(env, dt, NULL, NULL, NULL, th);
713         if (rc < 0) {
714                 CDEBUG(D_OTHER, "can't declare creation on #%u: %d\n",
715                        ost_idx, rc);
716                 lu_object_put(env, o);
717                 dt = ERR_PTR(rc);
718         }
719
720 out:
721         RETURN(dt);
722 }
723
724 /**
725  * Calculate a minimum acceptable stripe count.
726  *
727  * Return an acceptable stripe count depending on flag LOV_USES_DEFAULT_STRIPE:
728  * all stripes or 3/4 of stripes.
729  *
730  * \param[in] stripe_cnt        number of stripes requested
731  * \param[in] flags             0 or LOV_USES_DEFAULT_STRIPE
732  *
733  * \retval                      acceptable stripecount
734  */
735 static int min_stripe_count(__u32 stripe_cnt, int flags)
736 {
737         return (flags & LOV_USES_DEFAULT_STRIPE ?
738                         stripe_cnt - (stripe_cnt / 4) : stripe_cnt);
739 }
740
741 #define LOV_CREATE_RESEED_MULT 30
742 #define LOV_CREATE_RESEED_MIN  2000
743
744 /**
745  * Check if an OST is full.
746  *
747  * Check whether an OST should be considered full based
748  * on the given statfs data.
749  *
750  * \param[in] msfs      statfs data
751  *
752  * \retval false        not full
753  * \retval true         full
754  */
755 static int inline lod_qos_dev_is_full(struct obd_statfs *msfs)
756 {
757         __u64 used;
758         int   bs = msfs->os_bsize;
759
760         LASSERT(((bs - 1) & bs) == 0);
761
762         /* the minimum of 0.1% used blocks and 1GB bytes. */
763         used = min_t(__u64, (msfs->os_blocks - msfs->os_bfree) >> 10,
764                         1 << (31 - ffs(bs)));
765         return (msfs->os_bavail < used);
766 }
767
768 /**
769  * Initialize temporary OST-in-use array.
770  *
771  * Allocate or extend the array used to mark targets already assigned to a new
772  * striping so they are not used more than once.
773  *
774  * \param[in] env       execution environment for this thread
775  * \param[in] stripes   number of items needed in the array
776  *
777  * \retval 0            on success
778  * \retval -ENOMEM      on error
779  */
780 static inline int lod_qos_ost_in_use_clear(const struct lu_env *env,
781                                            __u32 stripes)
782 {
783         struct lod_thread_info *info = lod_env_info(env);
784
785         if (info->lti_ea_store_size < sizeof(int) * stripes)
786                 lod_ea_store_resize(info, stripes * sizeof(int));
787         if (info->lti_ea_store_size < sizeof(int) * stripes) {
788                 CERROR("can't allocate memory for ost-in-use array\n");
789                 return -ENOMEM;
790         }
791         memset(info->lti_ea_store, -1, sizeof(int) * stripes);
792         return 0;
793 }
794
795 /**
796  * Remember a target in the array of used targets.
797  *
798  * Mark the given target as used for a new striping being created. The status
799  * of an OST in a striping can be checked with lod_qos_is_ost_used().
800  *
801  * \param[in] env       execution environment for this thread
802  * \param[in] idx       index in the array
803  * \param[in] ost       OST target index to mark as used
804  */
805 static inline void lod_qos_ost_in_use(const struct lu_env *env,
806                                       int idx, int ost)
807 {
808         struct lod_thread_info *info = lod_env_info(env);
809         int *osts = info->lti_ea_store;
810
811         LASSERT(info->lti_ea_store_size >= idx * sizeof(int));
812         osts[idx] = ost;
813 }
814
815 /**
816  * Check is OST used in a striping.
817  *
818  * Checks whether OST with the given index is marked as used in the temporary
819  * array (see lod_qos_ost_in_use()).
820  *
821  * \param[in] env       execution environment for this thread
822  * \param[in] ost       OST target index to check
823  * \param[in] stripes   the number of items used in the array already
824  *
825  * \retval 0            not used
826  * \retval 1            used
827  */
828 static int lod_qos_is_ost_used(const struct lu_env *env, int ost, __u32 stripes)
829 {
830         struct lod_thread_info *info = lod_env_info(env);
831         int *osts = info->lti_ea_store;
832         __u32 j;
833
834         for (j = 0; j < stripes; j++) {
835                 if (osts[j] == ost)
836                         return 1;
837         }
838         return 0;
839 }
840
841 static int lod_check_and_reserve_ost(const struct lu_env *env,
842                                      struct lod_device *m,
843                                      struct obd_statfs *sfs, __u32 ost_idx,
844                                      __u32 speed, __u32 *s_idx,
845                                      struct dt_object **stripe,
846                                      struct thandle *th)
847 {
848         struct dt_object   *o;
849         __u32 stripe_idx = *s_idx;
850         int rc;
851
852         rc = lod_statfs_and_check(env, m, ost_idx, sfs);
853         if (rc) {
854                 /* this OSP doesn't feel well */
855                 goto out_return;
856         }
857
858         /*
859          * skip full devices
860          */
861         if (lod_qos_dev_is_full(sfs)) {
862                 QOS_DEBUG("#%d is full\n", ost_idx);
863                 goto out_return;
864         }
865
866         /*
867          * We expect number of precreated objects in f_ffree at
868          * the first iteration, skip OSPs with no objects ready
869          */
870         if (sfs->os_fprecreated == 0 && speed == 0) {
871                 QOS_DEBUG("#%d: precreation is empty\n", ost_idx);
872                 goto out_return;
873         }
874
875         /*
876          * try to use another OSP if this one is degraded
877          */
878         if (sfs->os_state & OS_STATE_DEGRADED && speed < 2) {
879                 QOS_DEBUG("#%d: degraded\n", ost_idx);
880                 goto out_return;
881         }
882
883         /*
884          * do not put >1 objects on a single OST
885          */
886         if (speed && lod_qos_is_ost_used(env, ost_idx, stripe_idx))
887                 goto out_return;
888
889         o = lod_qos_declare_object_on(env, m, ost_idx, th);
890         if (IS_ERR(o)) {
891                 CDEBUG(D_OTHER, "can't declare new object on #%u: %d\n",
892                        ost_idx, (int) PTR_ERR(o));
893                 rc = PTR_ERR(o);
894                 goto out_return;
895         }
896
897         /*
898          * We've successfully declared (reserved) an object
899          */
900         lod_qos_ost_in_use(env, stripe_idx, ost_idx);
901         stripe[stripe_idx] = o;
902         stripe_idx++;
903         *s_idx = stripe_idx;
904
905 out_return:
906         return rc;
907 }
908
909 /**
910  * Allocate a striping using round-robin algorithm.
911  *
912  * Allocates a new striping using round-robin algorithm. The function refreshes
913  * all the internal structures (statfs cache, array of available OSTs sorted
914  * with regard to OSS, etc). The number of stripes required is taken from the
915  * object (must be prepared by the caller), but can change if the flag
916  * LOV_USES_DEFAULT_STRIPE is supplied. The caller should ensure nobody else
917  * is trying to create a striping on the object in parallel. All the internal
918  * structures (like pools, etc) are protected and no additional locking is
919  * required. The function succeeds even if a single stripe is allocated. To save
920  * time we give priority to targets which already have objects precreated.
921  * Full OSTs are skipped (see lod_qos_dev_is_full() for the details).
922  *
923  * \param[in] env       execution environment for this thread
924  * \param[in] lo        LOD object
925  * \param[out] stripe   striping created
926  * \param[in] flags     allocation flags (0 or LOV_USES_DEFAULT_STRIPE)
927  * \param[in] th        transaction handle
928  *
929  * \retval 0            on success
930  * \retval -ENOSPC      if not enough OSTs are found
931  * \retval negative     negated errno for other failures
932  */
933 static int lod_alloc_rr(const struct lu_env *env, struct lod_object *lo,
934                         struct dt_object **stripe, int flags,
935                         struct thandle *th)
936 {
937         struct lod_device *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
938         struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
939         struct pool_desc  *pool = NULL;
940         struct ost_pool   *osts;
941         struct lod_qos_rr *lqr;
942         unsigned int       i, array_idx;
943         int                rc;
944         __u32              ost_start_idx_temp;
945         int                speed = 0;
946         int                ost_connecting = 0;
947         __u32              stripe_idx = 0;
948         __u32              stripe_cnt = lo->ldo_stripenr;
949         __u32              stripe_cnt_min = min_stripe_count(stripe_cnt, flags);
950         __u32              ost_idx;
951         ENTRY;
952
953         if (lo->ldo_pool)
954                 pool = lod_find_pool(m, lo->ldo_pool);
955
956         if (pool != NULL) {
957                 down_read(&pool_tgt_rw_sem(pool));
958                 osts = &(pool->pool_obds);
959                 lqr = &(pool->pool_rr);
960         } else {
961                 osts = &(m->lod_pool_info);
962                 lqr = &(m->lod_qos.lq_rr);
963         }
964
965         rc = lod_qos_calc_rr(m, osts, lqr);
966         if (rc)
967                 GOTO(out, rc);
968
969         rc = lod_qos_ost_in_use_clear(env, lo->ldo_stripenr);
970         if (rc)
971                 GOTO(out, rc);
972
973         down_read(&m->lod_qos.lq_rw_sem);
974         spin_lock(&lqr->lqr_alloc);
975         if (--lqr->lqr_start_count <= 0) {
976                 lqr->lqr_start_idx = cfs_rand() % osts->op_count;
977                 lqr->lqr_start_count =
978                         (LOV_CREATE_RESEED_MIN / max(osts->op_count, 1U) +
979                          LOV_CREATE_RESEED_MULT) * max(osts->op_count, 1U);
980         } else if (stripe_cnt_min >= osts->op_count ||
981                         lqr->lqr_start_idx > osts->op_count) {
982                 /* If we have allocated from all of the OSTs, slowly
983                  * precess the next start if the OST/stripe count isn't
984                  * already doing this for us. */
985                 lqr->lqr_start_idx %= osts->op_count;
986                 if (stripe_cnt > 1 && (osts->op_count % stripe_cnt) != 1)
987                         ++lqr->lqr_offset_idx;
988         }
989         ost_start_idx_temp = lqr->lqr_start_idx;
990
991 repeat_find:
992
993         QOS_DEBUG("pool '%s' want %d startidx %d startcnt %d offset %d "
994                   "active %d count %d\n",
995                   lo->ldo_pool ? lo->ldo_pool : "",
996                   stripe_cnt, lqr->lqr_start_idx, lqr->lqr_start_count,
997                   lqr->lqr_offset_idx, osts->op_count, osts->op_count);
998
999         for (i = 0; i < osts->op_count && stripe_idx < lo->ldo_stripenr; i++) {
1000                 array_idx = (lqr->lqr_start_idx + lqr->lqr_offset_idx) %
1001                                 osts->op_count;
1002                 ++lqr->lqr_start_idx;
1003                 ost_idx = lqr->lqr_pool.op_array[array_idx];
1004
1005                 QOS_DEBUG("#%d strt %d act %d strp %d ary %d idx %d\n",
1006                           i, lqr->lqr_start_idx, /* XXX: active*/ 0,
1007                           stripe_idx, array_idx, ost_idx);
1008
1009                 if ((ost_idx == LOV_QOS_EMPTY) ||
1010                     !cfs_bitmap_check(m->lod_ost_bitmap, ost_idx))
1011                         continue;
1012
1013                 /* Fail Check before osc_precreate() is called
1014                    so we can only 'fail' single OSC. */
1015                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
1016                         continue;
1017
1018                 spin_unlock(&lqr->lqr_alloc);
1019                 rc = lod_check_and_reserve_ost(env, m, sfs, ost_idx, speed,
1020                                                &stripe_idx, stripe, th);
1021                 spin_lock(&lqr->lqr_alloc);
1022
1023                 if (rc != 0 && OST_TGT(m, ost_idx)->ltd_connecting)
1024                         ost_connecting = 1;
1025         }
1026         if ((speed < 2) && (stripe_idx < stripe_cnt_min)) {
1027                 /* Try again, allowing slower OSCs */
1028                 speed++;
1029                 lqr->lqr_start_idx = ost_start_idx_temp;
1030
1031                 ost_connecting = 0;
1032                 goto repeat_find;
1033         }
1034
1035         spin_unlock(&lqr->lqr_alloc);
1036         up_read(&m->lod_qos.lq_rw_sem);
1037
1038         if (stripe_idx) {
1039                 lo->ldo_stripenr = stripe_idx;
1040                 /* at least one stripe is allocated */
1041                 rc = 0;
1042         } else {
1043                 /* nobody provided us with a single object */
1044                 if (ost_connecting)
1045                         rc = -EINPROGRESS;
1046                 else
1047                         rc = -ENOSPC;
1048         }
1049
1050 out:
1051         if (pool != NULL) {
1052                 up_read(&pool_tgt_rw_sem(pool));
1053                 /* put back ref got by lod_find_pool() */
1054                 lod_pool_putref(pool);
1055         }
1056
1057         RETURN(rc);
1058 }
1059
1060 /**
1061  * Allocate a specific striping layout on a user defined set of OSTs.
1062  *
1063  * Allocates new striping using the OST index range provided by the data from
1064  * the lmm_obejcts contained in the lov_user_md passed to this method. Full
1065  * OSTs are not considered. The exact order of OSTs requested by the user
1066  * is respected as much as possible depending on OST status. The number of
1067  * stripes needed and stripe offset are taken from the object. If that number
1068  * can not be met, then the function returns a failure and then it's the
1069  * caller's responsibility to release the stripes allocated. All the internal
1070  * structures are protected, but no concurrent allocation is allowed on the
1071  * same objects.
1072  *
1073  * \param[in] env       execution environment for this thread
1074  * \param[in] lo        LOD object
1075  * \param[out] stripe   striping created
1076  * \param[in] lum       stripe md to specify list of OSTs
1077  * \param[in] th        transaction handle
1078  *
1079  * \retval 0            on success
1080  * \retval -ENODEV      OST index does not exist on file system
1081  * \retval -EINVAL      requested OST index is invalid
1082  * \retval negative     negated errno on error
1083  */
1084 static int lod_alloc_ost_list(const struct lu_env *env,
1085                               struct lod_object *lo, struct dt_object **stripe,
1086                               struct lov_user_md *lum, struct thandle *th)
1087 {
1088         struct lod_device       *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1089         struct obd_statfs       *sfs = &lod_env_info(env)->lti_osfs;
1090         struct dt_object        *o;
1091         struct lov_user_md_v3   *v3;
1092         unsigned int            array_idx = 0;
1093         int                     stripe_count = 0;
1094         int                     i;
1095         int                     rc;
1096         ENTRY;
1097
1098         /* for specific OSTs layout */
1099         LASSERT(lum != NULL && lum->lmm_magic == LOV_USER_MAGIC_SPECIFIC);
1100         lustre_print_user_md(D_OTHER, lum, __func__);
1101
1102         rc = lod_qos_ost_in_use_clear(env, lo->ldo_stripenr);
1103         if (rc < 0)
1104                 RETURN(rc);
1105
1106         v3 = (struct lov_user_md_v3 *)lum;
1107         for (i = 0; i < lo->ldo_stripenr; i++) {
1108                 if (v3->lmm_objects[i].l_ost_idx == lo->ldo_def_stripe_offset) {
1109                         array_idx = i;
1110                         break;
1111                 }
1112         }
1113         if (i == lo->ldo_stripenr) {
1114                 CDEBUG(D_OTHER,
1115                        "%s: start index %d not in the specified list of OSTs\n",
1116                        lod2obd(m)->obd_name, lo->ldo_def_stripe_offset);
1117                 RETURN(-EINVAL);
1118         }
1119
1120         for (i = 0; i < lo->ldo_stripenr;
1121              i++, array_idx = (array_idx + 1) % lo->ldo_stripenr) {
1122                 __u32 ost_idx = v3->lmm_objects[array_idx].l_ost_idx;
1123
1124                 if (!cfs_bitmap_check(m->lod_ost_bitmap, ost_idx)) {
1125                         rc = -ENODEV;
1126                         break;
1127                 }
1128
1129                 /*
1130                  * do not put >1 objects on a single OST
1131                  */
1132                 if (lod_qos_is_ost_used(env, ost_idx, stripe_count)) {
1133                         rc = -EINVAL;
1134                         break;
1135                 }
1136
1137                 rc = lod_statfs_and_check(env, m, ost_idx, sfs);
1138                 if (rc < 0) /* this OSP doesn't feel well */
1139                         break;
1140
1141                 o = lod_qos_declare_object_on(env, m, ost_idx, th);
1142                 if (IS_ERR(o)) {
1143                         rc = PTR_ERR(o);
1144                         CDEBUG(D_OTHER,
1145                                "%s: can't declare new object on #%u: %d\n",
1146                                lod2obd(m)->obd_name, ost_idx, rc);
1147                         break;
1148                 }
1149
1150                 /*
1151                  * We've successfully declared (reserved) an object
1152                  */
1153                 lod_qos_ost_in_use(env, stripe_count, ost_idx);
1154                 stripe[stripe_count] = o;
1155                 stripe_count++;
1156         }
1157
1158         RETURN(rc);
1159 }
1160
1161 /**
1162  * Allocate a striping on a predefined set of OSTs.
1163  *
1164  * Allocates new layout starting from OST index in lo->ldo_def_stripe_offset.
1165  * Full OSTs are not considered. The exact order of OSTs is not important and
1166  * varies depending on OST status. The allocation procedure prefers the targets
1167  * with precreated objects ready. The number of stripes needed and stripe
1168  * offset are taken from the object. If that number cannot be met, then the
1169  * function returns an error and then it's the caller's responsibility to
1170  * release the stripes allocated. All the internal structures are protected,
1171  * but no concurrent allocation is allowed on the same objects.
1172  *
1173  * \param[in] env       execution environment for this thread
1174  * \param[in] lo        LOD object
1175  * \param[out] stripe   striping created
1176  * \param[in] flags     not used
1177  * \param[in] th        transaction handle
1178  *
1179  * \retval 0            on success
1180  * \retval -ENOSPC      if no OST objects are available at all
1181  * \retval -EFBIG       if not enough OST objects are found
1182  * \retval -EINVAL      requested offset is invalid
1183  * \retval negative     errno on failure
1184  */
1185 static int lod_alloc_specific(const struct lu_env *env, struct lod_object *lo,
1186                               struct dt_object **stripe, int flags,
1187                               struct thandle *th)
1188 {
1189         struct lod_device *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1190         struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
1191         struct dt_object  *o;
1192         __u32              ost_idx;
1193         unsigned int       i, array_idx, ost_count;
1194         int                rc, stripe_num = 0;
1195         int                speed = 0;
1196         struct pool_desc  *pool = NULL;
1197         struct ost_pool   *osts;
1198         ENTRY;
1199
1200         rc = lod_qos_ost_in_use_clear(env, lo->ldo_stripenr);
1201         if (rc)
1202                 GOTO(out, rc);
1203
1204         if (lo->ldo_pool)
1205                 pool = lod_find_pool(m, lo->ldo_pool);
1206
1207         if (pool != NULL) {
1208                 down_read(&pool_tgt_rw_sem(pool));
1209                 osts = &(pool->pool_obds);
1210         } else {
1211                 osts = &(m->lod_pool_info);
1212         }
1213
1214         ost_count = osts->op_count;
1215
1216 repeat_find:
1217         /* search loi_ost_idx in ost array */
1218         array_idx = 0;
1219         for (i = 0; i < ost_count; i++) {
1220                 if (osts->op_array[i] == lo->ldo_def_stripe_offset) {
1221                         array_idx = i;
1222                         break;
1223                 }
1224         }
1225         if (i == ost_count) {
1226                 CERROR("Start index %d not found in pool '%s'\n",
1227                        lo->ldo_def_stripe_offset,
1228                        lo->ldo_pool ? lo->ldo_pool : "");
1229                 GOTO(out, rc = -EINVAL);
1230         }
1231
1232         for (i = 0; i < ost_count;
1233                         i++, array_idx = (array_idx + 1) % ost_count) {
1234                 ost_idx = osts->op_array[array_idx];
1235
1236                 if (!cfs_bitmap_check(m->lod_ost_bitmap, ost_idx))
1237                         continue;
1238
1239                 /* Fail Check before osc_precreate() is called
1240                    so we can only 'fail' single OSC. */
1241                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
1242                         continue;
1243
1244                 /*
1245                  * do not put >1 objects on a single OST
1246                  */
1247                 if (lod_qos_is_ost_used(env, ost_idx, stripe_num))
1248                         continue;
1249
1250                 /* Drop slow OSCs if we can, but not for requested start idx.
1251                  *
1252                  * This means "if OSC is slow and it is not the requested
1253                  * start OST, then it can be skipped, otherwise skip it only
1254                  * if it is inactive/recovering/out-of-space." */
1255
1256                 rc = lod_statfs_and_check(env, m, ost_idx, sfs);
1257                 if (rc) {
1258                         /* this OSP doesn't feel well */
1259                         continue;
1260                 }
1261
1262                 /*
1263                  * We expect number of precreated objects in f_ffree at
1264                  * the first iteration, skip OSPs with no objects ready
1265                  * don't apply this logic to OST specified with stripe_offset
1266                  */
1267                 if (i != 0 && sfs->os_fprecreated == 0 && speed == 0)
1268                         continue;
1269
1270                 o = lod_qos_declare_object_on(env, m, ost_idx, th);
1271                 if (IS_ERR(o)) {
1272                         CDEBUG(D_OTHER, "can't declare new object on #%u: %d\n",
1273                                ost_idx, (int) PTR_ERR(o));
1274                         continue;
1275                 }
1276
1277                 /*
1278                  * We've successfully declared (reserved) an object
1279                  */
1280                 lod_qos_ost_in_use(env, stripe_num, ost_idx);
1281                 stripe[stripe_num] = o;
1282                 stripe_num++;
1283
1284                 /* We have enough stripes */
1285                 if (stripe_num == lo->ldo_stripenr)
1286                         GOTO(out, rc = 0);
1287         }
1288         if (speed < 2) {
1289                 /* Try again, allowing slower OSCs */
1290                 speed++;
1291                 goto repeat_find;
1292         }
1293
1294         /* If we were passed specific striping params, then a failure to
1295          * meet those requirements is an error, since we can't reallocate
1296          * that memory (it might be part of a larger array or something).
1297          */
1298         CERROR("can't lstripe objid "DFID": have %d want %u\n",
1299                PFID(lu_object_fid(lod2lu_obj(lo))), stripe_num,
1300                lo->ldo_stripenr);
1301         rc = stripe_num == 0 ? -ENOSPC : -EFBIG;
1302 out:
1303         if (pool != NULL) {
1304                 up_read(&pool_tgt_rw_sem(pool));
1305                 /* put back ref got by lod_find_pool() */
1306                 lod_pool_putref(pool);
1307         }
1308
1309         RETURN(rc);
1310 }
1311
1312 /**
1313  * Check whether QoS allocation should be used.
1314  *
1315  * A simple helper to decide when QoS allocation should be used:
1316  * if it's just a single available target or the used space is
1317  * evenly distributed among the targets at the moment, then QoS
1318  * allocation algorithm should not be used.
1319  *
1320  * \param[in] lod       LOD device
1321  *
1322  * \retval 0            should not be used
1323  * \retval 1            should be used
1324  */
1325 static inline int lod_qos_is_usable(struct lod_device *lod)
1326 {
1327 #ifdef FORCE_QOS
1328         /* to be able to debug QoS code */
1329         return 1;
1330 #endif
1331
1332         /* Detect -EAGAIN early, before expensive lock is taken. */
1333         if (!lod->lod_qos.lq_dirty && lod->lod_qos.lq_same_space)
1334                 return 0;
1335
1336         if (lod->lod_desc.ld_active_tgt_count < 2)
1337                 return 0;
1338
1339         return 1;
1340 }
1341
1342 /**
1343  * Allocate a striping using an algorithm with weights.
1344  *
1345  * The function allocates OST objects to create a striping. The algorithm
1346  * used is based on weights (currently only using the free space), and it's
1347  * trying to ensure the space is used evenly by OSTs and OSSs. The striping
1348  * configuration (# of stripes, offset, pool) is taken from the object and
1349  * is prepared by the caller.
1350  *
1351  * If LOV_USES_DEFAULT_STRIPE is not passed and prepared configuration can't
1352  * be met due to too few OSTs, then allocation fails. If the flag is passed
1353  * fewer than 3/4 of the requested number of stripes can be allocated, then
1354  * allocation fails.
1355  *
1356  * No concurrent allocation is allowed on the object and this must be ensured
1357  * by the caller. All the internal structures are protected by the function.
1358  *
1359  * The algorithm has two steps: find available OSTs and calculate their
1360  * weights, then select the OSTs with their weights used as the probability.
1361  * An OST with a higher weight is proportionately more likely to be selected
1362  * than one with a lower weight.
1363  *
1364  * \param[in] env       execution environment for this thread
1365  * \param[in] lo        LOD object
1366  * \param[out] stripe   striping created
1367  * \param[in] flags     0 or LOV_USES_DEFAULT_STRIPE
1368  * \param[in] th        transaction handle
1369  *
1370  * \retval 0            on success
1371  * \retval -EAGAIN      not enough OSTs are found for specified stripe count
1372  * \retval -EINVAL      requested OST index is invalid
1373  * \retval negative     errno on failure
1374  */
1375 static int lod_alloc_qos(const struct lu_env *env, struct lod_object *lo,
1376                          struct dt_object **stripe, int flags,
1377                          struct thandle *th)
1378 {
1379         struct lod_device   *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1380         struct obd_statfs   *sfs = &lod_env_info(env)->lti_osfs;
1381         struct lod_tgt_desc *ost;
1382         struct dt_object    *o;
1383         __u64                total_weight = 0;
1384         unsigned int         i;
1385         int                  rc = 0;
1386         __u32                nfound, good_osts;
1387         __u32                stripe_cnt = lo->ldo_stripenr;
1388         __u32                stripe_cnt_min;
1389         struct pool_desc    *pool = NULL;
1390         struct ost_pool    *osts;
1391         ENTRY;
1392
1393         stripe_cnt_min = min_stripe_count(stripe_cnt, flags);
1394         if (stripe_cnt_min < 1)
1395                 RETURN(-EINVAL);
1396
1397         if (lo->ldo_pool)
1398                 pool = lod_find_pool(m, lo->ldo_pool);
1399
1400         if (pool != NULL) {
1401                 down_read(&pool_tgt_rw_sem(pool));
1402                 osts = &(pool->pool_obds);
1403         } else {
1404                 osts = &(m->lod_pool_info);
1405         }
1406
1407         /* Detect -EAGAIN early, before expensive lock is taken. */
1408         if (!lod_qos_is_usable(m))
1409                 GOTO(out_nolock, rc = -EAGAIN);
1410
1411         /* Do actual allocation, use write lock here. */
1412         down_write(&m->lod_qos.lq_rw_sem);
1413
1414         /*
1415          * Check again, while we were sleeping on @lq_rw_sem things could
1416          * change.
1417          */
1418         if (!lod_qos_is_usable(m))
1419                 GOTO(out, rc = -EAGAIN);
1420
1421         rc = lod_qos_calc_ppo(m);
1422         if (rc)
1423                 GOTO(out, rc);
1424
1425         rc = lod_qos_ost_in_use_clear(env, lo->ldo_stripenr);
1426         if (rc)
1427                 GOTO(out, rc);
1428
1429         good_osts = 0;
1430         /* Find all the OSTs that are valid stripe candidates */
1431         for (i = 0; i < osts->op_count; i++) {
1432                 if (!cfs_bitmap_check(m->lod_ost_bitmap, osts->op_array[i]))
1433                         continue;
1434
1435                 rc = lod_statfs_and_check(env, m, osts->op_array[i], sfs);
1436                 if (rc) {
1437                         /* this OSP doesn't feel well */
1438                         continue;
1439                 }
1440
1441                 /*
1442                  * skip full devices
1443                  */
1444                 if (lod_qos_dev_is_full(sfs))
1445                         continue;
1446
1447                 /* Fail Check before osc_precreate() is called
1448                    so we can only 'fail' single OSC. */
1449                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) &&
1450                                    osts->op_array[i] == 0)
1451                         continue;
1452
1453                 ost = OST_TGT(m,osts->op_array[i]);
1454                 ost->ltd_qos.ltq_usable = 1;
1455                 lod_qos_calc_weight(m, osts->op_array[i]);
1456                 total_weight += ost->ltd_qos.ltq_weight;
1457
1458                 good_osts++;
1459         }
1460
1461         QOS_DEBUG("found %d good osts\n", good_osts);
1462
1463         if (good_osts < stripe_cnt_min)
1464                 GOTO(out, rc = -EAGAIN);
1465
1466         /* We have enough osts */
1467         if (good_osts < stripe_cnt)
1468                 stripe_cnt = good_osts;
1469
1470         /* Find enough OSTs with weighted random allocation. */
1471         nfound = 0;
1472         while (nfound < stripe_cnt) {
1473                 __u64 rand, cur_weight;
1474
1475                 cur_weight = 0;
1476                 rc = -ENOSPC;
1477
1478                 if (total_weight) {
1479 #if BITS_PER_LONG == 32
1480                         rand = cfs_rand() % (unsigned)total_weight;
1481                         /* If total_weight > 32-bit, first generate the high
1482                          * 32 bits of the random number, then add in the low
1483                          * 32 bits (truncated to the upper limit, if needed) */
1484                         if (total_weight > 0xffffffffULL)
1485                                 rand = (__u64)(cfs_rand() %
1486                                         (unsigned)(total_weight >> 32)) << 32;
1487                         else
1488                                 rand = 0;
1489
1490                         if (rand == (total_weight & 0xffffffff00000000ULL))
1491                                 rand |= cfs_rand() % (unsigned)total_weight;
1492                         else
1493                                 rand |= cfs_rand();
1494
1495 #else
1496                         rand = ((__u64)cfs_rand() << 32 | cfs_rand()) %
1497                                 total_weight;
1498 #endif
1499                 } else {
1500                         rand = 0;
1501                 }
1502
1503                 /* On average, this will hit larger-weighted OSTs more often.
1504                  * 0-weight OSTs will always get used last (only when rand=0) */
1505                 for (i = 0; i < osts->op_count; i++) {
1506                         __u32 idx = osts->op_array[i];
1507
1508                         if (!cfs_bitmap_check(m->lod_ost_bitmap, idx))
1509                                 continue;
1510
1511                         ost = OST_TGT(m,idx);
1512
1513                         if (!ost->ltd_qos.ltq_usable)
1514                                 continue;
1515
1516                         cur_weight += ost->ltd_qos.ltq_weight;
1517                         QOS_DEBUG("stripe_cnt=%d nfound=%d cur_weight="LPU64
1518                                   " rand="LPU64" total_weight="LPU64"\n",
1519                                   stripe_cnt, nfound, cur_weight, rand,
1520                                   total_weight);
1521
1522                         if (cur_weight < rand)
1523                                 continue;
1524
1525                         QOS_DEBUG("stripe=%d to idx=%d\n", nfound, idx);
1526
1527                         /*
1528                          * do not put >1 objects on a single OST
1529                          */
1530                         if (lod_qos_is_ost_used(env, idx, nfound))
1531                                 continue;
1532                         lod_qos_ost_in_use(env, nfound, idx);
1533
1534                         o = lod_qos_declare_object_on(env, m, idx, th);
1535                         if (IS_ERR(o)) {
1536                                 QOS_DEBUG("can't declare object on #%u: %d\n",
1537                                           idx, (int) PTR_ERR(o));
1538                                 continue;
1539                         }
1540                         stripe[nfound++] = o;
1541                         lod_qos_used(m, osts, idx, &total_weight);
1542                         rc = 0;
1543                         break;
1544                 }
1545
1546                 if (rc) {
1547                         /* no OST found on this iteration, give up */
1548                         break;
1549                 }
1550         }
1551
1552         if (unlikely(nfound != stripe_cnt)) {
1553                 /*
1554                  * when the decision to use weighted algorithm was made
1555                  * we had enough appropriate OSPs, but this state can
1556                  * change anytime (no space on OST, broken connection, etc)
1557                  * so it's possible OSP won't be able to provide us with
1558                  * an object due to just changed state
1559                  */
1560                 LCONSOLE_INFO("wanted %d, found %d\n", stripe_cnt, nfound);
1561                 for (i = 0; i < nfound; i++) {
1562                         LASSERT(stripe[i] != NULL);
1563                         lu_object_put(env, &stripe[i]->do_lu);
1564                         stripe[i] = NULL;
1565                 }
1566
1567                 /* makes sense to rebalance next time */
1568                 m->lod_qos.lq_dirty = 1;
1569                 m->lod_qos.lq_same_space = 0;
1570
1571                 rc = -EAGAIN;
1572         }
1573
1574 out:
1575         up_write(&m->lod_qos.lq_rw_sem);
1576
1577 out_nolock:
1578         if (pool != NULL) {
1579                 up_read(&pool_tgt_rw_sem(pool));
1580                 /* put back ref got by lod_find_pool() */
1581                 lod_pool_putref(pool);
1582         }
1583
1584         RETURN(rc);
1585 }
1586
1587 /**
1588  * Find largest stripe count the caller can use.
1589  *
1590  * Find the maximal possible stripe count not greater than \a stripe_count.
1591  * Sometimes suggested stripecount can't be reached for a number of reasons:
1592  * lack of enough active OSTs or the backend does not support EAs that large.
1593  * If the passed one is 0, then the filesystem's default one is used.
1594  *
1595  * \param[in] lod       LOD device
1596  * \param[in] magic     the format if striping
1597  * \param[in] stripe_count      count the caller would like to use
1598  *
1599  * \retval              the maximum usable stripe count
1600  */
1601 static __u16 lod_get_stripecnt(struct lod_device *lod, __u32 magic,
1602                                __u16 stripe_count)
1603 {
1604         __u32 max_stripes = LOV_MAX_STRIPE_COUNT_OLD;
1605
1606         if (!stripe_count)
1607                 stripe_count = lod->lod_desc.ld_default_stripe_count;
1608         if (stripe_count > lod->lod_desc.ld_active_tgt_count)
1609                 stripe_count = lod->lod_desc.ld_active_tgt_count;
1610         if (!stripe_count)
1611                 stripe_count = 1;
1612
1613         /* stripe count is based on whether OSD can handle larger EA sizes */
1614         if (lod->lod_osd_max_easize > 0)
1615                 max_stripes = lov_mds_md_max_stripe_count(
1616                         lod->lod_osd_max_easize, magic);
1617
1618         return (stripe_count < max_stripes) ? stripe_count : max_stripes;
1619 }
1620
1621 /**
1622  * Create in-core respresentation for a fully-defined striping
1623  *
1624  * When the caller passes a fully-defined striping (i.e. everything including
1625  * OST object FIDs are defined), then we still need to instantiate LU-cache
1626  * with the objects representing the stripes defined. This function completes
1627  * that task.
1628  *
1629  * \param[in] env       execution environment for this thread
1630  * \param[in] mo        LOD object
1631  * \param[in] buf       buffer containing the striping
1632  *
1633  * \retval 0            on success
1634  * \retval negative     negated errno on error
1635  */
1636 static int lod_use_defined_striping(const struct lu_env *env,
1637                                     struct lod_object *mo,
1638                                     const struct lu_buf *buf)
1639 {
1640         struct lov_mds_md_v1   *v1 = buf->lb_buf;
1641         struct lov_mds_md_v3   *v3 = buf->lb_buf;
1642         struct lov_ost_data_v1 *objs;
1643         __u32                   magic;
1644         int                     rc = 0;
1645         ENTRY;
1646
1647         magic = le32_to_cpu(v1->lmm_magic);
1648         if (magic == LOV_MAGIC_V1_DEF) {
1649                 magic = LOV_MAGIC_V1;
1650                 objs = &v1->lmm_objects[0];
1651         } else if (magic == LOV_MAGIC_V3_DEF) {
1652                 magic = LOV_MAGIC_V3;
1653                 objs = &v3->lmm_objects[0];
1654                 lod_object_set_pool(mo, v3->lmm_pool_name);
1655         } else {
1656                 GOTO(out, rc = -EINVAL);
1657         }
1658
1659         mo->ldo_pattern = le32_to_cpu(v1->lmm_pattern);
1660         mo->ldo_stripe_size = le32_to_cpu(v1->lmm_stripe_size);
1661         mo->ldo_stripenr = le16_to_cpu(v1->lmm_stripe_count);
1662         mo->ldo_layout_gen = le16_to_cpu(v1->lmm_layout_gen);
1663
1664         /* fixup for released file before object initialization */
1665         if (mo->ldo_pattern & LOV_PATTERN_F_RELEASED) {
1666                 mo->ldo_released_stripenr = mo->ldo_stripenr;
1667                 mo->ldo_stripenr = 0;
1668         }
1669
1670         LASSERT(buf->lb_len >= lov_mds_md_size(mo->ldo_stripenr, magic));
1671
1672         if (mo->ldo_stripenr > 0)
1673                 rc = lod_initialize_objects(env, mo, objs);
1674
1675 out:
1676         RETURN(rc);
1677 }
1678
1679 /**
1680  * Parse suggested striping configuration.
1681  *
1682  * The caller gets a suggested striping configuration from a number of sources
1683  * including per-directory default and applications. Then it needs to verify
1684  * the suggested striping is valid, apply missing bits and store the resulting
1685  * configuration in the object to be used by the allocator later. Must not be
1686  * called concurrently against the same object. It's OK to provide a
1687  * fully-defined striping.
1688  *
1689  * \param[in] env       execution environment for this thread
1690  * \param[in] lo        LOD object
1691  * \param[in] buf       buffer containing the striping
1692  *
1693  * \retval 0            on success
1694  * \retval negative     negated errno on error
1695  */
1696 static int lod_qos_parse_config(const struct lu_env *env,
1697                                 struct lod_object *lo,
1698                                 const struct lu_buf *buf)
1699 {
1700         struct lod_device     *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
1701         struct lov_user_md_v1 *v1 = NULL;
1702         struct lov_user_md_v3 *v3 = NULL;
1703         char                  *pool_name = NULL;
1704         __u32                  magic;
1705         int                    rc;
1706         unsigned int           size;
1707         ENTRY;
1708
1709         if (buf == NULL || buf->lb_buf == NULL || buf->lb_len == 0)
1710                 RETURN(0);
1711
1712         v3 = buf->lb_buf;
1713         v1 = buf->lb_buf;
1714         magic = v1->lmm_magic;
1715
1716         if (unlikely(magic == LOV_MAGIC_V1_DEF || magic == LOV_MAGIC_V3_DEF)) {
1717                 /* try to use as fully defined striping */
1718                 rc = lod_use_defined_striping(env, lo, buf);
1719                 RETURN(rc);
1720         }
1721
1722         switch (magic) {
1723         case __swab32(LOV_USER_MAGIC_V1):
1724                 lustre_swab_lov_user_md_v1(v1);
1725                 magic = v1->lmm_magic;
1726                 /* fall through */
1727         case LOV_USER_MAGIC_V1:
1728                 size = sizeof(*v1);
1729                 break;
1730
1731         case __swab32(LOV_USER_MAGIC_V3):
1732                 lustre_swab_lov_user_md_v3(v3);
1733                 magic = v3->lmm_magic;
1734                 /* fall through */
1735         case LOV_USER_MAGIC_V3:
1736                 size = sizeof(*v3);
1737                 pool_name = v3->lmm_pool_name;
1738                 break;
1739
1740         case __swab32(LOV_USER_MAGIC_SPECIFIC):
1741                 lustre_swab_lov_user_md_v3(v3);
1742                 lustre_swab_lov_user_md_objects(v3->lmm_objects,
1743                                                 v3->lmm_stripe_count);
1744                 magic = v3->lmm_magic;
1745                 /* fall through */
1746         case LOV_USER_MAGIC_SPECIFIC:
1747                 if (v3->lmm_stripe_offset == LOV_OFFSET_DEFAULT)
1748                         v3->lmm_stripe_offset = v3->lmm_objects[0].l_ost_idx;
1749                 if (v3->lmm_pool_name[0] != '\0')
1750                         pool_name = v3->lmm_pool_name;
1751                 size = lov_user_md_size(v3->lmm_stripe_count,
1752                                         LOV_USER_MAGIC_SPECIFIC);
1753                 break;
1754
1755         default:
1756                 CERROR("%s: unrecognized magic %X\n",
1757                        lod2obd(d)->obd_name, magic);
1758                 RETURN(-EINVAL);
1759         }
1760
1761         if (unlikely(buf->lb_len < size)) {
1762                 CERROR("%s: wrong size: %zd, expect: %u\n",
1763                        lod2obd(d)->obd_name, buf->lb_len, size);
1764                 RETURN(-EINVAL);
1765         }
1766
1767         lustre_print_user_md(D_OTHER, v1, "parse config");
1768
1769         v1->lmm_magic = magic;
1770         if (v1->lmm_pattern == 0)
1771                 v1->lmm_pattern = LOV_PATTERN_RAID0;
1772         if (lov_pattern(v1->lmm_pattern) != LOV_PATTERN_RAID0) {
1773                 CERROR("%s: invalid pattern: %x\n",
1774                        lod2obd(d)->obd_name, v1->lmm_pattern);
1775                 RETURN(-EINVAL);
1776         }
1777         lo->ldo_pattern = v1->lmm_pattern;
1778
1779         if (v1->lmm_stripe_size > 0)
1780                 lo->ldo_stripe_size = v1->lmm_stripe_size;
1781
1782         if (lo->ldo_stripe_size & (LOV_MIN_STRIPE_SIZE - 1))
1783                 lo->ldo_stripe_size = LOV_MIN_STRIPE_SIZE;
1784
1785         if (v1->lmm_stripe_count > 0)
1786                 lo->ldo_stripenr = v1->lmm_stripe_count;
1787
1788         lo->ldo_def_stripe_offset = v1->lmm_stripe_offset;
1789
1790         lod_object_set_pool(lo, NULL);
1791         if (pool_name != NULL) {
1792                 struct pool_desc *pool;
1793
1794                 /* In the function below, .hs_keycmp resolves to
1795                  * pool_hashkey_keycmp() */
1796                 /* coverity[overrun-buffer-val] */
1797                 pool = lod_find_pool(d, pool_name);
1798                 if (pool != NULL) {
1799                         if (lo->ldo_def_stripe_offset != LOV_OFFSET_DEFAULT) {
1800                                 rc = lod_check_index_in_pool(
1801                                                lo->ldo_def_stripe_offset, pool);
1802                                 if (rc < 0) {
1803                                         lod_pool_putref(pool);
1804                                         CERROR("%s: invalid offset, %u\n",
1805                                                lod2obd(d)->obd_name,
1806                                                lo->ldo_def_stripe_offset);
1807                                         RETURN(-EINVAL);
1808                                 }
1809                         }
1810
1811                         if (lo->ldo_stripenr > pool_tgt_count(pool))
1812                                 lo->ldo_stripenr = pool_tgt_count(pool);
1813
1814                         lod_pool_putref(pool);
1815                 }
1816
1817                 lod_object_set_pool(lo, pool_name);
1818         }
1819
1820         /* fixup for released file */
1821         if (lo->ldo_pattern & LOV_PATTERN_F_RELEASED) {
1822                 lo->ldo_released_stripenr = lo->ldo_stripenr;
1823                 lo->ldo_stripenr = 0;
1824         }
1825
1826         RETURN(0);
1827 }
1828
1829 /**
1830  * Create a striping for an obejct.
1831  *
1832  * The function creates a new striping for the object. A buffer containing
1833  * configuration hints can be provided optionally. The function tries QoS
1834  * algorithm first unless free space is distributed evenly among OSTs, but
1835  * by default RR algorithm is preferred due to internal concurrency (QoS is
1836  * serialized). The caller must ensure no concurrent calls to the function
1837  * are made against the same object.
1838  *
1839  * \param[in] env       execution environment for this thread
1840  * \param[in] lo        LOD object
1841  * \param[in] attr      attributes OST objects will be declared with
1842  * \param[in] buf       suggested striping configuration or NULL
1843  * \param[in] th        transaction handle
1844  *
1845  * \retval 0            on success
1846  * \retval negative     negated errno on error
1847  */
1848 int lod_qos_prep_create(const struct lu_env *env, struct lod_object *lo,
1849                         struct lu_attr *attr, const struct lu_buf *buf,
1850                         struct thandle *th)
1851 {
1852         struct lod_device      *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
1853         struct dt_object      **stripe;
1854         int                     stripe_len;
1855         int                     flag = LOV_USES_ASSIGNED_STRIPE;
1856         int                     i, rc;
1857         ENTRY;
1858
1859         LASSERT(lo);
1860
1861         /* no OST available */
1862         /* XXX: should we be waiting a bit to prevent failures during
1863          * cluster initialization? */
1864         if (d->lod_ostnr == 0)
1865                 GOTO(out, rc = -EIO);
1866
1867         /*
1868          * by this time, the object's ldo_stripenr and ldo_stripe_size
1869          * contain default value for striping: taken from the parent
1870          * or from filesystem defaults
1871          *
1872          * in case the caller is passing lovea with new striping config,
1873          * we may need to parse lovea and apply new configuration
1874          */
1875         rc = lod_qos_parse_config(env, lo, buf);
1876         if (rc)
1877                 GOTO(out, rc);
1878
1879         /* A released file is being created */
1880         if (lo->ldo_stripenr == 0)
1881                 GOTO(out, rc = 0);
1882
1883         if (likely(lo->ldo_stripe == NULL)) {
1884                 struct lov_user_md *lum = NULL;
1885
1886                 /*
1887                  * no striping has been created so far
1888                  */
1889                 LASSERT(lo->ldo_stripenr > 0);
1890                 /*
1891                  * statfs and check OST targets now, since ld_active_tgt_count
1892                  * could be changed if some OSTs are [de]activated manually.
1893                  */
1894                 lod_qos_statfs_update(env, d);
1895                 lo->ldo_stripenr = lod_get_stripecnt(d, LOV_MAGIC,
1896                                                      lo->ldo_stripenr);
1897
1898                 stripe_len = lo->ldo_stripenr;
1899                 OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_len);
1900                 if (stripe == NULL)
1901                         GOTO(out, rc = -ENOMEM);
1902
1903                 lod_getref(&d->lod_ost_descs);
1904                 /* XXX: support for non-0 files w/o objects */
1905                 CDEBUG(D_OTHER, "tgt_count %d stripenr %d\n",
1906                                 d->lod_desc.ld_tgt_count, stripe_len);
1907
1908                 if (buf != NULL && buf->lb_buf != NULL)
1909                         lum = buf->lb_buf;
1910
1911                 if (lum != NULL && lum->lmm_magic == LOV_USER_MAGIC_SPECIFIC) {
1912                         rc = lod_alloc_ost_list(env, lo, stripe, lum, th);
1913                 } else if (lo->ldo_def_stripe_offset == LOV_OFFSET_DEFAULT) {
1914                         rc = lod_alloc_qos(env, lo, stripe, flag, th);
1915                         if (rc == -EAGAIN)
1916                                 rc = lod_alloc_rr(env, lo, stripe, flag, th);
1917                 } else {
1918                         rc = lod_alloc_specific(env, lo, stripe, flag, th);
1919                 }
1920                 lod_putref(d, &d->lod_ost_descs);
1921
1922                 if (rc < 0) {
1923                         for (i = 0; i < stripe_len; i++)
1924                                 if (stripe[i] != NULL)
1925                                         lu_object_put(env, &stripe[i]->do_lu);
1926
1927                         OBD_FREE(stripe, sizeof(stripe[0]) * stripe_len);
1928                         lo->ldo_stripenr = 0;
1929                 } else {
1930                         lo->ldo_stripe = stripe;
1931                         lo->ldo_stripes_allocated = stripe_len;
1932                 }
1933         } else {
1934                 /*
1935                  * lod_qos_parse_config() found supplied buf as a predefined
1936                  * striping (not a hint), so it allocated all the object
1937                  * now we need to create them
1938                  */
1939                 for (i = 0; i < lo->ldo_stripenr; i++) {
1940                         struct dt_object  *o;
1941
1942                         o = lo->ldo_stripe[i];
1943                         LASSERT(o);
1944
1945                         rc = lod_sub_object_declare_create(env, o, attr, NULL,
1946                                                            NULL, th);
1947                         if (rc < 0) {
1948                                 CERROR("can't declare create: %d\n", rc);
1949                                 break;
1950                         }
1951                 }
1952         }
1953
1954 out:
1955         RETURN(rc);
1956 }
1957