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