lustre/ldlm/ldlm_request.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (C) 2002, 2003 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of the Lustre file system, http://www.lustre.org
   7  *   Lustre is a trademark of Cluster File Systems, Inc.
   8  *
   9  *   You may have signed or agreed to another license before downloading
  10  *   this software.  If so, you are bound by the terms and conditions
  11  *   of that agreement, and the following does not apply to you.  See the
  12  *   LICENSE file included with this distribution for more information.
  13  *
  14  *   If you did not agree to a different license, then this copy of Lustre
  15  *   is open source software; you can redistribute it and/or modify it
  16  *   under the terms of version 2 of the GNU General Public License as
  17  *   published by the Free Software Foundation.
  18  *
  19  *   In either case, Lustre is distributed in the hope that it will be
  20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   license text for more details.
  23  */
  24
  25 #define DEBUG_SUBSYSTEM S_LDLM
  26 #ifndef __KERNEL__
  27 #include <signal.h>
  28 #include <liblustre.h>
  29 #endif
  30
  31 #include <lustre_dlm.h>
  32 #include <obd_class.h>
  33 #include <obd.h>
  34
  35 #include "ldlm_internal.h"
  36
  37 static void interrupted_completion_wait(void *data)
  38 {
  39 }
  40
  41 struct lock_wait_data {
  42         struct ldlm_lock *lwd_lock;
  43         __u32             lwd_conn_cnt;
  44 };
  45
  46 int ldlm_expired_completion_wait(void *data)
  47 {
  48         struct lock_wait_data *lwd = data;
  49         struct ldlm_lock *lock = lwd->lwd_lock;
  50         struct obd_import *imp;
  51         struct obd_device *obd;
  52
  53         ENTRY;
  54         if (lock->l_conn_export == NULL) {
  55                 static cfs_time_t next_dump = 0, last_dump = 0;
  56
  57                 if (ptlrpc_check_suspend())
  58                         RETURN(0);
  59
  60                 LDLM_ERROR(lock, "lock timed out (enqueued at %lu, %lus ago); "
  61                            "not entering recovery in server code, just going "
  62                            "back to sleep", lock->l_enqueued_time.tv_sec,
  63                            CURRENT_SECONDS - lock->l_enqueued_time.tv_sec);
  64                 if (cfs_time_after(cfs_time_current(), next_dump)) {
  65                         last_dump = next_dump;
  66                         next_dump = cfs_time_shift(300);
  67                         ldlm_namespace_dump(D_DLMTRACE,
  68                                             lock->l_resource->lr_namespace);
  69                         if (last_dump == 0)
  70                                 libcfs_debug_dumplog();
  71                 }
  72                 RETURN(0);
  73         }
  74
  75         obd = lock->l_conn_export->exp_obd;
  76         imp = obd->u.cli.cl_import;
  77         ptlrpc_fail_import(imp, lwd->lwd_conn_cnt);
  78         LDLM_ERROR(lock, "lock timed out (enqueued at %lu, %lus ago), entering "
  79                    "recovery for %s@%s", lock->l_enqueued_time.tv_sec,
  80                    CURRENT_SECONDS - lock->l_enqueued_time.tv_sec,
  81                    obd2cli_tgt(obd), imp->imp_connection->c_remote_uuid.uuid);
  82
  83         RETURN(0);
  84 }
  85
  86 int ldlm_completion_ast(struct ldlm_lock *lock, int flags, void *data)
  87 {
  88         /* XXX ALLOCATE - 160 bytes */
  89         struct lock_wait_data lwd;
  90         struct obd_device *obd;
  91         struct obd_import *imp = NULL;
  92         struct l_wait_info lwi;
  93         int rc = 0;
  94         ENTRY;
  95
  96         if (flags == LDLM_FL_WAIT_NOREPROC) {
  97                 LDLM_DEBUG(lock, "client-side enqueue waiting on pending lock");
  98                 goto noreproc;
  99         }
 100
 101         if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
 102                        LDLM_FL_BLOCK_CONV))) {
 103                 cfs_waitq_signal(&lock->l_waitq);
 104                 RETURN(0);
 105         }
 106
 107         LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, "
 108                    "sleeping");
 109         ldlm_lock_dump(D_OTHER, lock, 0);
 110         ldlm_reprocess_all(lock->l_resource);
 111
 112 noreproc:
 113
 114         obd = class_exp2obd(lock->l_conn_export);
 115
 116         /* if this is a local lock, then there is no import */
 117         if (obd != NULL)
 118                 imp = obd->u.cli.cl_import;
 119
 120         lwd.lwd_lock = lock;
 121
 122         if (lock->l_flags & LDLM_FL_NO_TIMEOUT) {
 123                 LDLM_DEBUG(lock, "waiting indefinitely because of NO_TIMEOUT");
 124                 lwi = LWI_INTR(interrupted_completion_wait, &lwd);
 125         } else {
 126                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(obd_timeout),
 127                                        ldlm_expired_completion_wait,
 128                                        interrupted_completion_wait, &lwd);
 129         }
 130
 131         if (imp != NULL) {
 132                 spin_lock(&imp->imp_lock);
 133                 lwd.lwd_conn_cnt = imp->imp_conn_cnt;
 134                 spin_unlock(&imp->imp_lock);
 135         }
 136
 137         /* Go to sleep until the lock is granted or cancelled. */
 138         rc = l_wait_event(lock->l_waitq,
 139                           ((lock->l_req_mode == lock->l_granted_mode) ||
 140                            (lock->l_flags & LDLM_FL_FAILED)), &lwi);
 141
 142         if (lock->l_destroyed || lock->l_flags & LDLM_FL_FAILED) {
 143                 LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed");
 144                 RETURN(-EIO);
 145         }
 146
 147         if (rc) {
 148                 LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)",
 149                            rc);
 150                 RETURN(rc);
 151         }
 152
 153         LDLM_DEBUG(lock, "client-side enqueue waking up: granted");
 154         RETURN(0);
 155 }
 156
 157 /*
 158  * ->l_blocking_ast() callback for LDLM locks acquired by server-side OBDs.
 159  */
 160 int ldlm_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 161                       void *data, int flag)
 162 {
 163         int do_ast;
 164         ENTRY;
 165
 166         if (flag == LDLM_CB_CANCELING) {
 167                 /* Don't need to do anything here. */
 168                 RETURN(0);
 169         }
 170
 171         lock_res_and_lock(lock);
 172         /* Get this: if ldlm_blocking_ast is racing with intent_policy, such
 173          * that ldlm_blocking_ast is called just before intent_policy method
 174          * takes the ns_lock, then by the time we get the lock, we might not
 175          * be the correct blocking function anymore.  So check, and return
 176          * early, if so. */
 177         if (lock->l_blocking_ast != ldlm_blocking_ast) {
 178                 unlock_res_and_lock(lock);
 179                 RETURN(0);
 180         }
 181
 182         lock->l_flags |= LDLM_FL_CBPENDING;
 183         do_ast = (!lock->l_readers && !lock->l_writers);
 184         unlock_res_and_lock(lock);
 185
 186         if (do_ast) {
 187                 struct lustre_handle lockh;
 188                 int rc;
 189
 190                 LDLM_DEBUG(lock, "already unused, calling ldlm_cli_cancel");
 191                 ldlm_lock2handle(lock, &lockh);
 192                 rc = ldlm_cli_cancel(&lockh);
 193                 if (rc < 0)
 194                         CERROR("ldlm_cli_cancel: %d\n", rc);
 195         } else {
 196                 LDLM_DEBUG(lock, "Lock still has references, will be "
 197                            "cancelled later");
 198         }
 199         RETURN(0);
 200 }
 201
 202 /*
 203  * ->l_glimpse_ast() for DLM extent locks acquired on the server-side. See
 204  * comment in filter_intent_policy() on why you may need this.
 205  */
 206 int ldlm_glimpse_ast(struct ldlm_lock *lock, void *reqp)
 207 {
 208         /*
 209          * Returning -ELDLM_NO_LOCK_DATA actually works, but the reason for
 210          * that is rather subtle: with OST-side locking, it may so happen that
 211          * _all_ extent locks are held by the OST. If client wants to obtain
 212          * current file size it calls ll{,u}_glimpse_size(), and (as locks are
 213          * on the server), dummy glimpse callback fires and does
 214          * nothing. Client still receives correct file size due to the
 215          * following fragment in filter_intent_policy():
 216          *
 217          * rc = l->l_glimpse_ast(l, NULL); // this will update the LVB
 218          * if (rc != 0 && res->lr_namespace->ns_lvbo &&
 219          *     res->lr_namespace->ns_lvbo->lvbo_update) {
 220          *         res->lr_namespace->ns_lvbo->lvbo_update(res, NULL, 0, 1);
 221          * }
 222          *
 223          * that is, after glimpse_ast() fails, filter_lvbo_update() runs, and
 224          * returns correct file size to the client.
 225          */
 226         return -ELDLM_NO_LOCK_DATA;
 227 }
 228
 229 int ldlm_cli_enqueue_local(struct ldlm_namespace *ns,
 230                            const struct ldlm_res_id *res_id,
 231                            ldlm_type_t type, ldlm_policy_data_t *policy,
 232                            ldlm_mode_t mode, int *flags,
 233                            ldlm_blocking_callback blocking,
 234                            ldlm_completion_callback completion,
 235                            ldlm_glimpse_callback glimpse,
 236                            void *data, __u32 lvb_len, void *lvb_swabber,
 237                            struct lustre_handle *lockh)
 238 {
 239         struct ldlm_lock *lock;
 240         int err;
 241         ENTRY;
 242
 243         LASSERT(!(*flags & LDLM_FL_REPLAY));
 244         if (unlikely(ns->ns_client)) {
 245                 CERROR("Trying to enqueue local lock in a shadow namespace\n");
 246                 LBUG();
 247         }
 248
 249         lock = ldlm_lock_create(ns, res_id, type, mode, blocking,
 250                                 completion, glimpse, data, lvb_len);
 251         if (unlikely(!lock))
 252                 GOTO(out_nolock, err = -ENOMEM);
 253         LDLM_DEBUG(lock, "client-side local enqueue handler, new lock created");
 254
 255         ldlm_lock_addref_internal(lock, mode);
 256         ldlm_lock2handle(lock, lockh);
 257         lock_res_and_lock(lock);
 258         lock->l_flags |= LDLM_FL_LOCAL;
 259         if (*flags & LDLM_FL_ATOMIC_CB)
 260                 lock->l_flags |= LDLM_FL_ATOMIC_CB;
 261         lock->l_lvb_swabber = lvb_swabber;
 262         unlock_res_and_lock(lock);
 263         if (policy != NULL)
 264                 lock->l_policy_data = *policy;
 265         if (type == LDLM_EXTENT)
 266                 lock->l_req_extent = policy->l_extent;
 267
 268         err = ldlm_lock_enqueue(ns, &lock, policy, flags);
 269         if (unlikely(err != ELDLM_OK))
 270                 GOTO(out, err);
 271
 272         if (policy != NULL)
 273                 *policy = lock->l_policy_data;
 274
 275         LDLM_DEBUG_NOLOCK("client-side local enqueue handler END (lock %p)",
 276                           lock);
 277
 278         if (lock->l_completion_ast)
 279                 lock->l_completion_ast(lock, *flags, NULL);
 280
 281         LDLM_DEBUG(lock, "client-side local enqueue END");
 282         EXIT;
 283  out:
 284         LDLM_LOCK_PUT(lock);
 285  out_nolock:
 286         return err;
 287 }
 288
 289 static void failed_lock_cleanup(struct ldlm_namespace *ns,
 290                                 struct ldlm_lock *lock,
 291                                 struct lustre_handle *lockh, int mode)
 292 {
 293         /* Set a flag to prevent us from sending a CANCEL (bug 407) */
 294         lock_res_and_lock(lock);
 295         lock->l_flags |= LDLM_FL_LOCAL_ONLY;
 296         unlock_res_and_lock(lock);
 297         LDLM_DEBUG(lock, "setting FL_LOCAL_ONLY");
 298
 299         ldlm_lock_decref_and_cancel(lockh, mode);
 300
 301         /* XXX - HACK because we shouldn't call ldlm_lock_destroy()
 302          *       from llite/file.c/ll_file_flock(). */
 303         if (lock->l_resource->lr_type == LDLM_FLOCK) {
 304                 ldlm_lock_destroy(lock);
 305         }
 306 }
 307
 308 int ldlm_cli_enqueue_fini(struct obd_export *exp, struct ptlrpc_request *req,
 309                           ldlm_type_t type, __u8 with_policy, ldlm_mode_t mode,
 310                           int *flags, void *lvb, __u32 lvb_len,
 311                           void *lvb_swabber, struct lustre_handle *lockh,int rc)
 312 {
 313         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
 314         int is_replay = *flags & LDLM_FL_REPLAY;
 315         struct ldlm_lock *lock;
 316         struct ldlm_reply *reply;
 317         int cleanup_phase = 1;
 318         ENTRY;
 319
 320         lock = ldlm_handle2lock(lockh);
 321         /* ldlm_cli_enqueue is holding a reference on this lock. */
 322         LASSERT(lock != NULL);
 323         if (rc != ELDLM_OK) {
 324                 LASSERT(!is_replay);
 325                 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
 326                            rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
 327                 if (rc == ELDLM_LOCK_ABORTED) {
 328                         /* Before we return, swab the reply */
 329                         reply = lustre_swab_repbuf(req, DLM_LOCKREPLY_OFF,
 330                                                    sizeof(*reply),
 331                                                    lustre_swab_ldlm_reply);
 332                         if (reply == NULL) {
 333                                 CERROR("Can't unpack ldlm_reply\n");
 334                                 rc = -EPROTO;
 335                         }
 336                         if (lvb_len) {
 337                                 void *tmplvb;
 338                                 tmplvb = lustre_swab_repbuf(req,
 339                                                             DLM_REPLY_REC_OFF,
 340                                                             lvb_len,
 341                                                             lvb_swabber);
 342                                 if (tmplvb == NULL)
 343                                         GOTO(cleanup, rc = -EPROTO);
 344                                 if (lvb != NULL)
 345                                         memcpy(lvb, tmplvb, lvb_len);
 346                         }
 347                 }
 348                 GOTO(cleanup, rc);
 349         }
 350
 351         reply = lustre_swab_repbuf(req, DLM_LOCKREPLY_OFF, sizeof(*reply),
 352                                    lustre_swab_ldlm_reply);
 353         if (reply == NULL) {
 354                 CERROR("Can't unpack ldlm_reply\n");
 355                 GOTO(cleanup, rc = -EPROTO);
 356         }
 357
 358         /* lock enqueued on the server */
 359         cleanup_phase = 0;
 360
 361         lock_res_and_lock(lock);
 362         lock->l_remote_handle = reply->lock_handle;
 363         *flags = reply->lock_flags;
 364         lock->l_flags |= reply->lock_flags & LDLM_INHERIT_FLAGS;
 365         /* move NO_TIMEOUT flag to the lock to force ldlm_lock_match()
 366          * to wait with no timeout as well */
 367         lock->l_flags |= reply->lock_flags & LDLM_FL_NO_TIMEOUT;
 368         unlock_res_and_lock(lock);
 369
 370         CDEBUG(D_INFO, "local: %p, remote cookie: "LPX64", flags: 0x%x\n",
 371                lock, reply->lock_handle.cookie, *flags);
 372
 373         /* If enqueue returned a blocked lock but the completion handler has
 374          * already run, then it fixed up the resource and we don't need to do it
 375          * again. */
 376         if ((*flags) & LDLM_FL_LOCK_CHANGED) {
 377                 int newmode = reply->lock_desc.l_req_mode;
 378                 LASSERT(!is_replay);
 379                 if (newmode && newmode != lock->l_req_mode) {
 380                         LDLM_DEBUG(lock, "server returned different mode %s",
 381                                    ldlm_lockname[newmode]);
 382                         lock->l_req_mode = newmode;
 383                 }
 384
 385                 if (memcmp(reply->lock_desc.l_resource.lr_name.name,
 386                           lock->l_resource->lr_name.name,
 387                           sizeof(struct ldlm_res_id))) {
 388                         CDEBUG(D_INFO, "remote intent success, locking "
 389                                         "(%ld,%ld,%ld) instead of "
 390                                         "(%ld,%ld,%ld)\n",
 391                               (long)reply->lock_desc.l_resource.lr_name.name[0],
 392                               (long)reply->lock_desc.l_resource.lr_name.name[1],
 393                               (long)reply->lock_desc.l_resource.lr_name.name[2],
 394                               (long)lock->l_resource->lr_name.name[0],
 395                               (long)lock->l_resource->lr_name.name[1],
 396                               (long)lock->l_resource->lr_name.name[2]);
 397
 398                         ldlm_lock_change_resource(ns, lock,
 399                                           &reply->lock_desc.l_resource.lr_name);
 400                         if (lock->l_resource == NULL) {
 401                                 LBUG();
 402                                 GOTO(cleanup, rc = -ENOMEM);
 403                         }
 404                         LDLM_DEBUG(lock, "client-side enqueue, new resource");
 405                 }
 406                 if (with_policy)
 407                         if (!(type == LDLM_IBITS && !(exp->exp_connect_flags &
 408                                                     OBD_CONNECT_IBITS)))
 409                                 lock->l_policy_data =
 410                                                  reply->lock_desc.l_policy_data;
 411                 if (type != LDLM_PLAIN)
 412                         LDLM_DEBUG(lock,"client-side enqueue, new policy data");
 413         }
 414
 415         if ((*flags) & LDLM_FL_AST_SENT ||
 416             /* Cancel extent locks as soon as possible on a liblustre client,
 417              * because it cannot handle asynchronous ASTs robustly (see
 418              * bug 7311). */
 419             (LIBLUSTRE_CLIENT && type == LDLM_EXTENT)) {
 420                 lock_res_and_lock(lock);
 421                 lock->l_flags |= LDLM_FL_CBPENDING;
 422                 unlock_res_and_lock(lock);
 423                 LDLM_DEBUG(lock, "enqueue reply includes blocking AST");
 424         }
 425
 426         /* If the lock has already been granted by a completion AST, don't
 427          * clobber the LVB with an older one. */
 428         if (lvb_len && (lock->l_req_mode != lock->l_granted_mode)) {
 429                 void *tmplvb;
 430                 tmplvb = lustre_swab_repbuf(req, DLM_REPLY_REC_OFF, lvb_len,
 431                                             lvb_swabber);
 432                 if (tmplvb == NULL)
 433                         GOTO(cleanup, rc = -EPROTO);
 434                 memcpy(lock->l_lvb_data, tmplvb, lvb_len);
 435         }
 436
 437         if (!is_replay) {
 438                 rc = ldlm_lock_enqueue(ns, &lock, NULL, flags);
 439                 if (lock->l_completion_ast != NULL) {
 440                         int err = lock->l_completion_ast(lock, *flags, NULL);
 441                         if (!rc)
 442                                 rc = err;
 443                         if (rc && type != LDLM_FLOCK) /* bug 9425, bug 10250 */
 444                                 cleanup_phase = 1;
 445                 }
 446         }
 447
 448         if (lvb_len && lvb != NULL) {
 449                 /* Copy the LVB here, and not earlier, because the completion
 450                  * AST (if any) can override what we got in the reply */
 451                 memcpy(lvb, lock->l_lvb_data, lvb_len);
 452         }
 453
 454         LDLM_DEBUG(lock, "client-side enqueue END");
 455         EXIT;
 456 cleanup:
 457         if (cleanup_phase == 1 && rc)
 458                 failed_lock_cleanup(ns, lock, lockh, mode);
 459         /* Put lock 2 times, the second reference is held by ldlm_cli_enqueue */
 460         LDLM_LOCK_PUT(lock);
 461         LDLM_LOCK_PUT(lock);
 462         return rc;
 463 }
 464
 465 /* PAGE_SIZE-512 is to allow TCP/IP and LNET headers to fit into
 466  * a single page on the send/receive side. XXX: 512 should be changed
 467  * to more adequate value. */
 468 #define ldlm_req_handles_avail(exp, size, bufcount, off)                \
 469 ({                                                                      \
 470         int _avail = min_t(int, LDLM_MAXREQSIZE, PAGE_SIZE - 512);      \
 471         int _s = size[DLM_LOCKREQ_OFF];                                 \
 472         size[DLM_LOCKREQ_OFF] = sizeof(struct ldlm_request);            \
 473         _avail -= lustre_msg_size(class_exp2cliimp(exp)->imp_msg_magic, \
 474                                 bufcount, size);                        \
 475         _avail /= sizeof(struct lustre_handle);                         \
 476         _avail += LDLM_LOCKREQ_HANDLES - off;                           \
 477         size[DLM_LOCKREQ_OFF] = _s;                                     \
 478         _avail;                                                         \
 479 })
 480
 481 /* Cancel lru locks and pack them into the enqueue request. Pack there the given
 482  * @count locks in @cancel. */
 483 struct ptlrpc_request *ldlm_prep_enqueue_req(struct obd_export *exp,
 484                                              int bufcount, int *size,
 485                                              struct list_head *cancels,
 486                                              int count)
 487 {
 488         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
 489         struct ldlm_request *dlm = NULL;
 490         struct ptlrpc_request *req;
 491         CFS_LIST_HEAD(head);
 492         ENTRY;
 493
 494         if (cancels == NULL)
 495                 cancels = &head;
 496         if (exp_connect_cancelset(exp)) {
 497                 /* Estimate the amount of available space in the request. */
 498                 int avail = ldlm_req_handles_avail(exp, size, bufcount,
 499                                                    LDLM_ENQUEUE_CANCEL_OFF);
 500                 LASSERT(avail >= count);
 501
 502                 /* Cancel lru locks here _only_ if the server supports
 503                  * EARLY_CANCEL. Otherwise we have to send extra CANCEL
 504                  * rpc right on enqueue, what will make it slower, vs.
 505                  * asynchronous rpc in blocking thread. */
 506                 count += ldlm_cancel_lru_local(ns, cancels, 1, avail - count,
 507                                                LDLM_CANCEL_AGED);
 508                 size[DLM_LOCKREQ_OFF] =
 509                         ldlm_request_bufsize(count, LDLM_ENQUEUE);
 510         }
 511         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_DLM_VERSION,
 512                               LDLM_ENQUEUE, bufcount, size, NULL);
 513         if (exp_connect_cancelset(exp) && req) {
 514                 dlm = lustre_msg_buf(req->rq_reqmsg,
 515                                      DLM_LOCKREQ_OFF, sizeof(*dlm));
 516                 /* Skip first lock handler in ldlm_request_pack(), this method
 517                  * will incrment @lock_count according to the lock handle amount
 518                  * actually written to the buffer. */
 519                 dlm->lock_count = LDLM_ENQUEUE_CANCEL_OFF;
 520         }
 521         if (req)
 522                 ldlm_cli_cancel_list(cancels, count, req, DLM_LOCKREQ_OFF, 0);
 523         else
 524                 ldlm_lock_list_put(cancels, l_bl_ast, count);
 525         RETURN(req);
 526 }
 527
 528 /* If a request has some specific initialisation it is passed in @reqp,
 529  * otherwise it is created in ldlm_cli_enqueue.
 530  *
 531  * Supports sync and async requests, pass @async flag accordingly. If a
 532  * request was created in ldlm_cli_enqueue and it is the async request,
 533  * pass it to the caller in @reqp. */
 534 int ldlm_cli_enqueue(struct obd_export *exp, struct ptlrpc_request **reqp,
 535                      const struct ldlm_res_id *res_id,
 536                      ldlm_type_t type, ldlm_policy_data_t *policy,
 537                      ldlm_mode_t mode, int *flags,
 538                      ldlm_blocking_callback blocking,
 539                      ldlm_completion_callback completion,
 540                      ldlm_glimpse_callback glimpse,
 541                      void *data, void *lvb, __u32 lvb_len, void *lvb_swabber,
 542                      struct lustre_handle *lockh, int async)
 543 {
 544         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
 545         struct ldlm_lock *lock;
 546         struct ldlm_request *body;
 547         struct ldlm_reply *reply;
 548         int size[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 549                         [DLM_LOCKREQ_OFF]     = sizeof(*body),
 550                         [DLM_REPLY_REC_OFF]   = lvb_len };
 551         int is_replay = *flags & LDLM_FL_REPLAY;
 552         int req_passed_in = 1, rc;
 553         struct ptlrpc_request *req;
 554         ENTRY;
 555
 556         LASSERT(exp != NULL);
 557
 558         /* If we're replaying this lock, just check some invariants.
 559          * If we're creating a new lock, get everything all setup nice. */
 560         if (is_replay) {
 561                 lock = ldlm_handle2lock(lockh);
 562                 LASSERT(lock != NULL);
 563                 LDLM_DEBUG(lock, "client-side enqueue START");
 564                 LASSERT(exp == lock->l_conn_export);
 565         } else {
 566                 lock = ldlm_lock_create(ns, res_id, type, mode, blocking,
 567                                         completion, glimpse, data, lvb_len);
 568                 if (lock == NULL)
 569                         RETURN(-ENOMEM);
 570                 /* for the local lock, add the reference */
 571                 ldlm_lock_addref_internal(lock, mode);
 572                 ldlm_lock2handle(lock, lockh);
 573                 lock->l_lvb_swabber = lvb_swabber;
 574                 if (policy != NULL) {
 575                         /* INODEBITS_INTEROP: If the server does not support
 576                          * inodebits, we will request a plain lock in the
 577                          * descriptor (ldlm_lock2desc() below) but use an
 578                          * inodebits lock internally with both bits set.
 579                          */
 580                         if (type == LDLM_IBITS && !(exp->exp_connect_flags &
 581                                                     OBD_CONNECT_IBITS))
 582                                 lock->l_policy_data.l_inodebits.bits =
 583                                         MDS_INODELOCK_LOOKUP |
 584                                         MDS_INODELOCK_UPDATE;
 585                         else
 586                                 lock->l_policy_data = *policy;
 587                 }
 588
 589                 if (type == LDLM_EXTENT)
 590                         lock->l_req_extent = policy->l_extent;
 591                 LDLM_DEBUG(lock, "client-side enqueue START");
 592         }
 593
 594         /* lock not sent to server yet */
 595
 596         if (reqp == NULL || *reqp == NULL) {
 597                 req = ldlm_prep_enqueue_req(exp, 2, size, NULL, 0);
 598                 if (req == NULL) {
 599                         failed_lock_cleanup(ns, lock, lockh, mode);
 600                         LDLM_LOCK_PUT(lock);
 601                         RETURN(-ENOMEM);
 602                 }
 603                 req_passed_in = 0;
 604                 if (reqp)
 605                         *reqp = req;
 606         } else {
 607                 req = *reqp;
 608                 LASSERTF(lustre_msg_buflen(req->rq_reqmsg, DLM_LOCKREQ_OFF) >=
 609                          sizeof(*body), "buflen[%d] = %d, not "LPSZ"\n",
 610                          DLM_LOCKREQ_OFF,
 611                          lustre_msg_buflen(req->rq_reqmsg, DLM_LOCKREQ_OFF),
 612                          sizeof(*body));
 613         }
 614
 615         lock->l_conn_export = exp;
 616         lock->l_export = NULL;
 617         lock->l_blocking_ast = blocking;
 618
 619         /* Dump lock data into the request buffer */
 620         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 621         ldlm_lock2desc(lock, &body->lock_desc);
 622         body->lock_flags = *flags;
 623         body->lock_handle[0] = *lockh;
 624
 625         /* Continue as normal. */
 626         if (!req_passed_in) {
 627                 size[DLM_LOCKREPLY_OFF] = sizeof(*reply);
 628                 ptlrpc_req_set_repsize(req, 2 + (lvb_len > 0), size);
 629         }
 630
 631         /*
 632          * Liblustre client doesn't get extent locks, except for O_APPEND case
 633          * where [0, OBD_OBJECT_EOF] lock is taken, or truncate, where
 634          * [i_size, OBD_OBJECT_EOF] lock is taken.
 635          */
 636         LASSERT(ergo(LIBLUSTRE_CLIENT, type != LDLM_EXTENT ||
 637                      policy->l_extent.end == OBD_OBJECT_EOF));
 638
 639         if (async) {
 640                 LASSERT(reqp != NULL);
 641                 RETURN(0);
 642         }
 643
 644         LDLM_DEBUG(lock, "sending request");
 645         rc = ptlrpc_queue_wait(req);
 646         rc = ldlm_cli_enqueue_fini(exp, req, type, policy ? 1 : 0,
 647                                    mode, flags, lvb, lvb_len, lvb_swabber,
 648                                    lockh, rc);
 649
 650         if (!req_passed_in && req != NULL) {
 651                 ptlrpc_req_finished(req);
 652                 if (reqp)
 653                         *reqp = NULL;
 654         }
 655
 656         RETURN(rc);
 657 }
 658
 659 static int ldlm_cli_convert_local(struct ldlm_lock *lock, int new_mode,
 660                                   int *flags)
 661 {
 662         struct ldlm_resource *res;
 663         int rc;
 664         ENTRY;
 665         if (lock->l_resource->lr_namespace->ns_client) {
 666                 CERROR("Trying to cancel local lock\n");
 667                 LBUG();
 668         }
 669         LDLM_DEBUG(lock, "client-side local convert");
 670
 671         res = ldlm_lock_convert(lock, new_mode, flags);
 672         if (res) {
 673                 ldlm_reprocess_all(res);
 674                 rc = 0;
 675         } else {
 676                 rc = EDEADLOCK;
 677         }
 678         LDLM_DEBUG(lock, "client-side local convert handler END");
 679         LDLM_LOCK_PUT(lock);
 680         RETURN(rc);
 681 }
 682
 683 /* FIXME: one of ldlm_cli_convert or the server side should reject attempted
 684  * conversion of locks which are on the waiting or converting queue */
 685 /* Caller of this code is supposed to take care of lock readers/writers
 686    accounting */
 687 int ldlm_cli_convert(struct lustre_handle *lockh, int new_mode, int *flags)
 688 {
 689         struct ldlm_request *body;
 690         struct ldlm_reply *reply;
 691         struct ldlm_lock *lock;
 692         struct ldlm_resource *res;
 693         struct ptlrpc_request *req;
 694         int size[2] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 695                         [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 696         int rc;
 697         ENTRY;
 698
 699         lock = ldlm_handle2lock(lockh);
 700         if (!lock) {
 701                 LBUG();
 702                 RETURN(-EINVAL);
 703         }
 704         *flags = 0;
 705
 706         if (lock->l_conn_export == NULL)
 707                 RETURN(ldlm_cli_convert_local(lock, new_mode, flags));
 708
 709         LDLM_DEBUG(lock, "client-side convert");
 710
 711         req = ptlrpc_prep_req(class_exp2cliimp(lock->l_conn_export),
 712                               LUSTRE_DLM_VERSION, LDLM_CONVERT, 2, size, NULL);
 713         if (!req)
 714                 GOTO(out, rc = -ENOMEM);
 715
 716         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 717         body->lock_handle[0] = lock->l_remote_handle;
 718
 719         body->lock_desc.l_req_mode = new_mode;
 720         body->lock_flags = *flags;
 721
 722         size[DLM_LOCKREPLY_OFF] = sizeof(*reply);
 723         ptlrpc_req_set_repsize(req, 2, size);
 724
 725         rc = ptlrpc_queue_wait(req);
 726         if (rc != ELDLM_OK)
 727                 GOTO(out, rc);
 728
 729         reply = lustre_swab_repbuf(req, DLM_LOCKREPLY_OFF, sizeof(*reply),
 730                                    lustre_swab_ldlm_reply);
 731         if (reply == NULL) {
 732                 CERROR ("Can't unpack ldlm_reply\n");
 733                 GOTO (out, rc = -EPROTO);
 734         }
 735
 736         if (req->rq_status)
 737                 GOTO(out, rc = req->rq_status);
 738
 739         res = ldlm_lock_convert(lock, new_mode, &reply->lock_flags);
 740         if (res != NULL) {
 741                 ldlm_reprocess_all(res);
 742                 /* Go to sleep until the lock is granted. */
 743                 /* FIXME: or cancelled. */
 744                 if (lock->l_completion_ast) {
 745                         rc = lock->l_completion_ast(lock, LDLM_FL_WAIT_NOREPROC,
 746                                                     NULL);
 747                         if (rc)
 748                                 GOTO(out, rc);
 749                 }
 750         } else {
 751                 rc = EDEADLOCK;
 752         }
 753         EXIT;
 754  out:
 755         LDLM_LOCK_PUT(lock);
 756         ptlrpc_req_finished(req);
 757         return rc;
 758 }
 759
 760 /* Cancel locks locally.
 761  * Returns: 1 if there is a need to send a cancel RPC to server. 0 otherwise. */
 762 static int ldlm_cli_cancel_local(struct ldlm_lock *lock)
 763 {
 764         int rc = 0;
 765         ENTRY;
 766
 767         if (lock->l_conn_export) {
 768                 int local_only;
 769
 770                 LDLM_DEBUG(lock, "client-side cancel");
 771                 /* Set this flag to prevent others from getting new references*/
 772                 lock_res_and_lock(lock);
 773                 lock->l_flags |= LDLM_FL_CBPENDING;
 774                 local_only = (lock->l_flags &
 775                               (LDLM_FL_LOCAL_ONLY|LDLM_FL_CANCEL_ON_BLOCK));
 776                 ldlm_cancel_callback(lock);
 777                 unlock_res_and_lock(lock);
 778
 779                 if (local_only)
 780                         CDEBUG(D_INFO, "not sending request (at caller's "
 781                                "instruction)\n");
 782                 else
 783                         rc = 1;
 784
 785                 ldlm_lock_cancel(lock);
 786         } else {
 787                 if (lock->l_resource->lr_namespace->ns_client) {
 788                         LDLM_ERROR(lock, "Trying to cancel local lock");
 789                         LBUG();
 790                 }
 791                 LDLM_DEBUG(lock, "server-side local cancel");
 792                 ldlm_lock_cancel(lock);
 793                 ldlm_reprocess_all(lock->l_resource);
 794                 LDLM_DEBUG(lock, "server-side local cancel handler END");
 795         }
 796
 797         RETURN(rc);
 798 }
 799
 800 /* Pack @count locks in @head into ldlm_request buffer at the offset @off,
 801    of the request @req. */
 802 static void ldlm_cancel_pack(struct ptlrpc_request *req, int off,
 803                              struct list_head *head, int count)
 804 {
 805         struct ldlm_request *dlm;
 806         struct ldlm_lock *lock;
 807         int max;
 808         ENTRY;
 809
 810         dlm = lustre_msg_buf(req->rq_reqmsg, off, sizeof(*dlm));
 811         LASSERT(dlm != NULL);
 812
 813         /* Check the room in the request buffer. */
 814         max = lustre_msg_buflen(req->rq_reqmsg, off) -
 815                 sizeof(struct ldlm_request);
 816         max /= sizeof(struct lustre_handle);
 817         max += LDLM_LOCKREQ_HANDLES;
 818         LASSERT(max >= dlm->lock_count + count);
 819
 820         /* XXX: it would be better to pack lock handles grouped by resource.
 821          * so that the server cancel would call filter_lvbo_update() less
 822          * frequently. */
 823         list_for_each_entry(lock, head, l_bl_ast) {
 824                 if (!count--)
 825                         break;
 826                 /* Pack the lock handle to the given request buffer. */
 827                 LASSERT(lock->l_conn_export);
 828                 /* Cannot be set on a lock in a resource granted list.*/
 829                 LASSERT(!(lock->l_flags &
 830                           (LDLM_FL_LOCAL_ONLY|LDLM_FL_CANCEL_ON_BLOCK)));
 831                 /* If @lock is marked CANCEL_ON_BLOCK, cancel
 832                  * will not be sent in ldlm_cli_cancel(). It
 833                  * is used for liblustre clients, no cancel on
 834                  * block requests. However, even for liblustre
 835                  * clients, when the flag is set, batched cancel
 836                  * should be sent (what if no block rpc has
 837                  * come). To not send another separated rpc in
 838                  * this case, the caller pass CANCEL_ON_BLOCK
 839                  * flag to ldlm_cli_cancel_unused_resource(). */
 840                 dlm->lock_handle[dlm->lock_count++] = lock->l_remote_handle;
 841         }
 842         EXIT;
 843 }
 844
 845 /* Prepare and send a batched cancel rpc, it will include count lock handles
 846  * of locks given in @head. */
 847 int ldlm_cli_cancel_req(struct obd_export *exp, struct list_head *cancels,
 848                         int count, int flags)
 849 {
 850         struct ptlrpc_request *req = NULL;
 851         struct ldlm_request *body;
 852         int size[2] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 853                 [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 854         struct obd_import *imp;
 855         int free, sent = 0;
 856         int rc = 0;
 857         ENTRY;
 858
 859         LASSERT(exp != NULL);
 860         LASSERT(count > 0);
 861
 862         free = ldlm_req_handles_avail(exp, size, 2, 0);
 863         if (count > free)
 864                 count = free;
 865
 866         size[DLM_LOCKREQ_OFF] = ldlm_request_bufsize(count, LDLM_CANCEL);
 867         while (1) {
 868                 imp = class_exp2cliimp(exp);
 869                 if (imp == NULL || imp->imp_invalid) {
 870                         CDEBUG(D_HA, "skipping cancel on invalid import %p\n",
 871                                imp);
 872                         break;
 873                 }
 874
 875                 req = ptlrpc_prep_req(imp, LUSTRE_DLM_VERSION, LDLM_CANCEL, 2,
 876                                       size, NULL);
 877                 if (!req)
 878                         GOTO(out, rc = -ENOMEM);
 879
 880                 req->rq_no_resend = 1;
 881
 882                 /* XXX FIXME bug 249 */
 883                 req->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
 884                 req->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
 885
 886                 body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF,
 887                                       sizeof(*body));
 888                 ldlm_cancel_pack(req, DLM_LOCKREQ_OFF, cancels, count);
 889
 890                 ptlrpc_req_set_repsize(req, 1, NULL);
 891                 if (flags & LDLM_FL_ASYNC) {
 892                         ptlrpcd_add_req(req);
 893                         sent = count;
 894                         GOTO(out, 0);
 895                 } else {
 896                         rc = ptlrpc_queue_wait(req);
 897                 }
 898                 if (rc == ESTALE) {
 899                         CDEBUG(D_DLMTRACE, "client/server (nid %s) "
 900                                "out of sync -- not fatal\n",
 901                                libcfs_nid2str(req->rq_import->
 902                                               imp_connection->c_peer.nid));
 903                 } else if (rc == -ETIMEDOUT) {
 904                         ptlrpc_req_finished(req);
 905                         continue;
 906                 } else if (rc != ELDLM_OK) {
 907                         CERROR("Got rc %d from cancel RPC: canceling "
 908                                "anyway\n", rc);
 909                         break;
 910                 }
 911                 sent = count;
 912                 break;
 913         }
 914
 915         ptlrpc_req_finished(req);
 916         EXIT;
 917 out:
 918         return sent ? sent : rc;
 919 }
 920
 921 int ldlm_cli_cancel(struct lustre_handle *lockh)
 922 {
 923         struct ldlm_lock *lock;
 924         CFS_LIST_HEAD(head);
 925         int rc = 0;
 926         ENTRY;
 927
 928         /* concurrent cancels on the same handle can happen */
 929         lock = __ldlm_handle2lock(lockh, LDLM_FL_CANCELING);
 930         if (lock == NULL)
 931                 RETURN(0);
 932
 933         rc = ldlm_cli_cancel_local(lock);
 934         if (rc <= 0)
 935                 GOTO(out, rc);
 936
 937         list_add(&lock->l_bl_ast, &head);
 938         rc = ldlm_cli_cancel_req(lock->l_conn_export, &head, 1, 0);
 939         EXIT;
 940 out:
 941         LDLM_LOCK_PUT(lock);
 942         return rc < 0 ? rc : 0;
 943 }
 944
 945 /* - Free space in lru for @count new locks,
 946  *   redundant unused locks are canceled locally;
 947  * - also cancel locally unused aged locks;
 948  * - do not cancel more than @max locks;
 949  * - GET the found locks and add them into the @cancels list.
 950  *
 951  * A client lock can be added to the l_bl_ast list only when it is
 952  * marked LDLM_FL_CANCELING. Otherwise, somebody is already doing CANCEL.
 953  * There are the following use cases: ldlm_cancel_resource_local(),
 954  * ldlm_cancel_lru_local() and ldlm_cli_cancel(), which check&set this
 955  * flag properly. As any attempt to cancel a lock rely on this flag,
 956  * l_bl_ast list is accessed later without any special locking. */
 957 int ldlm_cancel_lru_local(struct ldlm_namespace *ns, struct list_head *cancels,
 958                           int count, int max, int flags)
 959 {
 960         cfs_time_t cur = cfs_time_current();
 961         struct ldlm_lock *lock, *next;
 962         int rc, added = 0, left;
 963         ENTRY;
 964
 965         spin_lock(&ns->ns_unused_lock);
 966         count += ns->ns_nr_unused - ns->ns_max_unused;
 967         while (!list_empty(&ns->ns_unused_list)) {
 968                 struct list_head *tmp = ns->ns_unused_list.next;
 969                 lock = list_entry(tmp, struct ldlm_lock, l_lru);
 970
 971                 if (max && added >= max)
 972                         break;
 973
 974                 if ((added >= count) &&
 975                     (!(flags & LDLM_CANCEL_AGED) ||
 976                      cfs_time_before_64(cur, ns->ns_max_age +
 977                                         lock->l_last_used)))
 978                         break;
 979
 980                 LDLM_LOCK_GET(lock); /* dropped by bl thread */
 981                 spin_unlock(&ns->ns_unused_lock);
 982
 983                 lock_res_and_lock(lock);
 984                 if ((ldlm_lock_remove_from_lru(lock) == 0) ||
 985                     (lock->l_flags & LDLM_FL_CANCELING)) {
 986                         /* other thread is removing lock from lru or
 987                          * somebody is already doing CANCEL. */
 988                         unlock_res_and_lock(lock);
 989                         LDLM_LOCK_PUT(lock);
 990                         spin_lock(&ns->ns_unused_lock);
 991                         continue;
 992                 }
 993                 LASSERT(!lock->l_readers && !lock->l_writers);
 994
 995                 /* If we have chosen to canecl this lock voluntarily, we better
 996                    send cancel notification to server, so that it frees
 997                    appropriate state. This might lead to a race where while
 998                    we are doing cancel here, server is also silently
 999                    cancelling this lock. */
1000                 lock->l_flags &= ~LDLM_FL_CANCEL_ON_BLOCK;
1001
1002                 /* Setting the CBPENDING flag is a little misleading, but
1003                  * prevents an important race; namely, once CBPENDING is set,
1004                  * the lock can accumulate no more readers/writers.  Since
1005                  * readers and writers are already zero here, ldlm_lock_decref
1006                  * won't see this flag and call l_blocking_ast */
1007                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_CANCELING;
1008                 /* We can't re-add to l_lru as it confuses the refcounting in
1009                  * ldlm_lock_remove_from_lru() if an AST arrives after we drop
1010                  * ns_lock below. We use l_bl_ast and can't use l_pending_chain
1011                  * as it is used both on server and client nevertheles bug 5666
1012                  * says it is used only on server. --umka */
1013
1014                 LASSERT(list_empty(&lock->l_bl_ast));
1015                 list_add(&lock->l_bl_ast, cancels);
1016                 unlock_res_and_lock(lock);
1017                 spin_lock(&ns->ns_unused_lock);
1018                 added++;
1019         }
1020         spin_unlock(&ns->ns_unused_lock);
1021
1022         /* Handle only @added inserted locks. */
1023         left = added;
1024         list_for_each_entry_safe(lock, next, cancels, l_bl_ast) {
1025                 if (left-- == 0)
1026                         break;
1027                 rc = ldlm_cli_cancel_local(lock);
1028                 if (rc == 0) {
1029                         /* CANCEL RPC should not be sent to server. */
1030                         list_del_init(&lock->l_bl_ast);
1031                         LDLM_LOCK_PUT(lock);
1032                         added--;
1033                 }
1034         }
1035         RETURN(added);
1036 }
1037
1038 /* when called with LDLM_ASYNC the blocking callback will be handled
1039  * in a thread and this function will return after the thread has been
1040  * asked to call the callback.  when called with LDLM_SYNC the blocking
1041  * callback will be performed in this function. */
1042 int ldlm_cancel_lru(struct ldlm_namespace *ns, ldlm_sync_t sync)
1043 {
1044         CFS_LIST_HEAD(cancels);
1045         int count, rc;
1046         ENTRY;
1047
1048 #ifndef __KERNEL__
1049         sync = LDLM_SYNC; /* force to be sync in user space */
1050 #endif
1051         count = ldlm_cancel_lru_local(ns, &cancels, 0, 0, 0);
1052         if (sync == LDLM_ASYNC) {
1053                 struct ldlm_lock *lock, *next;
1054                 list_for_each_entry_safe(lock, next, &cancels, l_bl_ast) {
1055                         /* Remove from the list to allow blocking thread to
1056                          * re-use l_bl_ast. */
1057                         list_del_init(&lock->l_bl_ast);
1058                         rc = ldlm_bl_to_thread(ns, NULL, lock,
1059                                                LDLM_FL_CANCELING);
1060                         if (rc)
1061                                 list_add_tail(&lock->l_bl_ast, &next->l_bl_ast);
1062                 }
1063         }
1064
1065         /* If some locks are left in the list in ASYNC mode, or
1066          * this is SYNC mode, cancel the list. */
1067         ldlm_cli_cancel_list(&cancels, count, NULL, DLM_LOCKREQ_OFF, 0);
1068         RETURN(0);
1069 }
1070
1071 /* Find and cancel locally unused locks found on resource, matched to the
1072  * given policy, mode. GET the found locks and add them into the @cancels
1073  * list. */
1074 int ldlm_cancel_resource_local(struct ldlm_resource *res,
1075                                struct list_head *cancels,
1076                                ldlm_policy_data_t *policy,
1077                                ldlm_mode_t mode, int lock_flags,
1078                                int flags, void *opaque)
1079 {
1080         struct ldlm_lock *lock, *next;
1081         int count = 0, left;
1082         ENTRY;
1083
1084         lock_res(res);
1085         list_for_each_entry(lock, &res->lr_granted, l_res_link) {
1086                 if (opaque != NULL && lock->l_ast_data != opaque) {
1087                         LDLM_ERROR(lock, "data %p doesn't match opaque %p",
1088                                    lock->l_ast_data, opaque);
1089                         //LBUG();
1090                         continue;
1091                 }
1092
1093                 if (lock->l_readers || lock->l_writers) {
1094                         if (flags & LDLM_FL_WARN) {
1095                                 LDLM_ERROR(lock, "lock in use");
1096                                 //LBUG();
1097                         }
1098                         continue;
1099                 }
1100
1101                 if (lockmode_compat(lock->l_granted_mode, mode))
1102                         continue;
1103
1104                 /* If policy is given and this is IBITS lock, add to list only
1105                  * those locks that match by policy. */
1106                 if (policy && (lock->l_resource->lr_type == LDLM_IBITS) &&
1107                     !(lock->l_policy_data.l_inodebits.bits &
1108                       policy->l_inodebits.bits))
1109                         continue;
1110
1111                 /* If somebody is already doing CANCEL, skip it. */
1112                 if (lock->l_flags & LDLM_FL_CANCELING)
1113                         continue;
1114
1115                 /* See CBPENDING comment in ldlm_cancel_lru */
1116                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_CANCELING |
1117                         lock_flags;
1118
1119                 LASSERT(list_empty(&lock->l_bl_ast));
1120                 list_add(&lock->l_bl_ast, cancels);
1121                 LDLM_LOCK_GET(lock);
1122                 count++;
1123         }
1124         unlock_res(res);
1125
1126         /* Handle only @count inserted locks. */
1127         left = count;
1128         list_for_each_entry_safe(lock, next, cancels, l_bl_ast) {
1129                 int rc = 0;
1130
1131                 if (left-- == 0)
1132                         break;
1133                 if (flags & LDLM_FL_LOCAL_ONLY)
1134                         ldlm_lock_cancel(lock);
1135                 else
1136                         rc = ldlm_cli_cancel_local(lock);
1137
1138                 if (rc == 0) {
1139                         /* CANCEL RPC should not be sent to server. */
1140                         list_del_init(&lock->l_bl_ast);
1141                         LDLM_LOCK_PUT(lock);
1142                         count--;
1143                 }
1144         }
1145         RETURN(count);
1146 }
1147
1148 /* If @req is NULL, send CANCEL request to server with handles of locks
1149  * in the @cancels. If EARLY_CANCEL is not supported, send CANCEL requests
1150  * separately per lock.
1151  * If @req is not NULL, put handles of locks in @cancels into the request
1152  * buffer at the offset @off.
1153  * Destroy @cancels at the end. */
1154 int ldlm_cli_cancel_list(struct list_head *cancels, int count,
1155                          struct ptlrpc_request *req, int off, int flags)
1156 {
1157         struct ldlm_lock *lock;
1158         int res = 0;
1159         ENTRY;
1160
1161         if (list_empty(cancels) || count == 0)
1162                 RETURN(0);
1163
1164         /* XXX: requests (both batched and not) could be sent in parallel.
1165          * Usually it is enough to have just 1 RPC, but it is possible that
1166          * there are to many locks to be cancelled in LRU or on a resource.
1167          * It would also speed up the case when the server does not support
1168          * the feature. */
1169         while (count > 0) {
1170                 LASSERT(!list_empty(cancels));
1171                 lock = list_entry(cancels->next, struct ldlm_lock, l_bl_ast);
1172                 LASSERT(lock->l_conn_export);
1173
1174                 if (exp_connect_cancelset(lock->l_conn_export)) {
1175                         res = count;
1176                         if (req)
1177                                 ldlm_cancel_pack(req, off, cancels, count);
1178                         else
1179                                 res = ldlm_cli_cancel_req(lock->l_conn_export,
1180                                                           cancels, count, flags);
1181                 } else {
1182                         res = ldlm_cli_cancel_req(lock->l_conn_export,
1183                                                   cancels, 1, flags);
1184                 }
1185
1186                 if (res < 0) {
1187                         CERROR("ldlm_cli_cancel_list: %d\n", res);
1188                         res = count;
1189                 }
1190
1191                 count -= res;
1192                 ldlm_lock_list_put(cancels, l_bl_ast, res);
1193         }
1194         LASSERT(list_empty(cancels));
1195         LASSERT(count == 0);
1196         RETURN(0);
1197 }
1198
1199 int ldlm_cli_cancel_unused_resource(struct ldlm_namespace *ns,
1200                                     const struct ldlm_res_id *res_id,
1201                                     ldlm_policy_data_t *policy,
1202                                     int mode, int flags, void *opaque)
1203 {
1204         struct ldlm_resource *res;
1205         CFS_LIST_HEAD(cancels);
1206         int count;
1207         int rc;
1208         ENTRY;
1209
1210         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
1211         if (res == NULL) {
1212                 /* This is not a problem. */
1213                 CDEBUG(D_INFO, "No resource "LPU64"\n", res_id->name[0]);
1214                 RETURN(0);
1215         }
1216
1217         count = ldlm_cancel_resource_local(res, &cancels, policy, mode,
1218                                            0, flags, opaque);
1219         rc = ldlm_cli_cancel_list(&cancels, count, NULL,
1220                                   DLM_LOCKREQ_OFF, flags);
1221         if (rc != ELDLM_OK)
1222                 CERROR("ldlm_cli_cancel_unused_resource: %d\n", rc);
1223
1224         ldlm_resource_putref(res);
1225         RETURN(0);
1226 }
1227
1228 static inline int have_no_nsresource(struct ldlm_namespace *ns)
1229 {
1230         int no_resource = 0;
1231
1232         spin_lock(&ns->ns_hash_lock);
1233         if (ns->ns_resources == 0)
1234                 no_resource = 1;
1235         spin_unlock(&ns->ns_hash_lock);
1236
1237         RETURN(no_resource);
1238 }
1239
1240 /* Cancel all locks on a namespace (or a specific resource, if given)
1241  * that have 0 readers/writers.
1242  *
1243  * If flags & LDLM_FL_LOCAL_ONLY, throw the locks away without trying
1244  * to notify the server. */
1245 int ldlm_cli_cancel_unused(struct ldlm_namespace *ns,
1246                            const struct ldlm_res_id *res_id,
1247                            int flags, void *opaque)
1248 {
1249         int i;
1250         ENTRY;
1251
1252         if (ns == NULL)
1253                 RETURN(ELDLM_OK);
1254
1255         if (res_id)
1256                 RETURN(ldlm_cli_cancel_unused_resource(ns, res_id, NULL,
1257                                                        LCK_MINMODE, flags,
1258                                                        opaque));
1259
1260         spin_lock(&ns->ns_hash_lock);
1261         for (i = 0; i < RES_HASH_SIZE; i++) {
1262                 struct list_head *tmp;
1263                 tmp = ns->ns_hash[i].next;
1264                 while (tmp != &(ns->ns_hash[i])) {
1265                         struct ldlm_resource *res;
1266                         int rc;
1267
1268                         res = list_entry(tmp, struct ldlm_resource, lr_hash);
1269                         ldlm_resource_getref(res);
1270                         spin_unlock(&ns->ns_hash_lock);
1271
1272                         rc = ldlm_cli_cancel_unused_resource(ns, &res->lr_name,
1273                                                              NULL, LCK_MINMODE,
1274                                                              flags, opaque);
1275
1276                         if (rc)
1277                                 CERROR("ldlm_cli_cancel_unused ("LPU64"): %d\n",
1278                                        res->lr_name.name[0], rc);
1279
1280                         spin_lock(&ns->ns_hash_lock);
1281                         tmp = tmp->next;
1282                         ldlm_resource_putref_locked(res);
1283                 }
1284         }
1285         spin_unlock(&ns->ns_hash_lock);
1286
1287         RETURN(ELDLM_OK);
1288 }
1289
1290 /* join/split resource locks to/from lru list */
1291 int ldlm_cli_join_lru(struct ldlm_namespace *ns,
1292                       const struct ldlm_res_id *res_id, int join)
1293 {
1294         struct ldlm_resource *res;
1295         struct ldlm_lock *lock, *n;
1296         int count = 0;
1297         ENTRY;
1298
1299         LASSERT(ns->ns_client == LDLM_NAMESPACE_CLIENT);
1300
1301         res = ldlm_resource_get(ns, NULL, res_id, LDLM_EXTENT, 0);
1302         if (res == NULL)
1303                 RETURN(count);
1304         LASSERT(res->lr_type == LDLM_EXTENT);
1305
1306         lock_res(res);
1307         if (!join)
1308                 goto split;
1309
1310         list_for_each_entry_safe (lock, n, &res->lr_granted, l_res_link) {
1311                 if (list_empty(&lock->l_lru) &&
1312                     !lock->l_readers && !lock->l_writers &&
1313                     !(lock->l_flags & LDLM_FL_LOCAL) &&
1314                     !(lock->l_flags & LDLM_FL_CBPENDING)) {
1315                         lock->l_last_used = cfs_time_current();
1316                         spin_lock(&ns->ns_unused_lock);
1317                         LASSERT(ns->ns_nr_unused >= 0);
1318                         list_add_tail(&lock->l_lru, &ns->ns_unused_list);
1319                         ns->ns_nr_unused++;
1320                         spin_unlock(&ns->ns_unused_lock);
1321                         lock->l_flags &= ~LDLM_FL_NO_LRU;
1322                         LDLM_DEBUG(lock, "join lock to lru");
1323                         count++;
1324                 }
1325         }
1326         goto unlock;
1327 split:
1328         spin_lock(&ns->ns_unused_lock);
1329         list_for_each_entry_safe (lock, n, &ns->ns_unused_list, l_lru) {
1330                 if (lock->l_resource == res) {
1331                         ldlm_lock_remove_from_lru_nolock(lock);
1332                         lock->l_flags |= LDLM_FL_NO_LRU;
1333                         LDLM_DEBUG(lock, "split lock from lru");
1334                         count++;
1335                 }
1336         }
1337         spin_unlock(&ns->ns_unused_lock);
1338 unlock:
1339         unlock_res(res);
1340         ldlm_resource_putref(res);
1341         RETURN(count);
1342 }
1343
1344 /* Lock iterators. */
1345
1346 int ldlm_resource_foreach(struct ldlm_resource *res, ldlm_iterator_t iter,
1347                           void *closure)
1348 {
1349         struct list_head *tmp, *next;
1350         struct ldlm_lock *lock;
1351         int rc = LDLM_ITER_CONTINUE;
1352
1353         ENTRY;
1354
1355         if (!res)
1356                 RETURN(LDLM_ITER_CONTINUE);
1357
1358         lock_res(res);
1359         list_for_each_safe(tmp, next, &res->lr_granted) {
1360                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
1361
1362                 if (iter(lock, closure) == LDLM_ITER_STOP)
1363                         GOTO(out, rc = LDLM_ITER_STOP);
1364         }
1365
1366         list_for_each_safe(tmp, next, &res->lr_converting) {
1367                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
1368
1369                 if (iter(lock, closure) == LDLM_ITER_STOP)
1370                         GOTO(out, rc = LDLM_ITER_STOP);
1371         }
1372
1373         list_for_each_safe(tmp, next, &res->lr_waiting) {
1374                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
1375
1376                 if (iter(lock, closure) == LDLM_ITER_STOP)
1377                         GOTO(out, rc = LDLM_ITER_STOP);
1378         }
1379  out:
1380         unlock_res(res);
1381         RETURN(rc);
1382 }
1383
1384 struct iter_helper_data {
1385         ldlm_iterator_t iter;
1386         void *closure;
1387 };
1388
1389 static int ldlm_iter_helper(struct ldlm_lock *lock, void *closure)
1390 {
1391         struct iter_helper_data *helper = closure;
1392         return helper->iter(lock, helper->closure);
1393 }
1394
1395 static int ldlm_res_iter_helper(struct ldlm_resource *res, void *closure)
1396 {
1397         return ldlm_resource_foreach(res, ldlm_iter_helper, closure);
1398 }
1399
1400 int ldlm_namespace_foreach(struct ldlm_namespace *ns, ldlm_iterator_t iter,
1401                            void *closure)
1402 {
1403         struct iter_helper_data helper = { iter: iter, closure: closure };
1404         return ldlm_namespace_foreach_res(ns, ldlm_res_iter_helper, &helper);
1405 }
1406
1407 int ldlm_namespace_foreach_res(struct ldlm_namespace *ns,
1408                                ldlm_res_iterator_t iter, void *closure)
1409 {
1410         int i, rc = LDLM_ITER_CONTINUE;
1411         struct ldlm_resource *res;
1412         struct list_head *tmp;
1413
1414         ENTRY;
1415         spin_lock(&ns->ns_hash_lock);
1416         for (i = 0; i < RES_HASH_SIZE; i++) {
1417                 tmp = ns->ns_hash[i].next;
1418                 while (tmp != &(ns->ns_hash[i])) {
1419                         res = list_entry(tmp, struct ldlm_resource, lr_hash);
1420                         ldlm_resource_getref(res);
1421                         spin_unlock(&ns->ns_hash_lock);
1422
1423                         rc = iter(res, closure);
1424
1425                         spin_lock(&ns->ns_hash_lock);
1426                         tmp = tmp->next;
1427                         ldlm_resource_putref_locked(res);
1428                         if (rc == LDLM_ITER_STOP)
1429                                 GOTO(out, rc);
1430                 }
1431         }
1432  out:
1433         spin_unlock(&ns->ns_hash_lock);
1434         RETURN(rc);
1435 }
1436
1437 /* non-blocking function to manipulate a lock whose cb_data is being put away.*/
1438 void ldlm_resource_iterate(struct ldlm_namespace *ns,
1439                            const struct ldlm_res_id *res_id,
1440                            ldlm_iterator_t iter, void *data)
1441 {
1442         struct ldlm_resource *res;
1443         ENTRY;
1444
1445         if (ns == NULL) {
1446                 CERROR("must pass in namespace\n");
1447                 LBUG();
1448         }
1449
1450         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
1451         if (res == NULL) {
1452                 EXIT;
1453                 return;
1454         }
1455
1456         ldlm_resource_foreach(res, iter, data);
1457         ldlm_resource_putref(res);
1458         EXIT;
1459 }
1460
1461 /* Lock replay */
1462
1463 static int ldlm_chain_lock_for_replay(struct ldlm_lock *lock, void *closure)
1464 {
1465         struct list_head *list = closure;
1466
1467         /* we use l_pending_chain here, because it's unused on clients. */
1468         LASSERTF(list_empty(&lock->l_pending_chain),"lock %p next %p prev %p\n",
1469                  lock, &lock->l_pending_chain.next,&lock->l_pending_chain.prev);
1470         /* bug 9573: don't replay locks left after eviction */
1471         if (!(lock->l_flags & LDLM_FL_FAILED))
1472                 list_add(&lock->l_pending_chain, list);
1473         return LDLM_ITER_CONTINUE;
1474 }
1475
1476 static int replay_lock_interpret(struct ptlrpc_request *req,
1477                                     void * data, int rc)
1478 {
1479         struct ldlm_lock *lock;
1480         struct ldlm_reply *reply;
1481
1482         ENTRY;
1483         atomic_dec(&req->rq_import->imp_replay_inflight);
1484         if (rc != ELDLM_OK)
1485                 GOTO(out, rc);
1486
1487         lock = req->rq_async_args.pointer_arg[0];
1488         LASSERT(lock != NULL);
1489
1490         reply = lustre_swab_repbuf(req, DLM_LOCKREPLY_OFF, sizeof(*reply),
1491                                    lustre_swab_ldlm_reply);
1492         if (reply == NULL) {
1493                 CERROR("Can't unpack ldlm_reply\n");
1494                 GOTO (out, rc = -EPROTO);
1495         }
1496
1497         lock->l_remote_handle = reply->lock_handle;
1498         LDLM_DEBUG(lock, "replayed lock:");
1499         ptlrpc_import_recovery_state_machine(req->rq_import);
1500  out:
1501         if (rc != ELDLM_OK)
1502                 ptlrpc_connect_import(req->rq_import, NULL);
1503
1504
1505         RETURN(rc);
1506 }
1507
1508 static int replay_one_lock(struct obd_import *imp, struct ldlm_lock *lock)
1509 {
1510         struct ptlrpc_request *req;
1511         struct ldlm_request *body;
1512         struct ldlm_reply *reply;
1513         int buffers = 2;
1514         int size[3] = { sizeof(struct ptlrpc_body) };
1515         int flags;
1516         ENTRY;
1517
1518         /* If this is reply-less callback lock, we cannot replay it, since
1519          * server might have long dropped it, but notification of that event was
1520          * lost by network. (and server granted conflicting lock already) */
1521         if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) {
1522                 LDLM_DEBUG(lock, "Not replaying reply-less lock:");
1523                 ldlm_lock_cancel(lock);
1524                 RETURN(0);
1525         }
1526         /*
1527          * If granted mode matches the requested mode, this lock is granted.
1528          *
1529          * If they differ, but we have a granted mode, then we were granted
1530          * one mode and now want another: ergo, converting.
1531          *
1532          * If we haven't been granted anything and are on a resource list,
1533          * then we're blocked/waiting.
1534          *
1535          * If we haven't been granted anything and we're NOT on a resource list,
1536          * then we haven't got a reply yet and don't have a known disposition.
1537          * This happens whenever a lock enqueue is the request that triggers
1538          * recovery.
1539          */
1540         if (lock->l_granted_mode == lock->l_req_mode)
1541                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_GRANTED;
1542         else if (lock->l_granted_mode)
1543                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_CONV;
1544         else if (!list_empty(&lock->l_res_link))
1545                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_WAIT;
1546         else
1547                 flags = LDLM_FL_REPLAY;
1548
1549         size[DLM_LOCKREQ_OFF] = sizeof(*body);
1550         req = ptlrpc_prep_req(imp, LUSTRE_DLM_VERSION, LDLM_ENQUEUE, 2, size,
1551                               NULL);
1552         if (!req)
1553                 RETURN(-ENOMEM);
1554
1555         /* We're part of recovery, so don't wait for it. */
1556         req->rq_send_state = LUSTRE_IMP_REPLAY_LOCKS;
1557
1558         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
1559         ldlm_lock2desc(lock, &body->lock_desc);
1560         body->lock_flags = flags;
1561
1562         ldlm_lock2handle(lock, &body->lock_handle[0]);
1563         size[DLM_LOCKREPLY_OFF] = sizeof(*reply);
1564         if (lock->l_lvb_len != 0) {
1565                 buffers = 3;
1566                 size[DLM_REPLY_REC_OFF] = lock->l_lvb_len;
1567         }
1568         ptlrpc_req_set_repsize(req, buffers, size);
1569         /* notify the server we've replayed all requests.
1570          * also, we mark the request to be put on a dedicated
1571          * queue to be processed after all request replayes.
1572          * bug 6063 */
1573         lustre_msg_set_flags(req->rq_reqmsg, MSG_REQ_REPLAY_DONE);
1574
1575         LDLM_DEBUG(lock, "replaying lock:");
1576
1577         atomic_inc(&req->rq_import->imp_replay_inflight);
1578         req->rq_async_args.pointer_arg[0] = lock;
1579         req->rq_interpret_reply = replay_lock_interpret;
1580         ptlrpcd_add_req(req);
1581
1582         RETURN(0);
1583 }
1584
1585 int ldlm_replay_locks(struct obd_import *imp)
1586 {
1587         struct ldlm_namespace *ns = imp->imp_obd->obd_namespace;
1588         struct list_head list;
1589         struct ldlm_lock *lock, *next;
1590         int rc = 0;
1591
1592         ENTRY;
1593         CFS_INIT_LIST_HEAD(&list);
1594
1595         LASSERT(atomic_read(&imp->imp_replay_inflight) == 0);
1596
1597         /* ensure this doesn't fall to 0 before all have been queued */
1598         atomic_inc(&imp->imp_replay_inflight);
1599
1600         (void)ldlm_namespace_foreach(ns, ldlm_chain_lock_for_replay, &list);
1601
1602         list_for_each_entry_safe(lock, next, &list, l_pending_chain) {
1603                 list_del_init(&lock->l_pending_chain);
1604                 if (rc)
1605                         continue; /* or try to do the rest? */
1606                 rc = replay_one_lock(imp, lock);
1607         }
1608
1609         atomic_dec(&imp->imp_replay_inflight);
1610
1611         RETURN(rc);
1612 }