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