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