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