lustre/ldlm/ldlm_lockd.c

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