lustre/obdclass/cl_lock.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * Client Extent Lock.
  37  *
  38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  39  *   Author: Jinshan Xiong <jinshan.xiong@intel.com>
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_CLASS
  43
  44 #include <libcfs/libcfs.h>
  45 #include <obd_class.h>
  46 #include <obd_support.h>
  47 #include <lustre_fid.h>
  48 #include <libcfs/list.h>
  49 #include <cl_object.h>
  50 #include "cl_internal.h"
  51
  52 /** Lock class of cl_lock::cll_guard */
  53 static struct lock_class_key cl_lock_guard_class;
  54 static struct kmem_cache *cl_lock_kmem;
  55
  56 static struct lu_kmem_descr cl_lock_caches[] = {
  57         {
  58                 .ckd_cache = &cl_lock_kmem,
  59                 .ckd_name  = "cl_lock_kmem",
  60                 .ckd_size  = sizeof (struct cl_lock)
  61         },
  62         {
  63                 .ckd_cache = NULL
  64         }
  65 };
  66
  67 #ifdef CONFIG_DEBUG_PAGESTATE_TRACKING
  68 #define CS_LOCK_INC(o, item) \
  69         atomic_inc(&cl_object_site(o)->cs_locks.cs_stats[CS_##item])
  70 #define CS_LOCK_DEC(o, item) \
  71         atomic_dec(&cl_object_site(o)->cs_locks.cs_stats[CS_##item])
  72 #define CS_LOCKSTATE_INC(o, state) \
  73         atomic_inc(&cl_object_site(o)->cs_locks_state[state])
  74 #define CS_LOCKSTATE_DEC(o, state) \
  75         atomic_dec(&cl_object_site(o)->cs_locks_state[state])
  76 #else
  77 #define CS_LOCK_INC(o, item)
  78 #define CS_LOCK_DEC(o, item)
  79 #define CS_LOCKSTATE_INC(o, state)
  80 #define CS_LOCKSTATE_DEC(o, state)
  81 #endif
  82
  83 /**
  84  * Basic lock invariant that is maintained at all times. Caller either has a
  85  * reference to \a lock, or somehow assures that \a lock cannot be freed.
  86  *
  87  * \see cl_lock_invariant()
  88  */
  89 static int cl_lock_invariant_trusted(const struct lu_env *env,
  90                                      const struct cl_lock *lock)
  91 {
  92         return  ergo(lock->cll_state == CLS_FREEING, lock->cll_holds == 0) &&
  93                 atomic_read(&lock->cll_ref) >= lock->cll_holds &&
  94                 lock->cll_holds >= lock->cll_users &&
  95                 lock->cll_holds >= 0 &&
  96                 lock->cll_users >= 0 &&
  97                 lock->cll_depth >= 0;
  98 }
  99
 100 /**
 101  * Stronger lock invariant, checking that caller has a reference on a lock.
 102  *
 103  * \see cl_lock_invariant_trusted()
 104  */
 105 static int cl_lock_invariant(const struct lu_env *env,
 106                              const struct cl_lock *lock)
 107 {
 108         int result;
 109
 110         result = atomic_read(&lock->cll_ref) > 0 &&
 111                 cl_lock_invariant_trusted(env, lock);
 112         if (!result && env != NULL)
 113                 CL_LOCK_DEBUG(D_ERROR, env, lock, "invariant broken\n");
 114         return result;
 115 }
 116
 117 /**
 118  * Returns lock "nesting": 0 for a top-lock and 1 for a sub-lock.
 119  */
 120 static enum clt_nesting_level cl_lock_nesting(const struct cl_lock *lock)
 121 {
 122         return cl_object_header(lock->cll_descr.cld_obj)->coh_nesting;
 123 }
 124
 125 /**
 126  * Returns a set of counters for this lock, depending on a lock nesting.
 127  */
 128 static struct cl_thread_counters *cl_lock_counters(const struct lu_env *env,
 129                                                    const struct cl_lock *lock)
 130 {
 131         struct cl_thread_info *info;
 132         enum clt_nesting_level nesting;
 133
 134         info = cl_env_info(env);
 135         nesting = cl_lock_nesting(lock);
 136         LASSERT(nesting < ARRAY_SIZE(info->clt_counters));
 137         return &info->clt_counters[nesting];
 138 }
 139
 140 static void cl_lock_trace0(int level, const struct lu_env *env,
 141                            const char *prefix, const struct cl_lock *lock,
 142                            const char *func, const int line)
 143 {
 144         struct cl_object_header *h = cl_object_header(lock->cll_descr.cld_obj);
 145         CDEBUG(level, "%s: %p@(%d %p %d %d %d %d %d %lx)"
 146                       "(%p/%d/%d) at %s():%d\n",
 147                prefix, lock, atomic_read(&lock->cll_ref),
 148                lock->cll_guarder, lock->cll_depth,
 149                lock->cll_state, lock->cll_error, lock->cll_holds,
 150                lock->cll_users, lock->cll_flags,
 151                env, h->coh_nesting, cl_lock_nr_mutexed(env),
 152                func, line);
 153 }
 154 #define cl_lock_trace(level, env, prefix, lock)                         \
 155         cl_lock_trace0(level, env, prefix, lock, __FUNCTION__, __LINE__)
 156
 157 #define RETIP ((unsigned long)__builtin_return_address(0))
 158
 159 #ifdef CONFIG_LOCKDEP
 160 static struct lock_class_key cl_lock_key;
 161
 162 static void cl_lock_lockdep_init(struct cl_lock *lock)
 163 {
 164         lockdep_set_class_and_name(lock, &cl_lock_key, "EXT");
 165 }
 166
 167 static void cl_lock_lockdep_acquire(const struct lu_env *env,
 168                                     struct cl_lock *lock, __u32 enqflags)
 169 {
 170         cl_lock_counters(env, lock)->ctc_nr_locks_acquired++;
 171         lock_map_acquire(&lock->dep_map);
 172 }
 173
 174 static void cl_lock_lockdep_release(const struct lu_env *env,
 175                                     struct cl_lock *lock)
 176 {
 177         cl_lock_counters(env, lock)->ctc_nr_locks_acquired--;
 178         lock_map_release(&lock->dep_map);
 179 }
 180
 181 #else /* !CONFIG_LOCKDEP */
 182
 183 static void cl_lock_lockdep_init(struct cl_lock *lock)
 184 {}
 185 static void cl_lock_lockdep_acquire(const struct lu_env *env,
 186                                     struct cl_lock *lock, __u32 enqflags)
 187 {}
 188 static void cl_lock_lockdep_release(const struct lu_env *env,
 189                                     struct cl_lock *lock)
 190 {}
 191
 192 #endif /* !CONFIG_LOCKDEP */
 193
 194 /**
 195  * Adds lock slice to the compound lock.
 196  *
 197  * This is called by cl_object_operations::coo_lock_init() methods to add a
 198  * per-layer state to the lock. New state is added at the end of
 199  * cl_lock::cll_layers list, that is, it is at the bottom of the stack.
 200  *
 201  * \see cl_req_slice_add(), cl_page_slice_add(), cl_io_slice_add()
 202  */
 203 void cl_lock_slice_add(struct cl_lock *lock, struct cl_lock_slice *slice,
 204                        struct cl_object *obj,
 205                        const struct cl_lock_operations *ops)
 206 {
 207         ENTRY;
 208         slice->cls_lock = lock;
 209         list_add_tail(&slice->cls_linkage, &lock->cll_layers);
 210         slice->cls_obj = obj;
 211         slice->cls_ops = ops;
 212         EXIT;
 213 }
 214 EXPORT_SYMBOL(cl_lock_slice_add);
 215
 216 /**
 217  * Returns true iff a lock with the mode \a has provides at least the same
 218  * guarantees as a lock with the mode \a need.
 219  */
 220 int cl_lock_mode_match(enum cl_lock_mode has, enum cl_lock_mode need)
 221 {
 222         LINVRNT(need == CLM_READ || need == CLM_WRITE ||
 223                 need == CLM_PHANTOM || need == CLM_GROUP);
 224         LINVRNT(has == CLM_READ || has == CLM_WRITE ||
 225                 has == CLM_PHANTOM || has == CLM_GROUP);
 226         CLASSERT(CLM_PHANTOM < CLM_READ);
 227         CLASSERT(CLM_READ < CLM_WRITE);
 228         CLASSERT(CLM_WRITE < CLM_GROUP);
 229
 230         if (has != CLM_GROUP)
 231                 return need <= has;
 232         else
 233                 return need == has;
 234 }
 235 EXPORT_SYMBOL(cl_lock_mode_match);
 236
 237 /**
 238  * Returns true iff extent portions of lock descriptions match.
 239  */
 240 int cl_lock_ext_match(const struct cl_lock_descr *has,
 241                       const struct cl_lock_descr *need)
 242 {
 243         return
 244                 has->cld_start <= need->cld_start &&
 245                 has->cld_end >= need->cld_end &&
 246                 cl_lock_mode_match(has->cld_mode, need->cld_mode) &&
 247                 (has->cld_mode != CLM_GROUP || has->cld_gid == need->cld_gid);
 248 }
 249 EXPORT_SYMBOL(cl_lock_ext_match);
 250
 251 /**
 252  * Returns true iff a lock with the description \a has provides at least the
 253  * same guarantees as a lock with the description \a need.
 254  */
 255 int cl_lock_descr_match(const struct cl_lock_descr *has,
 256                         const struct cl_lock_descr *need)
 257 {
 258         return
 259                 cl_object_same(has->cld_obj, need->cld_obj) &&
 260                 cl_lock_ext_match(has, need);
 261 }
 262 EXPORT_SYMBOL(cl_lock_descr_match);
 263
 264 static void cl_lock_free(const struct lu_env *env, struct cl_lock *lock)
 265 {
 266         struct cl_object *obj = lock->cll_descr.cld_obj;
 267
 268         LINVRNT(!cl_lock_is_mutexed(lock));
 269
 270         ENTRY;
 271         cl_lock_trace(D_DLMTRACE, env, "free lock", lock);
 272         while (!list_empty(&lock->cll_layers)) {
 273                 struct cl_lock_slice *slice;
 274
 275                 slice = list_entry(lock->cll_layers.next,
 276                                    struct cl_lock_slice, cls_linkage);
 277                 list_del_init(lock->cll_layers.next);
 278                 slice->cls_ops->clo_fini(env, slice);
 279         }
 280         CS_LOCK_DEC(obj, total);
 281         CS_LOCKSTATE_DEC(obj, lock->cll_state);
 282         lu_object_ref_del_at(&obj->co_lu, &lock->cll_obj_ref, "cl_lock", lock);
 283         cl_object_put(env, obj);
 284         lu_ref_fini(&lock->cll_reference);
 285         lu_ref_fini(&lock->cll_holders);
 286         mutex_destroy(&lock->cll_guard);
 287         OBD_SLAB_FREE_PTR(lock, cl_lock_kmem);
 288         EXIT;
 289 }
 290
 291 /**
 292  * Releases a reference on a lock.
 293  *
 294  * When last reference is released, lock is returned to the cache, unless it
 295  * is in cl_lock_state::CLS_FREEING state, in which case it is destroyed
 296  * immediately.
 297  *
 298  * \see cl_object_put(), cl_page_put()
 299  */
 300 void cl_lock_put(const struct lu_env *env, struct cl_lock *lock)
 301 {
 302         struct cl_object        *obj;
 303
 304         LINVRNT(cl_lock_invariant(env, lock));
 305         ENTRY;
 306         obj = lock->cll_descr.cld_obj;
 307         LINVRNT(obj != NULL);
 308
 309         CDEBUG(D_TRACE, "releasing reference: %d %p %lu\n",
 310                atomic_read(&lock->cll_ref), lock, RETIP);
 311
 312         if (atomic_dec_and_test(&lock->cll_ref)) {
 313                 if (lock->cll_state == CLS_FREEING) {
 314                         LASSERT(list_empty(&lock->cll_linkage));
 315                         cl_lock_free(env, lock);
 316                 }
 317                 CS_LOCK_DEC(obj, busy);
 318         }
 319         EXIT;
 320 }
 321 EXPORT_SYMBOL(cl_lock_put);
 322
 323 /**
 324  * Acquires an additional reference to a lock.
 325  *
 326  * This can be called only by caller already possessing a reference to \a
 327  * lock.
 328  *
 329  * \see cl_object_get(), cl_page_get()
 330  */
 331 void cl_lock_get(struct cl_lock *lock)
 332 {
 333         LINVRNT(cl_lock_invariant(NULL, lock));
 334         CDEBUG(D_TRACE, "acquiring reference: %d %p %lu\n",
 335                atomic_read(&lock->cll_ref), lock, RETIP);
 336         atomic_inc(&lock->cll_ref);
 337 }
 338 EXPORT_SYMBOL(cl_lock_get);
 339
 340 /**
 341  * Acquires a reference to a lock.
 342  *
 343  * This is much like cl_lock_get(), except that this function can be used to
 344  * acquire initial reference to the cached lock. Caller has to deal with all
 345  * possible races. Use with care!
 346  *
 347  * \see cl_page_get_trust()
 348  */
 349 void cl_lock_get_trust(struct cl_lock *lock)
 350 {
 351         CDEBUG(D_TRACE, "acquiring trusted reference: %d %p %lu\n",
 352                atomic_read(&lock->cll_ref), lock, RETIP);
 353         if (atomic_inc_return(&lock->cll_ref) == 1)
 354                 CS_LOCK_INC(lock->cll_descr.cld_obj, busy);
 355 }
 356 EXPORT_SYMBOL(cl_lock_get_trust);
 357
 358 /**
 359  * Helper function destroying the lock that wasn't completely initialized.
 360  *
 361  * Other threads can acquire references to the top-lock through its
 362  * sub-locks. Hence, it cannot be cl_lock_free()-ed immediately.
 363  */
 364 static void cl_lock_finish(const struct lu_env *env, struct cl_lock *lock)
 365 {
 366         cl_lock_mutex_get(env, lock);
 367         cl_lock_cancel(env, lock);
 368         cl_lock_delete(env, lock);
 369         cl_lock_mutex_put(env, lock);
 370         cl_lock_put(env, lock);
 371 }
 372
 373 static struct cl_lock *cl_lock_alloc(const struct lu_env *env,
 374                                      struct cl_object *obj,
 375                                      const struct cl_io *io,
 376                                      const struct cl_lock_descr *descr)
 377 {
 378         struct cl_lock          *lock;
 379         struct lu_object_header *head;
 380
 381         ENTRY;
 382         OBD_SLAB_ALLOC_PTR_GFP(lock, cl_lock_kmem, GFP_NOFS);
 383         if (lock != NULL) {
 384                 atomic_set(&lock->cll_ref, 1);
 385                 lock->cll_descr = *descr;
 386                 lock->cll_state = CLS_NEW;
 387                 cl_object_get(obj);
 388                 lu_object_ref_add_at(&obj->co_lu, &lock->cll_obj_ref, "cl_lock",
 389                                      lock);
 390                 INIT_LIST_HEAD(&lock->cll_layers);
 391                 INIT_LIST_HEAD(&lock->cll_linkage);
 392                 INIT_LIST_HEAD(&lock->cll_inclosure);
 393                 lu_ref_init(&lock->cll_reference);
 394                 lu_ref_init(&lock->cll_holders);
 395                 mutex_init(&lock->cll_guard);
 396                 lockdep_set_class(&lock->cll_guard, &cl_lock_guard_class);
 397                 init_waitqueue_head(&lock->cll_wq);
 398                 head = obj->co_lu.lo_header;
 399                 CS_LOCKSTATE_INC(obj, CLS_NEW);
 400                 CS_LOCK_INC(obj, total);
 401                 CS_LOCK_INC(obj, create);
 402                 cl_lock_lockdep_init(lock);
 403                 list_for_each_entry(obj, &head->loh_layers, co_lu.lo_linkage) {
 404                         int err;
 405
 406                         err = obj->co_ops->coo_lock_init(env, obj, lock, io);
 407                         if (err != 0) {
 408                                 cl_lock_finish(env, lock);
 409                                 lock = ERR_PTR(err);
 410                                 break;
 411                         }
 412                 }
 413         } else
 414                 lock = ERR_PTR(-ENOMEM);
 415         RETURN(lock);
 416 }
 417
 418 /**
 419  * Transfer the lock into INTRANSIT state and return the original state.
 420  *
 421  * \pre  state: CLS_CACHED, CLS_HELD or CLS_ENQUEUED
 422  * \post state: CLS_INTRANSIT
 423  * \see CLS_INTRANSIT
 424  */
 425 enum cl_lock_state cl_lock_intransit(const struct lu_env *env,
 426                                      struct cl_lock *lock)
 427 {
 428         enum cl_lock_state state = lock->cll_state;
 429
 430         LASSERT(cl_lock_is_mutexed(lock));
 431         LASSERT(state != CLS_INTRANSIT);
 432         LASSERTF(state >= CLS_ENQUEUED && state <= CLS_CACHED,
 433                  "Malformed lock state %d.\n", state);
 434
 435         cl_lock_state_set(env, lock, CLS_INTRANSIT);
 436         lock->cll_intransit_owner = current;
 437         cl_lock_hold_add(env, lock, "intransit", current);
 438         return state;
 439 }
 440 EXPORT_SYMBOL(cl_lock_intransit);
 441
 442 /**
 443  *  Exit the intransit state and restore the lock state to the original state
 444  */
 445 void cl_lock_extransit(const struct lu_env *env, struct cl_lock *lock,
 446                        enum cl_lock_state state)
 447 {
 448         LASSERT(cl_lock_is_mutexed(lock));
 449         LASSERT(lock->cll_state == CLS_INTRANSIT);
 450         LASSERT(state != CLS_INTRANSIT);
 451         LASSERT(lock->cll_intransit_owner == current);
 452
 453         lock->cll_intransit_owner = NULL;
 454         cl_lock_state_set(env, lock, state);
 455         cl_lock_unhold(env, lock, "intransit", current);
 456 }
 457 EXPORT_SYMBOL(cl_lock_extransit);
 458
 459 /**
 460  * Checking whether the lock is intransit state
 461  */
 462 int cl_lock_is_intransit(struct cl_lock *lock)
 463 {
 464         LASSERT(cl_lock_is_mutexed(lock));
 465         return lock->cll_state == CLS_INTRANSIT &&
 466                lock->cll_intransit_owner != current;
 467 }
 468 EXPORT_SYMBOL(cl_lock_is_intransit);
 469 /**
 470  * Returns true iff lock is "suitable" for given io. E.g., locks acquired by
 471  * truncate and O_APPEND cannot be reused for read/non-append-write, as they
 472  * cover multiple stripes and can trigger cascading timeouts.
 473  */
 474 static int cl_lock_fits_into(const struct lu_env *env,
 475                              const struct cl_lock *lock,
 476                              const struct cl_lock_descr *need,
 477                              const struct cl_io *io)
 478 {
 479         const struct cl_lock_slice *slice;
 480
 481         LINVRNT(cl_lock_invariant_trusted(env, lock));
 482         ENTRY;
 483         list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
 484                 if (slice->cls_ops->clo_fits_into != NULL &&
 485                     !slice->cls_ops->clo_fits_into(env, slice, need, io))
 486                         RETURN(0);
 487         }
 488         RETURN(1);
 489 }
 490
 491 static struct cl_lock *cl_lock_lookup(const struct lu_env *env,
 492                                       struct cl_object *obj,
 493                                       const struct cl_io *io,
 494                                       const struct cl_lock_descr *need)
 495 {
 496         struct cl_lock          *lock;
 497         struct cl_object_header *head;
 498
 499         ENTRY;
 500
 501         head = cl_object_header(obj);
 502         assert_spin_locked(&head->coh_lock_guard);
 503         CS_LOCK_INC(obj, lookup);
 504         list_for_each_entry(lock, &head->coh_locks, cll_linkage) {
 505                 int matched;
 506
 507                 matched = cl_lock_ext_match(&lock->cll_descr, need) &&
 508                           lock->cll_state < CLS_FREEING &&
 509                           lock->cll_error == 0 &&
 510                           !(lock->cll_flags & CLF_CANCELLED) &&
 511                           cl_lock_fits_into(env, lock, need, io);
 512                 CDEBUG(D_DLMTRACE, "has: "DDESCR"(%d) need: "DDESCR": %d\n",
 513                        PDESCR(&lock->cll_descr), lock->cll_state, PDESCR(need),
 514                        matched);
 515                 if (matched) {
 516                         cl_lock_get_trust(lock);
 517                         CS_LOCK_INC(obj, hit);
 518                         RETURN(lock);
 519                 }
 520         }
 521         RETURN(NULL);
 522 }
 523
 524 /**
 525  * Returns a lock matching description \a need.
 526  *
 527  * This is the main entry point into the cl_lock caching interface. First, a
 528  * cache (implemented as a per-object linked list) is consulted. If lock is
 529  * found there, it is returned immediately. Otherwise new lock is allocated
 530  * and returned. In any case, additional reference to lock is acquired.
 531  *
 532  * \see cl_object_find(), cl_page_find()
 533  */
 534 static struct cl_lock *cl_lock_find(const struct lu_env *env,
 535                                     const struct cl_io *io,
 536                                     const struct cl_lock_descr *need)
 537 {
 538         struct cl_object_header *head;
 539         struct cl_object        *obj;
 540         struct cl_lock          *lock;
 541
 542         ENTRY;
 543
 544         obj  = need->cld_obj;
 545         head = cl_object_header(obj);
 546
 547         spin_lock(&head->coh_lock_guard);
 548         lock = cl_lock_lookup(env, obj, io, need);
 549         spin_unlock(&head->coh_lock_guard);
 550
 551         if (lock == NULL) {
 552                 lock = cl_lock_alloc(env, obj, io, need);
 553                 if (!IS_ERR(lock)) {
 554                         struct cl_lock *ghost;
 555
 556                         spin_lock(&head->coh_lock_guard);
 557                         ghost = cl_lock_lookup(env, obj, io, need);
 558                         if (ghost == NULL) {
 559                                 cl_lock_get_trust(lock);
 560                                 list_add_tail(&lock->cll_linkage,
 561                                               &head->coh_locks);
 562                                 spin_unlock(&head->coh_lock_guard);
 563                                 CS_LOCK_INC(obj, busy);
 564                         } else {
 565                                 spin_unlock(&head->coh_lock_guard);
 566                                 /*
 567                                  * Other threads can acquire references to the
 568                                  * top-lock through its sub-locks. Hence, it
 569                                  * cannot be cl_lock_free()-ed immediately.
 570                                  */
 571                                 cl_lock_finish(env, lock);
 572                                 lock = ghost;
 573                         }
 574                 }
 575         }
 576         RETURN(lock);
 577 }
 578
 579 /**
 580  * Returns existing lock matching given description. This is similar to
 581  * cl_lock_find() except that no new lock is created, and returned lock is
 582  * guaranteed to be in enum cl_lock_state::CLS_HELD state.
 583  */
 584 struct cl_lock *cl_lock_peek(const struct lu_env *env, const struct cl_io *io,
 585                              const struct cl_lock_descr *need,
 586                              const char *scope, const void *source)
 587 {
 588         struct cl_object_header *head;
 589         struct cl_object        *obj;
 590         struct cl_lock          *lock;
 591
 592         obj  = need->cld_obj;
 593         head = cl_object_header(obj);
 594
 595         do {
 596                 spin_lock(&head->coh_lock_guard);
 597                 lock = cl_lock_lookup(env, obj, io, need);
 598                 spin_unlock(&head->coh_lock_guard);
 599                 if (lock == NULL)
 600                         return NULL;
 601
 602                 cl_lock_mutex_get(env, lock);
 603                 if (lock->cll_state == CLS_INTRANSIT)
 604                         /* Don't care return value. */
 605                         cl_lock_state_wait(env, lock);
 606                 if (lock->cll_state == CLS_FREEING) {
 607                         cl_lock_mutex_put(env, lock);
 608                         cl_lock_put(env, lock);
 609                         lock = NULL;
 610                 }
 611         } while (lock == NULL);
 612
 613         cl_lock_hold_add(env, lock, scope, source);
 614         cl_lock_user_add(env, lock);
 615         if (lock->cll_state == CLS_CACHED)
 616                 cl_use_try(env, lock, 1);
 617         if (lock->cll_state == CLS_HELD) {
 618                 cl_lock_mutex_put(env, lock);
 619                 cl_lock_lockdep_acquire(env, lock, 0);
 620                 cl_lock_put(env, lock);
 621         } else {
 622                 cl_unuse_try(env, lock);
 623                 cl_lock_unhold(env, lock, scope, source);
 624                 cl_lock_mutex_put(env, lock);
 625                 cl_lock_put(env, lock);
 626                 lock = NULL;
 627         }
 628
 629         return lock;
 630 }
 631 EXPORT_SYMBOL(cl_lock_peek);
 632
 633 /**
 634  * Returns a slice within a lock, corresponding to the given layer in the
 635  * device stack.
 636  *
 637  * \see cl_page_at()
 638  */
 639 const struct cl_lock_slice *cl_lock_at(const struct cl_lock *lock,
 640                                        const struct lu_device_type *dtype)
 641 {
 642         const struct cl_lock_slice *slice;
 643
 644         LINVRNT(cl_lock_invariant_trusted(NULL, lock));
 645         ENTRY;
 646
 647         list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
 648                 if (slice->cls_obj->co_lu.lo_dev->ld_type == dtype)
 649                         RETURN(slice);
 650         }
 651         RETURN(NULL);
 652 }
 653 EXPORT_SYMBOL(cl_lock_at);
 654
 655 static void cl_lock_mutex_tail(const struct lu_env *env, struct cl_lock *lock)
 656 {
 657         struct cl_thread_counters *counters;
 658
 659         counters = cl_lock_counters(env, lock);
 660         lock->cll_depth++;
 661         counters->ctc_nr_locks_locked++;
 662         lu_ref_add(&counters->ctc_locks_locked, "cll_guard", lock);
 663         cl_lock_trace(D_TRACE, env, "got mutex", lock);
 664 }
 665
 666 /**
 667  * Locks cl_lock object.
 668  *
 669  * This is used to manipulate cl_lock fields, and to serialize state
 670  * transitions in the lock state machine.
 671  *
 672  * \post cl_lock_is_mutexed(lock)
 673  *
 674  * \see cl_lock_mutex_put()
 675  */
 676 void cl_lock_mutex_get(const struct lu_env *env, struct cl_lock *lock)
 677 {
 678         LINVRNT(cl_lock_invariant(env, lock));
 679
 680         if (lock->cll_guarder == current) {
 681                 LINVRNT(cl_lock_is_mutexed(lock));
 682                 LINVRNT(lock->cll_depth > 0);
 683         } else {
 684                 struct cl_object_header *hdr;
 685                 struct cl_thread_info   *info;
 686                 int i;
 687
 688                 LINVRNT(lock->cll_guarder != current);
 689                 hdr = cl_object_header(lock->cll_descr.cld_obj);
 690                 /*
 691                  * Check that mutices are taken in the bottom-to-top order.
 692                  */
 693                 info = cl_env_info(env);
 694                 for (i = 0; i < hdr->coh_nesting; ++i)
 695                         LASSERT(info->clt_counters[i].ctc_nr_locks_locked == 0);
 696                 mutex_lock_nested(&lock->cll_guard, hdr->coh_nesting);
 697                 lock->cll_guarder = current;
 698                 LINVRNT(lock->cll_depth == 0);
 699         }
 700         cl_lock_mutex_tail(env, lock);
 701 }
 702 EXPORT_SYMBOL(cl_lock_mutex_get);
 703
 704 /**
 705  * Try-locks cl_lock object.
 706  *
 707  * \retval 0 \a lock was successfully locked
 708  *
 709  * \retval -EBUSY \a lock cannot be locked right now
 710  *
 711  * \post ergo(result == 0, cl_lock_is_mutexed(lock))
 712  *
 713  * \see cl_lock_mutex_get()
 714  */
 715 int cl_lock_mutex_try(const struct lu_env *env, struct cl_lock *lock)
 716 {
 717         int result;
 718
 719         LINVRNT(cl_lock_invariant_trusted(env, lock));
 720         ENTRY;
 721
 722         result = 0;
 723         if (lock->cll_guarder == current) {
 724                 LINVRNT(lock->cll_depth > 0);
 725                 cl_lock_mutex_tail(env, lock);
 726         } else if (mutex_trylock(&lock->cll_guard)) {
 727                 LINVRNT(lock->cll_depth == 0);
 728                 lock->cll_guarder = current;
 729                 cl_lock_mutex_tail(env, lock);
 730         } else
 731                 result = -EBUSY;
 732         RETURN(result);
 733 }
 734 EXPORT_SYMBOL(cl_lock_mutex_try);
 735
 736 /**
 737  {* Unlocks cl_lock object.
 738  *
 739  * \pre cl_lock_is_mutexed(lock)
 740  *
 741  * \see cl_lock_mutex_get()
 742  */
 743 void cl_lock_mutex_put(const struct lu_env *env, struct cl_lock *lock)
 744 {
 745         struct cl_thread_counters *counters;
 746
 747         LINVRNT(cl_lock_invariant(env, lock));
 748         LINVRNT(cl_lock_is_mutexed(lock));
 749         LINVRNT(lock->cll_guarder == current);
 750         LINVRNT(lock->cll_depth > 0);
 751
 752         counters = cl_lock_counters(env, lock);
 753         LINVRNT(counters->ctc_nr_locks_locked > 0);
 754
 755         cl_lock_trace(D_TRACE, env, "put mutex", lock);
 756         lu_ref_del(&counters->ctc_locks_locked, "cll_guard", lock);
 757         counters->ctc_nr_locks_locked--;
 758         if (--lock->cll_depth == 0) {
 759                 lock->cll_guarder = NULL;
 760                 mutex_unlock(&lock->cll_guard);
 761         }
 762 }
 763 EXPORT_SYMBOL(cl_lock_mutex_put);
 764
 765 /**
 766  * Returns true iff lock's mutex is owned by the current thread.
 767  */
 768 int cl_lock_is_mutexed(struct cl_lock *lock)
 769 {
 770         return lock->cll_guarder == current;
 771 }
 772 EXPORT_SYMBOL(cl_lock_is_mutexed);
 773
 774 /**
 775  * Returns number of cl_lock mutices held by the current thread (environment).
 776  */
 777 int cl_lock_nr_mutexed(const struct lu_env *env)
 778 {
 779         struct cl_thread_info *info;
 780         int i;
 781         int locked;
 782
 783         /*
 784          * NOTE: if summation across all nesting levels (currently 2) proves
 785          *       too expensive, a summary counter can be added to
 786          *       struct cl_thread_info.
 787          */
 788         info = cl_env_info(env);
 789         for (i = 0, locked = 0; i < ARRAY_SIZE(info->clt_counters); ++i)
 790                 locked += info->clt_counters[i].ctc_nr_locks_locked;
 791         return locked;
 792 }
 793 EXPORT_SYMBOL(cl_lock_nr_mutexed);
 794
 795 static void cl_lock_cancel0(const struct lu_env *env, struct cl_lock *lock)
 796 {
 797         LINVRNT(cl_lock_is_mutexed(lock));
 798         LINVRNT(cl_lock_invariant(env, lock));
 799         ENTRY;
 800         if (!(lock->cll_flags & CLF_CANCELLED)) {
 801                 const struct cl_lock_slice *slice;
 802
 803                 lock->cll_flags |= CLF_CANCELLED;
 804                 list_for_each_entry_reverse(slice, &lock->cll_layers,
 805                                             cls_linkage) {
 806                         if (slice->cls_ops->clo_cancel != NULL)
 807                                 slice->cls_ops->clo_cancel(env, slice);
 808                 }
 809         }
 810         EXIT;
 811 }
 812
 813 static void cl_lock_delete0(const struct lu_env *env, struct cl_lock *lock)
 814 {
 815         struct cl_object_header    *head;
 816         const struct cl_lock_slice *slice;
 817
 818         LINVRNT(cl_lock_is_mutexed(lock));
 819         LINVRNT(cl_lock_invariant(env, lock));
 820
 821         ENTRY;
 822         if (lock->cll_state < CLS_FREEING) {
 823                 bool in_cache;
 824
 825                 LASSERT(lock->cll_state != CLS_INTRANSIT);
 826                 cl_lock_state_set(env, lock, CLS_FREEING);
 827
 828                 head = cl_object_header(lock->cll_descr.cld_obj);
 829
 830                 spin_lock(&head->coh_lock_guard);
 831                 in_cache = !list_empty(&lock->cll_linkage);
 832                 if (in_cache)
 833                         list_del_init(&lock->cll_linkage);
 834                 spin_unlock(&head->coh_lock_guard);
 835
 836                 if (in_cache) /* coh_locks cache holds a refcount. */
 837                         cl_lock_put(env, lock);
 838
 839                 /*
 840                  * From now on, no new references to this lock can be acquired
 841                  * by cl_lock_lookup().
 842                  */
 843                 list_for_each_entry_reverse(slice, &lock->cll_layers,
 844                                             cls_linkage) {
 845                         if (slice->cls_ops->clo_delete != NULL)
 846                                 slice->cls_ops->clo_delete(env, slice);
 847                 }
 848                 /*
 849                  * From now on, no new references to this lock can be acquired
 850                  * by layer-specific means (like a pointer from struct
 851                  * ldlm_lock in osc, or a pointer from top-lock to sub-lock in
 852                  * lov).
 853                  *
 854                  * Lock will be finally freed in cl_lock_put() when last of
 855                  * existing references goes away.
 856                  */
 857         }
 858         EXIT;
 859 }
 860
 861 /**
 862  * Mod(ifie)s cl_lock::cll_holds counter for a given lock. Also, for a
 863  * top-lock (nesting == 0) accounts for this modification in the per-thread
 864  * debugging counters. Sub-lock holds can be released by a thread different
 865  * from one that acquired it.
 866  */
 867 static void cl_lock_hold_mod(const struct lu_env *env, struct cl_lock *lock,
 868                              int delta)
 869 {
 870         struct cl_thread_counters *counters;
 871         enum clt_nesting_level     nesting;
 872
 873         lock->cll_holds += delta;
 874         nesting = cl_lock_nesting(lock);
 875         if (nesting == CNL_TOP) {
 876                 counters = &cl_env_info(env)->clt_counters[CNL_TOP];
 877                 counters->ctc_nr_held += delta;
 878                 LASSERT(counters->ctc_nr_held >= 0);
 879         }
 880 }
 881
 882 /**
 883  * Mod(ifie)s cl_lock::cll_users counter for a given lock. See
 884  * cl_lock_hold_mod() for the explanation of the debugging code.
 885  */
 886 static void cl_lock_used_mod(const struct lu_env *env, struct cl_lock *lock,
 887                              int delta)
 888 {
 889         struct cl_thread_counters *counters;
 890         enum clt_nesting_level     nesting;
 891
 892         lock->cll_users += delta;
 893         nesting = cl_lock_nesting(lock);
 894         if (nesting == CNL_TOP) {
 895                 counters = &cl_env_info(env)->clt_counters[CNL_TOP];
 896                 counters->ctc_nr_used += delta;
 897                 LASSERT(counters->ctc_nr_used >= 0);
 898         }
 899 }
 900
 901 void cl_lock_hold_release(const struct lu_env *env, struct cl_lock *lock,
 902                           const char *scope, const void *source)
 903 {
 904         LINVRNT(cl_lock_is_mutexed(lock));
 905         LINVRNT(cl_lock_invariant(env, lock));
 906         LASSERT(lock->cll_holds > 0);
 907
 908         ENTRY;
 909         cl_lock_trace(D_DLMTRACE, env, "hold release lock", lock);
 910         lu_ref_del(&lock->cll_holders, scope, source);
 911         cl_lock_hold_mod(env, lock, -1);
 912         if (lock->cll_holds == 0) {
 913                 CL_LOCK_ASSERT(lock->cll_state != CLS_HELD, env, lock);
 914                 if (lock->cll_descr.cld_mode == CLM_PHANTOM ||
 915                     lock->cll_descr.cld_mode == CLM_GROUP ||
 916                     lock->cll_state != CLS_CACHED)
 917                         /*
 918                          * If lock is still phantom or grouplock when user is
 919                          * done with it---destroy the lock.
 920                          */
 921                         lock->cll_flags |= CLF_CANCELPEND|CLF_DOOMED;
 922                 if (lock->cll_flags & CLF_CANCELPEND) {
 923                         lock->cll_flags &= ~CLF_CANCELPEND;
 924                         cl_lock_cancel0(env, lock);
 925                 }
 926                 if (lock->cll_flags & CLF_DOOMED) {
 927                         /* no longer doomed: it's dead... Jim. */
 928                         lock->cll_flags &= ~CLF_DOOMED;
 929                         cl_lock_delete0(env, lock);
 930                 }
 931         }
 932         EXIT;
 933 }
 934 EXPORT_SYMBOL(cl_lock_hold_release);
 935
 936 /**
 937  * Waits until lock state is changed.
 938  *
 939  * This function is called with cl_lock mutex locked, atomically releases
 940  * mutex and goes to sleep, waiting for a lock state change (signaled by
 941  * cl_lock_signal()), and re-acquires the mutex before return.
 942  *
 943  * This function is used to wait until lock state machine makes some progress
 944  * and to emulate synchronous operations on top of asynchronous lock
 945  * interface.
 946  *
 947  * \retval -EINTR wait was interrupted
 948  *
 949  * \retval 0 wait wasn't interrupted
 950  *
 951  * \pre cl_lock_is_mutexed(lock)
 952  *
 953  * \see cl_lock_signal()
 954  */
 955 int cl_lock_state_wait(const struct lu_env *env, struct cl_lock *lock)
 956 {
 957         wait_queue_t waiter;
 958         sigset_t blocked;
 959         int result;
 960
 961         ENTRY;
 962         LINVRNT(cl_lock_is_mutexed(lock));
 963         LINVRNT(cl_lock_invariant(env, lock));
 964         LASSERT(lock->cll_depth == 1);
 965         LASSERT(lock->cll_state != CLS_FREEING); /* too late to wait */
 966
 967         cl_lock_trace(D_DLMTRACE, env, "state wait lock", lock);
 968         result = lock->cll_error;
 969         if (result == 0) {
 970                 /* To avoid being interrupted by the 'non-fatal' signals
 971                  * (SIGCHLD, for instance), we'd block them temporarily.
 972                  * LU-305 */
 973                 blocked = cfs_block_sigsinv(LUSTRE_FATAL_SIGS);
 974
 975                 init_waitqueue_entry_current(&waiter);
 976                 add_wait_queue(&lock->cll_wq, &waiter);
 977                 set_current_state(TASK_INTERRUPTIBLE);
 978                 cl_lock_mutex_put(env, lock);
 979
 980                 LASSERT(cl_lock_nr_mutexed(env) == 0);
 981
 982                 /* Returning ERESTARTSYS instead of EINTR so syscalls
 983                  * can be restarted if signals are pending here */
 984                 result = -ERESTARTSYS;
 985                 if (likely(!OBD_FAIL_CHECK(OBD_FAIL_LOCK_STATE_WAIT_INTR))) {
 986                         waitq_wait(&waiter, TASK_INTERRUPTIBLE);
 987                         if (!cfs_signal_pending())
 988                                 result = 0;
 989                 }
 990
 991                 cl_lock_mutex_get(env, lock);
 992                 set_current_state(TASK_RUNNING);
 993                 remove_wait_queue(&lock->cll_wq, &waiter);
 994
 995                 /* Restore old blocked signals */
 996                 cfs_restore_sigs(blocked);
 997         }
 998         RETURN(result);
 999 }
1000 EXPORT_SYMBOL(cl_lock_state_wait);
1001
1002 static void cl_lock_state_signal(const struct lu_env *env, struct cl_lock *lock,
1003                                  enum cl_lock_state state)
1004 {
1005         const struct cl_lock_slice *slice;
1006
1007         ENTRY;
1008         LINVRNT(cl_lock_is_mutexed(lock));
1009         LINVRNT(cl_lock_invariant(env, lock));
1010
1011         list_for_each_entry(slice, &lock->cll_layers, cls_linkage)
1012                 if (slice->cls_ops->clo_state != NULL)
1013                         slice->cls_ops->clo_state(env, slice, state);
1014         wake_up_all(&lock->cll_wq);
1015         EXIT;
1016 }
1017
1018 /**
1019  * Notifies waiters that lock state changed.
1020  *
1021  * Wakes up all waiters sleeping in cl_lock_state_wait(), also notifies all
1022  * layers about state change by calling cl_lock_operations::clo_state()
1023  * top-to-bottom.
1024  */
1025 void cl_lock_signal(const struct lu_env *env, struct cl_lock *lock)
1026 {
1027         ENTRY;
1028         cl_lock_trace(D_DLMTRACE, env, "state signal lock", lock);
1029         cl_lock_state_signal(env, lock, lock->cll_state);
1030         EXIT;
1031 }
1032 EXPORT_SYMBOL(cl_lock_signal);
1033
1034 /**
1035  * Changes lock state.
1036  *
1037  * This function is invoked to notify layers that lock state changed, possible
1038  * as a result of an asynchronous event such as call-back reception.
1039  *
1040  * \post lock->cll_state == state
1041  *
1042  * \see cl_lock_operations::clo_state()
1043  */
1044 void cl_lock_state_set(const struct lu_env *env, struct cl_lock *lock,
1045                        enum cl_lock_state state)
1046 {
1047         ENTRY;
1048         LASSERT(lock->cll_state <= state ||
1049                 (lock->cll_state == CLS_CACHED &&
1050                  (state == CLS_HELD || /* lock found in cache */
1051                   state == CLS_NEW  ||   /* sub-lock canceled */
1052                   state == CLS_INTRANSIT)) ||
1053                 /* lock is in transit state */
1054                 lock->cll_state == CLS_INTRANSIT);
1055
1056         if (lock->cll_state != state) {
1057                 CS_LOCKSTATE_DEC(lock->cll_descr.cld_obj, lock->cll_state);
1058                 CS_LOCKSTATE_INC(lock->cll_descr.cld_obj, state);
1059
1060                 cl_lock_state_signal(env, lock, state);
1061                 lock->cll_state = state;
1062         }
1063         EXIT;
1064 }
1065 EXPORT_SYMBOL(cl_lock_state_set);
1066
1067 static int cl_unuse_try_internal(const struct lu_env *env, struct cl_lock *lock)
1068 {
1069         const struct cl_lock_slice *slice;
1070         int result;
1071
1072         do {
1073                 result = 0;
1074
1075                 LINVRNT(cl_lock_is_mutexed(lock));
1076                 LINVRNT(cl_lock_invariant(env, lock));
1077                 LASSERT(lock->cll_state == CLS_INTRANSIT);
1078
1079                 result = -ENOSYS;
1080                 list_for_each_entry_reverse(slice, &lock->cll_layers,
1081                                             cls_linkage) {
1082                         if (slice->cls_ops->clo_unuse != NULL) {
1083                                 result = slice->cls_ops->clo_unuse(env, slice);
1084                                 if (result != 0)
1085                                         break;
1086                         }
1087                 }
1088                 LASSERT(result != -ENOSYS);
1089         } while (result == CLO_REPEAT);
1090
1091         return result;
1092 }
1093
1094 /**
1095  * Yanks lock from the cache (cl_lock_state::CLS_CACHED state) by calling
1096  * cl_lock_operations::clo_use() top-to-bottom to notify layers.
1097  * @atomic = 1, it must unuse the lock to recovery the lock to keep the
1098  *  use process atomic
1099  */
1100 int cl_use_try(const struct lu_env *env, struct cl_lock *lock, int atomic)
1101 {
1102         const struct cl_lock_slice *slice;
1103         int result;
1104         enum cl_lock_state state;
1105
1106         ENTRY;
1107         cl_lock_trace(D_DLMTRACE, env, "use lock", lock);
1108
1109         LASSERT(lock->cll_state == CLS_CACHED);
1110         if (lock->cll_error)
1111                 RETURN(lock->cll_error);
1112
1113         result = -ENOSYS;
1114         state = cl_lock_intransit(env, lock);
1115         list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
1116                 if (slice->cls_ops->clo_use != NULL) {
1117                         result = slice->cls_ops->clo_use(env, slice);
1118                         if (result != 0)
1119                                 break;
1120                 }
1121         }
1122         LASSERT(result != -ENOSYS);
1123
1124         LASSERTF(lock->cll_state == CLS_INTRANSIT, "Wrong state %d.\n",
1125                  lock->cll_state);
1126
1127         if (result == 0) {
1128                 state = CLS_HELD;
1129         } else {
1130                 if (result == -ESTALE) {
1131                         /*
1132                          * ESTALE means sublock being cancelled
1133                          * at this time, and set lock state to
1134                          * be NEW here and ask the caller to repeat.
1135                          */
1136                         state = CLS_NEW;
1137                         result = CLO_REPEAT;
1138                 }
1139
1140                 /* @atomic means back-off-on-failure. */
1141                 if (atomic) {
1142                         int rc;
1143                         rc = cl_unuse_try_internal(env, lock);
1144                         /* Vet the results. */
1145                         if (rc < 0 && result > 0)
1146                                 result = rc;
1147                 }
1148
1149         }
1150         cl_lock_extransit(env, lock, state);
1151         RETURN(result);
1152 }
1153 EXPORT_SYMBOL(cl_use_try);
1154
1155 /**
1156  * Helper for cl_enqueue_try() that calls ->clo_enqueue() across all layers
1157  * top-to-bottom.
1158  */
1159 static int cl_enqueue_kick(const struct lu_env *env,
1160                            struct cl_lock *lock,
1161                            struct cl_io *io, __u32 flags)
1162 {
1163         int result;
1164         const struct cl_lock_slice *slice;
1165
1166         ENTRY;
1167         result = -ENOSYS;
1168         list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
1169                 if (slice->cls_ops->clo_enqueue != NULL) {
1170                         result = slice->cls_ops->clo_enqueue(env,
1171                                                              slice, io, flags);
1172                         if (result != 0)
1173                                 break;
1174                 }
1175         }
1176         LASSERT(result != -ENOSYS);
1177         RETURN(result);
1178 }
1179
1180 /**
1181  * Tries to enqueue a lock.
1182  *
1183  * This function is called repeatedly by cl_enqueue() until either lock is
1184  * enqueued, or error occurs. This function does not block waiting for
1185  * networking communication to complete.
1186  *
1187  * \post ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
1188  *                         lock->cll_state == CLS_HELD)
1189  *
1190  * \see cl_enqueue() cl_lock_operations::clo_enqueue()
1191  * \see cl_lock_state::CLS_ENQUEUED
1192  */
1193 int cl_enqueue_try(const struct lu_env *env, struct cl_lock *lock,
1194                    struct cl_io *io, __u32 flags)
1195 {
1196         int result;
1197
1198         ENTRY;
1199         cl_lock_trace(D_DLMTRACE, env, "enqueue lock", lock);
1200         do {
1201                 LINVRNT(cl_lock_is_mutexed(lock));
1202
1203                 result = lock->cll_error;
1204                 if (result != 0)
1205                         break;
1206
1207                 switch (lock->cll_state) {
1208                 case CLS_NEW:
1209                         cl_lock_state_set(env, lock, CLS_QUEUING);
1210                         /* fall-through */
1211                 case CLS_QUEUING:
1212                         /* kick layers. */
1213                         result = cl_enqueue_kick(env, lock, io, flags);
1214                         /* For AGL case, the cl_lock::cll_state may
1215                          * become CLS_HELD already. */
1216                         if (result == 0 && lock->cll_state == CLS_QUEUING)
1217                                 cl_lock_state_set(env, lock, CLS_ENQUEUED);
1218                         break;
1219                 case CLS_INTRANSIT:
1220                         LASSERT(cl_lock_is_intransit(lock));
1221                         result = CLO_WAIT;
1222                         break;
1223                 case CLS_CACHED:
1224                         /* yank lock from the cache. */
1225                         result = cl_use_try(env, lock, 0);
1226                         break;
1227                 case CLS_ENQUEUED:
1228                 case CLS_HELD:
1229                         result = 0;
1230                         break;
1231                 default:
1232                 case CLS_FREEING:
1233                         /*
1234                          * impossible, only held locks with increased
1235                          * ->cll_holds can be enqueued, and they cannot be
1236                          * freed.
1237                          */
1238                         LBUG();
1239                 }
1240         } while (result == CLO_REPEAT);
1241         RETURN(result);
1242 }
1243 EXPORT_SYMBOL(cl_enqueue_try);
1244
1245 /**
1246  * Cancel the conflicting lock found during previous enqueue.
1247  *
1248  * \retval 0 conflicting lock has been canceled.
1249  * \retval -ve error code.
1250  */
1251 int cl_lock_enqueue_wait(const struct lu_env *env,
1252                          struct cl_lock *lock,
1253                          int keep_mutex)
1254 {
1255         struct cl_lock  *conflict;
1256         int              rc = 0;
1257         ENTRY;
1258
1259         LASSERT(cl_lock_is_mutexed(lock));
1260         LASSERT(lock->cll_state == CLS_QUEUING);
1261         LASSERT(lock->cll_conflict != NULL);
1262
1263         conflict = lock->cll_conflict;
1264         lock->cll_conflict = NULL;
1265
1266         cl_lock_mutex_put(env, lock);
1267         LASSERT(cl_lock_nr_mutexed(env) == 0);
1268
1269         cl_lock_mutex_get(env, conflict);
1270         cl_lock_trace(D_DLMTRACE, env, "enqueue wait", conflict);
1271         cl_lock_cancel(env, conflict);
1272         cl_lock_delete(env, conflict);
1273
1274         while (conflict->cll_state != CLS_FREEING) {
1275                 rc = cl_lock_state_wait(env, conflict);
1276                 if (rc != 0)
1277                         break;
1278         }
1279         cl_lock_mutex_put(env, conflict);
1280         lu_ref_del(&conflict->cll_reference, "cancel-wait", lock);
1281         cl_lock_put(env, conflict);
1282
1283         if (keep_mutex)
1284                 cl_lock_mutex_get(env, lock);
1285
1286         LASSERT(rc <= 0);
1287         RETURN(rc);
1288 }
1289 EXPORT_SYMBOL(cl_lock_enqueue_wait);
1290
1291 static int cl_enqueue_locked(const struct lu_env *env, struct cl_lock *lock,
1292                              struct cl_io *io, __u32 enqflags)
1293 {
1294         int result;
1295
1296         ENTRY;
1297
1298         LINVRNT(cl_lock_is_mutexed(lock));
1299         LINVRNT(cl_lock_invariant(env, lock));
1300         LASSERT(lock->cll_holds > 0);
1301
1302         cl_lock_user_add(env, lock);
1303         do {
1304                 result = cl_enqueue_try(env, lock, io, enqflags);
1305                 if (result == CLO_WAIT) {
1306                         if (lock->cll_conflict != NULL)
1307                                 result = cl_lock_enqueue_wait(env, lock, 1);
1308                         else
1309                                 result = cl_lock_state_wait(env, lock);
1310                         if (result == 0)
1311                                 continue;
1312                 }
1313                 break;
1314         } while (1);
1315         if (result != 0)
1316                 cl_unuse_try(env, lock);
1317         LASSERT(ergo(result == 0 && !(enqflags & CEF_AGL),
1318                      lock->cll_state == CLS_ENQUEUED ||
1319                      lock->cll_state == CLS_HELD));
1320         RETURN(result);
1321 }
1322
1323 /**
1324  * Enqueues a lock.
1325  *
1326  * \pre current thread or io owns a hold on lock.
1327  *
1328  * \post ergo(result == 0, lock->users increased)
1329  * \post ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
1330  *                         lock->cll_state == CLS_HELD)
1331  */
1332 int cl_enqueue(const struct lu_env *env, struct cl_lock *lock,
1333                struct cl_io *io, __u32 enqflags)
1334 {
1335         int result;
1336
1337         ENTRY;
1338
1339         cl_lock_lockdep_acquire(env, lock, enqflags);
1340         cl_lock_mutex_get(env, lock);
1341         result = cl_enqueue_locked(env, lock, io, enqflags);
1342         cl_lock_mutex_put(env, lock);
1343         if (result != 0)
1344                 cl_lock_lockdep_release(env, lock);
1345         LASSERT(ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
1346                      lock->cll_state == CLS_HELD));
1347         RETURN(result);
1348 }
1349 EXPORT_SYMBOL(cl_enqueue);
1350
1351 /**
1352  * Tries to unlock a lock.
1353  *
1354  * This function is called to release underlying resource:
1355  * 1. for top lock, the resource is sublocks it held;
1356  * 2. for sublock, the resource is the reference to dlmlock.
1357  *
1358  * cl_unuse_try is a one-shot operation, so it must NOT return CLO_WAIT.
1359  *
1360  * \see cl_unuse() cl_lock_operations::clo_unuse()
1361  * \see cl_lock_state::CLS_CACHED
1362  */
1363 int cl_unuse_try(const struct lu_env *env, struct cl_lock *lock)
1364 {
1365         int                         result;
1366         enum cl_lock_state          state = CLS_NEW;
1367
1368         ENTRY;
1369         cl_lock_trace(D_DLMTRACE, env, "unuse lock", lock);
1370
1371         if (lock->cll_users > 1) {
1372                 cl_lock_user_del(env, lock);
1373                 RETURN(0);
1374         }
1375
1376         /* Only if the lock is in CLS_HELD or CLS_ENQUEUED state, it can hold
1377          * underlying resources. */
1378         if (!(lock->cll_state == CLS_HELD || lock->cll_state == CLS_ENQUEUED)) {
1379                 cl_lock_user_del(env, lock);
1380                 RETURN(0);
1381         }
1382
1383         /*
1384          * New lock users (->cll_users) are not protecting unlocking
1385          * from proceeding. From this point, lock eventually reaches
1386          * CLS_CACHED, is reinitialized to CLS_NEW or fails into
1387          * CLS_FREEING.
1388          */
1389         state = cl_lock_intransit(env, lock);
1390
1391         result = cl_unuse_try_internal(env, lock);
1392         LASSERT(lock->cll_state == CLS_INTRANSIT);
1393         LASSERT(result != CLO_WAIT);
1394         cl_lock_user_del(env, lock);
1395         if (result == 0 || result == -ESTALE) {
1396                 /*
1397                  * Return lock back to the cache. This is the only
1398                  * place where lock is moved into CLS_CACHED state.
1399                  *
1400                  * If one of ->clo_unuse() methods returned -ESTALE, lock
1401                  * cannot be placed into cache and has to be
1402                  * re-initialized. This happens e.g., when a sub-lock was
1403                  * canceled while unlocking was in progress.
1404                  */
1405                 if (state == CLS_HELD && result == 0)
1406                         state = CLS_CACHED;
1407                 else
1408                         state = CLS_NEW;
1409                 cl_lock_extransit(env, lock, state);
1410
1411                 /*
1412                  * Hide -ESTALE error.
1413                  * If the lock is a glimpse lock, and it has multiple
1414                  * stripes. Assuming that one of its sublock returned -ENAVAIL,
1415                  * and other sublocks are matched write locks. In this case,
1416                  * we can't set this lock to error because otherwise some of
1417                  * its sublocks may not be canceled. This causes some dirty
1418                  * pages won't be written to OSTs. -jay
1419                  */
1420                 result = 0;
1421         } else {
1422                 CERROR("result = %d, this is unlikely!\n", result);
1423                 state = CLS_NEW;
1424                 cl_lock_extransit(env, lock, state);
1425         }
1426         RETURN(result ?: lock->cll_error);
1427 }
1428 EXPORT_SYMBOL(cl_unuse_try);
1429
1430 static void cl_unuse_locked(const struct lu_env *env, struct cl_lock *lock)
1431 {
1432         int result;
1433         ENTRY;
1434
1435         result = cl_unuse_try(env, lock);
1436         if (result)
1437                 CL_LOCK_DEBUG(D_ERROR, env, lock, "unuse return %d\n", result);
1438
1439         EXIT;
1440 }
1441
1442 /**
1443  * Unlocks a lock.
1444  */
1445 void cl_unuse(const struct lu_env *env, struct cl_lock *lock)
1446 {
1447         ENTRY;
1448         cl_lock_mutex_get(env, lock);
1449         cl_unuse_locked(env, lock);
1450         cl_lock_mutex_put(env, lock);
1451         cl_lock_lockdep_release(env, lock);
1452         EXIT;
1453 }
1454 EXPORT_SYMBOL(cl_unuse);
1455
1456 /**
1457  * Tries to wait for a lock.
1458  *
1459  * This function is called repeatedly by cl_wait() until either lock is
1460  * granted, or error occurs. This function does not block waiting for network
1461  * communication to complete.
1462  *
1463  * \see cl_wait() cl_lock_operations::clo_wait()
1464  * \see cl_lock_state::CLS_HELD
1465  */
1466 int cl_wait_try(const struct lu_env *env, struct cl_lock *lock)
1467 {
1468         const struct cl_lock_slice *slice;
1469         int                         result;
1470
1471         ENTRY;
1472         cl_lock_trace(D_DLMTRACE, env, "wait lock try", lock);
1473         do {
1474                 LINVRNT(cl_lock_is_mutexed(lock));
1475                 LINVRNT(cl_lock_invariant(env, lock));
1476                 LASSERTF(lock->cll_state == CLS_QUEUING ||
1477                          lock->cll_state == CLS_ENQUEUED ||
1478                          lock->cll_state == CLS_HELD ||
1479                          lock->cll_state == CLS_INTRANSIT,
1480                          "lock state: %d\n", lock->cll_state);
1481                 LASSERT(lock->cll_users > 0);
1482                 LASSERT(lock->cll_holds > 0);
1483
1484                 result = lock->cll_error;
1485                 if (result != 0)
1486                         break;
1487
1488                 if (cl_lock_is_intransit(lock)) {
1489                         result = CLO_WAIT;
1490                         break;
1491                 }
1492
1493                 if (lock->cll_state == CLS_HELD)
1494                         /* nothing to do */
1495                         break;
1496
1497                 result = -ENOSYS;
1498                 list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
1499                         if (slice->cls_ops->clo_wait != NULL) {
1500                                 result = slice->cls_ops->clo_wait(env, slice);
1501                                 if (result != 0)
1502                                         break;
1503                         }
1504                 }
1505                 LASSERT(result != -ENOSYS);
1506                 if (result == 0) {
1507                         LASSERT(lock->cll_state != CLS_INTRANSIT);
1508                         cl_lock_state_set(env, lock, CLS_HELD);
1509                 }
1510         } while (result == CLO_REPEAT);
1511         RETURN(result);
1512 }
1513 EXPORT_SYMBOL(cl_wait_try);
1514
1515 /**
1516  * Waits until enqueued lock is granted.
1517  *
1518  * \pre current thread or io owns a hold on the lock
1519  * \pre ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
1520  *                        lock->cll_state == CLS_HELD)
1521  *
1522  * \post ergo(result == 0, lock->cll_state == CLS_HELD)
1523  */
1524 int cl_wait(const struct lu_env *env, struct cl_lock *lock)
1525 {
1526         int result;
1527
1528         ENTRY;
1529         cl_lock_mutex_get(env, lock);
1530
1531         LINVRNT(cl_lock_invariant(env, lock));
1532         LASSERTF(lock->cll_state == CLS_ENQUEUED || lock->cll_state == CLS_HELD,
1533                  "Wrong state %d \n", lock->cll_state);
1534         LASSERT(lock->cll_holds > 0);
1535
1536         do {
1537                 result = cl_wait_try(env, lock);
1538                 if (result == CLO_WAIT) {
1539                         result = cl_lock_state_wait(env, lock);
1540                         if (result == 0)
1541                                 continue;
1542                 }
1543                 break;
1544         } while (1);
1545         if (result < 0) {
1546                 cl_unuse_try(env, lock);
1547                 cl_lock_lockdep_release(env, lock);
1548         }
1549         cl_lock_trace(D_DLMTRACE, env, "wait lock", lock);
1550         cl_lock_mutex_put(env, lock);
1551         LASSERT(ergo(result == 0, lock->cll_state == CLS_HELD));
1552         RETURN(result);
1553 }
1554 EXPORT_SYMBOL(cl_wait);
1555
1556 /**
1557  * Executes cl_lock_operations::clo_weigh(), and sums results to estimate lock
1558  * value.
1559  */
1560 unsigned long cl_lock_weigh(const struct lu_env *env, struct cl_lock *lock)
1561 {
1562         const struct cl_lock_slice *slice;
1563         unsigned long pound;
1564         unsigned long ounce;
1565
1566         ENTRY;
1567         LINVRNT(cl_lock_is_mutexed(lock));
1568         LINVRNT(cl_lock_invariant(env, lock));
1569
1570         pound = 0;
1571         list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
1572                 if (slice->cls_ops->clo_weigh != NULL) {
1573                         ounce = slice->cls_ops->clo_weigh(env, slice);
1574                         pound += ounce;
1575                         if (pound < ounce) /* over-weight^Wflow */
1576                                 pound = ~0UL;
1577                 }
1578         }
1579         RETURN(pound);
1580 }
1581 EXPORT_SYMBOL(cl_lock_weigh);
1582
1583 /**
1584  * Notifies layers that lock description changed.
1585  *
1586  * The server can grant client a lock different from one that was requested
1587  * (e.g., larger in extent). This method is called when actually granted lock
1588  * description becomes known to let layers to accommodate for changed lock
1589  * description.
1590  *
1591  * \see cl_lock_operations::clo_modify()
1592  */
1593 int cl_lock_modify(const struct lu_env *env, struct cl_lock *lock,
1594                    const struct cl_lock_descr *desc)
1595 {
1596         const struct cl_lock_slice *slice;
1597         struct cl_object           *obj = lock->cll_descr.cld_obj;
1598         struct cl_object_header    *hdr = cl_object_header(obj);
1599         int result;
1600
1601         ENTRY;
1602         cl_lock_trace(D_DLMTRACE, env, "modify lock", lock);
1603         /* don't allow object to change */
1604         LASSERT(obj == desc->cld_obj);
1605         LINVRNT(cl_lock_is_mutexed(lock));
1606         LINVRNT(cl_lock_invariant(env, lock));
1607
1608         list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
1609                 if (slice->cls_ops->clo_modify != NULL) {
1610                         result = slice->cls_ops->clo_modify(env, slice, desc);
1611                         if (result != 0)
1612                                 RETURN(result);
1613                 }
1614         }
1615         CL_LOCK_DEBUG(D_DLMTRACE, env, lock, " -> "DDESCR"@"DFID"\n",
1616                       PDESCR(desc), PFID(lu_object_fid(&desc->cld_obj->co_lu)));
1617         /*
1618          * Just replace description in place. Nothing more is needed for
1619          * now. If locks were indexed according to their extent and/or mode,
1620          * that index would have to be updated here.
1621          */
1622         spin_lock(&hdr->coh_lock_guard);
1623         lock->cll_descr = *desc;
1624         spin_unlock(&hdr->coh_lock_guard);
1625         RETURN(0);
1626 }
1627 EXPORT_SYMBOL(cl_lock_modify);
1628
1629 /**
1630  * Initializes lock closure with a given origin.
1631  *
1632  * \see cl_lock_closure
1633  */
1634 void cl_lock_closure_init(const struct lu_env *env,
1635                           struct cl_lock_closure *closure,
1636                           struct cl_lock *origin, int wait)
1637 {
1638         LINVRNT(cl_lock_is_mutexed(origin));
1639         LINVRNT(cl_lock_invariant(env, origin));
1640
1641         INIT_LIST_HEAD(&closure->clc_list);
1642         closure->clc_origin = origin;
1643         closure->clc_wait   = wait;
1644         closure->clc_nr     = 0;
1645 }
1646 EXPORT_SYMBOL(cl_lock_closure_init);
1647
1648 /**
1649  * Builds a closure of \a lock.
1650  *
1651  * Building of a closure consists of adding initial lock (\a lock) into it,
1652  * and calling cl_lock_operations::clo_closure() methods of \a lock. These
1653  * methods might call cl_lock_closure_build() recursively again, adding more
1654  * locks to the closure, etc.
1655  *
1656  * \see cl_lock_closure
1657  */
1658 int cl_lock_closure_build(const struct lu_env *env, struct cl_lock *lock,
1659                           struct cl_lock_closure *closure)
1660 {
1661         const struct cl_lock_slice *slice;
1662         int result;
1663
1664         ENTRY;
1665         LINVRNT(cl_lock_is_mutexed(closure->clc_origin));
1666         LINVRNT(cl_lock_invariant(env, closure->clc_origin));
1667
1668         result = cl_lock_enclosure(env, lock, closure);
1669         if (result == 0) {
1670                 list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
1671                         if (slice->cls_ops->clo_closure != NULL) {
1672                                 result = slice->cls_ops->clo_closure(env, slice,
1673                                                                      closure);
1674                                 if (result != 0)
1675                                         break;
1676                         }
1677                 }
1678         }
1679         if (result != 0)
1680                 cl_lock_disclosure(env, closure);
1681         RETURN(result);
1682 }
1683 EXPORT_SYMBOL(cl_lock_closure_build);
1684
1685 /**
1686  * Adds new lock to a closure.
1687  *
1688  * Try-locks \a lock and if succeeded, adds it to the closure (never more than
1689  * once). If try-lock failed, returns CLO_REPEAT, after optionally waiting
1690  * until next try-lock is likely to succeed.
1691  */
1692 int cl_lock_enclosure(const struct lu_env *env, struct cl_lock *lock,
1693                       struct cl_lock_closure *closure)
1694 {
1695         int result = 0;
1696         ENTRY;
1697         cl_lock_trace(D_DLMTRACE, env, "enclosure lock", lock);
1698         if (!cl_lock_mutex_try(env, lock)) {
1699                 /*
1700                  * If lock->cll_inclosure is not empty, lock is already in
1701                  * this closure.
1702                  */
1703                 if (list_empty(&lock->cll_inclosure)) {
1704                         cl_lock_get_trust(lock);
1705                         lu_ref_add(&lock->cll_reference, "closure", closure);
1706                         list_add(&lock->cll_inclosure, &closure->clc_list);
1707                         closure->clc_nr++;
1708                 } else
1709                         cl_lock_mutex_put(env, lock);
1710                 result = 0;
1711         } else {
1712                 cl_lock_disclosure(env, closure);
1713                 if (closure->clc_wait) {
1714                         cl_lock_get_trust(lock);
1715                         lu_ref_add(&lock->cll_reference, "closure-w", closure);
1716                         cl_lock_mutex_put(env, closure->clc_origin);
1717
1718                         LASSERT(cl_lock_nr_mutexed(env) == 0);
1719                         cl_lock_mutex_get(env, lock);
1720                         cl_lock_mutex_put(env, lock);
1721
1722                         cl_lock_mutex_get(env, closure->clc_origin);
1723                         lu_ref_del(&lock->cll_reference, "closure-w", closure);
1724                         cl_lock_put(env, lock);
1725                 }
1726                 result = CLO_REPEAT;
1727         }
1728         RETURN(result);
1729 }
1730 EXPORT_SYMBOL(cl_lock_enclosure);
1731
1732 /** Releases mutices of enclosed locks. */
1733 void cl_lock_disclosure(const struct lu_env *env,
1734                         struct cl_lock_closure *closure)
1735 {
1736         struct cl_lock *scan;
1737         struct cl_lock *temp;
1738
1739         cl_lock_trace(D_DLMTRACE, env, "disclosure lock", closure->clc_origin);
1740         list_for_each_entry_safe(scan, temp, &closure->clc_list,
1741                                  cll_inclosure){
1742                 list_del_init(&scan->cll_inclosure);
1743                 cl_lock_mutex_put(env, scan);
1744                 lu_ref_del(&scan->cll_reference, "closure", closure);
1745                 cl_lock_put(env, scan);
1746                 closure->clc_nr--;
1747         }
1748         LASSERT(closure->clc_nr == 0);
1749 }
1750 EXPORT_SYMBOL(cl_lock_disclosure);
1751
1752 /** Finalizes a closure. */
1753 void cl_lock_closure_fini(struct cl_lock_closure *closure)
1754 {
1755         LASSERT(closure->clc_nr == 0);
1756         LASSERT(list_empty(&closure->clc_list));
1757 }
1758 EXPORT_SYMBOL(cl_lock_closure_fini);
1759
1760 /**
1761  * Destroys this lock. Notifies layers (bottom-to-top) that lock is being
1762  * destroyed, then destroy the lock. If there are holds on the lock, postpone
1763  * destruction until all holds are released. This is called when a decision is
1764  * made to destroy the lock in the future. E.g., when a blocking AST is
1765  * received on it, or fatal communication error happens.
1766  *
1767  * Caller must have a reference on this lock to prevent a situation, when
1768  * deleted lock lingers in memory for indefinite time, because nobody calls
1769  * cl_lock_put() to finish it.
1770  *
1771  * \pre atomic_read(&lock->cll_ref) > 0
1772  * \pre ergo(cl_lock_nesting(lock) == CNL_TOP,
1773  *           cl_lock_nr_mutexed(env) == 1)
1774  *      [i.e., if a top-lock is deleted, mutices of no other locks can be
1775  *      held, as deletion of sub-locks might require releasing a top-lock
1776  *      mutex]
1777  *
1778  * \see cl_lock_operations::clo_delete()
1779  * \see cl_lock::cll_holds
1780  */
1781 void cl_lock_delete(const struct lu_env *env, struct cl_lock *lock)
1782 {
1783         LINVRNT(cl_lock_is_mutexed(lock));
1784         LINVRNT(cl_lock_invariant(env, lock));
1785         LASSERT(ergo(cl_lock_nesting(lock) == CNL_TOP,
1786                      cl_lock_nr_mutexed(env) == 1));
1787
1788         ENTRY;
1789         cl_lock_trace(D_DLMTRACE, env, "delete lock", lock);
1790         if (lock->cll_holds == 0)
1791                 cl_lock_delete0(env, lock);
1792         else
1793                 lock->cll_flags |= CLF_DOOMED;
1794         EXIT;
1795 }
1796 EXPORT_SYMBOL(cl_lock_delete);
1797
1798 /**
1799  * Mark lock as irrecoverably failed, and mark it for destruction. This
1800  * happens when, e.g., server fails to grant a lock to us, or networking
1801  * time-out happens.
1802  *
1803  * \pre atomic_read(&lock->cll_ref) > 0
1804  *
1805  * \see clo_lock_delete()
1806  * \see cl_lock::cll_holds
1807  */
1808 void cl_lock_error(const struct lu_env *env, struct cl_lock *lock, int error)
1809 {
1810         LINVRNT(cl_lock_is_mutexed(lock));
1811         LINVRNT(cl_lock_invariant(env, lock));
1812
1813         ENTRY;
1814         if (lock->cll_error == 0 && error != 0) {
1815                 cl_lock_trace(D_DLMTRACE, env, "set lock error", lock);
1816                 lock->cll_error = error;
1817                 cl_lock_signal(env, lock);
1818                 cl_lock_cancel(env, lock);
1819                 cl_lock_delete(env, lock);
1820         }
1821         EXIT;
1822 }
1823 EXPORT_SYMBOL(cl_lock_error);
1824
1825 /**
1826  * Cancels this lock. Notifies layers
1827  * (bottom-to-top) that lock is being cancelled, then destroy the lock. If
1828  * there are holds on the lock, postpone cancellation until
1829  * all holds are released.
1830  *
1831  * Cancellation notification is delivered to layers at most once.
1832  *
1833  * \see cl_lock_operations::clo_cancel()
1834  * \see cl_lock::cll_holds
1835  */
1836 void cl_lock_cancel(const struct lu_env *env, struct cl_lock *lock)
1837 {
1838         LINVRNT(cl_lock_is_mutexed(lock));
1839         LINVRNT(cl_lock_invariant(env, lock));
1840
1841         ENTRY;
1842         cl_lock_trace(D_DLMTRACE, env, "cancel lock", lock);
1843         if (lock->cll_holds == 0)
1844                 cl_lock_cancel0(env, lock);
1845         else
1846                 lock->cll_flags |= CLF_CANCELPEND;
1847         EXIT;
1848 }
1849 EXPORT_SYMBOL(cl_lock_cancel);
1850
1851 /**
1852  * Finds an existing lock covering given index and optionally different from a
1853  * given \a except lock.
1854  */
1855 struct cl_lock *cl_lock_at_pgoff(const struct lu_env *env,
1856                                  struct cl_object *obj, pgoff_t index,
1857                                  struct cl_lock *except,
1858                                  int pending, int canceld)
1859 {
1860         struct cl_object_header *head;
1861         struct cl_lock          *scan;
1862         struct cl_lock          *lock;
1863         struct cl_lock_descr    *need;
1864
1865         ENTRY;
1866
1867         head = cl_object_header(obj);
1868         need = &cl_env_info(env)->clt_descr;
1869         lock = NULL;
1870
1871         need->cld_mode = CLM_READ; /* CLM_READ matches both READ & WRITE, but
1872                                     * not PHANTOM */
1873         need->cld_start = need->cld_end = index;
1874         need->cld_enq_flags = 0;
1875
1876         spin_lock(&head->coh_lock_guard);
1877         /* It is fine to match any group lock since there could be only one
1878          * with a uniq gid and it conflicts with all other lock modes too */
1879         list_for_each_entry(scan, &head->coh_locks, cll_linkage) {
1880                 if (scan != except &&
1881                     (scan->cll_descr.cld_mode == CLM_GROUP ||
1882                     cl_lock_ext_match(&scan->cll_descr, need)) &&
1883                     scan->cll_state >= CLS_HELD &&
1884                     scan->cll_state < CLS_FREEING &&
1885                     /*
1886                      * This check is racy as the lock can be canceled right
1887                      * after it is done, but this is fine, because page exists
1888                      * already.
1889                      */
1890                     (canceld || !(scan->cll_flags & CLF_CANCELLED)) &&
1891                     (pending || !(scan->cll_flags & CLF_CANCELPEND))) {
1892                         /* Don't increase cs_hit here since this
1893                          * is just a helper function. */
1894                         cl_lock_get_trust(scan);
1895                         lock = scan;
1896                         break;
1897                 }
1898         }
1899         spin_unlock(&head->coh_lock_guard);
1900         RETURN(lock);
1901 }
1902 EXPORT_SYMBOL(cl_lock_at_pgoff);
1903
1904 /**
1905  * Eliminate all locks for a given object.
1906  *
1907  * Caller has to guarantee that no lock is in active use.
1908  *
1909  * \param cancel when this is set, cl_locks_prune() cancels locks before
1910  *               destroying.
1911  */
1912 void cl_locks_prune(const struct lu_env *env, struct cl_object *obj, int cancel)
1913 {
1914         struct cl_object_header *head;
1915         struct cl_lock          *lock;
1916
1917         ENTRY;
1918         head = cl_object_header(obj);
1919
1920         spin_lock(&head->coh_lock_guard);
1921         while (!list_empty(&head->coh_locks)) {
1922                 lock = container_of(head->coh_locks.next,
1923                                     struct cl_lock, cll_linkage);
1924                 cl_lock_get_trust(lock);
1925                 spin_unlock(&head->coh_lock_guard);
1926                 lu_ref_add(&lock->cll_reference, "prune", current);
1927
1928 again:
1929                 cl_lock_mutex_get(env, lock);
1930                 if (lock->cll_state < CLS_FREEING) {
1931                         LASSERT(lock->cll_users <= 1);
1932                         if (unlikely(lock->cll_users == 1)) {
1933                                 struct l_wait_info lwi = { 0 };
1934
1935                                 cl_lock_mutex_put(env, lock);
1936                                 l_wait_event(lock->cll_wq,
1937                                              lock->cll_users == 0,
1938                                              &lwi);
1939                                 goto again;
1940                         }
1941
1942                         if (cancel)
1943                                 cl_lock_cancel(env, lock);
1944                         cl_lock_delete(env, lock);
1945                 }
1946                 cl_lock_mutex_put(env, lock);
1947                 lu_ref_del(&lock->cll_reference, "prune", current);
1948                 cl_lock_put(env, lock);
1949                 spin_lock(&head->coh_lock_guard);
1950         }
1951         spin_unlock(&head->coh_lock_guard);
1952         EXIT;
1953 }
1954 EXPORT_SYMBOL(cl_locks_prune);
1955
1956 static struct cl_lock *cl_lock_hold_mutex(const struct lu_env *env,
1957                                           const struct cl_io *io,
1958                                           const struct cl_lock_descr *need,
1959                                           const char *scope, const void *source)
1960 {
1961         struct cl_lock *lock;
1962
1963         ENTRY;
1964
1965         while (1) {
1966                 lock = cl_lock_find(env, io, need);
1967                 if (IS_ERR(lock))
1968                         break;
1969                 cl_lock_mutex_get(env, lock);
1970                 if (lock->cll_state < CLS_FREEING &&
1971                     !(lock->cll_flags & CLF_CANCELLED)) {
1972                         cl_lock_hold_mod(env, lock, +1);
1973                         lu_ref_add(&lock->cll_holders, scope, source);
1974                         lu_ref_add(&lock->cll_reference, scope, source);
1975                         break;
1976                 }
1977                 cl_lock_mutex_put(env, lock);
1978                 cl_lock_put(env, lock);
1979         }
1980         RETURN(lock);
1981 }
1982
1983 /**
1984  * Returns a lock matching \a need description with a reference and a hold on
1985  * it.
1986  *
1987  * This is much like cl_lock_find(), except that cl_lock_hold() additionally
1988  * guarantees that lock is not in the CLS_FREEING state on return.
1989  */
1990 struct cl_lock *cl_lock_hold(const struct lu_env *env, const struct cl_io *io,
1991                              const struct cl_lock_descr *need,
1992                              const char *scope, const void *source)
1993 {
1994         struct cl_lock *lock;
1995
1996         ENTRY;
1997
1998         lock = cl_lock_hold_mutex(env, io, need, scope, source);
1999         if (!IS_ERR(lock))
2000                 cl_lock_mutex_put(env, lock);
2001         RETURN(lock);
2002 }
2003 EXPORT_SYMBOL(cl_lock_hold);
2004
2005 /**
2006  * Main high-level entry point of cl_lock interface that finds existing or
2007  * enqueues new lock matching given description.
2008  */
2009 struct cl_lock *cl_lock_request(const struct lu_env *env, struct cl_io *io,
2010                                 const struct cl_lock_descr *need,
2011                                 const char *scope, const void *source)
2012 {
2013         struct cl_lock       *lock;
2014         int                   rc;
2015         __u32                 enqflags = need->cld_enq_flags;
2016
2017         ENTRY;
2018         do {
2019                 lock = cl_lock_hold_mutex(env, io, need, scope, source);
2020                 if (IS_ERR(lock))
2021                         break;
2022
2023                 rc = cl_enqueue_locked(env, lock, io, enqflags);
2024                 if (rc == 0) {
2025                         if (cl_lock_fits_into(env, lock, need, io)) {
2026                                 if (!(enqflags & CEF_AGL)) {
2027                                         cl_lock_mutex_put(env, lock);
2028                                         cl_lock_lockdep_acquire(env, lock,
2029                                                                 enqflags);
2030                                         break;
2031                                 }
2032                                 rc = 1;
2033                         }
2034                         cl_unuse_locked(env, lock);
2035                 }
2036                 cl_lock_trace(D_DLMTRACE, env,
2037                               rc <= 0 ? "enqueue failed" : "agl succeed", lock);
2038                 cl_lock_hold_release(env, lock, scope, source);
2039                 cl_lock_mutex_put(env, lock);
2040                 lu_ref_del(&lock->cll_reference, scope, source);
2041                 cl_lock_put(env, lock);
2042                 if (rc > 0) {
2043                         LASSERT(enqflags & CEF_AGL);
2044                         lock = NULL;
2045                 } else if (rc != 0) {
2046                         lock = ERR_PTR(rc);
2047                 }
2048         } while (rc == 0);
2049         RETURN(lock);
2050 }
2051 EXPORT_SYMBOL(cl_lock_request);
2052
2053 /**
2054  * Adds a hold to a known lock.
2055  */
2056 void cl_lock_hold_add(const struct lu_env *env, struct cl_lock *lock,
2057                       const char *scope, const void *source)
2058 {
2059         LINVRNT(cl_lock_is_mutexed(lock));
2060         LINVRNT(cl_lock_invariant(env, lock));
2061         LASSERT(lock->cll_state != CLS_FREEING);
2062
2063         ENTRY;
2064         cl_lock_get(lock);
2065         cl_lock_hold_mod(env, lock, +1);
2066         lu_ref_add(&lock->cll_holders, scope, source);
2067         lu_ref_add(&lock->cll_reference, scope, source);
2068         EXIT;
2069 }
2070 EXPORT_SYMBOL(cl_lock_hold_add);
2071
2072 /**
2073  * Releases a hold and a reference on a lock, on which caller acquired a
2074  * mutex.
2075  */
2076 void cl_lock_unhold(const struct lu_env *env, struct cl_lock *lock,
2077                     const char *scope, const void *source)
2078 {
2079         LINVRNT(cl_lock_invariant(env, lock));
2080         ENTRY;
2081         cl_lock_hold_release(env, lock, scope, source);
2082         lu_ref_del(&lock->cll_reference, scope, source);
2083         cl_lock_put(env, lock);
2084         EXIT;
2085 }
2086 EXPORT_SYMBOL(cl_lock_unhold);
2087
2088 /**
2089  * Releases a hold and a reference on a lock, obtained by cl_lock_hold().
2090  */
2091 void cl_lock_release(const struct lu_env *env, struct cl_lock *lock,
2092                      const char *scope, const void *source)
2093 {
2094         LINVRNT(cl_lock_invariant(env, lock));
2095         ENTRY;
2096         cl_lock_trace(D_DLMTRACE, env, "release lock", lock);
2097         cl_lock_mutex_get(env, lock);
2098         cl_lock_hold_release(env, lock, scope, source);
2099         cl_lock_mutex_put(env, lock);
2100         lu_ref_del(&lock->cll_reference, scope, source);
2101         cl_lock_put(env, lock);
2102         EXIT;
2103 }
2104 EXPORT_SYMBOL(cl_lock_release);
2105
2106 void cl_lock_user_add(const struct lu_env *env, struct cl_lock *lock)
2107 {
2108         LINVRNT(cl_lock_is_mutexed(lock));
2109         LINVRNT(cl_lock_invariant(env, lock));
2110
2111         ENTRY;
2112         cl_lock_used_mod(env, lock, +1);
2113         EXIT;
2114 }
2115 EXPORT_SYMBOL(cl_lock_user_add);
2116
2117 void cl_lock_user_del(const struct lu_env *env, struct cl_lock *lock)
2118 {
2119         LINVRNT(cl_lock_is_mutexed(lock));
2120         LINVRNT(cl_lock_invariant(env, lock));
2121         LASSERT(lock->cll_users > 0);
2122
2123         ENTRY;
2124         cl_lock_used_mod(env, lock, -1);
2125         if (lock->cll_users == 0)
2126                 wake_up_all(&lock->cll_wq);
2127         EXIT;
2128 }
2129 EXPORT_SYMBOL(cl_lock_user_del);
2130
2131 const char *cl_lock_mode_name(const enum cl_lock_mode mode)
2132 {
2133         static const char *names[] = {
2134                 [CLM_PHANTOM] = "P",
2135                 [CLM_READ]    = "R",
2136                 [CLM_WRITE]   = "W",
2137                 [CLM_GROUP]   = "G"
2138         };
2139         CLASSERT(CLM_MAX == ARRAY_SIZE(names));
2140         return names[mode];
2141 }
2142 EXPORT_SYMBOL(cl_lock_mode_name);
2143
2144 /**
2145  * Prints human readable representation of a lock description.
2146  */
2147 void cl_lock_descr_print(const struct lu_env *env, void *cookie,
2148                        lu_printer_t printer,
2149                        const struct cl_lock_descr *descr)
2150 {
2151         const struct lu_fid  *fid;
2152
2153         fid = lu_object_fid(&descr->cld_obj->co_lu);
2154         (*printer)(env, cookie, DDESCR"@"DFID, PDESCR(descr), PFID(fid));
2155 }
2156 EXPORT_SYMBOL(cl_lock_descr_print);
2157
2158 /**
2159  * Prints human readable representation of \a lock to the \a f.
2160  */
2161 void cl_lock_print(const struct lu_env *env, void *cookie,
2162                    lu_printer_t printer, const struct cl_lock *lock)
2163 {
2164         const struct cl_lock_slice *slice;
2165         (*printer)(env, cookie, "lock@%p[%d %d %d %d %d %08lx] ",
2166                    lock, atomic_read(&lock->cll_ref),
2167                    lock->cll_state, lock->cll_error, lock->cll_holds,
2168                    lock->cll_users, lock->cll_flags);
2169         cl_lock_descr_print(env, cookie, printer, &lock->cll_descr);
2170         (*printer)(env, cookie, " {\n");
2171
2172         list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
2173                 (*printer)(env, cookie, "    %s@%p: ",
2174                            slice->cls_obj->co_lu.lo_dev->ld_type->ldt_name,
2175                            slice);
2176                 if (slice->cls_ops->clo_print != NULL)
2177                         slice->cls_ops->clo_print(env, cookie, printer, slice);
2178                 (*printer)(env, cookie, "\n");
2179         }
2180         (*printer)(env, cookie, "} lock@%p\n", lock);
2181 }
2182 EXPORT_SYMBOL(cl_lock_print);
2183
2184 int cl_lock_init(void)
2185 {
2186         return lu_kmem_init(cl_lock_caches);
2187 }
2188
2189 void cl_lock_fini(void)
2190 {
2191         lu_kmem_fini(cl_lock_caches);
2192 }