lustre/ldlm/ldlm_extent.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2002, 2003 Cluster File Systems, Inc.
   5  *   Author: Peter Braam <braam@clusterfs.com>
   6  *   Author: Phil Schwan <phil@clusterfs.com>
   7  *
   8  *   This file is part of the Lustre file system, http://www.lustre.org
   9  *   Lustre is a trademark of Cluster File Systems, Inc.
  10  *
  11  *   You may have signed or agreed to another license before downloading
  12  *   this software.  If so, you are bound by the terms and conditions
  13  *   of that agreement, and the following does not apply to you.  See the
  14  *   LICENSE file included with this distribution for more information.
  15  *
  16  *   If you did not agree to a different license, then this copy of Lustre
  17  *   is open source software; you can redistribute it and/or modify it
  18  *   under the terms of version 2 of the GNU General Public License as
  19  *   published by the Free Software Foundation.
  20  *
  21  *   In either case, Lustre is distributed in the hope that it will be
  22  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  23  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  24  *   license text for more details.
  25  */
  26
  27 #define DEBUG_SUBSYSTEM S_LDLM
  28 #ifndef __KERNEL__
  29 # include <liblustre.h>
  30 #endif
  31
  32 #include <lustre_dlm.h>
  33 #include <obd_support.h>
  34 #include <lustre_lib.h>
  35
  36 #include "ldlm_internal.h"
  37
  38 /* The purpose of this function is to return:
  39  * - the maximum extent
  40  * - containing the requested extent
  41  * - and not overlapping existing conflicting extents outside the requested one
  42  */
  43 static void
  44 ldlm_extent_internal_policy(struct list_head *queue, struct ldlm_lock *req,
  45                             struct ldlm_extent *new_ex)
  46 {
  47         struct list_head *tmp;
  48         ldlm_mode_t req_mode = req->l_req_mode;
  49         __u64 req_start = req->l_req_extent.start;
  50         __u64 req_end = req->l_req_extent.end;
  51         __u64 req_align, mask;
  52         int conflicting = 0;
  53         ENTRY;
  54
  55         lockmode_verify(req_mode);
  56
  57         list_for_each(tmp, queue) {
  58                 struct ldlm_lock *lock;
  59                 struct ldlm_extent *l_extent;
  60
  61                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
  62                 l_extent = &lock->l_policy_data.l_extent;
  63
  64                 /* We already hit the minimum requested size, search no more */
  65                 if (new_ex->start == req_start && new_ex->end == req_end) {
  66                         EXIT;
  67                         return;
  68                 }
  69
  70                 /* Don't conflict with ourselves */
  71                 if (req == lock)
  72                         continue;
  73
  74                 /* Locks are compatible, overlap doesn't matter */
  75                 /* Until bug 20 is fixed, try to avoid granting overlapping
  76                  * locks on one client (they take a long time to cancel) */
  77                 if (lockmode_compat(lock->l_req_mode, req_mode) &&
  78                     lock->l_export != req->l_export)
  79                         continue;
  80
  81                 /* If this is a high-traffic lock, don't grow downwards at all
  82                  * or grow upwards too much */
  83                 ++conflicting;
  84                 if (conflicting > 4)
  85                         new_ex->start = req_start;
  86
  87                 /* If lock doesn't overlap new_ex, skip it. */
  88                 if (l_extent->end < new_ex->start ||
  89                     l_extent->start > new_ex->end)
  90                         continue;
  91
  92                 /* Locks conflicting in requested extents and we can't satisfy
  93                  * both locks, so ignore it.  Either we will ping-pong this
  94                  * extent (we would regardless of what extent we granted) or
  95                  * lock is unused and it shouldn't limit our extent growth. */
  96                 if (lock->l_req_extent.end >= req_start &&
  97                     lock->l_req_extent.start <= req_end)
  98                         continue;
  99
 100                 /* We grow extents downwards only as far as they don't overlap
 101                  * with already-granted locks, on the assumtion that clients
 102                  * will be writing beyond the initial requested end and would
 103                  * then need to enqueue a new lock beyond previous request.
 104                  * l_req_extent->end strictly < req_start, checked above. */
 105                 if (l_extent->start < req_start && new_ex->start != req_start) {
 106                         if (l_extent->end >= req_start)
 107                                 new_ex->start = req_start;
 108                         else
 109                                 new_ex->start = min(l_extent->end+1, req_start);
 110                 }
 111
 112                 /* If we need to cancel this lock anyways because our request
 113                  * overlaps the granted lock, we grow up to its requested
 114                  * extent start instead of limiting this extent, assuming that
 115                  * clients are writing forwards and the lock had over grown
 116                  * its extent downwards before we enqueued our request. */
 117                 if (l_extent->end > req_end) {
 118                         if (l_extent->start <= req_end)
 119                                 new_ex->end = max(lock->l_req_extent.start - 1,
 120                                                   req_end);
 121                         else
 122                                 new_ex->end = max(l_extent->start - 1, req_end);
 123                 }
 124         }
 125
 126 #define LDLM_MAX_GROWN_EXTENT (32 * 1024 * 1024 - 1)
 127         if (conflicting > 32 && (req_mode == LCK_PW || req_mode == LCK_CW)) {
 128                 if (req_end < req_start + LDLM_MAX_GROWN_EXTENT)
 129                         new_ex->end = min(req_start + LDLM_MAX_GROWN_EXTENT,
 130                                           new_ex->end);
 131         }
 132
 133         if (new_ex->start == 0 && new_ex->end == OBD_OBJECT_EOF) {
 134                 EXIT;
 135                 return;
 136         }
 137
 138         /* we need to ensure that the lock extent is properly aligned to what
 139          * the client requested.  We align it to the lowest-common denominator
 140          * of the clients requested lock start and end alignment. */
 141         mask = 0x1000ULL;
 142         req_align = (req_end + 1) | req_start;
 143         if (req_align != 0) {
 144                 while ((req_align & mask) == 0)
 145                         mask <<= 1;
 146         }
 147         mask -= 1;
 148         /* We can only shrink the lock, not grow it.
 149          * This should never cause lock to be smaller than requested,
 150          * since requested lock was already aligned on these boundaries. */
 151         new_ex->start = ((new_ex->start - 1) | mask) + 1;
 152         new_ex->end = ((new_ex->end + 1) & ~mask) - 1;
 153         LASSERTF(new_ex->start <= req_start,
 154                  "mask "LPX64" grant start "LPU64" req start "LPU64"\n",
 155                  mask, new_ex->start, req_start);
 156         LASSERTF(new_ex->end >= req_end,
 157                  "mask "LPX64" grant end "LPU64" req end "LPU64"\n",
 158                  mask, new_ex->end, req_end);
 159
 160         EXIT;
 161 }
 162
 163 /* In order to determine the largest possible extent we can grant, we need
 164  * to scan all of the queues. */
 165 static void ldlm_extent_policy(struct ldlm_resource *res,
 166                                struct ldlm_lock *lock, int *flags)
 167 {
 168         struct ldlm_extent new_ex = { .start = 0, .end = OBD_OBJECT_EOF };
 169
 170         if (lock->l_export == NULL)
 171                 /*
 172                  * this is local lock taken by server (e.g., as a part of
 173                  * OST-side locking, or unlink handling). Expansion doesn't
 174                  * make a lot of sense for local locks, because they are
 175                  * dropped immediately on operation completion and would only
 176                  * conflict with other threads.
 177                  */
 178                 return;
 179
 180         if (lock->l_policy_data.l_extent.start == 0 &&
 181             lock->l_policy_data.l_extent.end == OBD_OBJECT_EOF)
 182                 /* fast-path whole file locks */
 183                 return;
 184
 185         ldlm_extent_internal_policy(&res->lr_granted, lock, &new_ex);
 186         ldlm_extent_internal_policy(&res->lr_waiting, lock, &new_ex);
 187
 188         if (new_ex.start != lock->l_policy_data.l_extent.start ||
 189             new_ex.end != lock->l_policy_data.l_extent.end) {
 190                 *flags |= LDLM_FL_LOCK_CHANGED;
 191                 lock->l_policy_data.l_extent.start = new_ex.start;
 192                 lock->l_policy_data.l_extent.end = new_ex.end;
 193         }
 194 }
 195
 196 /* Determine if the lock is compatible with all locks on the queue.
 197  * We stop walking the queue if we hit ourselves so we don't take
 198  * conflicting locks enqueued after us into accound, or we'd wait forever.
 199  *
 200  * 0 if the lock is not compatible
 201  * 1 if the lock is compatible
 202  * 2 if this group lock is compatible and requires no further checking
 203  * negative error, such as EWOULDBLOCK for group locks
 204  */
 205 static int
 206 ldlm_extent_compat_queue(struct list_head *queue, struct ldlm_lock *req,
 207                          int *flags, ldlm_error_t *err,
 208                          struct list_head *work_list)
 209 {
 210         struct list_head *tmp;
 211         struct ldlm_lock *lock;
 212         ldlm_mode_t req_mode = req->l_req_mode;
 213         __u64 req_start = req->l_req_extent.start;
 214         __u64 req_end = req->l_req_extent.end;
 215         int compat = 1;
 216         int scan = 0;
 217         ENTRY;
 218
 219         lockmode_verify(req_mode);
 220
 221         list_for_each(tmp, queue) {
 222                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 223
 224                 if (req == lock)
 225                         RETURN(compat);
 226
 227                 if (unlikely(scan)) {
 228                         /* We only get here if we are queuing GROUP lock
 229                            and met some incompatible one. The main idea of this
 230                            code is to insert GROUP lock past compatible GROUP
 231                            lock in the waiting queue or if there is not any,
 232                            then in front of first non-GROUP lock */
 233                         if (lock->l_req_mode != LCK_GROUP) {
 234                                 /* Ok, we hit non-GROUP lock, there should
 235                                  * be no more GROUP locks later on, queue in
 236                                  * front of first non-GROUP lock */
 237
 238                                 ldlm_resource_insert_lock_after(lock, req);
 239                                 list_del_init(&lock->l_res_link);
 240                                 ldlm_resource_insert_lock_after(req, lock);
 241                                 RETURN(0);
 242                         }
 243                         if (req->l_policy_data.l_extent.gid ==
 244                              lock->l_policy_data.l_extent.gid) {
 245                                 /* found it */
 246                                 ldlm_resource_insert_lock_after(lock, req);
 247                                 RETURN(0);
 248                         }
 249                         continue;
 250                 }
 251
 252                 /* locks are compatible, overlap doesn't matter */
 253                 if (lockmode_compat(lock->l_req_mode, req_mode)) {
 254                         if (req_mode == LCK_PR &&
 255                             ((lock->l_policy_data.l_extent.start <=
 256                              req->l_policy_data.l_extent.start) &&
 257                              (lock->l_policy_data.l_extent.end >=
 258                               req->l_policy_data.l_extent.end))) {
 259                                 /* If we met a PR lock just like us or wider,
 260                                    and nobody down the list conflicted with
 261                                    it, that means we can skip processing of
 262                                    the rest of the list and safely place
 263                                    ourselves at the end of the list, or grant
 264                                    (dependent if we met an conflicting locks
 265                                    before in the list).
 266                                    In case of 1st enqueue only we continue
 267                                    traversing if there is something conflicting
 268                                    down the list because we need to make sure
 269                                    that something is marked as AST_SENT as well,
 270                                    in cse of empy worklist we would exit on
 271                                    first conflict met. */
 272                                 /* There IS a case where such flag is
 273                                    not set for a lock, yet it blocks
 274                                    something. Luckily for us this is
 275                                    only during destroy, so lock is
 276                                    exclusive. So here we are safe */
 277                                 if (!(lock->l_flags & LDLM_FL_AST_SENT)) {
 278                                         RETURN(compat);
 279                                 }
 280                         }
 281
 282                         /* non-group locks are compatible, overlap doesn't
 283                            matter */
 284                         if (likely(req_mode != LCK_GROUP))
 285                                 continue;
 286
 287                         /* If we are trying to get a GROUP lock and there is
 288                            another one of this kind, we need to compare gid */
 289                         if (req->l_policy_data.l_extent.gid ==
 290                             lock->l_policy_data.l_extent.gid) {
 291                                 /* If existing lock with matched gid is granted,
 292                                    we grant new one too. */
 293                                 if (lock->l_req_mode == lock->l_granted_mode)
 294                                         RETURN(2);
 295
 296                                 /* Otherwise we are scanning queue of waiting
 297                                  * locks and it means current request would
 298                                  * block along with existing lock (that is
 299                                  * already blocked.
 300                                  * If we are in nonblocking mode - return
 301                                  * immediately */
 302                                 if (*flags & LDLM_FL_BLOCK_NOWAIT) {
 303                                         compat = -EWOULDBLOCK;
 304                                         goto destroylock;
 305                                 }
 306                                 /* If this group lock is compatible with another
 307                                  * group lock on the waiting list, they must be
 308                                  * together in the list, so they can be granted
 309                                  * at the same time.  Otherwise the later lock
 310                                  * can get stuck behind another, incompatible,
 311                                  * lock. */
 312                                 ldlm_resource_insert_lock_after(lock, req);
 313                                 /* Because 'lock' is not granted, we can stop
 314                                  * processing this queue and return immediately.
 315                                  * There is no need to check the rest of the
 316                                  * list. */
 317                                 RETURN(0);
 318                         }
 319                 }
 320
 321                 if (unlikely(req_mode == LCK_GROUP &&
 322                     (lock->l_req_mode != lock->l_granted_mode))) {
 323                         scan = 1;
 324                         compat = 0;
 325                         if (lock->l_req_mode != LCK_GROUP) {
 326                         /* Ok, we hit non-GROUP lock, there should be no
 327                            more GROUP locks later on, queue in front of
 328                            first non-GROUP lock */
 329
 330                                 ldlm_resource_insert_lock_after(lock, req);
 331                                 list_del_init(&lock->l_res_link);
 332                                 ldlm_resource_insert_lock_after(req, lock);
 333                                 RETURN(0);
 334                         }
 335                         if (req->l_policy_data.l_extent.gid ==
 336                              lock->l_policy_data.l_extent.gid) {
 337                                 /* found it */
 338                                 ldlm_resource_insert_lock_after(lock, req);
 339                                 RETURN(0);
 340                         }
 341                         continue;
 342                 }
 343
 344                 if (unlikely(lock->l_req_mode == LCK_GROUP)) {
 345                         /* If compared lock is GROUP, then requested is PR/PW/
 346                          * so this is not compatible; extent range does not
 347                          * matter */
 348                         if (*flags & LDLM_FL_BLOCK_NOWAIT) {
 349                                 compat = -EWOULDBLOCK;
 350                                 goto destroylock;
 351                         } else {
 352                                 *flags |= LDLM_FL_NO_TIMEOUT;
 353                         }
 354                 } else if (lock->l_policy_data.l_extent.end < req_start ||
 355                            lock->l_policy_data.l_extent.start > req_end) {
 356                         /* if a non group lock doesn't overlap skip it */
 357                         continue;
 358                 }
 359
 360                 if (!work_list)
 361                         RETURN(0);
 362
 363                 compat = 0;
 364                 if (lock->l_blocking_ast)
 365                         ldlm_add_ast_work_item(lock, req, work_list);
 366         }
 367
 368         RETURN(compat);
 369 destroylock:
 370         list_del_init(&req->l_res_link);
 371         ldlm_lock_destroy_nolock(req);
 372         *err = compat;
 373         RETURN(compat);
 374 }
 375
 376 /* If first_enq is 0 (ie, called from ldlm_reprocess_queue):
 377   *   - blocking ASTs have already been sent
 378   *   - must call this function with the ns lock held
 379   *
 380   * If first_enq is 1 (ie, called from ldlm_lock_enqueue):
 381   *   - blocking ASTs have not been sent
 382   *   - must call this function with the ns lock held once */
 383 int ldlm_process_extent_lock(struct ldlm_lock *lock, int *flags, int first_enq,
 384                              ldlm_error_t *err, struct list_head *work_list)
 385 {
 386         struct ldlm_resource *res = lock->l_resource;
 387         struct list_head rpc_list = CFS_LIST_HEAD_INIT(rpc_list);
 388         int rc, rc2;
 389         ENTRY;
 390
 391         LASSERT(list_empty(&res->lr_converting));
 392         check_res_locked(res);
 393         *err = ELDLM_OK;
 394
 395         if (!first_enq) {
 396                 /* Careful observers will note that we don't handle -EWOULDBLOCK
 397                  * here, but it's ok for a non-obvious reason -- compat_queue
 398                  * can only return -EWOULDBLOCK if (flags & BLOCK_NOWAIT).
 399                  * flags should always be zero here, and if that ever stops
 400                  * being true, we want to find out. */
 401                 LASSERT(*flags == 0);
 402                 rc = ldlm_extent_compat_queue(&res->lr_granted, lock, flags,
 403                                               err, NULL);
 404                 if (rc == 1) {
 405                         rc = ldlm_extent_compat_queue(&res->lr_waiting, lock,
 406                                                       flags, err, NULL);
 407                 }
 408                 if (rc == 0)
 409                         RETURN(LDLM_ITER_STOP);
 410
 411                 ldlm_resource_unlink_lock(lock);
 412
 413                 if (!OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_EVICT_RACE))
 414                         ldlm_extent_policy(res, lock, flags);
 415                 ldlm_grant_lock(lock, work_list);
 416                 RETURN(LDLM_ITER_CONTINUE);
 417         }
 418
 419  restart:
 420         rc = ldlm_extent_compat_queue(&res->lr_granted, lock, flags, err, &rpc_list);
 421         if (rc < 0)
 422                 GOTO(out, rc); /* lock was destroyed */
 423         if (rc == 2)
 424                 goto grant;
 425
 426         rc2 = ldlm_extent_compat_queue(&res->lr_waiting, lock, flags, err, &rpc_list);
 427         if (rc2 < 0)
 428                 GOTO(out, rc = rc2); /* lock was destroyed */
 429
 430         if (rc + rc2 == 2) {
 431         grant:
 432                 ldlm_extent_policy(res, lock, flags);
 433                 ldlm_resource_unlink_lock(lock);
 434                 ldlm_grant_lock(lock, NULL);
 435         } else {
 436                 /* If either of the compat_queue()s returned failure, then we
 437                  * have ASTs to send and must go onto the waiting list.
 438                  *
 439                  * bug 2322: we used to unlink and re-add here, which was a
 440                  * terrible folly -- if we goto restart, we could get
 441                  * re-ordered!  Causes deadlock, because ASTs aren't sent! */
 442                 if (list_empty(&lock->l_res_link))
 443                         ldlm_resource_add_lock(res, &res->lr_waiting, lock);
 444                 unlock_res(res);
 445                 rc = ldlm_run_ast_work(&rpc_list, LDLM_WORK_BL_AST);
 446                 lock_res(res);
 447                 if (rc == -ERESTART)
 448                         GOTO(restart, -ERESTART);
 449                 *flags |= LDLM_FL_BLOCK_GRANTED;
 450                 /* this way we force client to wait for the lock
 451                  * endlessly once the lock is enqueued -bzzz */
 452                 *flags |= LDLM_FL_NO_TIMEOUT;
 453
 454         }
 455         rc = 0;
 456 out:
 457         RETURN(rc);
 458 }
 459
 460 /* When a lock is cancelled by a client, the KMS may undergo change if this
 461  * is the "highest lock".  This function returns the new KMS value.
 462  * Caller must hold ns_lock already.
 463  *
 464  * NB: A lock on [x,y] protects a KMS of up to y + 1 bytes! */
 465 __u64 ldlm_extent_shift_kms(struct ldlm_lock *lock, __u64 old_kms)
 466 {
 467         struct ldlm_resource *res = lock->l_resource;
 468         struct list_head *tmp;
 469         struct ldlm_lock *lck;
 470         __u64 kms = 0;
 471         ENTRY;
 472
 473         /* don't let another thread in ldlm_extent_shift_kms race in
 474          * just after we finish and take our lock into account in its
 475          * calculation of the kms */
 476         lock->l_flags |= LDLM_FL_KMS_IGNORE;
 477
 478         list_for_each(tmp, &res->lr_granted) {
 479                 lck = list_entry(tmp, struct ldlm_lock, l_res_link);
 480
 481                 if (lck->l_flags & LDLM_FL_KMS_IGNORE)
 482                         continue;
 483
 484                 if (lck->l_policy_data.l_extent.end >= old_kms)
 485                         RETURN(old_kms);
 486
 487                 /* This extent _has_ to be smaller than old_kms (checked above)
 488                  * so kms can only ever be smaller or the same as old_kms. */
 489                 if (lck->l_policy_data.l_extent.end + 1 > kms)
 490                         kms = lck->l_policy_data.l_extent.end + 1;
 491         }
 492         LASSERTF(kms <= old_kms, "kms "LPU64" old_kms "LPU64"\n", kms, old_kms);
 493
 494         RETURN(kms);
 495 }