lustre/ldlm/ldlm_extent.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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2010, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/ldlm/ldlm_extent.c
  32  *
  33  * Author: Peter Braam <braam@clusterfs.com>
  34  * Author: Phil Schwan <phil@clusterfs.com>
  35  */
  36
  37 /**
  38  * This file contains implementation of EXTENT lock type
  39  *
  40  * EXTENT lock type is for locking a contiguous range of values, represented
  41  * by 64-bit starting and ending offsets (inclusive). There are several extent
  42  * lock modes, some of which may be mutually incompatible. Extent locks are
  43  * considered incompatible if their modes are incompatible and their extents
  44  * intersect.  See the lock mode compatibility matrix in lustre_dlm.h.
  45  */
  46
  47 #define DEBUG_SUBSYSTEM S_LDLM
  48
  49 #include <libcfs/libcfs.h>
  50 #include <lustre_dlm.h>
  51 #include <obd_support.h>
  52 #include <obd.h>
  53 #include <obd_class.h>
  54 #include <lustre_lib.h>
  55
  56 #include "ldlm_internal.h"
  57
  58 #ifdef HAVE_SERVER_SUPPORT
  59 # define LDLM_MAX_GROWN_EXTENT (32 * 1024 * 1024 - 1)
  60
  61 /**
  62  * Fix up the ldlm_extent after expanding it.
  63  *
  64  * After expansion has been done, we might still want to do certain adjusting
  65  * based on overall contention of the resource and the like to avoid granting
  66  * overly wide locks.
  67  */
  68 static void ldlm_extent_internal_policy_fixup(struct ldlm_lock *req,
  69                                               struct ldlm_extent *new_ex,
  70                                               int conflicting)
  71 {
  72         enum ldlm_mode req_mode = req->l_req_mode;
  73         __u64 req_start = req->l_req_extent.start;
  74         __u64 req_end = req->l_req_extent.end;
  75         __u64 req_align, mask;
  76
  77         if (conflicting > 32 && (req_mode == LCK_PW || req_mode == LCK_CW)) {
  78                 if (req_end < req_start + LDLM_MAX_GROWN_EXTENT)
  79                         new_ex->end = min(req_start + LDLM_MAX_GROWN_EXTENT,
  80                                           new_ex->end);
  81         }
  82
  83         if (new_ex->start == 0 && new_ex->end == OBD_OBJECT_EOF) {
  84                 EXIT;
  85                 return;
  86         }
  87
  88         /* we need to ensure that the lock extent is properly aligned to what
  89          * the client requested. Also we need to make sure it's also server
  90          * page size aligned otherwise a server page can be covered by two
  91          * write locks.
  92          */
  93         mask = PAGE_SIZE;
  94         req_align = (req_end + 1) | req_start;
  95         if (req_align != 0 && (req_align & (mask - 1)) == 0) {
  96                 while ((req_align & mask) == 0)
  97                         mask <<= 1;
  98         }
  99         mask -= 1;
 100         /* We can only shrink the lock, not grow it.
 101          * This should never cause lock to be smaller than requested,
 102          * since requested lock was already aligned on these boundaries.
 103          */
 104         new_ex->start = ((new_ex->start - 1) | mask) + 1;
 105         new_ex->end = ((new_ex->end + 1) & ~mask) - 1;
 106         LASSERTF(new_ex->start <= req_start,
 107                  "mask %#llx grant start %llu req start %llu\n",
 108                  mask, new_ex->start, req_start);
 109         LASSERTF(new_ex->end >= req_end,
 110                  "mask %#llx grant end %llu req end %llu\n",
 111                  mask, new_ex->end, req_end);
 112 }
 113
 114 /**
 115  * Return the maximum extent that:
 116  * - contains the requested extent
 117  * - does not overlap existing conflicting extents outside the requested one
 118  *
 119  * This allows clients to request a small required extent range, but if there
 120  * is no contention on the lock the full lock can be granted to the client.
 121  * This avoids the need for many smaller lock requests to be granted in the
 122  * common (uncontended) case.
 123  *
 124  * Use interval tree to expand the lock extent for granted lock.
 125  */
 126 static void ldlm_extent_internal_policy_granted(struct ldlm_lock *req,
 127                                                 struct ldlm_extent *new_ex)
 128 {
 129         struct ldlm_resource *res = req->l_resource;
 130         enum ldlm_mode req_mode = req->l_req_mode;
 131         __u64 req_start = req->l_req_extent.start;
 132         __u64 req_end = req->l_req_extent.end;
 133         struct ldlm_interval_tree *tree;
 134         struct interval_node_extent limiter = {
 135                 .start  = new_ex->start,
 136                 .end    = new_ex->end,
 137         };
 138         int conflicting = 0;
 139         int idx;
 140
 141         ENTRY;
 142
 143         lockmode_verify(req_mode);
 144
 145         /* Using interval tree to handle the LDLM extent granted locks. */
 146         for (idx = 0; idx < LCK_MODE_NUM; idx++) {
 147                 struct interval_node_extent ext = {
 148                         .start  = req_start,
 149                         .end    = req_end,
 150                 };
 151
 152                 tree = &res->lr_itree[idx];
 153                 if (lockmode_compat(tree->lit_mode, req_mode))
 154                         continue;
 155
 156                 conflicting += tree->lit_size;
 157                 if (conflicting > 4)
 158                         limiter.start = req_start;
 159
 160                 if (interval_is_overlapped(tree->lit_root, &ext))
 161                         CDEBUG(D_INFO,
 162                                "req_mode = %d, tree->lit_mode = %d, tree->lit_size = %d\n",
 163                                req_mode, tree->lit_mode, tree->lit_size);
 164                 interval_expand(tree->lit_root, &ext, &limiter);
 165                 limiter.start = max(limiter.start, ext.start);
 166                 limiter.end = min(limiter.end, ext.end);
 167                 if (limiter.start == req_start && limiter.end == req_end)
 168                         break;
 169         }
 170
 171         new_ex->start = limiter.start;
 172         new_ex->end = limiter.end;
 173         LASSERT(new_ex->start <= req_start);
 174         LASSERT(new_ex->end >= req_end);
 175
 176         ldlm_extent_internal_policy_fixup(req, new_ex, conflicting);
 177         EXIT;
 178 }
 179
 180 /* The purpose of this function is to return:
 181  * - the maximum extent
 182  * - containing the requested extent
 183  * - and not overlapping existing conflicting extents outside the requested one
 184  */
 185 static void
 186 ldlm_extent_internal_policy_waiting(struct ldlm_lock *req,
 187                                     struct ldlm_extent *new_ex)
 188 {
 189         struct ldlm_resource *res = req->l_resource;
 190         enum ldlm_mode req_mode = req->l_req_mode;
 191         __u64 req_start = req->l_req_extent.start;
 192         __u64 req_end = req->l_req_extent.end;
 193         struct ldlm_lock *lock;
 194         int conflicting = 0;
 195
 196         ENTRY;
 197
 198         lockmode_verify(req_mode);
 199
 200         /* for waiting locks */
 201         list_for_each_entry(lock, &res->lr_waiting, l_res_link) {
 202                 struct ldlm_extent *l_extent = &lock->l_policy_data.l_extent;
 203
 204                 /* We already hit the minimum requested size, search no more */
 205                 if (new_ex->start == req_start && new_ex->end == req_end) {
 206                         EXIT;
 207                         return;
 208                 }
 209
 210                 /* Don't conflict with ourselves */
 211                 if (req == lock)
 212                         continue;
 213
 214                 /* Locks are compatible, overlap doesn't matter
 215                  * Until bug 20 is fixed, try to avoid granting overlapping
 216                  * locks on one client (they take a long time to cancel)
 217                  */
 218                 if (lockmode_compat(lock->l_req_mode, req_mode) &&
 219                     lock->l_export != req->l_export)
 220                         continue;
 221
 222                 /* If this is a high-traffic lock, don't grow downwards at all
 223                  * or grow upwards too much
 224                  */
 225                 ++conflicting;
 226                 if (conflicting > 4)
 227                         new_ex->start = req_start;
 228
 229                 /* If lock doesn't overlap new_ex, skip it. */
 230                 if (!ldlm_extent_overlap(l_extent, new_ex))
 231                         continue;
 232
 233                 /* Locks conflicting in requested extents and we can't satisfy
 234                  * both locks, so ignore it.  Either we will ping-pong this
 235                  * extent (we would regardless of what extent we granted) or
 236                  * lock is unused and it shouldn't limit our extent growth.
 237                  */
 238                 if (ldlm_extent_overlap(&lock->l_req_extent,
 239                                          &req->l_req_extent))
 240                         continue;
 241
 242                 /* We grow extents downwards only as far as they don't overlap
 243                  * with already-granted locks, on the assumption that clients
 244                  * will be writing beyond the initial requested end and would
 245                  * then need to enqueue a new lock beyond previous request.
 246                  * l_req_extent->end strictly < req_start, checked above.
 247                  */
 248                 if (l_extent->start < req_start && new_ex->start != req_start) {
 249                         if (l_extent->end >= req_start)
 250                                 new_ex->start = req_start;
 251                         else
 252                                 new_ex->start = min(l_extent->end+1, req_start);
 253                 }
 254
 255                 /* If we need to cancel this lock anyways because our request
 256                  * overlaps the granted lock, we grow up to its requested
 257                  * extent start instead of limiting this extent, assuming that
 258                  * clients are writing forwards and the lock had over grown
 259                  * its extent downwards before we enqueued our request.
 260                  */
 261                 if (l_extent->end > req_end) {
 262                         if (l_extent->start <= req_end)
 263                                 new_ex->end = max(lock->l_req_extent.start - 1,
 264                                                   req_end);
 265                         else
 266                                 new_ex->end = max(l_extent->start - 1, req_end);
 267                 }
 268         }
 269
 270         ldlm_extent_internal_policy_fixup(req, new_ex, conflicting);
 271         EXIT;
 272 }
 273
 274
 275 /* In order to determine the largest possible extent we can grant, we need
 276  * to scan all of the queues.
 277  */
 278 static void ldlm_extent_policy(struct ldlm_resource *res,
 279                                struct ldlm_lock *lock, __u64 *flags)
 280 {
 281         struct ldlm_extent new_ex = { .start = 0, .end = OBD_OBJECT_EOF };
 282
 283         if (lock->l_export == NULL)
 284                 /* this is a local lock taken by server (e.g., as a part of
 285                  * OST-side locking, or unlink handling). Expansion doesn't
 286                  * make a lot of sense for local locks, because they are
 287                  * dropped immediately on operation completion and would only
 288                  * conflict with other threads.
 289                  */
 290                 return;
 291
 292         if (lock->l_policy_data.l_extent.start == 0 &&
 293             lock->l_policy_data.l_extent.end == OBD_OBJECT_EOF)
 294                 /* fast-path whole file locks */
 295                 return;
 296
 297         /* Because reprocess_queue zeroes flags and uses it to return
 298          * LDLM_FL_LOCK_CHANGED, we must check for the NO_EXPANSION flag
 299          * in the lock flags rather than the 'flags' argument
 300          */
 301         if (likely(!(lock->l_flags & LDLM_FL_NO_EXPANSION))) {
 302                 ldlm_extent_internal_policy_granted(lock, &new_ex);
 303                 ldlm_extent_internal_policy_waiting(lock, &new_ex);
 304         } else {
 305                 LDLM_DEBUG(lock, "Not expanding manually requested lock");
 306                 new_ex.start = lock->l_policy_data.l_extent.start;
 307                 new_ex.end = lock->l_policy_data.l_extent.end;
 308                 /* In case the request is not on correct boundaries, we call
 309                  * fixup. (normally called in ldlm_extent_internal_policy_*)
 310                  */
 311                 ldlm_extent_internal_policy_fixup(lock, &new_ex, 0);
 312         }
 313
 314         if (!ldlm_extent_equal(&new_ex, &lock->l_policy_data.l_extent)) {
 315                 *flags |= LDLM_FL_LOCK_CHANGED;
 316                 lock->l_policy_data.l_extent.start = new_ex.start;
 317                 lock->l_policy_data.l_extent.end = new_ex.end;
 318         }
 319 }
 320
 321 static bool ldlm_check_contention(struct ldlm_lock *lock, int contended_locks)
 322 {
 323         struct ldlm_resource *res = lock->l_resource;
 324         time64_t now = ktime_get_seconds();
 325
 326         if (CFS_FAIL_CHECK(OBD_FAIL_LDLM_SET_CONTENTION))
 327                 return true;
 328
 329         CDEBUG(D_DLMTRACE, "contended locks = %d\n", contended_locks);
 330         if (contended_locks > ldlm_res_to_ns(res)->ns_contended_locks)
 331                 res->lr_contention_time = now;
 332
 333         return now < res->lr_contention_time +
 334                 ldlm_res_to_ns(res)->ns_contention_time;
 335 }
 336
 337 struct ldlm_extent_compat_args {
 338         struct list_head *work_list;
 339         struct ldlm_lock *lock;
 340         enum ldlm_mode mode;
 341         int *locks;
 342         int *compat;
 343 };
 344
 345 static enum interval_iter ldlm_extent_compat_cb(struct interval_node *n,
 346                                                 void *data)
 347 {
 348         struct ldlm_lock *lock = container_of(n, struct ldlm_lock, l_tree_node);
 349         struct ldlm_extent_compat_args *priv = data;
 350         struct list_head *work_list = priv->work_list;
 351         struct ldlm_lock *enq = priv->lock;
 352         enum ldlm_mode mode = priv->mode;
 353
 354         ENTRY;
 355         /* interval tree is for granted lock */
 356         LASSERTF(mode == lock->l_granted_mode,
 357                  "mode = %s, lock->l_granted_mode = %s\n",
 358                  ldlm_lockname[mode],
 359                  ldlm_lockname[lock->l_granted_mode]);
 360         if (lock->l_blocking_ast && lock->l_granted_mode != LCK_GROUP)
 361                 ldlm_add_ast_work_item(lock, enq, work_list);
 362
 363         /* don't count conflicting glimpse locks */
 364         if (!(mode == LCK_PR && lock->l_policy_data.l_extent.start == 0 &&
 365               lock->l_policy_data.l_extent.end == OBD_OBJECT_EOF))
 366                 *priv->locks += 1;
 367
 368         if (priv->compat)
 369                 *priv->compat = 0;
 370
 371         RETURN(INTERVAL_ITER_CONT);
 372 }
 373
 374 /**
 375  * Determine if the lock is compatible with all locks on the queue.
 376  *
 377  * If \a work_list is provided, conflicting locks are linked there.
 378  * If \a work_list is not provided, we exit this function on first conflict.
 379  *
 380  * \retval 0 if the lock is not compatible
 381  * \retval 1 if the lock is compatible
 382  * \retval 2 if \a req is a group lock and it is compatible and requires
 383  *           no further checking
 384  * \retval negative error, such as EAGAIN for group locks
 385  */
 386 static int
 387 ldlm_extent_compat_queue(struct list_head *queue, struct ldlm_lock *req,
 388                          __u64 *flags, struct list_head *work_list,
 389                          int *contended_locks)
 390 {
 391         struct ldlm_resource *res = req->l_resource;
 392         enum ldlm_mode req_mode = req->l_req_mode;
 393         __u64 req_start = req->l_req_extent.start;
 394         __u64 req_end = req->l_req_extent.end;
 395         struct ldlm_lock *lock;
 396         int check_contention;
 397         int compat = 1;
 398
 399         ENTRY;
 400
 401         lockmode_verify(req_mode);
 402
 403         /* Using interval tree for granted lock */
 404         if (queue == &res->lr_granted) {
 405                 struct ldlm_interval_tree *tree;
 406                 struct ldlm_extent_compat_args data = {
 407                         .work_list = work_list,
 408                         .lock = req,
 409                         .locks = contended_locks,
 410                         .compat = &compat };
 411                 struct interval_node_extent ex = {
 412                         .start = req_start,
 413                         .end = req_end };
 414                 int idx, rc;
 415
 416                 for (idx = 0; idx < LCK_MODE_NUM; idx++) {
 417                         tree = &res->lr_itree[idx];
 418                         if (tree->lit_root == NULL) /* empty tree, skipped */
 419                                 continue;
 420
 421                         data.mode = tree->lit_mode;
 422                         if (lockmode_compat(req_mode, tree->lit_mode)) {
 423                                 struct ldlm_lock *lock;
 424
 425                                 if (req_mode != LCK_GROUP)
 426                                         continue;
 427
 428                                 /* group lock, grant it immediately if
 429                                  * compatible
 430                                  */
 431                                 lock = container_of(tree->lit_root,
 432                                                     struct ldlm_lock,
 433                                                     l_tree_node);
 434                                 if (req->l_policy_data.l_extent.gid ==
 435                                     lock->l_policy_data.l_extent.gid)
 436                                         RETURN(2);
 437                         }
 438
 439                         if (tree->lit_mode == LCK_GROUP) {
 440                                 if (*flags & (LDLM_FL_BLOCK_NOWAIT |
 441                                     LDLM_FL_SPECULATIVE)) {
 442                                         compat = -EAGAIN;
 443                                         goto destroylock;
 444                                 }
 445
 446                                 if (!work_list)
 447                                         RETURN(0);
 448
 449                                 /* if work list is not NULL,add all
 450                                  * locks in the tree to work list
 451                                  */
 452                                 compat = 0;
 453                                 interval_iterate(tree->lit_root,
 454                                                  ldlm_extent_compat_cb, &data);
 455                                 continue;
 456                         }
 457
 458                         /* We've found a potentially blocking lock, check
 459                          * compatibility.  This handles locks other than GROUP
 460                          * locks, which are handled separately above.
 461                          *
 462                          * Locks with FL_SPECULATIVE are asynchronous requests
 463                          * which must never wait behind another lock, so they
 464                          * fail if any conflicting lock is found.
 465                          */
 466                         if (!work_list || (*flags & LDLM_FL_SPECULATIVE)) {
 467                                 rc = interval_is_overlapped(tree->lit_root,
 468                                                             &ex);
 469                                 if (rc) {
 470                                         if (!work_list) {
 471                                                 RETURN(0);
 472                                         } else {
 473                                                 compat = -EAGAIN;
 474                                                 goto destroylock;
 475                                         }
 476                                 }
 477                         } else {
 478                                 interval_search(tree->lit_root, &ex,
 479                                                 ldlm_extent_compat_cb, &data);
 480                                 if (!list_empty(work_list) && compat)
 481                                         compat = 0;
 482                         }
 483                 }
 484         } else { /* for waiting queue */
 485                 list_for_each_entry(lock, queue, l_res_link) {
 486                         check_contention = 1;
 487
 488                         /* We stop walking the queue if we hit ourselves so
 489                          * we don't take conflicting locks enqueued after us
 490                          * into account, or we'd wait forever.
 491                          */
 492                         if (req == lock)
 493                                 break;
 494
 495                         /* locks are compatible, overlap doesn't matter */
 496                         if (lockmode_compat(lock->l_req_mode, req_mode)) {
 497                                 if (req_mode == LCK_PR &&
 498                                 ((lock->l_policy_data.l_extent.start <=
 499                                 req->l_policy_data.l_extent.start) &&
 500                                 (lock->l_policy_data.l_extent.end >=
 501                                 req->l_policy_data.l_extent.end))) {
 502                                         /* If we met a PR lock just like us or
 503                                          * wider, and nobody down the list
 504                                          * conflicted with it, that means we
 505                                          * can skip processing of the rest of
 506                                          * the list and safely place ourselves
 507                                          * at the end of the list, or grant
 508                                          * (dependent if we met an conflicting
 509                                          * locks before in the list).  In case
 510                                          * of 1st enqueue only we continue
 511                                          * traversing if there is something
 512                                          * conflicting down the list because
 513                                          * we need to make sure that something
 514                                          * is marked as AST_SENT as well, in
 515                                          * case of empy worklist we would exit
 516                                          * on first conflict met.
 517                                          */
 518
 519                                         /* There IS a case where such flag is
 520                                          * not set for a lock, yet it blocks
 521                                          * something. Luckily for us this is
 522                                          * only during destroy, so lock is
 523                                          * exclusive. So here we are safe
 524                                          */
 525                                         if (!ldlm_is_ast_sent(lock))
 526                                                 RETURN(compat);
 527                                 }
 528
 529                                 /* non-group locks are compatible,
 530                                  * overlap doesn't matter
 531                                  */
 532                                 if (likely(req_mode != LCK_GROUP))
 533                                         continue;
 534
 535                                 /* If we are trying to get a GROUP lock and
 536                                  * there is another one of this kind, we need
 537                                  * to compare gid
 538                                  */
 539                                 if (req->l_policy_data.l_extent.gid ==
 540                                 lock->l_policy_data.l_extent.gid) {
 541                                         /* If existing lock with matched gid
 542                                          * is granted, we grant new one too.
 543                                          */
 544                                         if (ldlm_is_granted(lock))
 545                                                 RETURN(2);
 546
 547                                         /* Otherwise we are scanning queue of
 548                                          * waiting locks and it means current
 549                                          * request would block along with
 550                                          * existing lock (that is already
 551                                          * blocked.  If we are in nonblocking
 552                                          * mode - return immediately
 553                                          */
 554                                         if (*flags & (LDLM_FL_BLOCK_NOWAIT |
 555                                                       LDLM_FL_SPECULATIVE)) {
 556                                                 compat = -EAGAIN;
 557                                                 goto destroylock;
 558                                         }
 559                                         /* If this group lock is compatible with
 560                                          * another group lock on the waiting
 561                                          * list, they must be together in the
 562                                          * list, so they can be granted at the
 563                                          * same time.  Otherwise the later lock
 564                                          * can get stuck behind another,
 565                                          * incompatible, lock.
 566                                          */
 567                                         ldlm_resource_insert_lock_after(lock,
 568                                                                         req);
 569                                         /* Because 'lock' is not granted, we can
 570                                          * stop processing this queue and return
 571                                          * immediately. There is no need to
 572                                          * check the rest of the list.
 573                                          */
 574                                         RETURN(0);
 575                                 }
 576                         }
 577
 578                         if (unlikely(req_mode == LCK_GROUP &&
 579                             !ldlm_is_granted(lock))) {
 580                                 compat = 0;
 581                                 if (lock->l_req_mode != LCK_GROUP) {
 582                                         /* Ok, we hit non-GROUP lock, there
 583                                          * should be no more GROUP locks later
 584                                          * on, queue in front of first
 585                                          * non-GROUP lock
 586                                          */
 587
 588                                         ldlm_resource_insert_lock_before(lock,
 589                                                                          req);
 590                                         break;
 591                                 }
 592                                 LASSERT(req->l_policy_data.l_extent.gid !=
 593                                         lock->l_policy_data.l_extent.gid);
 594                                 continue;
 595                         }
 596
 597                         if (unlikely(lock->l_req_mode == LCK_GROUP)) {
 598                                 /* If compared lock is GROUP, then requested is
 599                                  * PR/PW so this is not compatible; extent
 600                                  * range does not matter
 601                                  */
 602                                 if (*flags & (LDLM_FL_BLOCK_NOWAIT |
 603                                     LDLM_FL_SPECULATIVE)) {
 604                                         compat = -EAGAIN;
 605                                         goto destroylock;
 606                                 }
 607                         } else if (lock->l_policy_data.l_extent.end < req_start ||
 608                                    lock->l_policy_data.l_extent.start > req_end) {
 609                                 /* if non group lock doesn't overlap skip it */
 610                                 continue;
 611                         } else if (lock->l_req_extent.end < req_start ||
 612                                    lock->l_req_extent.start > req_end) {
 613                                 /* false contention, the requests doesn't
 614                                  * really overlap */
 615                                 check_contention = 0;
 616                         }
 617
 618                         if (!work_list)
 619                                 RETURN(0);
 620
 621                         if (*flags & LDLM_FL_SPECULATIVE) {
 622                                 compat = -EAGAIN;
 623                                 goto destroylock;
 624                         }
 625
 626                         /* don't count conflicting glimpse locks */
 627                         if (lock->l_req_mode == LCK_PR &&
 628                             lock->l_policy_data.l_extent.start == 0 &&
 629                              lock->l_policy_data.l_extent.end == OBD_OBJECT_EOF)
 630                                 check_contention = 0;
 631
 632                         *contended_locks += check_contention;
 633
 634                         compat = 0;
 635                         if (lock->l_blocking_ast &&
 636                             lock->l_req_mode != LCK_GROUP)
 637                                 ldlm_add_ast_work_item(lock, req, work_list);
 638                 }
 639         }
 640
 641         if (ldlm_check_contention(req, *contended_locks) &&
 642             compat == 0 && (*flags & LDLM_FL_DENY_ON_CONTENTION) &&
 643              req->l_req_mode != LCK_GROUP && req_end - req_start <=
 644               ldlm_res_to_ns(req->l_resource)->ns_max_nolock_size)
 645                 GOTO(destroylock, compat = -EUSERS);
 646
 647         RETURN(compat);
 648 destroylock:
 649         list_del_init(&req->l_res_link);
 650         ldlm_lock_destroy_nolock(req);
 651         RETURN(compat);
 652         }
 653
 654 /**
 655  * This function refresh eviction timer for cancelled lock.
 656  * \param[in] lock              ldlm lock for refresh
 657  * \param[in] arg               ldlm prolong arguments, timeout, export, extent
 658  *                              and counter are used
 659  */
 660 void ldlm_lock_prolong_one(struct ldlm_lock *lock,
 661                            struct ldlm_prolong_args *arg)
 662 {
 663         timeout_t timeout;
 664
 665         CFS_FAIL_TIMEOUT(OBD_FAIL_LDLM_PROLONG_PAUSE, 3);
 666
 667         if (arg->lpa_export != lock->l_export ||
 668             lock->l_flags & LDLM_FL_DESTROYED)
 669                 /* ignore unrelated locks */
 670                 return;
 671
 672         arg->lpa_locks_cnt++;
 673
 674         if (!(lock->l_flags & LDLM_FL_AST_SENT))
 675                 /* ignore locks not being cancelled */
 676                 return;
 677
 678         arg->lpa_blocks_cnt++;
 679
 680         /* OK. this is a possible lock the user holds doing I/O
 681          * let's refresh eviction timer for it.
 682          */
 683         timeout = ldlm_bl_timeout_by_rpc(arg->lpa_req);
 684         LDLM_DEBUG(lock, "refreshed to %ds", timeout);
 685         ldlm_refresh_waiting_lock(lock, timeout);
 686 }
 687 EXPORT_SYMBOL(ldlm_lock_prolong_one);
 688
 689 static enum interval_iter ldlm_resource_prolong_cb(struct interval_node *n,
 690                                                    void *data)
 691 {
 692         struct ldlm_lock *lock = container_of(n, struct ldlm_lock, l_tree_node);
 693         struct ldlm_prolong_args *arg = data;
 694
 695         ENTRY;
 696         ldlm_lock_prolong_one(lock, arg);
 697         RETURN(INTERVAL_ITER_CONT);
 698 }
 699
 700 /**
 701  * Walk through granted tree and prolong locks if they overlaps extent.
 702  *
 703  * \param[in] arg               prolong args
 704  */
 705 void ldlm_resource_prolong(struct ldlm_prolong_args *arg)
 706 {
 707         struct ldlm_interval_tree *tree;
 708         struct ldlm_resource *res;
 709         struct interval_node_extent ex = { .start = arg->lpa_extent.start,
 710                                         .end = arg->lpa_extent.end };
 711         int idx;
 712
 713         ENTRY;
 714
 715         res = ldlm_resource_get(arg->lpa_export->exp_obd->obd_namespace,
 716                                 &arg->lpa_resid, LDLM_EXTENT, 0);
 717         if (IS_ERR(res)) {
 718                 CDEBUG(D_DLMTRACE, "Failed to get resource for resid %llu/%llu\n",
 719                 arg->lpa_resid.name[0], arg->lpa_resid.name[1]);
 720                 RETURN_EXIT;
 721         }
 722
 723         lock_res(res);
 724         for (idx = 0; idx < LCK_MODE_NUM; idx++) {
 725                 tree = &res->lr_itree[idx];
 726                 if (tree->lit_root == NULL) /* empty tree, skipped */
 727                         continue;
 728
 729                 /* There is no possibility to check for the groupID
 730                  * so all the group locks are considered as valid
 731                  * here, especially because the client is supposed
 732                  * to check it has such a lock before sending an RPC.
 733                  */
 734                 if (!(tree->lit_mode & arg->lpa_mode))
 735                         continue;
 736
 737                 interval_search(tree->lit_root, &ex,
 738                                 ldlm_resource_prolong_cb, arg);
 739         }
 740         unlock_res(res);
 741         ldlm_resource_putref(res);
 742
 743         EXIT;
 744 }
 745 EXPORT_SYMBOL(ldlm_resource_prolong);
 746
 747 /**
 748  * Process a granting attempt for extent lock.
 749  * Must be called with ns lock held.
 750  *
 751  * This function looks for any conflicts for \a lock in the granted or
 752  * waiting queues. The lock is granted if no conflicts are found in
 753  * either queue.
 754  */
 755 int ldlm_process_extent_lock(struct ldlm_lock *lock, __u64 *flags,
 756                 enum ldlm_process_intention intention,
 757                 enum ldlm_error *err, struct list_head *work_list)
 758 {
 759         struct ldlm_resource *res = lock->l_resource;
 760         int rc, rc2 = 0;
 761         int contended_locks = 0;
 762         struct list_head *grant_work = intention == LDLM_PROCESS_ENQUEUE ?
 763                                                         NULL : work_list;
 764         ENTRY;
 765
 766         LASSERT(!ldlm_is_granted(lock));
 767         LASSERT(!(*flags & LDLM_FL_DENY_ON_CONTENTION) ||
 768                 !ldlm_is_ast_discard_data(lock));
 769         check_res_locked(res);
 770         *err = ELDLM_OK;
 771
 772         if (intention == LDLM_PROCESS_RESCAN) {
 773                 /* Careful observers will note that we don't handle -EAGAIN
 774                  * here, but it's ok for a non-obvious reason -- compat_queue
 775                  * can only return -EAGAIN if (flags & BLOCK_NOWAIT |
 776                  * SPECULATIVE). flags should always be zero here, and if that
 777                  * ever stops being true, we want to find out. */
 778                 LASSERT(*flags == 0);
 779                 rc = ldlm_extent_compat_queue(&res->lr_granted, lock, flags,
 780                                         NULL, &contended_locks);
 781                 if (rc == 1) {
 782                         rc = ldlm_extent_compat_queue(&res->lr_waiting, lock,
 783                                                 flags, NULL,
 784                                                 &contended_locks);
 785                 }
 786                 if (rc == 0)
 787                         RETURN(LDLM_ITER_STOP);
 788
 789                 ldlm_resource_unlink_lock(lock);
 790
 791                 if (!CFS_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_EVICT_RACE))
 792                         ldlm_extent_policy(res, lock, flags);
 793                 ldlm_grant_lock(lock, grant_work);
 794                 RETURN(LDLM_ITER_CONTINUE);
 795         }
 796
 797         contended_locks = 0;
 798         rc = ldlm_extent_compat_queue(&res->lr_granted, lock, flags,
 799                                 work_list, &contended_locks);
 800         if (rc < 0)
 801                 GOTO(out, *err = rc);
 802
 803         if (rc != 2) {
 804                 rc2 = ldlm_extent_compat_queue(&res->lr_waiting, lock,
 805                                         flags, work_list,
 806                                         &contended_locks);
 807                 if (rc2 < 0)
 808                         GOTO(out, *err = rc = rc2);
 809         }
 810
 811         if (rc + rc2 == 2) {
 812                 ldlm_extent_policy(res, lock, flags);
 813                 ldlm_resource_unlink_lock(lock);
 814                 ldlm_grant_lock(lock, grant_work);
 815         } else {
 816                 /* Adding LDLM_FL_NO_TIMEOUT flag to granted lock to
 817                  * force client to wait for the lock endlessly once
 818                  * the lock is enqueued -bzzz */
 819                 *flags |= LDLM_FL_NO_TIMEOUT;
 820         }
 821
 822         RETURN(LDLM_ITER_CONTINUE);
 823 out:
 824         return rc;
 825 }
 826 #endif /* HAVE_SERVER_SUPPORT */
 827
 828 struct ldlm_kms_shift_args {
 829         __u64   old_kms;
 830         __u64   kms;
 831         bool    complete;
 832 };
 833
 834 /* Callback for interval_iterate functions, used by ldlm_extent_shift_Kms */
 835 static enum interval_iter ldlm_kms_shift_cb(struct interval_node *n,
 836                                             void *args)
 837 {
 838         struct ldlm_lock *lock = container_of(n, struct ldlm_lock, l_tree_node);
 839         struct ldlm_kms_shift_args *arg = args;
 840
 841         ENTRY;
 842         /* Since all locks in an interval have the same extent, we can just
 843          * use the lock without kms_ignore set.
 844          */
 845         if (ldlm_is_kms_ignore(lock))
 846                 RETURN(INTERVAL_ITER_CONT);
 847
 848         /* If we find a lock with a greater or equal kms, we are not the
 849          * highest lock (or we share that distinction with another lock), and
 850          * don't need to update KMS.  Return old_kms and stop looking.
 851          */
 852         if (lock->l_policy_data.l_extent.end == OBD_OBJECT_EOF ||
 853         lock->l_policy_data.l_extent.end + 1 >= arg->old_kms) {
 854                 arg->kms = arg->old_kms;
 855                 arg->complete = true;
 856                 RETURN(INTERVAL_ITER_STOP);
 857         }
 858
 859         if (lock->l_policy_data.l_extent.end + 1 > arg->kms)
 860                 arg->kms = lock->l_policy_data.l_extent.end + 1;
 861
 862         /* Since interval_iterate_reverse starts with the highest lock and
 863          * works down, for PW locks, we only need to check if we should update
 864          * the kms, then stop walking the tree.  PR locks are not exclusive, so
 865          * the highest start does not imply the highest end and we must
 866          * continue. (Only one group lock is allowed per resource, so this is
 867          * irrelevant for group locks.)*/
 868         if (lock->l_granted_mode == LCK_PW)
 869                 RETURN(INTERVAL_ITER_STOP);
 870         else
 871                 RETURN(INTERVAL_ITER_CONT);
 872 }
 873
 874 /* When a lock is cancelled by a client, the KMS may undergo change if this
 875  * is the "highest lock".  This function returns the new KMS value, updating
 876  * it only if we were the highest lock.
 877  *
 878  * Caller must hold lr_lock already.
 879  *
 880  * NB: A lock on [x,y] protects a KMS of up to y + 1 bytes!
 881  */
 882 __u64 ldlm_extent_shift_kms(struct ldlm_lock *lock, __u64 old_kms)
 883 {
 884         struct ldlm_resource *res = lock->l_resource;
 885         struct ldlm_interval_tree *tree;
 886         struct ldlm_kms_shift_args args;
 887         int idx = 0;
 888
 889         ENTRY;
 890
 891         args.old_kms = old_kms;
 892         args.kms = 0;
 893         args.complete = false;
 894
 895         /* don't let another thread in ldlm_extent_shift_kms race in
 896          * just after we finish and take our lock into account in its
 897          * calculation of the kms */
 898         ldlm_set_kms_ignore(lock);
 899
 900         /* We iterate over the lock trees, looking for the largest kms smaller
 901          * than the current one. */
 902         for (idx = 0; idx < LCK_MODE_NUM; idx++) {
 903                 tree = &res->lr_itree[idx];
 904
 905                 /* If our already known kms is >= than the highest 'end' in
 906                  * this tree, we don't need to check this tree, because
 907                  * the kms from a tree can be lower than in_max_high (due to
 908                  * kms_ignore), but it can never be higher. */
 909                 if (!tree->lit_root || args.kms >= tree->lit_root->in_max_high)
 910                         continue;
 911
 912                 interval_iterate_reverse(tree->lit_root, ldlm_kms_shift_cb,
 913                                          &args);
 914
 915                 /* this tells us we're not the highest lock, so we don't need
 916                  * to check the remaining trees */
 917                 if (args.complete)
 918                         break;
 919         }
 920
 921         LASSERTF(args.kms <= args.old_kms, "kms %llu old_kms %llu\n", args.kms,
 922                  args.old_kms);
 923
 924         RETURN(args.kms);
 925 }
 926 EXPORT_SYMBOL(ldlm_extent_shift_kms);
 927
 928 static inline int ldlm_mode_to_index(enum ldlm_mode mode)
 929 {
 930         int index;
 931
 932         LASSERT(mode != 0);
 933         LASSERT(is_power_of_2(mode));
 934         index = ilog2(mode);
 935         LASSERT(index < LCK_MODE_NUM);
 936         return index;
 937 }
 938
 939 /** Add newly granted lock into interval tree for the resource. */
 940 void ldlm_extent_add_lock(struct ldlm_resource *res,
 941                           struct ldlm_lock *lock)
 942 {
 943         struct interval_node **root;
 944         struct ldlm_extent *extent;
 945         int idx, rc;
 946
 947         LASSERT(ldlm_is_granted(lock));
 948
 949         LASSERT(!interval_is_intree(&lock->l_tree_node));
 950
 951         idx = ldlm_mode_to_index(lock->l_granted_mode);
 952         LASSERT(lock->l_granted_mode == BIT(idx));
 953         LASSERT(lock->l_granted_mode == res->lr_itree[idx].lit_mode);
 954
 955         /* node extent initialize */
 956         extent = &lock->l_policy_data.l_extent;
 957
 958         rc = interval_set(&lock->l_tree_node, extent->start, extent->end);
 959         LASSERT(!rc);
 960
 961         root = &res->lr_itree[idx].lit_root;
 962         interval_insert(&lock->l_tree_node, root);
 963         res->lr_itree[idx].lit_size++;
 964
 965         /* even though we use interval tree to manage the extent lock, we also
 966          * add the locks into grant list, for debug purpose, .. */
 967         ldlm_resource_add_lock(res, &res->lr_granted, lock);
 968
 969         if (CFS_FAIL_CHECK(OBD_FAIL_LDLM_GRANT_CHECK)) {
 970                 struct ldlm_lock *lck;
 971
 972                 list_for_each_entry_reverse(lck, &res->lr_granted,
 973                                         l_res_link) {
 974                         if (lck == lock)
 975                                 continue;
 976                         if (lockmode_compat(lck->l_granted_mode,
 977                                         lock->l_granted_mode))
 978                                 continue;
 979                         if (ldlm_extent_overlap(&lck->l_req_extent,
 980                                                 &lock->l_req_extent)) {
 981                                 CDEBUG(D_ERROR,
 982                                        "granting conflicting lock %p %p\n",
 983                                        lck, lock);
 984                                 ldlm_resource_dump(D_ERROR, res);
 985                                 LBUG();
 986                         }
 987                 }
 988         }
 989 }
 990
 991 /** Remove cancelled lock from resource interval tree. */
 992 void ldlm_extent_unlink_lock(struct ldlm_lock *lock)
 993 {
 994         struct ldlm_resource *res = lock->l_resource;
 995         struct ldlm_interval_tree *tree;
 996         int idx;
 997
 998         if (!interval_is_intree(&lock->l_tree_node)) /* duplicate unlink */
 999                 return;
1000
1001         idx = ldlm_mode_to_index(lock->l_granted_mode);
1002         LASSERT(lock->l_granted_mode == BIT(idx));
1003         tree = &res->lr_itree[idx];
1004
1005         LASSERT(tree->lit_root != NULL); /* assure the tree is not null */
1006
1007         tree->lit_size--;
1008         interval_erase(&lock->l_tree_node, &tree->lit_root);
1009 }
1010
1011 void ldlm_extent_policy_wire_to_local(const union ldlm_wire_policy_data *wpolicy,
1012                                       union ldlm_policy_data *lpolicy)
1013 {
1014         lpolicy->l_extent.start = wpolicy->l_extent.start;
1015         lpolicy->l_extent.end = wpolicy->l_extent.end;
1016         lpolicy->l_extent.gid = wpolicy->l_extent.gid;
1017 }
1018
1019 void ldlm_extent_policy_local_to_wire(const union ldlm_policy_data *lpolicy,
1020                                       union ldlm_wire_policy_data *wpolicy)
1021 {
1022         memset(wpolicy, 0, sizeof(*wpolicy));
1023         wpolicy->l_extent.start = lpolicy->l_extent.start;
1024         wpolicy->l_extent.end = lpolicy->l_extent.end;
1025         wpolicy->l_extent.gid = lpolicy->l_extent.gid;
1026 }