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         /* It's unlikely but possible that someone marked the lock as
 550          * destroyed after we did handle2object on it
 551          */
 552         if (flags == 0 && !ldlm_is_destroyed(lock)) {
 553                 lu_ref_add(&lock->l_reference, "handle", current);
 554                 return lock;
 555         }
 556
 557         lock_res_and_lock(lock);
 558
 559         LASSERT(lock->l_resource);
 560
 561         lu_ref_add_atomic(&lock->l_reference, "handle", current);
 562         if (unlikely(ldlm_is_destroyed(lock))) {
 563                 unlock_res_and_lock(lock);
 564                 CDEBUG(D_INFO, "lock already destroyed: lock %p\n", lock);
 565                 LDLM_LOCK_PUT(lock);
 566                 return NULL;
 567         }
 568
 569         if (flags) {
 570                 if (lock->l_flags & flags) {
 571                         unlock_res_and_lock(lock);
 572                         LDLM_LOCK_PUT(lock);
 573                         return NULL;
 574                 }
 575
 576                 lock->l_flags |= flags;
 577         }
 578
 579         unlock_res_and_lock(lock);
 580         return lock;
 581 }
 582 EXPORT_SYMBOL(__ldlm_handle2lock);
 583 /** @} ldlm_handles */
 584
 585 /**
 586  * Fill in "on the wire" representation for given LDLM lock into supplied
 587  * lock descriptor \a desc structure.
 588  */
 589 void ldlm_lock2desc(struct ldlm_lock *lock, struct ldlm_lock_desc *desc)
 590 {
 591         ldlm_res2desc(lock->l_resource, &desc->l_resource);
 592         desc->l_req_mode = lock->l_req_mode;
 593         desc->l_granted_mode = lock->l_granted_mode;
 594         ldlm_convert_policy_to_wire(lock->l_resource->lr_type,
 595                                     &lock->l_policy_data,
 596                                     &desc->l_policy_data);
 597 }
 598 EXPORT_SYMBOL(ldlm_lock2desc);
 599
 600 /**
 601  * Add a lock to list of conflicting locks to send AST to.
 602  *
 603  * Only add if we have not sent a blocking AST to the lock yet.
 604  */
 605 static void ldlm_add_bl_work_item(struct ldlm_lock *lock, struct ldlm_lock *new,
 606                                   struct list_head *work_list)
 607 {
 608         if (!ldlm_is_ast_sent(lock)) {
 609                 LDLM_DEBUG(lock, "lock incompatible; sending blocking AST.");
 610                 ldlm_set_ast_sent(lock);
 611                 /* If the enqueuing client said so, tell the AST recipient to
 612                  * discard dirty data, rather than writing back.
 613                  */
 614                 if (ldlm_is_ast_discard_data(new))
 615                         ldlm_set_discard_data(lock);
 616                 LASSERT(list_empty(&lock->l_bl_ast));
 617                 list_add(&lock->l_bl_ast, work_list);
 618                 LDLM_LOCK_GET(lock);
 619                 LASSERT(!lock->l_blocking_lock);
 620                 lock->l_blocking_lock = LDLM_LOCK_GET(new);
 621         }
 622 }
 623
 624 /**
 625  * Add a lock to list of just granted locks to send completion AST to.
 626  */
 627 static void ldlm_add_cp_work_item(struct ldlm_lock *lock,
 628                                   struct list_head *work_list)
 629 {
 630         if (!ldlm_is_cp_reqd(lock)) {
 631                 ldlm_set_cp_reqd(lock);
 632                 LDLM_DEBUG(lock, "lock granted; sending completion AST.");
 633                 LASSERT(list_empty(&lock->l_cp_ast));
 634                 list_add(&lock->l_cp_ast, work_list);
 635                 LDLM_LOCK_GET(lock);
 636         }
 637 }
 638
 639 /**
 640  * Aggregator function to add AST work items into a list. Determines
 641  * what sort of an AST work needs to be done and calls the proper
 642  * adding function.
 643  * Must be called with lr_lock held.
 644  */
 645 static void ldlm_add_ast_work_item(struct ldlm_lock *lock,
 646                                    struct ldlm_lock *new,
 647                                    struct list_head *work_list)
 648 {
 649         check_res_locked(lock->l_resource);
 650         if (new)
 651                 ldlm_add_bl_work_item(lock, new, work_list);
 652         else
 653                 ldlm_add_cp_work_item(lock, work_list);
 654 }
 655
 656 /**
 657  * Add specified reader/writer reference to LDLM lock with handle \a lockh.
 658  * r/w reference type is determined by \a mode
 659  * Calls ldlm_lock_addref_internal.
 660  */
 661 void ldlm_lock_addref(const struct lustre_handle *lockh, __u32 mode)
 662 {
 663         struct ldlm_lock *lock;
 664
 665         lock = ldlm_handle2lock(lockh);
 666         LASSERTF(lock, "Non-existing lock: %llx\n", lockh->cookie);
 667         ldlm_lock_addref_internal(lock, mode);
 668         LDLM_LOCK_PUT(lock);
 669 }
 670 EXPORT_SYMBOL(ldlm_lock_addref);
 671
 672 /**
 673  * Helper function.
 674  * Add specified reader/writer reference to LDLM lock \a lock.
 675  * r/w reference type is determined by \a mode
 676  * Removes lock from LRU if it is there.
 677  * Assumes the LDLM lock is already locked.
 678  */
 679 void ldlm_lock_addref_internal_nolock(struct ldlm_lock *lock, __u32 mode)
 680 {
 681         ldlm_lock_remove_from_lru(lock);
 682         if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
 683                 lock->l_readers++;
 684                 lu_ref_add_atomic(&lock->l_reference, "reader", lock);
 685         }
 686         if (mode & (LCK_EX | LCK_CW | LCK_PW | LCK_GROUP | LCK_COS)) {
 687                 lock->l_writers++;
 688                 lu_ref_add_atomic(&lock->l_reference, "writer", lock);
 689         }
 690         LDLM_LOCK_GET(lock);
 691         lu_ref_add_atomic(&lock->l_reference, "user", lock);
 692         LDLM_DEBUG(lock, "ldlm_lock_addref(%s)", ldlm_lockname[mode]);
 693 }
 694
 695 /**
 696  * Attempts to add reader/writer reference to a lock with handle \a lockh, and
 697  * fails if lock is already LDLM_FL_CBPENDING or destroyed.
 698  *
 699  * \retval 0 success, lock was addref-ed
 700  *
 701  * \retval -EAGAIN lock is being canceled.
 702  */
 703 int ldlm_lock_addref_try(const struct lustre_handle *lockh, __u32 mode)
 704 {
 705         struct ldlm_lock *lock;
 706         int            result;
 707
 708         result = -EAGAIN;
 709         lock = ldlm_handle2lock(lockh);
 710         if (lock) {
 711                 lock_res_and_lock(lock);
 712                 if (lock->l_readers != 0 || lock->l_writers != 0 ||
 713                     !ldlm_is_cbpending(lock)) {
 714                         ldlm_lock_addref_internal_nolock(lock, mode);
 715                         result = 0;
 716                 }
 717                 unlock_res_and_lock(lock);
 718                 LDLM_LOCK_PUT(lock);
 719         }
 720         return result;
 721 }
 722 EXPORT_SYMBOL(ldlm_lock_addref_try);
 723
 724 /**
 725  * Add specified reader/writer reference to LDLM lock \a lock.
 726  * Locks LDLM lock and calls ldlm_lock_addref_internal_nolock to do the work.
 727  * Only called for local locks.
 728  */
 729 void ldlm_lock_addref_internal(struct ldlm_lock *lock, __u32 mode)
 730 {
 731         lock_res_and_lock(lock);
 732         ldlm_lock_addref_internal_nolock(lock, mode);
 733         unlock_res_and_lock(lock);
 734 }
 735
 736 /**
 737  * Removes reader/writer reference for LDLM lock \a lock.
 738  * Assumes LDLM lock is already locked.
 739  * only called in ldlm_flock_destroy and for local locks.
 740  * Does NOT add lock to LRU if no r/w references left to accommodate flock locks
 741  * that cannot be placed in LRU.
 742  */
 743 void ldlm_lock_decref_internal_nolock(struct ldlm_lock *lock, __u32 mode)
 744 {
 745         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
 746         if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
 747                 LASSERT(lock->l_readers > 0);
 748                 lu_ref_del(&lock->l_reference, "reader", lock);
 749                 lock->l_readers--;
 750         }
 751         if (mode & (LCK_EX | LCK_CW | LCK_PW | LCK_GROUP | LCK_COS)) {
 752                 LASSERT(lock->l_writers > 0);
 753                 lu_ref_del(&lock->l_reference, "writer", lock);
 754                 lock->l_writers--;
 755         }
 756
 757         lu_ref_del(&lock->l_reference, "user", lock);
 758         LDLM_LOCK_RELEASE(lock);    /* matches the LDLM_LOCK_GET() in addref */
 759 }
 760
 761 /**
 762  * Removes reader/writer reference for LDLM lock \a lock.
 763  * Locks LDLM lock first.
 764  * If the lock is determined to be client lock on a client and r/w refcount
 765  * drops to zero and the lock is not blocked, the lock is added to LRU lock
 766  * on the namespace.
 767  * For blocked LDLM locks if r/w count drops to zero, blocking_ast is called.
 768  */
 769 void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode)
 770 {
 771         struct ldlm_namespace *ns;
 772
 773         lock_res_and_lock(lock);
 774
 775         ns = ldlm_lock_to_ns(lock);
 776
 777         ldlm_lock_decref_internal_nolock(lock, mode);
 778
 779         if (ldlm_is_local(lock) &&
 780             !lock->l_readers && !lock->l_writers) {
 781                 /* If this is a local lock on a server namespace and this was
 782                  * the last reference, cancel the lock.
 783                  */
 784                 CDEBUG(D_INFO, "forcing cancel of local lock\n");
 785                 ldlm_set_cbpending(lock);
 786         }
 787
 788         if (!lock->l_readers && !lock->l_writers &&
 789             ldlm_is_cbpending(lock)) {
 790                 /* If we received a blocked AST and this was the last reference,
 791                  * run the callback.
 792                  */
 793
 794                 LDLM_DEBUG(lock, "final decref done on cbpending lock");
 795
 796                 LDLM_LOCK_GET(lock); /* dropped by bl thread */
 797                 ldlm_lock_remove_from_lru(lock);
 798                 unlock_res_and_lock(lock);
 799
 800                 if (ldlm_is_fail_loc(lock))
 801                         OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
 802
 803                 if (ldlm_is_atomic_cb(lock) ||
 804                     ldlm_bl_to_thread_lock(ns, NULL, lock) != 0)
 805                         ldlm_handle_bl_callback(ns, NULL, lock);
 806         } else if (!lock->l_readers && !lock->l_writers &&
 807                    !ldlm_is_no_lru(lock) && !ldlm_is_bl_ast(lock)) {
 808                 LDLM_DEBUG(lock, "add lock into lru list");
 809
 810                 /* If this is a client-side namespace and this was the last
 811                  * reference, put it on the LRU.
 812                  */
 813                 ldlm_lock_add_to_lru(lock);
 814                 unlock_res_and_lock(lock);
 815
 816                 if (ldlm_is_fail_loc(lock))
 817                         OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
 818
 819                 /* Call ldlm_cancel_lru() only if EARLY_CANCEL and LRU RESIZE
 820                  * are not supported by the server, otherwise, it is done on
 821                  * enqueue.
 822                  */
 823                 if (!exp_connect_cancelset(lock->l_conn_export) &&
 824                     !ns_connect_lru_resize(ns))
 825                         ldlm_cancel_lru(ns, 0, LCF_ASYNC, 0);
 826         } else {
 827                 LDLM_DEBUG(lock, "do not add lock into lru list");
 828                 unlock_res_and_lock(lock);
 829         }
 830 }
 831
 832 /**
 833  * Decrease reader/writer refcount for LDLM lock with handle \a lockh
 834  */
 835 void ldlm_lock_decref(const struct lustre_handle *lockh, __u32 mode)
 836 {
 837         struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0);
 838
 839         LASSERTF(lock, "Non-existing lock: %#llx\n", lockh->cookie);
 840         ldlm_lock_decref_internal(lock, mode);
 841         LDLM_LOCK_PUT(lock);
 842 }
 843 EXPORT_SYMBOL(ldlm_lock_decref);
 844
 845 /**
 846  * Decrease reader/writer refcount for LDLM lock with handle
 847  * \a lockh and mark it for subsequent cancellation once r/w refcount
 848  * drops to zero instead of putting into LRU.
 849  *
 850  * Typical usage is for GROUP locks which we cannot allow to be cached.
 851  */
 852 void ldlm_lock_decref_and_cancel(const struct lustre_handle *lockh, __u32 mode)
 853 {
 854         struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0);
 855
 856         LASSERT(lock);
 857
 858         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
 859         lock_res_and_lock(lock);
 860         ldlm_set_cbpending(lock);
 861         unlock_res_and_lock(lock);
 862         ldlm_lock_decref_internal(lock, mode);
 863         LDLM_LOCK_PUT(lock);
 864 }
 865 EXPORT_SYMBOL(ldlm_lock_decref_and_cancel);
 866
 867 struct sl_insert_point {
 868         struct list_head *res_link;
 869         struct list_head *mode_link;
 870         struct list_head *policy_link;
 871 };
 872
 873 /**
 874  * Finds a position to insert the new lock into granted lock list.
 875  *
 876  * Used for locks eligible for skiplist optimization.
 877  *
 878  * Parameters:
 879  *      queue [input]:  the granted list where search acts on;
 880  *      req [input]:    the lock whose position to be located;
 881  *      prev [output]:  positions within 3 lists to insert @req to
 882  * Return Value:
 883  *      filled @prev
 884  * NOTE: called by
 885  *  - ldlm_grant_lock_with_skiplist
 886  */
 887 static void search_granted_lock(struct list_head *queue,
 888                                 struct ldlm_lock *req,
 889                                 struct sl_insert_point *prev)
 890 {
 891         struct list_head *tmp;
 892         struct ldlm_lock *lock, *mode_end, *policy_end;
 893
 894         list_for_each(tmp, queue) {
 895                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 896
 897                 mode_end = list_prev_entry(lock, l_sl_mode);
 898
 899                 if (lock->l_req_mode != req->l_req_mode) {
 900                         /* jump to last lock of mode group */
 901                         tmp = &mode_end->l_res_link;
 902                         continue;
 903                 }
 904
 905                 /* suitable mode group is found */
 906                 if (lock->l_resource->lr_type == LDLM_PLAIN) {
 907                         /* insert point is last lock of the mode group */
 908                         prev->res_link = &mode_end->l_res_link;
 909                         prev->mode_link = &mode_end->l_sl_mode;
 910                         prev->policy_link = &req->l_sl_policy;
 911                         return;
 912                 }
 913
 914                 if (lock->l_resource->lr_type == LDLM_IBITS) {
 915                         for (;;) {
 916                                 policy_end =
 917                                         list_prev_entry(lock, l_sl_policy);
 918
 919                                 if (lock->l_policy_data.l_inodebits.bits ==
 920                                     req->l_policy_data.l_inodebits.bits) {
 921                                         /* insert point is last lock of
 922                                          * the policy group
 923                                          */
 924                                         prev->res_link =
 925                                                 &policy_end->l_res_link;
 926                                         prev->mode_link =
 927                                                 &policy_end->l_sl_mode;
 928                                         prev->policy_link =
 929                                                 &policy_end->l_sl_policy;
 930                                         return;
 931                                 }
 932
 933                                 if (policy_end == mode_end)
 934                                         /* done with mode group */
 935                                         break;
 936
 937                                 /* go to next policy group within mode group */
 938                                 tmp = policy_end->l_res_link.next;
 939                                 lock = list_entry(tmp, struct ldlm_lock,
 940                                                   l_res_link);
 941                         }  /* loop over policy groups within the mode group */
 942
 943                         /* insert point is last lock of the mode group,
 944                          * new policy group is started
 945                          */
 946                         prev->res_link = &mode_end->l_res_link;
 947                         prev->mode_link = &mode_end->l_sl_mode;
 948                         prev->policy_link = &req->l_sl_policy;
 949                         return;
 950                 }
 951
 952                 LDLM_ERROR(lock, "is not LDLM_PLAIN or LDLM_IBITS lock");
 953                 LBUG();
 954         }
 955
 956         /* insert point is last lock on the queue,
 957          * new mode group and new policy group are started
 958          */
 959         prev->res_link = queue->prev;
 960         prev->mode_link = &req->l_sl_mode;
 961         prev->policy_link = &req->l_sl_policy;
 962 }
 963
 964 /**
 965  * Add a lock into resource granted list after a position described by
 966  * \a prev.
 967  */
 968 static void ldlm_granted_list_add_lock(struct ldlm_lock *lock,
 969                                        struct sl_insert_point *prev)
 970 {
 971         struct ldlm_resource *res = lock->l_resource;
 972
 973         check_res_locked(res);
 974
 975         ldlm_resource_dump(D_INFO, res);
 976         LDLM_DEBUG(lock, "About to add lock:");
 977
 978         if (ldlm_is_destroyed(lock)) {
 979                 CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
 980                 return;
 981         }
 982
 983         LASSERT(list_empty(&lock->l_res_link));
 984         LASSERT(list_empty(&lock->l_sl_mode));
 985         LASSERT(list_empty(&lock->l_sl_policy));
 986
 987         /*
 988          * lock->link == prev->link means lock is first starting the group.
 989          * Don't re-add to itself to suppress kernel warnings.
 990          */
 991         if (&lock->l_res_link != prev->res_link)
 992                 list_add(&lock->l_res_link, prev->res_link);
 993         if (&lock->l_sl_mode != prev->mode_link)
 994                 list_add(&lock->l_sl_mode, prev->mode_link);
 995         if (&lock->l_sl_policy != prev->policy_link)
 996                 list_add(&lock->l_sl_policy, prev->policy_link);
 997 }
 998
 999 /**
1000  * Add a lock to granted list on a resource maintaining skiplist
1001  * correctness.
1002  */
1003 static void ldlm_grant_lock_with_skiplist(struct ldlm_lock *lock)
1004 {
1005         struct sl_insert_point prev;
1006
1007         LASSERT(lock->l_req_mode == lock->l_granted_mode);
1008
1009         search_granted_lock(&lock->l_resource->lr_granted, lock, &prev);
1010         ldlm_granted_list_add_lock(lock, &prev);
1011 }
1012
1013 /**
1014  * Perform lock granting bookkeeping.
1015  *
1016  * Includes putting the lock into granted list and updating lock mode.
1017  * NOTE: called by
1018  *  - ldlm_lock_enqueue
1019  *  - ldlm_reprocess_queue
1020  *  - ldlm_lock_convert
1021  *
1022  * must be called with lr_lock held
1023  */
1024 void ldlm_grant_lock(struct ldlm_lock *lock, struct list_head *work_list)
1025 {
1026         struct ldlm_resource *res = lock->l_resource;
1027
1028         check_res_locked(res);
1029
1030         lock->l_granted_mode = lock->l_req_mode;
1031
1032         if (work_list && lock->l_completion_ast)
1033                 ldlm_add_ast_work_item(lock, NULL, work_list);
1034
1035         if (res->lr_type == LDLM_PLAIN || res->lr_type == LDLM_IBITS)
1036                 ldlm_grant_lock_with_skiplist(lock);
1037         else if (res->lr_type == LDLM_EXTENT)
1038                 ldlm_extent_add_lock(res, lock);
1039         else if (res->lr_type == LDLM_FLOCK) {
1040                 /*
1041                  * We should not add locks to granted list in the following cases:
1042                  * - this is an UNLOCK but not a real lock;
1043                  * - this is a TEST lock;
1044                  * - this is a F_CANCELLK lock (async flock has req_mode == 0)
1045                  * - this is a deadlock (flock cannot be granted)
1046                  */
1047                 if (!lock->l_req_mode || lock->l_req_mode == LCK_NL ||
1048                     ldlm_is_test_lock(lock) || ldlm_is_flock_deadlock(lock))
1049                         return;
1050                 ldlm_resource_add_lock(res, &res->lr_granted, lock);
1051         } else
1052                 LBUG();
1053
1054         ldlm_pool_add(&ldlm_res_to_ns(res)->ns_pool, lock);
1055 }
1056
1057 /**
1058  * Search for a lock with given properties in a queue.
1059  *
1060  * \retval a referenced lock or NULL.  See the flag descriptions below, in the
1061  * comment above ldlm_lock_match
1062  */
1063 static struct ldlm_lock *search_queue(struct list_head *queue,
1064                                       enum ldlm_mode *mode,
1065                                       ldlm_policy_data_t *policy,
1066                                       struct ldlm_lock *old_lock,
1067                                       __u64 flags, int unref)
1068 {
1069         struct ldlm_lock *lock;
1070         struct list_head       *tmp;
1071
1072         list_for_each(tmp, queue) {
1073                 enum ldlm_mode match;
1074
1075                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
1076
1077                 if (lock == old_lock)
1078                         break;
1079
1080                 /* Check if this lock can be matched.
1081                  * Used by LU-2919(exclusive open) for open lease lock
1082                  */
1083                 if (ldlm_is_excl(lock))
1084                         continue;
1085
1086                 /* llite sometimes wants to match locks that will be
1087                  * canceled when their users drop, but we allow it to match
1088                  * if it passes in CBPENDING and the lock still has users.
1089                  * this is generally only going to be used by children
1090                  * whose parents already hold a lock so forward progress
1091                  * can still happen.
1092                  */
1093                 if (ldlm_is_cbpending(lock) && !(flags & LDLM_FL_CBPENDING))
1094                         continue;
1095                 if (!unref && ldlm_is_cbpending(lock) &&
1096                     lock->l_readers == 0 && lock->l_writers == 0)
1097                         continue;
1098
1099                 if (!(lock->l_req_mode & *mode))
1100                         continue;
1101                 match = lock->l_req_mode;
1102
1103                 if (lock->l_resource->lr_type == LDLM_EXTENT &&
1104                     (lock->l_policy_data.l_extent.start >
1105                      policy->l_extent.start ||
1106                      lock->l_policy_data.l_extent.end < policy->l_extent.end))
1107                         continue;
1108
1109                 if (unlikely(match == LCK_GROUP) &&
1110                     lock->l_resource->lr_type == LDLM_EXTENT &&
1111                     policy->l_extent.gid != LDLM_GID_ANY &&
1112                     lock->l_policy_data.l_extent.gid != policy->l_extent.gid)
1113                         continue;
1114
1115                 /* We match if we have existing lock with same or wider set
1116                  * of bits.
1117                  */
1118                 if (lock->l_resource->lr_type == LDLM_IBITS &&
1119                     ((lock->l_policy_data.l_inodebits.bits &
1120                       policy->l_inodebits.bits) !=
1121                       policy->l_inodebits.bits))
1122                         continue;
1123
1124                 if (!unref && LDLM_HAVE_MASK(lock, GONE))
1125                         continue;
1126
1127                 if ((flags & LDLM_FL_LOCAL_ONLY) && !ldlm_is_local(lock))
1128                         continue;
1129
1130                 if (flags & LDLM_FL_TEST_LOCK) {
1131                         LDLM_LOCK_GET(lock);
1132                         ldlm_lock_touch_in_lru(lock);
1133                 } else {
1134                         ldlm_lock_addref_internal_nolock(lock, match);
1135                 }
1136                 *mode = match;
1137                 return lock;
1138         }
1139
1140         return NULL;
1141 }
1142
1143 void ldlm_lock_fail_match_locked(struct ldlm_lock *lock)
1144 {
1145         if ((lock->l_flags & LDLM_FL_FAIL_NOTIFIED) == 0) {
1146                 lock->l_flags |= LDLM_FL_FAIL_NOTIFIED;
1147                 wake_up_all(&lock->l_waitq);
1148         }
1149 }
1150 EXPORT_SYMBOL(ldlm_lock_fail_match_locked);
1151
1152 /**
1153  * Mark lock as "matchable" by OST.
1154  *
1155  * Used to prevent certain races in LOV/OSC where the lock is granted, but LVB
1156  * is not yet valid.
1157  * Assumes LDLM lock is already locked.
1158  */
1159 void ldlm_lock_allow_match_locked(struct ldlm_lock *lock)
1160 {
1161         ldlm_set_lvb_ready(lock);
1162         wake_up_all(&lock->l_waitq);
1163 }
1164 EXPORT_SYMBOL(ldlm_lock_allow_match_locked);
1165
1166 /**
1167  * Mark lock as "matchable" by OST.
1168  * Locks the lock and then \see ldlm_lock_allow_match_locked
1169  */
1170 void ldlm_lock_allow_match(struct ldlm_lock *lock)
1171 {
1172         lock_res_and_lock(lock);
1173         ldlm_lock_allow_match_locked(lock);
1174         unlock_res_and_lock(lock);
1175 }
1176 EXPORT_SYMBOL(ldlm_lock_allow_match);
1177
1178 /**
1179  * Attempt to find a lock with specified properties.
1180  *
1181  * Typically returns a reference to matched lock unless LDLM_FL_TEST_LOCK is
1182  * set in \a flags
1183  *
1184  * Can be called in two ways:
1185  *
1186  * If 'ns' is NULL, then lockh describes an existing lock that we want to look
1187  * for a duplicate of.
1188  *
1189  * Otherwise, all of the fields must be filled in, to match against.
1190  *
1191  * If 'flags' contains LDLM_FL_LOCAL_ONLY, then only match local locks on the
1192  *     server (ie, connh is NULL)
1193  * If 'flags' contains LDLM_FL_BLOCK_GRANTED, then only locks on the granted
1194  *     list will be considered
1195  * If 'flags' contains LDLM_FL_CBPENDING, then locks that have been marked
1196  *     to be canceled can still be matched as long as they still have reader
1197  *     or writer referneces
1198  * If 'flags' contains LDLM_FL_TEST_LOCK, then don't actually reference a lock,
1199  *     just tell us if we would have matched.
1200  *
1201  * \retval 1 if it finds an already-existing lock that is compatible; in this
1202  * case, lockh is filled in with a addref()ed lock
1203  *
1204  * We also check security context, and if that fails we simply return 0 (to
1205  * keep caller code unchanged), the context failure will be discovered by
1206  * caller sometime later.
1207  */
1208 enum ldlm_mode ldlm_lock_match(struct ldlm_namespace *ns, __u64 flags,
1209                                const struct ldlm_res_id *res_id,
1210                                enum ldlm_type type,
1211                                ldlm_policy_data_t *policy,
1212                                enum ldlm_mode mode,
1213                                struct lustre_handle *lockh, int unref)
1214 {
1215         struct ldlm_resource *res;
1216         struct ldlm_lock *lock, *old_lock = NULL;
1217         int rc = 0;
1218
1219         if (!ns) {
1220                 old_lock = ldlm_handle2lock(lockh);
1221                 LASSERT(old_lock);
1222
1223                 ns = ldlm_lock_to_ns(old_lock);
1224                 res_id = &old_lock->l_resource->lr_name;
1225                 type = old_lock->l_resource->lr_type;
1226                 mode = old_lock->l_req_mode;
1227         }
1228
1229         res = ldlm_resource_get(ns, NULL, res_id, type, 0);
1230         if (IS_ERR(res)) {
1231                 LASSERT(!old_lock);
1232                 return 0;
1233         }
1234
1235         LDLM_RESOURCE_ADDREF(res);
1236         lock_res(res);
1237
1238         lock = search_queue(&res->lr_granted, &mode, policy, old_lock,
1239                             flags, unref);
1240         if (lock) {
1241                 rc = 1;
1242                 goto out;
1243         }
1244         if (flags & LDLM_FL_BLOCK_GRANTED) {
1245                 rc = 0;
1246                 goto out;
1247         }
1248         lock = search_queue(&res->lr_waiting, &mode, policy, old_lock,
1249                             flags, unref);
1250         if (lock) {
1251                 rc = 1;
1252                 goto out;
1253         }
1254
1255  out:
1256         unlock_res(res);
1257         LDLM_RESOURCE_DELREF(res);
1258         ldlm_resource_putref(res);
1259
1260         if (lock) {
1261                 ldlm_lock2handle(lock, lockh);
1262                 if ((flags & LDLM_FL_LVB_READY) && !ldlm_is_lvb_ready(lock)) {
1263                         __u64 wait_flags = LDLM_FL_LVB_READY |
1264                                 LDLM_FL_DESTROYED | LDLM_FL_FAIL_NOTIFIED;
1265                         struct l_wait_info lwi;
1266
1267                         if (lock->l_completion_ast) {
1268                                 int err = lock->l_completion_ast(lock,
1269                                                           LDLM_FL_WAIT_NOREPROC,
1270                                                                  NULL);
1271                                 if (err) {
1272                                         if (flags & LDLM_FL_TEST_LOCK)
1273                                                 LDLM_LOCK_RELEASE(lock);
1274                                         else
1275                                                 ldlm_lock_decref_internal(lock,
1276                                                                           mode);
1277                                         rc = 0;
1278                                         goto out2;
1279                                 }
1280                         }
1281
1282                         lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(obd_timeout),
1283                                                NULL, LWI_ON_SIGNAL_NOOP, NULL);
1284
1285                         /* XXX FIXME see comment on CAN_MATCH in lustre_dlm.h */
1286                         l_wait_event(lock->l_waitq,
1287                                      lock->l_flags & wait_flags,
1288                                      &lwi);
1289                         if (!ldlm_is_lvb_ready(lock)) {
1290                                 if (flags & LDLM_FL_TEST_LOCK)
1291                                         LDLM_LOCK_RELEASE(lock);
1292                                 else
1293                                         ldlm_lock_decref_internal(lock, mode);
1294                                 rc = 0;
1295                         }
1296                 }
1297         }
1298  out2:
1299         if (rc) {
1300                 LDLM_DEBUG(lock, "matched (%llu %llu)",
1301                            (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1302                                 res_id->name[2] : policy->l_extent.start,
1303                            (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1304                                 res_id->name[3] : policy->l_extent.end);
1305
1306                 /* check user's security context */
1307                 if (lock->l_conn_export &&
1308                     sptlrpc_import_check_ctx(
1309                                 class_exp2cliimp(lock->l_conn_export))) {
1310                         if (!(flags & LDLM_FL_TEST_LOCK))
1311                                 ldlm_lock_decref_internal(lock, mode);
1312                         rc = 0;
1313                 }
1314
1315                 if (flags & LDLM_FL_TEST_LOCK)
1316                         LDLM_LOCK_RELEASE(lock);
1317
1318         } else if (!(flags & LDLM_FL_TEST_LOCK)) {/*less verbose for test-only*/
1319                 LDLM_DEBUG_NOLOCK("not matched ns %p type %u mode %u res %llu/%llu (%llu %llu)",
1320                                   ns, type, mode, res_id->name[0],
1321                                   res_id->name[1],
1322                                   (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1323                                         res_id->name[2] : policy->l_extent.start,
1324                                   (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1325                                         res_id->name[3] : policy->l_extent.end);
1326         }
1327         if (old_lock)
1328                 LDLM_LOCK_PUT(old_lock);
1329
1330         return rc ? mode : 0;
1331 }
1332 EXPORT_SYMBOL(ldlm_lock_match);
1333
1334 enum ldlm_mode ldlm_revalidate_lock_handle(const struct lustre_handle *lockh,
1335                                            __u64 *bits)
1336 {
1337         struct ldlm_lock *lock;
1338         enum ldlm_mode mode = 0;
1339
1340         lock = ldlm_handle2lock(lockh);
1341         if (lock) {
1342                 lock_res_and_lock(lock);
1343                 if (LDLM_HAVE_MASK(lock, GONE))
1344                         goto out;
1345
1346                 if (ldlm_is_cbpending(lock) &&
1347                     lock->l_readers == 0 && lock->l_writers == 0)
1348                         goto out;
1349
1350                 if (bits)
1351                         *bits = lock->l_policy_data.l_inodebits.bits;
1352                 mode = lock->l_granted_mode;
1353                 ldlm_lock_addref_internal_nolock(lock, mode);
1354         }
1355
1356 out:
1357         if (lock) {
1358                 unlock_res_and_lock(lock);
1359                 LDLM_LOCK_PUT(lock);
1360         }
1361         return mode;
1362 }
1363 EXPORT_SYMBOL(ldlm_revalidate_lock_handle);
1364
1365 /** The caller must guarantee that the buffer is large enough. */
1366 int ldlm_fill_lvb(struct ldlm_lock *lock, struct req_capsule *pill,
1367                   enum req_location loc, void *data, int size)
1368 {
1369         void *lvb;
1370
1371         LASSERT(data);
1372         LASSERT(size >= 0);
1373
1374         switch (lock->l_lvb_type) {
1375         case LVB_T_OST:
1376                 if (size == sizeof(struct ost_lvb)) {
1377                         if (loc == RCL_CLIENT)
1378                                 lvb = req_capsule_client_swab_get(pill,
1379                                                                   &RMF_DLM_LVB,
1380                                                         lustre_swab_ost_lvb);
1381                         else
1382                                 lvb = req_capsule_server_swab_get(pill,
1383                                                                   &RMF_DLM_LVB,
1384                                                         lustre_swab_ost_lvb);
1385                         if (unlikely(!lvb)) {
1386                                 LDLM_ERROR(lock, "no LVB");
1387                                 return -EPROTO;
1388                         }
1389
1390                         memcpy(data, lvb, size);
1391                 } else if (size == sizeof(struct ost_lvb_v1)) {
1392                         struct ost_lvb *olvb = data;
1393
1394                         if (loc == RCL_CLIENT)
1395                                 lvb = req_capsule_client_swab_get(pill,
1396                                                                   &RMF_DLM_LVB,
1397                                                         lustre_swab_ost_lvb_v1);
1398                         else
1399                                 lvb = req_capsule_server_sized_swab_get(pill,
1400                                                 &RMF_DLM_LVB, size,
1401                                                 lustre_swab_ost_lvb_v1);
1402                         if (unlikely(!lvb)) {
1403                                 LDLM_ERROR(lock, "no LVB");
1404                                 return -EPROTO;
1405                         }
1406
1407                         memcpy(data, lvb, size);
1408                         olvb->lvb_mtime_ns = 0;
1409                         olvb->lvb_atime_ns = 0;
1410                         olvb->lvb_ctime_ns = 0;
1411                 } else {
1412                         LDLM_ERROR(lock, "Replied unexpected ost LVB size %d",
1413                                    size);
1414                         return -EINVAL;
1415                 }
1416                 break;
1417         case LVB_T_LQUOTA:
1418                 if (size == sizeof(struct lquota_lvb)) {
1419                         if (loc == RCL_CLIENT)
1420                                 lvb = req_capsule_client_swab_get(pill,
1421                                                                   &RMF_DLM_LVB,
1422                                                         lustre_swab_lquota_lvb);
1423                         else
1424                                 lvb = req_capsule_server_swab_get(pill,
1425                                                                   &RMF_DLM_LVB,
1426                                                         lustre_swab_lquota_lvb);
1427                         if (unlikely(!lvb)) {
1428                                 LDLM_ERROR(lock, "no LVB");
1429                                 return -EPROTO;
1430                         }
1431
1432                         memcpy(data, lvb, size);
1433                 } else {
1434                         LDLM_ERROR(lock,
1435                                    "Replied unexpected lquota LVB size %d",
1436                                    size);
1437                         return -EINVAL;
1438                 }
1439                 break;
1440         case LVB_T_LAYOUT:
1441                 if (size == 0)
1442                         break;
1443
1444                 if (loc == RCL_CLIENT)
1445                         lvb = req_capsule_client_get(pill, &RMF_DLM_LVB);
1446                 else
1447                         lvb = req_capsule_server_get(pill, &RMF_DLM_LVB);
1448                 if (unlikely(!lvb)) {
1449                         LDLM_ERROR(lock, "no LVB");
1450                         return -EPROTO;
1451                 }
1452
1453                 memcpy(data, lvb, size);
1454                 break;
1455         default:
1456                 LDLM_ERROR(lock, "Unknown LVB type: %d", lock->l_lvb_type);
1457                 dump_stack();
1458                 return -EINVAL;
1459         }
1460
1461         return 0;
1462 }
1463
1464 /**
1465  * Create and fill in new LDLM lock with specified properties.
1466  * Returns a referenced lock
1467  */
1468 struct ldlm_lock *ldlm_lock_create(struct ldlm_namespace *ns,
1469                                    const struct ldlm_res_id *res_id,
1470                                    enum ldlm_type type,
1471                                    enum ldlm_mode mode,
1472                                    const struct ldlm_callback_suite *cbs,
1473                                    void *data, __u32 lvb_len,
1474                                    enum lvb_type lvb_type)
1475 {
1476         struct ldlm_lock *lock;
1477         struct ldlm_resource *res;
1478         int rc;
1479
1480         res = ldlm_resource_get(ns, NULL, res_id, type, 1);
1481         if (IS_ERR(res))
1482                 return ERR_CAST(res);
1483
1484         lock = ldlm_lock_new(res);
1485         if (!lock)
1486                 return ERR_PTR(-ENOMEM);
1487
1488         lock->l_req_mode = mode;
1489         lock->l_ast_data = data;
1490         lock->l_pid = current_pid();
1491         if (cbs) {
1492                 lock->l_blocking_ast = cbs->lcs_blocking;
1493                 lock->l_completion_ast = cbs->lcs_completion;
1494                 lock->l_glimpse_ast = cbs->lcs_glimpse;
1495         }
1496
1497         lock->l_tree_node = NULL;
1498         /* if this is the extent lock, allocate the interval tree node */
1499         if (type == LDLM_EXTENT) {
1500                 if (!ldlm_interval_alloc(lock)) {
1501                         rc = -ENOMEM;
1502                         goto out;
1503                 }
1504         }
1505
1506         if (lvb_len) {
1507                 lock->l_lvb_len = lvb_len;
1508                 lock->l_lvb_data = kzalloc(lvb_len, GFP_NOFS);
1509                 if (!lock->l_lvb_data) {
1510                         rc = -ENOMEM;
1511                         goto out;
1512                 }
1513         }
1514
1515         lock->l_lvb_type = lvb_type;
1516         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_NEW_LOCK)) {
1517                 rc = -ENOENT;
1518                 goto out;
1519         }
1520
1521         return lock;
1522
1523 out:
1524         ldlm_lock_destroy(lock);
1525         LDLM_LOCK_RELEASE(lock);
1526         return ERR_PTR(rc);
1527 }
1528
1529 /**
1530  * Enqueue (request) a lock.
1531  * On the client this is called from ldlm_cli_enqueue_fini
1532  * after we already got an initial reply from the server with some status.
1533  *
1534  * Does not block. As a result of enqueue the lock would be put
1535  * into granted or waiting list.
1536  */
1537 enum ldlm_error ldlm_lock_enqueue(struct ldlm_namespace *ns,
1538                                   struct ldlm_lock **lockp,
1539                                   void *cookie, __u64 *flags)
1540 {
1541         struct ldlm_lock *lock = *lockp;
1542         struct ldlm_resource *res = lock->l_resource;
1543
1544         lock_res_and_lock(lock);
1545         if (lock->l_req_mode == lock->l_granted_mode) {
1546                 /* The server returned a blocked lock, but it was granted
1547                  * before we got a chance to actually enqueue it.  We don't
1548                  * need to do anything else.
1549                  */
1550                 *flags &= ~LDLM_FL_BLOCKED_MASK;
1551                 goto out;
1552         }
1553
1554         ldlm_resource_unlink_lock(lock);
1555
1556         /* Cannot happen unless on the server */
1557         if (res->lr_type == LDLM_EXTENT && !lock->l_tree_node)
1558                 LBUG();
1559
1560         /* Some flags from the enqueue want to make it into the AST, via the
1561          * lock's l_flags.
1562          */
1563         if (*flags & LDLM_FL_AST_DISCARD_DATA)
1564                 ldlm_set_ast_discard_data(lock);
1565         if (*flags & LDLM_FL_TEST_LOCK)
1566                 ldlm_set_test_lock(lock);
1567
1568         /*
1569          * This distinction between local lock trees is very important; a client
1570          * namespace only has information about locks taken by that client, and
1571          * thus doesn't have enough information to decide for itself if it can
1572          * be granted (below).  In this case, we do exactly what the server
1573          * tells us to do, as dictated by the 'flags'.
1574          */
1575         if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED))
1576                 ldlm_resource_add_lock(res, &res->lr_waiting, lock);
1577         else
1578                 ldlm_grant_lock(lock, NULL);
1579
1580 out:
1581         unlock_res_and_lock(lock);
1582         return ELDLM_OK;
1583 }
1584
1585 /**
1586  * Process a call to blocking AST callback for a lock in ast_work list
1587  */
1588 static int
1589 ldlm_work_bl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
1590 {
1591         struct ldlm_cb_set_arg *arg = opaq;
1592         struct ldlm_lock_desc   d;
1593         int                  rc;
1594         struct ldlm_lock       *lock;
1595
1596         if (list_empty(arg->list))
1597                 return -ENOENT;
1598
1599         lock = list_entry(arg->list->next, struct ldlm_lock, l_bl_ast);
1600
1601         /* nobody should touch l_bl_ast */
1602         lock_res_and_lock(lock);
1603         list_del_init(&lock->l_bl_ast);
1604
1605         LASSERT(ldlm_is_ast_sent(lock));
1606         LASSERT(lock->l_bl_ast_run == 0);
1607         LASSERT(lock->l_blocking_lock);
1608         lock->l_bl_ast_run++;
1609         unlock_res_and_lock(lock);
1610
1611         ldlm_lock2desc(lock->l_blocking_lock, &d);
1612
1613         rc = lock->l_blocking_ast(lock, &d, (void *)arg, LDLM_CB_BLOCKING);
1614         LDLM_LOCK_RELEASE(lock->l_blocking_lock);
1615         lock->l_blocking_lock = NULL;
1616         LDLM_LOCK_RELEASE(lock);
1617
1618         return rc;
1619 }
1620
1621 /**
1622  * Process a call to completion AST callback for a lock in ast_work list
1623  */
1624 static int
1625 ldlm_work_cp_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
1626 {
1627         struct ldlm_cb_set_arg  *arg = opaq;
1628         int                   rc = 0;
1629         struct ldlm_lock        *lock;
1630         ldlm_completion_callback completion_callback;
1631
1632         if (list_empty(arg->list))
1633                 return -ENOENT;
1634
1635         lock = list_entry(arg->list->next, struct ldlm_lock, l_cp_ast);
1636
1637         /* It's possible to receive a completion AST before we've set
1638          * the l_completion_ast pointer: either because the AST arrived
1639          * before the reply, or simply because there's a small race
1640          * window between receiving the reply and finishing the local
1641          * enqueue. (bug 842)
1642          *
1643          * This can't happen with the blocking_ast, however, because we
1644          * will never call the local blocking_ast until we drop our
1645          * reader/writer reference, which we won't do until we get the
1646          * reply and finish enqueueing.
1647          */
1648
1649         /* nobody should touch l_cp_ast */
1650         lock_res_and_lock(lock);
1651         list_del_init(&lock->l_cp_ast);
1652         LASSERT(ldlm_is_cp_reqd(lock));
1653         /* save l_completion_ast since it can be changed by
1654          * mds_intent_policy(), see bug 14225
1655          */
1656         completion_callback = lock->l_completion_ast;
1657         ldlm_clear_cp_reqd(lock);
1658         unlock_res_and_lock(lock);
1659
1660         if (completion_callback)
1661                 rc = completion_callback(lock, 0, (void *)arg);
1662         LDLM_LOCK_RELEASE(lock);
1663
1664         return rc;
1665 }
1666
1667 /**
1668  * Process a call to revocation AST callback for a lock in ast_work list
1669  */
1670 static int
1671 ldlm_work_revoke_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
1672 {
1673         struct ldlm_cb_set_arg *arg = opaq;
1674         struct ldlm_lock_desc   desc;
1675         int                  rc;
1676         struct ldlm_lock       *lock;
1677
1678         if (list_empty(arg->list))
1679                 return -ENOENT;
1680
1681         lock = list_entry(arg->list->next, struct ldlm_lock, l_rk_ast);
1682         list_del_init(&lock->l_rk_ast);
1683
1684         /* the desc just pretend to exclusive */
1685         ldlm_lock2desc(lock, &desc);
1686         desc.l_req_mode = LCK_EX;
1687         desc.l_granted_mode = 0;
1688
1689         rc = lock->l_blocking_ast(lock, &desc, (void *)arg, LDLM_CB_BLOCKING);
1690         LDLM_LOCK_RELEASE(lock);
1691
1692         return rc;
1693 }
1694
1695 /**
1696  * Process a call to glimpse AST callback for a lock in ast_work list
1697  */
1698 static int ldlm_work_gl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
1699 {
1700         struct ldlm_cb_set_arg          *arg = opaq;
1701         struct ldlm_glimpse_work        *gl_work;
1702         struct ldlm_lock                *lock;
1703         int                              rc = 0;
1704
1705         if (list_empty(arg->list))
1706                 return -ENOENT;
1707
1708         gl_work = list_entry(arg->list->next, struct ldlm_glimpse_work,
1709                              gl_list);
1710         list_del_init(&gl_work->gl_list);
1711
1712         lock = gl_work->gl_lock;
1713
1714         /* transfer the glimpse descriptor to ldlm_cb_set_arg */
1715         arg->gl_desc = gl_work->gl_desc;
1716
1717         /* invoke the actual glimpse callback */
1718         if (lock->l_glimpse_ast(lock, (void *)arg) == 0)
1719                 rc = 1;
1720
1721         LDLM_LOCK_RELEASE(lock);
1722
1723         if ((gl_work->gl_flags & LDLM_GL_WORK_NOFREE) == 0)
1724                 kfree(gl_work);
1725
1726         return rc;
1727 }
1728
1729 /**
1730  * Process list of locks in need of ASTs being sent.
1731  *
1732  * Used on server to send multiple ASTs together instead of sending one by
1733  * one.
1734  */
1735 int ldlm_run_ast_work(struct ldlm_namespace *ns, struct list_head *rpc_list,
1736                       enum ldlm_desc_ast_t ast_type)
1737 {
1738         struct ldlm_cb_set_arg *arg;
1739         set_producer_func       work_ast_lock;
1740         int                  rc;
1741
1742         if (list_empty(rpc_list))
1743                 return 0;
1744
1745         arg = kzalloc(sizeof(*arg), GFP_NOFS);
1746         if (!arg)
1747                 return -ENOMEM;
1748
1749         atomic_set(&arg->restart, 0);
1750         arg->list = rpc_list;
1751
1752         switch (ast_type) {
1753         case LDLM_WORK_BL_AST:
1754                 arg->type = LDLM_BL_CALLBACK;
1755                 work_ast_lock = ldlm_work_bl_ast_lock;
1756                 break;
1757         case LDLM_WORK_CP_AST:
1758                 arg->type = LDLM_CP_CALLBACK;
1759                 work_ast_lock = ldlm_work_cp_ast_lock;
1760                 break;
1761         case LDLM_WORK_REVOKE_AST:
1762                 arg->type = LDLM_BL_CALLBACK;
1763                 work_ast_lock = ldlm_work_revoke_ast_lock;
1764                 break;
1765         case LDLM_WORK_GL_AST:
1766                 arg->type = LDLM_GL_CALLBACK;
1767                 work_ast_lock = ldlm_work_gl_ast_lock;
1768                 break;
1769         default:
1770                 LBUG();
1771         }
1772
1773         /* We create a ptlrpc request set with flow control extension.
1774          * This request set will use the work_ast_lock function to produce new
1775          * requests and will send a new request each time one completes in order
1776          * to keep the number of requests in flight to ns_max_parallel_ast
1777          */
1778         arg->set = ptlrpc_prep_fcset(ns->ns_max_parallel_ast ? : UINT_MAX,
1779                                      work_ast_lock, arg);
1780         if (!arg->set) {
1781                 rc = -ENOMEM;
1782                 goto out;
1783         }
1784
1785         ptlrpc_set_wait(arg->set);
1786         ptlrpc_set_destroy(arg->set);
1787
1788         rc = atomic_read(&arg->restart) ? -ERESTART : 0;
1789         goto out;
1790 out:
1791         kfree(arg);
1792         return rc;
1793 }
1794
1795 /**
1796  * Helper function to call blocking AST for LDLM lock \a lock in a
1797  * "cancelling" mode.
1798  */
1799 void ldlm_cancel_callback(struct ldlm_lock *lock)
1800 {
1801         check_res_locked(lock->l_resource);
1802         if (!ldlm_is_cancel(lock)) {
1803                 ldlm_set_cancel(lock);
1804                 if (lock->l_blocking_ast) {
1805                         unlock_res_and_lock(lock);
1806                         lock->l_blocking_ast(lock, NULL, lock->l_ast_data,
1807                                              LDLM_CB_CANCELING);
1808                         lock_res_and_lock(lock);
1809                 } else {
1810                         LDLM_DEBUG(lock, "no blocking ast");
1811                 }
1812         }
1813         ldlm_set_bl_done(lock);
1814 }
1815
1816 /**
1817  * Remove skiplist-enabled LDLM lock \a req from granted list
1818  */
1819 void ldlm_unlink_lock_skiplist(struct ldlm_lock *req)
1820 {
1821         if (req->l_resource->lr_type != LDLM_PLAIN &&
1822             req->l_resource->lr_type != LDLM_IBITS)
1823                 return;
1824
1825         list_del_init(&req->l_sl_policy);
1826         list_del_init(&req->l_sl_mode);
1827 }
1828
1829 /**
1830  * Attempts to cancel LDLM lock \a lock that has no reader/writer references.
1831  */
1832 void ldlm_lock_cancel(struct ldlm_lock *lock)
1833 {
1834         struct ldlm_resource *res;
1835         struct ldlm_namespace *ns;
1836
1837         lock_res_and_lock(lock);
1838
1839         res = lock->l_resource;
1840         ns  = ldlm_res_to_ns(res);
1841
1842         /* Please do not, no matter how tempting, remove this LBUG without
1843          * talking to me first. -phik
1844          */
1845         if (lock->l_readers || lock->l_writers) {
1846                 LDLM_ERROR(lock, "lock still has references");
1847                 LBUG();
1848         }
1849
1850         /* Releases cancel callback. */
1851         ldlm_cancel_callback(lock);
1852
1853         ldlm_resource_unlink_lock(lock);
1854         ldlm_lock_destroy_nolock(lock);
1855
1856         if (lock->l_granted_mode == lock->l_req_mode)
1857                 ldlm_pool_del(&ns->ns_pool, lock);
1858
1859         /* Make sure we will not be called again for same lock what is possible
1860          * if not to zero out lock->l_granted_mode
1861          */
1862         lock->l_granted_mode = LCK_MINMODE;
1863         unlock_res_and_lock(lock);
1864 }
1865 EXPORT_SYMBOL(ldlm_lock_cancel);
1866
1867 /**
1868  * Set opaque data into the lock that only makes sense to upper layer.
1869  */
1870 int ldlm_lock_set_data(const struct lustre_handle *lockh, void *data)
1871 {
1872         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
1873         int rc = -EINVAL;
1874
1875         if (lock) {
1876                 if (!lock->l_ast_data)
1877                         lock->l_ast_data = data;
1878                 if (lock->l_ast_data == data)
1879                         rc = 0;
1880                 LDLM_LOCK_PUT(lock);
1881         }
1882         return rc;
1883 }
1884 EXPORT_SYMBOL(ldlm_lock_set_data);
1885
1886 struct export_cl_data {
1887         struct obd_export       *ecl_exp;
1888         int                     ecl_loop;
1889 };
1890
1891 /**
1892  * Print lock with lock handle \a lockh description into debug log.
1893  *
1894  * Used when printing all locks on a resource for debug purposes.
1895  */
1896 void ldlm_lock_dump_handle(int level, const struct lustre_handle *lockh)
1897 {
1898         struct ldlm_lock *lock;
1899
1900         if (!((libcfs_debug | D_ERROR) & level))
1901                 return;
1902
1903         lock = ldlm_handle2lock(lockh);
1904         if (!lock)
1905                 return;
1906
1907         LDLM_DEBUG_LIMIT(level, lock, "###");
1908
1909         LDLM_LOCK_PUT(lock);
1910 }
1911 EXPORT_SYMBOL(ldlm_lock_dump_handle);
1912
1913 /**
1914  * Print lock information with custom message into debug log.
1915  * Helper function.
1916  */
1917 void _ldlm_lock_debug(struct ldlm_lock *lock,
1918                       struct libcfs_debug_msg_data *msgdata,
1919                       const char *fmt, ...)
1920 {
1921         va_list args;
1922         struct obd_export *exp = lock->l_export;
1923         struct ldlm_resource *resource = lock->l_resource;
1924         char *nid = "local";
1925
1926         va_start(args, fmt);
1927
1928         if (exp && exp->exp_connection) {
1929                 nid = libcfs_nid2str(exp->exp_connection->c_peer.nid);
1930         } else if (exp && exp->exp_obd) {
1931                 struct obd_import *imp = exp->exp_obd->u.cli.cl_import;
1932
1933                 nid = libcfs_nid2str(imp->imp_connection->c_peer.nid);
1934         }
1935
1936         if (!resource) {
1937                 libcfs_debug_vmsg2(msgdata, fmt, args,
1938                                    " 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",
1939                                    lock,
1940                                    lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
1941                                    lock->l_readers, lock->l_writers,
1942                                    ldlm_lockname[lock->l_granted_mode],
1943                                    ldlm_lockname[lock->l_req_mode],
1944                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
1945                                    exp ? atomic_read(&exp->exp_refcount) : -99,
1946                                    lock->l_pid, lock->l_callback_timeout, lock->l_lvb_type);
1947                 va_end(args);
1948                 return;
1949         }
1950
1951         switch (resource->lr_type) {
1952         case LDLM_EXTENT:
1953                 libcfs_debug_vmsg2(msgdata, fmt, args,
1954                                    " 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",
1955                                    ldlm_lock_to_ns_name(lock), lock,
1956                                    lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
1957                                    lock->l_readers, lock->l_writers,
1958                                    ldlm_lockname[lock->l_granted_mode],
1959                                    ldlm_lockname[lock->l_req_mode],
1960                                    PLDLMRES(resource),
1961                                    atomic_read(&resource->lr_refcount),
1962                                    ldlm_typename[resource->lr_type],
1963                                    lock->l_policy_data.l_extent.start,
1964                                    lock->l_policy_data.l_extent.end,
1965                                    lock->l_req_extent.start, lock->l_req_extent.end,
1966                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
1967                                    exp ? atomic_read(&exp->exp_refcount) : -99,
1968                                    lock->l_pid, lock->l_callback_timeout,
1969                                    lock->l_lvb_type);
1970                 break;
1971
1972         case LDLM_FLOCK:
1973                 libcfs_debug_vmsg2(msgdata, fmt, args,
1974                                    " 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",
1975                                    ldlm_lock_to_ns_name(lock), lock,
1976                                    lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
1977                                    lock->l_readers, lock->l_writers,
1978                                    ldlm_lockname[lock->l_granted_mode],
1979                                    ldlm_lockname[lock->l_req_mode],
1980                                    PLDLMRES(resource),
1981                                    atomic_read(&resource->lr_refcount),
1982                                    ldlm_typename[resource->lr_type],
1983                                    lock->l_policy_data.l_flock.pid,
1984                                    lock->l_policy_data.l_flock.start,
1985                                    lock->l_policy_data.l_flock.end,
1986                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
1987                                    exp ? atomic_read(&exp->exp_refcount) : -99,
1988                                    lock->l_pid, lock->l_callback_timeout);
1989                 break;
1990
1991         case LDLM_IBITS:
1992                 libcfs_debug_vmsg2(msgdata, fmt, args,
1993                                    " 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",
1994                                    ldlm_lock_to_ns_name(lock),
1995                                    lock, lock->l_handle.h_cookie,
1996                                    atomic_read(&lock->l_refc),
1997                                    lock->l_readers, lock->l_writers,
1998                                    ldlm_lockname[lock->l_granted_mode],
1999                                    ldlm_lockname[lock->l_req_mode],
2000                                    PLDLMRES(resource),
2001                                    lock->l_policy_data.l_inodebits.bits,
2002                                    atomic_read(&resource->lr_refcount),
2003                                    ldlm_typename[resource->lr_type],
2004                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
2005                                    exp ? atomic_read(&exp->exp_refcount) : -99,
2006                                    lock->l_pid, lock->l_callback_timeout,
2007                                    lock->l_lvb_type);
2008                 break;
2009
2010         default:
2011                 libcfs_debug_vmsg2(msgdata, fmt, args,
2012                                    " 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",
2013                                    ldlm_lock_to_ns_name(lock),
2014                                    lock, lock->l_handle.h_cookie,
2015                                    atomic_read(&lock->l_refc),
2016                                    lock->l_readers, lock->l_writers,
2017                                    ldlm_lockname[lock->l_granted_mode],
2018                                    ldlm_lockname[lock->l_req_mode],
2019                                    PLDLMRES(resource),
2020                                    atomic_read(&resource->lr_refcount),
2021                                    ldlm_typename[resource->lr_type],
2022                                    lock->l_flags, nid, lock->l_remote_handle.cookie,
2023                                    exp ? atomic_read(&exp->exp_refcount) : -99,
2024                                    lock->l_pid, lock->l_callback_timeout,
2025                                    lock->l_lvb_type);
2026                 break;
2027         }
2028         va_end(args);
2029 }
2030 EXPORT_SYMBOL(_ldlm_lock_debug);