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