drivers/staging/lustre/lustre/ldlm/ldlm_lock.c

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