Whamcloud - gitweb
e00b172bca1fba969627ffa9cb269a40ac194cea
[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, 2017, 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 <uapi/linux/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
203         if (rc == 0 && ((sfs->os_state & OS_STATE_ENOSPC) ||
204             (sfs->os_state & OS_STATE_ENOINO && sfs->os_fprecreated == 0)))
205                 RETURN(-ENOSPC);
206
207         if (rc && rc != -ENOTCONN)
208                 CERROR("%s: statfs: rc = %d\n", lod2obd(d)->obd_name, rc);
209
210         /* If the OST is readonly then we can't allocate objects there */
211         if (sfs->os_state & OS_STATE_READONLY)
212                 rc = -EROFS;
213
214         /* check whether device has changed state (active, inactive) */
215         if (rc != 0 && ost->ltd_active) {
216                 /* turned inactive? */
217                 spin_lock(&d->lod_lock);
218                 if (ost->ltd_active) {
219                         ost->ltd_active = 0;
220                         if (rc == -ENOTCONN)
221                                 ost->ltd_connecting = 1;
222
223                         LASSERT(d->lod_desc.ld_active_tgt_count > 0);
224                         d->lod_desc.ld_active_tgt_count--;
225                         d->lod_qos.lq_dirty = 1;
226                         d->lod_qos.lq_rr.lqr_dirty = 1;
227                         CDEBUG(D_CONFIG, "%s: turns inactive\n",
228                                ost->ltd_exp->exp_obd->obd_name);
229                 }
230                 spin_unlock(&d->lod_lock);
231         } else if (rc == 0 && ost->ltd_active == 0) {
232                 /* turned active? */
233                 LASSERTF(d->lod_desc.ld_active_tgt_count < d->lod_ostnr,
234                          "active tgt count %d, ost nr %d\n",
235                          d->lod_desc.ld_active_tgt_count, d->lod_ostnr);
236                 spin_lock(&d->lod_lock);
237                 if (ost->ltd_active == 0) {
238                         ost->ltd_active = 1;
239                         ost->ltd_connecting = 0;
240                         d->lod_desc.ld_active_tgt_count++;
241                         d->lod_qos.lq_dirty = 1;
242                         d->lod_qos.lq_rr.lqr_dirty = 1;
243                         CDEBUG(D_CONFIG, "%s: turns active\n",
244                                ost->ltd_exp->exp_obd->obd_name);
245                 }
246                 spin_unlock(&d->lod_lock);
247         }
248
249         RETURN(rc);
250 }
251
252 /**
253  * Maintain per-target statfs data.
254  *
255  * The function refreshes statfs data for all the targets every N seconds.
256  * The actual N is controlled via procfs and set to LOV_DESC_QOS_MAXAGE_DEFAULT
257  * initially.
258  *
259  * \param[in] env       execution environment for this thread
260  * \param[in] lod       LOD device
261  */
262 void lod_qos_statfs_update(const struct lu_env *env, struct lod_device *lod)
263 {
264         struct obd_device *obd = lod2obd(lod);
265         struct ost_pool *osts = &(lod->lod_pool_info);
266         time64_t max_age;
267         unsigned int i;
268         u64 avail;
269         int idx;
270         ENTRY;
271
272         max_age = ktime_get_seconds() - 2 * lod->lod_desc.ld_qos_maxage;
273
274         if (obd->obd_osfs_age > max_age)
275                 /* statfs data are quite recent, don't need to refresh it */
276                 RETURN_EXIT;
277
278         down_write(&lod->lod_qos.lq_rw_sem);
279
280         if (obd->obd_osfs_age > max_age)
281                 goto out;
282
283         for (i = 0; i < osts->op_count; i++) {
284                 idx = osts->op_array[i];
285                 avail = OST_TGT(lod,idx)->ltd_statfs.os_bavail;
286                 if (lod_statfs_and_check(env, lod, idx,
287                                          &OST_TGT(lod, idx)->ltd_statfs))
288                         continue;
289                 if (OST_TGT(lod,idx)->ltd_statfs.os_bavail != avail)
290                         /* recalculate weigths */
291                         lod->lod_qos.lq_dirty = 1;
292         }
293         obd->obd_osfs_age = ktime_get_seconds();
294
295 out:
296         up_write(&lod->lod_qos.lq_rw_sem);
297         EXIT;
298 }
299
300 /**
301  * Calculate per-OST and per-OSS penalties
302  *
303  * Re-calculate penalties when the configuration changes, active targets
304  * change and after statfs refresh (all these are reflected by lq_dirty flag).
305  * On every OST and OSS: decay the penalty by half for every 8x the update
306  * interval that the device has been idle. That gives lots of time for the
307  * statfs information to be updated (which the penalty is only a proxy for),
308  * and avoids penalizing OSS/OSTs under light load.
309  * See lod_qos_calc_weight() for how penalties are factored into the weight.
310  *
311  * \param[in] lod       LOD device
312  *
313  * \retval 0            on success
314  * \retval -EAGAIN      the number of OSTs isn't enough
315  */
316 static int lod_qos_calc_ppo(struct lod_device *lod)
317 {
318         struct lod_qos_oss *oss;
319         __u64               ba_max, ba_min, temp;
320         __u32               num_active;
321         unsigned int        i;
322         int                 rc, prio_wide;
323         time64_t            now, age;
324         ENTRY;
325
326         if (!lod->lod_qos.lq_dirty)
327                 GOTO(out, rc = 0);
328
329         num_active = lod->lod_desc.ld_active_tgt_count - 1;
330         if (num_active < 1)
331                 GOTO(out, rc = -EAGAIN);
332
333         /* find bavail on each OSS */
334         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list)
335                             oss->lqo_bavail = 0;
336         lod->lod_qos.lq_active_oss_count = 0;
337
338         /*
339          * How badly user wants to select OSTs "widely" (not recently chosen
340          * and not on recent OSS's).  As opposed to "freely" (free space
341          * avail.) 0-256
342          */
343         prio_wide = 256 - lod->lod_qos.lq_prio_free;
344
345         ba_min = (__u64)(-1);
346         ba_max = 0;
347         now = ktime_get_real_seconds();
348         /* Calculate OST penalty per object
349          * (lod ref taken in lod_qos_prep_create())
350          */
351         cfs_foreach_bit(lod->lod_ost_bitmap, i) {
352                 LASSERT(OST_TGT(lod,i));
353                 temp = TGT_BAVAIL(i);
354                 if (!temp)
355                         continue;
356                 ba_min = min(temp, ba_min);
357                 ba_max = max(temp, ba_max);
358
359                 /* Count the number of usable OSS's */
360                 if (OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_bavail == 0)
361                         lod->lod_qos.lq_active_oss_count++;
362                 OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_bavail += temp;
363
364                 /* per-OST penalty is prio * TGT_bavail / (num_ost - 1) / 2 */
365                 temp >>= 1;
366                 do_div(temp, num_active);
367                 OST_TGT(lod,i)->ltd_qos.ltq_penalty_per_obj =
368                         (temp * prio_wide) >> 8;
369
370                 age = (now - OST_TGT(lod,i)->ltd_qos.ltq_used) >> 3;
371                 if (lod->lod_qos.lq_reset ||
372                     age > 32 * lod->lod_desc.ld_qos_maxage)
373                         OST_TGT(lod,i)->ltd_qos.ltq_penalty = 0;
374                 else if (age > lod->lod_desc.ld_qos_maxage)
375                         /* Decay OST penalty. */
376                         OST_TGT(lod,i)->ltd_qos.ltq_penalty >>=
377                                 (age / lod->lod_desc.ld_qos_maxage);
378         }
379
380         num_active = lod->lod_qos.lq_active_oss_count - 1;
381         if (num_active < 1) {
382                 /* If there's only 1 OSS, we can't penalize it, so instead
383                    we have to double the OST penalty */
384                 num_active = 1;
385                 cfs_foreach_bit(lod->lod_ost_bitmap, i)
386                         OST_TGT(lod,i)->ltd_qos.ltq_penalty_per_obj <<= 1;
387         }
388
389         /* Per-OSS penalty is prio * oss_avail / oss_osts / (num_oss - 1) / 2 */
390         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
391                 temp = oss->lqo_bavail >> 1;
392                 do_div(temp, oss->lqo_ost_count * num_active);
393                 oss->lqo_penalty_per_obj = (temp * prio_wide) >> 8;
394
395                 age = (now - oss->lqo_used) >> 3;
396                 if (lod->lod_qos.lq_reset ||
397                     age > 32 * lod->lod_desc.ld_qos_maxage)
398                         oss->lqo_penalty = 0;
399                 else if (age > lod->lod_desc.ld_qos_maxage)
400                         /* Decay OSS penalty. */
401                         oss->lqo_penalty >>= age / lod->lod_desc.ld_qos_maxage;
402         }
403
404         lod->lod_qos.lq_dirty = 0;
405         lod->lod_qos.lq_reset = 0;
406
407         /* If each ost has almost same free space,
408          * do rr allocation for better creation performance */
409         lod->lod_qos.lq_same_space = 0;
410         if ((ba_max * (256 - lod->lod_qos.lq_threshold_rr)) >> 8 < ba_min) {
411                 lod->lod_qos.lq_same_space = 1;
412                 /* Reset weights for the next time we enter qos mode */
413                 lod->lod_qos.lq_reset = 1;
414         }
415         rc = 0;
416
417 out:
418 #ifndef FORCE_QOS
419         if (!rc && lod->lod_qos.lq_same_space)
420                 RETURN(-EAGAIN);
421 #endif
422         RETURN(rc);
423 }
424
425 /**
426  * Calculate weight for a given OST target.
427  *
428  * The final OST weight is the number of bytes available minus the OST and
429  * OSS penalties.  See lod_qos_calc_ppo() for how penalties are calculated.
430  *
431  * \param[in] lod       LOD device, where OST targets are listed
432  * \param[in] i         OST target index
433  *
434  * \retval              0
435  */
436 static int lod_qos_calc_weight(struct lod_device *lod, int i)
437 {
438         __u64 temp, temp2;
439
440         temp = TGT_BAVAIL(i);
441         temp2 = OST_TGT(lod,i)->ltd_qos.ltq_penalty +
442                 OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_penalty;
443         if (temp < temp2)
444                 OST_TGT(lod,i)->ltd_qos.ltq_weight = 0;
445         else
446                 OST_TGT(lod,i)->ltd_qos.ltq_weight = temp - temp2;
447         return 0;
448 }
449
450 /**
451  * Re-calculate weights.
452  *
453  * The function is called when some OST target was used for a new object. In
454  * this case we should re-calculate all the weights to keep new allocations
455  * balanced well.
456  *
457  * \param[in] lod       LOD device
458  * \param[in] osts      OST pool where a new object was placed
459  * \param[in] index     OST target where a new object was placed
460  * \param[out] total_wt new total weight for the pool
461  *
462  * \retval              0
463  */
464 static int lod_qos_used(struct lod_device *lod, struct ost_pool *osts,
465                         __u32 index, __u64 *total_wt)
466 {
467         struct lod_tgt_desc *ost;
468         struct lod_qos_oss  *oss;
469         unsigned int j;
470         ENTRY;
471
472         ost = OST_TGT(lod,index);
473         LASSERT(ost);
474
475         /* Don't allocate on this devuce anymore, until the next alloc_qos */
476         ost->ltd_qos.ltq_usable = 0;
477
478         oss = ost->ltd_qos.ltq_oss;
479
480         /* Decay old penalty by half (we're adding max penalty, and don't
481            want it to run away.) */
482         ost->ltd_qos.ltq_penalty >>= 1;
483         oss->lqo_penalty >>= 1;
484
485         /* mark the OSS and OST as recently used */
486         ost->ltd_qos.ltq_used = oss->lqo_used = ktime_get_real_seconds();
487
488         /* Set max penalties for this OST and OSS */
489         ost->ltd_qos.ltq_penalty +=
490                 ost->ltd_qos.ltq_penalty_per_obj * lod->lod_ostnr;
491         oss->lqo_penalty += oss->lqo_penalty_per_obj *
492                 lod->lod_qos.lq_active_oss_count;
493
494         /* Decrease all OSS penalties */
495         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
496                 if (oss->lqo_penalty < oss->lqo_penalty_per_obj)
497                         oss->lqo_penalty = 0;
498                 else
499                         oss->lqo_penalty -= oss->lqo_penalty_per_obj;
500         }
501
502         *total_wt = 0;
503         /* Decrease all OST penalties */
504         for (j = 0; j < osts->op_count; j++) {
505                 int i;
506
507                 i = osts->op_array[j];
508                 if (!cfs_bitmap_check(lod->lod_ost_bitmap, i))
509                         continue;
510
511                 ost = OST_TGT(lod,i);
512                 LASSERT(ost);
513
514                 if (ost->ltd_qos.ltq_penalty <
515                                 ost->ltd_qos.ltq_penalty_per_obj)
516                         ost->ltd_qos.ltq_penalty = 0;
517                 else
518                         ost->ltd_qos.ltq_penalty -=
519                                 ost->ltd_qos.ltq_penalty_per_obj;
520
521                 lod_qos_calc_weight(lod, i);
522
523                 /* Recalc the total weight of usable osts */
524                 if (ost->ltd_qos.ltq_usable)
525                         *total_wt += ost->ltd_qos.ltq_weight;
526
527                 QOS_DEBUG("recalc tgt %d usable=%d avail=%llu"
528                           " ostppo=%llu ostp=%llu ossppo=%llu"
529                           " ossp=%llu wt=%llu\n",
530                           i, ost->ltd_qos.ltq_usable, TGT_BAVAIL(i) >> 10,
531                           ost->ltd_qos.ltq_penalty_per_obj >> 10,
532                           ost->ltd_qos.ltq_penalty >> 10,
533                           ost->ltd_qos.ltq_oss->lqo_penalty_per_obj >> 10,
534                           ost->ltd_qos.ltq_oss->lqo_penalty >> 10,
535                           ost->ltd_qos.ltq_weight >> 10);
536         }
537
538         RETURN(0);
539 }
540
541 void lod_qos_rr_init(struct lod_qos_rr *lqr)
542 {
543         spin_lock_init(&lqr->lqr_alloc);
544         lqr->lqr_dirty = 1;
545 }
546
547
548 #define LOV_QOS_EMPTY ((__u32)-1)
549
550 /**
551  * Calculate optimal round-robin order with regard to OSSes.
552  *
553  * Place all the OSTs from pool \a src_pool in a special array to be used for
554  * round-robin (RR) stripe allocation.  The placement algorithm interleaves
555  * OSTs from the different OSSs so that RR allocation can balance OSSs evenly.
556  * Resorts the targets when the number of active targets changes (because of
557  * a new target or activation/deactivation).
558  *
559  * \param[in] lod       LOD device
560  * \param[in] src_pool  OST pool
561  * \param[in] lqr       round-robin list
562  *
563  * \retval 0            on success
564  * \retval -ENOMEM      fails to allocate the array
565  */
566 static int lod_qos_calc_rr(struct lod_device *lod, struct ost_pool *src_pool,
567                            struct lod_qos_rr *lqr)
568 {
569         struct lod_qos_oss  *oss;
570         struct lod_tgt_desc *ost;
571         unsigned placed, real_count;
572         unsigned int i;
573         int rc;
574         ENTRY;
575
576         if (!lqr->lqr_dirty) {
577                 LASSERT(lqr->lqr_pool.op_size);
578                 RETURN(0);
579         }
580
581         /* Do actual allocation. */
582         down_write(&lod->lod_qos.lq_rw_sem);
583
584         /*
585          * Check again. While we were sleeping on @lq_rw_sem something could
586          * change.
587          */
588         if (!lqr->lqr_dirty) {
589                 LASSERT(lqr->lqr_pool.op_size);
590                 up_write(&lod->lod_qos.lq_rw_sem);
591                 RETURN(0);
592         }
593
594         real_count = src_pool->op_count;
595
596         /* Zero the pool array */
597         /* alloc_rr is holding a read lock on the pool, so nobody is adding/
598            deleting from the pool. The lq_rw_sem insures that nobody else
599            is reading. */
600         lqr->lqr_pool.op_count = real_count;
601         rc = lod_ost_pool_extend(&lqr->lqr_pool, real_count);
602         if (rc) {
603                 up_write(&lod->lod_qos.lq_rw_sem);
604                 RETURN(rc);
605         }
606         for (i = 0; i < lqr->lqr_pool.op_count; i++)
607                 lqr->lqr_pool.op_array[i] = LOV_QOS_EMPTY;
608
609         /* Place all the OSTs from 1 OSS at the same time. */
610         placed = 0;
611         list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
612                 int j = 0;
613
614                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
615                         int next;
616
617                         if (!cfs_bitmap_check(lod->lod_ost_bitmap,
618                                                 src_pool->op_array[i]))
619                                 continue;
620
621                         ost = OST_TGT(lod,src_pool->op_array[i]);
622                         LASSERT(ost && ost->ltd_ost);
623                         if (ost->ltd_qos.ltq_oss != oss)
624                                 continue;
625
626                         /* Evenly space these OSTs across arrayspace */
627                         next = j * lqr->lqr_pool.op_count / oss->lqo_ost_count;
628                         while (lqr->lqr_pool.op_array[next] != LOV_QOS_EMPTY)
629                                 next = (next + 1) % lqr->lqr_pool.op_count;
630
631                         lqr->lqr_pool.op_array[next] = src_pool->op_array[i];
632                         j++;
633                         placed++;
634                 }
635         }
636
637         lqr->lqr_dirty = 0;
638         up_write(&lod->lod_qos.lq_rw_sem);
639
640         if (placed != real_count) {
641                 /* This should never happen */
642                 LCONSOLE_ERROR_MSG(0x14e, "Failed to place all OSTs in the "
643                                    "round-robin list (%d of %d).\n",
644                                    placed, real_count);
645                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
646                         LCONSOLE(D_WARNING, "rr #%d ost idx=%d\n", i,
647                                  lqr->lqr_pool.op_array[i]);
648                 }
649                 lqr->lqr_dirty = 1;
650                 RETURN(-EAGAIN);
651         }
652
653 #if 0
654         for (i = 0; i < lqr->lqr_pool.op_count; i++)
655                 QOS_CONSOLE("rr #%d ost idx=%d\n", i, lqr->lqr_pool.op_array[i]);
656 #endif
657
658         RETURN(0);
659 }
660
661 /**
662  * Instantiate and declare creation of a new object.
663  *
664  * The function instantiates LU representation for a new object on the
665  * specified device. Also it declares an intention to create that
666  * object on the storage target.
667  *
668  * Note lu_object_anon() is used which is a trick with regard to LU/OSD
669  * infrastructure - in the existing precreation framework we can't assign FID
670  * at this moment, we do this later once a transaction is started. So the
671  * special method instantiates FID-less object in the cache and later it
672  * will get a FID and proper placement in LU cache.
673  *
674  * \param[in] env       execution environment for this thread
675  * \param[in] d         LOD device
676  * \param[in] ost_idx   OST target index where the object is being created
677  * \param[in] th        transaction handle
678  *
679  * \retval              object ptr on success, ERR_PTR() otherwise
680  */
681 static struct dt_object *lod_qos_declare_object_on(const struct lu_env *env,
682                                                    struct lod_device *d,
683                                                    __u32 ost_idx,
684                                                    struct thandle *th)
685 {
686         struct lod_tgt_desc *ost;
687         struct lu_object *o, *n;
688         struct lu_device *nd;
689         struct dt_object *dt;
690         int               rc;
691         ENTRY;
692
693         LASSERT(d);
694         LASSERT(ost_idx < d->lod_osts_size);
695         ost = OST_TGT(d,ost_idx);
696         LASSERT(ost);
697         LASSERT(ost->ltd_ost);
698
699         nd = &ost->ltd_ost->dd_lu_dev;
700
701         /*
702          * allocate anonymous object with zero fid, real fid
703          * will be assigned by OSP within transaction
704          * XXX: to be fixed with fully-functional OST fids
705          */
706         o = lu_object_anon(env, nd, NULL);
707         if (IS_ERR(o))
708                 GOTO(out, dt = ERR_PTR(PTR_ERR(o)));
709
710         n = lu_object_locate(o->lo_header, nd->ld_type);
711         if (unlikely(n == NULL)) {
712                 CERROR("can't find slice\n");
713                 lu_object_put(env, o);
714                 GOTO(out, dt = ERR_PTR(-EINVAL));
715         }
716
717         dt = container_of(n, struct dt_object, do_lu);
718
719         rc = lod_sub_declare_create(env, dt, NULL, NULL, NULL, th);
720         if (rc < 0) {
721                 CDEBUG(D_OTHER, "can't declare creation on #%u: %d\n",
722                        ost_idx, rc);
723                 lu_object_put(env, o);
724                 dt = ERR_PTR(rc);
725         }
726
727 out:
728         RETURN(dt);
729 }
730
731 /**
732  * Calculate a minimum acceptable stripe count.
733  *
734  * Return an acceptable stripe count depending on flag LOV_USES_DEFAULT_STRIPE:
735  * all stripes or 3/4 of stripes.
736  *
737  * \param[in] stripe_count      number of stripes requested
738  * \param[in] flags             0 or LOV_USES_DEFAULT_STRIPE
739  *
740  * \retval                      acceptable stripecount
741  */
742 static int min_stripe_count(__u32 stripe_count, int flags)
743 {
744         return (flags & LOV_USES_DEFAULT_STRIPE ?
745                 stripe_count - (stripe_count / 4) : stripe_count);
746 }
747
748 #define LOV_CREATE_RESEED_MULT 30
749 #define LOV_CREATE_RESEED_MIN  2000
750
751 /**
752  * Initialize temporary OST-in-use array.
753  *
754  * Allocate or extend the array used to mark targets already assigned to a new
755  * striping so they are not used more than once.
756  *
757  * \param[in] env       execution environment for this thread
758  * \param[in] stripes   number of items needed in the array
759  *
760  * \retval 0            on success
761  * \retval -ENOMEM      on error
762  */
763 static inline int lod_qos_ost_in_use_clear(const struct lu_env *env,
764                                            __u32 stripes)
765 {
766         struct lod_thread_info *info = lod_env_info(env);
767
768         if (info->lti_ea_store_size < sizeof(int) * stripes)
769                 lod_ea_store_resize(info, stripes * sizeof(int));
770         if (info->lti_ea_store_size < sizeof(int) * stripes) {
771                 CERROR("can't allocate memory for ost-in-use array\n");
772                 return -ENOMEM;
773         }
774         memset(info->lti_ea_store, -1, sizeof(int) * stripes);
775         return 0;
776 }
777
778 /**
779  * Remember a target in the array of used targets.
780  *
781  * Mark the given target as used for a new striping being created. The status
782  * of an OST in a striping can be checked with lod_qos_is_ost_used().
783  *
784  * \param[in] env       execution environment for this thread
785  * \param[in] idx       index in the array
786  * \param[in] ost       OST target index to mark as used
787  */
788 static inline void lod_qos_ost_in_use(const struct lu_env *env,
789                                       int idx, int ost)
790 {
791         struct lod_thread_info *info = lod_env_info(env);
792         int *osts = info->lti_ea_store;
793
794         LASSERT(info->lti_ea_store_size >= idx * sizeof(int));
795         osts[idx] = ost;
796 }
797
798 /**
799  * Check is OST used in a striping.
800  *
801  * Checks whether OST with the given index is marked as used in the temporary
802  * array (see lod_qos_ost_in_use()).
803  *
804  * \param[in] env       execution environment for this thread
805  * \param[in] ost       OST target index to check
806  * \param[in] stripes   the number of items used in the array already
807  *
808  * \retval 0            not used
809  * \retval 1            used
810  */
811 static int lod_qos_is_ost_used(const struct lu_env *env, int ost, __u32 stripes)
812 {
813         struct lod_thread_info *info = lod_env_info(env);
814         int *osts = info->lti_ea_store;
815         __u32 j;
816
817         for (j = 0; j < stripes; j++) {
818                 if (osts[j] == ost)
819                         return 1;
820         }
821         return 0;
822 }
823
824 /**
825  * Check is OST used in a composite layout
826  *
827  * \param[in] inuse     all inuse ost indexs
828  * \param[in] ost       OST target index to check
829  *
830  * \retval 0            not used
831  * \retval 1            used
832  */
833 static inline int lod_comp_is_ost_used(struct ost_pool *inuse, int ost)
834 {
835         __u32 j;
836         LASSERT(inuse != NULL);
837
838         if (inuse->op_size == 0)
839                 return 0;
840
841         LASSERT(inuse->op_count * sizeof(inuse->op_array[0]) <= inuse->op_size);
842         for (j = 0; j < inuse->op_count; j++) {
843                 if (inuse->op_array[j] == ost)
844                         return 1;
845         }
846         return 0;
847 }
848
849 /**
850  * Mark the given target as used for a composite layout
851  *
852  * \param[in] inuse     inuse ost index array
853  * \param[in] idx       index in the array
854  */
855 static inline void lod_comp_ost_in_use(struct ost_pool *inuse, int ost)
856 {
857         LASSERT(inuse != NULL);
858         if (inuse->op_size && !lod_comp_is_ost_used(inuse, ost)) {
859                 LASSERTF(inuse->op_count * sizeof(inuse->op_array[0]) <
860                          inuse->op_size,
861                          "count %d size %u", inuse->op_count, inuse->op_size);
862                 inuse->op_array[inuse->op_count] = ost;
863                 inuse->op_count++;
864         }
865 }
866
867 static int lod_check_and_reserve_ost(const struct lu_env *env,
868                                      struct lod_device *m,
869                                      struct obd_statfs *sfs, __u32 ost_idx,
870                                      __u32 speed, __u32 *s_idx,
871                                      struct dt_object **stripe,
872                                      struct thandle *th,
873                                      struct ost_pool *inuse)
874 {
875         struct dt_object   *o;
876         __u32 stripe_idx = *s_idx;
877         int rc;
878
879         rc = lod_statfs_and_check(env, m, ost_idx, sfs);
880         if (rc) {
881                 /* this OSP doesn't feel well */
882                 goto out_return;
883         }
884
885         /*
886          * We expect number of precreated objects in f_ffree at
887          * the first iteration, skip OSPs with no objects ready
888          */
889         if (sfs->os_fprecreated == 0 && speed == 0) {
890                 QOS_DEBUG("#%d: precreation is empty\n", ost_idx);
891                 goto out_return;
892         }
893
894         /*
895          * try to use another OSP if this one is degraded
896          */
897         if (sfs->os_state & OS_STATE_DEGRADED && speed < 2) {
898                 QOS_DEBUG("#%d: degraded\n", ost_idx);
899                 goto out_return;
900         }
901
902         /*
903          * try not allocate on OST which has been used by other
904          * component
905          */
906         if (speed == 0 && lod_comp_is_ost_used(inuse, ost_idx)) {
907                 QOS_DEBUG("#%d: used by other component\n", ost_idx);
908                 goto out_return;
909         }
910
911         /*
912          * do not put >1 objects on a single OST
913          */
914         if (lod_qos_is_ost_used(env, ost_idx, stripe_idx))
915                 goto out_return;
916
917         o = lod_qos_declare_object_on(env, m, ost_idx, th);
918         if (IS_ERR(o)) {
919                 CDEBUG(D_OTHER, "can't declare new object on #%u: %d\n",
920                        ost_idx, (int) PTR_ERR(o));
921                 rc = PTR_ERR(o);
922                 goto out_return;
923         }
924
925         /*
926          * We've successfully declared (reserved) an object
927          */
928         lod_qos_ost_in_use(env, stripe_idx, ost_idx);
929         lod_comp_ost_in_use(inuse, ost_idx);
930         stripe[stripe_idx] = o;
931         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_LOV_CREATE_RACE, 2);
932         stripe_idx++;
933         *s_idx = stripe_idx;
934
935 out_return:
936         return rc;
937 }
938
939 /**
940  * Allocate a striping using round-robin algorithm.
941  *
942  * Allocates a new striping using round-robin algorithm. The function refreshes
943  * all the internal structures (statfs cache, array of available OSTs sorted
944  * with regard to OSS, etc). The number of stripes required is taken from the
945  * object (must be prepared by the caller), but can change if the flag
946  * LOV_USES_DEFAULT_STRIPE is supplied. The caller should ensure nobody else
947  * is trying to create a striping on the object in parallel. All the internal
948  * structures (like pools, etc) are protected and no additional locking is
949  * required. The function succeeds even if a single stripe is allocated. To save
950  * time we give priority to targets which already have objects precreated.
951  * Full OSTs are skipped (see lod_qos_dev_is_full() for the details).
952  *
953  * \param[in] env       execution environment for this thread
954  * \param[in] lo        LOD object
955  * \param[out] stripe   striping created
956  * \param[in] flags     allocation flags (0 or LOV_USES_DEFAULT_STRIPE)
957  * \param[in] th        transaction handle
958  * \param[in] comp_idx  index of ldo_comp_entries
959  * \param[in|out] inuse array of inuse ost index
960  *
961  * \retval 0            on success
962  * \retval -ENOSPC      if not enough OSTs are found
963  * \retval negative     negated errno for other failures
964  */
965 static int lod_alloc_rr(const struct lu_env *env, struct lod_object *lo,
966                         struct dt_object **stripe, int flags,
967                         struct thandle *th, int comp_idx,
968                         struct ost_pool *inuse)
969 {
970         struct lod_layout_component *lod_comp;
971         struct lod_device *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
972         struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
973         struct pool_desc  *pool = NULL;
974         struct ost_pool   *osts;
975         struct lod_qos_rr *lqr;
976         unsigned int    i, array_idx;
977         __u32 ost_start_idx_temp;
978         __u32 stripe_idx = 0;
979         __u32 stripe_count, stripe_count_min, ost_idx;
980         int rc, speed = 0, ost_connecting = 0;
981         ENTRY;
982
983         LASSERT(lo->ldo_comp_cnt > comp_idx && lo->ldo_comp_entries != NULL);
984         lod_comp = &lo->ldo_comp_entries[comp_idx];
985         stripe_count = lod_comp->llc_stripe_count;
986         stripe_count_min = min_stripe_count(stripe_count, flags);
987
988         if (lod_comp->llc_pool != NULL)
989                 pool = lod_find_pool(m, lod_comp->llc_pool);
990
991         if (pool != NULL) {
992                 down_read(&pool_tgt_rw_sem(pool));
993                 osts = &(pool->pool_obds);
994                 lqr = &(pool->pool_rr);
995         } else {
996                 osts = &(m->lod_pool_info);
997                 lqr = &(m->lod_qos.lq_rr);
998         }
999
1000         rc = lod_qos_calc_rr(m, osts, lqr);
1001         if (rc)
1002                 GOTO(out, rc);
1003
1004         rc = lod_qos_ost_in_use_clear(env, stripe_count);
1005         if (rc)
1006                 GOTO(out, rc);
1007
1008         down_read(&m->lod_qos.lq_rw_sem);
1009         spin_lock(&lqr->lqr_alloc);
1010         if (--lqr->lqr_start_count <= 0) {
1011                 lqr->lqr_start_idx = cfs_rand() % osts->op_count;
1012                 lqr->lqr_start_count =
1013                         (LOV_CREATE_RESEED_MIN / max(osts->op_count, 1U) +
1014                          LOV_CREATE_RESEED_MULT) * max(osts->op_count, 1U);
1015         } else if (stripe_count_min >= osts->op_count ||
1016                         lqr->lqr_start_idx > osts->op_count) {
1017                 /* If we have allocated from all of the OSTs, slowly
1018                  * precess the next start if the OST/stripe count isn't
1019                  * already doing this for us. */
1020                 lqr->lqr_start_idx %= osts->op_count;
1021                 if (stripe_count > 1 && (osts->op_count % stripe_count) != 1)
1022                         ++lqr->lqr_offset_idx;
1023         }
1024         ost_start_idx_temp = lqr->lqr_start_idx;
1025
1026 repeat_find:
1027
1028         QOS_DEBUG("pool '%s' want %d start_idx %d start_count %d offset %d "
1029                   "active %d count %d\n",
1030                   lod_comp->llc_pool ? lod_comp->llc_pool : "",
1031                   stripe_count, lqr->lqr_start_idx, lqr->lqr_start_count,
1032                   lqr->lqr_offset_idx, osts->op_count, osts->op_count);
1033
1034         for (i = 0; i < osts->op_count && stripe_idx < stripe_count; i++) {
1035                 array_idx = (lqr->lqr_start_idx + lqr->lqr_offset_idx) %
1036                                 osts->op_count;
1037                 ++lqr->lqr_start_idx;
1038                 ost_idx = lqr->lqr_pool.op_array[array_idx];
1039
1040                 QOS_DEBUG("#%d strt %d act %d strp %d ary %d idx %d\n",
1041                           i, lqr->lqr_start_idx, /* XXX: active*/ 0,
1042                           stripe_idx, array_idx, ost_idx);
1043
1044                 if ((ost_idx == LOV_QOS_EMPTY) ||
1045                     !cfs_bitmap_check(m->lod_ost_bitmap, ost_idx))
1046                         continue;
1047
1048                 /* Fail Check before osc_precreate() is called
1049                    so we can only 'fail' single OSC. */
1050                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
1051                         continue;
1052
1053                 spin_unlock(&lqr->lqr_alloc);
1054                 rc = lod_check_and_reserve_ost(env, m, sfs, ost_idx, speed,
1055                                                &stripe_idx, stripe, th, inuse);
1056                 spin_lock(&lqr->lqr_alloc);
1057
1058                 if (rc != 0 && OST_TGT(m, ost_idx)->ltd_connecting)
1059                         ost_connecting = 1;
1060         }
1061         if ((speed < 2) && (stripe_idx < stripe_count_min)) {
1062                 /* Try again, allowing slower OSCs */
1063                 speed++;
1064                 lqr->lqr_start_idx = ost_start_idx_temp;
1065
1066                 ost_connecting = 0;
1067                 goto repeat_find;
1068         }
1069
1070         spin_unlock(&lqr->lqr_alloc);
1071         up_read(&m->lod_qos.lq_rw_sem);
1072
1073         if (stripe_idx) {
1074                 lod_comp->llc_stripe_count = stripe_idx;
1075                 /* at least one stripe is allocated */
1076                 rc = 0;
1077         } else {
1078                 /* nobody provided us with a single object */
1079                 if (ost_connecting)
1080                         rc = -EINPROGRESS;
1081                 else
1082                         rc = -ENOSPC;
1083         }
1084
1085 out:
1086         if (pool != NULL) {
1087                 up_read(&pool_tgt_rw_sem(pool));
1088                 /* put back ref got by lod_find_pool() */
1089                 lod_pool_putref(pool);
1090         }
1091
1092         RETURN(rc);
1093 }
1094
1095 /**
1096  * Allocate a specific striping layout on a user defined set of OSTs.
1097  *
1098  * Allocates new striping using the OST index range provided by the data from
1099  * the lmm_obejcts contained in the lov_user_md passed to this method. Full
1100  * OSTs are not considered. The exact order of OSTs requested by the user
1101  * is respected as much as possible depending on OST status. The number of
1102  * stripes needed and stripe offset are taken from the object. If that number
1103  * can not be met, then the function returns a failure and then it's the
1104  * caller's responsibility to release the stripes allocated. All the internal
1105  * structures are protected, but no concurrent allocation is allowed on the
1106  * same objects.
1107  *
1108  * \param[in] env       execution environment for this thread
1109  * \param[in] lo        LOD object
1110  * \param[out] stripe   striping created
1111  * \param[in] th        transaction handle
1112  * \param[in] comp_idx  index of ldo_comp_entries
1113  * \param[in|out] inuse array of inuse ost index
1114  *
1115  * \retval 0            on success
1116  * \retval -ENODEV      OST index does not exist on file system
1117  * \retval -EINVAL      requested OST index is invalid
1118  * \retval negative     negated errno on error
1119  */
1120 static int lod_alloc_ost_list(const struct lu_env *env, struct lod_object *lo,
1121                               struct dt_object **stripe, struct thandle *th,
1122                               int comp_idx, struct ost_pool *inuse)
1123 {
1124         struct lod_layout_component *lod_comp;
1125         struct lod_device       *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1126         struct obd_statfs       *sfs = &lod_env_info(env)->lti_osfs;
1127         struct dt_object        *o;
1128         unsigned int            array_idx = 0;
1129         int                     stripe_count = 0;
1130         int                     i;
1131         int                     rc = -EINVAL;
1132         ENTRY;
1133
1134         /* for specific OSTs layout */
1135         LASSERT(lo->ldo_comp_cnt > comp_idx && lo->ldo_comp_entries != NULL);
1136         lod_comp = &lo->ldo_comp_entries[comp_idx];
1137         LASSERT(lod_comp->llc_ostlist.op_array);
1138
1139         rc = lod_qos_ost_in_use_clear(env, lod_comp->llc_stripe_count);
1140         if (rc < 0)
1141                 RETURN(rc);
1142
1143         for (i = 0; i < lod_comp->llc_stripe_count; i++) {
1144                 if (lod_comp->llc_ostlist.op_array[i] ==
1145                     lod_comp->llc_stripe_offset) {
1146                         array_idx = i;
1147                         break;
1148                 }
1149         }
1150         if (i == lod_comp->llc_stripe_count) {
1151                 CDEBUG(D_OTHER,
1152                        "%s: start index %d not in the specified list of OSTs\n",
1153                        lod2obd(m)->obd_name, lod_comp->llc_stripe_offset);
1154                 RETURN(-EINVAL);
1155         }
1156
1157         for (i = 0; i < lod_comp->llc_stripe_count;
1158              i++, array_idx = (array_idx + 1) % lod_comp->llc_stripe_count) {
1159                 __u32 ost_idx = lod_comp->llc_ostlist.op_array[array_idx];
1160
1161                 if (!cfs_bitmap_check(m->lod_ost_bitmap, ost_idx)) {
1162                         rc = -ENODEV;
1163                         break;
1164                 }
1165
1166                 /*
1167                  * do not put >1 objects on a single OST
1168                  */
1169                 if (lod_qos_is_ost_used(env, ost_idx, stripe_count)) {
1170                         rc = -EINVAL;
1171                         break;
1172                 }
1173
1174                 rc = lod_statfs_and_check(env, m, ost_idx, sfs);
1175                 if (rc < 0) /* this OSP doesn't feel well */
1176                         break;
1177
1178                 o = lod_qos_declare_object_on(env, m, ost_idx, th);
1179                 if (IS_ERR(o)) {
1180                         rc = PTR_ERR(o);
1181                         CDEBUG(D_OTHER,
1182                                "%s: can't declare new object on #%u: %d\n",
1183                                lod2obd(m)->obd_name, ost_idx, rc);
1184                         break;
1185                 }
1186
1187                 /*
1188                  * We've successfully declared (reserved) an object
1189                  */
1190                 lod_qos_ost_in_use(env, stripe_count, ost_idx);
1191                 lod_comp_ost_in_use(inuse, ost_idx);
1192                 stripe[stripe_count] = o;
1193                 stripe_count++;
1194         }
1195
1196         RETURN(rc);
1197 }
1198
1199 /**
1200  * Allocate a striping on a predefined set of OSTs.
1201  *
1202  * Allocates new layout starting from OST index in lo->ldo_stripe_offset.
1203  * Full OSTs are not considered. The exact order of OSTs is not important and
1204  * varies depending on OST status. The allocation procedure prefers the targets
1205  * with precreated objects ready. The number of stripes needed and stripe
1206  * offset are taken from the object. If that number cannot be met, then the
1207  * function returns an error and then it's the caller's responsibility to
1208  * release the stripes allocated. All the internal structures are protected,
1209  * but no concurrent allocation is allowed on the same objects.
1210  *
1211  * \param[in] env       execution environment for this thread
1212  * \param[in] lo        LOD object
1213  * \param[out] stripe   striping created
1214  * \param[in] flags     not used
1215  * \param[in] th        transaction handle
1216  * \param[in] comp_idx  index of ldo_comp_entries
1217  * \param[in|out]inuse  array of inuse ost index
1218  *
1219  * \retval 0            on success
1220  * \retval -ENOSPC      if no OST objects are available at all
1221  * \retval -EFBIG       if not enough OST objects are found
1222  * \retval -EINVAL      requested offset is invalid
1223  * \retval negative     errno on failure
1224  */
1225 static int lod_alloc_specific(const struct lu_env *env, struct lod_object *lo,
1226                               struct dt_object **stripe, int flags,
1227                               struct thandle *th, int comp_idx,
1228                               struct ost_pool *inuse)
1229 {
1230         struct lod_layout_component *lod_comp;
1231         struct lod_device *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1232         struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
1233         struct dt_object  *o;
1234         __u32              ost_idx;
1235         unsigned int       i, array_idx, ost_count;
1236         int                rc, stripe_num = 0;
1237         int                speed = 0;
1238         struct pool_desc  *pool = NULL;
1239         struct ost_pool   *osts;
1240         ENTRY;
1241
1242         LASSERT(lo->ldo_comp_cnt > comp_idx && lo->ldo_comp_entries != NULL);
1243         lod_comp = &lo->ldo_comp_entries[comp_idx];
1244
1245         rc = lod_qos_ost_in_use_clear(env, lod_comp->llc_stripe_count);
1246         if (rc)
1247                 GOTO(out, rc);
1248
1249         if (lod_comp->llc_pool != NULL)
1250                 pool = lod_find_pool(m, lod_comp->llc_pool);
1251
1252         if (pool != NULL) {
1253                 down_read(&pool_tgt_rw_sem(pool));
1254                 osts = &(pool->pool_obds);
1255         } else {
1256                 osts = &(m->lod_pool_info);
1257         }
1258
1259         ost_count = osts->op_count;
1260
1261 repeat_find:
1262         /* search loi_ost_idx in ost array */
1263         array_idx = 0;
1264         for (i = 0; i < ost_count; i++) {
1265                 if (osts->op_array[i] == lod_comp->llc_stripe_offset) {
1266                         array_idx = i;
1267                         break;
1268                 }
1269         }
1270         if (i == ost_count) {
1271                 CERROR("Start index %d not found in pool '%s'\n",
1272                        lod_comp->llc_stripe_offset,
1273                        lod_comp->llc_pool ? lod_comp->llc_pool : "");
1274                 GOTO(out, rc = -EINVAL);
1275         }
1276
1277         for (i = 0; i < ost_count;
1278                         i++, array_idx = (array_idx + 1) % ost_count) {
1279                 ost_idx = osts->op_array[array_idx];
1280
1281                 if (!cfs_bitmap_check(m->lod_ost_bitmap, ost_idx))
1282                         continue;
1283
1284                 /* Fail Check before osc_precreate() is called
1285                    so we can only 'fail' single OSC. */
1286                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
1287                         continue;
1288
1289                 /*
1290                  * do not put >1 objects on a single OST
1291                  */
1292                 if (lod_qos_is_ost_used(env, ost_idx, stripe_num))
1293                         continue;
1294
1295                 /*
1296                  * try not allocate on the OST used by other component
1297                  */
1298                 if (speed == 0 && i != 0 &&
1299                     lod_comp_is_ost_used(inuse, ost_idx))
1300                         continue;
1301
1302                 /* Drop slow OSCs if we can, but not for requested start idx.
1303                  *
1304                  * This means "if OSC is slow and it is not the requested
1305                  * start OST, then it can be skipped, otherwise skip it only
1306                  * if it is inactive/recovering/out-of-space." */
1307
1308                 rc = lod_statfs_and_check(env, m, ost_idx, sfs);
1309                 if (rc) {
1310                         /* this OSP doesn't feel well */
1311                         continue;
1312                 }
1313
1314                 /*
1315                  * We expect number of precreated objects at the first
1316                  * iteration.  Skip OSPs with no objects ready.  Don't apply
1317                  * this logic to OST specified with stripe_offset.
1318                  */
1319                 if (i != 0 && sfs->os_fprecreated == 0 && speed == 0)
1320                         continue;
1321
1322                 o = lod_qos_declare_object_on(env, m, ost_idx, th);
1323                 if (IS_ERR(o)) {
1324                         CDEBUG(D_OTHER, "can't declare new object on #%u: %d\n",
1325                                ost_idx, (int) PTR_ERR(o));
1326                         continue;
1327                 }
1328
1329                 /*
1330                  * We've successfully declared (reserved) an object
1331                  */
1332                 lod_qos_ost_in_use(env, stripe_num, ost_idx);
1333                 lod_comp_ost_in_use(inuse, ost_idx);
1334                 stripe[stripe_num] = o;
1335                 stripe_num++;
1336
1337                 /* We have enough stripes */
1338                 if (stripe_num == lod_comp->llc_stripe_count)
1339                         GOTO(out, rc = 0);
1340         }
1341         if (speed < 2) {
1342                 /* Try again, allowing slower OSCs */
1343                 speed++;
1344                 goto repeat_find;
1345         }
1346
1347         /* If we were passed specific striping params, then a failure to
1348          * meet those requirements is an error, since we can't reallocate
1349          * that memory (it might be part of a larger array or something).
1350          */
1351         CERROR("can't lstripe objid "DFID": have %d want %u\n",
1352                PFID(lu_object_fid(lod2lu_obj(lo))), stripe_num,
1353                lod_comp->llc_stripe_count);
1354         rc = stripe_num == 0 ? -ENOSPC : -EFBIG;
1355 out:
1356         if (pool != NULL) {
1357                 up_read(&pool_tgt_rw_sem(pool));
1358                 /* put back ref got by lod_find_pool() */
1359                 lod_pool_putref(pool);
1360         }
1361
1362         RETURN(rc);
1363 }
1364
1365 /**
1366  * Check whether QoS allocation should be used.
1367  *
1368  * A simple helper to decide when QoS allocation should be used:
1369  * if it's just a single available target or the used space is
1370  * evenly distributed among the targets at the moment, then QoS
1371  * allocation algorithm should not be used.
1372  *
1373  * \param[in] lod       LOD device
1374  *
1375  * \retval 0            should not be used
1376  * \retval 1            should be used
1377  */
1378 static inline int lod_qos_is_usable(struct lod_device *lod)
1379 {
1380 #ifdef FORCE_QOS
1381         /* to be able to debug QoS code */
1382         return 1;
1383 #endif
1384
1385         /* Detect -EAGAIN early, before expensive lock is taken. */
1386         if (!lod->lod_qos.lq_dirty && lod->lod_qos.lq_same_space)
1387                 return 0;
1388
1389         if (lod->lod_desc.ld_active_tgt_count < 2)
1390                 return 0;
1391
1392         return 1;
1393 }
1394
1395 /**
1396  * Allocate a striping using an algorithm with weights.
1397  *
1398  * The function allocates OST objects to create a striping. The algorithm
1399  * used is based on weights (currently only using the free space), and it's
1400  * trying to ensure the space is used evenly by OSTs and OSSs. The striping
1401  * configuration (# of stripes, offset, pool) is taken from the object and
1402  * is prepared by the caller.
1403  *
1404  * If LOV_USES_DEFAULT_STRIPE is not passed and prepared configuration can't
1405  * be met due to too few OSTs, then allocation fails. If the flag is passed
1406  * fewer than 3/4 of the requested number of stripes can be allocated, then
1407  * allocation fails.
1408  *
1409  * No concurrent allocation is allowed on the object and this must be ensured
1410  * by the caller. All the internal structures are protected by the function.
1411  *
1412  * The algorithm has two steps: find available OSTs and calculate their
1413  * weights, then select the OSTs with their weights used as the probability.
1414  * An OST with a higher weight is proportionately more likely to be selected
1415  * than one with a lower weight.
1416  *
1417  * \param[in] env       execution environment for this thread
1418  * \param[in] lo        LOD object
1419  * \param[out] stripe   striping created
1420  * \param[in] flags     0 or LOV_USES_DEFAULT_STRIPE
1421  * \param[in] th        transaction handle
1422  * \param[in] comp_idx  index of ldo_comp_entries
1423  * \param[in|out]inuse  array of inuse ost index
1424  *
1425  * \retval 0            on success
1426  * \retval -EAGAIN      not enough OSTs are found for specified stripe count
1427  * \retval -EINVAL      requested OST index is invalid
1428  * \retval negative     errno on failure
1429  */
1430 static int lod_alloc_qos(const struct lu_env *env, struct lod_object *lo,
1431                          struct dt_object **stripe, int flags,
1432                          struct thandle *th, int comp_idx,
1433                          struct ost_pool *inuse)
1434 {
1435         struct lod_layout_component *lod_comp;
1436         struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1437         struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
1438         struct lod_tgt_desc *ost;
1439         struct dt_object *o;
1440         __u64 total_weight = 0;
1441         struct pool_desc *pool = NULL;
1442         struct ost_pool *osts;
1443         unsigned int i;
1444         __u32 nfound, good_osts, stripe_count, stripe_count_min;
1445         __u32 inuse_old_count = inuse->op_count;
1446         int rc = 0;
1447         ENTRY;
1448
1449         LASSERT(lo->ldo_comp_cnt > comp_idx && lo->ldo_comp_entries != NULL);
1450         lod_comp = &lo->ldo_comp_entries[comp_idx];
1451         stripe_count = lod_comp->llc_stripe_count;
1452         stripe_count_min = min_stripe_count(stripe_count, flags);
1453         if (stripe_count_min < 1)
1454                 RETURN(-EINVAL);
1455
1456         if (lod_comp->llc_pool != NULL)
1457                 pool = lod_find_pool(lod, lod_comp->llc_pool);
1458
1459         if (pool != NULL) {
1460                 down_read(&pool_tgt_rw_sem(pool));
1461                 osts = &(pool->pool_obds);
1462         } else {
1463                 osts = &(lod->lod_pool_info);
1464         }
1465
1466         /* Detect -EAGAIN early, before expensive lock is taken. */
1467         if (!lod_qos_is_usable(lod))
1468                 GOTO(out_nolock, rc = -EAGAIN);
1469
1470         /* Do actual allocation, use write lock here. */
1471         down_write(&lod->lod_qos.lq_rw_sem);
1472
1473         /*
1474          * Check again, while we were sleeping on @lq_rw_sem things could
1475          * change.
1476          */
1477         if (!lod_qos_is_usable(lod))
1478                 GOTO(out, rc = -EAGAIN);
1479
1480         rc = lod_qos_calc_ppo(lod);
1481         if (rc)
1482                 GOTO(out, rc);
1483
1484         rc = lod_qos_ost_in_use_clear(env, lod_comp->llc_stripe_count);
1485         if (rc)
1486                 GOTO(out, rc);
1487
1488         good_osts = 0;
1489         /* Find all the OSTs that are valid stripe candidates */
1490         for (i = 0; i < osts->op_count; i++) {
1491                 if (!cfs_bitmap_check(lod->lod_ost_bitmap, osts->op_array[i]))
1492                         continue;
1493
1494                 ost = OST_TGT(lod, osts->op_array[i]);
1495                 ost->ltd_qos.ltq_usable = 0;
1496
1497                 rc = lod_statfs_and_check(env, lod, osts->op_array[i], sfs);
1498                 if (rc) {
1499                         /* this OSP doesn't feel well */
1500                         continue;
1501                 }
1502
1503                 if (sfs->os_state & OS_STATE_DEGRADED)
1504                         continue;
1505
1506                 /* Fail Check before osc_precreate() is called
1507                    so we can only 'fail' single OSC. */
1508                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) &&
1509                                    osts->op_array[i] == 0)
1510                         continue;
1511
1512                 ost->ltd_qos.ltq_usable = 1;
1513                 lod_qos_calc_weight(lod, osts->op_array[i]);
1514                 total_weight += ost->ltd_qos.ltq_weight;
1515
1516                 good_osts++;
1517         }
1518
1519         QOS_DEBUG("found %d good osts\n", good_osts);
1520
1521         if (good_osts < stripe_count_min)
1522                 GOTO(out, rc = -EAGAIN);
1523
1524         /* We have enough osts */
1525         if (good_osts < stripe_count)
1526                 stripe_count = good_osts;
1527
1528         /* Find enough OSTs with weighted random allocation. */
1529         nfound = 0;
1530         while (nfound < stripe_count) {
1531                 __u64 rand, cur_weight;
1532
1533                 cur_weight = 0;
1534                 rc = -ENOSPC;
1535
1536                 if (total_weight) {
1537 #if BITS_PER_LONG == 32
1538                         rand = cfs_rand() % (unsigned)total_weight;
1539                         /* If total_weight > 32-bit, first generate the high
1540                          * 32 bits of the random number, then add in the low
1541                          * 32 bits (truncated to the upper limit, if needed) */
1542                         if (total_weight > 0xffffffffULL)
1543                                 rand = (__u64)(cfs_rand() %
1544                                         (unsigned)(total_weight >> 32)) << 32;
1545                         else
1546                                 rand = 0;
1547
1548                         if (rand == (total_weight & 0xffffffff00000000ULL))
1549                                 rand |= cfs_rand() % (unsigned)total_weight;
1550                         else
1551                                 rand |= cfs_rand();
1552
1553 #else
1554                         rand = ((__u64)cfs_rand() << 32 | cfs_rand()) %
1555                                 total_weight;
1556 #endif
1557                 } else {
1558                         rand = 0;
1559                 }
1560
1561                 /* On average, this will hit larger-weighted OSTs more often.
1562                  * 0-weight OSTs will always get used last (only when rand=0) */
1563                 for (i = 0; i < osts->op_count; i++) {
1564                         __u32 idx = osts->op_array[i];
1565
1566                         if (!cfs_bitmap_check(lod->lod_ost_bitmap, idx))
1567                                 continue;
1568
1569                         ost = OST_TGT(lod, idx);
1570
1571                         if (!ost->ltd_qos.ltq_usable)
1572                                 continue;
1573
1574                         cur_weight += ost->ltd_qos.ltq_weight;
1575                         QOS_DEBUG("stripe_count=%d nfound=%d cur_weight=%llu "
1576                                   "rand=%llu total_weight=%llu\n",
1577                                   stripe_count, nfound, cur_weight, rand,
1578                                   total_weight);
1579
1580                         if (cur_weight < rand)
1581                                 continue;
1582
1583                         QOS_DEBUG("stripe=%d to idx=%d\n", nfound, idx);
1584                         /*
1585                          * do not put >1 objects on a single OST
1586                          */
1587                         if (lod_qos_is_ost_used(env, idx, nfound) ||
1588                             lod_comp_is_ost_used(inuse, idx))
1589                                 continue;
1590
1591                         o = lod_qos_declare_object_on(env, lod, idx, th);
1592                         if (IS_ERR(o)) {
1593                                 QOS_DEBUG("can't declare object on #%u: %d\n",
1594                                           idx, (int) PTR_ERR(o));
1595                                 continue;
1596                         }
1597
1598                         lod_qos_ost_in_use(env, nfound, idx);
1599                         lod_comp_ost_in_use(inuse, idx);
1600                         stripe[nfound++] = o;
1601                         lod_qos_used(lod, osts, idx, &total_weight);
1602                         rc = 0;
1603                         break;
1604                 }
1605
1606                 if (rc) {
1607                         /* no OST found on this iteration, give up */
1608                         break;
1609                 }
1610         }
1611
1612         if (unlikely(nfound != stripe_count)) {
1613                 /*
1614                  * when the decision to use weighted algorithm was made
1615                  * we had enough appropriate OSPs, but this state can
1616                  * change anytime (no space on OST, broken connection, etc)
1617                  * so it's possible OSP won't be able to provide us with
1618                  * an object due to just changed state
1619                  */
1620                 QOS_DEBUG("%s: wanted %d objects, found only %d\n",
1621                           lod2obd(lod)->obd_name, stripe_count, nfound);
1622                 for (i = 0; i < nfound; i++) {
1623                         LASSERT(stripe[i] != NULL);
1624                         dt_object_put(env, stripe[i]);
1625                         stripe[i] = NULL;
1626                 }
1627                 LASSERTF(nfound <= inuse->op_count,
1628                          "nfound:%d, op_count:%u\n", nfound, inuse->op_count);
1629                 inuse->op_count = inuse_old_count;
1630
1631                 /* makes sense to rebalance next time */
1632                 lod->lod_qos.lq_dirty = 1;
1633                 lod->lod_qos.lq_same_space = 0;
1634
1635                 rc = -EAGAIN;
1636         }
1637
1638 out:
1639         up_write(&lod->lod_qos.lq_rw_sem);
1640
1641 out_nolock:
1642         if (pool != NULL) {
1643                 up_read(&pool_tgt_rw_sem(pool));
1644                 /* put back ref got by lod_find_pool() */
1645                 lod_pool_putref(pool);
1646         }
1647
1648         RETURN(rc);
1649 }
1650
1651 /**
1652  * Find largest stripe count the caller can use.
1653  *
1654  * Find the maximal possible stripe count not greater than \a stripe_count.
1655  * Sometimes suggested stripecount can't be reached for a number of reasons:
1656  * lack of enough active OSTs or the backend does not support EAs that large.
1657  * If the passed one is 0, then the filesystem's default one is used.
1658  *
1659  * \param[in] lod       LOD device
1660  * \param[in] lo        The lod_object
1661  * \param[in] stripe_count      count the caller would like to use
1662  *
1663  * \retval              the maximum usable stripe count
1664  */
1665 __u16 lod_get_stripe_count(struct lod_device *lod, struct lod_object *lo,
1666                            __u16 stripe_count)
1667 {
1668         __u32 max_stripes = LOV_MAX_STRIPE_COUNT_OLD;
1669
1670         if (!stripe_count)
1671                 stripe_count = lod->lod_desc.ld_default_stripe_count;
1672         if (stripe_count > lod->lod_desc.ld_active_tgt_count)
1673                 stripe_count = lod->lod_desc.ld_active_tgt_count;
1674         if (!stripe_count)
1675                 stripe_count = 1;
1676
1677         /* stripe count is based on whether OSD can handle larger EA sizes */
1678         if (lod->lod_osd_max_easize > 0) {
1679                 unsigned int easize = lod->lod_osd_max_easize;
1680                 int i;
1681
1682                 if (lo->ldo_is_composite) {
1683                         struct lod_layout_component *lod_comp;
1684                         unsigned int header_sz = sizeof(struct lov_comp_md_v1);
1685
1686                         header_sz += sizeof(struct lov_comp_md_entry_v1) *
1687                                         lo->ldo_comp_cnt;
1688                         for (i = 0; i < lo->ldo_comp_cnt; i++) {
1689                                 lod_comp = &lo->ldo_comp_entries[i];
1690                                 if (lod_comp->llc_flags & LCME_FL_INIT)
1691                                         header_sz += lov_mds_md_size(
1692                                                 lod_comp->llc_stripe_count,
1693                                                 LOV_MAGIC_V3);
1694                         }
1695                         if (easize > header_sz)
1696                                 easize -= header_sz;
1697                         else
1698                                 easize = 0;
1699                 }
1700
1701                 max_stripes = lov_mds_md_max_stripe_count(easize, LOV_MAGIC_V3);
1702         }
1703
1704         return (stripe_count < max_stripes) ? stripe_count : max_stripes;
1705 }
1706
1707 /**
1708  * Create in-core respresentation for a fully-defined striping
1709  *
1710  * When the caller passes a fully-defined striping (i.e. everything including
1711  * OST object FIDs are defined), then we still need to instantiate LU-cache
1712  * with the objects representing the stripes defined. This function completes
1713  * that task.
1714  *
1715  * \param[in] env       execution environment for this thread
1716  * \param[in] mo        LOD object
1717  * \param[in] buf       buffer containing the striping
1718  *
1719  * \retval 0            on success
1720  * \retval negative     negated errno on error
1721  */
1722 int lod_use_defined_striping(const struct lu_env *env,
1723                              struct lod_object *mo,
1724                              const struct lu_buf *buf)
1725 {
1726         struct lod_layout_component *lod_comp;
1727         struct lov_mds_md_v1   *v1 = buf->lb_buf;
1728         struct lov_mds_md_v3   *v3 = buf->lb_buf;
1729         struct lov_comp_md_v1  *comp_v1 = NULL;
1730         struct lov_ost_data_v1 *objs;
1731         __u32   magic;
1732         __u16   comp_cnt;
1733         __u16   mirror_cnt;
1734         int     rc = 0, i;
1735         ENTRY;
1736
1737         magic = le32_to_cpu(v1->lmm_magic) & ~LOV_MAGIC_DEFINED;
1738
1739         if (magic != LOV_MAGIC_V1 && magic != LOV_MAGIC_V3 &&
1740             magic != LOV_MAGIC_COMP_V1)
1741                 RETURN(-EINVAL);
1742
1743         if (magic == LOV_MAGIC_COMP_V1) {
1744                 comp_v1 = buf->lb_buf;
1745                 comp_cnt = le16_to_cpu(comp_v1->lcm_entry_count);
1746                 if (comp_cnt == 0)
1747                         RETURN(-EINVAL);
1748                 mirror_cnt = le16_to_cpu(comp_v1->lcm_mirror_count) + 1;
1749                 mo->ldo_flr_state = le16_to_cpu(comp_v1->lcm_flags) &
1750                                         LCM_FL_FLR_MASK;
1751                 mo->ldo_is_composite = 1;
1752         } else {
1753                 mo->ldo_is_composite = 0;
1754                 comp_cnt = 1;
1755                 mirror_cnt = 0;
1756         }
1757         mo->ldo_layout_gen = le16_to_cpu(v1->lmm_layout_gen);
1758
1759         rc = lod_alloc_comp_entries(mo, mirror_cnt, comp_cnt);
1760         if (rc)
1761                 RETURN(rc);
1762
1763         for (i = 0; i < comp_cnt; i++) {
1764                 struct lu_extent *ext;
1765                 char    *pool_name;
1766                 __u32   offs;
1767
1768                 lod_comp = &mo->ldo_comp_entries[i];
1769
1770                 if (mo->ldo_is_composite) {
1771                         offs = le32_to_cpu(comp_v1->lcm_entries[i].lcme_offset);
1772                         v1 = (struct lov_mds_md_v1 *)((char *)comp_v1 + offs);
1773                         magic = le32_to_cpu(v1->lmm_magic);
1774
1775                         ext = &comp_v1->lcm_entries[i].lcme_extent;
1776                         lod_comp->llc_extent.e_start =
1777                                 le64_to_cpu(ext->e_start);
1778                         lod_comp->llc_extent.e_end = le64_to_cpu(ext->e_end);
1779                         lod_comp->llc_flags =
1780                                 le32_to_cpu(comp_v1->lcm_entries[i].lcme_flags);
1781                         lod_comp->llc_id =
1782                                 le32_to_cpu(comp_v1->lcm_entries[i].lcme_id);
1783                         if (lod_comp->llc_id == LCME_ID_INVAL)
1784                                 GOTO(out, rc = -EINVAL);
1785                 }
1786
1787                 pool_name = NULL;
1788                 if (magic == LOV_MAGIC_V1) {
1789                         objs = &v1->lmm_objects[0];
1790                 } else if (magic == LOV_MAGIC_V3) {
1791                         objs = &v3->lmm_objects[0];
1792                         if (v3->lmm_pool_name[0] != '\0')
1793                                 pool_name = v3->lmm_pool_name;
1794                 } else {
1795                         CDEBUG(D_LAYOUT, "Invalid magic %x\n", magic);
1796                         GOTO(out, rc = -EINVAL);
1797                 }
1798
1799                 lod_comp->llc_pattern = le32_to_cpu(v1->lmm_pattern);
1800                 lod_comp->llc_stripe_size = le32_to_cpu(v1->lmm_stripe_size);
1801                 lod_comp->llc_stripe_count = le16_to_cpu(v1->lmm_stripe_count);
1802                 lod_comp->llc_layout_gen = le16_to_cpu(v1->lmm_layout_gen);
1803                 /**
1804                  * The stripe_offset of an uninit-ed component is stored in
1805                  * the lmm_layout_gen
1806                  */
1807                 if (mo->ldo_is_composite && !lod_comp_inited(lod_comp))
1808                         lod_comp->llc_stripe_offset = lod_comp->llc_layout_gen;
1809                 lod_obj_set_pool(mo, i, pool_name);
1810
1811                 if ((!mo->ldo_is_composite || lod_comp_inited(lod_comp)) &&
1812                     !(lod_comp->llc_pattern & LOV_PATTERN_F_RELEASED) &&
1813                     !(lod_comp->llc_pattern & LOV_PATTERN_MDT)) {
1814                         rc = lod_initialize_objects(env, mo, objs, i);
1815                         if (rc)
1816                                 GOTO(out, rc);
1817                 }
1818         }
1819
1820         rc = lod_fill_mirrors(mo);
1821         if (rc)
1822                 GOTO(out, rc);
1823 out:
1824         if (rc)
1825                 lod_object_free_striping(env, mo);
1826
1827         RETURN(rc);
1828 }
1829
1830 /**
1831  * Parse suggested striping configuration.
1832  *
1833  * The caller gets a suggested striping configuration from a number of sources
1834  * including per-directory default and applications. Then it needs to verify
1835  * the suggested striping is valid, apply missing bits and store the resulting
1836  * configuration in the object to be used by the allocator later. Must not be
1837  * called concurrently against the same object. It's OK to provide a
1838  * fully-defined striping.
1839  *
1840  * \param[in] env       execution environment for this thread
1841  * \param[in] lo        LOD object
1842  * \param[in] buf       buffer containing the striping
1843  *
1844  * \retval 0            on success
1845  * \retval negative     negated errno on error
1846  */
1847 int lod_qos_parse_config(const struct lu_env *env, struct lod_object *lo,
1848                          const struct lu_buf *buf)
1849 {
1850         struct lod_layout_component *lod_comp;
1851         struct lod_device       *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
1852         struct lov_desc         *desc = &d->lod_desc;
1853         struct lov_user_md_v1   *v1 = NULL;
1854         struct lov_user_md_v3   *v3 = NULL;
1855         struct lov_comp_md_v1   *comp_v1 = NULL;
1856         char    def_pool[LOV_MAXPOOLNAME + 1];
1857         __u32   magic;
1858         __u16   comp_cnt;
1859         __u16   mirror_cnt;
1860         int     i, rc;
1861         ENTRY;
1862
1863         if (buf == NULL || buf->lb_buf == NULL || buf->lb_len == 0)
1864                 RETURN(0);
1865
1866         memset(def_pool, 0, sizeof(def_pool));
1867         if (lo->ldo_comp_entries != NULL)
1868                 lod_layout_get_pool(lo->ldo_comp_entries, lo->ldo_comp_cnt,
1869                                     def_pool, sizeof(def_pool));
1870
1871         /* free default striping info */
1872         lod_free_comp_entries(lo);
1873
1874         rc = lod_verify_striping(d, lo, buf, false);
1875         if (rc)
1876                 RETURN(-EINVAL);
1877
1878         v3 = buf->lb_buf;
1879         v1 = buf->lb_buf;
1880         comp_v1 = buf->lb_buf;
1881         magic = v1->lmm_magic;
1882
1883         if (unlikely(le32_to_cpu(magic) & LOV_MAGIC_DEFINED)) {
1884                 /* try to use as fully defined striping */
1885                 rc = lod_use_defined_striping(env, lo, buf);
1886                 RETURN(rc);
1887         }
1888
1889         switch (magic) {
1890         case __swab32(LOV_USER_MAGIC_V1):
1891                 lustre_swab_lov_user_md_v1(v1);
1892                 magic = v1->lmm_magic;
1893                 /* fall through */
1894         case LOV_USER_MAGIC_V1:
1895                 break;
1896         case __swab32(LOV_USER_MAGIC_V3):
1897                 lustre_swab_lov_user_md_v3(v3);
1898                 magic = v3->lmm_magic;
1899                 /* fall through */
1900         case LOV_USER_MAGIC_V3:
1901                 break;
1902         case __swab32(LOV_USER_MAGIC_SPECIFIC):
1903                 lustre_swab_lov_user_md_v3(v3);
1904                 lustre_swab_lov_user_md_objects(v3->lmm_objects,
1905                                                 v3->lmm_stripe_count);
1906                 magic = v3->lmm_magic;
1907                 /* fall through */
1908         case LOV_USER_MAGIC_SPECIFIC:
1909                 break;
1910         case __swab32(LOV_USER_MAGIC_COMP_V1):
1911                 lustre_swab_lov_comp_md_v1(comp_v1);
1912                 magic = comp_v1->lcm_magic;
1913                 /* fall trhough */
1914         case LOV_USER_MAGIC_COMP_V1:
1915                 break;
1916         default:
1917                 CERROR("%s: unrecognized magic %X\n",
1918                        lod2obd(d)->obd_name, magic);
1919                 RETURN(-EINVAL);
1920         }
1921
1922         lustre_print_user_md(D_OTHER, v1, "parse config");
1923
1924         if (magic == LOV_USER_MAGIC_COMP_V1) {
1925                 comp_cnt = comp_v1->lcm_entry_count;
1926                 if (comp_cnt == 0)
1927                         RETURN(-EINVAL);
1928                 mirror_cnt =  comp_v1->lcm_mirror_count + 1;
1929                 if (mirror_cnt > 1)
1930                         lo->ldo_flr_state = LCM_FL_RDONLY;
1931                 lo->ldo_is_composite = 1;
1932         } else {
1933                 comp_cnt = 1;
1934                 mirror_cnt = 0;
1935                 lo->ldo_is_composite = 0;
1936         }
1937
1938         rc = lod_alloc_comp_entries(lo, mirror_cnt, comp_cnt);
1939         if (rc)
1940                 RETURN(rc);
1941
1942         LASSERT(lo->ldo_comp_entries);
1943
1944         for (i = 0; i < comp_cnt; i++) {
1945                 struct pool_desc        *pool;
1946                 struct lu_extent        *ext;
1947                 char    *pool_name;
1948
1949                 lod_comp = &lo->ldo_comp_entries[i];
1950
1951                 if (lo->ldo_is_composite) {
1952                         v1 = (struct lov_user_md *)((char *)comp_v1 +
1953                                         comp_v1->lcm_entries[i].lcme_offset);
1954                         ext = &comp_v1->lcm_entries[i].lcme_extent;
1955                         lod_comp->llc_extent = *ext;
1956                         lod_comp->llc_flags =
1957                                 comp_v1->lcm_entries[i].lcme_flags &
1958                                         LCME_USER_FLAGS;
1959                 }
1960
1961                 pool_name = NULL;
1962                 if (v1->lmm_magic == LOV_USER_MAGIC_V3 ||
1963                     v1->lmm_magic == LOV_USER_MAGIC_SPECIFIC) {
1964                         int j;
1965
1966                         v3 = (struct lov_user_md_v3 *)v1;
1967                         if (v3->lmm_pool_name[0] != '\0')
1968                                 pool_name = v3->lmm_pool_name;
1969
1970                         if (v3->lmm_magic == LOV_USER_MAGIC_SPECIFIC) {
1971                                 if (v3->lmm_stripe_offset == LOV_OFFSET_DEFAULT)
1972                                         v3->lmm_stripe_offset =
1973                                                 v3->lmm_objects[0].l_ost_idx;
1974
1975                                 /* copy ost list from lmm */
1976                                 lod_comp->llc_ostlist.op_count =
1977                                         v3->lmm_stripe_count;
1978                                 lod_comp->llc_ostlist.op_size =
1979                                         v3->lmm_stripe_count * sizeof(__u32);
1980                                 OBD_ALLOC(lod_comp->llc_ostlist.op_array,
1981                                           lod_comp->llc_ostlist.op_size);
1982                                 if (!lod_comp->llc_ostlist.op_array)
1983                                         GOTO(free_comp, rc = -ENOMEM);
1984
1985                                 for (j = 0; j < v3->lmm_stripe_count; j++)
1986                                         lod_comp->llc_ostlist.op_array[j] =
1987                                                 v3->lmm_objects[j].l_ost_idx;
1988                         }
1989                 }
1990
1991                 if (pool_name == NULL && def_pool[0] != '\0')
1992                         pool_name = def_pool;
1993
1994                 if (v1->lmm_pattern == 0)
1995                         v1->lmm_pattern = LOV_PATTERN_RAID0;
1996                 if (lov_pattern(v1->lmm_pattern) != LOV_PATTERN_RAID0 &&
1997                     lov_pattern(v1->lmm_pattern) != LOV_PATTERN_MDT) {
1998                         CDEBUG(D_LAYOUT, "%s: invalid pattern: %x\n",
1999                                lod2obd(d)->obd_name, v1->lmm_pattern);
2000                         GOTO(free_comp, rc = -EINVAL);
2001                 }
2002
2003                 lod_comp->llc_pattern = v1->lmm_pattern;
2004                 lod_comp->llc_stripe_size = desc->ld_default_stripe_size;
2005                 if (v1->lmm_stripe_size)
2006                         lod_comp->llc_stripe_size = v1->lmm_stripe_size;
2007
2008                 lod_comp->llc_stripe_count = desc->ld_default_stripe_count;
2009                 if (v1->lmm_stripe_count ||
2010                     lov_pattern(v1->lmm_pattern) == LOV_PATTERN_MDT)
2011                         lod_comp->llc_stripe_count = v1->lmm_stripe_count;
2012
2013                 lod_comp->llc_stripe_offset = v1->lmm_stripe_offset;
2014                 lod_obj_set_pool(lo, i, pool_name);
2015
2016                 LASSERT(ergo(lov_pattern(lod_comp->llc_pattern) ==
2017                              LOV_PATTERN_MDT, lod_comp->llc_stripe_count == 0));
2018
2019                 if (pool_name == NULL)
2020                         continue;
2021
2022                 /* In the function below, .hs_keycmp resolves to
2023                  * pool_hashkey_keycmp() */
2024                 /* coverity[overrun-buffer-val] */
2025                 pool = lod_find_pool(d, pool_name);
2026                 if (pool == NULL)
2027                         continue;
2028
2029                 if (lod_comp->llc_stripe_offset != LOV_OFFSET_DEFAULT) {
2030                         rc = lod_check_index_in_pool(
2031                                         lod_comp->llc_stripe_offset, pool);
2032                         if (rc < 0) {
2033                                 lod_pool_putref(pool);
2034                                 CDEBUG(D_LAYOUT, "%s: invalid offset, %u\n",
2035                                        lod2obd(d)->obd_name,
2036                                        lod_comp->llc_stripe_offset);
2037                                 GOTO(free_comp, rc = -EINVAL);
2038                         }
2039                 }
2040
2041                 if (lod_comp->llc_stripe_count > pool_tgt_count(pool))
2042                         lod_comp->llc_stripe_count = pool_tgt_count(pool);
2043
2044                 lod_pool_putref(pool);
2045         }
2046
2047         RETURN(0);
2048
2049 free_comp:
2050         lod_free_comp_entries(lo);
2051         RETURN(rc);
2052 }
2053
2054 /**
2055  * Create a striping for an obejct.
2056  *
2057  * The function creates a new striping for the object. The function tries QoS
2058  * algorithm first unless free space is distributed evenly among OSTs, but
2059  * by default RR algorithm is preferred due to internal concurrency (QoS is
2060  * serialized). The caller must ensure no concurrent calls to the function
2061  * are made against the same object.
2062  *
2063  * \param[in] env       execution environment for this thread
2064  * \param[in] lo        LOD object
2065  * \param[in] attr      attributes OST objects will be declared with
2066  * \param[in] th        transaction handle
2067  * \param[in] comp_idx  index of ldo_comp_entries
2068  * \param[in|out] inuse array of inuse ost index
2069  *
2070  * \retval 0            on success
2071  * \retval negative     negated errno on error
2072  */
2073 int lod_qos_prep_create(const struct lu_env *env, struct lod_object *lo,
2074                         struct lu_attr *attr, struct thandle *th,
2075                         int comp_idx, struct ost_pool *inuse)
2076 {
2077         struct lod_layout_component *lod_comp;
2078         struct lod_device      *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
2079         struct dt_object      **stripe;
2080         int                     stripe_len;
2081         int                     flag = LOV_USES_ASSIGNED_STRIPE;
2082         int                     i, rc = 0;
2083         ENTRY;
2084
2085         LASSERT(lo);
2086         LASSERT(lo->ldo_comp_cnt > comp_idx && lo->ldo_comp_entries != NULL);
2087         lod_comp = &lo->ldo_comp_entries[comp_idx];
2088
2089         /* A released component is being created */
2090         if (lod_comp->llc_pattern & LOV_PATTERN_F_RELEASED)
2091                 RETURN(0);
2092
2093         /* A Data-on-MDT component is being created */
2094         if (lov_pattern(lod_comp->llc_pattern) == LOV_PATTERN_MDT)
2095                 RETURN(0);
2096
2097         if (likely(lod_comp->llc_stripe == NULL)) {
2098                 /*
2099                  * no striping has been created so far
2100                  */
2101                 LASSERT(lod_comp->llc_stripe_count);
2102                 /*
2103                  * statfs and check OST targets now, since ld_active_tgt_count
2104                  * could be changed if some OSTs are [de]activated manually.
2105                  */
2106                 lod_qos_statfs_update(env, d);
2107                 stripe_len = lod_get_stripe_count(d, lo,
2108                                                   lod_comp->llc_stripe_count);
2109                 if (stripe_len == 0)
2110                         GOTO(out, rc = -ERANGE);
2111                 lod_comp->llc_stripe_count = stripe_len;
2112                 OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_len);
2113                 if (stripe == NULL)
2114                         GOTO(out, rc = -ENOMEM);
2115
2116                 lod_getref(&d->lod_ost_descs);
2117                 /* XXX: support for non-0 files w/o objects */
2118                 CDEBUG(D_OTHER, "tgt_count %d stripe_count %d\n",
2119                                 d->lod_desc.ld_tgt_count, stripe_len);
2120
2121                 if (lod_comp->llc_ostlist.op_array) {
2122                         rc = lod_alloc_ost_list(env, lo, stripe, th, comp_idx,
2123                                                 inuse);
2124                 } else if (lod_comp->llc_stripe_offset == LOV_OFFSET_DEFAULT) {
2125                         rc = lod_alloc_qos(env, lo, stripe, flag, th,
2126                                            comp_idx, inuse);
2127                         if (rc == -EAGAIN)
2128                                 rc = lod_alloc_rr(env, lo, stripe, flag, th,
2129                                                   comp_idx, inuse);
2130                 } else {
2131                         rc = lod_alloc_specific(env, lo, stripe, flag, th,
2132                                                 comp_idx, inuse);
2133                 }
2134                 lod_putref(d, &d->lod_ost_descs);
2135
2136                 if (rc < 0) {
2137                         for (i = 0; i < stripe_len; i++)
2138                                 if (stripe[i] != NULL)
2139                                         dt_object_put(env, stripe[i]);
2140
2141                         OBD_FREE(stripe, sizeof(stripe[0]) * stripe_len);
2142                         lod_comp->llc_stripe_count = 0;
2143                 } else {
2144                         lod_comp->llc_stripe = stripe;
2145                         lod_comp->llc_stripes_allocated = stripe_len;
2146                 }
2147         } else {
2148                 /*
2149                  * lod_qos_parse_config() found supplied buf as a predefined
2150                  * striping (not a hint), so it allocated all the object
2151                  * now we need to create them
2152                  */
2153                 for (i = 0; i < lod_comp->llc_stripe_count; i++) {
2154                         struct dt_object  *o;
2155
2156                         o = lod_comp->llc_stripe[i];
2157                         LASSERT(o);
2158
2159                         rc = lod_sub_declare_create(env, o, attr, NULL,
2160                                                     NULL, th);
2161                         if (rc < 0) {
2162                                 CERROR("can't declare create: %d\n", rc);
2163                                 break;
2164                         }
2165                 }
2166                 /**
2167                  * Clear LCME_FL_INIT for the component so that
2168                  * lod_striping_create() can create the striping objects
2169                  * in replay.
2170                  */
2171                 lod_comp_unset_init(lod_comp);
2172         }
2173
2174 out:
2175         RETURN(rc);
2176 }
2177
2178 int lod_obj_stripe_set_inuse_cb(const struct lu_env *env,
2179                                 struct lod_object *lo,
2180                                 struct dt_object *dt, struct thandle *th,
2181                                 int comp_idx, int stripe_idx,
2182                                 struct lod_obj_stripe_cb_data *data)
2183 {
2184         struct lod_thread_info  *info = lod_env_info(env);
2185         struct lod_device       *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
2186         struct lu_fid   *fid = &info->lti_fid;
2187         __u32   index;
2188         int     rc, type = LU_SEQ_RANGE_OST;
2189
2190         *fid = *lu_object_fid(&dt->do_lu);
2191         rc = lod_fld_lookup(env, d, fid, &index, &type);
2192         if (rc < 0) {
2193                 CERROR("%s: fail to locate "DFID": rc = %d\n",
2194                        lod2obd(d)->obd_name, PFID(fid), rc);
2195                 return rc;
2196         }
2197         lod_comp_ost_in_use(data->locd_inuse, index);
2198         return 0;
2199 }
2200
2201 /**
2202  * Resize per-thread ost list to hold OST target index list already used.
2203  *
2204  * \param[in,out] inuse         structure contains ost list array
2205  * \param[in] cnt               total stripe count of all components
2206  * \param[in] max               array's max size if @max > 0
2207  *
2208  * \retval 0            on success
2209  * \retval -ENOMEM      reallocation failed
2210  */
2211 static int lod_inuse_resize(struct ost_pool *inuse, __u16 cnt, __u16 max)
2212 {
2213         __u32 *array;
2214         __u32 new = cnt * sizeof(inuse->op_array[0]);
2215
2216         inuse->op_count = 0;
2217
2218         if (new <= inuse->op_size)
2219                 return 0;
2220
2221         if (max)
2222                 new = min_t(__u32, new, max);
2223
2224         OBD_ALLOC(array, new);
2225         if (!array)
2226                 return -ENOMEM;
2227
2228         if (inuse->op_array)
2229                 OBD_FREE(inuse->op_array, inuse->op_size);
2230
2231         inuse->op_array = array;
2232         inuse->op_size = new;
2233
2234         return 0;
2235 }
2236
2237 int lod_prepare_inuse(const struct lu_env *env, struct lod_object *lo)
2238 {
2239         struct lod_thread_info *info = lod_env_info(env);
2240         struct lod_device *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
2241         struct ost_pool *inuse = &info->lti_inuse_osts;
2242         struct lod_obj_stripe_cb_data data = { { 0 } };
2243         __u32 stripe_count = 0;
2244         int i;
2245         int rc;
2246
2247         for (i = 0; i < lo->ldo_comp_cnt; i++)
2248                 stripe_count += lod_comp_entry_stripe_count(lo,
2249                                         &lo->ldo_comp_entries[i], false);
2250         rc = lod_inuse_resize(inuse, stripe_count, d->lod_osd_max_easize);
2251         if (rc)
2252                 return rc;
2253
2254         data.locd_inuse = inuse;
2255         data.locd_stripe_cb = lod_obj_stripe_set_inuse_cb;
2256         return lod_obj_for_each_stripe(env, lo, NULL, &data);
2257 }
2258
2259 int lod_prepare_create(const struct lu_env *env, struct lod_object *lo,
2260                        struct lu_attr *attr, const struct lu_buf *buf,
2261                        struct thandle *th)
2262
2263 {
2264         struct lod_thread_info *info = lod_env_info(env);
2265         struct lod_device *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
2266         struct ost_pool *inuse = &info->lti_inuse_osts;
2267         uint64_t size = 0;
2268         int i;
2269         int rc;
2270         ENTRY;
2271
2272         LASSERT(lo);
2273
2274         /* no OST available */
2275         /* XXX: should we be waiting a bit to prevent failures during
2276          * cluster initialization? */
2277         if (d->lod_ostnr == 0)
2278                 RETURN(-EIO);
2279
2280         /*
2281          * by this time, the object's ldo_stripe_count and ldo_stripe_size
2282          * contain default value for striping: taken from the parent
2283          * or from filesystem defaults
2284          *
2285          * in case the caller is passing lovea with new striping config,
2286          * we may need to parse lovea and apply new configuration
2287          */
2288         rc = lod_qos_parse_config(env, lo, buf);
2289         if (rc)
2290                 RETURN(rc);
2291
2292         if (attr->la_valid & LA_SIZE)
2293                 size = attr->la_size;
2294
2295         /* prepare inuse */
2296         rc = lod_prepare_inuse(env, lo);
2297         if (rc)
2298                 RETURN(rc);
2299
2300         /**
2301          * prepare OST object creation for the component covering file's
2302          * size, the 1st component (including plain layout file) is always
2303          * instantiated.
2304          */
2305         for (i = 0; i < lo->ldo_comp_cnt; i++) {
2306                 struct lod_layout_component *lod_comp;
2307                 struct lu_extent *extent;
2308
2309                 lod_comp = &lo->ldo_comp_entries[i];
2310                 extent = &lod_comp->llc_extent;
2311                 CDEBUG(D_QOS, "%lld [%lld, %lld)\n",
2312                        size, extent->e_start, extent->e_end);
2313                 if (!lo->ldo_is_composite || size >= extent->e_start) {
2314                         rc = lod_qos_prep_create(env, lo, attr, th, i, inuse);
2315                         if (rc)
2316                                 break;
2317                 }
2318         }
2319
2320         RETURN(rc);
2321 }