lustre/ldlm/ldlm_lockd.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  *   Author: Peter Braam <braam@clusterfs.com>
   6  *   Author: Phil Schwan <phil@clusterfs.com>
   7  *
   8  *   This file is part of Lustre, http://www.lustre.org.
   9  *
  10  *   Lustre is free software; you can redistribute it and/or
  11  *   modify it under the terms of version 2 of the GNU General Public
  12  *   License as published by the Free Software Foundation.
  13  *
  14  *   Lustre is distributed in the hope that it will be useful,
  15  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  *   GNU General Public License for more details.
  18  *
  19  *   You should have received a copy of the GNU General Public License
  20  *   along with Lustre; if not, write to the Free Software
  21  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  22  */
  23
  24 #define EXPORT_SYMTAB
  25 #define DEBUG_SUBSYSTEM S_LDLM
  26
  27 #include <linux/module.h>
  28 #include <linux/slab.h>
  29 #include <linux/lustre_dlm.h>
  30 #include <linux/init.h>
  31 #include <linux/obd_class.h>
  32
  33 extern kmem_cache_t *ldlm_resource_slab;
  34 extern kmem_cache_t *ldlm_lock_slab;
  35 extern struct lustre_lock ldlm_handle_lock;
  36 extern struct list_head ldlm_namespace_list;
  37 extern int (*mds_reint_p)(int offset, struct ptlrpc_request *req);
  38 extern int (*mds_getattr_name_p)(int offset, struct ptlrpc_request *req);
  39
  40 inline unsigned long round_timeout(unsigned long timeout)
  41 {
  42         return ((timeout / HZ) + 1) * HZ;
  43 }
  44
  45 static struct list_head waiting_locks_list;
  46 static spinlock_t waiting_locks_spinlock;
  47 static struct timer_list waiting_locks_timer;
  48 static int ldlm_already_setup = 0;
  49
  50 static void waiting_locks_callback(unsigned long unused)
  51 {
  52         struct list_head *liter, *n;
  53
  54         spin_lock_bh(&waiting_locks_spinlock);
  55         list_for_each_safe(liter, n, &waiting_locks_list) {
  56                 struct ldlm_lock *l = list_entry(liter, struct ldlm_lock,
  57                                                  l_pending_chain);
  58                 if (l->l_callback_timeout > jiffies)
  59                         break;
  60                 CERROR("lock timer expired, lock %p\n", l);
  61                 LDLM_DEBUG(l, "timer expired, recovering exp %p on conn %p",
  62                            l->l_export, l->l_export->exp_connection);
  63                 recovd_conn_fail(l->l_export->exp_connection);
  64         }
  65         spin_unlock_bh(&waiting_locks_spinlock);
  66 }
  67
  68 /*
  69  * Indicate that we're waiting for a client to call us back cancelling a given
  70  * lock.  We add it to the pending-callback chain, and schedule the lock-timeout
  71  * timer to fire appropriately.  (We round up to the next second, to avoid
  72  * floods of timer firings during periods of high lock contention and traffic).
  73  */
  74 static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
  75 {
  76         unsigned long timeout_rounded;
  77         ENTRY;
  78
  79         LASSERT(list_empty(&lock->l_pending_chain));
  80
  81         spin_lock_bh(&waiting_locks_spinlock);
  82         lock->l_callback_timeout = jiffies + (obd_timeout * HZ / 2);
  83
  84         timeout_rounded = round_timeout(lock->l_callback_timeout);
  85
  86         if (timeout_rounded < waiting_locks_timer.expires ||
  87             !timer_pending(&waiting_locks_timer)) {
  88                 mod_timer(&waiting_locks_timer, timeout_rounded);
  89         }
  90         list_add_tail(&lock->l_pending_chain, &waiting_locks_list); /* FIFO */
  91         spin_unlock_bh(&waiting_locks_spinlock);
  92         RETURN(1);
  93 }
  94
  95 /*
  96  * Remove a lock from the pending list, likely because it had its cancellation
  97  * callback arrive without incident.  This adjusts the lock-timeout timer if
  98  * needed.  Returns 0 if the lock wasn't pending after all, 1 if it was.
  99  */
 100 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
 101 {
 102         struct list_head *list_next;
 103
 104         ENTRY;
 105
 106         spin_lock_bh(&waiting_locks_spinlock);
 107
 108         if (list_empty(&lock->l_pending_chain)) {
 109                 spin_unlock_bh(&waiting_locks_spinlock);
 110                 RETURN(0);
 111         }
 112
 113         list_next = lock->l_pending_chain.next;
 114         if (lock->l_pending_chain.prev == &waiting_locks_list) {
 115                 /* Removing the head of the list, adjust timer. */
 116                 if (list_next == &waiting_locks_list) {
 117                         /* No more, just cancel. */
 118                         del_timer(&waiting_locks_timer);
 119                 } else {
 120                         struct ldlm_lock *next;
 121                         next = list_entry(list_next, struct ldlm_lock,
 122                                           l_pending_chain);
 123                         mod_timer(&waiting_locks_timer,
 124                                   round_timeout(next->l_callback_timeout));
 125                 }
 126         }
 127         list_del_init(&lock->l_pending_chain);
 128         spin_unlock_bh(&waiting_locks_spinlock);
 129         RETURN(1);
 130 }
 131
 132 static int ldlm_server_blocking_ast(struct ldlm_lock *lock,
 133                                     struct ldlm_lock_desc *desc,
 134                                     void *data, __u32 data_len, int flag)
 135 {
 136         struct ldlm_request *body;
 137         struct ptlrpc_request *req;
 138         int rc = 0, size = sizeof(*body);
 139         ENTRY;
 140
 141         if (flag == LDLM_CB_CANCELING) {
 142                 /* Don't need to do anything here. */
 143                 RETURN(0);
 144         }
 145
 146         LASSERT(lock);
 147
 148         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 149         if (lock->l_destroyed) {
 150                 /* What's the point? */
 151                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 152                 RETURN(0);
 153         }
 154
 155         req = ptlrpc_prep_req(&lock->l_export->exp_ldlm_data.led_import,
 156                               LDLM_BL_CALLBACK, 1, &size, NULL);
 157         if (!req)
 158                 RETURN(-ENOMEM);
 159
 160         body = lustre_msg_buf(req->rq_reqmsg, 0);
 161         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 162                sizeof(body->lock_handle1));
 163         memcpy(&body->lock_desc, desc, sizeof(*desc));
 164
 165         LDLM_DEBUG(lock, "server preparing blocking AST");
 166         req->rq_replen = lustre_msg_size(0, NULL);
 167
 168         ldlm_add_waiting_lock(lock);
 169         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 170
 171         req->rq_level = LUSTRE_CONN_RECOVD;
 172         rc = ptlrpc_queue_wait(req);
 173         if (rc == -ETIMEDOUT || rc == -EINTR) {
 174                 ldlm_expired_completion_wait(lock);
 175         } else if (rc) {
 176                 CERROR("client returned %d from blocking AST for lock %p\n",
 177                        req->rq_status, lock);
 178                 LDLM_DEBUG(lock, "client returned error %d from blocking AST",
 179                            req->rq_status);
 180                 ldlm_lock_cancel(lock);
 181                 /* Server-side AST functions are called from ldlm_reprocess_all,
 182                  * which needs to be told to please restart its reprocessing. */
 183                 rc = -ERESTART;
 184         }
 185
 186         ptlrpc_req_finished(req);
 187
 188         RETURN(rc);
 189 }
 190
 191 static int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags)
 192 {
 193         struct ldlm_request *body;
 194         struct ptlrpc_request *req;
 195         int rc = 0, size = sizeof(*body);
 196         ENTRY;
 197
 198         if (lock == NULL) {
 199                 LBUG();
 200                 RETURN(-EINVAL);
 201         }
 202
 203         req = ptlrpc_prep_req(&lock->l_export->exp_ldlm_data.led_import,
 204                               LDLM_CP_CALLBACK, 1, &size, NULL);
 205         if (!req)
 206                 RETURN(-ENOMEM);
 207
 208         body = lustre_msg_buf(req->rq_reqmsg, 0);
 209         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 210                sizeof(body->lock_handle1));
 211         body->lock_flags = flags;
 212         ldlm_lock2desc(lock, &body->lock_desc);
 213
 214         LDLM_DEBUG(lock, "server preparing completion AST");
 215         req->rq_replen = lustre_msg_size(0, NULL);
 216
 217         req->rq_level = LUSTRE_CONN_RECOVD;
 218         rc = ptlrpc_queue_wait(req);
 219         if (rc == -ETIMEDOUT || rc == -EINTR) {
 220                 ldlm_expired_completion_wait(lock);
 221         } else if (rc) {
 222                 CERROR("client returned %d from completion AST for lock %p\n",
 223                        req->rq_status, lock);
 224                 LDLM_DEBUG(lock, "client returned error %d from completion AST",
 225                            req->rq_status);
 226                 ldlm_lock_cancel(lock);
 227                 /* Server-side AST functions are called from ldlm_reprocess_all,
 228                  * which needs to be told to please restart its reprocessing. */
 229                 rc = -ERESTART;
 230         }
 231         ptlrpc_req_finished(req);
 232
 233         RETURN(rc);
 234 }
 235
 236 int ldlm_handle_enqueue(struct ptlrpc_request *req)
 237 {
 238         struct obd_device *obddev = req->rq_export->exp_obd;
 239         struct ldlm_reply *dlm_rep;
 240         struct ldlm_request *dlm_req;
 241         int rc, size = sizeof(*dlm_rep), cookielen = 0;
 242         __u32 flags;
 243         ldlm_error_t err;
 244         struct ldlm_lock *lock = NULL;
 245         void *cookie = NULL;
 246         ENTRY;
 247
 248         LDLM_DEBUG_NOLOCK("server-side enqueue handler START");
 249
 250         dlm_req = lustre_msg_buf(req->rq_reqmsg, 0);
 251         flags = dlm_req->lock_flags;
 252         if (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN &&
 253             (flags & LDLM_FL_HAS_INTENT)) {
 254                 /* In this case, the reply buffer is allocated deep in
 255                  * local_lock_enqueue by the policy function. */
 256                 cookie = req;
 257                 cookielen = sizeof(*req);
 258         } else {
 259                 rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen,
 260                                      &req->rq_repmsg);
 261                 if (rc) {
 262                         CERROR("out of memory\n");
 263                         RETURN(-ENOMEM);
 264                 }
 265                 if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT) {
 266                         cookie = &dlm_req->lock_desc.l_extent;
 267                         cookielen = sizeof(struct ldlm_extent);
 268                 }
 269         }
 270
 271         /* XXX notice that this lock has no callback data: of course the
 272            export would be exactly what we may want to use here... */
 273         lock = ldlm_lock_create(obddev->obd_namespace,
 274                                 &dlm_req->lock_handle2,
 275                                 dlm_req->lock_desc.l_resource.lr_name,
 276                                 dlm_req->lock_desc.l_resource.lr_type,
 277                                 dlm_req->lock_desc.l_req_mode, NULL, 0);
 278         if (!lock)
 279                 GOTO(out, err = -ENOMEM);
 280
 281         memcpy(&lock->l_remote_handle, &dlm_req->lock_handle1,
 282                sizeof(lock->l_remote_handle));
 283         LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
 284
 285         LASSERT(req->rq_export);
 286         lock->l_export = req->rq_export;
 287         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 288         list_add(&lock->l_export_chain,
 289                  &lock->l_export->exp_ldlm_data.led_held_locks);
 290         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 291
 292         err = ldlm_lock_enqueue(obddev->obd_namespace, lock, cookie, cookielen,
 293                                 &flags, ldlm_server_completion_ast,
 294                                 ldlm_server_blocking_ast);
 295         if (err != ELDLM_OK)
 296                 GOTO(out, err);
 297
 298         dlm_rep = lustre_msg_buf(req->rq_repmsg, 0);
 299         dlm_rep->lock_flags = flags;
 300
 301         ldlm_lock2handle(lock, &dlm_rep->lock_handle);
 302         if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT)
 303                 memcpy(&dlm_rep->lock_extent, &lock->l_extent,
 304                        sizeof(lock->l_extent));
 305         if (dlm_rep->lock_flags & LDLM_FL_LOCK_CHANGED) {
 306                 memcpy(dlm_rep->lock_resource_name, lock->l_resource->lr_name,
 307                        sizeof(dlm_rep->lock_resource_name));
 308                 dlm_rep->lock_mode = lock->l_req_mode;
 309         }
 310
 311         EXIT;
 312  out:
 313         if (lock)
 314                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
 315                            "(err=%d)", err);
 316         req->rq_status = err;
 317
 318         if (lock) {
 319                 if (!err)
 320                         ldlm_reprocess_all(lock->l_resource);
 321                 LDLM_LOCK_PUT(lock);
 322         }
 323         LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p)", lock);
 324
 325         return 0;
 326 }
 327
 328 int ldlm_handle_convert(struct ptlrpc_request *req)
 329 {
 330         struct ldlm_request *dlm_req;
 331         struct ldlm_reply *dlm_rep;
 332         struct ldlm_lock *lock;
 333         int rc, size = sizeof(*dlm_rep);
 334         ENTRY;
 335
 336         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
 337         if (rc) {
 338                 CERROR("out of memory\n");
 339                 RETURN(-ENOMEM);
 340         }
 341         dlm_req = lustre_msg_buf(req->rq_reqmsg, 0);
 342         dlm_rep = lustre_msg_buf(req->rq_repmsg, 0);
 343         dlm_rep->lock_flags = dlm_req->lock_flags;
 344
 345         lock = ldlm_handle2lock(&dlm_req->lock_handle1);
 346         if (!lock) {
 347                 req->rq_status = EINVAL;
 348         } else {
 349                 LDLM_DEBUG(lock, "server-side convert handler START");
 350                 ldlm_lock_convert(lock, dlm_req->lock_desc.l_req_mode,
 351                                   &dlm_rep->lock_flags);
 352                 if (ldlm_del_waiting_lock(lock))
 353                         CDEBUG(D_DLMTRACE, "converted waiting lock %p\n", lock);
 354                 req->rq_status = 0;
 355         }
 356
 357         if (lock) {
 358                 ldlm_reprocess_all(lock->l_resource);
 359                 LDLM_DEBUG(lock, "server-side convert handler END");
 360                 LDLM_LOCK_PUT(lock);
 361         } else
 362                 LDLM_DEBUG_NOLOCK("server-side convert handler END");
 363
 364         RETURN(0);
 365 }
 366
 367 int ldlm_handle_cancel(struct ptlrpc_request *req)
 368 {
 369         struct ldlm_request *dlm_req;
 370         struct ldlm_lock *lock;
 371         int rc;
 372         ENTRY;
 373
 374         rc = lustre_pack_msg(0, NULL, NULL, &req->rq_replen, &req->rq_repmsg);
 375         if (rc) {
 376                 CERROR("out of memory\n");
 377                 RETURN(-ENOMEM);
 378         }
 379         dlm_req = lustre_msg_buf(req->rq_reqmsg, 0);
 380         if (!dlm_req) {
 381                 CERROR("bad request buffer for cancel\n");
 382                 RETURN(-EINVAL);
 383         }
 384
 385         lock = ldlm_handle2lock(&dlm_req->lock_handle1);
 386         if (!lock) {
 387                 LDLM_DEBUG_NOLOCK("server-side cancel handler stale lock (lock "
 388                                   "%p)", (void *)(unsigned long)
 389                                   dlm_req->lock_handle1.addr);
 390                 req->rq_status = ESTALE;
 391         } else {
 392                 LDLM_DEBUG(lock, "server-side cancel handler START");
 393                 ldlm_lock_cancel(lock);
 394                 if (ldlm_del_waiting_lock(lock))
 395                         CDEBUG(D_DLMTRACE, "cancelled waiting lock %p\n", lock);
 396                 req->rq_status = 0;
 397         }
 398
 399         if (ptlrpc_reply(req->rq_svc, req) != 0)
 400                 LBUG();
 401
 402         if (lock) {
 403                 ldlm_reprocess_all(lock->l_resource);
 404                 LDLM_DEBUG(lock, "server-side cancel handler END");
 405                 LDLM_LOCK_PUT(lock);
 406         }
 407
 408         RETURN(0);
 409 }
 410
 411 struct ldlm_lock *ldlm_handle2lock_ns(struct ldlm_namespace *ns,
 412                                       struct lustre_handle *handle);
 413
 414 static int ldlm_handle_bl_callback(struct ptlrpc_request *req,
 415                                    struct ldlm_namespace *ns)
 416 {
 417         struct ldlm_request *dlm_req;
 418         struct ldlm_lock *lock;
 419         int do_ast;
 420         ENTRY;
 421
 422         OBD_FAIL_RETURN(OBD_FAIL_OSC_LOCK_BL_AST, 0);
 423
 424         dlm_req = lustre_msg_buf(req->rq_reqmsg, 0);
 425
 426         lock = ldlm_handle2lock_ns(ns, &dlm_req->lock_handle1);
 427         if (!lock) {
 428                 CERROR("blocking callback on lock "LPX64" - lock disappeared\n",
 429                        dlm_req->lock_handle1.cookie);
 430                 RETURN(-EINVAL);
 431         }
 432
 433         LDLM_DEBUG(lock, "client blocking AST callback handler START");
 434
 435         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 436         lock->l_flags |= LDLM_FL_CBPENDING;
 437         do_ast = (!lock->l_readers && !lock->l_writers);
 438         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 439
 440         if (do_ast) {
 441                 LDLM_DEBUG(lock, "already unused, calling "
 442                            "callback (%p)", lock->l_blocking_ast);
 443                 if (lock->l_blocking_ast != NULL) {
 444                         lock->l_blocking_ast(lock, &dlm_req->lock_desc,
 445                                              lock->l_data, lock->l_data_len,
 446                                              LDLM_CB_BLOCKING);
 447                 }
 448         } else
 449                 LDLM_DEBUG(lock, "Lock still has references, will be"
 450                            " cancelled later");
 451
 452         LDLM_DEBUG(lock, "client blocking callback handler END");
 453         LDLM_LOCK_PUT(lock);
 454         RETURN(0);
 455 }
 456
 457 static int ldlm_handle_cp_callback(struct ptlrpc_request *req,
 458                                    struct ldlm_namespace *ns)
 459 {
 460         struct list_head ast_list = LIST_HEAD_INIT(ast_list);
 461         struct ldlm_request *dlm_req;
 462         struct ldlm_lock *lock;
 463         ENTRY;
 464
 465         OBD_FAIL_RETURN(OBD_FAIL_OSC_LOCK_CP_AST, 0);
 466
 467         dlm_req = lustre_msg_buf(req->rq_reqmsg, 0);
 468
 469         lock = ldlm_handle2lock_ns(ns, &dlm_req->lock_handle1);
 470         if (!lock) {
 471                 CERROR("completion callback on lock "LPX64" - lock "
 472                        "disappeared\n", dlm_req->lock_handle1.cookie);
 473                 RETURN(-EINVAL);
 474         }
 475
 476         LDLM_DEBUG(lock, "client completion callback handler START");
 477
 478         l_lock(&ns->ns_lock);
 479
 480         /* If we receive the completion AST before the actual enqueue returned,
 481          * then we might need to switch lock modes, resources, or extents. */
 482         if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
 483                 lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
 484                 LDLM_DEBUG(lock, "completion AST, new lock mode");
 485         }
 486         if (lock->l_resource->lr_type == LDLM_EXTENT)
 487                 memcpy(&lock->l_extent, &dlm_req->lock_desc.l_extent,
 488                        sizeof(lock->l_extent));
 489         ldlm_resource_unlink_lock(lock);
 490         if (memcmp(dlm_req->lock_desc.l_resource.lr_name,
 491                    lock->l_resource->lr_name,
 492                    sizeof(__u64) * RES_NAME_SIZE) != 0) {
 493                 ldlm_lock_change_resource(ns, lock,
 494                                          dlm_req->lock_desc.l_resource.lr_name);
 495                 LDLM_DEBUG(lock, "completion AST, new resource");
 496         }
 497         lock->l_resource->lr_tmp = &ast_list;
 498         ldlm_grant_lock(lock);
 499         lock->l_resource->lr_tmp = NULL;
 500         l_unlock(&ns->ns_lock);
 501         LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
 502         LDLM_LOCK_PUT(lock);
 503
 504         ldlm_run_ast_work(&ast_list);
 505
 506         LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
 507                           lock);
 508         RETURN(0);
 509 }
 510
 511 static int ldlm_callback_handler(struct ptlrpc_request *req)
 512 {
 513         struct ldlm_namespace *ns;
 514         int rc;
 515         ENTRY;
 516
 517         rc = lustre_unpack_msg(req->rq_reqmsg, req->rq_reqlen);
 518         if (rc) {
 519                 CERROR("Invalid request: %d\n", rc);
 520                 RETURN(rc);
 521         }
 522
 523         if (req->rq_export == NULL) {
 524                 struct ldlm_request *dlm_req;
 525
 526                 CERROR("operation %d with bad export (ptl req %d/rep %d)\n",
 527                        req->rq_reqmsg->opc, req->rq_request_portal,
 528                        req->rq_reply_portal);
 529                 CERROR("--> export addr: "LPX64", cookie: "LPX64"\n",
 530                        req->rq_reqmsg->addr, req->rq_reqmsg->cookie);
 531                 dlm_req = lustre_msg_buf(req->rq_reqmsg, 0);
 532                 CERROR("--> lock addr: "LPX64", cookie: "LPX64"\n",
 533                        dlm_req->lock_handle1.addr,dlm_req->lock_handle1.cookie);
 534                 RETURN(-ENOTCONN);
 535         }
 536
 537         LASSERT(req->rq_export != NULL);
 538         LASSERT(req->rq_export->exp_obd != NULL);
 539         ns = req->rq_export->exp_obd->obd_namespace;
 540         LASSERT(ns != NULL);
 541
 542         switch (req->rq_reqmsg->opc) {
 543         case LDLM_BL_CALLBACK:
 544                 CDEBUG(D_INODE, "blocking ast\n");
 545                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
 546                 rc = ldlm_handle_bl_callback(req, ns);
 547                 break;
 548         case LDLM_CP_CALLBACK:
 549                 CDEBUG(D_INODE, "completion ast\n");
 550                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CP_CALLBACK, 0);
 551                 rc = ldlm_handle_cp_callback(req, ns);
 552                 break;
 553         default:
 554                 CERROR("invalid opcode %d\n", req->rq_reqmsg->opc);
 555                 RETURN(-EINVAL);
 556         }
 557
 558         req->rq_status = rc;
 559         if (rc) {
 560                 ptlrpc_error(req->rq_svc, req);
 561         } else {
 562                 rc = lustre_pack_msg(0, NULL, NULL, &req->rq_replen,
 563                                      &req->rq_repmsg);
 564                 if (rc)
 565                         RETURN(rc);
 566                 ptlrpc_reply(req->rq_svc, req);
 567         }
 568
 569         RETURN(0);
 570 }
 571
 572 static int ldlm_cancel_handler(struct ptlrpc_request *req)
 573 {
 574         int rc;
 575         ENTRY;
 576
 577         rc = lustre_unpack_msg(req->rq_reqmsg, req->rq_reqlen);
 578         if (rc) {
 579                 CERROR("lustre_ldlm: Invalid request: %d\n", rc);
 580                 RETURN(rc);
 581         }
 582
 583         if (req->rq_export == NULL) {
 584                 struct ldlm_request *dlm_req;
 585                 CERROR("operation %d with bad export (ptl req %d/rep %d)\n",
 586                        req->rq_reqmsg->opc, req->rq_request_portal,
 587                        req->rq_reply_portal);
 588                 CERROR("--> export addr: "LPX64", cookie: "LPX64"\n",
 589                        req->rq_reqmsg->addr, req->rq_reqmsg->cookie);
 590                 dlm_req = lustre_msg_buf(req->rq_reqmsg, 0);
 591                 ldlm_lock_dump_handle(D_ERROR, &dlm_req->lock_handle1);
 592                 CERROR("--> ignoring this error as a temporary workaround!  "
 593                        "beware!\n");
 594                 //RETURN(-ENOTCONN);
 595         }
 596
 597         switch (req->rq_reqmsg->opc) {
 598
 599         /* XXX FIXME move this back to mds/handler.c, bug 625069 */
 600         case LDLM_CANCEL:
 601                 CDEBUG(D_INODE, "cancel\n");
 602                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CANCEL, 0);
 603                 rc = ldlm_handle_cancel(req);
 604                 if (rc)
 605                         break;
 606                 RETURN(0);
 607
 608         default:
 609                 CERROR("invalid opcode %d\n", req->rq_reqmsg->opc);
 610                 RETURN(-EINVAL);
 611         }
 612
 613         RETURN(0);
 614 }
 615
 616 static int ldlm_iocontrol(unsigned int cmd, struct lustre_handle *conn, int len,
 617                           void *karg, void *uarg)
 618 {
 619         struct obd_device *obddev = class_conn2obd(conn);
 620         struct ptlrpc_connection *connection;
 621         int err = 0;
 622         ENTRY;
 623
 624         if (_IOC_TYPE(cmd) != IOC_LDLM_TYPE || _IOC_NR(cmd) < IOC_LDLM_MIN_NR ||
 625             _IOC_NR(cmd) > IOC_LDLM_MAX_NR) {
 626                 CDEBUG(D_IOCTL, "invalid ioctl (type %d, nr %d, size %d)\n",
 627                        _IOC_TYPE(cmd), _IOC_NR(cmd), _IOC_SIZE(cmd));
 628                 RETURN(-EINVAL);
 629         }
 630
 631         OBD_ALLOC(obddev->u.ldlm.ldlm_client,
 632                   sizeof(*obddev->u.ldlm.ldlm_client));
 633         connection = ptlrpc_uuid_to_connection("ldlm");
 634         if (!connection)
 635                 CERROR("No LDLM UUID found: assuming ldlm is local.\n");
 636
 637         switch (cmd) {
 638         case IOC_LDLM_TEST:
 639                 err = ldlm_test(obddev, conn);
 640                 CERROR("-- done err %d\n", err);
 641                 GOTO(out, err);
 642         case IOC_LDLM_DUMP:
 643                 ldlm_dump_all_namespaces();
 644                 GOTO(out, err);
 645         default:
 646                 GOTO(out, err = -EINVAL);
 647         }
 648
 649  out:
 650         if (connection)
 651                 ptlrpc_put_connection(connection);
 652         OBD_FREE(obddev->u.ldlm.ldlm_client,
 653                  sizeof(*obddev->u.ldlm.ldlm_client));
 654         return err;
 655 }
 656
 657 static int ldlm_setup(struct obd_device *obddev, obd_count len, void *buf)
 658 {
 659         struct ldlm_obd *ldlm = &obddev->u.ldlm;
 660         int rc, i;
 661         ENTRY;
 662
 663         if (ldlm_already_setup)
 664                 RETURN(-EALREADY);
 665
 666         rc = ldlm_proc_setup(obddev);
 667         if (rc != 0)
 668                 RETURN(rc);
 669
 670         ldlm->ldlm_cb_service =
 671                 ptlrpc_init_svc(LDLM_NEVENTS, LDLM_NBUFS, LDLM_BUFSIZE,
 672                                 LDLM_MAXREQSIZE, LDLM_CB_REQUEST_PORTAL,
 673                                 LDLM_CB_REPLY_PORTAL, "self",
 674                                 ldlm_callback_handler, "ldlm_cbd");
 675
 676         if (!ldlm->ldlm_cb_service) {
 677                 CERROR("failed to start service\n");
 678                 GOTO(out_proc, rc = -ENOMEM);
 679         }
 680
 681         ldlm->ldlm_cancel_service =
 682                 ptlrpc_init_svc(LDLM_NEVENTS, LDLM_NBUFS, LDLM_BUFSIZE,
 683                                 LDLM_MAXREQSIZE, LDLM_CANCEL_REQUEST_PORTAL,
 684                                 LDLM_CANCEL_REPLY_PORTAL, "self",
 685                                 ldlm_cancel_handler, "ldlm_canceld");
 686
 687         if (!ldlm->ldlm_cancel_service) {
 688                 CERROR("failed to start service\n");
 689                 GOTO(out_proc, rc = -ENOMEM);
 690         }
 691
 692         for (i = 0; i < LDLM_NUM_THREADS; i++) {
 693                 char name[32];
 694                 sprintf(name, "ldlm_cn_%02d", i);
 695                 rc = ptlrpc_start_thread(obddev, ldlm->ldlm_cancel_service,
 696                                          name);
 697                 if (rc) {
 698                         CERROR("cannot start LDLM thread #%d: rc %d\n", i, rc);
 699                         LBUG();
 700                         GOTO(out_thread, rc);
 701                 }
 702         }
 703
 704         for (i = 0; i < LDLM_NUM_THREADS; i++) {
 705                 char name[32];
 706                 sprintf(name, "ldlm_cb_%02d", i);
 707                 rc = ptlrpc_start_thread(obddev, ldlm->ldlm_cb_service, name);
 708                 if (rc) {
 709                         CERROR("cannot start LDLM thread #%d: rc %d\n", i, rc);
 710                         LBUG();
 711                         GOTO(out_thread, rc);
 712                 }
 713         }
 714
 715         INIT_LIST_HEAD(&waiting_locks_list);
 716         spin_lock_init(&waiting_locks_spinlock);
 717         waiting_locks_timer.function = waiting_locks_callback;
 718         waiting_locks_timer.data = 0;
 719         init_timer(&waiting_locks_timer);
 720
 721         ldlm_already_setup = 1;
 722
 723         RETURN(0);
 724
 725  out_thread:
 726         ptlrpc_stop_all_threads(ldlm->ldlm_cancel_service);
 727         ptlrpc_unregister_service(ldlm->ldlm_cancel_service);
 728         ptlrpc_stop_all_threads(ldlm->ldlm_cb_service);
 729         ptlrpc_unregister_service(ldlm->ldlm_cb_service);
 730
 731  out_proc:
 732         ldlm_proc_cleanup(obddev);
 733
 734         return rc;
 735 }
 736
 737 static int ldlm_cleanup(struct obd_device *obddev)
 738 {
 739         struct ldlm_obd *ldlm = &obddev->u.ldlm;
 740         ENTRY;
 741
 742         if (!list_empty(&ldlm_namespace_list)) {
 743                 CERROR("ldlm still has namespaces; clean these up first.\n");
 744                 RETURN(-EBUSY);
 745         }
 746
 747         ptlrpc_stop_all_threads(ldlm->ldlm_cb_service);
 748         ptlrpc_unregister_service(ldlm->ldlm_cb_service);
 749         ptlrpc_stop_all_threads(ldlm->ldlm_cancel_service);
 750         ptlrpc_unregister_service(ldlm->ldlm_cancel_service);
 751         ldlm_proc_cleanup(obddev);
 752
 753         ldlm_already_setup = 0;
 754         RETURN(0);
 755 }
 756
 757 static int ldlm_connect(struct lustre_handle *conn, struct obd_device *src,
 758                         obd_uuid_t cluuid, struct recovd_obd *recovd,
 759                         ptlrpc_recovery_cb_t recover)
 760 {
 761         return class_connect(conn, src, cluuid);
 762 }
 763
 764 struct obd_ops ldlm_obd_ops = {
 765         o_owner:       THIS_MODULE,
 766         o_iocontrol:   ldlm_iocontrol,
 767         o_setup:       ldlm_setup,
 768         o_cleanup:     ldlm_cleanup,
 769         o_connect:     ldlm_connect,
 770         o_disconnect:  class_disconnect
 771 };
 772
 773 static int __init ldlm_init(void)
 774 {
 775         int rc = class_register_type(&ldlm_obd_ops, 0, OBD_LDLM_DEVICENAME);
 776         if (rc != 0)
 777                 return rc;
 778
 779         ldlm_resource_slab = kmem_cache_create("ldlm_resources",
 780                                                sizeof(struct ldlm_resource), 0,
 781                                                SLAB_HWCACHE_ALIGN, NULL, NULL);
 782         if (ldlm_resource_slab == NULL)
 783                 return -ENOMEM;
 784
 785         ldlm_lock_slab = kmem_cache_create("ldlm_locks",
 786                                            sizeof(struct ldlm_lock), 0,
 787                                            SLAB_HWCACHE_ALIGN, NULL, NULL);
 788         if (ldlm_lock_slab == NULL) {
 789                 kmem_cache_destroy(ldlm_resource_slab);
 790                 return -ENOMEM;
 791         }
 792
 793         l_lock_init(&ldlm_handle_lock);
 794
 795         return 0;
 796 }
 797
 798 static void __exit ldlm_exit(void)
 799 {
 800         class_unregister_type(OBD_LDLM_DEVICENAME);
 801         if (kmem_cache_destroy(ldlm_resource_slab) != 0)
 802                 CERROR("couldn't free ldlm resource slab\n");
 803         if (kmem_cache_destroy(ldlm_lock_slab) != 0)
 804                 CERROR("couldn't free ldlm lock slab\n");
 805 }
 806
 807 EXPORT_SYMBOL(ldlm_completion_ast);
 808 EXPORT_SYMBOL(ldlm_handle_enqueue);
 809 EXPORT_SYMBOL(ldlm_handle_cancel);
 810 EXPORT_SYMBOL(ldlm_handle_convert);
 811 EXPORT_SYMBOL(ldlm_register_intent);
 812 EXPORT_SYMBOL(ldlm_unregister_intent);
 813 EXPORT_SYMBOL(ldlm_lockname);
 814 EXPORT_SYMBOL(ldlm_typename);
 815 EXPORT_SYMBOL(__ldlm_handle2lock);
 816 EXPORT_SYMBOL(ldlm_lock2handle);
 817 EXPORT_SYMBOL(ldlm_lock_put);
 818 EXPORT_SYMBOL(ldlm_lock_match);
 819 EXPORT_SYMBOL(ldlm_lock_addref);
 820 EXPORT_SYMBOL(ldlm_lock_decref);
 821 EXPORT_SYMBOL(ldlm_lock_change_resource);
 822 EXPORT_SYMBOL(ldlm_lock_set_data);
 823 EXPORT_SYMBOL(ldlm_cli_convert);
 824 EXPORT_SYMBOL(ldlm_cli_enqueue);
 825 EXPORT_SYMBOL(ldlm_cli_cancel);
 826 EXPORT_SYMBOL(ldlm_cli_cancel_unused);
 827 EXPORT_SYMBOL(ldlm_match_or_enqueue);
 828 EXPORT_SYMBOL(ldlm_it2str);
 829 EXPORT_SYMBOL(ldlm_test);
 830 EXPORT_SYMBOL(ldlm_regression_start);
 831 EXPORT_SYMBOL(ldlm_regression_stop);
 832 EXPORT_SYMBOL(ldlm_lock_dump);
 833 EXPORT_SYMBOL(ldlm_lock_dump_handle);
 834 EXPORT_SYMBOL(ldlm_namespace_new);
 835 EXPORT_SYMBOL(ldlm_namespace_cleanup);
 836 EXPORT_SYMBOL(ldlm_namespace_free);
 837 EXPORT_SYMBOL(ldlm_namespace_dump);
 838 EXPORT_SYMBOL(ldlm_cancel_locks_for_export);
 839 EXPORT_SYMBOL(ldlm_replay_locks);
 840 EXPORT_SYMBOL(ldlm_resource_foreach);
 841 EXPORT_SYMBOL(ldlm_reprocess_all_ns);
 842 EXPORT_SYMBOL(ldlm_namespace_foreach);
 843 EXPORT_SYMBOL(ldlm_namespace_foreach_res);
 844 EXPORT_SYMBOL(l_lock);
 845 EXPORT_SYMBOL(l_unlock);
 846
 847 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 848 MODULE_DESCRIPTION("Lustre Lock Management Module v0.1");
 849 MODULE_LICENSE("GPL");
 850
 851 module_init(ldlm_init);
 852 module_exit(ldlm_exit);