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