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