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