lustre/ofd/ofd_grant.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, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/ofd/ofd_grant.c
  37  *
  38  * Author: Johann Lombardi <johann@whamcloud..com>
  39  */
  40
  41 #define DEBUG_SUBSYSTEM S_FILTER
  42
  43 #include "ofd_internal.h"
  44
  45 /* At least enough to send a couple of 1MB RPCs, even if not max sized */
  46 #define OFD_GRANT_CHUNK                 (2ULL * DT_MAX_BRW_SIZE)
  47
  48 /* Clients typically hold 2x their max_rpcs_in_flight of grant space */
  49 #define OFD_GRANT_SHRINK_LIMIT(exp)     (2ULL * 8 * exp_max_brw_size(exp))
  50
  51 static inline obd_size ofd_grant_from_cli(struct obd_export *exp,
  52                                           struct ofd_device *ofd, obd_size val)
  53 {
  54         if (ofd_grant_compat(exp, ofd))
  55                 /* clients not supporting OBD_CONNECT_GRANT_PARAM actually
  56                  * consume 4KB of grant per block, we should thus inflate
  57                  * the grant counters to reflect what was actually consumed */
  58                 return val << (ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT);
  59         return val;
  60 }
  61
  62 static inline obd_size ofd_grant_to_cli(struct obd_export *exp,
  63                                         struct ofd_device *ofd, obd_size val)
  64 {
  65         if (ofd_grant_compat(exp, ofd))
  66                 return val >> (ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT);
  67         return val;
  68 }
  69
  70 static inline obd_size ofd_grant_chunk(struct obd_export *exp,
  71                                        struct ofd_device *ofd)
  72 {
  73         if (ofd_obd(ofd)->obd_self_export == exp)
  74                 /* Grant enough space to handle a big precreate request */
  75                 return OST_MAX_PRECREATE * ofd->ofd_dt_conf.ddp_inodespace / 2;
  76
  77         if (ofd_grant_compat(exp, ofd))
  78                 /* Try to grant enough space to send a full-size RPC */
  79                 return exp_max_brw_size(exp) <<
  80                        (ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT);
  81
  82         /* Try to return enough to send two full RPCs, if needed */
  83         return exp_max_brw_size(exp) * 2;
  84 }
  85
  86 /**
  87  * Perform extra sanity checks for grant accounting. This is done at connect,
  88  * disconnect, and statfs RPC time, so it shouldn't be too bad. We can
  89  * always get rid of it or turn it off when we know accounting is good.
  90  *
  91  * \param obd - is the device to check
  92  * \param func - is the function to call if an inconsistency is found
  93  */
  94 void ofd_grant_sanity_check(struct obd_device *obd, const char *func)
  95 {
  96         struct filter_export_data       *fed;
  97         struct ofd_device               *ofd = ofd_dev(obd->obd_lu_dev);
  98         struct obd_export               *exp;
  99         obd_size                         maxsize;
 100         obd_size                         tot_dirty = 0;
 101         obd_size                         tot_pending = 0;
 102         obd_size                         tot_granted = 0;
 103         obd_size                         fo_tot_dirty, fo_tot_pending;
 104         obd_size                         fo_tot_granted;
 105
 106         if (cfs_list_empty(&obd->obd_exports))
 107                 return;
 108
 109         /* We don't want to do this for large machines that do lots of
 110          * mounts or unmounts.  It burns... */
 111         if (obd->obd_num_exports > 100)
 112                 return;
 113
 114         maxsize = ofd->ofd_osfs.os_blocks << ofd->ofd_blockbits;
 115
 116         spin_lock(&obd->obd_dev_lock);
 117         spin_lock(&ofd->ofd_grant_lock);
 118         cfs_list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain) {
 119                 int error = 0;
 120
 121                 fed = &exp->exp_filter_data;
 122
 123                 if (obd->obd_self_export == exp)
 124                         CDEBUG(D_CACHE, "%s: processing self export: %ld %ld "
 125                                "%ld\n", obd->obd_name, fed->fed_grant,
 126                                fed->fed_pending, fed->fed_dirty);
 127
 128                 if (fed->fed_grant < 0 || fed->fed_pending < 0 ||
 129                     fed->fed_dirty < 0)
 130                         error = 1;
 131                 if (fed->fed_grant + fed->fed_pending > maxsize) {
 132                         CERROR("%s: cli %s/%p fed_grant(%ld) + fed_pending(%ld)"
 133                                " > maxsize("LPU64")\n", obd->obd_name,
 134                                exp->exp_client_uuid.uuid, exp, fed->fed_grant,
 135                                fed->fed_pending, maxsize);
 136                         spin_unlock(&obd->obd_dev_lock);
 137                         spin_unlock(&ofd->ofd_grant_lock);
 138                         LBUG();
 139                 }
 140                 if (fed->fed_dirty > maxsize) {
 141                         CERROR("%s: cli %s/%p fed_dirty(%ld) > maxsize("LPU64
 142                                ")\n", obd->obd_name, exp->exp_client_uuid.uuid,
 143                                exp, fed->fed_dirty, maxsize);
 144                         spin_unlock(&obd->obd_dev_lock);
 145                         spin_unlock(&ofd->ofd_grant_lock);
 146                         LBUG();
 147                 }
 148                 CDEBUG_LIMIT(error ? D_ERROR : D_CACHE, "%s: cli %s/%p dirty "
 149                              "%ld pend %ld grant %ld\n", obd->obd_name,
 150                              exp->exp_client_uuid.uuid, exp, fed->fed_dirty,
 151                              fed->fed_pending, fed->fed_grant);
 152                 tot_granted += fed->fed_grant + fed->fed_pending;
 153                 tot_pending += fed->fed_pending;
 154                 tot_dirty += fed->fed_dirty;
 155         }
 156         spin_unlock(&obd->obd_dev_lock);
 157         fo_tot_granted = ofd->ofd_tot_granted;
 158         fo_tot_pending = ofd->ofd_tot_pending;
 159         fo_tot_dirty = ofd->ofd_tot_dirty;
 160
 161         if (tot_granted != fo_tot_granted)
 162                 CERROR("%s: tot_granted "LPU64" != fo_tot_granted "LPU64"\n",
 163                        func, tot_granted, fo_tot_granted);
 164         if (tot_pending != fo_tot_pending)
 165                 CERROR("%s: tot_pending "LPU64" != fo_tot_pending "LPU64"\n",
 166                        func, tot_pending, fo_tot_pending);
 167         if (tot_dirty != fo_tot_dirty)
 168                 CERROR("%s: tot_dirty "LPU64" != fo_tot_dirty "LPU64"\n",
 169                        func, tot_dirty, fo_tot_dirty);
 170         if (tot_pending > tot_granted)
 171                 CERROR("%s: tot_pending "LPU64" > tot_granted "LPU64"\n",
 172                        func, tot_pending, tot_granted);
 173         if (tot_granted > maxsize)
 174                 CERROR("%s: tot_granted "LPU64" > maxsize "LPU64"\n",
 175                        func, tot_granted, maxsize);
 176         if (tot_dirty > maxsize)
 177                 CERROR("%s: tot_dirty "LPU64" > maxsize "LPU64"\n",
 178                        func, tot_dirty, maxsize);
 179         spin_unlock(&ofd->ofd_grant_lock);
 180 }
 181
 182 /**
 183  * Get fresh statfs information from the OSD layer if the cache is older than 1s
 184  * or if force is set. The OSD layer is in charge of estimating data & metadata
 185  * overhead.
 186  *
 187  * \param env - is the lu environment passed by the caller
 188  * \param exp - export used to print client info in debug messages
 189  * \param force - is used to force a refresh of statfs information
 190  * \param from_cache - returns whether the statfs information are
 191  *                   taken from cache
 192  */
 193 static void ofd_grant_statfs(const struct lu_env *env, struct obd_export *exp,
 194                              int force, int *from_cache)
 195 {
 196         struct obd_device       *obd = exp->exp_obd;
 197         struct ofd_device       *ofd = ofd_exp(exp);
 198         struct obd_statfs       *osfs = &ofd_info(env)->fti_u.osfs;
 199         __u64                    max_age;
 200         int                      rc;
 201
 202         if (force)
 203                 max_age = 0; /* get fresh statfs data */
 204         else
 205                 max_age = cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS);
 206
 207         rc = ofd_statfs_internal(env, ofd, osfs, max_age, from_cache);
 208         if (unlikely(rc)) {
 209                 *from_cache = 0;
 210                 return;
 211         }
 212
 213         CDEBUG(D_CACHE, "%s: cli %s/%p free: "LPU64" avail: "LPU64"\n",
 214                obd->obd_name, exp->exp_client_uuid.uuid, exp,
 215                osfs->os_bfree << ofd->ofd_blockbits,
 216                osfs->os_bavail << ofd->ofd_blockbits);
 217 }
 218
 219 /**
 220  * Figure out how much space is available on the backend filesystem.
 221  * This is done by accessing cached statfs data previously populated by
 222  * ofd_grant_statfs(), from which we withdraw the space already granted to
 223  * clients and the reserved space.
 224  *
 225  * \param exp - export which received the write request
 226  */
 227 static obd_size ofd_grant_space_left(struct obd_export *exp)
 228 {
 229         struct obd_device       *obd = exp->exp_obd;
 230         struct ofd_device       *ofd = ofd_exp(exp);
 231         obd_size                 tot_granted;
 232         obd_size                 left, avail;
 233         obd_size                 unstable;
 234
 235         ENTRY;
 236         LASSERT_SPIN_LOCKED(&ofd->ofd_grant_lock);
 237
 238         spin_lock(&ofd->ofd_osfs_lock);
 239         /* get available space from cached statfs data */
 240         left = ofd->ofd_osfs.os_bavail << ofd->ofd_blockbits;
 241         unstable = ofd->ofd_osfs_unstable; /* those might be accounted twice */
 242         spin_unlock(&ofd->ofd_osfs_lock);
 243
 244         tot_granted = ofd->ofd_tot_granted;
 245
 246         if (left < tot_granted) {
 247                 int mask = (left + unstable <
 248                             tot_granted - ofd->ofd_tot_pending) ?
 249                             D_ERROR : D_CACHE;
 250
 251                 CDEBUG_LIMIT(mask, "%s: cli %s/%p left "LPU64" < tot_grant "
 252                              LPU64" unstable "LPU64" pending "LPU64"\n",
 253                              obd->obd_name, exp->exp_client_uuid.uuid, exp,
 254                              left, tot_granted, unstable,
 255                              ofd->ofd_tot_pending);
 256                 RETURN(0);
 257         }
 258
 259         avail = left;
 260         /* Withdraw space already granted to clients */
 261         left -= tot_granted;
 262
 263         /* If the left space is below the grant threshold x available space,
 264          * stop granting space to clients.
 265          * The purpose of this threshold is to keep some error margin on the
 266          * overhead estimate made by the OSD layer. If we grant all the free
 267          * space, we have no way (grant space cannot be revoked yet) to
 268          * adjust if the write overhead has been underestimated. */
 269         left -= min_t(obd_size, left, ofd_grant_reserved(ofd, avail));
 270
 271         /* Align left on block size */
 272         left &= ~((1ULL << ofd->ofd_blockbits) - 1);
 273
 274         CDEBUG(D_CACHE, "%s: cli %s/%p avail "LPU64" left "LPU64" unstable "
 275                LPU64" tot_grant "LPU64" pending "LPU64"\n", obd->obd_name,
 276                exp->exp_client_uuid.uuid, exp, avail, left, unstable,
 277                tot_granted, ofd->ofd_tot_pending);
 278
 279         RETURN(left);
 280 }
 281
 282 /**
 283  * Grab the dirty and seen grant announcements from the incoming obdo.
 284  * We will later calculate the client's new grant and return it.
 285  * Caller must hold ofd_grant_lock spinlock.
 286  *
 287  * \param env - is the lu environment supplying osfs storage
 288  * \param exp - is the export for which we received the request
 289  * \paral oa - is the incoming obdo sent by the client
 290  */
 291 static void ofd_grant_incoming(const struct lu_env *env, struct obd_export *exp,
 292                                struct obdo *oa)
 293 {
 294         struct filter_export_data       *fed;
 295         struct ofd_device               *ofd = ofd_exp(exp);
 296         struct obd_device               *obd = exp->exp_obd;
 297         long                             dirty, dropped, grant_chunk;
 298         ENTRY;
 299
 300         LASSERT_SPIN_LOCKED(&ofd->ofd_grant_lock);
 301
 302         if ((oa->o_valid & (OBD_MD_FLBLOCKS|OBD_MD_FLGRANT)) !=
 303                                         (OBD_MD_FLBLOCKS|OBD_MD_FLGRANT)) {
 304                 oa->o_valid &= ~OBD_MD_FLGRANT;
 305                 RETURN_EXIT;
 306         }
 307
 308         fed = &exp->exp_filter_data;
 309
 310         /* Add some margin, since there is a small race if other RPCs arrive
 311          * out-or-order and have already consumed some grant.  We want to
 312          * leave this here in case there is a large error in accounting. */
 313         CDEBUG(D_CACHE,
 314                "%s: cli %s/%p reports grant "LPU64" dropped %u, local %lu\n",
 315                obd->obd_name, exp->exp_client_uuid.uuid, exp, oa->o_grant,
 316                oa->o_dropped, fed->fed_grant);
 317
 318         if ((long long)oa->o_dirty < 0)
 319                 oa->o_dirty = 0;
 320
 321         dirty       = ofd_grant_from_cli(exp, ofd, oa->o_dirty);
 322         dropped     = ofd_grant_from_cli(exp, ofd, (obd_size)oa->o_dropped);
 323         grant_chunk = ofd_grant_chunk(exp, ofd);
 324
 325         /* Update our accounting now so that statfs takes it into account.
 326          * Note that fed_dirty is only approximate and can become incorrect
 327          * if RPCs arrive out-of-order.  No important calculations depend
 328          * on fed_dirty however, but we must check sanity to not assert. */
 329         if (dirty > fed->fed_grant + 4 * grant_chunk)
 330                 dirty = fed->fed_grant + 4 * grant_chunk;
 331         ofd->ofd_tot_dirty += dirty - fed->fed_dirty;
 332         if (fed->fed_grant < dropped) {
 333                 CDEBUG(D_CACHE,
 334                        "%s: cli %s/%p reports %lu dropped > grant %lu\n",
 335                        obd->obd_name, exp->exp_client_uuid.uuid, exp, dropped,
 336                        fed->fed_grant);
 337                 dropped = 0;
 338         }
 339         if (ofd->ofd_tot_granted < dropped) {
 340                 CERROR("%s: cli %s/%p reports %lu dropped > tot_grant "LPU64
 341                        "\n", obd->obd_name, exp->exp_client_uuid.uuid, exp,
 342                        dropped, ofd->ofd_tot_granted);
 343                 dropped = 0;
 344         }
 345         ofd->ofd_tot_granted -= dropped;
 346         fed->fed_grant -= dropped;
 347         fed->fed_dirty = dirty;
 348
 349         if (fed->fed_dirty < 0 || fed->fed_grant < 0 || fed->fed_pending < 0) {
 350                 CERROR("%s: cli %s/%p dirty %ld pend %ld grant %ld\n",
 351                        obd->obd_name, exp->exp_client_uuid.uuid, exp,
 352                        fed->fed_dirty, fed->fed_pending, fed->fed_grant);
 353                 spin_unlock(&ofd->ofd_grant_lock);
 354                 LBUG();
 355         }
 356         EXIT;
 357 }
 358
 359 /**
 360  * Called when the client is able to release some grants. Proceed with the
 361  * shrink request when there is less ungranted space remaining
 362  * than the amount all of the connected clients would consume if they
 363  * used their full grant.
 364  *
 365  * \param exp - is the export for which we received the request
 366  * \paral oa - is the incoming obdo sent by the client
 367  * \param left_space - is the remaining free space with space already granted
 368  *                   taken out
 369  */
 370 static void ofd_grant_shrink(struct obd_export *exp,
 371                              struct obdo *oa, obd_size left_space)
 372 {
 373         struct filter_export_data       *fed;
 374         struct ofd_device               *ofd = ofd_exp(exp);
 375         struct obd_device               *obd = exp->exp_obd;
 376         long                             grant_shrink;
 377
 378         LASSERT_SPIN_LOCKED(&ofd->ofd_grant_lock);
 379         LASSERT(exp);
 380         if (left_space >= ofd->ofd_tot_granted_clients *
 381                           OFD_GRANT_SHRINK_LIMIT(exp))
 382                 return;
 383
 384         grant_shrink = ofd_grant_from_cli(exp, ofd, oa->o_grant);
 385
 386         fed = &exp->exp_filter_data;
 387         fed->fed_grant       -= grant_shrink;
 388         ofd->ofd_tot_granted -= grant_shrink;
 389
 390         CDEBUG(D_CACHE, "%s: cli %s/%p shrink %ld fed_grant %ld total "
 391                LPU64"\n", obd->obd_name, exp->exp_client_uuid.uuid,
 392                exp, grant_shrink, fed->fed_grant, ofd->ofd_tot_granted);
 393
 394         /* client has just released some grant, don't grant any space back */
 395         oa->o_grant = 0;
 396 }
 397
 398 /**
 399  * Calculate how much space is required to write a given network buffer
 400  */
 401 static inline int ofd_grant_rnb_size(struct obd_export *exp,
 402                                      struct ofd_device *ofd,
 403                                      struct niobuf_remote *rnb)
 404 {
 405         obd_size blocksize, bytes, end;
 406
 407         if (exp && ofd_grant_compat(exp, ofd))
 408                 blocksize = 1ULL << COMPAT_BSIZE_SHIFT;
 409         else
 410                 blocksize = 1ULL << ofd->ofd_blockbits;
 411
 412         /* The network buffer might span several blocks, align it on block
 413          * boundaries */
 414         bytes  = rnb->rnb_offset & (blocksize - 1);
 415         bytes += rnb->rnb_len;
 416         end    = bytes & (blocksize - 1);
 417         if (end)
 418                 bytes += blocksize - end;
 419         if (exp)
 420                 /* Apply per-export pecularities if one is given */
 421                 bytes = ofd_grant_from_cli(exp, ofd, (obd_size)bytes);
 422         return bytes;
 423 }
 424
 425
 426 /**
 427  * When clients have dirtied as much space as they've been granted they
 428  * fall through to sync writes.  These sync writes haven't been expressed
 429  * in grants and need to error with ENOSPC when there isn't room in the
 430  * filesystem for them after grants are taken into account.  However,
 431  * writeback of the dirty data that was already granted space can write
 432  * right on through.
 433  * Caller must hold ofd_grant_lock spinlock.
 434  *
 435  * \param env - is the lu environment passed by the caller
 436  * \param exp - is the export identifying the client which sent the RPC
 437  * \param oa  - is the incoming obdo in which we should return the pack the
 438  *            additional grant
 439  * \param rnb - is the list of network buffers
 440  * \param niocont - is the number of network buffers in the list
 441  * \param left - is the remaining free space with space already granted
 442  *             taken out
 443  */
 444 static void ofd_grant_check(const struct lu_env *env, struct obd_export *exp,
 445                             struct obdo *oa, struct niobuf_remote *rnb,
 446                             int niocount, obd_size *left)
 447 {
 448         struct filter_export_data       *fed = &exp->exp_filter_data;
 449         struct obd_device               *obd = exp->exp_obd;
 450         struct ofd_device               *ofd = ofd_exp(exp);
 451         unsigned long                    ungranted = 0;
 452         unsigned long                    granted = 0;
 453         int                              i;
 454         int                              resend = 0;
 455         struct ofd_thread_info          *info = ofd_info(env);
 456
 457         ENTRY;
 458
 459         LASSERT_SPIN_LOCKED(&ofd->ofd_grant_lock);
 460
 461         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
 462             (oa->o_flags & OBD_FL_RECOV_RESEND)) {
 463                 resend = 1;
 464                 CDEBUG(D_CACHE, "Recoverable resend arrived, skipping "
 465                                 "accounting\n");
 466         }
 467
 468         for (i = 0; i < niocount; i++) {
 469                 int bytes;
 470
 471                 if (obd->obd_recovering) {
 472                         /* Replaying write. Grant info have been processed
 473                          * already so no need to do any enforcement here.
 474                          * It is worth noting that only bulk writes with all
 475                          * rnbs having OBD_BRW_FROM_GRANT can be replayed.
 476                          * If one page hasn't OBD_BRW_FROM_GRANT set, then
 477                          * the whole bulk is written synchronously */
 478                         if (rnb[i].rnb_flags & OBD_BRW_FROM_GRANT) {
 479                                  rnb[i].rnb_flags |= OBD_BRW_GRANTED;
 480                                  continue;
 481                         } else {
 482                                 CERROR("%s: cli %s is replaying OST_WRITE "
 483                                        "while one rnb hasn't OBD_BRW_FROM_GRANT"
 484                                        " set (0x%x)\n", exp->exp_obd->obd_name,
 485                                         exp->exp_client_uuid.uuid,
 486                                         rnb[i].rnb_flags);
 487
 488                         }
 489                 } else if ((oa->o_valid & OBD_MD_FLGRANT) &&
 490                            (rnb[i].rnb_flags & OBD_BRW_FROM_GRANT)) {
 491                         if (resend) {
 492                                 /* This is a recoverable resend so grant
 493                                  * information have already been processed */
 494                                 rnb[i].rnb_flags |= OBD_BRW_GRANTED;
 495                                 continue;
 496                         }
 497
 498                         /* inflate consumed space if needed */
 499                         bytes = ofd_grant_rnb_size(exp, ofd, &rnb[i]);
 500                         if (fed->fed_grant < granted + bytes) {
 501                                 CDEBUG(D_CACHE, "%s: cli %s/%p claims %ld+%d "
 502                                        "GRANT, real grant %lu idx %d\n",
 503                                        exp->exp_obd->obd_name,
 504                                        exp->exp_client_uuid.uuid, exp,
 505                                        granted, bytes, fed->fed_grant, i);
 506                         } else {
 507                                 granted += bytes;
 508                                 rnb[i].rnb_flags |= OBD_BRW_GRANTED;
 509                                 continue;
 510                         }
 511                 }
 512
 513                 /* Consume grant space on the server.
 514                  * Unlike above, ofd_grant_rnb_size() is called with exp = NULL
 515                  * so that the required grant space isn't inflated. This is
 516                  * done on purpose since the server can deal with large block
 517                  * size, unlike some clients */
 518                 bytes = ofd_grant_rnb_size(NULL, ofd, &rnb[i]);
 519                 if (*left > ungranted + bytes) {
 520                         /* if enough space, pretend it was granted */
 521                         ungranted += bytes;
 522                         rnb[i].rnb_flags |= OBD_BRW_GRANTED;
 523                         continue;
 524                 }
 525
 526                 /* We can't check for already-mapped blocks here (make sense
 527                  * when backend filesystem does not use COW) as it requires
 528                  * dropping the grant lock.
 529                  * Instead, we clear ~OBD_BRW_GRANTED and in that case we need
 530                  * to go through and verify if all of the blocks not marked
 531                  *  BRW_GRANTED are already mapped and we can ignore this error.
 532                  */
 533                 rnb[i].rnb_flags &= ~OBD_BRW_GRANTED;
 534                 CDEBUG(D_CACHE,"%s: cli %s/%p idx %d no space for %d\n",
 535                                 exp->exp_obd->obd_name,
 536                                 exp->exp_client_uuid.uuid, exp, i, bytes);
 537         }
 538
 539         /* record space used for the I/O, will be used in ofd_grant_commmit() */
 540         /* Now substract what the clients has used already.  We don't subtract
 541          * this from the tot_granted yet, so that other client's can't grab
 542          * that space before we have actually allocated our blocks. That
 543          * happens in ofd_grant_commit() after the writes are done. */
 544         info->fti_used = granted + ungranted;
 545         *left -= ungranted;
 546         fed->fed_grant -= granted;
 547         fed->fed_pending += info->fti_used;
 548         ofd->ofd_tot_granted += ungranted;
 549         ofd->ofd_tot_pending += info->fti_used;
 550
 551         CDEBUG(D_CACHE,
 552                "%s: cli %s/%p granted: %lu ungranted: %lu grant: %lu dirty: %lu"
 553                "\n", obd->obd_name, exp->exp_client_uuid.uuid, exp,
 554                granted, ungranted, fed->fed_grant, fed->fed_dirty);
 555
 556         if (obd->obd_recovering)
 557                 /* don't update dirty accounting during recovery */
 558                 RETURN_EXIT;
 559
 560         if (fed->fed_dirty < granted) {
 561                 CWARN("%s: cli %s/%p claims granted %lu > fed_dirty %lu\n",
 562                        obd->obd_name, exp->exp_client_uuid.uuid, exp,
 563                        granted, fed->fed_dirty);
 564                 granted = fed->fed_dirty;
 565         }
 566         ofd->ofd_tot_dirty -= granted;
 567         fed->fed_dirty -= granted;
 568
 569         if (fed->fed_dirty < 0 || fed->fed_grant < 0 || fed->fed_pending < 0) {
 570                 CERROR("%s: cli %s/%p dirty %ld pend %ld grant %ld\n",
 571                        obd->obd_name, exp->exp_client_uuid.uuid, exp,
 572                        fed->fed_dirty, fed->fed_pending, fed->fed_grant);
 573                 spin_unlock(&ofd->ofd_grant_lock);
 574                 LBUG();
 575         }
 576         EXIT;
 577 }
 578
 579 /**
 580  * Calculate how much grant space to return to client, based on how much space
 581  * is currently free and how much of that is already granted.
 582  * Caller must hold ofd_grant_lock spinlock.
 583  *
 584  * \param exp - is the export of the client which sent the request
 585  * \param curgrant - is the current grant claimed by the client
 586  * \param want - is how much grant space the client would like to have
 587  * \param left - is the remaining free space with granted space taken out
 588  * \param conservative - is how server grants, if true, a certain amount, else
 589  *        server will grant as client requested.
 590  */
 591 static long ofd_grant(struct obd_export *exp, obd_size curgrant,
 592                       obd_size want, obd_size left, bool conservative)
 593 {
 594         struct obd_device               *obd = exp->exp_obd;
 595         struct ofd_device               *ofd = ofd_exp(exp);
 596         struct filter_export_data       *fed = &exp->exp_filter_data;
 597         long                             grant_chunk;
 598         obd_size                         grant;
 599
 600         ENTRY;
 601
 602         if (ofd_grant_prohibit(exp, ofd) || left == 0 || exp->exp_failed)
 603                 RETURN(0);
 604
 605         if (want > 0x7fffffff) {
 606                 CERROR("%s: client %s/%p requesting > 2GB grant "LPU64"\n",
 607                        obd->obd_name, exp->exp_client_uuid.uuid, exp, want);
 608                 RETURN(0);
 609         }
 610
 611         /* client not supporting OBD_CONNECT_GRANT_PARAM works with a 4KB block
 612          * size while the reality is different */
 613         curgrant = ofd_grant_from_cli(exp, ofd, curgrant);
 614         want = ofd_grant_from_cli(exp, ofd, want);
 615         grant_chunk = ofd_grant_chunk(exp, ofd);
 616
 617         /* Grant some fraction of the client's requested grant space so that
 618          * they are not always waiting for write credits (not all of it to
 619          * avoid overgranting in face of multiple RPCs in flight).  This
 620          * essentially will be able to control the OSC_MAX_RIF for a client.
 621          *
 622          * If we do have a large disparity between what the client thinks it
 623          * has and what we think it has, don't grant very much and let the
 624          * client consume its grant first.  Either it just has lots of RPCs
 625          * in flight, or it was evicted and its grants will soon be used up. */
 626         if (curgrant >= want || curgrant >= fed->fed_grant + grant_chunk)
 627                    RETURN(0);
 628
 629         if (obd->obd_recovering)
 630                 conservative = false;
 631
 632         if (conservative)
 633                 /* don't grant more than 1/8th of the remaining free space in
 634                  * one chunk */
 635                 left >>= 3;
 636         grant = min(want, left);
 637         /* round grant upt to the next block size */
 638         grant = (grant + (1 << ofd->ofd_blockbits) - 1) &
 639                 ~((1ULL << ofd->ofd_blockbits) - 1);
 640
 641         if (!grant)
 642                 RETURN(0);
 643
 644         /* Limit to ofd_grant_chunk() if not reconnect/recovery */
 645         if ((grant > grant_chunk) && conservative)
 646                 grant = grant_chunk;
 647
 648         ofd->ofd_tot_granted += grant;
 649         fed->fed_grant += grant;
 650
 651         if (fed->fed_grant < 0) {
 652                 CERROR("%s: cli %s/%p grant %ld want "LPU64" current "LPU64"\n",
 653                        obd->obd_name, exp->exp_client_uuid.uuid, exp,
 654                        fed->fed_grant, want, curgrant);
 655                 spin_unlock(&ofd->ofd_grant_lock);
 656                 LBUG();
 657         }
 658
 659         CDEBUG(D_CACHE,
 660                "%s: cli %s/%p wants: "LPU64" current grant "LPU64
 661                " granting: "LPU64"\n", obd->obd_name, exp->exp_client_uuid.uuid,
 662                exp, want, curgrant, grant);
 663         CDEBUG(D_CACHE,
 664                "%s: cli %s/%p tot cached:"LPU64" granted:"LPU64
 665                " num_exports: %d\n", obd->obd_name, exp->exp_client_uuid.uuid,
 666                exp, ofd->ofd_tot_dirty, ofd->ofd_tot_granted,
 667                obd->obd_num_exports);
 668
 669         RETURN(ofd_grant_to_cli(exp, ofd, grant));
 670 }
 671
 672 /**
 673  * Client connection or reconnection.
 674  *
 675  * \param env - is the lu environment provided by the caller
 676  * \param exp - is the client's export which is reconnecting
 677  * \param want - is how much the client would like to get
 678  * \param conservative - is how server grants to client, if true server will
 679  *        only grant certain amount, else server will grant client requested
 680  *        amount.
 681  */
 682 long ofd_grant_connect(const struct lu_env *env, struct obd_export *exp,
 683                        obd_size want, bool conservative)
 684 {
 685         struct ofd_device               *ofd = ofd_exp(exp);
 686         struct filter_export_data       *fed = &exp->exp_filter_data;
 687         obd_size                         left = 0;
 688         long                             grant;
 689         int                              from_cache;
 690         int                              force = 0; /* can use cached data */
 691
 692         /* don't grant space to client with read-only access */
 693         if ((exp_connect_flags(exp) & OBD_CONNECT_RDONLY) ||
 694             ofd_grant_prohibit(exp, ofd))
 695                 return 0;
 696
 697 refresh:
 698         ofd_grant_statfs(env, exp, force, &from_cache);
 699
 700         spin_lock(&ofd->ofd_grant_lock);
 701
 702         /* Grab free space from cached info and take out space already granted
 703          * to clients as well as reserved space */
 704         left = ofd_grant_space_left(exp);
 705
 706         /* get fresh statfs data if we are short in ungranted space */
 707         if (from_cache && left < 32 * ofd_grant_chunk(exp, ofd)) {
 708                 spin_unlock(&ofd->ofd_grant_lock);
 709                 CDEBUG(D_CACHE, "fs has no space left and statfs too old\n");
 710                 force = 1;
 711                 goto refresh;
 712         }
 713
 714         ofd_grant(exp, ofd_grant_to_cli(exp, ofd, (obd_size)fed->fed_grant),
 715                   want, left, conservative);
 716
 717         /* return to client its current grant */
 718         grant = ofd_grant_to_cli(exp, ofd, (obd_size)fed->fed_grant);
 719         ofd->ofd_tot_granted_clients++;
 720
 721         spin_unlock(&ofd->ofd_grant_lock);
 722
 723         CDEBUG(D_CACHE, "%s: cli %s/%p ocd_grant: %ld want: "LPU64" left: "
 724                LPU64"\n", exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
 725                exp, grant, want, left);
 726
 727         return grant;
 728 }
 729
 730 /**
 731  * Remove a client from the grant accounting totals.  We also remove
 732  * the export from the obd device under the osfs and dev locks to ensure
 733  * that the ofd_grant_sanity_check() calculations are always valid.
 734  * The client should do something similar when it invalidates its import.
 735  *
 736  * \param exp - is the client's export to remove from grant accounting
 737  */
 738 void ofd_grant_discard(struct obd_export *exp)
 739 {
 740         struct obd_device               *obd = exp->exp_obd;
 741         struct ofd_device               *ofd = ofd_exp(exp);
 742         struct filter_export_data       *fed = &exp->exp_filter_data;
 743
 744         spin_lock(&ofd->ofd_grant_lock);
 745         LASSERTF(ofd->ofd_tot_granted >= fed->fed_grant,
 746                  "%s: tot_granted "LPU64" cli %s/%p fed_grant %ld\n",
 747                  obd->obd_name, ofd->ofd_tot_granted,
 748                  exp->exp_client_uuid.uuid, exp, fed->fed_grant);
 749         ofd->ofd_tot_granted -= fed->fed_grant;
 750         fed->fed_grant = 0;
 751         LASSERTF(ofd->ofd_tot_pending >= fed->fed_pending,
 752                  "%s: tot_pending "LPU64" cli %s/%p fed_pending %ld\n",
 753                  obd->obd_name, ofd->ofd_tot_pending,
 754                  exp->exp_client_uuid.uuid, exp, fed->fed_pending);
 755         /* ofd_tot_pending is handled in ofd_grant_commit as bulk
 756          * finishes */
 757         LASSERTF(ofd->ofd_tot_dirty >= fed->fed_dirty,
 758                  "%s: tot_dirty "LPU64" cli %s/%p fed_dirty %ld\n",
 759                  obd->obd_name, ofd->ofd_tot_dirty,
 760                  exp->exp_client_uuid.uuid, exp, fed->fed_dirty);
 761         ofd->ofd_tot_dirty -= fed->fed_dirty;
 762         fed->fed_dirty = 0;
 763         spin_unlock(&ofd->ofd_grant_lock);
 764 }
 765
 766 /**
 767  * Called at prepare time when handling read request. This function extracts
 768  * incoming grant information from the obdo and processes the grant shrink
 769  * request, if any.
 770  *
 771  * \param env - is the lu environment provided by the caller
 772  * \param exp - is the export of the client which sent the request
 773  * \paral oa - is the incoming obdo sent by the client
 774  */
 775 void ofd_grant_prepare_read(const struct lu_env *env,
 776                             struct obd_export *exp, struct obdo *oa)
 777 {
 778         struct ofd_device       *ofd = ofd_exp(exp);
 779         int                      do_shrink;
 780         obd_size                 left = 0;
 781
 782         if (!oa)
 783                 return;
 784
 785         if ((oa->o_valid & OBD_MD_FLGRANT) == 0)
 786                 /* The read request does not contain any grant
 787                  * information */
 788                 return;
 789
 790         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
 791             (oa->o_flags & OBD_FL_SHRINK_GRANT)) {
 792                 /* To process grant shrink request, we need to know how much
 793                  * available space remains on the backend filesystem.
 794                  * Shrink requests are not so common, we always get fresh
 795                  * statfs information. */
 796                 ofd_grant_statfs(env, exp, 1, NULL);
 797
 798                 /* protect all grant counters */
 799                 spin_lock(&ofd->ofd_grant_lock);
 800
 801                 /* Grab free space from cached statfs data and take out space
 802                  * already granted to clients as well as reserved space */
 803                 left = ofd_grant_space_left(exp);
 804
 805                 /* all set now to proceed with shrinking */
 806                 do_shrink = 1;
 807         } else {
 808                 /* no grant shrinking request packed in the obdo and
 809                  * since we don't grant space back on reads, no point
 810                  * in running statfs, so just skip it and process
 811                  * incoming grant data directly. */
 812                 spin_lock(&ofd->ofd_grant_lock);
 813                 do_shrink = 0;
 814         }
 815
 816         /* extract incoming grant infomation provided by the client */
 817         ofd_grant_incoming(env, exp, oa);
 818
 819         /* unlike writes, we don't return grants back on reads unless a grant
 820          * shrink request was packed and we decided to turn it down. */
 821         if (do_shrink)
 822                 ofd_grant_shrink(exp, oa, left);
 823         else
 824                 oa->o_grant = 0;
 825
 826         spin_unlock(&ofd->ofd_grant_lock);
 827 }
 828
 829 /**
 830  * Called at write prepare time to handle incoming grant, check that we have
 831  * enough space and grant some space back to the client if possible.
 832  *
 833  * \param env - is the lu environment provided by the caller
 834  * \param exp - is the export of the client which sent the request
 835  * \paral oa - is the incoming obdo sent by the client
 836  * \param rnb - is the list of network buffers
 837  * \param niocont - is the number of network buffers in the list
 838  */
 839 void ofd_grant_prepare_write(const struct lu_env *env,
 840                              struct obd_export *exp, struct obdo *oa,
 841                              struct niobuf_remote *rnb, int niocount)
 842 {
 843         struct obd_device       *obd = exp->exp_obd;
 844         struct ofd_device       *ofd = ofd_exp(exp);
 845         obd_size                 left;
 846         int                      from_cache;
 847         int                      force = 0; /* can use cached data intially */
 848         int                      rc;
 849
 850         ENTRY;
 851
 852 refresh:
 853         /* get statfs information from OSD layer */
 854         ofd_grant_statfs(env, exp, force, &from_cache);
 855
 856         spin_lock(&ofd->ofd_grant_lock); /* protect all grant counters */
 857
 858         /* Grab free space from cached statfs data and take out space already
 859          * granted to clients as well as reserved space */
 860         left = ofd_grant_space_left(exp);
 861
 862         /* Get fresh statfs data if we are short in ungranted space */
 863         if (from_cache && left < 32 * ofd_grant_chunk(exp, ofd)) {
 864                 spin_unlock(&ofd->ofd_grant_lock);
 865                 CDEBUG(D_CACHE, "%s: fs has no space left and statfs too old\n",
 866                        obd->obd_name);
 867                 force = 1;
 868                 goto refresh;
 869         }
 870
 871         /* When close to free space exhaustion, trigger a sync to force
 872          * writeback cache to consume required space immediately and release as
 873          * much space as possible. */
 874         if (!obd->obd_recovering && force != 2 && left < OFD_GRANT_CHUNK) {
 875                 bool from_grant = true;
 876                 int  i;
 877
 878                 /* That said, it is worth running a sync only if some pages did
 879                  * not consume grant space on the client and could thus fail
 880                  * with ENOSPC later in ofd_grant_check() */
 881                 for (i = 0; i < niocount; i++)
 882                         if (!(rnb[i].rnb_flags & OBD_BRW_FROM_GRANT))
 883                                 from_grant = false;
 884
 885                 if (!from_grant) {
 886                         /* at least one network buffer requires acquiring grant
 887                          * space on the server */
 888                         spin_unlock(&ofd->ofd_grant_lock);
 889                         /* discard errors, at least we tried ... */
 890                         rc = dt_sync(env, ofd->ofd_osd);
 891                         force = 2;
 892                         goto refresh;
 893                 }
 894         }
 895
 896         /* extract incoming grant information provided by the client */
 897         ofd_grant_incoming(env, exp, oa);
 898
 899         /* check limit */
 900         ofd_grant_check(env, exp, oa, rnb, niocount, &left);
 901
 902         if (!(oa->o_valid & OBD_MD_FLGRANT)) {
 903                 spin_unlock(&ofd->ofd_grant_lock);
 904                 RETURN_EXIT;
 905         }
 906
 907         /* if OBD_FL_SHRINK_GRANT is set, the client is willing to release some
 908          * grant space. */
 909         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
 910             (oa->o_flags & OBD_FL_SHRINK_GRANT))
 911                 ofd_grant_shrink(exp, oa, left);
 912         else
 913                 /* grant more space back to the client if possible */
 914                 oa->o_grant = ofd_grant(exp, oa->o_grant, oa->o_undirty, left,
 915                                         true);
 916         spin_unlock(&ofd->ofd_grant_lock);
 917 }
 918
 919 /**
 920  * Called during object precreation to consume grant space.
 921  * More space is granted for precreation if possible.
 922  *
 923  * \param env - is the lu environment provided by the caller
 924  * \param exp - is the export holding the grant space for precreation (= self
 925  *            export currently)
 926  * \paral nr - is the number of objects the caller wants to create objects
 927  */
 928 int ofd_grant_create(const struct lu_env *env, struct obd_export *exp, int *nr)
 929 {
 930         struct ofd_thread_info          *info = ofd_info(env);
 931         struct ofd_device               *ofd = ofd_exp(exp);
 932         struct filter_export_data       *fed = &exp->exp_filter_data;
 933         obd_size                         left = 0;
 934         unsigned long                    wanted;
 935         ENTRY;
 936
 937         info->fti_used = 0;
 938
 939         if (exp->exp_obd->obd_recovering ||
 940             ofd->ofd_dt_conf.ddp_inodespace == 0)
 941                 /* don't enforce grant during recovery */
 942                 RETURN(0);
 943
 944         /* Update statfs data if required */
 945         ofd_grant_statfs(env, exp, 1, NULL);
 946
 947         /* protect all grant counters */
 948         spin_lock(&ofd->ofd_grant_lock);
 949
 950         /* fail precreate request if there is not enough blocks available for
 951          * writing */
 952         if (ofd->ofd_osfs.os_bavail - (fed->fed_grant >> ofd->ofd_blockbits) <
 953             (ofd->ofd_osfs.os_blocks >> 10)) {
 954                 spin_unlock(&ofd->ofd_grant_lock);
 955                 CDEBUG(D_RPCTRACE, "%s: not enough space for create "LPU64"\n",
 956                        ofd_obd(ofd)->obd_name,
 957                        ofd->ofd_osfs.os_bavail * ofd->ofd_osfs.os_blocks);
 958                 RETURN(-ENOSPC);
 959         }
 960
 961         /* Grab free space from cached statfs data and take out space
 962          * already granted to clients as well as reserved space */
 963         left = ofd_grant_space_left(exp);
 964
 965         /* compute how much space is required to handle the precreation
 966          * request */
 967         wanted = *nr * ofd->ofd_dt_conf.ddp_inodespace;
 968         if (wanted > fed->fed_grant + left) {
 969                 /* that's beyond what remains, adjust the number of objects that
 970                  * can be safely precreated */
 971                 wanted = fed->fed_grant + left;
 972                 *nr = wanted / ofd->ofd_dt_conf.ddp_inodespace;
 973                 if (*nr == 0) {
 974                         /* we really have no space any more for precreation,
 975                          * fail the precreate request with ENOSPC */
 976                         spin_unlock(&ofd->ofd_grant_lock);
 977                         RETURN(-ENOSPC);
 978                 }
 979                 /* compute space needed for the new number of creations */
 980                 wanted = *nr * ofd->ofd_dt_conf.ddp_inodespace;
 981         }
 982         LASSERT(wanted <= fed->fed_grant + left);
 983
 984         if (wanted <= fed->fed_grant) {
 985                 /* we've enough grant space to handle this precreate request */
 986                 fed->fed_grant -= wanted;
 987         } else {
 988                 /* we need to take some space from the ungranted pool */
 989                 ofd->ofd_tot_granted += wanted - fed->fed_grant;
 990                 left -= wanted - fed->fed_grant;
 991                 fed->fed_grant = 0;
 992         }
 993         info->fti_used = wanted;
 994         fed->fed_pending += info->fti_used;
 995         ofd->ofd_tot_pending += info->fti_used;
 996
 997         /* grant more space for precreate purpose if possible. */
 998         wanted = OST_MAX_PRECREATE * ofd->ofd_dt_conf.ddp_inodespace / 2;
 999         if (wanted > fed->fed_grant) {
1000                 /* always try to book enough space to handle a large precreate
1001                  * request */
1002                 wanted -= fed->fed_grant;
1003                 ofd_grant(exp, fed->fed_grant, wanted, left, false);
1004         }
1005         spin_unlock(&ofd->ofd_grant_lock);
1006         RETURN(0);
1007 }
1008
1009 /**
1010  * Called at commit time to update pending grant counter for writes in flight
1011  *
1012  * \param env - is the lu environment provided by the caller
1013  * \param exp - is the export of the client which sent the request
1014  */
1015 void ofd_grant_commit(const struct lu_env *env, struct obd_export *exp,
1016                       int rc)
1017 {
1018         struct ofd_device       *ofd  = ofd_exp(exp);
1019         struct ofd_thread_info  *info = ofd_info(env);
1020         unsigned long            pending;
1021
1022         ENTRY;
1023
1024         /* get space accounted in tot_pending for the I/O, set in
1025          * ofd_grant_check() */
1026         pending = info->fti_used;
1027         if (pending == 0)
1028                 RETURN_EXIT;
1029
1030         spin_lock(&ofd->ofd_grant_lock);
1031         /* Don't update statfs data for errors raised before commit (e.g.
1032          * bulk transfer failed, ...) since we know those writes have not been
1033          * processed. For other errors hit during commit, we cannot really tell
1034          * whether or not something was written, so we update statfs data.
1035          * In any case, this should not be fatal since we always get fresh
1036          * statfs data before failing a request with ENOSPC */
1037         if (rc == 0) {
1038                 spin_lock(&ofd->ofd_osfs_lock);
1039                 /* Take pending out of cached statfs data */
1040                 ofd->ofd_osfs.os_bavail -= min_t(obd_size,
1041                                                  ofd->ofd_osfs.os_bavail,
1042                                                  pending >> ofd->ofd_blockbits);
1043                 if (ofd->ofd_statfs_inflight)
1044                         /* someone is running statfs and want to be notified of
1045                          * writes happening meanwhile */
1046                         ofd->ofd_osfs_inflight += pending;
1047                 spin_unlock(&ofd->ofd_osfs_lock);
1048         }
1049
1050         if (exp->exp_filter_data.fed_pending < pending) {
1051                 CERROR("%s: cli %s/%p fed_pending(%lu) < grant_used(%lu)\n",
1052                        exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp,
1053                        exp->exp_filter_data.fed_pending, pending);
1054                 spin_unlock(&ofd->ofd_grant_lock);
1055                 LBUG();
1056         }
1057         exp->exp_filter_data.fed_pending -= pending;
1058
1059         if (ofd->ofd_tot_granted < pending) {
1060                  CERROR("%s: cli %s/%p tot_granted("LPU64") < grant_used(%lu)"
1061                         "\n", exp->exp_obd->obd_name,
1062                         exp->exp_client_uuid.uuid, exp, ofd->ofd_tot_granted,
1063                         pending);
1064                 spin_unlock(&ofd->ofd_grant_lock);
1065                 LBUG();
1066         }
1067         ofd->ofd_tot_granted -= pending;
1068
1069         if (ofd->ofd_tot_pending < pending) {
1070                  CERROR("%s: cli %s/%p tot_pending("LPU64") < grant_used(%lu)"
1071                         "\n", exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
1072                         exp, ofd->ofd_tot_pending, pending);
1073                 spin_unlock(&ofd->ofd_grant_lock);
1074                 LBUG();
1075         }
1076         ofd->ofd_tot_pending -= pending;
1077         spin_unlock(&ofd->ofd_grant_lock);
1078         EXIT;
1079 }