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