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