Whamcloud - gitweb
LU-948 clio: add a callback to cl_page_gang_lookup()
[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 (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
30  * Use is subject to license terms.
31  *
32  * Copyright (c) 2011 Whamcloud, Inc.
33  *
34  */
35 /*
36  * This file is part of Lustre, http://www.lustre.org/
37  * Lustre is a trademark of Sun Microsystems, Inc.
38  *
39  * Client Extent Lock.
40  *
41  *   Author: Nikita Danilov <nikita.danilov@sun.com>
42  */
43
44 #define DEBUG_SUBSYSTEM S_CLASS
45 #ifndef EXPORT_SYMTAB
46 # define EXPORT_SYMTAB
47 #endif
48
49 #include <obd_class.h>
50 #include <obd_support.h>
51 #include <lustre_fid.h>
52 #include <libcfs/list.h>
53 /* lu_time_global_{init,fini}() */
54 #include <lu_time.h>
55
56 #include <cl_object.h>
57 #include "cl_internal.h"
58
59 /** Lock class of cl_lock::cll_guard */
60 static cfs_lock_class_key_t cl_lock_guard_class;
61 static cfs_mem_cache_t *cl_lock_kmem;
62
63 static struct lu_kmem_descr cl_lock_caches[] = {
64         {
65                 .ckd_cache = &cl_lock_kmem,
66                 .ckd_name  = "cl_lock_kmem",
67                 .ckd_size  = sizeof (struct cl_lock)
68         },
69         {
70                 .ckd_cache = NULL
71         }
72 };
73
74 /**
75  * Basic lock invariant that is maintained at all times. Caller either has a
76  * reference to \a lock, or somehow assures that \a lock cannot be freed.
77  *
78  * \see cl_lock_invariant()
79  */
80 static int cl_lock_invariant_trusted(const struct lu_env *env,
81                                      const struct cl_lock *lock)
82 {
83         return  ergo(lock->cll_state == CLS_FREEING, lock->cll_holds == 0) &&
84                 cfs_atomic_read(&lock->cll_ref) >= lock->cll_holds &&
85                 lock->cll_holds >= lock->cll_users &&
86                 lock->cll_holds >= 0 &&
87                 lock->cll_users >= 0 &&
88                 lock->cll_depth >= 0;
89 }
90
91 /**
92  * Stronger lock invariant, checking that caller has a reference on a lock.
93  *
94  * \see cl_lock_invariant_trusted()
95  */
96 static int cl_lock_invariant(const struct lu_env *env,
97                              const struct cl_lock *lock)
98 {
99         int result;
100
101         result = cfs_atomic_read(&lock->cll_ref) > 0 &&
102                 cl_lock_invariant_trusted(env, lock);
103         if (!result && env != NULL)
104                 CL_LOCK_DEBUG(D_ERROR, env, lock, "invariant broken");
105         return result;
106 }
107
108 /**
109  * Returns lock "nesting": 0 for a top-lock and 1 for a sub-lock.
110  */
111 static enum clt_nesting_level cl_lock_nesting(const struct cl_lock *lock)
112 {
113         return cl_object_header(lock->cll_descr.cld_obj)->coh_nesting;
114 }
115
116 /**
117  * Returns a set of counters for this lock, depending on a lock nesting.
118  */
119 static struct cl_thread_counters *cl_lock_counters(const struct lu_env *env,
120                                                    const struct cl_lock *lock)
121 {
122         struct cl_thread_info *info;
123         enum clt_nesting_level nesting;
124
125         info = cl_env_info(env);
126         nesting = cl_lock_nesting(lock);
127         LASSERT(nesting < ARRAY_SIZE(info->clt_counters));
128         return &info->clt_counters[nesting];
129 }
130
131 static void cl_lock_trace0(int level, const struct lu_env *env,
132                            const char *prefix, const struct cl_lock *lock,
133                            const char *func, const int line)
134 {
135         struct cl_object_header *h = cl_object_header(lock->cll_descr.cld_obj);
136         CDEBUG(level, "%s: %p@(%d %p %d %d %d %d %d %lx)"
137                       "(%p/%d/%d) at %s():%d\n",
138                prefix, lock, cfs_atomic_read(&lock->cll_ref),
139                lock->cll_guarder, lock->cll_depth,
140                lock->cll_state, lock->cll_error, lock->cll_holds,
141                lock->cll_users, lock->cll_flags,
142                env, h->coh_nesting, cl_lock_nr_mutexed(env),
143                func, line);
144 }
145 #define cl_lock_trace(level, env, prefix, lock)                         \
146         cl_lock_trace0(level, env, prefix, lock, __FUNCTION__, __LINE__)
147
148 #define RETIP ((unsigned long)__builtin_return_address(0))
149
150 #ifdef CONFIG_LOCKDEP
151 static cfs_lock_class_key_t cl_lock_key;
152
153 static void cl_lock_lockdep_init(struct cl_lock *lock)
154 {
155         lockdep_set_class_and_name(lock, &cl_lock_key, "EXT");
156 }
157
158 static void cl_lock_lockdep_acquire(const struct lu_env *env,
159                                     struct cl_lock *lock, __u32 enqflags)
160 {
161         cl_lock_counters(env, lock)->ctc_nr_locks_acquired++;
162 #ifdef HAVE_LOCK_MAP_ACQUIRE
163         lock_map_acquire(&lock->dep_map);
164 #else  /* HAVE_LOCK_MAP_ACQUIRE */
165         lock_acquire(&lock->dep_map, !!(enqflags & CEF_ASYNC),
166                      /* try: */ 0, lock->cll_descr.cld_mode <= CLM_READ,
167                      /* check: */ 2, RETIP);
168 #endif /* HAVE_LOCK_MAP_ACQUIRE */
169 }
170
171 static void cl_lock_lockdep_release(const struct lu_env *env,
172                                     struct cl_lock *lock)
173 {
174         cl_lock_counters(env, lock)->ctc_nr_locks_acquired--;
175         lock_release(&lock->dep_map, 0, RETIP);
176 }
177
178 #else /* !CONFIG_LOCKDEP */
179
180 static void cl_lock_lockdep_init(struct cl_lock *lock)
181 {}
182 static void cl_lock_lockdep_acquire(const struct lu_env *env,
183                                     struct cl_lock *lock, __u32 enqflags)
184 {}
185 static void cl_lock_lockdep_release(const struct lu_env *env,
186                                     struct cl_lock *lock)
187 {}
188
189 #endif /* !CONFIG_LOCKDEP */
190
191 /**
192  * Adds lock slice to the compound lock.
193  *
194  * This is called by cl_object_operations::coo_lock_init() methods to add a
195  * per-layer state to the lock. New state is added at the end of
196  * cl_lock::cll_layers list, that is, it is at the bottom of the stack.
197  *
198  * \see cl_req_slice_add(), cl_page_slice_add(), cl_io_slice_add()
199  */
200 void cl_lock_slice_add(struct cl_lock *lock, struct cl_lock_slice *slice,
201                        struct cl_object *obj,
202                        const struct cl_lock_operations *ops)
203 {
204         ENTRY;
205         slice->cls_lock = lock;
206         cfs_list_add_tail(&slice->cls_linkage, &lock->cll_layers);
207         slice->cls_obj = obj;
208         slice->cls_ops = ops;
209         EXIT;
210 }
211 EXPORT_SYMBOL(cl_lock_slice_add);
212
213 /**
214  * Returns true iff a lock with the mode \a has provides at least the same
215  * guarantees as a lock with the mode \a need.
216  */
217 int cl_lock_mode_match(enum cl_lock_mode has, enum cl_lock_mode need)
218 {
219         LINVRNT(need == CLM_READ || need == CLM_WRITE ||
220                 need == CLM_PHANTOM || need == CLM_GROUP);
221         LINVRNT(has == CLM_READ || has == CLM_WRITE ||
222                 has == CLM_PHANTOM || has == CLM_GROUP);
223         CLASSERT(CLM_PHANTOM < CLM_READ);
224         CLASSERT(CLM_READ < CLM_WRITE);
225         CLASSERT(CLM_WRITE < CLM_GROUP);
226
227         if (has != CLM_GROUP)
228                 return need <= has;
229         else
230                 return need == has;
231 }
232 EXPORT_SYMBOL(cl_lock_mode_match);
233
234 /**
235  * Returns true iff extent portions of lock descriptions match.
236  */
237 int cl_lock_ext_match(const struct cl_lock_descr *has,
238                       const struct cl_lock_descr *need)
239 {
240         return
241                 has->cld_start <= need->cld_start &&
242                 has->cld_end >= need->cld_end &&
243                 cl_lock_mode_match(has->cld_mode, need->cld_mode) &&
244                 (has->cld_mode != CLM_GROUP || has->cld_gid == need->cld_gid);
245 }
246 EXPORT_SYMBOL(cl_lock_ext_match);
247
248 /**
249  * Returns true iff a lock with the description \a has provides at least the
250  * same guarantees as a lock with the description \a need.
251  */
252 int cl_lock_descr_match(const struct cl_lock_descr *has,
253                         const struct cl_lock_descr *need)
254 {
255         return
256                 cl_object_same(has->cld_obj, need->cld_obj) &&
257                 cl_lock_ext_match(has, need);
258 }
259 EXPORT_SYMBOL(cl_lock_descr_match);
260
261 static void cl_lock_free(const struct lu_env *env, struct cl_lock *lock)
262 {
263         struct cl_object *obj = lock->cll_descr.cld_obj;
264
265         LINVRNT(!cl_lock_is_mutexed(lock));
266
267         ENTRY;
268         cl_lock_trace(D_DLMTRACE, env, "free lock", lock);
269         cfs_might_sleep();
270         while (!cfs_list_empty(&lock->cll_layers)) {
271                 struct cl_lock_slice *slice;
272
273                 slice = cfs_list_entry(lock->cll_layers.next,
274                                        struct cl_lock_slice, cls_linkage);
275                 cfs_list_del_init(lock->cll_layers.next);
276                 slice->cls_ops->clo_fini(env, slice);
277         }
278         cfs_atomic_dec(&cl_object_site(obj)->cs_locks.cs_total);
279         cfs_atomic_dec(&cl_object_site(obj)->cs_locks_state[lock->cll_state]);
280         lu_object_ref_del_at(&obj->co_lu, lock->cll_obj_ref, "cl_lock", lock);
281         cl_object_put(env, obj);
282         lu_ref_fini(&lock->cll_reference);
283         lu_ref_fini(&lock->cll_holders);
284         cfs_mutex_destroy(&lock->cll_guard);
285         OBD_SLAB_FREE_PTR(lock, cl_lock_kmem);
286         EXIT;
287 }
288
289 /**
290  * Releases a reference on a lock.
291  *
292  * When last reference is released, lock is returned to the cache, unless it
293  * is in cl_lock_state::CLS_FREEING state, in which case it is destroyed
294  * immediately.
295  *
296  * \see cl_object_put(), cl_page_put()
297  */
298 void cl_lock_put(const struct lu_env *env, struct cl_lock *lock)
299 {
300         struct cl_object        *obj;
301         struct cl_site          *site;
302
303         LINVRNT(cl_lock_invariant(env, lock));
304         ENTRY;
305         obj = lock->cll_descr.cld_obj;
306         LINVRNT(obj != NULL);
307         site = cl_object_site(obj);
308
309         CDEBUG(D_TRACE, "releasing reference: %d %p %lu\n",
310                cfs_atomic_read(&lock->cll_ref), lock, RETIP);
311
312         if (cfs_atomic_dec_and_test(&lock->cll_ref)) {
313                 if (lock->cll_state == CLS_FREEING) {
314                         LASSERT(cfs_list_empty(&lock->cll_linkage));
315                         cl_lock_free(env, lock);
316                 }
317                 cfs_atomic_dec(&site->cs_locks.cs_busy);
318         }
319         EXIT;
320 }
321 EXPORT_SYMBOL(cl_lock_put);
322
323 /**
324  * Acquires an additional reference to a lock.
325  *
326  * This can be called only by caller already possessing a reference to \a
327  * lock.
328  *
329  * \see cl_object_get(), cl_page_get()
330  */
331 void cl_lock_get(struct cl_lock *lock)
332 {
333         LINVRNT(cl_lock_invariant(NULL, lock));
334         CDEBUG(D_TRACE, "acquiring reference: %d %p %lu\n",
335                cfs_atomic_read(&lock->cll_ref), lock, RETIP);
336         cfs_atomic_inc(&lock->cll_ref);
337 }
338 EXPORT_SYMBOL(cl_lock_get);
339
340 /**
341  * Acquires a reference to a lock.
342  *
343  * This is much like cl_lock_get(), except that this function can be used to
344  * acquire initial reference to the cached lock. Caller has to deal with all
345  * possible races. Use with care!
346  *
347  * \see cl_page_get_trust()
348  */
349 void cl_lock_get_trust(struct cl_lock *lock)
350 {
351         struct cl_site *site = cl_object_site(lock->cll_descr.cld_obj);
352
353         CDEBUG(D_TRACE, "acquiring trusted reference: %d %p %lu\n",
354                cfs_atomic_read(&lock->cll_ref), lock, RETIP);
355         if (cfs_atomic_inc_return(&lock->cll_ref) == 1)
356                 cfs_atomic_inc(&site->cs_locks.cs_busy);
357 }
358 EXPORT_SYMBOL(cl_lock_get_trust);
359
360 /**
361  * Helper function destroying the lock that wasn't completely initialized.
362  *
363  * Other threads can acquire references to the top-lock through its
364  * sub-locks. Hence, it cannot be cl_lock_free()-ed immediately.
365  */
366 static void cl_lock_finish(const struct lu_env *env, struct cl_lock *lock)
367 {
368         cl_lock_mutex_get(env, lock);
369         cl_lock_cancel(env, lock);
370         cl_lock_delete(env, lock);
371         cl_lock_mutex_put(env, lock);
372         cl_lock_put(env, lock);
373 }
374
375 static struct cl_lock *cl_lock_alloc(const struct lu_env *env,
376                                      struct cl_object *obj,
377                                      const struct cl_io *io,
378                                      const struct cl_lock_descr *descr)
379 {
380         struct cl_lock          *lock;
381         struct lu_object_header *head;
382         struct cl_site          *site = cl_object_site(obj);
383
384         ENTRY;
385         OBD_SLAB_ALLOC_PTR_GFP(lock, cl_lock_kmem, CFS_ALLOC_IO);
386         if (lock != NULL) {
387                 cfs_atomic_set(&lock->cll_ref, 1);
388                 lock->cll_descr = *descr;
389                 lock->cll_state = CLS_NEW;
390                 cl_object_get(obj);
391                 lock->cll_obj_ref = lu_object_ref_add(&obj->co_lu,
392                                                       "cl_lock", lock);
393                 CFS_INIT_LIST_HEAD(&lock->cll_layers);
394                 CFS_INIT_LIST_HEAD(&lock->cll_linkage);
395                 CFS_INIT_LIST_HEAD(&lock->cll_inclosure);
396                 lu_ref_init(&lock->cll_reference);
397                 lu_ref_init(&lock->cll_holders);
398                 cfs_mutex_init(&lock->cll_guard);
399                 cfs_lockdep_set_class(&lock->cll_guard, &cl_lock_guard_class);
400                 cfs_waitq_init(&lock->cll_wq);
401                 head = obj->co_lu.lo_header;
402                 cfs_atomic_inc(&site->cs_locks_state[CLS_NEW]);
403                 cfs_atomic_inc(&site->cs_locks.cs_total);
404                 cfs_atomic_inc(&site->cs_locks.cs_created);
405                 cl_lock_lockdep_init(lock);
406                 cfs_list_for_each_entry(obj, &head->loh_layers,
407                                         co_lu.lo_linkage) {
408                         int err;
409
410                         err = obj->co_ops->coo_lock_init(env, obj, lock, io);
411                         if (err != 0) {
412                                 cl_lock_finish(env, lock);
413                                 lock = ERR_PTR(err);
414                                 break;
415                         }
416                 }
417         } else
418                 lock = ERR_PTR(-ENOMEM);
419         RETURN(lock);
420 }
421
422 /**
423  * Transfer the lock into INTRANSIT state and return the original state.
424  *
425  * \pre  state: CLS_CACHED, CLS_HELD or CLS_ENQUEUED
426  * \post state: CLS_INTRANSIT
427  * \see CLS_INTRANSIT
428  */
429 enum cl_lock_state cl_lock_intransit(const struct lu_env *env,
430                                      struct cl_lock *lock)
431 {
432         enum cl_lock_state state = lock->cll_state;
433
434         LASSERT(cl_lock_is_mutexed(lock));
435         LASSERT(state != CLS_INTRANSIT);
436         LASSERTF(state >= CLS_ENQUEUED && state <= CLS_CACHED,
437                  "Malformed lock state %d.\n", state);
438
439         cl_lock_state_set(env, lock, CLS_INTRANSIT);
440         lock->cll_intransit_owner = cfs_current();
441         cl_lock_hold_add(env, lock, "intransit", cfs_current());
442         return state;
443 }
444 EXPORT_SYMBOL(cl_lock_intransit);
445
446 /**
447  *  Exit the intransit state and restore the lock state to the original state
448  */
449 void cl_lock_extransit(const struct lu_env *env, struct cl_lock *lock,
450                        enum cl_lock_state state)
451 {
452         LASSERT(cl_lock_is_mutexed(lock));
453         LASSERT(lock->cll_state == CLS_INTRANSIT);
454         LASSERT(state != CLS_INTRANSIT);
455         LASSERT(lock->cll_intransit_owner == cfs_current());
456
457         lock->cll_intransit_owner = NULL;
458         cl_lock_state_set(env, lock, state);
459         cl_lock_unhold(env, lock, "intransit", cfs_current());
460 }
461 EXPORT_SYMBOL(cl_lock_extransit);
462
463 /**
464  * Checking whether the lock is intransit state
465  */
466 int cl_lock_is_intransit(struct cl_lock *lock)
467 {
468         LASSERT(cl_lock_is_mutexed(lock));
469         return lock->cll_state == CLS_INTRANSIT &&
470                lock->cll_intransit_owner != cfs_current();
471 }
472 EXPORT_SYMBOL(cl_lock_is_intransit);
473 /**
474  * Returns true iff lock is "suitable" for given io. E.g., locks acquired by
475  * truncate and O_APPEND cannot be reused for read/non-append-write, as they
476  * cover multiple stripes and can trigger cascading timeouts.
477  */
478 static int cl_lock_fits_into(const struct lu_env *env,
479                              const struct cl_lock *lock,
480                              const struct cl_lock_descr *need,
481                              const struct cl_io *io)
482 {
483         const struct cl_lock_slice *slice;
484
485         LINVRNT(cl_lock_invariant_trusted(env, lock));
486         ENTRY;
487         cfs_list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
488                 if (slice->cls_ops->clo_fits_into != NULL &&
489                     !slice->cls_ops->clo_fits_into(env, slice, need, io))
490                         RETURN(0);
491         }
492         RETURN(1);
493 }
494
495 static struct cl_lock *cl_lock_lookup(const struct lu_env *env,
496                                       struct cl_object *obj,
497                                       const struct cl_io *io,
498                                       const struct cl_lock_descr *need)
499 {
500         struct cl_lock          *lock;
501         struct cl_object_header *head;
502         struct cl_site          *site;
503
504         ENTRY;
505
506         head = cl_object_header(obj);
507         site = cl_object_site(obj);
508         LINVRNT_SPIN_LOCKED(&head->coh_lock_guard);
509         cfs_atomic_inc(&site->cs_locks.cs_lookup);
510         cfs_list_for_each_entry(lock, &head->coh_locks, cll_linkage) {
511                 int matched;
512
513                 matched = cl_lock_ext_match(&lock->cll_descr, need) &&
514                           lock->cll_state < CLS_FREEING &&
515                           lock->cll_error == 0 &&
516                           !(lock->cll_flags & CLF_CANCELLED) &&
517                           cl_lock_fits_into(env, lock, need, io);
518                 CDEBUG(D_DLMTRACE, "has: "DDESCR"(%d) need: "DDESCR": %d\n",
519                        PDESCR(&lock->cll_descr), lock->cll_state, PDESCR(need),
520                        matched);
521                 if (matched) {
522                         cl_lock_get_trust(lock);
523                         cfs_atomic_inc(&cl_object_site(obj)->cs_locks.cs_hit);
524                         RETURN(lock);
525                 }
526         }
527         RETURN(NULL);
528 }
529
530 /**
531  * Returns a lock matching description \a need.
532  *
533  * This is the main entry point into the cl_lock caching interface. First, a
534  * cache (implemented as a per-object linked list) is consulted. If lock is
535  * found there, it is returned immediately. Otherwise new lock is allocated
536  * and returned. In any case, additional reference to lock is acquired.
537  *
538  * \see cl_object_find(), cl_page_find()
539  */
540 static struct cl_lock *cl_lock_find(const struct lu_env *env,
541                                     const struct cl_io *io,
542                                     const struct cl_lock_descr *need)
543 {
544         struct cl_object_header *head;
545         struct cl_object        *obj;
546         struct cl_lock          *lock;
547         struct cl_site          *site;
548
549         ENTRY;
550
551         obj  = need->cld_obj;
552         head = cl_object_header(obj);
553         site = cl_object_site(obj);
554
555         cfs_spin_lock(&head->coh_lock_guard);
556         lock = cl_lock_lookup(env, obj, io, need);
557         cfs_spin_unlock(&head->coh_lock_guard);
558
559         if (lock == NULL) {
560                 lock = cl_lock_alloc(env, obj, io, need);
561                 if (!IS_ERR(lock)) {
562                         struct cl_lock *ghost;
563
564                         cfs_spin_lock(&head->coh_lock_guard);
565                         ghost = cl_lock_lookup(env, obj, io, need);
566                         if (ghost == NULL) {
567                                 cfs_list_add_tail(&lock->cll_linkage,
568                                                   &head->coh_locks);
569                                 cfs_spin_unlock(&head->coh_lock_guard);
570                                 cfs_atomic_inc(&site->cs_locks.cs_busy);
571                         } else {
572                                 cfs_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         cfs_spin_lock(&head->coh_lock_guard);
604         lock = cl_lock_lookup(env, obj, io, need);
605         cfs_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         cfs_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                 cfs_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 (cfs_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                 cfs_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                 cfs_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                 cfs_spin_lock(&head->coh_lock_guard);
831                 cfs_list_del_init(&lock->cll_linkage);
832
833                 cfs_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                 cfs_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  * 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         cfs_waitlink_t waiter;
950         cfs_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                 cfs_waitlink_init(&waiter);
968                 cfs_waitq_add(&lock->cll_wq, &waiter);
969                 cfs_set_current_state(CFS_TASK_INTERRUPTIBLE);
970                 cl_lock_mutex_put(env, lock);
971
972                 LASSERT(cl_lock_nr_mutexed(env) == 0);
973                 cfs_waitq_wait(&waiter, CFS_TASK_INTERRUPTIBLE);
974
975                 cl_lock_mutex_get(env, lock);
976                 cfs_set_current_state(CFS_TASK_RUNNING);
977                 cfs_waitq_del(&lock->cll_wq, &waiter);
978                 result = cfs_signal_pending() ? -EINTR : 0;
979
980                 /* Restore old blocked signals */
981                 cfs_restore_sigs(blocked);
982         }
983         RETURN(result);
984 }
985 EXPORT_SYMBOL(cl_lock_state_wait);
986
987 static void cl_lock_state_signal(const struct lu_env *env, struct cl_lock *lock,
988                                  enum cl_lock_state state)
989 {
990         const struct cl_lock_slice *slice;
991
992         ENTRY;
993         LINVRNT(cl_lock_is_mutexed(lock));
994         LINVRNT(cl_lock_invariant(env, lock));
995
996         cfs_list_for_each_entry(slice, &lock->cll_layers, cls_linkage)
997                 if (slice->cls_ops->clo_state != NULL)
998                         slice->cls_ops->clo_state(env, slice, state);
999         cfs_waitq_broadcast(&lock->cll_wq);
1000         EXIT;
1001 }
1002
1003 /**
1004  * Notifies waiters that lock state changed.
1005  *
1006  * Wakes up all waiters sleeping in cl_lock_state_wait(), also notifies all
1007  * layers about state change by calling cl_lock_operations::clo_state()
1008  * top-to-bottom.
1009  */
1010 void cl_lock_signal(const struct lu_env *env, struct cl_lock *lock)
1011 {
1012         ENTRY;
1013         cl_lock_trace(D_DLMTRACE, env, "state signal lock", lock);
1014         cl_lock_state_signal(env, lock, lock->cll_state);
1015         EXIT;
1016 }
1017 EXPORT_SYMBOL(cl_lock_signal);
1018
1019 /**
1020  * Changes lock state.
1021  *
1022  * This function is invoked to notify layers that lock state changed, possible
1023  * as a result of an asynchronous event such as call-back reception.
1024  *
1025  * \post lock->cll_state == state
1026  *
1027  * \see cl_lock_operations::clo_state()
1028  */
1029 void cl_lock_state_set(const struct lu_env *env, struct cl_lock *lock,
1030                        enum cl_lock_state state)
1031 {
1032         struct cl_site *site = cl_object_site(lock->cll_descr.cld_obj);
1033
1034         ENTRY;
1035         LASSERT(lock->cll_state <= state ||
1036                 (lock->cll_state == CLS_CACHED &&
1037                  (state == CLS_HELD || /* lock found in cache */
1038                   state == CLS_NEW  ||   /* sub-lock canceled */
1039                   state == CLS_INTRANSIT)) ||
1040                 /* lock is in transit state */
1041                 lock->cll_state == CLS_INTRANSIT);
1042
1043         if (lock->cll_state != state) {
1044                 cfs_atomic_dec(&site->cs_locks_state[lock->cll_state]);
1045                 cfs_atomic_inc(&site->cs_locks_state[state]);
1046
1047                 cl_lock_state_signal(env, lock, state);
1048                 lock->cll_state = state;
1049         }
1050         EXIT;
1051 }
1052 EXPORT_SYMBOL(cl_lock_state_set);
1053
1054 static int cl_unuse_try_internal(const struct lu_env *env, struct cl_lock *lock)
1055 {
1056         const struct cl_lock_slice *slice;
1057         int result;
1058
1059         do {
1060                 result = 0;
1061
1062                 LINVRNT(cl_lock_is_mutexed(lock));
1063                 LINVRNT(cl_lock_invariant(env, lock));
1064                 LASSERT(lock->cll_state == CLS_INTRANSIT);
1065
1066                 result = -ENOSYS;
1067                 cfs_list_for_each_entry_reverse(slice, &lock->cll_layers,
1068                                                 cls_linkage) {
1069                         if (slice->cls_ops->clo_unuse != NULL) {
1070                                 result = slice->cls_ops->clo_unuse(env, slice);
1071                                 if (result != 0)
1072                                         break;
1073                         }
1074                 }
1075                 LASSERT(result != -ENOSYS);
1076         } while (result == CLO_REPEAT);
1077
1078         return result;
1079 }
1080
1081 /**
1082  * Yanks lock from the cache (cl_lock_state::CLS_CACHED state) by calling
1083  * cl_lock_operations::clo_use() top-to-bottom to notify layers.
1084  * @atomic = 1, it must unuse the lock to recovery the lock to keep the
1085  *  use process atomic
1086  */
1087 int cl_use_try(const struct lu_env *env, struct cl_lock *lock, int atomic)
1088 {
1089         const struct cl_lock_slice *slice;
1090         int result;
1091         enum cl_lock_state state;
1092
1093         ENTRY;
1094         cl_lock_trace(D_DLMTRACE, env, "use lock", lock);
1095
1096         LASSERT(lock->cll_state == CLS_CACHED);
1097         if (lock->cll_error)
1098                 RETURN(lock->cll_error);
1099
1100         result = -ENOSYS;
1101         state = cl_lock_intransit(env, lock);
1102         cfs_list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
1103                 if (slice->cls_ops->clo_use != NULL) {
1104                         result = slice->cls_ops->clo_use(env, slice);
1105                         if (result != 0)
1106                                 break;
1107                 }
1108         }
1109         LASSERT(result != -ENOSYS);
1110
1111         LASSERTF(lock->cll_state == CLS_INTRANSIT, "Wrong state %d.\n",
1112                  lock->cll_state);
1113
1114         if (result == 0) {
1115                 state = CLS_HELD;
1116         } else {
1117                 if (result == -ESTALE) {
1118                         /*
1119                          * ESTALE means sublock being cancelled
1120                          * at this time, and set lock state to
1121                          * be NEW here and ask the caller to repeat.
1122                          */
1123                         state = CLS_NEW;
1124                         result = CLO_REPEAT;
1125                 }
1126
1127                 /* @atomic means back-off-on-failure. */
1128                 if (atomic) {
1129                         int rc;
1130                         rc = cl_unuse_try_internal(env, lock);
1131                         /* Vet the results. */
1132                         if (rc < 0 && result > 0)
1133                                 result = rc;
1134                 }
1135
1136         }
1137         cl_lock_extransit(env, lock, state);
1138         RETURN(result);
1139 }
1140 EXPORT_SYMBOL(cl_use_try);
1141
1142 /**
1143  * Helper for cl_enqueue_try() that calls ->clo_enqueue() across all layers
1144  * top-to-bottom.
1145  */
1146 static int cl_enqueue_kick(const struct lu_env *env,
1147                            struct cl_lock *lock,
1148                            struct cl_io *io, __u32 flags)
1149 {
1150         int result;
1151         const struct cl_lock_slice *slice;
1152
1153         ENTRY;
1154         result = -ENOSYS;
1155         cfs_list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
1156                 if (slice->cls_ops->clo_enqueue != NULL) {
1157                         result = slice->cls_ops->clo_enqueue(env,
1158                                                              slice, io, flags);
1159                         if (result != 0)
1160                                 break;
1161                 }
1162         }
1163         LASSERT(result != -ENOSYS);
1164         RETURN(result);
1165 }
1166
1167 /**
1168  * Tries to enqueue a lock.
1169  *
1170  * This function is called repeatedly by cl_enqueue() until either lock is
1171  * enqueued, or error occurs. This function does not block waiting for
1172  * networking communication to complete.
1173  *
1174  * \post ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
1175  *                         lock->cll_state == CLS_HELD)
1176  *
1177  * \see cl_enqueue() cl_lock_operations::clo_enqueue()
1178  * \see cl_lock_state::CLS_ENQUEUED
1179  */
1180 int cl_enqueue_try(const struct lu_env *env, struct cl_lock *lock,
1181                    struct cl_io *io, __u32 flags)
1182 {
1183         int result;
1184
1185         ENTRY;
1186         cl_lock_trace(D_DLMTRACE, env, "enqueue lock", lock);
1187         do {
1188                 result = 0;
1189
1190                 LINVRNT(cl_lock_is_mutexed(lock));
1191
1192                 if (lock->cll_error != 0)
1193                         break;
1194                 switch (lock->cll_state) {
1195                 case CLS_NEW:
1196                         cl_lock_state_set(env, lock, CLS_QUEUING);
1197                         /* fall-through */
1198                 case CLS_QUEUING:
1199                         /* kick layers. */
1200                         result = cl_enqueue_kick(env, lock, io, flags);
1201                         if (result == 0)
1202                                 cl_lock_state_set(env, lock, CLS_ENQUEUED);
1203                         break;
1204                 case CLS_INTRANSIT:
1205                         LASSERT(cl_lock_is_intransit(lock));
1206                         result = CLO_WAIT;
1207                         break;
1208                 case CLS_CACHED:
1209                         /* yank lock from the cache. */
1210                         result = cl_use_try(env, lock, 0);
1211                         break;
1212                 case CLS_ENQUEUED:
1213                 case CLS_HELD:
1214                         result = 0;
1215                         break;
1216                 default:
1217                 case CLS_FREEING:
1218                         /*
1219                          * impossible, only held locks with increased
1220                          * ->cll_holds can be enqueued, and they cannot be
1221                          * freed.
1222                          */
1223                         LBUG();
1224                 }
1225         } while (result == CLO_REPEAT);
1226         if (result < 0)
1227                 cl_lock_error(env, lock, result);
1228         RETURN(result ?: lock->cll_error);
1229 }
1230 EXPORT_SYMBOL(cl_enqueue_try);
1231
1232 /**
1233  * Cancel the conflicting lock found during previous enqueue.
1234  *
1235  * \retval 0 conflicting lock has been canceled.
1236  * \retval -ve error code.
1237  */
1238 int cl_lock_enqueue_wait(const struct lu_env *env,
1239                          struct cl_lock *lock,
1240                          int keep_mutex)
1241 {
1242         struct cl_lock  *conflict;
1243         int              rc = 0;
1244         ENTRY;
1245
1246         LASSERT(cl_lock_is_mutexed(lock));
1247         LASSERT(lock->cll_state == CLS_QUEUING);
1248         LASSERT(lock->cll_conflict != NULL);
1249
1250         conflict = lock->cll_conflict;
1251         lock->cll_conflict = NULL;
1252
1253         cl_lock_mutex_put(env, lock);
1254         LASSERT(cl_lock_nr_mutexed(env) == 0);
1255
1256         cl_lock_mutex_get(env, conflict);
1257         cl_lock_cancel(env, conflict);
1258         cl_lock_delete(env, conflict);
1259
1260         while (conflict->cll_state != CLS_FREEING) {
1261                 rc = cl_lock_state_wait(env, conflict);
1262                 if (rc != 0)
1263                         break;
1264         }
1265         cl_lock_mutex_put(env, conflict);
1266         lu_ref_del(&conflict->cll_reference, "cancel-wait", lock);
1267         cl_lock_put(env, conflict);
1268
1269         if (keep_mutex)
1270                 cl_lock_mutex_get(env, lock);
1271
1272         LASSERT(rc <= 0);
1273         RETURN(rc);
1274 }
1275 EXPORT_SYMBOL(cl_lock_enqueue_wait);
1276
1277 static int cl_enqueue_locked(const struct lu_env *env, struct cl_lock *lock,
1278                              struct cl_io *io, __u32 enqflags)
1279 {
1280         int result;
1281
1282         ENTRY;
1283
1284         LINVRNT(cl_lock_is_mutexed(lock));
1285         LINVRNT(cl_lock_invariant(env, lock));
1286         LASSERT(lock->cll_holds > 0);
1287
1288         cl_lock_user_add(env, lock);
1289         do {
1290                 result = cl_enqueue_try(env, lock, io, enqflags);
1291                 if (result == CLO_WAIT) {
1292                         if (lock->cll_conflict != NULL)
1293                                 result = cl_lock_enqueue_wait(env, lock, 1);
1294                         else
1295                                 result = cl_lock_state_wait(env, lock);
1296                         if (result == 0)
1297                                 continue;
1298                 }
1299                 break;
1300         } while (1);
1301         if (result != 0) {
1302                 cl_lock_user_del(env, lock);
1303                 cl_lock_error(env, lock, result);
1304         }
1305         LASSERT(ergo(result == 0 && !(enqflags & CEF_AGL),
1306                      lock->cll_state == CLS_ENQUEUED ||
1307                      lock->cll_state == CLS_HELD));
1308         RETURN(result);
1309 }
1310
1311 /**
1312  * Enqueues a lock.
1313  *
1314  * \pre current thread or io owns a hold on lock.
1315  *
1316  * \post ergo(result == 0, lock->users increased)
1317  * \post ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
1318  *                         lock->cll_state == CLS_HELD)
1319  */
1320 int cl_enqueue(const struct lu_env *env, struct cl_lock *lock,
1321                struct cl_io *io, __u32 enqflags)
1322 {
1323         int result;
1324
1325         ENTRY;
1326
1327         cl_lock_lockdep_acquire(env, lock, enqflags);
1328         cl_lock_mutex_get(env, lock);
1329         result = cl_enqueue_locked(env, lock, io, enqflags);
1330         cl_lock_mutex_put(env, lock);
1331         if (result != 0)
1332                 cl_lock_lockdep_release(env, lock);
1333         LASSERT(ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
1334                      lock->cll_state == CLS_HELD));
1335         RETURN(result);
1336 }
1337 EXPORT_SYMBOL(cl_enqueue);
1338
1339 /**
1340  * Tries to unlock a lock.
1341  *
1342  * This function is called repeatedly by cl_unuse() until either lock is
1343  * unlocked, or error occurs.
1344  * cl_unuse_try is a one-shot operation, so it must NOT return CLO_WAIT.
1345  *
1346  * \pre  lock->cll_state == CLS_HELD
1347  *
1348  * \post ergo(result == 0, lock->cll_state == CLS_CACHED)
1349  *
1350  * \see cl_unuse() cl_lock_operations::clo_unuse()
1351  * \see cl_lock_state::CLS_CACHED
1352  */
1353 int cl_unuse_try(const struct lu_env *env, struct cl_lock *lock)
1354 {
1355         int                         result;
1356         enum cl_lock_state          state = CLS_NEW;
1357
1358         ENTRY;
1359         cl_lock_trace(D_DLMTRACE, env, "unuse lock", lock);
1360
1361         LASSERT(lock->cll_state == CLS_HELD || lock->cll_state == CLS_ENQUEUED);
1362         if (lock->cll_users > 1) {
1363                 cl_lock_user_del(env, lock);
1364                 RETURN(0);
1365         }
1366
1367         /*
1368          * New lock users (->cll_users) are not protecting unlocking
1369          * from proceeding. From this point, lock eventually reaches
1370          * CLS_CACHED, is reinitialized to CLS_NEW or fails into
1371          * CLS_FREEING.
1372          */
1373         state = cl_lock_intransit(env, lock);
1374
1375         result = cl_unuse_try_internal(env, lock);
1376         LASSERT(lock->cll_state == CLS_INTRANSIT);
1377         LASSERT(result != CLO_WAIT);
1378         cl_lock_user_del(env, lock);
1379         if (result == 0 || result == -ESTALE) {
1380                 /*
1381                  * Return lock back to the cache. This is the only
1382                  * place where lock is moved into CLS_CACHED state.
1383                  *
1384                  * If one of ->clo_unuse() methods returned -ESTALE, lock
1385                  * cannot be placed into cache and has to be
1386                  * re-initialized. This happens e.g., when a sub-lock was
1387                  * canceled while unlocking was in progress.
1388                  */
1389                 if (state == CLS_HELD && result == 0)
1390                         state = CLS_CACHED;
1391                 else
1392                         state = CLS_NEW;
1393                 cl_lock_extransit(env, lock, state);
1394
1395                 /*
1396                  * Hide -ESTALE error.
1397                  * If the lock is a glimpse lock, and it has multiple
1398                  * stripes. Assuming that one of its sublock returned -ENAVAIL,
1399                  * and other sublocks are matched write locks. In this case,
1400                  * we can't set this lock to error because otherwise some of
1401                  * its sublocks may not be canceled. This causes some dirty
1402                  * pages won't be written to OSTs. -jay
1403                  */
1404                 result = 0;
1405         } else {
1406                 CERROR("result = %d, this is unlikely!\n", result);
1407                 cl_lock_extransit(env, lock, state);
1408         }
1409
1410         result = result ?: lock->cll_error;
1411         if (result < 0)
1412                 cl_lock_error(env, lock, result);
1413         RETURN(result);
1414 }
1415 EXPORT_SYMBOL(cl_unuse_try);
1416
1417 static void cl_unuse_locked(const struct lu_env *env, struct cl_lock *lock)
1418 {
1419         int result;
1420         ENTRY;
1421
1422         result = cl_unuse_try(env, lock);
1423         if (result)
1424                 CL_LOCK_DEBUG(D_ERROR, env, lock, "unuse return %d\n", result);
1425
1426         EXIT;
1427 }
1428
1429 /**
1430  * Unlocks a lock.
1431  */
1432 void cl_unuse(const struct lu_env *env, struct cl_lock *lock)
1433 {
1434         ENTRY;
1435         cl_lock_mutex_get(env, lock);
1436         cl_unuse_locked(env, lock);
1437         cl_lock_mutex_put(env, lock);
1438         cl_lock_lockdep_release(env, lock);
1439         EXIT;
1440 }
1441 EXPORT_SYMBOL(cl_unuse);
1442
1443 /**
1444  * Tries to wait for a lock.
1445  *
1446  * This function is called repeatedly by cl_wait() until either lock is
1447  * granted, or error occurs. This function does not block waiting for network
1448  * communication to complete.
1449  *
1450  * \see cl_wait() cl_lock_operations::clo_wait()
1451  * \see cl_lock_state::CLS_HELD
1452  */
1453 int cl_wait_try(const struct lu_env *env, struct cl_lock *lock)
1454 {
1455         const struct cl_lock_slice *slice;
1456         int                         result;
1457
1458         ENTRY;
1459         cl_lock_trace(D_DLMTRACE, env, "wait lock try", lock);
1460         do {
1461                 LINVRNT(cl_lock_is_mutexed(lock));
1462                 LINVRNT(cl_lock_invariant(env, lock));
1463                 LASSERT(lock->cll_state == CLS_ENQUEUED ||
1464                         lock->cll_state == CLS_HELD ||
1465                         lock->cll_state == CLS_INTRANSIT);
1466                 LASSERT(lock->cll_users > 0);
1467                 LASSERT(lock->cll_holds > 0);
1468
1469                 result = 0;
1470                 if (lock->cll_error != 0)
1471                         break;
1472
1473                 if (cl_lock_is_intransit(lock)) {
1474                         result = CLO_WAIT;
1475                         break;
1476                 }
1477
1478                 if (lock->cll_state == CLS_HELD)
1479                         /* nothing to do */
1480                         break;
1481
1482                 result = -ENOSYS;
1483                 cfs_list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
1484                         if (slice->cls_ops->clo_wait != NULL) {
1485                                 result = slice->cls_ops->clo_wait(env, slice);
1486                                 if (result != 0)
1487                                         break;
1488                         }
1489                 }
1490                 LASSERT(result != -ENOSYS);
1491                 if (result == 0) {
1492                         LASSERT(lock->cll_state != CLS_INTRANSIT);
1493                         cl_lock_state_set(env, lock, CLS_HELD);
1494                 }
1495         } while (result == CLO_REPEAT);
1496         RETURN(result ?: lock->cll_error);
1497 }
1498 EXPORT_SYMBOL(cl_wait_try);
1499
1500 /**
1501  * Waits until enqueued lock is granted.
1502  *
1503  * \pre current thread or io owns a hold on the lock
1504  * \pre ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
1505  *                        lock->cll_state == CLS_HELD)
1506  *
1507  * \post ergo(result == 0, lock->cll_state == CLS_HELD)
1508  */
1509 int cl_wait(const struct lu_env *env, struct cl_lock *lock)
1510 {
1511         int result;
1512
1513         ENTRY;
1514         cl_lock_mutex_get(env, lock);
1515
1516         LINVRNT(cl_lock_invariant(env, lock));
1517         LASSERTF(lock->cll_state == CLS_ENQUEUED || lock->cll_state == CLS_HELD,
1518                  "Wrong state %d \n", lock->cll_state);
1519         LASSERT(lock->cll_holds > 0);
1520
1521         do {
1522                 result = cl_wait_try(env, lock);
1523                 if (result == CLO_WAIT) {
1524                         result = cl_lock_state_wait(env, lock);
1525                         if (result == 0)
1526                                 continue;
1527                 }
1528                 break;
1529         } while (1);
1530         if (result < 0) {
1531                 cl_lock_user_del(env, lock);
1532                 cl_lock_error(env, lock, result);
1533                 cl_lock_lockdep_release(env, lock);
1534         }
1535         cl_lock_trace(D_DLMTRACE, env, "wait lock", lock);
1536         cl_lock_mutex_put(env, lock);
1537         LASSERT(ergo(result == 0, lock->cll_state == CLS_HELD));
1538         RETURN(result);
1539 }
1540 EXPORT_SYMBOL(cl_wait);
1541
1542 /**
1543  * Executes cl_lock_operations::clo_weigh(), and sums results to estimate lock
1544  * value.
1545  */
1546 unsigned long cl_lock_weigh(const struct lu_env *env, struct cl_lock *lock)
1547 {
1548         const struct cl_lock_slice *slice;
1549         unsigned long pound;
1550         unsigned long ounce;
1551
1552         ENTRY;
1553         LINVRNT(cl_lock_is_mutexed(lock));
1554         LINVRNT(cl_lock_invariant(env, lock));
1555
1556         pound = 0;
1557         cfs_list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
1558                 if (slice->cls_ops->clo_weigh != NULL) {
1559                         ounce = slice->cls_ops->clo_weigh(env, slice);
1560                         pound += ounce;
1561                         if (pound < ounce) /* over-weight^Wflow */
1562                                 pound = ~0UL;
1563                 }
1564         }
1565         RETURN(pound);
1566 }
1567 EXPORT_SYMBOL(cl_lock_weigh);
1568
1569 /**
1570  * Notifies layers that lock description changed.
1571  *
1572  * The server can grant client a lock different from one that was requested
1573  * (e.g., larger in extent). This method is called when actually granted lock
1574  * description becomes known to let layers to accommodate for changed lock
1575  * description.
1576  *
1577  * \see cl_lock_operations::clo_modify()
1578  */
1579 int cl_lock_modify(const struct lu_env *env, struct cl_lock *lock,
1580                    const struct cl_lock_descr *desc)
1581 {
1582         const struct cl_lock_slice *slice;
1583         struct cl_object           *obj = lock->cll_descr.cld_obj;
1584         struct cl_object_header    *hdr = cl_object_header(obj);
1585         int result;
1586
1587         ENTRY;
1588         cl_lock_trace(D_DLMTRACE, env, "modify lock", lock);
1589         /* don't allow object to change */
1590         LASSERT(obj == desc->cld_obj);
1591         LINVRNT(cl_lock_is_mutexed(lock));
1592         LINVRNT(cl_lock_invariant(env, lock));
1593
1594         cfs_list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
1595                 if (slice->cls_ops->clo_modify != NULL) {
1596                         result = slice->cls_ops->clo_modify(env, slice, desc);
1597                         if (result != 0)
1598                                 RETURN(result);
1599                 }
1600         }
1601         CL_LOCK_DEBUG(D_DLMTRACE, env, lock, " -> "DDESCR"@"DFID"\n",
1602                       PDESCR(desc), PFID(lu_object_fid(&desc->cld_obj->co_lu)));
1603         /*
1604          * Just replace description in place. Nothing more is needed for
1605          * now. If locks were indexed according to their extent and/or mode,
1606          * that index would have to be updated here.
1607          */
1608         cfs_spin_lock(&hdr->coh_lock_guard);
1609         lock->cll_descr = *desc;
1610         cfs_spin_unlock(&hdr->coh_lock_guard);
1611         RETURN(0);
1612 }
1613 EXPORT_SYMBOL(cl_lock_modify);
1614
1615 /**
1616  * Initializes lock closure with a given origin.
1617  *
1618  * \see cl_lock_closure
1619  */
1620 void cl_lock_closure_init(const struct lu_env *env,
1621                           struct cl_lock_closure *closure,
1622                           struct cl_lock *origin, int wait)
1623 {
1624         LINVRNT(cl_lock_is_mutexed(origin));
1625         LINVRNT(cl_lock_invariant(env, origin));
1626
1627         CFS_INIT_LIST_HEAD(&closure->clc_list);
1628         closure->clc_origin = origin;
1629         closure->clc_wait   = wait;
1630         closure->clc_nr     = 0;
1631 }
1632 EXPORT_SYMBOL(cl_lock_closure_init);
1633
1634 /**
1635  * Builds a closure of \a lock.
1636  *
1637  * Building of a closure consists of adding initial lock (\a lock) into it,
1638  * and calling cl_lock_operations::clo_closure() methods of \a lock. These
1639  * methods might call cl_lock_closure_build() recursively again, adding more
1640  * locks to the closure, etc.
1641  *
1642  * \see cl_lock_closure
1643  */
1644 int cl_lock_closure_build(const struct lu_env *env, struct cl_lock *lock,
1645                           struct cl_lock_closure *closure)
1646 {
1647         const struct cl_lock_slice *slice;
1648         int result;
1649
1650         ENTRY;
1651         LINVRNT(cl_lock_is_mutexed(closure->clc_origin));
1652         LINVRNT(cl_lock_invariant(env, closure->clc_origin));
1653
1654         result = cl_lock_enclosure(env, lock, closure);
1655         if (result == 0) {
1656                 cfs_list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
1657                         if (slice->cls_ops->clo_closure != NULL) {
1658                                 result = slice->cls_ops->clo_closure(env, slice,
1659                                                                      closure);
1660                                 if (result != 0)
1661                                         break;
1662                         }
1663                 }
1664         }
1665         if (result != 0)
1666                 cl_lock_disclosure(env, closure);
1667         RETURN(result);
1668 }
1669 EXPORT_SYMBOL(cl_lock_closure_build);
1670
1671 /**
1672  * Adds new lock to a closure.
1673  *
1674  * Try-locks \a lock and if succeeded, adds it to the closure (never more than
1675  * once). If try-lock failed, returns CLO_REPEAT, after optionally waiting
1676  * until next try-lock is likely to succeed.
1677  */
1678 int cl_lock_enclosure(const struct lu_env *env, struct cl_lock *lock,
1679                       struct cl_lock_closure *closure)
1680 {
1681         int result = 0;
1682         ENTRY;
1683         cl_lock_trace(D_DLMTRACE, env, "enclosure lock", lock);
1684         if (!cl_lock_mutex_try(env, lock)) {
1685                 /*
1686                  * If lock->cll_inclosure is not empty, lock is already in
1687                  * this closure.
1688                  */
1689                 if (cfs_list_empty(&lock->cll_inclosure)) {
1690                         cl_lock_get_trust(lock);
1691                         lu_ref_add(&lock->cll_reference, "closure", closure);
1692                         cfs_list_add(&lock->cll_inclosure, &closure->clc_list);
1693                         closure->clc_nr++;
1694                 } else
1695                         cl_lock_mutex_put(env, lock);
1696                 result = 0;
1697         } else {
1698                 cl_lock_disclosure(env, closure);
1699                 if (closure->clc_wait) {
1700                         cl_lock_get_trust(lock);
1701                         lu_ref_add(&lock->cll_reference, "closure-w", closure);
1702                         cl_lock_mutex_put(env, closure->clc_origin);
1703
1704                         LASSERT(cl_lock_nr_mutexed(env) == 0);
1705                         cl_lock_mutex_get(env, lock);
1706                         cl_lock_mutex_put(env, lock);
1707
1708                         cl_lock_mutex_get(env, closure->clc_origin);
1709                         lu_ref_del(&lock->cll_reference, "closure-w", closure);
1710                         cl_lock_put(env, lock);
1711                 }
1712                 result = CLO_REPEAT;
1713         }
1714         RETURN(result);
1715 }
1716 EXPORT_SYMBOL(cl_lock_enclosure);
1717
1718 /** Releases mutices of enclosed locks. */
1719 void cl_lock_disclosure(const struct lu_env *env,
1720                         struct cl_lock_closure *closure)
1721 {
1722         struct cl_lock *scan;
1723         struct cl_lock *temp;
1724
1725         cl_lock_trace(D_DLMTRACE, env, "disclosure lock", closure->clc_origin);
1726         cfs_list_for_each_entry_safe(scan, temp, &closure->clc_list,
1727                                      cll_inclosure){
1728                 cfs_list_del_init(&scan->cll_inclosure);
1729                 cl_lock_mutex_put(env, scan);
1730                 lu_ref_del(&scan->cll_reference, "closure", closure);
1731                 cl_lock_put(env, scan);
1732                 closure->clc_nr--;
1733         }
1734         LASSERT(closure->clc_nr == 0);
1735 }
1736 EXPORT_SYMBOL(cl_lock_disclosure);
1737
1738 /** Finalizes a closure. */
1739 void cl_lock_closure_fini(struct cl_lock_closure *closure)
1740 {
1741         LASSERT(closure->clc_nr == 0);
1742         LASSERT(cfs_list_empty(&closure->clc_list));
1743 }
1744 EXPORT_SYMBOL(cl_lock_closure_fini);
1745
1746 /**
1747  * Destroys this lock. Notifies layers (bottom-to-top) that lock is being
1748  * destroyed, then destroy the lock. If there are holds on the lock, postpone
1749  * destruction until all holds are released. This is called when a decision is
1750  * made to destroy the lock in the future. E.g., when a blocking AST is
1751  * received on it, or fatal communication error happens.
1752  *
1753  * Caller must have a reference on this lock to prevent a situation, when
1754  * deleted lock lingers in memory for indefinite time, because nobody calls
1755  * cl_lock_put() to finish it.
1756  *
1757  * \pre atomic_read(&lock->cll_ref) > 0
1758  * \pre ergo(cl_lock_nesting(lock) == CNL_TOP,
1759  *           cl_lock_nr_mutexed(env) == 1)
1760  *      [i.e., if a top-lock is deleted, mutices of no other locks can be
1761  *      held, as deletion of sub-locks might require releasing a top-lock
1762  *      mutex]
1763  *
1764  * \see cl_lock_operations::clo_delete()
1765  * \see cl_lock::cll_holds
1766  */
1767 void cl_lock_delete(const struct lu_env *env, struct cl_lock *lock)
1768 {
1769         LINVRNT(cl_lock_is_mutexed(lock));
1770         LINVRNT(cl_lock_invariant(env, lock));
1771         LASSERT(ergo(cl_lock_nesting(lock) == CNL_TOP,
1772                      cl_lock_nr_mutexed(env) == 1));
1773
1774         ENTRY;
1775         cl_lock_trace(D_DLMTRACE, env, "delete lock", lock);
1776         if (lock->cll_holds == 0)
1777                 cl_lock_delete0(env, lock);
1778         else
1779                 lock->cll_flags |= CLF_DOOMED;
1780         EXIT;
1781 }
1782 EXPORT_SYMBOL(cl_lock_delete);
1783
1784 /**
1785  * Mark lock as irrecoverably failed, and mark it for destruction. This
1786  * happens when, e.g., server fails to grant a lock to us, or networking
1787  * time-out happens.
1788  *
1789  * \pre atomic_read(&lock->cll_ref) > 0
1790  *
1791  * \see clo_lock_delete()
1792  * \see cl_lock::cll_holds
1793  */
1794 void cl_lock_error(const struct lu_env *env, struct cl_lock *lock, int error)
1795 {
1796         LINVRNT(cl_lock_is_mutexed(lock));
1797         LINVRNT(cl_lock_invariant(env, lock));
1798
1799         ENTRY;
1800         cl_lock_trace(D_DLMTRACE, env, "set lock error", lock);
1801         if (lock->cll_error == 0 && error != 0) {
1802                 lock->cll_error = error;
1803                 cl_lock_signal(env, lock);
1804                 cl_lock_cancel(env, lock);
1805                 cl_lock_delete(env, lock);
1806         }
1807         EXIT;
1808 }
1809 EXPORT_SYMBOL(cl_lock_error);
1810
1811 /**
1812  * Cancels this lock. Notifies layers
1813  * (bottom-to-top) that lock is being cancelled, then destroy the lock. If
1814  * there are holds on the lock, postpone cancellation until
1815  * all holds are released.
1816  *
1817  * Cancellation notification is delivered to layers at most once.
1818  *
1819  * \see cl_lock_operations::clo_cancel()
1820  * \see cl_lock::cll_holds
1821  */
1822 void cl_lock_cancel(const struct lu_env *env, struct cl_lock *lock)
1823 {
1824         LINVRNT(cl_lock_is_mutexed(lock));
1825         LINVRNT(cl_lock_invariant(env, lock));
1826
1827         ENTRY;
1828         cl_lock_trace(D_DLMTRACE, env, "cancel lock", lock);
1829         if (lock->cll_holds == 0)
1830                 cl_lock_cancel0(env, lock);
1831         else
1832                 lock->cll_flags |= CLF_CANCELPEND;
1833         EXIT;
1834 }
1835 EXPORT_SYMBOL(cl_lock_cancel);
1836
1837 /**
1838  * Finds an existing lock covering given page and optionally different from a
1839  * given \a except lock.
1840  */
1841 struct cl_lock *cl_lock_at_page(const struct lu_env *env, struct cl_object *obj,
1842                                 struct cl_page *page, struct cl_lock *except,
1843                                 int pending, int canceld)
1844 {
1845         struct cl_object_header *head;
1846         struct cl_lock          *scan;
1847         struct cl_lock          *lock;
1848         struct cl_lock_descr    *need;
1849
1850         ENTRY;
1851
1852         head = cl_object_header(obj);
1853         need = &cl_env_info(env)->clt_descr;
1854         lock = NULL;
1855
1856         need->cld_mode = CLM_READ; /* CLM_READ matches both READ & WRITE, but
1857                                     * not PHANTOM */
1858         need->cld_start = need->cld_end = page->cp_index;
1859         need->cld_enq_flags = 0;
1860
1861         cfs_spin_lock(&head->coh_lock_guard);
1862         /* It is fine to match any group lock since there could be only one
1863          * with a uniq gid and it conflicts with all other lock modes too */
1864         cfs_list_for_each_entry(scan, &head->coh_locks, cll_linkage) {
1865                 if (scan != except &&
1866                     (scan->cll_descr.cld_mode == CLM_GROUP ||
1867                     cl_lock_ext_match(&scan->cll_descr, need)) &&
1868                     scan->cll_state >= CLS_HELD &&
1869                     scan->cll_state < CLS_FREEING &&
1870                     /*
1871                      * This check is racy as the lock can be canceled right
1872                      * after it is done, but this is fine, because page exists
1873                      * already.
1874                      */
1875                     (canceld || !(scan->cll_flags & CLF_CANCELLED)) &&
1876                     (pending || !(scan->cll_flags & CLF_CANCELPEND))) {
1877                         /* Don't increase cs_hit here since this
1878                          * is just a helper function. */
1879                         cl_lock_get_trust(scan);
1880                         lock = scan;
1881                         break;
1882                 }
1883         }
1884         cfs_spin_unlock(&head->coh_lock_guard);
1885         RETURN(lock);
1886 }
1887 EXPORT_SYMBOL(cl_lock_at_page);
1888
1889 /**
1890  * Calculate the page offset at the layer of @lock.
1891  * At the time of this writing, @page is top page and @lock is sub lock.
1892  */
1893 static pgoff_t pgoff_at_lock(struct cl_page *page, struct cl_lock *lock)
1894 {
1895         struct lu_device_type *dtype;
1896         const struct cl_page_slice *slice;
1897
1898         dtype = lock->cll_descr.cld_obj->co_lu.lo_dev->ld_type;
1899         slice = cl_page_at(page, dtype);
1900         LASSERT(slice != NULL);
1901         return slice->cpl_page->cp_index;
1902 }
1903
1904 /**
1905  * Check if page @page is covered by an extra lock or discard it.
1906  */
1907 static int check_and_discard_cb(const struct lu_env *env, struct cl_io *io,
1908                                 struct cl_page *page, void *cbdata)
1909 {
1910         struct cl_thread_info *info = cl_env_info(env);
1911         struct cl_lock *lock = cbdata;
1912         pgoff_t index = pgoff_at_lock(page, lock);
1913
1914         if (index >= info->clt_fn_index) {
1915                 struct cl_lock *tmp;
1916
1917                 /* refresh non-overlapped index */
1918                 tmp = cl_lock_at_page(env, lock->cll_descr.cld_obj, page, lock,
1919                                       1, 0);
1920                 if (tmp != NULL) {
1921                         /* Cache the first-non-overlapped index so as to skip
1922                          * all pages within [index, clt_fn_index). This
1923                          * is safe because if tmp lock is canceled, it will
1924                          * discard these pages. */
1925                         info->clt_fn_index = tmp->cll_descr.cld_end + 1;
1926                         if (tmp->cll_descr.cld_end == CL_PAGE_EOF)
1927                                 info->clt_fn_index = CL_PAGE_EOF;
1928                         cl_lock_put(env, tmp);
1929                 } else { /* discard the page */
1930                         cl_page_own(env, io, page);
1931                         cl_page_unmap(env, io, page);
1932                         cl_page_discard(env, io, page);
1933                         cl_page_disown(env, io, page);
1934                 }
1935         }
1936
1937         info->clt_next_index = index + 1;
1938         return CLP_GANG_OKAY;
1939 }
1940
1941 static int pageout_cb(const struct lu_env *env, struct cl_io *io,
1942                       struct cl_page *page, void *cbdata)
1943 {
1944         struct cl_thread_info *info  = cl_env_info(env);
1945         struct cl_page_list   *queue = &info->clt_queue.c2_qin;
1946         struct cl_lock        *lock  = cbdata;
1947         typeof(cl_page_own)   *page_own;
1948         int rc = CLP_GANG_OKAY;
1949
1950         page_own = queue->pl_nr ? cl_page_own_try : cl_page_own;
1951         if (page_own(env, io, page) == 0) {
1952                 cl_page_list_add(queue, page);
1953                 info->clt_next_index = pgoff_at_lock(page, lock) + 1;
1954         } else if (page->cp_state != CPS_FREEING) {
1955                 /* cl_page_own() won't fail unless
1956                  * the page is being freed. */
1957                 LASSERT(queue->pl_nr != 0);
1958                 rc = CLP_GANG_AGAIN;
1959         }
1960
1961         return rc;
1962 }
1963
1964 /**
1965  * Invalidate pages protected by the given lock, sending them out to the
1966  * server first, if necessary.
1967  *
1968  * This function does the following:
1969  *
1970  *     - collects a list of pages to be invalidated,
1971  *
1972  *     - unmaps them from the user virtual memory,
1973  *
1974  *     - sends dirty pages to the server,
1975  *
1976  *     - waits for transfer completion,
1977  *
1978  *     - discards pages, and throws them out of memory.
1979  *
1980  * If \a discard is set, pages are discarded without sending them to the
1981  * server.
1982  *
1983  * If error happens on any step, the process continues anyway (the reasoning
1984  * behind this being that lock cancellation cannot be delayed indefinitely).
1985  */
1986 int cl_lock_page_out(const struct lu_env *env, struct cl_lock *lock,
1987                      int discard)
1988 {
1989         struct cl_thread_info *info  = cl_env_info(env);
1990         struct cl_io          *io    = &info->clt_io;
1991         struct cl_2queue      *queue = &info->clt_queue;
1992         struct cl_lock_descr  *descr = &lock->cll_descr;
1993         cl_page_gang_cb_t      cb;
1994         long page_count;
1995         int res;
1996         int result;
1997
1998         LINVRNT(cl_lock_invariant(env, lock));
1999         ENTRY;
2000
2001         io->ci_obj = cl_object_top(descr->cld_obj);
2002         result = cl_io_init(env, io, CIT_MISC, io->ci_obj);
2003         if (result != 0)
2004                 GOTO(out, result);
2005
2006         cb = descr->cld_mode == CLM_READ ? check_and_discard_cb : pageout_cb;
2007         info->clt_fn_index = info->clt_next_index = descr->cld_start;
2008         do {
2009                 cl_2queue_init(queue);
2010                 res = cl_page_gang_lookup(env, descr->cld_obj, io,
2011                                           info->clt_next_index, descr->cld_end,
2012                                           cb, (void *)lock);
2013                 page_count = queue->c2_qin.pl_nr;
2014                 if (page_count > 0) {
2015                         /* must be writeback case */
2016                         LASSERTF(descr->cld_mode >= CLM_WRITE, "lock mode %s\n",
2017                                  cl_lock_mode_name(descr->cld_mode));
2018
2019                         result = cl_page_list_unmap(env, io, &queue->c2_qin);
2020                         if (!discard) {
2021                                 long timeout = 600; /* 10 minutes. */
2022                                 /* for debug purpose, if this request can't be
2023                                  * finished in 10 minutes, we hope it can
2024                                  * notify us.
2025                                  */
2026                                 result = cl_io_submit_sync(env, io, CRT_WRITE,
2027                                                            queue, CRP_CANCEL,
2028                                                            timeout);
2029                                 if (result)
2030                                         CWARN("Writing %lu pages error: %d\n",
2031                                               page_count, result);
2032                         }
2033                         cl_2queue_discard(env, io, queue);
2034                         cl_2queue_disown(env, io, queue);
2035                         cl_2queue_fini(env, queue);
2036                 }
2037
2038                 if (info->clt_next_index > descr->cld_end)
2039                         break;
2040
2041                 if (res == CLP_GANG_RESCHED)
2042                         cfs_cond_resched();
2043         } while (res != CLP_GANG_OKAY);
2044 out:
2045         cl_io_fini(env, io);
2046         RETURN(result);
2047 }
2048 EXPORT_SYMBOL(cl_lock_page_out);
2049
2050 /**
2051  * Eliminate all locks for a given object.
2052  *
2053  * Caller has to guarantee that no lock is in active use.
2054  *
2055  * \param cancel when this is set, cl_locks_prune() cancels locks before
2056  *               destroying.
2057  */
2058 void cl_locks_prune(const struct lu_env *env, struct cl_object *obj, int cancel)
2059 {
2060         struct cl_object_header *head;
2061         struct cl_lock          *lock;
2062
2063         ENTRY;
2064         head = cl_object_header(obj);
2065         /*
2066          * If locks are destroyed without cancellation, all pages must be
2067          * already destroyed (as otherwise they will be left unprotected).
2068          */
2069         LASSERT(ergo(!cancel,
2070                      head->coh_tree.rnode == NULL && head->coh_pages == 0));
2071
2072         cfs_spin_lock(&head->coh_lock_guard);
2073         while (!cfs_list_empty(&head->coh_locks)) {
2074                 lock = container_of(head->coh_locks.next,
2075                                     struct cl_lock, cll_linkage);
2076                 cl_lock_get_trust(lock);
2077                 cfs_spin_unlock(&head->coh_lock_guard);
2078                 lu_ref_add(&lock->cll_reference, "prune", cfs_current());
2079                 cl_lock_mutex_get(env, lock);
2080                 if (lock->cll_state < CLS_FREEING) {
2081                         LASSERT(lock->cll_holds == 0);
2082                         LASSERT(lock->cll_users == 0);
2083                         if (cancel)
2084                                 cl_lock_cancel(env, lock);
2085                         cl_lock_delete(env, lock);
2086                 }
2087                 cl_lock_mutex_put(env, lock);
2088                 lu_ref_del(&lock->cll_reference, "prune", cfs_current());
2089                 cl_lock_put(env, lock);
2090                 cfs_spin_lock(&head->coh_lock_guard);
2091         }
2092         cfs_spin_unlock(&head->coh_lock_guard);
2093         EXIT;
2094 }
2095 EXPORT_SYMBOL(cl_locks_prune);
2096
2097 static struct cl_lock *cl_lock_hold_mutex(const struct lu_env *env,
2098                                           const struct cl_io *io,
2099                                           const struct cl_lock_descr *need,
2100                                           const char *scope, const void *source)
2101 {
2102         struct cl_lock *lock;
2103
2104         ENTRY;
2105
2106         while (1) {
2107                 lock = cl_lock_find(env, io, need);
2108                 if (IS_ERR(lock))
2109                         break;
2110                 cl_lock_mutex_get(env, lock);
2111                 if (lock->cll_state < CLS_FREEING &&
2112                     !(lock->cll_flags & CLF_CANCELLED)) {
2113                         cl_lock_hold_mod(env, lock, +1);
2114                         lu_ref_add(&lock->cll_holders, scope, source);
2115                         lu_ref_add(&lock->cll_reference, scope, source);
2116                         break;
2117                 }
2118                 cl_lock_mutex_put(env, lock);
2119                 cl_lock_put(env, lock);
2120         }
2121         RETURN(lock);
2122 }
2123
2124 /**
2125  * Returns a lock matching \a need description with a reference and a hold on
2126  * it.
2127  *
2128  * This is much like cl_lock_find(), except that cl_lock_hold() additionally
2129  * guarantees that lock is not in the CLS_FREEING state on return.
2130  */
2131 struct cl_lock *cl_lock_hold(const struct lu_env *env, const struct cl_io *io,
2132                              const struct cl_lock_descr *need,
2133                              const char *scope, const void *source)
2134 {
2135         struct cl_lock *lock;
2136
2137         ENTRY;
2138
2139         lock = cl_lock_hold_mutex(env, io, need, scope, source);
2140         if (!IS_ERR(lock))
2141                 cl_lock_mutex_put(env, lock);
2142         RETURN(lock);
2143 }
2144 EXPORT_SYMBOL(cl_lock_hold);
2145
2146 /**
2147  * Main high-level entry point of cl_lock interface that finds existing or
2148  * enqueues new lock matching given description.
2149  */
2150 struct cl_lock *cl_lock_request(const struct lu_env *env, struct cl_io *io,
2151                                 const struct cl_lock_descr *need,
2152                                 const char *scope, const void *source)
2153 {
2154         struct cl_lock       *lock;
2155         int                   rc;
2156         __u32                 enqflags = need->cld_enq_flags;
2157
2158         ENTRY;
2159         do {
2160                 lock = cl_lock_hold_mutex(env, io, need, scope, source);
2161                 if (IS_ERR(lock))
2162                         break;
2163
2164                 rc = cl_enqueue_locked(env, lock, io, enqflags);
2165                 if (rc == 0) {
2166                         if (cl_lock_fits_into(env, lock, need, io)) {
2167                                 if (!(enqflags & CEF_AGL)) {
2168                                         cl_lock_mutex_put(env, lock);
2169                                         cl_lock_lockdep_acquire(env, lock,
2170                                                                 enqflags);
2171                                         break;
2172                                 }
2173                                 rc = 1;
2174                         }
2175                         cl_unuse_locked(env, lock);
2176                 }
2177                 cl_lock_trace(D_DLMTRACE, env,
2178                               rc <= 0 ? "enqueue failed" : "agl succeed", lock);
2179                 cl_lock_hold_release(env, lock, scope, source);
2180                 cl_lock_mutex_put(env, lock);
2181                 lu_ref_del(&lock->cll_reference, scope, source);
2182                 cl_lock_put(env, lock);
2183                 if (rc > 0) {
2184                         LASSERT(enqflags & CEF_AGL);
2185                         lock = NULL;
2186                 } else if (rc != 0) {
2187                         lock = ERR_PTR(rc);
2188                 }
2189         } while (rc == 0);
2190         RETURN(lock);
2191 }
2192 EXPORT_SYMBOL(cl_lock_request);
2193
2194 /**
2195  * Adds a hold to a known lock.
2196  */
2197 void cl_lock_hold_add(const struct lu_env *env, struct cl_lock *lock,
2198                       const char *scope, const void *source)
2199 {
2200         LINVRNT(cl_lock_is_mutexed(lock));
2201         LINVRNT(cl_lock_invariant(env, lock));
2202         LASSERT(lock->cll_state != CLS_FREEING);
2203
2204         ENTRY;
2205         cl_lock_hold_mod(env, lock, +1);
2206         cl_lock_get(lock);
2207         lu_ref_add(&lock->cll_holders, scope, source);
2208         lu_ref_add(&lock->cll_reference, scope, source);
2209         EXIT;
2210 }
2211 EXPORT_SYMBOL(cl_lock_hold_add);
2212
2213 /**
2214  * Releases a hold and a reference on a lock, on which caller acquired a
2215  * mutex.
2216  */
2217 void cl_lock_unhold(const struct lu_env *env, struct cl_lock *lock,
2218                     const char *scope, const void *source)
2219 {
2220         LINVRNT(cl_lock_invariant(env, lock));
2221         ENTRY;
2222         cl_lock_hold_release(env, lock, scope, source);
2223         lu_ref_del(&lock->cll_reference, scope, source);
2224         cl_lock_put(env, lock);
2225         EXIT;
2226 }
2227 EXPORT_SYMBOL(cl_lock_unhold);
2228
2229 /**
2230  * Releases a hold and a reference on a lock, obtained by cl_lock_hold().
2231  */
2232 void cl_lock_release(const struct lu_env *env, struct cl_lock *lock,
2233                      const char *scope, const void *source)
2234 {
2235         LINVRNT(cl_lock_invariant(env, lock));
2236         ENTRY;
2237         cl_lock_trace(D_DLMTRACE, env, "release lock", lock);
2238         cl_lock_mutex_get(env, lock);
2239         cl_lock_hold_release(env, lock, scope, source);
2240         cl_lock_mutex_put(env, lock);
2241         lu_ref_del(&lock->cll_reference, scope, source);
2242         cl_lock_put(env, lock);
2243         EXIT;
2244 }
2245 EXPORT_SYMBOL(cl_lock_release);
2246
2247 void cl_lock_user_add(const struct lu_env *env, struct cl_lock *lock)
2248 {
2249         LINVRNT(cl_lock_is_mutexed(lock));
2250         LINVRNT(cl_lock_invariant(env, lock));
2251
2252         ENTRY;
2253         cl_lock_used_mod(env, lock, +1);
2254         EXIT;
2255 }
2256 EXPORT_SYMBOL(cl_lock_user_add);
2257
2258 int cl_lock_user_del(const struct lu_env *env, struct cl_lock *lock)
2259 {
2260         LINVRNT(cl_lock_is_mutexed(lock));
2261         LINVRNT(cl_lock_invariant(env, lock));
2262         LASSERT(lock->cll_users > 0);
2263
2264         ENTRY;
2265         cl_lock_used_mod(env, lock, -1);
2266         RETURN(lock->cll_users == 0);
2267 }
2268 EXPORT_SYMBOL(cl_lock_user_del);
2269
2270 const char *cl_lock_mode_name(const enum cl_lock_mode mode)
2271 {
2272         static const char *names[] = {
2273                 [CLM_PHANTOM] = "P",
2274                 [CLM_READ]    = "R",
2275                 [CLM_WRITE]   = "W",
2276                 [CLM_GROUP]   = "G"
2277         };
2278         if (0 <= mode && mode < ARRAY_SIZE(names))
2279                 return names[mode];
2280         else
2281                 return "U";
2282 }
2283 EXPORT_SYMBOL(cl_lock_mode_name);
2284
2285 /**
2286  * Prints human readable representation of a lock description.
2287  */
2288 void cl_lock_descr_print(const struct lu_env *env, void *cookie,
2289                        lu_printer_t printer,
2290                        const struct cl_lock_descr *descr)
2291 {
2292         const struct lu_fid  *fid;
2293
2294         fid = lu_object_fid(&descr->cld_obj->co_lu);
2295         (*printer)(env, cookie, DDESCR"@"DFID, PDESCR(descr), PFID(fid));
2296 }
2297 EXPORT_SYMBOL(cl_lock_descr_print);
2298
2299 /**
2300  * Prints human readable representation of \a lock to the \a f.
2301  */
2302 void cl_lock_print(const struct lu_env *env, void *cookie,
2303                    lu_printer_t printer, const struct cl_lock *lock)
2304 {
2305         const struct cl_lock_slice *slice;
2306         (*printer)(env, cookie, "lock@%p[%d %d %d %d %d %08lx] ",
2307                    lock, cfs_atomic_read(&lock->cll_ref),
2308                    lock->cll_state, lock->cll_error, lock->cll_holds,
2309                    lock->cll_users, lock->cll_flags);
2310         cl_lock_descr_print(env, cookie, printer, &lock->cll_descr);
2311         (*printer)(env, cookie, " {\n");
2312
2313         cfs_list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
2314                 (*printer)(env, cookie, "    %s@%p: ",
2315                            slice->cls_obj->co_lu.lo_dev->ld_type->ldt_name,
2316                            slice);
2317                 if (slice->cls_ops->clo_print != NULL)
2318                         slice->cls_ops->clo_print(env, cookie, printer, slice);
2319                 (*printer)(env, cookie, "\n");
2320         }
2321         (*printer)(env, cookie, "} lock@%p\n", lock);
2322 }
2323 EXPORT_SYMBOL(cl_lock_print);
2324
2325 int cl_lock_init(void)
2326 {
2327         return lu_kmem_init(cl_lock_caches);
2328 }
2329
2330 void cl_lock_fini(void)
2331 {
2332         lu_kmem_fini(cl_lock_caches);
2333 }