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 (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/ldlm/ldlm_lockd.c
  39  *
  40  * Author: Peter Braam <braam@clusterfs.com>
  41  * Author: Phil Schwan <phil@clusterfs.com>
  42  */
  43
  44 #ifndef EXPORT_SYMTAB
  45 # define EXPORT_SYMTAB
  46 #endif
  47 #define DEBUG_SUBSYSTEM S_LDLM
  48
  49 #ifdef __KERNEL__
  50 # include <libcfs/libcfs.h>
  51 #else
  52 # include <liblustre.h>
  53 #endif
  54
  55 #include <lustre_dlm.h>
  56 #include <obd_class.h>
  57 #include <libcfs/list.h>
  58 #include "ldlm_internal.h"
  59
  60 #ifdef __KERNEL__
  61 static int ldlm_num_threads;
  62 CFS_MODULE_PARM(ldlm_num_threads, "i", int, 0444,
  63                 "number of DLM service threads to start");
  64 #endif
  65
  66 extern cfs_mem_cache_t *ldlm_resource_slab;
  67 extern cfs_mem_cache_t *ldlm_lock_slab;
  68 static cfs_mutex_t      ldlm_ref_mutex;
  69 static int ldlm_refcount;
  70
  71 struct ldlm_cb_async_args {
  72         struct ldlm_cb_set_arg *ca_set_arg;
  73         struct ldlm_lock       *ca_lock;
  74 };
  75
  76 /* LDLM state */
  77
  78 static struct ldlm_state *ldlm_state;
  79
  80 inline cfs_time_t round_timeout(cfs_time_t timeout)
  81 {
  82         return cfs_time_seconds((int)cfs_duration_sec(cfs_time_sub(timeout, 0)) + 1);
  83 }
  84
  85 /* timeout for initial callback (AST) reply (bz10399) */
  86 static inline unsigned int ldlm_get_rq_timeout(void)
  87 {
  88         /* Non-AT value */
  89         unsigned int timeout = min(ldlm_timeout, obd_timeout / 3);
  90
  91         return timeout < 1 ? 1 : timeout;
  92 }
  93
  94 #ifdef __KERNEL__
  95 /* w_l_spinlock protects both waiting_locks_list and expired_lock_thread */
  96 static cfs_spinlock_t waiting_locks_spinlock;   /* BH lock (timer) */
  97 static cfs_list_t waiting_locks_list;
  98 static cfs_timer_t waiting_locks_timer;
  99
 100 static struct expired_lock_thread {
 101         cfs_waitq_t               elt_waitq;
 102         int                       elt_state;
 103         int                       elt_dump;
 104         cfs_list_t                elt_expired_locks;
 105 } expired_lock_thread;
 106 #endif
 107
 108 #define ELT_STOPPED   0
 109 #define ELT_READY     1
 110 #define ELT_TERMINATE 2
 111
 112 struct ldlm_bl_pool {
 113         cfs_spinlock_t          blp_lock;
 114
 115         /*
 116          * blp_prio_list is used for callbacks that should be handled
 117          * as a priority. It is used for LDLM_FL_DISCARD_DATA requests.
 118          * see bug 13843
 119          */
 120         cfs_list_t              blp_prio_list;
 121
 122         /*
 123          * blp_list is used for all other callbacks which are likely
 124          * to take longer to process.
 125          */
 126         cfs_list_t              blp_list;
 127
 128         cfs_waitq_t             blp_waitq;
 129         cfs_completion_t        blp_comp;
 130         cfs_atomic_t            blp_num_threads;
 131         cfs_atomic_t            blp_busy_threads;
 132         int                     blp_min_threads;
 133         int                     blp_max_threads;
 134 };
 135
 136 struct ldlm_bl_work_item {
 137         cfs_list_t              blwi_entry;
 138         struct ldlm_namespace  *blwi_ns;
 139         struct ldlm_lock_desc   blwi_ld;
 140         struct ldlm_lock       *blwi_lock;
 141         cfs_list_t              blwi_head;
 142         int                     blwi_count;
 143         cfs_completion_t        blwi_comp;
 144         int                     blwi_mode;
 145         int                     blwi_mem_pressure;
 146 };
 147
 148 #ifdef __KERNEL__
 149
 150 static inline int have_expired_locks(void)
 151 {
 152         int need_to_run;
 153
 154         ENTRY;
 155         cfs_spin_lock_bh(&waiting_locks_spinlock);
 156         need_to_run = !cfs_list_empty(&expired_lock_thread.elt_expired_locks);
 157         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 158
 159         RETURN(need_to_run);
 160 }
 161
 162 static int expired_lock_main(void *arg)
 163 {
 164         cfs_list_t *expired = &expired_lock_thread.elt_expired_locks;
 165         struct l_wait_info lwi = { 0 };
 166         int do_dump;
 167
 168         ENTRY;
 169         cfs_daemonize("ldlm_elt");
 170
 171         expired_lock_thread.elt_state = ELT_READY;
 172         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 173
 174         while (1) {
 175                 l_wait_event(expired_lock_thread.elt_waitq,
 176                              have_expired_locks() ||
 177                              expired_lock_thread.elt_state == ELT_TERMINATE,
 178                              &lwi);
 179
 180                 cfs_spin_lock_bh(&waiting_locks_spinlock);
 181                 if (expired_lock_thread.elt_dump) {
 182                         struct libcfs_debug_msg_data msgdata = {
 183                                 .msg_file = __FILE__,
 184                                 .msg_fn = "waiting_locks_callback",
 185                                 .msg_line = expired_lock_thread.elt_dump };
 186                         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 187
 188                         /* from waiting_locks_callback, but not in timer */
 189                         libcfs_debug_dumplog();
 190                         libcfs_run_lbug_upcall(&msgdata);
 191
 192                         cfs_spin_lock_bh(&waiting_locks_spinlock);
 193                         expired_lock_thread.elt_dump = 0;
 194                 }
 195
 196                 do_dump = 0;
 197
 198                 while (!cfs_list_empty(expired)) {
 199                         struct obd_export *export;
 200                         struct ldlm_lock *lock;
 201
 202                         lock = cfs_list_entry(expired->next, struct ldlm_lock,
 203                                           l_pending_chain);
 204                         if ((void *)lock < LP_POISON + CFS_PAGE_SIZE &&
 205                             (void *)lock >= LP_POISON) {
 206                                 cfs_spin_unlock_bh(&waiting_locks_spinlock);
 207                                 CERROR("free lock on elt list %p\n", lock);
 208                                 LBUG();
 209                         }
 210                         cfs_list_del_init(&lock->l_pending_chain);
 211                         if ((void *)lock->l_export < LP_POISON + CFS_PAGE_SIZE &&
 212                             (void *)lock->l_export >= LP_POISON) {
 213                                 CERROR("lock with free export on elt list %p\n",
 214                                        lock->l_export);
 215                                 lock->l_export = NULL;
 216                                 LDLM_ERROR(lock, "free export");
 217                                 /* release extra ref grabbed by
 218                                  * ldlm_add_waiting_lock() or
 219                                  * ldlm_failed_ast() */
 220                                 LDLM_LOCK_RELEASE(lock);
 221                                 continue;
 222                         }
 223                         export = class_export_lock_get(lock->l_export, lock);
 224                         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 225
 226                         do_dump++;
 227                         class_fail_export(export);
 228                         class_export_lock_put(export, lock);
 229
 230                         /* release extra ref grabbed by ldlm_add_waiting_lock()
 231                          * or ldlm_failed_ast() */
 232                         LDLM_LOCK_RELEASE(lock);
 233
 234                         cfs_spin_lock_bh(&waiting_locks_spinlock);
 235                 }
 236                 cfs_spin_unlock_bh(&waiting_locks_spinlock);
 237
 238                 if (do_dump && obd_dump_on_eviction) {
 239                         CERROR("dump the log upon eviction\n");
 240                         libcfs_debug_dumplog();
 241                 }
 242
 243                 if (expired_lock_thread.elt_state == ELT_TERMINATE)
 244                         break;
 245         }
 246
 247         expired_lock_thread.elt_state = ELT_STOPPED;
 248         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 249         RETURN(0);
 250 }
 251
 252 static int ldlm_add_waiting_lock(struct ldlm_lock *lock);
 253
 254 /**
 255  * Check if there is a request in the export request list
 256  * which prevents the lock canceling.
 257  */
 258 static int ldlm_lock_busy(struct ldlm_lock *lock)
 259 {
 260         struct ptlrpc_request *req;
 261         int match = 0;
 262         ENTRY;
 263
 264         if (lock->l_export == NULL)
 265                 return 0;
 266
 267         cfs_spin_lock_bh(&lock->l_export->exp_rpc_lock);
 268         cfs_list_for_each_entry(req, &lock->l_export->exp_hp_rpcs,
 269                                 rq_exp_list) {
 270                 if (req->rq_ops->hpreq_lock_match) {
 271                         match = req->rq_ops->hpreq_lock_match(req, lock);
 272                         if (match)
 273                                 break;
 274                 }
 275         }
 276         cfs_spin_unlock_bh(&lock->l_export->exp_rpc_lock);
 277         RETURN(match);
 278 }
 279
 280 /* This is called from within a timer interrupt and cannot schedule */
 281 static void waiting_locks_callback(unsigned long unused)
 282 {
 283         struct ldlm_lock *lock;
 284
 285 repeat:
 286         cfs_spin_lock_bh(&waiting_locks_spinlock);
 287         while (!cfs_list_empty(&waiting_locks_list)) {
 288                 lock = cfs_list_entry(waiting_locks_list.next, struct ldlm_lock,
 289                                       l_pending_chain);
 290                 if (cfs_time_after(lock->l_callback_timeout,
 291                                    cfs_time_current()) ||
 292                     (lock->l_req_mode == LCK_GROUP))
 293                         break;
 294
 295                 if (ptlrpc_check_suspend()) {
 296                         /* there is a case when we talk to one mds, holding
 297                          * lock from another mds. this way we easily can get
 298                          * here, if second mds is being recovered. so, we
 299                          * suspend timeouts. bug 6019 */
 300
 301                         LDLM_ERROR(lock, "recharge timeout: %s@%s nid %s ",
 302                                    lock->l_export->exp_client_uuid.uuid,
 303                                    lock->l_export->exp_connection->c_remote_uuid.uuid,
 304                                    libcfs_nid2str(lock->l_export->exp_connection->c_peer.nid));
 305
 306                         cfs_list_del_init(&lock->l_pending_chain);
 307                         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 308                         ldlm_add_waiting_lock(lock);
 309                         goto repeat;
 310                 }
 311
 312                 /* if timeout overlaps the activation time of suspended timeouts
 313                  * then extend it to give a chance for client to reconnect */
 314                 if (cfs_time_before(cfs_time_sub(lock->l_callback_timeout,
 315                                                  cfs_time_seconds(obd_timeout)/2),
 316                                     ptlrpc_suspend_wakeup_time())) {
 317                         LDLM_ERROR(lock, "extend timeout due to recovery: %s@%s nid %s ",
 318                                    lock->l_export->exp_client_uuid.uuid,
 319                                    lock->l_export->exp_connection->c_remote_uuid.uuid,
 320                                    libcfs_nid2str(lock->l_export->exp_connection->c_peer.nid));
 321
 322                         cfs_list_del_init(&lock->l_pending_chain);
 323                         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 324                         ldlm_add_waiting_lock(lock);
 325                         goto repeat;
 326                 }
 327
 328                 /* Check if we need to prolong timeout */
 329                 if (!OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT) &&
 330                     ldlm_lock_busy(lock)) {
 331                         int cont = 1;
 332
 333                         if (lock->l_pending_chain.next == &waiting_locks_list)
 334                                 cont = 0;
 335
 336                         LDLM_LOCK_GET(lock);
 337
 338                         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 339                         LDLM_DEBUG(lock, "prolong the busy lock");
 340                         ldlm_refresh_waiting_lock(lock,
 341                                                   ldlm_get_enq_timeout(lock));
 342                         cfs_spin_lock_bh(&waiting_locks_spinlock);
 343
 344                         if (!cont) {
 345                                 LDLM_LOCK_RELEASE(lock);
 346                                 break;
 347                         }
 348
 349                         LDLM_LOCK_RELEASE(lock);
 350                         continue;
 351                 }
 352                 ldlm_lock_to_ns(lock)->ns_timeouts++;
 353                 LDLM_ERROR(lock, "lock callback timer expired after %lds: "
 354                            "evicting client at %s ",
 355                            cfs_time_current_sec()- lock->l_last_activity,
 356                            libcfs_nid2str(
 357                                    lock->l_export->exp_connection->c_peer.nid));
 358
 359                 /* no needs to take an extra ref on the lock since it was in
 360                  * the waiting_locks_list and ldlm_add_waiting_lock()
 361                  * already grabbed a ref */
 362                 cfs_list_del(&lock->l_pending_chain);
 363                 cfs_list_add(&lock->l_pending_chain,
 364                              &expired_lock_thread.elt_expired_locks);
 365         }
 366
 367         if (!cfs_list_empty(&expired_lock_thread.elt_expired_locks)) {
 368                 if (obd_dump_on_timeout)
 369                         expired_lock_thread.elt_dump = __LINE__;
 370
 371                 cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 372         }
 373
 374         /*
 375          * Make sure the timer will fire again if we have any locks
 376          * left.
 377          */
 378         if (!cfs_list_empty(&waiting_locks_list)) {
 379                 cfs_time_t timeout_rounded;
 380                 lock = cfs_list_entry(waiting_locks_list.next, struct ldlm_lock,
 381                                       l_pending_chain);
 382                 timeout_rounded = (cfs_time_t)round_timeout(lock->l_callback_timeout);
 383                 cfs_timer_arm(&waiting_locks_timer, timeout_rounded);
 384         }
 385         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 386 }
 387
 388 /*
 389  * Indicate that we're waiting for a client to call us back cancelling a given
 390  * lock.  We add it to the pending-callback chain, and schedule the lock-timeout
 391  * timer to fire appropriately.  (We round up to the next second, to avoid
 392  * floods of timer firings during periods of high lock contention and traffic).
 393  * As done by ldlm_add_waiting_lock(), the caller must grab a lock reference
 394  * if it has been added to the waiting list (1 is returned).
 395  *
 396  * Called with the namespace lock held.
 397  */
 398 static int __ldlm_add_waiting_lock(struct ldlm_lock *lock, int seconds)
 399 {
 400         cfs_time_t timeout;
 401         cfs_time_t timeout_rounded;
 402
 403         if (!cfs_list_empty(&lock->l_pending_chain))
 404                 return 0;
 405
 406         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT) ||
 407             OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT))
 408                 seconds = 1;
 409
 410         timeout = cfs_time_shift(seconds);
 411         if (likely(cfs_time_after(timeout, lock->l_callback_timeout)))
 412                 lock->l_callback_timeout = timeout;
 413
 414         timeout_rounded = round_timeout(lock->l_callback_timeout);
 415
 416         if (cfs_time_before(timeout_rounded,
 417                             cfs_timer_deadline(&waiting_locks_timer)) ||
 418             !cfs_timer_is_armed(&waiting_locks_timer)) {
 419                 cfs_timer_arm(&waiting_locks_timer, timeout_rounded);
 420         }
 421         /* if the new lock has a shorter timeout than something earlier on
 422            the list, we'll wait the longer amount of time; no big deal. */
 423         /* FIFO */
 424         cfs_list_add_tail(&lock->l_pending_chain, &waiting_locks_list);
 425         return 1;
 426 }
 427
 428 static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
 429 {
 430         int ret;
 431         int timeout = ldlm_get_enq_timeout(lock);
 432
 433         LASSERT(!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK));
 434
 435         cfs_spin_lock_bh(&waiting_locks_spinlock);
 436         if (lock->l_destroyed) {
 437                 static cfs_time_t next;
 438                 cfs_spin_unlock_bh(&waiting_locks_spinlock);
 439                 LDLM_ERROR(lock, "not waiting on destroyed lock (bug 5653)");
 440                 if (cfs_time_after(cfs_time_current(), next)) {
 441                         next = cfs_time_shift(14400);
 442                         libcfs_debug_dumpstack(NULL);
 443                 }
 444                 return 0;
 445         }
 446
 447         ret = __ldlm_add_waiting_lock(lock, timeout);
 448         if (ret) {
 449                 /* grab ref on the lock if it has been added to the
 450                  * waiting list */
 451                 LDLM_LOCK_GET(lock);
 452         }
 453         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 454
 455         if (ret) {
 456                 cfs_spin_lock_bh(&lock->l_export->exp_bl_list_lock);
 457                 if (cfs_list_empty(&lock->l_exp_list))
 458                         cfs_list_add(&lock->l_exp_list,
 459                                      &lock->l_export->exp_bl_list);
 460                 cfs_spin_unlock_bh(&lock->l_export->exp_bl_list_lock);
 461         }
 462
 463         LDLM_DEBUG(lock, "%sadding to wait list(timeout: %d, AT: %s)",
 464                    ret == 0 ? "not re-" : "", timeout,
 465                    AT_OFF ? "off" : "on");
 466         return ret;
 467 }
 468
 469 /*
 470  * Remove a lock from the pending list, likely because it had its cancellation
 471  * callback arrive without incident.  This adjusts the lock-timeout timer if
 472  * needed.  Returns 0 if the lock wasn't pending after all, 1 if it was.
 473  * As done by ldlm_del_waiting_lock(), the caller must release the lock
 474  * reference when the lock is removed from any list (1 is returned).
 475  *
 476  * Called with namespace lock held.
 477  */
 478 static int __ldlm_del_waiting_lock(struct ldlm_lock *lock)
 479 {
 480         cfs_list_t *list_next;
 481
 482         if (cfs_list_empty(&lock->l_pending_chain))
 483                 return 0;
 484
 485         list_next = lock->l_pending_chain.next;
 486         if (lock->l_pending_chain.prev == &waiting_locks_list) {
 487                 /* Removing the head of the list, adjust timer. */
 488                 if (list_next == &waiting_locks_list) {
 489                         /* No more, just cancel. */
 490                         cfs_timer_disarm(&waiting_locks_timer);
 491                 } else {
 492                         struct ldlm_lock *next;
 493                         next = cfs_list_entry(list_next, struct ldlm_lock,
 494                                               l_pending_chain);
 495                         cfs_timer_arm(&waiting_locks_timer,
 496                                       round_timeout(next->l_callback_timeout));
 497                 }
 498         }
 499         cfs_list_del_init(&lock->l_pending_chain);
 500
 501         return 1;
 502 }
 503
 504 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
 505 {
 506         int ret;
 507
 508         if (lock->l_export == NULL) {
 509                 /* We don't have a "waiting locks list" on clients. */
 510                 CDEBUG(D_DLMTRACE, "Client lock %p : no-op\n", lock);
 511                 return 0;
 512         }
 513
 514         cfs_spin_lock_bh(&waiting_locks_spinlock);
 515         ret = __ldlm_del_waiting_lock(lock);
 516         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 517
 518         /* remove the lock out of export blocking list */
 519         cfs_spin_lock_bh(&lock->l_export->exp_bl_list_lock);
 520         cfs_list_del_init(&lock->l_exp_list);
 521         cfs_spin_unlock_bh(&lock->l_export->exp_bl_list_lock);
 522
 523         if (ret) {
 524                 /* release lock ref if it has indeed been removed
 525                  * from a list */
 526                 LDLM_LOCK_RELEASE(lock);
 527         }
 528
 529         LDLM_DEBUG(lock, "%s", ret == 0 ? "wasn't waiting" : "removed");
 530         return ret;
 531 }
 532
 533 /*
 534  * Prolong the lock
 535  *
 536  * Called with namespace lock held.
 537  */
 538 int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
 539 {
 540         if (lock->l_export == NULL) {
 541                 /* We don't have a "waiting locks list" on clients. */
 542                 LDLM_DEBUG(lock, "client lock: no-op");
 543                 return 0;
 544         }
 545
 546         cfs_spin_lock_bh(&waiting_locks_spinlock);
 547
 548         if (cfs_list_empty(&lock->l_pending_chain)) {
 549                 cfs_spin_unlock_bh(&waiting_locks_spinlock);
 550                 LDLM_DEBUG(lock, "wasn't waiting");
 551                 return 0;
 552         }
 553
 554         /* we remove/add the lock to the waiting list, so no needs to
 555          * release/take a lock reference */
 556         __ldlm_del_waiting_lock(lock);
 557         __ldlm_add_waiting_lock(lock, timeout);
 558         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 559
 560         LDLM_DEBUG(lock, "refreshed");
 561         return 1;
 562 }
 563 #else /* !__KERNEL__ */
 564
 565 int ldlm_del_waiting_lock(struct ldlm_lock *lock)
 566 {
 567         RETURN(0);
 568 }
 569
 570 int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
 571 {
 572         RETURN(0);
 573 }
 574 #endif /* __KERNEL__ */
 575
 576 #ifdef HAVE_SERVER_SUPPORT
 577 # ifndef __KERNEL__
 578 static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
 579 {
 580         LASSERT(!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK));
 581         RETURN(1);
 582 }
 583 # endif
 584
 585 static void ldlm_failed_ast(struct ldlm_lock *lock, int rc,
 586                             const char *ast_type)
 587 {
 588         LCONSOLE_ERROR_MSG(0x138, "%s: A client on nid %s was evicted due "
 589                            "to a lock %s callback time out: rc %d\n",
 590                            lock->l_export->exp_obd->obd_name,
 591                            obd_export_nid2str(lock->l_export), ast_type, rc);
 592
 593         if (obd_dump_on_timeout)
 594                 libcfs_debug_dumplog();
 595 #ifdef __KERNEL__
 596         cfs_spin_lock_bh(&waiting_locks_spinlock);
 597         if (__ldlm_del_waiting_lock(lock) == 0)
 598                 /* the lock was not in any list, grab an extra ref before adding
 599                  * the lock to the expired list */
 600                 LDLM_LOCK_GET(lock);
 601         cfs_list_add(&lock->l_pending_chain,
 602                      &expired_lock_thread.elt_expired_locks);
 603         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
 604         cfs_spin_unlock_bh(&waiting_locks_spinlock);
 605 #else
 606         class_fail_export(lock->l_export);
 607 #endif
 608 }
 609
 610 static int ldlm_handle_ast_error(struct ldlm_lock *lock,
 611                                  struct ptlrpc_request *req, int rc,
 612                                  const char *ast_type)
 613 {
 614         lnet_process_id_t peer = req->rq_import->imp_connection->c_peer;
 615
 616         if (rc == -ETIMEDOUT || rc == -EINTR || rc == -ENOTCONN) {
 617                 LASSERT(lock->l_export);
 618                 if (lock->l_export->exp_libclient) {
 619                         LDLM_DEBUG(lock, "%s AST to liblustre client (nid %s)"
 620                                    " timeout, just cancelling lock", ast_type,
 621                                    libcfs_nid2str(peer.nid));
 622                         ldlm_lock_cancel(lock);
 623                         rc = -ERESTART;
 624                 } else if (lock->l_flags & LDLM_FL_CANCEL) {
 625                         LDLM_DEBUG(lock, "%s AST timeout from nid %s, but "
 626                                    "cancel was received (AST reply lost?)",
 627                                    ast_type, libcfs_nid2str(peer.nid));
 628                         ldlm_lock_cancel(lock);
 629                         rc = -ERESTART;
 630                 } else {
 631                         ldlm_del_waiting_lock(lock);
 632                         ldlm_failed_ast(lock, rc, ast_type);
 633                 }
 634         } else if (rc) {
 635                 if (rc == -EINVAL) {
 636                         struct ldlm_resource *res = lock->l_resource;
 637                         LDLM_DEBUG(lock, "client (nid %s) returned %d"
 638                                " from %s AST - normal race",
 639                                libcfs_nid2str(peer.nid),
 640                                req->rq_repmsg ?
 641                                lustre_msg_get_status(req->rq_repmsg) : -1,
 642                                ast_type);
 643                         if (res) {
 644                                 /* update lvbo to return proper attributes.
 645                                  * see bug 23174 */
 646                                 ldlm_resource_getref(res);
 647                                 ldlm_res_lvbo_update(res, NULL, 1);
 648                                 ldlm_resource_putref(res);
 649                         }
 650
 651                 } else {
 652                         LDLM_ERROR(lock, "client (nid %s) returned %d "
 653                                    "from %s AST", libcfs_nid2str(peer.nid),
 654                                    (req->rq_repmsg != NULL) ?
 655                                    lustre_msg_get_status(req->rq_repmsg) : 0,
 656                                    ast_type);
 657                 }
 658                 ldlm_lock_cancel(lock);
 659                 /* Server-side AST functions are called from ldlm_reprocess_all,
 660                  * which needs to be told to please restart its reprocessing. */
 661                 rc = -ERESTART;
 662         }
 663
 664         return rc;
 665 }
 666
 667 static int ldlm_cb_interpret(const struct lu_env *env,
 668                              struct ptlrpc_request *req, void *data, int rc)
 669 {
 670         struct ldlm_cb_async_args *ca   = data;
 671         struct ldlm_lock          *lock = ca->ca_lock;
 672         struct ldlm_cb_set_arg    *arg  = ca->ca_set_arg;
 673         ENTRY;
 674
 675         LASSERT(lock != NULL);
 676         if (rc != 0) {
 677                 rc = ldlm_handle_ast_error(lock, req, rc,
 678                                            arg->type == LDLM_BL_CALLBACK
 679                                            ? "blocking" : "completion");
 680                 if (rc == -ERESTART)
 681                         cfs_atomic_inc(&arg->restart);
 682         }
 683         LDLM_LOCK_RELEASE(lock);
 684
 685         if (cfs_atomic_dec_return(&arg->rpcs) < arg->threshold)
 686                 cfs_waitq_signal(&arg->waitq);
 687
 688         ldlm_csa_put(arg);
 689         RETURN(0);
 690 }
 691
 692 static inline int ldlm_bl_and_cp_ast_tail(struct ptlrpc_request *req,
 693                                           struct ldlm_cb_set_arg *arg,
 694                                           struct ldlm_lock *lock,
 695                                           int instant_cancel)
 696 {
 697         int rc = 0;
 698         ENTRY;
 699
 700         if (unlikely(instant_cancel)) {
 701                 rc = ptl_send_rpc(req, 1);
 702                 ptlrpc_req_finished(req);
 703                 if (rc == 0)
 704                         cfs_atomic_inc(&arg->restart);
 705         } else {
 706                 LDLM_LOCK_GET(lock);
 707                 cfs_atomic_inc(&arg->rpcs);
 708                 cfs_atomic_inc(&arg->refcount);
 709                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 710         }
 711
 712         RETURN(rc);
 713 }
 714
 715 /**
 716  * Check if there are requests in the export request list which prevent
 717  * the lock canceling and make these requests high priority ones.
 718  */
 719 static void ldlm_lock_reorder_req(struct ldlm_lock *lock)
 720 {
 721         struct ptlrpc_request *req;
 722         ENTRY;
 723
 724         if (lock->l_export == NULL) {
 725                 LDLM_DEBUG(lock, "client lock: no-op");
 726                 RETURN_EXIT;
 727         }
 728
 729         cfs_spin_lock_bh(&lock->l_export->exp_rpc_lock);
 730         cfs_list_for_each_entry(req, &lock->l_export->exp_hp_rpcs,
 731                                 rq_exp_list) {
 732                 /* Do not process requests that were not yet added to there
 733                  * incoming queue or were already removed from there for
 734                  * processing */
 735                 if (!req->rq_hp && !cfs_list_empty(&req->rq_list) &&
 736                     req->rq_ops->hpreq_lock_match &&
 737                     req->rq_ops->hpreq_lock_match(req, lock))
 738                         ptlrpc_hpreq_reorder(req);
 739         }
 740         cfs_spin_unlock_bh(&lock->l_export->exp_rpc_lock);
 741         EXIT;
 742 }
 743
 744 /*
 745  * ->l_blocking_ast() method for server-side locks. This is invoked when newly
 746  * enqueued server lock conflicts with given one.
 747  *
 748  * Sends blocking ast rpc to the client owning that lock; arms timeout timer
 749  * to wait for client response.
 750  */
 751 int ldlm_server_blocking_ast(struct ldlm_lock *lock,
 752                              struct ldlm_lock_desc *desc,
 753                              void *data, int flag)
 754 {
 755         struct ldlm_cb_async_args *ca;
 756         struct ldlm_cb_set_arg *arg = data;
 757         struct ldlm_request    *body;
 758         struct ptlrpc_request  *req;
 759         int                     instant_cancel = 0;
 760         int                     rc = 0;
 761         ENTRY;
 762
 763         if (flag == LDLM_CB_CANCELING)
 764                 /* Don't need to do anything here. */
 765                 RETURN(0);
 766
 767         LASSERT(lock);
 768         LASSERT(data != NULL);
 769         if (lock->l_export->exp_obd->obd_recovering != 0) {
 770                 LDLM_ERROR(lock, "BUG 6063: lock collide during recovery");
 771                 ldlm_lock_dump(D_ERROR, lock, 0);
 772         }
 773
 774         ldlm_lock_reorder_req(lock);
 775
 776         req = ptlrpc_request_alloc_pack(lock->l_export->exp_imp_reverse,
 777                                         &RQF_LDLM_BL_CALLBACK,
 778                                         LUSTRE_DLM_VERSION, LDLM_BL_CALLBACK);
 779         if (req == NULL)
 780                 RETURN(-ENOMEM);
 781
 782         CLASSERT(sizeof(*ca) <= sizeof(req->rq_async_args));
 783         ca = ptlrpc_req_async_args(req);
 784         ca->ca_set_arg = arg;
 785         ca->ca_lock = lock;
 786
 787         req->rq_interpret_reply = ldlm_cb_interpret;
 788         req->rq_no_resend = 1;
 789
 790         lock_res(lock->l_resource);
 791         if (lock->l_granted_mode != lock->l_req_mode) {
 792                 /* this blocking AST will be communicated as part of the
 793                  * completion AST instead */
 794                 unlock_res(lock->l_resource);
 795                 ptlrpc_req_finished(req);
 796                 LDLM_DEBUG(lock, "lock not granted, not sending blocking AST");
 797                 RETURN(0);
 798         }
 799
 800         if (lock->l_destroyed) {
 801                 /* What's the point? */
 802                 unlock_res(lock->l_resource);
 803                 ptlrpc_req_finished(req);
 804                 RETURN(0);
 805         }
 806
 807         if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)
 808                 instant_cancel = 1;
 809
 810         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
 811         body->lock_handle[0] = lock->l_remote_handle;
 812         body->lock_desc = *desc;
 813         body->lock_flags |= (lock->l_flags & LDLM_AST_FLAGS);
 814
 815         LDLM_DEBUG(lock, "server preparing blocking AST");
 816
 817         ptlrpc_request_set_replen(req);
 818         if (instant_cancel) {
 819                 unlock_res(lock->l_resource);
 820                 ldlm_lock_cancel(lock);
 821         } else {
 822                 LASSERT(lock->l_granted_mode == lock->l_req_mode);
 823                 ldlm_add_waiting_lock(lock);
 824                 unlock_res(lock->l_resource);
 825         }
 826
 827         req->rq_send_state = LUSTRE_IMP_FULL;
 828         /* ptlrpc_request_alloc_pack already set timeout */
 829         if (AT_OFF)
 830                 req->rq_timeout = ldlm_get_rq_timeout();
 831
 832         if (lock->l_export && lock->l_export->exp_nid_stats &&
 833             lock->l_export->exp_nid_stats->nid_ldlm_stats)
 834                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
 835                                      LDLM_BL_CALLBACK - LDLM_FIRST_OPC);
 836
 837         rc = ldlm_bl_and_cp_ast_tail(req, arg, lock, instant_cancel);
 838
 839         RETURN(rc);
 840 }
 841
 842 int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
 843 {
 844         struct ldlm_cb_set_arg *arg = data;
 845         struct ldlm_request    *body;
 846         struct ptlrpc_request  *req;
 847         struct ldlm_cb_async_args *ca;
 848         long                    total_enqueue_wait;
 849         int                     instant_cancel = 0;
 850         int                     rc = 0;
 851         ENTRY;
 852
 853         LASSERT(lock != NULL);
 854         LASSERT(data != NULL);
 855
 856         total_enqueue_wait = cfs_time_sub(cfs_time_current_sec(),
 857                                           lock->l_last_activity);
 858
 859         req = ptlrpc_request_alloc(lock->l_export->exp_imp_reverse,
 860                                     &RQF_LDLM_CP_CALLBACK);
 861         if (req == NULL)
 862                 RETURN(-ENOMEM);
 863
 864         /* server namespace, doesn't need lock */
 865         if (lock->l_resource->lr_lvb_len) {
 866                  req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_CLIENT,
 867                                       lock->l_resource->lr_lvb_len);
 868         }
 869
 870         rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_CP_CALLBACK);
 871         if (rc) {
 872                 ptlrpc_request_free(req);
 873                 RETURN(rc);
 874         }
 875
 876         CLASSERT(sizeof(*ca) <= sizeof(req->rq_async_args));
 877         ca = ptlrpc_req_async_args(req);
 878         ca->ca_set_arg = arg;
 879         ca->ca_lock = lock;
 880
 881         req->rq_interpret_reply = ldlm_cb_interpret;
 882         req->rq_no_resend = 1;
 883         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
 884
 885         body->lock_handle[0] = lock->l_remote_handle;
 886         body->lock_flags = flags;
 887         ldlm_lock2desc(lock, &body->lock_desc);
 888         if (lock->l_resource->lr_lvb_len) {
 889                 void *lvb = req_capsule_client_get(&req->rq_pill, &RMF_DLM_LVB);
 890
 891                 lock_res(lock->l_resource);
 892                 memcpy(lvb, lock->l_resource->lr_lvb_data,
 893                        lock->l_resource->lr_lvb_len);
 894                 unlock_res(lock->l_resource);
 895         }
 896
 897         LDLM_DEBUG(lock, "server preparing completion AST (after %lds wait)",
 898                    total_enqueue_wait);
 899
 900         /* Server-side enqueue wait time estimate, used in
 901             __ldlm_add_waiting_lock to set future enqueue timers */
 902         if (total_enqueue_wait < ldlm_get_enq_timeout(lock))
 903                 at_measured(ldlm_lock_to_ns_at(lock),
 904                             total_enqueue_wait);
 905         else
 906                 /* bz18618. Don't add lock enqueue time we spend waiting for a
 907                    previous callback to fail. Locks waiting legitimately will
 908                    get extended by ldlm_refresh_waiting_lock regardless of the
 909                    estimate, so it's okay to underestimate here. */
 910                 LDLM_DEBUG(lock, "lock completed after %lus; estimate was %ds. "
 911                        "It is likely that a previous callback timed out.",
 912                        total_enqueue_wait,
 913                        at_get(ldlm_lock_to_ns_at(lock)));
 914
 915         ptlrpc_request_set_replen(req);
 916
 917         req->rq_send_state = LUSTRE_IMP_FULL;
 918         /* ptlrpc_request_pack already set timeout */
 919         if (AT_OFF)
 920                 req->rq_timeout = ldlm_get_rq_timeout();
 921
 922         /* We only send real blocking ASTs after the lock is granted */
 923         lock_res_and_lock(lock);
 924         if (lock->l_flags & LDLM_FL_AST_SENT) {
 925                 body->lock_flags |= LDLM_FL_AST_SENT;
 926                 /* copy ast flags like LDLM_FL_DISCARD_DATA */
 927                 body->lock_flags |= (lock->l_flags & LDLM_AST_FLAGS);
 928
 929                 /* We might get here prior to ldlm_handle_enqueue setting
 930                  * LDLM_FL_CANCEL_ON_BLOCK flag. Then we will put this lock
 931                  * into waiting list, but this is safe and similar code in
 932                  * ldlm_handle_enqueue will call ldlm_lock_cancel() still,
 933                  * that would not only cancel the lock, but will also remove
 934                  * it from waiting list */
 935                 if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) {
 936                         unlock_res_and_lock(lock);
 937                         ldlm_lock_cancel(lock);
 938                         instant_cancel = 1;
 939                         lock_res_and_lock(lock);
 940                 } else {
 941                         /* start the lock-timeout clock */
 942                         ldlm_add_waiting_lock(lock);
 943                 }
 944         }
 945         unlock_res_and_lock(lock);
 946
 947         if (lock->l_export && lock->l_export->exp_nid_stats &&
 948             lock->l_export->exp_nid_stats->nid_ldlm_stats)
 949                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
 950                                      LDLM_CP_CALLBACK - LDLM_FIRST_OPC);
 951
 952         rc = ldlm_bl_and_cp_ast_tail(req, arg, lock, instant_cancel);
 953
 954         RETURN(rc);
 955 }
 956
 957 int ldlm_server_glimpse_ast(struct ldlm_lock *lock, void *data)
 958 {
 959         struct ldlm_resource  *res = lock->l_resource;
 960         struct ldlm_request   *body;
 961         struct ptlrpc_request *req;
 962         int                    rc;
 963         ENTRY;
 964
 965         LASSERT(lock != NULL);
 966
 967         req = ptlrpc_request_alloc_pack(lock->l_export->exp_imp_reverse,
 968                                         &RQF_LDLM_GL_CALLBACK,
 969                                         LUSTRE_DLM_VERSION, LDLM_GL_CALLBACK);
 970
 971         if (req == NULL)
 972                 RETURN(-ENOMEM);
 973
 974         body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
 975         body->lock_handle[0] = lock->l_remote_handle;
 976         ldlm_lock2desc(lock, &body->lock_desc);
 977
 978         /* server namespace, doesn't need lock */
 979         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
 980                              lock->l_resource->lr_lvb_len);
 981         res = lock->l_resource;
 982         ptlrpc_request_set_replen(req);
 983
 984
 985         req->rq_send_state = LUSTRE_IMP_FULL;
 986         /* ptlrpc_request_alloc_pack already set timeout */
 987         if (AT_OFF)
 988                 req->rq_timeout = ldlm_get_rq_timeout();
 989
 990         if (lock->l_export && lock->l_export->exp_nid_stats &&
 991             lock->l_export->exp_nid_stats->nid_ldlm_stats)
 992                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
 993                                      LDLM_GL_CALLBACK - LDLM_FIRST_OPC);
 994
 995         rc = ptlrpc_queue_wait(req);
 996         /* Update the LVB from disk if the AST failed (this is a legal race)
 997          *
 998          * - Glimpse callback of local lock just return -ELDLM_NO_LOCK_DATA.
 999          * - Glimpse callback of remote lock might return -ELDLM_NO_LOCK_DATA
1000          *   when inode is cleared. LU-274
1001          */
1002         if (rc == -ELDLM_NO_LOCK_DATA) {
1003                 LDLM_DEBUG(lock, "lost race - client has a lock but no inode");
1004                 ldlm_res_lvbo_update(res, NULL, 1);
1005         } else if (rc != 0) {
1006                 rc = ldlm_handle_ast_error(lock, req, rc, "glimpse");
1007         } else {
1008                 rc = ldlm_res_lvbo_update(res, req, 1);
1009         }
1010
1011         ptlrpc_req_finished(req);
1012         if (rc == -ERESTART)
1013                 ldlm_reprocess_all(res);
1014
1015         RETURN(rc);
1016 }
1017
1018 static void ldlm_svc_get_eopc(const struct ldlm_request *dlm_req,
1019                        struct lprocfs_stats *srv_stats)
1020 {
1021         int lock_type = 0, op = 0;
1022
1023         lock_type = dlm_req->lock_desc.l_resource.lr_type;
1024
1025         switch (lock_type) {
1026         case LDLM_PLAIN:
1027                 op = PTLRPC_LAST_CNTR + LDLM_PLAIN_ENQUEUE;
1028                 break;
1029         case LDLM_EXTENT:
1030                 if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT)
1031                         op = PTLRPC_LAST_CNTR + LDLM_GLIMPSE_ENQUEUE;
1032                 else
1033                         op = PTLRPC_LAST_CNTR + LDLM_EXTENT_ENQUEUE;
1034                 break;
1035         case LDLM_FLOCK:
1036                 op = PTLRPC_LAST_CNTR + LDLM_FLOCK_ENQUEUE;
1037                 break;
1038         case LDLM_IBITS:
1039                 op = PTLRPC_LAST_CNTR + LDLM_IBITS_ENQUEUE;
1040                 break;
1041         default:
1042                 op = 0;
1043                 break;
1044         }
1045
1046         if (op)
1047                 lprocfs_counter_incr(srv_stats, op);
1048
1049         return;
1050 }
1051
1052 /*
1053  * Main server-side entry point into LDLM. This is called by ptlrpc service
1054  * threads to carry out client lock enqueueing requests.
1055  */
1056 int ldlm_handle_enqueue0(struct ldlm_namespace *ns,
1057                          struct ptlrpc_request *req,
1058                          const struct ldlm_request *dlm_req,
1059                          const struct ldlm_callback_suite *cbs)
1060 {
1061         struct ldlm_reply *dlm_rep;
1062         __u32 flags;
1063         ldlm_error_t err = ELDLM_OK;
1064         struct ldlm_lock *lock = NULL;
1065         void *cookie = NULL;
1066         int rc = 0;
1067         ENTRY;
1068
1069         LDLM_DEBUG_NOLOCK("server-side enqueue handler START");
1070
1071         ldlm_request_cancel(req, dlm_req, LDLM_ENQUEUE_CANCEL_OFF);
1072         flags = dlm_req->lock_flags;
1073
1074         LASSERT(req->rq_export);
1075
1076         if (req->rq_rqbd->rqbd_service->srv_stats)
1077                 ldlm_svc_get_eopc(dlm_req,
1078                                   req->rq_rqbd->rqbd_service->srv_stats);
1079
1080         if (req->rq_export && req->rq_export->exp_nid_stats &&
1081             req->rq_export->exp_nid_stats->nid_ldlm_stats)
1082                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1083                                      LDLM_ENQUEUE - LDLM_FIRST_OPC);
1084
1085         if (unlikely(dlm_req->lock_desc.l_resource.lr_type < LDLM_MIN_TYPE ||
1086                      dlm_req->lock_desc.l_resource.lr_type >= LDLM_MAX_TYPE)) {
1087                 DEBUG_REQ(D_ERROR, req, "invalid lock request type %d",
1088                           dlm_req->lock_desc.l_resource.lr_type);
1089                 GOTO(out, rc = -EFAULT);
1090         }
1091
1092         if (unlikely(dlm_req->lock_desc.l_req_mode <= LCK_MINMODE ||
1093                      dlm_req->lock_desc.l_req_mode >= LCK_MAXMODE ||
1094                      dlm_req->lock_desc.l_req_mode &
1095                      (dlm_req->lock_desc.l_req_mode-1))) {
1096                 DEBUG_REQ(D_ERROR, req, "invalid lock request mode %d",
1097                           dlm_req->lock_desc.l_req_mode);
1098                 GOTO(out, rc = -EFAULT);
1099         }
1100
1101         if (req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS) {
1102                 if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
1103                              LDLM_PLAIN)) {
1104                         DEBUG_REQ(D_ERROR, req,
1105                                   "PLAIN lock request from IBITS client?");
1106                         GOTO(out, rc = -EPROTO);
1107                 }
1108         } else if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
1109                             LDLM_IBITS)) {
1110                 DEBUG_REQ(D_ERROR, req,
1111                           "IBITS lock request from unaware client?");
1112                 GOTO(out, rc = -EPROTO);
1113         }
1114
1115 #if 0
1116         /* FIXME this makes it impossible to use LDLM_PLAIN locks -- check
1117            against server's _CONNECT_SUPPORTED flags? (I don't want to use
1118            ibits for mgc/mgs) */
1119
1120         /* INODEBITS_INTEROP: Perform conversion from plain lock to
1121          * inodebits lock if client does not support them. */
1122         if (!(req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS) &&
1123             (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN)) {
1124                 dlm_req->lock_desc.l_resource.lr_type = LDLM_IBITS;
1125                 dlm_req->lock_desc.l_policy_data.l_inodebits.bits =
1126                         MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE;
1127                 if (dlm_req->lock_desc.l_req_mode == LCK_PR)
1128                         dlm_req->lock_desc.l_req_mode = LCK_CR;
1129         }
1130 #endif
1131
1132         if (unlikely(flags & LDLM_FL_REPLAY)) {
1133                 /* Find an existing lock in the per-export lock hash */
1134                 lock = cfs_hash_lookup(req->rq_export->exp_lock_hash,
1135                                        (void *)&dlm_req->lock_handle[0]);
1136                 if (lock != NULL) {
1137                         DEBUG_REQ(D_DLMTRACE, req, "found existing lock cookie "
1138                                   LPX64, lock->l_handle.h_cookie);
1139                         GOTO(existing_lock, rc = 0);
1140                 }
1141         }
1142
1143         /* The lock's callback data might be set in the policy function */
1144         lock = ldlm_lock_create(ns, &dlm_req->lock_desc.l_resource.lr_name,
1145                                 dlm_req->lock_desc.l_resource.lr_type,
1146                                 dlm_req->lock_desc.l_req_mode,
1147                                 cbs, NULL, 0);
1148
1149         if (!lock)
1150                 GOTO(out, rc = -ENOMEM);
1151
1152         lock->l_last_activity = cfs_time_current_sec();
1153         lock->l_remote_handle = dlm_req->lock_handle[0];
1154         LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
1155
1156         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_BLOCKED, obd_timeout * 2);
1157         /* Don't enqueue a lock onto the export if it is been disonnected
1158          * due to eviction (bug 3822) or server umount (bug 24324).
1159          * Cancel it now instead. */
1160         if (req->rq_export->exp_disconnected) {
1161                 LDLM_ERROR(lock, "lock on disconnected export %p",
1162                            req->rq_export);
1163                 GOTO(out, rc = -ENOTCONN);
1164         }
1165
1166         lock->l_export = class_export_lock_get(req->rq_export, lock);
1167         if (lock->l_export->exp_lock_hash)
1168                 cfs_hash_add(lock->l_export->exp_lock_hash,
1169                              &lock->l_remote_handle,
1170                              &lock->l_exp_hash);
1171
1172 existing_lock:
1173
1174         if (flags & LDLM_FL_HAS_INTENT) {
1175                 /* In this case, the reply buffer is allocated deep in
1176                  * local_lock_enqueue by the policy function. */
1177                 cookie = req;
1178         } else {
1179                 /* based on the assumption that lvb size never changes during
1180                  * resource life time otherwise it need resource->lr_lock's
1181                  * protection */
1182                 if (lock->l_resource->lr_lvb_len) {
1183                         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
1184                                              RCL_SERVER,
1185                                              lock->l_resource->lr_lvb_len);
1186                 }
1187
1188                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
1189                         GOTO(out, rc = -ENOMEM);
1190
1191                 rc = req_capsule_server_pack(&req->rq_pill);
1192                 if (rc)
1193                         GOTO(out, rc);
1194         }
1195
1196         if (dlm_req->lock_desc.l_resource.lr_type != LDLM_PLAIN)
1197                 ldlm_convert_policy_to_local(req->rq_export,
1198                                           dlm_req->lock_desc.l_resource.lr_type,
1199                                           &dlm_req->lock_desc.l_policy_data,
1200                                           &lock->l_policy_data);
1201         if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT)
1202                 lock->l_req_extent = lock->l_policy_data.l_extent;
1203
1204         err = ldlm_lock_enqueue(ns, &lock, cookie, (int *)&flags);
1205         if (err)
1206                 GOTO(out, err);
1207
1208         dlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1209         dlm_rep->lock_flags = flags;
1210
1211         ldlm_lock2desc(lock, &dlm_rep->lock_desc);
1212         ldlm_lock2handle(lock, &dlm_rep->lock_handle);
1213
1214         /* We never send a blocking AST until the lock is granted, but
1215          * we can tell it right now */
1216         lock_res_and_lock(lock);
1217
1218         /* Now take into account flags to be inherited from original lock
1219            request both in reply to client and in our own lock flags. */
1220         dlm_rep->lock_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
1221         lock->l_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
1222
1223         /* Don't move a pending lock onto the export if it has already been
1224          * disconnected due to eviction (bug 5683) or server umount (bug 24324).
1225          * Cancel it now instead. */
1226         if (unlikely(req->rq_export->exp_disconnected ||
1227                      OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_OLD_EXPORT))) {
1228                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
1229                 rc = -ENOTCONN;
1230         } else if (lock->l_flags & LDLM_FL_AST_SENT) {
1231                 dlm_rep->lock_flags |= LDLM_FL_AST_SENT;
1232                 if (lock->l_granted_mode == lock->l_req_mode) {
1233                         /*
1234                          * Only cancel lock if it was granted, because it would
1235                          * be destroyed immediately and would never be granted
1236                          * in the future, causing timeouts on client.  Not
1237                          * granted lock will be cancelled immediately after
1238                          * sending completion AST.
1239                          */
1240                         if (dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK) {
1241                                 unlock_res_and_lock(lock);
1242                                 ldlm_lock_cancel(lock);
1243                                 lock_res_and_lock(lock);
1244                         } else
1245                                 ldlm_add_waiting_lock(lock);
1246                 }
1247         }
1248         /* Make sure we never ever grant usual metadata locks to liblustre
1249            clients */
1250         if ((dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN ||
1251             dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS) &&
1252              req->rq_export->exp_libclient) {
1253                 if (unlikely(!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK) ||
1254                              !(dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK))){
1255                         CERROR("Granting sync lock to libclient. "
1256                                "req fl %d, rep fl %d, lock fl "LPX64"\n",
1257                                dlm_req->lock_flags, dlm_rep->lock_flags,
1258                                lock->l_flags);
1259                         LDLM_ERROR(lock, "sync lock");
1260                         if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT) {
1261                                 struct ldlm_intent *it;
1262
1263                                 it = req_capsule_client_get(&req->rq_pill,
1264                                                             &RMF_LDLM_INTENT);
1265                                 if (it != NULL) {
1266                                         CERROR("This is intent %s ("LPU64")\n",
1267                                                ldlm_it2str(it->opc), it->opc);
1268                                 }
1269                         }
1270                 }
1271         }
1272
1273         unlock_res_and_lock(lock);
1274
1275         EXIT;
1276  out:
1277         req->rq_status = rc ?: err; /* return either error - bug 11190 */
1278         if (!req->rq_packed_final) {
1279                 err = lustre_pack_reply(req, 1, NULL, NULL);
1280                 if (rc == 0)
1281                         rc = err;
1282         }
1283
1284         /* The LOCK_CHANGED code in ldlm_lock_enqueue depends on this
1285          * ldlm_reprocess_all.  If this moves, revisit that code. -phil */
1286         if (lock) {
1287                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
1288                            "(err=%d, rc=%d)", err, rc);
1289
1290                 if (rc == 0) {
1291                         if (lock->l_resource->lr_lvb_len > 0) {
1292                                 /* MDT path won't handle lr_lvb_data, so
1293                                  * lock/unlock better be contained in the
1294                                  * if block */
1295                                 void *lvb;
1296
1297                                 lvb = req_capsule_server_get(&req->rq_pill,
1298                                                              &RMF_DLM_LVB);
1299                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
1300                                          req, lock);
1301                                 lock_res(lock->l_resource);
1302                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
1303                                        lock->l_resource->lr_lvb_len);
1304                                 unlock_res(lock->l_resource);
1305                         }
1306                 } else {
1307                         lock_res_and_lock(lock);
1308                         ldlm_resource_unlink_lock(lock);
1309                         ldlm_lock_destroy_nolock(lock);
1310                         unlock_res_and_lock(lock);
1311                 }
1312
1313                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
1314                         ldlm_reprocess_all(lock->l_resource);
1315
1316                 LDLM_LOCK_RELEASE(lock);
1317         }
1318
1319         LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p, rc %d)",
1320                           lock, rc);
1321
1322         return rc;
1323 }
1324
1325 int ldlm_handle_enqueue(struct ptlrpc_request *req,
1326                         ldlm_completion_callback completion_callback,
1327                         ldlm_blocking_callback blocking_callback,
1328                         ldlm_glimpse_callback glimpse_callback)
1329 {
1330         struct ldlm_request *dlm_req;
1331         struct ldlm_callback_suite cbs = {
1332                 .lcs_completion = completion_callback,
1333                 .lcs_blocking   = blocking_callback,
1334                 .lcs_glimpse    = glimpse_callback
1335         };
1336         int rc;
1337
1338         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1339         if (dlm_req != NULL) {
1340                 rc = ldlm_handle_enqueue0(req->rq_export->exp_obd->obd_namespace,
1341                                           req, dlm_req, &cbs);
1342         } else {
1343                 rc = -EFAULT;
1344         }
1345         return rc;
1346 }
1347
1348 int ldlm_handle_convert0(struct ptlrpc_request *req,
1349                          const struct ldlm_request *dlm_req)
1350 {
1351         struct ldlm_reply *dlm_rep;
1352         struct ldlm_lock *lock;
1353         int rc;
1354         ENTRY;
1355
1356         if (req->rq_export && req->rq_export->exp_nid_stats &&
1357             req->rq_export->exp_nid_stats->nid_ldlm_stats)
1358                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1359                                      LDLM_CONVERT - LDLM_FIRST_OPC);
1360
1361         rc = req_capsule_server_pack(&req->rq_pill);
1362         if (rc)
1363                 RETURN(rc);
1364
1365         dlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1366         dlm_rep->lock_flags = dlm_req->lock_flags;
1367
1368         lock = ldlm_handle2lock(&dlm_req->lock_handle[0]);
1369         if (!lock) {
1370                 req->rq_status = EINVAL;
1371         } else {
1372                 void *res = NULL;
1373
1374                 LDLM_DEBUG(lock, "server-side convert handler START");
1375
1376                 lock->l_last_activity = cfs_time_current_sec();
1377                 res = ldlm_lock_convert(lock, dlm_req->lock_desc.l_req_mode,
1378                                         &dlm_rep->lock_flags);
1379                 if (res) {
1380                         if (ldlm_del_waiting_lock(lock))
1381                                 LDLM_DEBUG(lock, "converted waiting lock");
1382                         req->rq_status = 0;
1383                 } else {
1384                         req->rq_status = EDEADLOCK;
1385                 }
1386         }
1387
1388         if (lock) {
1389                 if (!req->rq_status)
1390                         ldlm_reprocess_all(lock->l_resource);
1391                 LDLM_DEBUG(lock, "server-side convert handler END");
1392                 LDLM_LOCK_PUT(lock);
1393         } else
1394                 LDLM_DEBUG_NOLOCK("server-side convert handler END");
1395
1396         RETURN(0);
1397 }
1398
1399 int ldlm_handle_convert(struct ptlrpc_request *req)
1400 {
1401         int rc;
1402         struct ldlm_request *dlm_req;
1403
1404         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1405         if (dlm_req != NULL) {
1406                 rc = ldlm_handle_convert0(req, dlm_req);
1407         } else {
1408                 CERROR ("Can't unpack dlm_req\n");
1409                 rc = -EFAULT;
1410         }
1411         return rc;
1412 }
1413
1414 /* Cancel all the locks whos handles are packed into ldlm_request */
1415 int ldlm_request_cancel(struct ptlrpc_request *req,
1416                         const struct ldlm_request *dlm_req, int first)
1417 {
1418         struct ldlm_resource *res, *pres = NULL;
1419         struct ldlm_lock *lock;
1420         int i, count, done = 0;
1421         ENTRY;
1422
1423         count = dlm_req->lock_count ? dlm_req->lock_count : 1;
1424         if (first >= count)
1425                 RETURN(0);
1426
1427         /* There is no lock on the server at the replay time,
1428          * skip lock cancelling to make replay tests to pass. */
1429         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
1430                 RETURN(0);
1431
1432         LDLM_DEBUG_NOLOCK("server-side cancel handler START: %d locks, "
1433                           "starting at %d", count, first);
1434
1435         for (i = first; i < count; i++) {
1436                 lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
1437                 if (!lock) {
1438                         LDLM_DEBUG_NOLOCK("server-side cancel handler stale "
1439                                           "lock (cookie "LPU64")",
1440                                           dlm_req->lock_handle[i].cookie);
1441                         continue;
1442                 }
1443
1444                 res = lock->l_resource;
1445                 done++;
1446
1447                 if (res != pres) {
1448                         if (pres != NULL) {
1449                                 ldlm_reprocess_all(pres);
1450                                 LDLM_RESOURCE_DELREF(pres);
1451                                 ldlm_resource_putref(pres);
1452                         }
1453                         if (res != NULL) {
1454                                 ldlm_resource_getref(res);
1455                                 LDLM_RESOURCE_ADDREF(res);
1456                                 ldlm_res_lvbo_update(res, NULL, 1);
1457                         }
1458                         pres = res;
1459                 }
1460                 ldlm_lock_cancel(lock);
1461                 LDLM_LOCK_PUT(lock);
1462         }
1463         if (pres != NULL) {
1464                 ldlm_reprocess_all(pres);
1465                 LDLM_RESOURCE_DELREF(pres);
1466                 ldlm_resource_putref(pres);
1467         }
1468         LDLM_DEBUG_NOLOCK("server-side cancel handler END");
1469         RETURN(done);
1470 }
1471
1472 int ldlm_handle_cancel(struct ptlrpc_request *req)
1473 {
1474         struct ldlm_request *dlm_req;
1475         int rc;
1476         ENTRY;
1477
1478         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1479         if (dlm_req == NULL) {
1480                 CDEBUG(D_INFO, "bad request buffer for cancel\n");
1481                 RETURN(-EFAULT);
1482         }
1483
1484         if (req->rq_export && req->rq_export->exp_nid_stats &&
1485             req->rq_export->exp_nid_stats->nid_ldlm_stats)
1486                 lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
1487                                      LDLM_CANCEL - LDLM_FIRST_OPC);
1488
1489         rc = req_capsule_server_pack(&req->rq_pill);
1490         if (rc)
1491                 RETURN(rc);
1492
1493         if (!ldlm_request_cancel(req, dlm_req, 0))
1494                 req->rq_status = ESTALE;
1495
1496         RETURN(ptlrpc_reply(req));
1497 }
1498 #endif /* HAVE_SERVER_SUPPORT */
1499
1500 void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
1501                              struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
1502 {
1503         int do_ast;
1504         ENTRY;
1505
1506         LDLM_DEBUG(lock, "client blocking AST callback handler");
1507
1508         lock_res_and_lock(lock);
1509         lock->l_flags |= LDLM_FL_CBPENDING;
1510
1511         if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)
1512                 lock->l_flags |= LDLM_FL_CANCEL;
1513
1514         do_ast = (!lock->l_readers && !lock->l_writers);
1515         unlock_res_and_lock(lock);
1516
1517         if (do_ast) {
1518                 CDEBUG(D_DLMTRACE, "Lock %p already unused, calling callback (%p)\n",
1519                        lock, lock->l_blocking_ast);
1520                 if (lock->l_blocking_ast != NULL)
1521                         lock->l_blocking_ast(lock, ld, lock->l_ast_data,
1522                                              LDLM_CB_BLOCKING);
1523         } else {
1524                 CDEBUG(D_DLMTRACE, "Lock %p is referenced, will be cancelled later\n",
1525                        lock);
1526         }
1527
1528         LDLM_DEBUG(lock, "client blocking callback handler END");
1529         LDLM_LOCK_RELEASE(lock);
1530         EXIT;
1531 }
1532
1533 static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
1534                                     struct ldlm_namespace *ns,
1535                                     struct ldlm_request *dlm_req,
1536                                     struct ldlm_lock *lock)
1537 {
1538         CFS_LIST_HEAD(ast_list);
1539         ENTRY;
1540
1541         LDLM_DEBUG(lock, "client completion callback handler START");
1542
1543         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE)) {
1544                 int to = cfs_time_seconds(1);
1545                 while (to > 0) {
1546                         cfs_schedule_timeout_and_set_state(
1547                                 CFS_TASK_INTERRUPTIBLE, to);
1548                         if (lock->l_granted_mode == lock->l_req_mode ||
1549                             lock->l_destroyed)
1550                                 break;
1551                 }
1552         }
1553
1554         lock_res_and_lock(lock);
1555         if (lock->l_destroyed ||
1556             lock->l_granted_mode == lock->l_req_mode) {
1557                 /* bug 11300: the lock has already been granted */
1558                 unlock_res_and_lock(lock);
1559                 LDLM_DEBUG(lock, "Double grant race happened");
1560                 LDLM_LOCK_RELEASE(lock);
1561                 EXIT;
1562                 return;
1563         }
1564
1565         /* If we receive the completion AST before the actual enqueue returned,
1566          * then we might need to switch lock modes, resources, or extents. */
1567         if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
1568                 lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
1569                 LDLM_DEBUG(lock, "completion AST, new lock mode");
1570         }
1571
1572         if (lock->l_resource->lr_type != LDLM_PLAIN) {
1573                 ldlm_convert_policy_to_local(req->rq_export,
1574                                           dlm_req->lock_desc.l_resource.lr_type,
1575                                           &dlm_req->lock_desc.l_policy_data,
1576                                           &lock->l_policy_data);
1577                 LDLM_DEBUG(lock, "completion AST, new policy data");
1578         }
1579
1580         ldlm_resource_unlink_lock(lock);
1581         if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
1582                    &lock->l_resource->lr_name,
1583                    sizeof(lock->l_resource->lr_name)) != 0) {
1584                 unlock_res_and_lock(lock);
1585                 if (ldlm_lock_change_resource(ns, lock,
1586                                 &dlm_req->lock_desc.l_resource.lr_name) != 0) {
1587                         LDLM_ERROR(lock, "Failed to allocate resource");
1588                         LDLM_LOCK_RELEASE(lock);
1589                         EXIT;
1590                         return;
1591                 }
1592                 LDLM_DEBUG(lock, "completion AST, new resource");
1593                 CERROR("change resource!\n");
1594                 lock_res_and_lock(lock);
1595         }
1596
1597         if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
1598                 /* BL_AST locks are not needed in lru.
1599                  * let ldlm_cancel_lru() be fast. */
1600                 ldlm_lock_remove_from_lru(lock);
1601                 lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_BL_AST;
1602                 LDLM_DEBUG(lock, "completion AST includes blocking AST");
1603         }
1604
1605         if (lock->l_lvb_len) {
1606                 if (req_capsule_get_size(&req->rq_pill, &RMF_DLM_LVB,
1607                                          RCL_CLIENT) < lock->l_lvb_len) {
1608                         LDLM_ERROR(lock, "completion AST did not contain "
1609                                    "expected LVB!");
1610                 } else {
1611                         void *lvb = req_capsule_client_get(&req->rq_pill,
1612                                                            &RMF_DLM_LVB);
1613                         memcpy(lock->l_lvb_data, lvb, lock->l_lvb_len);
1614                 }
1615         }
1616
1617         ldlm_grant_lock(lock, &ast_list);
1618         unlock_res_and_lock(lock);
1619
1620         LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
1621
1622         /* Let Enqueue to call osc_lock_upcall() and initialize
1623          * l_ast_data */
1624         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 2);
1625
1626         ldlm_run_ast_work(ns, &ast_list, LDLM_WORK_CP_AST);
1627
1628         LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
1629                           lock);
1630         LDLM_LOCK_RELEASE(lock);
1631         EXIT;
1632 }
1633
1634 static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
1635                                     struct ldlm_namespace *ns,
1636                                     struct ldlm_request *dlm_req,
1637                                     struct ldlm_lock *lock)
1638 {
1639         int rc = -ENOSYS;
1640         ENTRY;
1641
1642         LDLM_DEBUG(lock, "client glimpse AST callback handler");
1643
1644         if (lock->l_glimpse_ast != NULL)
1645                 rc = lock->l_glimpse_ast(lock, req);
1646
1647         if (req->rq_repmsg != NULL) {
1648                 ptlrpc_reply(req);
1649         } else {
1650                 req->rq_status = rc;
1651                 ptlrpc_error(req);
1652         }
1653
1654         lock_res_and_lock(lock);
1655         if (lock->l_granted_mode == LCK_PW &&
1656             !lock->l_readers && !lock->l_writers &&
1657             cfs_time_after(cfs_time_current(),
1658                            cfs_time_add(lock->l_last_used,
1659                                         cfs_time_seconds(10)))) {
1660                 unlock_res_and_lock(lock);
1661                 if (ldlm_bl_to_thread_lock(ns, NULL, lock))
1662                         ldlm_handle_bl_callback(ns, NULL, lock);
1663
1664                 EXIT;
1665                 return;
1666         }
1667         unlock_res_and_lock(lock);
1668         LDLM_LOCK_RELEASE(lock);
1669         EXIT;
1670 }
1671
1672 static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
1673 {
1674         if (req->rq_no_reply)
1675                 return 0;
1676
1677         req->rq_status = rc;
1678         if (!req->rq_packed_final) {
1679                 rc = lustre_pack_reply(req, 1, NULL, NULL);
1680                 if (rc)
1681                         return rc;
1682         }
1683         return ptlrpc_reply(req);
1684 }
1685
1686 #ifdef __KERNEL__
1687 static int __ldlm_bl_to_thread(struct ldlm_bl_work_item *blwi, int mode)
1688 {
1689         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
1690         ENTRY;
1691
1692         cfs_spin_lock(&blp->blp_lock);
1693         if (blwi->blwi_lock && blwi->blwi_lock->l_flags & LDLM_FL_DISCARD_DATA) {
1694                 /* add LDLM_FL_DISCARD_DATA requests to the priority list */
1695                 cfs_list_add_tail(&blwi->blwi_entry, &blp->blp_prio_list);
1696         } else {
1697                 /* other blocking callbacks are added to the regular list */
1698                 cfs_list_add_tail(&blwi->blwi_entry, &blp->blp_list);
1699         }
1700         cfs_spin_unlock(&blp->blp_lock);
1701
1702         cfs_waitq_signal(&blp->blp_waitq);
1703
1704         /* can not use blwi->blwi_mode as blwi could be already freed in
1705            LDLM_ASYNC mode */
1706         if (mode == LDLM_SYNC)
1707                 cfs_wait_for_completion(&blwi->blwi_comp);
1708
1709         RETURN(0);
1710 }
1711
1712 static inline void init_blwi(struct ldlm_bl_work_item *blwi,
1713                              struct ldlm_namespace *ns,
1714                              struct ldlm_lock_desc *ld,
1715                              cfs_list_t *cancels, int count,
1716                              struct ldlm_lock *lock,
1717                              int mode)
1718 {
1719         cfs_init_completion(&blwi->blwi_comp);
1720         CFS_INIT_LIST_HEAD(&blwi->blwi_head);
1721
1722         if (cfs_memory_pressure_get())
1723                 blwi->blwi_mem_pressure = 1;
1724
1725         blwi->blwi_ns = ns;
1726         blwi->blwi_mode = mode;
1727         if (ld != NULL)
1728                 blwi->blwi_ld = *ld;
1729         if (count) {
1730                 cfs_list_add(&blwi->blwi_head, cancels);
1731                 cfs_list_del_init(cancels);
1732                 blwi->blwi_count = count;
1733         } else {
1734                 blwi->blwi_lock = lock;
1735         }
1736 }
1737
1738 static int ldlm_bl_to_thread(struct ldlm_namespace *ns,
1739                              struct ldlm_lock_desc *ld, struct ldlm_lock *lock,
1740                              cfs_list_t *cancels, int count, int mode)
1741 {
1742         ENTRY;
1743
1744         if (cancels && count == 0)
1745                 RETURN(0);
1746
1747         if (mode == LDLM_SYNC) {
1748                 /* if it is synchronous call do minimum mem alloc, as it could
1749                  * be triggered from kernel shrinker
1750                  */
1751                 struct ldlm_bl_work_item blwi;
1752                 memset(&blwi, 0, sizeof(blwi));
1753                 init_blwi(&blwi, ns, ld, cancels, count, lock, LDLM_SYNC);
1754                 RETURN(__ldlm_bl_to_thread(&blwi, LDLM_SYNC));
1755         } else {
1756                 struct ldlm_bl_work_item *blwi;
1757                 OBD_ALLOC(blwi, sizeof(*blwi));
1758                 if (blwi == NULL)
1759                         RETURN(-ENOMEM);
1760                 init_blwi(blwi, ns, ld, cancels, count, lock, LDLM_ASYNC);
1761
1762                 RETURN(__ldlm_bl_to_thread(blwi, LDLM_ASYNC));
1763         }
1764 }
1765
1766 #endif
1767
1768 int ldlm_bl_to_thread_lock(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
1769                            struct ldlm_lock *lock)
1770 {
1771 #ifdef __KERNEL__
1772         RETURN(ldlm_bl_to_thread(ns, ld, lock, NULL, 0, LDLM_ASYNC));
1773 #else
1774         RETURN(-ENOSYS);
1775 #endif
1776 }
1777
1778 int ldlm_bl_to_thread_list(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
1779                            cfs_list_t *cancels, int count, int mode)
1780 {
1781 #ifdef __KERNEL__
1782         RETURN(ldlm_bl_to_thread(ns, ld, NULL, cancels, count, mode));
1783 #else
1784         RETURN(-ENOSYS);
1785 #endif
1786 }
1787
1788 /* Setinfo coming from Server (eg MDT) to Client (eg MDC)! */
1789 static int ldlm_handle_setinfo(struct ptlrpc_request *req)
1790 {
1791         struct obd_device *obd = req->rq_export->exp_obd;
1792         char *key;
1793         void *val;
1794         int keylen, vallen;
1795         int rc = -ENOSYS;
1796         ENTRY;
1797
1798         DEBUG_REQ(D_HSM, req, "%s: handle setinfo\n", obd->obd_name);
1799
1800         req_capsule_set(&req->rq_pill, &RQF_OBD_SET_INFO);
1801
1802         key = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
1803         if (key == NULL) {
1804                 DEBUG_REQ(D_IOCTL, req, "no set_info key");
1805                 RETURN(-EFAULT);
1806         }
1807         keylen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_KEY,
1808                                       RCL_CLIENT);
1809         val = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
1810         if (val == NULL) {
1811                 DEBUG_REQ(D_IOCTL, req, "no set_info val");
1812                 RETURN(-EFAULT);
1813         }
1814         vallen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_VAL,
1815                                       RCL_CLIENT);
1816
1817         /* We are responsible for swabbing contents of val */
1818
1819         if (KEY_IS(KEY_HSM_COPYTOOL_SEND))
1820                 /* Pass it on to mdc (the "export" in this case) */
1821                 rc = obd_set_info_async(req->rq_export,
1822                                         sizeof(KEY_HSM_COPYTOOL_SEND),
1823                                         KEY_HSM_COPYTOOL_SEND,
1824                                         vallen, val, NULL);
1825         else
1826                 DEBUG_REQ(D_WARNING, req, "ignoring unknown key %s", key);
1827
1828         return rc;
1829 }
1830
1831 static inline void ldlm_callback_errmsg(struct ptlrpc_request *req,
1832                                         const char *msg, int rc,
1833                                         struct lustre_handle *handle)
1834 {
1835         DEBUG_REQ((req->rq_no_reply || rc) ? D_WARNING : D_DLMTRACE, req,
1836                   "%s: [nid %s] [rc %d] [lock "LPX64"]",
1837                   msg, libcfs_id2str(req->rq_peer), rc,
1838                   handle ? handle->cookie : 0);
1839         if (req->rq_no_reply)
1840                 CWARN("No reply was sent, maybe cause bug 21636.\n");
1841         else if (rc)
1842                 CWARN("Send reply failed, maybe cause bug 21636.\n");
1843 }
1844
1845 /* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
1846 static int ldlm_callback_handler(struct ptlrpc_request *req)
1847 {
1848         struct ldlm_namespace *ns;
1849         struct ldlm_request *dlm_req;
1850         struct ldlm_lock *lock;
1851         int rc;
1852         ENTRY;
1853
1854         /* Requests arrive in sender's byte order.  The ptlrpc service
1855          * handler has already checked and, if necessary, byte-swapped the
1856          * incoming request message body, but I am responsible for the
1857          * message buffers. */
1858
1859         /* do nothing for sec context finalize */
1860         if (lustre_msg_get_opc(req->rq_reqmsg) == SEC_CTX_FINI)
1861                 RETURN(0);
1862
1863         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
1864
1865         if (req->rq_export == NULL) {
1866                 rc = ldlm_callback_reply(req, -ENOTCONN);
1867                 ldlm_callback_errmsg(req, "Operate on unconnected server",
1868                                      rc, NULL);
1869                 RETURN(0);
1870         }
1871
1872         LASSERT(req->rq_export != NULL);
1873         LASSERT(req->rq_export->exp_obd != NULL);
1874
1875         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1876         case LDLM_BL_CALLBACK:
1877                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK))
1878                         RETURN(0);
1879                 break;
1880         case LDLM_CP_CALLBACK:
1881                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CP_CALLBACK))
1882                         RETURN(0);
1883                 break;
1884         case LDLM_GL_CALLBACK:
1885                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_GL_CALLBACK))
1886                         RETURN(0);
1887                 break;
1888         case LDLM_SET_INFO:
1889                 rc = ldlm_handle_setinfo(req);
1890                 ldlm_callback_reply(req, rc);
1891                 RETURN(0);
1892         case OBD_LOG_CANCEL: /* remove this eventually - for 1.4.0 compat */
1893                 CERROR("shouldn't be handling OBD_LOG_CANCEL on DLM thread\n");
1894                 req_capsule_set(&req->rq_pill, &RQF_LOG_CANCEL);
1895                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_NET))
1896                         RETURN(0);
1897                 rc = llog_origin_handle_cancel(req);
1898                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_REP))
1899                         RETURN(0);
1900                 ldlm_callback_reply(req, rc);
1901                 RETURN(0);
1902         case OBD_QC_CALLBACK:
1903                 req_capsule_set(&req->rq_pill, &RQF_QC_CALLBACK);
1904                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_QC_CALLBACK_NET))
1905                         RETURN(0);
1906                 rc = target_handle_qc_callback(req);
1907                 ldlm_callback_reply(req, rc);
1908                 RETURN(0);
1909         case QUOTA_DQACQ:
1910         case QUOTA_DQREL:
1911                 /* reply in handler */
1912                 req_capsule_set(&req->rq_pill, &RQF_MDS_QUOTA_DQACQ);
1913                 rc = target_handle_dqacq_callback(req);
1914                 RETURN(0);
1915         case LLOG_ORIGIN_HANDLE_CREATE:
1916                 req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
1917                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
1918                         RETURN(0);
1919                 rc = llog_origin_handle_create(req);
1920                 ldlm_callback_reply(req, rc);
1921                 RETURN(0);
1922         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
1923                 req_capsule_set(&req->rq_pill,
1924                                 &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
1925                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
1926                         RETURN(0);
1927                 rc = llog_origin_handle_next_block(req);
1928                 ldlm_callback_reply(req, rc);
1929                 RETURN(0);
1930         case LLOG_ORIGIN_HANDLE_READ_HEADER:
1931                 req_capsule_set(&req->rq_pill,
1932                                 &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER);
1933                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
1934                         RETURN(0);
1935                 rc = llog_origin_handle_read_header(req);
1936                 ldlm_callback_reply(req, rc);
1937                 RETURN(0);
1938         case LLOG_ORIGIN_HANDLE_CLOSE:
1939                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
1940                         RETURN(0);
1941                 rc = llog_origin_handle_close(req);
1942                 ldlm_callback_reply(req, rc);
1943                 RETURN(0);
1944         default:
1945                 CERROR("unknown opcode %u\n",
1946                        lustre_msg_get_opc(req->rq_reqmsg));
1947                 ldlm_callback_reply(req, -EPROTO);
1948                 RETURN(0);
1949         }
1950
1951         ns = req->rq_export->exp_obd->obd_namespace;
1952         LASSERT(ns != NULL);
1953
1954         req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
1955
1956         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
1957         if (dlm_req == NULL) {
1958                 rc = ldlm_callback_reply(req, -EPROTO);
1959                 ldlm_callback_errmsg(req, "Operate without parameter", rc,
1960                                      NULL);
1961                 RETURN(0);
1962         }
1963
1964         /* Force a known safe race, send a cancel to the server for a lock
1965          * which the server has already started a blocking callback on. */
1966         if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE) &&
1967             lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
1968                 rc = ldlm_cli_cancel(&dlm_req->lock_handle[0]);
1969                 if (rc < 0)
1970                         CERROR("ldlm_cli_cancel: %d\n", rc);
1971         }
1972
1973         lock = ldlm_handle2lock_long(&dlm_req->lock_handle[0], 0);
1974         if (!lock) {
1975                 CDEBUG(D_DLMTRACE, "callback on lock "LPX64" - lock "
1976                        "disappeared\n", dlm_req->lock_handle[0].cookie);
1977                 rc = ldlm_callback_reply(req, -EINVAL);
1978                 ldlm_callback_errmsg(req, "Operate with invalid parameter", rc,
1979                                      &dlm_req->lock_handle[0]);
1980                 RETURN(0);
1981         }
1982
1983         if ((lock->l_flags & LDLM_FL_FAIL_LOC) &&
1984             lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK)
1985                 OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
1986
1987         /* Copy hints/flags (e.g. LDLM_FL_DISCARD_DATA) from AST. */
1988         lock_res_and_lock(lock);
1989         lock->l_flags |= (dlm_req->lock_flags & LDLM_AST_FLAGS);
1990         if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
1991                 /* If somebody cancels lock and cache is already dropped,
1992                  * or lock is failed before cp_ast received on client,
1993                  * we can tell the server we have no lock. Otherwise, we
1994                  * should send cancel after dropping the cache. */
1995                 if (((lock->l_flags & LDLM_FL_CANCELING) &&
1996                     (lock->l_flags & LDLM_FL_BL_DONE)) ||
1997                     (lock->l_flags & LDLM_FL_FAILED)) {
1998                         LDLM_DEBUG(lock, "callback on lock "
1999                                    LPX64" - lock disappeared\n",
2000                                    dlm_req->lock_handle[0].cookie);
2001                         unlock_res_and_lock(lock);
2002                         LDLM_LOCK_RELEASE(lock);
2003                         rc = ldlm_callback_reply(req, -EINVAL);
2004                         ldlm_callback_errmsg(req, "Operate on stale lock", rc,
2005                                              &dlm_req->lock_handle[0]);
2006                         RETURN(0);
2007                 }
2008                 /* BL_AST locks are not needed in lru.
2009                  * let ldlm_cancel_lru() be fast. */
2010                 ldlm_lock_remove_from_lru(lock);
2011                 lock->l_flags |= LDLM_FL_BL_AST;
2012         }
2013         unlock_res_and_lock(lock);
2014
2015         /* We want the ost thread to get this reply so that it can respond
2016          * to ost requests (write cache writeback) that might be triggered
2017          * in the callback.
2018          *
2019          * But we'd also like to be able to indicate in the reply that we're
2020          * cancelling right now, because it's unused, or have an intent result
2021          * in the reply, so we might have to push the responsibility for sending
2022          * the reply down into the AST handlers, alas. */
2023
2024         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2025         case LDLM_BL_CALLBACK:
2026                 CDEBUG(D_INODE, "blocking ast\n");
2027                 req_capsule_extend(&req->rq_pill, &RQF_LDLM_BL_CALLBACK);
2028                 if (!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)) {
2029                         rc = ldlm_callback_reply(req, 0);
2030                         if (req->rq_no_reply || rc)
2031                                 ldlm_callback_errmsg(req, "Normal process", rc,
2032                                                      &dlm_req->lock_handle[0]);
2033                 }
2034                 if (ldlm_bl_to_thread_lock(ns, &dlm_req->lock_desc, lock))
2035                         ldlm_handle_bl_callback(ns, &dlm_req->lock_desc, lock);
2036                 break;
2037         case LDLM_CP_CALLBACK:
2038                 CDEBUG(D_INODE, "completion ast\n");
2039                 req_capsule_extend(&req->rq_pill, &RQF_LDLM_CP_CALLBACK);
2040                 ldlm_callback_reply(req, 0);
2041                 ldlm_handle_cp_callback(req, ns, dlm_req, lock);
2042                 break;
2043         case LDLM_GL_CALLBACK:
2044                 CDEBUG(D_INODE, "glimpse ast\n");
2045                 req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_CALLBACK);
2046                 ldlm_handle_gl_callback(req, ns, dlm_req, lock);
2047                 break;
2048         default:
2049                 LBUG();                         /* checked above */
2050         }
2051
2052         RETURN(0);
2053 }
2054
2055 #ifdef HAVE_SERVER_SUPPORT
2056 static int ldlm_cancel_handler(struct ptlrpc_request *req)
2057 {
2058         int rc;
2059         ENTRY;
2060
2061         /* Requests arrive in sender's byte order.  The ptlrpc service
2062          * handler has already checked and, if necessary, byte-swapped the
2063          * incoming request message body, but I am responsible for the
2064          * message buffers. */
2065
2066         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2067
2068         if (req->rq_export == NULL) {
2069                 struct ldlm_request *dlm_req;
2070
2071                 CERROR("%s from %s arrived at %lu with bad export cookie "
2072                        LPU64"\n",
2073                        ll_opcode2str(lustre_msg_get_opc(req->rq_reqmsg)),
2074                        libcfs_nid2str(req->rq_peer.nid),
2075                        req->rq_arrival_time.tv_sec,
2076                        lustre_msg_get_handle(req->rq_reqmsg)->cookie);
2077
2078                 if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_CANCEL) {
2079                         req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
2080                         dlm_req = req_capsule_client_get(&req->rq_pill,
2081                                                          &RMF_DLM_REQ);
2082                         if (dlm_req != NULL)
2083                                 ldlm_lock_dump_handle(D_ERROR,
2084                                                       &dlm_req->lock_handle[0]);
2085                 }
2086                 ldlm_callback_reply(req, -ENOTCONN);
2087                 RETURN(0);
2088         }
2089
2090         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2091
2092         /* XXX FIXME move this back to mds/handler.c, bug 249 */
2093         case LDLM_CANCEL:
2094                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
2095                 CDEBUG(D_INODE, "cancel\n");
2096                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL))
2097                         RETURN(0);
2098                 rc = ldlm_handle_cancel(req);
2099                 if (rc)
2100                         break;
2101                 RETURN(0);
2102         case OBD_LOG_CANCEL:
2103                 req_capsule_set(&req->rq_pill, &RQF_LOG_CANCEL);
2104                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_NET))
2105                         RETURN(0);
2106                 rc = llog_origin_handle_cancel(req);
2107                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_REP))
2108                         RETURN(0);
2109                 ldlm_callback_reply(req, rc);
2110                 RETURN(0);
2111         default:
2112                 CERROR("invalid opcode %d\n",
2113                        lustre_msg_get_opc(req->rq_reqmsg));
2114                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
2115                 ldlm_callback_reply(req, -EINVAL);
2116         }
2117
2118         RETURN(0);
2119 }
2120
2121 static int ldlm_cancel_hpreq_lock_match(struct ptlrpc_request *req,
2122                                         struct ldlm_lock *lock)
2123 {
2124         struct ldlm_request *dlm_req;
2125         struct lustre_handle lockh;
2126         int rc = 0;
2127         int i;
2128         ENTRY;
2129
2130         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
2131         if (dlm_req == NULL)
2132                 RETURN(0);
2133
2134         ldlm_lock2handle(lock, &lockh);
2135         for (i = 0; i < dlm_req->lock_count; i++) {
2136                 if (lustre_handle_equal(&dlm_req->lock_handle[i],
2137                                         &lockh)) {
2138                         DEBUG_REQ(D_RPCTRACE, req,
2139                                   "Prio raised by lock "LPX64".", lockh.cookie);
2140
2141                         rc = 1;
2142                         break;
2143                 }
2144         }
2145
2146         RETURN(rc);
2147
2148 }
2149
2150 static int ldlm_cancel_hpreq_check(struct ptlrpc_request *req)
2151 {
2152         struct ldlm_request *dlm_req;
2153         int rc = 0;
2154         int i;
2155         ENTRY;
2156
2157         /* no prolong in recovery */
2158         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
2159                 RETURN(0);
2160
2161         dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
2162         if (dlm_req == NULL)
2163                 RETURN(-EFAULT);
2164
2165         for (i = 0; i < dlm_req->lock_count; i++) {
2166                 struct ldlm_lock *lock;
2167
2168                 lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
2169                 if (lock == NULL)
2170                         continue;
2171
2172                 rc = !!(lock->l_flags & LDLM_FL_AST_SENT);
2173                 if (rc)
2174                         LDLM_DEBUG(lock, "hpreq cancel lock");
2175                 LDLM_LOCK_PUT(lock);
2176
2177                 if (rc)
2178                         break;
2179         }
2180
2181         RETURN(rc);
2182 }
2183
2184 static struct ptlrpc_hpreq_ops ldlm_cancel_hpreq_ops = {
2185         .hpreq_lock_match = ldlm_cancel_hpreq_lock_match,
2186         .hpreq_check      = ldlm_cancel_hpreq_check
2187 };
2188
2189 static int ldlm_hpreq_handler(struct ptlrpc_request *req)
2190 {
2191         ENTRY;
2192
2193         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2194
2195         if (req->rq_export == NULL)
2196                 RETURN(0);
2197
2198         if (LDLM_CANCEL == lustre_msg_get_opc(req->rq_reqmsg)) {
2199                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
2200                 req->rq_ops = &ldlm_cancel_hpreq_ops;
2201         }
2202         RETURN(0);
2203 }
2204
2205 int ldlm_revoke_lock_cb(cfs_hash_t *hs, cfs_hash_bd_t *bd,
2206                         cfs_hlist_node_t *hnode, void *data)
2207
2208 {
2209         cfs_list_t         *rpc_list = data;
2210         struct ldlm_lock   *lock = cfs_hash_object(hs, hnode);
2211
2212         lock_res_and_lock(lock);
2213
2214         if (lock->l_req_mode != lock->l_granted_mode) {
2215                 unlock_res_and_lock(lock);
2216                 return 0;
2217         }
2218
2219         LASSERT(lock->l_resource);
2220         if (lock->l_resource->lr_type != LDLM_IBITS &&
2221             lock->l_resource->lr_type != LDLM_PLAIN) {
2222                 unlock_res_and_lock(lock);
2223                 return 0;
2224         }
2225
2226         if (lock->l_flags & LDLM_FL_AST_SENT) {
2227                 unlock_res_and_lock(lock);
2228                 return 0;
2229         }
2230
2231         LASSERT(lock->l_blocking_ast);
2232         LASSERT(!lock->l_blocking_lock);
2233
2234         lock->l_flags |= LDLM_FL_AST_SENT;
2235         if (lock->l_export && lock->l_export->exp_lock_hash &&
2236             !cfs_hlist_unhashed(&lock->l_exp_hash))
2237                 cfs_hash_del(lock->l_export->exp_lock_hash,
2238                              &lock->l_remote_handle, &lock->l_exp_hash);
2239         cfs_list_add_tail(&lock->l_rk_ast, rpc_list);
2240         LDLM_LOCK_GET(lock);
2241
2242         unlock_res_and_lock(lock);
2243         return 0;
2244 }
2245
2246 void ldlm_revoke_export_locks(struct obd_export *exp)
2247 {
2248         cfs_list_t  rpc_list;
2249         ENTRY;
2250
2251         CFS_INIT_LIST_HEAD(&rpc_list);
2252         cfs_hash_for_each_empty(exp->exp_lock_hash,
2253                                 ldlm_revoke_lock_cb, &rpc_list);
2254         ldlm_run_ast_work(exp->exp_obd->obd_namespace, &rpc_list,
2255                           LDLM_WORK_REVOKE_AST);
2256
2257         EXIT;
2258 }
2259 #endif /* HAVE_SERVER_SUPPORT */
2260
2261 #ifdef __KERNEL__
2262 static struct ldlm_bl_work_item *ldlm_bl_get_work(struct ldlm_bl_pool *blp)
2263 {
2264         struct ldlm_bl_work_item *blwi = NULL;
2265         static unsigned int num_bl = 0;
2266
2267         cfs_spin_lock(&blp->blp_lock);
2268         /* process a request from the blp_list at least every blp_num_threads */
2269         if (!cfs_list_empty(&blp->blp_list) &&
2270             (cfs_list_empty(&blp->blp_prio_list) || num_bl == 0))
2271                 blwi = cfs_list_entry(blp->blp_list.next,
2272                                       struct ldlm_bl_work_item, blwi_entry);
2273         else
2274                 if (!cfs_list_empty(&blp->blp_prio_list))
2275                         blwi = cfs_list_entry(blp->blp_prio_list.next,
2276                                               struct ldlm_bl_work_item,
2277                                               blwi_entry);
2278
2279         if (blwi) {
2280                 if (++num_bl >= cfs_atomic_read(&blp->blp_num_threads))
2281                         num_bl = 0;
2282                 cfs_list_del(&blwi->blwi_entry);
2283         }
2284         cfs_spin_unlock(&blp->blp_lock);
2285
2286         return blwi;
2287 }
2288
2289 /* This only contains temporary data until the thread starts */
2290 struct ldlm_bl_thread_data {
2291         char                    bltd_name[CFS_CURPROC_COMM_MAX];
2292         struct ldlm_bl_pool     *bltd_blp;
2293         cfs_completion_t        bltd_comp;
2294         int                     bltd_num;
2295 };
2296
2297 static int ldlm_bl_thread_main(void *arg);
2298
2299 static int ldlm_bl_thread_start(struct ldlm_bl_pool *blp)
2300 {
2301         struct ldlm_bl_thread_data bltd = { .bltd_blp = blp };
2302         int rc;
2303
2304         cfs_init_completion(&bltd.bltd_comp);
2305         rc = cfs_create_thread(ldlm_bl_thread_main, &bltd, 0);
2306         if (rc < 0) {
2307                 CERROR("cannot start LDLM thread ldlm_bl_%02d: rc %d\n",
2308                        cfs_atomic_read(&blp->blp_num_threads), rc);
2309                 return rc;
2310         }
2311         cfs_wait_for_completion(&bltd.bltd_comp);
2312
2313         return 0;
2314 }
2315
2316 static int ldlm_bl_thread_main(void *arg)
2317 {
2318         struct ldlm_bl_pool *blp;
2319         ENTRY;
2320
2321         {
2322                 struct ldlm_bl_thread_data *bltd = arg;
2323
2324                 blp = bltd->bltd_blp;
2325
2326                 bltd->bltd_num =
2327                         cfs_atomic_inc_return(&blp->blp_num_threads) - 1;
2328                 cfs_atomic_inc(&blp->blp_busy_threads);
2329
2330                 snprintf(bltd->bltd_name, sizeof(bltd->bltd_name) - 1,
2331                         "ldlm_bl_%02d", bltd->bltd_num);
2332                 cfs_daemonize(bltd->bltd_name);
2333
2334                 cfs_complete(&bltd->bltd_comp);
2335                 /* cannot use bltd after this, it is only on caller's stack */
2336         }
2337
2338         while (1) {
2339                 struct l_wait_info lwi = { 0 };
2340                 struct ldlm_bl_work_item *blwi = NULL;
2341                 int busy;
2342
2343                 blwi = ldlm_bl_get_work(blp);
2344
2345                 if (blwi == NULL) {
2346                         cfs_atomic_dec(&blp->blp_busy_threads);
2347                         l_wait_event_exclusive(blp->blp_waitq,
2348                                          (blwi = ldlm_bl_get_work(blp)) != NULL,
2349                                          &lwi);
2350                         busy = cfs_atomic_inc_return(&blp->blp_busy_threads);
2351                 } else {
2352                         busy = cfs_atomic_read(&blp->blp_busy_threads);
2353                 }
2354
2355                 if (blwi->blwi_ns == NULL)
2356                         /* added by ldlm_cleanup() */
2357                         break;
2358
2359                 /* Not fatal if racy and have a few too many threads */
2360                 if (unlikely(busy < blp->blp_max_threads &&
2361                              busy >= cfs_atomic_read(&blp->blp_num_threads) &&
2362                              !blwi->blwi_mem_pressure))
2363                         /* discard the return value, we tried */
2364                         ldlm_bl_thread_start(blp);
2365
2366                 if (blwi->blwi_mem_pressure)
2367                         cfs_memory_pressure_set();
2368
2369                 if (blwi->blwi_count) {
2370                         int count;
2371                         /* The special case when we cancel locks in lru
2372                          * asynchronously, we pass the list of locks here.
2373                          * Thus locks are marked LDLM_FL_CANCELING, but NOT
2374                          * canceled locally yet. */
2375                         count = ldlm_cli_cancel_list_local(&blwi->blwi_head,
2376                                                            blwi->blwi_count,
2377                                                            LCF_BL_AST);
2378                         ldlm_cli_cancel_list(&blwi->blwi_head, count, NULL, 0);
2379                 } else {
2380                         ldlm_handle_bl_callback(blwi->blwi_ns, &blwi->blwi_ld,
2381                                                 blwi->blwi_lock);
2382                 }
2383                 if (blwi->blwi_mem_pressure)
2384                         cfs_memory_pressure_clr();
2385
2386                 if (blwi->blwi_mode == LDLM_ASYNC)
2387                         OBD_FREE(blwi, sizeof(*blwi));
2388                 else
2389                         cfs_complete(&blwi->blwi_comp);
2390         }
2391
2392         cfs_atomic_dec(&blp->blp_busy_threads);
2393         cfs_atomic_dec(&blp->blp_num_threads);
2394         cfs_complete(&blp->blp_comp);
2395         RETURN(0);
2396 }
2397
2398 #endif
2399
2400 static int ldlm_setup(void);
2401 static int ldlm_cleanup(void);
2402
2403 int ldlm_get_ref(void)
2404 {
2405         int rc = 0;
2406         ENTRY;
2407         cfs_mutex_lock(&ldlm_ref_mutex);
2408         if (++ldlm_refcount == 1) {
2409                 rc = ldlm_setup();
2410                 if (rc)
2411                         ldlm_refcount--;
2412         }
2413         cfs_mutex_unlock(&ldlm_ref_mutex);
2414
2415         RETURN(rc);
2416 }
2417
2418 void ldlm_put_ref(void)
2419 {
2420         ENTRY;
2421         cfs_mutex_lock(&ldlm_ref_mutex);
2422         if (ldlm_refcount == 1) {
2423                 int rc = ldlm_cleanup();
2424                 if (rc)
2425                         CERROR("ldlm_cleanup failed: %d\n", rc);
2426                 else
2427                         ldlm_refcount--;
2428         } else {
2429                 ldlm_refcount--;
2430         }
2431         cfs_mutex_unlock(&ldlm_ref_mutex);
2432
2433         EXIT;
2434 }
2435
2436 /*
2437  * Export handle<->lock hash operations.
2438  */
2439 static unsigned
2440 ldlm_export_lock_hash(cfs_hash_t *hs, const void *key, unsigned mask)
2441 {
2442         return cfs_hash_u64_hash(((struct lustre_handle *)key)->cookie, mask);
2443 }
2444
2445 static void *
2446 ldlm_export_lock_key(cfs_hlist_node_t *hnode)
2447 {
2448         struct ldlm_lock *lock;
2449
2450         lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2451         return &lock->l_remote_handle;
2452 }
2453
2454 static void
2455 ldlm_export_lock_keycpy(cfs_hlist_node_t *hnode, void *key)
2456 {
2457         struct ldlm_lock     *lock;
2458
2459         lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2460         lock->l_remote_handle = *(struct lustre_handle *)key;
2461 }
2462
2463 static int
2464 ldlm_export_lock_keycmp(const void *key, cfs_hlist_node_t *hnode)
2465 {
2466         return lustre_handle_equal(ldlm_export_lock_key(hnode), key);
2467 }
2468
2469 static void *
2470 ldlm_export_lock_object(cfs_hlist_node_t *hnode)
2471 {
2472         return cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2473 }
2474
2475 static void
2476 ldlm_export_lock_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
2477 {
2478         struct ldlm_lock *lock;
2479
2480         lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2481         LDLM_LOCK_GET(lock);
2482 }
2483
2484 static void
2485 ldlm_export_lock_put(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
2486 {
2487         struct ldlm_lock *lock;
2488
2489         lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
2490         LDLM_LOCK_RELEASE(lock);
2491 }
2492
2493 static cfs_hash_ops_t ldlm_export_lock_ops = {
2494         .hs_hash        = ldlm_export_lock_hash,
2495         .hs_key         = ldlm_export_lock_key,
2496         .hs_keycmp      = ldlm_export_lock_keycmp,
2497         .hs_keycpy      = ldlm_export_lock_keycpy,
2498         .hs_object      = ldlm_export_lock_object,
2499         .hs_get         = ldlm_export_lock_get,
2500         .hs_put         = ldlm_export_lock_put,
2501         .hs_put_locked  = ldlm_export_lock_put,
2502 };
2503
2504 int ldlm_init_export(struct obd_export *exp)
2505 {
2506         ENTRY;
2507
2508         exp->exp_lock_hash =
2509                 cfs_hash_create(obd_uuid2str(&exp->exp_client_uuid),
2510                                 HASH_EXP_LOCK_CUR_BITS,
2511                                 HASH_EXP_LOCK_MAX_BITS,
2512                                 HASH_EXP_LOCK_BKT_BITS, 0,
2513                                 CFS_HASH_MIN_THETA, CFS_HASH_MAX_THETA,
2514                                 &ldlm_export_lock_ops,
2515                                 CFS_HASH_DEFAULT | CFS_HASH_REHASH_KEY |
2516                                 CFS_HASH_NBLK_CHANGE);
2517
2518         if (!exp->exp_lock_hash)
2519                 RETURN(-ENOMEM);
2520
2521         RETURN(0);
2522 }
2523 EXPORT_SYMBOL(ldlm_init_export);
2524
2525 void ldlm_destroy_export(struct obd_export *exp)
2526 {
2527         ENTRY;
2528         cfs_hash_putref(exp->exp_lock_hash);
2529         exp->exp_lock_hash = NULL;
2530         EXIT;
2531 }
2532 EXPORT_SYMBOL(ldlm_destroy_export);
2533
2534 static int ldlm_setup(void)
2535 {
2536         struct ldlm_bl_pool *blp;
2537         int rc = 0;
2538         int ldlm_min_threads = LDLM_THREADS_AUTO_MIN;
2539         int ldlm_max_threads = LDLM_THREADS_AUTO_MAX;
2540 #ifdef __KERNEL__
2541         int i;
2542 #endif
2543         ENTRY;
2544
2545         if (ldlm_state != NULL)
2546                 RETURN(-EALREADY);
2547
2548         OBD_ALLOC(ldlm_state, sizeof(*ldlm_state));
2549         if (ldlm_state == NULL)
2550                 RETURN(-ENOMEM);
2551
2552 #ifdef LPROCFS
2553         rc = ldlm_proc_setup();
2554         if (rc != 0)
2555                 GOTO(out_free, rc);
2556 #endif
2557
2558 #ifdef __KERNEL__
2559         if (ldlm_num_threads) {
2560                 /* If ldlm_num_threads is set, it is the min and the max. */
2561                 if (ldlm_num_threads > LDLM_THREADS_AUTO_MAX)
2562                         ldlm_num_threads = LDLM_THREADS_AUTO_MAX;
2563                 if (ldlm_num_threads < LDLM_THREADS_AUTO_MIN)
2564                         ldlm_num_threads = LDLM_THREADS_AUTO_MIN;
2565                 ldlm_min_threads = ldlm_max_threads = ldlm_num_threads;
2566         }
2567 #endif
2568
2569         ldlm_state->ldlm_cb_service =
2570                 ptlrpc_init_svc(LDLM_NBUFS, LDLM_BUFSIZE, LDLM_MAXREQSIZE,
2571                                 LDLM_MAXREPSIZE, LDLM_CB_REQUEST_PORTAL,
2572                                 LDLM_CB_REPLY_PORTAL, 2,
2573                                 ldlm_callback_handler, "ldlm_cbd",
2574                                 ldlm_svc_proc_dir, NULL,
2575                                 ldlm_min_threads, ldlm_max_threads,
2576                                 "ldlm_cb",
2577                                 LCT_MD_THREAD|LCT_DT_THREAD, NULL);
2578
2579         if (!ldlm_state->ldlm_cb_service) {
2580                 CERROR("failed to start service\n");
2581                 GOTO(out_proc, rc = -ENOMEM);
2582         }
2583
2584 #ifdef HAVE_SERVER_SUPPORT
2585         ldlm_state->ldlm_cancel_service =
2586                 ptlrpc_init_svc(LDLM_NBUFS, LDLM_BUFSIZE, LDLM_MAXREQSIZE,
2587                                 LDLM_MAXREPSIZE, LDLM_CANCEL_REQUEST_PORTAL,
2588                                 LDLM_CANCEL_REPLY_PORTAL, 6,
2589                                 ldlm_cancel_handler, "ldlm_canceld",
2590                                 ldlm_svc_proc_dir, NULL,
2591                                 ldlm_min_threads, ldlm_max_threads,
2592                                 "ldlm_cn",
2593                                 LCT_MD_THREAD|LCT_DT_THREAD|LCT_CL_THREAD,
2594                                 ldlm_hpreq_handler);
2595
2596         if (!ldlm_state->ldlm_cancel_service) {
2597                 CERROR("failed to start service\n");
2598                 GOTO(out_proc, rc = -ENOMEM);
2599         }
2600 #endif
2601
2602         OBD_ALLOC(blp, sizeof(*blp));
2603         if (blp == NULL)
2604                 GOTO(out_proc, rc = -ENOMEM);
2605         ldlm_state->ldlm_bl_pool = blp;
2606
2607         cfs_spin_lock_init(&blp->blp_lock);
2608         CFS_INIT_LIST_HEAD(&blp->blp_list);
2609         CFS_INIT_LIST_HEAD(&blp->blp_prio_list);
2610         cfs_waitq_init(&blp->blp_waitq);
2611         cfs_atomic_set(&blp->blp_num_threads, 0);
2612         cfs_atomic_set(&blp->blp_busy_threads, 0);
2613         blp->blp_min_threads = ldlm_min_threads;
2614         blp->blp_max_threads = ldlm_max_threads;
2615
2616 #ifdef __KERNEL__
2617         for (i = 0; i < blp->blp_min_threads; i++) {
2618                 rc = ldlm_bl_thread_start(blp);
2619                 if (rc < 0)
2620                         GOTO(out_thread, rc);
2621         }
2622
2623 # ifdef HAVE_SERVER_SUPPORT
2624         rc = ptlrpc_start_threads(ldlm_state->ldlm_cancel_service);
2625         if (rc)
2626                 GOTO(out_thread, rc);
2627 # endif
2628
2629         rc = ptlrpc_start_threads(ldlm_state->ldlm_cb_service);
2630         if (rc)
2631                 GOTO(out_thread, rc);
2632
2633         CFS_INIT_LIST_HEAD(&expired_lock_thread.elt_expired_locks);
2634         expired_lock_thread.elt_state = ELT_STOPPED;
2635         cfs_waitq_init(&expired_lock_thread.elt_waitq);
2636
2637         CFS_INIT_LIST_HEAD(&waiting_locks_list);
2638         cfs_spin_lock_init(&waiting_locks_spinlock);
2639         cfs_timer_init(&waiting_locks_timer, waiting_locks_callback, 0);
2640
2641         rc = cfs_create_thread(expired_lock_main, NULL, CFS_DAEMON_FLAGS);
2642         if (rc < 0) {
2643                 CERROR("Cannot start ldlm expired-lock thread: %d\n", rc);
2644                 GOTO(out_thread, rc);
2645         }
2646
2647         cfs_wait_event(expired_lock_thread.elt_waitq,
2648                        expired_lock_thread.elt_state == ELT_READY);
2649 #endif
2650
2651 #ifdef __KERNEL__
2652         rc = ldlm_pools_init();
2653         if (rc)
2654                 GOTO(out_thread, rc);
2655 #endif
2656         RETURN(0);
2657
2658 #ifdef __KERNEL__
2659  out_thread:
2660 # ifdef HAVE_SERVER_SUPPORT
2661         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
2662 # endif
2663         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
2664 #endif
2665
2666  out_proc:
2667 #ifdef LPROCFS
2668         ldlm_proc_cleanup();
2669  out_free:
2670 #endif
2671         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
2672         ldlm_state = NULL;
2673         return rc;
2674 }
2675
2676 static int ldlm_cleanup(void)
2677 {
2678 #ifdef __KERNEL__
2679         struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
2680 #endif
2681         ENTRY;
2682
2683         if (!cfs_list_empty(ldlm_namespace_list(LDLM_NAMESPACE_SERVER)) ||
2684             !cfs_list_empty(ldlm_namespace_list(LDLM_NAMESPACE_CLIENT))) {
2685                 CERROR("ldlm still has namespaces; clean these up first.\n");
2686                 ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
2687                 ldlm_dump_all_namespaces(LDLM_NAMESPACE_CLIENT, D_DLMTRACE);
2688                 RETURN(-EBUSY);
2689         }
2690
2691 #ifdef __KERNEL__
2692         ldlm_pools_fini();
2693 #endif
2694
2695 #ifdef __KERNEL__
2696         while (cfs_atomic_read(&blp->blp_num_threads) > 0) {
2697                 struct ldlm_bl_work_item blwi = { .blwi_ns = NULL };
2698
2699                 cfs_init_completion(&blp->blp_comp);
2700
2701                 cfs_spin_lock(&blp->blp_lock);
2702                 cfs_list_add_tail(&blwi.blwi_entry, &blp->blp_list);
2703                 cfs_waitq_signal(&blp->blp_waitq);
2704                 cfs_spin_unlock(&blp->blp_lock);
2705
2706                 cfs_wait_for_completion(&blp->blp_comp);
2707         }
2708         OBD_FREE(blp, sizeof(*blp));
2709
2710         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
2711 # ifdef HAVE_SERVER_SUPPORT
2712         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
2713 # endif
2714         ldlm_proc_cleanup();
2715
2716         expired_lock_thread.elt_state = ELT_TERMINATE;
2717         cfs_waitq_signal(&expired_lock_thread.elt_waitq);
2718         cfs_wait_event(expired_lock_thread.elt_waitq,
2719                        expired_lock_thread.elt_state == ELT_STOPPED);
2720 #else /* !__KERNEL__ */
2721         ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);
2722 # ifdef HAVE_SERVER_SUPPORT
2723         ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
2724 # endif
2725 #endif /* __KERNEL__ */
2726
2727         OBD_FREE(ldlm_state, sizeof(*ldlm_state));
2728         ldlm_state = NULL;
2729
2730         RETURN(0);
2731 }
2732
2733 int ldlm_init(void)
2734 {
2735         cfs_mutex_init(&ldlm_ref_mutex);
2736         cfs_mutex_init(ldlm_namespace_lock(LDLM_NAMESPACE_SERVER));
2737         cfs_mutex_init(ldlm_namespace_lock(LDLM_NAMESPACE_CLIENT));
2738         ldlm_resource_slab = cfs_mem_cache_create("ldlm_resources",
2739                                                sizeof(struct ldlm_resource), 0,
2740                                                CFS_SLAB_HWCACHE_ALIGN);
2741         if (ldlm_resource_slab == NULL)
2742                 return -ENOMEM;
2743
2744         ldlm_lock_slab = cfs_mem_cache_create("ldlm_locks",
2745                               sizeof(struct ldlm_lock), 0,
2746                               CFS_SLAB_HWCACHE_ALIGN | CFS_SLAB_DESTROY_BY_RCU);
2747         if (ldlm_lock_slab == NULL) {
2748                 cfs_mem_cache_destroy(ldlm_resource_slab);
2749                 return -ENOMEM;
2750         }
2751
2752         ldlm_interval_slab = cfs_mem_cache_create("interval_node",
2753                                         sizeof(struct ldlm_interval),
2754                                         0, CFS_SLAB_HWCACHE_ALIGN);
2755         if (ldlm_interval_slab == NULL) {
2756                 cfs_mem_cache_destroy(ldlm_resource_slab);
2757                 cfs_mem_cache_destroy(ldlm_lock_slab);
2758                 return -ENOMEM;
2759         }
2760 #if LUSTRE_TRACKS_LOCK_EXP_REFS
2761         class_export_dump_hook = ldlm_dump_export_locks;
2762 #endif
2763         return 0;
2764 }
2765
2766 void ldlm_exit(void)
2767 {
2768         int rc;
2769         if (ldlm_refcount)
2770                 CERROR("ldlm_refcount is %d in ldlm_exit!\n", ldlm_refcount);
2771         rc = cfs_mem_cache_destroy(ldlm_resource_slab);
2772         LASSERTF(rc == 0, "couldn't free ldlm resource slab\n");
2773 #ifdef __KERNEL__
2774         /* ldlm_lock_put() use RCU to call ldlm_lock_free, so need call
2775          * synchronize_rcu() to wait a grace period elapsed, so that
2776          * ldlm_lock_free() get a chance to be called. */
2777         synchronize_rcu();
2778 #endif
2779         rc = cfs_mem_cache_destroy(ldlm_lock_slab);
2780         LASSERTF(rc == 0, "couldn't free ldlm lock slab\n");
2781         rc = cfs_mem_cache_destroy(ldlm_interval_slab);
2782         LASSERTF(rc == 0, "couldn't free interval node slab\n");
2783 }
2784
2785 /* ldlm_extent.c */
2786 EXPORT_SYMBOL(ldlm_extent_shift_kms);
2787
2788 /* ldlm_lock.c */
2789 #ifdef HAVE_SERVER_SUPPORT
2790 EXPORT_SYMBOL(ldlm_get_processing_policy);
2791 #endif
2792 EXPORT_SYMBOL(ldlm_lock2desc);
2793 EXPORT_SYMBOL(ldlm_register_intent);
2794 EXPORT_SYMBOL(ldlm_lockname);
2795 EXPORT_SYMBOL(ldlm_typename);
2796 EXPORT_SYMBOL(ldlm_lock2handle);
2797 EXPORT_SYMBOL(__ldlm_handle2lock);
2798 EXPORT_SYMBOL(ldlm_lock_get);
2799 EXPORT_SYMBOL(ldlm_lock_put);
2800 EXPORT_SYMBOL(ldlm_lock_match);
2801 EXPORT_SYMBOL(ldlm_lock_cancel);
2802 EXPORT_SYMBOL(ldlm_lock_addref);
2803 EXPORT_SYMBOL(ldlm_lock_addref_try);
2804 EXPORT_SYMBOL(ldlm_lock_decref);
2805 EXPORT_SYMBOL(ldlm_lock_decref_and_cancel);
2806 EXPORT_SYMBOL(ldlm_lock_change_resource);
2807 EXPORT_SYMBOL(ldlm_it2str);
2808 EXPORT_SYMBOL(ldlm_lock_dump);
2809 EXPORT_SYMBOL(ldlm_lock_dump_handle);
2810 EXPORT_SYMBOL(ldlm_reprocess_all_ns);
2811 EXPORT_SYMBOL(ldlm_lock_allow_match_locked);
2812 EXPORT_SYMBOL(ldlm_lock_allow_match);
2813 EXPORT_SYMBOL(ldlm_lock_downgrade);
2814 EXPORT_SYMBOL(ldlm_lock_convert);
2815
2816 /* ldlm_request.c */
2817 EXPORT_SYMBOL(ldlm_completion_ast_async);
2818 EXPORT_SYMBOL(ldlm_blocking_ast_nocheck);
2819 EXPORT_SYMBOL(ldlm_completion_ast);
2820 EXPORT_SYMBOL(ldlm_blocking_ast);
2821 EXPORT_SYMBOL(ldlm_glimpse_ast);
2822 EXPORT_SYMBOL(ldlm_expired_completion_wait);
2823 EXPORT_SYMBOL(ldlm_prep_enqueue_req);
2824 EXPORT_SYMBOL(ldlm_prep_elc_req);
2825 EXPORT_SYMBOL(ldlm_cli_convert);
2826 EXPORT_SYMBOL(ldlm_cli_enqueue);
2827 EXPORT_SYMBOL(ldlm_cli_enqueue_fini);
2828 EXPORT_SYMBOL(ldlm_cli_enqueue_local);
2829 EXPORT_SYMBOL(ldlm_cli_cancel);
2830 EXPORT_SYMBOL(ldlm_cli_cancel_unused);
2831 EXPORT_SYMBOL(ldlm_cli_cancel_unused_resource);
2832 EXPORT_SYMBOL(ldlm_cli_cancel_req);
2833 EXPORT_SYMBOL(ldlm_replay_locks);
2834 EXPORT_SYMBOL(ldlm_resource_foreach);
2835 EXPORT_SYMBOL(ldlm_namespace_foreach);
2836 EXPORT_SYMBOL(ldlm_resource_iterate);
2837 EXPORT_SYMBOL(ldlm_cancel_resource_local);
2838 EXPORT_SYMBOL(ldlm_cli_cancel_list_local);
2839 EXPORT_SYMBOL(ldlm_cli_cancel_list);
2840
2841 /* ldlm_lockd.c */
2842 #ifdef HAVE_SERVER_SUPPORT
2843 EXPORT_SYMBOL(ldlm_server_blocking_ast);
2844 EXPORT_SYMBOL(ldlm_server_completion_ast);
2845 EXPORT_SYMBOL(ldlm_server_glimpse_ast);
2846 EXPORT_SYMBOL(ldlm_handle_enqueue);
2847 EXPORT_SYMBOL(ldlm_handle_enqueue0);
2848 EXPORT_SYMBOL(ldlm_handle_cancel);
2849 EXPORT_SYMBOL(ldlm_request_cancel);
2850 EXPORT_SYMBOL(ldlm_handle_convert);
2851 EXPORT_SYMBOL(ldlm_handle_convert0);
2852 EXPORT_SYMBOL(ldlm_revoke_export_locks);
2853 #endif
2854 EXPORT_SYMBOL(ldlm_del_waiting_lock);
2855 EXPORT_SYMBOL(ldlm_get_ref);
2856 EXPORT_SYMBOL(ldlm_put_ref);
2857 EXPORT_SYMBOL(ldlm_refresh_waiting_lock);
2858
2859 /* ldlm_resource.c */
2860 EXPORT_SYMBOL(ldlm_namespace_new);
2861 EXPORT_SYMBOL(ldlm_namespace_cleanup);
2862 EXPORT_SYMBOL(ldlm_namespace_free);
2863 EXPORT_SYMBOL(ldlm_namespace_dump);
2864 EXPORT_SYMBOL(ldlm_dump_all_namespaces);
2865 EXPORT_SYMBOL(ldlm_resource_get);
2866 EXPORT_SYMBOL(ldlm_resource_putref);
2867 EXPORT_SYMBOL(ldlm_resource_unlink_lock);
2868
2869 /* ldlm_lib.c */
2870 EXPORT_SYMBOL(client_import_add_conn);
2871 EXPORT_SYMBOL(client_import_del_conn);
2872 EXPORT_SYMBOL(client_obd_setup);
2873 EXPORT_SYMBOL(client_obd_cleanup);
2874 EXPORT_SYMBOL(client_connect_import);
2875 EXPORT_SYMBOL(client_disconnect_export);
2876 EXPORT_SYMBOL(target_send_reply);
2877 EXPORT_SYMBOL(target_pack_pool_reply);
2878
2879 #ifdef HAVE_SERVER_SUPPORT
2880 EXPORT_SYMBOL(server_disconnect_export);
2881 EXPORT_SYMBOL(target_stop_recovery_thread);
2882 EXPORT_SYMBOL(target_handle_connect);
2883 EXPORT_SYMBOL(target_cleanup_recovery);
2884 EXPORT_SYMBOL(target_destroy_export);
2885 EXPORT_SYMBOL(target_cancel_recovery_timer);
2886 EXPORT_SYMBOL(target_queue_recovery_request);
2887 EXPORT_SYMBOL(target_handle_ping);
2888 EXPORT_SYMBOL(target_handle_disconnect);
2889 #endif
2890
2891 /* l_lock.c */
2892 EXPORT_SYMBOL(lock_res_and_lock);
2893 EXPORT_SYMBOL(unlock_res_and_lock);