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