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