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