lustre/lov/lov_qos.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #ifndef EXPORT_SYMTAB
  38 # define EXPORT_SYMTAB
  39 #endif
  40 #define DEBUG_SUBSYSTEM S_LOV
  41
  42 #ifdef __KERNEL__
  43 #include <libcfs/libcfs.h>
  44 #else
  45 #include <liblustre.h>
  46 #endif
  47
  48 #include <obd_class.h>
  49 #include <obd_lov.h>
  50 #include "lov_internal.h"
  51
  52 /* #define QOS_DEBUG 1 */
  53 #define D_QOS D_OTHER
  54
  55 #define TGT_BAVAIL(i) (lov->lov_tgts[i]->ltd_exp->exp_obd->obd_osfs.os_bavail *\
  56                        lov->lov_tgts[i]->ltd_exp->exp_obd->obd_osfs.os_bsize)
  57 #define TGT_FFREE(i)  (lov->lov_tgts[i]->ltd_exp->exp_obd->obd_osfs.os_ffree)
  58
  59
  60 int qos_add_tgt(struct obd_device *obd, __u32 index)
  61 {
  62         struct lov_obd *lov = &obd->u.lov;
  63         struct lov_qos_oss *oss, *temposs;
  64         struct obd_export *exp = lov->lov_tgts[index]->ltd_exp;
  65         int rc = 0, found = 0;
  66         ENTRY;
  67
  68         /* We only need this QOS struct on MDT, not clients - but we may not
  69          * have registered the LOV's observer yet, so there's no way to know */
  70         if (!exp || !exp->exp_connection) {
  71                 CERROR("Missing connection\n");
  72                 RETURN(-ENOTCONN);
  73         }
  74
  75         down_write(&lov->lov_qos.lq_rw_sem);
  76         mutex_down(&lov->lov_lock);
  77         list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
  78                 if (obd_uuid_equals(&oss->lqo_uuid,
  79                                     &exp->exp_connection->c_remote_uuid)) {
  80                         found++;
  81                         break;
  82                 }
  83         }
  84
  85         if (!found) {
  86                 OBD_ALLOC_PTR(oss);
  87                 if (!oss)
  88                         GOTO(out, rc = -ENOMEM);
  89                 memcpy(&oss->lqo_uuid,
  90                        &exp->exp_connection->c_remote_uuid,
  91                        sizeof(oss->lqo_uuid));
  92         } else {
  93                 /* Assume we have to move this one */
  94                 list_del(&oss->lqo_oss_list);
  95         }
  96
  97         oss->lqo_ost_count++;
  98         lov->lov_tgts[index]->ltd_qos.ltq_oss = oss;
  99
 100         /* Add sorted by # of OSTs.  Find the first entry that we're
 101            bigger than... */
 102         list_for_each_entry(temposs, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
 103                 if (oss->lqo_ost_count > temposs->lqo_ost_count)
 104                         break;
 105         }
 106         /* ...and add before it.  If we're the first or smallest, temposs
 107            points to the list head, and we add to the end. */
 108         list_add_tail(&oss->lqo_oss_list, &temposs->lqo_oss_list);
 109
 110         lov->lov_qos.lq_dirty = 1;
 111         lov->lov_qos.lq_dirty_rr = 1;
 112
 113         CDEBUG(D_QOS, "add tgt %s to OSS %s (%d OSTs)\n",
 114                obd_uuid2str(&lov->lov_tgts[index]->ltd_uuid),
 115                obd_uuid2str(&oss->lqo_uuid),
 116                oss->lqo_ost_count);
 117
 118 out:
 119         mutex_up(&lov->lov_lock);
 120         up_write(&lov->lov_qos.lq_rw_sem);
 121         RETURN(rc);
 122 }
 123
 124 int qos_del_tgt(struct obd_device *obd, __u32 index)
 125 {
 126         struct lov_obd *lov = &obd->u.lov;
 127         struct lov_qos_oss *oss;
 128         int rc = 0;
 129         ENTRY;
 130
 131         if (!lov->lov_tgts[index])
 132                 RETURN(0);
 133
 134         down_write(&lov->lov_qos.lq_rw_sem);
 135
 136         oss = lov->lov_tgts[index]->ltd_qos.ltq_oss;
 137         if (!oss)
 138                 GOTO(out, rc = -ENOENT);
 139
 140         oss->lqo_ost_count--;
 141         if (oss->lqo_ost_count == 0) {
 142                 CDEBUG(D_QOS, "removing OSS %s\n",
 143                        obd_uuid2str(&oss->lqo_uuid));
 144                 list_del(&oss->lqo_oss_list);
 145                 OBD_FREE_PTR(oss);
 146         }
 147
 148         lov->lov_qos.lq_dirty = 1;
 149         lov->lov_qos.lq_dirty_rr = 1;
 150 out:
 151         up_write(&lov->lov_qos.lq_rw_sem);
 152         RETURN(rc);
 153 }
 154
 155 /* Recalculate per-object penalties for OSSs and OSTs,
 156    depends on size of each ost in an oss */
 157 static int qos_calc_ppo(struct obd_device *obd)
 158 {
 159         struct lov_obd *lov = &obd->u.lov;
 160         struct lov_qos_oss *oss;
 161         __u64 ba_max, ba_min, temp;
 162         __u32 num_active;
 163         int rc, i, prio_wide;
 164         ENTRY;
 165
 166         if (!lov->lov_qos.lq_dirty)
 167                 GOTO(out, rc = 0);
 168
 169         num_active = lov->desc.ld_active_tgt_count - 1;
 170         if (num_active < 1)
 171                 GOTO(out, rc = -EAGAIN);
 172
 173         /* find bavail on each OSS */
 174         list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
 175                 oss->lqo_bavail = 0;
 176         }
 177         lov->lov_qos.lq_active_oss_count = 0;
 178
 179         /* How badly user wants to select osts "widely" (not recently chosen
 180            and not on recent oss's).  As opposed to "freely" (free space
 181            avail.) 0-256. */
 182         prio_wide = 256 - lov->lov_qos.lq_prio_free;
 183
 184         ba_min = (__u64)(-1);
 185         ba_max = 0;
 186         /* Calculate OST penalty per object */
 187         /* (lov ref taken in alloc_qos) */
 188         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
 189                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active)
 190                         continue;
 191                 temp = TGT_BAVAIL(i);
 192                 if (!temp)
 193                         continue;
 194                 ba_min = min(temp, ba_min);
 195                 ba_max = max(temp, ba_max);
 196
 197                 /* Count the number of usable OSS's */
 198                 if (lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_bavail == 0)
 199                         lov->lov_qos.lq_active_oss_count++;
 200                 lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_bavail += temp;
 201
 202                 /* per-OST penalty is prio * TGT_bavail / (num_ost - 1) / 2 */
 203                 temp >>= 1;
 204                 do_div(temp, num_active);
 205                 lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj =
 206                         (temp * prio_wide) >> 8;
 207
 208                 if (lov->lov_qos.lq_reset == 0)
 209                         lov->lov_tgts[i]->ltd_qos.ltq_penalty = 0;
 210         }
 211
 212         num_active = lov->lov_qos.lq_active_oss_count - 1;
 213         if (num_active < 1) {
 214                 /* If there's only 1 OSS, we can't penalize it, so instead
 215                    we have to double the OST penalty */
 216                 num_active = 1;
 217                 for (i = 0; i < lov->desc.ld_tgt_count; i++) {
 218                         if (lov->lov_tgts[i] == NULL)
 219                                 continue;
 220                         lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj <<= 1;
 221                 }
 222         }
 223
 224         /* Per-OSS penalty is prio * oss_avail / oss_osts / (num_oss - 1) / 2 */
 225         list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
 226                 temp = oss->lqo_bavail >> 1;
 227                 do_div(temp, oss->lqo_ost_count * num_active);
 228                 oss->lqo_penalty_per_obj = (temp * prio_wide) >> 8;
 229                 if (lov->lov_qos.lq_reset == 0)
 230                         oss->lqo_penalty = 0;
 231         }
 232
 233         lov->lov_qos.lq_dirty = 0;
 234         lov->lov_qos.lq_reset = 0;
 235
 236         /* If each ost has almost same free space,
 237          * do rr allocation for better creation performance */
 238         lov->lov_qos.lq_same_space = 0;
 239         temp = ba_max - ba_min;
 240         ba_min = (ba_min * 51) >> 8;     /* 51/256 = .20 */
 241         if (temp < ba_min) {
 242                 /* Difference is less than 20% */
 243                 lov->lov_qos.lq_same_space = 1;
 244                 /* Reset weights for the next time we enter qos mode */
 245                 lov->lov_qos.lq_reset = 0;
 246         }
 247         rc = 0;
 248
 249 out:
 250         if (!rc && lov->lov_qos.lq_same_space)
 251                 RETURN(-EAGAIN);
 252         RETURN(rc);
 253 }
 254
 255 static int qos_calc_weight(struct lov_obd *lov, int i)
 256 {
 257         __u64 temp, temp2;
 258
 259         /* Final ost weight = TGT_BAVAIL - ost_penalty - oss_penalty */
 260         temp = TGT_BAVAIL(i);
 261         temp2 = lov->lov_tgts[i]->ltd_qos.ltq_penalty +
 262                 lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_penalty;
 263         if (temp < temp2)
 264                 lov->lov_tgts[i]->ltd_qos.ltq_weight = 0;
 265         else
 266                 lov->lov_tgts[i]->ltd_qos.ltq_weight = temp - temp2;
 267         return 0;
 268 }
 269
 270 /* We just used this index for a stripe; adjust everyone's weights */
 271 static int qos_used(struct lov_obd *lov, __u32 index, __u64 *total_wt)
 272 {
 273         struct lov_qos_oss *oss;
 274         int i;
 275         ENTRY;
 276
 277         /* Don't allocate from this stripe anymore, until the next alloc_qos */
 278         lov->lov_tgts[index]->ltd_qos.ltq_usable = 0;
 279
 280         oss = lov->lov_tgts[index]->ltd_qos.ltq_oss;
 281
 282         /* Decay old penalty by half (we're adding max penalty, and don't
 283            want it to run away.) */
 284         lov->lov_tgts[index]->ltd_qos.ltq_penalty >>= 1;
 285         oss->lqo_penalty >>= 1;
 286
 287         /* Set max penalties for this OST and OSS */
 288         lov->lov_tgts[index]->ltd_qos.ltq_penalty +=
 289                 lov->lov_tgts[index]->ltd_qos.ltq_penalty_per_obj *
 290                 lov->desc.ld_active_tgt_count;
 291         oss->lqo_penalty += oss->lqo_penalty_per_obj *
 292                 lov->lov_qos.lq_active_oss_count;
 293
 294         /* Decrease all OSS penalties */
 295         list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
 296                 if (oss->lqo_penalty < oss->lqo_penalty_per_obj)
 297                         oss->lqo_penalty = 0;
 298                 else
 299                         oss->lqo_penalty -= oss->lqo_penalty_per_obj;
 300         }
 301
 302         *total_wt = 0;
 303         /* Decrease all OST penalties */
 304         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
 305                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active)
 306                         continue;
 307                 if (lov->lov_tgts[i]->ltd_qos.ltq_penalty <
 308                     lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj)
 309                         lov->lov_tgts[i]->ltd_qos.ltq_penalty = 0;
 310                 else
 311                         lov->lov_tgts[i]->ltd_qos.ltq_penalty -=
 312                         lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj;
 313
 314                 qos_calc_weight(lov, i);
 315
 316                 /* Recalc the total weight of usable osts */
 317                 if (lov->lov_tgts[i]->ltd_qos.ltq_usable)
 318                         *total_wt += lov->lov_tgts[i]->ltd_qos.ltq_weight;
 319
 320 #ifdef QOS_DEBUG
 321                 CDEBUG(D_QOS, "recalc tgt %d avail="LPU64
 322                        " ostppo="LPU64" ostp="LPU64" ossppo="LPU64
 323                        " ossp="LPU64" wt="LPU64"\n",
 324                        i, TGT_BAVAIL(i) >> 10,
 325                        lov->lov_tgts[i]->ltd_qos.ltq_penalty_per_obj >> 10,
 326                        lov->lov_tgts[i]->ltd_qos.ltq_penalty >> 10,
 327                        lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_penalty_per_obj>>10,
 328                        lov->lov_tgts[i]->ltd_qos.ltq_oss->lqo_penalty >> 10,
 329                        lov->lov_tgts[i]->ltd_qos.ltq_weight >> 10);
 330 #endif
 331         }
 332
 333         RETURN(0);
 334 }
 335
 336 #define LOV_QOS_EMPTY ((__u32)-1)
 337 /* compute optimal round-robin order, based on OSTs per OSS */
 338 static int qos_calc_rr(struct lov_obd *lov)
 339 {
 340         struct lov_qos_oss *oss;
 341         unsigned ost_count, placed, real_count;
 342         int i;
 343         ENTRY;
 344
 345         if (!lov->lov_qos.lq_dirty_rr) {
 346                 LASSERT(lov->lov_qos.lq_rr_size);
 347                 RETURN(0);
 348         }
 349
 350         /* Do actual allocation. */
 351         down_write(&lov->lov_qos.lq_rw_sem);
 352
 353         /*
 354          * Check again. While we were sleeping on @lq_rw_sem something could
 355          * change.
 356          */
 357         if (!lov->lov_qos.lq_dirty_rr) {
 358                 LASSERT(lov->lov_qos.lq_rr_size);
 359                 up_write(&lov->lov_qos.lq_rw_sem);
 360                 RETURN(0);
 361         }
 362
 363         ost_count = lov->desc.ld_tgt_count;
 364
 365         if (lov->lov_qos.lq_rr_size)
 366                 OBD_FREE(lov->lov_qos.lq_rr_array, lov->lov_qos.lq_rr_size);
 367         lov->lov_qos.lq_rr_size = ost_count *
 368                 sizeof(lov->lov_qos.lq_rr_array[0]);
 369         OBD_ALLOC(lov->lov_qos.lq_rr_array, lov->lov_qos.lq_rr_size);
 370         if (!lov->lov_qos.lq_rr_array) {
 371                 lov->lov_qos.lq_rr_size = 0;
 372                 up_write(&lov->lov_qos.lq_rw_sem);
 373                 RETURN(-ENOMEM);
 374         }
 375
 376         real_count = 0;
 377         for (i = 0; i < ost_count; i++) {
 378                 lov->lov_qos.lq_rr_array[i] = LOV_QOS_EMPTY;
 379                 if (lov->lov_tgts[i])
 380                         real_count++;
 381         }
 382
 383         /* Place all the OSTs from 1 OSS at the same time. */
 384         placed = 0;
 385         list_for_each_entry(oss, &lov->lov_qos.lq_oss_list, lqo_oss_list) {
 386                 int j = 0;
 387                 for (i = 0; i < ost_count; i++) {
 388                         if (lov->lov_tgts[i] &&
 389                             lov->lov_tgts[i]->ltd_qos.ltq_oss == oss) {
 390                               /* Evenly space these OSTs across arrayspace */
 391                               int next = j * ost_count / oss->lqo_ost_count;
 392                               LASSERT(next < ost_count);
 393                               while (lov->lov_qos.lq_rr_array[next] !=
 394                                      LOV_QOS_EMPTY)
 395                                       next = (next + 1) % ost_count;
 396                               lov->lov_qos.lq_rr_array[next] = i;
 397                               j++;
 398                               placed++;
 399                         }
 400                 }
 401                 LASSERT(j == oss->lqo_ost_count);
 402         }
 403
 404         lov->lov_qos.lq_dirty_rr = 0;
 405         up_write(&lov->lov_qos.lq_rw_sem);
 406
 407         if (placed != real_count) {
 408                 /* This should never happen */
 409                 LCONSOLE_ERROR_MSG(0x14e, "Failed to place all OSTs in the "
 410                                    "round-robin list (%d of %d).\n",
 411                                    placed, real_count);
 412                 for (i = 0; i < ost_count; i++) {
 413                         LCONSOLE(D_WARNING, "rr #%d ost idx=%d\n", i,
 414                                  lov->lov_qos.lq_rr_array[i]);
 415                 }
 416                 lov->lov_qos.lq_dirty_rr = 1;
 417                 RETURN(-EAGAIN);
 418         }
 419
 420 #ifdef QOS_DEBUG
 421         for (i = 0; i < ost_count; i++) {
 422                 LCONSOLE(D_QOS, "rr #%d ost idx=%d\n", i,
 423                          lov->lov_qos.lq_rr_array[i]);
 424         }
 425 #endif
 426
 427         RETURN(0);
 428 }
 429
 430
 431 void qos_shrink_lsm(struct lov_request_set *set)
 432 {
 433         struct lov_stripe_md *lsm = set->set_oi->oi_md, *lsm_new;
 434         /* XXX LOV STACKING call into osc for sizes */
 435         unsigned oldsize, newsize;
 436
 437         if (set->set_oti && set->set_cookies && set->set_cookie_sent) {
 438                 struct llog_cookie *cookies;
 439                 oldsize = lsm->lsm_stripe_count * sizeof(*cookies);
 440                 newsize = set->set_count * sizeof(*cookies);
 441
 442                 cookies = set->set_cookies;
 443                 oti_alloc_cookies(set->set_oti, set->set_count);
 444                 if (set->set_oti->oti_logcookies) {
 445                         memcpy(set->set_oti->oti_logcookies, cookies, newsize);
 446                         OBD_FREE(cookies, oldsize);
 447                         set->set_cookies = set->set_oti->oti_logcookies;
 448                 } else {
 449                         CWARN("'leaking' %d bytes\n", oldsize - newsize);
 450                 }
 451         }
 452
 453         CWARN("using fewer stripes for object "LPU64": old %u new %u\n",
 454               lsm->lsm_object_id, lsm->lsm_stripe_count, set->set_count);
 455         LASSERT(lsm->lsm_stripe_count >= set->set_count);
 456
 457         newsize = lov_stripe_md_size(set->set_count);
 458         OBD_ALLOC(lsm_new, newsize);
 459         if (lsm_new != NULL) {
 460                 int i;
 461                 memcpy(lsm_new, lsm, sizeof(*lsm));
 462                 for (i = 0; i < lsm->lsm_stripe_count; i++) {
 463                         if (i < set->set_count) {
 464                                 lsm_new->lsm_oinfo[i] = lsm->lsm_oinfo[i];
 465                                 continue;
 466                         }
 467                         OBD_SLAB_FREE(lsm->lsm_oinfo[i], lov_oinfo_slab,
 468                                       sizeof(struct lov_oinfo));
 469                 }
 470                 lsm_new->lsm_stripe_count = set->set_count;
 471                 OBD_FREE(lsm, sizeof(struct lov_stripe_md) +
 472                          lsm->lsm_stripe_count * sizeof(struct lov_oinfo *));
 473                 set->set_oi->oi_md = lsm_new;
 474         } else {
 475                 CWARN("'leaking' few bytes\n");
 476         }
 477 }
 478
 479 int qos_remedy_create(struct lov_request_set *set, struct lov_request *req)
 480 {
 481         struct lov_stripe_md *lsm = set->set_oi->oi_md;
 482         struct lov_obd *lov = &set->set_exp->exp_obd->u.lov;
 483         unsigned ost_idx, ost_count = lov->desc.ld_tgt_count;
 484         int stripe, i, rc = -EIO;
 485         ENTRY;
 486
 487         ost_idx = (req->rq_idx + lsm->lsm_stripe_count) % ost_count;
 488         for (i = 0; i < ost_count; i++, ost_idx = (ost_idx + 1) % ost_count) {
 489                 if (!lov->lov_tgts[ost_idx] ||
 490                     !lov->lov_tgts[ost_idx]->ltd_active)
 491                         continue;
 492                 /* check if objects has been created on this ost */
 493                 for (stripe = 0; stripe < lsm->lsm_stripe_count; stripe++) {
 494                         if (stripe == req->rq_stripe)
 495                                 continue;
 496                         if (ost_idx == lsm->lsm_oinfo[stripe]->loi_ost_idx)
 497                                 break;
 498                 }
 499
 500                 if (stripe >= lsm->lsm_stripe_count) {
 501                         req->rq_idx = ost_idx;
 502                         rc = obd_create(lov->lov_tgts[ost_idx]->ltd_exp,
 503                                         req->rq_oi.oi_oa, &req->rq_oi.oi_md,
 504                                         set->set_oti);
 505                         if (!rc)
 506                                 break;
 507                 }
 508         }
 509         RETURN(rc);
 510 }
 511
 512 static int min_stripe_count(int stripe_cnt, int flags)
 513 {
 514         return (flags & LOV_USES_DEFAULT_STRIPE ?
 515                 stripe_cnt - (stripe_cnt / 4) : stripe_cnt);
 516 }
 517
 518 #define LOV_CREATE_RESEED_MULT 4
 519 #define LOV_CREATE_RESEED_MIN  1000
 520 /* Allocate objects on osts with round-robin algorithm */
 521 static int alloc_rr(struct lov_obd *lov, int *idx_arr, int *stripe_cnt,
 522                     int flags)
 523 {
 524         unsigned array_idx, ost_count = lov->desc.ld_tgt_count;
 525         unsigned ost_active_count = lov->desc.ld_active_tgt_count;
 526         int i, *idx_pos;
 527         __u32 ost_idx;
 528         int ost_start_idx_temp;
 529         int speed = 0;
 530         int stripe_cnt_min = min_stripe_count(*stripe_cnt, flags);
 531         ENTRY;
 532
 533         i = qos_calc_rr(lov);
 534         if (i)
 535                 RETURN(i);
 536
 537         if (--lov->lov_start_count <= 0) {
 538                 lov->lov_start_idx = ll_rand() % ost_count;
 539                 lov->lov_start_count =
 540                         (LOV_CREATE_RESEED_MIN / max(ost_active_count, 1U) +
 541                          LOV_CREATE_RESEED_MULT) * max(ost_active_count, 1U);
 542         } else if (stripe_cnt_min >= ost_active_count ||
 543                    lov->lov_start_idx > ost_count) {
 544                 /* If we have allocated from all of the OSTs, slowly
 545                  * precess the next start if the OST/stripe count isn't
 546                  * already doing this for us. */
 547                 lov->lov_start_idx %= ost_count;
 548                 if (*stripe_cnt > 1 && (ost_active_count % (*stripe_cnt)) != 1)
 549                         ++lov->lov_offset_idx;
 550         }
 551         down_read(&lov->lov_qos.lq_rw_sem);
 552         ost_start_idx_temp = lov->lov_start_idx;
 553
 554 repeat_find:
 555         array_idx = (lov->lov_start_idx + lov->lov_offset_idx) % ost_count;
 556         idx_pos = idx_arr;
 557 #ifdef QOS_DEBUG
 558         CDEBUG(D_QOS, "want %d startidx %d startcnt %d offset %d active %d "
 559                "count %d arrayidx %d\n",
 560                stripe_cnt, lov->lov_start_idx, lov->lov_start_count,
 561                lov->lov_offset_idx, ost_active_count, ost_count, array_idx);
 562 #endif
 563
 564         for (i = 0; i < ost_count; i++, array_idx=(array_idx + 1) % ost_count) {
 565                 ++lov->lov_start_idx;
 566                 ost_idx = lov->lov_qos.lq_rr_array[array_idx];
 567 #ifdef QOS_DEBUG
 568                 CDEBUG(D_QOS, "#%d strt %d act %d strp %d ary %d idx %d\n",
 569                        i, lov->lov_start_idx,
 570                        ((ost_idx != LOV_QOS_EMPTY) && lov->lov_tgts[ost_idx]) ?
 571                        lov->lov_tgts[ost_idx]->ltd_active : 0,
 572                        idx_pos - idx_arr, array_idx, ost_idx);
 573 #endif
 574                 if ((ost_idx == LOV_QOS_EMPTY) || !lov->lov_tgts[ost_idx] ||
 575                     !lov->lov_tgts[ost_idx]->ltd_active)
 576                         continue;
 577
 578                 /* Fail Check before osc_precreate() is called
 579                    so we can only 'fail' single OSC. */
 580                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
 581                         continue;
 582
 583                 /* Drop slow OSCs if we can */
 584                 if (obd_precreate(lov->lov_tgts[ost_idx]->ltd_exp) > speed)
 585                         continue;
 586
 587                 *idx_pos = ost_idx;
 588                 idx_pos++;
 589                 /* We have enough stripes */
 590                 if (idx_pos - idx_arr == *stripe_cnt)
 591                         break;
 592         }
 593         if ((speed < 2) && (idx_pos - idx_arr < stripe_cnt_min)) {
 594                 /* Try again, allowing slower OSCs */
 595                 speed++;
 596                 lov->lov_start_idx = ost_start_idx_temp;
 597                 goto repeat_find;
 598         }
 599
 600         up_read(&lov->lov_qos.lq_rw_sem);
 601
 602         *stripe_cnt = idx_pos - idx_arr;
 603         RETURN(0);
 604 }
 605
 606 /* alloc objects on osts with specific stripe offset */
 607 static int alloc_specific(struct lov_obd *lov, struct lov_stripe_md *lsm,
 608                           int *idx_arr)
 609 {
 610         unsigned ost_idx, ost_count = lov->desc.ld_tgt_count;
 611         int i, *idx_pos;
 612         int speed = 0;
 613         ENTRY;
 614
 615 repeat_find:
 616         ost_idx = lsm->lsm_oinfo[0]->loi_ost_idx;
 617         idx_pos = idx_arr;
 618         for (i = 0; i < ost_count; i++, ost_idx = (ost_idx + 1) % ost_count) {
 619                 if (!lov->lov_tgts[ost_idx] ||
 620                     !lov->lov_tgts[ost_idx]->ltd_active) {
 621                         continue;
 622                 }
 623
 624                 /* Fail Check before osc_precreate() is called
 625                    so we can only 'fail' single OSC. */
 626                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && ost_idx == 0)
 627                         continue;
 628
 629                 /* Drop slow OSCs if we can, but not for requested start idx */
 630                 if ((obd_precreate(lov->lov_tgts[ost_idx]->ltd_exp) > speed) &&
 631                     (i != 0 || speed < 2))
 632                         continue;
 633
 634                 *idx_pos = ost_idx;
 635                 idx_pos++;
 636                 /* We have enough stripes */
 637                 if (idx_pos - idx_arr == lsm->lsm_stripe_count)
 638                         RETURN(0);
 639         }
 640         if (speed < 2) {
 641                 /* Try again, allowing slower OSCs */
 642                 speed++;
 643                 goto repeat_find;
 644         }
 645
 646         /* If we were passed specific striping params, then a failure to
 647          * meet those requirements is an error, since we can't reallocate
 648          * that memory (it might be part of a larger array or something).
 649          *
 650          * We can only get here if lsm_stripe_count was originally > 1.
 651          */
 652         CERROR("can't lstripe objid "LPX64": have %d want %u\n",
 653                lsm->lsm_object_id, (int)(idx_pos - idx_arr),
 654                lsm->lsm_stripe_count);
 655         RETURN(-EFBIG);
 656 }
 657
 658 /* Alloc objects on osts with optimization based on:
 659    - free space
 660    - network resources (shared OSS's)
 661 */
 662 static int alloc_qos(struct obd_export *exp, int *idx_arr, int *stripe_cnt,
 663                      int flags)
 664 {
 665         struct lov_obd *lov = &exp->exp_obd->u.lov;
 666         static time_t last_warn = 0;
 667         time_t now = cfs_time_current_sec();
 668         __u64 total_bavail, total_weight = 0;
 669         __u32 ost_count;
 670         int nfound, good_osts, i, warn = 0, rc = 0;
 671         int stripe_cnt_min = min_stripe_count(*stripe_cnt, flags);
 672         ENTRY;
 673
 674         if (stripe_cnt_min < 1)
 675                 GOTO(out_nolock, rc = -EINVAL);
 676
 677         lov_getref(exp->exp_obd);
 678
 679         /* Detect -EAGAIN early, before expensive lock is taken. */
 680         if (!lov->lov_qos.lq_dirty && lov->lov_qos.lq_same_space)
 681                 GOTO(out_nolock, rc = -EAGAIN);
 682
 683         /* Do actual allocation, use write lock here. */
 684         down_write(&lov->lov_qos.lq_rw_sem);
 685
 686         /*
 687          * Check again, while we were sleeping on @lq_rw_sem things could
 688          * change.
 689          */
 690         if (!lov->lov_qos.lq_dirty && lov->lov_qos.lq_same_space)
 691                 GOTO(out, rc = -EAGAIN);
 692
 693         ost_count = lov->desc.ld_tgt_count;
 694
 695         if (lov->desc.ld_active_tgt_count < 2)
 696                 GOTO(out, rc = -EAGAIN);
 697
 698         rc = qos_calc_ppo(exp->exp_obd);
 699         if (rc)
 700                 GOTO(out, rc);
 701
 702         total_bavail = 0;
 703         good_osts = 0;
 704         /* Warn users about zero available space/inode every 30 min */
 705         if (cfs_time_sub(now, last_warn) > 60 * 30)
 706                 warn = 1;
 707         /* Find all the OSTs that are valid stripe candidates */
 708         for (i = 0; i < ost_count; i++) {
 709                 __u64 bavail;
 710
 711                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active)
 712                         continue;
 713                 bavail = TGT_BAVAIL(i);
 714                 if (!bavail) {
 715                         if (warn) {
 716                                 CDEBUG(D_QOS, "no free space on %s\n",
 717                                      obd_uuid2str(&lov->lov_tgts[i]->ltd_uuid));
 718                                 last_warn = now;
 719                         }
 720                         continue;
 721                 }
 722                 if (!TGT_FFREE(i)) {
 723                         if (warn) {
 724                                 CDEBUG(D_QOS, "no free inodes on %s\n",
 725                                      obd_uuid2str(&lov->lov_tgts[i]->ltd_uuid));
 726                                 last_warn = now;
 727                         }
 728                         continue;
 729                 }
 730
 731                 /* Fail Check before osc_precreate() is called
 732                    so we can only 'fail' single OSC. */
 733                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_PRECREATE) && i == 0)
 734                         continue;
 735
 736                 if (obd_precreate(lov->lov_tgts[i]->ltd_exp) > 2)
 737                         continue;
 738
 739                 lov->lov_tgts[i]->ltd_qos.ltq_usable = 1;
 740                 qos_calc_weight(lov, i);
 741                 total_bavail += bavail;
 742                 total_weight += lov->lov_tgts[i]->ltd_qos.ltq_weight;
 743
 744                 good_osts++;
 745         }
 746
 747         if (good_osts < stripe_cnt_min)
 748                 GOTO(out, rc = -EAGAIN);
 749
 750         if (!total_bavail)
 751                 GOTO(out, rc = -ENOSPC);
 752
 753         /* We have enough osts */
 754         if (good_osts < *stripe_cnt)
 755                 *stripe_cnt = good_osts;
 756
 757         if (!*stripe_cnt)
 758                 GOTO(out, rc = -EAGAIN);
 759
 760         /* Find enough OSTs with weighted random allocation. */
 761         nfound = 0;
 762         while (nfound < *stripe_cnt) {
 763                 __u64 rand, cur_weight;
 764
 765                 cur_weight = 0;
 766                 rc = -ENODEV;
 767
 768                 if (total_weight) {
 769 #if BITS_PER_LONG == 32
 770                         rand = ll_rand() % (unsigned)total_weight;
 771                         /* If total_weight > 32-bit, first generate the high
 772                          * 32 bits of the random number, then add in the low
 773                          * 32 bits (truncated to the upper limit, if needed) */
 774                         if (total_weight > 0xffffffffULL)
 775                                 rand = (__u64)(ll_rand() %
 776                                           (unsigned)(total_weight >> 32)) << 32;
 777                         else
 778                                 rand = 0;
 779
 780                         if (rand == (total_weight & 0xffffffff00000000ULL))
 781                                 rand |= ll_rand() % (unsigned)total_weight;
 782                         else
 783                                 rand |= ll_rand();
 784
 785 #else
 786                         rand = ((__u64)ll_rand() << 32 | ll_rand()) %
 787                                 total_weight;
 788 #endif
 789                 } else {
 790                         rand = 0;
 791                 }
 792
 793                 /* On average, this will hit larger-weighted osts more often.
 794                    0-weight osts will always get used last (only when rand=0).*/
 795                 for (i = 0; i < ost_count; i++) {
 796                         if (!lov->lov_tgts[i] ||
 797                             !lov->lov_tgts[i]->ltd_qos.ltq_usable)
 798                                 continue;
 799
 800                         cur_weight += lov->lov_tgts[i]->ltd_qos.ltq_weight;
 801                         if (cur_weight >= rand) {
 802 #ifdef QOS_DEBUG
 803                                 CDEBUG(D_QOS, "assigned stripe=%d to idx=%d\n",
 804                                        nfound, i);
 805 #endif
 806                                 idx_arr[nfound++] = i;
 807                                 qos_used(lov, i, &total_weight);
 808                                 rc = 0;
 809                                 break;
 810                         }
 811                 }
 812                 /* should never satisfy below condition */
 813                 if (rc) {
 814                         CERROR("Didn't find any OSTs?\n");
 815                         break;
 816                 }
 817         }
 818         LASSERT(nfound == *stripe_cnt);
 819
 820 out:
 821         up_write(&lov->lov_qos.lq_rw_sem);
 822
 823 out_nolock:
 824         if (rc == -EAGAIN)
 825                 rc = alloc_rr(lov, idx_arr, stripe_cnt, flags);
 826
 827         lov_putref(exp->exp_obd);
 828         RETURN(rc);
 829 }
 830
 831 /* return new alloced stripe count on success */
 832 static int alloc_idx_array(struct obd_export *exp, struct lov_stripe_md *lsm,
 833                            int newea, int **idx_arr, int *arr_cnt, int flags)
 834 {
 835         struct lov_obd *lov = &exp->exp_obd->u.lov;
 836         int stripe_cnt = lsm->lsm_stripe_count;
 837         int i, rc = 0;
 838         int *tmp_arr = NULL;
 839         ENTRY;
 840
 841         *arr_cnt = stripe_cnt;
 842         OBD_ALLOC(tmp_arr, *arr_cnt * sizeof(int));
 843         if (tmp_arr == NULL)
 844                 RETURN(-ENOMEM);
 845         for (i = 0; i < *arr_cnt; i++)
 846                 tmp_arr[i] = -1;
 847
 848         if (newea ||
 849             lsm->lsm_oinfo[0]->loi_ost_idx >= lov->desc.ld_tgt_count)
 850                 rc = alloc_qos(exp, tmp_arr, &stripe_cnt, flags);
 851         else
 852                 rc = alloc_specific(lov, lsm, tmp_arr);
 853
 854         if (rc)
 855                 GOTO(out_arr, rc);
 856
 857         *idx_arr = tmp_arr;
 858         RETURN(stripe_cnt);
 859 out_arr:
 860         OBD_FREE(tmp_arr, *arr_cnt * sizeof(int));
 861         *arr_cnt = 0;
 862         RETURN(rc);
 863 }
 864
 865 static void free_idx_array(int *idx_arr, int arr_cnt)
 866 {
 867         if (arr_cnt)
 868                 OBD_FREE(idx_arr, arr_cnt * sizeof(int));
 869 }
 870
 871 int qos_prep_create(struct obd_export *exp, struct lov_request_set *set)
 872 {
 873         struct lov_obd *lov = &exp->exp_obd->u.lov;
 874         struct lov_stripe_md *lsm;
 875         struct obdo *src_oa = set->set_oi->oi_oa;
 876         struct obd_trans_info *oti = set->set_oti;
 877         int i, stripes, rc = 0, newea = 0;
 878         int flag = LOV_USES_ASSIGNED_STRIPE;
 879         int *idx_arr = NULL, idx_cnt = 0;
 880         ENTRY;
 881
 882         LASSERT(src_oa->o_valid & OBD_MD_FLID);
 883         LASSERT(src_oa->o_valid & OBD_MD_FLGROUP);
 884
 885         if (set->set_oi->oi_md == NULL) {
 886                 int stripes_def = lov_get_stripecnt(lov, 0);
 887
 888                 /* If the MDS file was truncated up to some size, stripe over
 889                  * enough OSTs to allow the file to be created at that size.
 890                  * This may mean we use more than the default # of stripes. */
 891                 if (src_oa->o_valid & OBD_MD_FLSIZE) {
 892                         obd_size min_bavail = LUSTRE_STRIPE_MAXBYTES;
 893
 894                         /* Find a small number of stripes we can use
 895                            (up to # of active osts). */
 896                         stripes = 1;
 897                         lov_getref(exp->exp_obd);
 898                         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
 899                                 if (!lov->lov_tgts[i] ||
 900                                     !lov->lov_tgts[i]->ltd_active)
 901                                         continue;
 902                                 min_bavail = min(min_bavail, TGT_BAVAIL(i));
 903                                 if (min_bavail * stripes > src_oa->o_size)
 904                                         break;
 905                                 stripes++;
 906                         }
 907                         lov_putref(exp->exp_obd);
 908
 909                         if (stripes < stripes_def)
 910                                 stripes = stripes_def;
 911                 } else {
 912                          flag = LOV_USES_DEFAULT_STRIPE;
 913                          stripes = stripes_def;
 914                 }
 915
 916                 rc = lov_alloc_memmd(&set->set_oi->oi_md, stripes,
 917                                      lov->desc.ld_pattern ?
 918                                      lov->desc.ld_pattern : LOV_PATTERN_RAID0,
 919                                      LOV_MAGIC);
 920                 if (rc < 0)
 921                         GOTO(out_err, rc);
 922                 newea = 1;
 923                 rc = 0;
 924         }
 925
 926         lsm = set->set_oi->oi_md;
 927         lsm->lsm_object_id = src_oa->o_id;
 928         lsm->lsm_object_gr = src_oa->o_gr;
 929
 930         if (!lsm->lsm_stripe_size)
 931                 lsm->lsm_stripe_size = lov->desc.ld_default_stripe_size;
 932         if (!lsm->lsm_pattern) {
 933                 LASSERT(lov->desc.ld_pattern);
 934                 lsm->lsm_pattern = lov->desc.ld_pattern;
 935         }
 936
 937         stripes = alloc_idx_array(exp, lsm, newea, &idx_arr, &idx_cnt, flag);
 938         if (stripes <= 0)
 939                 GOTO(out_err, rc = stripes ? stripes : -EIO);
 940         LASSERTF(stripes <= lsm->lsm_stripe_count,"requested %d allocated %d\n",
 941                  lsm->lsm_stripe_count, stripes);
 942
 943         for (i = 0; i < stripes; i++) {
 944                 struct lov_request *req;
 945                 int ost_idx = idx_arr[i];
 946                 LASSERT(ost_idx >= 0);
 947
 948                 OBD_ALLOC(req, sizeof(*req));
 949                 if (req == NULL)
 950                         GOTO(out_err, rc = -ENOMEM);
 951                 lov_set_add_req(req, set);
 952
 953                 req->rq_buflen = sizeof(*req->rq_oi.oi_md);
 954                 OBD_ALLOC(req->rq_oi.oi_md, req->rq_buflen);
 955                 if (req->rq_oi.oi_md == NULL)
 956                         GOTO(out_err, rc = -ENOMEM);
 957
 958                 OBDO_ALLOC(req->rq_oi.oi_oa);
 959                 if (req->rq_oi.oi_oa == NULL)
 960                         GOTO(out_err, rc = -ENOMEM);
 961
 962                 req->rq_idx = ost_idx;
 963                 req->rq_stripe = i;
 964                 /* create data objects with "parent" OA */
 965                 memcpy(req->rq_oi.oi_oa, src_oa, sizeof(*req->rq_oi.oi_oa));
 966
 967                 /* XXX When we start creating objects on demand, we need to
 968                  *     make sure that we always create the object on the
 969                  *     stripe which holds the existing file size.
 970                  */
 971                 if (src_oa->o_valid & OBD_MD_FLSIZE) {
 972                         req->rq_oi.oi_oa->o_size =
 973                                 lov_size_to_stripe(lsm, src_oa->o_size, i);
 974
 975                         CDEBUG(D_INODE, "stripe %d has size "LPU64"/"LPU64"\n",
 976                                i, req->rq_oi.oi_oa->o_size, src_oa->o_size);
 977                 }
 978         }
 979         LASSERT(set->set_count == stripes);
 980
 981         if (stripes < lsm->lsm_stripe_count)
 982                 qos_shrink_lsm(set);
 983
 984         if (oti && (src_oa->o_valid & OBD_MD_FLCOOKIE)) {
 985                 oti_alloc_cookies(oti, set->set_count);
 986                 if (!oti->oti_logcookies)
 987                         GOTO(out_err, rc = -ENOMEM);
 988                 set->set_cookies = oti->oti_logcookies;
 989         }
 990 out_err:
 991         if (newea && rc)
 992                 obd_free_memmd(exp, &set->set_oi->oi_md);
 993         if (idx_arr)
 994                 free_idx_array(idx_arr, idx_cnt);
 995         EXIT;
 996         return rc;
 997 }
 998
 999 void qos_update(struct lov_obd *lov)
1000 {
1001         ENTRY;
1002         lov->lov_qos.lq_dirty = 1;
1003 }