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