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  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/ldlm/ldlm_lockd.c
  37  *
  38  * Author: Peter Braam <braam@clusterfs.com>
  39  * Author: Phil Schwan <phil@clusterfs.com>
  40  */
  41
  42 #ifndef EXPORT_SYMTAB
  43 # define EXPORT_SYMTAB
  44 #endif
  45 #define DEBUG_SUBSYSTEM S_LDLM
  46
  47 #ifdef __KERNEL__
  48 # include <libcfs/libcfs.h>
  49 #else
  50 # include <liblustre.h>
  51 #endif
  52
  53 #include <lustre_dlm.h>
  54 #include <obd_class.h>
  55 #include <libcfs/list.h>
  56 #include "ldlm_internal.h"
  57
  58 #ifdef __KERNEL__
  59 static int ldlm_num_threads;
  60 CFS_MODULE_PARM(ldlm_num_threads, "i", int, 0444,
  61                 "number of DLM service threads to start");
  62 #endif
  63
  64 extern cfs_mem_cache_t *ldlm_resource_slab;
  65 extern cfs_mem_cache_t *ldlm_lock_slab;
  66 extern struct lustre_lock ldlm_handle_lock;
  67
  68 static struct semaphore ldlm_ref_sem;
  69 static int ldlm_refcount;
  70
  71 static struct ldlm_state *ldlm_state;
  72
  73 inline cfs_time_t round_timeout(cfs_time_t timeout)
  74 {
  75         return cfs_time_seconds((int)cfs_duration_sec(cfs_time_sub(timeout, 0)) + 1);
  76 }
  77
  78 /* timeout for initial callback (AST) reply (bz10399) */
  79 static inline unsigned int ldlm_get_rq_timeout(void)
  80 {
  81         /* Non-AT value */
  82         unsigned int timeout = min(ldlm_timeout, obd_timeout / 3);
  83
  84         return timeout < 1 ? 1 : timeout;
  85 }
  86
  87 #ifdef __KERNEL__
  88 /* w_l_spinlock protects both waiting_locks_list and expired_lock_thread */
  89 static spinlock_t waiting_locks_spinlock;   /* BH lock (timer) */
  90 static struct list_head waiting_locks_list;
  91 static cfs_timer_t waiting_locks_timer;
  92
  93 static struct expired_lock_thread {
  94         cfs_waitq_t               elt_waitq;
  95         int                       elt_state;
  96         int                       elt_dump;
  97         struct list_head          elt_expired_locks;
  98 } expired_lock_thread;
  99 #endif
 100
 101 #define ELT_STOPPED   0
 102 #define ELT_READY     1
 103 #define ELT_TERMINATE 2
 104
 105 struct ldlm_bl_pool {
 106         spinlock_t              blp_lock;
 107
 108         /*
 109          * blp_prio_list is used for callbacks that should be handled
 110          * as a priority. It is used for LDLM_FL_DISCARD_DATA requests.
 111          * see bug 13843
 112          */
 113         struct list_head        blp_prio_list;
 114
 115         /*
 116          * blp_list is used for all other callbacks which are likely
 117          * to take longer to process.
 118          */
 119         struct list_head        blp_list;
 120
 121         cfs_waitq_t             blp_waitq;
 122         struct completion       blp_comp;
 123         atomic_t                blp_num_threads;
 124         atomic_t                blp_busy_threads;
 125         int                     blp_min_threads;
 126         int                     blp_max_threads;
 127 };
 128
 129 struct ldlm_bl_work_item {
 130         struct list_head        blwi_entry;
 131         struct ldlm_namespace   *blwi_ns;
 132         struct ldlm_lock_desc   blwi_ld;
 133         struct ldlm_lock        *blwi_lock;
 134         struct list_head        blwi_head;
 135         int                     blwi_count;
 136 };
 137
 138 #ifdef __KERNEL__
 139
 140 static inline int have_expired_locks(void)
 141 {
 142         int need_to_run;
 143         ENTRY;
 144
 145         spin_lock_bh(&waiting_locks_spinlock);
 146         need_to_run = !list_empty(&expired_lock_thread.elt_expired_locks);
 147         spin_unlock_bh(&waiting_locks_spinlock);
 148
 149         RETURN(need_to_run);
 150 }
 151
 152 static int expired_lock_main(void *arg)
 153 {
 154         struct list_head *expired = &expired_lock_thread.elt_expired_locks;
 155         struct l_wait_info lwi = { 0 };
 156         int do_dump;
 157
 158         ENTRY;
 159         cfs_daemonize("ldlm_elt");
 160
 161         expired_lock_thread.elt_state = ELT_READY;
 162         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 163
 164         while (1) {
 165                 l_wait_event(expired_lock_thread.elt_waitq,
 166                              have_expired_locks() ||
 167                              expired_lock_thread.elt_state == ELT_TERMINATE,
 168                              &lwi);
 169
 170                 spin_lock_bh(&waiting_locks_spinlock);
 171                 if (expired_lock_thread.elt_dump) {
 172                         spin_unlock_bh(&waiting_locks_spinlock);
 173
 174                         /* from waiting_locks_callback, but not in timer */
 175                         libcfs_debug_dumplog();
 176                         libcfs_run_lbug_upcall(__FILE__,
 177                                                 "waiting_locks_callback",
 178                                                 expired_lock_thread.elt_dump);
 179
 180                         spin_lock_bh(&waiting_locks_spinlock);
 181                         expired_lock_thread.elt_dump = 0;
 182                 }
 183
 184                 do_dump = 0;
 185
 186                 while (!list_empty(expired)) {
 187                         struct obd_export *export;
 188                         struct ldlm_lock *lock;
 189
 190                         lock = list_entry(expired->next, struct ldlm_lock,
 191                                           l_pending_chain);
 192                         if ((void *)lock < LP_POISON + CFS_PAGE_SIZE &&
 193                             (void *)lock >= LP_POISON) {
 194                                 spin_unlock_bh(&waiting_locks_spinlock);
 195                                 CERROR("free lock on elt list %p\n", lock);
 196                                 LBUG();
 197                         }
 198                         list_del_init(&lock->l_pending_chain);
 199                         if ((void *)lock->l_export < LP_POISON + CFS_PAGE_SIZE &&
 200                             (void *)lock->l_export >= LP_POISON) {
 201                                 CERROR("lock with free export on elt list %p\n",
 202                                        lock->l_export);
 203                                 lock->l_export = NULL;
 204                                 LDLM_ERROR(lock, "free export");
 205                                 /* release extra ref grabbed by
 206                                  * ldlm_add_waiting_lock() or
 207                                  * ldlm_failed_ast() */
 208                                 LDLM_LOCK_PUT(lock);
 209                                 continue;
 210                         }
 211                         export = class_export_get(lock->l_export);
 212                         spin_unlock_bh(&waiting_locks_spinlock);
 213
 214                         /* release extra ref grabbed by ldlm_add_waiting_lock()
 215                          * or ldlm_failed_ast() */
 216                         LDLM_LOCK_PUT(lock);
 217
 218                         do_dump++;
 219                         class_fail_export(export);
 220                         class_export_put(export);
 221                         spin_lock_bh(&waiting_locks_spinlock);
 222                 }
 223                 spin_unlock_bh(&waiting_locks_spinlock);
 224
 225                 if (do_dump && obd_dump_on_eviction) {
 226                         CERROR("dump the log upon eviction\n");
 227                         libcfs_debug_dumplog();
 228                 }
 229
 230                 if (expired_lock_thread.elt_state == ELT_TERMINATE)
 231                         break;
 232         }
 233
 234         expired_lock_thread.elt_state = ELT_STOPPED;
 235         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 236         RETURN(0);
 237 }
 238
 239 /**
 240  * Check if there is a request in the export request list
 241  * which prevents the lock canceling.
 242  */
 243 static int ldlm_lock_busy(struct ldlm_lock *lock)
 244 {
 245         struct ptlrpc_request *req;
 246         int match = 0;
 247         ENTRY;
 248
 249         if (lock->l_export == NULL)
 250                 return 0;
 251
 252         spin_lock(&lock->l_export->exp_lock);
 253         list_for_each_entry(req, &lock->l_export->exp_queued_rpc, rq_exp_list) {
 254                 if (req->rq_ops->hpreq_lock_match) {
 255                         match = req->rq_ops->hpreq_lock_match(req, lock);
 256                         if (match)
 257                                 break;
 258                 }
 259         }
 260         spin_unlock(&lock->l_export->exp_lock);
 261         RETURN(match);
 262 }
 263
 264 /* This is called from within a timer interrupt and cannot schedule */
 265 static void waiting_locks_callback(unsigned long unused)
 266 {
 267         struct ldlm_lock *lock, *last = NULL;
 268
 269         spin_lock_bh(&waiting_locks_spinlock);
 270         while (!list_empty(&waiting_locks_list)) {
 271                 lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
 272                                   l_pending_chain);
 273                 if (cfs_time_after(lock->l_callback_timeout, cfs_time_current())
 274                     || (lock->l_req_mode == LCK_GROUP))
 275                         break;
 276
 277                 /* Check if we need to prolong timeout */
 278                 if (!OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT) &&
 279                     ldlm_lock_busy(lock)) {
 280                         int cont = 1;
 281
 282                         if (lock->l_pending_chain.next == &waiting_locks_list)
 283                                 cont = 0;
 284
 285                         LDLM_LOCK_GET(lock);
 286                         spin_unlock_bh(&waiting_locks_spinlock);
 287                         LDLM_DEBUG(lock, "prolong the busy lock");
 288                         ldlm_refresh_waiting_lock(lock,
 289                                                   ldlm_get_enq_timeout(lock));
 290                         spin_lock_bh(&waiting_locks_spinlock);
 291
 292                         if (!cont) {
 293                                 LDLM_LOCK_PUT(lock);
 294                                 break;
 295                         }
 296
 297                         LDLM_LOCK_PUT(lock);
 298                         continue;
 299                 }
 300                 lock->l_resource->lr_namespace->ns_timeouts++;
 301                 LDLM_ERROR(lock, "lock callback timer expired after %lds: "
 302                            "evicting client at %s ",
 303                            cfs_time_current_sec()- lock->l_last_activity,
 304                            libcfs_nid2str(
 305                                    lock->l_export->exp_connection->c_peer.nid));
 306                 if (lock == last) {
 307                         LDLM_ERROR(lock, "waiting on lock multiple times");
 308                         CERROR("wll %p n/p %p/%p, l_pending %p n/p %p/%p\n",
 309                                &waiting_locks_list,
 310                                waiting_locks_list.next, waiting_locks_list.prev,
 311                                &lock->l_pending_chain,
 312                                lock->l_pending_chain.next,
 313                                lock->l_pending_chain.prev);
 314
 315                         CFS_INIT_LIST_HEAD(&waiting_locks_list);    /* HACK */
 316                         expired_lock_thread.elt_dump = __LINE__;
 317
 318                         /* LBUG(); */
 319                         CEMERG("would be an LBUG, but isn't (bug 5653)\n");
 320                         libcfs_debug_dumpstack(NULL);
 321                         /*blocks* libcfs_debug_dumplog(); */
 322                         /*blocks* libcfs_run_lbug_upcall(file, func, line); */
 323                         break;
 324                 }
 325                 last = lock;
 326
 327                 /* no needs to take an extra ref on the lock since it was in
 328                  * the waiting_locks_list and ldlm_add_waiting_lock()
 329                  * already grabbed a ref */
 330                 list_del(&lock->l_pending_chain);
 331                 list_add(&lock->l_pending_chain,
 332                          &expired_lock_thread.elt_expired_locks);
 333         }
 334
 335         if (!list_empty(&expired_lock_thread.elt_expired_locks)) {
 336                 if (obd_dump_on_timeout)
 337                         expired_lock_thread.elt_dump = __LINE__;
 338
 339                 cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 340         }
 341
 342         /*
 343          * Make sure the timer will fire again if we have any locks
 344          * left.
 345          */
 346         if (!list_empty(&waiting_locks_list)) {
 347                 cfs_time_t timeout_rounded;
 348                 lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
 349                                   l_pending_chain);
 350                 timeout_rounded = (cfs_time_t)round_timeout(lock->l_callback_timeout);
 351                 cfs_timer_arm(&waiting_locks_timer, timeout_rounded);
 352         }
 353         spin_unlock_bh(&waiting_locks_spinlock);
 354 }
 355
 356 /*
 357  * Indicate that we're waiting for a client to call us back cancelling a given
 358  * lock.  We add it to the pending-callback chain, and schedule the lock-timeout
 359  * timer to fire appropriately.  (We round up to the next second, to avoid
 360  * floods of timer firings during periods of high lock contention and traffic).
 361  * As done by ldlm_add_waiting_lock(), the caller must grab a lock reference
 362  * if it has been added to the waiting list (1 is returned).
 363  *
 364  * Called with the namespace lock held.
 365  */
 366 static int __ldlm_add_waiting_lock(struct ldlm_lock *lock, int seconds)
 367 {
 368         cfs_time_t timeout;
 369         cfs_time_t timeout_rounded;
 370
 371         if (!list_empty(&lock->l_pending_chain))
 372                 return 0;
 373
 374         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT) ||
 375             OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT))
 376                 seconds = 2;
 377
 378         timeout = cfs_time_shift(seconds);
 379         if (likely(cfs_time_after(timeout, lock->l_callback_timeout)))
 380                 lock->l_callback_timeout = timeout;
 381
 382         timeout_rounded = round_timeout(lock->l_callback_timeout);
 383
 384         if (cfs_time_before(timeout_rounded,
 385                             cfs_timer_deadline(&waiting_locks_timer)) ||
 386             !cfs_timer_is_armed(&waiting_locks_timer)) {
 387                 cfs_timer_arm(&waiting_locks_timer, timeout_rounded);
 388         }
 389         /* if the new lock has a shorter timeout than something earlier on
 390            the list, we'll wait the longer amount of time; no big deal. */
 391         list_add_tail(&lock->l_pending_chain, &waiting_locks_list); /* FIFO */
 392         return 1;
 393 }
 394
 395 static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
 396 {
 397         int ret;
 398
 399         LASSERT(!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK));
 400
 401         spin_lock_bh(&waiting_locks_spinlock);
 402         if (lock->l_destroyed) {
 403                 static cfs_time_t next;
 404                 spin_unlock_bh(&waiting_locks_spinlock);
 405                 LDLM_ERROR(lock, "not waiting on destroyed lock (bug 5653)");
 406                 if (cfs_time_after(cfs_time_current(), next)) {
 407                         next = cfs_time_shift(14400);
 408                         libcfs_debug_dumpstack(NULL);
 409                 }
 410                 return 0;
 411         }
 412
 413         ret = __ldlm_add_waiting_lock(lock, ldlm_get_enq_timeout(lock));
 414         if (ret)
 415                 /* grab ref on the lock if it has been added to the
 416                  * waiting list */
 417                 LDLM_LOCK_GET(lock);
 418         spin_unlock_bh(&waiting_locks_spinlock);
 419
 420         LDLM_DEBUG(lock, "%sadding to wait list",
 421                    ret == 0 ? "not re-" : "");
 422         return ret;
 423 }
 424
 425 /*
 426  * Remove a lock from the pending list, likely because it had its cancellation
 427  * callback arrive without incident.  This adjusts the lock-timeout timer if
 428  * needed.  Returns 0 if the lock wasn't pending after all, 1 if it was.
 429  * As done by ldlm_del_waiting_lock(), the caller must release the lock
 430  * reference when the lock is removed from any list (1 is returned).
 431  *
 432  * Called with namespace lock held.
 433  */
 434 static int __ldlm_del_waiting_lock(struct ldlm_lock *lock)
 435 {
 436         struct list_head *list_next;
 437
 438         if (list_empty(&lock->l_pending_chain))
 439                 return 0;
 440
 441         list_next = lock->l_pending_chain.next;
 442         if (lock->l_pending_chain.prev == &waiting_locks_list) {
 443                 /* Removing the head of the list, adjust timer. */
 444                 if (list_next == &waiting_locks_list) {
 445                         /* No more, just cancel. */
 446                         cfs_timer_disarm(&waiting_locks_timer);
 447                 } else {
 448                         struct ldlm_lock *next;
 449                         next = list_entry(list_next, struct ldlm_lock,
 450                                           l_pending_chain);
 451                         cfs_timer_arm(&waiting_locks_timer,
 452                                       round_timeout(next->l_callback_timeout));
 453                 }
 454         }
 455         list_del_init(&lock->l_pending_chain);
 456
 457         return 1;
 458 }
 459
 460 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
 461 {
 462         int ret;
 463
 464         if (lock->l_export == NULL) {
 465                 /* We don't have a "waiting locks list" on clients. */
 466                 LDLM_DEBUG(lock, "client lock: no-op");
 467                 return 0;
 468         }
 469
 470         spin_lock_bh(&waiting_locks_spinlock);
 471         ret = __ldlm_del_waiting_lock(lock);
 472         spin_unlock_bh(&waiting_locks_spinlock);
 473         if (ret)
 474                 /* release lock ref if it has indeed been removed
 475                  * from a list */
 476                 LDLM_LOCK_PUT(lock);
 477
 478         LDLM_DEBUG(lock, "%s", ret == 0 ? "wasn't waiting" : "removed");
 479         return ret;
 480 }
 481
 482 /*
 483  * Prolong the lock
 484  *
 485  * Called with namespace lock held.
 486  */
 487 int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
 488 {
 489         if (lock->l_export == NULL) {
 490                 /* We don't have a "waiting locks list" on clients. */
 491                 LDLM_DEBUG(lock, "client lock: no-op");
 492                 return 0;
 493         }
 494
 495         spin_lock_bh(&waiting_locks_spinlock);
 496
 497         if (list_empty(&lock->l_pending_chain)) {
 498                 spin_unlock_bh(&waiting_locks_spinlock);
 499                 LDLM_DEBUG(lock, "wasn't waiting");
 500                 return 0;
 501         }
 502
 503         /* we remove/add the lock to the waiting list, so no needs to
 504          * release/take a lock reference */
 505         __ldlm_del_waiting_lock(lock);
 506         __ldlm_add_waiting_lock(lock, timeout);
 507         spin_unlock_bh(&waiting_locks_spinlock);
 508
 509         LDLM_DEBUG(lock, "refreshed");
 510         return 1;
 511 }
 512 #else /* !__KERNEL__ */
 513
 514 static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
 515 {
 516         LASSERT(!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK));
 517         RETURN(1);
 518 }
 519
 520 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
 521 {
 522         RETURN(0);
 523 }
 524
 525 int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
 526 {
 527         RETURN(0);
 528 }
 529 #endif /* __KERNEL__ */
 530
 531 static void ldlm_failed_ast(struct ldlm_lock *lock, int rc,
 532                             const char *ast_type)
 533 {
 534         struct ptlrpc_connection *conn = lock->l_export->exp_connection;
 535         char                     *str = libcfs_nid2str(conn->c_peer.nid);
 536
 537         LCONSOLE_ERROR_MSG(0x138, "%s: A client on nid %s was evicted due "
 538                              "to a lock %s callback to %s timed out: rc %d\n",
 539                              lock->l_export->exp_obd->obd_name, str,
 540                              ast_type, obd_export_nid2str(lock->l_export), rc);
 541
 542         if (obd_dump_on_timeout)
 543                 libcfs_debug_dumplog();
 544 #ifdef __KERNEL__
 545         spin_lock_bh(&waiting_locks_spinlock);
 546         if (__ldlm_del_waiting_lock(lock) == 0)
 547                 /* the lock was not in any list, grab an extra ref before adding
 548                  * the lock to the expired list */
 549                 LDLM_LOCK_GET(lock);
 550         list_add(&lock->l_pending_chain, &expired_lock_thread.elt_expired_locks);
 551         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 552         spin_unlock_bh(&waiting_locks_spinlock);
 553 #else
 554         class_fail_export(lock->l_export);
 555 #endif
 556 }
 557
 558 static int ldlm_handle_ast_error(struct ldlm_lock *lock,
 559                                  struct ptlrpc_request *req, int rc,
 560                                  const char *ast_type)
 561 {
 562         lnet_process_id_t peer = req->rq_import->imp_connection->c_peer;
 563
 564         if (rc == -ETIMEDOUT || rc == -EINTR || rc == -ENOTCONN) {
 565                 LASSERT(lock->l_export);
 566                 if (lock->l_export->exp_libclient) {
 567                         LDLM_DEBUG(lock, "%s AST to liblustre client (nid %s)"
 568                                    " timeout, just cancelling lock", ast_type,
 569                                    libcfs_nid2str(peer.nid));
 570                         ldlm_lock_cancel(lock);
 571                         rc = -ERESTART;
 572                 } else if (lock->l_flags & LDLM_FL_CANCEL) {
 573                         LDLM_DEBUG(lock, "%s AST timeout from nid %s, but "
 574                                    "cancel was received (AST reply lost?)",
 575                                    ast_type, libcfs_nid2str(peer.nid));
 576                         ldlm_lock_cancel(lock);
 577                         rc = -ERESTART;
 578                 } else {
 579                         ldlm_del_waiting_lock(lock);
 580                         ldlm_failed_ast(lock, rc, ast_type);
 581                 }
 582         } else if (rc) {
 583                 if (rc == -EINVAL)
 584                         LDLM_DEBUG(lock, "client (nid %s) returned %d"
 585                                    " from %s AST - normal race",
 586                                    libcfs_nid2str(peer.nid),
 587                                    lustre_msg_get_status(req->rq_repmsg),
 588                                    ast_type);
 589                 else
 590                         LDLM_ERROR(lock, "client (nid %s) returned %d "
 591                                    "from %s AST", libcfs_nid2str(peer.nid),
 592                                    (req->rq_repmsg != NULL) ?
 593                                    lustre_msg_get_status(req->rq_repmsg) : 0,
 594                                    ast_type);
 595                 ldlm_lock_cancel(lock);
 596                 /* Server-side AST functions are called from ldlm_reprocess_all,
 597                  * which needs to be told to please restart its reprocessing. */
 598                 rc = -ERESTART;
 599         }
 600
 601         return rc;
 602 }
 603
 604 static int ldlm_cb_interpret(struct ptlrpc_request *req, void *data, int rc)
 605 {
 606         struct ldlm_cb_set_arg *arg;
 607         struct ldlm_lock *lock;
 608         ENTRY;
 609
 610         LASSERT(data != NULL);
 611
 612         arg = req->rq_async_args.pointer_arg[0];
 613         lock = req->rq_async_args.pointer_arg[1];
 614         LASSERT(lock != NULL);
 615         if (rc != 0) {
 616                 /* If client canceled the lock but the cancel has not
 617                  * been recieved yet, we need to update lvbo to have the
 618                  * proper attributes cached. */
 619                 if (rc == -EINVAL && arg->type == LDLM_BL_CALLBACK)
 620                         ldlm_res_lvbo_update(lock->l_resource, NULL,
 621                                              0, 1);
 622                 rc = ldlm_handle_ast_error(lock, req, rc,
 623                                            arg->type == LDLM_BL_CALLBACK
 624                                            ? "blocking" : "completion");
 625         }
 626
 627         LDLM_LOCK_PUT(lock);
 628
 629         if (rc == -ERESTART)
 630                 atomic_set(&arg->restart, 1);
 631
 632         RETURN(0);
 633 }
 634
 635 static inline int ldlm_bl_and_cp_ast_fini(struct ptlrpc_request *req,
 636                                           struct ldlm_cb_set_arg *arg,
 637                                           struct ldlm_lock *lock,
 638                                           int instant_cancel)
 639 {
 640         int rc = 0;
 641         ENTRY;
 642
 643         if (unlikely(instant_cancel)) {
 644                 rc = ptl_send_rpc(req, 1);
 645                 ptlrpc_req_finished(req);
 646                 if (rc == 0)
 647                         /* If we cancelled the lock, we need to restart
 648                          * ldlm_reprocess_queue */
 649                         atomic_set(&arg->restart, 1);
 650         } else {
 651                 LDLM_LOCK_GET(lock);
 652                 ptlrpc_set_add_req(arg->set, req);
 653         }
 654
 655         RETURN(rc);
 656 }
 657
 658 /**
 659  * Check if there are requests in the export request list which prevent
 660  * the lock canceling and make these requests high priority ones.
 661  */
 662 static void ldlm_lock_reorder_req(struct ldlm_lock *lock)
 663 {
 664         struct ptlrpc_request *req;
 665         ENTRY;
 666
 667         if (lock->l_export == NULL) {
 668                 LDLM_DEBUG(lock, "client lock: no-op");
 669                 RETURN_EXIT;
 670         }
 671
 672         spin_lock(&lock->l_export->exp_lock);
 673         list_for_each_entry(req, &lock->l_export->exp_queued_rpc, rq_exp_list) {
 674                 if (!req->rq_hp && req->rq_ops->hpreq_lock_match &&
 675                     req->rq_ops->hpreq_lock_match(req, lock))
 676                         ptlrpc_hpreq_reorder(req);
 677         }
 678         spin_unlock(&lock->l_export->exp_lock);
 679         EXIT;
 680 }
 681
 682 /*
 683  * ->l_blocking_ast() method for server-side locks. This is invoked when newly
 684  * enqueued server lock conflicts with given one.
 685  *
 686  * Sends blocking ast rpc to the client owning that lock; arms timeout timer
 687  * to wait for client response.
 688  */
 689 int ldlm_server_blocking_ast(struct ldlm_lock *lock,
 690                              struct ldlm_lock_desc *desc,
 691                              void *data, int flag)
 692 {
 693         struct ldlm_cb_set_arg *arg = data;
 694         struct ldlm_request *body;
 695         struct ptlrpc_request *req;
 696         __u32 size[] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 697                        [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 698         int instant_cancel = 0, rc;
 699         ENTRY;
 700
 701         if (flag == LDLM_CB_CANCELING) {
 702                 /* Don't need to do anything here. */
 703                 RETURN(0);
 704         }
 705
 706         LASSERT(lock);
 707         LASSERT(data != NULL);
 708
 709         ldlm_lock_reorder_req(lock);
 710
 711         req = ptlrpc_prep_req(lock->l_export->exp_imp_reverse,
 712                               LUSTRE_DLM_VERSION, LDLM_BL_CALLBACK, 2, size,
 713                               NULL);
 714         if (req == NULL)
 715                 RETURN(-ENOMEM);
 716
 717         req->rq_async_args.pointer_arg[0] = arg;
 718         req->rq_async_args.pointer_arg[1] = lock;
 719         req->rq_interpret_reply = ldlm_cb_interpret;
 720         req->rq_no_resend = 1;
 721
 722         lock_res(lock->l_resource);
 723         if (lock->l_granted_mode != lock->l_req_mode) {
 724                 /* this blocking AST will be communicated as part of the
 725                  * completion AST instead */
 726                 unlock_res(lock->l_resource);
 727                 ptlrpc_req_finished(req);
 728                 LDLM_DEBUG(lock, "lock not granted, not sending blocking AST");
 729                 RETURN(0);
 730         }
 731
 732         if (lock->l_destroyed) {
 733                 /* What's the point? */
 734                 unlock_res(lock->l_resource);
 735                 ptlrpc_req_finished(req);
 736                 RETURN(0);
 737         }
 738
 739 #if 0
 740         if (CURRENT_SECONDS - lock->l_export->exp_last_request_time > 30){
 741                 unlock_res(lock->l_resource);
 742                 ptlrpc_req_finished(req);
 743                 ldlm_failed_ast(lock, -ETIMEDOUT, "Not-attempted blocking");
 744                 RETURN(-ETIMEDOUT);
 745         }
 746 #endif
 747
 748         if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)
 749                 instant_cancel = 1;
 750
 751         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 752         body->lock_handle[0] = lock->l_remote_handle;
 753         body->lock_desc = *desc;
 754         body->lock_flags |= (lock->l_flags & LDLM_AST_FLAGS);
 755
 756         LDLM_DEBUG(lock, "server preparing blocking AST");
 757
 758         lock->l_last_activity = cfs_time_current_sec();
 759
 760         ptlrpc_req_set_repsize(req, 1, NULL);
 761         if (instant_cancel) {
 762                 unlock_res(lock->l_resource);
 763                 ldlm_lock_cancel(lock);
 764         } else {
 765                 LASSERT(lock->l_granted_mode == lock->l_req_mode);
 766                 ldlm_add_waiting_lock(lock);
 767                 unlock_res(lock->l_resource);
 768         }
 769
 770         req->rq_send_state = LUSTRE_IMP_FULL;
 771         /* ptlrpc_prep_req already set timeout */
 772         if (AT_OFF)
 773                 req->rq_timeout = ldlm_get_rq_timeout();
 774
 775         if (lock->l_export && lock->l_export->exp_nid_stats &&
 776             lock->l_export->exp_nid_stats->nid_ldlm_stats) {
 777                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
 778                                      LDLM_BL_CALLBACK - LDLM_FIRST_OPC);
 779         }
 780
 781         rc = ldlm_bl_and_cp_ast_fini(req, arg, lock, instant_cancel);
 782
 783         RETURN(rc);
 784 }
 785
 786 int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
 787 {
 788         struct ldlm_cb_set_arg *arg = data;
 789         struct ldlm_request *body;
 790         struct ptlrpc_request *req;
 791         long total_enqueue_wait;
 792         __u32 size[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 793                         [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 794         int rc, buffers = 2, instant_cancel = 0;
 795         ENTRY;
 796
 797         LASSERT(lock != NULL);
 798         LASSERT(data != NULL);
 799
 800         total_enqueue_wait = cfs_time_sub(cfs_time_current_sec(),
 801                                           lock->l_last_activity);
 802
 803         if (total_enqueue_wait > obd_timeout)
 804                 /* non-fatal with AT - change to LDLM_DEBUG? */
 805                 LDLM_WARN(lock, "enqueue wait took %lus from %lu",
 806                           total_enqueue_wait, lock->l_last_activity);
 807
 808         lock_res_and_lock(lock);
 809         if (lock->l_resource->lr_lvb_len) {
 810                 size[DLM_REQ_REC_OFF] = lock->l_resource->lr_lvb_len;
 811                 buffers = 3;
 812         }
 813         unlock_res_and_lock(lock);
 814
 815         req = ptlrpc_prep_req(lock->l_export->exp_imp_reverse,
 816                               LUSTRE_DLM_VERSION, LDLM_CP_CALLBACK, buffers,
 817                               size, NULL);
 818         if (req == NULL)
 819                 RETURN(-ENOMEM);
 820
 821         req->rq_async_args.pointer_arg[0] = arg;
 822         req->rq_async_args.pointer_arg[1] = lock;
 823         req->rq_interpret_reply = ldlm_cb_interpret;
 824         req->rq_no_resend = 1;
 825
 826         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 827         body->lock_handle[0] = lock->l_remote_handle;
 828         body->lock_flags = flags;
 829         ldlm_lock2desc(lock, &body->lock_desc);
 830
 831         if (buffers == 3) {
 832                 void *lvb;
 833
 834                 lvb = lustre_msg_buf(req->rq_reqmsg, DLM_REQ_REC_OFF,
 835                                      lock->l_resource->lr_lvb_len);
 836                 lock_res_and_lock(lock);
 837                 memcpy(lvb, lock->l_resource->lr_lvb_data,
 838                        lock->l_resource->lr_lvb_len);
 839                 unlock_res_and_lock(lock);
 840         }
 841
 842         LDLM_DEBUG(lock, "server preparing completion AST (after %lds wait)",
 843                    total_enqueue_wait);
 844
 845         /* Server-side enqueue wait time estimate, used in
 846             __ldlm_add_waiting_lock to set future enqueue timers */
 847         at_add(&lock->l_resource->lr_namespace->ns_at_estimate,
 848                total_enqueue_wait);
 849
 850         ptlrpc_req_set_repsize(req, 1, NULL);
 851
 852         req->rq_send_state = LUSTRE_IMP_FULL;
 853         /* ptlrpc_prep_req already set timeout */
 854         if (AT_OFF)
 855                 req->rq_timeout = ldlm_get_rq_timeout();
 856
 857         /* We only send real blocking ASTs after the lock is granted */
 858         lock_res_and_lock(lock);
 859         if (lock->l_flags & LDLM_FL_AST_SENT) {
 860                 body->lock_flags |= LDLM_FL_AST_SENT;
 861
 862                 /* We might get here prior to ldlm_handle_enqueue setting
 863                  * LDLM_FL_CANCEL_ON_BLOCK flag. Then we will put this lock
 864                  * into waiting list, but this is safe and similar code in
 865                  * ldlm_handle_enqueue will call ldlm_lock_cancel() still,
 866                  * that would not only cancel the lock, but will also remove
 867                  * it from waiting list */
 868                 if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) {
 869                         unlock_res_and_lock(lock);
 870                         ldlm_lock_cancel(lock);
 871                         instant_cancel = 1;
 872                         lock_res_and_lock(lock);
 873                 } else {
 874                         ldlm_add_waiting_lock(lock); /* start the lock-timeout
 875                                                          clock */
 876                 }
 877         }
 878         unlock_res_and_lock(lock);
 879
 880         if (lock->l_export && lock->l_export->exp_nid_stats &&
 881             lock->l_export->exp_nid_stats->nid_ldlm_stats) {
 882                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
 883                                      LDLM_CP_CALLBACK - LDLM_FIRST_OPC);
 884         }
 885
 886         rc = ldlm_bl_and_cp_ast_fini(req, arg, lock, instant_cancel);
 887
 888         RETURN(rc);
 889 }
 890
 891 int ldlm_server_glimpse_ast(struct ldlm_lock *lock, void *data)
 892 {
 893         struct ldlm_resource *res = lock->l_resource;
 894         struct ldlm_request *body;
 895         struct ptlrpc_request *req;
 896         __u32 size[] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 897                        [DLM_LOCKREQ_OFF]     = sizeof(*body) };
 898         int rc = 0;
 899         ENTRY;
 900
 901         LASSERT(lock != NULL);
 902
 903         req = ptlrpc_prep_req(lock->l_export->exp_imp_reverse,
 904                               LUSTRE_DLM_VERSION, LDLM_GL_CALLBACK, 2, size,
 905                               NULL);
 906         if (req == NULL)
 907                 RETURN(-ENOMEM);
 908
 909         body = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF, sizeof(*body));
 910         body->lock_handle[0] = lock->l_remote_handle;
 911         ldlm_lock2desc(lock, &body->lock_desc);
 912
 913         lock_res_and_lock(lock);
 914         size[REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
 915         unlock_res_and_lock(lock);
 916         res = lock->l_resource;
 917         ptlrpc_req_set_repsize(req, 2, size);
 918
 919         req->rq_send_state = LUSTRE_IMP_FULL;
 920         /* ptlrpc_prep_req already set timeout */
 921         if (AT_OFF)
 922                 req->rq_timeout = ldlm_get_rq_timeout();
 923
 924         if (lock->l_export && lock->l_export->exp_nid_stats &&
 925             lock->l_export->exp_nid_stats->nid_ldlm_stats) {
 926                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
 927                                      LDLM_GL_CALLBACK - LDLM_FIRST_OPC);
 928         }
 929
 930         rc = ptlrpc_queue_wait(req);
 931         if (rc == -ELDLM_NO_LOCK_DATA)
 932                 LDLM_DEBUG(lock, "lost race - client has a lock but no inode");
 933         else if (rc != 0)
 934                 rc = ldlm_handle_ast_error(lock, req, rc, "glimpse");
 935         else
 936                 rc = ldlm_res_lvbo_update(res, req,
 937                                           REPLY_REC_OFF, 1);
 938         ptlrpc_req_finished(req);
 939         if (rc == -ERESTART)
 940                 ldlm_reprocess_all(res);
 941
 942         RETURN(rc);
 943 }
 944
 945 static void ldlm_svc_get_eopc(struct ldlm_request *dlm_req,
 946                        struct lprocfs_stats *srv_stats)
 947 {
 948         int lock_type = 0, op = 0;
 949
 950         lock_type = dlm_req->lock_desc.l_resource.lr_type;
 951
 952         switch (lock_type) {
 953         case LDLM_PLAIN:
 954                 op = PTLRPC_LAST_CNTR + LDLM_PLAIN_ENQUEUE;
 955                 break;
 956         case LDLM_EXTENT:
 957                 if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT)
 958                         op = PTLRPC_LAST_CNTR + LDLM_GLIMPSE_ENQUEUE;
 959                 else
 960                         op = PTLRPC_LAST_CNTR + LDLM_EXTENT_ENQUEUE;
 961                 break;
 962         case LDLM_FLOCK:
 963                 op = PTLRPC_LAST_CNTR + LDLM_FLOCK_ENQUEUE;
 964                 break;
 965         case LDLM_IBITS:
 966                 op = PTLRPC_LAST_CNTR + LDLM_IBITS_ENQUEUE;
 967                 break;
 968         default:
 969                 op = 0;
 970                 break;
 971         }
 972
 973         if (op)
 974                 lprocfs_counter_incr(srv_stats, op);
 975
 976         return ;
 977 }
 978
 979 /*
 980  * Main server-side entry point into LDLM. This is called by ptlrpc service
 981  * threads to carry out client lock enqueueing requests.
 982  */
 983 int ldlm_handle_enqueue(struct ptlrpc_request *req,
 984                         ldlm_completion_callback completion_callback,
 985                         ldlm_blocking_callback blocking_callback,
 986                         ldlm_glimpse_callback glimpse_callback)
 987 {
 988         struct obd_device *obddev = req->rq_export->exp_obd;
 989         struct ldlm_reply *dlm_rep;
 990         struct ldlm_request *dlm_req;
 991         __u32 size[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 992                         [DLM_LOCKREPLY_OFF]   = sizeof(*dlm_rep) };
 993         int rc = 0;
 994         __u32 flags;
 995         ldlm_error_t err = ELDLM_OK;
 996         struct ldlm_lock *lock = NULL;
 997         void *cookie = NULL;
 998         ENTRY;
 999
1000         LDLM_DEBUG_NOLOCK("server-side enqueue handler START");
1001
1002         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1003                                      lustre_swab_ldlm_request);
1004         if (dlm_req == NULL) {
1005                 CERROR ("Can't unpack dlm_req\n");
1006                 GOTO(out, rc = -EFAULT);
1007         }
1008
1009         ldlm_request_cancel(req, dlm_req, LDLM_ENQUEUE_CANCEL_OFF);
1010         flags = dlm_req->lock_flags;
1011
1012         LASSERT(req->rq_export);
1013
1014         if (req->rq_rqbd->rqbd_service->srv_stats)
1015                 ldlm_svc_get_eopc(dlm_req,
1016                                   req->rq_rqbd->rqbd_service->srv_stats);
1017
1018         if (req->rq_export && req->rq_export->exp_nid_stats &&
1019             req->rq_export->exp_nid_stats->nid_ldlm_stats) {
1020                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1021                                      LDLM_ENQUEUE - LDLM_FIRST_OPC);
1022         }
1023
1024         if (dlm_req->lock_desc.l_resource.lr_type < LDLM_MIN_TYPE ||
1025             dlm_req->lock_desc.l_resource.lr_type >= LDLM_MAX_TYPE) {
1026                 DEBUG_REQ(D_ERROR, req, "invalid lock request type %d",
1027                           dlm_req->lock_desc.l_resource.lr_type);
1028                 GOTO(out, rc = -EFAULT);
1029         }
1030
1031         if (dlm_req->lock_desc.l_req_mode <= LCK_MINMODE ||
1032             dlm_req->lock_desc.l_req_mode >= LCK_MAXMODE ||
1033             dlm_req->lock_desc.l_req_mode & (dlm_req->lock_desc.l_req_mode-1)) {
1034                 DEBUG_REQ(D_ERROR, req, "invalid lock request mode %d",
1035                           dlm_req->lock_desc.l_req_mode);
1036                 GOTO(out, rc = -EFAULT);
1037         }
1038
1039         if (req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS) {
1040                 if (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN) {
1041                         DEBUG_REQ(D_ERROR, req,
1042                                   "PLAIN lock request from IBITS client?");
1043                         GOTO(out, rc = -EPROTO);
1044                 }
1045         } else if (dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS) {
1046                 DEBUG_REQ(D_ERROR, req,
1047                           "IBITS lock request from unaware client?");
1048                 GOTO(out, rc = -EPROTO);
1049         }
1050
1051 #if 0
1052         /* FIXME this makes it impossible to use LDLM_PLAIN locks -- check
1053            against server's _CONNECT_SUPPORTED flags? (I don't want to use
1054            ibits for mgc/mgs) */
1055
1056         /* INODEBITS_INTEROP: Perform conversion from plain lock to
1057          * inodebits lock if client does not support them. */
1058         if (!(req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS) &&
1059             (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN)) {
1060                 dlm_req->lock_desc.l_resource.lr_type = LDLM_IBITS;
1061                 dlm_req->lock_desc.l_policy_data.l_inodebits.bits =
1062                         MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE;
1063                 if (dlm_req->lock_desc.l_req_mode == LCK_PR)
1064                         dlm_req->lock_desc.l_req_mode = LCK_CR;
1065         }
1066 #endif
1067
1068         if (flags & LDLM_FL_REPLAY) {
1069                 /* Find an existing lock in the per-export lock hash */
1070                 lock = lustre_hash_lookup(req->rq_export->exp_lock_hash,
1071                                           (void *)&dlm_req->lock_handle[0]);
1072                 if (lock != NULL) {
1073                         DEBUG_REQ(D_DLMTRACE, req, "found existing lock cookie "
1074                                   LPX64, lock->l_handle.h_cookie);
1075                         GOTO(existing_lock, rc = 0);
1076                 }
1077         }
1078
1079         /* The lock's callback data might be set in the policy function */
1080         lock = ldlm_lock_create(obddev->obd_namespace,
1081                                 dlm_req->lock_desc.l_resource.lr_name,
1082                                 dlm_req->lock_desc.l_resource.lr_type,
1083                                 dlm_req->lock_desc.l_req_mode,
1084                                 blocking_callback, completion_callback,
1085                                 glimpse_callback, NULL, 0);
1086         if (!lock)
1087                 GOTO(out, rc = -ENOMEM);
1088
1089         lock->l_last_activity = cfs_time_current_sec();
1090         lock->l_remote_handle = dlm_req->lock_handle[0];
1091         LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
1092
1093         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_BLOCKED, obd_timeout * 2);
1094         /* Don't enqueue a lock onto the export if it has already
1095          * been evicted.  Cancel it now instead. (bug 3822) */
1096         if (req->rq_export->exp_failed) {
1097                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
1098                 GOTO(out, rc = -ENOTCONN);
1099         }
1100         lock->l_export = class_export_get(req->rq_export);
1101
1102         if (lock->l_export->exp_lock_hash)
1103                 lustre_hash_add(lock->l_export->exp_lock_hash,
1104                                 &lock->l_remote_handle, &lock->l_exp_hash);
1105
1106 existing_lock:
1107
1108         if (flags & LDLM_FL_HAS_INTENT) {
1109                 /* In this case, the reply buffer is allocated deep in
1110                  * local_lock_enqueue by the policy function. */
1111                 cookie = req;
1112         } else {
1113                 int buffers = 2;
1114
1115                 lock_res_and_lock(lock);
1116                 if (lock->l_resource->lr_lvb_len) {
1117                         size[DLM_REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
1118                         buffers = 3;
1119                 }
1120                 unlock_res_and_lock(lock);
1121
1122                 if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
1123                         GOTO(out, rc = -ENOMEM);
1124
1125                 rc = lustre_pack_reply(req, buffers, size, NULL);
1126                 if (rc)
1127                         GOTO(out, rc);
1128         }
1129
1130         if (dlm_req->lock_desc.l_resource.lr_type != LDLM_PLAIN)
1131                 lock->l_policy_data = dlm_req->lock_desc.l_policy_data;
1132         if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT)
1133                 lock->l_req_extent = lock->l_policy_data.l_extent;
1134
1135         err = ldlm_lock_enqueue(obddev->obd_namespace, &lock, cookie, (int *)&flags);
1136         if (err)
1137                 GOTO(out, err);
1138
1139         dlm_rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
1140                                  sizeof(*dlm_rep));
1141         dlm_rep->lock_flags = flags;
1142
1143         ldlm_lock2desc(lock, &dlm_rep->lock_desc);
1144         ldlm_lock2handle(lock, &dlm_rep->lock_handle);
1145
1146         /* We never send a blocking AST until the lock is granted, but
1147          * we can tell it right now */
1148         lock_res_and_lock(lock);
1149
1150         /* Now take into account flags to be inherited from original lock
1151            request both in reply to client and in our own lock flags. */
1152         dlm_rep->lock_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
1153         lock->l_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
1154
1155         /* Don't move a pending lock onto the export if it has already
1156          * been evicted.  Cancel it now instead. (bug 5683) */
1157         if (req->rq_export->exp_failed ||
1158             OBD_FAIL_CHECK_ONCE(OBD_FAIL_LDLM_ENQUEUE_OLD_EXPORT)) {
1159                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
1160                 rc = -ENOTCONN;
1161         } else if (lock->l_flags & LDLM_FL_AST_SENT) {
1162                 dlm_rep->lock_flags |= LDLM_FL_AST_SENT;
1163                 if (lock->l_granted_mode == lock->l_req_mode) {
1164                         /* Only cancel lock if it was granted, because it
1165                          * would be destroyed immediatelly and would never
1166                          * be granted in the future, causing timeouts on client.
1167                          * Not granted lock will be cancelled immediatelly after
1168                          * sending completion AST.
1169                          */
1170                         if (dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK) {
1171                                 unlock_res_and_lock(lock);
1172                                 ldlm_lock_cancel(lock);
1173                                 lock_res_and_lock(lock);
1174                         } else
1175                                 ldlm_add_waiting_lock(lock);
1176                 }
1177         }
1178         /* Make sure we never ever grant usual metadata locks to liblustre
1179            clients */
1180         if ((dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN ||
1181             dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS) &&
1182              req->rq_export->exp_libclient) {
1183                 if (!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) ||
1184                     !(dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK)) {
1185                         CERROR("Granting sync lock to libclient. "
1186                                "req fl %d, rep fl %d, lock fl "LPX64"\n",
1187                                dlm_req->lock_flags, dlm_rep->lock_flags,
1188                                lock->l_flags);
1189                         LDLM_ERROR(lock, "sync lock");
1190                         if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT) {
1191                                 struct ldlm_intent *it;
1192                                 it = lustre_msg_buf(req->rq_reqmsg,
1193                                                     DLM_INTENT_IT_OFF,
1194                                                     sizeof(*it));
1195                                 if (it != NULL) {
1196                                         CERROR("This is intent %s ("LPU64")\n",
1197                                                ldlm_it2str(it->opc), it->opc);
1198                                 }
1199                         }
1200                 }
1201         }
1202
1203         unlock_res_and_lock(lock);
1204
1205         EXIT;
1206  out:
1207         req->rq_status = rc ?: err;  /* return either error - bug 11190 */
1208         if (!req->rq_packed_final) {
1209                 err = lustre_pack_reply(req, 1, NULL, NULL);
1210                 if (rc == 0)
1211                         rc = err;
1212         }
1213
1214         /* The LOCK_CHANGED code in ldlm_lock_enqueue depends on this
1215          * ldlm_reprocess_all.  If this moves, revisit that code. -phil */
1216         if (lock) {
1217                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
1218                            "(err=%d, rc=%d)", err, rc);
1219
1220                 if (rc == 0) {
1221                         lock_res_and_lock(lock);
1222                         size[DLM_REPLY_REC_OFF] = lock->l_resource->lr_lvb_len;
1223                         if (size[DLM_REPLY_REC_OFF] > 0) {
1224                                 void *lvb = lustre_msg_buf(req->rq_repmsg,
1225                                                        DLM_REPLY_REC_OFF,
1226                                                        size[DLM_REPLY_REC_OFF]);
1227                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
1228                                          req, lock);
1229
1230                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
1231                                        size[DLM_REPLY_REC_OFF]);
1232                         }
1233                         unlock_res_and_lock(lock);
1234                 } else {
1235                         lock_res_and_lock(lock);
1236                         ldlm_resource_unlink_lock(lock);
1237                         ldlm_lock_destroy_nolock(lock);
1238                         unlock_res_and_lock(lock);
1239                 }
1240
1241                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
1242                         ldlm_reprocess_all(lock->l_resource);
1243
1244                 LDLM_LOCK_PUT(lock);
1245         }
1246
1247         LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p, rc %d)",
1248                           lock, rc);
1249
1250         return rc;
1251 }
1252
1253 int ldlm_handle_convert(struct ptlrpc_request *req)
1254 {
1255         struct ldlm_request *dlm_req;
1256         struct ldlm_reply *dlm_rep;
1257         struct ldlm_lock *lock;
1258         int rc;
1259         __u32 size[2] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
1260                         [DLM_LOCKREPLY_OFF]   = sizeof(*dlm_rep) };
1261         ENTRY;
1262
1263         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1264                                      lustre_swab_ldlm_request);
1265         if (dlm_req == NULL) {
1266                 CERROR ("Can't unpack dlm_req\n");
1267                 RETURN (-EFAULT);
1268         }
1269
1270         if (req->rq_export && req->rq_export->exp_nid_stats &&
1271             req->rq_export->exp_nid_stats->nid_ldlm_stats) {
1272                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1273                                      LDLM_CONVERT - LDLM_FIRST_OPC);
1274         }
1275
1276         rc = lustre_pack_reply(req, 2, size, NULL);
1277         if (rc)
1278                 RETURN(rc);
1279
1280         dlm_rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
1281                                  sizeof(*dlm_rep));
1282         dlm_rep->lock_flags = dlm_req->lock_flags;
1283
1284         lock = ldlm_handle2lock(&dlm_req->lock_handle[0]);
1285         if (!lock) {
1286                 req->rq_status = EINVAL;
1287         } else {
1288                 void *res = NULL;
1289
1290                 LDLM_DEBUG(lock, "server-side convert handler START");
1291
1292                 lock->l_last_activity = cfs_time_current_sec();
1293                 res = ldlm_lock_convert(lock, dlm_req->lock_desc.l_req_mode,
1294                                         &dlm_rep->lock_flags);
1295                 if (res) {
1296                         if (ldlm_del_waiting_lock(lock))
1297                                 LDLM_DEBUG(lock, "converted waiting lock");
1298                         req->rq_status = 0;
1299                 } else {
1300                         req->rq_status = EDEADLOCK;
1301                 }
1302         }
1303
1304         if (lock) {
1305                 if (!req->rq_status)
1306                         ldlm_reprocess_all(lock->l_resource);
1307                 LDLM_DEBUG(lock, "server-side convert handler END");
1308                 LDLM_LOCK_PUT(lock);
1309         } else
1310                 LDLM_DEBUG_NOLOCK("server-side convert handler END");
1311
1312         RETURN(0);
1313 }
1314
1315 /* Cancel all the locks whos handles are packed into ldlm_request */
1316 int ldlm_request_cancel(struct ptlrpc_request *req,
1317                         struct ldlm_request *dlm_req, int first)
1318 {
1319         struct ldlm_resource *res, *pres = NULL;
1320         struct ldlm_lock *lock;
1321         int i, count, done = 0;
1322         ENTRY;
1323
1324         count = dlm_req->lock_count ? dlm_req->lock_count : 1;
1325         if (first >= count)
1326                 RETURN(0);
1327
1328         /* There is no lock on the server at the replay time,
1329          * skip lock cancelling to make replay tests to pass. */
1330         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
1331                 RETURN(0);
1332
1333         LDLM_DEBUG_NOLOCK("server-side cancel handler START: %d locks",
1334                           count - first);
1335         for (i = first; i < count; i++) {
1336                 lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
1337                 if (!lock) {
1338                         LDLM_DEBUG_NOLOCK("server-side cancel handler stale "
1339                                           "lock (cookie "LPU64")",
1340                                           dlm_req->lock_handle[i].cookie);
1341                         continue;
1342                 }
1343
1344                 done++;
1345                 res = lock->l_resource;
1346                 if (res != pres) {
1347                         if (pres != NULL) {
1348                                 ldlm_reprocess_all(pres);
1349                                 ldlm_resource_putref(pres);
1350                         }
1351                         if (res != NULL) {
1352                                 ldlm_resource_getref(res);
1353                                 ldlm_res_lvbo_update(res, NULL, 0, 1);
1354                         }
1355                         pres = res;
1356                 }
1357                 ldlm_lock_cancel(lock);
1358                 LDLM_LOCK_PUT(lock);
1359         }
1360         if (pres != NULL) {
1361                 ldlm_reprocess_all(pres);
1362                 ldlm_resource_putref(pres);
1363         }
1364         LDLM_DEBUG_NOLOCK("server-side cancel handler END");
1365         RETURN(done);
1366 }
1367
1368 int ldlm_handle_cancel(struct ptlrpc_request *req)
1369 {
1370         struct ldlm_request *dlm_req;
1371         int rc;
1372         ENTRY;
1373
1374         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1375                                      lustre_swab_ldlm_request);
1376         if (dlm_req == NULL) {
1377                 CERROR("bad request buffer for cancel\n");
1378                 RETURN(-EFAULT);
1379         }
1380
1381         if (req->rq_export && req->rq_export->exp_nid_stats &&
1382             req->rq_export->exp_nid_stats->nid_ldlm_stats) {
1383                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1384                                      LDLM_CANCEL - LDLM_FIRST_OPC);
1385         }
1386
1387         rc = lustre_pack_reply(req, 1, NULL, NULL);
1388         if (rc)
1389                 RETURN(rc);
1390
1391         if (!ldlm_request_cancel(req, dlm_req, 0))
1392                 req->rq_status = ESTALE;
1393
1394         if (ptlrpc_reply(req) != 0)
1395                 LBUG();
1396
1397         RETURN(0);
1398 }
1399
1400 void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
1401                              struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
1402 {
1403         int do_ast;
1404         ENTRY;
1405
1406         LDLM_DEBUG(lock, "client blocking AST callback handler START");
1407
1408         lock_res_and_lock(lock);
1409         lock->l_flags |= LDLM_FL_CBPENDING;
1410
1411         if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)
1412                 lock->l_flags |= LDLM_FL_CANCEL;
1413
1414         do_ast = (!lock->l_readers && !lock->l_writers);
1415         unlock_res_and_lock(lock);
1416
1417         if (do_ast) {
1418                 LDLM_DEBUG(lock, "already unused, calling "
1419                            "callback (%p)", lock->l_blocking_ast);
1420                 if (lock->l_blocking_ast != NULL)
1421                         lock->l_blocking_ast(lock, ld, lock->l_ast_data,
1422                                              LDLM_CB_BLOCKING);
1423         } else {
1424                 LDLM_DEBUG(lock, "Lock still has references, will be"
1425                            " cancelled later");
1426         }
1427
1428         LDLM_DEBUG(lock, "client blocking callback handler END");
1429         LDLM_LOCK_PUT(lock);
1430         EXIT;
1431 }
1432
1433 static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
1434                                     struct ldlm_namespace *ns,
1435                                     struct ldlm_request *dlm_req,
1436                                     struct ldlm_lock *lock)
1437 {
1438         CFS_LIST_HEAD(ast_list);
1439         ENTRY;
1440
1441         LDLM_DEBUG(lock, "client completion callback handler START");
1442
1443         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE)) {
1444                 int to = cfs_time_seconds(1);
1445                 while (to > 0) {
1446                         to = schedule_timeout(to);
1447                         if (lock->l_granted_mode == lock->l_req_mode ||
1448                             lock->l_destroyed)
1449                                 break;
1450                 }
1451         }
1452
1453         lock_res_and_lock(lock);
1454         if (lock->l_destroyed ||
1455             lock->l_granted_mode == lock->l_req_mode) {
1456                 /* bug 11300: the lock has already been granted */
1457                 unlock_res_and_lock(lock);
1458                 LDLM_DEBUG(lock, "Double grant race happened");
1459                 LDLM_LOCK_PUT(lock);
1460                 EXIT;
1461                 return;
1462         }
1463
1464         /* If we receive the completion AST before the actual enqueue returned,
1465          * then we might need to switch lock modes, resources, or extents. */
1466         if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
1467                 lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
1468                 LDLM_DEBUG(lock, "completion AST, new lock mode");
1469         }
1470
1471         if (lock->l_resource->lr_type != LDLM_PLAIN) {
1472                 lock->l_policy_data = dlm_req->lock_desc.l_policy_data;
1473                 LDLM_DEBUG(lock, "completion AST, new policy data");
1474         }
1475
1476         ldlm_resource_unlink_lock(lock);
1477         if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
1478                    &lock->l_resource->lr_name,
1479                    sizeof(lock->l_resource->lr_name)) != 0) {
1480                 unlock_res_and_lock(lock);
1481                 if (ldlm_lock_change_resource(ns, lock,
1482                                 dlm_req->lock_desc.l_resource.lr_name)) {
1483                         LDLM_ERROR(lock, "Failed to allocate resource");
1484                         LDLM_LOCK_PUT(lock);
1485                         EXIT;
1486                         return;
1487                 }
1488                 LDLM_DEBUG(lock, "completion AST, new resource");
1489                 CERROR("change resource!\n");
1490                 lock_res_and_lock(lock);
1491         }
1492
1493         if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
1494                 /* BL_AST locks are not needed in lru.
1495                  * let ldlm_cancel_lru() be fast. */
1496                 ldlm_lock_remove_from_lru(lock);
1497                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_BL_AST;
1498                 LDLM_DEBUG(lock, "completion AST includes blocking AST");
1499         }
1500
1501         if (lock->l_lvb_len) {
1502                 void *lvb;
1503                 lvb = lustre_swab_reqbuf(req, DLM_REQ_REC_OFF, lock->l_lvb_len,
1504                                          lock->l_lvb_swabber);
1505                 if (lvb == NULL) {
1506                         LDLM_ERROR(lock, "completion AST did not contain "
1507                                    "expected LVB!");
1508                 } else {
1509                         memcpy(lock->l_lvb_data, lvb, lock->l_lvb_len);
1510                 }
1511         }
1512
1513         ldlm_grant_lock(lock, &ast_list);
1514         unlock_res_and_lock(lock);
1515
1516         LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
1517
1518         ldlm_run_cp_ast_work(&ast_list);
1519
1520         LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
1521                           lock);
1522         LDLM_LOCK_PUT(lock);
1523         EXIT;
1524 }
1525
1526 static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
1527                                     struct ldlm_namespace *ns,
1528                                     struct ldlm_request *dlm_req,
1529                                     struct ldlm_lock *lock)
1530 {
1531         int rc = -ENOSYS;
1532         ENTRY;
1533
1534         LDLM_DEBUG(lock, "client glimpse AST callback handler");
1535
1536         if (lock->l_glimpse_ast != NULL)
1537                 rc = lock->l_glimpse_ast(lock, req);
1538
1539         if (req->rq_repmsg != NULL) {
1540                 ptlrpc_reply(req);
1541         } else {
1542                 req->rq_status = rc;
1543                 ptlrpc_error(req);
1544         }
1545
1546         lock_res_and_lock(lock);
1547         if (lock->l_granted_mode == LCK_PW &&
1548             !lock->l_readers && !lock->l_writers &&
1549             cfs_time_after(cfs_time_current(),
1550                            cfs_time_add(lock->l_last_used,
1551                                         cfs_time_seconds(10)))) {
1552                 unlock_res_and_lock(lock);
1553                 if (ldlm_bl_to_thread_lock(ns, NULL, lock))
1554                         ldlm_handle_bl_callback(ns, NULL, lock);
1555
1556                 EXIT;
1557                 return;
1558         }
1559         unlock_res_and_lock(lock);
1560         LDLM_LOCK_PUT(lock);
1561         EXIT;
1562 }
1563
1564 static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
1565 {
1566         req->rq_status = rc;
1567         if (!req->rq_packed_final) {
1568                 rc = lustre_pack_reply(req, 1, NULL, NULL);
1569                 if (rc)
1570                         return rc;
1571         }
1572         return ptlrpc_reply(req);
1573 }
1574
1575 #ifdef __KERNEL__
1576 static int ldlm_bl_to_thread(struct ldlm_namespace *ns,
1577                              struct ldlm_lock_desc *ld, struct ldlm_lock *lock,
1578                              struct list_head *cancels, int count)
1579 {
1580         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
1581         struct ldlm_bl_work_item *blwi;
1582         ENTRY;
1583
1584         if (cancels && count == 0)
1585                 RETURN(0);
1586
1587         OBD_ALLOC(blwi, sizeof(*blwi));
1588         if (blwi == NULL)
1589                 RETURN(-ENOMEM);
1590
1591         blwi->blwi_ns = ns;
1592         if (ld != NULL)
1593                 blwi->blwi_ld = *ld;
1594         if (count) {
1595                 list_add(&blwi->blwi_head, cancels);
1596                 list_del_init(cancels);
1597                 blwi->blwi_count = count;
1598         } else {
1599                 blwi->blwi_lock = lock;
1600         }
1601         spin_lock(&blp->blp_lock);
1602         if (lock && lock->l_flags & LDLM_FL_DISCARD_DATA) {
1603                 /* add LDLM_FL_DISCARD_DATA requests to the priority list */
1604                 list_add_tail(&blwi->blwi_entry, &blp->blp_prio_list);
1605         } else {
1606                 /* other blocking callbacks are added to the regular list */
1607                 list_add_tail(&blwi->blwi_entry, &blp->blp_list);
1608         }
1609         cfs_waitq_signal(&blp->blp_waitq);
1610         spin_unlock(&blp->blp_lock);
1611
1612         RETURN(0);
1613 }
1614 #endif
1615
1616 int ldlm_bl_to_thread_lock(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
1617                            struct ldlm_lock *lock)
1618 {
1619 #ifdef __KERNEL__
1620         RETURN(ldlm_bl_to_thread(ns, ld, lock, NULL, 0));
1621 #else
1622         RETURN(-ENOSYS);
1623 #endif
1624 }
1625
1626 int ldlm_bl_to_thread_list(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
1627                            struct list_head *cancels, int count)
1628 {
1629 #ifdef __KERNEL__
1630         RETURN(ldlm_bl_to_thread(ns, ld, NULL, cancels, count));
1631 #else
1632         RETURN(-ENOSYS);
1633 #endif
1634 }
1635
1636 static int ldlm_callback_handler(struct ptlrpc_request *req)
1637 {
1638         struct ldlm_namespace *ns;
1639         struct ldlm_request *dlm_req;
1640         struct ldlm_lock *lock;
1641         int rc;
1642         ENTRY;
1643
1644         /* Requests arrive in sender's byte order.  The ptlrpc service
1645          * handler has already checked and, if necessary, byte-swapped the
1646          * incoming request message body, but I am responsible for the
1647          * message buffers. */
1648
1649         if (req->rq_export == NULL) {
1650                 struct ldlm_request *dlm_req;
1651
1652                 CDEBUG(D_RPCTRACE, "operation %d from %s with bad "
1653                        "export cookie "LPX64"; this is "
1654                        "normal if this node rebooted with a lock held\n",
1655                        lustre_msg_get_opc(req->rq_reqmsg),
1656                        libcfs_id2str(req->rq_peer),
1657                        lustre_msg_get_handle(req->rq_reqmsg)->cookie);
1658
1659                 dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF,
1660                                              sizeof(*dlm_req),
1661                                              lustre_swab_ldlm_request);
1662                 if (dlm_req != NULL)
1663                         CDEBUG(D_RPCTRACE, "--> lock cookie: "LPX64"\n",
1664                                dlm_req->lock_handle[0].cookie);
1665
1666                 ldlm_callback_reply(req, -ENOTCONN);
1667                 RETURN(0);
1668         }
1669
1670         LASSERT(req->rq_export != NULL);
1671         LASSERT(req->rq_export->exp_obd != NULL);
1672
1673         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1674         case LDLM_BL_CALLBACK:
1675                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
1676                 break;
1677         case LDLM_CP_CALLBACK:
1678                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CP_CALLBACK, 0);
1679                 break;
1680         case LDLM_GL_CALLBACK:
1681                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_GL_CALLBACK, 0);
1682                 break;
1683         case OBD_LOG_CANCEL: /* remove this eventually - for 1.4.0 compat */
1684                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
1685                 rc = llog_origin_handle_cancel(req);
1686                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_REP, 0);
1687                 ldlm_callback_reply(req, rc);
1688                 RETURN(0);
1689         case OBD_QC_CALLBACK:
1690                 OBD_FAIL_RETURN(OBD_FAIL_OBD_QC_CALLBACK_NET, 0);
1691                 rc = target_handle_qc_callback(req);
1692                 ldlm_callback_reply(req, rc);
1693                 RETURN(0);
1694         case QUOTA_DQACQ:
1695         case QUOTA_DQREL:
1696                 /* reply in handler */
1697                 rc = target_handle_dqacq_callback(req);
1698                 RETURN(0);
1699         case LLOG_ORIGIN_HANDLE_CREATE:
1700                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1701                 rc = llog_origin_handle_create(req);
1702                 ldlm_callback_reply(req, rc);
1703                 RETURN(0);
1704         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
1705                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1706                 rc = llog_origin_handle_next_block(req);
1707                 ldlm_callback_reply(req, rc);
1708                 RETURN(0);
1709         case LLOG_ORIGIN_HANDLE_READ_HEADER:
1710                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1711                 rc = llog_origin_handle_read_header(req);
1712                 ldlm_callback_reply(req, rc);
1713                 RETURN(0);
1714         case LLOG_ORIGIN_HANDLE_CLOSE:
1715                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1716                 rc = llog_origin_handle_close(req);
1717                 ldlm_callback_reply(req, rc);
1718                 RETURN(0);
1719         default:
1720                 CERROR("unknown opcode %u\n",
1721                        lustre_msg_get_opc(req->rq_reqmsg));
1722                 ldlm_callback_reply(req, -EPROTO);
1723                 RETURN(0);
1724         }
1725
1726         ns = req->rq_export->exp_obd->obd_namespace;
1727         LASSERT(ns != NULL);
1728
1729         dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF, sizeof(*dlm_req),
1730                                      lustre_swab_ldlm_request);
1731         if (dlm_req == NULL) {
1732                 CERROR ("can't unpack dlm_req\n");
1733                 ldlm_callback_reply(req, -EPROTO);
1734                 RETURN (0);
1735         }
1736
1737         /* Force a known safe race, send a cancel to the server for a lock
1738          * which the server has already started a blocking callback on. */
1739         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE) &&
1740             lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
1741                 rc = ldlm_cli_cancel(&dlm_req->lock_handle[0]);
1742                 if (rc < 0)
1743                         CERROR("ldlm_cli_cancel: %d\n", rc);
1744         }
1745
1746         lock = ldlm_handle2lock_ns(ns, &dlm_req->lock_handle[0]);
1747         if (!lock) {
1748                 CDEBUG(D_DLMTRACE, "callback on lock "LPX64" - lock "
1749                        "disappeared\n", dlm_req->lock_handle[0].cookie);
1750                 ldlm_callback_reply(req, -EINVAL);
1751                 RETURN(0);
1752         }
1753
1754         if ((lock->l_flags & LDLM_FL_FAIL_LOC) &&
1755             lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK)
1756                 OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
1757
1758         /* Copy hints/flags (e.g. LDLM_FL_DISCARD_DATA) from AST. */
1759         lock_res_and_lock(lock);
1760         lock->l_flags |= (dlm_req->lock_flags & LDLM_AST_FLAGS);
1761         if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
1762                 /* If somebody cancels lock and cache is already droped,
1763                  * or lock is failed before cp_ast received on client,
1764                  * we can tell the server we have no lock. Otherwise, we
1765                  * should send cancel after dropping the cache. */
1766                 if (((lock->l_flags & LDLM_FL_CANCELING) &&
1767                     (lock->l_flags & LDLM_FL_BL_DONE)) ||
1768                     (lock->l_flags & LDLM_FL_FAILED)) {
1769                         LDLM_DEBUG(lock, "callback on lock "
1770                                    LPX64" - lock disappeared\n",
1771                                    dlm_req->lock_handle[0].cookie);
1772                         unlock_res_and_lock(lock);
1773                         LDLM_LOCK_PUT(lock);
1774                         ldlm_callback_reply(req, -EINVAL);
1775                         RETURN(0);
1776                 }
1777                 /* BL_AST locks are not needed in lru.
1778                  * let ldlm_cancel_lru() be fast. */
1779                 ldlm_lock_remove_from_lru(lock);
1780                 lock->l_flags |= LDLM_FL_BL_AST;
1781         }
1782         unlock_res_and_lock(lock);
1783
1784         /* We want the ost thread to get this reply so that it can respond
1785          * to ost requests (write cache writeback) that might be triggered
1786          * in the callback.
1787          *
1788          * But we'd also like to be able to indicate in the reply that we're
1789          * cancelling right now, because it's unused, or have an intent result
1790          * in the reply, so we might have to push the responsibility for sending
1791          * the reply down into the AST handlers, alas. */
1792
1793         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1794         case LDLM_BL_CALLBACK:
1795                 CDEBUG(D_INODE, "blocking ast\n");
1796                 if (!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK))
1797                         ldlm_callback_reply(req, 0);
1798                 if (ldlm_bl_to_thread_lock(ns, &dlm_req->lock_desc, lock))
1799                         ldlm_handle_bl_callback(ns, &dlm_req->lock_desc, lock);
1800                 break;
1801         case LDLM_CP_CALLBACK:
1802                 CDEBUG(D_INODE, "completion ast\n");
1803                 ldlm_callback_reply(req, 0);
1804                 ldlm_handle_cp_callback(req, ns, dlm_req, lock);
1805                 break;
1806         case LDLM_GL_CALLBACK:
1807                 CDEBUG(D_INODE, "glimpse ast\n");
1808                 ldlm_handle_gl_callback(req, ns, dlm_req, lock);
1809                 break;
1810         default:
1811                 LBUG();                         /* checked above */
1812         }
1813
1814         RETURN(0);
1815 }
1816
1817 static int ldlm_cancel_handler(struct ptlrpc_request *req)
1818 {
1819         int rc;
1820         ENTRY;
1821
1822         /* Requests arrive in sender's byte order.  The ptlrpc service
1823          * handler has already checked and, if necessary, byte-swapped the
1824          * incoming request message body, but I am responsible for the
1825          * message buffers. */
1826
1827         if (req->rq_export == NULL) {
1828                 struct ldlm_request *dlm_req;
1829
1830                 CERROR("operation %d from %s with bad export cookie "LPU64"\n",
1831                        lustre_msg_get_opc(req->rq_reqmsg),
1832                        libcfs_id2str(req->rq_peer),
1833                        lustre_msg_get_handle(req->rq_reqmsg)->cookie);
1834
1835                 dlm_req = lustre_swab_reqbuf(req, DLM_LOCKREQ_OFF,
1836                                              sizeof(*dlm_req),
1837                                              lustre_swab_ldlm_request);
1838                 if (dlm_req != NULL)
1839                         ldlm_lock_dump_handle(D_ERROR,
1840                                               &dlm_req->lock_handle[0]);
1841
1842                 ldlm_callback_reply(req, -ENOTCONN);
1843                 RETURN(0);
1844         }
1845
1846         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1847
1848         /* XXX FIXME move this back to mds/handler.c, bug 249 */
1849         case LDLM_CANCEL:
1850                 CDEBUG(D_INODE, "cancel\n");
1851                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CANCEL, 0);
1852                 rc = ldlm_handle_cancel(req);
1853                 if (rc)
1854                         break;
1855                 RETURN(0);
1856         case OBD_LOG_CANCEL:
1857                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
1858                 rc = llog_origin_handle_cancel(req);
1859                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_REP, 0);
1860                 ldlm_callback_reply(req, rc);
1861                 RETURN(0);
1862         default:
1863                 CERROR("invalid opcode %d\n",
1864                        lustre_msg_get_opc(req->rq_reqmsg));
1865                 ldlm_callback_reply(req, -EINVAL);
1866         }
1867
1868         RETURN(0);
1869 }
1870
1871 #ifdef __KERNEL__
1872 static struct ldlm_bl_work_item *ldlm_bl_get_work(struct ldlm_bl_pool *blp)
1873 {
1874         struct ldlm_bl_work_item *blwi = NULL;
1875         static unsigned int num_bl = 0;
1876
1877         spin_lock(&blp->blp_lock);
1878         /* process a request from the blp_list at least every blp_num_threads */
1879         if (!list_empty(&blp->blp_list) &&
1880             (list_empty(&blp->blp_prio_list) || num_bl == 0))
1881                 blwi = list_entry(blp->blp_list.next,
1882                                   struct ldlm_bl_work_item, blwi_entry);
1883         else
1884                 if (!list_empty(&blp->blp_prio_list))
1885                         blwi = list_entry(blp->blp_prio_list.next,
1886                                           struct ldlm_bl_work_item, blwi_entry);
1887
1888         if (blwi) {
1889                 if (++num_bl >= atomic_read(&blp->blp_num_threads))
1890                         num_bl = 0;
1891                 list_del(&blwi->blwi_entry);
1892         }
1893         spin_unlock(&blp->blp_lock);
1894
1895         return blwi;
1896 }
1897
1898 /* This only contains temporary data until the thread starts */
1899 struct ldlm_bl_thread_data {
1900         char                    bltd_name[CFS_CURPROC_COMM_MAX];
1901         struct ldlm_bl_pool     *bltd_blp;
1902         struct completion       bltd_comp;
1903         int                     bltd_num;
1904 };
1905
1906 static int ldlm_bl_thread_main(void *arg);
1907
1908 static int ldlm_bl_thread_start(struct ldlm_bl_pool *blp)
1909 {
1910         struct ldlm_bl_thread_data bltd = { .bltd_blp = blp };
1911         int rc;
1912
1913         init_completion(&bltd.bltd_comp);
1914         rc = cfs_kernel_thread(ldlm_bl_thread_main, &bltd, 0);
1915         if (rc < 0) {
1916                 CERROR("cannot start LDLM thread ldlm_bl_%02d: rc %d\n",
1917                        atomic_read(&blp->blp_num_threads), rc);
1918                 return rc;
1919         }
1920         wait_for_completion(&bltd.bltd_comp);
1921
1922         return 0;
1923 }
1924
1925 static int ldlm_bl_thread_main(void *arg)
1926 {
1927         struct ldlm_bl_pool *blp;
1928         ENTRY;
1929
1930         {
1931                 struct ldlm_bl_thread_data *bltd = arg;
1932
1933                 blp = bltd->bltd_blp;
1934
1935                 bltd->bltd_num = atomic_inc_return(&blp->blp_num_threads) - 1;
1936                 atomic_inc(&blp->blp_busy_threads);
1937
1938                 snprintf(bltd->bltd_name, sizeof(bltd->bltd_name) - 1,
1939                         "ldlm_bl_%02d", bltd->bltd_num);
1940                 cfs_daemonize(bltd->bltd_name);
1941
1942                 complete(&bltd->bltd_comp);
1943                 /* cannot use bltd after this, it is only on caller's stack */
1944         }
1945
1946         while (1) {
1947                 struct l_wait_info lwi = { 0 };
1948                 struct ldlm_bl_work_item *blwi = NULL;
1949
1950                 blwi = ldlm_bl_get_work(blp);
1951
1952                 if (blwi == NULL) {
1953                         int busy;
1954
1955                         atomic_dec(&blp->blp_busy_threads);
1956                         l_wait_event_exclusive(blp->blp_waitq,
1957                                          (blwi = ldlm_bl_get_work(blp)) != NULL,
1958                                          &lwi);
1959                         busy = atomic_inc_return(&blp->blp_busy_threads);
1960
1961                         if (blwi->blwi_ns == NULL)
1962                                 /* added by ldlm_cleanup() */
1963                                 break;
1964
1965                         /* Not fatal if racy and have a few too many threads */
1966                         if (unlikely(busy < blp->blp_max_threads &&
1967                                     busy >= atomic_read(&blp->blp_num_threads)))
1968                                 /* discard the return value, we tried */
1969                                 ldlm_bl_thread_start(blp);
1970                 } else {
1971                         if (blwi->blwi_ns == NULL)
1972                                 /* added by ldlm_cleanup() */
1973                                 break;
1974                 }
1975
1976                 if (blwi->blwi_count) {
1977                         /* The special case when we cancel locks in lru
1978                          * asynchronously, we pass the list of locks here.
1979                          * Thus lock is marked LDLM_FL_CANCELING, and already
1980                          * canceled locally. */
1981                         ldlm_cli_cancel_list(&blwi->blwi_head,
1982                                              blwi->blwi_count, NULL, 0);
1983                 } else {
1984                         ldlm_handle_bl_callback(blwi->blwi_ns, &blwi->blwi_ld,
1985                                                 blwi->blwi_lock);
1986                 }
1987                 OBD_FREE(blwi, sizeof(*blwi));
1988         }
1989
1990         atomic_dec(&blp->blp_busy_threads);
1991         atomic_dec(&blp->blp_num_threads);
1992         complete(&blp->blp_comp);
1993         RETURN(0);
1994 }
1995
1996 #endif
1997
1998 /*
1999  * Export handle<->lock hash operations.
2000  */
2001 static unsigned
2002 ldlm_export_lock_hash(lustre_hash_t *lh, void *key, unsigned mask)
2003 {
2004         return lh_u64_hash(((struct lustre_handle *)key)->cookie, mask);
2005 }
2006
2007 static void *
2008 ldlm_export_lock_key(struct hlist_node *hnode)
2009 {
2010         struct ldlm_lock *lock;
2011         ENTRY;
2012
2013         lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2014         RETURN(&lock->l_remote_handle);
2015 }
2016
2017 static int
2018 ldlm_export_lock_compare(void *key, struct hlist_node *hnode)
2019 {
2020         ENTRY;
2021         RETURN(lustre_handle_equal(ldlm_export_lock_key(hnode), key));
2022 }
2023
2024 static void *
2025 ldlm_export_lock_get(struct hlist_node *hnode)
2026 {
2027         struct ldlm_lock *lock;
2028         ENTRY;
2029
2030         lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2031         LDLM_LOCK_GET(lock);
2032
2033         RETURN(lock);
2034 }
2035
2036 static void *
2037 ldlm_export_lock_put(struct hlist_node *hnode)
2038 {
2039         struct ldlm_lock *lock;
2040         ENTRY;
2041
2042         lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2043         LDLM_LOCK_PUT(lock);
2044
2045         RETURN(lock);
2046 }
2047
2048 static lustre_hash_ops_t ldlm_export_lock_ops = {
2049         .lh_hash    = ldlm_export_lock_hash,
2050         .lh_key     = ldlm_export_lock_key,
2051         .lh_compare = ldlm_export_lock_compare,
2052         .lh_get     = ldlm_export_lock_get,
2053         .lh_put     = ldlm_export_lock_put
2054 };
2055
2056 int ldlm_init_export(struct obd_export *exp)
2057 {
2058         ENTRY;
2059
2060         exp->exp_lock_hash =
2061                 lustre_hash_init(obd_uuid2str(&exp->exp_client_uuid),
2062                                  7, 16, &ldlm_export_lock_ops, LH_REHASH);
2063
2064         if (!exp->exp_lock_hash)
2065                 RETURN(-ENOMEM);
2066
2067         RETURN(0);
2068 }
2069 EXPORT_SYMBOL(ldlm_init_export);
2070
2071 void ldlm_destroy_export(struct obd_export *exp)
2072 {
2073         ENTRY;
2074         lustre_hash_exit(exp->exp_lock_hash);
2075         exp->exp_lock_hash = NULL;
2076         EXIT;
2077 }
2078 EXPORT_SYMBOL(ldlm_destroy_export);
2079
2080 static int ldlm_setup(void);
2081 static int ldlm_cleanup(void);
2082
2083 int ldlm_get_ref(void)
2084 {
2085         int rc = 0;
2086         ENTRY;
2087         mutex_down(&ldlm_ref_sem);
2088         if (++ldlm_refcount == 1) {
2089                 rc = ldlm_setup();
2090                 if (rc)
2091                         ldlm_refcount--;
2092         }
2093         mutex_up(&ldlm_ref_sem);
2094
2095         RETURN(rc);
2096 }
2097
2098 void ldlm_put_ref(void)
2099 {
2100         ENTRY;
2101         mutex_down(&ldlm_ref_sem);
2102         if (ldlm_refcount == 1) {
2103                 int rc = ldlm_cleanup();
2104                 if (rc)
2105                         CERROR("ldlm_cleanup failed: %d\n", rc);
2106                 else
2107                         ldlm_refcount--;
2108         } else {
2109                 ldlm_refcount--;
2110         }
2111         mutex_up(&ldlm_ref_sem);
2112
2113         EXIT;
2114 }
2115
2116 static int ldlm_setup(void)
2117 {
2118         struct ldlm_bl_pool *blp;
2119         int rc = 0;
2120         int ldlm_min_threads = LDLM_THREADS_AUTO_MIN;
2121         int ldlm_max_threads = LDLM_THREADS_AUTO_MAX;
2122 #ifdef __KERNEL__
2123         int i;
2124 #endif
2125         ENTRY;
2126
2127         if (ldlm_state != NULL)
2128                 RETURN(-EALREADY);
2129
2130         OBD_ALLOC(ldlm_state, sizeof(*ldlm_state));
2131         if (ldlm_state == NULL)
2132                 RETURN(-ENOMEM);
2133
2134 #ifdef LPROCFS
2135         rc = ldlm_proc_setup();
2136         if (rc != 0)
2137                 GOTO(out_free, rc);
2138 #endif
2139
2140 #ifdef __KERNEL__
2141         if (ldlm_num_threads) {
2142                 /* If ldlm_num_threads is set, it is the min and the max. */
2143                 if (ldlm_num_threads > LDLM_THREADS_AUTO_MAX)
2144                         ldlm_num_threads = LDLM_THREADS_AUTO_MAX;
2145                 if (ldlm_num_threads < LDLM_THREADS_AUTO_MIN)
2146                         ldlm_num_threads = LDLM_THREADS_AUTO_MIN;
2147                 ldlm_min_threads = ldlm_max_threads = ldlm_num_threads;
2148         }
2149 #endif
2150
2151         ldlm_state->ldlm_cb_service =
2152                 ptlrpc_init_svc(LDLM_NBUFS, LDLM_BUFSIZE, LDLM_MAXREQSIZE,
2153                                 LDLM_MAXREPSIZE, LDLM_CB_REQUEST_PORTAL,
2154                                 LDLM_CB_REPLY_PORTAL, 1800,
2155                                 ldlm_callback_handler, "ldlm_cbd",
2156                                 ldlm_svc_proc_dir, NULL,
2157                                 ldlm_min_threads, ldlm_max_threads,
2158                                 "ldlm_cb", NULL);
2159
2160         if (!ldlm_state->ldlm_cb_service) {
2161                 CERROR("failed to start service\n");
2162                 GOTO(out_proc, rc = -ENOMEM);
2163         }
2164
2165         ldlm_state->ldlm_cancel_service =
2166                 ptlrpc_init_svc(LDLM_NBUFS, LDLM_BUFSIZE, LDLM_MAXREQSIZE,
2167                                 LDLM_MAXREPSIZE, LDLM_CANCEL_REQUEST_PORTAL,
2168                                 LDLM_CANCEL_REPLY_PORTAL, 6000,
2169                                 ldlm_cancel_handler, "ldlm_canceld",
2170                                 ldlm_svc_proc_dir, NULL,
2171                                 ldlm_min_threads, ldlm_max_threads,
2172                                 "ldlm_cn", NULL);
2173
2174         if (!ldlm_state->ldlm_cancel_service) {
2175                 CERROR("failed to start service\n");
2176                 GOTO(out_proc, rc = -ENOMEM);
2177         }
2178
2179         OBD_ALLOC(blp, sizeof(*blp));
2180         if (blp == NULL)
2181                 GOTO(out_proc, rc = -ENOMEM);
2182         ldlm_state->ldlm_bl_pool = blp;
2183
2184         spin_lock_init(&blp->blp_lock);
2185         CFS_INIT_LIST_HEAD(&blp->blp_list);
2186         CFS_INIT_LIST_HEAD(&blp->blp_prio_list);
2187         cfs_waitq_init(&blp->blp_waitq);
2188         atomic_set(&blp->blp_num_threads, 0);
2189         atomic_set(&blp->blp_busy_threads, 0);
2190         blp->blp_min_threads = ldlm_min_threads;
2191         blp->blp_max_threads = ldlm_max_threads;
2192
2193 #ifdef __KERNEL__
2194         for (i = 0; i < blp->blp_min_threads; i++) {
2195                 rc = ldlm_bl_thread_start(blp);
2196                 if (rc < 0)
2197                         GOTO(out_thread, rc);
2198         }
2199
2200         rc = ptlrpc_start_threads(NULL, ldlm_state->ldlm_cancel_service);
2201         if (rc)
2202                 GOTO(out_thread, rc);
2203
2204         rc = ptlrpc_start_threads(NULL, ldlm_state->ldlm_cb_service);
2205         if (rc)
2206                 GOTO(out_thread, rc);
2207
2208         CFS_INIT_LIST_HEAD(&expired_lock_thread.elt_expired_locks);
2209         expired_lock_thread.elt_state = ELT_STOPPED;
2210         cfs_waitq_init(&expired_lock_thread.elt_waitq);
2211
2212         CFS_INIT_LIST_HEAD(&waiting_locks_list);
2213         spin_lock_init(&waiting_locks_spinlock);
2214         cfs_timer_init(&waiting_locks_timer, waiting_locks_callback, 0);
2215
2216         rc = cfs_kernel_thread(expired_lock_main, NULL, CLONE_VM | CLONE_FILES);
2217         if (rc < 0) {
2218                 CERROR("Cannot start ldlm expired-lock thread: %d\n", rc);
2219                 GOTO(out_thread, rc);
2220         }
2221
2222         wait_event(expired_lock_thread.elt_waitq,
2223                    expired_lock_thread.elt_state == ELT_READY);
2224 #endif
2225
2226 #ifdef __KERNEL__
2227         rc = ldlm_pools_init();
2228         if (rc)
2229                 GOTO(out_thread, rc);
2230 #endif
2231
2232         RETURN(0);
2233
2234 #ifdef __KERNEL__
2235  out_thread:
2236         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
2237         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
2238 #endif
2239
2240  out_proc:
2241 #ifdef LPROCFS
2242         ldlm_proc_cleanup();
2243  out_free:
2244 #endif
2245         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
2246         ldlm_state = NULL;
2247         return rc;
2248 }
2249
2250 static int ldlm_cleanup(void)
2251 {
2252 #ifdef __KERNEL__
2253         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
2254 #endif
2255         ENTRY;
2256
2257         if (!list_empty(ldlm_namespace_list(LDLM_NAMESPACE_SERVER)) ||
2258             !list_empty(ldlm_namespace_list(LDLM_NAMESPACE_CLIENT))) {
2259                 CERROR("ldlm still has namespaces; clean these up first.\n");
2260                 ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
2261                 ldlm_dump_all_namespaces(LDLM_NAMESPACE_CLIENT, D_DLMTRACE);
2262                 RETURN(-EBUSY);
2263         }
2264
2265 #ifdef __KERNEL__
2266         ldlm_pools_fini();
2267 #endif
2268
2269 #ifdef __KERNEL__
2270         while (atomic_read(&blp->blp_num_threads) > 0) {
2271                 struct ldlm_bl_work_item blwi = { .blwi_ns = NULL };
2272
2273                 init_completion(&blp->blp_comp);
2274
2275                 spin_lock(&blp->blp_lock);
2276                 list_add_tail(&blwi.blwi_entry, &blp->blp_list);
2277                 cfs_waitq_signal(&blp->blp_waitq);
2278                 spin_unlock(&blp->blp_lock);
2279
2280                 wait_for_completion(&blp->blp_comp);
2281         }
2282         OBD_FREE(blp, sizeof(*blp));
2283
2284         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
2285         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
2286         ldlm_proc_cleanup();
2287
2288         expired_lock_thread.elt_state = ELT_TERMINATE;
2289         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
2290         wait_event(expired_lock_thread.elt_waitq,
2291                    expired_lock_thread.elt_state == ELT_STOPPED);
2292 #else
2293         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
2294         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
2295 #endif
2296
2297         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
2298         ldlm_state = NULL;
2299
2300         RETURN(0);
2301 }
2302
2303 int __init ldlm_init(void)
2304 {
2305         init_mutex(&ldlm_ref_sem);
2306         init_mutex(ldlm_namespace_lock(LDLM_NAMESPACE_SERVER));
2307         init_mutex(ldlm_namespace_lock(LDLM_NAMESPACE_CLIENT));
2308         ldlm_resource_slab = cfs_mem_cache_create("ldlm_resources",
2309                                                sizeof(struct ldlm_resource), 0,
2310                                                SLAB_HWCACHE_ALIGN);
2311         if (ldlm_resource_slab == NULL)
2312                 return -ENOMEM;
2313
2314         ldlm_lock_slab = cfs_mem_cache_create("ldlm_locks",
2315                                            sizeof(struct ldlm_lock), 0,
2316                                            SLAB_HWCACHE_ALIGN);
2317         if (ldlm_lock_slab == NULL) {
2318                 cfs_mem_cache_destroy(ldlm_resource_slab);
2319                 return -ENOMEM;
2320         }
2321
2322         ldlm_interval_slab = cfs_mem_cache_create("interval_node",
2323                                         sizeof(struct ldlm_interval),
2324                                         0, SLAB_HWCACHE_ALIGN);
2325         if (ldlm_interval_slab == NULL) {
2326                 cfs_mem_cache_destroy(ldlm_resource_slab);
2327                 cfs_mem_cache_destroy(ldlm_lock_slab);
2328                 return -ENOMEM;
2329         }
2330
2331         return 0;
2332 }
2333
2334 void __exit ldlm_exit(void)
2335 {
2336         int rc;
2337         if (ldlm_refcount)
2338                 CERROR("ldlm_refcount is %d in ldlm_exit!\n", ldlm_refcount);
2339         rc = cfs_mem_cache_destroy(ldlm_resource_slab);
2340         LASSERTF(rc == 0, "couldn't free ldlm resource slab\n");
2341         rc = cfs_mem_cache_destroy(ldlm_lock_slab);
2342         LASSERTF(rc == 0, "couldn't free ldlm lock slab\n");
2343         rc = cfs_mem_cache_destroy(ldlm_interval_slab);
2344         LASSERTF(rc == 0, "couldn't free interval node slab\n");
2345 }
2346
2347 /* ldlm_extent.c */
2348 EXPORT_SYMBOL(ldlm_extent_shift_kms);
2349
2350 /* ldlm_lock.c */
2351 EXPORT_SYMBOL(ldlm_get_processing_policy);
2352 EXPORT_SYMBOL(ldlm_lock2desc);
2353 EXPORT_SYMBOL(ldlm_register_intent);
2354 EXPORT_SYMBOL(ldlm_lockname);
2355 EXPORT_SYMBOL(ldlm_typename);
2356 EXPORT_SYMBOL(ldlm_lock2handle);
2357 EXPORT_SYMBOL(__ldlm_handle2lock);
2358 EXPORT_SYMBOL(ldlm_lock_get);
2359 EXPORT_SYMBOL(ldlm_lock_put);
2360 EXPORT_SYMBOL(ldlm_lock_fast_match);
2361 EXPORT_SYMBOL(ldlm_lock_fast_release);
2362 EXPORT_SYMBOL(ldlm_lock_match);
2363 EXPORT_SYMBOL(ldlm_lock_cancel);
2364 EXPORT_SYMBOL(ldlm_lock_addref);
2365 EXPORT_SYMBOL(ldlm_lock_decref);
2366 EXPORT_SYMBOL(ldlm_lock_decref_and_cancel);
2367 EXPORT_SYMBOL(ldlm_lock_change_resource);
2368 EXPORT_SYMBOL(ldlm_lock_set_data);
2369 EXPORT_SYMBOL(ldlm_it2str);
2370 EXPORT_SYMBOL(ldlm_lock_dump);
2371 EXPORT_SYMBOL(ldlm_lock_dump_handle);
2372 EXPORT_SYMBOL(ldlm_cancel_locks_for_export);
2373 EXPORT_SYMBOL(ldlm_reprocess_all_ns);
2374 EXPORT_SYMBOL(ldlm_lock_allow_match);
2375
2376 /* ldlm_request.c */
2377 EXPORT_SYMBOL(ldlm_completion_ast);
2378 EXPORT_SYMBOL(ldlm_blocking_ast);
2379 EXPORT_SYMBOL(ldlm_glimpse_ast);
2380 EXPORT_SYMBOL(ldlm_expired_completion_wait);
2381 EXPORT_SYMBOL(ldlm_prep_enqueue_req);
2382 EXPORT_SYMBOL(ldlm_prep_elc_req);
2383 EXPORT_SYMBOL(ldlm_cli_convert);
2384 EXPORT_SYMBOL(ldlm_cli_enqueue);
2385 EXPORT_SYMBOL(ldlm_cli_enqueue_fini);
2386 EXPORT_SYMBOL(ldlm_cli_enqueue_local);
2387 EXPORT_SYMBOL(ldlm_cli_cancel);
2388 EXPORT_SYMBOL(ldlm_cli_cancel_unused);
2389 EXPORT_SYMBOL(ldlm_cli_cancel_req);
2390 EXPORT_SYMBOL(ldlm_cli_join_lru);
2391 EXPORT_SYMBOL(ldlm_replay_locks);
2392 EXPORT_SYMBOL(ldlm_resource_foreach);
2393 EXPORT_SYMBOL(ldlm_namespace_foreach);
2394 EXPORT_SYMBOL(ldlm_namespace_foreach_res);
2395 EXPORT_SYMBOL(ldlm_resource_iterate);
2396 EXPORT_SYMBOL(ldlm_cancel_resource_local);
2397 EXPORT_SYMBOL(ldlm_cli_cancel_list);
2398
2399 /* ldlm_lockd.c */
2400 EXPORT_SYMBOL(ldlm_server_blocking_ast);
2401 EXPORT_SYMBOL(ldlm_server_completion_ast);
2402 EXPORT_SYMBOL(ldlm_server_glimpse_ast);
2403 EXPORT_SYMBOL(ldlm_handle_enqueue);
2404 EXPORT_SYMBOL(ldlm_handle_cancel);
2405 EXPORT_SYMBOL(ldlm_request_cancel);
2406 EXPORT_SYMBOL(ldlm_handle_convert);
2407 EXPORT_SYMBOL(ldlm_del_waiting_lock);
2408 EXPORT_SYMBOL(ldlm_get_ref);
2409 EXPORT_SYMBOL(ldlm_put_ref);
2410 EXPORT_SYMBOL(ldlm_refresh_waiting_lock);
2411
2412 /* ldlm_resource.c */
2413 EXPORT_SYMBOL(ldlm_namespace_new);
2414 EXPORT_SYMBOL(ldlm_namespace_cleanup);
2415 EXPORT_SYMBOL(ldlm_namespace_free);
2416 EXPORT_SYMBOL(ldlm_namespace_dump);
2417 EXPORT_SYMBOL(ldlm_dump_all_namespaces);
2418 EXPORT_SYMBOL(ldlm_resource_get);
2419 EXPORT_SYMBOL(ldlm_resource_putref);
2420 EXPORT_SYMBOL(ldlm_resource_unlink_lock);
2421
2422 /* ldlm_lib.c */
2423 EXPORT_SYMBOL(client_import_add_conn);
2424 EXPORT_SYMBOL(client_import_del_conn);
2425 EXPORT_SYMBOL(client_obd_setup);
2426 EXPORT_SYMBOL(client_obd_cleanup);
2427 EXPORT_SYMBOL(client_connect_import);
2428 EXPORT_SYMBOL(client_disconnect_export);
2429 EXPORT_SYMBOL(target_abort_recovery);
2430 EXPORT_SYMBOL(target_cleanup_recovery);
2431 EXPORT_SYMBOL(target_handle_connect);
2432 EXPORT_SYMBOL(target_destroy_export);
2433 EXPORT_SYMBOL(target_cancel_recovery_timer);
2434 EXPORT_SYMBOL(target_send_reply);
2435 EXPORT_SYMBOL(target_queue_recovery_request);
2436 EXPORT_SYMBOL(target_handle_ping);
2437 EXPORT_SYMBOL(target_pack_pool_reply);
2438 EXPORT_SYMBOL(target_handle_disconnect);
2439 EXPORT_SYMBOL(target_handle_reply);
2440
2441 /* l_lock.c */
2442 EXPORT_SYMBOL(lock_res_and_lock);
2443 EXPORT_SYMBOL(unlock_res_and_lock);