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 };
 102
 103 #ifdef __KERNEL__
 104
 105 static inline int have_expired_locks(void)
 106 {
 107         int need_to_run;
 108
 109         ENTRY;
 110         spin_lock_bh(&waiting_locks_spinlock);
 111         need_to_run = !list_empty(&expired_lock_thread.elt_expired_locks);
 112         spin_unlock_bh(&waiting_locks_spinlock);
 113
 114         RETURN(need_to_run);
 115 }
 116
 117 static int expired_lock_main(void *arg)
 118 {
 119         struct list_head *expired = &expired_lock_thread.elt_expired_locks;
 120         struct l_wait_info lwi = { 0 };
 121         int do_dump;
 122
 123         ENTRY;
 124         cfs_daemonize("ldlm_elt");
 125
 126         expired_lock_thread.elt_state = ELT_READY;
 127         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 128
 129         while (1) {
 130                 l_wait_event(expired_lock_thread.elt_waitq,
 131                              have_expired_locks() ||
 132                              expired_lock_thread.elt_state == ELT_TERMINATE,
 133                              &lwi);
 134
 135                 spin_lock_bh(&waiting_locks_spinlock);
 136                 if (expired_lock_thread.elt_dump) {
 137                         spin_unlock_bh(&waiting_locks_spinlock);
 138
 139                         /* from waiting_locks_callback, but not in timer */
 140                         libcfs_debug_dumplog();
 141                         libcfs_run_lbug_upcall(__FILE__,
 142                                                 "waiting_locks_callback",
 143                                                 expired_lock_thread.elt_dump);
 144
 145                         spin_lock_bh(&waiting_locks_spinlock);
 146                         expired_lock_thread.elt_dump = 0;
 147                 }
 148
 149                 do_dump = 0;
 150
 151                 while (!list_empty(expired)) {
 152                         struct obd_export *export;
 153                         struct ldlm_lock *lock;
 154
 155                         lock = list_entry(expired->next, struct ldlm_lock,
 156                                           l_pending_chain);
 157                         if ((void *)lock < LP_POISON + PAGE_SIZE &&
 158                             (void *)lock >= LP_POISON) {
 159                                 spin_unlock_bh(&waiting_locks_spinlock);
 160                                 CERROR("free lock on elt list %p\n", lock);
 161                                 LBUG();
 162                         }
 163                         list_del_init(&lock->l_pending_chain);
 164                         if ((void *)lock->l_export < LP_POISON + PAGE_SIZE &&
 165                             (void *)lock->l_export >= LP_POISON) {
 166                                 CERROR("lock with free export on elt list %p\n",
 167                                        lock->l_export);
 168                                 lock->l_export = NULL;
 169                                 LDLM_ERROR(lock, "free export");
 170                                 continue;
 171                         }
 172                         export = class_export_get(lock->l_export);
 173                         spin_unlock_bh(&waiting_locks_spinlock);
 174
 175                         do_dump++;
 176                         class_fail_export(export);
 177                         class_export_put(export);
 178                         spin_lock_bh(&waiting_locks_spinlock);
 179                 }
 180                 spin_unlock_bh(&waiting_locks_spinlock);
 181
 182                 if (do_dump && obd_dump_on_eviction) {
 183                         CERROR("dump the log upon eviction\n");
 184                         libcfs_debug_dumplog();
 185                 }
 186
 187                 if (expired_lock_thread.elt_state == ELT_TERMINATE)
 188                         break;
 189         }
 190
 191         expired_lock_thread.elt_state = ELT_STOPPED;
 192         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 193         RETURN(0);
 194 }
 195
 196 /* This is called from within a timer interrupt and cannot schedule */
 197 static void waiting_locks_callback(unsigned long unused)
 198 {
 199         struct ldlm_lock *lock, *last = NULL;
 200
 201         spin_lock_bh(&waiting_locks_spinlock);
 202         while (!list_empty(&waiting_locks_list)) {
 203                 lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
 204                                   l_pending_chain);
 205
 206                 if (cfs_time_after(lock->l_callback_timeout, cfs_time_current()) ||
 207                     (lock->l_req_mode == LCK_GROUP))
 208                         break;
 209
 210                 LDLM_ERROR(lock, "lock callback timer expired: evicting client "
 211                            "%s@%s nid %s\n",
 212                            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("A client on nid %s was evicted from service %s.\n",
 424                        str, lock->l_export->exp_obd->obd_name);
 425
 426         LDLM_ERROR(lock, "%s AST failed (%d): evicting client %s@%s NID %s"
 427                    " (%s)", ast_type, rc, lock->l_export->exp_client_uuid.uuid,
 428                    conn->c_remote_uuid.uuid, libcfs_nid2str(conn->c_peer.nid),
 429                    str);
 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_handle1 = 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         ptlrpc_req_set_repsize(req, 1, NULL);
 542         if (instant_cancel) {
 543                 unlock_res(lock->l_resource);
 544                 ldlm_lock_cancel(lock);
 545         } else if (lock->l_granted_mode == lock->l_req_mode) {
 546                 ldlm_add_waiting_lock(lock);
 547                 unlock_res(lock->l_resource);
 548         }
 549
 550         req->rq_send_state = LUSTRE_IMP_FULL;
 551         req->rq_timeout = ldlm_get_rq_timeout(ldlm_timeout, obd_timeout);
 552         if (unlikely(instant_cancel)) {
 553                 rc = ptl_send_rpc(req, 1);
 554         } else {
 555                 rc = ptlrpc_queue_wait(req);
 556         }
 557         if (rc != 0)
 558                 rc = ldlm_handle_ast_error(lock, req, rc, "blocking");
 559
 560         ptlrpc_req_finished(req);
 561
 562         /* If we cancelled the lock, we need to restart ldlm_reprocess_queue */
 563         if (!rc && instant_cancel)
 564                 rc = -ERESTART;
 565
 566         RETURN(rc);
 567 }
 568
 569 int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
 570 {
 571         struct ldlm_request *body;
 572         struct ptlrpc_request *req;
 573         struct timeval granted_time;
 574         long total_enqueue_wait;
 575         int size[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 576                         [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 577         int rc = 0, buffers = 2, instant_cancel = 0;
 578         ENTRY;
 579
 580         LASSERT(lock != NULL);
 581
 582         do_gettimeofday(&granted_time);
 583         total_enqueue_wait = cfs_timeval_sub(&granted_time,
 584                                              &lock->l_enqueued_time, NULL);
 585
 586         if (total_enqueue_wait / 1000000 > obd_timeout)
 587                 LDLM_ERROR(lock, "enqueue wait took %luus from %lu",
 588                            total_enqueue_wait, lock->l_enqueued_time.tv_sec);
 589
 590         lock_res_and_lock(lock);
 591         if (lock->l_resource->lr_lvb_len) {
 592                 size[DLM_REQ_REC_OFF] = lock->l_resource->lr_lvb_len;
 593                 buffers = 3;
 594         }
 595         unlock_res_and_lock(lock);
 596
 597         req = ptlrpc_prep_req(lock->l_export->exp_imp_reverse,
 598                               LUSTRE_DLM_VERSION, LDLM_CP_CALLBACK, buffers,
 599                               size, NULL);
 600         if (req == NULL)
 601                 RETURN(-ENOMEM);
 602
 603         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 604         body->lock_handle1 = lock->l_remote_handle;
 605         body->lock_flags = flags;
 606         ldlm_lock2desc(lock, &body->lock_desc);
 607
 608         if (buffers == 3) {
 609                 void *lvb;
 610
 611                 lvb = lustre_msg_buf(req->rq_reqmsg, DLM_REQ_REC_OFF,
 612                                      lock->l_resource->lr_lvb_len);
 613                 lock_res_and_lock(lock);
 614                 memcpy(lvb, lock->l_resource->lr_lvb_data,
 615                        lock->l_resource->lr_lvb_len);
 616                 unlock_res_and_lock(lock);
 617         }
 618
 619         LDLM_DEBUG(lock, "server preparing completion AST (after %ldus wait)",
 620                    total_enqueue_wait);
 621         ptlrpc_req_set_repsize(req, 1, NULL);
 622
 623         req->rq_send_state = LUSTRE_IMP_FULL;
 624         req->rq_timeout = ldlm_get_rq_timeout(ldlm_timeout, obd_timeout);
 625
 626         /* We only send real blocking ASTs after the lock is granted */
 627         lock_res_and_lock(lock);
 628         if (lock->l_flags & LDLM_FL_AST_SENT) {
 629                 body->lock_flags |= LDLM_FL_AST_SENT;
 630
 631                 /* We might get here prior to ldlm_handle_enqueue setting
 632                  * LDLM_FL_CANCEL_ON_BLOCK flag. Then we will put this lock
 633                  * into waiting list, but this is safe and similar code in
 634                  * ldlm_handle_enqueue will call ldlm_lock_cancel() still,
 635                  * that would not only cancel the lock, but will also remove
 636                  * it from waiting list */
 637                 if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) {
 638                         unlock_res_and_lock(lock);
 639                         ldlm_lock_cancel(lock);
 640                         instant_cancel = 1;
 641                         lock_res_and_lock(lock);
 642                 } else {
 643                         ldlm_add_waiting_lock(lock); /* start the lock-timeout
 644                                                          clock */
 645                 }
 646         }
 647         unlock_res_and_lock(lock);
 648
 649         rc = ptlrpc_queue_wait(req);
 650         if (rc != 0)
 651                 rc = ldlm_handle_ast_error(lock, req, rc, "completion");
 652
 653         ptlrpc_req_finished(req);
 654
 655         /* If we cancelled the lock, we need to restart ldlm_reprocess_queue */
 656         if (!rc && instant_cancel)
 657                 rc = -ERESTART;
 658
 659         RETURN(rc);
 660 }
 661
 662 int ldlm_server_glimpse_ast(struct ldlm_lock *lock, void *data)
 663 {
 664         struct ldlm_resource *res = lock->l_resource;
 665         struct ldlm_request *body;
 666         struct ptlrpc_request *req;
 667         int size[] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 668                        [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 669         int rc = 0;
 670         ENTRY;
 671
 672         LASSERT(lock != NULL);
 673
 674         req = ptlrpc_prep_req(lock->l_export->exp_imp_reverse,
 675                               LUSTRE_DLM_VERSION, LDLM_GL_CALLBACK, 2, size,
 676                               NULL);
 677         if (req == NULL)
 678                 RETURN(-ENOMEM);
 679
 680         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 681         body->lock_handle1 = lock->l_remote_handle;
 682         ldlm_lock2desc(lock, &body->lock_desc);
 683
 684         lock_res_and_lock(lock);
 685         size[REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
 686         unlock_res_and_lock(lock);
 687         res = lock->l_resource;
 688         ptlrpc_req_set_repsize(req, 2, size);
 689
 690         req->rq_send_state = LUSTRE_IMP_FULL;
 691         req->rq_timeout = ldlm_get_rq_timeout(ldlm_timeout, obd_timeout);
 692
 693         rc = ptlrpc_queue_wait(req);
 694         if (rc == -ELDLM_NO_LOCK_DATA)
 695                 LDLM_DEBUG(lock, "lost race - client has a lock but no inode");
 696         else if (rc != 0)
 697                 rc = ldlm_handle_ast_error(lock, req, rc, "glimpse");
 698         else
 699                 rc = res->lr_namespace->ns_lvbo->lvbo_update
 700                         (res, req->rq_repmsg, REPLY_REC_OFF, 1);
 701         ptlrpc_req_finished(req);
 702         RETURN(rc);
 703 }
 704
 705 static struct ldlm_lock *
 706 find_existing_lock(struct obd_export *exp,
 707                    const struct lustre_handle *remote_hdl)
 708 {
 709         struct list_head *iter;
 710
 711         spin_lock(&exp->exp_ldlm_data.led_lock);
 712         list_for_each(iter, &exp->exp_ldlm_data.led_held_locks) {
 713                 struct ldlm_lock *lock;
 714                 lock = list_entry(iter, struct ldlm_lock, l_export_chain);
 715                 if (lock->l_remote_handle.cookie == remote_hdl->cookie) {
 716                         LDLM_LOCK_GET(lock);
 717                         spin_unlock(&exp->exp_ldlm_data.led_lock);
 718                         return lock;
 719                 }
 720         }
 721         spin_unlock(&exp->exp_ldlm_data.led_lock);
 722         return NULL;
 723 }
 724
 725
 726 /*
 727  * Main server-side entry point into LDLM. This is called by ptlrpc service
 728  * threads to carry out client lock enqueueing requests.
 729  */
 730 int ldlm_handle_enqueue0(struct ldlm_namespace *ns,
 731                          struct ptlrpc_request *req,
 732                          const struct ldlm_request *dlm_req,
 733                          const struct ldlm_callback_suite *cbs)
 734 {
 735         struct ldlm_reply *dlm_rep;
 736         int size[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 737                         [DLM_LOCKREPLY_OFF]   = sizeof(*dlm_rep) };
 738         int rc = 0;
 739         __u32 flags;
 740         ldlm_error_t err = ELDLM_OK;
 741         struct ldlm_lock *lock = NULL;
 742         void *cookie = NULL;
 743         ENTRY;
 744
 745         LDLM_DEBUG_NOLOCK("server-side enqueue handler START");
 746
 747         flags = dlm_req->lock_flags;
 748
 749         LASSERT(req->rq_export);
 750
 751         if (dlm_req->lock_desc.l_resource.lr_type < LDLM_MIN_TYPE ||
 752             dlm_req->lock_desc.l_resource.lr_type >= LDLM_MAX_TYPE) {
 753                 DEBUG_REQ(D_ERROR, req, "invalid lock request type %d\n",
 754                           dlm_req->lock_desc.l_resource.lr_type);
 755                 GOTO(out, rc = -EFAULT);
 756         }
 757
 758         if (dlm_req->lock_desc.l_req_mode <= LCK_MINMODE ||
 759             dlm_req->lock_desc.l_req_mode >= LCK_MAXMODE ||
 760             dlm_req->lock_desc.l_req_mode & (dlm_req->lock_desc.l_req_mode-1)) {
 761                 DEBUG_REQ(D_ERROR, req, "invalid lock request mode %d\n",
 762                           dlm_req->lock_desc.l_req_mode);
 763                 GOTO(out, rc = -EFAULT);
 764         }
 765
 766         if (req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS) {
 767                 if (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN) {
 768                         DEBUG_REQ(D_ERROR, req,
 769                                   "PLAIN lock request from IBITS client?\n");
 770                         GOTO(out, rc = -EPROTO);
 771                 }
 772         } else if (dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS) {
 773                 DEBUG_REQ(D_ERROR, req,
 774                           "IBITS lock request from unaware client?\n");
 775                 GOTO(out, rc = -EPROTO);
 776         }
 777
 778 #if 0
 779         /* FIXME this makes it impossible to use LDLM_PLAIN locks -- check
 780            against server's _CONNECT_SUPPORTED flags? (I don't want to use
 781            ibits for mgc/mgs) */
 782
 783         /* INODEBITS_INTEROP: Perform conversion from plain lock to
 784          * inodebits lock if client does not support them. */
 785         if (!(req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS) &&
 786             (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN)) {
 787                 dlm_req->lock_desc.l_resource.lr_type = LDLM_IBITS;
 788                 dlm_req->lock_desc.l_policy_data.l_inodebits.bits =
 789                         MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE;
 790                 if (dlm_req->lock_desc.l_req_mode == LCK_PR)
 791                         dlm_req->lock_desc.l_req_mode = LCK_CR;
 792         }
 793 #endif
 794
 795         if (flags & LDLM_FL_REPLAY) {
 796                 lock = find_existing_lock(req->rq_export,
 797                                           &dlm_req->lock_handle1);
 798                 if (lock != NULL) {
 799                         DEBUG_REQ(D_HA, req, "found existing lock cookie "LPX64,
 800                                   lock->l_handle.h_cookie);
 801                         GOTO(existing_lock, rc = 0);
 802                 }
 803         }
 804
 805         /* The lock's callback data might be set in the policy function */
 806         lock = ldlm_lock_create(ns, &dlm_req->lock_handle2,
 807                                 &dlm_req->lock_desc.l_resource.lr_name,
 808                                 dlm_req->lock_desc.l_resource.lr_type,
 809                                 dlm_req->lock_desc.l_req_mode,
 810                                 cbs->lcs_blocking, cbs->lcs_completion,
 811                                 cbs->lcs_glimpse, NULL, 0);
 812         if (!lock)
 813                 GOTO(out, rc = -ENOMEM);
 814
 815         do_gettimeofday(&lock->l_enqueued_time);
 816         lock->l_remote_handle = dlm_req->lock_handle1;
 817         LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
 818
 819         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_BLOCKED, obd_timeout * 2);
 820         /* Don't enqueue a lock onto the export if it has already
 821          * been evicted.  Cancel it now instead. (bug 3822) */
 822         if (req->rq_export->exp_failed) {
 823                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
 824                 GOTO(out, rc = -ENOTCONN);
 825         }
 826         lock->l_export = class_export_get(req->rq_export);
 827         spin_lock(&lock->l_export->exp_ldlm_data.led_lock);
 828         list_add(&lock->l_export_chain,
 829                  &lock->l_export->exp_ldlm_data.led_held_locks);
 830         spin_unlock(&lock->l_export->exp_ldlm_data.led_lock);
 831
 832 existing_lock:
 833
 834         if (flags & LDLM_FL_HAS_INTENT) {
 835                 /* In this case, the reply buffer is allocated deep in
 836                  * local_lock_enqueue by the policy function. */
 837                 cookie = req;
 838         } else {
 839                 int buffers = 2;
 840
 841                 lock_res_and_lock(lock);
 842                 if (lock->l_resource->lr_lvb_len) {
 843                         size[DLM_REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
 844                         buffers = 3;
 845                 }
 846                 unlock_res_and_lock(lock);
 847
 848                 if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
 849                         GOTO(out, rc = -ENOMEM);
 850
 851                 rc = lustre_pack_reply(req, buffers, size, NULL);
 852                 if (rc)
 853                         GOTO(out, rc);
 854         }
 855
 856         if (dlm_req->lock_desc.l_resource.lr_type != LDLM_PLAIN)
 857                 lock->l_policy_data = dlm_req->lock_desc.l_policy_data;
 858         if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT)
 859                 lock->l_req_extent = lock->l_policy_data.l_extent;
 860
 861         err = ldlm_lock_enqueue(ns, &lock, cookie, &flags);
 862         if (err)
 863                 GOTO(out, err);
 864
 865         dlm_rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
 866                                  sizeof(*dlm_rep));
 867         dlm_rep->lock_flags = flags;
 868
 869         ldlm_lock2desc(lock, &dlm_rep->lock_desc);
 870         ldlm_lock2handle(lock, &dlm_rep->lock_handle);
 871
 872         /* We never send a blocking AST until the lock is granted, but
 873          * we can tell it right now */
 874         lock_res_and_lock(lock);
 875
 876         /* Now take into account flags to be inherited from original lock
 877            request both in reply to client and in our own lock flags. */
 878         dlm_rep->lock_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
 879         lock->l_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
 880
 881         /* Don't move a pending lock onto the export if it has already
 882          * been evicted.  Cancel it now instead. (bug 5683) */
 883         if (req->rq_export->exp_failed ||
 884             OBD_FAIL_CHECK_ONCE(OBD_FAIL_LDLM_ENQUEUE_OLD_EXPORT)) {
 885                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
 886                 rc = -ENOTCONN;
 887         } else if (lock->l_flags & LDLM_FL_AST_SENT) {
 888                 dlm_rep->lock_flags |= LDLM_FL_AST_SENT;
 889                 if (lock->l_granted_mode == lock->l_req_mode) {
 890                         /*
 891                          * Only cancel lock if it was granted, because it would
 892                          * be destroyed immediatelly and would never be granted
 893                          * in the future, causing timeouts on client.  Not
 894                          * granted lock will be cancelled immediatelly after
 895                          * sending completion AST.
 896                          */
 897                         if (dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK) {
 898                                 unlock_res_and_lock(lock);
 899                                 ldlm_lock_cancel(lock);
 900                                 lock_res_and_lock(lock);
 901                         } else if (lock->l_granted_mode == lock->l_req_mode)
 902                                 ldlm_add_waiting_lock(lock);
 903                 }
 904         }
 905         /* Make sure we never ever grant usual metadata locks to liblustre
 906            clients */
 907         if ((dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN ||
 908             dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS) &&
 909              req->rq_export->exp_libclient) {
 910                 if (!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) ||
 911                     !(dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK)) {
 912                         CERROR("Granting sync lock to libclient. "
 913                                "req fl %d, rep fl %d, lock fl %d\n",
 914                                dlm_req->lock_flags, dlm_rep->lock_flags,
 915                                lock->l_flags);
 916                         LDLM_ERROR(lock, "sync lock");
 917                         if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT) {
 918                                 struct ldlm_intent *it;
 919                                 it = lustre_msg_buf(req->rq_reqmsg,
 920                                                     DLM_INTENT_IT_OFF,
 921                                                     sizeof(*it));
 922                                 if (it != NULL) {
 923                                         CERROR("This is intent %s ("LPU64")\n",
 924                                                ldlm_it2str(it->opc), it->opc);
 925                                 }
 926                         }
 927                 }
 928         }
 929
 930         unlock_res_and_lock(lock);
 931
 932         EXIT;
 933  out:
 934         req->rq_status = err;
 935         if (req->rq_reply_state == NULL) {
 936                 err = lustre_pack_reply(req, 1, NULL, NULL);
 937                 if (rc == 0)
 938                         rc = err;
 939                 req->rq_status = rc;
 940         }
 941
 942         /* The LOCK_CHANGED code in ldlm_lock_enqueue depends on this
 943          * ldlm_reprocess_all.  If this moves, revisit that code. -phil */
 944         if (lock) {
 945                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
 946                            "(err=%d, rc=%d)", err, rc);
 947
 948                 if (rc == 0) {
 949                         lock_res_and_lock(lock);
 950                         size[DLM_REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
 951                         if (size[DLM_REPLY_REC_OFF] > 0) {
 952                                 void *lvb = lustre_msg_buf(req->rq_repmsg,
 953                                                        DLM_REPLY_REC_OFF,
 954                                                        size[DLM_REPLY_REC_OFF]);
 955                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
 956                                          req, lock);
 957
 958                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
 959                                        size[DLM_REPLY_REC_OFF]);
 960                         }
 961                         unlock_res_and_lock(lock);
 962                 } else {
 963                         lock_res_and_lock(lock);
 964                         ldlm_resource_unlink_lock(lock);
 965                         ldlm_lock_destroy_nolock(lock);
 966                         unlock_res_and_lock(lock);
 967                 }
 968
 969                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
 970                         ldlm_reprocess_all(lock->l_resource);
 971
 972                 LDLM_LOCK_PUT(lock);
 973         }
 974
 975         LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p, rc %d)",
 976                           lock, rc);
 977
 978         return rc;
 979 }
 980
 981 int ldlm_handle_enqueue(struct ptlrpc_request *req,
 982                         ldlm_completion_callback completion_callback,
 983                         ldlm_blocking_callback blocking_callback,
 984                         ldlm_glimpse_callback glimpse_callback)
 985 {
 986         int rc;
 987         struct ldlm_request *dlm_req;
 988         struct ldlm_callback_suite cbs = {
 989                 .lcs_completion = completion_callback,
 990                 .lcs_blocking   = blocking_callback,
 991                 .lcs_glimpse    = glimpse_callback
 992         };
 993
 994
 995         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF,
 996                                      sizeof *dlm_req, lustre_swab_ldlm_request);
 997         if (dlm_req != NULL) {
 998                 rc = ldlm_handle_enqueue0(req->rq_export->exp_obd->obd_namespace,
 999                                           req, dlm_req, &cbs);
1000         } else {
1001                 CERROR ("Can't unpack dlm_req\n");
1002                 rc = -EFAULT;
1003         }
1004         return rc;
1005 }
1006
1007 int ldlm_handle_convert0(struct ptlrpc_request *req,
1008                          const struct ldlm_request *dlm_req)
1009 {
1010         struct ldlm_reply *dlm_rep;
1011         struct ldlm_lock *lock;
1012         int rc;
1013         int size[2] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
1014                         [DLM_LOCKREPLY_OFF]   = sizeof(*dlm_rep) };
1015         ENTRY;
1016
1017         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1018                                      lustre_swab_ldlm_request);
1019         if (dlm_req == NULL) {
1020                 CERROR ("Can't unpack dlm_req\n");
1021                 RETURN (-EFAULT);
1022         }
1023
1024         rc = lustre_pack_reply(req, 2, size, NULL);
1025         if (rc) {
1026                 CERROR("out of memory\n");
1027                 RETURN(-ENOMEM);
1028         }
1029         dlm_rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
1030                                  sizeof(*dlm_rep));
1031         dlm_rep->lock_flags = dlm_req->lock_flags;
1032
1033         lock = ldlm_handle2lock(&dlm_req->lock_handle1);
1034         if (!lock) {
1035                 req->rq_status = EINVAL;
1036         } else {
1037                 void *res = NULL;
1038
1039                 LDLM_DEBUG(lock, "server-side convert handler START");
1040
1041                 do_gettimeofday(&lock->l_enqueued_time);
1042                 res = ldlm_lock_convert(lock, dlm_req->lock_desc.l_req_mode,
1043                                         &dlm_rep->lock_flags);
1044                 if (res) {
1045                         if (ldlm_del_waiting_lock(lock))
1046                                 LDLM_DEBUG(lock, "converted waiting lock");
1047                         req->rq_status = 0;
1048                 } else {
1049                         req->rq_status = EDEADLOCK;
1050                 }
1051         }
1052
1053         if (lock) {
1054                 if (!req->rq_status)
1055                         ldlm_reprocess_all(lock->l_resource);
1056                 LDLM_DEBUG(lock, "server-side convert handler END");
1057                 LDLM_LOCK_PUT(lock);
1058         } else
1059                 LDLM_DEBUG_NOLOCK("server-side convert handler END");
1060
1061         RETURN(0);
1062 }
1063
1064 int ldlm_handle_convert(struct ptlrpc_request *req)
1065 {
1066         int rc;
1067         struct ldlm_request *dlm_req;
1068
1069         dlm_req = lustre_swab_reqbuf(req, 0, sizeof *dlm_req,
1070                                      lustre_swab_ldlm_request);
1071         if (dlm_req != NULL) {
1072                 rc = ldlm_handle_convert0(req, dlm_req);
1073         } else {
1074                 CERROR ("Can't unpack dlm_req\n");
1075                 rc = -EFAULT;
1076         }
1077         return rc;
1078 }
1079
1080 int ldlm_handle_cancel(struct ptlrpc_request *req)
1081 {
1082         struct ldlm_request *dlm_req;
1083         struct ldlm_lock *lock;
1084         struct ldlm_resource *res;
1085         int rc;
1086         ENTRY;
1087
1088         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1089                                      lustre_swab_ldlm_request);
1090         if (dlm_req == NULL) {
1091                 CERROR("bad request buffer for cancel\n");
1092                 RETURN(-EFAULT);
1093         }
1094
1095         rc = lustre_pack_reply(req, 1, NULL, NULL);
1096         if (rc) {
1097                 CERROR("out of memory\n");
1098                 RETURN(-ENOMEM);
1099         }
1100
1101         lock = ldlm_handle2lock(&dlm_req->lock_handle1);
1102         if (!lock) {
1103                 CERROR("received cancel for unknown lock cookie "LPX64
1104                        " from client %s id %s\n",
1105                        dlm_req->lock_handle1.cookie,
1106                        req->rq_export->exp_client_uuid.uuid,
1107                        libcfs_id2str(req->rq_peer));
1108                 LDLM_DEBUG_NOLOCK("server-side cancel handler stale lock "
1109                                   "(cookie "LPU64")",
1110                                   dlm_req->lock_handle1.cookie);
1111                 req->rq_status = ESTALE;
1112         } else {
1113                 LDLM_DEBUG(lock, "server-side cancel handler START");
1114                 res = lock->l_resource;
1115                 if (res && res->lr_namespace->ns_lvbo &&
1116                     res->lr_namespace->ns_lvbo->lvbo_update) {
1117                         (void)res->lr_namespace->ns_lvbo->lvbo_update
1118                                 (res, NULL, 0, 0);
1119                                 //(res, req->rq_reqmsg, 1);
1120                 }
1121
1122                 ldlm_lock_cancel(lock);
1123                 if (ldlm_del_waiting_lock(lock))
1124                         CDEBUG(D_DLMTRACE, "cancelled waiting lock %p\n", lock);
1125                 req->rq_status = rc;
1126         }
1127
1128         if (ptlrpc_reply(req) != 0)
1129                 LBUG();
1130
1131         if (lock) {
1132                 ldlm_reprocess_all(lock->l_resource);
1133                 LDLM_DEBUG(lock, "server-side cancel handler END");
1134                 LDLM_LOCK_PUT(lock);
1135         }
1136
1137         RETURN(0);
1138 }
1139
1140 void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
1141                              struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
1142 {
1143         int do_ast;
1144         ENTRY;
1145
1146         LDLM_DEBUG(lock, "client blocking AST callback handler START");
1147
1148         lock_res_and_lock(lock);
1149         lock->l_flags |= LDLM_FL_CBPENDING;
1150
1151         if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)
1152                 lock->l_flags |= LDLM_FL_CANCEL;
1153
1154         do_ast = (!lock->l_readers && !lock->l_writers);
1155         unlock_res_and_lock(lock);
1156
1157         if (do_ast) {
1158                 LDLM_DEBUG(lock, "already unused, calling "
1159                            "callback (%p)", lock->l_blocking_ast);
1160                 if (lock->l_blocking_ast != NULL)
1161                         lock->l_blocking_ast(lock, ld, lock->l_ast_data,
1162                                              LDLM_CB_BLOCKING);
1163         } else {
1164                 LDLM_DEBUG(lock, "Lock still has references, will be"
1165                            " cancelled later");
1166         }
1167
1168         LDLM_DEBUG(lock, "client blocking callback handler END");
1169         LDLM_LOCK_PUT(lock);
1170         EXIT;
1171 }
1172
1173 static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
1174                                     struct ldlm_namespace *ns,
1175                                     struct ldlm_request *dlm_req,
1176                                     struct ldlm_lock *lock)
1177 {
1178         CFS_LIST_HEAD(ast_list);
1179         ENTRY;
1180
1181         LDLM_DEBUG(lock, "client completion callback handler START");
1182
1183         lock_res_and_lock(lock);
1184
1185         /* If we receive the completion AST before the actual enqueue returned,
1186          * then we might need to switch lock modes, resources, or extents. */
1187         if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
1188                 lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
1189                 LDLM_DEBUG(lock, "completion AST, new lock mode");
1190         }
1191
1192         if (lock->l_resource->lr_type != LDLM_PLAIN) {
1193                 lock->l_policy_data = dlm_req->lock_desc.l_policy_data;
1194                 LDLM_DEBUG(lock, "completion AST, new policy data");
1195         }
1196
1197         ldlm_resource_unlink_lock(lock);
1198         if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
1199                    &lock->l_resource->lr_name,
1200                    sizeof(lock->l_resource->lr_name)) != 0) {
1201                 unlock_res_and_lock(lock);
1202                 ldlm_lock_change_resource(ns, lock,
1203                                         &dlm_req->lock_desc.l_resource.lr_name);
1204                 LDLM_DEBUG(lock, "completion AST, new resource");
1205                 CERROR("change resource!\n");
1206                 lock_res_and_lock(lock);
1207         }
1208
1209         if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
1210                 lock->l_flags |= LDLM_FL_CBPENDING;
1211                 LDLM_DEBUG(lock, "completion AST includes blocking AST");
1212         }
1213
1214         if (lock->l_lvb_len) {
1215                 void *lvb;
1216                 lvb = lustre_swab_reqbuf(req, DLM_REQ_REC_OFF, lock->l_lvb_len,
1217                                          lock->l_lvb_swabber);
1218                 if (lvb == NULL) {
1219                         LDLM_ERROR(lock, "completion AST did not contain "
1220                                    "expected LVB!");
1221                 } else {
1222                         memcpy(lock->l_lvb_data, lvb, lock->l_lvb_len);
1223                 }
1224         }
1225
1226         ldlm_grant_lock(lock, &ast_list);
1227         unlock_res_and_lock(lock);
1228
1229         LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
1230
1231         ldlm_run_cp_ast_work(&ast_list);
1232
1233         LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
1234                           lock);
1235         LDLM_LOCK_PUT(lock);
1236         EXIT;
1237 }
1238
1239 static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
1240                                     struct ldlm_namespace *ns,
1241                                     struct ldlm_request *dlm_req,
1242                                     struct ldlm_lock *lock)
1243 {
1244         int rc = -ENOSYS;
1245         ENTRY;
1246
1247         LDLM_DEBUG(lock, "client glimpse AST callback handler");
1248
1249         if (lock->l_glimpse_ast != NULL)
1250                 rc = lock->l_glimpse_ast(lock, req);
1251
1252         if (req->rq_repmsg != NULL) {
1253                 ptlrpc_reply(req);
1254         } else {
1255                 req->rq_status = rc;
1256                 ptlrpc_error(req);
1257         }
1258
1259         lock_res_and_lock(lock);
1260         if (lock->l_granted_mode == LCK_PW &&
1261             !lock->l_readers && !lock->l_writers &&
1262             cfs_time_after(cfs_time_current(),
1263                            cfs_time_add(lock->l_last_used, cfs_time_seconds(10)))) {
1264                 unlock_res_and_lock(lock);
1265                 if (ldlm_bl_to_thread(ns, NULL, lock))
1266                         ldlm_handle_bl_callback(ns, NULL, lock);
1267
1268                 EXIT;
1269                 return;
1270         }
1271         unlock_res_and_lock(lock);
1272         LDLM_LOCK_PUT(lock);
1273         EXIT;
1274 }
1275
1276 static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
1277 {
1278         req->rq_status = rc;
1279         if (req->rq_reply_state == NULL) {
1280                 rc = lustre_pack_reply(req, 1, NULL, NULL);
1281                 if (rc)
1282                         return rc;
1283         }
1284         return ptlrpc_reply(req);
1285 }
1286
1287 int ldlm_bl_to_thread(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
1288                       struct ldlm_lock *lock)
1289 {
1290 #ifdef __KERNEL__
1291         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
1292         struct ldlm_bl_work_item *blwi;
1293         ENTRY;
1294
1295         OBD_ALLOC(blwi, sizeof(*blwi));
1296         if (blwi == NULL)
1297                 RETURN(-ENOMEM);
1298
1299         blwi->blwi_ns = ns;
1300         if (ld != NULL)
1301                 blwi->blwi_ld = *ld;
1302         blwi->blwi_lock = lock;
1303
1304         spin_lock(&blp->blp_lock);
1305         list_add_tail(&blwi->blwi_entry, &blp->blp_list);
1306         cfs_waitq_signal(&blp->blp_waitq);
1307         spin_unlock(&blp->blp_lock);
1308
1309         RETURN(0);
1310 #else
1311         RETURN(-ENOSYS);
1312 #endif
1313 }
1314
1315 static int ldlm_callback_handler(struct ptlrpc_request *req)
1316 {
1317         struct ldlm_namespace *ns;
1318         struct ldlm_request *dlm_req;
1319         struct ldlm_lock *lock;
1320         int rc;
1321         ENTRY;
1322
1323         /* Requests arrive in sender's byte order.  The ptlrpc service
1324          * handler has already checked and, if necessary, byte-swapped the
1325          * incoming request message body, but I am responsible for the
1326          * message buffers. */
1327
1328         if (req->rq_export == NULL) {
1329                 struct ldlm_request *dlm_req;
1330
1331                 CDEBUG(D_RPCTRACE, "operation %d from %s with bad "
1332                        "export cookie "LPX64"; this is "
1333                        "normal if this node rebooted with a lock held\n",
1334                        lustre_msg_get_opc(req->rq_reqmsg),
1335                        libcfs_id2str(req->rq_peer),
1336                        lustre_msg_get_handle(req->rq_reqmsg)->cookie);
1337
1338                 dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF,
1339                                              sizeof(*dlm_req),
1340                                              lustre_swab_ldlm_request);
1341                 if (dlm_req != NULL)
1342                         CDEBUG(D_RPCTRACE, "--> lock cookie: "LPX64"\n",
1343                                dlm_req->lock_handle1.cookie);
1344
1345                 ldlm_callback_reply(req, -ENOTCONN);
1346                 RETURN(0);
1347         }
1348
1349         LASSERT(req->rq_export != NULL);
1350         LASSERT(req->rq_export->exp_obd != NULL);
1351
1352         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1353         case LDLM_BL_CALLBACK:
1354                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
1355                 break;
1356         case LDLM_CP_CALLBACK:
1357                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CP_CALLBACK, 0);
1358                 break;
1359         case LDLM_GL_CALLBACK:
1360                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_GL_CALLBACK, 0);
1361                 break;
1362         case OBD_LOG_CANCEL: /* remove this eventually - for 1.4.0 compat */
1363                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
1364                 rc = llog_origin_handle_cancel(req);
1365                 ldlm_callback_reply(req, rc);
1366                 RETURN(0);
1367         case OBD_QC_CALLBACK:
1368                 OBD_FAIL_RETURN(OBD_FAIL_OBD_QC_CALLBACK_NET, 0);
1369                 rc = target_handle_qc_callback(req);
1370                 ldlm_callback_reply(req, rc);
1371                 RETURN(0);
1372         case QUOTA_DQACQ:
1373         case QUOTA_DQREL:
1374                 /* reply in handler */
1375                 rc = target_handle_dqacq_callback(req);
1376                 RETURN(0);
1377         case LLOG_ORIGIN_HANDLE_CREATE:
1378                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1379                 rc = llog_origin_handle_create(req);
1380                 ldlm_callback_reply(req, rc);
1381                 RETURN(0);
1382         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
1383                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1384                 rc = llog_origin_handle_next_block(req);
1385                 ldlm_callback_reply(req, rc);
1386                 RETURN(0);
1387         case LLOG_ORIGIN_HANDLE_READ_HEADER:
1388                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1389                 rc = llog_origin_handle_read_header(req);
1390                 ldlm_callback_reply(req, rc);
1391                 RETURN(0);
1392         case LLOG_ORIGIN_HANDLE_CLOSE:
1393                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1394                 rc = llog_origin_handle_close(req);
1395                 ldlm_callback_reply(req, rc);
1396                 RETURN(0);
1397         default:
1398                 CERROR("unknown opcode %u\n",
1399                        lustre_msg_get_opc(req->rq_reqmsg));
1400                 ldlm_callback_reply(req, -EPROTO);
1401                 RETURN(0);
1402         }
1403
1404         ns = req->rq_export->exp_obd->obd_namespace;
1405         LASSERT(ns != NULL);
1406
1407         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1408                                      lustre_swab_ldlm_request);
1409         if (dlm_req == NULL) {
1410                 CERROR ("can't unpack dlm_req\n");
1411                 ldlm_callback_reply(req, -EPROTO);
1412                 RETURN (0);
1413         }
1414
1415         lock = ldlm_handle2lock_ns(ns, &dlm_req->lock_handle1);
1416         if (!lock) {
1417                 CDEBUG(D_INODE, "callback on lock "LPX64" - lock disappeared\n",
1418                        dlm_req->lock_handle1.cookie);
1419                 ldlm_callback_reply(req, -EINVAL);
1420                 RETURN(0);
1421         }
1422
1423         /* Copy hints/flags (e.g. LDLM_FL_DISCARD_DATA) from AST. */
1424         lock_res_and_lock(lock);
1425         lock->l_flags |= (dlm_req->lock_flags & LDLM_AST_FLAGS);
1426         unlock_res_and_lock(lock);
1427
1428         /* We want the ost thread to get this reply so that it can respond
1429          * to ost requests (write cache writeback) that might be triggered
1430          * in the callback.
1431          *
1432          * But we'd also like to be able to indicate in the reply that we're
1433          * cancelling right now, because it's unused, or have an intent result
1434          * in the reply, so we might have to push the responsibility for sending
1435          * the reply down into the AST handlers, alas. */
1436
1437         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1438         case LDLM_BL_CALLBACK:
1439                 CDEBUG(D_INODE, "blocking ast\n");
1440                 if (!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK))
1441                         ldlm_callback_reply(req, 0);
1442                 if (ldlm_bl_to_thread(ns, &dlm_req->lock_desc, lock))
1443                         ldlm_handle_bl_callback(ns, &dlm_req->lock_desc, lock);
1444                 break;
1445         case LDLM_CP_CALLBACK:
1446                 CDEBUG(D_INODE, "completion ast\n");
1447                 ldlm_callback_reply(req, 0);
1448                 ldlm_handle_cp_callback(req, ns, dlm_req, lock);
1449                 break;
1450         case LDLM_GL_CALLBACK:
1451                 CDEBUG(D_INODE, "glimpse ast\n");
1452                 ldlm_handle_gl_callback(req, ns, dlm_req, lock);
1453                 break;
1454         default:
1455                 LBUG();                         /* checked above */
1456         }
1457
1458         RETURN(0);
1459 }
1460
1461 static int ldlm_cancel_handler(struct ptlrpc_request *req)
1462 {
1463         int rc;
1464         ENTRY;
1465
1466         /* Requests arrive in sender's byte order.  The ptlrpc service
1467          * handler has already checked and, if necessary, byte-swapped the
1468          * incoming request message body, but I am responsible for the
1469          * message buffers. */
1470
1471         if (req->rq_export == NULL) {
1472                 struct ldlm_request *dlm_req;
1473
1474                 CERROR("operation %d from %s with bad export cookie "LPU64"\n",
1475                        lustre_msg_get_opc(req->rq_reqmsg),
1476                        libcfs_id2str(req->rq_peer),
1477                        lustre_msg_get_handle(req->rq_reqmsg)->cookie);
1478
1479                 dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF,
1480                                              sizeof(*dlm_req),
1481                                              lustre_swab_ldlm_request);
1482                 if (dlm_req != NULL)
1483                         ldlm_lock_dump_handle(D_ERROR, &dlm_req->lock_handle1);
1484
1485                 ldlm_callback_reply(req, -ENOTCONN);
1486                 RETURN(0);
1487         }
1488
1489         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1490
1491         /* XXX FIXME move this back to mds/handler.c, bug 249 */
1492         case LDLM_CANCEL:
1493                 CDEBUG(D_INODE, "cancel\n");
1494                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CANCEL, 0);
1495                 rc = ldlm_handle_cancel(req);
1496                 if (rc)
1497                         break;
1498                 RETURN(0);
1499         case OBD_LOG_CANCEL:
1500                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
1501                 rc = llog_origin_handle_cancel(req);
1502                 ldlm_callback_reply(req, rc);
1503                 RETURN(0);
1504         default:
1505                 CERROR("invalid opcode %d\n",
1506                        lustre_msg_get_opc(req->rq_reqmsg));
1507                 ldlm_callback_reply(req, -EINVAL);
1508         }
1509
1510         RETURN(0);
1511 }
1512
1513 void ldlm_revoke_export_locks(struct obd_export *exp)
1514 {
1515         struct list_head *locklist = &exp->exp_ldlm_data.led_held_locks;
1516         struct list_head  rpc_list;
1517         struct ldlm_lock *lock, *next;
1518         struct ldlm_lock_desc desc;
1519
1520         ENTRY;
1521         INIT_LIST_HEAD(&rpc_list);
1522
1523         spin_lock(&exp->exp_ldlm_data.led_lock);
1524         list_for_each_entry_safe(lock, next, locklist, l_export_chain) {
1525                 lock_res_and_lock(lock);
1526                 if (lock->l_req_mode != lock->l_granted_mode) {
1527                         unlock_res_and_lock(lock);
1528                         continue;
1529                 }
1530
1531                 LASSERT(lock->l_resource);
1532                 if (lock->l_resource->lr_type != LDLM_IBITS &&
1533                     lock->l_resource->lr_type != LDLM_PLAIN) {
1534                         unlock_res_and_lock(lock);
1535                         continue;
1536                 }
1537
1538                 if (lock->l_flags & LDLM_FL_AST_SENT) {
1539                         unlock_res_and_lock(lock);
1540                         continue;
1541                 }
1542
1543                 LASSERT(lock->l_blocking_ast);
1544                 LASSERT(!lock->l_blocking_lock);
1545
1546                 lock->l_flags |= LDLM_FL_AST_SENT;
1547                 unlock_res_and_lock(lock);
1548
1549                 list_move(&lock->l_export_chain, &rpc_list);
1550         }
1551         spin_unlock(&exp->exp_ldlm_data.led_lock);
1552
1553         while (!list_empty(&rpc_list)) {
1554                 lock = list_entry(rpc_list.next, struct ldlm_lock,
1555                                   l_export_chain);
1556                 list_del_init(&lock->l_export_chain);
1557
1558                 /* the desc just pretend to exclusive */
1559                 ldlm_lock2desc(lock, &desc);
1560                 desc.l_req_mode = LCK_EX;
1561                 desc.l_granted_mode = 0;
1562
1563                 lock->l_blocking_ast(lock, &desc, NULL, LDLM_CB_BLOCKING);
1564         }
1565         EXIT;
1566 }
1567
1568 #ifdef __KERNEL__
1569 static struct ldlm_bl_work_item *ldlm_bl_get_work(struct ldlm_bl_pool *blp)
1570 {
1571         struct ldlm_bl_work_item *blwi = NULL;
1572
1573         spin_lock(&blp->blp_lock);
1574         if (!list_empty(&blp->blp_list)) {
1575                 blwi = list_entry(blp->blp_list.next, struct ldlm_bl_work_item,
1576                                   blwi_entry);
1577                 list_del(&blwi->blwi_entry);
1578         }
1579         spin_unlock(&blp->blp_lock);
1580
1581         return blwi;
1582 }
1583
1584 struct ldlm_bl_thread_data {
1585         int                     bltd_num;
1586         struct ldlm_bl_pool     *bltd_blp;
1587 };
1588
1589 static int ldlm_bl_thread_main(void *arg)
1590 {
1591         struct ldlm_bl_thread_data *bltd = arg;
1592         struct ldlm_bl_pool *blp = bltd->bltd_blp;
1593         ENTRY;
1594
1595         {
1596                 char name[CFS_CURPROC_COMM_MAX];
1597                 snprintf(name, sizeof(name) - 1, "ldlm_bl_%02d",
1598                          bltd->bltd_num);
1599                 cfs_daemonize(name);
1600         }
1601
1602         atomic_inc(&blp->blp_num_threads);
1603         complete(&blp->blp_comp);
1604
1605         while(1) {
1606                 struct l_wait_info lwi = { 0 };
1607                 struct ldlm_bl_work_item *blwi = NULL;
1608
1609                 l_wait_event_exclusive(blp->blp_waitq,
1610                                        (blwi = ldlm_bl_get_work(blp)) != NULL,
1611                                        &lwi);
1612
1613                 if (blwi->blwi_ns == NULL)
1614                         break;
1615
1616                 ldlm_handle_bl_callback(blwi->blwi_ns, &blwi->blwi_ld,
1617                                         blwi->blwi_lock);
1618                 OBD_FREE(blwi, sizeof(*blwi));
1619         }
1620
1621         atomic_dec(&blp->blp_num_threads);
1622         complete(&blp->blp_comp);
1623         RETURN(0);
1624 }
1625
1626 #endif
1627
1628 static int ldlm_setup(void);
1629 static int ldlm_cleanup(int force);
1630
1631 int ldlm_get_ref(void)
1632 {
1633         int rc = 0;
1634         ENTRY;
1635         mutex_down(&ldlm_ref_sem);
1636         if (++ldlm_refcount == 1) {
1637                 rc = ldlm_setup();
1638                 if (rc)
1639                         ldlm_refcount--;
1640         }
1641         mutex_up(&ldlm_ref_sem);
1642
1643         RETURN(rc);
1644 }
1645
1646 void ldlm_put_ref(int force)
1647 {
1648         ENTRY;
1649         mutex_down(&ldlm_ref_sem);
1650         if (ldlm_refcount == 1) {
1651                 int rc = ldlm_cleanup(force);
1652                 if (rc)
1653                         CERROR("ldlm_cleanup failed: %d\n", rc);
1654                 else
1655                         ldlm_refcount--;
1656         } else {
1657                 ldlm_refcount--;
1658         }
1659         mutex_up(&ldlm_ref_sem);
1660
1661         EXIT;
1662 }
1663
1664 static int ldlm_setup(void)
1665 {
1666         struct ldlm_bl_pool *blp;
1667         int rc = 0;
1668 #ifdef __KERNEL__
1669         int i;
1670 #endif
1671         ENTRY;
1672
1673         if (ldlm_state != NULL)
1674                 RETURN(-EALREADY);
1675
1676         OBD_ALLOC(ldlm_state, sizeof(*ldlm_state));
1677         if (ldlm_state == NULL)
1678                 RETURN(-ENOMEM);
1679
1680 #ifdef LPROCFS
1681         rc = ldlm_proc_setup();
1682         if (rc != 0)
1683                 GOTO(out_free, rc);
1684 #endif
1685
1686         ldlm_state->ldlm_cb_service =
1687                 ptlrpc_init_svc(LDLM_NBUFS, LDLM_BUFSIZE, LDLM_MAXREQSIZE,
1688                                 LDLM_MAXREPSIZE, LDLM_CB_REQUEST_PORTAL,
1689                                 LDLM_CB_REPLY_PORTAL, ldlm_timeout * 900,
1690                                 ldlm_callback_handler, "ldlm_cbd",
1691                                 ldlm_svc_proc_dir, NULL, LDLM_NUM_THREADS,
1692                                 LCT_MD_THREAD|LCT_DT_THREAD);
1693
1694         if (!ldlm_state->ldlm_cb_service) {
1695                 CERROR("failed to start service\n");
1696                 GOTO(out_proc, rc = -ENOMEM);
1697         }
1698
1699         ldlm_state->ldlm_cancel_service =
1700                 ptlrpc_init_svc(LDLM_NBUFS, LDLM_BUFSIZE, LDLM_MAXREQSIZE,
1701                                 LDLM_MAXREPSIZE, LDLM_CANCEL_REQUEST_PORTAL,
1702                                 LDLM_CANCEL_REPLY_PORTAL, 30000,
1703                                 ldlm_cancel_handler, "ldlm_canceld",
1704                                 ldlm_svc_proc_dir, NULL, LDLM_NUM_THREADS,
1705                                 LCT_MD_THREAD|LCT_DT_THREAD);
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_NUM_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                                   "ldlm_cn");
1740         if (rc)
1741                 GOTO(out_thread, rc);
1742
1743         rc = ptlrpc_start_threads(NULL, ldlm_state->ldlm_cb_service,
1744                                   "ldlm_cb");
1745         if (rc)
1746                 GOTO(out_thread, rc);
1747
1748         CFS_INIT_LIST_HEAD(&expired_lock_thread.elt_expired_locks);
1749         expired_lock_thread.elt_state = ELT_STOPPED;
1750         cfs_waitq_init(&expired_lock_thread.elt_waitq);
1751
1752         CFS_INIT_LIST_HEAD(&waiting_locks_list);
1753         spin_lock_init(&waiting_locks_spinlock);
1754         cfs_timer_init(&waiting_locks_timer, waiting_locks_callback, 0);
1755
1756         rc = cfs_kernel_thread(expired_lock_main, NULL, CLONE_VM | CLONE_FILES);
1757         if (rc < 0) {
1758                 CERROR("Cannot start ldlm expired-lock thread: %d\n", rc);
1759                 GOTO(out_thread, rc);
1760         }
1761
1762         wait_event(expired_lock_thread.elt_waitq,
1763                    expired_lock_thread.elt_state == ELT_READY);
1764 #endif
1765
1766         RETURN(0);
1767
1768 #ifdef __KERNEL__
1769  out_thread:
1770         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
1771         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
1772 #endif
1773
1774  out_proc:
1775 #ifdef LPROCFS
1776         ldlm_proc_cleanup();
1777  out_free:
1778 #endif
1779         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
1780         ldlm_state = NULL;
1781         return rc;
1782 }
1783
1784 static int ldlm_cleanup(int force)
1785 {
1786 #ifdef __KERNEL__
1787         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
1788 #endif
1789         ENTRY;
1790
1791         if (!list_empty(&ldlm_namespace_list)) {
1792                 CERROR("ldlm still has namespaces; clean these up first.\n");
1793                 ldlm_dump_all_namespaces(D_DLMTRACE);
1794                 RETURN(-EBUSY);
1795         }
1796
1797 #ifdef __KERNEL__
1798         while (atomic_read(&blp->blp_num_threads) > 0) {
1799                 struct ldlm_bl_work_item blwi = { .blwi_ns = NULL };
1800
1801                 init_completion(&blp->blp_comp);
1802
1803                 spin_lock(&blp->blp_lock);
1804                 list_add_tail(&blwi.blwi_entry, &blp->blp_list);
1805                 cfs_waitq_signal(&blp->blp_waitq);
1806                 spin_unlock(&blp->blp_lock);
1807
1808                 wait_for_completion(&blp->blp_comp);
1809         }
1810         OBD_FREE(blp, sizeof(*blp));
1811
1812         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
1813         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
1814         ldlm_proc_cleanup();
1815
1816         expired_lock_thread.elt_state = ELT_TERMINATE;
1817         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
1818         wait_event(expired_lock_thread.elt_waitq,
1819                    expired_lock_thread.elt_state == ELT_STOPPED);
1820 #else
1821         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
1822         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
1823 #endif
1824
1825         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
1826         ldlm_state = NULL;
1827
1828         RETURN(0);
1829 }
1830
1831 int __init ldlm_init(void)
1832 {
1833         init_mutex(&ldlm_ref_sem);
1834         init_mutex(&ldlm_namespace_lock);
1835         ldlm_resource_slab = cfs_mem_cache_create("ldlm_resources",
1836                                                sizeof(struct ldlm_resource), 0,
1837                                                SLAB_HWCACHE_ALIGN);
1838         if (ldlm_resource_slab == NULL)
1839                 return -ENOMEM;
1840
1841         ldlm_lock_slab = cfs_mem_cache_create("ldlm_locks",
1842                                            sizeof(struct ldlm_lock), 0,
1843                                            SLAB_HWCACHE_ALIGN);
1844         if (ldlm_lock_slab == NULL) {
1845                 cfs_mem_cache_destroy(ldlm_resource_slab);
1846                 return -ENOMEM;
1847         }
1848
1849         return 0;
1850 }
1851
1852 void __exit ldlm_exit(void)
1853 {
1854         int rc;
1855         if (ldlm_refcount)
1856                 CERROR("ldlm_refcount is %d in ldlm_exit!\n", ldlm_refcount);
1857         rc = cfs_mem_cache_destroy(ldlm_resource_slab);
1858         LASSERTF(rc == 0, "couldn't free ldlm resource slab\n");
1859         rc = cfs_mem_cache_destroy(ldlm_lock_slab);
1860         LASSERTF(rc == 0, "couldn't free ldlm lock slab\n");
1861 }
1862
1863 /* ldlm_extent.c */
1864 EXPORT_SYMBOL(ldlm_extent_shift_kms);
1865
1866 /* ldlm_lock.c */
1867 EXPORT_SYMBOL(ldlm_get_processing_policy);
1868 EXPORT_SYMBOL(ldlm_lock2desc);
1869 EXPORT_SYMBOL(ldlm_register_intent);
1870 EXPORT_SYMBOL(ldlm_lockname);
1871 EXPORT_SYMBOL(ldlm_typename);
1872 EXPORT_SYMBOL(ldlm_lock2handle);
1873 EXPORT_SYMBOL(__ldlm_handle2lock);
1874 EXPORT_SYMBOL(ldlm_lock_get);
1875 EXPORT_SYMBOL(ldlm_lock_put);
1876 EXPORT_SYMBOL(ldlm_lock_match);
1877 EXPORT_SYMBOL(ldlm_lock_cancel);
1878 EXPORT_SYMBOL(ldlm_lock_addref);
1879 EXPORT_SYMBOL(ldlm_lock_decref);
1880 EXPORT_SYMBOL(ldlm_lock_decref_and_cancel);
1881 EXPORT_SYMBOL(ldlm_lock_change_resource);
1882 EXPORT_SYMBOL(ldlm_lock_set_data);
1883 EXPORT_SYMBOL(ldlm_it2str);
1884 EXPORT_SYMBOL(ldlm_lock_dump);
1885 EXPORT_SYMBOL(ldlm_lock_dump_handle);
1886 EXPORT_SYMBOL(ldlm_cancel_locks_for_export);
1887 EXPORT_SYMBOL(ldlm_reprocess_all_ns);
1888 EXPORT_SYMBOL(ldlm_lock_allow_match);
1889
1890 /* ldlm_request.c */
1891 EXPORT_SYMBOL(ldlm_completion_ast);
1892 EXPORT_SYMBOL(ldlm_blocking_ast);
1893 EXPORT_SYMBOL(ldlm_glimpse_ast);
1894 EXPORT_SYMBOL(ldlm_expired_completion_wait);
1895 EXPORT_SYMBOL(ldlm_cli_convert);
1896 EXPORT_SYMBOL(ldlm_cli_enqueue);
1897 EXPORT_SYMBOL(ldlm_cli_enqueue_fini);
1898 EXPORT_SYMBOL(ldlm_cli_enqueue_local);
1899 EXPORT_SYMBOL(ldlm_cli_cancel);
1900 EXPORT_SYMBOL(ldlm_cli_cancel_unused);
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
1908 /* ldlm_lockd.c */
1909 EXPORT_SYMBOL(ldlm_server_blocking_ast);
1910 EXPORT_SYMBOL(ldlm_server_completion_ast);
1911 EXPORT_SYMBOL(ldlm_server_glimpse_ast);
1912 EXPORT_SYMBOL(ldlm_handle_enqueue);
1913 EXPORT_SYMBOL(ldlm_handle_enqueue0);
1914 EXPORT_SYMBOL(ldlm_handle_cancel);
1915 EXPORT_SYMBOL(ldlm_handle_convert);
1916 EXPORT_SYMBOL(ldlm_handle_convert0);
1917 EXPORT_SYMBOL(ldlm_del_waiting_lock);
1918 EXPORT_SYMBOL(ldlm_get_ref);
1919 EXPORT_SYMBOL(ldlm_put_ref);
1920 EXPORT_SYMBOL(ldlm_refresh_waiting_lock);
1921 EXPORT_SYMBOL(ldlm_revoke_export_locks);
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_start_recovery_thread);
1941 EXPORT_SYMBOL(target_stop_recovery_thread);
1942 EXPORT_SYMBOL(target_handle_connect);
1943 EXPORT_SYMBOL(target_cleanup_recovery);
1944 EXPORT_SYMBOL(target_destroy_export);
1945 EXPORT_SYMBOL(target_cancel_recovery_timer);
1946 EXPORT_SYMBOL(target_send_reply);
1947 EXPORT_SYMBOL(target_queue_recovery_request);
1948 EXPORT_SYMBOL(target_handle_ping);
1949 EXPORT_SYMBOL(target_handle_disconnect);
1950
1951 /* l_lock.c */
1952 EXPORT_SYMBOL(lock_res_and_lock);
1953 EXPORT_SYMBOL(unlock_res_and_lock);
1954