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