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