Whamcloud - gitweb
LU-9679 modules: Use LIST_HEAD for declaring list_heads
[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 /**
587  * Finishing portion of client lock enqueue code.
588  *
589  * Called after receiving reply from server.
590  */
591 int ldlm_cli_enqueue_fini(struct obd_export *exp, struct ptlrpc_request *req,
592                           enum ldlm_type type, __u8 with_policy,
593                           enum ldlm_mode mode, __u64 *flags, void *lvb,
594                           __u32 lvb_len, const struct lustre_handle *lockh,
595                           int rc)
596 {
597         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
598         const struct lu_env *env = NULL;
599         int is_replay = *flags & LDLM_FL_REPLAY;
600         struct ldlm_lock *lock;
601         struct ldlm_reply *reply;
602         int cleanup_phase = 1;
603
604         ENTRY;
605
606         if (req && req->rq_svc_thread)
607                 env = req->rq_svc_thread->t_env;
608
609         lock = ldlm_handle2lock(lockh);
610         /* ldlm_cli_enqueue is holding a reference on this lock. */
611         if (!lock) {
612                 LASSERT(type == LDLM_FLOCK);
613                 RETURN(-ENOLCK);
614         }
615
616         LASSERTF(ergo(lvb_len != 0, lvb_len == lock->l_lvb_len),
617                  "lvb_len = %d, l_lvb_len = %d\n", lvb_len, lock->l_lvb_len);
618
619         if (rc != ELDLM_OK) {
620                 LASSERT(!is_replay);
621                 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
622                            rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
623
624                 if (rc != ELDLM_LOCK_ABORTED)
625                         GOTO(cleanup, rc);
626         }
627
628         /* Before we return, swab the reply */
629         reply = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
630         if (reply == NULL)
631                 GOTO(cleanup, rc = -EPROTO);
632
633         if (lvb_len > 0) {
634                 int size = 0;
635
636                 size = req_capsule_get_size(&req->rq_pill, &RMF_DLM_LVB,
637                                             RCL_SERVER);
638                 if (size < 0) {
639                         LDLM_ERROR(lock, "Fail to get lvb_len, rc = %d", size);
640                         GOTO(cleanup, rc = size);
641                 } else if (unlikely(size > lvb_len)) {
642                         LDLM_ERROR(lock,
643                                    "Replied LVB is larger than expectation, expected = %d, replied = %d",
644                                    lvb_len, size);
645                         GOTO(cleanup, rc = -EINVAL);
646                 }
647                 lvb_len = size;
648         }
649
650         if (rc == ELDLM_LOCK_ABORTED) {
651                 if (lvb_len > 0 && lvb != NULL)
652                         rc = ldlm_fill_lvb(lock, &req->rq_pill, RCL_SERVER,
653                                            lvb, lvb_len);
654                 GOTO(cleanup, rc = rc ? : ELDLM_LOCK_ABORTED);
655         }
656
657         /* lock enqueued on the server */
658         cleanup_phase = 0;
659
660         lock_res_and_lock(lock);
661         /* Key change rehash lock in per-export hash with new key */
662         if (exp->exp_lock_hash) {
663                 /*
664                  * In the function below, .hs_keycmp resolves to
665                  * ldlm_export_lock_keycmp()
666                  */
667                 /* coverity[overrun-buffer-val] */
668                 cfs_hash_rehash_key(exp->exp_lock_hash,
669                                     &lock->l_remote_handle,
670                                     &reply->lock_handle,
671                                     &lock->l_exp_hash);
672         } else {
673                 lock->l_remote_handle = reply->lock_handle;
674         }
675
676         *flags = ldlm_flags_from_wire(reply->lock_flags);
677         lock->l_flags |= ldlm_flags_from_wire(reply->lock_flags &
678                                               LDLM_FL_INHERIT_MASK);
679         unlock_res_and_lock(lock);
680
681         CDEBUG(D_INFO, "local: %p, remote cookie: %#llx, flags: %#llx\n",
682                lock, reply->lock_handle.cookie, *flags);
683
684         /*
685          * If enqueue returned a blocked lock but the completion handler has
686          * already run, then it fixed up the resource and we don't need to do it
687          * again.
688          */
689         if ((*flags) & LDLM_FL_LOCK_CHANGED) {
690                 int newmode = reply->lock_desc.l_req_mode;
691
692                 LASSERT(!is_replay);
693                 if (newmode && newmode != lock->l_req_mode) {
694                         LDLM_DEBUG(lock, "server returned different mode %s",
695                                    ldlm_lockname[newmode]);
696                         lock->l_req_mode = newmode;
697                 }
698
699                 if (!ldlm_res_eq(&reply->lock_desc.l_resource.lr_name,
700                                  &lock->l_resource->lr_name)) {
701                         CDEBUG(D_INFO,
702                                "remote intent success, locking "DLDLMRES", instead of "DLDLMRES"\n",
703                                PLDLMRES(&reply->lock_desc.l_resource),
704                                PLDLMRES(lock->l_resource));
705
706                         rc = ldlm_lock_change_resource(ns, lock,
707                                         &reply->lock_desc.l_resource.lr_name);
708                         if (rc || lock->l_resource == NULL)
709                                 GOTO(cleanup, rc = -ENOMEM);
710                         LDLM_DEBUG(lock, "client-side enqueue, new resource");
711                 }
712
713                 if (with_policy) {
714                         /* We assume lock type cannot change on server*/
715                         ldlm_convert_policy_to_local(exp,
716                                                 lock->l_resource->lr_type,
717                                                 &reply->lock_desc.l_policy_data,
718                                                 &lock->l_policy_data);
719                 }
720
721                 if (type != LDLM_PLAIN)
722                         LDLM_DEBUG(lock,
723                                    "client-side enqueue, new policy data");
724         }
725
726         if ((*flags) & LDLM_FL_AST_SENT) {
727                 lock_res_and_lock(lock);
728                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_BL_AST;
729                 unlock_res_and_lock(lock);
730                 LDLM_DEBUG(lock, "enqueue reply includes blocking AST");
731         }
732
733         /*
734          * If the lock has already been granted by a completion AST, don't
735          * clobber the LVB with an older one.
736          */
737         if (lvb_len > 0) {
738                 /*
739                  * We must lock or a racing completion might update lvb without
740                  * letting us know and we'll clobber the correct value.
741                  * Cannot unlock after the check either, a that still leaves
742                  * a tiny window for completion to get in
743                  */
744                 lock_res_and_lock(lock);
745                 if (!ldlm_is_granted(lock))
746                         rc = ldlm_fill_lvb(lock, &req->rq_pill, RCL_SERVER,
747                                            lock->l_lvb_data, lvb_len);
748                 unlock_res_and_lock(lock);
749                 if (rc < 0) {
750                         cleanup_phase = 1;
751                         GOTO(cleanup, rc);
752                 }
753         }
754
755         if (!is_replay) {
756                 rc = ldlm_lock_enqueue(env, ns, &lock, NULL, flags);
757                 if (lock->l_completion_ast != NULL) {
758                         int err = lock->l_completion_ast(lock, *flags, NULL);
759
760                         if (!rc)
761                                 rc = err;
762                         if (rc)
763                                 cleanup_phase = 1;
764                 }
765         }
766
767         if (lvb_len > 0 && lvb != NULL) {
768                 /*
769                  * Copy the LVB here, and not earlier, because the completion
770                  * AST (if any) can override what we got in the reply
771                  */
772                 memcpy(lvb, lock->l_lvb_data, lvb_len);
773         }
774
775         LDLM_DEBUG(lock, "client-side enqueue END");
776         EXIT;
777 cleanup:
778         if (cleanup_phase == 1 && rc)
779                 failed_lock_cleanup(ns, lock, mode);
780         /* Put lock 2 times, the second reference is held by ldlm_cli_enqueue */
781         LDLM_LOCK_PUT(lock);
782         LDLM_LOCK_RELEASE(lock);
783         return rc;
784 }
785 EXPORT_SYMBOL(ldlm_cli_enqueue_fini);
786
787 /**
788  * Estimate number of lock handles that would fit into request of given
789  * size.  PAGE_SIZE-512 is to allow TCP/IP and LNET headers to fit into
790  * a single page on the send/receive side. XXX: 512 should be changed to
791  * more adequate value.
792  */
793 static inline int ldlm_req_handles_avail(int req_size, int off)
794 {
795         int avail;
796
797         avail = min_t(int, LDLM_MAXREQSIZE, PAGE_SIZE - 512) - req_size;
798         if (likely(avail >= 0))
799                 avail /= (int)sizeof(struct lustre_handle);
800         else
801                 avail = 0;
802         avail += LDLM_LOCKREQ_HANDLES - off;
803
804         return avail;
805 }
806
807 static inline int ldlm_capsule_handles_avail(struct req_capsule *pill,
808                                              enum req_location loc,
809                                              int off)
810 {
811         __u32 size = req_capsule_msg_size(pill, loc);
812
813         return ldlm_req_handles_avail(size, off);
814 }
815
816 static inline int ldlm_format_handles_avail(struct obd_import *imp,
817                                             const struct req_format *fmt,
818                                             enum req_location loc, int off)
819 {
820         __u32 size = req_capsule_fmt_size(imp->imp_msg_magic, fmt, loc);
821
822         return ldlm_req_handles_avail(size, off);
823 }
824
825 /**
826  * Cancel LRU locks and pack them into the enqueue request. Pack there the given
827  * \a count locks in \a cancels.
828  *
829  * This is to be called by functions preparing their own requests that
830  * might contain lists of locks to cancel in addition to actual operation
831  * that needs to be performed.
832  */
833 int ldlm_prep_elc_req(struct obd_export *exp, struct ptlrpc_request *req,
834                       int version, int opc, int canceloff,
835                       struct list_head *cancels, int count)
836 {
837         struct ldlm_namespace   *ns = exp->exp_obd->obd_namespace;
838         struct req_capsule      *pill = &req->rq_pill;
839         struct ldlm_request     *dlm = NULL;
840         LIST_HEAD(head);
841         enum ldlm_lru_flags lru_flags;
842         int avail, to_free, pack = 0;
843         int rc;
844
845         ENTRY;
846
847         if (cancels == NULL)
848                 cancels = &head;
849         if (ns_connect_cancelset(ns)) {
850                 /* Estimate the amount of available space in the request. */
851                 req_capsule_filled_sizes(pill, RCL_CLIENT);
852                 avail = ldlm_capsule_handles_avail(pill, RCL_CLIENT, canceloff);
853
854                 lru_flags = LDLM_LRU_FLAG_NO_WAIT | (ns_connect_lru_resize(ns) ?
855                         LDLM_LRU_FLAG_LRUR : LDLM_LRU_FLAG_AGED);
856                 to_free = !ns_connect_lru_resize(ns) &&
857                         opc == LDLM_ENQUEUE ? 1 : 0;
858
859                 /*
860                  * Cancel LRU locks here _only_ if the server supports
861                  * EARLY_CANCEL. Otherwise we have to send extra CANCEL
862                  * RPC, which will make us slower.
863                  */
864                 if (avail > count)
865                         count += ldlm_cancel_lru_local(ns, cancels, to_free,
866                                                        avail - count, 0,
867                                                        lru_flags);
868                 if (avail > count)
869                         pack = count;
870                 else
871                         pack = avail;
872                 req_capsule_set_size(pill, &RMF_DLM_REQ, RCL_CLIENT,
873                                      ldlm_request_bufsize(pack, opc));
874         }
875
876         rc = ptlrpc_request_pack(req, version, opc);
877         if (rc) {
878                 ldlm_lock_list_put(cancels, l_bl_ast, count);
879                 RETURN(rc);
880         }
881
882         if (ns_connect_cancelset(ns)) {
883                 if (canceloff) {
884                         dlm = req_capsule_client_get(pill, &RMF_DLM_REQ);
885                         LASSERT(dlm);
886                         /*
887                          * Skip first lock handler in ldlm_request_pack(),
888                          * this method will increment @lock_count according
889                          * to the lock handle amount actually written to
890                          * the buffer.
891                          */
892                         dlm->lock_count = canceloff;
893                 }
894                 /* Pack into the request @pack lock handles. */
895                 ldlm_cli_cancel_list(cancels, pack, req, 0);
896                 /* Prepare and send separate cancel RPC for others. */
897                 ldlm_cli_cancel_list(cancels, count - pack, NULL, 0);
898         } else {
899                 ldlm_lock_list_put(cancels, l_bl_ast, count);
900         }
901         RETURN(0);
902 }
903 EXPORT_SYMBOL(ldlm_prep_elc_req);
904
905 int ldlm_prep_enqueue_req(struct obd_export *exp, struct ptlrpc_request *req,
906                           struct list_head *cancels, int count)
907 {
908         return ldlm_prep_elc_req(exp, req, LUSTRE_DLM_VERSION, LDLM_ENQUEUE,
909                                  LDLM_ENQUEUE_CANCEL_OFF, cancels, count);
910 }
911 EXPORT_SYMBOL(ldlm_prep_enqueue_req);
912
913 struct ptlrpc_request *ldlm_enqueue_pack(struct obd_export *exp, int lvb_len)
914 {
915         struct ptlrpc_request *req;
916         int rc;
917
918         ENTRY;
919
920         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LDLM_ENQUEUE);
921         if (req == NULL)
922                 RETURN(ERR_PTR(-ENOMEM));
923
924         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
925         if (rc) {
926                 ptlrpc_request_free(req);
927                 RETURN(ERR_PTR(rc));
928         }
929
930         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, lvb_len);
931         ptlrpc_request_set_replen(req);
932         RETURN(req);
933 }
934 EXPORT_SYMBOL(ldlm_enqueue_pack);
935
936 /**
937  * Client-side lock enqueue.
938  *
939  * If a request has some specific initialisation it is passed in \a reqp,
940  * otherwise it is created in ldlm_cli_enqueue.
941  *
942  * Supports sync and async requests, pass \a async flag accordingly. If a
943  * request was created in ldlm_cli_enqueue and it is the async request,
944  * pass it to the caller in \a reqp.
945  */
946 int ldlm_cli_enqueue(struct obd_export *exp, struct ptlrpc_request **reqp,
947                      struct ldlm_enqueue_info *einfo,
948                      const struct ldlm_res_id *res_id,
949                      union ldlm_policy_data const *policy, __u64 *flags,
950                      void *lvb, __u32 lvb_len, enum lvb_type lvb_type,
951                      struct lustre_handle *lockh, int async)
952 {
953         struct ldlm_namespace *ns;
954         struct ldlm_lock      *lock;
955         struct ldlm_request   *body;
956         int                    is_replay = *flags & LDLM_FL_REPLAY;
957         int                    req_passed_in = 1;
958         int                    rc, err;
959         struct ptlrpc_request *req;
960
961         ENTRY;
962
963         LASSERT(exp != NULL);
964
965         ns = exp->exp_obd->obd_namespace;
966
967         /*
968          * If we're replaying this lock, just check some invariants.
969          * If we're creating a new lock, get everything all setup nice.
970          */
971         if (is_replay) {
972                 lock = ldlm_handle2lock_long(lockh, 0);
973                 LASSERT(lock != NULL);
974                 LDLM_DEBUG(lock, "client-side enqueue START");
975                 LASSERT(exp == lock->l_conn_export);
976         } else {
977                 const struct ldlm_callback_suite cbs = {
978                         .lcs_completion = einfo->ei_cb_cp,
979                         .lcs_blocking   = einfo->ei_cb_bl,
980                         .lcs_glimpse    = einfo->ei_cb_gl
981                 };
982                 lock = ldlm_lock_create(ns, res_id, einfo->ei_type,
983                                         einfo->ei_mode, &cbs, einfo->ei_cbdata,
984                                         lvb_len, lvb_type);
985                 if (IS_ERR(lock))
986                         RETURN(PTR_ERR(lock));
987
988                 if (einfo->ei_cb_created)
989                         einfo->ei_cb_created(lock);
990
991                 /* for the local lock, add the reference */
992                 ldlm_lock_addref_internal(lock, einfo->ei_mode);
993                 ldlm_lock2handle(lock, lockh);
994                 if (policy != NULL)
995                         lock->l_policy_data = *policy;
996
997                 if (einfo->ei_type == LDLM_EXTENT) {
998                         /* extent lock without policy is a bug */
999                         if (policy == NULL)
1000                                 LBUG();
1001
1002                         lock->l_req_extent = policy->l_extent;
1003                 }
1004                 LDLM_DEBUG(lock, "client-side enqueue START, flags %#llx",
1005                            *flags);
1006         }
1007
1008         lock->l_conn_export = exp;
1009         lock->l_export = NULL;
1010         lock->l_blocking_ast = einfo->ei_cb_bl;
1011         lock->l_flags |= (*flags & (LDLM_FL_NO_LRU | LDLM_FL_EXCL |
1012                                     LDLM_FL_ATOMIC_CB));
1013         lock->l_activity = ktime_get_real_seconds();
1014
1015         /* lock not sent to server yet */
1016         if (reqp == NULL || *reqp == NULL) {
1017                 req = ldlm_enqueue_pack(exp, lvb_len);
1018                 if (IS_ERR(req)) {
1019                         failed_lock_cleanup(ns, lock, einfo->ei_mode);
1020                         LDLM_LOCK_RELEASE(lock);
1021                         RETURN(PTR_ERR(req));
1022                 }
1023
1024                 req_passed_in = 0;
1025                 if (reqp)
1026                         *reqp = req;
1027         } else {
1028                 int len;
1029
1030                 req = *reqp;
1031                 len = req_capsule_get_size(&req->rq_pill, &RMF_DLM_REQ,
1032                                            RCL_CLIENT);
1033                 LASSERTF(len >= sizeof(*body), "buflen[%d] = %d, not %d\n",
1034                          DLM_LOCKREQ_OFF, len, (int)sizeof(*body));
1035         }
1036
1037         if (*flags & LDLM_FL_NDELAY) {
1038                 DEBUG_REQ(D_DLMTRACE, req, "enqueue lock with no delay");
1039                 req->rq_no_resend = req->rq_no_delay = 1;
1040                 /*
1041                  * probably set a shorter timeout value and handle ETIMEDOUT
1042                  * in osc_lock_upcall() correctly
1043                  */
1044                 /* lustre_msg_set_timeout(req, req->rq_timeout / 2); */
1045         }
1046
1047         /* Dump lock data into the request buffer */
1048         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1049         ldlm_lock2desc(lock, &body->lock_desc);
1050         body->lock_flags = ldlm_flags_to_wire(*flags);
1051         body->lock_handle[0] = *lockh;
1052
1053         /* extended LDLM opcodes in client stats */
1054         if (exp->exp_obd->obd_svc_stats != NULL) {
1055                 bool glimpse = *flags & LDLM_FL_HAS_INTENT;
1056
1057                 /* OST glimpse has no intent buffer */
1058                 if (req_capsule_has_field(&req->rq_pill, &RMF_LDLM_INTENT,
1059                                           RCL_CLIENT)) {
1060                         struct ldlm_intent *it;
1061
1062                         it = req_capsule_client_get(&req->rq_pill,
1063                                                     &RMF_LDLM_INTENT);
1064                         glimpse = (it && (it->opc == IT_GLIMPSE));
1065                 }
1066
1067                 if (!glimpse)
1068                         ldlm_svc_get_eopc(body, exp->exp_obd->obd_svc_stats);
1069                 else
1070                         lprocfs_counter_incr(exp->exp_obd->obd_svc_stats,
1071                                              PTLRPC_LAST_CNTR +
1072                                              LDLM_GLIMPSE_ENQUEUE);
1073         }
1074
1075         if (async) {
1076                 LASSERT(reqp != NULL);
1077                 RETURN(0);
1078         }
1079
1080         LDLM_DEBUG(lock, "sending request");
1081
1082         rc = ptlrpc_queue_wait(req);
1083
1084         err = ldlm_cli_enqueue_fini(exp, req, einfo->ei_type, policy ? 1 : 0,
1085                                     einfo->ei_mode, flags, lvb, lvb_len,
1086                                     lockh, rc);
1087
1088         /*
1089          * If ldlm_cli_enqueue_fini did not find the lock, we need to free
1090          * one reference that we took
1091          */
1092         if (err == -ENOLCK)
1093                 LDLM_LOCK_RELEASE(lock);
1094         else
1095                 rc = err;
1096
1097         if (!req_passed_in && req != NULL) {
1098                 ptlrpc_req_finished(req);
1099                 if (reqp)
1100                         *reqp = NULL;
1101         }
1102
1103         RETURN(rc);
1104 }
1105 EXPORT_SYMBOL(ldlm_cli_enqueue);
1106
1107 /**
1108  * Client-side lock convert reply handling.
1109  *
1110  * Finish client lock converting, checks for concurrent converts
1111  * and clear 'converting' flag so lock can be placed back into LRU.
1112  */
1113 static int lock_convert_interpret(const struct lu_env *env,
1114                                   struct ptlrpc_request *req,
1115                                   void *args, int rc)
1116 {
1117         struct ldlm_async_args *aa = args;
1118         struct ldlm_lock *lock;
1119         struct ldlm_reply *reply;
1120
1121         ENTRY;
1122
1123         lock = ldlm_handle2lock(&aa->lock_handle);
1124         if (!lock) {
1125                 LDLM_DEBUG_NOLOCK("convert ACK for unknown local cookie %#llx",
1126                         aa->lock_handle.cookie);
1127                 RETURN(-ESTALE);
1128         }
1129
1130         LDLM_DEBUG(lock, "CONVERTED lock:");
1131
1132         if (rc != ELDLM_OK)
1133                 GOTO(out, rc);
1134
1135         reply = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1136         if (reply == NULL)
1137                 GOTO(out, rc = -EPROTO);
1138
1139         if (reply->lock_handle.cookie != aa->lock_handle.cookie) {
1140                 LDLM_ERROR(lock,
1141                            "convert ACK with wrong lock cookie %#llx but cookie %#llx from server %s id %s\n",
1142                            aa->lock_handle.cookie, reply->lock_handle.cookie,
1143                            req->rq_export->exp_client_uuid.uuid,
1144                            libcfs_id2str(req->rq_peer));
1145                 GOTO(out, rc = ELDLM_NO_LOCK_DATA);
1146         }
1147
1148         lock_res_and_lock(lock);
1149         /*
1150          * Lock convert is sent for any new bits to drop, the converting flag
1151          * is dropped when ibits on server are the same as on client. Meanwhile
1152          * that can be so that more later convert will be replied first with
1153          * and clear converting flag, so in case of such race just exit here.
1154          * if lock has no converting bits then
1155          */
1156         if (!ldlm_is_converting(lock)) {
1157                 LDLM_DEBUG(lock,
1158                            "convert ACK for lock without converting flag, reply ibits %#llx",
1159                            reply->lock_desc.l_policy_data.l_inodebits.bits);
1160         } else if (reply->lock_desc.l_policy_data.l_inodebits.bits !=
1161                    lock->l_policy_data.l_inodebits.bits) {
1162                 /*
1163                  * Compare server returned lock ibits and local lock ibits
1164                  * if they are the same we consider convertion is done,
1165                  * otherwise we have more converts inflight and keep
1166                  * converting flag.
1167                  */
1168                 LDLM_DEBUG(lock, "convert ACK with ibits %#llx\n",
1169                            reply->lock_desc.l_policy_data.l_inodebits.bits);
1170         } else {
1171                 ldlm_clear_converting(lock);
1172
1173                 /*
1174                  * Concurrent BL AST may arrive and cause another convert
1175                  * or cancel so just do nothing here if bl_ast is set,
1176                  * finish with convert otherwise.
1177                  */
1178                 if (!ldlm_is_bl_ast(lock)) {
1179                         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
1180
1181                         /*
1182                          * Drop cancel_bits since there are no more converts
1183                          * and put lock into LRU if it is still not used and
1184                          * is not there yet.
1185                          */
1186                         lock->l_policy_data.l_inodebits.cancel_bits = 0;
1187                         if (!lock->l_readers && !lock->l_writers &&
1188                             !ldlm_is_canceling(lock)) {
1189                                 spin_lock(&ns->ns_lock);
1190                                 /* there is check for list_empty() inside */
1191                                 ldlm_lock_remove_from_lru_nolock(lock);
1192                                 ldlm_lock_add_to_lru_nolock(lock);
1193                                 spin_unlock(&ns->ns_lock);
1194                         }
1195                 }
1196         }
1197         unlock_res_and_lock(lock);
1198 out:
1199         if (rc) {
1200                 int flag;
1201
1202                 lock_res_and_lock(lock);
1203                 if (ldlm_is_converting(lock)) {
1204                         ldlm_clear_converting(lock);
1205                         ldlm_set_cbpending(lock);
1206                         ldlm_set_bl_ast(lock);
1207                         lock->l_policy_data.l_inodebits.cancel_bits = 0;
1208                 }
1209                 unlock_res_and_lock(lock);
1210
1211                 /*
1212                  * fallback to normal lock cancel. If rc means there is no
1213                  * valid lock on server, do only local cancel
1214                  */
1215                 if (rc == ELDLM_NO_LOCK_DATA)
1216                         flag = LCF_LOCAL;
1217                 else
1218                         flag = LCF_ASYNC;
1219
1220                 rc = ldlm_cli_cancel(&aa->lock_handle, flag);
1221                 if (rc < 0)
1222                         LDLM_DEBUG(lock, "failed to cancel lock: rc = %d\n",
1223                                    rc);
1224         }
1225         LDLM_LOCK_PUT(lock);
1226         RETURN(rc);
1227 }
1228
1229 /**
1230  * Client-side IBITS lock convert.
1231  *
1232  * Inform server that lock has been converted instead of canceling.
1233  * Server finishes convert on own side and does reprocess to grant
1234  * all related waiting locks.
1235  *
1236  * Since convert means only ibits downgrading, client doesn't need to
1237  * wait for server reply to finish local converting process so this request
1238  * is made asynchronous.
1239  *
1240  */
1241 int ldlm_cli_convert(struct ldlm_lock *lock, __u32 *flags)
1242 {
1243         struct ldlm_request *body;
1244         struct ptlrpc_request *req;
1245         struct ldlm_async_args *aa;
1246         struct obd_export *exp = lock->l_conn_export;
1247
1248         ENTRY;
1249
1250         if (exp == NULL) {
1251                 LDLM_ERROR(lock, "convert must not be called on local locks.");
1252                 RETURN(-EINVAL);
1253         }
1254
1255         /*
1256          * this is better to check earlier and it is done so already,
1257          * but this check is kept too as final one to issue an error
1258          * if any new code will miss such check.
1259          */
1260         if (!exp_connect_lock_convert(exp)) {
1261                 LDLM_ERROR(lock, "server doesn't support lock convert\n");
1262                 RETURN(-EPROTO);
1263         }
1264
1265         if (lock->l_resource->lr_type != LDLM_IBITS) {
1266                 LDLM_ERROR(lock, "convert works with IBITS locks only.");
1267                 RETURN(-EINVAL);
1268         }
1269
1270         LDLM_DEBUG(lock, "client-side convert");
1271
1272         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
1273                                         &RQF_LDLM_CONVERT, LUSTRE_DLM_VERSION,
1274                                         LDLM_CONVERT);
1275         if (req == NULL)
1276                 RETURN(-ENOMEM);
1277
1278         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1279         body->lock_handle[0] = lock->l_remote_handle;
1280
1281         body->lock_desc.l_req_mode = lock->l_req_mode;
1282         body->lock_desc.l_granted_mode = lock->l_granted_mode;
1283
1284         body->lock_desc.l_policy_data.l_inodebits.bits =
1285                                         lock->l_policy_data.l_inodebits.bits;
1286         body->lock_desc.l_policy_data.l_inodebits.cancel_bits = 0;
1287
1288         body->lock_flags = ldlm_flags_to_wire(*flags);
1289         body->lock_count = 1;
1290
1291         ptlrpc_request_set_replen(req);
1292
1293         /*
1294          * Use cancel portals for convert as well as high-priority handling.
1295          */
1296         req->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
1297         req->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
1298
1299         ptlrpc_at_set_req_timeout(req);
1300
1301         if (exp->exp_obd->obd_svc_stats != NULL)
1302                 lprocfs_counter_incr(exp->exp_obd->obd_svc_stats,
1303                                      LDLM_CONVERT - LDLM_FIRST_OPC);
1304
1305         aa = ptlrpc_req_async_args(aa, req);
1306         ldlm_lock2handle(lock, &aa->lock_handle);
1307         req->rq_interpret_reply = lock_convert_interpret;
1308
1309         ptlrpcd_add_req(req);
1310         RETURN(0);
1311 }
1312
1313 /**
1314  * Cancel locks locally.
1315  * Returns:
1316  * \retval LDLM_FL_LOCAL_ONLY if there is no need for a CANCEL RPC to the server
1317  * \retval LDLM_FL_CANCELING otherwise;
1318  * \retval LDLM_FL_BL_AST if there is a need for a separate CANCEL RPC.
1319  */
1320 static __u64 ldlm_cli_cancel_local(struct ldlm_lock *lock)
1321 {
1322         __u64 rc = LDLM_FL_LOCAL_ONLY;
1323
1324         ENTRY;
1325
1326         if (lock->l_conn_export) {
1327                 bool local_only;
1328
1329                 LDLM_DEBUG(lock, "client-side cancel");
1330                 /* Set this flag to prevent others from getting new references*/
1331                 lock_res_and_lock(lock);
1332                 ldlm_set_cbpending(lock);
1333                 local_only = !!(lock->l_flags &
1334                                 (LDLM_FL_LOCAL_ONLY|LDLM_FL_CANCEL_ON_BLOCK));
1335                 ldlm_cancel_callback(lock);
1336                 rc = (ldlm_is_bl_ast(lock)) ?
1337                         LDLM_FL_BL_AST : LDLM_FL_CANCELING;
1338                 unlock_res_and_lock(lock);
1339
1340                 if (local_only) {
1341                         CDEBUG(D_DLMTRACE,
1342                                "not sending request (at caller's instruction)\n");
1343                         rc = LDLM_FL_LOCAL_ONLY;
1344                 }
1345                 ldlm_lock_cancel(lock);
1346         } else {
1347                 if (ns_is_client(ldlm_lock_to_ns(lock))) {
1348                         LDLM_ERROR(lock, "Trying to cancel local lock");
1349                         LBUG();
1350                 }
1351                 LDLM_DEBUG(lock, "server-side local cancel");
1352                 ldlm_lock_cancel(lock);
1353                 ldlm_reprocess_all(lock->l_resource, lock);
1354         }
1355
1356         RETURN(rc);
1357 }
1358
1359 /**
1360  * Pack \a count locks in \a head into ldlm_request buffer of request \a req.
1361  */
1362 static void ldlm_cancel_pack(struct ptlrpc_request *req,
1363                              struct list_head *head, int count)
1364 {
1365         struct ldlm_request *dlm;
1366         struct ldlm_lock *lock;
1367         int max, packed = 0;
1368
1369         ENTRY;
1370
1371         dlm = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1372         LASSERT(dlm != NULL);
1373
1374         /* Check the room in the request buffer. */
1375         max = req_capsule_get_size(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT) -
1376                 sizeof(struct ldlm_request);
1377         max /= sizeof(struct lustre_handle);
1378         max += LDLM_LOCKREQ_HANDLES;
1379         LASSERT(max >= dlm->lock_count + count);
1380
1381         /*
1382          * XXX: it would be better to pack lock handles grouped by resource.
1383          * so that the server cancel would call filter_lvbo_update() less
1384          * frequently.
1385          */
1386         list_for_each_entry(lock, head, l_bl_ast) {
1387                 if (!count--)
1388                         break;
1389                 LASSERT(lock->l_conn_export);
1390                 /* Pack the lock handle to the given request buffer. */
1391                 LDLM_DEBUG(lock, "packing");
1392                 dlm->lock_handle[dlm->lock_count++] = lock->l_remote_handle;
1393                 packed++;
1394         }
1395         CDEBUG(D_DLMTRACE, "%d locks packed\n", packed);
1396         EXIT;
1397 }
1398
1399 /**
1400  * Prepare and send a batched cancel RPC. It will include \a count lock
1401  * handles of locks given in \a cancels list.
1402  */
1403 int ldlm_cli_cancel_req(struct obd_export *exp, struct list_head *cancels,
1404                         int count, enum ldlm_cancel_flags flags)
1405 {
1406         struct ptlrpc_request *req = NULL;
1407         struct obd_import *imp;
1408         int free, sent = 0;
1409         int rc = 0;
1410
1411         ENTRY;
1412
1413         LASSERT(exp != NULL);
1414         LASSERT(count > 0);
1415
1416         CFS_FAIL_TIMEOUT(OBD_FAIL_LDLM_PAUSE_CANCEL, cfs_fail_val);
1417
1418         if (CFS_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_RACE))
1419                 RETURN(count);
1420
1421         free = ldlm_format_handles_avail(class_exp2cliimp(exp),
1422                                          &RQF_LDLM_CANCEL, RCL_CLIENT, 0);
1423         if (count > free)
1424                 count = free;
1425
1426         while (1) {
1427                 imp = class_exp2cliimp(exp);
1428                 if (imp == NULL || imp->imp_invalid) {
1429                         CDEBUG(D_DLMTRACE,
1430                                "skipping cancel on invalid import %p\n", imp);
1431                         RETURN(count);
1432                 }
1433
1434                 req = ptlrpc_request_alloc(imp, &RQF_LDLM_CANCEL);
1435                 if (req == NULL)
1436                         GOTO(out, rc = -ENOMEM);
1437
1438                 req_capsule_filled_sizes(&req->rq_pill, RCL_CLIENT);
1439                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT,
1440                                      ldlm_request_bufsize(count, LDLM_CANCEL));
1441
1442                 rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_CANCEL);
1443                 if (rc) {
1444                         ptlrpc_request_free(req);
1445                         GOTO(out, rc);
1446                 }
1447
1448                 /*
1449                  * If OSP want cancel cross-MDT lock, let's not block it in
1450                  * in recovery, otherwise the lock will not released, if
1451                  * the remote target is also in recovery, and it also need
1452                  * this lock, it might cause deadlock.
1453                  */
1454                 if (exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS &&
1455                     exp->exp_obd->obd_lu_dev != NULL &&
1456                     exp->exp_obd->obd_lu_dev->ld_site != NULL) {
1457                         struct lu_device *top_dev;
1458
1459                         top_dev = exp->exp_obd->obd_lu_dev->ld_site->ls_top_dev;
1460                         if (top_dev != NULL &&
1461                             top_dev->ld_obd->obd_recovering)
1462                                 req->rq_allow_replay = 1;
1463                 }
1464
1465                 req->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
1466                 req->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
1467                 ptlrpc_at_set_req_timeout(req);
1468
1469                 ldlm_cancel_pack(req, cancels, count);
1470
1471                 ptlrpc_request_set_replen(req);
1472                 if (flags & LCF_ASYNC) {
1473                         ptlrpcd_add_req(req);
1474                         sent = count;
1475                         GOTO(out, 0);
1476                 }
1477
1478                 rc = ptlrpc_queue_wait(req);
1479                 if (rc == LUSTRE_ESTALE) {
1480                         CDEBUG(D_DLMTRACE,
1481                                "client/server (nid %s) out of sync -- not fatal\n",
1482                                libcfs_nid2str(req->rq_import->imp_connection->c_peer.nid));
1483                         rc = 0;
1484                 } else if (rc == -ETIMEDOUT && /* check there was no reconnect*/
1485                            req->rq_import_generation == imp->imp_generation) {
1486                         ptlrpc_req_finished(req);
1487                         continue;
1488                 } else if (rc != ELDLM_OK) {
1489                         /* -ESHUTDOWN is common on umount */
1490                         CDEBUG_LIMIT(rc == -ESHUTDOWN ? D_DLMTRACE : D_ERROR,
1491                                      "Got rc %d from cancel RPC: canceling anyway\n",
1492                                      rc);
1493                         break;
1494                 }
1495                 sent = count;
1496                 break;
1497         }
1498
1499         ptlrpc_req_finished(req);
1500         EXIT;
1501 out:
1502         return sent ? sent : rc;
1503 }
1504
1505 static inline struct ldlm_pool *ldlm_imp2pl(struct obd_import *imp)
1506 {
1507         LASSERT(imp != NULL);
1508         return &imp->imp_obd->obd_namespace->ns_pool;
1509 }
1510
1511 /**
1512  * Update client's OBD pool related fields with new SLV and Limit from \a req.
1513  */
1514 int ldlm_cli_update_pool(struct ptlrpc_request *req)
1515 {
1516         struct obd_device *obd;
1517         __u64 new_slv;
1518         __u32 new_limit;
1519
1520         ENTRY;
1521         if (unlikely(!req->rq_import || !req->rq_import->imp_obd ||
1522                      !imp_connect_lru_resize(req->rq_import)))
1523                 /* Do nothing for corner cases. */
1524                 RETURN(0);
1525
1526         /*
1527          * In some cases RPC may contain SLV and limit zeroed out. This
1528          * is the case when server does not support LRU resize feature.
1529          * This is also possible in some recovery cases when server-side
1530          * reqs have no reference to the OBD export and thus access to
1531          * server-side namespace is not possible.
1532          */
1533         if (lustre_msg_get_slv(req->rq_repmsg) == 0 ||
1534             lustre_msg_get_limit(req->rq_repmsg) == 0) {
1535                 DEBUG_REQ(D_HA, req,
1536                           "Zero SLV or limit found (SLV=%llu, limit=%u)",
1537                           lustre_msg_get_slv(req->rq_repmsg),
1538                           lustre_msg_get_limit(req->rq_repmsg));
1539                 RETURN(0);
1540         }
1541
1542         new_limit = lustre_msg_get_limit(req->rq_repmsg);
1543         new_slv = lustre_msg_get_slv(req->rq_repmsg);
1544         obd = req->rq_import->imp_obd;
1545
1546         /*
1547          * Set new SLV and limit in OBD fields to make them accessible
1548          * to the pool thread. We do not access obd_namespace and pool
1549          * directly here as there is no reliable way to make sure that
1550          * they are still alive at cleanup time. Evil races are possible
1551          * which may cause Oops at that time.
1552          */
1553         write_lock(&obd->obd_pool_lock);
1554         obd->obd_pool_slv = new_slv;
1555         obd->obd_pool_limit = new_limit;
1556         write_unlock(&obd->obd_pool_lock);
1557
1558         RETURN(0);
1559 }
1560
1561 /**
1562  * Client side lock cancel.
1563  *
1564  * Lock must not have any readers or writers by this time.
1565  */
1566 int ldlm_cli_cancel(const struct lustre_handle *lockh,
1567                     enum ldlm_cancel_flags cancel_flags)
1568 {
1569         struct obd_export *exp;
1570         enum ldlm_lru_flags lru_flags;
1571         int avail, count = 1;
1572         __u64 rc = 0;
1573         struct ldlm_namespace *ns;
1574         struct ldlm_lock *lock;
1575         LIST_HEAD(cancels);
1576
1577         ENTRY;
1578
1579         lock = ldlm_handle2lock_long(lockh, 0);
1580         if (lock == NULL) {
1581                 LDLM_DEBUG_NOLOCK("lock is already being destroyed");
1582                 RETURN(0);
1583         }
1584
1585         /* Convert lock bits instead of cancel for IBITS locks */
1586         if (cancel_flags & LCF_CONVERT) {
1587                 LASSERT(lock->l_resource->lr_type == LDLM_IBITS);
1588                 LASSERT(lock->l_policy_data.l_inodebits.cancel_bits != 0);
1589
1590                 rc = ldlm_cli_dropbits(lock,
1591                                 lock->l_policy_data.l_inodebits.cancel_bits);
1592                 if (rc == 0) {
1593                         LDLM_LOCK_RELEASE(lock);
1594                         RETURN(0);
1595                 }
1596         }
1597
1598         lock_res_and_lock(lock);
1599         /* Lock is being canceled and the caller doesn't want to wait */
1600         if (ldlm_is_canceling(lock)) {
1601                 if (cancel_flags & LCF_ASYNC) {
1602                         unlock_res_and_lock(lock);
1603                 } else {
1604                         unlock_res_and_lock(lock);
1605                         wait_event_idle(lock->l_waitq, is_bl_done(lock));
1606                 }
1607                 LDLM_LOCK_RELEASE(lock);
1608                 RETURN(0);
1609         }
1610
1611         /*
1612          * Lock is being converted, cancel it immediately.
1613          * When convert will end, it releases lock and it will be gone.
1614          */
1615         if (ldlm_is_converting(lock)) {
1616                 /* set back flags removed by convert */
1617                 ldlm_set_cbpending(lock);
1618                 ldlm_set_bl_ast(lock);
1619         }
1620
1621         ldlm_set_canceling(lock);
1622         unlock_res_and_lock(lock);
1623
1624         if (cancel_flags & LCF_LOCAL)
1625                 OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_LOCAL_CANCEL_PAUSE,
1626                                  cfs_fail_val);
1627
1628         rc = ldlm_cli_cancel_local(lock);
1629         if (rc == LDLM_FL_LOCAL_ONLY || cancel_flags & LCF_LOCAL) {
1630                 LDLM_LOCK_RELEASE(lock);
1631                 RETURN(0);
1632         }
1633         /*
1634          * Even if the lock is marked as LDLM_FL_BL_AST, this is a LDLM_CANCEL
1635          * RPC which goes to canceld portal, so we can cancel other LRU locks
1636          * here and send them all as one LDLM_CANCEL RPC.
1637          */
1638         LASSERT(list_empty(&lock->l_bl_ast));
1639         list_add(&lock->l_bl_ast, &cancels);
1640
1641         exp = lock->l_conn_export;
1642         if (exp_connect_cancelset(exp)) {
1643                 avail = ldlm_format_handles_avail(class_exp2cliimp(exp),
1644                                                   &RQF_LDLM_CANCEL,
1645                                                   RCL_CLIENT, 0);
1646                 LASSERT(avail > 0);
1647
1648                 ns = ldlm_lock_to_ns(lock);
1649                 lru_flags = ns_connect_lru_resize(ns) ?
1650                         LDLM_LRU_FLAG_LRUR : LDLM_LRU_FLAG_AGED;
1651                 count += ldlm_cancel_lru_local(ns, &cancels, 0, avail - 1,
1652                                                LCF_BL_AST, lru_flags);
1653         }
1654         ldlm_cli_cancel_list(&cancels, count, NULL, cancel_flags);
1655         RETURN(0);
1656 }
1657 EXPORT_SYMBOL(ldlm_cli_cancel);
1658
1659 /**
1660  * Locally cancel up to \a count locks in list \a cancels.
1661  * Return the number of cancelled locks.
1662  */
1663 int ldlm_cli_cancel_list_local(struct list_head *cancels, int count,
1664                                enum ldlm_cancel_flags cancel_flags)
1665 {
1666         LIST_HEAD(head);
1667         struct ldlm_lock *lock, *next;
1668         int left = 0, bl_ast = 0;
1669         __u64 rc;
1670
1671         left = count;
1672         list_for_each_entry_safe(lock, next, cancels, l_bl_ast) {
1673                 if (left-- == 0)
1674                         break;
1675
1676                 if (cancel_flags & LCF_LOCAL) {
1677                         rc = LDLM_FL_LOCAL_ONLY;
1678                         ldlm_lock_cancel(lock);
1679                 } else {
1680                         rc = ldlm_cli_cancel_local(lock);
1681                 }
1682                 /*
1683                  * Until we have compound requests and can send LDLM_CANCEL
1684                  * requests batched with generic RPCs, we need to send cancels
1685                  * with the LDLM_FL_BL_AST flag in a separate RPC from
1686                  * the one being generated now.
1687                  */
1688                 if (!(cancel_flags & LCF_BL_AST) && (rc == LDLM_FL_BL_AST)) {
1689                         LDLM_DEBUG(lock, "Cancel lock separately");
1690                         list_move(&lock->l_bl_ast, &head);
1691                         bl_ast++;
1692                         continue;
1693                 }
1694                 if (rc == LDLM_FL_LOCAL_ONLY) {
1695                         /* CANCEL RPC should not be sent to server. */
1696                         list_del_init(&lock->l_bl_ast);
1697                         LDLM_LOCK_RELEASE(lock);
1698                         count--;
1699                 }
1700         }
1701         if (bl_ast > 0) {
1702                 count -= bl_ast;
1703                 ldlm_cli_cancel_list(&head, bl_ast, NULL, 0);
1704         }
1705
1706         RETURN(count);
1707 }
1708
1709 /**
1710  * Cancel as many locks as possible w/o sending any RPCs (e.g. to write back
1711  * dirty data, to close a file, ...) or waiting for any RPCs in-flight (e.g.
1712  * readahead requests, ...)
1713  */
1714 static enum ldlm_policy_res
1715 ldlm_cancel_no_wait_policy(struct ldlm_namespace *ns, struct ldlm_lock *lock,
1716                            int unused, int added, int count)
1717 {
1718         enum ldlm_policy_res result = LDLM_POLICY_CANCEL_LOCK;
1719
1720         /*
1721          * don't check added & count since we want to process all locks
1722          * from unused list.
1723          * It's fine to not take lock to access lock->l_resource since
1724          * the lock has already been granted so it won't change.
1725          */
1726         switch (lock->l_resource->lr_type) {
1727                 case LDLM_EXTENT:
1728                 case LDLM_IBITS:
1729                         if (ns->ns_cancel != NULL && ns->ns_cancel(lock) != 0)
1730                                 break;
1731                         /* fallthrough */
1732                 default:
1733                         result = LDLM_POLICY_SKIP_LOCK;
1734                         break;
1735         }
1736
1737         RETURN(result);
1738 }
1739
1740 /**
1741  * Callback function for LRU-resize policy. Decides whether to keep
1742  * \a lock in LRU for current \a LRU size \a unused, added in current
1743  * scan \a added and number of locks to be preferably canceled \a count.
1744  *
1745  * \retval LDLM_POLICY_KEEP_LOCK keep lock in LRU in stop scanning
1746  *
1747  * \retval LDLM_POLICY_CANCEL_LOCK cancel lock from LRU
1748  */
1749 static enum ldlm_policy_res ldlm_cancel_lrur_policy(struct ldlm_namespace *ns,
1750                                                     struct ldlm_lock *lock,
1751                                                     int unused, int added,
1752                                                     int count)
1753 {
1754         ktime_t cur = ktime_get();
1755         struct ldlm_pool *pl = &ns->ns_pool;
1756         u64 slv, lvf, lv;
1757         s64 la;
1758
1759         /*
1760          * Stop LRU processing when we reach past @count or have checked all
1761          * locks in LRU.
1762          */
1763         if (count && added >= count)
1764                 return LDLM_POLICY_KEEP_LOCK;
1765
1766         /*
1767          * Despite of the LV, It doesn't make sense to keep the lock which
1768          * is unused for ns_max_age time.
1769          */
1770         if (ktime_after(ktime_get(),
1771                         ktime_add(lock->l_last_used, ns->ns_max_age)))
1772                 return LDLM_POLICY_CANCEL_LOCK;
1773
1774         slv = ldlm_pool_get_slv(pl);
1775         lvf = ldlm_pool_get_lvf(pl);
1776         la = div_u64(ktime_to_ns(ktime_sub(cur, lock->l_last_used)),
1777                      NSEC_PER_SEC);
1778         lv = lvf * la * unused;
1779
1780         /* Inform pool about current CLV to see it via debugfs. */
1781         ldlm_pool_set_clv(pl, lv);
1782
1783         /*
1784          * Stop when SLV is not yet come from server or lv is smaller than
1785          * it is.
1786          */
1787         if (slv == 0 || lv < slv)
1788                 return LDLM_POLICY_KEEP_LOCK;
1789
1790         return LDLM_POLICY_CANCEL_LOCK;
1791 }
1792
1793 static enum ldlm_policy_res
1794 ldlm_cancel_lrur_no_wait_policy(struct ldlm_namespace *ns,
1795                                 struct ldlm_lock *lock,
1796                                 int unused, int added,
1797                                 int count)
1798 {
1799         enum ldlm_policy_res result;
1800
1801         result = ldlm_cancel_lrur_policy(ns, lock, unused, added, count);
1802         if (result == LDLM_POLICY_KEEP_LOCK)
1803                 return result;
1804
1805         return ldlm_cancel_no_wait_policy(ns, lock, unused, added, count);
1806 }
1807
1808 /**
1809  * Callback function for debugfs used policy. Makes decision whether to keep
1810  * \a lock in LRU for current \a LRU size \a unused, added in current scan \a
1811  * added and number of locks to be preferably canceled \a count.
1812  *
1813  * \retval LDLM_POLICY_KEEP_LOCK keep lock in LRU in stop scanning
1814  *
1815  * \retval LDLM_POLICY_CANCEL_LOCK cancel lock from LRU
1816  */
1817 static enum ldlm_policy_res ldlm_cancel_passed_policy(struct ldlm_namespace *ns,
1818                                                       struct ldlm_lock *lock,
1819                                                       int unused, int added,
1820                                                       int count)
1821 {
1822         /*
1823          * Stop LRU processing when we reach past @count or have checked all
1824          * locks in LRU.
1825          */
1826         return (added >= count) ?
1827                 LDLM_POLICY_KEEP_LOCK : LDLM_POLICY_CANCEL_LOCK;
1828 }
1829
1830 /**
1831  * Callback function for aged policy. Makes decision whether to keep \a lock in
1832  * LRU for current LRU size \a unused, added in current scan \a added and
1833  * number of locks to be preferably canceled \a count.
1834  *
1835  * \retval LDLM_POLICY_KEEP_LOCK keep lock in LRU in stop scanning
1836  *
1837  * \retval LDLM_POLICY_CANCEL_LOCK cancel lock from LRU
1838  */
1839 static enum ldlm_policy_res ldlm_cancel_aged_policy(struct ldlm_namespace *ns,
1840                                                     struct ldlm_lock *lock,
1841                                                     int unused, int added,
1842                                                     int count)
1843 {
1844         if ((added >= count) &&
1845             ktime_before(ktime_get(),
1846                          ktime_add(lock->l_last_used, ns->ns_max_age)))
1847                 return LDLM_POLICY_KEEP_LOCK;
1848
1849         return LDLM_POLICY_CANCEL_LOCK;
1850 }
1851
1852 static enum ldlm_policy_res
1853 ldlm_cancel_aged_no_wait_policy(struct ldlm_namespace *ns,
1854                                 struct ldlm_lock *lock,
1855                                 int unused, int added, int count)
1856 {
1857         enum ldlm_policy_res result;
1858
1859         result = ldlm_cancel_aged_policy(ns, lock, unused, added, count);
1860         if (result == LDLM_POLICY_KEEP_LOCK)
1861                 return result;
1862
1863         return ldlm_cancel_no_wait_policy(ns, lock, unused, added, count);
1864 }
1865
1866 /**
1867  * Callback function for default policy. Makes decision whether to keep \a lock
1868  * in LRU for current LRU size \a unused, added in current scan \a added and
1869  * number of locks to be preferably canceled \a count.
1870  *
1871  * \retval LDLM_POLICY_KEEP_LOCK keep lock in LRU in stop scanning
1872  *
1873  * \retval LDLM_POLICY_CANCEL_LOCK cancel lock from LRU
1874  */
1875 static
1876 enum ldlm_policy_res ldlm_cancel_default_policy(struct ldlm_namespace *ns,
1877                                                 struct ldlm_lock *lock,
1878                                                 int unused, int added,
1879                                                 int count)
1880 {
1881         /*
1882          * Stop LRU processing when we reach past count or have checked all
1883          * locks in LRU.
1884          */
1885         return (added >= count) ?
1886                 LDLM_POLICY_KEEP_LOCK : LDLM_POLICY_CANCEL_LOCK;
1887 }
1888
1889 typedef enum ldlm_policy_res
1890 (*ldlm_cancel_lru_policy_t)(struct ldlm_namespace *ns, struct ldlm_lock *lock,
1891                             int unused, int added, int count);
1892
1893 static ldlm_cancel_lru_policy_t
1894 ldlm_cancel_lru_policy(struct ldlm_namespace *ns, enum ldlm_lru_flags lru_flags)
1895 {
1896         if (ns_connect_lru_resize(ns)) {
1897                 if (lru_flags & LDLM_LRU_FLAG_SHRINK)
1898                         /* We kill passed number of old locks. */
1899                         return ldlm_cancel_passed_policy;
1900                 if (lru_flags & LDLM_LRU_FLAG_LRUR) {
1901                         if (lru_flags & LDLM_LRU_FLAG_NO_WAIT)
1902                                 return ldlm_cancel_lrur_no_wait_policy;
1903                         else
1904                                 return ldlm_cancel_lrur_policy;
1905                 }
1906                 if (lru_flags & LDLM_LRU_FLAG_PASSED)
1907                         return ldlm_cancel_passed_policy;
1908         } else {
1909                 if (lru_flags & LDLM_LRU_FLAG_AGED) {
1910                         if (lru_flags & LDLM_LRU_FLAG_NO_WAIT)
1911                                 return ldlm_cancel_aged_no_wait_policy;
1912                         else
1913                                 return ldlm_cancel_aged_policy;
1914                 }
1915         }
1916         if (lru_flags & LDLM_LRU_FLAG_NO_WAIT)
1917                 return ldlm_cancel_no_wait_policy;
1918
1919         return ldlm_cancel_default_policy;
1920 }
1921
1922 /**
1923  * - Free space in LRU for \a count new locks,
1924  *   redundant unused locks are canceled locally;
1925  * - also cancel locally unused aged locks;
1926  * - do not cancel more than \a max locks;
1927  * - GET the found locks and add them into the \a cancels list.
1928  *
1929  * A client lock can be added to the l_bl_ast list only when it is
1930  * marked LDLM_FL_CANCELING. Otherwise, somebody is already doing
1931  * CANCEL.  There are the following use cases:
1932  * ldlm_cancel_resource_local(), ldlm_cancel_lru_local() and
1933  * ldlm_cli_cancel(), which check and set this flag properly. As any
1934  * attempt to cancel a lock rely on this flag, l_bl_ast list is accessed
1935  * later without any special locking.
1936  *
1937  * Calling policies for enabled LRU resize:
1938  * ----------------------------------------
1939  * flags & LDLM_LRU_FLAG_LRUR - use LRU resize policy (SLV from server) to
1940  *                              cancel not more than \a count locks;
1941  *
1942  * flags & LDLM_LRU_FLAG_PASSED - cancel \a count number of old locks (located
1943  *                              at the beginning of LRU list);
1944  *
1945  * flags & LDLM_LRU_FLAG_SHRINK - cancel not more than \a count locks according
1946  *                              to memory pressre policy function;
1947  *
1948  * flags & LDLM_LRU_FLAG_AGED - cancel \a count locks according to "aged policy"
1949  *
1950  * flags & LDLM_LRU_FLAG_NO_WAIT - cancel as many unused locks as possible
1951  *                              (typically before replaying locks) w/o
1952  *                              sending any RPCs or waiting for any
1953  *                              outstanding RPC to complete.
1954  *
1955  * flags & LDLM_CANCEL_CLEANUP - when cancelling read locks, do not check for
1956  *                              other read locks covering the same pages, just
1957  *                              discard those pages.
1958  */
1959 static int ldlm_prepare_lru_list(struct ldlm_namespace *ns,
1960                                  struct list_head *cancels, int count, int max,
1961                                  enum ldlm_lru_flags lru_flags)
1962 {
1963         ldlm_cancel_lru_policy_t pf;
1964         int added = 0;
1965         int no_wait = lru_flags & LDLM_LRU_FLAG_NO_WAIT;
1966
1967         ENTRY;
1968
1969         if (!ns_connect_lru_resize(ns))
1970                 count += ns->ns_nr_unused - ns->ns_max_unused;
1971
1972         pf = ldlm_cancel_lru_policy(ns, lru_flags);
1973         LASSERT(pf != NULL);
1974
1975         /* For any flags, stop scanning if @max is reached. */
1976         while (!list_empty(&ns->ns_unused_list) && (max == 0 || added < max)) {
1977                 struct ldlm_lock *lock;
1978                 struct list_head *item, *next;
1979                 enum ldlm_policy_res result;
1980                 ktime_t last_use = ktime_set(0, 0);
1981
1982                 spin_lock(&ns->ns_lock);
1983                 item = no_wait ? ns->ns_last_pos : &ns->ns_unused_list;
1984                 for (item = item->next, next = item->next;
1985                      item != &ns->ns_unused_list;
1986                      item = next, next = item->next) {
1987                         lock = list_entry(item, struct ldlm_lock, l_lru);
1988
1989                         /* No locks which got blocking requests. */
1990                         LASSERT(!ldlm_is_bl_ast(lock));
1991
1992                         if (!ldlm_is_canceling(lock) &&
1993                             !ldlm_is_converting(lock))
1994                                 break;
1995
1996                         /*
1997                          * Somebody is already doing CANCEL. No need for this
1998                          * lock in LRU, do not traverse it again.
1999                          */
2000                         ldlm_lock_remove_from_lru_nolock(lock);
2001                 }
2002                 if (item == &ns->ns_unused_list) {
2003                         spin_unlock(&ns->ns_lock);
2004                         break;
2005                 }
2006
2007                 last_use = lock->l_last_used;
2008
2009                 LDLM_LOCK_GET(lock);
2010                 spin_unlock(&ns->ns_lock);
2011                 lu_ref_add(&lock->l_reference, __FUNCTION__, current);
2012
2013                 /*
2014                  * Pass the lock through the policy filter and see if it
2015                  * should stay in LRU.
2016                  *
2017                  * Even for shrinker policy we stop scanning if
2018                  * we find a lock that should stay in the cache.
2019                  * We should take into account lock age anyway
2020                  * as a new lock is a valuable resource even if
2021                  * it has a low weight.
2022                  *
2023                  * That is, for shrinker policy we drop only
2024                  * old locks, but additionally choose them by
2025                  * their weight. Big extent locks will stay in
2026                  * the cache.
2027                  */
2028                 result = pf(ns, lock, ns->ns_nr_unused, added, count);
2029                 if (result == LDLM_POLICY_KEEP_LOCK) {
2030                         lu_ref_del(&lock->l_reference, __func__, current);
2031                         LDLM_LOCK_RELEASE(lock);
2032                         break;
2033                 }
2034
2035                 if (result == LDLM_POLICY_SKIP_LOCK) {
2036                         lu_ref_del(&lock->l_reference, __func__, current);
2037                         if (no_wait) {
2038                                 spin_lock(&ns->ns_lock);
2039                                 if (!list_empty(&lock->l_lru) &&
2040                                     lock->l_lru.prev == ns->ns_last_pos)
2041                                         ns->ns_last_pos = &lock->l_lru;
2042                                 spin_unlock(&ns->ns_lock);
2043                         }
2044
2045                         LDLM_LOCK_RELEASE(lock);
2046                         continue;
2047                 }
2048
2049                 lock_res_and_lock(lock);
2050                 /* Check flags again under the lock. */
2051                 if (ldlm_is_canceling(lock) || ldlm_is_converting(lock) ||
2052                     ldlm_lock_remove_from_lru_check(lock, last_use) == 0) {
2053                         /*
2054                          * Another thread is removing lock from LRU, or
2055                          * somebody is already doing CANCEL, or there
2056                          * is a blocking request which will send cancel
2057                          * by itself, or the lock is no longer unused or
2058                          * the lock has been used since the pf() call and
2059                          * pages could be put under it.
2060                          */
2061                         unlock_res_and_lock(lock);
2062                         lu_ref_del(&lock->l_reference, __FUNCTION__, current);
2063                         LDLM_LOCK_RELEASE(lock);
2064                         continue;
2065                 }
2066                 LASSERT(!lock->l_readers && !lock->l_writers);
2067
2068                 /*
2069                  * If we have chosen to cancel this lock voluntarily, we
2070                  * better send cancel notification to server, so that it
2071                  * frees appropriate state. This might lead to a race
2072                  * where while we are doing cancel here, server is also
2073                  * silently cancelling this lock.
2074                  */
2075                 ldlm_clear_cancel_on_block(lock);
2076
2077                 /*
2078                  * Setting the CBPENDING flag is a little misleading,
2079                  * but prevents an important race; namely, once
2080                  * CBPENDING is set, the lock can accumulate no more
2081                  * readers/writers. Since readers and writers are
2082                  * already zero here, ldlm_lock_decref() won't see
2083                  * this flag and call l_blocking_ast
2084                  */
2085                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_CANCELING;
2086
2087                 if ((lru_flags & LDLM_LRU_FLAG_CLEANUP) &&
2088                     (lock->l_resource->lr_type == LDLM_EXTENT ||
2089                      ldlm_has_dom(lock)) && lock->l_granted_mode == LCK_PR)
2090                         ldlm_set_discard_data(lock);
2091
2092                 /*
2093                  * We can't re-add to l_lru as it confuses the
2094                  * refcounting in ldlm_lock_remove_from_lru() if an AST
2095                  * arrives after we drop lr_lock below. We use l_bl_ast
2096                  * and can't use l_pending_chain as it is used both on
2097                  * server and client nevertheless b=5666 says it is
2098                  * used only on server
2099                  */
2100                 LASSERT(list_empty(&lock->l_bl_ast));
2101                 list_add(&lock->l_bl_ast, cancels);
2102                 unlock_res_and_lock(lock);
2103                 lu_ref_del(&lock->l_reference, __FUNCTION__, current);
2104                 added++;
2105         }
2106         RETURN(added);
2107 }
2108
2109 int ldlm_cancel_lru_local(struct ldlm_namespace *ns, struct list_head *cancels,
2110                           int count, int max,
2111                           enum ldlm_cancel_flags cancel_flags,
2112                           enum ldlm_lru_flags lru_flags)
2113 {
2114         int added;
2115
2116         added = ldlm_prepare_lru_list(ns, cancels, count, max, lru_flags);
2117         if (added <= 0)
2118                 return added;
2119
2120         return ldlm_cli_cancel_list_local(cancels, added, cancel_flags);
2121 }
2122
2123 /**
2124  * Cancel at least \a nr locks from given namespace LRU.
2125  *
2126  * When called with LCF_ASYNC the blocking callback will be handled
2127  * in a thread and this function will return after the thread has been
2128  * asked to call the callback.  When called with LCF_ASYNC the blocking
2129  * callback will be performed in this function.
2130  */
2131 int ldlm_cancel_lru(struct ldlm_namespace *ns, int nr,
2132                     enum ldlm_cancel_flags cancel_flags,
2133                     enum ldlm_lru_flags lru_flags)
2134 {
2135         LIST_HEAD(cancels);
2136         int count, rc;
2137
2138         ENTRY;
2139
2140         /*
2141          * Just prepare the list of locks, do not actually cancel them yet.
2142          * Locks are cancelled later in a separate thread.
2143          */
2144         count = ldlm_prepare_lru_list(ns, &cancels, nr, 0, lru_flags);
2145         rc = ldlm_bl_to_thread_list(ns, NULL, &cancels, count, cancel_flags);
2146         if (rc == 0)
2147                 RETURN(count);
2148
2149         RETURN(0);
2150 }
2151
2152 /**
2153  * Find and cancel locally unused locks found on resource, matched to the
2154  * given policy, mode. GET the found locks and add them into the \a cancels
2155  * list.
2156  */
2157 int ldlm_cancel_resource_local(struct ldlm_resource *res,
2158                                struct list_head *cancels,
2159                                union ldlm_policy_data *policy,
2160                                enum ldlm_mode mode, __u64 lock_flags,
2161                                enum ldlm_cancel_flags cancel_flags,
2162                                void *opaque)
2163 {
2164         struct ldlm_lock *lock;
2165         int count = 0;
2166
2167         ENTRY;
2168
2169         lock_res(res);
2170         list_for_each_entry(lock, &res->lr_granted, l_res_link) {
2171                 if (opaque != NULL && lock->l_ast_data != opaque) {
2172                         LDLM_ERROR(lock, "data %p doesn't match opaque %p",
2173                                    lock->l_ast_data, opaque);
2174                         continue;
2175                 }
2176
2177                 if (lock->l_readers || lock->l_writers)
2178                         continue;
2179
2180                 /*
2181                  * If somebody is already doing CANCEL, or blocking AST came,
2182                  * or lock is being converted then skip this lock.
2183                  */
2184                 if (ldlm_is_bl_ast(lock) || ldlm_is_canceling(lock) ||
2185                     ldlm_is_converting(lock))
2186                         continue;
2187
2188                 if (lockmode_compat(lock->l_granted_mode, mode))
2189                         continue;
2190
2191                 /*
2192                  * If policy is given and this is IBITS lock, add to list only
2193                  * those locks that match by policy.
2194                  */
2195                 if (policy && (lock->l_resource->lr_type == LDLM_IBITS)) {
2196                         if (!(lock->l_policy_data.l_inodebits.bits &
2197                               policy->l_inodebits.bits))
2198                                 continue;
2199                         /* Skip locks with DoM bit if it is not set in policy
2200                          * to don't flush data by side-bits. Lock convert will
2201                          * drop those bits separately.
2202                          */
2203                         if (ldlm_has_dom(lock) &&
2204                             !(policy->l_inodebits.bits & MDS_INODELOCK_DOM))
2205                                 continue;
2206                 }
2207
2208                 /* See CBPENDING comment in ldlm_cancel_lru */
2209                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_CANCELING |
2210                                  lock_flags;
2211
2212                 LASSERT(list_empty(&lock->l_bl_ast));
2213                 list_add(&lock->l_bl_ast, cancels);
2214                 LDLM_LOCK_GET(lock);
2215                 count++;
2216         }
2217         unlock_res(res);
2218
2219         RETURN(ldlm_cli_cancel_list_local(cancels, count, cancel_flags));
2220 }
2221 EXPORT_SYMBOL(ldlm_cancel_resource_local);
2222
2223 /**
2224  * Cancel client-side locks from a list and send/prepare cancel RPCs to the
2225  * server.
2226  * If \a req is NULL, send CANCEL request to server with handles of locks
2227  * in the \a cancels. If EARLY_CANCEL is not supported, send CANCEL requests
2228  * separately per lock.
2229  * If \a req is not NULL, put handles of locks in \a cancels into the request
2230  * buffer at the offset \a off.
2231  * Destroy \a cancels at the end.
2232  */
2233 int ldlm_cli_cancel_list(struct list_head *cancels, int count,
2234                          struct ptlrpc_request *req,
2235                          enum ldlm_cancel_flags flags)
2236 {
2237         struct ldlm_lock *lock;
2238         int res = 0;
2239
2240         ENTRY;
2241
2242         if (list_empty(cancels) || count == 0)
2243                 RETURN(0);
2244
2245         /*
2246          * XXX: requests (both batched and not) could be sent in parallel.
2247          * Usually it is enough to have just 1 RPC, but it is possible that
2248          * there are too many locks to be cancelled in LRU or on a resource.
2249          * It would also speed up the case when the server does not support
2250          * the feature.
2251          */
2252         while (count > 0) {
2253                 LASSERT(!list_empty(cancels));
2254                 lock = list_entry(cancels->next, struct ldlm_lock,
2255                                   l_bl_ast);
2256                 LASSERT(lock->l_conn_export);
2257
2258                 if (exp_connect_cancelset(lock->l_conn_export)) {
2259                         res = count;
2260                         if (req)
2261                                 ldlm_cancel_pack(req, cancels, count);
2262                         else
2263                                 res = ldlm_cli_cancel_req(lock->l_conn_export,
2264                                                           cancels, count,
2265                                                           flags);
2266                 } else {
2267                         res = ldlm_cli_cancel_req(lock->l_conn_export,
2268                                                   cancels, 1, flags);
2269                 }
2270
2271                 if (res < 0) {
2272                         CDEBUG_LIMIT(res == -ESHUTDOWN ? D_DLMTRACE : D_ERROR,
2273                                      "ldlm_cli_cancel_list: %d\n", res);
2274                         res = count;
2275                 }
2276
2277                 count -= res;
2278                 ldlm_lock_list_put(cancels, l_bl_ast, res);
2279         }
2280         LASSERT(count == 0);
2281         RETURN(0);
2282 }
2283 EXPORT_SYMBOL(ldlm_cli_cancel_list);
2284
2285 /**
2286  * Cancel all locks on a resource that have 0 readers/writers.
2287  *
2288  * If flags & LDLM_FL_LOCAL_ONLY, throw the locks away without trying
2289  * to notify the server.
2290  */
2291 int ldlm_cli_cancel_unused_resource(struct ldlm_namespace *ns,
2292                                     const struct ldlm_res_id *res_id,
2293                                     union ldlm_policy_data *policy,
2294                                     enum ldlm_mode mode,
2295                                     enum ldlm_cancel_flags flags, void *opaque)
2296 {
2297         struct ldlm_resource *res;
2298         LIST_HEAD(cancels);
2299         int count;
2300         int rc;
2301
2302         ENTRY;
2303
2304         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
2305         if (IS_ERR(res)) {
2306                 /* This is not a problem. */
2307                 CDEBUG(D_INFO, "No resource %llu\n", res_id->name[0]);
2308                 RETURN(0);
2309         }
2310
2311         LDLM_RESOURCE_ADDREF(res);
2312         count = ldlm_cancel_resource_local(res, &cancels, policy, mode,
2313                                            0, flags | LCF_BL_AST, opaque);
2314         rc = ldlm_cli_cancel_list(&cancels, count, NULL, flags);
2315         if (rc != ELDLM_OK)
2316                 CERROR("canceling unused lock "DLDLMRES": rc = %d\n",
2317                        PLDLMRES(res), rc);
2318
2319         LDLM_RESOURCE_DELREF(res);
2320         ldlm_resource_putref(res);
2321         RETURN(0);
2322 }
2323 EXPORT_SYMBOL(ldlm_cli_cancel_unused_resource);
2324
2325 struct ldlm_cli_cancel_arg {
2326         int     lc_flags;
2327         void   *lc_opaque;
2328 };
2329
2330 static int
2331 ldlm_cli_hash_cancel_unused(struct cfs_hash *hs, struct cfs_hash_bd *bd,
2332                             struct hlist_node *hnode, void *arg)
2333 {
2334         struct ldlm_resource           *res = cfs_hash_object(hs, hnode);
2335         struct ldlm_cli_cancel_arg     *lc = arg;
2336
2337         ldlm_cli_cancel_unused_resource(ldlm_res_to_ns(res), &res->lr_name,
2338                                         NULL, LCK_MINMODE, lc->lc_flags,
2339                                         lc->lc_opaque);
2340         /* must return 0 for hash iteration */
2341         return 0;
2342 }
2343
2344 /**
2345  * Cancel all locks on a namespace (or a specific resource, if given)
2346  * that have 0 readers/writers.
2347  *
2348  * If flags & LCF_LOCAL, throw the locks away without trying
2349  * to notify the server.
2350  */
2351 int ldlm_cli_cancel_unused(struct ldlm_namespace *ns,
2352                            const struct ldlm_res_id *res_id,
2353                            enum ldlm_cancel_flags flags, void *opaque)
2354 {
2355         struct ldlm_cli_cancel_arg arg = {
2356                 .lc_flags       = flags,
2357                 .lc_opaque      = opaque,
2358         };
2359
2360         ENTRY;
2361
2362         if (ns == NULL)
2363                 RETURN(ELDLM_OK);
2364
2365         if (res_id != NULL) {
2366                 RETURN(ldlm_cli_cancel_unused_resource(ns, res_id, NULL,
2367                                                        LCK_MINMODE, flags,
2368                                                        opaque));
2369         } else {
2370                 cfs_hash_for_each_nolock(ns->ns_rs_hash,
2371                                          ldlm_cli_hash_cancel_unused, &arg, 0);
2372                 RETURN(ELDLM_OK);
2373         }
2374 }
2375
2376 /* Lock iterators. */
2377
2378 int ldlm_resource_foreach(struct ldlm_resource *res, ldlm_iterator_t iter,
2379                           void *closure)
2380 {
2381         struct list_head *tmp, *next;
2382         struct ldlm_lock *lock;
2383         int rc = LDLM_ITER_CONTINUE;
2384
2385         ENTRY;
2386
2387         if (!res)
2388                 RETURN(LDLM_ITER_CONTINUE);
2389
2390         lock_res(res);
2391         list_for_each_safe(tmp, next, &res->lr_granted) {
2392                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
2393
2394                 if (iter(lock, closure) == LDLM_ITER_STOP)
2395                         GOTO(out, rc = LDLM_ITER_STOP);
2396         }
2397
2398         list_for_each_safe(tmp, next, &res->lr_waiting) {
2399                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
2400
2401                 if (iter(lock, closure) == LDLM_ITER_STOP)
2402                         GOTO(out, rc = LDLM_ITER_STOP);
2403         }
2404 out:
2405         unlock_res(res);
2406         RETURN(rc);
2407 }
2408
2409 struct iter_helper_data {
2410         ldlm_iterator_t iter;
2411         void *closure;
2412 };
2413
2414 static int ldlm_iter_helper(struct ldlm_lock *lock, void *closure)
2415 {
2416         struct iter_helper_data *helper = closure;
2417
2418         return helper->iter(lock, helper->closure);
2419 }
2420
2421 static int ldlm_res_iter_helper(struct cfs_hash *hs, struct cfs_hash_bd *bd,
2422                                 struct hlist_node *hnode, void *arg)
2423
2424 {
2425         struct ldlm_resource *res = cfs_hash_object(hs, hnode);
2426
2427         return ldlm_resource_foreach(res, ldlm_iter_helper, arg) ==
2428                                      LDLM_ITER_STOP;
2429 }
2430
2431 void ldlm_namespace_foreach(struct ldlm_namespace *ns,
2432                             ldlm_iterator_t iter, void *closure)
2433
2434 {
2435         struct iter_helper_data helper = { .iter = iter, .closure = closure };
2436
2437         cfs_hash_for_each_nolock(ns->ns_rs_hash,
2438                                  ldlm_res_iter_helper, &helper, 0);
2439
2440 }
2441
2442 /*
2443  * non-blocking function to manipulate a lock whose cb_data is being put away.
2444  * return  0:  find no resource
2445  *       > 0:  must be LDLM_ITER_STOP/LDLM_ITER_CONTINUE.
2446  *       < 0:  errors
2447  */
2448 int ldlm_resource_iterate(struct ldlm_namespace *ns,
2449                           const struct ldlm_res_id *res_id,
2450                           ldlm_iterator_t iter, void *data)
2451 {
2452         struct ldlm_resource *res;
2453         int rc;
2454
2455         ENTRY;
2456
2457         LASSERTF(ns != NULL, "must pass in namespace\n");
2458
2459         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
2460         if (IS_ERR(res))
2461                 RETURN(0);
2462
2463         LDLM_RESOURCE_ADDREF(res);
2464         rc = ldlm_resource_foreach(res, iter, data);
2465         LDLM_RESOURCE_DELREF(res);
2466         ldlm_resource_putref(res);
2467         RETURN(rc);
2468 }
2469 EXPORT_SYMBOL(ldlm_resource_iterate);
2470
2471 /* Lock replay */
2472 static int ldlm_chain_lock_for_replay(struct ldlm_lock *lock, void *closure)
2473 {
2474         struct list_head *list = closure;
2475
2476         /* we use l_pending_chain here, because it's unused on clients. */
2477         LASSERTF(list_empty(&lock->l_pending_chain),
2478                  "lock %p next %p prev %p\n",
2479                  lock, &lock->l_pending_chain.next,
2480                  &lock->l_pending_chain.prev);
2481         /*
2482          * b=9573: don't replay locks left after eviction, or
2483          * b=17614: locks being actively cancelled. Get a reference
2484          * on a lock so that it does not disapear under us (e.g. due to cancel)
2485          */
2486         if (!(lock->l_flags & (LDLM_FL_FAILED|LDLM_FL_BL_DONE))) {
2487                 list_add(&lock->l_pending_chain, list);
2488                 LDLM_LOCK_GET(lock);
2489         }
2490
2491         return LDLM_ITER_CONTINUE;
2492 }
2493
2494 static int replay_lock_interpret(const struct lu_env *env,
2495                                  struct ptlrpc_request *req, void *args, int rc)
2496 {
2497         struct ldlm_async_args *aa = args;
2498         struct ldlm_lock     *lock;
2499         struct ldlm_reply    *reply;
2500         struct obd_export    *exp;
2501
2502         ENTRY;
2503         atomic_dec(&req->rq_import->imp_replay_inflight);
2504         if (rc != ELDLM_OK)
2505                 GOTO(out, rc);
2506
2507         reply = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
2508         if (reply == NULL)
2509                 GOTO(out, rc = -EPROTO);
2510
2511         lock = ldlm_handle2lock(&aa->lock_handle);
2512         if (!lock) {
2513                 CERROR("received replay ack for unknown local cookie %#llx remote cookie %#llx from server %s id %s\n",
2514                        aa->lock_handle.cookie, reply->lock_handle.cookie,
2515                        req->rq_export->exp_client_uuid.uuid,
2516                        libcfs_id2str(req->rq_peer));
2517                 GOTO(out, rc = -ESTALE);
2518         }
2519
2520         /* Key change rehash lock in per-export hash with new key */
2521         exp = req->rq_export;
2522         if (exp && exp->exp_lock_hash) {
2523                 /*
2524                  * In the function below, .hs_keycmp resolves to
2525                  * ldlm_export_lock_keycmp()
2526                  */
2527                 /* coverity[overrun-buffer-val] */
2528                 cfs_hash_rehash_key(exp->exp_lock_hash,
2529                                     &lock->l_remote_handle,
2530                                     &reply->lock_handle,
2531                                     &lock->l_exp_hash);
2532         } else {
2533                 lock->l_remote_handle = reply->lock_handle;
2534         }
2535
2536         LDLM_DEBUG(lock, "replayed lock:");
2537         ptlrpc_import_recovery_state_machine(req->rq_import);
2538         LDLM_LOCK_PUT(lock);
2539 out:
2540         if (rc != ELDLM_OK)
2541                 ptlrpc_connect_import(req->rq_import);
2542
2543         RETURN(rc);
2544 }
2545
2546 static int replay_one_lock(struct obd_import *imp, struct ldlm_lock *lock)
2547 {
2548         struct ptlrpc_request *req;
2549         struct ldlm_async_args *aa;
2550         struct ldlm_request   *body;
2551         int flags;
2552
2553         ENTRY;
2554
2555
2556         /* b=11974: Do not replay a lock which is actively being canceled */
2557         if (ldlm_is_bl_done(lock)) {
2558                 LDLM_DEBUG(lock, "Not replaying canceled lock:");
2559                 RETURN(0);
2560         }
2561
2562         /*
2563          * If this is reply-less callback lock, we cannot replay it, since
2564          * server might have long dropped it, but notification of that event was
2565          * lost by network. (and server granted conflicting lock already)
2566          */
2567         if (ldlm_is_cancel_on_block(lock)) {
2568                 LDLM_DEBUG(lock, "Not replaying reply-less lock:");
2569                 ldlm_lock_cancel(lock);
2570                 RETURN(0);
2571         }
2572
2573         /*
2574          * If granted mode matches the requested mode, this lock is granted.
2575          *
2576          * If we haven't been granted anything and are on a resource list,
2577          * then we're blocked/waiting.
2578          *
2579          * If we haven't been granted anything and we're NOT on a resource list,
2580          * then we haven't got a reply yet and don't have a known disposition.
2581          * This happens whenever a lock enqueue is the request that triggers
2582          * recovery.
2583          */
2584         if (ldlm_is_granted(lock))
2585                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_GRANTED;
2586         else if (!list_empty(&lock->l_res_link))
2587                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_WAIT;
2588         else
2589                 flags = LDLM_FL_REPLAY;
2590
2591         req = ptlrpc_request_alloc_pack(imp, &RQF_LDLM_ENQUEUE,
2592                                         LUSTRE_DLM_VERSION, LDLM_ENQUEUE);
2593         if (req == NULL)
2594                 RETURN(-ENOMEM);
2595
2596         /* We're part of recovery, so don't wait for it. */
2597         req->rq_send_state = LUSTRE_IMP_REPLAY_LOCKS;
2598
2599         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
2600         ldlm_lock2desc(lock, &body->lock_desc);
2601         body->lock_flags = ldlm_flags_to_wire(flags);
2602
2603         ldlm_lock2handle(lock, &body->lock_handle[0]);
2604         if (lock->l_lvb_len > 0)
2605                 req_capsule_extend(&req->rq_pill, &RQF_LDLM_ENQUEUE_LVB);
2606         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
2607                              lock->l_lvb_len);
2608         ptlrpc_request_set_replen(req);
2609         /*
2610          * notify the server we've replayed all requests.
2611          * also, we mark the request to be put on a dedicated
2612          * queue to be processed after all request replayes.
2613          * b=6063
2614          */
2615         lustre_msg_set_flags(req->rq_reqmsg, MSG_REQ_REPLAY_DONE);
2616
2617         LDLM_DEBUG(lock, "replaying lock:");
2618
2619         atomic_inc(&req->rq_import->imp_replay_inflight);
2620         aa = ptlrpc_req_async_args(aa, req);
2621         aa->lock_handle = body->lock_handle[0];
2622         req->rq_interpret_reply = replay_lock_interpret;
2623         ptlrpcd_add_req(req);
2624
2625         RETURN(0);
2626 }
2627
2628 /**
2629  * Cancel as many unused locks as possible before replay. since we are
2630  * in recovery, we can't wait for any outstanding RPCs to send any RPC
2631  * to the server.
2632  *
2633  * Called only in recovery before replaying locks. there is no need to
2634  * replay locks that are unused. since the clients may hold thousands of
2635  * cached unused locks, dropping the unused locks can greatly reduce the
2636  * load on the servers at recovery time.
2637  */
2638 static void ldlm_cancel_unused_locks_for_replay(struct ldlm_namespace *ns)
2639 {
2640         int canceled;
2641         LIST_HEAD(cancels);
2642
2643         CDEBUG(D_DLMTRACE,
2644                "Dropping as many unused locks as possible before replay for namespace %s (%d)\n",
2645                ldlm_ns_name(ns), ns->ns_nr_unused);
2646
2647         /*
2648          * We don't need to care whether or not LRU resize is enabled
2649          * because the LDLM_LRU_FLAG_NO_WAIT policy doesn't use the
2650          * count parameter
2651          */
2652         canceled = ldlm_cancel_lru_local(ns, &cancels, ns->ns_nr_unused, 0,
2653                                          LCF_LOCAL, LDLM_LRU_FLAG_NO_WAIT);
2654
2655         CDEBUG(D_DLMTRACE, "Canceled %d unused locks from namespace %s\n",
2656                            canceled, ldlm_ns_name(ns));
2657 }
2658
2659 int ldlm_replay_locks(struct obd_import *imp)
2660 {
2661         struct ldlm_namespace *ns = imp->imp_obd->obd_namespace;
2662         LIST_HEAD(list);
2663         struct ldlm_lock *lock, *next;
2664         int rc = 0;
2665
2666         ENTRY;
2667
2668         LASSERT(atomic_read(&imp->imp_replay_inflight) == 0);
2669
2670         /* don't replay locks if import failed recovery */
2671         if (imp->imp_vbr_failed)
2672                 RETURN(0);
2673
2674         /* ensure this doesn't fall to 0 before all have been queued */
2675         atomic_inc(&imp->imp_replay_inflight);
2676
2677         if (ldlm_cancel_unused_locks_before_replay)
2678                 ldlm_cancel_unused_locks_for_replay(ns);
2679
2680         ldlm_namespace_foreach(ns, ldlm_chain_lock_for_replay, &list);
2681
2682         list_for_each_entry_safe(lock, next, &list, l_pending_chain) {
2683                 list_del_init(&lock->l_pending_chain);
2684                 if (rc) {
2685                         LDLM_LOCK_RELEASE(lock);
2686                         continue; /* or try to do the rest? */
2687                 }
2688                 rc = replay_one_lock(imp, lock);
2689                 LDLM_LOCK_RELEASE(lock);
2690         }
2691
2692         atomic_dec(&imp->imp_replay_inflight);
2693
2694         RETURN(rc);
2695 }