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 Lustre, http://www.lustre.org.
   7  *
   8  *   Lustre is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Lustre is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Lustre; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22 #define DEBUG_SUBSYSTEM S_LDLM
  23
  24 #include <linux/lustre_dlm.h>
  25 #include <linux/obd_class.h>
  26 #include <linux/obd.h>
  27
  28 static int interrupted_completion_wait(void *data)
  29 {
  30         RETURN(1);
  31 }
  32
  33 int ldlm_expired_completion_wait(void *data)
  34 {
  35         struct ldlm_lock *lock = data;
  36         struct ptlrpc_connection *conn;
  37         struct obd_device *obd;
  38
  39         if (!lock)
  40                 CERROR("NULL lock\n");
  41         else if (!lock->l_connh)
  42                 CERROR("lock %p has NULL connh\n", lock);
  43         else if (!(obd = class_conn2obd(lock->l_connh)))
  44                 CERROR("lock %p has NULL obd\n", lock);
  45         else if (!(conn = obd->u.cli.cl_import.imp_connection))
  46                 CERROR("lock %p has NULL connection\n", lock);
  47         else {
  48                 LDLM_DEBUG0(lock, "timed out waiting for completion");
  49                 CERROR("lock %p timed out from %s\n", lock,
  50                        conn->c_remote_uuid.uuid);
  51                 ldlm_lock_dump(D_ERROR, lock);
  52                 class_signal_connection_failure(conn);
  53         }
  54         RETURN(0);
  55 }
  56
  57 int ldlm_completion_ast(struct ldlm_lock *lock, int flags, void *data)
  58 {
  59         struct l_wait_info lwi =
  60                 LWI_TIMEOUT_INTR(obd_timeout * HZ, ldlm_expired_completion_wait,
  61                                  interrupted_completion_wait, lock);
  62         int rc = 0;
  63         ENTRY;
  64
  65         if (flags == LDLM_FL_WAIT_NOREPROC)
  66                 goto noreproc;
  67
  68         if (flags == 0) {
  69                 wake_up(&lock->l_waitq);
  70                 RETURN(0);
  71         }
  72
  73         if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
  74                        LDLM_FL_BLOCK_CONV)))
  75                 RETURN(0);
  76
  77         LDLM_DEBUG0(lock, "client-side enqueue returned a blocked lock, "
  78                     "sleeping");
  79         ldlm_lock_dump(D_OTHER, lock);
  80         ldlm_reprocess_all(lock->l_resource);
  81
  82  noreproc:
  83         /* Go to sleep until the lock is granted or cancelled. */
  84         rc = l_wait_event(lock->l_waitq,
  85                           ((lock->l_req_mode == lock->l_granted_mode) ||
  86                            lock->l_destroyed), &lwi);
  87
  88         if (lock->l_destroyed) {
  89                 LDLM_DEBUG0(lock, "client-side enqueue waking up: destroyed");
  90                 RETURN(-EIO);
  91         }
  92
  93         if (rc) {
  94                 LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)",
  95                            rc);
  96                 RETURN(rc);
  97         }
  98
  99         LDLM_DEBUG0(lock, "client-side enqueue waking up: granted");
 100         RETURN(0);
 101 }
 102
 103 static int ldlm_cli_enqueue_local(struct ldlm_namespace *ns,
 104                                   struct lustre_handle *parent_lockh,
 105                                   struct ldlm_res_id res_id,
 106                                   __u32 type,
 107                                   void *cookie, int cookielen,
 108                                   ldlm_mode_t mode,
 109                                   int *flags,
 110                                   ldlm_completion_callback completion,
 111                                   ldlm_blocking_callback blocking,
 112                                   void *data,
 113                                   void *cp_data,
 114                                   struct lustre_handle *lockh)
 115 {
 116         struct ldlm_lock *lock;
 117         int err;
 118         ENTRY;
 119
 120         if (ns->ns_client) {
 121                 CERROR("Trying to enqueue local lock in a shadow namespace\n");
 122                 LBUG();
 123         }
 124
 125         lock = ldlm_lock_create(ns, parent_lockh, res_id, type, mode,
 126                                 data, cp_data);
 127         if (!lock)
 128                 GOTO(out_nolock, err = -ENOMEM);
 129         LDLM_DEBUG0(lock,
 130                     "client-side local enqueue handler, new lock created");
 131
 132         ldlm_lock_addref_internal(lock, mode);
 133         ldlm_lock2handle(lock, lockh);
 134         lock->l_flags |= LDLM_FL_LOCAL;
 135
 136         err = ldlm_lock_enqueue(ns, &lock, cookie, cookielen, flags, completion,
 137                                 blocking);
 138         if (err != ELDLM_OK)
 139                 GOTO(out, err);
 140
 141         if (type == LDLM_EXTENT)
 142                 memcpy(cookie, &lock->l_extent, sizeof(lock->l_extent));
 143         if ((*flags) & LDLM_FL_LOCK_CHANGED)
 144                 memcpy(&res_id, &lock->l_resource->lr_name, sizeof(res_id));
 145
 146         LDLM_DEBUG_NOLOCK("client-side local enqueue handler END (lock %p)",
 147                           lock);
 148
 149         if (lock->l_completion_ast)
 150                 lock->l_completion_ast(lock, *flags, NULL);
 151
 152         LDLM_DEBUG0(lock, "client-side local enqueue END");
 153         EXIT;
 154  out:
 155         LDLM_LOCK_PUT(lock);
 156  out_nolock:
 157         return err;
 158 }
 159
 160 int ldlm_cli_enqueue(struct lustre_handle *connh,
 161                      struct ptlrpc_request *req,
 162                      struct ldlm_namespace *ns,
 163                      struct lustre_handle *parent_lock_handle,
 164                      struct ldlm_res_id res_id,
 165                      __u32 type,
 166                      void *cookie, int cookielen,
 167                      ldlm_mode_t mode,
 168                      int *flags,
 169                      ldlm_completion_callback completion,
 170                      ldlm_blocking_callback blocking,
 171                      void *data,
 172                      void *cp_data,
 173                      struct lustre_handle *lockh)
 174 {
 175         struct ldlm_lock *lock;
 176         struct ldlm_request *body;
 177         struct ldlm_reply *reply;
 178         int rc, size = sizeof(*body), req_passed_in = 1, is_replay;
 179         ENTRY;
 180
 181         is_replay = *flags & LDLM_FL_REPLAY;
 182         LASSERT(connh != NULL || !is_replay);
 183
 184         if (connh == NULL) {
 185                 rc = ldlm_cli_enqueue_local(ns, parent_lock_handle, res_id,
 186                                             type, cookie, cookielen, mode,
 187                                             flags, completion, blocking, data,
 188                                             cp_data, lockh);
 189                 RETURN(rc);
 190         }
 191
 192         /* If we're replaying this lock, just check some invariants.
 193          * If we're creating a new lock, get everything all setup nice. */
 194         if (is_replay) {
 195                 lock = ldlm_handle2lock(lockh);
 196                 LDLM_DEBUG0(lock, "client-side enqueue START");
 197                 LASSERT(connh == lock->l_connh);
 198         } else {
 199                 lock = ldlm_lock_create(ns, parent_lock_handle, res_id, type,
 200                                         mode, data, cp_data);
 201                 if (lock == NULL)
 202                         GOTO(out_nolock, rc = -ENOMEM);
 203                 /* ugh.  I set this early (instead of waiting for _enqueue)
 204                  * because the completion AST might arrive early, and we need
 205                  * (in just this one case) to run the completion_cb even if it
 206                  * arrives before the reply. */
 207                 lock->l_completion_ast = completion;
 208                 LDLM_DEBUG0(lock, "client-side enqueue START");
 209                 /* for the local lock, add the reference */
 210                 ldlm_lock_addref_internal(lock, mode);
 211                 ldlm_lock2handle(lock, lockh);
 212                 if (type == LDLM_EXTENT)
 213                         memcpy(&lock->l_extent, cookie,
 214                                sizeof(body->lock_desc.l_extent));
 215         }
 216
 217         if (req == NULL) {
 218                 req = ptlrpc_prep_req(class_conn2cliimp(connh), LDLM_ENQUEUE, 1,
 219                                       &size, NULL);
 220                 if (!req)
 221                         GOTO(out, rc = -ENOMEM);
 222                 req_passed_in = 0;
 223         } else if (req->rq_reqmsg->buflens[0] != sizeof(*body))
 224                 LBUG();
 225
 226         /* Dump lock data into the request buffer */
 227         body = lustre_msg_buf(req->rq_reqmsg, 0);
 228         ldlm_lock2desc(lock, &body->lock_desc);
 229         body->lock_flags = *flags;
 230
 231         memcpy(&body->lock_handle1, lockh, sizeof(*lockh));
 232         if (parent_lock_handle)
 233                 memcpy(&body->lock_handle2, parent_lock_handle,
 234                        sizeof(body->lock_handle2));
 235
 236         /* Continue as normal. */
 237         if (!req_passed_in) {
 238                 size = sizeof(*reply);
 239                 req->rq_replen = lustre_msg_size(1, &size);
 240         }
 241         lock->l_connh = connh;
 242         lock->l_export = NULL;
 243
 244         LDLM_DEBUG0(lock, "sending request");
 245         rc = ptlrpc_queue_wait(req);
 246
 247         if (rc != ELDLM_OK) {
 248                 LASSERT(!is_replay);
 249                 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
 250                            rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
 251                 /* Set a flag to prevent us from sending a CANCEL (bug 407) */
 252                 l_lock(&ns->ns_lock);
 253                 lock->l_flags |= LDLM_FL_CANCELING;
 254                 l_unlock(&ns->ns_lock);
 255
 256                 ldlm_lock_decref(lockh, mode);
 257                 ldlm_lock_destroy(lock);
 258                 GOTO(out_req, rc);
 259         }
 260
 261         reply = lustre_msg_buf(req->rq_repmsg, 0);
 262         memcpy(&lock->l_remote_handle, &reply->lock_handle,
 263                sizeof(lock->l_remote_handle));
 264         *flags = reply->lock_flags;
 265
 266         CDEBUG(D_INFO, "local: %p, remote: %p, flags: %d\n", lock,
 267                (void *)(unsigned long)reply->lock_handle.addr, *flags);
 268         if (type == LDLM_EXTENT) {
 269                 CDEBUG(D_INFO, "requested extent: "LPU64" -> "LPU64", got "
 270                        "extent "LPU64" -> "LPU64"\n",
 271                        body->lock_desc.l_extent.start,
 272                        body->lock_desc.l_extent.end,
 273                        reply->lock_extent.start, reply->lock_extent.end);
 274                 cookie = &reply->lock_extent; /* FIXME bug 267 */
 275                 cookielen = sizeof(reply->lock_extent);
 276         }
 277
 278         /* If enqueue returned a blocked lock but the completion handler has
 279          * already run, then it fixed up the resource and we don't need to do it
 280          * again. */
 281         if ((*flags) & LDLM_FL_LOCK_CHANGED) {
 282                 int newmode = reply->lock_mode;
 283                 LASSERT(!is_replay);
 284                 if (newmode && newmode != lock->l_req_mode) {
 285                         LDLM_DEBUG(lock, "server returned different mode %s",
 286                                    ldlm_lockname[newmode]);
 287                         lock->l_req_mode = newmode;
 288                 }
 289
 290                 if (reply->lock_resource_name.name[0] !=
 291                     lock->l_resource->lr_name.name[0]) {
 292                         CDEBUG(D_INFO, "remote intent success, locking %ld "
 293                                "instead of %ld\n",
 294                                (long)reply->lock_resource_name.name[0],
 295                                (long)lock->l_resource->lr_name.name[0]);
 296
 297                         ldlm_lock_change_resource(ns, lock,
 298                                                   reply->lock_resource_name);
 299                         if (lock->l_resource == NULL) {
 300                                 LBUG();
 301                                 GOTO(out_req, rc = -ENOMEM);
 302                         }
 303                         LDLM_DEBUG0(lock, "client-side enqueue, new resource");
 304                 }
 305         }
 306
 307         if (!is_replay) {
 308                 l_lock(&ns->ns_lock);
 309                 lock->l_completion_ast = NULL;
 310                 rc = ldlm_lock_enqueue(ns, &lock, cookie, cookielen, flags,
 311                                        completion, blocking);
 312                 l_unlock(&ns->ns_lock);
 313                 if (lock->l_completion_ast)
 314                         lock->l_completion_ast(lock, *flags, NULL);
 315         }
 316
 317         LDLM_DEBUG0(lock, "client-side enqueue END");
 318         EXIT;
 319  out_req:
 320         if (!req_passed_in)
 321                 ptlrpc_req_finished(req);
 322  out:
 323         LDLM_LOCK_PUT(lock);
 324  out_nolock:
 325         return rc;
 326 }
 327
 328 int ldlm_match_or_enqueue(struct lustre_handle *connh,
 329                           struct ptlrpc_request *req,
 330                           struct ldlm_namespace *ns,
 331                           struct lustre_handle *parent_lock_handle,
 332                           struct ldlm_res_id res_id,
 333                           __u32 type,
 334                           void *cookie, int cookielen,
 335                           ldlm_mode_t mode,
 336                           int *flags,
 337                           ldlm_completion_callback completion,
 338                           ldlm_blocking_callback blocking,
 339                           void *data,
 340                           void *cp_data,
 341                           struct lustre_handle *lockh)
 342 {
 343         int rc;
 344         ENTRY;
 345         if (connh == NULL) {
 346                 /* Just to make sure that I understand things --phil */
 347                 LASSERT(*flags & LDLM_FL_LOCAL_ONLY);
 348         }
 349
 350         LDLM_DEBUG_NOLOCK("resource "LPU64"/"LPU64, res_id.name[0],
 351                           res_id.name[1]);
 352         rc = ldlm_lock_match(ns, *flags, &res_id, type, cookie, cookielen, mode,
 353                              lockh);
 354         if (rc == 0) {
 355                 rc = ldlm_cli_enqueue(connh, req, ns, parent_lock_handle,
 356                                       res_id, type, cookie, cookielen, mode,
 357                                       flags, completion, blocking, data,
 358                                       cp_data, lockh);
 359                 if (rc != ELDLM_OK)
 360                         CERROR("ldlm_cli_enqueue: err: %d\n", rc);
 361                 RETURN(rc);
 362         }
 363         RETURN(0);
 364 }
 365
 366 int ldlm_cli_replay_enqueue(struct ldlm_lock *lock)
 367 {
 368         struct lustre_handle lockh;
 369         struct ldlm_res_id junk;
 370         int flags = LDLM_FL_REPLAY;
 371         ldlm_lock2handle(lock, &lockh);
 372         return ldlm_cli_enqueue(lock->l_connh, NULL, NULL, NULL, junk,
 373                                 lock->l_resource->lr_type, NULL, 0, -1, &flags,
 374                                 NULL, NULL, NULL, 0, &lockh);
 375 }
 376
 377 static int ldlm_cli_convert_local(struct ldlm_lock *lock, int new_mode,
 378                                   int *flags)
 379 {
 380         ENTRY;
 381         if (lock->l_resource->lr_namespace->ns_client) {
 382                 CERROR("Trying to cancel local lock\n");
 383                 LBUG();
 384         }
 385         LDLM_DEBUG0(lock, "client-side local convert");
 386
 387         ldlm_lock_convert(lock, new_mode, flags);
 388         ldlm_reprocess_all(lock->l_resource);
 389
 390         LDLM_DEBUG0(lock, "client-side local convert handler END");
 391         LDLM_LOCK_PUT(lock);
 392         RETURN(0);
 393 }
 394
 395 /* FIXME: one of ldlm_cli_convert or the server side should reject attempted
 396  * conversion of locks which are on the waiting or converting queue */
 397 int ldlm_cli_convert(struct lustre_handle *lockh, int new_mode, int *flags)
 398 {
 399         struct ldlm_request *body;
 400         struct lustre_handle *connh;
 401         struct ldlm_reply *reply;
 402         struct ldlm_lock *lock;
 403         struct ldlm_resource *res;
 404         struct ptlrpc_request *req;
 405         int rc, size = sizeof(*body);
 406         ENTRY;
 407
 408         lock = ldlm_handle2lock(lockh);
 409         if (!lock) {
 410                 LBUG();
 411                 RETURN(-EINVAL);
 412         }
 413         *flags = 0;
 414         connh = lock->l_connh;
 415
 416         if (!connh)
 417                 RETURN(ldlm_cli_convert_local(lock, new_mode, flags));
 418
 419         LDLM_DEBUG0(lock, "client-side convert");
 420
 421         req = ptlrpc_prep_req(class_conn2cliimp(connh), LDLM_CONVERT, 1, &size,
 422                               NULL);
 423         if (!req)
 424                 GOTO(out, rc = -ENOMEM);
 425
 426         body = lustre_msg_buf(req->rq_reqmsg, 0);
 427         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 428                sizeof(body->lock_handle1));
 429
 430         body->lock_desc.l_req_mode = new_mode;
 431         body->lock_flags = *flags;
 432
 433         size = sizeof(*reply);
 434         req->rq_replen = lustre_msg_size(1, &size);
 435
 436         rc = ptlrpc_queue_wait(req);
 437         if (rc != ELDLM_OK)
 438                 GOTO(out, rc);
 439
 440         reply = lustre_msg_buf(req->rq_repmsg, 0);
 441         res = ldlm_lock_convert(lock, new_mode, &reply->lock_flags);
 442         if (res != NULL)
 443                 ldlm_reprocess_all(res);
 444         /* Go to sleep until the lock is granted. */
 445         /* FIXME: or cancelled. */
 446         if (lock->l_completion_ast)
 447                 lock->l_completion_ast(lock, LDLM_FL_WAIT_NOREPROC, NULL);
 448         EXIT;
 449  out:
 450         LDLM_LOCK_PUT(lock);
 451         ptlrpc_req_finished(req);
 452         return rc;
 453 }
 454
 455 int ldlm_cli_cancel(struct lustre_handle *lockh)
 456 {
 457         struct ptlrpc_request *req;
 458         struct ldlm_lock *lock;
 459         struct ldlm_request *body;
 460         int rc = 0, size = sizeof(*body);
 461         ENTRY;
 462
 463         /* concurrent cancels on the same handle can happen */
 464         lock = __ldlm_handle2lock(lockh, LDLM_FL_CANCELING);
 465         if (lock == NULL)
 466                 RETURN(0);
 467
 468         if (lock->l_connh) {
 469                 int local_only;
 470
 471                 LDLM_DEBUG0(lock, "client-side cancel");
 472                 /* Set this flag to prevent others from getting new references*/
 473                 l_lock(&lock->l_resource->lr_namespace->ns_lock);
 474                 lock->l_flags |= LDLM_FL_CBPENDING;
 475                 ldlm_cancel_callback(lock);
 476                 local_only = (lock->l_flags & LDLM_FL_LOCAL_ONLY);
 477                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 478
 479                 if (local_only) {
 480                         CDEBUG(D_INFO, "not sending request (at caller's "
 481                                "instruction\n");
 482                         goto local_cancel;
 483                 }
 484
 485                 req = ptlrpc_prep_req(class_conn2cliimp(lock->l_connh),
 486                                       LDLM_CANCEL, 1, &size, NULL);
 487                 if (!req)
 488                         GOTO(out, rc = -ENOMEM);
 489
 490                 /* XXX FIXME bug 249 */
 491                 req->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
 492                 req->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
 493
 494                 body = lustre_msg_buf(req->rq_reqmsg, 0);
 495                 memcpy(&body->lock_handle1, &lock->l_remote_handle,
 496                        sizeof(body->lock_handle1));
 497
 498                 req->rq_replen = lustre_msg_size(0, NULL);
 499
 500                 rc = ptlrpc_queue_wait(req);
 501                 ptlrpc_req_finished(req);
 502                 if (rc == ESTALE) {
 503                         CERROR("client/server out of sync\n");
 504                         LBUG();
 505                 }
 506                 if (rc != ELDLM_OK)
 507                         CERROR("Got rc %d from cancel RPC: canceling "
 508                                "anyway\n", rc);
 509         local_cancel:
 510                 ldlm_lock_cancel(lock);
 511         } else {
 512                 LDLM_DEBUG0(lock, "client-side local cancel");
 513                 if (lock->l_resource->lr_namespace->ns_client) {
 514                         CERROR("Trying to cancel local lock\n");
 515                         LBUG();
 516                 }
 517                 ldlm_lock_cancel(lock);
 518                 ldlm_reprocess_all(lock->l_resource);
 519                 LDLM_DEBUG0(lock, "client-side local cancel handler END");
 520         }
 521
 522         EXIT;
 523  out:
 524         LDLM_LOCK_PUT(lock);
 525         return rc;
 526 }
 527
 528 int ldlm_cancel_lru(struct ldlm_namespace *ns)
 529 {
 530         struct list_head *tmp, *next, list = LIST_HEAD_INIT(list);
 531         int count, rc = 0;
 532         struct ldlm_ast_work *w;
 533         ENTRY;
 534
 535         l_lock(&ns->ns_lock);
 536         count = ns->ns_nr_unused - ns->ns_max_unused;
 537
 538         if (count <= 0) {
 539                 l_unlock(&ns->ns_lock);
 540                 RETURN(0);
 541         }
 542
 543         list_for_each_safe(tmp, next, &ns->ns_unused_list) {
 544                 struct ldlm_lock *lock;
 545                 lock = list_entry(tmp, struct ldlm_lock, l_lru);
 546
 547                 LASSERT(!lock->l_readers && !lock->l_writers);
 548
 549                 /* Setting the CBPENDING flag is a little misleading, but
 550                  * prevents an important race; namely, once CBPENDING is set,
 551                  * the lock can accumulate no more readers/writers.  Since
 552                  * readers and writers are already zero here, ldlm_lock_decref
 553                  * won't see this flag and call l_blocking_ast */
 554                 lock->l_flags |= LDLM_FL_CBPENDING;
 555
 556                 OBD_ALLOC(w, sizeof(*w));
 557                 LASSERT(w);
 558
 559                 w->w_lock = LDLM_LOCK_GET(lock);
 560                 list_add(&w->w_list, &list);
 561                 ldlm_lock_remove_from_lru(lock);
 562
 563                 if (--count == 0)
 564                         break;
 565         }
 566         l_unlock(&ns->ns_lock);
 567
 568         list_for_each_safe(tmp, next, &list) {
 569                 struct lustre_handle lockh;
 570                 int rc;
 571                 w = list_entry(tmp, struct ldlm_ast_work, w_list);
 572
 573                 ldlm_lock2handle(w->w_lock, &lockh);
 574                 rc = ldlm_cli_cancel(&lockh);
 575                 if (rc != ELDLM_OK)
 576                         CDEBUG(D_INFO, "ldlm_cli_cancel: %d\n", rc);
 577
 578                 list_del(&w->w_list);
 579                 LDLM_LOCK_PUT(w->w_lock);
 580                 OBD_FREE(w, sizeof(*w));
 581         }
 582
 583         RETURN(rc);
 584 }
 585
 586 int ldlm_cli_cancel_unused_resource(struct ldlm_namespace *ns,
 587                                     struct ldlm_res_id res_id, int flags)
 588 {
 589         struct ldlm_resource *res;
 590         struct list_head *tmp, *next, list = LIST_HEAD_INIT(list);
 591         struct ldlm_ast_work *w;
 592         ENTRY;
 593
 594         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
 595         if (res == NULL) {
 596                 /* This is not a problem. */
 597                 CDEBUG(D_INFO, "No resource "LPU64"\n", res_id.name[0]);
 598                 RETURN(0);
 599         }
 600
 601         l_lock(&ns->ns_lock);
 602         list_for_each(tmp, &res->lr_granted) {
 603                 struct ldlm_lock *lock;
 604                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 605
 606                 if (lock->l_readers || lock->l_writers)
 607                         continue;
 608
 609                 /* See CBPENDING comment in ldlm_cancel_lru */
 610                 lock->l_flags |= LDLM_FL_CBPENDING;
 611
 612                 OBD_ALLOC(w, sizeof(*w));
 613                 LASSERT(w);
 614
 615                 w->w_lock = LDLM_LOCK_GET(lock);
 616                 list_add(&w->w_list, &list);
 617         }
 618         l_unlock(&ns->ns_lock);
 619
 620         list_for_each_safe(tmp, next, &list) {
 621                 struct lustre_handle lockh;
 622                 int rc;
 623                 w = list_entry(tmp, struct ldlm_ast_work, w_list);
 624
 625                 /* Prevent the cancel callback from being called by setting
 626                  * LDLM_FL_CANCEL in the lock.  Very sneaky. -p */
 627                 if (flags & LDLM_FL_NO_CALLBACK)
 628                         w->w_lock->l_flags |= LDLM_FL_CANCEL;
 629
 630                 if (flags & LDLM_FL_LOCAL_ONLY) {
 631                         ldlm_lock_cancel(w->w_lock);
 632                 } else {
 633                         ldlm_lock2handle(w->w_lock, &lockh);
 634                         rc = ldlm_cli_cancel(&lockh);
 635                         if (rc != ELDLM_OK)
 636                                 CERROR("ldlm_cli_cancel: %d\n", rc);
 637                 }
 638                 list_del(&w->w_list);
 639                 LDLM_LOCK_PUT(w->w_lock);
 640                 OBD_FREE(w, sizeof(*w));
 641         }
 642
 643         ldlm_resource_putref(res);
 644
 645         RETURN(0);
 646 }
 647
 648 /* Cancel all locks on a namespace (or a specific resource, if given) that have
 649  * 0 readers/writers.
 650  *
 651  * If 'local_only' is true, throw the locks away without trying to notify the
 652  * server. */
 653 int ldlm_cli_cancel_unused(struct ldlm_namespace *ns,
 654                            struct ldlm_res_id *res_id, int flags)
 655 {
 656         int i;
 657         ENTRY;
 658
 659         if (ns == NULL)
 660                 RETURN(ELDLM_OK);
 661
 662         if (res_id)
 663                 RETURN(ldlm_cli_cancel_unused_resource(ns, *res_id, flags));
 664
 665         l_lock(&ns->ns_lock);
 666         for (i = 0; i < RES_HASH_SIZE; i++) {
 667                 struct list_head *tmp, *pos;
 668                 list_for_each_safe(tmp, pos, &(ns->ns_hash[i])) {
 669                         int rc;
 670                         struct ldlm_resource *res;
 671                         res = list_entry(tmp, struct ldlm_resource, lr_hash);
 672                         ldlm_resource_getref(res);
 673
 674                         rc = ldlm_cli_cancel_unused_resource(ns, res->lr_name,
 675                                                              flags);
 676
 677                         if (rc)
 678                                 CERROR("cancel_unused_res ("LPU64"): %d\n",
 679                                        res->lr_name.name[0], rc);
 680                         ldlm_resource_putref(res);
 681                 }
 682         }
 683         l_unlock(&ns->ns_lock);
 684
 685         RETURN(ELDLM_OK);
 686 }
 687
 688 /* Lock iterators. */
 689
 690 int ldlm_resource_foreach(struct ldlm_resource *res, ldlm_iterator_t iter,
 691                           void *closure)
 692 {
 693         struct list_head *tmp, *next;
 694         struct ldlm_lock *lock;
 695         int rc = LDLM_ITER_CONTINUE;
 696         struct ldlm_namespace *ns = res->lr_namespace;
 697
 698         ENTRY;
 699
 700         if (!res)
 701                 RETURN(LDLM_ITER_CONTINUE);
 702
 703         l_lock(&ns->ns_lock);
 704         list_for_each_safe(tmp, next, &res->lr_granted) {
 705                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 706
 707                 if (iter(lock, closure) == LDLM_ITER_STOP)
 708                         GOTO(out, rc = LDLM_ITER_STOP);
 709         }
 710
 711         list_for_each_safe(tmp, next, &res->lr_converting) {
 712                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 713
 714                 if (iter(lock, closure) == LDLM_ITER_STOP)
 715                         GOTO(out, rc = LDLM_ITER_STOP);
 716         }
 717
 718         list_for_each_safe(tmp, next, &res->lr_waiting) {
 719                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 720
 721                 if (iter(lock, closure) == LDLM_ITER_STOP)
 722                         GOTO(out, rc = LDLM_ITER_STOP);
 723         }
 724  out:
 725         l_unlock(&ns->ns_lock);
 726         RETURN(rc);
 727 }
 728
 729 struct iter_helper_data {
 730         ldlm_iterator_t iter;
 731         void *closure;
 732 };
 733
 734 static int ldlm_iter_helper(struct ldlm_lock *lock, void *closure)
 735 {
 736         struct iter_helper_data *helper = closure;
 737         return helper->iter(lock, helper->closure);
 738 }
 739
 740 static int ldlm_res_iter_helper(struct ldlm_resource *res, void *closure)
 741 {
 742         return ldlm_resource_foreach(res, ldlm_iter_helper, closure);
 743 }
 744
 745 int ldlm_namespace_foreach(struct ldlm_namespace *ns, ldlm_iterator_t iter,
 746                            void *closure)
 747 {
 748         struct iter_helper_data helper = { iter: iter, closure: closure };
 749         return ldlm_namespace_foreach_res(ns, ldlm_res_iter_helper, &helper);
 750 }
 751
 752 int ldlm_namespace_foreach_res(struct ldlm_namespace *ns,
 753                                ldlm_res_iterator_t iter, void *closure)
 754 {
 755         int i, rc = LDLM_ITER_CONTINUE;
 756
 757         l_lock(&ns->ns_lock);
 758         for (i = 0; i < RES_HASH_SIZE; i++) {
 759                 struct list_head *tmp, *next;
 760                 list_for_each_safe(tmp, next, &(ns->ns_hash[i])) {
 761                         struct ldlm_resource *res =
 762                                 list_entry(tmp, struct ldlm_resource, lr_hash);
 763
 764                         ldlm_resource_getref(res);
 765                         rc = iter(res, closure);
 766                         ldlm_resource_putref(res);
 767                         if (rc == LDLM_ITER_STOP)
 768                                 GOTO(out, rc);
 769                 }
 770         }
 771  out:
 772         l_unlock(&ns->ns_lock);
 773         RETURN(rc);
 774 }
 775
 776 /* Lock replay */
 777
 778 static int ldlm_chain_lock_for_replay(struct ldlm_lock *lock, void *closure)
 779 {
 780         struct list_head *list = closure;
 781
 782         /* we use l_pending_chain here, because it's unused on clients. */
 783         list_add(&lock->l_pending_chain, list);
 784         return LDLM_ITER_CONTINUE;
 785 }
 786
 787 static int replay_one_lock(struct obd_import *imp, struct ldlm_lock *lock)
 788 {
 789         struct ptlrpc_request *req;
 790         struct ldlm_request *body;
 791         struct ldlm_reply *reply;
 792         int rc, size;
 793         int flags;
 794
 795         /*
 796          * If granted mode matches the requested mode, this lock is granted.
 797          *
 798          * If they differ, but we have a granted mode, then we were granted
 799          * one mode and now want another: ergo, converting.
 800          *
 801          * If we haven't been granted anything and are on a resource list,
 802          * then we're blocked/waiting.
 803          *
 804          * If we haven't been granted anything and we're NOT on a resource list,
 805          * then we haven't got a reply yet and don't have a known disposition.
 806          * This happens whenever a lock enqueue is the request that triggers
 807          * recovery.
 808          */
 809         if (lock->l_granted_mode == lock->l_req_mode)
 810                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_GRANTED;
 811         else if (lock->l_granted_mode)
 812                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_CONV;
 813         else if (!list_empty(&lock->l_res_link))
 814                 flags = LDLM_FL_REPLAY | LDLM_FL_BLOCK_WAIT;
 815         else
 816                 flags = LDLM_FL_REPLAY;
 817
 818         size = sizeof(*body);
 819         req = ptlrpc_prep_req(imp, LDLM_ENQUEUE, 1, &size, NULL);
 820         if (!req)
 821                 RETURN(-ENOMEM);
 822
 823         /* We're part of recovery, so don't wait for it. */
 824         req->rq_level = LUSTRE_CONN_RECOVD;
 825
 826         body = lustre_msg_buf(req->rq_reqmsg, 0);
 827         ldlm_lock2desc(lock, &body->lock_desc);
 828         body->lock_flags = flags;
 829
 830         ldlm_lock2handle(lock, &body->lock_handle1);
 831         size = sizeof(*reply);
 832         req->rq_replen = lustre_msg_size(1, &size);
 833
 834         LDLM_DEBUG0(lock, "replaying lock:");
 835         rc = ptlrpc_queue_wait(req);
 836         if (rc != ELDLM_OK)
 837                 GOTO(out, rc);
 838
 839         reply = lustre_msg_buf(req->rq_repmsg, 0);
 840         memcpy(&lock->l_remote_handle, &reply->lock_handle,
 841                sizeof(lock->l_remote_handle));
 842         LDLM_DEBUG0(lock, "replayed lock:");
 843  out:
 844         ptlrpc_req_finished(req);
 845         RETURN(rc);
 846 }
 847
 848 int ldlm_replay_locks(struct obd_import *imp)
 849 {
 850         struct ldlm_namespace *ns = imp->imp_obd->obd_namespace;
 851         struct list_head list, *pos, *next;
 852         struct ldlm_lock *lock;
 853         int rc = 0;
 854
 855         ENTRY;
 856         INIT_LIST_HEAD(&list);
 857
 858         l_lock(&ns->ns_lock);
 859         (void)ldlm_namespace_foreach(ns, ldlm_chain_lock_for_replay, &list);
 860
 861         list_for_each_safe(pos, next, &list) {
 862                 lock = list_entry(pos, struct ldlm_lock, l_pending_chain);
 863                 rc = replay_one_lock(imp, lock);
 864                 if (rc)
 865                         break; /* or try to do the rest? */
 866         }
 867         l_unlock(&ns->ns_lock);
 868         RETURN(rc);
 869 }