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