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, 2013, 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  */
  35
  36 #define DEBUG_SUBSYSTEM S_LOV
  37
  38 #include <asm/div64.h>
  39 #include <libcfs/libcfs.h>
  40 #include <obd_class.h>
  41 #include <lustre/lustre_idl.h>
  42 #include "lod_internal.h"
  43
  44 /*
  45  * force QoS policy (not RR) to be used for testing purposes
  46  */
  47 #define FORCE_QOS_
  48
  49 #define D_QOS   D_OTHER
  50
  51 #if 0
  52 #define QOS_DEBUG(fmt, ...)     CDEBUG(D_OTHER, fmt, ## __VA_ARGS__)
  53 #define QOS_CONSOLE(fmt, ...)   LCONSOLE(D_OTHER, fmt, ## __VA_ARGS__)
  54 #else
  55 #define QOS_DEBUG(fmt, ...)
  56 #define QOS_CONSOLE(fmt, ...)
  57 #endif
  58
  59 #define TGT_BAVAIL(i) (OST_TGT(lod,i)->ltd_statfs.os_bavail * \
  60                        OST_TGT(lod,i)->ltd_statfs.os_bsize)
  61
  62 int qos_add_tgt(struct lod_device *lod, struct lod_tgt_desc *ost_desc)
  63 {
  64         struct lod_qos_oss *oss = NULL, *temposs;
  65         struct obd_export  *exp = ost_desc->ltd_exp;
  66         int                 rc = 0, found = 0;
  67         cfs_list_t         *list;
  68         ENTRY;
  69
  70         down_write(&lod->lod_qos.lq_rw_sem);
  71         /*
  72          * a bit hacky approach to learn NID of corresponding connection
  73          * but there is no official API to access information like this
  74          * with OSD API.
  75          */
  76         cfs_list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
  77                 if (obd_uuid_equals(&oss->lqo_uuid,
  78                                     &exp->exp_connection->c_remote_uuid)) {
  79                         found++;
  80                         break;
  81                 }
  82         }
  83
  84         if (!found) {
  85                 OBD_ALLOC_PTR(oss);
  86                 if (!oss)
  87                         GOTO(out, rc = -ENOMEM);
  88                 memcpy(&oss->lqo_uuid, &exp->exp_connection->c_remote_uuid,
  89                        sizeof(oss->lqo_uuid));
  90         } else {
  91                 /* Assume we have to move this one */
  92                 cfs_list_del(&oss->lqo_oss_list);
  93         }
  94
  95         oss->lqo_ost_count++;
  96         ost_desc->ltd_qos.ltq_oss = oss;
  97
  98         CDEBUG(D_QOS, "add tgt %s to OSS %s (%d OSTs)\n",
  99                obd_uuid2str(&ost_desc->ltd_uuid), obd_uuid2str(&oss->lqo_uuid),
 100                oss->lqo_ost_count);
 101
 102         /* Add sorted by # of OSTs.  Find the first entry that we're
 103            bigger than... */
 104         list = &lod->lod_qos.lq_oss_list;
 105         cfs_list_for_each_entry(temposs, list, lqo_oss_list) {
 106                 if (oss->lqo_ost_count > temposs->lqo_ost_count)
 107                         break;
 108         }
 109         /* ...and add before it.  If we're the first or smallest, temposs
 110            points to the list head, and we add to the end. */
 111         cfs_list_add_tail(&oss->lqo_oss_list, &temposs->lqo_oss_list);
 112
 113         lod->lod_qos.lq_dirty = 1;
 114         lod->lod_qos.lq_rr.lqr_dirty = 1;
 115
 116 out:
 117         up_write(&lod->lod_qos.lq_rw_sem);
 118         RETURN(rc);
 119 }
 120
 121 int qos_del_tgt(struct lod_device *lod, struct lod_tgt_desc *ost_desc)
 122 {
 123         struct lod_qos_oss *oss;
 124         int                 rc = 0;
 125         ENTRY;
 126
 127         down_write(&lod->lod_qos.lq_rw_sem);
 128         oss = ost_desc->ltd_qos.ltq_oss;
 129         if (!oss)
 130                 GOTO(out, rc = -ENOENT);
 131
 132         oss->lqo_ost_count--;
 133         if (oss->lqo_ost_count == 0) {
 134                 CDEBUG(D_QOS, "removing OSS %s\n",
 135                        obd_uuid2str(&oss->lqo_uuid));
 136                 cfs_list_del(&oss->lqo_oss_list);
 137                 ost_desc->ltd_qos.ltq_oss = NULL;
 138                 OBD_FREE_PTR(oss);
 139         }
 140
 141         lod->lod_qos.lq_dirty = 1;
 142         lod->lod_qos.lq_rr.lqr_dirty = 1;
 143 out:
 144         up_write(&lod->lod_qos.lq_rw_sem);
 145         RETURN(rc);
 146 }
 147
 148 static int lod_statfs_and_check(const struct lu_env *env, struct lod_device *d,
 149                                 int index, struct obd_statfs *sfs)
 150 {
 151         struct lod_tgt_desc *ost;
 152         int                  rc;
 153
 154         LASSERT(d);
 155         ost = OST_TGT(d,index);
 156         LASSERT(ost);
 157
 158         rc = dt_statfs(env, ost->ltd_ost, sfs);
 159         if (rc && rc != -ENOTCONN)
 160                 CERROR("%s: statfs: rc = %d\n", lod2obd(d)->obd_name, rc);
 161
 162         /* If the OST is readonly then we can't allocate objects there */
 163         if (sfs->os_state & OS_STATE_READONLY)
 164                 rc = -EROFS;
 165
 166         /* check whether device has changed state (active, inactive) */
 167         if (rc != 0 && ost->ltd_active) {
 168                 /* turned inactive? */
 169                 spin_lock(&d->lod_desc_lock);
 170                 if (ost->ltd_active) {
 171                         ost->ltd_active = 0;
 172                         LASSERT(d->lod_desc.ld_active_tgt_count > 0);
 173                         d->lod_desc.ld_active_tgt_count--;
 174                         d->lod_qos.lq_dirty = 1;
 175                         d->lod_qos.lq_rr.lqr_dirty = 1;
 176                         CDEBUG(D_CONFIG, "%s: turns inactive\n",
 177                                ost->ltd_exp->exp_obd->obd_name);
 178                 }
 179                 spin_unlock(&d->lod_desc_lock);
 180         } else if (rc == 0 && ost->ltd_active == 0) {
 181                 /* turned active? */
 182                 LASSERTF(d->lod_desc.ld_active_tgt_count < d->lod_ostnr,
 183                          "active tgt count %d, ost nr %d\n",
 184                          d->lod_desc.ld_active_tgt_count, d->lod_ostnr);
 185                 spin_lock(&d->lod_desc_lock);
 186                 if (ost->ltd_active == 0) {
 187                         ost->ltd_active = 1;
 188                         d->lod_desc.ld_active_tgt_count++;
 189                         d->lod_qos.lq_dirty = 1;
 190                         d->lod_qos.lq_rr.lqr_dirty = 1;
 191                         CDEBUG(D_CONFIG, "%s: turns active\n",
 192                                ost->ltd_exp->exp_obd->obd_name);
 193                 }
 194                 spin_unlock(&d->lod_desc_lock);
 195         }
 196
 197         RETURN(rc);
 198 }
 199
 200 static void lod_qos_statfs_update(const struct lu_env *env,
 201                                   struct lod_device *lod)
 202 {
 203         struct obd_device *obd = lod2obd(lod);
 204         struct ost_pool   *osts = &(lod->lod_pool_info);
 205         int                i, idx, rc = 0;
 206         __u64              max_age, avail;
 207         ENTRY;
 208
 209         max_age = cfs_time_shift_64(-2 * lod->lod_desc.ld_qos_maxage);
 210
 211         if (cfs_time_beforeq_64(max_age, obd->obd_osfs_age))
 212                 /* statfs data are quite recent, don't need to refresh it */
 213                 RETURN_EXIT;
 214
 215         down_write(&lod->lod_qos.lq_rw_sem);
 216         if (cfs_time_beforeq_64(max_age, obd->obd_osfs_age))
 217                 GOTO(out, rc = 0);
 218
 219         for (i = 0; i < osts->op_count; i++) {
 220                 idx = osts->op_array[i];
 221                 avail = OST_TGT(lod,idx)->ltd_statfs.os_bavail;
 222                 rc = lod_statfs_and_check(env, lod, idx,
 223                                           &OST_TGT(lod,idx)->ltd_statfs);
 224                 if (rc)
 225                         break;
 226                 if (OST_TGT(lod,idx)->ltd_statfs.os_bavail != avail)
 227                         /* recalculate weigths */
 228                         lod->lod_qos.lq_dirty = 1;
 229         }
 230         obd->obd_osfs_age = cfs_time_current_64();
 231
 232 out:
 233         up_write(&lod->lod_qos.lq_rw_sem);
 234         EXIT;
 235 }
 236
 237 /* Recalculate per-object penalties for OSSs and OSTs,
 238    depends on size of each ost in an oss */
 239 static int lod_qos_calc_ppo(struct lod_device *lod)
 240 {
 241         struct lod_qos_oss *oss;
 242         __u64               ba_max, ba_min, temp;
 243         __u32               num_active;
 244         int                 rc, i, prio_wide;
 245         time_t              now, age;
 246         ENTRY;
 247
 248         if (!lod->lod_qos.lq_dirty)
 249                 GOTO(out, rc = 0);
 250
 251         num_active = lod->lod_desc.ld_active_tgt_count - 1;
 252         if (num_active < 1)
 253                 GOTO(out, rc = -EAGAIN);
 254
 255         /* find bavail on each OSS */
 256         cfs_list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list)
 257                                 oss->lqo_bavail = 0;
 258         lod->lod_qos.lq_active_oss_count = 0;
 259
 260         /*
 261          * How badly user wants to select OSTs "widely" (not recently chosen
 262          * and not on recent OSS's).  As opposed to "freely" (free space
 263          * avail.) 0-256
 264          */
 265         prio_wide = 256 - lod->lod_qos.lq_prio_free;
 266
 267         ba_min = (__u64)(-1);
 268         ba_max = 0;
 269         now = cfs_time_current_sec();
 270         /* Calculate OST penalty per object
 271          * (lod ref taken in lod_qos_prep_create()) */
 272         cfs_foreach_bit(lod->lod_ost_bitmap, i) {
 273                 LASSERT(OST_TGT(lod,i));
 274                 temp = TGT_BAVAIL(i);
 275                 if (!temp)
 276                         continue;
 277                 ba_min = min(temp, ba_min);
 278                 ba_max = max(temp, ba_max);
 279
 280                 /* Count the number of usable OSS's */
 281                 if (OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_bavail == 0)
 282                         lod->lod_qos.lq_active_oss_count++;
 283                 OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_bavail += temp;
 284
 285                 /* per-OST penalty is prio * TGT_bavail / (num_ost - 1) / 2 */
 286                 temp >>= 1;
 287                 do_div(temp, num_active);
 288                 OST_TGT(lod,i)->ltd_qos.ltq_penalty_per_obj =
 289                         (temp * prio_wide) >> 8;
 290
 291                 age = (now - OST_TGT(lod,i)->ltd_qos.ltq_used) >> 3;
 292                 if (lod->lod_qos.lq_reset ||
 293                     age > 32 * lod->lod_desc.ld_qos_maxage)
 294                         OST_TGT(lod,i)->ltd_qos.ltq_penalty = 0;
 295                 else if (age > lod->lod_desc.ld_qos_maxage)
 296                         /* Decay the penalty by half for every 8x the update
 297                          * interval that the device has been idle.  That gives
 298                          * lots of time for the statfs information to be
 299                          * updated (which the penalty is only a proxy for),
 300                          * and avoids penalizing OSS/OSTs under light load. */
 301                         OST_TGT(lod,i)->ltd_qos.ltq_penalty >>=
 302                                 (age / lod->lod_desc.ld_qos_maxage);
 303         }
 304
 305         num_active = lod->lod_qos.lq_active_oss_count - 1;
 306         if (num_active < 1) {
 307                 /* If there's only 1 OSS, we can't penalize it, so instead
 308                    we have to double the OST penalty */
 309                 num_active = 1;
 310                 cfs_foreach_bit(lod->lod_ost_bitmap, i)
 311                         OST_TGT(lod,i)->ltd_qos.ltq_penalty_per_obj <<= 1;
 312         }
 313
 314         /* Per-OSS penalty is prio * oss_avail / oss_osts / (num_oss - 1) / 2 */
 315         cfs_list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
 316                 temp = oss->lqo_bavail >> 1;
 317                 do_div(temp, oss->lqo_ost_count * num_active);
 318                 oss->lqo_penalty_per_obj = (temp * prio_wide) >> 8;
 319
 320                 age = (now - oss->lqo_used) >> 3;
 321                 if (lod->lod_qos.lq_reset ||
 322                     age > 32 * lod->lod_desc.ld_qos_maxage)
 323                         oss->lqo_penalty = 0;
 324                 else if (age > lod->lod_desc.ld_qos_maxage)
 325                         /* Decay the penalty by half for every 8x the update
 326                          * interval that the device has been idle.  That gives
 327                          * lots of time for the statfs information to be
 328                          * updated (which the penalty is only a proxy for),
 329                          * and avoids penalizing OSS/OSTs under light load. */
 330                         oss->lqo_penalty >>= age / lod->lod_desc.ld_qos_maxage;
 331         }
 332
 333         lod->lod_qos.lq_dirty = 0;
 334         lod->lod_qos.lq_reset = 0;
 335
 336         /* If each ost has almost same free space,
 337          * do rr allocation for better creation performance */
 338         lod->lod_qos.lq_same_space = 0;
 339         if ((ba_max * (256 - lod->lod_qos.lq_threshold_rr)) >> 8 < ba_min) {
 340                 lod->lod_qos.lq_same_space = 1;
 341                 /* Reset weights for the next time we enter qos mode */
 342                 lod->lod_qos.lq_reset = 1;
 343         }
 344         rc = 0;
 345
 346 out:
 347 #ifndef FORCE_QOS
 348         if (!rc && lod->lod_qos.lq_same_space)
 349                 RETURN(-EAGAIN);
 350 #endif
 351         RETURN(rc);
 352 }
 353
 354 static int lod_qos_calc_weight(struct lod_device *lod, int i)
 355 {
 356         __u64 temp, temp2;
 357
 358         /* Final ost weight = TGT_BAVAIL - ost_penalty - oss_penalty */
 359         temp = TGT_BAVAIL(i);
 360         temp2 = OST_TGT(lod,i)->ltd_qos.ltq_penalty +
 361                 OST_TGT(lod,i)->ltd_qos.ltq_oss->lqo_penalty;
 362         if (temp < temp2)
 363                 OST_TGT(lod,i)->ltd_qos.ltq_weight = 0;
 364         else
 365                 OST_TGT(lod,i)->ltd_qos.ltq_weight = temp - temp2;
 366         return 0;
 367 }
 368
 369 /* We just used this index for a stripe; adjust everyone's weights */
 370 static int lod_qos_used(struct lod_device *lod, struct ost_pool *osts,
 371                         __u32 index, __u64 *total_wt)
 372 {
 373         struct lod_tgt_desc *ost;
 374         struct lod_qos_oss  *oss;
 375         int j;
 376         ENTRY;
 377
 378         ost = OST_TGT(lod,index);
 379         LASSERT(ost);
 380
 381         /* Don't allocate on this devuce anymore, until the next alloc_qos */
 382         ost->ltd_qos.ltq_usable = 0;
 383
 384         oss = ost->ltd_qos.ltq_oss;
 385
 386         /* Decay old penalty by half (we're adding max penalty, and don't
 387            want it to run away.) */
 388         ost->ltd_qos.ltq_penalty >>= 1;
 389         oss->lqo_penalty >>= 1;
 390
 391         /* mark the OSS and OST as recently used */
 392         ost->ltd_qos.ltq_used = oss->lqo_used = cfs_time_current_sec();
 393
 394         /* Set max penalties for this OST and OSS */
 395         ost->ltd_qos.ltq_penalty +=
 396                 ost->ltd_qos.ltq_penalty_per_obj * lod->lod_ostnr;
 397         oss->lqo_penalty += oss->lqo_penalty_per_obj *
 398                 lod->lod_qos.lq_active_oss_count;
 399
 400         /* Decrease all OSS penalties */
 401         cfs_list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
 402                 if (oss->lqo_penalty < oss->lqo_penalty_per_obj)
 403                         oss->lqo_penalty = 0;
 404                 else
 405                         oss->lqo_penalty -= oss->lqo_penalty_per_obj;
 406         }
 407
 408         *total_wt = 0;
 409         /* Decrease all OST penalties */
 410         for (j = 0; j < osts->op_count; j++) {
 411                 int i;
 412
 413                 i = osts->op_array[j];
 414                 if (!cfs_bitmap_check(lod->lod_ost_bitmap, i))
 415                         continue;
 416
 417                 ost = OST_TGT(lod,i);
 418                 LASSERT(ost);
 419
 420                 if (ost->ltd_qos.ltq_penalty <
 421                                 ost->ltd_qos.ltq_penalty_per_obj)
 422                         ost->ltd_qos.ltq_penalty = 0;
 423                 else
 424                         ost->ltd_qos.ltq_penalty -=
 425                                 ost->ltd_qos.ltq_penalty_per_obj;
 426
 427                 lod_qos_calc_weight(lod, i);
 428
 429                 /* Recalc the total weight of usable osts */
 430                 if (ost->ltd_qos.ltq_usable)
 431                         *total_wt += ost->ltd_qos.ltq_weight;
 432
 433                 QOS_DEBUG("recalc tgt %d usable=%d avail="LPU64
 434                           " ostppo="LPU64" ostp="LPU64" ossppo="LPU64
 435                           " ossp="LPU64" wt="LPU64"\n",
 436                           i, ost->ltd_qos.ltq_usable, TGT_BAVAIL(i) >> 10,
 437                           ost->ltd_qos.ltq_penalty_per_obj >> 10,
 438                           ost->ltd_qos.ltq_penalty >> 10,
 439                           ost->ltd_qos.ltq_oss->lqo_penalty_per_obj >> 10,
 440                           ost->ltd_qos.ltq_oss->lqo_penalty >> 10,
 441                           ost->ltd_qos.ltq_weight >> 10);
 442         }
 443
 444         RETURN(0);
 445 }
 446
 447 #define LOV_QOS_EMPTY ((__u32)-1)
 448 /* compute optimal round-robin order, based on OSTs per OSS */
 449 static int lod_qos_calc_rr(struct lod_device *lod, struct ost_pool *src_pool,
 450                            struct lod_qos_rr *lqr)
 451 {
 452         struct lod_qos_oss  *oss;
 453         struct lod_tgt_desc *ost;
 454         unsigned placed, real_count;
 455         int i, rc;
 456         ENTRY;
 457
 458         if (!lqr->lqr_dirty) {
 459                 LASSERT(lqr->lqr_pool.op_size);
 460                 RETURN(0);
 461         }
 462
 463         /* Do actual allocation. */
 464         down_write(&lod->lod_qos.lq_rw_sem);
 465
 466         /*
 467          * Check again. While we were sleeping on @lq_rw_sem something could
 468          * change.
 469          */
 470         if (!lqr->lqr_dirty) {
 471                 LASSERT(lqr->lqr_pool.op_size);
 472                 up_write(&lod->lod_qos.lq_rw_sem);
 473                 RETURN(0);
 474         }
 475
 476         real_count = src_pool->op_count;
 477
 478         /* Zero the pool array */
 479         /* alloc_rr is holding a read lock on the pool, so nobody is adding/
 480            deleting from the pool. The lq_rw_sem insures that nobody else
 481            is reading. */
 482         lqr->lqr_pool.op_count = real_count;
 483         rc = lod_ost_pool_extend(&lqr->lqr_pool, real_count);
 484         if (rc) {
 485                 up_write(&lod->lod_qos.lq_rw_sem);
 486                 RETURN(rc);
 487         }
 488         for (i = 0; i < lqr->lqr_pool.op_count; i++)
 489                 lqr->lqr_pool.op_array[i] = LOV_QOS_EMPTY;
 490
 491         /* Place all the OSTs from 1 OSS at the same time. */
 492         placed = 0;
 493         cfs_list_for_each_entry(oss, &lod->lod_qos.lq_oss_list, lqo_oss_list) {
 494                 int j = 0;
 495
 496                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
 497                         int next;
 498
 499                         if (!cfs_bitmap_check(lod->lod_ost_bitmap,
 500                                                 src_pool->op_array[i]))
 501                                 continue;
 502
 503                         ost = OST_TGT(lod,src_pool->op_array[i]);
 504                         LASSERT(ost && ost->ltd_ost);
 505                         if (ost->ltd_qos.ltq_oss != oss)
 506                                 continue;
 507
 508                         /* Evenly space these OSTs across arrayspace */
 509                         next = j * lqr->lqr_pool.op_count / oss->lqo_ost_count;
 510                         while (lqr->lqr_pool.op_array[next] != LOV_QOS_EMPTY)
 511                                 next = (next + 1) % lqr->lqr_pool.op_count;
 512
 513                         lqr->lqr_pool.op_array[next] = src_pool->op_array[i];
 514                         j++;
 515                         placed++;
 516                 }
 517         }
 518
 519         lqr->lqr_dirty = 0;
 520         up_write(&lod->lod_qos.lq_rw_sem);
 521
 522         if (placed != real_count) {
 523                 /* This should never happen */
 524                 LCONSOLE_ERROR_MSG(0x14e, "Failed to place all OSTs in the "
 525                                    "round-robin list (%d of %d).\n",
 526                                    placed, real_count);
 527                 for (i = 0; i < lqr->lqr_pool.op_count; i++) {
 528                         LCONSOLE(D_WARNING, "rr #%d ost idx=%d\n", i,
 529                                  lqr->lqr_pool.op_array[i]);
 530                 }
 531                 lqr->lqr_dirty = 1;
 532                 RETURN(-EAGAIN);
 533         }
 534
 535 #if 0
 536         for (i = 0; i < lqr->lqr_pool.op_count; i++)
 537                 QOS_CONSOLE("rr #%d ost idx=%d\n", i, lqr->lqr_pool.op_array[i]);
 538 #endif
 539
 540         RETURN(0);
 541 }
 542
 543 /**
 544  * A helper function to:
 545  *   create in-core lu object on the specified OSP
 546  *   declare creation of the object
 547  * IMPORTANT: at this stage object is anonymouos - it has no fid assigned
 548  *            this is a workaround till we have natural FIDs on OST
 549  *
 550  *            at this point we want to declare (reserve) object for us as
 551  *            we can't block at execution (when create method is called).
 552  *            otherwise we'd block whole transaction batch
 553  */
 554 static struct dt_object *lod_qos_declare_object_on(const struct lu_env *env,
 555                                                    struct lod_device *d,
 556                                                    int ost_idx,
 557                                                    struct thandle *th)
 558 {
 559         struct lod_tgt_desc *ost;
 560         struct lu_object *o, *n;
 561         struct lu_device *nd;
 562         struct dt_object *dt;
 563         int               rc;
 564         ENTRY;
 565
 566         LASSERT(d);
 567         LASSERT(ost_idx >= 0);
 568         LASSERT(ost_idx < d->lod_osts_size);
 569         ost = OST_TGT(d,ost_idx);
 570         LASSERT(ost);
 571         LASSERT(ost->ltd_ost);
 572
 573         nd = &ost->ltd_ost->dd_lu_dev;
 574
 575         /*
 576          * allocate anonymous object with zero fid, real fid
 577          * will be assigned by OSP within transaction
 578          * XXX: to be fixed with fully-functional OST fids
 579          */
 580         o = lu_object_anon(env, nd, NULL);
 581         if (IS_ERR(o))
 582                 GOTO(out, dt = ERR_PTR(PTR_ERR(o)));
 583
 584         n = lu_object_locate(o->lo_header, nd->ld_type);
 585         if (unlikely(n == NULL)) {
 586                 CERROR("can't find slice\n");
 587                 lu_object_put(env, o);
 588                 GOTO(out, dt = ERR_PTR(-EINVAL));
 589         }
 590
 591         dt = container_of(n, struct dt_object, do_lu);
 592
 593         rc = dt_declare_create(env, dt, NULL, NULL, NULL, th);
 594         if (rc) {
 595                 CDEBUG(D_OTHER, "can't declare creation on #%u: %d\n",
 596                        ost_idx, rc);
 597                 lu_object_put(env, o);
 598                 dt = ERR_PTR(rc);
 599         }
 600
 601 out:
 602         RETURN(dt);
 603 }
 604
 605 static int min_stripe_count(int stripe_cnt, int flags)
 606 {
 607         return (flags & LOV_USES_DEFAULT_STRIPE ?
 608                         stripe_cnt - (stripe_cnt / 4) : stripe_cnt);
 609 }
 610
 611 #define LOV_CREATE_RESEED_MULT 30
 612 #define LOV_CREATE_RESEED_MIN  2000
 613
 614 static int inline lod_qos_dev_is_full(struct obd_statfs *msfs)
 615 {
 616         __u64 used;
 617         int   bs = msfs->os_bsize;
 618
 619         LASSERT(((bs - 1) & bs) == 0);
 620
 621         /* the minimum of 0.1% used blocks and 1GB bytes. */
 622         used = min_t(__u64, (msfs->os_blocks - msfs->os_bfree) >> 10,
 623                         1 << (31 - ffs(bs)));
 624         return (msfs->os_bavail < used);
 625 }
 626
 627 static inline int lod_qos_ost_in_use_clear(const struct lu_env *env, int stripes)
 628 {
 629         struct lod_thread_info *info = lod_env_info(env);
 630
 631         if (info->lti_ea_store_size < sizeof(int) * stripes)
 632                 lod_ea_store_resize(info, stripes * sizeof(int));
 633         if (info->lti_ea_store_size < sizeof(int) * stripes) {
 634                 CERROR("can't allocate memory for ost-in-use array\n");
 635                 return -ENOMEM;
 636         }
 637         memset(info->lti_ea_store, -1, sizeof(int) * stripes);
 638         return 0;
 639 }
 640
 641 static inline void lod_qos_ost_in_use(const struct lu_env *env, int idx, int ost)
 642 {
 643         struct lod_thread_info *info = lod_env_info(env);
 644         int *osts = info->lti_ea_store;
 645
 646         LASSERT(info->lti_ea_store_size >= idx * sizeof(int));
 647         osts[idx] = ost;
 648 }
 649
 650 static int lod_qos_is_ost_used(const struct lu_env *env, int ost, int stripes)
 651 {
 652         struct lod_thread_info *info = lod_env_info(env);
 653         int *osts = info->lti_ea_store;
 654         int j;
 655
 656         for (j = 0; j < stripes; j++) {
 657                 if (osts[j] == ost)
 658                         return 1;
 659         }
 660         return 0;
 661 }
 662
 663 /* Allocate objects on OSTs with round-robin algorithm */
 664 static int lod_alloc_rr(const struct lu_env *env, struct lod_object *lo,
 665                         struct dt_object **stripe, int flags,
 666                         struct thandle *th)
 667 {
 668         struct lod_device *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
 669         struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
 670         struct pool_desc  *pool = NULL;
 671         struct ost_pool   *osts;
 672         struct lod_qos_rr *lqr;
 673         struct dt_object  *o;
 674         unsigned           array_idx;
 675         int                i, rc;
 676         int                ost_start_idx_temp;
 677         int                speed = 0;
 678         int                stripe_idx = 0;
 679         int                stripe_cnt = lo->ldo_stripenr;
 680         int                stripe_cnt_min = min_stripe_count(stripe_cnt, flags);
 681         __u32              ost_idx;
 682         ENTRY;
 683
 684         if (lo->ldo_pool)
 685                 pool = lod_find_pool(m, lo->ldo_pool);
 686
 687         if (pool != NULL) {
 688                 down_read(&pool_tgt_rw_sem(pool));
 689                 osts = &(pool->pool_obds);
 690                 lqr = &(pool->pool_rr);
 691         } else {
 692                 osts = &(m->lod_pool_info);
 693                 lqr = &(m->lod_qos.lq_rr);
 694         }
 695
 696         rc = lod_qos_calc_rr(m, osts, lqr);
 697         if (rc)
 698                 GOTO(out, rc);
 699
 700         rc = lod_qos_ost_in_use_clear(env, lo->ldo_stripenr);
 701         if (rc)
 702                 GOTO(out, rc);
 703
 704         if (--lqr->lqr_start_count <= 0) {
 705                 lqr->lqr_start_idx = cfs_rand() % osts->op_count;
 706                 lqr->lqr_start_count =
 707                         (LOV_CREATE_RESEED_MIN / max(osts->op_count, 1U) +
 708                          LOV_CREATE_RESEED_MULT) * max(osts->op_count, 1U);
 709         } else if (stripe_cnt_min >= osts->op_count ||
 710                         lqr->lqr_start_idx > osts->op_count) {
 711                 /* If we have allocated from all of the OSTs, slowly
 712                  * precess the next start if the OST/stripe count isn't
 713                  * already doing this for us. */
 714                 lqr->lqr_start_idx %= osts->op_count;
 715                 if (stripe_cnt > 1 && (osts->op_count % stripe_cnt) != 1)
 716                         ++lqr->lqr_offset_idx;
 717         }
 718         down_read(&m->lod_qos.lq_rw_sem);
 719         ost_start_idx_temp = lqr->lqr_start_idx;
 720
 721 repeat_find:
 722         array_idx = (lqr->lqr_start_idx + lqr->lqr_offset_idx) %
 723                         osts->op_count;
 724
 725         QOS_DEBUG("pool '%s' want %d startidx %d startcnt %d offset %d "
 726                   "active %d count %d arrayidx %d\n",
 727                   lo->ldo_pool ? lo->ldo_pool : "",
 728                   stripe_cnt, lqr->lqr_start_idx, lqr->lqr_start_count,
 729                   lqr->lqr_offset_idx, osts->op_count, osts->op_count,
 730                   array_idx);
 731
 732         for (i = 0; i < osts->op_count && stripe_idx < lo->ldo_stripenr;
 733              i++, array_idx = (array_idx + 1) % osts->op_count) {
 734                 ++lqr->lqr_start_idx;
 735                 ost_idx = lqr->lqr_pool.op_array[array_idx];
 736
 737                 QOS_DEBUG("#%d strt %d act %d strp %d ary %d idx %d\n",
 738                           i, lqr->lqr_start_idx, /* XXX: active*/ 0,
 739                           stripe_idx, array_idx, ost_idx);
 740
 741                 if ((ost_idx == LOV_QOS_EMPTY) ||
 742                     !cfs_bitmap_check(m->lod_ost_bitmap, ost_idx))
 743                         continue;
 744
 745                 /* Fail Check before osc_precreate() is called
 746                    so we can only 'fail' single OSC. */
 747                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
 748                         continue;
 749
 750                 rc = lod_statfs_and_check(env, m, ost_idx, sfs);
 751                 if (rc) {
 752                         /* this OSP doesn't feel well */
 753                         continue;
 754                 }
 755
 756                 /*
 757                  * skip full devices
 758                  */
 759                 if (lod_qos_dev_is_full(sfs)) {
 760                         QOS_DEBUG("#%d is full\n", ost_idx);
 761                         continue;
 762                 }
 763
 764                 /*
 765                  * We expect number of precreated objects in f_ffree at
 766                  * the first iteration, skip OSPs with no objects ready
 767                  */
 768                 if (sfs->os_fprecreated == 0 && speed == 0) {
 769                         QOS_DEBUG("#%d: precreation is empty\n", ost_idx);
 770                         continue;
 771                 }
 772
 773                 /*
 774                  * try to use another OSP if this one is degraded
 775                  */
 776                 if (sfs->os_state & OS_STATE_DEGRADED && speed < 2) {
 777                         QOS_DEBUG("#%d: degraded\n", ost_idx);
 778                         continue;
 779                 }
 780
 781                 /*
 782                  * do not put >1 objects on a single OST
 783                  */
 784                 if (speed && lod_qos_is_ost_used(env, ost_idx, stripe_idx))
 785                         continue;
 786
 787                 o = lod_qos_declare_object_on(env, m, ost_idx, th);
 788                 if (IS_ERR(o)) {
 789                         CDEBUG(D_OTHER, "can't declare new object on #%u: %d\n",
 790                                ost_idx, (int) PTR_ERR(o));
 791                         rc = PTR_ERR(o);
 792                         continue;
 793                 }
 794
 795                 /*
 796                  * We've successfuly declared (reserved) an object
 797                  */
 798                 lod_qos_ost_in_use(env, stripe_idx, ost_idx);
 799                 stripe[stripe_idx] = o;
 800                 stripe_idx++;
 801
 802         }
 803         if ((speed < 2) && (stripe_idx < stripe_cnt_min)) {
 804                 /* Try again, allowing slower OSCs */
 805                 speed++;
 806                 lqr->lqr_start_idx = ost_start_idx_temp;
 807                 goto repeat_find;
 808         }
 809
 810         up_read(&m->lod_qos.lq_rw_sem);
 811
 812         if (stripe_idx) {
 813                 lo->ldo_stripenr = stripe_idx;
 814                 /* at least one stripe is allocated */
 815                 rc = 0;
 816         } else {
 817                 /* nobody provided us with a single object */
 818                 rc = -ENOSPC;
 819         }
 820
 821 out:
 822         if (pool != NULL) {
 823                 up_read(&pool_tgt_rw_sem(pool));
 824                 /* put back ref got by lod_find_pool() */
 825                 lod_pool_putref(pool);
 826         }
 827
 828         RETURN(rc);
 829 }
 830
 831 /* alloc objects on osts with specific stripe offset */
 832 static int lod_alloc_specific(const struct lu_env *env, struct lod_object *lo,
 833                               struct dt_object **stripe, int flags,
 834                               struct thandle *th)
 835 {
 836         struct lod_device *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
 837         struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
 838         struct dt_object  *o;
 839         unsigned           ost_idx, array_idx, ost_count;
 840         int                i, rc, stripe_num = 0;
 841         int                speed = 0;
 842         struct pool_desc  *pool = NULL;
 843         struct ost_pool   *osts;
 844         ENTRY;
 845
 846         rc = lod_qos_ost_in_use_clear(env, lo->ldo_stripenr);
 847         if (rc)
 848                 GOTO(out, rc);
 849
 850         if (lo->ldo_pool)
 851                 pool = lod_find_pool(m, lo->ldo_pool);
 852
 853         if (pool != NULL) {
 854                 down_read(&pool_tgt_rw_sem(pool));
 855                 osts = &(pool->pool_obds);
 856         } else {
 857                 osts = &(m->lod_pool_info);
 858         }
 859
 860         ost_count = osts->op_count;
 861
 862 repeat_find:
 863         /* search loi_ost_idx in ost array */
 864         array_idx = 0;
 865         for (i = 0; i < ost_count; i++) {
 866                 if (osts->op_array[i] == lo->ldo_def_stripe_offset) {
 867                         array_idx = i;
 868                         break;
 869                 }
 870         }
 871         if (i == ost_count) {
 872                 CERROR("Start index %d not found in pool '%s'\n",
 873                        lo->ldo_def_stripe_offset,
 874                        lo->ldo_pool ? lo->ldo_pool : "");
 875                 GOTO(out, rc = -EINVAL);
 876         }
 877
 878         for (i = 0; i < ost_count;
 879                         i++, array_idx = (array_idx + 1) % ost_count) {
 880                 ost_idx = osts->op_array[array_idx];
 881
 882                 if (!cfs_bitmap_check(m->lod_ost_bitmap, ost_idx))
 883                         continue;
 884
 885                 /* Fail Check before osc_precreate() is called
 886                    so we can only 'fail' single OSC. */
 887                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
 888                         continue;
 889
 890                 /*
 891                  * do not put >1 objects on a single OST
 892                  */
 893                 if (lod_qos_is_ost_used(env, ost_idx, stripe_num))
 894                         continue;
 895
 896                 /* Drop slow OSCs if we can, but not for requested start idx.
 897                  *
 898                  * This means "if OSC is slow and it is not the requested
 899                  * start OST, then it can be skipped, otherwise skip it only
 900                  * if it is inactive/recovering/out-of-space." */
 901
 902                 rc = lod_statfs_and_check(env, m, ost_idx, sfs);
 903                 if (rc) {
 904                         /* this OSP doesn't feel well */
 905                         continue;
 906                 }
 907
 908                 /*
 909                  * We expect number of precreated objects in f_ffree at
 910                  * the first iteration, skip OSPs with no objects ready
 911                  * don't apply this logic to OST specified with stripe_offset
 912                  */
 913                 if (i != 0 && sfs->os_fprecreated == 0 && speed == 0)
 914                         continue;
 915
 916                 o = lod_qos_declare_object_on(env, m, ost_idx, th);
 917                 if (IS_ERR(o)) {
 918                         CDEBUG(D_OTHER, "can't declare new object on #%u: %d\n",
 919                                ost_idx, (int) PTR_ERR(o));
 920                         continue;
 921                 }
 922
 923                 /*
 924                  * We've successfuly declared (reserved) an object
 925                  */
 926                 lod_qos_ost_in_use(env, stripe_num, ost_idx);
 927                 stripe[stripe_num] = o;
 928                 stripe_num++;
 929
 930                 /* We have enough stripes */
 931                 if (stripe_num == lo->ldo_stripenr)
 932                         GOTO(out, rc = 0);
 933         }
 934         if (speed < 2) {
 935                 /* Try again, allowing slower OSCs */
 936                 speed++;
 937                 goto repeat_find;
 938         }
 939
 940         /* If we were passed specific striping params, then a failure to
 941          * meet those requirements is an error, since we can't reallocate
 942          * that memory (it might be part of a larger array or something).
 943          *
 944          * We can only get here if lsm_stripe_count was originally > 1.
 945          */
 946         CERROR("can't lstripe objid "DFID": have %d want %u\n",
 947                PFID(lu_object_fid(lod2lu_obj(lo))), stripe_num,
 948                lo->ldo_stripenr);
 949         rc = -EFBIG;
 950 out:
 951         if (pool != NULL) {
 952                 up_read(&pool_tgt_rw_sem(pool));
 953                 /* put back ref got by lod_find_pool() */
 954                 lod_pool_putref(pool);
 955         }
 956
 957         RETURN(rc);
 958 }
 959
 960 static inline int lod_qos_is_usable(struct lod_device *lod)
 961 {
 962 #ifdef FORCE_QOS
 963         /* to be able to debug QoS code */
 964         return 1;
 965 #endif
 966
 967         /* Detect -EAGAIN early, before expensive lock is taken. */
 968         if (!lod->lod_qos.lq_dirty && lod->lod_qos.lq_same_space)
 969                 return 0;
 970
 971         if (lod->lod_desc.ld_active_tgt_count < 2)
 972                 return 0;
 973
 974         return 1;
 975 }
 976
 977 /* Alloc objects on OSTs with optimization based on:
 978    - free space
 979    - network resources (shared OSS's)
 980  */
 981 static int lod_alloc_qos(const struct lu_env *env, struct lod_object *lo,
 982                          struct dt_object **stripe, int flags,
 983                          struct thandle *th)
 984 {
 985         struct lod_device   *m = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
 986         struct obd_statfs   *sfs = &lod_env_info(env)->lti_osfs;
 987         struct lod_tgt_desc *ost;
 988         struct dt_object    *o;
 989         __u64                total_weight = 0;
 990         int                  nfound, good_osts, i, rc = 0;
 991         int                  stripe_cnt = lo->ldo_stripenr;
 992         int                  stripe_cnt_min;
 993         struct pool_desc    *pool = NULL;
 994         struct ost_pool    *osts;
 995         ENTRY;
 996
 997         stripe_cnt_min = min_stripe_count(stripe_cnt, flags);
 998         if (stripe_cnt_min < 1)
 999                 RETURN(-EINVAL);
1000
1001         if (lo->ldo_pool)
1002                 pool = lod_find_pool(m, lo->ldo_pool);
1003
1004         if (pool != NULL) {
1005                 down_read(&pool_tgt_rw_sem(pool));
1006                 osts = &(pool->pool_obds);
1007         } else {
1008                 osts = &(m->lod_pool_info);
1009         }
1010
1011         /* Detect -EAGAIN early, before expensive lock is taken. */
1012         if (!lod_qos_is_usable(m))
1013                 GOTO(out_nolock, rc = -EAGAIN);
1014
1015         /* Do actual allocation, use write lock here. */
1016         down_write(&m->lod_qos.lq_rw_sem);
1017
1018         /*
1019          * Check again, while we were sleeping on @lq_rw_sem things could
1020          * change.
1021          */
1022         if (!lod_qos_is_usable(m))
1023                 GOTO(out, rc = -EAGAIN);
1024
1025         rc = lod_qos_calc_ppo(m);
1026         if (rc)
1027                 GOTO(out, rc);
1028
1029         rc = lod_qos_ost_in_use_clear(env, lo->ldo_stripenr);
1030         if (rc)
1031                 GOTO(out, rc);
1032
1033         good_osts = 0;
1034         /* Find all the OSTs that are valid stripe candidates */
1035         for (i = 0; i < osts->op_count; i++) {
1036                 if (!cfs_bitmap_check(m->lod_ost_bitmap, osts->op_array[i]))
1037                         continue;
1038
1039                 rc = lod_statfs_and_check(env, m, osts->op_array[i], sfs);
1040                 if (rc) {
1041                         /* this OSP doesn't feel well */
1042                         continue;
1043                 }
1044
1045                 /*
1046                  * skip full devices
1047                  */
1048                 if (lod_qos_dev_is_full(sfs))
1049                         continue;
1050
1051                 /* Fail Check before osc_precreate() is called
1052                    so we can only 'fail' single OSC. */
1053                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) &&
1054                                    osts->op_array[i] == 0)
1055                         continue;
1056
1057                 ost = OST_TGT(m,osts->op_array[i]);
1058                 ost->ltd_qos.ltq_usable = 1;
1059                 lod_qos_calc_weight(m, osts->op_array[i]);
1060                 total_weight += ost->ltd_qos.ltq_weight;
1061
1062                 good_osts++;
1063         }
1064
1065         QOS_DEBUG("found %d good osts\n", good_osts);
1066
1067         if (good_osts < stripe_cnt_min)
1068                 GOTO(out, rc = -EAGAIN);
1069
1070         /* We have enough osts */
1071         if (good_osts < stripe_cnt)
1072                 stripe_cnt = good_osts;
1073
1074         /* Find enough OSTs with weighted random allocation. */
1075         nfound = 0;
1076         while (nfound < stripe_cnt) {
1077                 __u64 rand, cur_weight;
1078
1079                 cur_weight = 0;
1080                 rc = -ENOSPC;
1081
1082                 if (total_weight) {
1083 #if BITS_PER_LONG == 32
1084                         rand = cfs_rand() % (unsigned)total_weight;
1085                         /* If total_weight > 32-bit, first generate the high
1086                          * 32 bits of the random number, then add in the low
1087                          * 32 bits (truncated to the upper limit, if needed) */
1088                         if (total_weight > 0xffffffffULL)
1089                                 rand = (__u64)(cfs_rand() %
1090                                         (unsigned)(total_weight >> 32)) << 32;
1091                         else
1092                                 rand = 0;
1093
1094                         if (rand == (total_weight & 0xffffffff00000000ULL))
1095                                 rand |= cfs_rand() % (unsigned)total_weight;
1096                         else
1097                                 rand |= cfs_rand();
1098
1099 #else
1100                         rand = ((__u64)cfs_rand() << 32 | cfs_rand()) %
1101                                 total_weight;
1102 #endif
1103                 } else {
1104                         rand = 0;
1105                 }
1106
1107                 /* On average, this will hit larger-weighted osts more often.
1108                    0-weight osts will always get used last (only when rand=0) */
1109                 for (i = 0; i < osts->op_count; i++) {
1110                         int idx = osts->op_array[i];
1111
1112                         if (!cfs_bitmap_check(m->lod_ost_bitmap, idx))
1113                                 continue;
1114
1115                         ost = OST_TGT(m,idx);
1116
1117                         if (!ost->ltd_qos.ltq_usable)
1118                                 continue;
1119
1120                         cur_weight += ost->ltd_qos.ltq_weight;
1121                         QOS_DEBUG("stripe_cnt=%d nfound=%d cur_weight="LPU64
1122                                   " rand="LPU64" total_weight="LPU64"\n",
1123                                   stripe_cnt, nfound, cur_weight, rand,
1124                                   total_weight);
1125
1126                         if (cur_weight < rand)
1127                                 continue;
1128
1129                         QOS_DEBUG("stripe=%d to idx=%d\n", nfound, idx);
1130
1131                         /*
1132                          * do not put >1 objects on a single OST
1133                          */
1134                         if (lod_qos_is_ost_used(env, idx, nfound))
1135                                 continue;
1136                         lod_qos_ost_in_use(env, nfound, idx);
1137
1138                         o = lod_qos_declare_object_on(env, m, idx, th);
1139                         if (IS_ERR(o)) {
1140                                 QOS_DEBUG("can't declare object on #%u: %d\n",
1141                                           idx, (int) PTR_ERR(o));
1142                                 continue;
1143                         }
1144                         stripe[nfound++] = o;
1145                         lod_qos_used(m, osts, idx, &total_weight);
1146                         rc = 0;
1147                         break;
1148                 }
1149
1150                 if (rc) {
1151                         /* no OST found on this iteration, give up */
1152                         break;
1153                 }
1154         }
1155
1156         if (unlikely(nfound != stripe_cnt)) {
1157                 /*
1158                  * when the decision to use weighted algorithm was made
1159                  * we had enough appropriate OSPs, but this state can
1160                  * change anytime (no space on OST, broken connection, etc)
1161                  * so it's possible OSP won't be able to provide us with
1162                  * an object due to just changed state
1163                  */
1164                 LCONSOLE_INFO("wanted %d, found %d\n", stripe_cnt, nfound);
1165                 for (i = 0; i < nfound; i++) {
1166                         LASSERT(stripe[i] != NULL);
1167                         lu_object_put(env, &stripe[i]->do_lu);
1168                         stripe[i] = NULL;
1169                 }
1170
1171                 /* makes sense to rebalance next time */
1172                 m->lod_qos.lq_dirty = 1;
1173                 m->lod_qos.lq_same_space = 0;
1174
1175                 rc = -EAGAIN;
1176         }
1177
1178 out:
1179         up_write(&m->lod_qos.lq_rw_sem);
1180
1181 out_nolock:
1182         if (pool != NULL) {
1183                 up_read(&pool_tgt_rw_sem(pool));
1184                 /* put back ref got by lod_find_pool() */
1185                 lod_pool_putref(pool);
1186         }
1187
1188         RETURN(rc);
1189 }
1190
1191 /* Find the max stripecount we should use */
1192 static __u16 lod_get_stripecnt(struct lod_device *lod, __u32 magic,
1193                                __u16 stripe_count)
1194 {
1195         __u32 max_stripes = LOV_MAX_STRIPE_COUNT_OLD;
1196
1197         if (!stripe_count)
1198                 stripe_count = lod->lod_desc.ld_default_stripe_count;
1199         if (stripe_count > lod->lod_desc.ld_active_tgt_count)
1200                 stripe_count = lod->lod_desc.ld_active_tgt_count;
1201         if (!stripe_count)
1202                 stripe_count = 1;
1203
1204         /* stripe count is based on whether OSD can handle larger EA sizes */
1205         if (lod->lod_osd_max_easize > 0)
1206                 max_stripes = lov_mds_md_max_stripe_count(
1207                         lod->lod_osd_max_easize, magic);
1208
1209         return (stripe_count < max_stripes) ? stripe_count : max_stripes;
1210 }
1211
1212 static int lod_use_defined_striping(const struct lu_env *env,
1213                                     struct lod_object *mo,
1214                                     const struct lu_buf *buf)
1215 {
1216         struct lov_mds_md_v1   *v1 = buf->lb_buf;
1217         struct lov_mds_md_v3   *v3 = buf->lb_buf;
1218         struct lov_ost_data_v1 *objs;
1219         __u32                   magic;
1220         int                     rc = 0;
1221         ENTRY;
1222
1223         magic = le32_to_cpu(v1->lmm_magic);
1224         if (magic == LOV_MAGIC_V1_DEF) {
1225                 magic = LOV_MAGIC_V1;
1226                 objs = &v1->lmm_objects[0];
1227         } else if (magic == LOV_MAGIC_V3_DEF) {
1228                 magic = LOV_MAGIC_V3;
1229                 objs = &v3->lmm_objects[0];
1230                 lod_object_set_pool(mo, v3->lmm_pool_name);
1231         } else {
1232                 GOTO(out, rc = -EINVAL);
1233         }
1234
1235         mo->ldo_pattern = le32_to_cpu(v1->lmm_pattern);
1236         mo->ldo_stripe_size = le32_to_cpu(v1->lmm_stripe_size);
1237         mo->ldo_stripenr = le16_to_cpu(v1->lmm_stripe_count);
1238         mo->ldo_layout_gen = le16_to_cpu(v1->lmm_layout_gen);
1239
1240         /* fixup for released file before object initialization */
1241         if (mo->ldo_pattern & LOV_PATTERN_F_RELEASED) {
1242                 mo->ldo_released_stripenr = mo->ldo_stripenr;
1243                 mo->ldo_stripenr = 0;
1244         }
1245
1246         LASSERT(buf->lb_len >= lov_mds_md_size(mo->ldo_stripenr, magic));
1247
1248         if (mo->ldo_stripenr > 0)
1249                 rc = lod_initialize_objects(env, mo, objs);
1250
1251 out:
1252         RETURN(rc);
1253 }
1254
1255 static int lod_qos_parse_config(const struct lu_env *env,
1256                                 struct lod_object *lo,
1257                                 const struct lu_buf *buf)
1258 {
1259         struct lod_device     *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
1260         struct lov_user_md_v1 *v1 = NULL;
1261         struct lov_user_md_v3 *v3 = NULL;
1262         struct pool_desc      *pool;
1263         __u32                  magic;
1264         int                    rc;
1265         ENTRY;
1266
1267         if (buf == NULL || buf->lb_buf == NULL || buf->lb_len == 0)
1268                 RETURN(0);
1269
1270         v1 = buf->lb_buf;
1271         magic = v1->lmm_magic;
1272
1273         if (magic == __swab32(LOV_USER_MAGIC_V1)) {
1274                 lustre_swab_lov_user_md_v1(v1);
1275                 magic = v1->lmm_magic;
1276         } else if (magic == __swab32(LOV_USER_MAGIC_V3)) {
1277                 v3 = buf->lb_buf;
1278                 lustre_swab_lov_user_md_v3(v3);
1279                 magic = v3->lmm_magic;
1280         }
1281
1282         if (unlikely(magic != LOV_MAGIC_V1 && magic != LOV_MAGIC_V3)) {
1283                 /* try to use as fully defined striping */
1284                 rc = lod_use_defined_striping(env, lo, buf);
1285                 RETURN(rc);
1286         }
1287
1288         if (unlikely(buf->lb_len < sizeof(*v1))) {
1289                 CERROR("wrong size: %u\n", (unsigned) buf->lb_len);
1290                 RETURN(-EINVAL);
1291         }
1292
1293         v1->lmm_magic = magic;
1294         if (v1->lmm_pattern == 0)
1295                 v1->lmm_pattern = LOV_PATTERN_RAID0;
1296         if (lov_pattern(v1->lmm_pattern) != LOV_PATTERN_RAID0) {
1297                 CERROR("invalid pattern: %x\n", v1->lmm_pattern);
1298                 RETURN(-EINVAL);
1299         }
1300         lo->ldo_pattern = v1->lmm_pattern;
1301
1302         if (v1->lmm_stripe_size)
1303                 lo->ldo_stripe_size = v1->lmm_stripe_size;
1304         if (lo->ldo_stripe_size & (LOV_MIN_STRIPE_SIZE - 1))
1305                 lo->ldo_stripe_size = LOV_MIN_STRIPE_SIZE;
1306
1307         if (v1->lmm_stripe_count)
1308                 lo->ldo_stripenr = v1->lmm_stripe_count;
1309
1310         if ((v1->lmm_stripe_offset >= d->lod_desc.ld_tgt_count) &&
1311             (v1->lmm_stripe_offset != (typeof(v1->lmm_stripe_offset))(-1))) {
1312                 CERROR("invalid offset: %x\n", v1->lmm_stripe_offset);
1313                 RETURN(-EINVAL);
1314         }
1315         lo->ldo_def_stripe_offset = v1->lmm_stripe_offset;
1316
1317         CDEBUG(D_OTHER, "lsm: %u size, %u stripes, %u offset\n",
1318                v1->lmm_stripe_size, v1->lmm_stripe_count,
1319                v1->lmm_stripe_offset);
1320
1321         if (v1->lmm_magic == LOV_MAGIC_V3) {
1322                 if (buf->lb_len < sizeof(*v3)) {
1323                         CERROR("wrong size: %u\n", (unsigned) buf->lb_len);
1324                         RETURN(-EINVAL);
1325                 }
1326
1327                 v3 = buf->lb_buf;
1328                 lod_object_set_pool(lo, v3->lmm_pool_name);
1329
1330                 /* In the function below, .hs_keycmp resolves to
1331                  * pool_hashkey_keycmp() */
1332                 /* coverity[overrun-buffer-val] */
1333                 pool = lod_find_pool(d, v3->lmm_pool_name);
1334                 if (pool != NULL) {
1335                         if (lo->ldo_def_stripe_offset !=
1336                             (typeof(v1->lmm_stripe_offset))(-1)) {
1337                                 rc = lo->ldo_def_stripe_offset;
1338                                 rc = lod_check_index_in_pool(rc, pool);
1339                                 if (rc < 0) {
1340                                         lod_pool_putref(pool);
1341                                         CERROR("invalid offset\n");
1342                                         RETURN(-EINVAL);
1343                                 }
1344                         }
1345
1346                         if (lo->ldo_stripenr > pool_tgt_count(pool))
1347                                 lo->ldo_stripenr= pool_tgt_count(pool);
1348
1349                         lod_pool_putref(pool);
1350                 }
1351         } else
1352                 lod_object_set_pool(lo, NULL);
1353
1354         /* fixup for released file */
1355         if (lo->ldo_pattern & LOV_PATTERN_F_RELEASED) {
1356                 lo->ldo_released_stripenr = lo->ldo_stripenr;
1357                 lo->ldo_stripenr = 0;
1358         }
1359
1360         RETURN(0);
1361 }
1362
1363 /*
1364  * buf should be NULL or contain striping settings
1365  */
1366 int lod_qos_prep_create(const struct lu_env *env, struct lod_object *lo,
1367                         struct lu_attr *attr, const struct lu_buf *buf,
1368                         struct thandle *th)
1369 {
1370         struct lod_device      *d = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
1371         struct dt_object      **stripe;
1372         int                     stripe_len;
1373         int                     flag = LOV_USES_ASSIGNED_STRIPE;
1374         int                     i, rc;
1375         ENTRY;
1376
1377         LASSERT(lo);
1378
1379         /* no OST available */
1380         /* XXX: should we be waiting a bit to prevent failures during
1381          * cluster initialization? */
1382         if (d->lod_ostnr == 0)
1383                 GOTO(out, rc = -EIO);
1384
1385         /*
1386          * by this time, the object's ldo_stripenr and ldo_stripe_size
1387          * contain default value for striping: taken from the parent
1388          * or from filesystem defaults
1389          *
1390          * in case the caller is passing lovea with new striping config,
1391          * we may need to parse lovea and apply new configuration
1392          */
1393         rc = lod_qos_parse_config(env, lo, buf);
1394         if (rc)
1395                 GOTO(out, rc);
1396
1397         /* A released file is being created */
1398         if (lo->ldo_stripenr == 0)
1399                 GOTO(out, rc = 0);
1400
1401         if (likely(lo->ldo_stripe == NULL)) {
1402                 /*
1403                  * no striping has been created so far
1404                  */
1405                 LASSERT(lo->ldo_stripenr > 0);
1406                 /*
1407                  * statfs and check OST targets now, since ld_active_tgt_count
1408                  * could be changed if some OSTs are [de]activated manually.
1409                  */
1410                 lod_qos_statfs_update(env, d);
1411                 lo->ldo_stripenr = lod_get_stripecnt(d, LOV_MAGIC,
1412                                 lo->ldo_stripenr);
1413
1414                 stripe_len = lo->ldo_stripenr;
1415                 OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_len);
1416                 if (stripe == NULL)
1417                         GOTO(out, rc = -ENOMEM);
1418
1419                 lod_getref(&d->lod_ost_descs);
1420                 /* XXX: support for non-0 files w/o objects */
1421                 CDEBUG(D_OTHER, "tgt_count %d stripenr %d\n",
1422                                 d->lod_desc.ld_tgt_count, stripe_len);
1423                 if (lo->ldo_def_stripe_offset >= d->lod_desc.ld_tgt_count) {
1424                         rc = lod_alloc_qos(env, lo, stripe, flag, th);
1425                         if (rc == -EAGAIN)
1426                                 rc = lod_alloc_rr(env, lo, stripe, flag, th);
1427                 } else {
1428                         rc = lod_alloc_specific(env, lo, stripe, flag, th);
1429                 }
1430                 lod_putref(d, &d->lod_ost_descs);
1431
1432                 if (rc < 0) {
1433                         for (i = 0; i < stripe_len; i++)
1434                                 if (stripe[i] != NULL)
1435                                         lu_object_put(env, &stripe[i]->do_lu);
1436
1437                         OBD_FREE(stripe, sizeof(stripe[0]) * stripe_len);
1438                 } else {
1439                         lo->ldo_stripe = stripe;
1440                         lo->ldo_stripes_allocated = stripe_len;
1441                 }
1442         } else {
1443                 /*
1444                  * lod_qos_parse_config() found supplied buf as a predefined
1445                  * striping (not a hint), so it allocated all the object
1446                  * now we need to create them
1447                  */
1448                 for (i = 0; i < lo->ldo_stripenr; i++) {
1449                         struct dt_object  *o;
1450
1451                         o = lo->ldo_stripe[i];
1452                         LASSERT(o);
1453
1454                         rc = dt_declare_create(env, o, attr, NULL, NULL, th);
1455                         if (rc) {
1456                                 CERROR("can't declare create: %d\n", rc);
1457                                 break;
1458                         }
1459                 }
1460         }
1461
1462 out:
1463         RETURN(rc);
1464 }
1465