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-2004 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 the Lustre file system, http://www.lustre.org
   9  *   Lustre is a trademark of Cluster File Systems, Inc.
  10  *
  11  *   You may have signed or agreed to another license before downloading
  12  *   this software.  If so, you are bound by the terms and conditions
  13  *   of that agreement, and the following does not apply to you.  See the
  14  *   LICENSE file included with this distribution for more information.
  15  *
  16  *   If you did not agree to a different license, then this copy of Lustre
  17  *   is open source software; you can redistribute it and/or modify it
  18  *   under the terms of version 2 of the GNU General Public License as
  19  *   published by the Free Software Foundation.
  20  *
  21  *   In either case, Lustre is distributed in the hope that it will be
  22  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  23  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  24  *   license text for more details.
  25  */
  26
  27 #ifndef EXPORT_SYMTAB
  28 # define EXPORT_SYMTAB
  29 #endif
  30 #define DEBUG_SUBSYSTEM S_LDLM
  31
  32 #ifdef __KERNEL__
  33 # include <libcfs/libcfs.h>
  34 #else
  35 # include <liblustre.h>
  36 #endif
  37
  38 #include <lustre_dlm.h>
  39 #include <obd_class.h>
  40 #include <libcfs/list.h>
  41 #include "ldlm_internal.h"
  42
  43 extern cfs_mem_cache_t *ldlm_resource_slab;
  44 extern cfs_mem_cache_t *ldlm_lock_slab;
  45 extern struct lustre_lock ldlm_handle_lock;
  46 extern struct list_head ldlm_namespace_list;
  47
  48 extern struct semaphore ldlm_namespace_lock;
  49 static struct semaphore ldlm_ref_sem;
  50 static int ldlm_refcount;
  51
  52 /* LDLM state */
  53
  54 static struct ldlm_state *ldlm_state;
  55
  56 inline cfs_time_t round_timeout(cfs_time_t timeout)
  57 {
  58         return cfs_time_seconds((int)cfs_duration_sec(cfs_time_sub(timeout, 0)) + 1);
  59 }
  60
  61 /* timeout for initial callback (AST) reply */
  62 static inline unsigned int ldlm_get_rq_timeout(unsigned int ldlm_timeout,
  63                                                unsigned int obd_timeout)
  64 {
  65         unsigned int timeout = min(ldlm_timeout, obd_timeout / 3);
  66
  67         return timeout < 1 ? 1 : timeout;
  68 }
  69
  70 #ifdef __KERNEL__
  71 /* w_l_spinlock protects both waiting_locks_list and expired_lock_thread */
  72 static spinlock_t waiting_locks_spinlock;   /* BH lock (timer) */
  73 static struct list_head waiting_locks_list;
  74 static cfs_timer_t waiting_locks_timer;
  75
  76 static struct expired_lock_thread {
  77         cfs_waitq_t               elt_waitq;
  78         int                       elt_state;
  79         int                       elt_dump;
  80         struct list_head          elt_expired_locks;
  81 } expired_lock_thread;
  82 #endif
  83
  84 #define ELT_STOPPED   0
  85 #define ELT_READY     1
  86 #define ELT_TERMINATE 2
  87
  88 struct ldlm_bl_pool {
  89         spinlock_t              blp_lock;
  90         struct list_head        blp_list;
  91         cfs_waitq_t             blp_waitq;
  92         atomic_t                blp_num_threads;
  93         struct completion       blp_comp;
  94 };
  95
  96 struct ldlm_bl_work_item {
  97         struct list_head        blwi_entry;
  98         struct ldlm_namespace   *blwi_ns;
  99         struct ldlm_lock_desc   blwi_ld;
 100         struct ldlm_lock        *blwi_lock;
 101         int                     blwi_flags;
 102 };
 103
 104 #ifdef __KERNEL__
 105
 106 static inline int have_expired_locks(void)
 107 {
 108         int need_to_run;
 109
 110         ENTRY;
 111         spin_lock_bh(&waiting_locks_spinlock);
 112         need_to_run = !list_empty(&expired_lock_thread.elt_expired_locks);
 113         spin_unlock_bh(&waiting_locks_spinlock);
 114
 115         RETURN(need_to_run);
 116 }
 117
 118 static int expired_lock_main(void *arg)
 119 {
 120         struct list_head *expired = &expired_lock_thread.elt_expired_locks;
 121         struct l_wait_info lwi = { 0 };
 122         int do_dump;
 123
 124         ENTRY;
 125         cfs_daemonize("ldlm_elt");
 126
 127         expired_lock_thread.elt_state = ELT_READY;
 128         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 129
 130         while (1) {
 131                 l_wait_event(expired_lock_thread.elt_waitq,
 132                              have_expired_locks() ||
 133                              expired_lock_thread.elt_state == ELT_TERMINATE,
 134                              &lwi);
 135
 136                 spin_lock_bh(&waiting_locks_spinlock);
 137                 if (expired_lock_thread.elt_dump) {
 138                         spin_unlock_bh(&waiting_locks_spinlock);
 139
 140                         /* from waiting_locks_callback, but not in timer */
 141                         libcfs_debug_dumplog();
 142                         libcfs_run_lbug_upcall(__FILE__,
 143                                                 "waiting_locks_callback",
 144                                                 expired_lock_thread.elt_dump);
 145
 146                         spin_lock_bh(&waiting_locks_spinlock);
 147                         expired_lock_thread.elt_dump = 0;
 148                 }
 149
 150                 do_dump = 0;
 151
 152                 while (!list_empty(expired)) {
 153                         struct obd_export *export;
 154                         struct ldlm_lock *lock;
 155
 156                         lock = list_entry(expired->next, struct ldlm_lock,
 157                                           l_pending_chain);
 158                         if ((void *)lock < LP_POISON + CFS_PAGE_SIZE &&
 159                             (void *)lock >= LP_POISON) {
 160                                 spin_unlock_bh(&waiting_locks_spinlock);
 161                                 CERROR("free lock on elt list %p\n", lock);
 162                                 LBUG();
 163                         }
 164                         list_del_init(&lock->l_pending_chain);
 165                         if ((void *)lock->l_export < LP_POISON + CFS_PAGE_SIZE &&
 166                             (void *)lock->l_export >= LP_POISON) {
 167                                 CERROR("lock with free export on elt list %p\n",
 168                                        lock->l_export);
 169                                 lock->l_export = NULL;
 170                                 LDLM_ERROR(lock, "free export");
 171                                 continue;
 172                         }
 173                         export = class_export_get(lock->l_export);
 174                         spin_unlock_bh(&waiting_locks_spinlock);
 175
 176                         do_dump++;
 177                         class_fail_export(export);
 178                         class_export_put(export);
 179                         spin_lock_bh(&waiting_locks_spinlock);
 180                 }
 181                 spin_unlock_bh(&waiting_locks_spinlock);
 182
 183                 if (do_dump && obd_dump_on_eviction) {
 184                         CERROR("dump the log upon eviction\n");
 185                         libcfs_debug_dumplog();
 186                 }
 187
 188                 if (expired_lock_thread.elt_state == ELT_TERMINATE)
 189                         break;
 190         }
 191
 192         expired_lock_thread.elt_state = ELT_STOPPED;
 193         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 194         RETURN(0);
 195 }
 196
 197 /* This is called from within a timer interrupt and cannot schedule */
 198 static void waiting_locks_callback(unsigned long unused)
 199 {
 200         struct ldlm_lock *lock, *last = NULL;
 201
 202         spin_lock_bh(&waiting_locks_spinlock);
 203         while (!list_empty(&waiting_locks_list)) {
 204                 lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
 205                                   l_pending_chain);
 206
 207                 if (cfs_time_after(lock->l_callback_timeout, cfs_time_current()) ||
 208                     (lock->l_req_mode == LCK_GROUP))
 209                         break;
 210
 211                 LDLM_ERROR(lock, "lock callback timer expired: evicting client "
 212                            "%s@%s nid %s ",lock->l_export->exp_client_uuid.uuid,
 213                            lock->l_export->exp_connection->c_remote_uuid.uuid,
 214                            libcfs_nid2str(lock->l_export->exp_connection->c_peer.nid));
 215
 216                 if (lock == last) {
 217                         LDLM_ERROR(lock, "waiting on lock multiple times");
 218                         CERROR("wll %p n/p %p/%p, l_pending %p n/p %p/%p\n",
 219                                &waiting_locks_list,
 220                                waiting_locks_list.next, waiting_locks_list.prev,
 221                                &lock->l_pending_chain,
 222                                lock->l_pending_chain.next,
 223                                lock->l_pending_chain.prev);
 224
 225                         CFS_INIT_LIST_HEAD(&waiting_locks_list);    /* HACK */
 226                         expired_lock_thread.elt_dump = __LINE__;
 227
 228                         /* LBUG(); */
 229                         CEMERG("would be an LBUG, but isn't (bug 5653)\n");
 230                         libcfs_debug_dumpstack(NULL);
 231                         /*blocks* libcfs_debug_dumplog(); */
 232                         /*blocks* libcfs_run_lbug_upcall(file, func, line); */
 233                         break;
 234                 }
 235                 last = lock;
 236
 237                 list_del(&lock->l_pending_chain);
 238                 list_add(&lock->l_pending_chain,
 239                          &expired_lock_thread.elt_expired_locks);
 240         }
 241
 242         if (!list_empty(&expired_lock_thread.elt_expired_locks)) {
 243                 if (obd_dump_on_timeout)
 244                         expired_lock_thread.elt_dump = __LINE__;
 245
 246                 cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 247         }
 248
 249         /*
 250          * Make sure the timer will fire again if we have any locks
 251          * left.
 252          */
 253         if (!list_empty(&waiting_locks_list)) {
 254                 cfs_time_t timeout_rounded;
 255                 lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
 256                                   l_pending_chain);
 257                 timeout_rounded = (cfs_time_t)round_timeout(lock->l_callback_timeout);
 258                 cfs_timer_arm(&waiting_locks_timer, timeout_rounded);
 259         }
 260         spin_unlock_bh(&waiting_locks_spinlock);
 261 }
 262
 263 /*
 264  * Indicate that we're waiting for a client to call us back cancelling a given
 265  * lock.  We add it to the pending-callback chain, and schedule the lock-timeout
 266  * timer to fire appropriately.  (We round up to the next second, to avoid
 267  * floods of timer firings during periods of high lock contention and traffic).
 268  *
 269  * Called with the namespace lock held.
 270  */
 271 static int __ldlm_add_waiting_lock(struct ldlm_lock *lock)
 272 {
 273         cfs_time_t timeout_rounded;
 274
 275         if (!list_empty(&lock->l_pending_chain))
 276                 return 0;
 277
 278         lock->l_callback_timeout =cfs_time_add(cfs_time_current(),
 279                                                cfs_time_seconds(obd_timeout)/2);
 280
 281         timeout_rounded = round_timeout(lock->l_callback_timeout);
 282
 283         if (cfs_time_before(timeout_rounded, cfs_timer_deadline(&waiting_locks_timer)) ||
 284             !cfs_timer_is_armed(&waiting_locks_timer)) {
 285                 cfs_timer_arm(&waiting_locks_timer, timeout_rounded);
 286
 287         }
 288         list_add_tail(&lock->l_pending_chain, &waiting_locks_list); /* FIFO */
 289         return 1;
 290 }
 291
 292 static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
 293 {
 294         int ret;
 295
 296         LASSERT(!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK));
 297
 298         spin_lock_bh(&waiting_locks_spinlock);
 299         if (lock->l_destroyed) {
 300                 static cfs_time_t next;
 301                 spin_unlock_bh(&waiting_locks_spinlock);
 302                 LDLM_ERROR(lock, "not waiting on destroyed lock (bug 5653)");
 303                 if (cfs_time_after(cfs_time_current(), next)) {
 304                         next = cfs_time_shift(14400);
 305                         libcfs_debug_dumpstack(NULL);
 306                 }
 307                 return 0;
 308         }
 309
 310         ret = __ldlm_add_waiting_lock(lock);
 311         spin_unlock_bh(&waiting_locks_spinlock);
 312
 313         LDLM_DEBUG(lock, "%sadding to wait list",
 314                    ret == 0 ? "not re-" : "");
 315         return ret;
 316 }
 317
 318 /*
 319  * Remove a lock from the pending list, likely because it had its cancellation
 320  * callback arrive without incident.  This adjusts the lock-timeout timer if
 321  * needed.  Returns 0 if the lock wasn't pending after all, 1 if it was.
 322  *
 323  * Called with namespace lock held.
 324  */
 325 int __ldlm_del_waiting_lock(struct ldlm_lock *lock)
 326 {
 327         struct list_head *list_next;
 328
 329         if (list_empty(&lock->l_pending_chain))
 330                 return 0;
 331
 332         list_next = lock->l_pending_chain.next;
 333         if (lock->l_pending_chain.prev == &waiting_locks_list) {
 334                 /* Removing the head of the list, adjust timer. */
 335                 if (list_next == &waiting_locks_list) {
 336                         /* No more, just cancel. */
 337                         cfs_timer_disarm(&waiting_locks_timer);
 338                 } else {
 339                         struct ldlm_lock *next;
 340                         next = list_entry(list_next, struct ldlm_lock,
 341                                           l_pending_chain);
 342                         cfs_timer_arm(&waiting_locks_timer,
 343                                       round_timeout(next->l_callback_timeout));
 344                 }
 345         }
 346         list_del_init(&lock->l_pending_chain);
 347
 348         return 1;
 349 }
 350
 351 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
 352 {
 353         int ret;
 354
 355         if (lock->l_export == NULL) {
 356                 /* We don't have a "waiting locks list" on clients. */
 357                 LDLM_DEBUG(lock, "client lock: no-op");
 358                 return 0;
 359         }
 360
 361         spin_lock_bh(&waiting_locks_spinlock);
 362         ret = __ldlm_del_waiting_lock(lock);
 363         spin_unlock_bh(&waiting_locks_spinlock);
 364
 365         LDLM_DEBUG(lock, "%s", ret == 0 ? "wasn't waiting" : "removed");
 366         return ret;
 367 }
 368
 369 /*
 370  * Prolong the lock
 371  *
 372  * Called with namespace lock held.
 373  */
 374 int ldlm_refresh_waiting_lock(struct ldlm_lock *lock)
 375 {
 376         if (lock->l_export == NULL) {
 377                 /* We don't have a "waiting locks list" on clients. */
 378                 LDLM_DEBUG(lock, "client lock: no-op");
 379                 return 0;
 380         }
 381
 382         spin_lock_bh(&waiting_locks_spinlock);
 383
 384         if (list_empty(&lock->l_pending_chain)) {
 385                 spin_unlock_bh(&waiting_locks_spinlock);
 386                 LDLM_DEBUG(lock, "wasn't waiting");
 387                 return 0;
 388         }
 389
 390         __ldlm_del_waiting_lock(lock);
 391         __ldlm_add_waiting_lock(lock);
 392         spin_unlock_bh(&waiting_locks_spinlock);
 393
 394         LDLM_DEBUG(lock, "refreshed");
 395         return 1;
 396 }
 397
 398 #else /* !__KERNEL__ */
 399
 400 static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
 401 {
 402         LASSERT(!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK));
 403         RETURN(1);
 404 }
 405
 406 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
 407 {
 408         RETURN(0);
 409 }
 410
 411 int ldlm_refresh_waiting_lock(struct ldlm_lock *lock)
 412 {
 413         RETURN(0);
 414 }
 415 #endif /* __KERNEL__ */
 416
 417 static void ldlm_failed_ast(struct ldlm_lock *lock, int rc,
 418                             const char *ast_type)
 419 {
 420         struct ptlrpc_connection *conn = lock->l_export->exp_connection;
 421         char                     *str = libcfs_nid2str(conn->c_peer.nid);
 422
 423         LCONSOLE_ERROR_MSG(0x138, "A client on nid %s was evicted from "
 424                            "service %s.\n", str,
 425                            lock->l_export->exp_obd->obd_name);
 426
 427         LCONSOLE_ERROR_MSG(0x012, "Lock %s callback to %s timed out for "
 428                            "resource %d\n", ast_type,
 429                            obd_export_nid2str(lock->l_export), rc);
 430
 431         if (obd_dump_on_timeout)
 432                 libcfs_debug_dumplog();
 433         class_fail_export(lock->l_export);
 434 }
 435
 436 static int ldlm_handle_ast_error(struct ldlm_lock *lock,
 437                                  struct ptlrpc_request *req, int rc,
 438                                  const char *ast_type)
 439 {
 440         lnet_process_id_t peer = req->rq_import->imp_connection->c_peer;
 441
 442         if (rc == -ETIMEDOUT || rc == -EINTR || rc == -ENOTCONN) {
 443                 LASSERT(lock->l_export);
 444                 if (lock->l_export->exp_libclient) {
 445                         LDLM_DEBUG(lock, "%s AST to liblustre client (nid %s)"
 446                                    " timeout, just cancelling lock", ast_type,
 447                                    libcfs_nid2str(peer.nid));
 448                         ldlm_lock_cancel(lock);
 449                         rc = -ERESTART;
 450                 } else if (lock->l_flags & LDLM_FL_CANCEL) {
 451                         LDLM_DEBUG(lock, "%s AST timeout from nid %s, but "
 452                                    "cancel was received (AST reply lost?)",
 453                                    ast_type, libcfs_nid2str(peer.nid));
 454                         ldlm_lock_cancel(lock);
 455                         rc = -ERESTART;
 456                 } else {
 457                         ldlm_del_waiting_lock(lock);
 458                         ldlm_failed_ast(lock, rc, ast_type);
 459                 }
 460         } else if (rc) {
 461                 if (rc == -EINVAL)
 462                         LDLM_DEBUG(lock, "client (nid %s) returned %d"
 463                                    " from %s AST - normal race",
 464                                    libcfs_nid2str(peer.nid),
 465                                    lustre_msg_get_status(req->rq_repmsg),
 466                                    ast_type);
 467                 else
 468                         LDLM_ERROR(lock, "client (nid %s) returned %d "
 469                                    "from %s AST", libcfs_nid2str(peer.nid),
 470                                    (req->rq_repmsg != NULL) ?
 471                                    lustre_msg_get_status(req->rq_repmsg) : 0,
 472                                    ast_type);
 473                 ldlm_lock_cancel(lock);
 474                 /* Server-side AST functions are called from ldlm_reprocess_all,
 475                  * which needs to be told to please restart its reprocessing. */
 476                 rc = -ERESTART;
 477         }
 478
 479         return rc;
 480 }
 481
 482 int ldlm_server_blocking_ast(struct ldlm_lock *lock,
 483                              struct ldlm_lock_desc *desc,
 484                              void *data, int flag)
 485 {
 486         struct ldlm_request *body;
 487         struct ptlrpc_request *req;
 488         int size[] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 489                        [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 490         int instant_cancel = 0, rc = 0;
 491         ENTRY;
 492
 493         if (flag == LDLM_CB_CANCELING) {
 494                 /* Don't need to do anything here. */
 495                 RETURN(0);
 496         }
 497
 498         LASSERT(lock);
 499
 500         req = ptlrpc_prep_req(lock->l_export->exp_imp_reverse,
 501                               LUSTRE_DLM_VERSION, LDLM_BL_CALLBACK, 2, size,
 502                               NULL);
 503         if (req == NULL)
 504                 RETURN(-ENOMEM);
 505
 506         lock_res(lock->l_resource);
 507         if (lock->l_granted_mode != lock->l_req_mode) {
 508                 /* this blocking AST will be communicated as part of the
 509                  * completion AST instead */
 510                 unlock_res(lock->l_resource);
 511                 ptlrpc_req_finished(req);
 512                 LDLM_DEBUG(lock, "lock not granted, not sending blocking AST");
 513                 RETURN(0);
 514         }
 515
 516         if (lock->l_destroyed) {
 517                 /* What's the point? */
 518                 unlock_res(lock->l_resource);
 519                 ptlrpc_req_finished(req);
 520                 RETURN(0);
 521         }
 522
 523 #if 0
 524         if (CURRENT_SECONDS - lock->l_export->exp_last_request_time > 30){
 525                 unlock_res(lock->l_resource);
 526                 ptlrpc_req_finished(req);
 527                 ldlm_failed_ast(lock, -ETIMEDOUT, "Not-attempted blocking");
 528                 RETURN(-ETIMEDOUT);
 529         }
 530 #endif
 531
 532         if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)
 533                 instant_cancel = 1;
 534
 535         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 536         body->lock_handle[0] = lock->l_remote_handle;
 537         body->lock_desc = *desc;
 538         body->lock_flags |= (lock->l_flags & LDLM_AST_FLAGS);
 539
 540         LDLM_DEBUG(lock, "server preparing blocking AST");
 541
 542         ptlrpc_req_set_repsize(req, 1, NULL);
 543         if (instant_cancel) {
 544                 unlock_res(lock->l_resource);
 545                 ldlm_lock_cancel(lock);
 546         } else if (lock->l_granted_mode == lock->l_req_mode) {
 547                 ldlm_add_waiting_lock(lock);
 548                 unlock_res(lock->l_resource);
 549         }
 550
 551         req->rq_send_state = LUSTRE_IMP_FULL;
 552         req->rq_timeout = ldlm_get_rq_timeout(ldlm_timeout, obd_timeout);
 553
 554         if (lock->l_export && lock->l_export->exp_ldlm_stats)
 555                 lprocfs_counter_incr(lock->l_export->exp_ldlm_stats,
 556                                      LDLM_BL_CALLBACK - LDLM_FIRST_OPC);
 557
 558         if (unlikely(instant_cancel)) {
 559                 rc = ptl_send_rpc(req, 1);
 560         } else {
 561                 rc = ptlrpc_queue_wait(req);
 562                 OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_GLIMPSE, 2);
 563         }
 564         if (rc != 0) {
 565                 /* If client canceled the lock but the cancel has not been
 566                  * recieved yet, we need to update lvbo to have the proper
 567                  * attributes cached. */
 568                 if (rc == -EINVAL)
 569                         ldlm_res_lvbo_update(lock->l_resource, NULL, 0, 1);
 570                 rc = ldlm_handle_ast_error(lock, req, rc, "blocking");
 571         }
 572
 573         ptlrpc_req_finished(req);
 574
 575         /* If we cancelled the lock, we need to restart ldlm_reprocess_queue */
 576         if (!rc && instant_cancel)
 577                 rc = -ERESTART;
 578
 579         RETURN(rc);
 580 }
 581
 582 int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
 583 {
 584         struct ldlm_request *body;
 585         struct ptlrpc_request *req;
 586         struct timeval granted_time;
 587         long total_enqueue_wait;
 588         int size[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 589                         [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 590         int rc = 0, buffers = 2, instant_cancel = 0;
 591         ENTRY;
 592
 593         LASSERT(lock != NULL);
 594
 595         do_gettimeofday(&granted_time);
 596         total_enqueue_wait = cfs_timeval_sub(&granted_time,
 597                                              &lock->l_enqueued_time, NULL);
 598
 599         if (total_enqueue_wait / 1000000 > obd_timeout)
 600                 LDLM_ERROR(lock, "enqueue wait took %luus from %lu",
 601                            total_enqueue_wait, lock->l_enqueued_time.tv_sec);
 602
 603         lock_res_and_lock(lock);
 604         if (lock->l_resource->lr_lvb_len) {
 605                 size[DLM_REQ_REC_OFF] = lock->l_resource->lr_lvb_len;
 606                 buffers = 3;
 607         }
 608         unlock_res_and_lock(lock);
 609
 610         req = ptlrpc_prep_req(lock->l_export->exp_imp_reverse,
 611                               LUSTRE_DLM_VERSION, LDLM_CP_CALLBACK, buffers,
 612                               size, NULL);
 613         if (req == NULL)
 614                 RETURN(-ENOMEM);
 615
 616         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 617         body->lock_handle[0] = lock->l_remote_handle;
 618         body->lock_flags = flags;
 619         ldlm_lock2desc(lock, &body->lock_desc);
 620
 621         if (buffers == 3) {
 622                 void *lvb;
 623
 624                 lvb = lustre_msg_buf(req->rq_reqmsg, DLM_REQ_REC_OFF,
 625                                      lock->l_resource->lr_lvb_len);
 626                 lock_res_and_lock(lock);
 627                 memcpy(lvb, lock->l_resource->lr_lvb_data,
 628                        lock->l_resource->lr_lvb_len);
 629                 unlock_res_and_lock(lock);
 630         }
 631
 632         LDLM_DEBUG(lock, "server preparing completion AST (after %ldus wait)",
 633                    total_enqueue_wait);
 634
 635         ptlrpc_req_set_repsize(req, 1, NULL);
 636
 637         req->rq_send_state = LUSTRE_IMP_FULL;
 638         req->rq_timeout = ldlm_get_rq_timeout(ldlm_timeout, obd_timeout);
 639
 640         /* We only send real blocking ASTs after the lock is granted */
 641         lock_res_and_lock(lock);
 642         if (lock->l_flags & LDLM_FL_AST_SENT) {
 643                 body->lock_flags |= LDLM_FL_AST_SENT;
 644
 645                 /* We might get here prior to ldlm_handle_enqueue setting
 646                  * LDLM_FL_CANCEL_ON_BLOCK flag. Then we will put this lock
 647                  * into waiting list, but this is safe and similar code in
 648                  * ldlm_handle_enqueue will call ldlm_lock_cancel() still,
 649                  * that would not only cancel the lock, but will also remove
 650                  * it from waiting list */
 651                 if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) {
 652                         unlock_res_and_lock(lock);
 653                         ldlm_lock_cancel(lock);
 654                         instant_cancel = 1;
 655                         lock_res_and_lock(lock);
 656                 } else {
 657                         ldlm_add_waiting_lock(lock); /* start the lock-timeout
 658                                                          clock */
 659                 }
 660         }
 661         unlock_res_and_lock(lock);
 662
 663         if (lock->l_export && lock->l_export->exp_ldlm_stats)
 664                 lprocfs_counter_incr(lock->l_export->exp_ldlm_stats,
 665                                      LDLM_CP_CALLBACK - LDLM_FIRST_OPC);
 666
 667         rc = ptlrpc_queue_wait(req);
 668         if (rc != 0)
 669                 rc = ldlm_handle_ast_error(lock, req, rc, "completion");
 670
 671         ptlrpc_req_finished(req);
 672
 673         /* If we cancelled the lock, we need to restart ldlm_reprocess_queue */
 674         if (!rc && instant_cancel)
 675                 rc = -ERESTART;
 676
 677         RETURN(rc);
 678 }
 679
 680 int ldlm_server_glimpse_ast(struct ldlm_lock *lock, void *data)
 681 {
 682         struct ldlm_resource *res = lock->l_resource;
 683         struct ldlm_request *body;
 684         struct ptlrpc_request *req;
 685         int size[] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 686                        [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 687         int rc = 0;
 688         ENTRY;
 689
 690         LASSERT(lock != NULL);
 691
 692         req = ptlrpc_prep_req(lock->l_export->exp_imp_reverse,
 693                               LUSTRE_DLM_VERSION, LDLM_GL_CALLBACK, 2, size,
 694                               NULL);
 695         if (req == NULL)
 696                 RETURN(-ENOMEM);
 697
 698         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 699         body->lock_handle[0] = lock->l_remote_handle;
 700         ldlm_lock2desc(lock, &body->lock_desc);
 701
 702         lock_res_and_lock(lock);
 703         size[REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
 704         unlock_res_and_lock(lock);
 705         res = lock->l_resource;
 706         ptlrpc_req_set_repsize(req, 2, size);
 707
 708         req->rq_send_state = LUSTRE_IMP_FULL;
 709         req->rq_timeout = ldlm_get_rq_timeout(ldlm_timeout, obd_timeout);
 710
 711         if (lock->l_export && lock->l_export->exp_ldlm_stats)
 712                 lprocfs_counter_incr(lock->l_export->exp_ldlm_stats,
 713                                      LDLM_GL_CALLBACK - LDLM_FIRST_OPC);
 714
 715         rc = ptlrpc_queue_wait(req);
 716         if (rc == -ELDLM_NO_LOCK_DATA)
 717                 LDLM_DEBUG(lock, "lost race - client has a lock but no inode");
 718         else if (rc != 0)
 719                 rc = ldlm_handle_ast_error(lock, req, rc, "glimpse");
 720         else
 721                 rc = ldlm_res_lvbo_update(res, req->rq_repmsg,
 722                                           REPLY_REC_OFF, 1);
 723         ptlrpc_req_finished(req);
 724         RETURN(rc);
 725 }
 726
 727 static struct ldlm_lock *
 728 find_existing_lock(struct obd_export *exp, struct lustre_handle *remote_hdl)
 729 {
 730         struct list_head *iter;
 731
 732         spin_lock(&exp->exp_ldlm_data.led_lock);
 733         list_for_each(iter, &exp->exp_ldlm_data.led_held_locks) {
 734                 struct ldlm_lock *lock;
 735                 lock = list_entry(iter, struct ldlm_lock, l_export_chain);
 736                 if (lock->l_remote_handle.cookie == remote_hdl->cookie) {
 737                         LDLM_LOCK_GET(lock);
 738                         spin_unlock(&exp->exp_ldlm_data.led_lock);
 739                         return lock;
 740                 }
 741         }
 742         spin_unlock(&exp->exp_ldlm_data.led_lock);
 743         return NULL;
 744 }
 745
 746
 747 /*
 748  * Main server-side entry point into LDLM. This is called by ptlrpc service
 749  * threads to carry out client lock enqueueing requests.
 750  */
 751 int ldlm_handle_enqueue(struct ptlrpc_request *req,
 752                         ldlm_completion_callback completion_callback,
 753                         ldlm_blocking_callback blocking_callback,
 754                         ldlm_glimpse_callback glimpse_callback)
 755 {
 756         struct obd_device *obddev = req->rq_export->exp_obd;
 757         struct ldlm_reply *dlm_rep;
 758         struct ldlm_request *dlm_req;
 759         int size[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 760                         [DLM_LOCKREPLY_OFF]   = sizeof(*dlm_rep) };
 761         int rc = 0;
 762         __u32 flags;
 763         ldlm_error_t err = ELDLM_OK;
 764         struct ldlm_lock *lock = NULL;
 765         void *cookie = NULL;
 766         ENTRY;
 767
 768         LDLM_DEBUG_NOLOCK("server-side enqueue handler START");
 769
 770         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
 771                                      lustre_swab_ldlm_request);
 772         if (dlm_req == NULL) {
 773                 CERROR ("Can't unpack dlm_req\n");
 774                 GOTO(out, rc = -EFAULT);
 775         }
 776
 777         ldlm_request_cancel(req, dlm_req, LDLM_ENQUEUE_CANCEL_OFF);
 778         flags = dlm_req->lock_flags;
 779
 780         LASSERT(req->rq_export);
 781
 782         if (req->rq_export->exp_ldlm_stats)
 783                 lprocfs_counter_incr(req->rq_export->exp_ldlm_stats,
 784                                      LDLM_ENQUEUE - LDLM_FIRST_OPC);
 785
 786         if (dlm_req->lock_desc.l_resource.lr_type < LDLM_MIN_TYPE ||
 787             dlm_req->lock_desc.l_resource.lr_type >= LDLM_MAX_TYPE) {
 788                 DEBUG_REQ(D_ERROR, req, "invalid lock request type %d\n",
 789                           dlm_req->lock_desc.l_resource.lr_type);
 790                 GOTO(out, rc = -EFAULT);
 791         }
 792
 793         if (dlm_req->lock_desc.l_req_mode <= LCK_MINMODE ||
 794             dlm_req->lock_desc.l_req_mode >= LCK_MAXMODE ||
 795             dlm_req->lock_desc.l_req_mode & (dlm_req->lock_desc.l_req_mode-1)) {
 796                 DEBUG_REQ(D_ERROR, req, "invalid lock request mode %d\n",
 797                           dlm_req->lock_desc.l_req_mode);
 798                 GOTO(out, rc = -EFAULT);
 799         }
 800
 801         if (req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS) {
 802                 if (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN) {
 803                         DEBUG_REQ(D_ERROR, req,
 804                                   "PLAIN lock request from IBITS client?\n");
 805                         GOTO(out, rc = -EPROTO);
 806                 }
 807         } else if (dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS) {
 808                 DEBUG_REQ(D_ERROR, req,
 809                           "IBITS lock request from unaware client?\n");
 810                 GOTO(out, rc = -EPROTO);
 811         }
 812
 813 #if 0
 814         /* FIXME this makes it impossible to use LDLM_PLAIN locks -- check
 815            against server's _CONNECT_SUPPORTED flags? (I don't want to use
 816            ibits for mgc/mgs) */
 817
 818         /* INODEBITS_INTEROP: Perform conversion from plain lock to
 819          * inodebits lock if client does not support them. */
 820         if (!(req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS) &&
 821             (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN)) {
 822                 dlm_req->lock_desc.l_resource.lr_type = LDLM_IBITS;
 823                 dlm_req->lock_desc.l_policy_data.l_inodebits.bits =
 824                         MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE;
 825                 if (dlm_req->lock_desc.l_req_mode == LCK_PR)
 826                         dlm_req->lock_desc.l_req_mode = LCK_CR;
 827         }
 828 #endif
 829
 830         if (flags & LDLM_FL_REPLAY) {
 831                 lock = find_existing_lock(req->rq_export,
 832                                           &dlm_req->lock_handle[0]);
 833                 if (lock != NULL) {
 834                         DEBUG_REQ(D_HA, req, "found existing lock cookie "LPX64,
 835                                   lock->l_handle.h_cookie);
 836                         GOTO(existing_lock, rc = 0);
 837                 }
 838         }
 839
 840         /* The lock's callback data might be set in the policy function */
 841         lock = ldlm_lock_create(obddev->obd_namespace,
 842                                 dlm_req->lock_desc.l_resource.lr_name,
 843                                 dlm_req->lock_desc.l_resource.lr_type,
 844                                 dlm_req->lock_desc.l_req_mode,
 845                                 blocking_callback, completion_callback,
 846                                 glimpse_callback, NULL, 0);
 847         if (!lock)
 848                 GOTO(out, rc = -ENOMEM);
 849
 850         do_gettimeofday(&lock->l_enqueued_time);
 851         lock->l_remote_handle = dlm_req->lock_handle[0];
 852         LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
 853
 854         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_BLOCKED, obd_timeout * 2);
 855         /* Don't enqueue a lock onto the export if it has already
 856          * been evicted.  Cancel it now instead. (bug 3822) */
 857         if (req->rq_export->exp_failed) {
 858                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
 859                 GOTO(out, rc = -ENOTCONN);
 860         }
 861         lock->l_export = class_export_get(req->rq_export);
 862         spin_lock(&lock->l_export->exp_ldlm_data.led_lock);
 863         list_add(&lock->l_export_chain,
 864                  &lock->l_export->exp_ldlm_data.led_held_locks);
 865         spin_unlock(&lock->l_export->exp_ldlm_data.led_lock);
 866
 867 existing_lock:
 868
 869         if (flags & LDLM_FL_HAS_INTENT) {
 870                 /* In this case, the reply buffer is allocated deep in
 871                  * local_lock_enqueue by the policy function. */
 872                 cookie = req;
 873         } else {
 874                 int buffers = 2;
 875
 876                 lock_res_and_lock(lock);
 877                 if (lock->l_resource->lr_lvb_len) {
 878                         size[DLM_REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
 879                         buffers = 3;
 880                 }
 881                 unlock_res_and_lock(lock);
 882
 883                 if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
 884                         GOTO(out, rc = -ENOMEM);
 885
 886                 rc = lustre_pack_reply(req, buffers, size, NULL);
 887                 if (rc)
 888                         GOTO(out, rc);
 889         }
 890
 891         if (dlm_req->lock_desc.l_resource.lr_type != LDLM_PLAIN)
 892                 lock->l_policy_data = dlm_req->lock_desc.l_policy_data;
 893         if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT)
 894                 lock->l_req_extent = lock->l_policy_data.l_extent;
 895
 896         err = ldlm_lock_enqueue(obddev->obd_namespace, &lock, cookie, &flags);
 897         if (err)
 898                 GOTO(out, err);
 899
 900         dlm_rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
 901                                  sizeof(*dlm_rep));
 902         dlm_rep->lock_flags = flags;
 903
 904         ldlm_lock2desc(lock, &dlm_rep->lock_desc);
 905         ldlm_lock2handle(lock, &dlm_rep->lock_handle);
 906
 907         /* We never send a blocking AST until the lock is granted, but
 908          * we can tell it right now */
 909         lock_res_and_lock(lock);
 910
 911         /* Now take into account flags to be inherited from original lock
 912            request both in reply to client and in our own lock flags. */
 913         dlm_rep->lock_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
 914         lock->l_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
 915
 916         /* Don't move a pending lock onto the export if it has already
 917          * been evicted.  Cancel it now instead. (bug 5683) */
 918         if (req->rq_export->exp_failed ||
 919             OBD_FAIL_CHECK_ONCE(OBD_FAIL_LDLM_ENQUEUE_OLD_EXPORT)) {
 920                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
 921                 rc = -ENOTCONN;
 922         } else if (lock->l_flags & LDLM_FL_AST_SENT) {
 923                 dlm_rep->lock_flags |= LDLM_FL_AST_SENT;
 924                 if (lock->l_granted_mode == lock->l_req_mode) {
 925                         /* Only cancel lock if it was granted, because it
 926                          * would be destroyed immediatelly and would never
 927                          * be granted in the future, causing timeouts on client.
 928                          * Not granted lock will be cancelled immediatelly after
 929                          * sending completion AST.
 930                          */
 931                         if (dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK) {
 932                                 unlock_res_and_lock(lock);
 933                                 ldlm_lock_cancel(lock);
 934                                 lock_res_and_lock(lock);
 935                         } else if (lock->l_granted_mode == lock->l_req_mode)
 936                                 ldlm_add_waiting_lock(lock);
 937                 }
 938         }
 939         /* Make sure we never ever grant usual metadata locks to liblustre
 940            clients */
 941         if ((dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN ||
 942             dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS) &&
 943              req->rq_export->exp_libclient) {
 944                 if (!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) ||
 945                     !(dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK)) {
 946                         CERROR("Granting sync lock to libclient. "
 947                                "req fl %d, rep fl %d, lock fl %d\n",
 948                                dlm_req->lock_flags, dlm_rep->lock_flags,
 949                                lock->l_flags);
 950                         LDLM_ERROR(lock, "sync lock");
 951                         if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT) {
 952                                 struct ldlm_intent *it;
 953                                 it = lustre_msg_buf(req->rq_reqmsg,
 954                                                     DLM_INTENT_IT_OFF,
 955                                                     sizeof(*it));
 956                                 if (it != NULL) {
 957                                         CERROR("This is intent %s ("LPU64")\n",
 958                                                ldlm_it2str(it->opc), it->opc);
 959                                 }
 960                         }
 961                 }
 962         }
 963
 964         unlock_res_and_lock(lock);
 965
 966         EXIT;
 967  out:
 968         req->rq_status = err;
 969         if (req->rq_reply_state == NULL) {
 970                 err = lustre_pack_reply(req, 1, NULL, NULL);
 971                 if (rc == 0)
 972                         rc = err;
 973                 req->rq_status = rc;
 974         }
 975
 976         /* The LOCK_CHANGED code in ldlm_lock_enqueue depends on this
 977          * ldlm_reprocess_all.  If this moves, revisit that code. -phil */
 978         if (lock) {
 979                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
 980                            "(err=%d, rc=%d)", err, rc);
 981
 982                 if (rc == 0) {
 983                         lock_res_and_lock(lock);
 984                         size[DLM_REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
 985                         if (size[DLM_REPLY_REC_OFF] > 0) {
 986                                 void *lvb = lustre_msg_buf(req->rq_repmsg,
 987                                                        DLM_REPLY_REC_OFF,
 988                                                        size[DLM_REPLY_REC_OFF]);
 989                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
 990                                          req, lock);
 991
 992                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
 993                                        size[DLM_REPLY_REC_OFF]);
 994                         }
 995                         unlock_res_and_lock(lock);
 996                 } else {
 997                         lock_res_and_lock(lock);
 998                         ldlm_resource_unlink_lock(lock);
 999                         ldlm_lock_destroy_nolock(lock);
1000                         unlock_res_and_lock(lock);
1001                 }
1002
1003                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
1004                         ldlm_reprocess_all(lock->l_resource);
1005
1006                 LDLM_LOCK_PUT(lock);
1007         }
1008
1009         LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p, rc %d)",
1010                           lock, rc);
1011
1012         return rc;
1013 }
1014
1015 int ldlm_handle_convert(struct ptlrpc_request *req)
1016 {
1017         struct ldlm_request *dlm_req;
1018         struct ldlm_reply *dlm_rep;
1019         struct ldlm_lock *lock;
1020         int rc;
1021         int size[2] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
1022                         [DLM_LOCKREPLY_OFF]   = sizeof(*dlm_rep) };
1023         ENTRY;
1024
1025         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1026                                      lustre_swab_ldlm_request);
1027         if (dlm_req == NULL) {
1028                 CERROR ("Can't unpack dlm_req\n");
1029                 RETURN (-EFAULT);
1030         }
1031
1032         if (req->rq_export && req->rq_export->exp_ldlm_stats)
1033                 lprocfs_counter_incr(req->rq_export->exp_ldlm_stats,
1034                                      LDLM_CONVERT - LDLM_FIRST_OPC);
1035
1036         rc = lustre_pack_reply(req, 2, size, NULL);
1037         if (rc) {
1038                 CERROR("out of memory\n");
1039                 RETURN(-ENOMEM);
1040         }
1041         dlm_rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
1042                                  sizeof(*dlm_rep));
1043         dlm_rep->lock_flags = dlm_req->lock_flags;
1044
1045         lock = ldlm_handle2lock(&dlm_req->lock_handle[0]);
1046         if (!lock) {
1047                 req->rq_status = EINVAL;
1048         } else {
1049                 void *res = NULL;
1050
1051                 LDLM_DEBUG(lock, "server-side convert handler START");
1052
1053                 do_gettimeofday(&lock->l_enqueued_time);
1054                 res = ldlm_lock_convert(lock, dlm_req->lock_desc.l_req_mode,
1055                                         &dlm_rep->lock_flags);
1056                 if (res) {
1057                         if (ldlm_del_waiting_lock(lock))
1058                                 LDLM_DEBUG(lock, "converted waiting lock");
1059                         req->rq_status = 0;
1060                 } else {
1061                         req->rq_status = EDEADLOCK;
1062                 }
1063         }
1064
1065         if (lock) {
1066                 if (!req->rq_status)
1067                         ldlm_reprocess_all(lock->l_resource);
1068                 LDLM_DEBUG(lock, "server-side convert handler END");
1069                 LDLM_LOCK_PUT(lock);
1070         } else
1071                 LDLM_DEBUG_NOLOCK("server-side convert handler END");
1072
1073         RETURN(0);
1074 }
1075
1076 /* Cancel all the locks, which handles are packed into ldlm_request */
1077 int ldlm_request_cancel(struct ptlrpc_request *req,
1078                         struct ldlm_request *dlm_req, int first)
1079 {
1080         struct ldlm_resource *res, *pres = NULL;
1081         struct ldlm_lock *lock;
1082         int i, count, done = 0;
1083         ENTRY;
1084
1085         LDLM_DEBUG_NOLOCK("server-side cancel handler START: %d locks, "
1086                           "starting at %d", dlm_req->lock_count, first);
1087         count = dlm_req->lock_count ? dlm_req->lock_count : 1;
1088         if (first >= count)
1089                 RETURN(0);
1090
1091         /* There is no lock on the server at the replay time,
1092          * skip lock cancelling to make replay tests to pass. */
1093         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
1094                 RETURN(0);
1095
1096         for (i = first; i < count; i++) {
1097                 lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
1098                 if (!lock) {
1099                         LDLM_DEBUG_NOLOCK("server-side cancel handler stale "
1100                                           "lock (cookie "LPU64")",
1101                                           dlm_req->lock_handle[i].cookie);
1102                         continue;
1103                 }
1104
1105                 done++;
1106                 res = lock->l_resource;
1107                 if (res != pres) {
1108                         if (pres != NULL) {
1109                                 ldlm_reprocess_all(pres);
1110                                 ldlm_resource_putref(pres);
1111                         }
1112                         if (res != NULL) {
1113                                 ldlm_resource_getref(res);
1114                                 ldlm_res_lvbo_update(res, NULL, 0, 1);
1115                         }
1116                         pres = res;
1117                 }
1118                 ldlm_lock_cancel(lock);
1119                 LDLM_LOCK_PUT(lock);
1120         }
1121         if (pres != NULL) {
1122                 ldlm_reprocess_all(pres);
1123                 ldlm_resource_putref(pres);
1124         }
1125         LDLM_DEBUG_NOLOCK("server-side cancel handler END");
1126         RETURN(done);
1127 }
1128
1129 int ldlm_handle_cancel(struct ptlrpc_request *req)
1130 {
1131         struct ldlm_request *dlm_req;
1132         int rc;
1133         ENTRY;
1134
1135         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1136                                      lustre_swab_ldlm_request);
1137         if (dlm_req == NULL) {
1138                 CERROR("bad request buffer for cancel\n");
1139                 RETURN(-EFAULT);
1140         }
1141
1142         if (req->rq_export && req->rq_export->exp_ldlm_stats)
1143                 lprocfs_counter_incr(req->rq_export->exp_ldlm_stats,
1144                                      LDLM_CANCEL - LDLM_FIRST_OPC);
1145
1146         rc = lustre_pack_reply(req, 1, NULL, NULL);
1147         if (rc) {
1148                 CERROR("out of memory\n");
1149                 RETURN(-ENOMEM);
1150         }
1151
1152         if (!ldlm_request_cancel(req, dlm_req, 0))
1153                 req->rq_status = ESTALE;
1154
1155         if (ptlrpc_reply(req) != 0)
1156                 LBUG();
1157
1158         RETURN(0);
1159 }
1160
1161 void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
1162                              struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
1163 {
1164         int do_ast;
1165         ENTRY;
1166
1167         LDLM_DEBUG(lock, "client blocking AST callback handler START");
1168
1169         lock_res_and_lock(lock);
1170         lock->l_flags |= LDLM_FL_CBPENDING;
1171
1172         if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)
1173                 lock->l_flags |= LDLM_FL_CANCEL;
1174
1175         do_ast = (!lock->l_readers && !lock->l_writers);
1176         unlock_res_and_lock(lock);
1177
1178         if (do_ast) {
1179                 LDLM_DEBUG(lock, "already unused, calling "
1180                            "callback (%p)", lock->l_blocking_ast);
1181                 if (lock->l_blocking_ast != NULL)
1182                         lock->l_blocking_ast(lock, ld, lock->l_ast_data,
1183                                              LDLM_CB_BLOCKING);
1184         } else {
1185                 LDLM_DEBUG(lock, "Lock still has references, will be"
1186                            " cancelled later");
1187         }
1188
1189         LDLM_DEBUG(lock, "client blocking callback handler END");
1190         LDLM_LOCK_PUT(lock);
1191         EXIT;
1192 }
1193
1194 static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
1195                                     struct ldlm_namespace *ns,
1196                                     struct ldlm_request *dlm_req,
1197                                     struct ldlm_lock *lock)
1198 {
1199         CFS_LIST_HEAD(ast_list);
1200         ENTRY;
1201
1202         LDLM_DEBUG(lock, "client completion callback handler START");
1203
1204         lock_res_and_lock(lock);
1205
1206         /* If we receive the completion AST before the actual enqueue returned,
1207          * then we might need to switch lock modes, resources, or extents. */
1208         if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
1209                 lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
1210                 LDLM_DEBUG(lock, "completion AST, new lock mode");
1211         }
1212
1213         if (lock->l_resource->lr_type != LDLM_PLAIN) {
1214                 lock->l_policy_data = dlm_req->lock_desc.l_policy_data;
1215                 LDLM_DEBUG(lock, "completion AST, new policy data");
1216         }
1217
1218         ldlm_resource_unlink_lock(lock);
1219         if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
1220                    &lock->l_resource->lr_name,
1221                    sizeof(lock->l_resource->lr_name)) != 0) {
1222                 unlock_res_and_lock(lock);
1223                 ldlm_lock_change_resource(ns, lock,
1224                                          dlm_req->lock_desc.l_resource.lr_name);
1225                 LDLM_DEBUG(lock, "completion AST, new resource");
1226                 CERROR("change resource!\n");
1227                 lock_res_and_lock(lock);
1228         }
1229
1230         if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
1231                 lock->l_flags |= LDLM_FL_CBPENDING;
1232                 LDLM_DEBUG(lock, "completion AST includes blocking AST");
1233         }
1234
1235         if (lock->l_lvb_len) {
1236                 void *lvb;
1237                 lvb = lustre_swab_reqbuf(req, DLM_REQ_REC_OFF, lock->l_lvb_len,
1238                                          lock->l_lvb_swabber);
1239                 if (lvb == NULL) {
1240                         LDLM_ERROR(lock, "completion AST did not contain "
1241                                    "expected LVB!");
1242                 } else {
1243                         memcpy(lock->l_lvb_data, lvb, lock->l_lvb_len);
1244                 }
1245         }
1246
1247         ldlm_grant_lock(lock, &ast_list);
1248         unlock_res_and_lock(lock);
1249
1250         LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
1251
1252         ldlm_run_cp_ast_work(&ast_list);
1253
1254         LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
1255                           lock);
1256         LDLM_LOCK_PUT(lock);
1257         EXIT;
1258 }
1259
1260 static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
1261                                     struct ldlm_namespace *ns,
1262                                     struct ldlm_request *dlm_req,
1263                                     struct ldlm_lock *lock)
1264 {
1265         int rc = -ENOSYS;
1266         ENTRY;
1267
1268         LDLM_DEBUG(lock, "client glimpse AST callback handler");
1269
1270         if (lock->l_glimpse_ast != NULL)
1271                 rc = lock->l_glimpse_ast(lock, req);
1272
1273         if (req->rq_repmsg != NULL) {
1274                 ptlrpc_reply(req);
1275         } else {
1276                 req->rq_status = rc;
1277                 ptlrpc_error(req);
1278         }
1279
1280         lock_res_and_lock(lock);
1281         if (lock->l_granted_mode == LCK_PW &&
1282             !lock->l_readers && !lock->l_writers &&
1283             cfs_time_after(cfs_time_current(),
1284                            cfs_time_add(lock->l_last_used, cfs_time_seconds(10)))) {
1285                 unlock_res_and_lock(lock);
1286                 if (ldlm_bl_to_thread(ns, NULL, lock, 0))
1287                         ldlm_handle_bl_callback(ns, NULL, lock);
1288
1289                 EXIT;
1290                 return;
1291         }
1292         unlock_res_and_lock(lock);
1293         LDLM_LOCK_PUT(lock);
1294         EXIT;
1295 }
1296
1297 static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
1298 {
1299         req->rq_status = rc;
1300         if (req->rq_reply_state == NULL) {
1301                 rc = lustre_pack_reply(req, 1, NULL, NULL);
1302                 if (rc)
1303                         return rc;
1304         }
1305         return ptlrpc_reply(req);
1306 }
1307
1308 int ldlm_bl_to_thread(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
1309                       struct ldlm_lock *lock, int flags)
1310 {
1311 #ifdef __KERNEL__
1312         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
1313         struct ldlm_bl_work_item *blwi;
1314         ENTRY;
1315
1316         OBD_ALLOC(blwi, sizeof(*blwi));
1317         if (blwi == NULL)
1318                 RETURN(-ENOMEM);
1319
1320         blwi->blwi_ns = ns;
1321         if (ld != NULL)
1322                 blwi->blwi_ld = *ld;
1323         blwi->blwi_lock = lock;
1324         blwi->blwi_flags = flags;
1325
1326         spin_lock(&blp->blp_lock);
1327         list_add_tail(&blwi->blwi_entry, &blp->blp_list);
1328         cfs_waitq_signal(&blp->blp_waitq);
1329         spin_unlock(&blp->blp_lock);
1330
1331         RETURN(0);
1332 #else
1333         RETURN(-ENOSYS);
1334 #endif
1335 }
1336
1337 static int ldlm_callback_handler(struct ptlrpc_request *req)
1338 {
1339         struct ldlm_namespace *ns;
1340         struct ldlm_request *dlm_req;
1341         struct ldlm_lock *lock;
1342         int rc;
1343         ENTRY;
1344
1345         /* Requests arrive in sender's byte order.  The ptlrpc service
1346          * handler has already checked and, if necessary, byte-swapped the
1347          * incoming request message body, but I am responsible for the
1348          * message buffers. */
1349
1350         if (req->rq_export == NULL) {
1351                 struct ldlm_request *dlm_req;
1352
1353                 CDEBUG(D_RPCTRACE, "operation %d from %s with bad "
1354                        "export cookie "LPX64"; this is "
1355                        "normal if this node rebooted with a lock held\n",
1356                        lustre_msg_get_opc(req->rq_reqmsg),
1357                        libcfs_id2str(req->rq_peer),
1358                        lustre_msg_get_handle(req->rq_reqmsg)->cookie);
1359
1360                 dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF,
1361                                              sizeof(*dlm_req),
1362                                              lustre_swab_ldlm_request);
1363                 if (dlm_req != NULL)
1364                         CDEBUG(D_RPCTRACE, "--> lock cookie: "LPX64"\n",
1365                                dlm_req->lock_handle[0].cookie);
1366
1367                 ldlm_callback_reply(req, -ENOTCONN);
1368                 RETURN(0);
1369         }
1370
1371         LASSERT(req->rq_export != NULL);
1372         LASSERT(req->rq_export->exp_obd != NULL);
1373
1374         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1375         case LDLM_BL_CALLBACK:
1376                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
1377                 break;
1378         case LDLM_CP_CALLBACK:
1379                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CP_CALLBACK, 0);
1380                 break;
1381         case LDLM_GL_CALLBACK:
1382                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_GL_CALLBACK, 0);
1383                 break;
1384         case OBD_LOG_CANCEL: /* remove this eventually - for 1.4.0 compat */
1385                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
1386                 rc = llog_origin_handle_cancel(req);
1387                 ldlm_callback_reply(req, rc);
1388                 RETURN(0);
1389         case OBD_QC_CALLBACK:
1390                 OBD_FAIL_RETURN(OBD_FAIL_OBD_QC_CALLBACK_NET, 0);
1391                 rc = target_handle_qc_callback(req);
1392                 ldlm_callback_reply(req, rc);
1393                 RETURN(0);
1394         case QUOTA_DQACQ:
1395         case QUOTA_DQREL:
1396                 /* reply in handler */
1397                 rc = target_handle_dqacq_callback(req);
1398                 RETURN(0);
1399         case LLOG_ORIGIN_HANDLE_CREATE:
1400                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1401                 rc = llog_origin_handle_create(req);
1402                 ldlm_callback_reply(req, rc);
1403                 RETURN(0);
1404         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
1405                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1406                 rc = llog_origin_handle_next_block(req);
1407                 ldlm_callback_reply(req, rc);
1408                 RETURN(0);
1409         case LLOG_ORIGIN_HANDLE_READ_HEADER:
1410                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1411                 rc = llog_origin_handle_read_header(req);
1412                 ldlm_callback_reply(req, rc);
1413                 RETURN(0);
1414         case LLOG_ORIGIN_HANDLE_CLOSE:
1415                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1416                 rc = llog_origin_handle_close(req);
1417                 ldlm_callback_reply(req, rc);
1418                 RETURN(0);
1419         default:
1420                 CERROR("unknown opcode %u\n",
1421                        lustre_msg_get_opc(req->rq_reqmsg));
1422                 ldlm_callback_reply(req, -EPROTO);
1423                 RETURN(0);
1424         }
1425
1426         ns = req->rq_export->exp_obd->obd_namespace;
1427         LASSERT(ns != NULL);
1428
1429         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1430                                      lustre_swab_ldlm_request);
1431         if (dlm_req == NULL) {
1432                 CERROR ("can't unpack dlm_req\n");
1433                 ldlm_callback_reply(req, -EPROTO);
1434                 RETURN (0);
1435         }
1436
1437         lock = ldlm_handle2lock_ns(ns, &dlm_req->lock_handle[0]);
1438         if (!lock) {
1439                 CDEBUG(D_DLMTRACE, "callback on lock "LPX64" - lock "
1440                        "disappeared\n", dlm_req->lock_handle[0].cookie);
1441                 ldlm_callback_reply(req, -EINVAL);
1442                 RETURN(0);
1443         }
1444
1445         /* Copy hints/flags (e.g. LDLM_FL_DISCARD_DATA) from AST. */
1446         lock_res_and_lock(lock);
1447         lock->l_flags |= (dlm_req->lock_flags & LDLM_AST_FLAGS);
1448         if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
1449                 /* If somebody cancels locks and cache is already droped,
1450                  * we can tell the server we have no lock. Otherwise, we
1451                  * should send cancel after dropping the cache. */
1452                 if ((lock->l_flags & LDLM_FL_CANCELING) &&
1453                     (lock->l_flags & LDLM_FL_BL_DONE)) {
1454                         LDLM_DEBUG(lock, "callback on lock "
1455                                    LPX64" - lock disappeared\n",
1456                                    dlm_req->lock_handle[0].cookie);
1457                         LDLM_LOCK_PUT(lock);
1458                         unlock_res_and_lock(lock);
1459                         ldlm_callback_reply(req, -EINVAL);
1460                         RETURN(0);
1461                 }
1462                 lock->l_flags |= LDLM_FL_BL_AST;
1463         }
1464         unlock_res_and_lock(lock);
1465
1466         /* We want the ost thread to get this reply so that it can respond
1467          * to ost requests (write cache writeback) that might be triggered
1468          * in the callback.
1469          *
1470          * But we'd also like to be able to indicate in the reply that we're
1471          * cancelling right now, because it's unused, or have an intent result
1472          * in the reply, so we might have to push the responsibility for sending
1473          * the reply down into the AST handlers, alas. */
1474
1475         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1476         case LDLM_BL_CALLBACK:
1477                 CDEBUG(D_INODE, "blocking ast\n");
1478                 if (!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK))
1479                         ldlm_callback_reply(req, 0);
1480                 if (ldlm_bl_to_thread(ns, &dlm_req->lock_desc, lock, 0))
1481                         ldlm_handle_bl_callback(ns, &dlm_req->lock_desc, lock);
1482                 break;
1483         case LDLM_CP_CALLBACK:
1484                 CDEBUG(D_INODE, "completion ast\n");
1485                 ldlm_callback_reply(req, 0);
1486                 ldlm_handle_cp_callback(req, ns, dlm_req, lock);
1487                 break;
1488         case LDLM_GL_CALLBACK:
1489                 CDEBUG(D_INODE, "glimpse ast\n");
1490                 ldlm_handle_gl_callback(req, ns, dlm_req, lock);
1491                 break;
1492         default:
1493                 LBUG();                         /* checked above */
1494         }
1495
1496         RETURN(0);
1497 }
1498
1499 static int ldlm_cancel_handler(struct ptlrpc_request *req)
1500 {
1501         int rc;
1502         ENTRY;
1503
1504         /* Requests arrive in sender's byte order.  The ptlrpc service
1505          * handler has already checked and, if necessary, byte-swapped the
1506          * incoming request message body, but I am responsible for the
1507          * message buffers. */
1508
1509         if (req->rq_export == NULL) {
1510                 struct ldlm_request *dlm_req;
1511
1512                 CERROR("operation %d from %s with bad export cookie "LPU64"\n",
1513                        lustre_msg_get_opc(req->rq_reqmsg),
1514                        libcfs_id2str(req->rq_peer),
1515                        lustre_msg_get_handle(req->rq_reqmsg)->cookie);
1516
1517                 dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF,
1518                                              sizeof(*dlm_req),
1519                                              lustre_swab_ldlm_request);
1520                 if (dlm_req != NULL)
1521                         ldlm_lock_dump_handle(D_ERROR,
1522                                               &dlm_req->lock_handle[0]);
1523
1524                 ldlm_callback_reply(req, -ENOTCONN);
1525                 RETURN(0);
1526         }
1527
1528         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1529
1530         /* XXX FIXME move this back to mds/handler.c, bug 249 */
1531         case LDLM_CANCEL:
1532                 CDEBUG(D_INODE, "cancel\n");
1533                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CANCEL, 0);
1534                 rc = ldlm_handle_cancel(req);
1535                 if (rc)
1536                         break;
1537                 RETURN(0);
1538         case OBD_LOG_CANCEL:
1539                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
1540                 rc = llog_origin_handle_cancel(req);
1541                 ldlm_callback_reply(req, rc);
1542                 RETURN(0);
1543         default:
1544                 CERROR("invalid opcode %d\n",
1545                        lustre_msg_get_opc(req->rq_reqmsg));
1546                 ldlm_callback_reply(req, -EINVAL);
1547         }
1548
1549         RETURN(0);
1550 }
1551
1552 #ifdef __KERNEL__
1553 static struct ldlm_bl_work_item *ldlm_bl_get_work(struct ldlm_bl_pool *blp)
1554 {
1555         struct ldlm_bl_work_item *blwi = NULL;
1556
1557         spin_lock(&blp->blp_lock);
1558         if (!list_empty(&blp->blp_list)) {
1559                 blwi = list_entry(blp->blp_list.next, struct ldlm_bl_work_item,
1560                                   blwi_entry);
1561                 list_del(&blwi->blwi_entry);
1562         }
1563         spin_unlock(&blp->blp_lock);
1564
1565         return blwi;
1566 }
1567
1568 struct ldlm_bl_thread_data {
1569         int                     bltd_num;
1570         struct ldlm_bl_pool     *bltd_blp;
1571 };
1572
1573 static int ldlm_bl_thread_main(void *arg)
1574 {
1575         struct ldlm_bl_thread_data *bltd = arg;
1576         struct ldlm_bl_pool *blp = bltd->bltd_blp;
1577         ENTRY;
1578
1579         {
1580                 char name[CFS_CURPROC_COMM_MAX];
1581                 snprintf(name, sizeof(name) - 1, "ldlm_bl_%02d",
1582                          bltd->bltd_num);
1583                 cfs_daemonize(name);
1584         }
1585
1586         atomic_inc(&blp->blp_num_threads);
1587         complete(&blp->blp_comp);
1588
1589         while(1) {
1590                 struct l_wait_info lwi = { 0 };
1591                 struct ldlm_bl_work_item *blwi = NULL;
1592
1593                 l_wait_event_exclusive(blp->blp_waitq,
1594                                        (blwi = ldlm_bl_get_work(blp)) != NULL,
1595                                        &lwi);
1596
1597                 if (blwi->blwi_ns == NULL)
1598                         break;
1599
1600                 if (blwi->blwi_flags == LDLM_FL_CANCELING) {
1601                         /* The special case when we cancel locks in lru
1602                          * asynchronously, then we first remove the lock from
1603                          * l_bl_ast explicitely in ldlm_cancel_lru before
1604                          * sending it to this thread. Thus lock is marked
1605                          * LDLM_FL_CANCELING, and already cancelled locally. */
1606                         CFS_LIST_HEAD(head);
1607                         LASSERT(list_empty(&blwi->blwi_lock->l_bl_ast));
1608                         list_add(&blwi->blwi_lock->l_bl_ast, &head);
1609                         ldlm_cli_cancel_req(blwi->blwi_lock->l_conn_export,
1610                                             &head, 1);
1611                         LDLM_LOCK_PUT(blwi->blwi_lock);
1612                 } else {
1613                         ldlm_handle_bl_callback(blwi->blwi_ns, &blwi->blwi_ld,
1614                                                 blwi->blwi_lock);
1615                 }
1616                 OBD_FREE(blwi, sizeof(*blwi));
1617         }
1618
1619         atomic_dec(&blp->blp_num_threads);
1620         complete(&blp->blp_comp);
1621         RETURN(0);
1622 }
1623
1624 #endif
1625
1626 static int ldlm_setup(void);
1627 static int ldlm_cleanup(int force);
1628
1629 int ldlm_get_ref(void)
1630 {
1631         int rc = 0;
1632         ENTRY;
1633         mutex_down(&ldlm_ref_sem);
1634         if (++ldlm_refcount == 1) {
1635                 rc = ldlm_setup();
1636                 if (rc)
1637                         ldlm_refcount--;
1638         }
1639         mutex_up(&ldlm_ref_sem);
1640
1641         RETURN(rc);
1642 }
1643
1644 void ldlm_put_ref(int force)
1645 {
1646         ENTRY;
1647         mutex_down(&ldlm_ref_sem);
1648         if (ldlm_refcount == 1) {
1649                 int rc = ldlm_cleanup(force);
1650                 if (rc)
1651                         CERROR("ldlm_cleanup failed: %d\n", rc);
1652                 else
1653                         ldlm_refcount--;
1654         } else {
1655                 ldlm_refcount--;
1656         }
1657         mutex_up(&ldlm_ref_sem);
1658
1659         EXIT;
1660 }
1661
1662 static int ldlm_setup(void)
1663 {
1664         struct ldlm_bl_pool *blp;
1665         int rc = 0;
1666 #ifdef __KERNEL__
1667         int i;
1668 #endif
1669         ENTRY;
1670
1671         if (ldlm_state != NULL)
1672                 RETURN(-EALREADY);
1673
1674         OBD_ALLOC(ldlm_state, sizeof(*ldlm_state));
1675         if (ldlm_state == NULL)
1676                 RETURN(-ENOMEM);
1677
1678 #ifdef LPROCFS
1679         rc = ldlm_proc_setup();
1680         if (rc != 0)
1681                 GOTO(out_free, rc);
1682 #endif
1683
1684         ldlm_state->ldlm_cb_service =
1685                 ptlrpc_init_svc(LDLM_NBUFS, LDLM_BUFSIZE, LDLM_MAXREQSIZE,
1686                                 LDLM_MAXREPSIZE, LDLM_CB_REQUEST_PORTAL,
1687                                 LDLM_CB_REPLY_PORTAL, ldlm_timeout * 900,
1688                                 ldlm_callback_handler, "ldlm_cbd",
1689                                 ldlm_svc_proc_dir, NULL,
1690                                 LDLM_THREADS_AUTO_MIN, LDLM_THREADS_AUTO_MAX,
1691                                 "ldlm_cb");
1692
1693         if (!ldlm_state->ldlm_cb_service) {
1694                 CERROR("failed to start service\n");
1695                 GOTO(out_proc, rc = -ENOMEM);
1696         }
1697
1698         ldlm_state->ldlm_cancel_service =
1699                 ptlrpc_init_svc(LDLM_NBUFS, LDLM_BUFSIZE, LDLM_MAXREQSIZE,
1700                                 LDLM_MAXREPSIZE, LDLM_CANCEL_REQUEST_PORTAL,
1701                                 LDLM_CANCEL_REPLY_PORTAL, ldlm_timeout * 6000,
1702                                 ldlm_cancel_handler, "ldlm_canceld",
1703                                 ldlm_svc_proc_dir, NULL,
1704                                 LDLM_THREADS_AUTO_MIN, LDLM_THREADS_AUTO_MAX,
1705                                 "ldlm_cn");
1706
1707         if (!ldlm_state->ldlm_cancel_service) {
1708                 CERROR("failed to start service\n");
1709                 GOTO(out_proc, rc = -ENOMEM);
1710         }
1711
1712         OBD_ALLOC(blp, sizeof(*blp));
1713         if (blp == NULL)
1714                 GOTO(out_proc, rc = -ENOMEM);
1715         ldlm_state->ldlm_bl_pool = blp;
1716
1717         atomic_set(&blp->blp_num_threads, 0);
1718         cfs_waitq_init(&blp->blp_waitq);
1719         spin_lock_init(&blp->blp_lock);
1720
1721         CFS_INIT_LIST_HEAD(&blp->blp_list);
1722
1723 #ifdef __KERNEL__
1724         for (i = 0; i < LDLM_BL_THREADS; i++) {
1725                 struct ldlm_bl_thread_data bltd = {
1726                         .bltd_num = i,
1727                         .bltd_blp = blp,
1728                 };
1729                 init_completion(&blp->blp_comp);
1730                 rc = cfs_kernel_thread(ldlm_bl_thread_main, &bltd, 0);
1731                 if (rc < 0) {
1732                         CERROR("cannot start LDLM thread #%d: rc %d\n", i, rc);
1733                         GOTO(out_thread, rc);
1734                 }
1735                 wait_for_completion(&blp->blp_comp);
1736         }
1737
1738         rc = ptlrpc_start_threads(NULL, ldlm_state->ldlm_cancel_service);
1739         if (rc)
1740                 GOTO(out_thread, rc);
1741
1742         rc = ptlrpc_start_threads(NULL, ldlm_state->ldlm_cb_service);
1743         if (rc)
1744                 GOTO(out_thread, rc);
1745
1746         CFS_INIT_LIST_HEAD(&expired_lock_thread.elt_expired_locks);
1747         expired_lock_thread.elt_state = ELT_STOPPED;
1748         cfs_waitq_init(&expired_lock_thread.elt_waitq);
1749
1750         CFS_INIT_LIST_HEAD(&waiting_locks_list);
1751         spin_lock_init(&waiting_locks_spinlock);
1752         cfs_timer_init(&waiting_locks_timer, waiting_locks_callback, 0);
1753
1754         rc = cfs_kernel_thread(expired_lock_main, NULL, CLONE_VM | CLONE_FILES);
1755         if (rc < 0) {
1756                 CERROR("Cannot start ldlm expired-lock thread: %d\n", rc);
1757                 GOTO(out_thread, rc);
1758         }
1759
1760         wait_event(expired_lock_thread.elt_waitq,
1761                    expired_lock_thread.elt_state == ELT_READY);
1762 #endif
1763
1764         RETURN(0);
1765
1766 #ifdef __KERNEL__
1767  out_thread:
1768         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
1769         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
1770 #endif
1771
1772  out_proc:
1773 #ifdef LPROCFS
1774         ldlm_proc_cleanup();
1775  out_free:
1776 #endif
1777         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
1778         ldlm_state = NULL;
1779         return rc;
1780 }
1781
1782 static int ldlm_cleanup(int force)
1783 {
1784 #ifdef __KERNEL__
1785         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
1786 #endif
1787         ENTRY;
1788
1789         if (!list_empty(&ldlm_namespace_list)) {
1790                 CERROR("ldlm still has namespaces; clean these up first.\n");
1791                 ldlm_dump_all_namespaces(D_DLMTRACE);
1792                 RETURN(-EBUSY);
1793         }
1794
1795 #ifdef __KERNEL__
1796         while (atomic_read(&blp->blp_num_threads) > 0) {
1797                 struct ldlm_bl_work_item blwi = { .blwi_ns = NULL };
1798
1799                 init_completion(&blp->blp_comp);
1800
1801                 spin_lock(&blp->blp_lock);
1802                 list_add_tail(&blwi.blwi_entry, &blp->blp_list);
1803                 cfs_waitq_signal(&blp->blp_waitq);
1804                 spin_unlock(&blp->blp_lock);
1805
1806                 wait_for_completion(&blp->blp_comp);
1807         }
1808         OBD_FREE(blp, sizeof(*blp));
1809
1810         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
1811         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
1812         ldlm_proc_cleanup();
1813
1814         expired_lock_thread.elt_state = ELT_TERMINATE;
1815         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
1816         wait_event(expired_lock_thread.elt_waitq,
1817                    expired_lock_thread.elt_state == ELT_STOPPED);
1818 #else
1819         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
1820         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
1821 #endif
1822
1823         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
1824         ldlm_state = NULL;
1825
1826         RETURN(0);
1827 }
1828
1829 int __init ldlm_init(void)
1830 {
1831         init_mutex(&ldlm_ref_sem);
1832         init_mutex(&ldlm_namespace_lock);
1833         ldlm_resource_slab = cfs_mem_cache_create("ldlm_resources",
1834                                                sizeof(struct ldlm_resource), 0,
1835                                                SLAB_HWCACHE_ALIGN);
1836         if (ldlm_resource_slab == NULL)
1837                 return -ENOMEM;
1838
1839         ldlm_lock_slab = cfs_mem_cache_create("ldlm_locks",
1840                                            sizeof(struct ldlm_lock), 0,
1841                                            SLAB_HWCACHE_ALIGN);
1842         if (ldlm_lock_slab == NULL) {
1843                 cfs_mem_cache_destroy(ldlm_resource_slab);
1844                 return -ENOMEM;
1845         }
1846
1847         return 0;
1848 }
1849
1850 void __exit ldlm_exit(void)
1851 {
1852         int rc;
1853         if (ldlm_refcount)
1854                 CERROR("ldlm_refcount is %d in ldlm_exit!\n", ldlm_refcount);
1855         rc = cfs_mem_cache_destroy(ldlm_resource_slab);
1856         LASSERTF(rc == 0, "couldn't free ldlm resource slab\n");
1857         rc = cfs_mem_cache_destroy(ldlm_lock_slab);
1858         LASSERTF(rc == 0, "couldn't free ldlm lock slab\n");
1859 }
1860
1861 /* ldlm_extent.c */
1862 EXPORT_SYMBOL(ldlm_extent_shift_kms);
1863
1864 /* ldlm_lock.c */
1865 EXPORT_SYMBOL(ldlm_get_processing_policy);
1866 EXPORT_SYMBOL(ldlm_lock2desc);
1867 EXPORT_SYMBOL(ldlm_register_intent);
1868 EXPORT_SYMBOL(ldlm_lockname);
1869 EXPORT_SYMBOL(ldlm_typename);
1870 EXPORT_SYMBOL(ldlm_lock2handle);
1871 EXPORT_SYMBOL(__ldlm_handle2lock);
1872 EXPORT_SYMBOL(ldlm_lock_get);
1873 EXPORT_SYMBOL(ldlm_lock_put);
1874 EXPORT_SYMBOL(ldlm_lock_match);
1875 EXPORT_SYMBOL(ldlm_lock_cancel);
1876 EXPORT_SYMBOL(ldlm_lock_addref);
1877 EXPORT_SYMBOL(ldlm_lock_decref);
1878 EXPORT_SYMBOL(ldlm_lock_decref_and_cancel);
1879 EXPORT_SYMBOL(ldlm_lock_change_resource);
1880 EXPORT_SYMBOL(ldlm_lock_set_data);
1881 EXPORT_SYMBOL(ldlm_it2str);
1882 EXPORT_SYMBOL(ldlm_lock_dump);
1883 EXPORT_SYMBOL(ldlm_lock_dump_handle);
1884 EXPORT_SYMBOL(ldlm_cancel_locks_for_export);
1885 EXPORT_SYMBOL(ldlm_reprocess_all_ns);
1886 EXPORT_SYMBOL(ldlm_lock_allow_match);
1887
1888 /* ldlm_request.c */
1889 EXPORT_SYMBOL(ldlm_completion_ast);
1890 EXPORT_SYMBOL(ldlm_blocking_ast);
1891 EXPORT_SYMBOL(ldlm_glimpse_ast);
1892 EXPORT_SYMBOL(ldlm_expired_completion_wait);
1893 EXPORT_SYMBOL(ldlm_prep_enqueue_req);
1894 EXPORT_SYMBOL(ldlm_cli_convert);
1895 EXPORT_SYMBOL(ldlm_cli_enqueue);
1896 EXPORT_SYMBOL(ldlm_cli_enqueue_fini);
1897 EXPORT_SYMBOL(ldlm_cli_enqueue_local);
1898 EXPORT_SYMBOL(ldlm_cli_cancel);
1899 EXPORT_SYMBOL(ldlm_cli_cancel_unused);
1900 EXPORT_SYMBOL(ldlm_cli_cancel_req);
1901 EXPORT_SYMBOL(ldlm_cli_join_lru);
1902 EXPORT_SYMBOL(ldlm_replay_locks);
1903 EXPORT_SYMBOL(ldlm_resource_foreach);
1904 EXPORT_SYMBOL(ldlm_namespace_foreach);
1905 EXPORT_SYMBOL(ldlm_namespace_foreach_res);
1906 EXPORT_SYMBOL(ldlm_resource_iterate);
1907 EXPORT_SYMBOL(ldlm_cancel_resource_local);
1908 EXPORT_SYMBOL(ldlm_cli_cancel_list);
1909
1910 /* ldlm_lockd.c */
1911 EXPORT_SYMBOL(ldlm_server_blocking_ast);
1912 EXPORT_SYMBOL(ldlm_server_completion_ast);
1913 EXPORT_SYMBOL(ldlm_server_glimpse_ast);
1914 EXPORT_SYMBOL(ldlm_handle_enqueue);
1915 EXPORT_SYMBOL(ldlm_handle_cancel);
1916 EXPORT_SYMBOL(ldlm_request_cancel);
1917 EXPORT_SYMBOL(ldlm_handle_convert);
1918 EXPORT_SYMBOL(ldlm_del_waiting_lock);
1919 EXPORT_SYMBOL(ldlm_get_ref);
1920 EXPORT_SYMBOL(ldlm_put_ref);
1921 EXPORT_SYMBOL(ldlm_refresh_waiting_lock);
1922
1923 /* ldlm_resource.c */
1924 EXPORT_SYMBOL(ldlm_namespace_new);
1925 EXPORT_SYMBOL(ldlm_namespace_cleanup);
1926 EXPORT_SYMBOL(ldlm_namespace_free);
1927 EXPORT_SYMBOL(ldlm_namespace_dump);
1928 EXPORT_SYMBOL(ldlm_dump_all_namespaces);
1929 EXPORT_SYMBOL(ldlm_resource_get);
1930 EXPORT_SYMBOL(ldlm_resource_putref);
1931 EXPORT_SYMBOL(ldlm_resource_unlink_lock);
1932
1933 /* ldlm_lib.c */
1934 EXPORT_SYMBOL(client_import_add_conn);
1935 EXPORT_SYMBOL(client_import_del_conn);
1936 EXPORT_SYMBOL(client_obd_setup);
1937 EXPORT_SYMBOL(client_obd_cleanup);
1938 EXPORT_SYMBOL(client_connect_import);
1939 EXPORT_SYMBOL(client_disconnect_export);
1940 EXPORT_SYMBOL(target_abort_recovery);
1941 EXPORT_SYMBOL(target_cleanup_recovery);
1942 EXPORT_SYMBOL(target_handle_connect);
1943 EXPORT_SYMBOL(target_destroy_export);
1944 EXPORT_SYMBOL(target_cancel_recovery_timer);
1945 EXPORT_SYMBOL(target_send_reply);
1946 EXPORT_SYMBOL(target_queue_recovery_request);
1947 EXPORT_SYMBOL(target_handle_ping);
1948 EXPORT_SYMBOL(target_handle_disconnect);
1949 EXPORT_SYMBOL(target_queue_final_reply);
1950
1951 /* l_lock.c */
1952 EXPORT_SYMBOL(lock_res_and_lock);
1953 EXPORT_SYMBOL(unlock_res_and_lock);
1954