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