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