lustre/ldlm/ldlm_lockd.c

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