Whamcloud - gitweb
LU-10841 ldlm: ASSERTION(lock->l_granted_mode!=lock->l_req_mode)
[fs/lustre-release.git] / lustre / ldlm / ldlm_lock.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2010, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * lustre/ldlm/ldlm_lock.c
33  *
34  * Author: Peter Braam <braam@clusterfs.com>
35  * Author: Phil Schwan <phil@clusterfs.com>
36  */
37
38 #define DEBUG_SUBSYSTEM S_LDLM
39
40 #include <libcfs/libcfs.h>
41
42 #include <lustre_swab.h>
43 #include <obd_class.h>
44
45 #include "ldlm_internal.h"
46
47 struct kmem_cache *ldlm_glimpse_work_kmem;
48 EXPORT_SYMBOL(ldlm_glimpse_work_kmem);
49
50 /* lock types */
51 char *ldlm_lockname[] = {
52         [0] = "--",
53         [LCK_EX] = "EX",
54         [LCK_PW] = "PW",
55         [LCK_PR] = "PR",
56         [LCK_CW] = "CW",
57         [LCK_CR] = "CR",
58         [LCK_NL] = "NL",
59         [LCK_GROUP] = "GROUP",
60         [LCK_COS] = "COS"
61 };
62 EXPORT_SYMBOL(ldlm_lockname);
63
64 char *ldlm_typename[] = {
65         [LDLM_PLAIN] = "PLN",
66         [LDLM_EXTENT] = "EXT",
67         [LDLM_FLOCK] = "FLK",
68         [LDLM_IBITS] = "IBT",
69 };
70
71 static ldlm_policy_wire_to_local_t ldlm_policy_wire_to_local[] = {
72         [LDLM_PLAIN - LDLM_MIN_TYPE]  = ldlm_plain_policy_wire_to_local,
73         [LDLM_EXTENT - LDLM_MIN_TYPE] = ldlm_extent_policy_wire_to_local,
74         [LDLM_FLOCK - LDLM_MIN_TYPE]  = ldlm_flock_policy_wire_to_local,
75         [LDLM_IBITS - LDLM_MIN_TYPE]  = ldlm_ibits_policy_wire_to_local,
76 };
77
78 static ldlm_policy_local_to_wire_t ldlm_policy_local_to_wire[] = {
79         [LDLM_PLAIN - LDLM_MIN_TYPE]  = ldlm_plain_policy_local_to_wire,
80         [LDLM_EXTENT - LDLM_MIN_TYPE] = ldlm_extent_policy_local_to_wire,
81         [LDLM_FLOCK - LDLM_MIN_TYPE]  = ldlm_flock_policy_local_to_wire,
82         [LDLM_IBITS - LDLM_MIN_TYPE]  = ldlm_ibits_policy_local_to_wire,
83 };
84
85 /**
86  * Converts lock policy from local format to on the wire lock_desc format
87  */
88 void ldlm_convert_policy_to_wire(enum ldlm_type type,
89                                  const union ldlm_policy_data *lpolicy,
90                                  union ldlm_wire_policy_data *wpolicy)
91 {
92         ldlm_policy_local_to_wire_t convert;
93
94         convert = ldlm_policy_local_to_wire[type - LDLM_MIN_TYPE];
95
96         convert(lpolicy, wpolicy);
97 }
98
99 /**
100  * Converts lock policy from on the wire lock_desc format to local format
101  */
102 void ldlm_convert_policy_to_local(struct obd_export *exp, enum ldlm_type type,
103                                   const union ldlm_wire_policy_data *wpolicy,
104                                   union ldlm_policy_data *lpolicy)
105 {
106         ldlm_policy_wire_to_local_t convert;
107
108         convert = ldlm_policy_wire_to_local[type - LDLM_MIN_TYPE];
109
110         convert(wpolicy, lpolicy);
111 }
112
113 const char *ldlm_it2str(enum ldlm_intent_flags it)
114 {
115         switch (it) {
116         case IT_OPEN:
117                 return "open";
118         case IT_CREAT:
119                 return "creat";
120         case (IT_OPEN | IT_CREAT):
121                 return "open|creat";
122         case IT_READDIR:
123                 return "readdir";
124         case IT_GETATTR:
125                 return "getattr";
126         case IT_LOOKUP:
127                 return "lookup";
128         case IT_UNLINK:
129                 return "unlink";
130         case IT_GETXATTR:
131                 return "getxattr";
132         case IT_LAYOUT:
133                 return "layout";
134         default:
135                 CERROR("Unknown intent 0x%08x\n", it);
136                 return "UNKNOWN";
137         }
138 }
139 EXPORT_SYMBOL(ldlm_it2str);
140
141 extern struct kmem_cache *ldlm_lock_slab;
142
143 #ifdef HAVE_SERVER_SUPPORT
144 static ldlm_processing_policy ldlm_processing_policy_table[] = {
145         [LDLM_PLAIN]    = ldlm_process_plain_lock,
146         [LDLM_EXTENT]   = ldlm_process_extent_lock,
147         [LDLM_FLOCK]    = ldlm_process_flock_lock,
148         [LDLM_IBITS]    = ldlm_process_inodebits_lock,
149 };
150
151 ldlm_processing_policy ldlm_get_processing_policy(struct ldlm_resource *res)
152 {
153         return ldlm_processing_policy_table[res->lr_type];
154 }
155 EXPORT_SYMBOL(ldlm_get_processing_policy);
156 #endif /* HAVE_SERVER_SUPPORT */
157
158 void ldlm_register_intent(struct ldlm_namespace *ns, ldlm_res_policy arg)
159 {
160         ns->ns_policy = arg;
161 }
162 EXPORT_SYMBOL(ldlm_register_intent);
163
164 /*
165  * REFCOUNTED LOCK OBJECTS
166  */
167
168
169 /**
170  * Get a reference on a lock.
171  *
172  * Lock refcounts, during creation:
173  *   - one special one for allocation, dec'd only once in destroy
174  *   - one for being a lock that's in-use
175  *   - one for the addref associated with a new lock
176  */
177 struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock)
178 {
179         atomic_inc(&lock->l_refc);
180         return lock;
181 }
182 EXPORT_SYMBOL(ldlm_lock_get);
183
184 /**
185  * Release lock reference.
186  *
187  * Also frees the lock if it was last reference.
188  */
189 void ldlm_lock_put(struct ldlm_lock *lock)
190 {
191         ENTRY;
192
193         LASSERT(lock->l_resource != LP_POISON);
194         LASSERT(atomic_read(&lock->l_refc) > 0);
195         if (atomic_dec_and_test(&lock->l_refc)) {
196                 struct ldlm_resource *res;
197
198                 LDLM_DEBUG(lock,
199                            "final lock_put on destroyed lock, freeing it.");
200
201                 res = lock->l_resource;
202                 LASSERT(ldlm_is_destroyed(lock));
203                 LASSERT(list_empty(&lock->l_exp_list));
204                 LASSERT(list_empty(&lock->l_res_link));
205                 LASSERT(list_empty(&lock->l_pending_chain));
206
207                 lprocfs_counter_decr(ldlm_res_to_ns(res)->ns_stats,
208                                      LDLM_NSS_LOCKS);
209                 lu_ref_del(&res->lr_reference, "lock", lock);
210                 ldlm_resource_putref(res);
211                 lock->l_resource = NULL;
212                 if (lock->l_export) {
213                         class_export_lock_put(lock->l_export, lock);
214                         lock->l_export = NULL;
215                 }
216
217                 if (lock->l_lvb_data != NULL)
218                         OBD_FREE_LARGE(lock->l_lvb_data, lock->l_lvb_len);
219
220                 ldlm_interval_free(ldlm_interval_detach(lock));
221                 lu_ref_fini(&lock->l_reference);
222                 OBD_FREE_RCU(lock, sizeof(*lock), &lock->l_handle);
223         }
224
225         EXIT;
226 }
227 EXPORT_SYMBOL(ldlm_lock_put);
228
229 /**
230  * Removes LDLM lock \a lock from LRU. Assumes LRU is already locked.
231  */
232 int ldlm_lock_remove_from_lru_nolock(struct ldlm_lock *lock)
233 {
234         int rc = 0;
235         if (!list_empty(&lock->l_lru)) {
236                 struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
237
238                 LASSERT(lock->l_resource->lr_type != LDLM_FLOCK);
239                 list_del_init(&lock->l_lru);
240                 LASSERT(ns->ns_nr_unused > 0);
241                 ns->ns_nr_unused--;
242                 rc = 1;
243         }
244         return rc;
245 }
246
247 /**
248  * Removes LDLM lock \a lock from LRU. Obtains the LRU lock first.
249  *
250  * If \a last_use is non-zero, it will remove the lock from LRU only if
251  * it matches lock's l_last_used.
252  *
253  * \retval 0 if \a last_use is set, the lock is not in LRU list or \a last_use
254  *           doesn't match lock's l_last_used;
255  *           otherwise, the lock hasn't been in the LRU list.
256  * \retval 1 the lock was in LRU list and removed.
257  */
258 int ldlm_lock_remove_from_lru_check(struct ldlm_lock *lock, ktime_t last_use)
259 {
260         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
261         int rc = 0;
262
263         ENTRY;
264         if (ldlm_is_ns_srv(lock)) {
265                 LASSERT(list_empty(&lock->l_lru));
266                 RETURN(0);
267         }
268
269         spin_lock(&ns->ns_lock);
270         if (!ktime_compare(last_use, ktime_set(0, 0)) ||
271             !ktime_compare(last_use, lock->l_last_used))
272                 rc = ldlm_lock_remove_from_lru_nolock(lock);
273         spin_unlock(&ns->ns_lock);
274
275         RETURN(rc);
276 }
277
278 /**
279  * Adds LDLM lock \a lock to namespace LRU. Assumes LRU is already locked.
280  */
281 void ldlm_lock_add_to_lru_nolock(struct ldlm_lock *lock)
282 {
283         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
284
285         lock->l_last_used = ktime_get();
286         LASSERT(list_empty(&lock->l_lru));
287         LASSERT(lock->l_resource->lr_type != LDLM_FLOCK);
288         list_add_tail(&lock->l_lru, &ns->ns_unused_list);
289         ldlm_clear_skipped(lock);
290         LASSERT(ns->ns_nr_unused >= 0);
291         ns->ns_nr_unused++;
292 }
293
294 /**
295  * Adds LDLM lock \a lock to namespace LRU. Obtains necessary LRU locks
296  * first.
297  */
298 void ldlm_lock_add_to_lru(struct ldlm_lock *lock)
299 {
300         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
301
302         ENTRY;
303         spin_lock(&ns->ns_lock);
304         ldlm_lock_add_to_lru_nolock(lock);
305         spin_unlock(&ns->ns_lock);
306         EXIT;
307 }
308
309 /**
310  * Moves LDLM lock \a lock that is already in namespace LRU to the tail of
311  * the LRU. Performs necessary LRU locking
312  */
313 void ldlm_lock_touch_in_lru(struct ldlm_lock *lock)
314 {
315         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
316
317         ENTRY;
318         if (ldlm_is_ns_srv(lock)) {
319                 LASSERT(list_empty(&lock->l_lru));
320                 EXIT;
321                 return;
322         }
323
324         spin_lock(&ns->ns_lock);
325         if (!list_empty(&lock->l_lru)) {
326                 ldlm_lock_remove_from_lru_nolock(lock);
327                 ldlm_lock_add_to_lru_nolock(lock);
328         }
329         spin_unlock(&ns->ns_lock);
330         EXIT;
331 }
332
333 /**
334  * Helper to destroy a locked lock.
335  *
336  * Used by ldlm_lock_destroy and ldlm_lock_destroy_nolock
337  * Must be called with l_lock and lr_lock held.
338  *
339  * Does not actually free the lock data, but rather marks the lock as
340  * destroyed by setting l_destroyed field in the lock to 1.  Destroys a
341  * handle->lock association too, so that the lock can no longer be found
342  * and removes the lock from LRU list.  Actual lock freeing occurs when
343  * last lock reference goes away.
344  *
345  * Original comment (of some historical value):
346  * This used to have a 'strict' flag, which recovery would use to mark an
347  * in-use lock as needing-to-die.  Lest I am ever tempted to put it back, I
348  * shall explain why it's gone: with the new hash table scheme, once you call
349  * ldlm_lock_destroy, you can never drop your final references on this lock.
350  * Because it's not in the hash table anymore.  -phil
351  */
352 static int ldlm_lock_destroy_internal(struct ldlm_lock *lock)
353 {
354         ENTRY;
355
356         if (lock->l_readers || lock->l_writers) {
357                 LDLM_ERROR(lock, "lock still has references");
358                 LBUG();
359         }
360
361         if (!list_empty(&lock->l_res_link)) {
362                 LDLM_ERROR(lock, "lock still on resource");
363                 LBUG();
364         }
365
366         if (ldlm_is_destroyed(lock)) {
367                 LASSERT(list_empty(&lock->l_lru));
368                 EXIT;
369                 return 0;
370         }
371         ldlm_set_destroyed(lock);
372
373         if (lock->l_export && lock->l_export->exp_lock_hash) {
374                 /* NB: it's safe to call cfs_hash_del() even lock isn't
375                  * in exp_lock_hash. */
376                 /* In the function below, .hs_keycmp resolves to
377                  * ldlm_export_lock_keycmp() */
378                 /* coverity[overrun-buffer-val] */
379                 cfs_hash_del(lock->l_export->exp_lock_hash,
380                              &lock->l_remote_handle, &lock->l_exp_hash);
381         }
382
383         ldlm_lock_remove_from_lru(lock);
384         class_handle_unhash(&lock->l_handle);
385
386         EXIT;
387         return 1;
388 }
389
390 /**
391  * Destroys a LDLM lock \a lock. Performs necessary locking first.
392  */
393 void ldlm_lock_destroy(struct ldlm_lock *lock)
394 {
395         int first;
396         ENTRY;
397         lock_res_and_lock(lock);
398         first = ldlm_lock_destroy_internal(lock);
399         unlock_res_and_lock(lock);
400
401         /* drop reference from hashtable only for first destroy */
402         if (first) {
403                 lu_ref_del(&lock->l_reference, "hash", lock);
404                 LDLM_LOCK_RELEASE(lock);
405         }
406         EXIT;
407 }
408
409 /**
410  * Destroys a LDLM lock \a lock that is already locked.
411  */
412 void ldlm_lock_destroy_nolock(struct ldlm_lock *lock)
413 {
414         int first;
415         ENTRY;
416         first = ldlm_lock_destroy_internal(lock);
417         /* drop reference from hashtable only for first destroy */
418         if (first) {
419                 lu_ref_del(&lock->l_reference, "hash", lock);
420                 LDLM_LOCK_RELEASE(lock);
421         }
422         EXIT;
423 }
424
425 /* this is called by portals_handle2object with the handle lock taken */
426 static void lock_handle_addref(void *lock)
427 {
428         LDLM_LOCK_GET((struct ldlm_lock *)lock);
429 }
430
431 static void lock_handle_free(void *lock, int size)
432 {
433         LASSERT(size == sizeof(struct ldlm_lock));
434         OBD_SLAB_FREE(lock, ldlm_lock_slab, size);
435 }
436
437 static struct portals_handle_ops lock_handle_ops = {
438         .hop_addref = lock_handle_addref,
439         .hop_free   = lock_handle_free,
440 };
441
442 /**
443  *
444  * Allocate and initialize new lock structure.
445  *
446  * usage: pass in a resource on which you have done ldlm_resource_get
447  *        new lock will take over the refcount.
448  * returns: lock with refcount 2 - one for current caller and one for remote
449  */
450 static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource)
451 {
452         struct ldlm_lock *lock;
453         ENTRY;
454
455         if (resource == NULL)
456                 LBUG();
457
458         OBD_SLAB_ALLOC_PTR_GFP(lock, ldlm_lock_slab, GFP_NOFS);
459         if (lock == NULL)
460                 RETURN(NULL);
461
462         spin_lock_init(&lock->l_lock);
463         lock->l_resource = resource;
464         lu_ref_add(&resource->lr_reference, "lock", lock);
465
466         atomic_set(&lock->l_refc, 2);
467         INIT_LIST_HEAD(&lock->l_res_link);
468         INIT_LIST_HEAD(&lock->l_lru);
469         INIT_LIST_HEAD(&lock->l_pending_chain);
470         INIT_LIST_HEAD(&lock->l_bl_ast);
471         INIT_LIST_HEAD(&lock->l_cp_ast);
472         INIT_LIST_HEAD(&lock->l_rk_ast);
473         init_waitqueue_head(&lock->l_waitq);
474         lock->l_blocking_lock = NULL;
475         INIT_LIST_HEAD(&lock->l_sl_mode);
476         INIT_LIST_HEAD(&lock->l_sl_policy);
477         INIT_HLIST_NODE(&lock->l_exp_hash);
478         INIT_HLIST_NODE(&lock->l_exp_flock_hash);
479
480         lprocfs_counter_incr(ldlm_res_to_ns(resource)->ns_stats,
481                              LDLM_NSS_LOCKS);
482         INIT_LIST_HEAD(&lock->l_handle.h_link);
483         class_handle_hash(&lock->l_handle, &lock_handle_ops);
484
485         lu_ref_init(&lock->l_reference);
486         lu_ref_add(&lock->l_reference, "hash", lock);
487         lock->l_callback_timeout = 0;
488
489 #if LUSTRE_TRACKS_LOCK_EXP_REFS
490         INIT_LIST_HEAD(&lock->l_exp_refs_link);
491         lock->l_exp_refs_nr = 0;
492         lock->l_exp_refs_target = NULL;
493 #endif
494         INIT_LIST_HEAD(&lock->l_exp_list);
495
496         RETURN(lock);
497 }
498
499 /**
500  * Moves LDLM lock \a lock to another resource.
501  * This is used on client when server returns some other lock than requested
502  * (typically as a result of intent operation)
503  */
504 int ldlm_lock_change_resource(struct ldlm_namespace *ns, struct ldlm_lock *lock,
505                               const struct ldlm_res_id *new_resid)
506 {
507         struct ldlm_resource *oldres = lock->l_resource;
508         struct ldlm_resource *newres;
509         int type;
510         ENTRY;
511
512         LASSERT(ns_is_client(ns));
513
514         lock_res_and_lock(lock);
515         if (memcmp(new_resid, &lock->l_resource->lr_name,
516                    sizeof(lock->l_resource->lr_name)) == 0) {
517                 /* Nothing to do */
518                 unlock_res_and_lock(lock);
519                 RETURN(0);
520         }
521
522         LASSERT(new_resid->name[0] != 0);
523
524         /* This function assumes that the lock isn't on any lists */
525         LASSERT(list_empty(&lock->l_res_link));
526
527         type = oldres->lr_type;
528         unlock_res_and_lock(lock);
529
530         newres = ldlm_resource_get(ns, NULL, new_resid, type, 1);
531         if (IS_ERR(newres))
532                 RETURN(PTR_ERR(newres));
533
534         lu_ref_add(&newres->lr_reference, "lock", lock);
535         /*
536          * To flip the lock from the old to the new resource, lock, oldres and
537          * newres have to be locked. Resource spin-locks are nested within
538          * lock->l_lock, and are taken in the memory address order to avoid
539          * dead-locks.
540          */
541         spin_lock(&lock->l_lock);
542         oldres = lock->l_resource;
543         if (oldres < newres) {
544                 lock_res(oldres);
545                 lock_res_nested(newres, LRT_NEW);
546         } else {
547                 lock_res(newres);
548                 lock_res_nested(oldres, LRT_NEW);
549         }
550         LASSERT(memcmp(new_resid, &oldres->lr_name,
551                        sizeof oldres->lr_name) != 0);
552         lock->l_resource = newres;
553         unlock_res(oldres);
554         unlock_res_and_lock(lock);
555
556         /* ...and the flowers are still standing! */
557         lu_ref_del(&oldres->lr_reference, "lock", lock);
558         ldlm_resource_putref(oldres);
559
560         RETURN(0);
561 }
562
563 /** \defgroup ldlm_handles LDLM HANDLES
564  * Ways to get hold of locks without any addresses.
565  * @{
566  */
567
568 /**
569  * Fills in handle for LDLM lock \a lock into supplied \a lockh
570  * Does not take any references.
571  */
572 void ldlm_lock2handle(const struct ldlm_lock *lock, struct lustre_handle *lockh)
573 {
574         lockh->cookie = lock->l_handle.h_cookie;
575 }
576 EXPORT_SYMBOL(ldlm_lock2handle);
577
578 /**
579  * Obtain a lock reference by handle.
580  *
581  * if \a flags: atomically get the lock and set the flags.
582  *              Return NULL if flag already set
583  */
584 struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle,
585                                      __u64 flags)
586 {
587         struct ldlm_lock *lock;
588         ENTRY;
589
590         LASSERT(handle);
591
592         lock = class_handle2object(handle->cookie, NULL);
593         if (lock == NULL)
594                 RETURN(NULL);
595
596         if (lock->l_export != NULL && lock->l_export->exp_failed) {
597                 CDEBUG(D_INFO, "lock export failed: lock %p, exp %p\n",
598                        lock, lock->l_export);
599                 LDLM_LOCK_PUT(lock);
600                 RETURN(NULL);
601         }
602
603         /* It's unlikely but possible that someone marked the lock as
604          * destroyed after we did handle2object on it */
605         if ((flags == 0) && !ldlm_is_destroyed(lock)) {
606                 lu_ref_add(&lock->l_reference, "handle", current);
607                 RETURN(lock);
608         }
609
610         lock_res_and_lock(lock);
611
612         LASSERT(lock->l_resource != NULL);
613
614         lu_ref_add_atomic(&lock->l_reference, "handle", current);
615         if (unlikely(ldlm_is_destroyed(lock))) {
616                 unlock_res_and_lock(lock);
617                 CDEBUG(D_INFO, "lock already destroyed: lock %p\n", lock);
618                 LDLM_LOCK_PUT(lock);
619                 RETURN(NULL);
620         }
621
622         /* If we're setting flags, make sure none of them are already set. */
623         if (flags != 0) {
624                 if ((lock->l_flags & flags) != 0) {
625                         unlock_res_and_lock(lock);
626                         LDLM_LOCK_PUT(lock);
627                         RETURN(NULL);
628                 }
629
630                 lock->l_flags |= flags;
631         }
632
633         unlock_res_and_lock(lock);
634         RETURN(lock);
635 }
636 EXPORT_SYMBOL(__ldlm_handle2lock);
637 /** @} ldlm_handles */
638
639 /**
640  * Fill in "on the wire" representation for given LDLM lock into supplied
641  * lock descriptor \a desc structure.
642  */
643 void ldlm_lock2desc(struct ldlm_lock *lock, struct ldlm_lock_desc *desc)
644 {
645         ldlm_res2desc(lock->l_resource, &desc->l_resource);
646         desc->l_req_mode = lock->l_req_mode;
647         desc->l_granted_mode = lock->l_granted_mode;
648         ldlm_convert_policy_to_wire(lock->l_resource->lr_type,
649                                     &lock->l_policy_data,
650                                     &desc->l_policy_data);
651 }
652
653 /**
654  * Add a lock to list of conflicting locks to send AST to.
655  *
656  * Only add if we have not sent a blocking AST to the lock yet.
657  */
658 static void ldlm_add_bl_work_item(struct ldlm_lock *lock, struct ldlm_lock *new,
659                                   struct list_head *work_list)
660 {
661         if (!ldlm_is_ast_sent(lock)) {
662                 LDLM_DEBUG(lock, "lock incompatible; sending blocking AST.");
663                 ldlm_set_ast_sent(lock);
664                 /* If the enqueuing client said so, tell the AST recipient to
665                  * discard dirty data, rather than writing back. */
666                 if (ldlm_is_ast_discard_data(new))
667                         ldlm_set_discard_data(lock);
668                 LASSERT(list_empty(&lock->l_bl_ast));
669                 list_add(&lock->l_bl_ast, work_list);
670                 LDLM_LOCK_GET(lock);
671                 LASSERT(lock->l_blocking_lock == NULL);
672                 lock->l_blocking_lock = LDLM_LOCK_GET(new);
673         }
674 }
675
676 /**
677  * Add a lock to list of just granted locks to send completion AST to.
678  */
679 static void ldlm_add_cp_work_item(struct ldlm_lock *lock,
680                                   struct list_head *work_list)
681 {
682         if (!ldlm_is_cp_reqd(lock)) {
683                 ldlm_set_cp_reqd(lock);
684                 LDLM_DEBUG(lock, "lock granted; sending completion AST.");
685                 LASSERT(list_empty(&lock->l_cp_ast));
686                 list_add(&lock->l_cp_ast, work_list);
687                 LDLM_LOCK_GET(lock);
688         }
689 }
690
691 /**
692  * Aggregator function to add AST work items into a list. Determines
693  * what sort of an AST work needs to be done and calls the proper
694  * adding function.
695  * Must be called with lr_lock held.
696  */
697 void ldlm_add_ast_work_item(struct ldlm_lock *lock, struct ldlm_lock *new,
698                             struct list_head *work_list)
699 {
700         ENTRY;
701         check_res_locked(lock->l_resource);
702         if (new)
703                 ldlm_add_bl_work_item(lock, new, work_list);
704         else
705                 ldlm_add_cp_work_item(lock, work_list);
706         EXIT;
707 }
708
709 /**
710  * Add specified reader/writer reference to LDLM lock with handle \a lockh.
711  * r/w reference type is determined by \a mode
712  * Calls ldlm_lock_addref_internal.
713  */
714 void ldlm_lock_addref(const struct lustre_handle *lockh, enum ldlm_mode mode)
715 {
716         struct ldlm_lock *lock;
717
718         lock = ldlm_handle2lock(lockh);
719         LASSERTF(lock != NULL, "Non-existing lock: %#llx\n", lockh->cookie);
720         ldlm_lock_addref_internal(lock, mode);
721         LDLM_LOCK_PUT(lock);
722 }
723 EXPORT_SYMBOL(ldlm_lock_addref);
724
725 /**
726  * Helper function.
727  * Add specified reader/writer reference to LDLM lock \a lock.
728  * r/w reference type is determined by \a mode
729  * Removes lock from LRU if it is there.
730  * Assumes the LDLM lock is already locked.
731  */
732 void ldlm_lock_addref_internal_nolock(struct ldlm_lock *lock,
733                                       enum ldlm_mode mode)
734 {
735         ldlm_lock_remove_from_lru(lock);
736         if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
737                 lock->l_readers++;
738                 lu_ref_add_atomic(&lock->l_reference, "reader", lock);
739         }
740         if (mode & (LCK_EX | LCK_CW | LCK_PW | LCK_GROUP | LCK_COS)) {
741                 lock->l_writers++;
742                 lu_ref_add_atomic(&lock->l_reference, "writer", lock);
743         }
744         LDLM_LOCK_GET(lock);
745         lu_ref_add_atomic(&lock->l_reference, "user", lock);
746         LDLM_DEBUG(lock, "ldlm_lock_addref(%s)", ldlm_lockname[mode]);
747 }
748
749 /**
750  * Attempts to add reader/writer reference to a lock with handle \a lockh, and
751  * fails if lock is already LDLM_FL_CBPENDING or destroyed.
752  *
753  * \retval 0 success, lock was addref-ed
754  *
755  * \retval -EAGAIN lock is being canceled.
756  */
757 int ldlm_lock_addref_try(const struct lustre_handle *lockh, enum ldlm_mode mode)
758 {
759         struct ldlm_lock *lock;
760         int               result;
761
762         result = -EAGAIN;
763         lock = ldlm_handle2lock(lockh);
764         if (lock != NULL) {
765                 lock_res_and_lock(lock);
766                 if (lock->l_readers != 0 || lock->l_writers != 0 ||
767                     !ldlm_is_cbpending(lock)) {
768                         ldlm_lock_addref_internal_nolock(lock, mode);
769                         result = 0;
770                 }
771                 unlock_res_and_lock(lock);
772                 LDLM_LOCK_PUT(lock);
773         }
774         return result;
775 }
776 EXPORT_SYMBOL(ldlm_lock_addref_try);
777
778 /**
779  * Add specified reader/writer reference to LDLM lock \a lock.
780  * Locks LDLM lock and calls ldlm_lock_addref_internal_nolock to do the work.
781  * Only called for local locks.
782  */
783 void ldlm_lock_addref_internal(struct ldlm_lock *lock, enum ldlm_mode mode)
784 {
785         lock_res_and_lock(lock);
786         ldlm_lock_addref_internal_nolock(lock, mode);
787         unlock_res_and_lock(lock);
788 }
789
790 /**
791  * Removes reader/writer reference for LDLM lock \a lock.
792  * Assumes LDLM lock is already locked.
793  * only called in ldlm_flock_destroy and for local locks.
794  * Does NOT add lock to LRU if no r/w references left to accomodate flock locks
795  * that cannot be placed in LRU.
796  */
797 void ldlm_lock_decref_internal_nolock(struct ldlm_lock *lock,
798                                       enum ldlm_mode mode)
799 {
800         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
801         if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
802                 LASSERT(lock->l_readers > 0);
803                 lu_ref_del(&lock->l_reference, "reader", lock);
804                 lock->l_readers--;
805         }
806         if (mode & (LCK_EX | LCK_CW | LCK_PW | LCK_GROUP | LCK_COS)) {
807                 LASSERT(lock->l_writers > 0);
808                 lu_ref_del(&lock->l_reference, "writer", lock);
809                 lock->l_writers--;
810         }
811
812         lu_ref_del(&lock->l_reference, "user", lock);
813         LDLM_LOCK_RELEASE(lock);    /* matches the LDLM_LOCK_GET() in addref */
814 }
815
816 /**
817  * Removes reader/writer reference for LDLM lock \a lock.
818  * Locks LDLM lock first.
819  * If the lock is determined to be client lock on a client and r/w refcount
820  * drops to zero and the lock is not blocked, the lock is added to LRU lock
821  * on the namespace.
822  * For blocked LDLM locks if r/w count drops to zero, blocking_ast is called.
823  */
824 void ldlm_lock_decref_internal(struct ldlm_lock *lock, enum ldlm_mode mode)
825 {
826         struct ldlm_namespace *ns;
827         ENTRY;
828
829         lock_res_and_lock(lock);
830
831         ns = ldlm_lock_to_ns(lock);
832
833         ldlm_lock_decref_internal_nolock(lock, mode);
834
835         if ((ldlm_is_local(lock) || lock->l_req_mode == LCK_GROUP) &&
836             !lock->l_readers && !lock->l_writers) {
837                 /* If this is a local lock on a server namespace and this was
838                  * the last reference, cancel the lock.
839                  *
840                  * Group locks are special:
841                  * They must not go in LRU, but they are not called back
842                  * like non-group locks, instead they are manually released.
843                  * They have an l_writers reference which they keep until
844                  * they are manually released, so we remove them when they have
845                  * no more reader or writer references. - LU-6368 */
846                 ldlm_set_cbpending(lock);
847         }
848
849         if (!lock->l_readers && !lock->l_writers && ldlm_is_cbpending(lock)) {
850                 /* If we received a blocked AST and this was the last reference,
851                  * run the callback. */
852                 if (ldlm_is_ns_srv(lock) && lock->l_export)
853                         CERROR("FL_CBPENDING set on non-local lock--just a "
854                                "warning\n");
855
856                 LDLM_DEBUG(lock, "final decref done on cbpending lock");
857
858                 LDLM_LOCK_GET(lock); /* dropped by bl thread */
859                 ldlm_lock_remove_from_lru(lock);
860                 unlock_res_and_lock(lock);
861
862                 if (ldlm_is_fail_loc(lock))
863                         OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
864
865                 if (ldlm_is_atomic_cb(lock) ||
866                     ldlm_bl_to_thread_lock(ns, NULL, lock) != 0)
867                         ldlm_handle_bl_callback(ns, NULL, lock);
868         } else if (ns_is_client(ns) &&
869                    !lock->l_readers && !lock->l_writers &&
870                    !ldlm_is_no_lru(lock) &&
871                    !ldlm_is_bl_ast(lock)) {
872
873                 LDLM_DEBUG(lock, "add lock into lru list");
874
875                 /* If this is a client-side namespace and this was the last
876                  * reference, put it on the LRU. */
877                 ldlm_lock_add_to_lru(lock);
878                 unlock_res_and_lock(lock);
879
880                 if (ldlm_is_fail_loc(lock))
881                         OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
882
883                 /* Call ldlm_cancel_lru() only if EARLY_CANCEL and LRU RESIZE
884                  * are not supported by the server, otherwise, it is done on
885                  * enqueue. */
886                 if (!exp_connect_cancelset(lock->l_conn_export) &&
887                     !ns_connect_lru_resize(ns))
888                         ldlm_cancel_lru(ns, 0, LCF_ASYNC, 0);
889         } else {
890                 LDLM_DEBUG(lock, "do not add lock into lru list");
891                 unlock_res_and_lock(lock);
892         }
893
894         EXIT;
895 }
896
897 /**
898  * Decrease reader/writer refcount for LDLM lock with handle \a lockh
899  */
900 void ldlm_lock_decref(const struct lustre_handle *lockh, enum ldlm_mode mode)
901 {
902         struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0);
903         LASSERTF(lock != NULL, "Non-existing lock: %#llx\n", lockh->cookie);
904         ldlm_lock_decref_internal(lock, mode);
905         LDLM_LOCK_PUT(lock);
906 }
907 EXPORT_SYMBOL(ldlm_lock_decref);
908
909 /**
910  * Decrease reader/writer refcount for LDLM lock with handle
911  * \a lockh and mark it for subsequent cancellation once r/w refcount
912  * drops to zero instead of putting into LRU.
913  *
914  */
915 void ldlm_lock_decref_and_cancel(const struct lustre_handle *lockh,
916                                  enum ldlm_mode mode)
917 {
918         struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0);
919         ENTRY;
920
921         LASSERT(lock != NULL);
922
923         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
924         lock_res_and_lock(lock);
925         ldlm_set_cbpending(lock);
926         unlock_res_and_lock(lock);
927         ldlm_lock_decref_internal(lock, mode);
928         LDLM_LOCK_PUT(lock);
929 }
930 EXPORT_SYMBOL(ldlm_lock_decref_and_cancel);
931
932 struct sl_insert_point {
933         struct list_head *res_link;
934         struct list_head *mode_link;
935         struct list_head *policy_link;
936 };
937
938 /**
939  * Finds a position to insert the new lock into granted lock list.
940  *
941  * Used for locks eligible for skiplist optimization.
942  *
943  * Parameters:
944  *      queue [input]:  the granted list where search acts on;
945  *      req [input]:    the lock whose position to be located;
946  *      prev [output]:  positions within 3 lists to insert @req to
947  * Return Value:
948  *      filled @prev
949  * NOTE: called by
950  *  - ldlm_grant_lock_with_skiplist
951  */
952 static void search_granted_lock(struct list_head *queue,
953                                 struct ldlm_lock *req,
954                                 struct sl_insert_point *prev)
955 {
956         struct list_head *tmp;
957         struct ldlm_lock *lock, *mode_end, *policy_end;
958         ENTRY;
959
960         list_for_each(tmp, queue) {
961                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
962
963                 mode_end = list_entry(lock->l_sl_mode.prev,
964                                           struct ldlm_lock, l_sl_mode);
965
966                 if (lock->l_req_mode != req->l_req_mode) {
967                         /* jump to last lock of mode group */
968                         tmp = &mode_end->l_res_link;
969                         continue;
970                 }
971
972                 /* suitable mode group is found */
973                 if (lock->l_resource->lr_type == LDLM_PLAIN) {
974                         /* insert point is last lock of the mode group */
975                         prev->res_link = &mode_end->l_res_link;
976                         prev->mode_link = &mode_end->l_sl_mode;
977                         prev->policy_link = &req->l_sl_policy;
978                         EXIT;
979                         return;
980                 } else if (lock->l_resource->lr_type == LDLM_IBITS) {
981                         for (;;) {
982                                 policy_end =
983                                         list_entry(lock->l_sl_policy.prev,
984                                                        struct ldlm_lock,
985                                                        l_sl_policy);
986
987                                 if (lock->l_policy_data.l_inodebits.bits ==
988                                     req->l_policy_data.l_inodebits.bits) {
989                                         /* insert point is last lock of
990                                          * the policy group */
991                                         prev->res_link =
992                                                 &policy_end->l_res_link;
993                                         prev->mode_link =
994                                                 &policy_end->l_sl_mode;
995                                         prev->policy_link =
996                                                 &policy_end->l_sl_policy;
997                                         EXIT;
998                                         return;
999                                 }
1000
1001                                 if (policy_end == mode_end)
1002                                         /* done with mode group */
1003                                         break;
1004
1005                                 /* go to next policy group within mode group */
1006                                 tmp = policy_end->l_res_link.next;
1007                                 lock = list_entry(tmp, struct ldlm_lock,
1008                                                       l_res_link);
1009                         }  /* loop over policy groups within the mode group */
1010
1011                         /* insert point is last lock of the mode group,
1012                          * new policy group is started */
1013                         prev->res_link = &mode_end->l_res_link;
1014                         prev->mode_link = &mode_end->l_sl_mode;
1015                         prev->policy_link = &req->l_sl_policy;
1016                         EXIT;
1017                         return;
1018                 } else {
1019                         LDLM_ERROR(lock,"is not LDLM_PLAIN or LDLM_IBITS lock");
1020                         LBUG();
1021                 }
1022         }
1023
1024         /* insert point is last lock on the queue,
1025          * new mode group and new policy group are started */
1026         prev->res_link = queue->prev;
1027         prev->mode_link = &req->l_sl_mode;
1028         prev->policy_link = &req->l_sl_policy;
1029         EXIT;
1030         return;
1031 }
1032
1033 /**
1034  * Add a lock into resource granted list after a position described by
1035  * \a prev.
1036  */
1037 static void ldlm_granted_list_add_lock(struct ldlm_lock *lock,
1038                                        struct sl_insert_point *prev)
1039 {
1040         struct ldlm_resource *res = lock->l_resource;
1041         ENTRY;
1042
1043         check_res_locked(res);
1044
1045         ldlm_resource_dump(D_INFO, res);
1046         LDLM_DEBUG(lock, "About to add lock:");
1047
1048         if (ldlm_is_destroyed(lock)) {
1049                 CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
1050                 return;
1051         }
1052
1053         LASSERT(list_empty(&lock->l_res_link));
1054         LASSERT(list_empty(&lock->l_sl_mode));
1055         LASSERT(list_empty(&lock->l_sl_policy));
1056
1057         /*
1058          * lock->link == prev->link means lock is first starting the group.
1059          * Don't re-add to itself to suppress kernel warnings.
1060          */
1061         if (&lock->l_res_link != prev->res_link)
1062                 list_add(&lock->l_res_link, prev->res_link);
1063         if (&lock->l_sl_mode != prev->mode_link)
1064                 list_add(&lock->l_sl_mode, prev->mode_link);
1065         if (&lock->l_sl_policy != prev->policy_link)
1066                 list_add(&lock->l_sl_policy, prev->policy_link);
1067
1068         EXIT;
1069 }
1070
1071 /**
1072  * Add a lock to granted list on a resource maintaining skiplist
1073  * correctness.
1074  */
1075 void ldlm_grant_lock_with_skiplist(struct ldlm_lock *lock)
1076 {
1077         struct sl_insert_point prev;
1078
1079         LASSERT(lock->l_req_mode == lock->l_granted_mode);
1080
1081         search_granted_lock(&lock->l_resource->lr_granted, lock, &prev);
1082         ldlm_granted_list_add_lock(lock, &prev);
1083 }
1084
1085 /**
1086  * Perform lock granting bookkeeping.
1087  *
1088  * Includes putting the lock into granted list and updating lock mode.
1089  * NOTE: called by
1090  *  - ldlm_lock_enqueue
1091  *  - ldlm_reprocess_queue
1092  *
1093  * must be called with lr_lock held
1094  */
1095 void ldlm_grant_lock(struct ldlm_lock *lock, struct list_head *work_list)
1096 {
1097         struct ldlm_resource *res = lock->l_resource;
1098         ENTRY;
1099
1100         check_res_locked(res);
1101
1102         lock->l_granted_mode = lock->l_req_mode;
1103
1104         if (work_list && lock->l_completion_ast != NULL)
1105                 ldlm_add_ast_work_item(lock, NULL, work_list);
1106
1107         if (res->lr_type == LDLM_PLAIN || res->lr_type == LDLM_IBITS)
1108                 ldlm_grant_lock_with_skiplist(lock);
1109         else if (res->lr_type == LDLM_EXTENT)
1110                 ldlm_extent_add_lock(res, lock);
1111         else if (res->lr_type == LDLM_FLOCK) {
1112                 /* We should not add locks to granted list in the following
1113                  * cases:
1114                  * - this is an UNLOCK but not a real lock;
1115                  * - this is a TEST lock;
1116                  * - this is a F_CANCELLK lock (async flock has req_mode == 0)
1117                  * - this is a deadlock (flock cannot be granted) */
1118                 if (lock->l_req_mode == 0 ||
1119                     lock->l_req_mode == LCK_NL ||
1120                     ldlm_is_test_lock(lock) ||
1121                     ldlm_is_flock_deadlock(lock))
1122                         RETURN_EXIT;
1123                 ldlm_resource_add_lock(res, &res->lr_granted, lock);
1124         } else {
1125                 LBUG();
1126         }
1127
1128         ldlm_pool_add(&ldlm_res_to_ns(res)->ns_pool, lock);
1129         EXIT;
1130 }
1131
1132 /**
1133  * Describe the overlap between two locks.  itree_overlap_cb data.
1134  */
1135 struct lock_match_data {
1136         struct ldlm_lock        *lmd_old;
1137         struct ldlm_lock        *lmd_lock;
1138         enum ldlm_mode          *lmd_mode;
1139         union ldlm_policy_data  *lmd_policy;
1140         __u64                    lmd_flags;
1141         int                      lmd_unref;
1142 };
1143
1144 /**
1145  * Check if the given @lock meets the criteria for a match.
1146  * A reference on the lock is taken if matched.
1147  *
1148  * \param lock     test-against this lock
1149  * \param data     parameters
1150  */
1151 static int lock_matches(struct ldlm_lock *lock, struct lock_match_data *data)
1152 {
1153         union ldlm_policy_data *lpol = &lock->l_policy_data;
1154         enum ldlm_mode match;
1155
1156         if (lock == data->lmd_old)
1157                 return INTERVAL_ITER_STOP;
1158
1159         /* Check if this lock can be matched.
1160          * Used by LU-2919(exclusive open) for open lease lock */
1161         if (ldlm_is_excl(lock))
1162                 return INTERVAL_ITER_CONT;
1163
1164         /* llite sometimes wants to match locks that will be
1165          * canceled when their users drop, but we allow it to match
1166          * if it passes in CBPENDING and the lock still has users.
1167          * this is generally only going to be used by children
1168          * whose parents already hold a lock so forward progress
1169          * can still happen. */
1170         if (ldlm_is_cbpending(lock) &&
1171             !(data->lmd_flags & LDLM_FL_CBPENDING))
1172                 return INTERVAL_ITER_CONT;
1173         if (!data->lmd_unref && ldlm_is_cbpending(lock) &&
1174             lock->l_readers == 0 && lock->l_writers == 0)
1175                 return INTERVAL_ITER_CONT;
1176
1177         if (!(lock->l_req_mode & *data->lmd_mode))
1178                 return INTERVAL_ITER_CONT;
1179         match = lock->l_req_mode;
1180
1181         switch (lock->l_resource->lr_type) {
1182         case LDLM_EXTENT:
1183                 if (lpol->l_extent.start > data->lmd_policy->l_extent.start ||
1184                     lpol->l_extent.end < data->lmd_policy->l_extent.end)
1185                         return INTERVAL_ITER_CONT;
1186
1187                 if (unlikely(match == LCK_GROUP) &&
1188                     data->lmd_policy->l_extent.gid != LDLM_GID_ANY &&
1189                     lpol->l_extent.gid != data->lmd_policy->l_extent.gid)
1190                         return INTERVAL_ITER_CONT;
1191                 break;
1192         case LDLM_IBITS:
1193                 /* We match if we have existing lock with same or wider set
1194                    of bits. */
1195                 if ((lpol->l_inodebits.bits &
1196                      data->lmd_policy->l_inodebits.bits) !=
1197                     data->lmd_policy->l_inodebits.bits)
1198                         return INTERVAL_ITER_CONT;
1199                 break;
1200         default:
1201                 ;
1202         }
1203
1204         /* We match if we have existing lock with same or wider set
1205            of bits. */
1206         if (!data->lmd_unref && LDLM_HAVE_MASK(lock, GONE))
1207                 return INTERVAL_ITER_CONT;
1208
1209         if (!equi(data->lmd_flags & LDLM_FL_LOCAL_ONLY, ldlm_is_local(lock)))
1210                 return INTERVAL_ITER_CONT;
1211
1212         if (data->lmd_flags & LDLM_FL_TEST_LOCK) {
1213                 LDLM_LOCK_GET(lock);
1214                 ldlm_lock_touch_in_lru(lock);
1215         } else {
1216                 ldlm_lock_addref_internal_nolock(lock, match);
1217         }
1218
1219         *data->lmd_mode = match;
1220         data->lmd_lock = lock;
1221
1222         return INTERVAL_ITER_STOP;
1223 }
1224
1225 static unsigned int itree_overlap_cb(struct interval_node *in, void *args)
1226 {
1227         struct ldlm_interval *node = to_ldlm_interval(in);
1228         struct lock_match_data *data = args;
1229         struct ldlm_lock *lock;
1230         int rc;
1231
1232         list_for_each_entry(lock, &node->li_group, l_sl_policy) {
1233                 rc = lock_matches(lock, data);
1234                 if (rc == INTERVAL_ITER_STOP)
1235                         return INTERVAL_ITER_STOP;
1236         }
1237         return INTERVAL_ITER_CONT;
1238 }
1239
1240 /**
1241  * Search for a lock with given parameters in interval trees.
1242  *
1243  * \param res      search for a lock in this resource
1244  * \param data     parameters
1245  *
1246  * \retval a referenced lock or NULL.
1247  */
1248 static struct ldlm_lock *search_itree(struct ldlm_resource *res,
1249                                       struct lock_match_data *data)
1250 {
1251         struct interval_node_extent ext = {
1252                 .start     = data->lmd_policy->l_extent.start,
1253                 .end       = data->lmd_policy->l_extent.end
1254         };
1255         int idx;
1256
1257         for (idx = 0; idx < LCK_MODE_NUM; idx++) {
1258                 struct ldlm_interval_tree *tree = &res->lr_itree[idx];
1259
1260                 if (tree->lit_root == NULL)
1261                         continue;
1262
1263                 if (!(tree->lit_mode & *data->lmd_mode))
1264                         continue;
1265
1266                 interval_search(tree->lit_root, &ext,
1267                                 itree_overlap_cb, data);
1268         }
1269         return data->lmd_lock;
1270 }
1271
1272
1273 /**
1274  * Search for a lock with given properties in a queue.
1275  *
1276  * \param queue    search for a lock in this queue
1277  * \param data     parameters
1278  *
1279  * \retval a referenced lock or NULL.
1280  */
1281 static struct ldlm_lock *search_queue(struct list_head *queue,
1282                                       struct lock_match_data *data)
1283 {
1284         struct ldlm_lock *lock;
1285         int rc;
1286
1287         list_for_each_entry(lock, queue, l_res_link) {
1288                 rc = lock_matches(lock, data);
1289                 if (rc == INTERVAL_ITER_STOP)
1290                         return data->lmd_lock;
1291         }
1292         return NULL;
1293 }
1294
1295 void ldlm_lock_fail_match_locked(struct ldlm_lock *lock)
1296 {
1297         if ((lock->l_flags & LDLM_FL_FAIL_NOTIFIED) == 0) {
1298                 lock->l_flags |= LDLM_FL_FAIL_NOTIFIED;
1299                 wake_up_all(&lock->l_waitq);
1300         }
1301 }
1302 EXPORT_SYMBOL(ldlm_lock_fail_match_locked);
1303
1304 void ldlm_lock_fail_match(struct ldlm_lock *lock)
1305 {
1306         lock_res_and_lock(lock);
1307         ldlm_lock_fail_match_locked(lock);
1308         unlock_res_and_lock(lock);
1309 }
1310
1311 /**
1312  * Mark lock as "matchable" by OST.
1313  *
1314  * Used to prevent certain races in LOV/OSC where the lock is granted, but LVB
1315  * is not yet valid.
1316  * Assumes LDLM lock is already locked.
1317  */
1318 void ldlm_lock_allow_match_locked(struct ldlm_lock *lock)
1319 {
1320         ldlm_set_lvb_ready(lock);
1321         wake_up_all(&lock->l_waitq);
1322 }
1323 EXPORT_SYMBOL(ldlm_lock_allow_match_locked);
1324
1325 /**
1326  * Mark lock as "matchable" by OST.
1327  * Locks the lock and then \see ldlm_lock_allow_match_locked
1328  */
1329 void ldlm_lock_allow_match(struct ldlm_lock *lock)
1330 {
1331         lock_res_and_lock(lock);
1332         ldlm_lock_allow_match_locked(lock);
1333         unlock_res_and_lock(lock);
1334 }
1335 EXPORT_SYMBOL(ldlm_lock_allow_match);
1336
1337 /**
1338  * Attempt to find a lock with specified properties.
1339  *
1340  * Typically returns a reference to matched lock unless LDLM_FL_TEST_LOCK is
1341  * set in \a flags
1342  *
1343  * Can be called in two ways:
1344  *
1345  * If 'ns' is NULL, then lockh describes an existing lock that we want to look
1346  * for a duplicate of.
1347  *
1348  * Otherwise, all of the fields must be filled in, to match against.
1349  *
1350  * If 'flags' contains LDLM_FL_LOCAL_ONLY, then only match local locks on the
1351  *     server (ie, connh is NULL)
1352  * If 'flags' contains LDLM_FL_BLOCK_GRANTED, then only locks on the granted
1353  *     list will be considered
1354  * If 'flags' contains LDLM_FL_CBPENDING, then locks that have been marked
1355  *     to be canceled can still be matched as long as they still have reader
1356  *     or writer refernces
1357  * If 'flags' contains LDLM_FL_TEST_LOCK, then don't actually reference a lock,
1358  *     just tell us if we would have matched.
1359  *
1360  * \retval 1 if it finds an already-existing lock that is compatible; in this
1361  * case, lockh is filled in with a addref()ed lock
1362  *
1363  * We also check security context, and if that fails we simply return 0 (to
1364  * keep caller code unchanged), the context failure will be discovered by
1365  * caller sometime later.
1366  */
1367 enum ldlm_mode ldlm_lock_match(struct ldlm_namespace *ns, __u64 flags,
1368                                const struct ldlm_res_id *res_id,
1369                                enum ldlm_type type,
1370                                union ldlm_policy_data *policy,
1371                                enum ldlm_mode mode,
1372                                struct lustre_handle *lockh, int unref)
1373 {
1374         struct lock_match_data data = {
1375                 .lmd_old        = NULL,
1376                 .lmd_lock       = NULL,
1377                 .lmd_mode       = &mode,
1378                 .lmd_policy     = policy,
1379                 .lmd_flags      = flags,
1380                 .lmd_unref      = unref,
1381         };
1382         struct ldlm_resource *res;
1383         struct ldlm_lock *lock;
1384         int rc = 0;
1385         ENTRY;
1386
1387         if (ns == NULL) {
1388                 data.lmd_old = ldlm_handle2lock(lockh);
1389                 LASSERT(data.lmd_old != NULL);
1390
1391                 ns = ldlm_lock_to_ns(data.lmd_old);
1392                 res_id = &data.lmd_old->l_resource->lr_name;
1393                 type = data.lmd_old->l_resource->lr_type;
1394                 *data.lmd_mode = data.lmd_old->l_req_mode;
1395         }
1396
1397         res = ldlm_resource_get(ns, NULL, res_id, type, 0);
1398         if (IS_ERR(res)) {
1399                 LASSERT(data.lmd_old == NULL);
1400                 RETURN(0);
1401         }
1402
1403         LDLM_RESOURCE_ADDREF(res);
1404         lock_res(res);
1405
1406         if (res->lr_type == LDLM_EXTENT)
1407                 lock = search_itree(res, &data);
1408         else
1409                 lock = search_queue(&res->lr_granted, &data);
1410         if (lock != NULL)
1411                 GOTO(out, rc = 1);
1412         if (flags & LDLM_FL_BLOCK_GRANTED)
1413                 GOTO(out, rc = 0);
1414         lock = search_queue(&res->lr_waiting, &data);
1415         if (lock != NULL)
1416                 GOTO(out, rc = 1);
1417
1418         EXIT;
1419  out:
1420         unlock_res(res);
1421         LDLM_RESOURCE_DELREF(res);
1422         ldlm_resource_putref(res);
1423
1424         if (lock) {
1425                 ldlm_lock2handle(lock, lockh);
1426                 if ((flags & LDLM_FL_LVB_READY) &&
1427                     (!ldlm_is_lvb_ready(lock))) {
1428                         __u64 wait_flags = LDLM_FL_LVB_READY |
1429                                 LDLM_FL_DESTROYED | LDLM_FL_FAIL_NOTIFIED;
1430                         struct l_wait_info lwi;
1431                         if (lock->l_completion_ast) {
1432                                 int err = lock->l_completion_ast(lock,
1433                                                           LDLM_FL_WAIT_NOREPROC,
1434                                                                  NULL);
1435                                 if (err) {
1436                                         if (flags & LDLM_FL_TEST_LOCK)
1437                                                 LDLM_LOCK_RELEASE(lock);
1438                                         else
1439                                                 ldlm_lock_decref_internal(lock,
1440                                                                           mode);
1441                                         rc = 0;
1442                                         goto out2;
1443                                 }
1444                         }
1445
1446                         lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(obd_timeout),
1447                                                NULL, LWI_ON_SIGNAL_NOOP, NULL);
1448
1449                         /* XXX FIXME see comment on CAN_MATCH in lustre_dlm.h */
1450                         l_wait_event(lock->l_waitq,
1451                                      lock->l_flags & wait_flags,
1452                                      &lwi);
1453                         if (!ldlm_is_lvb_ready(lock)) {
1454                                 if (flags & LDLM_FL_TEST_LOCK)
1455                                         LDLM_LOCK_RELEASE(lock);
1456                                 else
1457                                         ldlm_lock_decref_internal(lock, mode);
1458                                 rc = 0;
1459                         }
1460                 }
1461         }
1462  out2:
1463         if (rc) {
1464                 LDLM_DEBUG(lock, "matched (%llu %llu)",
1465                            (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1466                                 res_id->name[2] : policy->l_extent.start,
1467                            (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1468                                 res_id->name[3] : policy->l_extent.end);
1469
1470                 /* check user's security context */
1471                 if (lock->l_conn_export &&
1472                     sptlrpc_import_check_ctx(
1473                                 class_exp2cliimp(lock->l_conn_export))) {
1474                         if (!(flags & LDLM_FL_TEST_LOCK))
1475                                 ldlm_lock_decref_internal(lock, mode);
1476                         rc = 0;
1477                 }
1478
1479                 if (flags & LDLM_FL_TEST_LOCK)
1480                         LDLM_LOCK_RELEASE(lock);
1481
1482         } else if (!(flags & LDLM_FL_TEST_LOCK)) {/*less verbose for test-only*/
1483                 LDLM_DEBUG_NOLOCK("not matched ns %p type %u mode %u res "
1484                                   "%llu/%llu (%llu %llu)", ns,
1485                                   type, mode, res_id->name[0], res_id->name[1],
1486                                   (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1487                                         res_id->name[2] :policy->l_extent.start,
1488                                   (type == LDLM_PLAIN || type == LDLM_IBITS) ?
1489                                         res_id->name[3] : policy->l_extent.end);
1490         }
1491         if (data.lmd_old != NULL)
1492                 LDLM_LOCK_PUT(data.lmd_old);
1493
1494         return rc ? mode : 0;
1495 }
1496 EXPORT_SYMBOL(ldlm_lock_match);
1497
1498 enum ldlm_mode ldlm_revalidate_lock_handle(const struct lustre_handle *lockh,
1499                                            __u64 *bits)
1500 {
1501         struct ldlm_lock *lock;
1502         enum ldlm_mode mode = 0;
1503         ENTRY;
1504
1505         lock = ldlm_handle2lock(lockh);
1506         if (lock != NULL) {
1507                 lock_res_and_lock(lock);
1508                 if (LDLM_HAVE_MASK(lock, GONE))
1509                         GOTO(out, mode);
1510
1511                 if (ldlm_is_cbpending(lock) &&
1512                     lock->l_readers == 0 && lock->l_writers == 0)
1513                         GOTO(out, mode);
1514
1515                 if (bits)
1516                         *bits = lock->l_policy_data.l_inodebits.bits;
1517                 mode = lock->l_granted_mode;
1518                 ldlm_lock_addref_internal_nolock(lock, mode);
1519         }
1520
1521         EXIT;
1522
1523 out:
1524         if (lock != NULL) {
1525                 unlock_res_and_lock(lock);
1526                 LDLM_LOCK_PUT(lock);
1527         }
1528         return mode;
1529 }
1530 EXPORT_SYMBOL(ldlm_revalidate_lock_handle);
1531
1532 /** The caller must guarantee that the buffer is large enough. */
1533 int ldlm_fill_lvb(struct ldlm_lock *lock, struct req_capsule *pill,
1534                   enum req_location loc, void *data, int size)
1535 {
1536         void *lvb;
1537         ENTRY;
1538
1539         LASSERT(data != NULL);
1540         LASSERT(size >= 0);
1541
1542         switch (lock->l_lvb_type) {
1543         case LVB_T_OST:
1544                 if (size == sizeof(struct ost_lvb)) {
1545                         if (loc == RCL_CLIENT)
1546                                 lvb = req_capsule_client_swab_get(pill,
1547                                                 &RMF_DLM_LVB,
1548                                                 lustre_swab_ost_lvb);
1549                         else
1550                                 lvb = req_capsule_server_swab_get(pill,
1551                                                 &RMF_DLM_LVB,
1552                                                 lustre_swab_ost_lvb);
1553                         if (unlikely(lvb == NULL)) {
1554                                 LDLM_ERROR(lock, "no LVB");
1555                                 RETURN(-EPROTO);
1556                         }
1557
1558                         memcpy(data, lvb, size);
1559                 } else if (size == sizeof(struct ost_lvb_v1)) {
1560                         struct ost_lvb *olvb = data;
1561
1562                         if (loc == RCL_CLIENT)
1563                                 lvb = req_capsule_client_swab_get(pill,
1564                                                 &RMF_DLM_LVB,
1565                                                 lustre_swab_ost_lvb_v1);
1566                         else
1567                                 lvb = req_capsule_server_sized_swab_get(pill,
1568                                                 &RMF_DLM_LVB, size,
1569                                                 lustre_swab_ost_lvb_v1);
1570                         if (unlikely(lvb == NULL)) {
1571                                 LDLM_ERROR(lock, "no LVB");
1572                                 RETURN(-EPROTO);
1573                         }
1574
1575                         memcpy(data, lvb, size);
1576                         olvb->lvb_mtime_ns = 0;
1577                         olvb->lvb_atime_ns = 0;
1578                         olvb->lvb_ctime_ns = 0;
1579                 } else {
1580                         LDLM_ERROR(lock, "Replied unexpected ost LVB size %d",
1581                                    size);
1582                         RETURN(-EINVAL);
1583                 }
1584                 break;
1585         case LVB_T_LQUOTA:
1586                 if (size == sizeof(struct lquota_lvb)) {
1587                         if (loc == RCL_CLIENT)
1588                                 lvb = req_capsule_client_swab_get(pill,
1589                                                 &RMF_DLM_LVB,
1590                                                 lustre_swab_lquota_lvb);
1591                         else
1592                                 lvb = req_capsule_server_swab_get(pill,
1593                                                 &RMF_DLM_LVB,
1594                                                 lustre_swab_lquota_lvb);
1595                         if (unlikely(lvb == NULL)) {
1596                                 LDLM_ERROR(lock, "no LVB");
1597                                 RETURN(-EPROTO);
1598                         }
1599
1600                         memcpy(data, lvb, size);
1601                 } else {
1602                         LDLM_ERROR(lock, "Replied unexpected lquota LVB size %d",
1603                                    size);
1604                         RETURN(-EINVAL);
1605                 }
1606                 break;
1607         case LVB_T_LAYOUT:
1608                 if (size == 0)
1609                         break;
1610
1611                 if (loc == RCL_CLIENT)
1612                         lvb = req_capsule_client_get(pill, &RMF_DLM_LVB);
1613                 else
1614                         lvb = req_capsule_server_get(pill, &RMF_DLM_LVB);
1615                 if (unlikely(lvb == NULL)) {
1616                         LDLM_ERROR(lock, "no LVB");
1617                         RETURN(-EPROTO);
1618                 }
1619
1620                 memcpy(data, lvb, size);
1621                 break;
1622         default:
1623                 LDLM_ERROR(lock, "Unknown LVB type: %d", lock->l_lvb_type);
1624                 libcfs_debug_dumpstack(NULL);
1625                 RETURN(-EINVAL);
1626         }
1627
1628         RETURN(0);
1629 }
1630
1631 /**
1632  * Create and fill in new LDLM lock with specified properties.
1633  * Returns a referenced lock
1634  */
1635 struct ldlm_lock *ldlm_lock_create(struct ldlm_namespace *ns,
1636                                    const struct ldlm_res_id *res_id,
1637                                    enum ldlm_type type,
1638                                    enum ldlm_mode mode,
1639                                    const struct ldlm_callback_suite *cbs,
1640                                    void *data, __u32 lvb_len,
1641                                    enum lvb_type lvb_type)
1642 {
1643         struct ldlm_lock        *lock;
1644         struct ldlm_resource    *res;
1645         int                     rc;
1646         ENTRY;
1647
1648         res = ldlm_resource_get(ns, NULL, res_id, type, 1);
1649         if (IS_ERR(res))
1650                 RETURN(ERR_CAST(res));
1651
1652         lock = ldlm_lock_new(res);
1653         if (lock == NULL)
1654                 RETURN(ERR_PTR(-ENOMEM));
1655
1656         lock->l_req_mode = mode;
1657         lock->l_ast_data = data;
1658         lock->l_pid = current_pid();
1659         if (ns_is_server(ns))
1660                 ldlm_set_ns_srv(lock);
1661         if (cbs) {
1662                 lock->l_blocking_ast = cbs->lcs_blocking;
1663                 lock->l_completion_ast = cbs->lcs_completion;
1664                 lock->l_glimpse_ast = cbs->lcs_glimpse;
1665         }
1666
1667         lock->l_tree_node = NULL;
1668         /* if this is the extent lock, allocate the interval tree node */
1669         if (type == LDLM_EXTENT)
1670                 if (ldlm_interval_alloc(lock) == NULL)
1671                         GOTO(out, rc = -ENOMEM);
1672
1673         if (lvb_len) {
1674                 lock->l_lvb_len = lvb_len;
1675                 OBD_ALLOC_LARGE(lock->l_lvb_data, lvb_len);
1676                 if (lock->l_lvb_data == NULL)
1677                         GOTO(out, rc = -ENOMEM);
1678         }
1679
1680         lock->l_lvb_type = lvb_type;
1681         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_NEW_LOCK))
1682                 GOTO(out, rc = -ENOENT);
1683
1684         RETURN(lock);
1685
1686 out:
1687         ldlm_lock_destroy(lock);
1688         LDLM_LOCK_RELEASE(lock);
1689         RETURN(ERR_PTR(rc));
1690 }
1691
1692 #ifdef HAVE_SERVER_SUPPORT
1693 static enum ldlm_error ldlm_lock_enqueue_helper(struct ldlm_lock *lock,
1694                                              __u64 *flags)
1695 {
1696         struct ldlm_resource *res = lock->l_resource;
1697         enum ldlm_error rc = ELDLM_OK;
1698         struct list_head rpc_list = LIST_HEAD_INIT(rpc_list);
1699         ldlm_processing_policy policy;
1700         ENTRY;
1701
1702         policy = ldlm_processing_policy_table[res->lr_type];
1703 restart:
1704         policy(lock, flags, LDLM_PROCESS_ENQUEUE, &rc, &rpc_list);
1705         if (rc == ELDLM_OK && lock->l_granted_mode != lock->l_req_mode &&
1706             res->lr_type != LDLM_FLOCK) {
1707                 rc = ldlm_handle_conflict_lock(lock, flags, &rpc_list);
1708                 if (rc == -ERESTART)
1709                         GOTO(restart, rc);
1710         }
1711
1712         if (!list_empty(&rpc_list))
1713                 ldlm_discard_bl_list(&rpc_list);
1714
1715         RETURN(rc);
1716 }
1717 #endif
1718
1719 /**
1720  * Enqueue (request) a lock.
1721  *
1722  * Does not block. As a result of enqueue the lock would be put
1723  * into granted or waiting list.
1724  *
1725  * If namespace has intent policy sent and the lock has LDLM_FL_HAS_INTENT flag
1726  * set, skip all the enqueueing and delegate lock processing to intent policy
1727  * function.
1728  */
1729 enum ldlm_error ldlm_lock_enqueue(struct ldlm_namespace *ns,
1730                                   struct ldlm_lock **lockp,
1731                                   void *cookie, __u64 *flags)
1732 {
1733         struct ldlm_lock *lock = *lockp;
1734         struct ldlm_resource *res = lock->l_resource;
1735         int local = ns_is_client(ldlm_res_to_ns(res));
1736         enum ldlm_error rc = ELDLM_OK;
1737         struct ldlm_interval *node = NULL;
1738         ENTRY;
1739
1740         /* policies are not executed on the client or during replay */
1741         if ((*flags & (LDLM_FL_HAS_INTENT|LDLM_FL_REPLAY)) == LDLM_FL_HAS_INTENT
1742             && !local && ns->ns_policy) {
1743                 rc = ns->ns_policy(ns, lockp, cookie, lock->l_req_mode, *flags,
1744                                    NULL);
1745                 if (rc == ELDLM_LOCK_REPLACED) {
1746                         /* The lock that was returned has already been granted,
1747                          * and placed into lockp.  If it's not the same as the
1748                          * one we passed in, then destroy the old one and our
1749                          * work here is done. */
1750                         if (lock != *lockp) {
1751                                 ldlm_lock_destroy(lock);
1752                                 LDLM_LOCK_RELEASE(lock);
1753                         }
1754                         *flags |= LDLM_FL_LOCK_CHANGED;
1755                         RETURN(0);
1756                 } else if (rc != ELDLM_OK &&
1757                            lock->l_req_mode == lock->l_granted_mode) {
1758                         LASSERT(*flags & LDLM_FL_RESENT);
1759                         /* It may happen that ns_policy returns an error in
1760                          * resend case, object may be unlinked or just some
1761                          * error occurs. It is unclear if lock reached the
1762                          * client in the original reply, just leave the lock on
1763                          * server, not returning it again to client. Due to
1764                          * LU-6529, the server will not OOM. */
1765                         RETURN(rc);
1766                 } else if (rc != ELDLM_OK ||
1767                            (rc == ELDLM_OK && (*flags & LDLM_FL_INTENT_ONLY))) {
1768                         ldlm_lock_destroy(lock);
1769                         RETURN(rc);
1770                 }
1771         }
1772
1773         if (*flags & LDLM_FL_RESENT) {
1774                 /* Reconstruct LDLM_FL_SRV_ENQ_MASK @flags for reply.
1775                  * Set LOCK_CHANGED always.
1776                  * Check if the lock is granted for BLOCK_GRANTED.
1777                  * Take NO_TIMEOUT from the lock as it is inherited through
1778                  * LDLM_FL_INHERIT_MASK */
1779                 *flags |= LDLM_FL_LOCK_CHANGED;
1780                 if (lock->l_req_mode != lock->l_granted_mode)
1781                         *flags |= LDLM_FL_BLOCK_GRANTED;
1782                 *flags |= lock->l_flags & LDLM_FL_NO_TIMEOUT;
1783                 RETURN(ELDLM_OK);
1784         }
1785
1786         /* For a replaying lock, it might be already in granted list. So
1787          * unlinking the lock will cause the interval node to be freed, we
1788          * have to allocate the interval node early otherwise we can't regrant
1789          * this lock in the future. - jay */
1790         if (!local && (*flags & LDLM_FL_REPLAY) && res->lr_type == LDLM_EXTENT)
1791                 OBD_SLAB_ALLOC_PTR_GFP(node, ldlm_interval_slab, GFP_NOFS);
1792
1793         lock_res_and_lock(lock);
1794         if (local && lock->l_req_mode == lock->l_granted_mode) {
1795                 /* The server returned a blocked lock, but it was granted
1796                  * before we got a chance to actually enqueue it.  We don't
1797                  * need to do anything else. */
1798                 *flags &= ~LDLM_FL_BLOCKED_MASK;
1799                 GOTO(out, rc = ELDLM_OK);
1800         }
1801
1802         ldlm_resource_unlink_lock(lock);
1803         if (res->lr_type == LDLM_EXTENT && lock->l_tree_node == NULL) {
1804                 if (node == NULL) {
1805                         ldlm_lock_destroy_nolock(lock);
1806                         GOTO(out, rc = -ENOMEM);
1807                 }
1808
1809                 INIT_LIST_HEAD(&node->li_group);
1810                 ldlm_interval_attach(node, lock);
1811                 node = NULL;
1812         }
1813
1814         /* Some flags from the enqueue want to make it into the AST, via the
1815          * lock's l_flags. */
1816         if (*flags & LDLM_FL_AST_DISCARD_DATA)
1817                 ldlm_set_ast_discard_data(lock);
1818         if (*flags & LDLM_FL_TEST_LOCK)
1819                 ldlm_set_test_lock(lock);
1820         if (*flags & LDLM_FL_COS_INCOMPAT)
1821                 ldlm_set_cos_incompat(lock);
1822         if (*flags & LDLM_FL_COS_ENABLED)
1823                 ldlm_set_cos_enabled(lock);
1824
1825         /* This distinction between local lock trees is very important; a client
1826          * namespace only has information about locks taken by that client, and
1827          * thus doesn't have enough information to decide for itself if it can
1828          * be granted (below).  In this case, we do exactly what the server
1829          * tells us to do, as dictated by the 'flags'.
1830          *
1831          * We do exactly the same thing during recovery, when the server is
1832          * more or less trusting the clients not to lie.
1833          *
1834          * FIXME (bug 268): Detect obvious lies by checking compatibility in
1835          * granted queue. */
1836         if (local) {
1837                 if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED))
1838                         ldlm_resource_add_lock(res, &res->lr_waiting, lock);
1839                 else
1840                         ldlm_grant_lock(lock, NULL);
1841                 GOTO(out, rc = ELDLM_OK);
1842 #ifdef HAVE_SERVER_SUPPORT
1843         } else if (*flags & LDLM_FL_REPLAY) {
1844                 if (*flags & LDLM_FL_BLOCK_WAIT) {
1845                         ldlm_resource_add_lock(res, &res->lr_waiting, lock);
1846                         GOTO(out, rc = ELDLM_OK);
1847                 } else if (*flags & LDLM_FL_BLOCK_GRANTED) {
1848                         ldlm_grant_lock(lock, NULL);
1849                         GOTO(out, rc = ELDLM_OK);
1850                 }
1851                 /* If no flags, fall through to normal enqueue path. */
1852         }
1853
1854         rc = ldlm_lock_enqueue_helper(lock, flags);
1855         GOTO(out, rc);
1856 #else
1857         } else {
1858                 CERROR("This is client-side-only module, cannot handle "
1859                        "LDLM_NAMESPACE_SERVER resource type lock.\n");
1860                 LBUG();
1861         }
1862 #endif
1863
1864 out:
1865         unlock_res_and_lock(lock);
1866         if (node)
1867                 OBD_SLAB_FREE(node, ldlm_interval_slab, sizeof(*node));
1868         return rc;
1869 }
1870
1871 #ifdef HAVE_SERVER_SUPPORT
1872 /**
1873  * Iterate through all waiting locks on a given resource queue and attempt to
1874  * grant them.
1875  *
1876  * Must be called with resource lock held.
1877  */
1878 int ldlm_reprocess_queue(struct ldlm_resource *res, struct list_head *queue,
1879                          struct list_head *work_list,
1880                          enum ldlm_process_intention intention)
1881 {
1882         struct list_head *tmp, *pos;
1883         ldlm_processing_policy policy;
1884         __u64 flags;
1885         int rc = LDLM_ITER_CONTINUE;
1886         enum ldlm_error err;
1887         struct list_head bl_ast_list = LIST_HEAD_INIT(bl_ast_list);
1888         ENTRY;
1889
1890         check_res_locked(res);
1891
1892         policy = ldlm_processing_policy_table[res->lr_type];
1893         LASSERT(policy);
1894         LASSERT(intention == LDLM_PROCESS_RESCAN ||
1895                 intention == LDLM_PROCESS_RECOVERY);
1896
1897 restart:
1898         list_for_each_safe(tmp, pos, queue) {
1899                 struct ldlm_lock *pending;
1900                 struct list_head rpc_list = LIST_HEAD_INIT(rpc_list);
1901
1902                 pending = list_entry(tmp, struct ldlm_lock, l_res_link);
1903
1904                 CDEBUG(D_INFO, "Reprocessing lock %p\n", pending);
1905
1906                 flags = 0;
1907                 rc = policy(pending, &flags, intention, &err, &rpc_list);
1908                 if (pending->l_granted_mode == pending->l_req_mode ||
1909                     res->lr_type == LDLM_FLOCK) {
1910                         list_splice(&rpc_list, work_list);
1911                 } else {
1912                         list_splice(&rpc_list, &bl_ast_list);
1913                 }
1914                 /*
1915                  * When this is called from recovery done, we always want
1916                  * to scan the whole list no matter what 'rc' is returned.
1917                  */
1918                 if (rc != LDLM_ITER_CONTINUE &&
1919                     intention == LDLM_PROCESS_RESCAN)
1920                         break;
1921         }
1922
1923         if (!list_empty(&bl_ast_list)) {
1924                 unlock_res(res);
1925
1926                 LASSERT(intention == LDLM_PROCESS_RECOVERY);
1927
1928                 rc = ldlm_run_ast_work(ldlm_res_to_ns(res), &bl_ast_list,
1929                                        LDLM_WORK_BL_AST);
1930
1931                 lock_res(res);
1932                 if (rc == -ERESTART)
1933                         GOTO(restart, rc);
1934         }
1935
1936         if (!list_empty(&bl_ast_list))
1937                 ldlm_discard_bl_list(&bl_ast_list);
1938
1939         RETURN(intention == LDLM_PROCESS_RESCAN ? rc : LDLM_ITER_CONTINUE);
1940 }
1941
1942 /**
1943  * Conflicting locks are detected for a lock to be enqueued, add the lock
1944  * into waiting list and send blocking ASTs to the conflicting locks.
1945  *
1946  * \param[in] lock              The lock to be enqueued.
1947  * \param[out] flags            Lock flags for the lock to be enqueued.
1948  * \param[in] rpc_list          Conflicting locks list.
1949  *
1950  * \retval -ERESTART:   Some lock was instantly canceled while sending
1951  *                      blocking ASTs, caller needs to re-check conflicting
1952  *                      locks.
1953  * \retval -EAGAIN:     Lock was destroyed, caller should return error.
1954  * \reval 0:            Lock is successfully added in waiting list.
1955  */
1956 int ldlm_handle_conflict_lock(struct ldlm_lock *lock, __u64 *flags,
1957                               struct list_head *rpc_list)
1958 {
1959         struct ldlm_resource *res = lock->l_resource;
1960         int rc;
1961         ENTRY;
1962
1963         check_res_locked(res);
1964
1965         /* If either of the compat_queue()s returned failure, then we
1966          * have ASTs to send and must go onto the waiting list.
1967          *
1968          * bug 2322: we used to unlink and re-add here, which was a
1969          * terrible folly -- if we goto restart, we could get
1970          * re-ordered!  Causes deadlock, because ASTs aren't sent! */
1971         if (list_empty(&lock->l_res_link))
1972                 ldlm_resource_add_lock(res, &res->lr_waiting, lock);
1973         unlock_res(res);
1974
1975         rc = ldlm_run_ast_work(ldlm_res_to_ns(res), rpc_list,
1976                                LDLM_WORK_BL_AST);
1977
1978         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_OST_FAIL_RACE) &&
1979             !ns_is_client(ldlm_res_to_ns(res)))
1980                 class_fail_export(lock->l_export);
1981
1982         lock_res(res);
1983         if (rc == -ERESTART) {
1984                 /* 15715: The lock was granted and destroyed after
1985                  * resource lock was dropped. Interval node was freed
1986                  * in ldlm_lock_destroy. Anyway, this always happens
1987                  * when a client is being evicted. So it would be
1988                  * ok to return an error. -jay */
1989                 if (ldlm_is_destroyed(lock))
1990                         RETURN(-EAGAIN);
1991
1992                 /* lock was granted while resource was unlocked. */
1993                 if (lock->l_granted_mode == lock->l_req_mode) {
1994                         /* bug 11300: if the lock has been granted,
1995                          * break earlier because otherwise, we will go
1996                          * to restart and ldlm_resource_unlink will be
1997                          * called and it causes the interval node to be
1998                          * freed. Then we will fail at
1999                          * ldlm_extent_add_lock() */
2000                         *flags &= ~LDLM_FL_BLOCKED_MASK;
2001                         RETURN(0);
2002                 }
2003
2004                 RETURN(rc);
2005         }
2006         *flags |= LDLM_FL_BLOCK_GRANTED;
2007
2008         RETURN(0);
2009 }
2010
2011 /**
2012  * Discard all AST work items from list.
2013  *
2014  * If for whatever reason we do not want to send ASTs to conflicting locks
2015  * anymore, disassemble the list with this function.
2016  */
2017 void ldlm_discard_bl_list(struct list_head *bl_list)
2018 {
2019         struct list_head *tmp, *pos;
2020         ENTRY;
2021
2022         list_for_each_safe(pos, tmp, bl_list) {
2023                 struct ldlm_lock *lock =
2024                         list_entry(pos, struct ldlm_lock, l_bl_ast);
2025
2026                 list_del_init(&lock->l_bl_ast);
2027                 LASSERT(ldlm_is_ast_sent(lock));
2028                 ldlm_clear_ast_sent(lock);
2029                 LASSERT(lock->l_bl_ast_run == 0);
2030                 LASSERT(lock->l_blocking_lock);
2031                 LDLM_LOCK_RELEASE(lock->l_blocking_lock);
2032                 lock->l_blocking_lock = NULL;
2033                 LDLM_LOCK_RELEASE(lock);
2034         }
2035         EXIT;
2036 }
2037
2038 #endif
2039
2040 /**
2041  * Process a call to blocking AST callback for a lock in ast_work list
2042  */
2043 static int
2044 ldlm_work_bl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
2045 {
2046         struct ldlm_cb_set_arg *arg = opaq;
2047         struct ldlm_lock_desc   d;
2048         int                     rc;
2049         struct ldlm_lock       *lock;
2050         ENTRY;
2051
2052         if (list_empty(arg->list))
2053                 RETURN(-ENOENT);
2054
2055         lock = list_entry(arg->list->next, struct ldlm_lock, l_bl_ast);
2056
2057         /* nobody should touch l_bl_ast */
2058         lock_res_and_lock(lock);
2059         list_del_init(&lock->l_bl_ast);
2060
2061         LASSERT(ldlm_is_ast_sent(lock));
2062         LASSERT(lock->l_bl_ast_run == 0);
2063         LASSERT(lock->l_blocking_lock);
2064         lock->l_bl_ast_run++;
2065         unlock_res_and_lock(lock);
2066
2067         ldlm_lock2desc(lock->l_blocking_lock, &d);
2068
2069         rc = lock->l_blocking_ast(lock, &d, (void *)arg, LDLM_CB_BLOCKING);
2070         LDLM_LOCK_RELEASE(lock->l_blocking_lock);
2071         lock->l_blocking_lock = NULL;
2072         LDLM_LOCK_RELEASE(lock);
2073
2074         RETURN(rc);
2075 }
2076
2077 /**
2078  * Process a call to completion AST callback for a lock in ast_work list
2079  */
2080 static int
2081 ldlm_work_cp_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
2082 {
2083         struct ldlm_cb_set_arg  *arg = opaq;
2084         int                      rc = 0;
2085         struct ldlm_lock        *lock;
2086         ldlm_completion_callback completion_callback;
2087         ENTRY;
2088
2089         if (list_empty(arg->list))
2090                 RETURN(-ENOENT);
2091
2092         lock = list_entry(arg->list->next, struct ldlm_lock, l_cp_ast);
2093
2094         /* It's possible to receive a completion AST before we've set
2095          * the l_completion_ast pointer: either because the AST arrived
2096          * before the reply, or simply because there's a small race
2097          * window between receiving the reply and finishing the local
2098          * enqueue. (bug 842)
2099          *
2100          * This can't happen with the blocking_ast, however, because we
2101          * will never call the local blocking_ast until we drop our
2102          * reader/writer reference, which we won't do until we get the
2103          * reply and finish enqueueing. */
2104
2105         /* nobody should touch l_cp_ast */
2106         lock_res_and_lock(lock);
2107         list_del_init(&lock->l_cp_ast);
2108         LASSERT(ldlm_is_cp_reqd(lock));
2109         /* save l_completion_ast since it can be changed by
2110          * mds_intent_policy(), see bug 14225 */
2111         completion_callback = lock->l_completion_ast;
2112         ldlm_clear_cp_reqd(lock);
2113         unlock_res_and_lock(lock);
2114
2115         if (completion_callback != NULL)
2116                 rc = completion_callback(lock, 0, (void *)arg);
2117         LDLM_LOCK_RELEASE(lock);
2118
2119         RETURN(rc);
2120 }
2121
2122 /**
2123  * Process a call to revocation AST callback for a lock in ast_work list
2124  */
2125 static int
2126 ldlm_work_revoke_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
2127 {
2128         struct ldlm_cb_set_arg *arg = opaq;
2129         struct ldlm_lock_desc   desc;
2130         int                     rc;
2131         struct ldlm_lock       *lock;
2132         ENTRY;
2133
2134         if (list_empty(arg->list))
2135                 RETURN(-ENOENT);
2136
2137         lock = list_entry(arg->list->next, struct ldlm_lock, l_rk_ast);
2138         list_del_init(&lock->l_rk_ast);
2139
2140         /* the desc just pretend to exclusive */
2141         ldlm_lock2desc(lock, &desc);
2142         desc.l_req_mode = LCK_EX;
2143         desc.l_granted_mode = 0;
2144
2145         rc = lock->l_blocking_ast(lock, &desc, (void*)arg, LDLM_CB_BLOCKING);
2146         LDLM_LOCK_RELEASE(lock);
2147
2148         RETURN(rc);
2149 }
2150
2151 /**
2152  * Process a call to glimpse AST callback for a lock in ast_work list
2153  */
2154 int ldlm_work_gl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
2155 {
2156         struct ldlm_cb_set_arg          *arg = opaq;
2157         struct ldlm_glimpse_work        *gl_work;
2158         struct ldlm_lock                *lock;
2159         int                              rc = 0;
2160         ENTRY;
2161
2162         if (list_empty(arg->list))
2163                 RETURN(-ENOENT);
2164
2165         gl_work = list_entry(arg->list->next, struct ldlm_glimpse_work,
2166                                  gl_list);
2167         list_del_init(&gl_work->gl_list);
2168
2169         lock = gl_work->gl_lock;
2170
2171         /* transfer the glimpse descriptor to ldlm_cb_set_arg */
2172         arg->gl_desc = gl_work->gl_desc;
2173         arg->gl_interpret_reply = gl_work->gl_interpret_reply;
2174         arg->gl_interpret_data = gl_work->gl_interpret_data;
2175
2176         /* invoke the actual glimpse callback */
2177         if (lock->l_glimpse_ast(lock, (void*)arg) == 0)
2178                 rc = 1;
2179
2180         LDLM_LOCK_RELEASE(lock);
2181         if (gl_work->gl_flags & LDLM_GL_WORK_SLAB_ALLOCATED)
2182                 OBD_SLAB_FREE_PTR(gl_work, ldlm_glimpse_work_kmem);
2183         else
2184                 OBD_FREE_PTR(gl_work);
2185
2186         RETURN(rc);
2187 }
2188
2189 /**
2190  * Process list of locks in need of ASTs being sent.
2191  *
2192  * Used on server to send multiple ASTs together instead of sending one by
2193  * one.
2194  */
2195 int ldlm_run_ast_work(struct ldlm_namespace *ns, struct list_head *rpc_list,
2196                       ldlm_desc_ast_t ast_type)
2197 {
2198         struct ldlm_cb_set_arg *arg;
2199         set_producer_func       work_ast_lock;
2200         int                     rc;
2201
2202         if (list_empty(rpc_list))
2203                 RETURN(0);
2204
2205         OBD_ALLOC_PTR(arg);
2206         if (arg == NULL)
2207                 RETURN(-ENOMEM);
2208
2209         atomic_set(&arg->restart, 0);
2210         arg->list = rpc_list;
2211
2212         switch (ast_type) {
2213                 case LDLM_WORK_BL_AST:
2214                         arg->type = LDLM_BL_CALLBACK;
2215                         work_ast_lock = ldlm_work_bl_ast_lock;
2216                         break;
2217                 case LDLM_WORK_CP_AST:
2218                         arg->type = LDLM_CP_CALLBACK;
2219                         work_ast_lock = ldlm_work_cp_ast_lock;
2220                         break;
2221                 case LDLM_WORK_REVOKE_AST:
2222                         arg->type = LDLM_BL_CALLBACK;
2223                         work_ast_lock = ldlm_work_revoke_ast_lock;
2224                         break;
2225                 case LDLM_WORK_GL_AST:
2226                         arg->type = LDLM_GL_CALLBACK;
2227                         work_ast_lock = ldlm_work_gl_ast_lock;
2228                         break;
2229                 default:
2230                         LBUG();
2231         }
2232
2233         /* We create a ptlrpc request set with flow control extension.
2234          * This request set will use the work_ast_lock function to produce new
2235          * requests and will send a new request each time one completes in order
2236          * to keep the number of requests in flight to ns_max_parallel_ast */
2237         arg->set = ptlrpc_prep_fcset(ns->ns_max_parallel_ast ? : UINT_MAX,
2238                                      work_ast_lock, arg);
2239         if (arg->set == NULL)
2240                 GOTO(out, rc = -ENOMEM);
2241
2242         ptlrpc_set_wait(arg->set);
2243         ptlrpc_set_destroy(arg->set);
2244
2245         rc = atomic_read(&arg->restart) ? -ERESTART : 0;
2246         GOTO(out, rc);
2247 out:
2248         OBD_FREE_PTR(arg);
2249         return rc;
2250 }
2251
2252 /**
2253  * Try to grant all waiting locks on a resource.
2254  *
2255  * Calls ldlm_reprocess_queue on waiting queue.
2256  *
2257  * Typically called after some resource locks are cancelled to see
2258  * if anything could be granted as a result of the cancellation.
2259  */
2260 static void __ldlm_reprocess_all(struct ldlm_resource *res,
2261                                  enum ldlm_process_intention intention)
2262 {
2263         struct list_head rpc_list;
2264 #ifdef HAVE_SERVER_SUPPORT
2265         struct obd_device *obd;
2266         int rc;
2267         ENTRY;
2268
2269         INIT_LIST_HEAD(&rpc_list);
2270         /* Local lock trees don't get reprocessed. */
2271         if (ns_is_client(ldlm_res_to_ns(res))) {
2272                 EXIT;
2273                 return;
2274         }
2275
2276         /* Disable reprocess during lock replay stage but allow during
2277          * request replay stage.
2278          */
2279         obd = ldlm_res_to_ns(res)->ns_obd;
2280         if (obd->obd_recovering &&
2281             atomic_read(&obd->obd_req_replay_clients) == 0)
2282                 RETURN_EXIT;
2283 restart:
2284         lock_res(res);
2285         ldlm_reprocess_queue(res, &res->lr_waiting, &rpc_list, intention);
2286         unlock_res(res);
2287
2288         rc = ldlm_run_ast_work(ldlm_res_to_ns(res), &rpc_list,
2289                                LDLM_WORK_CP_AST);
2290         if (rc == -ERESTART) {
2291                 LASSERT(list_empty(&rpc_list));
2292                 goto restart;
2293         }
2294 #else
2295         ENTRY;
2296
2297         INIT_LIST_HEAD(&rpc_list);
2298         if (!ns_is_client(ldlm_res_to_ns(res))) {
2299                 CERROR("This is client-side-only module, cannot handle "
2300                        "LDLM_NAMESPACE_SERVER resource type lock.\n");
2301                 LBUG();
2302         }
2303 #endif
2304         EXIT;
2305 }
2306
2307 void ldlm_reprocess_all(struct ldlm_resource *res)
2308 {
2309         __ldlm_reprocess_all(res, LDLM_PROCESS_RESCAN);
2310 }
2311 EXPORT_SYMBOL(ldlm_reprocess_all);
2312
2313 static int ldlm_reprocess_res(struct cfs_hash *hs, struct cfs_hash_bd *bd,
2314                               struct hlist_node *hnode, void *arg)
2315 {
2316         struct ldlm_resource *res = cfs_hash_object(hs, hnode);
2317
2318         /* This is only called once after recovery done. LU-8306. */
2319         __ldlm_reprocess_all(res, LDLM_PROCESS_RECOVERY);
2320         return 0;
2321 }
2322
2323 /**
2324  * Iterate through all resources on a namespace attempting to grant waiting
2325  * locks.
2326  */
2327 void ldlm_reprocess_recovery_done(struct ldlm_namespace *ns)
2328 {
2329         ENTRY;
2330
2331         if (ns != NULL) {
2332                 cfs_hash_for_each_nolock(ns->ns_rs_hash,
2333                                          ldlm_reprocess_res, NULL, 0);
2334         }
2335         EXIT;
2336 }
2337
2338 /**
2339  * Helper function to call blocking AST for LDLM lock \a lock in a
2340  * "cancelling" mode.
2341  */
2342 void ldlm_cancel_callback(struct ldlm_lock *lock)
2343 {
2344         check_res_locked(lock->l_resource);
2345         if (!ldlm_is_cancel(lock)) {
2346                 ldlm_set_cancel(lock);
2347                 if (lock->l_blocking_ast) {
2348                         unlock_res_and_lock(lock);
2349                         lock->l_blocking_ast(lock, NULL, lock->l_ast_data,
2350                                              LDLM_CB_CANCELING);
2351                         lock_res_and_lock(lock);
2352                 } else {
2353                         LDLM_DEBUG(lock, "no blocking ast");
2354                 }
2355
2356                 /* only canceller can set bl_done bit */
2357                 ldlm_set_bl_done(lock);
2358                 wake_up_all(&lock->l_waitq);
2359         } else if (!ldlm_is_bl_done(lock)) {
2360                 struct l_wait_info lwi = { 0 };
2361
2362                 /* The lock is guaranteed to have been canceled once
2363                  * returning from this function. */
2364                 unlock_res_and_lock(lock);
2365                 l_wait_event(lock->l_waitq, is_bl_done(lock), &lwi);
2366                 lock_res_and_lock(lock);
2367         }
2368 }
2369
2370 /**
2371  * Remove skiplist-enabled LDLM lock \a req from granted list
2372  */
2373 void ldlm_unlink_lock_skiplist(struct ldlm_lock *req)
2374 {
2375         if (req->l_resource->lr_type != LDLM_PLAIN &&
2376             req->l_resource->lr_type != LDLM_IBITS)
2377                 return;
2378
2379         list_del_init(&req->l_sl_policy);
2380         list_del_init(&req->l_sl_mode);
2381 }
2382
2383 /**
2384  * Attempts to cancel LDLM lock \a lock that has no reader/writer references.
2385  */
2386 void ldlm_lock_cancel(struct ldlm_lock *lock)
2387 {
2388         struct ldlm_resource *res;
2389         struct ldlm_namespace *ns;
2390         ENTRY;
2391
2392         lock_res_and_lock(lock);
2393
2394         res = lock->l_resource;
2395         ns  = ldlm_res_to_ns(res);
2396
2397         /* Please do not, no matter how tempting, remove this LBUG without
2398          * talking to me first. -phik */
2399         if (lock->l_readers || lock->l_writers) {
2400                 LDLM_ERROR(lock, "lock still has references");
2401                 LBUG();
2402         }
2403
2404         if (ldlm_is_waited(lock))
2405                 ldlm_del_waiting_lock(lock);
2406
2407         /* Releases cancel callback. */
2408         ldlm_cancel_callback(lock);
2409
2410         /* Yes, second time, just in case it was added again while we were
2411          * running with no res lock in ldlm_cancel_callback */
2412         if (ldlm_is_waited(lock))
2413                 ldlm_del_waiting_lock(lock);
2414
2415         ldlm_resource_unlink_lock(lock);
2416         ldlm_lock_destroy_nolock(lock);
2417
2418         if (lock->l_granted_mode == lock->l_req_mode)
2419                 ldlm_pool_del(&ns->ns_pool, lock);
2420
2421         /* Make sure we will not be called again for same lock what is possible
2422          * if not to zero out lock->l_granted_mode */
2423         lock->l_granted_mode = LCK_MINMODE;
2424         unlock_res_and_lock(lock);
2425
2426         EXIT;
2427 }
2428 EXPORT_SYMBOL(ldlm_lock_cancel);
2429
2430 /**
2431  * Set opaque data into the lock that only makes sense to upper layer.
2432  */
2433 int ldlm_lock_set_data(const struct lustre_handle *lockh, void *data)
2434 {
2435         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
2436         int rc = -EINVAL;
2437         ENTRY;
2438
2439         if (lock) {
2440                 if (lock->l_ast_data == NULL)
2441                         lock->l_ast_data = data;
2442                 if (lock->l_ast_data == data)
2443                         rc = 0;
2444                 LDLM_LOCK_PUT(lock);
2445         }
2446         RETURN(rc);
2447 }
2448 EXPORT_SYMBOL(ldlm_lock_set_data);
2449
2450 struct export_cl_data {
2451         struct obd_export       *ecl_exp;
2452         int                     ecl_loop;
2453 };
2454
2455 static void ldlm_cancel_lock_for_export(struct obd_export *exp,
2456                                         struct ldlm_lock *lock,
2457                                         struct export_cl_data *ecl)
2458 {
2459         struct ldlm_resource *res;
2460
2461         res = ldlm_resource_getref(lock->l_resource);
2462
2463         ldlm_lvbo_update(res, lock, NULL, 1);
2464         ldlm_lock_cancel(lock);
2465         if (!exp->exp_obd->obd_stopping)
2466                 ldlm_reprocess_all(res);
2467         ldlm_resource_putref(res);
2468
2469         ecl->ecl_loop++;
2470         if ((ecl->ecl_loop & -ecl->ecl_loop) == ecl->ecl_loop) {
2471                 CDEBUG(D_INFO, "Export %p, %d locks cancelled.\n",
2472                        exp, ecl->ecl_loop);
2473         }
2474 }
2475
2476 /**
2477  * Iterator function for ldlm_export_cancel_locks.
2478  * Cancels passed locks.
2479  */
2480 static int
2481 ldlm_cancel_locks_for_export_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
2482                                 struct hlist_node *hnode, void *data)
2483
2484 {
2485         struct export_cl_data   *ecl = (struct export_cl_data *)data;
2486         struct obd_export       *exp  = ecl->ecl_exp;
2487         struct ldlm_lock        *lock = cfs_hash_object(hs, hnode);
2488
2489         LDLM_LOCK_GET(lock);
2490         ldlm_cancel_lock_for_export(exp, lock, ecl);
2491         LDLM_LOCK_RELEASE(lock);
2492
2493         return 0;
2494 }
2495
2496 /**
2497  * Cancel all blocked locks for given export.
2498  *
2499  * Typically called on client disconnection/eviction
2500  */
2501 int ldlm_export_cancel_blocked_locks(struct obd_export *exp)
2502 {
2503         struct export_cl_data   ecl = {
2504                 .ecl_exp        = exp,
2505                 .ecl_loop       = 0,
2506         };
2507
2508         while (!list_empty(&exp->exp_bl_list)) {
2509                 struct ldlm_lock *lock;
2510
2511                 spin_lock_bh(&exp->exp_bl_list_lock);
2512                 if (!list_empty(&exp->exp_bl_list)) {
2513                         lock = list_entry(exp->exp_bl_list.next,
2514                                           struct ldlm_lock, l_exp_list);
2515                         LDLM_LOCK_GET(lock);
2516                         list_del_init(&lock->l_exp_list);
2517                 } else {
2518                         lock = NULL;
2519                 }
2520                 spin_unlock_bh(&exp->exp_bl_list_lock);
2521
2522                 if (lock == NULL)
2523                         break;
2524
2525                 ldlm_cancel_lock_for_export(exp, lock, &ecl);
2526                 LDLM_LOCK_RELEASE(lock);
2527         }
2528
2529         CDEBUG(D_DLMTRACE, "Export %p, canceled %d locks, "
2530                "left on hash table %d.\n", exp, ecl.ecl_loop,
2531                atomic_read(&exp->exp_lock_hash->hs_count));
2532
2533         return ecl.ecl_loop;
2534 }
2535
2536 /**
2537  * Cancel all locks for given export.
2538  *
2539  * Typically called after client disconnection/eviction
2540  */
2541 int ldlm_export_cancel_locks(struct obd_export *exp)
2542 {
2543         struct export_cl_data   ecl = {
2544                 .ecl_exp        = exp,
2545                 .ecl_loop       = 0,
2546         };
2547
2548         cfs_hash_for_each_empty(exp->exp_lock_hash,
2549                                 ldlm_cancel_locks_for_export_cb, &ecl);
2550
2551         CDEBUG(D_DLMTRACE, "Export %p, canceled %d locks, "
2552                "left on hash table %d.\n", exp, ecl.ecl_loop,
2553                atomic_read(&exp->exp_lock_hash->hs_count));
2554
2555         if (ecl.ecl_loop > 0 &&
2556             atomic_read(&exp->exp_lock_hash->hs_count) == 0 &&
2557             exp->exp_obd->obd_stopping)
2558                 ldlm_reprocess_recovery_done(exp->exp_obd->obd_namespace);
2559
2560         return ecl.ecl_loop;
2561 }
2562
2563 /**
2564  * Downgrade an PW/EX lock to COS mode.
2565  *
2566  * A lock mode convertion from PW/EX mode to less conflict mode. The
2567  * convertion may fail if lock was canceled before downgrade, but it doesn't
2568  * indicate any problem, because such lock has no reader or writer, and will
2569  * be released soon.
2570  * Used by Commit on Sharing (COS) code only for now.
2571  *
2572  * \param lock A lock to convert
2573  * \param new_mode new lock mode
2574  */
2575 void ldlm_lock_mode_downgrade(struct ldlm_lock *lock, enum ldlm_mode new_mode)
2576 {
2577         ENTRY;
2578
2579         LASSERT(new_mode == LCK_COS);
2580
2581         lock_res_and_lock(lock);
2582
2583         if (!(lock->l_granted_mode & (LCK_PW | LCK_EX))) {
2584                 unlock_res_and_lock(lock);
2585
2586                 LASSERT(lock->l_granted_mode == LCK_MINMODE);
2587                 LDLM_DEBUG(lock, "lock was canceled before downgrade");
2588                 RETURN_EXIT;
2589         }
2590
2591         ldlm_resource_unlink_lock(lock);
2592         /*
2593          * Remove the lock from pool as it will be added again in
2594          * ldlm_grant_lock() called below.
2595          */
2596         ldlm_pool_del(&ldlm_lock_to_ns(lock)->ns_pool, lock);
2597         lock->l_req_mode = new_mode;
2598         ldlm_grant_lock(lock, NULL);
2599
2600         unlock_res_and_lock(lock);
2601
2602         ldlm_reprocess_all(lock->l_resource);
2603
2604         EXIT;
2605 }
2606 EXPORT_SYMBOL(ldlm_lock_mode_downgrade);
2607
2608 /**
2609  * Print lock with lock handle \a lockh description into debug log.
2610  *
2611  * Used when printing all locks on a resource for debug purposes.
2612  */
2613 void ldlm_lock_dump_handle(int level, const struct lustre_handle *lockh)
2614 {
2615         struct ldlm_lock *lock;
2616
2617         if (!((libcfs_debug | D_ERROR) & level))
2618                 return;
2619
2620         lock = ldlm_handle2lock(lockh);
2621         if (lock == NULL)
2622                 return;
2623
2624         LDLM_DEBUG_LIMIT(level, lock, "###");
2625
2626         LDLM_LOCK_PUT(lock);
2627 }
2628 EXPORT_SYMBOL(ldlm_lock_dump_handle);
2629
2630 /**
2631  * Print lock information with custom message into debug log.
2632  * Helper function.
2633  */
2634 void _ldlm_lock_debug(struct ldlm_lock *lock,
2635                       struct libcfs_debug_msg_data *msgdata,
2636                       const char *fmt, ...)
2637 {
2638         va_list args;
2639         struct obd_export *exp = lock->l_export;
2640         struct ldlm_resource *resource = NULL;
2641         char *nid = "local";
2642
2643         /* on server-side resource of lock doesn't change */
2644         if ((lock->l_flags & LDLM_FL_NS_SRV) != 0) {
2645                 if (lock->l_resource != NULL)
2646                         resource = ldlm_resource_getref(lock->l_resource);
2647         } else if (spin_trylock(&lock->l_lock)) {
2648                 if (lock->l_resource != NULL)
2649                         resource = ldlm_resource_getref(lock->l_resource);
2650                 spin_unlock(&lock->l_lock);
2651         }
2652
2653         va_start(args, fmt);
2654
2655         if (exp && exp->exp_connection) {
2656                 nid = obd_export_nid2str(exp);
2657         } else if (exp && exp->exp_obd != NULL) {
2658                 struct obd_import *imp = exp->exp_obd->u.cli.cl_import;
2659                 nid = obd_import_nid2str(imp);
2660         }
2661
2662         if (resource == NULL) {
2663                 libcfs_debug_vmsg2(msgdata, fmt, args,
2664                        " ns: \?\? lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s "
2665                        "res: \?\? rrc=\?\? type: \?\?\? flags: %#llx nid: %s "
2666                        "remote: %#llx expref: %d pid: %u timeout: %lld "
2667                        "lvb_type: %d\n",
2668                        lock,
2669                        lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
2670                        lock->l_readers, lock->l_writers,
2671                        ldlm_lockname[lock->l_granted_mode],
2672                        ldlm_lockname[lock->l_req_mode],
2673                        lock->l_flags, nid, lock->l_remote_handle.cookie,
2674                        exp ? atomic_read(&exp->exp_refcount) : -99,
2675                        lock->l_pid, lock->l_callback_timeout, lock->l_lvb_type);
2676                 va_end(args);
2677                 return;
2678         }
2679
2680         switch (resource->lr_type) {
2681         case LDLM_EXTENT:
2682                 libcfs_debug_vmsg2(msgdata, fmt, args,
2683                         " ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s "
2684                         "res: "DLDLMRES" rrc: %d type: %s [%llu->%llu] "
2685                         "(req %llu->%llu) flags: %#llx nid: %s remote: "
2686                         "%#llx expref: %d pid: %u timeout: %lld lvb_type: %d\n",
2687                         ldlm_lock_to_ns_name(lock), lock,
2688                         lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
2689                         lock->l_readers, lock->l_writers,
2690                         ldlm_lockname[lock->l_granted_mode],
2691                         ldlm_lockname[lock->l_req_mode],
2692                         PLDLMRES(resource),
2693                         atomic_read(&resource->lr_refcount),
2694                         ldlm_typename[resource->lr_type],
2695                         lock->l_policy_data.l_extent.start,
2696                         lock->l_policy_data.l_extent.end,
2697                         lock->l_req_extent.start, lock->l_req_extent.end,
2698                         lock->l_flags, nid, lock->l_remote_handle.cookie,
2699                         exp ? atomic_read(&exp->exp_refcount) : -99,
2700                         lock->l_pid, lock->l_callback_timeout,
2701                         lock->l_lvb_type);
2702                 break;
2703
2704         case LDLM_FLOCK:
2705                 libcfs_debug_vmsg2(msgdata, fmt, args,
2706                         " ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s "
2707                         "res: "DLDLMRES" rrc: %d type: %s pid: %d "
2708                         "[%llu->%llu] flags: %#llx nid: %s "
2709                         "remote: %#llx expref: %d pid: %u timeout: %lld\n",
2710                         ldlm_lock_to_ns_name(lock), lock,
2711                         lock->l_handle.h_cookie, atomic_read(&lock->l_refc),
2712                         lock->l_readers, lock->l_writers,
2713                         ldlm_lockname[lock->l_granted_mode],
2714                         ldlm_lockname[lock->l_req_mode],
2715                         PLDLMRES(resource),
2716                         atomic_read(&resource->lr_refcount),
2717                         ldlm_typename[resource->lr_type],
2718                         lock->l_policy_data.l_flock.pid,
2719                         lock->l_policy_data.l_flock.start,
2720                         lock->l_policy_data.l_flock.end,
2721                         lock->l_flags, nid, lock->l_remote_handle.cookie,
2722                         exp ? atomic_read(&exp->exp_refcount) : -99,
2723                         lock->l_pid, lock->l_callback_timeout);
2724                 break;
2725
2726         case LDLM_IBITS:
2727                 libcfs_debug_vmsg2(msgdata, fmt, args,
2728                         " ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s "
2729                         "res: "DLDLMRES" bits %#llx/%#llx rrc: %d type: %s "
2730                         "flags: %#llx nid: %s remote: %#llx expref: %d "
2731                         "pid: %u timeout: %lld lvb_type: %d\n",
2732                         ldlm_lock_to_ns_name(lock),
2733                         lock, lock->l_handle.h_cookie,
2734                         atomic_read(&lock->l_refc),
2735                         lock->l_readers, lock->l_writers,
2736                         ldlm_lockname[lock->l_granted_mode],
2737                         ldlm_lockname[lock->l_req_mode],
2738                         PLDLMRES(resource),
2739                         lock->l_policy_data.l_inodebits.bits,
2740                         lock->l_policy_data.l_inodebits.try_bits,
2741                         atomic_read(&resource->lr_refcount),
2742                         ldlm_typename[resource->lr_type],
2743                         lock->l_flags, nid, lock->l_remote_handle.cookie,
2744                         exp ? atomic_read(&exp->exp_refcount) : -99,
2745                         lock->l_pid, lock->l_callback_timeout,
2746                         lock->l_lvb_type);
2747                 break;
2748
2749         default:
2750                 libcfs_debug_vmsg2(msgdata, fmt, args,
2751                         " ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s "
2752                         "res: "DLDLMRES" rrc: %d type: %s flags: %#llx "
2753                         "nid: %s remote: %#llx expref: %d pid: %u "
2754                         "timeout: %lld lvb_type: %d\n",
2755                         ldlm_lock_to_ns_name(lock),
2756                         lock, lock->l_handle.h_cookie,
2757                         atomic_read(&lock->l_refc),
2758                         lock->l_readers, lock->l_writers,
2759                         ldlm_lockname[lock->l_granted_mode],
2760                         ldlm_lockname[lock->l_req_mode],
2761                         PLDLMRES(resource),
2762                         atomic_read(&resource->lr_refcount),
2763                         ldlm_typename[resource->lr_type],
2764                         lock->l_flags, nid, lock->l_remote_handle.cookie,
2765                         exp ? atomic_read(&exp->exp_refcount) : -99,
2766                         lock->l_pid, lock->l_callback_timeout,
2767                         lock->l_lvb_type);
2768                 break;
2769         }
2770         va_end(args);
2771         ldlm_resource_putref(resource);
2772 }
2773 EXPORT_SYMBOL(_ldlm_lock_debug);