Whamcloud - gitweb
5d3b781cb5e9d0a212f0d3faf6bf63a087e3885b
[fs/lustre-release.git] / lustre / ldlm / ldlm_lockd.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2010, 2016, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * lustre/ldlm/ldlm_lockd.c
33  *
34  * Author: Peter Braam <braam@clusterfs.com>
35  * Author: Phil Schwan <phil@clusterfs.com>
36  */
37
38 #define DEBUG_SUBSYSTEM S_LDLM
39
40 #include <linux/kthread.h>
41 #include <linux/list.h>
42 #include <libcfs/libcfs.h>
43 #include <lustre_errno.h>
44 #include <lustre_dlm.h>
45 #include <obd_class.h>
46 #include "ldlm_internal.h"
47
48 static int ldlm_num_threads;
49 module_param(ldlm_num_threads, int, 0444);
50 MODULE_PARM_DESC(ldlm_num_threads, "number of DLM service threads to start");
51
52 static char *ldlm_cpts;
53 module_param(ldlm_cpts, charp, 0444);
54 MODULE_PARM_DESC(ldlm_cpts, "CPU partitions ldlm threads should run on");
55
56 static DEFINE_MUTEX(ldlm_ref_mutex);
57 static int ldlm_refcount;
58
59 struct kobject *ldlm_kobj;
60 struct kset *ldlm_ns_kset;
61 struct kset *ldlm_svc_kset;
62
63 /* LDLM state */
64
65 static struct ldlm_state *ldlm_state;
66
67 static inline cfs_time_t round_timeout(cfs_time_t timeout)
68 {
69         return cfs_time_seconds((int)cfs_duration_sec(cfs_time_sub(timeout, 0)) + 1);
70 }
71
72 /* timeout for initial callback (AST) reply (bz10399) */
73 static inline unsigned int ldlm_get_rq_timeout(void)
74 {
75         /* Non-AT value */
76         unsigned int timeout = min(ldlm_timeout, obd_timeout / 3);
77
78         return timeout < 1 ? 1 : timeout;
79 }
80
81 struct ldlm_bl_pool {
82         spinlock_t              blp_lock;
83
84         /*
85          * blp_prio_list is used for callbacks that should be handled
86          * as a priority. It is used for LDLM_FL_DISCARD_DATA requests.
87          * see bug 13843
88          */
89         struct list_head              blp_prio_list;
90
91         /*
92          * blp_list is used for all other callbacks which are likely
93          * to take longer to process.
94          */
95         struct list_head              blp_list;
96
97         wait_queue_head_t       blp_waitq;
98         struct completion       blp_comp;
99         atomic_t            blp_num_threads;
100         atomic_t            blp_busy_threads;
101         int                     blp_min_threads;
102         int                     blp_max_threads;
103 };
104
105 struct ldlm_bl_work_item {
106         struct list_head        blwi_entry;
107         struct ldlm_namespace   *blwi_ns;
108         struct ldlm_lock_desc   blwi_ld;
109         struct ldlm_lock        *blwi_lock;
110         struct list_head        blwi_head;
111         int                     blwi_count;
112         struct completion       blwi_comp;
113         enum ldlm_cancel_flags  blwi_flags;
114         int                     blwi_mem_pressure;
115 };
116
117 #ifdef HAVE_SERVER_SUPPORT
118
119 /**
120  * Protects both waiting_locks_list and expired_lock_thread.
121  */
122 static DEFINE_SPINLOCK(waiting_locks_spinlock); /* BH lock (timer) */
123
124 /**
125  * List for contended locks.
126  *
127  * As soon as a lock is contended, it gets placed on this list and
128  * expected time to get a response is filled in the lock. A special
129  * thread walks the list looking for locks that should be released and
130  * schedules client evictions for those that have not been released in
131  * time.
132  *
133  * All access to it should be under waiting_locks_spinlock.
134  */
135 static LIST_HEAD(waiting_locks_list);
136 static void waiting_locks_callback(unsigned long unused);
137 static DEFINE_TIMER(waiting_locks_timer, waiting_locks_callback, 0, 0);
138
139 enum elt_state {
140         ELT_STOPPED,
141         ELT_READY,
142         ELT_TERMINATE,
143 };
144
145 static DECLARE_WAIT_QUEUE_HEAD(expired_lock_wait_queue);
146 static enum elt_state expired_lock_thread_state = ELT_STOPPED;
147 static int expired_lock_dump;
148 static LIST_HEAD(expired_lock_list);
149
150 static inline int have_expired_locks(void)
151 {
152         int need_to_run;
153
154         ENTRY;
155         spin_lock_bh(&waiting_locks_spinlock);
156         need_to_run = !list_empty(&expired_lock_list);
157         spin_unlock_bh(&waiting_locks_spinlock);
158
159         RETURN(need_to_run);
160 }
161
162 /**
163  * Check expired lock list for expired locks and time them out.
164  */
165 static int expired_lock_main(void *arg)
166 {
167         struct list_head *expired = &expired_lock_list;
168         struct l_wait_info lwi = { 0 };
169         int do_dump;
170
171         ENTRY;
172
173         expired_lock_thread_state = ELT_READY;
174         wake_up(&expired_lock_wait_queue);
175
176         while (1) {
177                 l_wait_event(expired_lock_wait_queue,
178                              have_expired_locks() ||
179                              expired_lock_thread_state == ELT_TERMINATE,
180                              &lwi);
181
182                 spin_lock_bh(&waiting_locks_spinlock);
183                 if (expired_lock_dump) {
184                         spin_unlock_bh(&waiting_locks_spinlock);
185
186                         /* from waiting_locks_callback, but not in timer */
187                         libcfs_debug_dumplog();
188
189                         spin_lock_bh(&waiting_locks_spinlock);
190                         expired_lock_dump = 0;
191                 }
192
193                 do_dump = 0;
194
195                 while (!list_empty(expired)) {
196                         struct obd_export *export;
197                         struct ldlm_lock *lock;
198
199                         lock = list_entry(expired->next, struct ldlm_lock,
200                                           l_pending_chain);
201                         if ((void *)lock < LP_POISON + PAGE_SIZE &&
202                             (void *)lock >= LP_POISON) {
203                                 spin_unlock_bh(&waiting_locks_spinlock);
204                                 CERROR("free lock on elt list %p\n", lock);
205                                 LBUG();
206                         }
207                         list_del_init(&lock->l_pending_chain);
208                         if ((void *)lock->l_export <
209                              LP_POISON + PAGE_SIZE &&
210                             (void *)lock->l_export >= LP_POISON) {
211                                 CERROR("lock with free export on elt list %p\n",
212                                        lock->l_export);
213                                 lock->l_export = NULL;
214                                 LDLM_ERROR(lock, "free export");
215                                 /* release extra ref grabbed by
216                                  * ldlm_add_waiting_lock() or
217                                  * ldlm_failed_ast() */
218                                 LDLM_LOCK_RELEASE(lock);
219                                 continue;
220                         }
221
222                         if (ldlm_is_destroyed(lock)) {
223                                 /* release the lock refcount where
224                                  * waiting_locks_callback() founds */
225                                 LDLM_LOCK_RELEASE(lock);
226                                 continue;
227                         }
228                         export = class_export_lock_get(lock->l_export, lock);
229                         spin_unlock_bh(&waiting_locks_spinlock);
230
231                         spin_lock_bh(&export->exp_bl_list_lock);
232                         list_del_init(&lock->l_exp_list);
233                         spin_unlock_bh(&export->exp_bl_list_lock);
234
235                         do_dump++;
236                         class_fail_export(export);
237                         class_export_lock_put(export, lock);
238
239                         /* release extra ref grabbed by ldlm_add_waiting_lock()
240                          * or ldlm_failed_ast() */
241                         LDLM_LOCK_RELEASE(lock);
242
243                         spin_lock_bh(&waiting_locks_spinlock);
244                 }
245                 spin_unlock_bh(&waiting_locks_spinlock);
246
247                 if (do_dump && obd_dump_on_eviction) {
248                         CERROR("dump the log upon eviction\n");
249                         libcfs_debug_dumplog();
250                 }
251
252                 if (expired_lock_thread_state == ELT_TERMINATE)
253                         break;
254         }
255
256         expired_lock_thread_state = ELT_STOPPED;
257         wake_up(&expired_lock_wait_queue);
258         RETURN(0);
259 }
260
261 static int ldlm_add_waiting_lock(struct ldlm_lock *lock);
262 static int __ldlm_add_waiting_lock(struct ldlm_lock *lock, int seconds);
263
264 /**
265  * Check if there is a request in the export request list
266  * which prevents the lock canceling.
267  */
268 static int ldlm_lock_busy(struct ldlm_lock *lock)
269 {
270         struct ptlrpc_request *req;
271         int match = 0;
272         ENTRY;
273
274         if (lock->l_export == NULL)
275                 return 0;
276
277         spin_lock_bh(&lock->l_export->exp_rpc_lock);
278         list_for_each_entry(req, &lock->l_export->exp_hp_rpcs,
279                                 rq_exp_list) {
280                 if (req->rq_ops->hpreq_lock_match) {
281                         match = req->rq_ops->hpreq_lock_match(req, lock);
282                         if (match)
283                                 break;
284                 }
285         }
286         spin_unlock_bh(&lock->l_export->exp_rpc_lock);
287         RETURN(match);
288 }
289
290 /* This is called from within a timer interrupt and cannot schedule */
291 static void waiting_locks_callback(unsigned long unused)
292 {
293         struct ldlm_lock        *lock;
294         int                     need_dump = 0;
295
296         spin_lock_bh(&waiting_locks_spinlock);
297         while (!list_empty(&waiting_locks_list)) {
298                 lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
299                                       l_pending_chain);
300                 if (cfs_time_after(lock->l_callback_timeout,
301                                    cfs_time_current()) ||
302                     (lock->l_req_mode == LCK_GROUP))
303                         break;
304
305                 /* Check if we need to prolong timeout */
306                 if (!OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT) &&
307                     ldlm_lock_busy(lock)) {
308                         int cont = 1;
309
310                         if (lock->l_pending_chain.next == &waiting_locks_list)
311                                 cont = 0;
312
313                         LDLM_LOCK_GET(lock);
314
315                         spin_unlock_bh(&waiting_locks_spinlock);
316                         LDLM_DEBUG(lock, "prolong the busy lock");
317                         ldlm_refresh_waiting_lock(lock,
318                                                   ldlm_bl_timeout(lock) >> 1);
319                         spin_lock_bh(&waiting_locks_spinlock);
320
321                         if (!cont) {
322                                 LDLM_LOCK_RELEASE(lock);
323                                 break;
324                         }
325
326                         LDLM_LOCK_RELEASE(lock);
327                         continue;
328                 }
329                 ldlm_lock_to_ns(lock)->ns_timeouts++;
330                 LDLM_ERROR(lock, "lock callback timer expired after %llds: "
331                            "evicting client at %s ",
332                            ktime_get_real_seconds() - lock->l_last_activity,
333                            libcfs_nid2str(
334                                    lock->l_export->exp_connection->c_peer.nid));
335
336                 /* no needs to take an extra ref on the lock since it was in
337                  * the waiting_locks_list and ldlm_add_waiting_lock()
338                  * already grabbed a ref */
339                 list_del(&lock->l_pending_chain);
340                 list_add(&lock->l_pending_chain, &expired_lock_list);
341                 need_dump = 1;
342         }
343
344         if (!list_empty(&expired_lock_list)) {
345                 if (obd_dump_on_timeout && need_dump)
346                         expired_lock_dump = __LINE__;
347
348                 wake_up(&expired_lock_wait_queue);
349         }
350
351         /*
352          * Make sure the timer will fire again if we have any locks
353          * left.
354          */
355         if (!list_empty(&waiting_locks_list)) {
356                 cfs_time_t timeout_rounded;
357                 lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
358                                       l_pending_chain);
359                 timeout_rounded = (cfs_time_t)round_timeout(lock->l_callback_timeout);
360                 mod_timer(&waiting_locks_timer, timeout_rounded);
361         }
362         spin_unlock_bh(&waiting_locks_spinlock);
363 }
364
365 /**
366  * Add lock to the list of contended locks.
367  *
368  * Indicate that we're waiting for a client to call us back cancelling a given
369  * lock.  We add it to the pending-callback chain, and schedule the lock-timeout
370  * timer to fire appropriately.  (We round up to the next second, to avoid
371  * floods of timer firings during periods of high lock contention and traffic).
372  * As done by ldlm_add_waiting_lock(), the caller must grab a lock reference
373  * if it has been added to the waiting list (1 is returned).
374  *
375  * Called with the namespace lock held.
376  */
377 static int __ldlm_add_waiting_lock(struct ldlm_lock *lock, int seconds)
378 {
379         cfs_time_t timeout;
380         cfs_time_t timeout_rounded;
381
382         if (!list_empty(&lock->l_pending_chain))
383                 return 0;
384
385         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT) ||
386             OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT))
387                 seconds = 1;
388
389         timeout = cfs_time_shift(seconds);
390         if (likely(cfs_time_after(timeout, lock->l_callback_timeout)))
391                 lock->l_callback_timeout = timeout;
392
393         timeout_rounded = round_timeout(lock->l_callback_timeout);
394
395         if (cfs_time_before(timeout_rounded, waiting_locks_timer.expires) ||
396             !timer_pending(&waiting_locks_timer)) {
397                 mod_timer(&waiting_locks_timer, timeout_rounded);
398         }
399         /* if the new lock has a shorter timeout than something earlier on
400            the list, we'll wait the longer amount of time; no big deal. */
401         /* FIFO */
402         list_add_tail(&lock->l_pending_chain, &waiting_locks_list);
403         return 1;
404 }
405
406 static void ldlm_add_blocked_lock(struct ldlm_lock *lock)
407 {
408         spin_lock_bh(&lock->l_export->exp_bl_list_lock);
409         if (list_empty(&lock->l_exp_list)) {
410                 if (lock->l_granted_mode != lock->l_req_mode)
411                         list_add_tail(&lock->l_exp_list,
412                                       &lock->l_export->exp_bl_list);
413                 else
414                         list_add(&lock->l_exp_list,
415                                  &lock->l_export->exp_bl_list);
416         }
417         spin_unlock_bh(&lock->l_export->exp_bl_list_lock);
418
419         /* A blocked lock is added. Adjust the position in
420          * the stale list if the export is in the list.
421          * If export is stale and not in the list - it is being
422          * processed and will be placed on the right position
423          * on obd_stale_export_put(). */
424         if (!list_empty(&lock->l_export->exp_stale_list))
425                 obd_stale_export_adjust(lock->l_export);
426 }
427
428 static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
429 {
430         int ret;
431         int timeout = ldlm_bl_timeout(lock);
432
433         /* NB: must be called with hold of lock_res_and_lock() */
434         LASSERT(ldlm_is_res_locked(lock));
435         LASSERT(!ldlm_is_cancel_on_block(lock));
436
437         /* Do not put cross-MDT lock in the waiting list, since we
438          * will not evict it due to timeout for now */
439         if (lock->l_export != NULL &&
440             (exp_connect_flags(lock->l_export) & OBD_CONNECT_MDS_MDS))
441                 return 0;
442
443         spin_lock_bh(&waiting_locks_spinlock);
444         if (ldlm_is_cancel(lock)) {
445                 spin_unlock_bh(&waiting_locks_spinlock);
446                 return 0;
447         }
448
449         if (ldlm_is_destroyed(lock)) {
450                 static cfs_time_t next;
451
452                 spin_unlock_bh(&waiting_locks_spinlock);
453                 LDLM_ERROR(lock, "not waiting on destroyed lock (bug 5653)");
454                 if (cfs_time_after(cfs_time_current(), next)) {
455                         next = cfs_time_shift(14400);
456                         libcfs_debug_dumpstack(NULL);
457                 }
458                 return 0;
459         }
460
461         ldlm_set_waited(lock);
462         lock->l_last_activity = ktime_get_real_seconds();
463         ret = __ldlm_add_waiting_lock(lock, timeout);
464         if (ret) {
465                 /* grab ref on the lock if it has been added to the
466                  * waiting list */
467                 LDLM_LOCK_GET(lock);
468         }
469         spin_unlock_bh(&waiting_locks_spinlock);
470
471         if (ret)
472                 ldlm_add_blocked_lock(lock);
473
474         LDLM_DEBUG(lock, "%sadding to wait list(timeout: %d, AT: %s)",
475                    ret == 0 ? "not re-" : "", timeout,
476                    AT_OFF ? "off" : "on");
477         return ret;
478 }
479
480 /**
481  * Remove a lock from the pending list, likely because it had its cancellation
482  * callback arrive without incident.  This adjusts the lock-timeout timer if
483  * needed.  Returns 0 if the lock wasn't pending after all, 1 if it was.
484  * As done by ldlm_del_waiting_lock(), the caller must release the lock
485  * reference when the lock is removed from any list (1 is returned).
486  *
487  * Called with namespace lock held.
488  */
489 static int __ldlm_del_waiting_lock(struct ldlm_lock *lock)
490 {
491         struct list_head *list_next;
492
493         if (list_empty(&lock->l_pending_chain))
494                 return 0;
495
496         list_next = lock->l_pending_chain.next;
497         if (lock->l_pending_chain.prev == &waiting_locks_list) {
498                 /* Removing the head of the list, adjust timer. */
499                 if (list_next == &waiting_locks_list) {
500                         /* No more, just cancel. */
501                         del_timer(&waiting_locks_timer);
502                 } else {
503                         struct ldlm_lock *next;
504                         next = list_entry(list_next, struct ldlm_lock,
505                                               l_pending_chain);
506                         mod_timer(&waiting_locks_timer,
507                                   round_timeout(next->l_callback_timeout));
508                 }
509         }
510         list_del_init(&lock->l_pending_chain);
511
512         return 1;
513 }
514
515 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
516 {
517         int ret;
518
519         if (lock->l_export == NULL) {
520                 /* We don't have a "waiting locks list" on clients. */
521                 CDEBUG(D_DLMTRACE, "Client lock %p : no-op\n", lock);
522                 return 0;
523         }
524
525         spin_lock_bh(&waiting_locks_spinlock);
526         ret = __ldlm_del_waiting_lock(lock);
527         ldlm_clear_waited(lock);
528         spin_unlock_bh(&waiting_locks_spinlock);
529
530         /* remove the lock out of export blocking list */
531         spin_lock_bh(&lock->l_export->exp_bl_list_lock);
532         list_del_init(&lock->l_exp_list);
533         spin_unlock_bh(&lock->l_export->exp_bl_list_lock);
534
535         if (ret) {
536                 /* release lock ref if it has indeed been removed
537                  * from a list */
538                 LDLM_LOCK_RELEASE(lock);
539         }
540
541         LDLM_DEBUG(lock, "%s", ret == 0 ? "wasn't waiting" : "removed");
542         return ret;
543 }
544
545 /**
546  * Prolong the contended lock waiting time.
547  *
548  * Called with namespace lock held.
549  */
550 int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
551 {
552         if (lock->l_export == NULL) {
553                 /* We don't have a "waiting locks list" on clients. */
554                 LDLM_DEBUG(lock, "client lock: no-op");
555                 return 0;
556         }
557
558         if (exp_connect_flags(lock->l_export) & OBD_CONNECT_MDS_MDS) {
559                 /* We don't have a "waiting locks list" on OSP. */
560                 LDLM_DEBUG(lock, "MDS-MDS lock: no-op");
561                 return 0;
562         }
563
564         spin_lock_bh(&waiting_locks_spinlock);
565
566         if (list_empty(&lock->l_pending_chain)) {
567                 spin_unlock_bh(&waiting_locks_spinlock);
568                 LDLM_DEBUG(lock, "wasn't waiting");
569                 return 0;
570         }
571
572         /* we remove/add the lock to the waiting list, so no needs to
573          * release/take a lock reference */
574         __ldlm_del_waiting_lock(lock);
575         __ldlm_add_waiting_lock(lock, timeout);
576         spin_unlock_bh(&waiting_locks_spinlock);
577
578         LDLM_DEBUG(lock, "refreshed");
579         return 1;
580 }
581 EXPORT_SYMBOL(ldlm_refresh_waiting_lock);
582
583 #else /* HAVE_SERVER_SUPPORT */
584
585 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
586 {
587         RETURN(0);
588 }
589
590 int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
591 {
592         RETURN(0);
593 }
594
595 #endif /* !HAVE_SERVER_SUPPORT */
596
597 #ifdef HAVE_SERVER_SUPPORT
598
599 /**
600  * Calculate the per-export Blocking timeout (covering BL AST, data flush,
601  * lock cancel, and their replies). Used for lock callback timeout and AST
602  * re-send period.
603  *
604  * \param[in] lock        lock which is getting the blocking callback
605  *
606  * \retval            timeout in seconds to wait for the client reply
607  */
608 unsigned int ldlm_bl_timeout(struct ldlm_lock *lock)
609 {
610         unsigned int timeout;
611
612         if (AT_OFF)
613                 return obd_timeout / 2;
614
615         /* Since these are non-updating timeouts, we should be conservative.
616          * Take more than usually, 150%
617          * It would be nice to have some kind of "early reply" mechanism for
618          * lock callbacks too... */
619         timeout = at_get(&lock->l_export->exp_bl_lock_at);
620         return max(timeout + (timeout >> 1), ldlm_enqueue_min);
621 }
622 EXPORT_SYMBOL(ldlm_bl_timeout);
623
624 /**
625  * Perform lock cleanup if AST sending failed.
626  */
627 static void ldlm_failed_ast(struct ldlm_lock *lock, int rc,
628                             const char *ast_type)
629 {
630         LCONSOLE_ERROR_MSG(0x138, "%s: A client on nid %s was evicted due "
631                            "to a lock %s callback time out: rc %d\n",
632                            lock->l_export->exp_obd->obd_name,
633                            obd_export_nid2str(lock->l_export), ast_type, rc);
634
635         if (obd_dump_on_timeout)
636                 libcfs_debug_dumplog();
637         spin_lock_bh(&waiting_locks_spinlock);
638         if (__ldlm_del_waiting_lock(lock) == 0)
639                 /* the lock was not in any list, grab an extra ref before adding
640                  * the lock to the expired list */
641                 LDLM_LOCK_GET(lock);
642         list_add(&lock->l_pending_chain, &expired_lock_list);
643         wake_up(&expired_lock_wait_queue);
644         spin_unlock_bh(&waiting_locks_spinlock);
645 }
646
647 /**
648  * Perform lock cleanup if AST reply came with error.
649  */
650 static int ldlm_handle_ast_error(struct ldlm_lock *lock,
651                                  struct ptlrpc_request *req, int rc,
652                                  const char *ast_type)
653 {
654         struct lnet_process_id peer = req->rq_import->imp_connection->c_peer;
655
656         if (!req->rq_replied || (rc && rc != -EINVAL)) {
657                 if (ldlm_is_cancel(lock)) {
658                         LDLM_DEBUG(lock,
659                                    "%s AST (req@%p x%llu) timeout from nid %s, but cancel was received (AST reply lost?)",
660                                    ast_type, req, req->rq_xid,
661                                    libcfs_nid2str(peer.nid));
662                         ldlm_lock_cancel(lock);
663                         rc = -ERESTART;
664                 } else if (rc == -ENODEV || rc == -ESHUTDOWN ||
665                            (rc == -EIO &&
666                             req->rq_import->imp_state == LUSTRE_IMP_CLOSED)) {
667                         /* Upon umount process the AST fails because cannot be
668                          * sent. This shouldn't lead to the client eviction.
669                          * -ENODEV error is returned by ptl_send_rpc() for
670                          *  new request in such import.
671                          * -SHUTDOWN is returned by ptlrpc_import_delay_req()
672                          *  if imp_invalid is set or obd_no_recov.
673                          * Meanwhile there is also check for LUSTRE_IMP_CLOSED
674                          * in ptlrpc_import_delay_req() as well with -EIO code.
675                          * In all such cases errors are ignored.
676                          */
677                         LDLM_DEBUG(lock, "%s AST can't be sent due to a server"
678                                          " %s failure or umount process: rc = %d\n",
679                                          ast_type,
680                                          req->rq_import->imp_obd->obd_name, rc);
681                 } else {
682                         LDLM_ERROR(lock,
683                                    "client (nid %s) %s %s AST (req@%p x%llu status %d rc %d), evict it",
684                                    libcfs_nid2str(peer.nid),
685                                    req->rq_replied ? "returned error from" :
686                                    "failed to reply to",
687                                    ast_type, req, req->rq_xid,
688                                    (req->rq_repmsg != NULL) ?
689                                    lustre_msg_get_status(req->rq_repmsg) : 0,
690                                    rc);
691                         ldlm_failed_ast(lock, rc, ast_type);
692                 }
693                 return rc;
694         }
695
696         if (rc == -EINVAL) {
697                 struct ldlm_resource *res = lock->l_resource;
698
699                 LDLM_DEBUG(lock,
700                            "client (nid %s) returned %d from %s AST (req@%p x%llu) - normal race",
701                            libcfs_nid2str(peer.nid),
702                            req->rq_repmsg ?
703                            lustre_msg_get_status(req->rq_repmsg) : -1,
704                            ast_type, req, req->rq_xid);
705                 if (res) {
706                         /* update lvbo to return proper attributes.
707                          * see bug 23174 */
708                         ldlm_resource_getref(res);
709                         ldlm_res_lvbo_update(res, NULL, 1);
710                         ldlm_resource_putref(res);
711                 }
712                 ldlm_lock_cancel(lock);
713                 rc = -ERESTART;
714         }
715
716         return rc;
717 }
718
719 static int ldlm_cb_interpret(const struct lu_env *env,
720                              struct ptlrpc_request *req, void *data, int rc)
721 {
722         struct ldlm_cb_async_args *ca   = data;
723         struct ldlm_lock          *lock = ca->ca_lock;
724         struct ldlm_cb_set_arg    *arg  = ca->ca_set_arg;
725         ENTRY;
726
727         LASSERT(lock != NULL);
728
729         switch (arg->type) {
730         case LDLM_GL_CALLBACK:
731                 /* Update the LVB from disk if the AST failed
732                  * (this is a legal race)
733                  *
734                  * - Glimpse callback of local lock just returns
735                  *   -ELDLM_NO_LOCK_DATA.
736                  * - Glimpse callback of remote lock might return
737                  *   -ELDLM_NO_LOCK_DATA when inode is cleared. LU-274
738                  */
739                 if (unlikely(arg->gl_interpret_reply)) {
740                         rc = arg->gl_interpret_reply(env, req, data, rc);
741                 } else if (rc == -ELDLM_NO_LOCK_DATA) {
742                         LDLM_DEBUG(lock, "lost race - client has a lock but no "
743                                    "inode");
744                         ldlm_res_lvbo_update(lock->l_resource, NULL, 1);
745                 } else if (rc != 0) {
746                         rc = ldlm_handle_ast_error(lock, req, rc, "glimpse");
747                 } else {
748                         rc = ldlm_res_lvbo_update(lock->l_resource, req, 1);
749                 }
750                 break;
751         case LDLM_BL_CALLBACK:
752                 if (rc != 0)
753                         rc = ldlm_handle_ast_error(lock, req, rc, "blocking");
754                 break;
755         case LDLM_CP_CALLBACK:
756                 if (rc != 0)
757                         rc = ldlm_handle_ast_error(lock, req, rc, "completion");
758                 break;
759         default:
760                 LDLM_ERROR(lock, "invalid opcode for lock callback %d",
761                            arg->type);
762                 LBUG();
763         }
764
765         /* release extra reference taken in ldlm_ast_fini() */
766         LDLM_LOCK_RELEASE(lock);
767
768         if (rc == -ERESTART)
769                 atomic_inc(&arg->restart);
770
771         RETURN(0);
772 }
773
774 static void ldlm_update_resend(struct ptlrpc_request *req, void *data)
775 {
776         struct ldlm_cb_async_args *ca   = data;
777         struct ldlm_lock          *lock = ca->ca_lock;
778
779         ldlm_refresh_waiting_lock(lock, ldlm_bl_timeout(lock));
780 }
781
782 static inline int ldlm_ast_fini(struct ptlrpc_request *req,
783                                 struct ldlm_cb_set_arg *arg,
784                                 struct ldlm_lock *lock,
785                                 int instant_cancel)
786 {
787         int rc = 0;
788         ENTRY;
789
790         if (unlikely(instant_cancel)) {
791                 rc = ptl_send_rpc(req, 1);
792                 ptlrpc_req_finished(req);
793                 if (rc == 0)
794                         atomic_inc(&arg->restart);
795         } else {
796                 LDLM_LOCK_GET(lock);
797                 ptlrpc_set_add_req(arg->set, req);
798         }
799
800         RETURN(rc);
801 }
802
803 /**
804  * Check if there are requests in the export request list which prevent
805  * the lock canceling and make these requests high priority ones.
806  */
807 static void ldlm_lock_reorder_req(struct ldlm_lock *lock)
808 {
809         struct ptlrpc_request *req;
810         ENTRY;
811
812         if (lock->l_export == NULL) {
813                 LDLM_DEBUG(lock, "client lock: no-op");
814                 RETURN_EXIT;
815         }
816
817         spin_lock_bh(&lock->l_export->exp_rpc_lock);
818         list_for_each_entry(req, &lock->l_export->exp_hp_rpcs,
819                             rq_exp_list) {
820                 /* Do not process requests that were not yet added to there
821                  * incoming queue or were already removed from there for
822                  * processing. We evaluate ptlrpc_nrs_req_can_move() without
823                  * holding svcpt->scp_req_lock, and then redo the check with
824                  * the lock held once we need to obtain a reliable result.
825                  */
826                 if (ptlrpc_nrs_req_can_move(req) &&
827                     req->rq_ops->hpreq_lock_match &&
828                     req->rq_ops->hpreq_lock_match(req, lock))
829                         ptlrpc_nrs_req_hp_move(req);
830         }
831         spin_unlock_bh(&lock->l_export->exp_rpc_lock);
832         EXIT;
833 }
834
835 /**
836  * ->l_blocking_ast() method for server-side locks. This is invoked when newly
837  * enqueued server lock conflicts with given one.
838  *
839  * Sends blocking AST RPC to the client owning that lock; arms timeout timer
840  * to wait for client response.
841  */
842 int ldlm_server_blocking_ast(struct ldlm_lock *lock,
843                              struct ldlm_lock_desc *desc,
844                              void *data, int flag)
845 {
846         struct ldlm_cb_async_args *ca;
847         struct ldlm_cb_set_arg *arg = data;
848         struct ldlm_request    *body;
849         struct ptlrpc_request  *req;
850         int                     instant_cancel = 0;
851         int                     rc = 0;
852         ENTRY;
853
854         if (flag == LDLM_CB_CANCELING)
855                 /* Don't need to do anything here. */
856                 RETURN(0);
857
858         if (OBD_FAIL_PRECHECK(OBD_FAIL_LDLM_SRV_BL_AST)) {
859                 LDLM_DEBUG(lock, "dropping BL AST");
860                 RETURN(0);
861         }
862
863         LASSERT(lock);
864         LASSERT(data != NULL);
865         if (lock->l_export->exp_obd->obd_recovering != 0)
866                 LDLM_ERROR(lock, "BUG 6063: lock collide during recovery");
867
868         ldlm_lock_reorder_req(lock);
869
870         req = ptlrpc_request_alloc_pack(lock->l_export->exp_imp_reverse,
871                                         &RQF_LDLM_BL_CALLBACK,
872                                         LUSTRE_DLM_VERSION, LDLM_BL_CALLBACK);
873         if (req == NULL)
874                 RETURN(-ENOMEM);
875
876         CLASSERT(sizeof(*ca) <= sizeof(req->rq_async_args));
877         ca = ptlrpc_req_async_args(req);
878         ca->ca_set_arg = arg;
879         ca->ca_lock = lock;
880
881         req->rq_interpret_reply = ldlm_cb_interpret;
882
883         lock_res_and_lock(lock);
884         if (ldlm_is_destroyed(lock)) {
885                 /* What's the point? */
886                 unlock_res_and_lock(lock);
887                 ptlrpc_req_finished(req);
888                 RETURN(0);
889         }
890
891         if (lock->l_granted_mode != lock->l_req_mode) {
892                 /* this blocking AST will be communicated as part of the
893                  * completion AST instead */
894                 ldlm_add_blocked_lock(lock);
895                 ldlm_set_waited(lock);
896                 unlock_res_and_lock(lock);
897
898                 ptlrpc_req_finished(req);
899                 LDLM_DEBUG(lock, "lock not granted, not sending blocking AST");
900                 RETURN(0);
901         }
902
903         if (ldlm_is_cancel_on_block(lock))
904                 instant_cancel = 1;
905
906         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
907         body->lock_handle[0] = lock->l_remote_handle;
908         body->lock_desc = *desc;
909         body->lock_flags |= ldlm_flags_to_wire(lock->l_flags & LDLM_FL_AST_MASK);
910
911         LDLM_DEBUG(lock, "server preparing blocking AST");
912
913         ptlrpc_request_set_replen(req);
914         ldlm_set_cbpending(lock);
915         if (instant_cancel) {
916                 unlock_res_and_lock(lock);
917                 ldlm_lock_cancel(lock);
918
919                 req->rq_no_resend = 1;
920         } else {
921                 LASSERT(lock->l_granted_mode == lock->l_req_mode);
922                 ldlm_add_waiting_lock(lock);
923                 unlock_res_and_lock(lock);
924
925                 /* Do not resend after lock callback timeout */
926                 req->rq_delay_limit = ldlm_bl_timeout(lock);
927                 req->rq_resend_cb = ldlm_update_resend;
928         }
929
930         req->rq_send_state = LUSTRE_IMP_FULL;
931         /* ptlrpc_request_alloc_pack already set timeout */
932         if (AT_OFF)
933                 req->rq_timeout = ldlm_get_rq_timeout();
934
935         lock->l_last_activity = ktime_get_real_seconds();
936
937         if (lock->l_export && lock->l_export->exp_nid_stats &&
938             lock->l_export->exp_nid_stats->nid_ldlm_stats)
939                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
940                                      LDLM_BL_CALLBACK - LDLM_FIRST_OPC);
941
942         rc = ldlm_ast_fini(req, arg, lock, instant_cancel);
943
944         RETURN(rc);
945 }
946
947 /**
948  * ->l_completion_ast callback for a remote lock in server namespace.
949  *
950  *  Sends AST to the client notifying it of lock granting.  If initial
951  *  lock response was not sent yet, instead of sending another RPC, just
952  *  mark the lock as granted and client will understand
953  */
954 int ldlm_server_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data)
955 {
956         struct ldlm_cb_set_arg *arg = data;
957         struct ldlm_request    *body;
958         struct ptlrpc_request  *req;
959         struct ldlm_cb_async_args *ca;
960         int                     instant_cancel = 0;
961         int                     rc = 0;
962         int                     lvb_len;
963         ENTRY;
964
965         LASSERT(lock != NULL);
966         LASSERT(data != NULL);
967
968         if (OBD_FAIL_PRECHECK(OBD_FAIL_LDLM_SRV_CP_AST)) {
969                 LDLM_DEBUG(lock, "dropping CP AST");
970                 RETURN(0);
971         }
972
973         req = ptlrpc_request_alloc(lock->l_export->exp_imp_reverse,
974                                     &RQF_LDLM_CP_CALLBACK);
975         if (req == NULL)
976                 RETURN(-ENOMEM);
977
978         /* server namespace, doesn't need lock */
979         lvb_len = ldlm_lvbo_size(lock);
980         /* LU-3124 & LU-2187: to not return layout in completion AST because
981          * it may deadlock for LU-2187, or client may not have enough space
982          * for large layout. The layout will be returned to client with an
983          * extra RPC to fetch xattr.lov */
984         if (ldlm_has_layout(lock))
985                 lvb_len = 0;
986
987         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_CLIENT, lvb_len);
988         rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_CP_CALLBACK);
989         if (rc) {
990                 ptlrpc_request_free(req);
991                 RETURN(rc);
992         }
993
994         CLASSERT(sizeof(*ca) <= sizeof(req->rq_async_args));
995         ca = ptlrpc_req_async_args(req);
996         ca->ca_set_arg = arg;
997         ca->ca_lock = lock;
998
999         req->rq_interpret_reply = ldlm_cb_interpret;
1000         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1001
1002         body->lock_handle[0] = lock->l_remote_handle;
1003         body->lock_flags = ldlm_flags_to_wire(flags);
1004         ldlm_lock2desc(lock, &body->lock_desc);
1005         if (lvb_len > 0) {
1006                 void *lvb = req_capsule_client_get(&req->rq_pill, &RMF_DLM_LVB);
1007
1008                 lvb_len = ldlm_lvbo_fill(lock, lvb, lvb_len);
1009                 if (lvb_len < 0) {
1010                         /* We still need to send the RPC to wake up the blocked
1011                          * enqueue thread on the client.
1012                          *
1013                          * Consider old client, there is no better way to notify
1014                          * the failure, just zero-sized the LVB, then the client
1015                          * will fail out as "-EPROTO". */
1016                         req_capsule_shrink(&req->rq_pill, &RMF_DLM_LVB, 0,
1017                                            RCL_CLIENT);
1018                         instant_cancel = 1;
1019                 } else {
1020                         req_capsule_shrink(&req->rq_pill, &RMF_DLM_LVB, lvb_len,
1021                                            RCL_CLIENT);
1022                 }
1023         }
1024
1025         lock->l_last_activity = ktime_get_real_seconds();
1026
1027         LDLM_DEBUG(lock, "server preparing completion AST");
1028
1029         ptlrpc_request_set_replen(req);
1030
1031         req->rq_send_state = LUSTRE_IMP_FULL;
1032         /* ptlrpc_request_pack already set timeout */
1033         if (AT_OFF)
1034                 req->rq_timeout = ldlm_get_rq_timeout();
1035
1036         /* We only send real blocking ASTs after the lock is granted */
1037         lock_res_and_lock(lock);
1038         if (ldlm_is_ast_sent(lock)) {
1039                 body->lock_flags |= ldlm_flags_to_wire(LDLM_FL_AST_SENT);
1040                 /* Copy AST flags like LDLM_FL_DISCARD_DATA. */
1041                 body->lock_flags |= ldlm_flags_to_wire(lock->l_flags &
1042                                                        LDLM_FL_AST_MASK);
1043
1044                 /* We might get here prior to ldlm_handle_enqueue setting
1045                  * LDLM_FL_CANCEL_ON_BLOCK flag. Then we will put this lock
1046                  * into waiting list, but this is safe and similar code in
1047                  * ldlm_handle_enqueue will call ldlm_lock_cancel() still,
1048                  * that would not only cancel the lock, but will also remove
1049                  * it from waiting list */
1050                 if (ldlm_is_cancel_on_block(lock)) {
1051                         unlock_res_and_lock(lock);
1052                         ldlm_lock_cancel(lock);
1053
1054                         instant_cancel = 1;
1055                         req->rq_no_resend = 1;
1056
1057                         lock_res_and_lock(lock);
1058                 } else {
1059                         /* start the lock-timeout clock */
1060                         ldlm_add_waiting_lock(lock);
1061                         /* Do not resend after lock callback timeout */
1062                         req->rq_delay_limit = ldlm_bl_timeout(lock);
1063                         req->rq_resend_cb = ldlm_update_resend;
1064                 }
1065         }
1066         unlock_res_and_lock(lock);
1067
1068         if (lock->l_export && lock->l_export->exp_nid_stats &&
1069             lock->l_export->exp_nid_stats->nid_ldlm_stats)
1070                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
1071                                      LDLM_CP_CALLBACK - LDLM_FIRST_OPC);
1072
1073         rc = ldlm_ast_fini(req, arg, lock, instant_cancel);
1074
1075         RETURN(lvb_len < 0 ? lvb_len : rc);
1076 }
1077
1078 /**
1079  * Server side ->l_glimpse_ast handler for client locks.
1080  *
1081  * Sends glimpse AST to the client and waits for reply. Then updates
1082  * lvbo with the result.
1083  */
1084 int ldlm_server_glimpse_ast(struct ldlm_lock *lock, void *data)
1085 {
1086         struct ldlm_cb_set_arg          *arg = data;
1087         struct ldlm_request             *body;
1088         struct ptlrpc_request           *req;
1089         struct ldlm_cb_async_args       *ca;
1090         int                              rc;
1091         struct req_format               *req_fmt;
1092         ENTRY;
1093
1094         LASSERT(lock != NULL);
1095
1096         if (arg->gl_desc != NULL)
1097                 /* There is a glimpse descriptor to pack */
1098                 req_fmt = &RQF_LDLM_GL_DESC_CALLBACK;
1099         else
1100                 req_fmt = &RQF_LDLM_GL_CALLBACK;
1101
1102         req = ptlrpc_request_alloc_pack(lock->l_export->exp_imp_reverse,
1103                                         req_fmt, LUSTRE_DLM_VERSION,
1104                                         LDLM_GL_CALLBACK);
1105
1106         if (req == NULL)
1107                 RETURN(-ENOMEM);
1108
1109         if (arg->gl_desc != NULL) {
1110                 /* copy the GL descriptor */
1111                 union ldlm_gl_desc      *desc;
1112                 desc = req_capsule_client_get(&req->rq_pill, &RMF_DLM_GL_DESC);
1113                 *desc = *arg->gl_desc;
1114         }
1115
1116         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1117         body->lock_handle[0] = lock->l_remote_handle;
1118         ldlm_lock2desc(lock, &body->lock_desc);
1119
1120         CLASSERT(sizeof(*ca) <= sizeof(req->rq_async_args));
1121         ca = ptlrpc_req_async_args(req);
1122         ca->ca_set_arg = arg;
1123         ca->ca_lock = lock;
1124
1125         /* server namespace, doesn't need lock */
1126         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
1127                              ldlm_lvbo_size(lock));
1128         ptlrpc_request_set_replen(req);
1129
1130         req->rq_send_state = LUSTRE_IMP_FULL;
1131         /* ptlrpc_request_alloc_pack already set timeout */
1132         if (AT_OFF)
1133                 req->rq_timeout = ldlm_get_rq_timeout();
1134
1135         lock->l_last_activity = ktime_get_real_seconds();
1136
1137         req->rq_interpret_reply = ldlm_cb_interpret;
1138
1139         if (lock->l_export && lock->l_export->exp_nid_stats &&
1140             lock->l_export->exp_nid_stats->nid_ldlm_stats)
1141                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
1142                                      LDLM_GL_CALLBACK - LDLM_FIRST_OPC);
1143
1144         rc = ldlm_ast_fini(req, arg, lock, 0);
1145
1146         RETURN(rc);
1147 }
1148
1149 int ldlm_glimpse_locks(struct ldlm_resource *res,
1150                        struct list_head *gl_work_list)
1151 {
1152         int     rc;
1153         ENTRY;
1154
1155         rc = ldlm_run_ast_work(ldlm_res_to_ns(res), gl_work_list,
1156                                LDLM_WORK_GL_AST);
1157         if (rc == -ERESTART)
1158                 ldlm_reprocess_all(res);
1159
1160         RETURN(rc);
1161 }
1162 EXPORT_SYMBOL(ldlm_glimpse_locks);
1163
1164 /* return LDLM lock associated with a lock callback request */
1165 struct ldlm_lock *ldlm_request_lock(struct ptlrpc_request *req)
1166 {
1167         struct ldlm_cb_async_args       *ca;
1168         struct ldlm_lock                *lock;
1169         ENTRY;
1170
1171         ca = ptlrpc_req_async_args(req);
1172         lock = ca->ca_lock;
1173         if (lock == NULL)
1174                 RETURN(ERR_PTR(-EFAULT));
1175
1176         RETURN(lock);
1177 }
1178 EXPORT_SYMBOL(ldlm_request_lock);
1179
1180 static void ldlm_svc_get_eopc(const struct ldlm_request *dlm_req,
1181                        struct lprocfs_stats *srv_stats)
1182 {
1183         int lock_type = 0, op = 0;
1184
1185         lock_type = dlm_req->lock_desc.l_resource.lr_type;
1186
1187         switch (lock_type) {
1188         case LDLM_PLAIN:
1189                 op = PTLRPC_LAST_CNTR + LDLM_PLAIN_ENQUEUE;
1190                 break;
1191         case LDLM_EXTENT:
1192                 if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT)
1193                         op = PTLRPC_LAST_CNTR + LDLM_GLIMPSE_ENQUEUE;
1194                 else
1195                         op = PTLRPC_LAST_CNTR + LDLM_EXTENT_ENQUEUE;
1196                 break;
1197         case LDLM_FLOCK:
1198                 op = PTLRPC_LAST_CNTR + LDLM_FLOCK_ENQUEUE;
1199                 break;
1200         case LDLM_IBITS:
1201                 op = PTLRPC_LAST_CNTR + LDLM_IBITS_ENQUEUE;
1202                 break;
1203         default:
1204                 op = 0;
1205                 break;
1206         }
1207
1208         if (op)
1209                 lprocfs_counter_incr(srv_stats, op);
1210
1211         return;
1212 }
1213
1214 /**
1215  * Main server-side entry point into LDLM for enqueue. This is called by ptlrpc
1216  * service threads to carry out client lock enqueueing requests.
1217  */
1218 int ldlm_handle_enqueue0(struct ldlm_namespace *ns,
1219                          struct ptlrpc_request *req,
1220                          const struct ldlm_request *dlm_req,
1221                          const struct ldlm_callback_suite *cbs)
1222 {
1223         struct ldlm_reply *dlm_rep;
1224         __u64 flags;
1225         enum ldlm_error err = ELDLM_OK;
1226         struct ldlm_lock *lock = NULL;
1227         void *cookie = NULL;
1228         int rc = 0;
1229         struct ldlm_resource *res = NULL;
1230         ENTRY;
1231
1232         LDLM_DEBUG_NOLOCK("server-side enqueue handler START");
1233
1234         ldlm_request_cancel(req, dlm_req, LDLM_ENQUEUE_CANCEL_OFF, LATF_SKIP);
1235         flags = ldlm_flags_from_wire(dlm_req->lock_flags);
1236
1237         LASSERT(req->rq_export);
1238
1239         if (ptlrpc_req2svc(req)->srv_stats != NULL)
1240                 ldlm_svc_get_eopc(dlm_req, ptlrpc_req2svc(req)->srv_stats);
1241
1242         if (req->rq_export && req->rq_export->exp_nid_stats &&
1243             req->rq_export->exp_nid_stats->nid_ldlm_stats)
1244                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1245                                      LDLM_ENQUEUE - LDLM_FIRST_OPC);
1246
1247         if (unlikely(dlm_req->lock_desc.l_resource.lr_type < LDLM_MIN_TYPE ||
1248                      dlm_req->lock_desc.l_resource.lr_type >= LDLM_MAX_TYPE)) {
1249                 DEBUG_REQ(D_ERROR, req, "invalid lock request type %d",
1250                           dlm_req->lock_desc.l_resource.lr_type);
1251                 GOTO(out, rc = -EFAULT);
1252         }
1253
1254         if (unlikely(dlm_req->lock_desc.l_req_mode <= LCK_MINMODE ||
1255                      dlm_req->lock_desc.l_req_mode >= LCK_MAXMODE ||
1256                      dlm_req->lock_desc.l_req_mode &
1257                      (dlm_req->lock_desc.l_req_mode-1))) {
1258                 DEBUG_REQ(D_ERROR, req, "invalid lock request mode %d",
1259                           dlm_req->lock_desc.l_req_mode);
1260                 GOTO(out, rc = -EFAULT);
1261         }
1262
1263         if (exp_connect_flags(req->rq_export) & OBD_CONNECT_IBITS) {
1264                 if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
1265                              LDLM_PLAIN)) {
1266                         DEBUG_REQ(D_ERROR, req,
1267                                   "PLAIN lock request from IBITS client?");
1268                         GOTO(out, rc = -EPROTO);
1269                 }
1270         } else if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
1271                             LDLM_IBITS)) {
1272                 DEBUG_REQ(D_ERROR, req,
1273                           "IBITS lock request from unaware client?");
1274                 GOTO(out, rc = -EPROTO);
1275         }
1276
1277         if (unlikely((flags & LDLM_FL_REPLAY) ||
1278                      (lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT))) {
1279                 /* Find an existing lock in the per-export lock hash */
1280                 /* In the function below, .hs_keycmp resolves to
1281                  * ldlm_export_lock_keycmp() */
1282                 /* coverity[overrun-buffer-val] */
1283                 lock = cfs_hash_lookup(req->rq_export->exp_lock_hash,
1284                                        (void *)&dlm_req->lock_handle[0]);
1285                 if (lock != NULL) {
1286                         DEBUG_REQ(D_DLMTRACE, req, "found existing lock cookie %#llx",
1287                                   lock->l_handle.h_cookie);
1288                         flags |= LDLM_FL_RESENT;
1289                         GOTO(existing_lock, rc = 0);
1290                 }
1291         } else {
1292                 if (ldlm_reclaim_full()) {
1293                         DEBUG_REQ(D_DLMTRACE, req, "Too many granted locks, "
1294                                   "reject current enqueue request and let the "
1295                                   "client retry later.\n");
1296                         GOTO(out, rc = -EINPROGRESS);
1297                 }
1298         }
1299
1300         /* The lock's callback data might be set in the policy function */
1301         lock = ldlm_lock_create(ns, &dlm_req->lock_desc.l_resource.lr_name,
1302                                 dlm_req->lock_desc.l_resource.lr_type,
1303                                 dlm_req->lock_desc.l_req_mode,
1304                                 cbs, NULL, 0, LVB_T_NONE);
1305         if (IS_ERR(lock)) {
1306                 rc = PTR_ERR(lock);
1307                 lock = NULL;
1308                 GOTO(out, rc);
1309         }
1310
1311         lock->l_remote_handle = dlm_req->lock_handle[0];
1312         LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
1313
1314         /* Initialize resource lvb but not for a lock being replayed since
1315          * Client already got lvb sent in this case.
1316          * This must occur early since some policy methods assume resource
1317          * lvb is available (lr_lvb_data != NULL).
1318          */
1319         res = lock->l_resource;
1320         if (!(flags & LDLM_FL_REPLAY)) {
1321                 /* non-replayed lock, delayed lvb init may need to be done */
1322                 rc = ldlm_lvbo_init(res);
1323                 if (rc < 0) {
1324                         LDLM_DEBUG(lock, "delayed lvb init failed (rc %d)", rc);
1325                         GOTO(out, rc);
1326                 }
1327         }
1328
1329         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_BLOCKED, obd_timeout * 2);
1330         /* Don't enqueue a lock onto the export if it is been disonnected
1331          * due to eviction (bug 3822) or server umount (bug 24324).
1332          * Cancel it now instead. */
1333         if (req->rq_export->exp_disconnected) {
1334                 LDLM_ERROR(lock, "lock on disconnected export %p",
1335                            req->rq_export);
1336                 GOTO(out, rc = -ENOTCONN);
1337         }
1338
1339         lock->l_export = class_export_lock_get(req->rq_export, lock);
1340         if (lock->l_export->exp_lock_hash)
1341                 cfs_hash_add(lock->l_export->exp_lock_hash,
1342                              &lock->l_remote_handle,
1343                              &lock->l_exp_hash);
1344
1345         /* Inherit the enqueue flags before the operation, because we do not
1346          * keep the res lock on return and next operations (BL AST) may proceed
1347          * without them. */
1348         lock->l_flags |= ldlm_flags_from_wire(dlm_req->lock_flags &
1349                                               LDLM_FL_INHERIT_MASK);
1350
1351         ldlm_convert_policy_to_local(req->rq_export,
1352                                      dlm_req->lock_desc.l_resource.lr_type,
1353                                      &dlm_req->lock_desc.l_policy_data,
1354                                      &lock->l_policy_data);
1355         if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT)
1356                 lock->l_req_extent = lock->l_policy_data.l_extent;
1357
1358 existing_lock:
1359
1360         if (flags & LDLM_FL_HAS_INTENT) {
1361                 /* In this case, the reply buffer is allocated deep in
1362                  * local_lock_enqueue by the policy function. */
1363                 cookie = req;
1364         } else {
1365                 /* based on the assumption that lvb size never changes during
1366                  * resource life time otherwise it need resource->lr_lock's
1367                  * protection */
1368                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
1369                                      RCL_SERVER, ldlm_lvbo_size(lock));
1370
1371                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
1372                         GOTO(out, rc = -ENOMEM);
1373
1374                 rc = req_capsule_server_pack(&req->rq_pill);
1375                 if (rc)
1376                         GOTO(out, rc);
1377         }
1378
1379         err = ldlm_lock_enqueue(ns, &lock, cookie, &flags);
1380         if (err) {
1381                 if ((int)err < 0)
1382                         rc = (int)err;
1383                 GOTO(out, err);
1384         }
1385
1386         dlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1387
1388         ldlm_lock2desc(lock, &dlm_rep->lock_desc);
1389         ldlm_lock2handle(lock, &dlm_rep->lock_handle);
1390
1391         if (lock && lock->l_resource->lr_type == LDLM_EXTENT)
1392                 OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_BL_EVICT, 6);
1393
1394         /* We never send a blocking AST until the lock is granted, but
1395          * we can tell it right now */
1396         lock_res_and_lock(lock);
1397
1398         /* Now take into account flags to be inherited from original lock
1399            request both in reply to client and in our own lock flags. */
1400         dlm_rep->lock_flags = ldlm_flags_to_wire(flags);
1401         lock->l_flags |= flags & LDLM_FL_INHERIT_MASK;
1402
1403         /* Don't move a pending lock onto the export if it has already been
1404          * disconnected due to eviction (bug 5683) or server umount (bug 24324).
1405          * Cancel it now instead. */
1406         if (unlikely(req->rq_export->exp_disconnected ||
1407                      OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_OLD_EXPORT))) {
1408                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
1409                 rc = -ENOTCONN;
1410         } else if (ldlm_is_ast_sent(lock)) {
1411                 dlm_rep->lock_flags |= ldlm_flags_to_wire(LDLM_FL_AST_SENT);
1412                 if (lock->l_granted_mode == lock->l_req_mode) {
1413                         /*
1414                          * Only cancel lock if it was granted, because it would
1415                          * be destroyed immediately and would never be granted
1416                          * in the future, causing timeouts on client.  Not
1417                          * granted lock will be cancelled immediately after
1418                          * sending completion AST.
1419                          */
1420                         if (dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK) {
1421                                 unlock_res_and_lock(lock);
1422                                 ldlm_lock_cancel(lock);
1423                                 lock_res_and_lock(lock);
1424                         } else
1425                                 ldlm_add_waiting_lock(lock);
1426                 }
1427         }
1428         unlock_res_and_lock(lock);
1429
1430         EXIT;
1431  out:
1432         req->rq_status = rc ?: err; /* return either error - bug 11190 */
1433         if (!req->rq_packed_final) {
1434                 err = lustre_pack_reply(req, 1, NULL, NULL);
1435                 if (rc == 0)
1436                         rc = err;
1437         }
1438
1439         /* The LOCK_CHANGED code in ldlm_lock_enqueue depends on this
1440          * ldlm_reprocess_all.  If this moves, revisit that code. -phil */
1441         if (lock != NULL) {
1442                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
1443                            "(err=%d, rc=%d)", err, rc);
1444
1445                 if (rc == 0) {
1446                         if (req_capsule_has_field(&req->rq_pill, &RMF_DLM_LVB,
1447                                                   RCL_SERVER) &&
1448                             ldlm_lvbo_size(lock) > 0) {
1449                                 void *buf;
1450                                 int buflen;
1451
1452                                 buf = req_capsule_server_get(&req->rq_pill,
1453                                                              &RMF_DLM_LVB);
1454                                 LASSERTF(buf != NULL, "req %p, lock %p\n",
1455                                          req, lock);
1456                                 buflen = req_capsule_get_size(&req->rq_pill,
1457                                                 &RMF_DLM_LVB, RCL_SERVER);
1458                                 /* non-replayed lock, delayed lvb init may
1459                                  * need to be occur now */
1460                                 if ((buflen > 0) && !(flags & LDLM_FL_REPLAY)) {
1461                                         buflen = ldlm_lvbo_fill(lock, buf,
1462                                                                 buflen);
1463                                         if (buflen >= 0)
1464                                                 req_capsule_shrink(
1465                                                         &req->rq_pill,
1466                                                         &RMF_DLM_LVB,
1467                                                         buflen, RCL_SERVER);
1468                                         else
1469                                                 rc = buflen;
1470                                 } else if (flags & LDLM_FL_REPLAY) {
1471                                         /* no LVB resend upon replay */
1472                                         if (buflen > 0)
1473                                                 req_capsule_shrink(
1474                                                         &req->rq_pill,
1475                                                         &RMF_DLM_LVB,
1476                                                         0, RCL_SERVER);
1477                                         else
1478                                                 rc = buflen;
1479                                 } else {
1480                                         rc = buflen;
1481                                 }
1482                         }
1483                 }
1484
1485                 if (rc != 0 && !(flags & LDLM_FL_RESENT)) {
1486                         if (lock->l_export) {
1487                                 ldlm_lock_cancel(lock);
1488                         } else {
1489                                 lock_res_and_lock(lock);
1490                                 ldlm_resource_unlink_lock(lock);
1491                                 ldlm_lock_destroy_nolock(lock);
1492                                 unlock_res_and_lock(lock);
1493
1494                         }
1495                 }
1496
1497                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
1498                         ldlm_reprocess_all(lock->l_resource);
1499
1500                 LDLM_LOCK_RELEASE(lock);
1501         }
1502
1503         LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p, rc %d)",
1504                           lock, rc);
1505
1506         return rc;
1507 }
1508
1509 /**
1510  * Old-style LDLM main entry point for server code enqueue.
1511  */
1512 int ldlm_handle_enqueue(struct ptlrpc_request *req,
1513                         ldlm_completion_callback completion_callback,
1514                         ldlm_blocking_callback blocking_callback,
1515                         ldlm_glimpse_callback glimpse_callback)
1516 {
1517         struct ldlm_request *dlm_req;
1518         struct ldlm_callback_suite cbs = {
1519                 .lcs_completion = completion_callback,
1520                 .lcs_blocking   = blocking_callback,
1521                 .lcs_glimpse    = glimpse_callback
1522         };
1523         int rc;
1524
1525         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1526         if (dlm_req != NULL) {
1527                 rc = ldlm_handle_enqueue0(req->rq_export->exp_obd->obd_namespace,
1528                                           req, dlm_req, &cbs);
1529         } else {
1530                 rc = -EFAULT;
1531         }
1532         return rc;
1533 }
1534
1535 /**
1536  * Main LDLM entry point for server code to process lock conversion requests.
1537  */
1538 int ldlm_handle_convert0(struct ptlrpc_request *req,
1539                          const struct ldlm_request *dlm_req)
1540 {
1541         struct ldlm_reply *dlm_rep;
1542         struct ldlm_lock *lock;
1543         int rc;
1544         ENTRY;
1545
1546         if (req->rq_export && req->rq_export->exp_nid_stats &&
1547             req->rq_export->exp_nid_stats->nid_ldlm_stats)
1548                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1549                                      LDLM_CONVERT - LDLM_FIRST_OPC);
1550
1551         rc = req_capsule_server_pack(&req->rq_pill);
1552         if (rc)
1553                 RETURN(rc);
1554
1555         dlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1556         dlm_rep->lock_flags = dlm_req->lock_flags;
1557
1558         lock = ldlm_handle2lock(&dlm_req->lock_handle[0]);
1559         if (!lock) {
1560                 req->rq_status = LUSTRE_EINVAL;
1561         } else {
1562                 void *res = NULL;
1563
1564                 LDLM_DEBUG(lock, "server-side convert handler START");
1565
1566                 res = ldlm_lock_convert(lock, dlm_req->lock_desc.l_req_mode,
1567                                         &dlm_rep->lock_flags);
1568                 if (res) {
1569                         if (ldlm_del_waiting_lock(lock))
1570                                 LDLM_DEBUG(lock, "converted waiting lock");
1571                         req->rq_status = 0;
1572                 } else {
1573                         req->rq_status = LUSTRE_EDEADLK;
1574                 }
1575         }
1576
1577         if (lock) {
1578                 if (!req->rq_status)
1579                         ldlm_reprocess_all(lock->l_resource);
1580                 LDLM_DEBUG(lock, "server-side convert handler END");
1581                 LDLM_LOCK_PUT(lock);
1582         } else
1583                 LDLM_DEBUG_NOLOCK("server-side convert handler END");
1584
1585         RETURN(0);
1586 }
1587
1588 /**
1589  * Old-style main LDLM entry point for server code to process lock conversion
1590  * requests.
1591  */
1592 int ldlm_handle_convert(struct ptlrpc_request *req)
1593 {
1594         int rc;
1595         struct ldlm_request *dlm_req;
1596
1597         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1598         if (dlm_req != NULL) {
1599                 rc = ldlm_handle_convert0(req, dlm_req);
1600         } else {
1601                 CERROR ("Can't unpack dlm_req\n");
1602                 rc = -EFAULT;
1603         }
1604         return rc;
1605 }
1606
1607 /**
1608  * Cancel all the locks whose handles are packed into ldlm_request
1609  *
1610  * Called by server code expecting such combined cancel activity
1611  * requests.
1612  */
1613 int ldlm_request_cancel(struct ptlrpc_request *req,
1614                         const struct ldlm_request *dlm_req,
1615                         int first, enum lustre_at_flags flags)
1616 {
1617         struct ldlm_resource *res, *pres = NULL;
1618         struct ldlm_lock *lock;
1619         int i, count, done = 0;
1620         ENTRY;
1621
1622         count = dlm_req->lock_count ? dlm_req->lock_count : 1;
1623         if (first >= count)
1624                 RETURN(0);
1625
1626         if (count == 1 && dlm_req->lock_handle[0].cookie == 0)
1627                 RETURN(0);
1628
1629         /* There is no lock on the server at the replay time,
1630          * skip lock cancelling to make replay tests to pass. */
1631         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
1632                 RETURN(0);
1633
1634         LDLM_DEBUG_NOLOCK("server-side cancel handler START: %d locks, "
1635                           "starting at %d", count, first);
1636
1637         for (i = first; i < count; i++) {
1638                 lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
1639                 if (!lock) {
1640                         LDLM_DEBUG_NOLOCK("server-side cancel handler stale "
1641                                           "lock (cookie %llu)",
1642                                           dlm_req->lock_handle[i].cookie);
1643                         continue;
1644                 }
1645
1646                 res = lock->l_resource;
1647                 done++;
1648
1649                 /* This code is an optimization to only attempt lock
1650                  * granting on the resource (that could be CPU-expensive)
1651                  * after we are done cancelling lock in that resource. */
1652                 if (res != pres) {
1653                         if (pres != NULL) {
1654                                 ldlm_reprocess_all(pres);
1655                                 LDLM_RESOURCE_DELREF(pres);
1656                                 ldlm_resource_putref(pres);
1657                         }
1658                         if (res != NULL) {
1659                                 ldlm_resource_getref(res);
1660                                 LDLM_RESOURCE_ADDREF(res);
1661                                 ldlm_res_lvbo_update(res, NULL, 1);
1662                         }
1663                         pres = res;
1664                 }
1665
1666                 if ((flags & LATF_STATS) && ldlm_is_ast_sent(lock)) {
1667                         time64_t delay = ktime_get_real_seconds() -
1668                                          lock->l_last_activity;
1669                         LDLM_DEBUG(lock, "server cancels blocked lock after %llds",
1670                                    (s64)delay);
1671                         at_measured(&lock->l_export->exp_bl_lock_at, delay);
1672                 }
1673                 ldlm_lock_cancel(lock);
1674                 LDLM_LOCK_PUT(lock);
1675         }
1676         if (pres != NULL) {
1677                 ldlm_reprocess_all(pres);
1678                 LDLM_RESOURCE_DELREF(pres);
1679                 ldlm_resource_putref(pres);
1680         }
1681         LDLM_DEBUG_NOLOCK("server-side cancel handler END");
1682         RETURN(done);
1683 }
1684 EXPORT_SYMBOL(ldlm_request_cancel);
1685
1686 /**
1687  * Main LDLM entry point for server code to cancel locks.
1688  *
1689  * Typically gets called from service handler on LDLM_CANCEL opc.
1690  */
1691 int ldlm_handle_cancel(struct ptlrpc_request *req)
1692 {
1693         struct ldlm_request *dlm_req;
1694         int rc;
1695         ENTRY;
1696
1697         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1698         if (dlm_req == NULL) {
1699                 CDEBUG(D_INFO, "bad request buffer for cancel\n");
1700                 RETURN(-EFAULT);
1701         }
1702
1703         if (req->rq_export && req->rq_export->exp_nid_stats &&
1704             req->rq_export->exp_nid_stats->nid_ldlm_stats)
1705                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1706                                      LDLM_CANCEL - LDLM_FIRST_OPC);
1707
1708         rc = req_capsule_server_pack(&req->rq_pill);
1709         if (rc)
1710                 RETURN(rc);
1711
1712         if (!ldlm_request_cancel(req, dlm_req, 0, LATF_STATS))
1713                 req->rq_status = LUSTRE_ESTALE;
1714
1715         RETURN(ptlrpc_reply(req));
1716 }
1717 #endif /* HAVE_SERVER_SUPPORT */
1718
1719 /**
1720  * Callback handler for receiving incoming blocking ASTs.
1721  *
1722  * This can only happen on client side.
1723  */
1724 void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
1725                              struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
1726 {
1727         int do_ast;
1728         ENTRY;
1729
1730         LDLM_DEBUG(lock, "client blocking AST callback handler");
1731
1732         lock_res_and_lock(lock);
1733         ldlm_set_cbpending(lock);
1734
1735         if (ldlm_is_cancel_on_block(lock))
1736                 ldlm_set_cancel(lock);
1737
1738         do_ast = (!lock->l_readers && !lock->l_writers);
1739         unlock_res_and_lock(lock);
1740
1741         if (do_ast) {
1742                 CDEBUG(D_DLMTRACE, "Lock %p already unused, calling callback (%p)\n",
1743                        lock, lock->l_blocking_ast);
1744                 if (lock->l_blocking_ast != NULL)
1745                         lock->l_blocking_ast(lock, ld, lock->l_ast_data,
1746                                              LDLM_CB_BLOCKING);
1747         } else {
1748                 CDEBUG(D_DLMTRACE, "Lock %p is referenced, will be cancelled later\n",
1749                        lock);
1750         }
1751
1752         LDLM_DEBUG(lock, "client blocking callback handler END");
1753         LDLM_LOCK_RELEASE(lock);
1754         EXIT;
1755 }
1756
1757 /**
1758  * Callback handler for receiving incoming completion ASTs.
1759  *
1760  * This only can happen on client side.
1761  */
1762 static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
1763                                     struct ldlm_namespace *ns,
1764                                     struct ldlm_request *dlm_req,
1765                                     struct ldlm_lock *lock)
1766 {
1767         struct list_head ast_list;
1768         int lvb_len;
1769         int rc = 0;
1770         ENTRY;
1771
1772         LDLM_DEBUG(lock, "client completion callback handler START");
1773
1774         INIT_LIST_HEAD(&ast_list);
1775         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE)) {
1776                 int to = cfs_time_seconds(1);
1777                 while (to > 0) {
1778                         set_current_state(TASK_INTERRUPTIBLE);
1779                         schedule_timeout(to);
1780                         if (lock->l_granted_mode == lock->l_req_mode ||
1781                             ldlm_is_destroyed(lock))
1782                                 break;
1783                 }
1784         }
1785
1786         lvb_len = req_capsule_get_size(&req->rq_pill, &RMF_DLM_LVB, RCL_CLIENT);
1787         if (lvb_len < 0) {
1788                 LDLM_ERROR(lock, "Fail to get lvb_len, rc = %d", lvb_len);
1789                 GOTO(out, rc = lvb_len);
1790         } else if (lvb_len > 0) {
1791                 if (lock->l_lvb_len > 0) {
1792                         /* for extent lock, lvb contains ost_lvb{}. */
1793                         LASSERT(lock->l_lvb_data != NULL);
1794
1795                         if (unlikely(lock->l_lvb_len < lvb_len)) {
1796                                 LDLM_ERROR(lock, "Replied LVB is larger than "
1797                                            "expectation, expected = %d, "
1798                                            "replied = %d",
1799                                            lock->l_lvb_len, lvb_len);
1800                                 GOTO(out, rc = -EINVAL);
1801                         }
1802                 }
1803         }
1804
1805         lock_res_and_lock(lock);
1806         if (ldlm_is_destroyed(lock) ||
1807             lock->l_granted_mode == lock->l_req_mode) {
1808                 /* bug 11300: the lock has already been granted */
1809                 unlock_res_and_lock(lock);
1810                 LDLM_DEBUG(lock, "Double grant race happened");
1811                 GOTO(out, rc = 0);
1812         }
1813
1814         /* If we receive the completion AST before the actual enqueue returned,
1815          * then we might need to switch lock modes, resources, or extents. */
1816         if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
1817                 lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
1818                 LDLM_DEBUG(lock, "completion AST, new lock mode");
1819         }
1820
1821         if (lock->l_resource->lr_type != LDLM_PLAIN) {
1822                 ldlm_convert_policy_to_local(req->rq_export,
1823                                           dlm_req->lock_desc.l_resource.lr_type,
1824                                           &dlm_req->lock_desc.l_policy_data,
1825                                           &lock->l_policy_data);
1826                 LDLM_DEBUG(lock, "completion AST, new policy data");
1827         }
1828
1829         ldlm_resource_unlink_lock(lock);
1830         if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
1831                    &lock->l_resource->lr_name,
1832                    sizeof(lock->l_resource->lr_name)) != 0) {
1833                 unlock_res_and_lock(lock);
1834                 rc = ldlm_lock_change_resource(ns, lock,
1835                                 &dlm_req->lock_desc.l_resource.lr_name);
1836                 if (rc < 0) {
1837                         LDLM_ERROR(lock, "Failed to allocate resource");
1838                         GOTO(out, rc);
1839                 }
1840                 LDLM_DEBUG(lock, "completion AST, new resource");
1841                 CERROR("change resource!\n");
1842                 lock_res_and_lock(lock);
1843         }
1844
1845         if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
1846                 /* BL_AST locks are not needed in LRU.
1847                  * Let ldlm_cancel_lru() be fast. */
1848                 ldlm_lock_remove_from_lru(lock);
1849                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_BL_AST;
1850                 LDLM_DEBUG(lock, "completion AST includes blocking AST");
1851         }
1852
1853         if (lock->l_lvb_len > 0) {
1854                 rc = ldlm_fill_lvb(lock, &req->rq_pill, RCL_CLIENT,
1855                                    lock->l_lvb_data, lvb_len);
1856                 if (rc < 0) {
1857                         unlock_res_and_lock(lock);
1858                         GOTO(out, rc);
1859                 }
1860         }
1861
1862         ldlm_grant_lock(lock, &ast_list);
1863         unlock_res_and_lock(lock);
1864
1865         LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
1866
1867         /* Let Enqueue to call osc_lock_upcall() and initialize
1868          * l_ast_data */
1869         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 2);
1870
1871         ldlm_run_ast_work(ns, &ast_list, LDLM_WORK_CP_AST);
1872
1873         LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
1874                           lock);
1875         GOTO(out, rc);
1876
1877 out:
1878         if (rc < 0) {
1879                 lock_res_and_lock(lock);
1880                 ldlm_set_failed(lock);
1881                 unlock_res_and_lock(lock);
1882                 wake_up(&lock->l_waitq);
1883         }
1884         LDLM_LOCK_RELEASE(lock);
1885 }
1886
1887 /**
1888  * Callback handler for receiving incoming glimpse ASTs.
1889  *
1890  * This only can happen on client side.  After handling the glimpse AST
1891  * we also consider dropping the lock here if it is unused locally for a
1892  * long time.
1893  */
1894 static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
1895                                     struct ldlm_namespace *ns,
1896                                     struct ldlm_request *dlm_req,
1897                                     struct ldlm_lock *lock)
1898 {
1899         int rc = -ENOSYS;
1900         ENTRY;
1901
1902         LDLM_DEBUG(lock, "client glimpse AST callback handler");
1903
1904         if (lock->l_glimpse_ast != NULL)
1905                 rc = lock->l_glimpse_ast(lock, req);
1906
1907         if (req->rq_repmsg != NULL) {
1908                 ptlrpc_reply(req);
1909         } else {
1910                 req->rq_status = rc;
1911                 ptlrpc_error(req);
1912         }
1913
1914         lock_res_and_lock(lock);
1915         if (lock->l_granted_mode == LCK_PW &&
1916             !lock->l_readers && !lock->l_writers &&
1917             ktime_after(ktime_get(),
1918                         ktime_add(lock->l_last_used,
1919                                   ktime_set(10, 0)))) {
1920                 unlock_res_and_lock(lock);
1921                 if (ldlm_bl_to_thread_lock(ns, NULL, lock))
1922                         ldlm_handle_bl_callback(ns, NULL, lock);
1923
1924                 EXIT;
1925                 return;
1926         }
1927         unlock_res_and_lock(lock);
1928         LDLM_LOCK_RELEASE(lock);
1929         EXIT;
1930 }
1931
1932 static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
1933 {
1934         if (req->rq_no_reply)
1935                 return 0;
1936
1937         req->rq_status = rc;
1938         if (!req->rq_packed_final) {
1939                 rc = lustre_pack_reply(req, 1, NULL, NULL);
1940                 if (rc)
1941                         return rc;
1942         }
1943         return ptlrpc_reply(req);
1944 }
1945
1946 static int __ldlm_bl_to_thread(struct ldlm_bl_work_item *blwi,
1947                                enum ldlm_cancel_flags cancel_flags)
1948 {
1949         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
1950         ENTRY;
1951
1952         spin_lock(&blp->blp_lock);
1953         if (blwi->blwi_lock &&
1954             ldlm_is_discard_data(blwi->blwi_lock)) {
1955                 /* add LDLM_FL_DISCARD_DATA requests to the priority list */
1956                 list_add_tail(&blwi->blwi_entry, &blp->blp_prio_list);
1957         } else {
1958                 /* other blocking callbacks are added to the regular list */
1959                 list_add_tail(&blwi->blwi_entry, &blp->blp_list);
1960         }
1961         spin_unlock(&blp->blp_lock);
1962
1963         wake_up(&blp->blp_waitq);
1964
1965         /* can not check blwi->blwi_flags as blwi could be already freed in
1966            LCF_ASYNC mode */
1967         if (!(cancel_flags & LCF_ASYNC))
1968                 wait_for_completion(&blwi->blwi_comp);
1969
1970         RETURN(0);
1971 }
1972
1973 static inline void init_blwi(struct ldlm_bl_work_item *blwi,
1974                              struct ldlm_namespace *ns,
1975                              struct ldlm_lock_desc *ld,
1976                              struct list_head *cancels, int count,
1977                              struct ldlm_lock *lock,
1978                              enum ldlm_cancel_flags cancel_flags)
1979 {
1980         init_completion(&blwi->blwi_comp);
1981         INIT_LIST_HEAD(&blwi->blwi_head);
1982
1983         if (memory_pressure_get())
1984                 blwi->blwi_mem_pressure = 1;
1985
1986         blwi->blwi_ns = ns;
1987         blwi->blwi_flags = cancel_flags;
1988         if (ld != NULL)
1989                 blwi->blwi_ld = *ld;
1990         if (count) {
1991                 list_add(&blwi->blwi_head, cancels);
1992                 list_del_init(cancels);
1993                 blwi->blwi_count = count;
1994         } else {
1995                 blwi->blwi_lock = lock;
1996         }
1997 }
1998
1999 /**
2000  * Queues a list of locks \a cancels containing \a count locks
2001  * for later processing by a blocking thread.  If \a count is zero,
2002  * then the lock referenced as \a lock is queued instead.
2003  *
2004  * The blocking thread would then call ->l_blocking_ast callback in the lock.
2005  * If list addition fails an error is returned and caller is supposed to
2006  * call ->l_blocking_ast itself.
2007  */
2008 static int ldlm_bl_to_thread(struct ldlm_namespace *ns,
2009                              struct ldlm_lock_desc *ld,
2010                              struct ldlm_lock *lock,
2011                              struct list_head *cancels, int count,
2012                              enum ldlm_cancel_flags cancel_flags)
2013 {
2014         ENTRY;
2015
2016         if (cancels && count == 0)
2017                 RETURN(0);
2018
2019         if (cancel_flags & LCF_ASYNC) {
2020                 struct ldlm_bl_work_item *blwi;
2021
2022                 OBD_ALLOC(blwi, sizeof(*blwi));
2023                 if (blwi == NULL)
2024                         RETURN(-ENOMEM);
2025                 init_blwi(blwi, ns, ld, cancels, count, lock, cancel_flags);
2026
2027                 RETURN(__ldlm_bl_to_thread(blwi, cancel_flags));
2028         } else {
2029                 /* if it is synchronous call do minimum mem alloc, as it could
2030                  * be triggered from kernel shrinker
2031                  */
2032                 struct ldlm_bl_work_item blwi;
2033
2034                 memset(&blwi, 0, sizeof(blwi));
2035                 init_blwi(&blwi, ns, ld, cancels, count, lock, cancel_flags);
2036                 RETURN(__ldlm_bl_to_thread(&blwi, cancel_flags));
2037         }
2038 }
2039
2040
2041 int ldlm_bl_to_thread_lock(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
2042                            struct ldlm_lock *lock)
2043 {
2044         return ldlm_bl_to_thread(ns, ld, lock, NULL, 0, LCF_ASYNC);
2045 }
2046
2047 int ldlm_bl_to_thread_list(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
2048                            struct list_head *cancels, int count,
2049                            enum ldlm_cancel_flags cancel_flags)
2050 {
2051         return ldlm_bl_to_thread(ns, ld, NULL, cancels, count, cancel_flags);
2052 }
2053
2054 int ldlm_bl_thread_wakeup(void)
2055 {
2056         wake_up(&ldlm_state->ldlm_bl_pool->blp_waitq);
2057         return 0;
2058 }
2059
2060 /* Setinfo coming from Server (eg MDT) to Client (eg MDC)! */
2061 static int ldlm_handle_setinfo(struct ptlrpc_request *req)
2062 {
2063         struct obd_device *obd = req->rq_export->exp_obd;
2064         char *key;
2065         void *val;
2066         int keylen, vallen;
2067         int rc = -ENOSYS;
2068         ENTRY;
2069
2070         DEBUG_REQ(D_HSM, req, "%s: handle setinfo\n", obd->obd_name);
2071
2072         req_capsule_set(&req->rq_pill, &RQF_OBD_SET_INFO);
2073
2074         key = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
2075         if (key == NULL) {
2076                 DEBUG_REQ(D_IOCTL, req, "no set_info key");
2077                 RETURN(-EFAULT);
2078         }
2079         keylen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_KEY,
2080                                       RCL_CLIENT);
2081         val = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
2082         if (val == NULL) {
2083                 DEBUG_REQ(D_IOCTL, req, "no set_info val");
2084                 RETURN(-EFAULT);
2085         }
2086         vallen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_VAL,
2087                                       RCL_CLIENT);
2088
2089         /* We are responsible for swabbing contents of val */
2090
2091         if (KEY_IS(KEY_HSM_COPYTOOL_SEND))
2092                 /* Pass it on to mdc (the "export" in this case) */
2093                 rc = obd_set_info_async(req->rq_svc_thread->t_env,
2094                                         req->rq_export,
2095                                         sizeof(KEY_HSM_COPYTOOL_SEND),
2096                                         KEY_HSM_COPYTOOL_SEND,
2097                                         vallen, val, NULL);
2098         else
2099                 DEBUG_REQ(D_WARNING, req, "ignoring unknown key %s", key);
2100
2101         return rc;
2102 }
2103
2104 static inline void ldlm_callback_errmsg(struct ptlrpc_request *req,
2105                                         const char *msg, int rc,
2106                                         const struct lustre_handle *handle)
2107 {
2108         DEBUG_REQ((req->rq_no_reply || rc) ? D_WARNING : D_DLMTRACE, req,
2109                   "%s: [nid %s] [rc %d] [lock %#llx]",
2110                   msg, libcfs_id2str(req->rq_peer), rc,
2111                   handle ? handle->cookie : 0);
2112         if (req->rq_no_reply)
2113                 CWARN("No reply was sent, maybe cause bug 21636.\n");
2114         else if (rc)
2115                 CWARN("Send reply failed, maybe cause bug 21636.\n");
2116 }
2117
2118 /* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
2119 static int ldlm_callback_handler(struct ptlrpc_request *req)
2120 {
2121         struct ldlm_namespace *ns;
2122         struct ldlm_request *dlm_req;
2123         struct ldlm_lock *lock;
2124         int rc;
2125         ENTRY;
2126
2127         /* Requests arrive in sender's byte order.  The ptlrpc service
2128          * handler has already checked and, if necessary, byte-swapped the
2129          * incoming request message body, but I am responsible for the
2130          * message buffers. */
2131
2132         /* do nothing for sec context finalize */
2133         if (lustre_msg_get_opc(req->rq_reqmsg) == SEC_CTX_FINI)
2134                 RETURN(0);
2135
2136         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2137
2138         if (req->rq_export == NULL) {
2139                 rc = ldlm_callback_reply(req, -ENOTCONN);
2140                 ldlm_callback_errmsg(req, "Operate on unconnected server",
2141                                      rc, NULL);
2142                 RETURN(0);
2143         }
2144
2145         LASSERT(req->rq_export != NULL);
2146         LASSERT(req->rq_export->exp_obd != NULL);
2147
2148         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2149         case LDLM_BL_CALLBACK:
2150                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET)) {
2151                         if (cfs_fail_err)
2152                                 ldlm_callback_reply(req, -(int)cfs_fail_err);
2153                         RETURN(0);
2154                 }
2155                 break;
2156         case LDLM_CP_CALLBACK:
2157                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CP_CALLBACK_NET))
2158                         RETURN(0);
2159                 break;
2160         case LDLM_GL_CALLBACK:
2161                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_GL_CALLBACK_NET))
2162                         RETURN(0);
2163                 break;
2164         case LDLM_SET_INFO:
2165                 rc = ldlm_handle_setinfo(req);
2166                 ldlm_callback_reply(req, rc);
2167                 RETURN(0);
2168         case LLOG_ORIGIN_HANDLE_CREATE:
2169                 req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
2170                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
2171                         RETURN(0);
2172                 rc = llog_origin_handle_open(req);
2173                 ldlm_callback_reply(req, rc);
2174                 RETURN(0);
2175         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
2176                 req_capsule_set(&req->rq_pill,
2177                                 &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
2178                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
2179                         RETURN(0);
2180                 rc = llog_origin_handle_next_block(req);
2181                 ldlm_callback_reply(req, rc);
2182                 RETURN(0);
2183         case LLOG_ORIGIN_HANDLE_READ_HEADER:
2184                 req_capsule_set(&req->rq_pill,
2185                                 &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER);
2186                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
2187                         RETURN(0);
2188                 rc = llog_origin_handle_read_header(req);
2189                 ldlm_callback_reply(req, rc);
2190                 RETURN(0);
2191         case LLOG_ORIGIN_HANDLE_CLOSE:
2192                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
2193                         RETURN(0);
2194                 rc = llog_origin_handle_close(req);
2195                 ldlm_callback_reply(req, rc);
2196                 RETURN(0);
2197         default:
2198                 CERROR("unknown opcode %u\n",
2199                        lustre_msg_get_opc(req->rq_reqmsg));
2200                 ldlm_callback_reply(req, -EPROTO);
2201                 RETURN(0);
2202         }
2203
2204         ns = req->rq_export->exp_obd->obd_namespace;
2205         LASSERT(ns != NULL);
2206
2207         req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
2208
2209         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
2210         if (dlm_req == NULL) {
2211                 rc = ldlm_callback_reply(req, -EPROTO);
2212                 ldlm_callback_errmsg(req, "Operate without parameter", rc,
2213                                      NULL);
2214                 RETURN(0);
2215         }
2216
2217         /* Force a known safe race, send a cancel to the server for a lock
2218          * which the server has already started a blocking callback on. */
2219         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE) &&
2220             lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
2221                 rc = ldlm_cli_cancel(&dlm_req->lock_handle[0], 0);
2222                 if (rc < 0)
2223                         CERROR("ldlm_cli_cancel: %d\n", rc);
2224         }
2225
2226         lock = ldlm_handle2lock_long(&dlm_req->lock_handle[0], 0);
2227         if (!lock) {
2228                 CDEBUG(D_DLMTRACE, "callback on lock %#llx - lock "
2229                        "disappeared\n", dlm_req->lock_handle[0].cookie);
2230                 rc = ldlm_callback_reply(req, -EINVAL);
2231                 ldlm_callback_errmsg(req, "Operate with invalid parameter", rc,
2232                                      &dlm_req->lock_handle[0]);
2233                 RETURN(0);
2234         }
2235
2236         if (ldlm_is_fail_loc(lock) &&
2237             lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK)
2238                 OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
2239
2240         /* Copy hints/flags (e.g. LDLM_FL_DISCARD_DATA) from AST. */
2241         lock_res_and_lock(lock);
2242         lock->l_flags |= ldlm_flags_from_wire(dlm_req->lock_flags &
2243                                               LDLM_FL_AST_MASK);
2244         if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
2245                 /* If somebody cancels lock and cache is already dropped,
2246                  * or lock is failed before cp_ast received on client,
2247                  * we can tell the server we have no lock. Otherwise, we
2248                  * should send cancel after dropping the cache. */
2249                 if ((ldlm_is_canceling(lock) && ldlm_is_bl_done(lock)) ||
2250                      ldlm_is_failed(lock)) {
2251                         LDLM_DEBUG(lock, "callback on lock %llx - lock disappeared",
2252                                    dlm_req->lock_handle[0].cookie);
2253                         unlock_res_and_lock(lock);
2254                         LDLM_LOCK_RELEASE(lock);
2255                         rc = ldlm_callback_reply(req, -EINVAL);
2256                         ldlm_callback_errmsg(req, "Operate on stale lock", rc,
2257                                              &dlm_req->lock_handle[0]);
2258                         RETURN(0);
2259                 }
2260                 /* BL_AST locks are not needed in LRU.
2261                  * Let ldlm_cancel_lru() be fast. */
2262                 ldlm_lock_remove_from_lru(lock);
2263                 ldlm_set_bl_ast(lock);
2264         }
2265         unlock_res_and_lock(lock);
2266
2267         /* We want the ost thread to get this reply so that it can respond
2268          * to ost requests (write cache writeback) that might be triggered
2269          * in the callback.
2270          *
2271          * But we'd also like to be able to indicate in the reply that we're
2272          * cancelling right now, because it's unused, or have an intent result
2273          * in the reply, so we might have to push the responsibility for sending
2274          * the reply down into the AST handlers, alas. */
2275
2276         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2277         case LDLM_BL_CALLBACK:
2278                 CDEBUG(D_INODE, "blocking ast\n");
2279                 req_capsule_extend(&req->rq_pill, &RQF_LDLM_BL_CALLBACK);
2280                 if (!ldlm_is_cancel_on_block(lock)) {
2281                         rc = ldlm_callback_reply(req, 0);
2282                         if (req->rq_no_reply || rc)
2283                                 ldlm_callback_errmsg(req, "Normal process", rc,
2284                                                      &dlm_req->lock_handle[0]);
2285                 }
2286                 if (ldlm_bl_to_thread_lock(ns, &dlm_req->lock_desc, lock))
2287                         ldlm_handle_bl_callback(ns, &dlm_req->lock_desc, lock);
2288                 break;
2289         case LDLM_CP_CALLBACK:
2290                 CDEBUG(D_INODE, "completion ast\n");
2291                 req_capsule_extend(&req->rq_pill, &RQF_LDLM_CP_CALLBACK);
2292                 ldlm_callback_reply(req, 0);
2293                 ldlm_handle_cp_callback(req, ns, dlm_req, lock);
2294                 break;
2295         case LDLM_GL_CALLBACK:
2296                 CDEBUG(D_INODE, "glimpse ast\n");
2297                 req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_CALLBACK);
2298                 ldlm_handle_gl_callback(req, ns, dlm_req, lock);
2299                 break;
2300         default:
2301                 LBUG();                         /* checked above */
2302         }
2303
2304         RETURN(0);
2305 }
2306
2307 #ifdef HAVE_SERVER_SUPPORT
2308 /**
2309  * Main handler for canceld thread.
2310  *
2311  * Separated into its own thread to avoid deadlocks.
2312  */
2313 static int ldlm_cancel_handler(struct ptlrpc_request *req)
2314 {
2315         int rc;
2316         ENTRY;
2317
2318         /* Requests arrive in sender's byte order.  The ptlrpc service
2319          * handler has already checked and, if necessary, byte-swapped the
2320          * incoming request message body, but I am responsible for the
2321          * message buffers. */
2322
2323         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2324
2325         if (req->rq_export == NULL) {
2326                 struct ldlm_request *dlm_req;
2327
2328                 CERROR("%s from %s arrived at %lu with bad export cookie "
2329                        "%llu\n",
2330                        ll_opcode2str(lustre_msg_get_opc(req->rq_reqmsg)),
2331                        libcfs_nid2str(req->rq_peer.nid),
2332                        req->rq_arrival_time.tv_sec,
2333                        lustre_msg_get_handle(req->rq_reqmsg)->cookie);
2334
2335                 if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_CANCEL) {
2336                         req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
2337                         dlm_req = req_capsule_client_get(&req->rq_pill,
2338                                                          &RMF_DLM_REQ);
2339                         if (dlm_req != NULL)
2340                                 ldlm_lock_dump_handle(D_ERROR,
2341                                                       &dlm_req->lock_handle[0]);
2342                 }
2343                 ldlm_callback_reply(req, -ENOTCONN);
2344                 RETURN(0);
2345         }
2346
2347         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2348
2349         /* XXX FIXME move this back to mds/handler.c, bug 249 */
2350         case LDLM_CANCEL:
2351                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
2352                 CDEBUG(D_INODE, "cancel\n");
2353                 if (CFS_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_NET) ||
2354                     CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_CANCEL_RESEND) ||
2355                     CFS_FAIL_CHECK(OBD_FAIL_LDLM_BL_EVICT))
2356                         RETURN(0);
2357                 rc = ldlm_handle_cancel(req);
2358                 if (rc)
2359                         break;
2360                 RETURN(0);
2361         default:
2362                 CERROR("invalid opcode %d\n",
2363                        lustre_msg_get_opc(req->rq_reqmsg));
2364                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
2365                 ldlm_callback_reply(req, -EINVAL);
2366         }
2367
2368         RETURN(0);
2369 }
2370
2371 static int ldlm_cancel_hpreq_lock_match(struct ptlrpc_request *req,
2372                                         struct ldlm_lock *lock)
2373 {
2374         struct ldlm_request *dlm_req;
2375         struct lustre_handle lockh;
2376         int rc = 0;
2377         int i;
2378         ENTRY;
2379
2380         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
2381         if (dlm_req == NULL)
2382                 RETURN(0);
2383
2384         ldlm_lock2handle(lock, &lockh);
2385         for (i = 0; i < dlm_req->lock_count; i++) {
2386                 if (lustre_handle_equal(&dlm_req->lock_handle[i],
2387                                         &lockh)) {
2388                         DEBUG_REQ(D_RPCTRACE, req,
2389                                   "Prio raised by lock %#llx.", lockh.cookie);
2390
2391                         rc = 1;
2392                         break;
2393                 }
2394         }
2395
2396         RETURN(rc);
2397
2398 }
2399
2400 static int ldlm_cancel_hpreq_check(struct ptlrpc_request *req)
2401 {
2402         struct ldlm_request *dlm_req;
2403         int rc = 0;
2404         int i;
2405         ENTRY;
2406
2407         /* no prolong in recovery */
2408         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
2409                 RETURN(0);
2410
2411         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
2412         if (dlm_req == NULL)
2413                 RETURN(-EFAULT);
2414
2415         for (i = 0; i < dlm_req->lock_count; i++) {
2416                 struct ldlm_lock *lock;
2417
2418                 lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
2419                 if (lock == NULL)
2420                         continue;
2421
2422                 rc = ldlm_is_ast_sent(lock) ? 1 : 0;
2423                 if (rc)
2424                         LDLM_DEBUG(lock, "hpreq cancel lock");
2425                 LDLM_LOCK_PUT(lock);
2426
2427                 if (rc)
2428                         break;
2429         }
2430
2431         RETURN(rc);
2432 }
2433
2434 static struct ptlrpc_hpreq_ops ldlm_cancel_hpreq_ops = {
2435         .hpreq_lock_match = ldlm_cancel_hpreq_lock_match,
2436         .hpreq_check      = ldlm_cancel_hpreq_check,
2437         .hpreq_fini       = NULL,
2438 };
2439
2440 static int ldlm_hpreq_handler(struct ptlrpc_request *req)
2441 {
2442         ENTRY;
2443
2444         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2445
2446         if (req->rq_export == NULL)
2447                 RETURN(0);
2448
2449         if (LDLM_CANCEL == lustre_msg_get_opc(req->rq_reqmsg)) {
2450                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
2451                 req->rq_ops = &ldlm_cancel_hpreq_ops;
2452         }
2453         RETURN(0);
2454 }
2455
2456 static int ldlm_revoke_lock_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
2457                                struct hlist_node *hnode, void *data)
2458
2459 {
2460         struct list_head         *rpc_list = data;
2461         struct ldlm_lock   *lock = cfs_hash_object(hs, hnode);
2462
2463         lock_res_and_lock(lock);
2464
2465         if (lock->l_req_mode != lock->l_granted_mode) {
2466                 unlock_res_and_lock(lock);
2467                 return 0;
2468         }
2469
2470         LASSERT(lock->l_resource);
2471         if (lock->l_resource->lr_type != LDLM_IBITS &&
2472             lock->l_resource->lr_type != LDLM_PLAIN) {
2473                 unlock_res_and_lock(lock);
2474                 return 0;
2475         }
2476
2477         if (ldlm_is_ast_sent(lock)) {
2478                 unlock_res_and_lock(lock);
2479                 return 0;
2480         }
2481
2482         LASSERT(lock->l_blocking_ast);
2483         LASSERT(!lock->l_blocking_lock);
2484
2485         ldlm_set_ast_sent(lock);
2486         if (lock->l_export && lock->l_export->exp_lock_hash) {
2487                 /* NB: it's safe to call cfs_hash_del() even lock isn't
2488                  * in exp_lock_hash. */
2489                 /* In the function below, .hs_keycmp resolves to
2490                  * ldlm_export_lock_keycmp() */
2491                 /* coverity[overrun-buffer-val] */
2492                 cfs_hash_del(lock->l_export->exp_lock_hash,
2493                              &lock->l_remote_handle, &lock->l_exp_hash);
2494         }
2495
2496         list_add_tail(&lock->l_rk_ast, rpc_list);
2497         LDLM_LOCK_GET(lock);
2498
2499         unlock_res_and_lock(lock);
2500         return 0;
2501 }
2502
2503 void ldlm_revoke_export_locks(struct obd_export *exp)
2504 {
2505         struct list_head  rpc_list;
2506         ENTRY;
2507
2508         INIT_LIST_HEAD(&rpc_list);
2509         cfs_hash_for_each_nolock(exp->exp_lock_hash,
2510                                  ldlm_revoke_lock_cb, &rpc_list, 0);
2511         ldlm_run_ast_work(exp->exp_obd->obd_namespace, &rpc_list,
2512                           LDLM_WORK_REVOKE_AST);
2513
2514         EXIT;
2515 }
2516 EXPORT_SYMBOL(ldlm_revoke_export_locks);
2517 #endif /* HAVE_SERVER_SUPPORT */
2518
2519 static int ldlm_bl_get_work(struct ldlm_bl_pool *blp,
2520                             struct ldlm_bl_work_item **p_blwi,
2521                             struct obd_export **p_exp)
2522 {
2523         struct ldlm_bl_work_item *blwi = NULL;
2524         static unsigned int num_bl = 0;
2525         static unsigned int num_stale;
2526         int num_th = atomic_read(&blp->blp_num_threads);
2527
2528         *p_exp = obd_stale_export_get();
2529
2530         spin_lock(&blp->blp_lock);
2531         if (*p_exp != NULL) {
2532                 if (num_th == 1 || ++num_stale < num_th) {
2533                         spin_unlock(&blp->blp_lock);
2534                         return 1;
2535                 } else {
2536                         num_stale = 0;
2537                 }
2538         }
2539
2540         /* process a request from the blp_list at least every blp_num_threads */
2541         if (!list_empty(&blp->blp_list) &&
2542             (list_empty(&blp->blp_prio_list) || num_bl == 0))
2543                 blwi = list_entry(blp->blp_list.next,
2544                                   struct ldlm_bl_work_item, blwi_entry);
2545         else
2546                 if (!list_empty(&blp->blp_prio_list))
2547                         blwi = list_entry(blp->blp_prio_list.next,
2548                                           struct ldlm_bl_work_item,
2549                                           blwi_entry);
2550
2551         if (blwi) {
2552                 if (++num_bl >= num_th)
2553                         num_bl = 0;
2554                 list_del(&blwi->blwi_entry);
2555         }
2556         spin_unlock(&blp->blp_lock);
2557         *p_blwi = blwi;
2558
2559         if (*p_exp != NULL && *p_blwi != NULL) {
2560                 obd_stale_export_put(*p_exp);
2561                 *p_exp = NULL;
2562         }
2563
2564         return (*p_blwi != NULL || *p_exp != NULL) ? 1 : 0;
2565 }
2566
2567 /* This only contains temporary data until the thread starts */
2568 struct ldlm_bl_thread_data {
2569         struct ldlm_bl_pool     *bltd_blp;
2570         struct completion       bltd_comp;
2571         int                     bltd_num;
2572 };
2573
2574 static int ldlm_bl_thread_main(void *arg);
2575
2576 static int ldlm_bl_thread_start(struct ldlm_bl_pool *blp, bool check_busy)
2577 {
2578         struct ldlm_bl_thread_data bltd = { .bltd_blp = blp };
2579         struct task_struct *task;
2580
2581         init_completion(&bltd.bltd_comp);
2582
2583         bltd.bltd_num = atomic_inc_return(&blp->blp_num_threads);
2584         if (bltd.bltd_num >= blp->blp_max_threads) {
2585                 atomic_dec(&blp->blp_num_threads);
2586                 return 0;
2587         }
2588
2589         LASSERTF(bltd.bltd_num > 0, "thread num:%d\n", bltd.bltd_num);
2590         if (check_busy &&
2591             atomic_read(&blp->blp_busy_threads) < (bltd.bltd_num - 1)) {
2592                 atomic_dec(&blp->blp_num_threads);
2593                 return 0;
2594         }
2595
2596         task = kthread_run(ldlm_bl_thread_main, &bltd, "ldlm_bl_%02d",
2597                            bltd.bltd_num);
2598         if (IS_ERR(task)) {
2599                 CERROR("cannot start LDLM thread ldlm_bl_%02d: rc %ld\n",
2600                        bltd.bltd_num, PTR_ERR(task));
2601                 atomic_dec(&blp->blp_num_threads);
2602                 return PTR_ERR(task);
2603         }
2604         wait_for_completion(&bltd.bltd_comp);
2605
2606         return 0;
2607 }
2608
2609 /* Not fatal if racy and have a few too many threads */
2610 static int ldlm_bl_thread_need_create(struct ldlm_bl_pool *blp,
2611                                       struct ldlm_bl_work_item *blwi)
2612 {
2613         if (atomic_read(&blp->blp_num_threads) >= blp->blp_max_threads)
2614                 return 0;
2615
2616         if (atomic_read(&blp->blp_busy_threads) <
2617             atomic_read(&blp->blp_num_threads))
2618                 return 0;
2619
2620         if (blwi != NULL && (blwi->blwi_ns == NULL ||
2621                              blwi->blwi_mem_pressure))
2622                 return 0;
2623
2624         return 1;
2625 }
2626
2627 static int ldlm_bl_thread_blwi(struct ldlm_bl_pool *blp,
2628                                struct ldlm_bl_work_item *blwi)
2629 {
2630         ENTRY;
2631
2632         if (blwi->blwi_ns == NULL)
2633                 /* added by ldlm_cleanup() */
2634                 RETURN(LDLM_ITER_STOP);
2635
2636         if (blwi->blwi_mem_pressure)
2637                 memory_pressure_set();
2638
2639         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_PAUSE_CANCEL2, 4);
2640
2641         if (blwi->blwi_count) {
2642                 int count;
2643                 /* The special case when we cancel locks in lru
2644                  * asynchronously, we pass the list of locks here.
2645                  * Thus locks are marked LDLM_FL_CANCELING, but NOT
2646                  * canceled locally yet. */
2647                 count = ldlm_cli_cancel_list_local(&blwi->blwi_head,
2648                                                    blwi->blwi_count,
2649                                                    LCF_BL_AST);
2650                 ldlm_cli_cancel_list(&blwi->blwi_head, count, NULL,
2651                                      blwi->blwi_flags);
2652         } else {
2653                 ldlm_handle_bl_callback(blwi->blwi_ns, &blwi->blwi_ld,
2654                                         blwi->blwi_lock);
2655         }
2656         if (blwi->blwi_mem_pressure)
2657                 memory_pressure_clr();
2658
2659         if (blwi->blwi_flags & LCF_ASYNC)
2660                 OBD_FREE(blwi, sizeof(*blwi));
2661         else
2662                 complete(&blwi->blwi_comp);
2663
2664         RETURN(0);
2665 }
2666
2667 /**
2668  * Cancel stale locks on export. Cancel blocked locks first.
2669  * If the given export has blocked locks, the next in the list may have
2670  * them too, thus cancel not blocked locks only if the current export has
2671  * no blocked locks.
2672  **/
2673 static int ldlm_bl_thread_exports(struct ldlm_bl_pool *blp,
2674                                   struct obd_export *exp)
2675 {
2676         int num;
2677         ENTRY;
2678
2679         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_BL_EVICT, 4);
2680
2681         num = ldlm_export_cancel_blocked_locks(exp);
2682         if (num == 0)
2683                 ldlm_export_cancel_locks(exp);
2684
2685         obd_stale_export_put(exp);
2686
2687         RETURN(0);
2688 }
2689
2690
2691 /**
2692  * Main blocking requests processing thread.
2693  *
2694  * Callers put locks into its queue by calling ldlm_bl_to_thread.
2695  * This thread in the end ends up doing actual call to ->l_blocking_ast
2696  * for queued locks.
2697  */
2698 static int ldlm_bl_thread_main(void *arg)
2699 {
2700         struct ldlm_bl_pool *blp;
2701         struct ldlm_bl_thread_data *bltd = arg;
2702         ENTRY;
2703
2704         blp = bltd->bltd_blp;
2705
2706         complete(&bltd->bltd_comp);
2707         /* cannot use bltd after this, it is only on caller's stack */
2708
2709         while (1) {
2710                 struct l_wait_info lwi = { 0 };
2711                 struct ldlm_bl_work_item *blwi = NULL;
2712                 struct obd_export *exp = NULL;
2713                 int rc;
2714
2715                 rc = ldlm_bl_get_work(blp, &blwi, &exp);
2716
2717                 if (rc == 0)
2718                         l_wait_event_exclusive(blp->blp_waitq,
2719                                                ldlm_bl_get_work(blp, &blwi,
2720                                                                 &exp),
2721                                                &lwi);
2722                 atomic_inc(&blp->blp_busy_threads);
2723
2724                 if (ldlm_bl_thread_need_create(blp, blwi))
2725                         /* discard the return value, we tried */
2726                         ldlm_bl_thread_start(blp, true);
2727
2728                 if (exp)
2729                         rc = ldlm_bl_thread_exports(blp, exp);
2730                 else if (blwi)
2731                         rc = ldlm_bl_thread_blwi(blp, blwi);
2732
2733                 atomic_dec(&blp->blp_busy_threads);
2734
2735                 if (rc == LDLM_ITER_STOP)
2736                         break;
2737
2738                 /* If there are many namespaces, we will not sleep waiting for
2739                  * work, and must do a cond_resched to avoid holding the CPU
2740                  * for too long */
2741                 cond_resched();
2742         }
2743
2744         atomic_dec(&blp->blp_num_threads);
2745         complete(&blp->blp_comp);
2746         RETURN(0);
2747 }
2748
2749
2750 static int ldlm_setup(void);
2751 static int ldlm_cleanup(void);
2752
2753 int ldlm_get_ref(void)
2754 {
2755         int rc = 0;
2756         ENTRY;
2757         mutex_lock(&ldlm_ref_mutex);
2758         if (++ldlm_refcount == 1) {
2759                 rc = ldlm_setup();
2760                 if (rc)
2761                         ldlm_refcount--;
2762         }
2763         mutex_unlock(&ldlm_ref_mutex);
2764
2765         RETURN(rc);
2766 }
2767
2768 void ldlm_put_ref(void)
2769 {
2770         ENTRY;
2771         mutex_lock(&ldlm_ref_mutex);
2772         if (ldlm_refcount == 1) {
2773                 int rc = ldlm_cleanup();
2774                 if (rc)
2775                         CERROR("ldlm_cleanup failed: %d\n", rc);
2776                 else
2777                         ldlm_refcount--;
2778         } else {
2779                 ldlm_refcount--;
2780         }
2781         mutex_unlock(&ldlm_ref_mutex);
2782
2783         EXIT;
2784 }
2785
2786 /*
2787  * Export handle<->lock hash operations.
2788  */
2789 static unsigned
2790 ldlm_export_lock_hash(struct cfs_hash *hs, const void *key, unsigned mask)
2791 {
2792         return cfs_hash_u64_hash(((struct lustre_handle *)key)->cookie, mask);
2793 }
2794
2795 static void *
2796 ldlm_export_lock_key(struct hlist_node *hnode)
2797 {
2798         struct ldlm_lock *lock;
2799
2800         lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2801         return &lock->l_remote_handle;
2802 }
2803
2804 static void
2805 ldlm_export_lock_keycpy(struct hlist_node *hnode, void *key)
2806 {
2807         struct ldlm_lock     *lock;
2808
2809         lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2810         lock->l_remote_handle = *(struct lustre_handle *)key;
2811 }
2812
2813 static int
2814 ldlm_export_lock_keycmp(const void *key, struct hlist_node *hnode)
2815 {
2816         return lustre_handle_equal(ldlm_export_lock_key(hnode), key);
2817 }
2818
2819 static void *
2820 ldlm_export_lock_object(struct hlist_node *hnode)
2821 {
2822         return hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2823 }
2824
2825 static void
2826 ldlm_export_lock_get(struct cfs_hash *hs, struct hlist_node *hnode)
2827 {
2828         struct ldlm_lock *lock;
2829
2830         lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2831         LDLM_LOCK_GET(lock);
2832 }
2833
2834 static void
2835 ldlm_export_lock_put(struct cfs_hash *hs, struct hlist_node *hnode)
2836 {
2837         struct ldlm_lock *lock;
2838
2839         lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2840         LDLM_LOCK_RELEASE(lock);
2841 }
2842
2843 static struct cfs_hash_ops ldlm_export_lock_ops = {
2844         .hs_hash        = ldlm_export_lock_hash,
2845         .hs_key         = ldlm_export_lock_key,
2846         .hs_keycmp      = ldlm_export_lock_keycmp,
2847         .hs_keycpy      = ldlm_export_lock_keycpy,
2848         .hs_object      = ldlm_export_lock_object,
2849         .hs_get         = ldlm_export_lock_get,
2850         .hs_put         = ldlm_export_lock_put,
2851         .hs_put_locked  = ldlm_export_lock_put,
2852 };
2853
2854 int ldlm_init_export(struct obd_export *exp)
2855 {
2856         int rc;
2857         ENTRY;
2858
2859         exp->exp_lock_hash =
2860                 cfs_hash_create(obd_uuid2str(&exp->exp_client_uuid),
2861                                 HASH_EXP_LOCK_CUR_BITS,
2862                                 HASH_EXP_LOCK_MAX_BITS,
2863                                 HASH_EXP_LOCK_BKT_BITS, 0,
2864                                 CFS_HASH_MIN_THETA, CFS_HASH_MAX_THETA,
2865                                 &ldlm_export_lock_ops,
2866                                 CFS_HASH_DEFAULT | CFS_HASH_REHASH_KEY |
2867                                 CFS_HASH_NBLK_CHANGE);
2868
2869         if (!exp->exp_lock_hash)
2870                 RETURN(-ENOMEM);
2871
2872         rc = ldlm_init_flock_export(exp);
2873         if (rc)
2874                 GOTO(err, rc);
2875
2876         RETURN(0);
2877 err:
2878         ldlm_destroy_export(exp);
2879         RETURN(rc);
2880 }
2881 EXPORT_SYMBOL(ldlm_init_export);
2882
2883 void ldlm_destroy_export(struct obd_export *exp)
2884 {
2885         ENTRY;
2886         cfs_hash_putref(exp->exp_lock_hash);
2887         exp->exp_lock_hash = NULL;
2888
2889         ldlm_destroy_flock_export(exp);
2890         EXIT;
2891 }
2892 EXPORT_SYMBOL(ldlm_destroy_export);
2893
2894 static ssize_t cancel_unused_locks_before_replay_show(struct kobject *kobj,
2895                                                       struct attribute *attr,
2896                                                       char *buf)
2897 {
2898         return sprintf(buf, "%d\n", ldlm_cancel_unused_locks_before_replay);
2899 }
2900
2901 static ssize_t cancel_unused_locks_before_replay_store(struct kobject *kobj,
2902                                                        struct attribute *attr,
2903                                                        const char *buffer,
2904                                                        size_t count)
2905 {
2906         int rc;
2907         unsigned long val;
2908
2909         rc = kstrtoul(buffer, 10, &val);
2910         if (rc)
2911                 return rc;
2912
2913         ldlm_cancel_unused_locks_before_replay = val;
2914
2915         return count;
2916 }
2917 LUSTRE_RW_ATTR(cancel_unused_locks_before_replay);
2918
2919 static struct attribute *ldlm_attrs[] = {
2920         &lustre_attr_cancel_unused_locks_before_replay.attr,
2921         NULL,
2922 };
2923
2924 static struct attribute_group ldlm_attr_group = {
2925         .attrs = ldlm_attrs,
2926 };
2927
2928 static int ldlm_setup(void)
2929 {
2930         static struct ptlrpc_service_conf       conf;
2931         struct ldlm_bl_pool                    *blp = NULL;
2932 #ifdef HAVE_SERVER_SUPPORT
2933         struct task_struct *task;
2934 #endif /* HAVE_SERVER_SUPPORT */
2935         int i;
2936         int rc = 0;
2937
2938         ENTRY;
2939
2940         if (ldlm_state != NULL)
2941                 RETURN(-EALREADY);
2942
2943         OBD_ALLOC(ldlm_state, sizeof(*ldlm_state));
2944         if (ldlm_state == NULL)
2945                 RETURN(-ENOMEM);
2946
2947         ldlm_kobj = kobject_create_and_add("ldlm", &lustre_kset->kobj);
2948         if (!ldlm_kobj)
2949                 GOTO(out, -ENOMEM);
2950
2951         rc = sysfs_create_group(ldlm_kobj, &ldlm_attr_group);
2952         if (rc)
2953                 GOTO(out, rc);
2954
2955         ldlm_ns_kset = kset_create_and_add("namespaces", NULL, ldlm_kobj);
2956         if (!ldlm_ns_kset)
2957                 GOTO(out, -ENOMEM);
2958
2959         ldlm_svc_kset = kset_create_and_add("services", NULL, ldlm_kobj);
2960         if (!ldlm_svc_kset)
2961                 GOTO(out, -ENOMEM);
2962
2963 #ifdef CONFIG_PROC_FS
2964         rc = ldlm_proc_setup();
2965         if (rc != 0)
2966                 GOTO(out, rc);
2967 #endif /* CONFIG_PROC_FS */
2968
2969         memset(&conf, 0, sizeof(conf));
2970         conf = (typeof(conf)) {
2971                 .psc_name               = "ldlm_cbd",
2972                 .psc_watchdog_factor    = 2,
2973                 .psc_buf                = {
2974                         .bc_nbufs               = LDLM_CLIENT_NBUFS,
2975                         .bc_buf_size            = LDLM_BUFSIZE,
2976                         .bc_req_max_size        = LDLM_MAXREQSIZE,
2977                         .bc_rep_max_size        = LDLM_MAXREPSIZE,
2978                         .bc_req_portal          = LDLM_CB_REQUEST_PORTAL,
2979                         .bc_rep_portal          = LDLM_CB_REPLY_PORTAL,
2980                 },
2981                 .psc_thr                = {
2982                         .tc_thr_name            = "ldlm_cb",
2983                         .tc_thr_factor          = LDLM_THR_FACTOR,
2984                         .tc_nthrs_init          = LDLM_NTHRS_INIT,
2985                         .tc_nthrs_base          = LDLM_NTHRS_BASE,
2986                         .tc_nthrs_max           = LDLM_NTHRS_MAX,
2987                         .tc_nthrs_user          = ldlm_num_threads,
2988                         .tc_cpu_affinity        = 1,
2989                         .tc_ctx_tags            = LCT_MD_THREAD | LCT_DT_THREAD,
2990                 },
2991                 .psc_cpt                = {
2992                         .cc_pattern             = ldlm_cpts,
2993                 },
2994                 .psc_ops                = {
2995                         .so_req_handler         = ldlm_callback_handler,
2996                 },
2997         };
2998         ldlm_state->ldlm_cb_service = \
2999                         ptlrpc_register_service(&conf, ldlm_svc_kset,
3000                                                 ldlm_svc_proc_dir);
3001         if (IS_ERR(ldlm_state->ldlm_cb_service)) {
3002                 CERROR("failed to start service\n");
3003                 rc = PTR_ERR(ldlm_state->ldlm_cb_service);
3004                 ldlm_state->ldlm_cb_service = NULL;
3005                 GOTO(out, rc);
3006         }
3007
3008 #ifdef HAVE_SERVER_SUPPORT
3009         memset(&conf, 0, sizeof(conf));
3010         conf = (typeof(conf)) {
3011                 .psc_name               = "ldlm_canceld",
3012                 .psc_watchdog_factor    = 6,
3013                 .psc_buf                = {
3014                         .bc_nbufs               = LDLM_SERVER_NBUFS,
3015                         .bc_buf_size            = LDLM_BUFSIZE,
3016                         .bc_req_max_size        = LDLM_MAXREQSIZE,
3017                         .bc_rep_max_size        = LDLM_MAXREPSIZE,
3018                         .bc_req_portal          = LDLM_CANCEL_REQUEST_PORTAL,
3019                         .bc_rep_portal          = LDLM_CANCEL_REPLY_PORTAL,
3020
3021                 },
3022                 .psc_thr                = {
3023                         .tc_thr_name            = "ldlm_cn",
3024                         .tc_thr_factor          = LDLM_THR_FACTOR,
3025                         .tc_nthrs_init          = LDLM_NTHRS_INIT,
3026                         .tc_nthrs_base          = LDLM_NTHRS_BASE,
3027                         .tc_nthrs_max           = LDLM_NTHRS_MAX,
3028                         .tc_nthrs_user          = ldlm_num_threads,
3029                         .tc_cpu_affinity        = 1,
3030                         .tc_ctx_tags            = LCT_MD_THREAD | \
3031                                                   LCT_DT_THREAD | \
3032                                                   LCT_CL_THREAD,
3033                 },
3034                 .psc_cpt                = {
3035                         .cc_pattern             = ldlm_cpts,
3036                 },
3037                 .psc_ops                = {
3038                         .so_req_handler         = ldlm_cancel_handler,
3039                         .so_hpreq_handler       = ldlm_hpreq_handler,
3040                 },
3041         };
3042         ldlm_state->ldlm_cancel_service = \
3043                         ptlrpc_register_service(&conf, ldlm_svc_kset,
3044                                                 ldlm_svc_proc_dir);
3045         if (IS_ERR(ldlm_state->ldlm_cancel_service)) {
3046                 CERROR("failed to start service\n");
3047                 rc = PTR_ERR(ldlm_state->ldlm_cancel_service);
3048                 ldlm_state->ldlm_cancel_service = NULL;
3049                 GOTO(out, rc);
3050         }
3051 #endif /* HAVE_SERVER_SUPPORT */
3052
3053         OBD_ALLOC(blp, sizeof(*blp));
3054         if (blp == NULL)
3055                 GOTO(out, rc = -ENOMEM);
3056         ldlm_state->ldlm_bl_pool = blp;
3057
3058         spin_lock_init(&blp->blp_lock);
3059         INIT_LIST_HEAD(&blp->blp_list);
3060         INIT_LIST_HEAD(&blp->blp_prio_list);
3061         init_waitqueue_head(&blp->blp_waitq);
3062         atomic_set(&blp->blp_num_threads, 0);
3063         atomic_set(&blp->blp_busy_threads, 0);
3064
3065         if (ldlm_num_threads == 0) {
3066                 blp->blp_min_threads = LDLM_NTHRS_INIT;
3067                 blp->blp_max_threads = LDLM_NTHRS_MAX;
3068         } else {
3069                 blp->blp_min_threads = blp->blp_max_threads = \
3070                         min_t(int, LDLM_NTHRS_MAX, max_t(int, LDLM_NTHRS_INIT,
3071                                                          ldlm_num_threads));
3072         }
3073
3074         for (i = 0; i < blp->blp_min_threads; i++) {
3075                 rc = ldlm_bl_thread_start(blp, false);
3076                 if (rc < 0)
3077                         GOTO(out, rc);
3078         }
3079
3080 #ifdef HAVE_SERVER_SUPPORT
3081         task = kthread_run(expired_lock_main, NULL, "ldlm_elt");
3082         if (IS_ERR(task)) {
3083                 rc = PTR_ERR(task);
3084                 CERROR("Cannot start ldlm expired-lock thread: %d\n", rc);
3085                 GOTO(out, rc);
3086         }
3087
3088         wait_event(expired_lock_wait_queue,
3089                    expired_lock_thread_state == ELT_READY);
3090 #endif /* HAVE_SERVER_SUPPORT */
3091
3092         rc = ldlm_pools_init();
3093         if (rc) {
3094                 CERROR("Failed to initialize LDLM pools: %d\n", rc);
3095                 GOTO(out, rc);
3096         }
3097
3098         rc = ldlm_reclaim_setup();
3099         if (rc) {
3100                 CERROR("Failed to setup reclaim thread: rc = %d\n", rc);
3101                 GOTO(out, rc);
3102         }
3103         RETURN(0);
3104
3105  out:
3106         ldlm_cleanup();
3107         RETURN(rc);
3108 }
3109
3110 static int ldlm_cleanup(void)
3111 {
3112         ENTRY;
3113
3114         if (!list_empty(ldlm_namespace_list(LDLM_NAMESPACE_SERVER)) ||
3115             !list_empty(ldlm_namespace_list(LDLM_NAMESPACE_CLIENT))) {
3116                 CERROR("ldlm still has namespaces; clean these up first.\n");
3117                 ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
3118                 ldlm_dump_all_namespaces(LDLM_NAMESPACE_CLIENT, D_DLMTRACE);
3119                 RETURN(-EBUSY);
3120         }
3121
3122         ldlm_reclaim_cleanup();
3123         ldlm_pools_fini();
3124
3125         if (ldlm_state->ldlm_bl_pool != NULL) {
3126                 struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
3127
3128                 while (atomic_read(&blp->blp_num_threads) > 0) {
3129                         struct ldlm_bl_work_item blwi = { .blwi_ns = NULL };
3130
3131                         init_completion(&blp->blp_comp);
3132
3133                         spin_lock(&blp->blp_lock);
3134                         list_add_tail(&blwi.blwi_entry, &blp->blp_list);
3135                         wake_up(&blp->blp_waitq);
3136                         spin_unlock(&blp->blp_lock);
3137
3138                         wait_for_completion(&blp->blp_comp);
3139                 }
3140
3141                 OBD_FREE(blp, sizeof(*blp));
3142         }
3143
3144         if (ldlm_state->ldlm_cb_service != NULL)
3145                 ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
3146 #ifdef HAVE_SERVER_SUPPORT
3147         if (ldlm_state->ldlm_cancel_service != NULL)
3148                 ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
3149 #endif
3150
3151         if (ldlm_ns_kset)
3152                 kset_unregister(ldlm_ns_kset);
3153         if (ldlm_svc_kset)
3154                 kset_unregister(ldlm_svc_kset);
3155         if (ldlm_kobj) {
3156                 sysfs_remove_group(ldlm_kobj, &ldlm_attr_group);
3157                 kobject_put(ldlm_kobj);
3158         }
3159
3160         ldlm_proc_cleanup();
3161
3162 #ifdef HAVE_SERVER_SUPPORT
3163         if (expired_lock_thread_state != ELT_STOPPED) {
3164                 expired_lock_thread_state = ELT_TERMINATE;
3165                 wake_up(&expired_lock_wait_queue);
3166                 wait_event(expired_lock_wait_queue,
3167                            expired_lock_thread_state == ELT_STOPPED);
3168         }
3169 #endif
3170
3171         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
3172         ldlm_state = NULL;
3173
3174         RETURN(0);
3175 }
3176
3177 int ldlm_init(void)
3178 {
3179         ldlm_resource_slab = kmem_cache_create("ldlm_resources",
3180                                                sizeof(struct ldlm_resource), 0,
3181                                                SLAB_HWCACHE_ALIGN, NULL);
3182         if (ldlm_resource_slab == NULL)
3183                 return -ENOMEM;
3184
3185         ldlm_lock_slab = kmem_cache_create("ldlm_locks",
3186                               sizeof(struct ldlm_lock), 0,
3187                               SLAB_HWCACHE_ALIGN | SLAB_DESTROY_BY_RCU, NULL);
3188         if (ldlm_lock_slab == NULL)
3189                 goto out_resource;
3190
3191         ldlm_interval_slab = kmem_cache_create("interval_node",
3192                                         sizeof(struct ldlm_interval),
3193                                         0, SLAB_HWCACHE_ALIGN, NULL);
3194         if (ldlm_interval_slab == NULL)
3195                 goto out_lock;
3196
3197         ldlm_interval_tree_slab = kmem_cache_create("interval_tree",
3198                         sizeof(struct ldlm_interval_tree) * LCK_MODE_NUM,
3199                         0, SLAB_HWCACHE_ALIGN, NULL);
3200         if (ldlm_interval_tree_slab == NULL)
3201                 goto out_interval;
3202
3203 #if LUSTRE_TRACKS_LOCK_EXP_REFS
3204         class_export_dump_hook = ldlm_dump_export_locks;
3205 #endif
3206         return 0;
3207
3208 out_interval:
3209         kmem_cache_destroy(ldlm_interval_slab);
3210 out_lock:
3211         kmem_cache_destroy(ldlm_lock_slab);
3212 out_resource:
3213         kmem_cache_destroy(ldlm_resource_slab);
3214
3215         return -ENOMEM;
3216 }
3217
3218 void ldlm_exit(void)
3219 {
3220         if (ldlm_refcount)
3221                 CERROR("ldlm_refcount is %d in ldlm_exit!\n", ldlm_refcount);
3222         kmem_cache_destroy(ldlm_resource_slab);
3223         /* ldlm_lock_put() use RCU to call ldlm_lock_free, so need call
3224          * synchronize_rcu() to wait a grace period elapsed, so that
3225          * ldlm_lock_free() get a chance to be called. */
3226         synchronize_rcu();
3227         kmem_cache_destroy(ldlm_lock_slab);
3228         kmem_cache_destroy(ldlm_interval_slab);
3229         kmem_cache_destroy(ldlm_interval_tree_slab);
3230 }