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