lustre/ldlm/ldlm_lockd.c

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