drivers/staging/lustre/lustre/ldlm/ldlm_lock.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, 2015, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/ldlm/ldlm_lock.c
  33  *
  34  * Author: Peter Braam <braam@clusterfs.com>
  35  * Author: Phil Schwan <phil@clusterfs.com>
  36  */
  37
  38 #define DEBUG_SUBSYSTEM S_LDLM
  39
  40 #include "../../include/linux/libcfs/libcfs.h"
  41 #include "../include/lustre_intent.h"
  42 #include "../include/obd_class.h"
  43 #include "ldlm_internal.h"
  44
  45 /* lock types */
  46 char *ldlm_lockname[] = {
  47         [0]             = "--",
  48         [LCK_EX]        = "EX",
  49         [LCK_PW]        = "PW",
  50         [LCK_PR]        = "PR",
  51         [LCK_CW]        = "CW",
  52         [LCK_CR]        = "CR",
  53         [LCK_NL]        = "NL",
  54         [LCK_GROUP]     = "GROUP",
  55         [LCK_COS]       = "COS",
  56 };
  57 EXPORT_SYMBOL(ldlm_lockname);
  58
  59 static char *ldlm_typename[] = {
  60         [LDLM_PLAIN]    = "PLN",
  61         [LDLM_EXTENT]   = "EXT",
  62         [LDLM_FLOCK]    = "FLK",
  63         [LDLM_IBITS]    = "IBT",
  64 };
  65
  66 static ldlm_policy_wire_to_local_t ldlm_policy_wire18_to_local[] = {
  67         [LDLM_PLAIN - LDLM_MIN_TYPE]    = ldlm_plain_policy_wire_to_local,
  68         [LDLM_EXTENT - LDLM_MIN_TYPE]   = ldlm_extent_policy_wire_to_local,
  69         [LDLM_FLOCK - LDLM_MIN_TYPE]    = ldlm_flock_policy_wire18_to_local,
  70         [LDLM_IBITS - LDLM_MIN_TYPE]    = ldlm_ibits_policy_wire_to_local,
  71 };
  72
  73 static ldlm_policy_wire_to_local_t ldlm_policy_wire21_to_local[] = {
  74         [LDLM_PLAIN - LDLM_MIN_TYPE]    = ldlm_plain_policy_wire_to_local,
  75         [LDLM_EXTENT - LDLM_MIN_TYPE]   = ldlm_extent_policy_wire_to_local,
  76         [LDLM_FLOCK - LDLM_MIN_TYPE]    = ldlm_flock_policy_wire21_to_local,
  77         [LDLM_IBITS - LDLM_MIN_TYPE]    = ldlm_ibits_policy_wire_to_local,
  78 };
  79
  80 static ldlm_policy_local_to_wire_t ldlm_policy_local_to_wire[] = {
  81         [LDLM_PLAIN - LDLM_MIN_TYPE]    = ldlm_plain_policy_local_to_wire,
  82         [LDLM_EXTENT - LDLM_MIN_TYPE]   = ldlm_extent_policy_local_to_wire,
  83         [LDLM_FLOCK - LDLM_MIN_TYPE]    = ldlm_flock_policy_local_to_wire,
  84         [LDLM_IBITS - LDLM_MIN_TYPE]    = ldlm_ibits_policy_local_to_wire,
  85 };
  86
  87 /**
  88  * Converts lock policy from local format to on the wire lock_desc format
  89  */
  90 static void ldlm_convert_policy_to_wire(enum ldlm_type type,
  91                                         const ldlm_policy_data_t *lpolicy,
  92                                         ldlm_wire_policy_data_t *wpolicy)
  93 {
  94         ldlm_policy_local_to_wire_t convert;
  95
  96         convert = ldlm_policy_local_to_wire[type - LDLM_MIN_TYPE];
  97
  98         convert(lpolicy, wpolicy);
  99 }
 100
 101 /**
 102  * Converts lock policy from on the wire lock_desc format to local format
 103  */
 104 void ldlm_convert_policy_to_local(struct obd_export *exp, enum ldlm_type type,
 105                                   const ldlm_wire_policy_data_t *wpolicy,
 106                                   ldlm_policy_data_t *lpolicy)
 107 {
 108         ldlm_policy_wire_to_local_t convert;
 109         int new_client;
 110
 111         /** some badness for 2.0.0 clients, but 2.0.0 isn't supported */
 112         new_client = (exp_connect_flags(exp) & OBD_CONNECT_FULL20) != 0;
 113         if (new_client)
 114                 convert = ldlm_policy_wire21_to_local[type - LDLM_MIN_TYPE];
 115         else
 116                 convert = ldlm_policy_wire18_to_local[type - LDLM_MIN_TYPE];
 117
 118         convert(wpolicy, lpolicy);
 119 }
 120
 121 char *ldlm_it2str(int it)
 122 {
 123         switch (it) {
 124         case IT_OPEN:
 125                 return "open";
 126         case IT_CREAT:
 127                 return "creat";
 128         case (IT_OPEN | IT_CREAT):
 129                 return "open|creat";
 130         case IT_READDIR:
 131                 return "readdir";
 132         case IT_GETATTR:
 133                 return "getattr";
 134         case IT_LOOKUP:
 135                 return "lookup";
 136         case IT_UNLINK:
 137                 return "unlink";
 138         case IT_GETXATTR:
 139                 return "getxattr";
 140         case IT_LAYOUT:
 141                 return "layout";
 142         default:
 143                 CERROR("Unknown intent %d\n", it);
 144                 return "UNKNOWN";
 145         }
 146 }
 147 EXPORT_SYMBOL(ldlm_it2str);
 148
 149 /*
 150  * REFCOUNTED LOCK OBJECTS
 151  */
 152
 153 /**
 154  * Get a reference on a lock.
 155  *
 156  * Lock refcounts, during creation:
 157  *   - one special one for allocation, dec'd only once in destroy
 158  *   - one for being a lock that's in-use
 159  *   - one for the addref associated with a new lock
 160  */
 161 struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock)
 162 {
 163         atomic_inc(&lock->l_refc);
 164         return lock;
 165 }
 166 EXPORT_SYMBOL(ldlm_lock_get);
 167
 168 /**
 169  * Release lock reference.
 170  *
 171  * Also frees the lock if it was last reference.
 172  */
 173 void ldlm_lock_put(struct ldlm_lock *lock)
 174 {
 175         LASSERT(lock->l_resource != LP_POISON);
 176         LASSERT(atomic_read(&lock->l_refc) > 0);
 177         if (atomic_dec_and_test(&lock->l_refc)) {
 178                 struct ldlm_resource *res;
 179
 180                 LDLM_DEBUG(lock,
 181                            "final lock_put on destroyed lock, freeing it.");
 182
 183                 res = lock->l_resource;
 184                 LASSERT(ldlm_is_destroyed(lock));
 185                 LASSERT(list_empty(&lock->l_res_link));
 186                 LASSERT(list_empty(&lock->l_pending_chain));
 187
 188                 lprocfs_counter_decr(ldlm_res_to_ns(res)->ns_stats,
 189                                      LDLM_NSS_LOCKS);
 190                 lu_ref_del(&res->lr_reference, "lock", lock);
 191                 ldlm_resource_putref(res);
 192                 lock->l_resource = NULL;
 193                 if (lock->l_export) {
 194                         class_export_lock_put(lock->l_export, lock);
 195                         lock->l_export = NULL;
 196                 }
 197
 198                 kfree(lock->l_lvb_data);
 199
 200                 ldlm_interval_free(ldlm_interval_detach(lock));
 201                 lu_ref_fini(&lock->l_reference);
 202                 OBD_FREE_RCU(lock, sizeof(*lock), &lock->l_handle);
 203         }
 204 }
 205 EXPORT_SYMBOL(ldlm_lock_put);
 206
 207 /**
 208  * Removes LDLM lock \a lock from LRU. Assumes LRU is already locked.
 209  */
 210 int ldlm_lock_remove_from_lru_nolock(struct ldlm_lock *lock)
 211 {
 212         int rc = 0;
 213
 214         if (!list_empty(&lock->l_lru)) {
 215                 struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
 216
 217                 LASSERT(lock->l_resource->lr_type != LDLM_FLOCK);
 218                 list_del_init(&lock->l_lru);
 219                 LASSERT(ns->ns_nr_unused > 0);
 220                 ns->ns_nr_unused--;
 221                 rc = 1;
 222         }
 223         return rc;
 224 }
 225
 226 /**
 227  * Removes LDLM lock \a lock from LRU. Obtains the LRU lock first.
 228  *
 229  * If \a last_use is non-zero, it will remove the lock from LRU only if
 230  * it matches lock's l_last_used.
 231  *
 232  * \retval 0 if \a last_use is set, the lock is not in LRU list or \a last_use
 233  *           doesn't match lock's l_last_used;
 234  *           otherwise, the lock hasn't been in the LRU list.
 235  * \retval 1 the lock was in LRU list and removed.
 236  */
 237 int ldlm_lock_remove_from_lru_check(struct ldlm_lock *lock, time_t last_use)
 238 {
 239         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
 240         int rc = 0;
 241
 242         spin_lock(&ns->ns_lock);
 243         if (last_use == 0 || last_use == lock->l_last_used)
 244                 rc = ldlm_lock_remove_from_lru_nolock(lock);
 245         spin_unlock(&ns->ns_lock);
 246
 247         return rc;
 248 }
 249
 250 /**
 251  * Adds LDLM lock \a lock to namespace LRU. Assumes LRU is already locked.
 252  */
 253 static void ldlm_lock_add_to_lru_nolock(struct ldlm_lock *lock)
 254 {
 255         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
 256
 257         lock->l_last_used = cfs_time_current();
 258         LASSERT(list_empty(&lock->l_lru));
 259         LASSERT(lock->l_resource->lr_type != LDLM_FLOCK);
 260         list_add_tail(&lock->l_lru, &ns->ns_unused_list);
 261         ldlm_clear_skipped(lock);
 262         LASSERT(ns->ns_nr_unused >= 0);
 263         ns->ns_nr_unused++;
 264 }
 265
 266 /**
 267  * Adds LDLM lock \a lock to namespace LRU. Obtains necessary LRU locks
 268  * first.
 269  */
 270 static void ldlm_lock_add_to_lru(struct ldlm_lock *lock)
 271 {
 272         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
 273
 274         spin_lock(&ns->ns_lock);
 275         ldlm_lock_add_to_lru_nolock(lock);
 276         spin_unlock(&ns->ns_lock);
 277 }
 278
 279 /**
 280  * Moves LDLM lock \a lock that is already in namespace LRU to the tail of
 281  * the LRU. Performs necessary LRU locking
 282  */
 283 static void ldlm_lock_touch_in_lru(struct ldlm_lock *lock)
 284 {
 285         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
 286
 287         spin_lock(&ns->ns_lock);
 288         if (!list_empty(&lock->l_lru)) {
 289                 ldlm_lock_remove_from_lru_nolock(lock);
 290                 ldlm_lock_add_to_lru_nolock(lock);
 291         }
 292         spin_unlock(&ns->ns_lock);
 293 }
 294
 295 /**
 296  * Helper to destroy a locked lock.
 297  *
 298  * Used by ldlm_lock_destroy and ldlm_lock_destroy_nolock
 299  * Must be called with l_lock and lr_lock held.
 300  *
 301  * Does not actually free the lock data, but rather marks the lock as
 302  * destroyed by setting l_destroyed field in the lock to 1.  Destroys a
 303  * handle->lock association too, so that the lock can no longer be found
 304  * and removes the lock from LRU list.  Actual lock freeing occurs when
 305  * last lock reference goes away.
 306  *
 307  * Original comment (of some historical value):
 308  * This used to have a 'strict' flag, which recovery would use to mark an
 309  * in-use lock as needing-to-die.  Lest I am ever tempted to put it back, I
 310  * shall explain why it's gone: with the new hash table scheme, once you call
 311  * ldlm_lock_destroy, you can never drop your final references on this lock.
 312  * Because it's not in the hash table anymore.  -phil
 313  */
 314 static int ldlm_lock_destroy_internal(struct ldlm_lock *lock)
 315 {
 316         if (lock->l_readers || lock->l_writers) {
 317                 LDLM_ERROR(lock, "lock still has references");
 318                 LBUG();
 319         }
 320
 321         if (!list_empty(&lock->l_res_link)) {
 322                 LDLM_ERROR(lock, "lock still on resource");
 323                 LBUG();
 324         }
 325
 326         if (ldlm_is_destroyed(lock)) {
 327                 LASSERT(list_empty(&lock->l_lru));
 328                 return 0;
 329         }
 330         ldlm_set_destroyed(lock);
 331
 332         if (lock->l_export && lock->l_export->exp_lock_hash) {
 333                 /* NB: it's safe to call cfs_hash_del() even lock isn't
 334                  * in exp_lock_hash.
 335                  */
 336                 /* In the function below, .hs_keycmp resolves to
 337                  * ldlm_export_lock_keycmp()
 338                  */
 339                 /* coverity[overrun-buffer-val] */
 340                 cfs_hash_del(lock->l_export->exp_lock_hash,
 341                              &lock->l_remote_handle, &lock->l_exp_hash);
 342         }
 343
 344         ldlm_lock_remove_from_lru(lock);
 345         class_handle_unhash(&lock->l_handle);
 346
 347         return 1;
 348 }
 349
 350 /**
 351  * Destroys a LDLM lock \a lock. Performs necessary locking first.
 352  */
 353 static void ldlm_lock_destroy(struct ldlm_lock *lock)
 354 {
 355         int first;
 356
 357         lock_res_and_lock(lock);
 358         first = ldlm_lock_destroy_internal(lock);
 359         unlock_res_and_lock(lock);
 360
 361         /* drop reference from hashtable only for first destroy */
 362         if (first) {
 363                 lu_ref_del(&lock->l_reference, "hash", lock);
 364                 LDLM_LOCK_RELEASE(lock);
 365         }
 366 }
 367
 368 /**
 369  * Destroys a LDLM lock \a lock that is already locked.
 370  */
 371 void ldlm_lock_destroy_nolock(struct ldlm_lock *lock)
 372 {
 373         int first;
 374
 375         first = ldlm_lock_destroy_internal(lock);
 376         /* drop reference from hashtable only for first destroy */
 377         if (first) {
 378                 lu_ref_del(&lock->l_reference, "hash", lock);
 379                 LDLM_LOCK_RELEASE(lock);
 380         }
 381 }
 382
 383 /* this is called by portals_handle2object with the handle lock taken */
 384 static void lock_handle_addref(void *lock)
 385 {
 386         LDLM_LOCK_GET((struct ldlm_lock *)lock);
 387 }
 388
 389 static void lock_handle_free(void *lock, int size)
 390 {
 391         LASSERT(size == sizeof(struct ldlm_lock));
 392         kmem_cache_free(ldlm_lock_slab, lock);
 393 }
 394
 395 static struct portals_handle_ops lock_handle_ops = {
 396         .hop_addref = lock_handle_addref,
 397         .hop_free   = lock_handle_free,
 398 };
 399
 400 /**
 401  *
 402  * Allocate and initialize new lock structure.
 403  *
 404  * usage: pass in a resource on which you have done ldlm_resource_get
 405  *      new lock will take over the refcount.
 406  * returns: lock with refcount 2 - one for current caller and one for remote
 407  */
 408 static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource)
 409 {
 410         struct ldlm_lock *lock;
 411
 412         LASSERT(resource);
 413
 414         lock = kmem_cache_zalloc(ldlm_lock_slab, GFP_NOFS);
 415         if (!lock)
 416                 return NULL;
 417
 418         spin_lock_init(&lock->l_lock);
 419         lock->l_resource = resource;
 420         lu_ref_add(&resource->lr_reference, "lock", lock);
 421
 422         atomic_set(&lock->l_refc, 2);
 423         INIT_LIST_HEAD(&lock->l_res_link);
 424         INIT_LIST_HEAD(&lock->l_lru);
 425         INIT_LIST_HEAD(&lock->l_pending_chain);
 426         INIT_LIST_HEAD(&lock->l_bl_ast);
 427         INIT_LIST_HEAD(&lock->l_cp_ast);
 428         INIT_LIST_HEAD(&lock->l_rk_ast);
 429         init_waitqueue_head(&lock->l_waitq);
 430         lock->l_blocking_lock = NULL;
 431         INIT_LIST_HEAD(&lock->l_sl_mode);
 432         INIT_LIST_HEAD(&lock->l_sl_policy);
 433         INIT_HLIST_NODE(&lock->l_exp_hash);
 434         INIT_HLIST_NODE(&lock->l_exp_flock_hash);
 435
 436         lprocfs_counter_incr(ldlm_res_to_ns(resource)->ns_stats,
 437                              LDLM_NSS_LOCKS);
 438         INIT_LIST_HEAD(&lock->l_handle.h_link);
 439         class_handle_hash(&lock->l_handle, &lock_handle_ops);
 440
 441         lu_ref_init(&lock->l_reference);
 442         lu_ref_add(&lock->l_reference, "hash", lock);
 443         lock->l_callback_timeout = 0;
 444
 445 #if LUSTRE_TRACKS_LOCK_EXP_REFS
 446         INIT_LIST_HEAD(&lock->l_exp_refs_link);
 447         lock->l_exp_refs_nr = 0;
 448         lock->l_exp_refs_target = NULL;
 449 #endif
 450         INIT_LIST_HEAD(&lock->l_exp_list);
 451
 452         return lock;
 453 }
 454
 455 /**
 456  * Moves LDLM lock \a lock to another resource.
 457  * This is used on client when server returns some other lock than requested
 458  * (typically as a result of intent operation)
 459  */
 460 int ldlm_lock_change_resource(struct ldlm_namespace *ns, struct ldlm_lock *lock,
 461                               const struct ldlm_res_id *new_resid)
 462 {
 463         struct ldlm_resource *oldres = lock->l_resource;
 464         struct ldlm_resource *newres;
 465         int type;
 466
 467         lock_res_and_lock(lock);
 468         if (memcmp(new_resid, &lock->l_resource->lr_name,
 469                    sizeof(lock->l_resource->lr_name)) == 0) {
 470                 /* Nothing to do */
 471                 unlock_res_and_lock(lock);
 472                 return 0;
 473         }
 474
 475         LASSERT(new_resid->name[0] != 0);
 476
 477         /* This function assumes that the lock isn't on any lists */
 478         LASSERT(list_empty(&lock->l_res_link));
 479
 480         type = oldres->lr_type;
 481         unlock_res_and_lock(lock);
 482
 483         newres = ldlm_resource_get(ns, NULL, new_resid, type, 1);
 484         if (IS_ERR(newres))
 485                 return PTR_ERR(newres);
 486
 487         lu_ref_add(&newres->lr_reference, "lock", lock);
 488         /*
 489          * To flip the lock from the old to the new resource, lock, oldres and
 490          * newres have to be locked. Resource spin-locks are nested within
 491          * lock->l_lock, and are taken in the memory address order to avoid
 492          * dead-locks.
 493          */
 494         spin_lock(&lock->l_lock);
 495         oldres = lock->l_resource;
 496         if (oldres < newres) {
 497                 lock_res(oldres);
 498                 lock_res_nested(newres, LRT_NEW);
 499         } else {
 500                 lock_res(newres);
 501                 lock_res_nested(oldres, LRT_NEW);
 502         }
 503         LASSERT(memcmp(new_resid, &oldres->lr_name,
 504                        sizeof(oldres->lr_name)) != 0);
 505         lock->l_resource = newres;
 506         unlock_res(oldres);
 507         unlock_res_and_lock(lock);
 508
 509         /* ...and the flowers are still standing! */
 510         lu_ref_del(&oldres->lr_reference, "lock", lock);
 511         ldlm_resource_putref(oldres);
 512
 513         return 0;
 514 }
 515 EXPORT_SYMBOL(ldlm_lock_change_resource);
 516
 517 /** \defgroup ldlm_handles LDLM HANDLES
 518  * Ways to get hold of locks without any addresses.
 519  * @{
 520  */
 521
 522 /**
 523  * Fills in handle for LDLM lock \a lock into supplied \a lockh
 524  * Does not take any references.
 525  */
 526 void ldlm_lock2handle(const struct ldlm_lock *lock, struct lustre_handle *lockh)
 527 {
 528         lockh->cookie = lock->l_handle.h_cookie;
 529 }
 530 EXPORT_SYMBOL(ldlm_lock2handle);
 531
 532 /**
 533  * Obtain a lock reference by handle.
 534  *
 535  * if \a flags: atomically get the lock and set the flags.
 536  *            Return NULL if flag already set
 537  */
 538 struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle,
 539                                      __u64 flags)
 540 {
 541         struct ldlm_lock *lock;
 542
 543         LASSERT(handle);
 544
 545         lock = class_handle2object(handle->cookie, NULL);
 546         if (!lock)
 547                 return NULL;
 548
 549         if (lock->l_export && lock->l_export->exp_failed) {
 550                 CDEBUG(D_INFO, "lock export failed: lock %p, exp %p\n",
 551                        lock, lock->l_export);
 552                 LDLM_LOCK_PUT(lock);
 553                 return NULL;
 554         }
 555
 556         /* It's unlikely but possible that someone marked the lock as
 557          * destroyed after we did handle2object on it
 558          */
 559         if (flags == 0 && !ldlm_is_destroyed(lock)) {
 560                 lu_ref_add(&lock->l_reference, "handle", current);
 561                 return lock;
 562         }
 563
 564         lock_res_and_lock(lock);
 565
 566         LASSERT(lock->l_resource);
 567
 568         lu_ref_add_atomic(&lock->l_reference, "handle", current);
 569         if (unlikely(ldlm_is_destroyed(lock))) {
 570                 unlock_res_and_lock(lock);
 571                 CDEBUG(D_INFO, "lock already destroyed: lock %p\n", lock);
 572                 LDLM_LOCK_PUT(lock);
 573                 return NULL;
 574         }
 575
 576         if (flags) {
 577                 if (lock->l_flags & flags) {
 578                         unlock_res_and_lock(lock);
 579                         LDLM_LOCK_PUT(lock);
 580                         return NULL;
 581                 }
 582
 583                 lock->l_flags |= flags;
 584         }
 585
 586         unlock_res_and_lock(lock);
 587         return lock;
 588 }
 589 EXPORT_SYMBOL(__ldlm_handle2lock);
 590 /** @} ldlm_handles */
 591
 592 /**
 593  * Fill in "on the wire" representation for given LDLM lock into supplied
 594  * lock descriptor \a desc structure.
 595  */
 596 void ldlm_lock2desc(struct ldlm_lock *lock, struct ldlm_lock_desc *desc)
 597 {
 598         ldlm_res2desc(lock->l_resource, &desc->l_resource);
 599         desc->l_req_mode = lock->l_req_mode;
 600         desc->l_granted_mode = lock->l_granted_mode;
 601         ldlm_convert_policy_to_wire(lock->l_resource->lr_type,
 602                                     &lock->l_policy_data,
 603                                     &desc->l_policy_data);
 604 }
 605 EXPORT_SYMBOL(ldlm_lock2desc);
 606
 607 /**
 608  * Add a lock to list of conflicting locks to send AST to.
 609  *
 610  * Only add if we have not sent a blocking AST to the lock yet.
 611  */
 612 static void ldlm_add_bl_work_item(struct ldlm_lock *lock, struct ldlm_lock *new,
 613                                   struct list_head *work_list)
 614 {
 615         if (!ldlm_is_ast_sent(lock)) {
 616                 LDLM_DEBUG(lock, "lock incompatible; sending blocking AST.");
 617                 ldlm_set_ast_sent(lock);
 618                 /* If the enqueuing client said so, tell the AST recipient to
 619                  * discard dirty data, rather than writing back.
 620                  */
 621                 if (ldlm_is_ast_discard_data(new))
 622                         ldlm_set_discard_data(lock);
 623                 LASSERT(list_empty(&lock->l_bl_ast));
 624                 list_add(&lock->l_bl_ast, work_list);
 625                 LDLM_LOCK_GET(lock);
 626                 LASSERT(!lock->l_blocking_lock);
 627                 lock->l_blocking_lock = LDLM_LOCK_GET(new);
 628         }
 629 }
 630
 631 /**
 632  * Add a lock to list of just granted locks to send completion AST to.
 633  */
 634 static void ldlm_add_cp_work_item(struct ldlm_lock *lock,
 635                                   struct list_head *work_list)
 636 {
 637         if (!ldlm_is_cp_reqd(lock)) {
 638                 ldlm_set_cp_reqd(lock);
 639                 LDLM_DEBUG(lock, "lock granted; sending completion AST.");
 640                 LASSERT(list_empty(&lock->l_cp_ast));
 641                 list_add(&lock->l_cp_ast, work_list);
 642                 LDLM_LOCK_GET(lock);
 643         }
 644 }
 645
 646 /**
 647  * Aggregator function to add AST work items into a list. Determines
 648  * what sort of an AST work needs to be done and calls the proper
 649  * adding function.
 650  * Must be called with lr_lock held.
 651  */
 652 static void ldlm_add_ast_work_item(struct ldlm_lock *lock,
 653                                    struct ldlm_lock *new,
 654                                    struct list_head *work_list)
 655 {
 656         check_res_locked(lock->l_resource);
 657         if (new)
 658                 ldlm_add_bl_work_item(lock, new, work_list);
 659         else
 660                 ldlm_add_cp_work_item(lock, work_list);
 661 }
 662
 663 /**
 664  * Add specified reader/writer reference to LDLM lock with handle \a lockh.
 665  * r/w reference type is determined by \a mode
 666  * Calls ldlm_lock_addref_internal.
 667  */
 668 void ldlm_lock_addref(const struct lustre_handle *lockh, __u32 mode)
 669 {
 670         struct ldlm_lock *lock;
 671
 672         lock = ldlm_handle2lock(lockh);
 673         LASSERTF(lock, "Non-existing lock: %llx\n", lockh->cookie);
 674         ldlm_lock_addref_internal(lock, mode);
 675         LDLM_LOCK_PUT(lock);
 676 }
 677 EXPORT_SYMBOL(ldlm_lock_addref);
 678
 679 /**
 680  * Helper function.
 681  * Add specified reader/writer reference to LDLM lock \a lock.
 682  * r/w reference type is determined by \a mode
 683  * Removes lock from LRU if it is there.
 684  * Assumes the LDLM lock is already locked.
 685  */
 686 void ldlm_lock_addref_internal_nolock(struct ldlm_lock *lock, __u32 mode)
 687 {
 688         ldlm_lock_remove_from_lru(lock);
 689         if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
 690                 lock->l_readers++;
 691                 lu_ref_add_atomic(&lock->l_reference, "reader", lock);
 692         }
 693         if (mode & (LCK_EX | LCK_CW | LCK_PW | LCK_GROUP | LCK_COS)) {
 694                 lock->l_writers++;
 695                 lu_ref_add_atomic(&lock->l_reference, "writer", lock);
 696         }
 697         LDLM_LOCK_GET(lock);
 698         lu_ref_add_atomic(&lock->l_reference, "user", lock);
 699         LDLM_DEBUG(lock, "ldlm_lock_addref(%s)", ldlm_lockname[mode]);
 700 }
 701
 702 /**
 703  * Attempts to add reader/writer reference to a lock with handle \a lockh, and
 704  * fails if lock is already LDLM_FL_CBPENDING or destroyed.
 705  *
 706  * \retval 0 success, lock was addref-ed
 707  *
 708  * \retval -EAGAIN lock is being canceled.
 709  */
 710 int ldlm_lock_addref_try(const struct lustre_handle *lockh, __u32 mode)
 711 {
 712         struct ldlm_lock *lock;
 713         int            result;
 714
 715         result = -EAGAIN;
 716         lock = ldlm_handle2lock(lockh);
 717         if (lock) {
 718                 lock_res_and_lock(lock);
 719                 if (lock->l_readers != 0 || lock->l_writers != 0 ||
 720                     !ldlm_is_cbpending(lock)) {
 721                         ldlm_lock_addref_internal_nolock(lock, mode);
 722                         result = 0;
 723                 }
 724                 unlock_res_and_lock(lock);
 725                 LDLM_LOCK_PUT(lock);
 726         }
 727         return result;
 728 }
 729 EXPORT_SYMBOL(ldlm_lock_addref_try);
 730
 731 /**
 732  * Add specified reader/writer reference to LDLM lock \a lock.
 733  * Locks LDLM lock and calls ldlm_lock_addref_internal_nolock to do the work.
 734  * Only called for local locks.
 735  */
 736 void ldlm_lock_addref_internal(struct ldlm_lock *lock, __u32 mode)
 737 {
 738         lock_res_and_lock(lock);
 739         ldlm_lock_addref_internal_nolock(lock, mode);
 740         unlock_res_and_lock(lock);
 741 }
 742
 743 /**
 744  * Removes reader/writer reference for LDLM lock \a lock.
 745  * Assumes LDLM lock is already locked.
 746  * only called in ldlm_flock_destroy and for local locks.
 747  * Does NOT add lock to LRU if no r/w references left to accommodate flock locks
 748  * that cannot be placed in LRU.
 749  */
 750 void ldlm_lock_decref_internal_nolock(struct ldlm_lock *lock, __u32 mode)
 751 {
 752         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
 753         if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
 754                 LASSERT(lock->l_readers > 0);
 755                 lu_ref_del(&lock->l_reference, "reader", lock);
 756                 lock->l_readers--;
 757         }
 758         if (mode & (LCK_EX | LCK_CW | LCK_PW | LCK_GROUP | LCK_COS)) {
 759                 LASSERT(lock->l_writers > 0);
 760                 lu_ref_del(&lock->l_reference, "writer", lock);
 761                 lock->l_writers--;
 762         }
 763
 764         lu_ref_del(&lock->l_reference, "user", lock);
 765         LDLM_LOCK_RELEASE(lock);    /* matches the LDLM_LOCK_GET() in addref */
 766 }
 767
 768 /**
 769  * Removes reader/writer reference for LDLM lock \a lock.
 770  * Locks LDLM lock first.
 771  * If the lock is determined to be client lock on a client and r/w refcount
 772  * drops to zero and the lock is not blocked, the lock is added to LRU lock
 773  * on the namespace.
 774  * For blocked LDLM locks if r/w count drops to zero, blocking_ast is called.
 775  */
 776 void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode)
 777 {
 778         struct ldlm_namespace *ns;
 779
 780         lock_res_and_lock(lock);
 781
 782         ns = ldlm_lock_to_ns(lock);
 783
 784         ldlm_lock_decref_internal_nolock(lock, mode);
 785
 786         if (ldlm_is_local(lock) &&
 787             !lock->l_readers && !lock->l_writers) {
 788                 /* If this is a local lock on a server namespace and this was
 789                  * the last reference, cancel the lock.
 790                  */
 791                 CDEBUG(D_INFO, "forcing cancel of local lock\n");
 792                 ldlm_set_cbpending(lock);
 793         }
 794
 795         if (!lock->l_readers && !lock->l_writers &&
 796             ldlm_is_cbpending(lock)) {
 797                 /* If we received a blocked AST and this was the last reference,
 798                  * run the callback.
 799                  */
 800
 801                 LDLM_DEBUG(lock, "final decref done on cbpending lock");
 802
 803                 LDLM_LOCK_GET(lock); /* dropped by bl thread */
 804                 ldlm_lock_remove_from_lru(lock);
 805                 unlock_res_and_lock(lock);
 806
 807                 if (ldlm_is_fail_loc(lock))
 808                         OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
 809
 810                 if (ldlm_is_atomic_cb(lock) ||
 811                     ldlm_bl_to_thread_lock(ns, NULL, lock) != 0)
 812                         ldlm_handle_bl_callback(ns, NULL, lock);
 813         } else if (!lock->l_readers && !lock->l_writers &&
 814                    !ldlm_is_no_lru(lock) && !ldlm_is_bl_ast(lock)) {
 815                 LDLM_DEBUG(lock, "add lock into lru list");
 816
 817                 /* If this is a client-side namespace and this was the last
 818                  * reference, put it on the LRU.
 819                  */
 820                 ldlm_lock_add_to_lru(lock);
 821                 unlock_res_and_lock(lock);
 822
 823                 if (ldlm_is_fail_loc(lock))
 824                         OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
 825
 826                 /* Call ldlm_cancel_lru() only if EARLY_CANCEL and LRU RESIZE
 827                  * are not supported by the server, otherwise, it is done on
 828                  * enqueue.
 829                  */
 830                 if (!exp_connect_cancelset(lock->l_conn_export) &&
 831                     !ns_connect_lru_resize(ns))
 832                         ldlm_cancel_lru(ns, 0, LCF_ASYNC, 0);
 833         } else {
 834                 LDLM_DEBUG(lock, "do not add lock into lru list");
 835                 unlock_res_and_lock(lock);
 836         }
 837 }
 838
 839 /**
 840  * Decrease reader/writer refcount for LDLM lock with handle \a lockh
 841  */
 842 void ldlm_lock_decref(const struct lustre_handle *lockh, __u32 mode)
 843 {
 844         struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0);
 845
 846         LASSERTF(lock, "Non-existing lock: %#llx\n", lockh->cookie);
 847         ldlm_lock_decref_internal(lock, mode);
 848         LDLM_LOCK_PUT(lock);
 849 }
 850 EXPORT_SYMBOL(ldlm_lock_decref);
 851
 852 /**
 853  * Decrease reader/writer refcount for LDLM lock with handle
 854  * \a lockh and mark it for subsequent cancellation once r/w refcount
 855  * drops to zero instead of putting into LRU.
 856  *
 857  * Typical usage is for GROUP locks which we cannot allow to be cached.
 858  */
 859 void ldlm_lock_decref_and_cancel(const struct lustre_handle *lockh, __u32 mode)
 860 {
 861         struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0);
 862
 863         LASSERT(lock);
 864
 865         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
 866         lock_res_and_lock(lock);
 867         ldlm_set_cbpending(lock);
 868         unlock_res_and_lock(lock);
 869         ldlm_lock_decref_internal(lock, mode);
 870         LDLM_LOCK_PUT(lock);
 871 }
 872 EXPORT_SYMBOL(ldlm_lock_decref_and_cancel);
 873
 874 struct sl_insert_point {
 875         struct list_head *res_link;
 876         struct list_head *mode_link;
 877         struct list_head *policy_link;
 878 };
 879
 880 /**
 881  * Finds a position to insert the new lock into granted lock list.
 882  *
 883  * Used for locks eligible for skiplist optimization.
 884  *
 885  * Parameters:
 886  *      queue [input]:  the granted list where search acts on;
 887  *      req [input]:    the lock whose position to be located;
 888  *      prev [output]:  positions within 3 lists to insert @req to
 889  * Return Value:
 890  *      filled @prev
 891  * NOTE: called by
 892  *  - ldlm_grant_lock_with_skiplist
 893  */
 894 static void search_granted_lock(struct list_head *queue,
 895                                 struct ldlm_lock *req,
 896                                 struct sl_insert_point *prev)
 897 {
 898         struct list_head *tmp;
 899         struct ldlm_lock *lock, *mode_end, *policy_end;
 900
 901         list_for_each(tmp, queue) {
 902                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 903
 904                 mode_end = list_prev_entry(lock, l_sl_mode);
 905
 906                 if (lock->l_req_mode != req->l_req_mode) {
 907                         /* jump to last lock of mode group */
 908                         tmp = &mode_end->l_res_link;
 909                         continue;
 910                 }
 911
 912                 /* suitable mode group is found */
 913                 if (lock->l_resource->lr_type == LDLM_PLAIN) {
 914                         /* insert point is last lock of the mode group */
 915                         prev->res_link = &mode_end->l_res_link;
 916                         prev->mode_link = &mode_end->l_sl_mode;
 917                         prev->policy_link = &req->l_sl_policy;
 918                         return;
 919                 }
 920
 921                 if (lock->l_resource->lr_type == LDLM_IBITS) {
 922                         for (;;) {
 923                                 policy_end =
 924                                         list_prev_entry(lock, l_sl_policy);
 925
 926                                 if (lock->l_policy_data.l_inodebits.bits ==
 927                                     req->l_policy_data.l_inodebits.bits) {
 928                                         /* insert point is last lock of
 929                                          * the policy group
 930                                          */
 931                                         prev->res_link =
 932                                                 &policy_end->l_res_link;
 933                                         prev->mode_link =
 934                                                 &policy_end->l_sl_mode;
 935                                         prev->policy_link =
 936                                                 &policy_end->l_sl_policy;
 937                                         return;
 938                                 }
 939
 940                                 if (policy_end == mode_end)
 941                                         /* done with mode group */
 942                                         break;
 943
 944                                 /* go to next policy group within mode group */
 945                                 tmp = policy_end->l_res_link.next;
 946                                 lock = list_entry(tmp, struct ldlm_lock,
 947                                                   l_res_link);
 948                         }  /* loop over policy groups within the mode group */
 949
 950                         /* insert point is last lock of the mode group,
 951                          * new policy group is started
 952                          */
 953                         prev->res_link = &mode_end->l_res_link;
 954                         prev->mode_link = &mode_end->l_sl_mode;
 955                         prev->policy_link = &req->l_sl_policy;
 956                         return;
 957                 }
 958
 959                 LDLM_ERROR(lock, "is not LDLM_PLAIN or LDLM_IBITS lock");
 960                 LBUG();
 961         }
 962
 963         /* insert point is last lock on the queue,
 964          * new mode group and new policy group are started
 965          */
 966         prev->res_link = queue->prev;
 967         prev->mode_link = &req->l_sl_mode;
 968         prev->policy_link = &req->l_sl_policy;
 969 }
 970
 971 /**
 972  * Add a lock into resource granted list after a position described by
 973  * \a prev.
 974  */
 975 static void ldlm_granted_list_add_lock(struct ldlm_lock *lock,
 976                                        struct sl_insert_point *prev)
 977 {
 978         struct ldlm_resource *res = lock->l_resource;
 979
 980         check_res_locked(res);
 981
 982         ldlm_resource_dump(D_INFO, res);
 983         LDLM_DEBUG(lock, "About to add lock:");
 984
 985         if (ldlm_is_destroyed(lock)) {
 986                 CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
 987                 return;
 988         }
 989
 990         LASSERT(list_empty(&lock->l_res_link));
 991         LASSERT(list_empty(&lock->l_sl_mode));
 992         LASSERT(list_empty(&lock->l_sl_policy));
 993
 994         /*
 995          * lock->link == prev->link means lock is first starting the group.
 996          * Don't re-add to itself to suppress kernel warnings.
 997          */
 998         if (&lock->l_res_link != prev->res_link)
 999                 list_add(&lock->l_res_link, prev->res_link);
1000         if (&lock->l_sl_mode != prev->mode_link)
1001                 list_add(&lock->l_sl_mode, prev->mode_link);
1002         if (&lock->l_sl_policy != prev->policy_link)
1003                 list_add(&lock->l_sl_policy, prev->policy_link);
1004 }
1005
1006 /**
1007  * Add a lock to granted list on a resource maintaining skiplist
1008  * correctness.
1009  */
1010 static void ldlm_grant_lock_with_skiplist(struct ldlm_lock *lock)
1011 {
1012         struct sl_insert_point prev;
1013
1014         LASSERT(lock->l_req_mode == lock->l_granted_mode);
1015
1016         search_granted_lock(&lock->l_resource->lr_granted, lock, &prev);
1017         ldlm_granted_list_add_lock(lock, &prev);
1018 }
1019
1020 /**
1021  * Perform lock granting bookkeeping.
1022  *
1023  * Includes putting the lock into granted list and updating lock mode.
1024  * NOTE: called by
1025  *  - ldlm_lock_enqueue
1026  *  - ldlm_reprocess_queue
1027  *  - ldlm_lock_convert
1028  *
1029  * must be called with lr_lock held
1030  */
1031 void ldlm_grant_lock(struct ldlm_lock *lock, struct list_head *work_list)
1032 {
1033         struct ldlm_resource *res = lock->l_resource;
1034
1035         check_res_locked(res);
1036
1037         lock->l_granted_mode = lock->l_req_mode;
1038
1039         if (work_list && lock->l_completion_ast)
1040                 ldlm_add_ast_work_item(lock, NULL, work_list);
1041
1042         if (res->lr_type == LDLM_PLAIN || res->lr_type == LDLM_IBITS)
1043                 ldlm_grant_lock_with_skiplist(lock);
1044         else if (res->lr_type == LDLM_EXTENT)
1045                 ldlm_extent_add_lock(res, lock);
1046         else if (res->lr_type == LDLM_FLOCK) {
1047                 /*
1048                  * We should not add locks to granted list in the following cases:
1049                  * - this is an UNLOCK but not a real lock;
1050                  * - this is a TEST lock;
1051                  * - this is a F_CANCELLK lock (async flock has req_mode == 0)
1052                  * - this is a deadlock (flock cannot be granted)
1053                  */
1054                 if (!lock->l_req_mode || lock->l_req_mode == LCK_NL ||
1055                     ldlm_is_test_lock(lock) || ldlm_is_flock_deadlock(lock))
1056                         return;
1057                 ldlm_resource_add_lock(res, &res->lr_granted, lock);
1058         } else
1059                 LBUG();
1060
1061         ldlm_pool_add(&ldlm_res_to_ns(res)->ns_pool, lock);
1062 }
1063
1064 /**
1065  * Search for a lock with given properties in a queue.
1066  *
1067  * \retval a referenced lock or NULL.  See the flag descriptions below, in the
1068  * comment above ldlm_lock_match
1069  */
1070 static struct ldlm_lock *search_queue(struct list_head *queue,
1071                                       enum ldlm_mode *mode,
1072                                       ldlm_policy_data_t *policy,
1073                                       struct ldlm_lock *old_lock,
1074                                       __u64 flags, int unref)
1075 {
1076         struct ldlm_lock *lock;
1077         struct list_head       *tmp;
1078
1079         list_for_each(tmp, queue) {
1080                 enum ldlm_mode match;
1081
1082                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
1083
1084                 if (lock == old_lock)
1085                         break;
1086
1087                 /* Check if this lock can be matched.
1088                  * Used by LU-2919(exclusive open) for open lease lock
1089                  */
1090                 if (ldlm_is_excl(lock))
1091                         continue;
1092
1093                 /* llite sometimes wants to match locks that will be
1094                  * canceled when their users drop, but we allow it to match
1095                  * if it passes in CBPENDING and the lock still has users.
1096                  * this is generally only going to be used by children
1097                  * whose parents already hold a lock so forward progress
1098                  * can still happen.
1099                  */
1100                 if (ldlm_is_cbpending(lock) && !(flags & LDLM_FL_CBPENDING))
1101                         continue;
1102                 if (!unref && ldlm_is_cbpending(lock) &&
1103                     lock->l_readers == 0 && lock->l_writers == 0)
1104                         continue;
1105
1106                 if (!(lock->l_req_mode & *mode))
1107                         continue;
1108                 match = lock->l_req_mode;
1109
1110                 if (lock->l_resource->lr_type == LDLM_EXTENT &&
1111                     (lock->l_policy_data.l_extent.start >
1112                      policy->l_extent.start ||
1113                      lock->l_policy_data.l_extent.end < policy->l_extent.end))
1114                         continue;
1115
1116                 if (unlikely(match == LCK_GROUP) &&
1117                     lock->l_resource->lr_type == LDLM_EXTENT &&
1118                     policy->l_extent.gid != LDLM_GID_ANY &&
1119                     lock->l_policy_data.l_extent.gid != policy->l_extent.gid)
1120                         continue;
1121
1122                 /* We match if we have existing lock with same or wider set
1123                  * of bits.
1124                  */
1125                 if (lock->l_resource->lr_type == LDLM_IBITS &&
1126                     ((lock->l_policy_data.l_inodebits.bits &
1127                       policy->l_inodebits.bits) !=
1128                       policy->l_inodebits.bits))
1129                         continue;
1130
1131                 if (!unref && LDLM_HAVE_MASK(lock, GONE))
1132                         continue;
1133
1134                 if ((flags & LDLM_FL_LOCAL_ONLY) && !ldlm_is_local(lock))
1135                         continue;
1136
1137                 if (flags & LDLM_FL_TEST_LOCK) {
1138                         LDLM_LOCK_GET(lock);
1139                         ldlm_lock_touch_in_lru(lock);
1140                 } else {
1141                         ldlm_lock_addref_internal_nolock(lock, match);
1142                 }
1143                 *mode = match;
1144                 return lock;
1145         }
1146
1147         return NULL;
1148 }
1149
1150 void ldlm_lock_fail_match_locked(struct ldlm_lock *lock)
1151 {
1152         if ((lock->l_flags & LDLM_FL_FAIL_NOTIFIED) == 0) {
1153                 lock->l_flags |= LDLM_FL_FAIL_NOTIFIED;
1154                 wake_up_all(&lock->l_waitq);
1155         }
1156 }
1157 EXPORT_SYMBOL(ldlm_lock_fail_match_locked);
1158
1159 /**
1160  * Mark lock as "matchable" by OST.
1161  *
1162  * Used to prevent certain races in LOV/OSC where the lock is granted, but LVB
1163  * is not yet valid.
1164  * Assumes LDLM lock is already locked.
1165  */
1166 void ldlm_lock_allow_match_locked(struct ldlm_lock *lock)
1167 {
1168         ldlm_set_lvb_ready(lock);
1169         wake_up_all(&lock->l_waitq);
1170 }
1171 EXPORT_SYMBOL(ldlm_lock_allow_match_locked);
1172
1173 /**
1174  * Mark lock as "matchable" by OST.
1175  * Locks the lock and then \see ldlm_lock_allow_match_locked
1176  */
1177 void ldlm_lock_allow_match(struct ldlm_lock *lock)
1178 {
1179         lock_res_and_lock(lock);
1180         ldlm_lock_allow_match_locked(lock);
1181         unlock_res_and_lock(lock);
1182 }
1183 EXPORT_SYMBOL(ldlm_lock_allow_match);
1184
1185 /**
1186  * Attempt to find a lock with specified properties.
1187  *
1188  * Typically returns a reference to matched lock unless LDLM_FL_TEST_LOCK is
1189  * set in \a flags
1190  *
1191  * Can be called in two ways:
1192  *
1193  * If 'ns' is NULL, then lockh describes an existing lock that we want to look
1194  * for a duplicate of.
1195  *
1196  * Otherwise, all of the fields must be filled in, to match against.
1197  *
1198  * If 'flags' contains LDLM_FL_LOCAL_ONLY, then only match local locks on the
1199  *     server (ie, connh is NULL)
1200  * If 'flags' contains LDLM_FL_BLOCK_GRANTED, then only locks on the granted
1201  *     list will be considered
1202  * If 'flags' contains LDLM_FL_CBPENDING, then locks that have been marked
1203  *     to be canceled can still be matched as long as they still have reader
1204  *     or writer referneces
1205  * If 'flags' contains LDLM_FL_TEST_LOCK, then don't actually reference a lock,
1206  *     just tell us if we would have matched.
1207  *
1208  * \retval 1 if it finds an already-existing lock that is compatible; in this
1209  * case, lockh is filled in with a addref()ed lock
1210  *
1211  * We also check security context, and if that fails we simply return 0 (to
1212  * keep caller code unchanged), the context failure will be discovered by
1213  * caller sometime later.
1214  */
1215 enum ldlm_mode ldlm_lock_match(struct ldlm_namespace *ns, __u64 flags,
1216                                const struct ldlm_res_id *res_id,
1217                                enum ldlm_type type,
1218                                ldlm_policy_data_t *policy,
1219                                enum ldlm_mode mode,
1220                                struct lustre_handle *lockh, int unref)
1221 {
1222         struct ldlm_resource *res;
1223         struct ldlm_lock *lock, *old_lock = NULL;
1224         int rc = 0;
1225
1226         if (!ns) {
1227                 old_lock = ldlm_handle2lock(lockh);
1228                 LASSERT(old_lock);
1229
1230                 ns = ldlm_lock_to_ns(old_lock);
1231                 res_id = &old_lock->l_resource->lr_name;
1232                 type = old_lock->l_resource->lr_type;
1233                 mode = old_lock->l_req_mode;
1234         }
1235
1236         res = ldlm_resource_get(ns, NULL, res_id, type, 0);
1237         if (IS_ERR(res)) {
1238                 LASSERT(!old_lock);
1239                 return 0;
1240         }
1241
1242         LDLM_RESOURCE_ADDREF(res);
1243         lock_res(res);
1244
1245         lock = search_queue(&res->lr_granted, &mode, policy, old_lock,
1246                             flags, unref);
1247         if (lock) {
1248                 rc = 1;
1249                 goto out;
1250         }
1251         if (flags & LDLM_FL_BLOCK_GRANTED) {
1252                 rc = 0;
1253                 goto out;
1254         }
1255         lock = search_queue(&res->lr_waiting, &mode, policy, old_lock,
1256                             flags, unref);
1257         if (lock) {
1258                 rc = 1;
1259                 goto out;
1260         }
1261
1262  out:
1263         unlock_res(res);
1264         LDLM_RESOURCE_DELREF(res);
1265         ldlm_resource_putref(res);
1266
1267         if (lock) {
1268                 ldlm_lock2handle(lock, lockh);
1269                 if ((flags & LDLM_FL_LVB_READY) && !ldlm_is_lvb_ready(lock)) {
1270                         __u64 wait_flags = LDLM_FL_LVB_READY |
1271                                 LDLM_FL_DESTROYED | LDLM_FL_FAIL_NOTIFIED;
1272                         struct l_wait_info lwi;
1273
1274                         if (lock->l_completion_ast) {
1275                                 int err = lock->l_completion_ast(lock,
1276                                                           LDLM_FL_WAIT_NOREPROC,
1277                                                                  NULL);
1278                                 if (err) {
1279                                         if (flags & LDLM_FL_TEST_LOCK)
1280                                                 LDLM_LOCK_RELEASE(lock);
1281                                         else
1282                                                 ldlm_lock_decref_internal(lock,
1283                                                                           mode);
1284                                         rc = 0;
1285                                         goto out2;
1286                                 }
1287                         }
1288
1289                         lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(obd_timeout),
1290                                                NULL, LWI_ON_SIGNAL_NOOP, NULL);
1291
1292                         /* XXX FIXME see comment on CAN_MATCH in lustre_dlm.h */
1293                         l_wait_event(lock->l_waitq,
1294                                      lock->l_flags & wait_flags,
1295                                      &lwi);
1296                         if (!ldlm_is_lvb_ready(lock)) {
1297                                 if (flags & LDLM_FL_TEST_LOCK)
1298                                         LDLM_LOCK_RELEASE(lock);
1299                                 else
1300                                         ldlm_lock_decref_internal(lock, mode);
1301                                 rc = 0;
1302                         }
1303                 }
1304         }
1305  out2:
1306         if (rc) {
1307                 LDLM_DEBUG(lock, "matched (%llu %llu)",
1308                            (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1309                                 res_id->name[2] : policy->l_extent.start,
1310                            (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1311                                 res_id->name[3] : policy->l_extent.end);
1312
1313                 /* check user's security context */
1314                 if (lock->l_conn_export &&
1315                     sptlrpc_import_check_ctx(
1316                                 class_exp2cliimp(lock->l_conn_export))) {
1317                         if (!(flags & LDLM_FL_TEST_LOCK))
1318                                 ldlm_lock_decref_internal(lock, mode);
1319                         rc = 0;
1320                 }
1321
1322                 if (flags & LDLM_FL_TEST_LOCK)
1323                         LDLM_LOCK_RELEASE(lock);
1324
1325         } else if (!(flags & LDLM_FL_TEST_LOCK)) {/*less verbose for test-only*/
1326                 LDLM_DEBUG_NOLOCK("not matched ns %p type %u mode %u res %llu/%llu (%llu %llu)",
1327                                   ns, type, mode, res_id->name[0],
1328                                   res_id->name[1],
1329                                   (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1330                                         res_id->name[2] : policy->l_extent.start,
1331                                   (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1332                                         res_id->name[3] : policy->l_extent.end);
1333         }
1334         if (old_lock)
1335                 LDLM_LOCK_PUT(old_lock);
1336
1337         return rc ? mode : 0;
1338 }
1339 EXPORT_SYMBOL(ldlm_lock_match);
1340
1341 enum ldlm_mode ldlm_revalidate_lock_handle(const struct lustre_handle *lockh,
1342                                            __u64 *bits)
1343 {
1344         struct ldlm_lock *lock;
1345         enum ldlm_mode mode = 0;
1346
1347         lock = ldlm_handle2lock(lockh);
1348         if (lock) {
1349                 lock_res_and_lock(lock);
1350                 if (LDLM_HAVE_MASK(lock, GONE))
1351                         goto out;
1352
1353                 if (ldlm_is_cbpending(lock) &&
1354                     lock->l_readers == 0 && lock->l_writers == 0)
1355                         goto out;
1356
1357                 if (bits)
1358                         *bits = lock->l_policy_data.l_inodebits.bits;
1359                 mode = lock->l_granted_mode;
1360                 ldlm_lock_addref_internal_nolock(lock, mode);
1361         }
1362
1363 out:
1364         if (lock) {
1365                 unlock_res_and_lock(lock);
1366                 LDLM_LOCK_PUT(lock);
1367         }
1368         return mode;
1369 }
1370 EXPORT_SYMBOL(ldlm_revalidate_lock_handle);
1371
1372 /** The caller must guarantee that the buffer is large enough. */
1373 int ldlm_fill_lvb(struct ldlm_lock *lock, struct req_capsule *pill,
1374                   enum req_location loc, void *data, int size)
1375 {
1376         void *lvb;
1377
1378         LASSERT(data);
1379         LASSERT(size >= 0);
1380
1381         switch (lock->l_lvb_type) {
1382         case LVB_T_OST:
1383                 if (size == sizeof(struct ost_lvb)) {
1384                         if (loc == RCL_CLIENT)
1385                                 lvb = req_capsule_client_swab_get(pill,
1386                                                                   &RMF_DLM_LVB,
1387                                                         lustre_swab_ost_lvb);
1388                         else
1389                                 lvb = req_capsule_server_swab_get(pill,
1390                                                                   &RMF_DLM_LVB,
1391                                                         lustre_swab_ost_lvb);
1392                         if (unlikely(!lvb)) {
1393                                 LDLM_ERROR(lock, "no LVB");
1394                                 return -EPROTO;
1395                         }
1396
1397                         memcpy(data, lvb, size);
1398                 } else if (size == sizeof(struct ost_lvb_v1)) {
1399                         struct ost_lvb *olvb = data;
1400
1401                         if (loc == RCL_CLIENT)
1402                                 lvb = req_capsule_client_swab_get(pill,
1403                                                                   &RMF_DLM_LVB,
1404                                                         lustre_swab_ost_lvb_v1);
1405                         else
1406                                 lvb = req_capsule_server_sized_swab_get(pill,
1407                                                 &RMF_DLM_LVB, size,
1408                                                 lustre_swab_ost_lvb_v1);
1409                         if (unlikely(!lvb)) {
1410                                 LDLM_ERROR(lock, "no LVB");
1411                                 return -EPROTO;
1412                         }
1413
1414                         memcpy(data, lvb, size);
1415                         olvb->lvb_mtime_ns = 0;
1416                         olvb->lvb_atime_ns = 0;
1417                         olvb->lvb_ctime_ns = 0;
1418                 } else {
1419                         LDLM_ERROR(lock, "Replied unexpected ost LVB size %d",
1420                                    size);
1421                         return -EINVAL;
1422                 }
1423                 break;
1424         case LVB_T_LQUOTA:
1425                 if (size == sizeof(struct lquota_lvb)) {
1426                         if (loc == RCL_CLIENT)
1427                                 lvb = req_capsule_client_swab_get(pill,
1428                                                                   &RMF_DLM_LVB,
1429                                                         lustre_swab_lquota_lvb);
1430                         else
1431                                 lvb = req_capsule_server_swab_get(pill,
1432                                                                   &RMF_DLM_LVB,
1433                                                         lustre_swab_lquota_lvb);
1434                         if (unlikely(!lvb)) {
1435                                 LDLM_ERROR(lock, "no LVB");
1436                                 return -EPROTO;
1437                         }
1438
1439                         memcpy(data, lvb, size);
1440                 } else {
1441                         LDLM_ERROR(lock,
1442                                    "Replied unexpected lquota LVB size %d",
1443                                    size);
1444                         return -EINVAL;
1445                 }
1446                 break;
1447         case LVB_T_LAYOUT:
1448                 if (size == 0)
1449                         break;
1450
1451                 if (loc == RCL_CLIENT)
1452                         lvb = req_capsule_client_get(pill, &RMF_DLM_LVB);
1453                 else
1454                         lvb = req_capsule_server_get(pill, &RMF_DLM_LVB);
1455                 if (unlikely(!lvb)) {
1456                         LDLM_ERROR(lock, "no LVB");
1457                         return -EPROTO;
1458                 }
1459
1460                 memcpy(data, lvb, size);
1461                 break;
1462         default:
1463                 LDLM_ERROR(lock, "Unknown LVB type: %d", lock->l_lvb_type);
1464                 dump_stack();
1465                 return -EINVAL;
1466         }
1467
1468         return 0;
1469 }
1470
1471 /**
1472  * Create and fill in new LDLM lock with specified properties.
1473  * Returns a referenced lock
1474  */
1475 struct ldlm_lock *ldlm_lock_create(struct ldlm_namespace *ns,
1476                                    const struct ldlm_res_id *res_id,
1477                                    enum ldlm_type type,
1478                                    enum ldlm_mode mode,
1479                                    const struct ldlm_callback_suite *cbs,
1480                                    void *data, __u32 lvb_len,
1481                                    enum lvb_type lvb_type)
1482 {
1483         struct ldlm_lock *lock;
1484         struct ldlm_resource *res;
1485         int rc;
1486
1487         res = ldlm_resource_get(ns, NULL, res_id, type, 1);
1488         if (IS_ERR(res))
1489                 return ERR_CAST(res);
1490
1491         lock = ldlm_lock_new(res);
1492         if (!lock) {
1493                 ldlm_resource_putref(res);
1494                 return ERR_PTR(-ENOMEM);
1495         }
1496
1497         lock->l_req_mode = mode;
1498         lock->l_ast_data = data;
1499         lock->l_pid = current_pid();
1500         if (cbs) {
1501                 lock->l_blocking_ast = cbs->lcs_blocking;
1502                 lock->l_completion_ast = cbs->lcs_completion;
1503                 lock->l_glimpse_ast = cbs->lcs_glimpse;
1504         }
1505
1506         lock->l_tree_node = NULL;
1507         /* if this is the extent lock, allocate the interval tree node */
1508         if (type == LDLM_EXTENT) {
1509                 if (!ldlm_interval_alloc(lock)) {
1510                         rc = -ENOMEM;
1511                         goto out;
1512                 }
1513         }
1514
1515         if (lvb_len) {
1516                 lock->l_lvb_len = lvb_len;
1517                 lock->l_lvb_data = kzalloc(lvb_len, GFP_NOFS);
1518                 if (!lock->l_lvb_data) {
1519                         rc = -ENOMEM;
1520                         goto out;
1521                 }
1522         }
1523
1524         lock->l_lvb_type = lvb_type;
1525         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_NEW_LOCK)) {
1526                 rc = -ENOENT;
1527                 goto out;
1528         }
1529
1530         return lock;
1531
1532 out:
1533         ldlm_lock_destroy(lock);
1534         LDLM_LOCK_RELEASE(lock);
1535         return ERR_PTR(rc);
1536 }
1537
1538
1539
1540 /**
1541  * Enqueue (request) a lock.
1542  * On the client this is called from ldlm_cli_enqueue_fini
1543  * after we already got an initial reply from the server with some status.
1544  *
1545  * Does not block. As a result of enqueue the lock would be put
1546  * into granted or waiting list.
1547  */
1548 enum ldlm_error ldlm_lock_enqueue(struct ldlm_namespace *ns,
1549                                   struct ldlm_lock **lockp,
1550                                   void *cookie, __u64 *flags)
1551 {
1552         struct ldlm_lock *lock = *lockp;
1553         struct ldlm_resource *res = lock->l_resource;
1554
1555         lock_res_and_lock(lock);
1556         if (lock->l_req_mode == lock->l_granted_mode) {
1557                 /* The server returned a blocked lock, but it was granted
1558                  * before we got a chance to actually enqueue it.  We don't
1559                  * need to do anything else.
1560                  */
1561                 *flags &= ~LDLM_FL_BLOCKED_MASK;
1562                 goto out;
1563         }
1564
1565         ldlm_resource_unlink_lock(lock);
1566
1567         /* Cannot happen unless on the server */
1568         if (res->lr_type == LDLM_EXTENT && !lock->l_tree_node)
1569                 LBUG();
1570
1571         /* Some flags from the enqueue want to make it into the AST, via the
1572          * lock's l_flags.
1573          */
1574         if (*flags & LDLM_FL_AST_DISCARD_DATA)
1575                 ldlm_set_ast_discard_data(lock);
1576         if (*flags & LDLM_FL_TEST_LOCK)
1577                 ldlm_set_test_lock(lock);
1578
1579         /*
1580          * This distinction between local lock trees is very important; a client
1581          * namespace only has information about locks taken by that client, and
1582          * thus doesn't have enough information to decide for itself if it can
1583          * be granted (below).  In this case, we do exactly what the server
1584          * tells us to do, as dictated by the 'flags'.
1585          */
1586         if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED))
1587                 ldlm_resource_add_lock(res, &res->lr_waiting, lock);
1588         else
1589                 ldlm_grant_lock(lock, NULL);
1590
1591 out:
1592         unlock_res_and_lock(lock);
1593         return ELDLM_OK;
1594 }
1595
1596 /**
1597  * Process a call to blocking AST callback for a lock in ast_work list
1598  */
1599 static int
1600 ldlm_work_bl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
1601 {
1602         struct ldlm_cb_set_arg *arg = opaq;
1603         struct ldlm_lock_desc   d;
1604         int                  rc;
1605         struct ldlm_lock       *lock;
1606
1607         if (list_empty(arg->list))
1608                 return -ENOENT;
1609
1610         lock = list_entry(arg->list->next, struct ldlm_lock, l_bl_ast);
1611
1612         /* nobody should touch l_bl_ast */
1613         lock_res_and_lock(lock);
1614         list_del_init(&lock->l_bl_ast);
1615
1616         LASSERT(ldlm_is_ast_sent(lock));
1617         LASSERT(lock->l_bl_ast_run == 0);
1618         LASSERT(lock->l_blocking_lock);
1619         lock->l_bl_ast_run++;
1620         unlock_res_and_lock(lock);
1621
1622         ldlm_lock2desc(lock->l_blocking_lock, &d);
1623
1624         rc = lock->l_blocking_ast(lock, &d, (void *)arg, LDLM_CB_BLOCKING);
1625         LDLM_LOCK_RELEASE(lock->l_blocking_lock);
1626         lock->l_blocking_lock = NULL;
1627         LDLM_LOCK_RELEASE(lock);
1628
1629         return rc;
1630 }
1631
1632 /**
1633  * Process a call to completion AST callback for a lock in ast_work list
1634  */
1635 static int
1636 ldlm_work_cp_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
1637 {
1638         struct ldlm_cb_set_arg  *arg = opaq;
1639         int                   rc = 0;
1640         struct ldlm_lock        *lock;
1641         ldlm_completion_callback completion_callback;
1642
1643         if (list_empty(arg->list))
1644                 return -ENOENT;
1645
1646         lock = list_entry(arg->list->next, struct ldlm_lock, l_cp_ast);
1647
1648         /* It's possible to receive a completion AST before we've set
1649          * the l_completion_ast pointer: either because the AST arrived
1650          * before the reply, or simply because there's a small race
1651          * window between receiving the reply and finishing the local
1652          * enqueue. (bug 842)
1653          *
1654          * This can't happen with the blocking_ast, however, because we
1655          * will never call the local blocking_ast until we drop our
1656          * reader/writer reference, which we won't do until we get the
1657          * reply and finish enqueueing.
1658          */
1659
1660         /* nobody should touch l_cp_ast */
1661         lock_res_and_lock(lock);
1662         list_del_init(&lock->l_cp_ast);
1663         LASSERT(ldlm_is_cp_reqd(lock));
1664         /* save l_completion_ast since it can be changed by
1665          * mds_intent_policy(), see bug 14225
1666          */
1667         completion_callback = lock->l_completion_ast;
1668         ldlm_clear_cp_reqd(lock);
1669         unlock_res_and_lock(lock);
1670
1671         if (completion_callback)
1672                 rc = completion_callback(lock, 0, (void *)arg);
1673         LDLM_LOCK_RELEASE(lock);
1674
1675         return rc;
1676 }
1677
1678 /**
1679  * Process a call to revocation AST callback for a lock in ast_work list
1680  */
1681 static int
1682 ldlm_work_revoke_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
1683 {
1684         struct ldlm_cb_set_arg *arg = opaq;
1685         struct ldlm_lock_desc   desc;
1686         int                  rc;
1687         struct ldlm_lock       *lock;
1688
1689         if (list_empty(arg->list))
1690                 return -ENOENT;
1691
1692         lock = list_entry(arg->list->next, struct ldlm_lock, l_rk_ast);
1693         list_del_init(&lock->l_rk_ast);
1694
1695         /* the desc just pretend to exclusive */
1696         ldlm_lock2desc(lock, &desc);
1697         desc.l_req_mode = LCK_EX;
1698         desc.l_granted_mode = 0;
1699
1700         rc = lock->l_blocking_ast(lock, &desc, (void *)arg, LDLM_CB_BLOCKING);
1701         LDLM_LOCK_RELEASE(lock);
1702
1703         return rc;
1704 }
1705
1706 /**
1707  * Process a call to glimpse AST callback for a lock in ast_work list
1708  */
1709 static int ldlm_work_gl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
1710 {
1711         struct ldlm_cb_set_arg          *arg = opaq;
1712         struct ldlm_glimpse_work        *gl_work;
1713         struct ldlm_lock                *lock;
1714         int                              rc = 0;
1715
1716         if (list_empty(arg->list))
1717                 return -ENOENT;
1718
1719         gl_work = list_entry(arg->list->next, struct ldlm_glimpse_work,
1720                              gl_list);
1721         list_del_init(&gl_work->gl_list);
1722
1723         lock = gl_work->gl_lock;
1724
1725         /* transfer the glimpse descriptor to ldlm_cb_set_arg */
1726         arg->gl_desc = gl_work->gl_desc;
1727
1728         /* invoke the actual glimpse callback */
1729         if (lock->l_glimpse_ast(lock, (void *)arg) == 0)
1730                 rc = 1;
1731
1732         LDLM_LOCK_RELEASE(lock);
1733
1734         if ((gl_work->gl_flags & LDLM_GL_WORK_NOFREE) == 0)
1735                 kfree(gl_work);
1736
1737         return rc;
1738 }
1739
1740 /**
1741  * Process list of locks in need of ASTs being sent.
1742  *
1743  * Used on server to send multiple ASTs together instead of sending one by
1744  * one.
1745  */
1746 int ldlm_run_ast_work(struct ldlm_namespace *ns, struct list_head *rpc_list,
1747                       enum ldlm_desc_ast_t ast_type)
1748 {
1749         struct ldlm_cb_set_arg *arg;
1750         set_producer_func       work_ast_lock;
1751         int                  rc;
1752
1753         if (list_empty(rpc_list))
1754                 return 0;
1755
1756         arg = kzalloc(sizeof(*arg), GFP_NOFS);
1757         if (!arg)
1758                 return -ENOMEM;
1759
1760         atomic_set(&arg->restart, 0);
1761         arg->list = rpc_list;
1762
1763         switch (ast_type) {
1764         case LDLM_WORK_BL_AST:
1765                 arg->type = LDLM_BL_CALLBACK;
1766                 work_ast_lock = ldlm_work_bl_ast_lock;
1767                 break;
1768         case LDLM_WORK_CP_AST:
1769                 arg->type = LDLM_CP_CALLBACK;
1770                 work_ast_lock = ldlm_work_cp_ast_lock;
1771                 break;
1772         case LDLM_WORK_REVOKE_AST:
1773                 arg->type = LDLM_BL_CALLBACK;
1774                 work_ast_lock = ldlm_work_revoke_ast_lock;
1775                 break;
1776         case LDLM_WORK_GL_AST:
1777                 arg->type = LDLM_GL_CALLBACK;
1778                 work_ast_lock = ldlm_work_gl_ast_lock;
1779                 break;
1780         default:
1781                 LBUG();
1782         }
1783
1784         /* We create a ptlrpc request set with flow control extension.
1785          * This request set will use the work_ast_lock function to produce new
1786          * requests and will send a new request each time one completes in order
1787          * to keep the number of requests in flight to ns_max_parallel_ast
1788          */
1789         arg->set = ptlrpc_prep_fcset(ns->ns_max_parallel_ast ? : UINT_MAX,
1790                                      work_ast_lock, arg);
1791         if (!arg->set) {
1792                 rc = -ENOMEM;
1793                 goto out;
1794         }
1795
1796         ptlrpc_set_wait(arg->set);
1797         ptlrpc_set_destroy(arg->set);
1798
1799         rc = atomic_read(&arg->restart) ? -ERESTART : 0;
1800         goto out;
1801 out:
1802         kfree(arg);
1803         return rc;
1804 }
1805
1806 /**
1807  * Helper function to call blocking AST for LDLM lock \a lock in a
1808  * "cancelling" mode.
1809  */
1810 void ldlm_cancel_callback(struct ldlm_lock *lock)
1811 {
1812         check_res_locked(lock->l_resource);
1813         if (!ldlm_is_cancel(lock)) {
1814                 ldlm_set_cancel(lock);
1815                 if (lock->l_blocking_ast) {
1816                         unlock_res_and_lock(lock);
1817                         lock->l_blocking_ast(lock, NULL, lock->l_ast_data,
1818                                              LDLM_CB_CANCELING);
1819                         lock_res_and_lock(lock);
1820                 } else {
1821                         LDLM_DEBUG(lock, "no blocking ast");
1822                 }
1823         }
1824         ldlm_set_bl_done(lock);
1825 }
1826
1827 /**
1828  * Remove skiplist-enabled LDLM lock \a req from granted list
1829  */
1830 void ldlm_unlink_lock_skiplist(struct ldlm_lock *req)
1831 {
1832         if (req->l_resource->lr_type != LDLM_PLAIN &&
1833             req->l_resource->lr_type != LDLM_IBITS)
1834                 return;
1835
1836         list_del_init(&req->l_sl_policy);
1837         list_del_init(&req->l_sl_mode);
1838 }
1839
1840 /**
1841  * Attempts to cancel LDLM lock \a lock that has no reader/writer references.
1842  */
1843 void ldlm_lock_cancel(struct ldlm_lock *lock)
1844 {
1845         struct ldlm_resource *res;
1846         struct ldlm_namespace *ns;
1847
1848         lock_res_and_lock(lock);
1849
1850         res = lock->l_resource;
1851         ns  = ldlm_res_to_ns(res);
1852
1853         /* Please do not, no matter how tempting, remove this LBUG without
1854          * talking to me first. -phik
1855          */
1856         if (lock->l_readers || lock->l_writers) {
1857                 LDLM_ERROR(lock, "lock still has references");
1858                 LBUG();
1859         }
1860
1861         /* Releases cancel callback. */
1862         ldlm_cancel_callback(lock);
1863
1864         ldlm_resource_unlink_lock(lock);
1865         ldlm_lock_destroy_nolock(lock);
1866
1867         if (lock->l_granted_mode == lock->l_req_mode)
1868                 ldlm_pool_del(&ns->ns_pool, lock);
1869
1870         /* Make sure we will not be called again for same lock what is possible
1871          * if not to zero out lock->l_granted_mode
1872          */
1873         lock->l_granted_mode = LCK_MINMODE;
1874         unlock_res_and_lock(lock);
1875 }
1876 EXPORT_SYMBOL(ldlm_lock_cancel);
1877
1878 /**
1879  * Set opaque data into the lock that only makes sense to upper layer.
1880  */
1881 int ldlm_lock_set_data(const struct lustre_handle *lockh, void *data)
1882 {
1883         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
1884         int rc = -EINVAL;
1885
1886         if (lock) {
1887                 if (!lock->l_ast_data)
1888                         lock->l_ast_data = data;
1889                 if (lock->l_ast_data == data)
1890                         rc = 0;
1891                 LDLM_LOCK_PUT(lock);
1892         }
1893         return rc;
1894 }
1895 EXPORT_SYMBOL(ldlm_lock_set_data);
1896
1897 struct export_cl_data {
1898         struct obd_export       *ecl_exp;
1899         int                     ecl_loop;
1900 };
1901
1902 /**
1903  * Print lock with lock handle \a lockh description into debug log.
1904  *
1905  * Used when printing all locks on a resource for debug purposes.
1906  */
1907 void ldlm_lock_dump_handle(int level, const struct lustre_handle *lockh)
1908 {
1909         struct ldlm_lock *lock;
1910
1911         if (!((libcfs_debug | D_ERROR) & level))
1912                 return;
1913
1914         lock = ldlm_handle2lock(lockh);
1915         if (!lock)
1916                 return;
1917
1918         LDLM_DEBUG_LIMIT(level, lock, "###");
1919
1920         LDLM_LOCK_PUT(lock);
1921 }
1922 EXPORT_SYMBOL(ldlm_lock_dump_handle);
1923
1924 /**
1925  * Print lock information with custom message into debug log.
1926  * Helper function.
1927  */
1928 void _ldlm_lock_debug(struct ldlm_lock *lock,
1929                       struct libcfs_debug_msg_data *msgdata,
1930                       const char *fmt, ...)
1931 {
1932         va_list args;
1933         struct obd_export *exp = lock->l_export;
1934         struct ldlm_resource *resource = lock->l_resource;
1935         char *nid = "local";
1936
1937         va_start(args, fmt);
1938
1939         if (exp && exp->exp_connection) {
1940                 nid = libcfs_nid2str(exp->exp_connection->c_peer.nid);
1941         } else if (exp && exp->exp_obd) {
1942                 struct obd_import *imp = exp->exp_obd->u.cli.cl_import;
1943
1944                 nid = libcfs_nid2str(imp->imp_connection->c_peer.nid);
1945         }
1946
1947         if (!resource) {
1948                 libcfs_debug_vmsg2(msgdata, fmt, args,
1949                                    " ns: \?\? lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: \?\? rrc=\?\? type: \?\?\? flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lu lvb_type: %d\n",
1950                                    lock,
1951                                    lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
1952                                    lock->l_readers, lock->l_writers,
1953                                    ldlm_lockname[lock->l_granted_mode],
1954                                    ldlm_lockname[lock->l_req_mode],
1955                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
1956                                    exp ? atomic_read(&exp->exp_refcount) : -99,
1957                                    lock->l_pid, lock->l_callback_timeout, lock->l_lvb_type);
1958                 va_end(args);
1959                 return;
1960         }
1961
1962         switch (resource->lr_type) {
1963         case LDLM_EXTENT:
1964                 libcfs_debug_vmsg2(msgdata, fmt, args,
1965                                    " ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: " DLDLMRES " rrc: %d type: %s [%llu->%llu] (req %llu->%llu) flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lu lvb_type: %d\n",
1966                                    ldlm_lock_to_ns_name(lock), lock,
1967                                    lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
1968                                    lock->l_readers, lock->l_writers,
1969                                    ldlm_lockname[lock->l_granted_mode],
1970                                    ldlm_lockname[lock->l_req_mode],
1971                                    PLDLMRES(resource),
1972                                    atomic_read(&resource->lr_refcount),
1973                                    ldlm_typename[resource->lr_type],
1974                                    lock->l_policy_data.l_extent.start,
1975                                    lock->l_policy_data.l_extent.end,
1976                                    lock->l_req_extent.start, lock->l_req_extent.end,
1977                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
1978                                    exp ? atomic_read(&exp->exp_refcount) : -99,
1979                                    lock->l_pid, lock->l_callback_timeout,
1980                                    lock->l_lvb_type);
1981                 break;
1982
1983         case LDLM_FLOCK:
1984                 libcfs_debug_vmsg2(msgdata, fmt, args,
1985                                    " ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: " DLDLMRES " rrc: %d type: %s pid: %d [%llu->%llu] flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lu\n",
1986                                    ldlm_lock_to_ns_name(lock), lock,
1987                                    lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
1988                                    lock->l_readers, lock->l_writers,
1989                                    ldlm_lockname[lock->l_granted_mode],
1990                                    ldlm_lockname[lock->l_req_mode],
1991                                    PLDLMRES(resource),
1992                                    atomic_read(&resource->lr_refcount),
1993                                    ldlm_typename[resource->lr_type],
1994                                    lock->l_policy_data.l_flock.pid,
1995                                    lock->l_policy_data.l_flock.start,
1996                                    lock->l_policy_data.l_flock.end,
1997                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
1998                                    exp ? atomic_read(&exp->exp_refcount) : -99,
1999                                    lock->l_pid, lock->l_callback_timeout);
2000                 break;
2001
2002         case LDLM_IBITS:
2003                 libcfs_debug_vmsg2(msgdata, fmt, args,
2004                                    " ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: " DLDLMRES " bits %#llx rrc: %d type: %s flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lu lvb_type: %d\n",
2005                                    ldlm_lock_to_ns_name(lock),
2006                                    lock, lock->l_handle.h_cookie,
2007                                    atomic_read(&lock->l_refc),
2008                                    lock->l_readers, lock->l_writers,
2009                                    ldlm_lockname[lock->l_granted_mode],
2010                                    ldlm_lockname[lock->l_req_mode],
2011                                    PLDLMRES(resource),
2012                                    lock->l_policy_data.l_inodebits.bits,
2013                                    atomic_read(&resource->lr_refcount),
2014                                    ldlm_typename[resource->lr_type],
2015                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
2016                                    exp ? atomic_read(&exp->exp_refcount) : -99,
2017                                    lock->l_pid, lock->l_callback_timeout,
2018                                    lock->l_lvb_type);
2019                 break;
2020
2021         default:
2022                 libcfs_debug_vmsg2(msgdata, fmt, args,
2023                                    " ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: " DLDLMRES " rrc: %d type: %s flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lu lvb_type: %d\n",
2024                                    ldlm_lock_to_ns_name(lock),
2025                                    lock, lock->l_handle.h_cookie,
2026                                    atomic_read(&lock->l_refc),
2027                                    lock->l_readers, lock->l_writers,
2028                                    ldlm_lockname[lock->l_granted_mode],
2029                                    ldlm_lockname[lock->l_req_mode],
2030                                    PLDLMRES(resource),
2031                                    atomic_read(&resource->lr_refcount),
2032                                    ldlm_typename[resource->lr_type],
2033                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
2034                                    exp ? atomic_read(&exp->exp_refcount) : -99,
2035                                    lock->l_pid, lock->l_callback_timeout,
2036                                    lock->l_lvb_type);
2037                 break;
2038         }
2039         va_end(args);
2040 }
2041 EXPORT_SYMBOL(_ldlm_lock_debug);