Whamcloud - gitweb
143d0a9c376b07c8d9dd1cfbce72ca892e97f13a
[fs/lustre-release.git] / lustre / ldlm / ldlm_request.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2010, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  */
32 /**
33  * This file contains Asynchronous System Trap (AST) handlers and related
34  * LDLM request-processing routines.
35  *
36  * An AST is a callback issued on a lock when its state is changed. There are
37  * several different types of ASTs (callbacks) registered for each lock:
38  *
39  * - completion AST: when a lock is enqueued by some process, but cannot be
40  *   granted immediately due to other conflicting locks on the same resource,
41  *   the completion AST is sent to notify the caller when the lock is
42  *   eventually granted
43  *
44  * - blocking AST: when a lock is granted to some process, if another process
45  *   enqueues a conflicting (blocking) lock on a resource, a blocking AST is
46  *   sent to notify the holder(s) of the lock(s) of the conflicting lock
47  *   request. The lock holder(s) must release their lock(s) on that resource in
48  *   a timely manner or be evicted by the server.
49  *
50  * - glimpse AST: this is used when a process wants information about a lock
51  *   (i.e. the lock value block (LVB)) but does not necessarily require holding
52  *   the lock. If the resource is locked, the lock holder(s) are sent glimpse
53  *   ASTs and the LVB is returned to the caller, and lock holder(s) may CANCEL
54  *   their lock(s) if they are idle. If the resource is not locked, the server
55  *   may grant the lock.
56  */
57
58 #define DEBUG_SUBSYSTEM S_LDLM
59
60 #include <lustre_errno.h>
61 #include <lustre_dlm.h>
62 #include <obd_class.h>
63 #include <obd.h>
64
65 #include "ldlm_internal.h"
66
67 unsigned int ldlm_enqueue_min = OBD_TIMEOUT_DEFAULT;
68 module_param(ldlm_enqueue_min, uint, 0644);
69 MODULE_PARM_DESC(ldlm_enqueue_min, "lock enqueue timeout minimum");
70
71 /* in client side, whether the cached locks will be canceled before replay */
72 unsigned int ldlm_cancel_unused_locks_before_replay = 1;
73
74 static void interrupted_completion_wait(void *data)
75 {
76 }
77
78 struct lock_wait_data {
79         struct ldlm_lock *lwd_lock;
80         __u32             lwd_conn_cnt;
81 };
82
83 struct ldlm_async_args {
84         struct lustre_handle lock_handle;
85 };
86
87 /**
88  * ldlm_request_bufsize
89  *
90  * If opcode=LDLM_ENQUEUE, 1 slot is already occupied,
91  * LDLM_LOCKREQ_HANDLE -1 slots are available.
92  * Otherwise, LDLM_LOCKREQ_HANDLE slots are available.
93  *
94  * \param[in] count
95  * \param[in] type
96  *
97  * \retval size of the request buffer
98  */
99 int ldlm_request_bufsize(int count, int type)
100 {
101         int avail = LDLM_LOCKREQ_HANDLES;
102
103         if (type == LDLM_ENQUEUE)
104                 avail -= LDLM_ENQUEUE_CANCEL_OFF;
105
106         if (count > avail)
107                 avail = (count - avail) * sizeof(struct lustre_handle);
108         else
109                 avail = 0;
110
111         return sizeof(struct ldlm_request) + avail;
112 }
113
114 int ldlm_expired_completion_wait(void *data)
115 {
116         struct lock_wait_data *lwd = data;
117         struct ldlm_lock *lock = lwd->lwd_lock;
118         struct obd_import *imp;
119         struct obd_device *obd;
120
121         ENTRY;
122         if (lock->l_conn_export == NULL) {
123                 static time64_t next_dump, last_dump;
124
125                 LDLM_ERROR(lock,
126                            "lock timed out (enqueued at %lld, %llds ago); not entering recovery in server code, just going back to sleep",
127                            (s64)lock->l_activity,
128                            (s64)(ktime_get_real_seconds() -
129                                  lock->l_activity));
130                 if (ktime_get_seconds() > next_dump) {
131                         last_dump = next_dump;
132                         next_dump = ktime_get_seconds() + 300;
133                         ldlm_namespace_dump(D_DLMTRACE,
134                                             ldlm_lock_to_ns(lock));
135                         if (last_dump == 0)
136                                 libcfs_debug_dumplog();
137                 }
138                 RETURN(0);
139         }
140
141         obd = lock->l_conn_export->exp_obd;
142         imp = obd->u.cli.cl_import;
143         ptlrpc_fail_import(imp, lwd->lwd_conn_cnt);
144         LDLM_ERROR(lock,
145                    "lock timed out (enqueued at %lld, %llds ago), entering recovery for %s@%s",
146                    (s64)lock->l_activity,
147                    (s64)(ktime_get_real_seconds() - lock->l_activity),
148                    obd2cli_tgt(obd), imp->imp_connection->c_remote_uuid.uuid);
149
150         RETURN(0);
151 }
152
153 int is_granted_or_cancelled_nolock(struct ldlm_lock *lock)
154 {
155         int ret = 0;
156
157         check_res_locked(lock->l_resource);
158         if (ldlm_is_granted(lock) && !ldlm_is_cp_reqd(lock))
159                 ret = 1;
160         else if (ldlm_is_failed(lock) || ldlm_is_cancel(lock))
161                 ret = 1;
162         return ret;
163 }
164 EXPORT_SYMBOL(is_granted_or_cancelled_nolock);
165
166 /**
167  * Calculate the Completion timeout (covering enqueue, BL AST, data flush,
168  * lock cancel, and their replies). Used for lock completion timeout on the
169  * client side.
170  *
171  * \param[in] lock        lock which is waiting the completion callback
172  *
173  * \retval            timeout in seconds to wait for the server reply
174  */
175 /*
176  * We use the same basis for both server side and client side functions
177  * from a single node.
178  */
179 static time64_t ldlm_cp_timeout(struct ldlm_lock *lock)
180 {
181         time64_t timeout;
182
183         if (AT_OFF)
184                 return obd_timeout;
185
186         /*
187          * Wait a long time for enqueue - server may have to callback a
188          * lock from another client.  Server will evict the other client if it
189          * doesn't respond reasonably, and then give us the lock.
190          */
191         timeout = at_get(ldlm_lock_to_ns_at(lock));
192         return max(3 * timeout, (time64_t) ldlm_enqueue_min);
193 }
194
195 /**
196  * Helper function for ldlm_completion_ast(), updating timings when lock is
197  * actually granted.
198  */
199 static int ldlm_completion_tail(struct ldlm_lock *lock, void *data)
200 {
201         time64_t delay;
202         int result = 0;
203
204         if (ldlm_is_destroyed(lock) || ldlm_is_failed(lock)) {
205                 LDLM_DEBUG(lock, "client-side enqueue: destroyed");
206                 result = -EIO;
207         } else if (data == NULL) {
208                 LDLM_DEBUG(lock, "client-side enqueue: granted");
209         } else {
210                 /* Take into AT only CP RPC, not immediately granted locks */
211                 delay = ktime_get_real_seconds() - lock->l_activity;
212                 LDLM_DEBUG(lock, "client-side enqueue: granted after %llds",
213                            (s64)delay);
214
215                 /* Update our time estimate */
216                 at_measured(ldlm_lock_to_ns_at(lock), delay);
217         }
218         return result;
219 }
220
221 /**
222  * Implementation of ->l_completion_ast() for a client, that doesn't wait
223  * until lock is granted. Suitable for locks enqueued through ptlrpcd, of
224  * other threads that cannot block for long.
225  */
226 int ldlm_completion_ast_async(struct ldlm_lock *lock, __u64 flags, void *data)
227 {
228         ENTRY;
229
230         if (flags == LDLM_FL_WAIT_NOREPROC) {
231                 LDLM_DEBUG(lock, "client-side enqueue waiting on pending lock");
232                 RETURN(0);
233         }
234
235         if (!(flags & LDLM_FL_BLOCKED_MASK)) {
236                 wake_up(&lock->l_waitq);
237                 RETURN(ldlm_completion_tail(lock, data));
238         }
239
240         LDLM_DEBUG(lock,
241                    "client-side enqueue returned a blocked lock, going forward");
242         ldlm_reprocess_all(lock->l_resource, NULL);
243         RETURN(0);
244 }
245 EXPORT_SYMBOL(ldlm_completion_ast_async);
246
247 /**
248  * Generic LDLM "completion" AST. This is called in several cases:
249  *
250  *     - when a reply to an ENQUEUE RPC is received from the server
251  *       (ldlm_cli_enqueue_fini()). Lock might be granted or not granted at
252  *       this point (determined by flags);
253  *
254  *     - when LDLM_CP_CALLBACK RPC comes to client to notify it that lock has
255  *       been granted;
256  *
257  *     - when ldlm_lock_match(LDLM_FL_LVB_READY) is about to wait until lock
258  *       gets correct lvb;
259  *
260  *     - to force all locks when resource is destroyed (cleanup_resource());
261  *
262  * If lock is not granted in the first case, this function waits until second
263  * or penultimate cases happen in some other thread.
264  *
265  */
266 int ldlm_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data)
267 {
268         /* XXX ALLOCATE - 160 bytes */
269         struct lock_wait_data lwd;
270         struct obd_device *obd;
271         struct obd_import *imp = NULL;
272         struct l_wait_info lwi;
273         time64_t timeout;
274         int rc = 0;
275
276         ENTRY;
277
278         if (flags == LDLM_FL_WAIT_NOREPROC) {
279                 LDLM_DEBUG(lock, "client-side enqueue waiting on pending lock");
280                 goto noreproc;
281         }
282
283         if (!(flags & LDLM_FL_BLOCKED_MASK)) {
284                 wake_up(&lock->l_waitq);
285                 RETURN(0);
286         }
287
288         LDLM_DEBUG(lock, "client-side enqueue returned a blocked locksleeping");
289
290 noreproc:
291
292         obd = class_exp2obd(lock->l_conn_export);
293
294         /* if this is a local lock, then there is no import */
295         if (obd != NULL)
296                 imp = obd->u.cli.cl_import;
297
298         timeout = ldlm_cp_timeout(lock);
299
300         lwd.lwd_lock = lock;
301         lock->l_activity = ktime_get_real_seconds();
302
303         if (ldlm_is_no_timeout(lock)) {
304                 LDLM_DEBUG(lock, "waiting indefinitely because of NO_TIMEOUT");
305                 lwi = LWI_INTR(interrupted_completion_wait, &lwd);
306         } else {
307                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(timeout),
308                                        ldlm_expired_completion_wait,
309                                        interrupted_completion_wait, &lwd);
310         }
311
312         if (imp != NULL) {
313                 spin_lock(&imp->imp_lock);
314                 lwd.lwd_conn_cnt = imp->imp_conn_cnt;
315                 spin_unlock(&imp->imp_lock);
316         }
317
318         if (ns_is_client(ldlm_lock_to_ns(lock)) &&
319             OBD_FAIL_CHECK_RESET(OBD_FAIL_LDLM_INTR_CP_AST,
320                                  OBD_FAIL_LDLM_CP_BL_RACE | OBD_FAIL_ONCE)) {
321                 ldlm_set_fail_loc(lock);
322                 rc = -EINTR;
323         } else {
324                 /* Go to sleep until the lock is granted or cancelled. */
325                 rc = l_wait_event(lock->l_waitq,
326                                   is_granted_or_cancelled(lock), &lwi);
327         }
328
329         if (rc) {
330                 LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)",
331                            rc);
332                 RETURN(rc);
333         }
334
335         RETURN(ldlm_completion_tail(lock, data));
336 }
337 EXPORT_SYMBOL(ldlm_completion_ast);
338
339 /**
340  * A helper to build a blocking AST function
341  *
342  * Perform a common operation for blocking ASTs:
343  * defferred lock cancellation.
344  *
345  * \param lock the lock blocking or canceling AST was called on
346  * \retval 0
347  * \see mdt_blocking_ast
348  * \see ldlm_blocking_ast
349  */
350 int ldlm_blocking_ast_nocheck(struct ldlm_lock *lock)
351 {
352         int do_ast;
353
354         ENTRY;
355
356         ldlm_set_cbpending(lock);
357         do_ast = (!lock->l_readers && !lock->l_writers);
358         unlock_res_and_lock(lock);
359
360         if (do_ast) {
361                 struct lustre_handle lockh;
362                 int rc;
363
364                 LDLM_DEBUG(lock, "already unused, calling ldlm_cli_cancel");
365                 ldlm_lock2handle(lock, &lockh);
366                 rc = ldlm_cli_cancel(&lockh, LCF_ASYNC);
367                 if (rc < 0)
368                         CERROR("ldlm_cli_cancel: %d\n", rc);
369         } else {
370                 LDLM_DEBUG(lock,
371                            "Lock still has references, will be cancelled later");
372         }
373         RETURN(0);
374 }
375 EXPORT_SYMBOL(ldlm_blocking_ast_nocheck);
376
377 /**
378  * Server blocking AST
379  *
380  * ->l_blocking_ast() callback for LDLM locks acquired by server-side
381  * OBDs.
382  *
383  * \param lock the lock which blocks a request or cancelling lock
384  * \param desc unused
385  * \param data unused
386  * \param flag indicates whether this cancelling or blocking callback
387  * \retval 0
388  * \see ldlm_blocking_ast_nocheck
389  */
390 int ldlm_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
391                       void *data, int flag)
392 {
393         ENTRY;
394
395         if (flag == LDLM_CB_CANCELING) {
396                 /* Don't need to do anything here. */
397                 RETURN(0);
398         }
399
400         lock_res_and_lock(lock);
401         /*
402          * Get this: if ldlm_blocking_ast is racing with intent_policy, such
403          * that ldlm_blocking_ast is called just before intent_policy method
404          * takes the lr_lock, then by the time we get the lock, we might not
405          * be the correct blocking function anymore.  So check, and return
406          * early, if so.
407          */
408         if (lock->l_blocking_ast != ldlm_blocking_ast) {
409                 unlock_res_and_lock(lock);
410                 RETURN(0);
411         }
412         RETURN(ldlm_blocking_ast_nocheck(lock));
413 }
414 EXPORT_SYMBOL(ldlm_blocking_ast);
415
416 /**
417  * Implements ldlm_lock::l_glimpse_ast for extent locks acquired on the server.
418  *
419  * Returning -ELDLM_NO_LOCK_DATA actually works, but the reason for that is
420  * rather subtle: with OST-side locking, it may so happen that _all_ extent
421  * locks are held by the OST. If client wants to obtain the current file size
422  * it calls ll_glimpse_size(), and (as all locks are held only on the server),
423  * this dummy glimpse callback fires and does nothing. The client still
424  * receives the correct file size due to the following fragment of code in
425  * ldlm_cb_interpret():
426  *
427  *      if (rc == -ELDLM_NO_LOCK_DATA) {
428  *              LDLM_DEBUG(lock, "lost race - client has a lock but no"
429  *                         "inode");
430  *              ldlm_res_lvbo_update(lock->l_resource, NULL, 1);
431  *      }
432  *
433  * That is, after the glimpse returns this error, ofd_lvbo_update() is called
434  * and returns the updated file attributes from the inode to the client.
435  *
436  * See also comment in ofd_intent_policy() on why servers must set a non-NULL
437  * l_glimpse_ast when grabbing DLM locks.  Otherwise, the server will assume
438  * that the object is in the process of being destroyed.
439  *
440  * \param[in] lock      DLM lock being glimpsed, unused
441  * \param[in] reqp      pointer to ptlrpc_request, unused
442  *
443  * \retval              -ELDLM_NO_LOCK_DATA to get attributes from disk object
444  */
445 int ldlm_glimpse_ast(struct ldlm_lock *lock, void *reqp)
446 {
447         return -ELDLM_NO_LOCK_DATA;
448 }
449
450 /**
451  * Enqueue a local lock (typically on a server).
452  */
453 int ldlm_cli_enqueue_local(const struct lu_env *env,
454                            struct ldlm_namespace *ns,
455                            const struct ldlm_res_id *res_id,
456                            enum ldlm_type type, union ldlm_policy_data *policy,
457                            enum ldlm_mode mode, __u64 *flags,
458                            ldlm_blocking_callback blocking,
459                            ldlm_completion_callback completion,
460                            ldlm_glimpse_callback glimpse,
461                            void *data, __u32 lvb_len, enum lvb_type lvb_type,
462                            const __u64 *client_cookie,
463                            struct lustre_handle *lockh)
464 {
465         struct ldlm_lock *lock;
466         int err;
467         const struct ldlm_callback_suite cbs = { .lcs_completion = completion,
468                                                  .lcs_blocking   = blocking,
469                                                  .lcs_glimpse    = glimpse,
470         };
471
472         ENTRY;
473
474         LASSERT(!(*flags & LDLM_FL_REPLAY));
475         if (unlikely(ns_is_client(ns))) {
476                 CERROR("Trying to enqueue local lock in a shadow namespace\n");
477                 LBUG();
478         }
479
480         lock = ldlm_lock_create(ns, res_id, type, mode, &cbs, data, lvb_len,
481                                 lvb_type);
482         if (IS_ERR(lock))
483                 GOTO(out_nolock, err = PTR_ERR(lock));
484
485         err = ldlm_lvbo_init(lock->l_resource);
486         if (err < 0) {
487                 LDLM_ERROR(lock, "delayed lvb init failed (rc %d)", err);
488                 ldlm_lock_destroy_nolock(lock);
489                 GOTO(out, err);
490         }
491
492         ldlm_lock2handle(lock, lockh);
493
494         /*
495          * NB: we don't have any lock now (lock_res_and_lock)
496          * because it's a new lock
497          */
498         ldlm_lock_addref_internal_nolock(lock, mode);
499         ldlm_set_local(lock);
500         if (*flags & LDLM_FL_ATOMIC_CB)
501                 ldlm_set_atomic_cb(lock);
502
503         if (*flags & LDLM_FL_CANCEL_ON_BLOCK)
504                 ldlm_set_cancel_on_block(lock);
505
506         if (policy != NULL)
507                 lock->l_policy_data = *policy;
508         if (client_cookie != NULL)
509                 lock->l_client_cookie = *client_cookie;
510         if (type == LDLM_EXTENT) {
511                 /* extent lock without policy is a bug */
512                 if (policy == NULL)
513                         LBUG();
514
515                 lock->l_req_extent = policy->l_extent;
516         }
517
518         err = ldlm_lock_enqueue(env, ns, &lock, policy, flags);
519         if (unlikely(err != ELDLM_OK))
520                 GOTO(out, err);
521
522         if (policy != NULL)
523                 *policy = lock->l_policy_data;
524
525         if (lock->l_completion_ast)
526                 lock->l_completion_ast(lock, *flags, NULL);
527
528         LDLM_DEBUG(lock, "client-side local enqueue handler, new lock created");
529         EXIT;
530  out:
531         LDLM_LOCK_RELEASE(lock);
532  out_nolock:
533         return err;
534 }
535 EXPORT_SYMBOL(ldlm_cli_enqueue_local);
536
537 static void failed_lock_cleanup(struct ldlm_namespace *ns,
538                                 struct ldlm_lock *lock, int mode)
539 {
540         int need_cancel = 0;
541
542         /* Set a flag to prevent us from sending a CANCEL (b=407) */
543         lock_res_and_lock(lock);
544         /* Check that lock is not granted or failed, we might race. */
545         if (!ldlm_is_granted(lock) && !ldlm_is_failed(lock)) {
546                 /*
547                  * Make sure that this lock will not be found by raced
548                  * bl_ast and -EINVAL reply is sent to server anyways.
549                  * b=17645
550                  */
551                 lock->l_flags |= LDLM_FL_LOCAL_ONLY | LDLM_FL_FAILED |
552                                  LDLM_FL_ATOMIC_CB | LDLM_FL_CBPENDING;
553                 need_cancel = 1;
554         }
555         unlock_res_and_lock(lock);
556
557         if (need_cancel)
558                 LDLM_DEBUG(lock,
559                            "setting FL_LOCAL_ONLY | LDLM_FL_FAILED | LDLM_FL_ATOMIC_CB | LDLM_FL_CBPENDING");
560         else
561                 LDLM_DEBUG(lock, "lock was granted or failed in race");
562
563         /*
564          * XXX - HACK because we shouldn't call ldlm_lock_destroy()
565          *       from llite/file.c/ll_file_flock().
566          */
567         /*
568          * This code makes for the fact that we do not have blocking handler on
569          * a client for flock locks. As such this is the place where we must
570          * completely kill failed locks. (interrupted and those that
571          * were waiting to be granted when server evicted us.
572          */
573         if (lock->l_resource->lr_type == LDLM_FLOCK) {
574                 lock_res_and_lock(lock);
575                 if (!ldlm_is_destroyed(lock)) {
576                         ldlm_resource_unlink_lock(lock);
577                         ldlm_lock_decref_internal_nolock(lock, mode);
578                         ldlm_lock_destroy_nolock(lock);
579                 }
580                 unlock_res_and_lock(lock);
581         } else {
582                 ldlm_lock_decref_internal(lock, mode);
583         }
584 }
585
586 static bool ldlm_request_slot_needed(enum ldlm_type type)
587 {
588         return type == LDLM_FLOCK || type == LDLM_IBITS;
589 }
590
591 /**
592  * Finishing portion of client lock enqueue code.
593  *
594  * Called after receiving reply from server.
595  */
596 int ldlm_cli_enqueue_fini(struct obd_export *exp, struct ptlrpc_request *req,
597                           enum ldlm_type type, __u8 with_policy,
598                           enum ldlm_mode mode, __u64 *flags, void *lvb,
599                           __u32 lvb_len, const struct lustre_handle *lockh,
600                           int rc)
601 {
602         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
603         const struct lu_env *env = NULL;
604         int is_replay = *flags & LDLM_FL_REPLAY;
605         struct ldlm_lock *lock;
606         struct ldlm_reply *reply;
607         int cleanup_phase = 1;
608
609         ENTRY;
610
611         if (ldlm_request_slot_needed(type))
612                 obd_put_request_slot(&req->rq_import->imp_obd->u.cli);
613
614         ptlrpc_put_mod_rpc_slot(req);
615
616         if (req && req->rq_svc_thread)
617                 env = req->rq_svc_thread->t_env;
618
619         lock = ldlm_handle2lock(lockh);
620         /* ldlm_cli_enqueue is holding a reference on this lock. */
621         if (!lock) {
622                 LASSERT(type == LDLM_FLOCK);
623                 RETURN(-ENOLCK);
624         }
625
626         LASSERTF(ergo(lvb_len != 0, lvb_len == lock->l_lvb_len),
627                  "lvb_len = %d, l_lvb_len = %d\n", lvb_len, lock->l_lvb_len);
628
629         if (rc != ELDLM_OK) {
630                 LASSERT(!is_replay);
631                 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
632                            rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
633
634                 if (rc != ELDLM_LOCK_ABORTED)
635                         GOTO(cleanup, rc);
636         }
637
638         /* Before we return, swab the reply */
639         reply = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
640         if (reply == NULL)
641                 GOTO(cleanup, rc = -EPROTO);
642
643         if (lvb_len > 0) {
644                 int size = 0;
645
646                 size = req_capsule_get_size(&req->rq_pill, &RMF_DLM_LVB,
647                                             RCL_SERVER);
648                 if (size < 0) {
649                         LDLM_ERROR(lock, "Fail to get lvb_len, rc = %d", size);
650                         GOTO(cleanup, rc = size);
651                 } else if (unlikely(size > lvb_len)) {
652                         LDLM_ERROR(lock,
653                                    "Replied LVB is larger than expectation, expected = %d, replied = %d",
654                                    lvb_len, size);
655                         GOTO(cleanup, rc = -EINVAL);
656                 }
657                 lvb_len = size;
658         }
659
660         if (rc == ELDLM_LOCK_ABORTED) {
661                 if (lvb_len > 0 && lvb != NULL)
662                         rc = ldlm_fill_lvb(lock, &req->rq_pill, RCL_SERVER,
663                                            lvb, lvb_len);
664                 GOTO(cleanup, rc = rc ? : ELDLM_LOCK_ABORTED);
665         }
666
667         /* lock enqueued on the server */
668         cleanup_phase = 0;
669
670         lock_res_and_lock(lock);
671         /* Key change rehash lock in per-export hash with new key */
672         if (exp->exp_lock_hash) {
673                 /*
674                  * In the function below, .hs_keycmp resolves to
675                  * ldlm_export_lock_keycmp()
676                  */
677                 /* coverity[overrun-buffer-val] */
678                 cfs_hash_rehash_key(exp->exp_lock_hash,
679                                     &lock->l_remote_handle,
680                                     &reply->lock_handle,
681                                     &lock->l_exp_hash);
682         } else {
683                 lock->l_remote_handle = reply->lock_handle;
684         }
685
686         *flags = ldlm_flags_from_wire(reply->lock_flags);
687         lock->l_flags |= ldlm_flags_from_wire(reply->lock_flags &
688                                               LDLM_FL_INHERIT_MASK);
689         unlock_res_and_lock(lock);
690
691         CDEBUG(D_INFO, "local: %p, remote cookie: %#llx, flags: %#llx\n",
692                lock, reply->lock_handle.cookie, *flags);
693
694         /*
695          * If enqueue returned a blocked lock but the completion handler has
696          * already run, then it fixed up the resource and we don't need to do it
697          * again.
698          */
699         if ((*flags) & LDLM_FL_LOCK_CHANGED) {
700                 int newmode = reply->lock_desc.l_req_mode;
701
702                 LASSERT(!is_replay);
703                 if (newmode && newmode != lock->l_req_mode) {
704                         LDLM_DEBUG(lock, "server returned different mode %s",
705                                    ldlm_lockname[newmode]);
706                         lock->l_req_mode = newmode;
707                 }
708
709                 if (!ldlm_res_eq(&reply->lock_desc.l_resource.lr_name,
710                                  &lock->l_resource->lr_name)) {
711                         CDEBUG(D_INFO,
712                                "remote intent success, locking "DLDLMRES", instead of "DLDLMRES"\n",
713                                PLDLMRES(&reply->lock_desc.l_resource),
714                                PLDLMRES(lock->l_resource));
715
716                         rc = ldlm_lock_change_resource(ns, lock,
717                                         &reply->lock_desc.l_resource.lr_name);
718                         if (rc || lock->l_resource == NULL)
719                                 GOTO(cleanup, rc = -ENOMEM);
720                         LDLM_DEBUG(lock, "client-side enqueue, new resource");
721                 }
722
723                 if (with_policy) {
724                         /* We assume lock type cannot change on server*/
725                         ldlm_convert_policy_to_local(exp,
726                                                 lock->l_resource->lr_type,
727                                                 &reply->lock_desc.l_policy_data,
728                                                 &lock->l_policy_data);
729                 }
730
731                 if (type != LDLM_PLAIN)
732                         LDLM_DEBUG(lock,
733                                    "client-side enqueue, new policy data");
734         }
735
736         if ((*flags) & LDLM_FL_AST_SENT) {
737                 lock_res_and_lock(lock);
738                 ldlm_bl_desc2lock(&reply->lock_desc, lock);
739                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_BL_AST;
740                 unlock_res_and_lock(lock);
741                 LDLM_DEBUG(lock, "enqueue reply includes blocking AST");
742         }
743
744         /*
745          * If the lock has already been granted by a completion AST, don't
746          * clobber the LVB with an older one.
747          */
748         if (lvb_len > 0) {
749                 /*
750                  * We must lock or a racing completion might update lvb without
751                  * letting us know and we'll clobber the correct value.
752                  * Cannot unlock after the check either, a that still leaves
753                  * a tiny window for completion to get in
754                  */
755                 lock_res_and_lock(lock);
756                 if (!ldlm_is_granted(lock))
757                         rc = ldlm_fill_lvb(lock, &req->rq_pill, RCL_SERVER,
758                                            lock->l_lvb_data, lvb_len);
759                 unlock_res_and_lock(lock);
760                 if (rc < 0) {
761                         cleanup_phase = 1;
762                         GOTO(cleanup, rc);
763                 }
764         }
765
766         if (!is_replay) {
767                 rc = ldlm_lock_enqueue(env, ns, &lock, NULL, flags);
768                 if (lock->l_completion_ast != NULL) {
769                         int err = lock->l_completion_ast(lock, *flags, NULL);
770
771                         if (!rc)
772                                 rc = err;
773                         if (rc)
774                                 cleanup_phase = 1;
775                 }
776         }
777
778         if (lvb_len > 0 && lvb != NULL) {
779                 /*
780                  * Copy the LVB here, and not earlier, because the completion
781                  * AST (if any) can override what we got in the reply
782                  */
783                 memcpy(lvb, lock->l_lvb_data, lvb_len);
784         }
785
786         LDLM_DEBUG(lock, "client-side enqueue END");
787         EXIT;
788 cleanup:
789         if (cleanup_phase == 1 && rc)
790                 failed_lock_cleanup(ns, lock, mode);
791         /* Put lock 2 times, the second reference is held by ldlm_cli_enqueue */
792         LDLM_LOCK_PUT(lock);
793         LDLM_LOCK_RELEASE(lock);
794         return rc;
795 }
796 EXPORT_SYMBOL(ldlm_cli_enqueue_fini);
797
798 /**
799  * Estimate number of lock handles that would fit into request of given
800  * size.  PAGE_SIZE-512 is to allow TCP/IP and LNET headers to fit into
801  * a single page on the send/receive side. XXX: 512 should be changed to
802  * more adequate value.
803  */
804 static inline int ldlm_req_handles_avail(int req_size, int off)
805 {
806         int avail;
807
808         avail = min_t(int, LDLM_MAXREQSIZE, PAGE_SIZE - 512) - req_size;
809         if (likely(avail >= 0))
810                 avail /= (int)sizeof(struct lustre_handle);
811         else
812                 avail = 0;
813         avail += LDLM_LOCKREQ_HANDLES - off;
814
815         return avail;
816 }
817
818 static inline int ldlm_capsule_handles_avail(struct req_capsule *pill,
819                                              enum req_location loc,
820                                              int off)
821 {
822         __u32 size = req_capsule_msg_size(pill, loc);
823
824         return ldlm_req_handles_avail(size, off);
825 }
826
827 static inline int ldlm_format_handles_avail(struct obd_import *imp,
828                                             const struct req_format *fmt,
829                                             enum req_location loc, int off)
830 {
831         __u32 size = req_capsule_fmt_size(imp->imp_msg_magic, fmt, loc);
832
833         return ldlm_req_handles_avail(size, off);
834 }
835
836 /**
837  * Cancel LRU locks and pack them into the enqueue request. Pack there the given
838  * \a count locks in \a cancels.
839  *
840  * This is to be called by functions preparing their own requests that
841  * might contain lists of locks to cancel in addition to actual operation
842  * that needs to be performed.
843  */
844 int ldlm_prep_elc_req(struct obd_export *exp, struct ptlrpc_request *req,
845                       int version, int opc, int canceloff,
846                       struct list_head *cancels, int count)
847 {
848         struct ldlm_namespace   *ns = exp->exp_obd->obd_namespace;
849         struct req_capsule      *pill = &req->rq_pill;
850         struct ldlm_request     *dlm = NULL;
851         LIST_HEAD(head);
852         enum ldlm_lru_flags lru_flags;
853         int avail, to_free, pack = 0;
854         int rc;
855
856         ENTRY;
857
858         if (cancels == NULL)
859                 cancels = &head;
860         if (ns_connect_cancelset(ns)) {
861                 /* Estimate the amount of available space in the request. */
862                 req_capsule_filled_sizes(pill, RCL_CLIENT);
863                 avail = ldlm_capsule_handles_avail(pill, RCL_CLIENT, canceloff);
864
865                 lru_flags = LDLM_LRU_FLAG_NO_WAIT | (ns_connect_lru_resize(ns) ?
866                         LDLM_LRU_FLAG_LRUR : LDLM_LRU_FLAG_AGED);
867                 to_free = !ns_connect_lru_resize(ns) &&
868                         opc == LDLM_ENQUEUE ? 1 : 0;
869
870                 /*
871                  * Cancel LRU locks here _only_ if the server supports
872                  * EARLY_CANCEL. Otherwise we have to send extra CANCEL
873                  * RPC, which will make us slower.
874                  */
875                 if (avail > count)
876                         count += ldlm_cancel_lru_local(ns, cancels, to_free,
877                                                        avail - count, 0,
878                                                        lru_flags);
879                 if (avail > count)
880                         pack = count;
881                 else
882                         pack = avail;
883                 req_capsule_set_size(pill, &RMF_DLM_REQ, RCL_CLIENT,
884                                      ldlm_request_bufsize(pack, opc));
885         }
886
887         rc = ptlrpc_request_pack(req, version, opc);
888         if (rc) {
889                 ldlm_lock_list_put(cancels, l_bl_ast, count);
890                 RETURN(rc);
891         }
892
893         if (ns_connect_cancelset(ns)) {
894                 if (canceloff) {
895                         dlm = req_capsule_client_get(pill, &RMF_DLM_REQ);
896                         LASSERT(dlm);
897                         /*
898                          * Skip first lock handler in ldlm_request_pack(),
899                          * this method will increment @lock_count according
900                          * to the lock handle amount actually written to
901                          * the buffer.
902                          */
903                         dlm->lock_count = canceloff;
904                 }
905                 /* Pack into the request @pack lock handles. */
906                 ldlm_cli_cancel_list(cancels, pack, req, 0);
907                 /* Prepare and send separate cancel RPC for others. */
908                 ldlm_cli_cancel_list(cancels, count - pack, NULL, 0);
909         } else {
910                 ldlm_lock_list_put(cancels, l_bl_ast, count);
911         }
912         RETURN(0);
913 }
914 EXPORT_SYMBOL(ldlm_prep_elc_req);
915
916 int ldlm_prep_enqueue_req(struct obd_export *exp, struct ptlrpc_request *req,
917                           struct list_head *cancels, int count)
918 {
919         return ldlm_prep_elc_req(exp, req, LUSTRE_DLM_VERSION, LDLM_ENQUEUE,
920                                  LDLM_ENQUEUE_CANCEL_OFF, cancels, count);
921 }
922 EXPORT_SYMBOL(ldlm_prep_enqueue_req);
923
924 struct ptlrpc_request *ldlm_enqueue_pack(struct obd_export *exp, int lvb_len)
925 {
926         struct ptlrpc_request *req;
927         int rc;
928
929         ENTRY;
930
931         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LDLM_ENQUEUE);
932         if (req == NULL)
933                 RETURN(ERR_PTR(-ENOMEM));
934
935         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
936         if (rc) {
937                 ptlrpc_request_free(req);
938                 RETURN(ERR_PTR(rc));
939         }
940
941         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, lvb_len);
942         ptlrpc_request_set_replen(req);
943         RETURN(req);
944 }
945 EXPORT_SYMBOL(ldlm_enqueue_pack);
946
947 /**
948  * Client-side lock enqueue.
949  *
950  * If a request has some specific initialisation it is passed in \a reqp,
951  * otherwise it is created in ldlm_cli_enqueue.
952  *
953  * Supports sync and async requests, pass \a async flag accordingly. If a
954  * request was created in ldlm_cli_enqueue and it is the async request,
955  * pass it to the caller in \a reqp.
956  */
957 int ldlm_cli_enqueue(struct obd_export *exp, struct ptlrpc_request **reqp,
958                      struct ldlm_enqueue_info *einfo,
959                      const struct ldlm_res_id *res_id,
960                      union ldlm_policy_data const *policy, __u64 *flags,
961                      void *lvb, __u32 lvb_len, enum lvb_type lvb_type,
962                      struct lustre_handle *lockh, int async)
963 {
964         struct ldlm_namespace *ns;
965         struct ldlm_lock      *lock;
966         struct ldlm_request   *body;
967         int                    is_replay = *flags & LDLM_FL_REPLAY;
968         int                    req_passed_in = 1;
969         int                    rc, err;
970         struct ptlrpc_request *req;
971
972         ENTRY;
973
974         LASSERT(exp != NULL);
975
976         ns = exp->exp_obd->obd_namespace;
977
978         /*
979          * If we're replaying this lock, just check some invariants.
980          * If we're creating a new lock, get everything all setup nice.
981          */
982         if (is_replay) {
983                 lock = ldlm_handle2lock_long(lockh, 0);
984                 LASSERT(lock != NULL);
985                 LDLM_DEBUG(lock, "client-side enqueue START");
986                 LASSERT(exp == lock->l_conn_export);
987         } else {
988                 const struct ldlm_callback_suite cbs = {
989                         .lcs_completion = einfo->ei_cb_cp,
990                         .lcs_blocking   = einfo->ei_cb_bl,
991                         .lcs_glimpse    = einfo->ei_cb_gl
992                 };
993                 lock = ldlm_lock_create(ns, res_id, einfo->ei_type,
994                                         einfo->ei_mode, &cbs, einfo->ei_cbdata,
995                                         lvb_len, lvb_type);
996                 if (IS_ERR(lock))
997                         RETURN(PTR_ERR(lock));
998
999                 if (einfo->ei_cb_created)
1000                         einfo->ei_cb_created(lock);
1001
1002                 /* for the local lock, add the reference */
1003                 ldlm_lock_addref_internal(lock, einfo->ei_mode);
1004                 ldlm_lock2handle(lock, lockh);
1005                 if (policy != NULL)
1006                         lock->l_policy_data = *policy;
1007
1008                 if (einfo->ei_type == LDLM_EXTENT) {
1009                         /* extent lock without policy is a bug */
1010                         if (policy == NULL)
1011                                 LBUG();
1012
1013                         lock->l_req_extent = policy->l_extent;
1014                 }
1015                 LDLM_DEBUG(lock, "client-side enqueue START, flags %#llx",
1016                            *flags);
1017         }
1018
1019         lock->l_conn_export = exp;
1020         lock->l_export = NULL;
1021         lock->l_blocking_ast = einfo->ei_cb_bl;
1022         lock->l_flags |= (*flags & (LDLM_FL_NO_LRU | LDLM_FL_EXCL |
1023                                     LDLM_FL_ATOMIC_CB));
1024         lock->l_activity = ktime_get_real_seconds();
1025
1026         /* lock not sent to server yet */
1027         if (reqp == NULL || *reqp == NULL) {
1028                 req = ldlm_enqueue_pack(exp, lvb_len);
1029                 if (IS_ERR(req)) {
1030                         failed_lock_cleanup(ns, lock, einfo->ei_mode);
1031                         LDLM_LOCK_RELEASE(lock);
1032                         RETURN(PTR_ERR(req));
1033                 }
1034
1035                 req_passed_in = 0;
1036                 if (reqp)
1037                         *reqp = req;
1038         } else {
1039                 int len;
1040
1041                 req = *reqp;
1042                 len = req_capsule_get_size(&req->rq_pill, &RMF_DLM_REQ,
1043                                            RCL_CLIENT);
1044                 LASSERTF(len >= sizeof(*body), "buflen[%d] = %d, not %d\n",
1045                          DLM_LOCKREQ_OFF, len, (int)sizeof(*body));
1046         }
1047
1048         if (*flags & LDLM_FL_NDELAY) {
1049                 DEBUG_REQ(D_DLMTRACE, req, "enqueue lock with no delay");
1050                 req->rq_no_resend = req->rq_no_delay = 1;
1051                 /*
1052                  * probably set a shorter timeout value and handle ETIMEDOUT
1053                  * in osc_lock_upcall() correctly
1054                  */
1055                 /* lustre_msg_set_timeout(req, req->rq_timeout / 2); */
1056         }
1057
1058         /* Dump lock data into the request buffer */
1059         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1060         ldlm_lock2desc(lock, &body->lock_desc);
1061         body->lock_flags = ldlm_flags_to_wire(*flags);
1062         body->lock_handle[0] = *lockh;
1063
1064         /* extended LDLM opcodes in client stats */
1065         if (exp->exp_obd->obd_svc_stats != NULL) {
1066                 bool glimpse = *flags & LDLM_FL_HAS_INTENT;
1067
1068                 /* OST glimpse has no intent buffer */
1069                 if (req_capsule_has_field(&req->rq_pill, &RMF_LDLM_INTENT,
1070                                           RCL_CLIENT)) {
1071                         struct ldlm_intent *it;
1072
1073                         it = req_capsule_client_get(&req->rq_pill,
1074                                                     &RMF_LDLM_INTENT);
1075                         glimpse = (it && (it->opc == IT_GLIMPSE));
1076                 }
1077
1078                 if (!glimpse)
1079                         ldlm_svc_get_eopc(body, exp->exp_obd->obd_svc_stats);
1080                 else
1081                         lprocfs_counter_incr(exp->exp_obd->obd_svc_stats,
1082                                              PTLRPC_LAST_CNTR +
1083                                              LDLM_GLIMPSE_ENQUEUE);
1084         }
1085
1086         /* It is important to obtain modify RPC slot first (if applicable), so
1087          * that threads that are waiting for a modify RPC slot are not polluting
1088          * our rpcs in flight counter. */
1089
1090         if (einfo->ei_enq_slot)
1091                 ptlrpc_get_mod_rpc_slot(req);
1092
1093         if (ldlm_request_slot_needed(einfo->ei_type)) {
1094                 rc = obd_get_request_slot(&req->rq_import->imp_obd->u.cli);
1095                 if (rc) {
1096                         if (einfo->ei_enq_slot)
1097                                 ptlrpc_put_mod_rpc_slot(req);
1098                         failed_lock_cleanup(ns, lock, einfo->ei_mode);
1099                         LDLM_LOCK_RELEASE(lock);
1100                         GOTO(out, rc);
1101                 }
1102         }
1103
1104         if (async) {
1105                 LASSERT(reqp != NULL);
1106                 RETURN(0);
1107         }
1108
1109         LDLM_DEBUG(lock, "sending request");
1110
1111         rc = ptlrpc_queue_wait(req);
1112
1113         err = ldlm_cli_enqueue_fini(exp, req, einfo->ei_type, policy ? 1 : 0,
1114                                     einfo->ei_mode, flags, lvb, lvb_len,
1115                                     lockh, rc);
1116
1117         /*
1118          * If ldlm_cli_enqueue_fini did not find the lock, we need to free
1119          * one reference that we took
1120          */
1121         if (err == -ENOLCK)
1122                 LDLM_LOCK_RELEASE(lock);
1123         else
1124                 rc = err;
1125
1126 out:
1127         if (!req_passed_in && req != NULL) {
1128                 ptlrpc_req_finished(req);
1129                 if (reqp)
1130                         *reqp = NULL;
1131         }
1132
1133         RETURN(rc);
1134 }
1135 EXPORT_SYMBOL(ldlm_cli_enqueue);
1136
1137 /**
1138  * Client-side IBITS lock convert.
1139  *
1140  * Inform server that lock has been converted instead of canceling.
1141  * Server finishes convert on own side and does reprocess to grant
1142  * all related waiting locks.
1143  *
1144  * Since convert means only ibits downgrading, client doesn't need to
1145  * wait for server reply to finish local converting process so this request
1146  * is made asynchronous.
1147  *
1148  */
1149 int ldlm_cli_convert_req(struct ldlm_lock *lock, __u32 *flags, __u64 new_bits)
1150 {
1151         struct ldlm_request *body;
1152         struct ptlrpc_request *req;
1153         struct obd_export *exp = lock->l_conn_export;
1154
1155         ENTRY;
1156
1157         LASSERT(exp != NULL);
1158
1159         /*
1160          * this is better to check earlier and it is done so already,
1161          * but this check is kept too as final one to issue an error
1162          * if any new code will miss such check.
1163          */
1164         if (!exp_connect_lock_convert(exp)) {
1165                 LDLM_ERROR(lock, "server doesn't support lock convert\n");
1166                 RETURN(-EPROTO);
1167         }
1168
1169         if (lock->l_resource->lr_type != LDLM_IBITS) {
1170                 LDLM_ERROR(lock, "convert works with IBITS locks only.");
1171                 RETURN(-EINVAL);
1172         }
1173
1174         LDLM_DEBUG(lock, "client-side convert");
1175
1176         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
1177                                         &RQF_LDLM_CONVERT, LUSTRE_DLM_VERSION,
1178                                         LDLM_CONVERT);
1179         if (req == NULL)
1180                 RETURN(-ENOMEM);
1181
1182         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1183         body->lock_handle[0] = lock->l_remote_handle;
1184
1185         body->lock_desc.l_req_mode = lock->l_req_mode;
1186         body->lock_desc.l_granted_mode = lock->l_granted_mode;
1187
1188         body->lock_desc.l_policy_data.l_inodebits.bits = new_bits;
1189         body->lock_desc.l_policy_data.l_inodebits.cancel_bits = 0;
1190
1191         body->lock_flags = ldlm_flags_to_wire(*flags);
1192         body->lock_count = 1;
1193
1194         ptlrpc_request_set_replen(req);
1195
1196         /*
1197          * Use cancel portals for convert as well as high-priority handling.
1198          */
1199         req->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
1200         req->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
1201
1202         ptlrpc_at_set_req_timeout(req);
1203
1204         if (exp->exp_obd->obd_svc_stats != NULL)
1205                 lprocfs_counter_incr(exp->exp_obd->obd_svc_stats,
1206                                      LDLM_CONVERT - LDLM_FIRST_OPC);
1207
1208         ptlrpcd_add_req(req);
1209         RETURN(0);
1210 }
1211
1212 /**
1213  * Cancel locks locally.
1214  * Returns:
1215  * \retval LDLM_FL_LOCAL_ONLY if there is no need for a CANCEL RPC to the server
1216  * \retval LDLM_FL_CANCELING otherwise;
1217  * \retval LDLM_FL_BL_AST if there is a need for a separate CANCEL RPC.
1218  */
1219 static __u64 ldlm_cli_cancel_local(struct ldlm_lock *lock)
1220 {
1221         __u64 rc = LDLM_FL_LOCAL_ONLY;
1222
1223         ENTRY;
1224
1225         if (lock->l_conn_export) {
1226                 bool local_only;
1227
1228                 LDLM_DEBUG(lock, "client-side cancel");
1229                 OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_PAUSE_CANCEL_LOCAL,
1230                                  cfs_fail_val);
1231
1232                 /* Set this flag to prevent others from getting new references*/
1233                 lock_res_and_lock(lock);
1234                 ldlm_set_cbpending(lock);
1235                 local_only = !!(lock->l_flags &
1236                                 (LDLM_FL_LOCAL_ONLY|LDLM_FL_CANCEL_ON_BLOCK));
1237                 ldlm_cancel_callback(lock);
1238                 rc = (ldlm_is_bl_ast(lock)) ?
1239                         LDLM_FL_BL_AST : LDLM_FL_CANCELING;
1240                 unlock_res_and_lock(lock);
1241
1242                 if (local_only) {
1243                         CDEBUG(D_DLMTRACE,
1244                                "not sending request (at caller's instruction)\n");
1245                         rc = LDLM_FL_LOCAL_ONLY;
1246                 }
1247                 ldlm_lock_cancel(lock);
1248         } else {
1249                 if (ns_is_client(ldlm_lock_to_ns(lock))) {
1250                         LDLM_ERROR(lock, "Trying to cancel local lock");
1251                         LBUG();
1252                 }
1253                 LDLM_DEBUG(lock, "server-side local cancel");
1254                 ldlm_lock_cancel(lock);
1255                 ldlm_reprocess_all(lock->l_resource, lock);
1256         }
1257
1258         RETURN(rc);
1259 }
1260
1261 /**
1262  * Pack \a count locks in \a head into ldlm_request buffer of request \a req.
1263  */
1264 static void ldlm_cancel_pack(struct ptlrpc_request *req,
1265                              struct list_head *head, int count)
1266 {
1267         struct ldlm_request *dlm;
1268         struct ldlm_lock *lock;
1269         int max, packed = 0;
1270
1271         ENTRY;
1272
1273         dlm = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1274         LASSERT(dlm != NULL);
1275
1276         /* Check the room in the request buffer. */
1277         max = req_capsule_get_size(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT) -
1278                 sizeof(struct ldlm_request);
1279         max /= sizeof(struct lustre_handle);
1280         max += LDLM_LOCKREQ_HANDLES;
1281         LASSERT(max >= dlm->lock_count + count);
1282
1283         /*
1284          * XXX: it would be better to pack lock handles grouped by resource.
1285          * so that the server cancel would call filter_lvbo_update() less
1286          * frequently.
1287          */
1288         list_for_each_entry(lock, head, l_bl_ast) {
1289                 if (!count--)
1290                         break;
1291                 LASSERT(lock->l_conn_export);
1292                 /* Pack the lock handle to the given request buffer. */
1293                 LDLM_DEBUG(lock, "packing");
1294                 dlm->lock_handle[dlm->lock_count++] = lock->l_remote_handle;
1295                 packed++;
1296         }
1297         CDEBUG(D_DLMTRACE, "%d locks packed\n", packed);
1298         EXIT;
1299 }
1300
1301 /**
1302  * Prepare and send a batched cancel RPC. It will include \a count lock
1303  * handles of locks given in \a cancels list.
1304  */
1305 int ldlm_cli_cancel_req(struct obd_export *exp, struct list_head *cancels,
1306                         int count, enum ldlm_cancel_flags flags)
1307 {
1308         struct ptlrpc_request *req = NULL;
1309         struct obd_import *imp;
1310         int free, sent = 0;
1311         int rc = 0;
1312
1313         ENTRY;
1314
1315         LASSERT(exp != NULL);
1316         LASSERT(count > 0);
1317
1318         CFS_FAIL_TIMEOUT(OBD_FAIL_LDLM_PAUSE_CANCEL, cfs_fail_val);
1319
1320         if (CFS_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_RACE))
1321                 RETURN(count);
1322
1323         free = ldlm_format_handles_avail(class_exp2cliimp(exp),
1324                                          &RQF_LDLM_CANCEL, RCL_CLIENT, 0);
1325         if (count > free)
1326                 count = free;
1327
1328         while (1) {
1329                 imp = class_exp2cliimp(exp);
1330                 if (imp == NULL || imp->imp_invalid) {
1331                         CDEBUG(D_DLMTRACE,
1332                                "skipping cancel on invalid import %p\n", imp);
1333                         RETURN(count);
1334                 }
1335
1336                 req = ptlrpc_request_alloc(imp, &RQF_LDLM_CANCEL);
1337                 if (req == NULL)
1338                         GOTO(out, rc = -ENOMEM);
1339
1340                 req_capsule_filled_sizes(&req->rq_pill, RCL_CLIENT);
1341                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT,
1342                                      ldlm_request_bufsize(count, LDLM_CANCEL));
1343
1344                 rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_CANCEL);
1345                 if (rc) {
1346                         ptlrpc_request_free(req);
1347                         GOTO(out, rc);
1348                 }
1349
1350                 /*
1351                  * If OSP want cancel cross-MDT lock, let's not block it in
1352                  * in recovery, otherwise the lock will not released, if
1353                  * the remote target is also in recovery, and it also need
1354                  * this lock, it might cause deadlock.
1355                  */
1356                 if (exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS &&
1357                     exp->exp_obd->obd_lu_dev != NULL &&
1358                     exp->exp_obd->obd_lu_dev->ld_site != NULL) {
1359                         struct lu_device *top_dev;
1360
1361                         top_dev = exp->exp_obd->obd_lu_dev->ld_site->ls_top_dev;
1362                         if (top_dev != NULL &&
1363                             top_dev->ld_obd->obd_recovering)
1364                                 req->rq_allow_replay = 1;
1365                 }
1366
1367                 req->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
1368                 req->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
1369                 ptlrpc_at_set_req_timeout(req);
1370
1371                 ldlm_cancel_pack(req, cancels, count);
1372
1373                 ptlrpc_request_set_replen(req);
1374                 if (flags & LCF_ASYNC) {
1375                         ptlrpcd_add_req(req);
1376                         sent = count;
1377                         GOTO(out, 0);
1378                 }
1379
1380                 rc = ptlrpc_queue_wait(req);
1381                 if (rc == LUSTRE_ESTALE) {
1382                         CDEBUG(D_DLMTRACE,
1383                                "client/server (nid %s) out of sync -- not fatal\n",
1384                                libcfs_nid2str(req->rq_import->imp_connection->c_peer.nid));
1385                         rc = 0;
1386                 } else if (rc == -ETIMEDOUT && /* check there was no reconnect*/
1387                            req->rq_import_generation == imp->imp_generation) {
1388                         ptlrpc_req_finished(req);
1389                         continue;
1390                 } else if (rc != ELDLM_OK) {
1391                         /* -ESHUTDOWN is common on umount */
1392                         CDEBUG_LIMIT(rc == -ESHUTDOWN ? D_DLMTRACE : D_ERROR,
1393                                      "Got rc %d from cancel RPC: canceling anyway\n",
1394                                      rc);
1395                         break;
1396                 }
1397                 sent = count;
1398                 break;
1399         }
1400
1401         ptlrpc_req_finished(req);
1402         EXIT;
1403 out:
1404         return sent ? sent : rc;
1405 }
1406
1407 static inline struct ldlm_pool *ldlm_imp2pl(struct obd_import *imp)
1408 {
1409         LASSERT(imp != NULL);
1410         return &imp->imp_obd->obd_namespace->ns_pool;
1411 }
1412
1413 /**
1414  * Update client's OBD pool related fields with new SLV and Limit from \a req.
1415  */
1416 int ldlm_cli_update_pool(struct ptlrpc_request *req)
1417 {
1418         struct obd_device *obd;
1419         __u64 new_slv;
1420         __u32 new_limit;
1421
1422         ENTRY;
1423         if (unlikely(!req->rq_import || !req->rq_import->imp_obd ||
1424                      !imp_connect_lru_resize(req->rq_import)))
1425                 /* Do nothing for corner cases. */
1426                 RETURN(0);
1427
1428         /*
1429          * In some cases RPC may contain SLV and limit zeroed out. This
1430          * is the case when server does not support LRU resize feature.
1431          * This is also possible in some recovery cases when server-side
1432          * reqs have no reference to the OBD export and thus access to
1433          * server-side namespace is not possible.
1434          */
1435         if (lustre_msg_get_slv(req->rq_repmsg) == 0 ||
1436             lustre_msg_get_limit(req->rq_repmsg) == 0) {
1437                 DEBUG_REQ(D_HA, req,
1438                           "Zero SLV or limit found (SLV=%llu, limit=%u)",
1439                           lustre_msg_get_slv(req->rq_repmsg),
1440                           lustre_msg_get_limit(req->rq_repmsg));
1441                 RETURN(0);
1442         }
1443
1444         new_limit = lustre_msg_get_limit(req->rq_repmsg);
1445         new_slv = lustre_msg_get_slv(req->rq_repmsg);
1446         obd = req->rq_import->imp_obd;
1447
1448         /*
1449          * Set new SLV and limit in OBD fields to make them accessible
1450          * to the pool thread. We do not access obd_namespace and pool
1451          * directly here as there is no reliable way to make sure that
1452          * they are still alive at cleanup time. Evil races are possible
1453          * which may cause Oops at that time.
1454          */
1455         write_lock(&obd->obd_pool_lock);
1456         obd->obd_pool_slv = new_slv;
1457         obd->obd_pool_limit = new_limit;
1458         write_unlock(&obd->obd_pool_lock);
1459
1460         RETURN(0);
1461 }
1462
1463 int ldlm_cli_convert(struct ldlm_lock *lock,
1464                      enum ldlm_cancel_flags cancel_flags)
1465 {
1466         int rc = -EINVAL;
1467
1468         LASSERT(!lock->l_readers && !lock->l_writers);
1469         LDLM_DEBUG(lock, "client lock convert START");
1470
1471         if (lock->l_resource->lr_type == LDLM_IBITS) {
1472                 lock_res_and_lock(lock);
1473                 do {
1474                         rc = ldlm_cli_inodebits_convert(lock, cancel_flags);
1475                 } while (rc == -EAGAIN);
1476                 unlock_res_and_lock(lock);
1477         }
1478
1479         LDLM_DEBUG(lock, "client lock convert END");
1480         RETURN(rc);
1481 }
1482 EXPORT_SYMBOL(ldlm_cli_convert);
1483
1484 /**
1485  * Client side lock cancel.
1486  *
1487  * Lock must not have any readers or writers by this time.
1488  */
1489 int ldlm_cli_cancel(const struct lustre_handle *lockh,
1490                     enum ldlm_cancel_flags cancel_flags)
1491 {
1492         struct obd_export *exp;
1493         enum ldlm_lru_flags lru_flags;
1494         int avail, count = 1;
1495         __u64 rc = 0;
1496         struct ldlm_namespace *ns;
1497         struct ldlm_lock *lock;
1498         LIST_HEAD(cancels);
1499
1500         ENTRY;
1501
1502         lock = ldlm_handle2lock_long(lockh, 0);
1503         if (lock == NULL) {
1504                 LDLM_DEBUG_NOLOCK("lock is already being destroyed");
1505                 RETURN(0);
1506         }
1507
1508         lock_res_and_lock(lock);
1509         LASSERT(!ldlm_is_converting(lock));
1510
1511         /* Lock is being canceled and the caller doesn't want to wait */
1512         if (ldlm_is_canceling(lock)) {
1513                 if (cancel_flags & LCF_ASYNC) {
1514                         unlock_res_and_lock(lock);
1515                 } else {
1516                         unlock_res_and_lock(lock);
1517                         wait_event_idle(lock->l_waitq, is_bl_done(lock));
1518                 }
1519                 LDLM_LOCK_RELEASE(lock);
1520                 RETURN(0);
1521         }
1522
1523         ldlm_set_canceling(lock);
1524         unlock_res_and_lock(lock);
1525
1526         if (cancel_flags & LCF_LOCAL)
1527                 OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_LOCAL_CANCEL_PAUSE,
1528                                  cfs_fail_val);
1529
1530         rc = ldlm_cli_cancel_local(lock);
1531         if (rc == LDLM_FL_LOCAL_ONLY || cancel_flags & LCF_LOCAL) {
1532                 LDLM_LOCK_RELEASE(lock);
1533                 RETURN(0);
1534         }
1535         /*
1536          * Even if the lock is marked as LDLM_FL_BL_AST, this is a LDLM_CANCEL
1537          * RPC which goes to canceld portal, so we can cancel other LRU locks
1538          * here and send them all as one LDLM_CANCEL RPC.
1539          */
1540         LASSERT(list_empty(&lock->l_bl_ast));
1541         list_add(&lock->l_bl_ast, &cancels);
1542
1543         exp = lock->l_conn_export;
1544         if (exp_connect_cancelset(exp)) {
1545                 avail = ldlm_format_handles_avail(class_exp2cliimp(exp),
1546                                                   &RQF_LDLM_CANCEL,
1547                                                   RCL_CLIENT, 0);
1548                 LASSERT(avail > 0);
1549
1550                 ns = ldlm_lock_to_ns(lock);
1551                 lru_flags = ns_connect_lru_resize(ns) ?
1552                         LDLM_LRU_FLAG_LRUR : LDLM_LRU_FLAG_AGED;
1553                 count += ldlm_cancel_lru_local(ns, &cancels, 0, avail - 1,
1554                                                LCF_BL_AST, lru_flags);
1555         }
1556         ldlm_cli_cancel_list(&cancels, count, NULL, cancel_flags);
1557         RETURN(0);
1558 }
1559 EXPORT_SYMBOL(ldlm_cli_cancel);
1560
1561 /**
1562  * Locally cancel up to \a count locks in list \a cancels.
1563  * Return the number of cancelled locks.
1564  */
1565 int ldlm_cli_cancel_list_local(struct list_head *cancels, int count,
1566                                enum ldlm_cancel_flags cancel_flags)
1567 {
1568         LIST_HEAD(head);
1569         struct ldlm_lock *lock, *next;
1570         int left = 0, bl_ast = 0;
1571         __u64 rc;
1572
1573         left = count;
1574         list_for_each_entry_safe(lock, next, cancels, l_bl_ast) {
1575                 if (left-- == 0)
1576                         break;
1577
1578                 if (cancel_flags & LCF_LOCAL) {
1579                         rc = LDLM_FL_LOCAL_ONLY;
1580                         ldlm_lock_cancel(lock);
1581                 } else {
1582                         rc = ldlm_cli_cancel_local(lock);
1583                 }
1584                 /*
1585                  * Until we have compound requests and can send LDLM_CANCEL
1586                  * requests batched with generic RPCs, we need to send cancels
1587                  * with the LDLM_FL_BL_AST flag in a separate RPC from
1588                  * the one being generated now.
1589                  */
1590                 if (!(cancel_flags & LCF_BL_AST) && (rc == LDLM_FL_BL_AST)) {
1591                         LDLM_DEBUG(lock, "Cancel lock separately");
1592                         list_move(&lock->l_bl_ast, &head);
1593                         bl_ast++;
1594                         continue;
1595                 }
1596                 if (rc == LDLM_FL_LOCAL_ONLY) {
1597                         /* CANCEL RPC should not be sent to server. */
1598                         list_del_init(&lock->l_bl_ast);
1599                         LDLM_LOCK_RELEASE(lock);
1600                         count--;
1601                 }
1602         }
1603         if (bl_ast > 0) {
1604                 count -= bl_ast;
1605                 ldlm_cli_cancel_list(&head, bl_ast, NULL, 0);
1606         }
1607
1608         RETURN(count);
1609 }
1610
1611 /**
1612  * Cancel as many locks as possible w/o sending any RPCs (e.g. to write back
1613  * dirty data, to close a file, ...) or waiting for any RPCs in-flight (e.g.
1614  * readahead requests, ...)
1615  */
1616 static enum ldlm_policy_res
1617 ldlm_cancel_no_wait_policy(struct ldlm_namespace *ns, struct ldlm_lock *lock,
1618                            int unused, int added, int count)
1619 {
1620         enum ldlm_policy_res result = LDLM_POLICY_CANCEL_LOCK;
1621
1622         /*
1623          * don't check added & count since we want to process all locks
1624          * from unused list.
1625          * It's fine to not take lock to access lock->l_resource since
1626          * the lock has already been granted so it won't change.
1627          */
1628         switch (lock->l_resource->lr_type) {
1629                 case LDLM_EXTENT:
1630                 case LDLM_IBITS:
1631                         if (ns->ns_cancel != NULL && ns->ns_cancel(lock) != 0)
1632                                 break;
1633                         /* fallthrough */
1634                 default:
1635                         result = LDLM_POLICY_SKIP_LOCK;
1636                         break;
1637         }
1638
1639         RETURN(result);
1640 }
1641
1642 /**
1643  * Callback function for LRU-resize policy. Decides whether to keep
1644  * \a lock in LRU for current \a LRU size \a unused, added in current
1645  * scan \a added and number of locks to be preferably canceled \a count.
1646  *
1647  * \retval LDLM_POLICY_KEEP_LOCK keep lock in LRU in stop scanning
1648  *
1649  * \retval LDLM_POLICY_CANCEL_LOCK cancel lock from LRU
1650  */
1651 static enum ldlm_policy_res ldlm_cancel_lrur_policy(struct ldlm_namespace *ns,
1652                                                     struct ldlm_lock *lock,
1653                                                     int unused, int added,
1654                                                     int count)
1655 {
1656         ktime_t cur = ktime_get();
1657         struct ldlm_pool *pl = &ns->ns_pool;
1658         u64 slv, lvf, lv;
1659         s64 la;
1660
1661         /*
1662          * Stop LRU processing when we reach past @count or have checked all
1663          * locks in LRU.
1664          */
1665         if (count && added >= count)
1666                 return LDLM_POLICY_KEEP_LOCK;
1667
1668         /*
1669          * Despite of the LV, It doesn't make sense to keep the lock which
1670          * is unused for ns_max_age time.
1671          */
1672         if (ktime_after(ktime_get(),
1673                         ktime_add(lock->l_last_used, ns->ns_max_age)))
1674                 return LDLM_POLICY_CANCEL_LOCK;
1675
1676         slv = ldlm_pool_get_slv(pl);
1677         lvf = ldlm_pool_get_lvf(pl);
1678         la = div_u64(ktime_to_ns(ktime_sub(cur, lock->l_last_used)),
1679                      NSEC_PER_SEC);
1680         lv = lvf * la * unused;
1681
1682         /* Inform pool about current CLV to see it via debugfs. */
1683         ldlm_pool_set_clv(pl, lv);
1684
1685         /*
1686          * Stop when SLV is not yet come from server or lv is smaller than
1687          * it is.
1688          */
1689         if (slv == 0 || lv < slv)
1690                 return LDLM_POLICY_KEEP_LOCK;
1691
1692         return LDLM_POLICY_CANCEL_LOCK;
1693 }
1694
1695 static enum ldlm_policy_res
1696 ldlm_cancel_lrur_no_wait_policy(struct ldlm_namespace *ns,
1697                                 struct ldlm_lock *lock,
1698                                 int unused, int added,
1699                                 int count)
1700 {
1701         enum ldlm_policy_res result;
1702
1703         result = ldlm_cancel_lrur_policy(ns, lock, unused, added, count);
1704         if (result == LDLM_POLICY_KEEP_LOCK)
1705                 return result;
1706
1707         return ldlm_cancel_no_wait_policy(ns, lock, unused, added, count);
1708 }
1709
1710 /**
1711  * Callback function for debugfs used policy. Makes decision whether to keep
1712  * \a lock in LRU for current \a LRU size \a unused, added in current scan \a
1713  * added and number of locks to be preferably canceled \a count.
1714  *
1715  * \retval LDLM_POLICY_KEEP_LOCK keep lock in LRU in stop scanning
1716  *
1717  * \retval LDLM_POLICY_CANCEL_LOCK cancel lock from LRU
1718  */
1719 static enum ldlm_policy_res ldlm_cancel_passed_policy(struct ldlm_namespace *ns,
1720                                                       struct ldlm_lock *lock,
1721                                                       int unused, int added,
1722                                                       int count)
1723 {
1724         /*
1725          * Stop LRU processing when we reach past @count or have checked all
1726          * locks in LRU.
1727          */
1728         return (added >= count) ?
1729                 LDLM_POLICY_KEEP_LOCK : LDLM_POLICY_CANCEL_LOCK;
1730 }
1731
1732 /**
1733  * Callback function for aged policy. Makes decision whether to keep \a lock in
1734  * LRU for current LRU size \a unused, added in current scan \a added and
1735  * number of locks to be preferably canceled \a count.
1736  *
1737  * \retval LDLM_POLICY_KEEP_LOCK keep lock in LRU in stop scanning
1738  *
1739  * \retval LDLM_POLICY_CANCEL_LOCK cancel lock from LRU
1740  */
1741 static enum ldlm_policy_res ldlm_cancel_aged_policy(struct ldlm_namespace *ns,
1742                                                     struct ldlm_lock *lock,
1743                                                     int unused, int added,
1744                                                     int count)
1745 {
1746         if ((added >= count) &&
1747             ktime_before(ktime_get(),
1748                          ktime_add(lock->l_last_used, ns->ns_max_age)))
1749                 return LDLM_POLICY_KEEP_LOCK;
1750
1751         return LDLM_POLICY_CANCEL_LOCK;
1752 }
1753
1754 static enum ldlm_policy_res
1755 ldlm_cancel_aged_no_wait_policy(struct ldlm_namespace *ns,
1756                                 struct ldlm_lock *lock,
1757                                 int unused, int added, int count)
1758 {
1759         enum ldlm_policy_res result;
1760
1761         result = ldlm_cancel_aged_policy(ns, lock, unused, added, count);
1762         if (result == LDLM_POLICY_KEEP_LOCK)
1763                 return result;
1764
1765         return ldlm_cancel_no_wait_policy(ns, lock, unused, added, count);
1766 }
1767
1768 /**
1769  * Callback function for default policy. Makes decision whether to keep \a lock
1770  * in LRU for current LRU size \a unused, added in current scan \a added and
1771  * number of locks to be preferably canceled \a count.
1772  *
1773  * \retval LDLM_POLICY_KEEP_LOCK keep lock in LRU in stop scanning
1774  *
1775  * \retval LDLM_POLICY_CANCEL_LOCK cancel lock from LRU
1776  */
1777 static
1778 enum ldlm_policy_res ldlm_cancel_default_policy(struct ldlm_namespace *ns,
1779                                                 struct ldlm_lock *lock,
1780                                                 int unused, int added,
1781                                                 int count)
1782 {
1783         /*
1784          * Stop LRU processing when we reach past count or have checked all
1785          * locks in LRU.
1786          */
1787         return (added >= count) ?
1788                 LDLM_POLICY_KEEP_LOCK : LDLM_POLICY_CANCEL_LOCK;
1789 }
1790
1791 typedef enum ldlm_policy_res
1792 (*ldlm_cancel_lru_policy_t)(struct ldlm_namespace *ns, struct ldlm_lock *lock,
1793                             int unused, int added, int count);
1794
1795 static ldlm_cancel_lru_policy_t
1796 ldlm_cancel_lru_policy(struct ldlm_namespace *ns, enum ldlm_lru_flags lru_flags)
1797 {
1798         if (ns_connect_lru_resize(ns)) {
1799                 if (lru_flags & LDLM_LRU_FLAG_SHRINK)
1800                         /* We kill passed number of old locks. */
1801                         return ldlm_cancel_passed_policy;
1802                 if (lru_flags & LDLM_LRU_FLAG_LRUR) {
1803                         if (lru_flags & LDLM_LRU_FLAG_NO_WAIT)
1804                                 return ldlm_cancel_lrur_no_wait_policy;
1805                         else
1806                                 return ldlm_cancel_lrur_policy;
1807                 }
1808                 if (lru_flags & LDLM_LRU_FLAG_PASSED)
1809                         return ldlm_cancel_passed_policy;
1810         } else {
1811                 if (lru_flags & LDLM_LRU_FLAG_AGED) {
1812                         if (lru_flags & LDLM_LRU_FLAG_NO_WAIT)
1813                                 return ldlm_cancel_aged_no_wait_policy;
1814                         else
1815                                 return ldlm_cancel_aged_policy;
1816                 }
1817         }
1818         if (lru_flags & LDLM_LRU_FLAG_NO_WAIT)
1819                 return ldlm_cancel_no_wait_policy;
1820
1821         return ldlm_cancel_default_policy;
1822 }
1823
1824 /**
1825  * - Free space in LRU for \a count new locks,
1826  *   redundant unused locks are canceled locally;
1827  * - also cancel locally unused aged locks;
1828  * - do not cancel more than \a max locks;
1829  * - GET the found locks and add them into the \a cancels list.
1830  *
1831  * A client lock can be added to the l_bl_ast list only when it is
1832  * marked LDLM_FL_CANCELING. Otherwise, somebody is already doing
1833  * CANCEL.  There are the following use cases:
1834  * ldlm_cancel_resource_local(), ldlm_cancel_lru_local() and
1835  * ldlm_cli_cancel(), which check and set this flag properly. As any
1836  * attempt to cancel a lock rely on this flag, l_bl_ast list is accessed
1837  * later without any special locking.
1838  *
1839  * Calling policies for enabled LRU resize:
1840  * ----------------------------------------
1841  * flags & LDLM_LRU_FLAG_LRUR - use LRU resize policy (SLV from server) to
1842  *                              cancel not more than \a count locks;
1843  *
1844  * flags & LDLM_LRU_FLAG_PASSED - cancel \a count number of old locks (located
1845  *                              at the beginning of LRU list);
1846  *
1847  * flags & LDLM_LRU_FLAG_SHRINK - cancel not more than \a count locks according
1848  *                              to memory pressre policy function;
1849  *
1850  * flags & LDLM_LRU_FLAG_AGED - cancel \a count locks according to "aged policy"
1851  *
1852  * flags & LDLM_LRU_FLAG_NO_WAIT - cancel as many unused locks as possible
1853  *                              (typically before replaying locks) w/o
1854  *                              sending any RPCs or waiting for any
1855  *                              outstanding RPC to complete.
1856  *
1857  * flags & LDLM_CANCEL_CLEANUP - when cancelling read locks, do not check for
1858  *                              other read locks covering the same pages, just
1859  *                              discard those pages.
1860  */
1861 static int ldlm_prepare_lru_list(struct ldlm_namespace *ns,
1862                                  struct list_head *cancels, int count, int max,
1863                                  enum ldlm_lru_flags lru_flags)
1864 {
1865         ldlm_cancel_lru_policy_t pf;
1866         int added = 0;
1867         int no_wait = lru_flags & LDLM_LRU_FLAG_NO_WAIT;
1868
1869         ENTRY;
1870
1871         if (!ns_connect_lru_resize(ns))
1872                 count += ns->ns_nr_unused - ns->ns_max_unused;
1873
1874         pf = ldlm_cancel_lru_policy(ns, lru_flags);
1875         LASSERT(pf != NULL);
1876
1877         /* For any flags, stop scanning if @max is reached. */
1878         while (!list_empty(&ns->ns_unused_list) && (max == 0 || added < max)) {
1879                 struct ldlm_lock *lock;
1880                 struct list_head *item, *next;
1881                 enum ldlm_policy_res result;
1882                 ktime_t last_use = ktime_set(0, 0);
1883
1884                 spin_lock(&ns->ns_lock);
1885                 item = no_wait ? ns->ns_last_pos : &ns->ns_unused_list;
1886                 for (item = item->next, next = item->next;
1887                      item != &ns->ns_unused_list;
1888                      item = next, next = item->next) {
1889                         lock = list_entry(item, struct ldlm_lock, l_lru);
1890
1891                         /* No locks which got blocking requests. */
1892                         LASSERT(!ldlm_is_bl_ast(lock));
1893
1894                         if (!ldlm_is_canceling(lock))
1895                                 break;
1896
1897                         /*
1898                          * Somebody is already doing CANCEL. No need for this
1899                          * lock in LRU, do not traverse it again.
1900                          */
1901                         ldlm_lock_remove_from_lru_nolock(lock);
1902                 }
1903                 if (item == &ns->ns_unused_list) {
1904                         spin_unlock(&ns->ns_lock);
1905                         break;
1906                 }
1907
1908                 last_use = lock->l_last_used;
1909
1910                 LDLM_LOCK_GET(lock);
1911                 spin_unlock(&ns->ns_lock);
1912                 lu_ref_add(&lock->l_reference, __FUNCTION__, current);
1913
1914                 /*
1915                  * Pass the lock through the policy filter and see if it
1916                  * should stay in LRU.
1917                  *
1918                  * Even for shrinker policy we stop scanning if
1919                  * we find a lock that should stay in the cache.
1920                  * We should take into account lock age anyway
1921                  * as a new lock is a valuable resource even if
1922                  * it has a low weight.
1923                  *
1924                  * That is, for shrinker policy we drop only
1925                  * old locks, but additionally choose them by
1926                  * their weight. Big extent locks will stay in
1927                  * the cache.
1928                  */
1929                 result = pf(ns, lock, ns->ns_nr_unused, added, count);
1930                 if (result == LDLM_POLICY_KEEP_LOCK) {
1931                         lu_ref_del(&lock->l_reference, __func__, current);
1932                         LDLM_LOCK_RELEASE(lock);
1933                         break;
1934                 }
1935
1936                 if (result == LDLM_POLICY_SKIP_LOCK) {
1937                         lu_ref_del(&lock->l_reference, __func__, current);
1938                         if (no_wait) {
1939                                 spin_lock(&ns->ns_lock);
1940                                 if (!list_empty(&lock->l_lru) &&
1941                                     lock->l_lru.prev == ns->ns_last_pos)
1942                                         ns->ns_last_pos = &lock->l_lru;
1943                                 spin_unlock(&ns->ns_lock);
1944                         }
1945
1946                         LDLM_LOCK_RELEASE(lock);
1947                         continue;
1948                 }
1949
1950                 lock_res_and_lock(lock);
1951                 /* Check flags again under the lock. */
1952                 if (ldlm_is_canceling(lock) ||
1953                     ldlm_lock_remove_from_lru_check(lock, last_use) == 0) {
1954                         /*
1955                          * Another thread is removing lock from LRU, or
1956                          * somebody is already doing CANCEL, or there
1957                          * is a blocking request which will send cancel
1958                          * by itself, or the lock is no longer unused or
1959                          * the lock has been used since the pf() call and
1960                          * pages could be put under it.
1961                          */
1962                         unlock_res_and_lock(lock);
1963                         lu_ref_del(&lock->l_reference, __FUNCTION__, current);
1964                         LDLM_LOCK_RELEASE(lock);
1965                         continue;
1966                 }
1967                 LASSERT(!lock->l_readers && !lock->l_writers);
1968
1969                 /*
1970                  * If we have chosen to cancel this lock voluntarily, we
1971                  * better send cancel notification to server, so that it
1972                  * frees appropriate state. This might lead to a race
1973                  * where while we are doing cancel here, server is also
1974                  * silently cancelling this lock.
1975                  */
1976                 ldlm_clear_cancel_on_block(lock);
1977
1978                 /*
1979                  * Setting the CBPENDING flag is a little misleading,
1980                  * but prevents an important race; namely, once
1981                  * CBPENDING is set, the lock can accumulate no more
1982                  * readers/writers. Since readers and writers are
1983                  * already zero here, ldlm_lock_decref() won't see
1984                  * this flag and call l_blocking_ast
1985                  */
1986                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_CANCELING;
1987
1988                 if ((lru_flags & LDLM_LRU_FLAG_CLEANUP) &&
1989                     (lock->l_resource->lr_type == LDLM_EXTENT ||
1990                      ldlm_has_dom(lock)) && lock->l_granted_mode == LCK_PR)
1991                         ldlm_set_discard_data(lock);
1992
1993                 /*
1994                  * We can't re-add to l_lru as it confuses the
1995                  * refcounting in ldlm_lock_remove_from_lru() if an AST
1996                  * arrives after we drop lr_lock below. We use l_bl_ast
1997                  * and can't use l_pending_chain as it is used both on
1998                  * server and client nevertheless b=5666 says it is
1999                  * used only on server
2000                  */
2001                 LASSERT(list_empty(&lock->l_bl_ast));
2002                 list_add(&lock->l_bl_ast, cancels);
2003                 unlock_res_and_lock(lock);
2004                 lu_ref_del(&lock->l_reference, __FUNCTION__, current);
2005                 added++;
2006         }
2007         RETURN(added);
2008 }
2009
2010 int ldlm_cancel_lru_local(struct ldlm_namespace *ns, struct list_head *cancels,
2011                           int count, int max,
2012                           enum ldlm_cancel_flags cancel_flags,
2013                           enum ldlm_lru_flags lru_flags)
2014 {
2015         int added;
2016
2017         added = ldlm_prepare_lru_list(ns, cancels, count, max, lru_flags);
2018         if (added <= 0)
2019                 return added;
2020
2021         return ldlm_cli_cancel_list_local(cancels, added, cancel_flags);
2022 }
2023
2024 /**
2025  * Cancel at least \a nr locks from given namespace LRU.
2026  *
2027  * When called with LCF_ASYNC the blocking callback will be handled
2028  * in a thread and this function will return after the thread has been
2029  * asked to call the callback.  When called with LCF_ASYNC the blocking
2030  * callback will be performed in this function.
2031  */
2032 int ldlm_cancel_lru(struct ldlm_namespace *ns, int nr,
2033                     enum ldlm_cancel_flags cancel_flags,
2034                     enum ldlm_lru_flags lru_flags)
2035 {
2036         LIST_HEAD(cancels);
2037         int count, rc;
2038
2039         ENTRY;
2040
2041         /*
2042          * Just prepare the list of locks, do not actually cancel them yet.
2043          * Locks are cancelled later in a separate thread.
2044          */
2045         count = ldlm_prepare_lru_list(ns, &cancels, nr, 0, lru_flags);
2046         rc = ldlm_bl_to_thread_list(ns, NULL, &cancels, count, cancel_flags);
2047         if (rc == 0)
2048                 RETURN(count);
2049
2050         RETURN(0);
2051 }
2052
2053 /**
2054  * Find and cancel locally unused locks found on resource, matched to the
2055  * given policy, mode. GET the found locks and add them into the \a cancels
2056  * list.
2057  */
2058 int ldlm_cancel_resource_local(struct ldlm_resource *res,
2059                                struct list_head *cancels,
2060                                union ldlm_policy_data *policy,
2061                                enum ldlm_mode mode, __u64 lock_flags,
2062                                enum ldlm_cancel_flags cancel_flags,
2063                                void *opaque)
2064 {
2065         struct ldlm_lock *lock;
2066         int count = 0;
2067
2068         ENTRY;
2069
2070         lock_res(res);
2071         list_for_each_entry(lock, &res->lr_granted, l_res_link) {
2072                 if (opaque != NULL && lock->l_ast_data != opaque) {
2073                         LDLM_ERROR(lock, "data %p doesn't match opaque %p",
2074                                    lock->l_ast_data, opaque);
2075                         continue;
2076                 }
2077
2078                 if (lock->l_readers || lock->l_writers)
2079                         continue;
2080
2081                 /*
2082                  * If somebody is already doing CANCEL, or blocking AST came
2083                  * then skip this lock.
2084                  */
2085                 if (ldlm_is_bl_ast(lock) || ldlm_is_canceling(lock))
2086                         continue;
2087
2088                 if (lockmode_compat(lock->l_granted_mode, mode))
2089                         continue;
2090
2091                 /*
2092                  * If policy is given and this is IBITS lock, add to list only
2093                  * those locks that match by policy.
2094                  */
2095                 if (policy && (lock->l_resource->lr_type == LDLM_IBITS)) {
2096                         if (!(lock->l_policy_data.l_inodebits.bits &
2097                               policy->l_inodebits.bits))
2098                                 continue;
2099                         /* Skip locks with DoM bit if it is not set in policy
2100                          * to don't flush data by side-bits. Lock convert will
2101                          * drop those bits separately.
2102                          */
2103                         if (ldlm_has_dom(lock) &&
2104                             !(policy->l_inodebits.bits & MDS_INODELOCK_DOM))
2105                                 continue;
2106                 }
2107
2108                 /* See CBPENDING comment in ldlm_cancel_lru */
2109                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_CANCELING |
2110                                  lock_flags;
2111                 LASSERT(list_empty(&lock->l_bl_ast));
2112                 list_add(&lock->l_bl_ast, cancels);
2113                 LDLM_LOCK_GET(lock);
2114                 count++;
2115         }
2116         unlock_res(res);
2117
2118         RETURN(ldlm_cli_cancel_list_local(cancels, count, cancel_flags));
2119 }
2120 EXPORT_SYMBOL(ldlm_cancel_resource_local);
2121
2122 /**
2123  * Cancel client-side locks from a list and send/prepare cancel RPCs to the
2124  * server.
2125  * If \a req is NULL, send CANCEL request to server with handles of locks
2126  * in the \a cancels. If EARLY_CANCEL is not supported, send CANCEL requests
2127  * separately per lock.
2128  * If \a req is not NULL, put handles of locks in \a cancels into the request
2129  * buffer at the offset \a off.
2130  * Destroy \a cancels at the end.
2131  */
2132 int ldlm_cli_cancel_list(struct list_head *cancels, int count,
2133                          struct ptlrpc_request *req,
2134                          enum ldlm_cancel_flags flags)
2135 {
2136         struct ldlm_lock *lock;
2137         int res = 0;
2138
2139         ENTRY;
2140
2141         if (list_empty(cancels) || count == 0)
2142                 RETURN(0);
2143
2144         /*
2145          * XXX: requests (both batched and not) could be sent in parallel.
2146          * Usually it is enough to have just 1 RPC, but it is possible that
2147          * there are too many locks to be cancelled in LRU or on a resource.
2148          * It would also speed up the case when the server does not support
2149          * the feature.
2150          */
2151         while (count > 0) {
2152                 LASSERT(!list_empty(cancels));
2153                 lock = list_entry(cancels->next, struct ldlm_lock,
2154                                   l_bl_ast);
2155                 LASSERT(lock->l_conn_export);
2156
2157                 if (exp_connect_cancelset(lock->l_conn_export)) {
2158                         res = count;
2159                         if (req)
2160                                 ldlm_cancel_pack(req, cancels, count);
2161                         else
2162                                 res = ldlm_cli_cancel_req(lock->l_conn_export,
2163                                                           cancels, count,
2164                                                           flags);
2165                 } else {
2166                         res = ldlm_cli_cancel_req(lock->l_conn_export,
2167                                                   cancels, 1, flags);
2168                 }
2169
2170                 if (res < 0) {
2171                         CDEBUG_LIMIT(res == -ESHUTDOWN ? D_DLMTRACE : D_ERROR,
2172                                      "ldlm_cli_cancel_list: %d\n", res);
2173                         res = count;
2174                 }
2175
2176                 count -= res;
2177                 ldlm_lock_list_put(cancels, l_bl_ast, res);
2178         }
2179         LASSERT(count == 0);
2180         RETURN(0);
2181 }
2182 EXPORT_SYMBOL(ldlm_cli_cancel_list);
2183
2184 /**
2185  * Cancel all locks on a resource that have 0 readers/writers.
2186  *
2187  * If flags & LDLM_FL_LOCAL_ONLY, throw the locks away without trying
2188  * to notify the server.
2189  */
2190 int ldlm_cli_cancel_unused_resource(struct ldlm_namespace *ns,
2191                                     const struct ldlm_res_id *res_id,
2192                                     union ldlm_policy_data *policy,
2193                                     enum ldlm_mode mode,
2194                                     enum ldlm_cancel_flags flags, void *opaque)
2195 {
2196         struct ldlm_resource *res;
2197         LIST_HEAD(cancels);
2198         int count;
2199         int rc;
2200
2201         ENTRY;
2202
2203         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
2204         if (IS_ERR(res)) {
2205                 /* This is not a problem. */
2206                 CDEBUG(D_INFO, "No resource %llu\n", res_id->name[0]);
2207                 RETURN(0);
2208         }
2209
2210         LDLM_RESOURCE_ADDREF(res);
2211         count = ldlm_cancel_resource_local(res, &cancels, policy, mode,
2212                                            0, flags | LCF_BL_AST, opaque);
2213         rc = ldlm_cli_cancel_list(&cancels, count, NULL, flags);
2214         if (rc != ELDLM_OK)
2215                 CERROR("canceling unused lock "DLDLMRES": rc = %d\n",
2216                        PLDLMRES(res), rc);
2217
2218         LDLM_RESOURCE_DELREF(res);
2219         ldlm_resource_putref(res);
2220         RETURN(0);
2221 }
2222 EXPORT_SYMBOL(ldlm_cli_cancel_unused_resource);
2223
2224 struct ldlm_cli_cancel_arg {
2225         int     lc_flags;
2226         void   *lc_opaque;
2227 };
2228
2229 static int
2230 ldlm_cli_hash_cancel_unused(struct cfs_hash *hs, struct cfs_hash_bd *bd,
2231                             struct hlist_node *hnode, void *arg)
2232 {
2233         struct ldlm_resource           *res = cfs_hash_object(hs, hnode);
2234         struct ldlm_cli_cancel_arg     *lc = arg;
2235
2236         ldlm_cli_cancel_unused_resource(ldlm_res_to_ns(res), &res->lr_name,
2237                                         NULL, LCK_MINMODE, lc->lc_flags,
2238                                         lc->lc_opaque);
2239         /* must return 0 for hash iteration */
2240         return 0;
2241 }
2242
2243 /**
2244  * Cancel all locks on a namespace (or a specific resource, if given)
2245  * that have 0 readers/writers.
2246  *
2247  * If flags & LCF_LOCAL, throw the locks away without trying
2248  * to notify the server.
2249  */
2250 int ldlm_cli_cancel_unused(struct ldlm_namespace *ns,
2251                            const struct ldlm_res_id *res_id,
2252                            enum ldlm_cancel_flags flags, void *opaque)
2253 {
2254         struct ldlm_cli_cancel_arg arg = {
2255                 .lc_flags       = flags,
2256                 .lc_opaque      = opaque,
2257         };
2258
2259         ENTRY;
2260
2261         if (ns == NULL)
2262                 RETURN(ELDLM_OK);
2263
2264         if (res_id != NULL) {
2265                 RETURN(ldlm_cli_cancel_unused_resource(ns, res_id, NULL,
2266                                                        LCK_MINMODE, flags,
2267                                                        opaque));
2268         } else {
2269                 cfs_hash_for_each_nolock(ns->ns_rs_hash,
2270                                          ldlm_cli_hash_cancel_unused, &arg, 0);
2271                 RETURN(ELDLM_OK);
2272         }
2273 }
2274
2275 /* Lock iterators. */
2276
2277 int ldlm_resource_foreach(struct ldlm_resource *res, ldlm_iterator_t iter,
2278                           void *closure)
2279 {
2280         struct list_head *tmp, *next;
2281         struct ldlm_lock *lock;
2282         int rc = LDLM_ITER_CONTINUE;
2283
2284         ENTRY;
2285
2286         if (!res)
2287                 RETURN(LDLM_ITER_CONTINUE);
2288
2289         lock_res(res);
2290         list_for_each_safe(tmp, next, &res->lr_granted) {
2291                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
2292
2293                 if (iter(lock, closure) == LDLM_ITER_STOP)
2294                         GOTO(out, rc = LDLM_ITER_STOP);
2295         }
2296
2297         list_for_each_safe(tmp, next, &res->lr_waiting) {
2298                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
2299
2300                 if (iter(lock, closure) == LDLM_ITER_STOP)
2301                         GOTO(out, rc = LDLM_ITER_STOP);
2302         }
2303 out:
2304         unlock_res(res);
2305         RETURN(rc);
2306 }
2307
2308 struct iter_helper_data {
2309         ldlm_iterator_t iter;
2310         void *closure;
2311 };
2312
2313 static int ldlm_iter_helper(struct ldlm_lock *lock, void *closure)
2314 {
2315         struct iter_helper_data *helper = closure;
2316
2317         return helper->iter(lock, helper->closure);
2318 }
2319
2320 static int ldlm_res_iter_helper(struct cfs_hash *hs, struct cfs_hash_bd *bd,
2321                                 struct hlist_node *hnode, void *arg)
2322
2323 {
2324         struct ldlm_resource *res = cfs_hash_object(hs, hnode);
2325
2326         return ldlm_resource_foreach(res, ldlm_iter_helper, arg) ==
2327                                      LDLM_ITER_STOP;
2328 }
2329
2330 void ldlm_namespace_foreach(struct ldlm_namespace *ns,
2331                             ldlm_iterator_t iter, void *closure)
2332
2333 {
2334         struct iter_helper_data helper = { .iter = iter, .closure = closure };
2335
2336         cfs_hash_for_each_nolock(ns->ns_rs_hash,
2337                                  ldlm_res_iter_helper, &helper, 0);
2338
2339 }
2340
2341 /*
2342  * non-blocking function to manipulate a lock whose cb_data is being put away.
2343  * return  0:  find no resource
2344  *       > 0:  must be LDLM_ITER_STOP/LDLM_ITER_CONTINUE.
2345  *       < 0:  errors
2346  */
2347 int ldlm_resource_iterate(struct ldlm_namespace *ns,
2348                           const struct ldlm_res_id *res_id,
2349                           ldlm_iterator_t iter, void *data)
2350 {
2351         struct ldlm_resource *res;
2352         int rc;
2353
2354         ENTRY;
2355
2356         LASSERTF(ns != NULL, "must pass in namespace\n");
2357
2358         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
2359         if (IS_ERR(res))
2360                 RETURN(0);
2361
2362         LDLM_RESOURCE_ADDREF(res);
2363         rc = ldlm_resource_foreach(res, iter, data);
2364         LDLM_RESOURCE_DELREF(res);
2365         ldlm_resource_putref(res);
2366         RETURN(rc);
2367 }
2368 EXPORT_SYMBOL(ldlm_resource_iterate);
2369
2370 /* Lock replay */
2371 static int ldlm_chain_lock_for_replay(struct ldlm_lock *lock, void *closure)
2372 {
2373         struct list_head *list = closure;
2374
2375         /* we use l_pending_chain here, because it's unused on clients. */
2376         LASSERTF(list_empty(&lock->l_pending_chain),
2377                  "lock %p next %p prev %p\n",
2378                  lock, &lock->l_pending_chain.next,
2379                  &lock->l_pending_chain.prev);
2380         /*
2381          * b=9573: don't replay locks left after eviction, or
2382          * b=17614: locks being actively cancelled. Get a reference
2383          * on a lock so that it does not disapear under us (e.g. due to cancel)
2384          */
2385         if (!(lock->l_flags & (LDLM_FL_FAILED|LDLM_FL_BL_DONE))) {
2386                 list_add(&lock->l_pending_chain, list);
2387                 LDLM_LOCK_GET(lock);
2388         }
2389
2390         return LDLM_ITER_CONTINUE;
2391 }
2392
2393 static int replay_lock_interpret(const struct lu_env *env,
2394                                  struct ptlrpc_request *req, void *args, int rc)
2395 {
2396         struct ldlm_async_args *aa = args;
2397         struct ldlm_lock     *lock;
2398         struct ldlm_reply    *reply;
2399         struct obd_export    *exp;
2400
2401         ENTRY;
2402         atomic_dec(&req->rq_import->imp_replay_inflight);
2403         if (rc != ELDLM_OK)
2404                 GOTO(out, rc);
2405
2406         reply = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
2407         if (reply == NULL)
2408                 GOTO(out, rc = -EPROTO);
2409
2410         lock = ldlm_handle2lock(&aa->lock_handle);
2411         if (!lock) {
2412                 CERROR("received replay ack for unknown local cookie %#llx remote cookie %#llx from server %s id %s\n",
2413                        aa->lock_handle.cookie, reply->lock_handle.cookie,
2414                        req->rq_export->exp_client_uuid.uuid,
2415                        libcfs_id2str(req->rq_peer));
2416                 GOTO(out, rc = -ESTALE);
2417         }
2418
2419         /* Key change rehash lock in per-export hash with new key */
2420         exp = req->rq_export;
2421         if (exp && exp->exp_lock_hash) {
2422                 /*
2423                  * In the function below, .hs_keycmp resolves to
2424                  * ldlm_export_lock_keycmp()
2425                  */
2426                 /* coverity[overrun-buffer-val] */
2427                 cfs_hash_rehash_key(exp->exp_lock_hash,
2428                                     &lock->l_remote_handle,
2429                                     &reply->lock_handle,
2430                                     &lock->l_exp_hash);
2431         } else {
2432                 lock->l_remote_handle = reply->lock_handle;
2433         }
2434
2435         LDLM_DEBUG(lock, "replayed lock:");
2436         ptlrpc_import_recovery_state_machine(req->rq_import);
2437         LDLM_LOCK_PUT(lock);
2438 out:
2439         if (rc != ELDLM_OK)
2440                 ptlrpc_connect_import(req->rq_import);
2441
2442         RETURN(rc);
2443 }
2444
2445 static int replay_one_lock(struct obd_import *imp, struct ldlm_lock *lock)
2446 {
2447         struct ptlrpc_request *req;
2448         struct ldlm_async_args *aa;
2449         struct ldlm_request   *body;
2450         int flags;
2451
2452         ENTRY;
2453
2454
2455         /* b=11974: Do not replay a lock which is actively being canceled */
2456         if (ldlm_is_bl_done(lock)) {
2457                 LDLM_DEBUG(lock, "Not replaying canceled lock:");
2458                 RETURN(0);
2459         }
2460
2461         /*
2462          * If this is reply-less callback lock, we cannot replay it, since
2463          * server might have long dropped it, but notification of that event was
2464          * lost by network. (and server granted conflicting lock already)
2465          */
2466         if (ldlm_is_cancel_on_block(lock)) {
2467                 LDLM_DEBUG(lock, "Not replaying reply-less lock:");
2468                 ldlm_lock_cancel(lock);
2469                 RETURN(0);
2470         }
2471
2472         /*
2473          * If granted mode matches the requested mode, this lock is granted.
2474          *
2475          * If we haven't been granted anything and are on a resource list,
2476          * then we're blocked/waiting.
2477          *
2478          * If we haven't been granted anything and we're NOT on a resource list,
2479          * then we haven't got a reply yet and don't have a known disposition.
2480          * This happens whenever a lock enqueue is the request that triggers
2481          * recovery.
2482          */
2483         if (ldlm_is_granted(lock))
2484                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_GRANTED;
2485         else if (!list_empty(&lock->l_res_link))
2486                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_WAIT;
2487         else
2488                 flags = LDLM_FL_REPLAY;
2489
2490         req = ptlrpc_request_alloc_pack(imp, &RQF_LDLM_ENQUEUE,
2491                                         LUSTRE_DLM_VERSION, LDLM_ENQUEUE);
2492         if (req == NULL)
2493                 RETURN(-ENOMEM);
2494
2495         /* We're part of recovery, so don't wait for it. */
2496         req->rq_send_state = LUSTRE_IMP_REPLAY_LOCKS;
2497
2498         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
2499         ldlm_lock2desc(lock, &body->lock_desc);
2500         body->lock_flags = ldlm_flags_to_wire(flags);
2501
2502         ldlm_lock2handle(lock, &body->lock_handle[0]);
2503         if (lock->l_lvb_len > 0)
2504                 req_capsule_extend(&req->rq_pill, &RQF_LDLM_ENQUEUE_LVB);
2505         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
2506                              lock->l_lvb_len);
2507         ptlrpc_request_set_replen(req);
2508         /*
2509          * notify the server we've replayed all requests.
2510          * also, we mark the request to be put on a dedicated
2511          * queue to be processed after all request replayes.
2512          * b=6063
2513          */
2514         lustre_msg_set_flags(req->rq_reqmsg, MSG_REQ_REPLAY_DONE);
2515
2516         LDLM_DEBUG(lock, "replaying lock:");
2517
2518         atomic_inc(&req->rq_import->imp_replay_inflight);
2519         aa = ptlrpc_req_async_args(aa, req);
2520         aa->lock_handle = body->lock_handle[0];
2521         req->rq_interpret_reply = replay_lock_interpret;
2522         ptlrpcd_add_req(req);
2523
2524         RETURN(0);
2525 }
2526
2527 /**
2528  * Cancel as many unused locks as possible before replay. since we are
2529  * in recovery, we can't wait for any outstanding RPCs to send any RPC
2530  * to the server.
2531  *
2532  * Called only in recovery before replaying locks. there is no need to
2533  * replay locks that are unused. since the clients may hold thousands of
2534  * cached unused locks, dropping the unused locks can greatly reduce the
2535  * load on the servers at recovery time.
2536  */
2537 static void ldlm_cancel_unused_locks_for_replay(struct ldlm_namespace *ns)
2538 {
2539         int canceled;
2540         LIST_HEAD(cancels);
2541
2542         CDEBUG(D_DLMTRACE,
2543                "Dropping as many unused locks as possible before replay for namespace %s (%d)\n",
2544                ldlm_ns_name(ns), ns->ns_nr_unused);
2545
2546         /*
2547          * We don't need to care whether or not LRU resize is enabled
2548          * because the LDLM_LRU_FLAG_NO_WAIT policy doesn't use the
2549          * count parameter
2550          */
2551         canceled = ldlm_cancel_lru_local(ns, &cancels, ns->ns_nr_unused, 0,
2552                                          LCF_LOCAL, LDLM_LRU_FLAG_NO_WAIT);
2553
2554         CDEBUG(D_DLMTRACE, "Canceled %d unused locks from namespace %s\n",
2555                            canceled, ldlm_ns_name(ns));
2556 }
2557
2558 int ldlm_replay_locks(struct obd_import *imp)
2559 {
2560         struct ldlm_namespace *ns = imp->imp_obd->obd_namespace;
2561         LIST_HEAD(list);
2562         struct ldlm_lock *lock, *next;
2563         int rc = 0;
2564
2565         ENTRY;
2566
2567         LASSERT(atomic_read(&imp->imp_replay_inflight) == 0);
2568
2569         /* don't replay locks if import failed recovery */
2570         if (imp->imp_vbr_failed)
2571                 RETURN(0);
2572
2573         /* ensure this doesn't fall to 0 before all have been queued */
2574         atomic_inc(&imp->imp_replay_inflight);
2575
2576         if (ldlm_cancel_unused_locks_before_replay)
2577                 ldlm_cancel_unused_locks_for_replay(ns);
2578
2579         ldlm_namespace_foreach(ns, ldlm_chain_lock_for_replay, &list);
2580
2581         list_for_each_entry_safe(lock, next, &list, l_pending_chain) {
2582                 list_del_init(&lock->l_pending_chain);
2583                 if (rc) {
2584                         LDLM_LOCK_RELEASE(lock);
2585                         continue; /* or try to do the rest? */
2586                 }
2587                 rc = replay_one_lock(imp, lock);
2588                 LDLM_LOCK_RELEASE(lock);
2589         }
2590
2591         atomic_dec(&imp->imp_replay_inflight);
2592
2593         RETURN(rc);
2594 }