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