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