lustre/ldlm/ldlm_request.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (C) 2002 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of Lustre, http://www.lustre.org.
   7  *
   8  *   Lustre is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Lustre is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Lustre; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22 #define DEBUG_SUBSYSTEM S_LDLM
  23
  24 #include <linux/lustre_dlm.h>
  25 #include <linux/obd_class.h>
  26 #include <linux/obd.h>
  27
  28 static int interrupted_completion_wait(void *data)
  29 {
  30         RETURN(1);
  31 }
  32
  33 static int expired_completion_wait(void *data)
  34 {
  35         struct ldlm_lock *lock = data;
  36         struct ptlrpc_connection *conn;
  37         struct obd_device *obd;
  38
  39         if (!lock)
  40                 CERROR("NULL lock\n");
  41         else if (!lock->l_connh)
  42                 CERROR("lock %p has NULL connh\n", lock);
  43         else if (!(obd = class_conn2obd(lock->l_connh)))
  44                 CERROR("lock %p has NULL obd\n", lock);
  45         else if (!(conn = obd->u.cli.cl_import.imp_connection))
  46                 CERROR("lock %p has NULL connection\n", lock);
  47         else
  48                 class_signal_connection_failure(conn);
  49         RETURN(0);
  50 }
  51
  52 int ldlm_completion_ast(struct ldlm_lock *lock, int flags)
  53 {
  54         struct l_wait_info lwi =
  55                 LWI_TIMEOUT_INTR(obd_timeout * HZ, expired_completion_wait,
  56                                  interrupted_completion_wait, lock);
  57         int rc = 0;
  58         ENTRY;
  59
  60         if (flags == LDLM_FL_WAIT_NOREPROC)
  61                 goto noreproc;
  62
  63         if (flags == 0) {
  64                 wake_up(&lock->l_waitq);
  65                 RETURN(0);
  66         }
  67
  68         if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
  69                        LDLM_FL_BLOCK_CONV)))
  70                 RETURN(0);
  71
  72         LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, "
  73                    "sleeping");
  74         ldlm_lock_dump(lock);
  75         ldlm_reprocess_all(lock->l_resource);
  76
  77  noreproc:
  78         /* Go to sleep until the lock is granted or cancelled. */
  79         rc = l_wait_event(lock->l_waitq,
  80                           ((lock->l_req_mode == lock->l_granted_mode) ||
  81                            (lock->l_flags & LDLM_FL_DESTROYED)), &lwi);
  82
  83         if (lock->l_flags & LDLM_FL_DESTROYED) {
  84                 LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed");
  85                 RETURN(-EIO);
  86         }
  87
  88         if (rc) {
  89                 LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)",
  90                            rc);
  91                 RETURN(rc);
  92         }
  93
  94         LDLM_DEBUG(lock, "client-side enqueue waking up: granted");
  95         RETURN(0);
  96 }
  97
  98 static int ldlm_cli_enqueue_local(struct ldlm_namespace *ns,
  99                                   struct lustre_handle *parent_lockh,
 100                                   __u64 *res_id,
 101                                   __u32 type,
 102                                   void *cookie, int cookielen,
 103                                   ldlm_mode_t mode,
 104                                   int *flags,
 105                                   ldlm_completion_callback completion,
 106                                   ldlm_blocking_callback blocking,
 107                                   void *data,
 108                                   __u32 data_len,
 109                                   struct lustre_handle *lockh)
 110 {
 111         struct ldlm_lock *lock;
 112         int err;
 113         ENTRY;
 114
 115         if (ns->ns_client) {
 116                 CERROR("Trying to enqueue local lock in a shadow namespace\n");
 117                 LBUG();
 118         }
 119
 120         lock = ldlm_lock_create(ns, parent_lockh, res_id, type, mode, data,
 121                                 data_len);
 122         if (!lock)
 123                 GOTO(out_nolock, err = -ENOMEM);
 124         LDLM_DEBUG(lock, "client-side local enqueue handler, new lock created");
 125
 126         ldlm_lock_addref_internal(lock, mode);
 127         ldlm_lock2handle(lock, lockh);
 128         lock->l_connh = NULL;
 129
 130         err = ldlm_lock_enqueue(lock, cookie, cookielen, flags, completion,
 131                                 blocking);
 132         if (err != ELDLM_OK)
 133                 GOTO(out, err);
 134
 135         if (type == LDLM_EXTENT)
 136                 memcpy(cookie, &lock->l_extent, sizeof(lock->l_extent));
 137         if ((*flags) & LDLM_FL_LOCK_CHANGED)
 138                 memcpy(res_id, lock->l_resource->lr_name, sizeof(*res_id));
 139
 140         LDLM_DEBUG_NOLOCK("client-side local enqueue handler END (lock %p)",
 141                           lock);
 142
 143         if (lock->l_completion_ast)
 144                 lock->l_completion_ast(lock, *flags);
 145
 146         LDLM_DEBUG(lock, "client-side local enqueue END");
 147         EXIT;
 148  out:
 149         LDLM_LOCK_PUT(lock);
 150  out_nolock:
 151         return err;
 152 }
 153
 154 int ldlm_cli_enqueue(struct lustre_handle *connh,
 155                      struct ptlrpc_request *req,
 156                      struct ldlm_namespace *ns,
 157                      struct lustre_handle *parent_lock_handle,
 158                      __u64 *res_id,
 159                      __u32 type,
 160                      void *cookie, int cookielen,
 161                      ldlm_mode_t mode,
 162                      int *flags,
 163                      ldlm_completion_callback completion,
 164                      ldlm_blocking_callback blocking,
 165                      void *data,
 166                      __u32 data_len,
 167                      struct lustre_handle *lockh)
 168 {
 169         struct ldlm_lock *lock;
 170         struct ldlm_request *body;
 171         struct ldlm_reply *reply;
 172         int rc, size = sizeof(*body), req_passed_in = 1;
 173         ENTRY;
 174
 175         if (connh == NULL)
 176                 return ldlm_cli_enqueue_local(ns, parent_lock_handle, res_id,
 177                                               type, cookie, cookielen, mode,
 178                                               flags, completion, blocking, data,
 179                                               data_len, lockh);
 180
 181         *flags = 0;
 182         lock = ldlm_lock_create(ns, parent_lock_handle, res_id, type, mode,
 183                                 data, data_len);
 184         if (lock == NULL)
 185                 GOTO(out_nolock, rc = -ENOMEM);
 186         LDLM_DEBUG(lock, "client-side enqueue START");
 187         /* for the local lock, add the reference */
 188         ldlm_lock_addref_internal(lock, mode);
 189         ldlm_lock2handle(lock, lockh);
 190
 191         if (req == NULL) {
 192                 req = ptlrpc_prep_req(class_conn2cliimp(connh), LDLM_ENQUEUE, 1,
 193                                       &size, NULL);
 194                 if (!req)
 195                         GOTO(out, rc = -ENOMEM);
 196                 req_passed_in = 0;
 197         } else if (req->rq_reqmsg->buflens[0] != sizeof(*body))
 198                 LBUG();
 199
 200         /* Dump all of this data into the request buffer */
 201         body = lustre_msg_buf(req->rq_reqmsg, 0);
 202         ldlm_lock2desc(lock, &body->lock_desc);
 203         /* Phil: make this part of ldlm_lock2desc */
 204         if (type == LDLM_EXTENT) {
 205                 memcpy(&body->lock_desc.l_extent, cookie,
 206                        sizeof(body->lock_desc.l_extent));
 207                 CDEBUG(D_INFO, "extent in body: "LPU64" -> "LPU64"\n",
 208                            body->lock_desc.l_extent.start,
 209                            body->lock_desc.l_extent.end);
 210         }
 211         body->lock_flags = *flags;
 212
 213         memcpy(&body->lock_handle1, lockh, sizeof(*lockh));
 214         if (parent_lock_handle)
 215                 memcpy(&body->lock_handle2, parent_lock_handle,
 216                        sizeof(body->lock_handle2));
 217
 218         /* Continue as normal. */
 219         if (!req_passed_in) {
 220                 size = sizeof(*reply);
 221                 req->rq_replen = lustre_msg_size(1, &size);
 222         }
 223         lock->l_connh = connh;
 224         lock->l_export = NULL;
 225
 226         rc = ptlrpc_queue_wait(req);
 227         /* FIXME: status check here? */
 228         rc = ptlrpc_check_status(req, rc);
 229
 230         if (rc != ELDLM_OK) {
 231                 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
 232                            rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
 233                 ldlm_lock_decref(lockh, mode);
 234                 /* FIXME: if we've already received a completion AST, this will
 235                  * LBUG! */
 236                 ldlm_lock_destroy(lock);
 237                 GOTO(out, rc);
 238         }
 239
 240         reply = lustre_msg_buf(req->rq_repmsg, 0);
 241         memcpy(&lock->l_remote_handle, &reply->lock_handle,
 242                sizeof(lock->l_remote_handle));
 243         if (type == LDLM_EXTENT)
 244                 memcpy(cookie, &reply->lock_extent, sizeof(reply->lock_extent));
 245         *flags = reply->lock_flags;
 246
 247         CDEBUG(D_INFO, "remote handle: %p, flags: %d\n",
 248                (void *)(unsigned long)reply->lock_handle.addr, *flags);
 249         CDEBUG(D_INFO, "requested extent: "LPU64" -> "LPU64", got extent "
 250                LPU64" -> "LPU64"\n",
 251                body->lock_desc.l_extent.start, body->lock_desc.l_extent.end,
 252                reply->lock_extent.start, reply->lock_extent.end);
 253
 254         /* If enqueue returned a blocked lock but the completion handler has
 255          * already run, then it fixed up the resource and we don't need to do it
 256          * again. */
 257         if ((*flags) & LDLM_FL_LOCK_CHANGED) {
 258                 int newmode = reply->lock_mode;
 259                 if (newmode && newmode != lock->l_req_mode) {
 260                         LDLM_DEBUG(lock, "server returned different mode %s",
 261                                    ldlm_lockname[newmode]);
 262                         lock->l_req_mode = newmode;
 263                 }
 264
 265                 if (reply->lock_resource_name[0] !=
 266                     lock->l_resource->lr_name[0]) {
 267                         CDEBUG(D_INFO, "remote intent success, locking %ld "
 268                                "instead of %ld\n",
 269                                (long)reply->lock_resource_name[0],
 270                                (long)lock->l_resource->lr_name[0]);
 271
 272                         ldlm_lock_change_resource(lock,
 273                                                   reply->lock_resource_name);
 274                         if (lock->l_resource == NULL) {
 275                                 LBUG();
 276                                 RETURN(-ENOMEM);
 277                         }
 278                         LDLM_DEBUG(lock, "client-side enqueue, new resource");
 279                 }
 280         }
 281
 282         if (!req_passed_in)
 283                 ptlrpc_req_finished(req);
 284
 285         rc = ldlm_lock_enqueue(lock, cookie, cookielen, flags, completion,
 286                                blocking);
 287         if (lock->l_completion_ast)
 288                 lock->l_completion_ast(lock, *flags);
 289
 290         LDLM_DEBUG(lock, "client-side enqueue END");
 291         EXIT;
 292  out:
 293         LDLM_LOCK_PUT(lock);
 294  out_nolock:
 295         return rc;
 296 }
 297
 298 int ldlm_match_or_enqueue(struct lustre_handle *connh,
 299                           struct ptlrpc_request *req,
 300                           struct ldlm_namespace *ns,
 301                           struct lustre_handle *parent_lock_handle,
 302                           __u64 *res_id,
 303                           __u32 type,
 304                           void *cookie, int cookielen,
 305                           ldlm_mode_t mode,
 306                           int *flags,
 307                           ldlm_completion_callback completion,
 308                           ldlm_blocking_callback blocking,
 309                           void *data,
 310                           __u32 data_len,
 311                           struct lustre_handle *lockh)
 312 {
 313         int rc;
 314         ENTRY;
 315         rc = ldlm_lock_match(ns, res_id, type, cookie, cookielen, mode, lockh);
 316         if (rc == 0) {
 317                 rc = ldlm_cli_enqueue(connh, req, ns,
 318                                       parent_lock_handle, res_id, type, cookie,
 319                                       cookielen, mode, flags, completion,
 320                                       blocking, data, data_len, lockh);
 321                 if (rc != ELDLM_OK)
 322                         CERROR("ldlm_cli_enqueue: err: %d\n", rc);
 323                 RETURN(rc);
 324         } else
 325                 RETURN(0);
 326 }
 327
 328 static int ldlm_cli_convert_local(struct ldlm_lock *lock, int new_mode,
 329                                   int *flags)
 330 {
 331
 332         if (lock->l_resource->lr_namespace->ns_client) {
 333                 CERROR("Trying to cancel local lock\n");
 334                 LBUG();
 335         }
 336         LDLM_DEBUG(lock, "client-side local convert");
 337
 338         ldlm_lock_convert(lock, new_mode, flags);
 339         ldlm_reprocess_all(lock->l_resource);
 340
 341         LDLM_DEBUG(lock, "client-side local convert handler END");
 342         LDLM_LOCK_PUT(lock);
 343         RETURN(0);
 344 }
 345
 346 /* FIXME: one of ldlm_cli_convert or the server side should reject attempted
 347  * conversion of locks which are on the waiting or converting queue */
 348 int ldlm_cli_convert(struct lustre_handle *lockh, int new_mode, int *flags)
 349 {
 350         struct ldlm_request *body;
 351         struct lustre_handle *connh;
 352         struct ldlm_reply *reply;
 353         struct ldlm_lock *lock;
 354         struct ldlm_resource *res;
 355         struct ptlrpc_request *req;
 356         int rc, size = sizeof(*body);
 357         ENTRY;
 358
 359         lock = ldlm_handle2lock(lockh);
 360         if (!lock) {
 361                 LBUG();
 362                 RETURN(-EINVAL);
 363         }
 364         *flags = 0;
 365         connh = lock->l_connh;
 366
 367         if (!connh)
 368                 return ldlm_cli_convert_local(lock, new_mode, flags);
 369
 370         LDLM_DEBUG(lock, "client-side convert");
 371
 372         req = ptlrpc_prep_req(class_conn2cliimp(connh), LDLM_CONVERT, 1, &size,
 373                               NULL);
 374         if (!req)
 375                 GOTO(out, rc = -ENOMEM);
 376
 377         body = lustre_msg_buf(req->rq_reqmsg, 0);
 378         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 379                sizeof(body->lock_handle1));
 380
 381         body->lock_desc.l_req_mode = new_mode;
 382         body->lock_flags = *flags;
 383
 384         size = sizeof(*reply);
 385         req->rq_replen = lustre_msg_size(1, &size);
 386
 387         rc = ptlrpc_queue_wait(req);
 388         rc = ptlrpc_check_status(req, rc);
 389         if (rc != ELDLM_OK)
 390                 GOTO(out, rc);
 391
 392         reply = lustre_msg_buf(req->rq_repmsg, 0);
 393         res = ldlm_lock_convert(lock, new_mode, &reply->lock_flags);
 394         if (res != NULL)
 395                 ldlm_reprocess_all(res);
 396         /* Go to sleep until the lock is granted. */
 397         /* FIXME: or cancelled. */
 398         if (lock->l_completion_ast)
 399                 lock->l_completion_ast(lock, LDLM_FL_WAIT_NOREPROC);
 400         EXIT;
 401  out:
 402         LDLM_LOCK_PUT(lock);
 403         ptlrpc_req_finished(req);
 404         return rc;
 405 }
 406
 407 int ldlm_cli_cancel(struct lustre_handle *lockh)
 408 {
 409         struct ptlrpc_request *req;
 410         struct ldlm_lock *lock;
 411         struct ldlm_request *body;
 412         int rc = 0, size = sizeof(*body);
 413         ENTRY;
 414
 415         lock = ldlm_handle2lock(lockh);
 416         if (!lock) {
 417                 /* It's possible that the decref that we did just before this
 418                  * cancel was the last reader/writer, and caused a cancel before
 419                  * we could call this function.  If we want to make this
 420                  * impossible (by adding a dec_and_cancel() or similar), then
 421                  * we can put the LBUG back. */
 422                 //LBUG();
 423                 RETURN(-EINVAL);
 424         }
 425
 426         if (lock->l_connh) {
 427                 LDLM_DEBUG(lock, "client-side cancel");
 428                 /* Set this flag to prevent others from getting new references*/
 429                 l_lock(&ldlm_everything_lock);
 430                 lock->l_flags |= LDLM_FL_CBPENDING;
 431                 ldlm_cancel_callback(lock);
 432                 l_unlock(&ldlm_everything_lock);
 433
 434                 req = ptlrpc_prep_req(class_conn2cliimp(lock->l_connh),
 435                                       LDLM_CANCEL, 1, &size, NULL);
 436                 if (!req)
 437                         GOTO(out, rc = -ENOMEM);
 438
 439                 body = lustre_msg_buf(req->rq_reqmsg, 0);
 440                 memcpy(&body->lock_handle1, &lock->l_remote_handle,
 441                        sizeof(body->lock_handle1));
 442
 443                 req->rq_replen = lustre_msg_size(0, NULL);
 444
 445                 rc = ptlrpc_queue_wait(req);
 446                 rc = ptlrpc_check_status(req, rc);
 447                 ptlrpc_req_finished(req);
 448                 if (rc != ELDLM_OK)
 449                         GOTO(out, rc);
 450
 451                 ldlm_lock_cancel(lock);
 452         } else {
 453                 LDLM_DEBUG(lock, "client-side local cancel");
 454                 if (lock->l_resource->lr_namespace->ns_client) {
 455                         CERROR("Trying to cancel local lock\n");
 456                         LBUG();
 457                 }
 458                 ldlm_lock_cancel(lock);
 459                 ldlm_reprocess_all(lock->l_resource);
 460                 LDLM_DEBUG(lock, "client-side local cancel handler END");
 461         }
 462
 463         EXIT;
 464  out:
 465         LDLM_LOCK_PUT(lock);
 466         return rc;
 467 }
 468
 469 /* Cancel all locks on a given resource that have 0 readers/writers.
 470  *
 471  * If 'local_only' is true, throw the locks away without trying to notify the
 472  * server. */
 473 int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id,
 474                            int local_only)
 475 {
 476         struct ldlm_resource *res;
 477         struct list_head *tmp, *next, list = LIST_HEAD_INIT(list);
 478         struct ldlm_ast_work *w;
 479         ENTRY;
 480
 481         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
 482         if (res == NULL) {
 483                 /* This is not a problem. */
 484                 CDEBUG(D_INFO, "No resource "LPU64"\n", res_id[0]);
 485                 RETURN(0);
 486         }
 487
 488         l_lock(&ldlm_everything_lock);
 489         list_for_each(tmp, &res->lr_granted) {
 490                 struct ldlm_lock *lock;
 491                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 492
 493                 if (lock->l_readers || lock->l_writers)
 494                         continue;
 495
 496                 /* Setting the CBPENDING flag is a little misleading, but
 497                  * prevents an important race; namely, once CBPENDING is set,
 498                  * the lock can accumulate no more readers/writers.  Since
 499                  * readers and writers are already zero here, ldlm_lock_decref
 500                  * won't see this flag and call l_blocking_ast */
 501                 lock->l_flags |= LDLM_FL_CBPENDING;
 502
 503                 OBD_ALLOC(w, sizeof(*w));
 504                 LASSERT(w);
 505
 506                 w->w_lock = LDLM_LOCK_GET(lock);
 507                 list_add(&w->w_list, &list);
 508         }
 509         l_unlock(&ldlm_everything_lock);
 510
 511         list_for_each_safe(tmp, next, &list) {
 512                 struct lustre_handle lockh;
 513                 int rc;
 514                 w = list_entry(tmp, struct ldlm_ast_work, w_list);
 515
 516                 if (local_only)
 517                         ldlm_lock_cancel(w->w_lock);
 518                 else {
 519                         ldlm_lock2handle(w->w_lock, &lockh);
 520                         rc = ldlm_cli_cancel(&lockh);
 521                         if (rc != ELDLM_OK)
 522                                 CERROR("ldlm_cli_cancel: %d\n", rc);
 523                 }
 524                 LDLM_LOCK_PUT(w->w_lock);
 525                 list_del(&w->w_list);
 526                 OBD_FREE(w, sizeof(*w));
 527         }
 528
 529         ldlm_resource_put(res);
 530
 531         RETURN(0);
 532 }