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