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