/*
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 only,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License version 2 for more details (a copy is included
 * in the LICENSE file that accompanied this code).
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; If not, see
 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
 *
 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
 * CA 95054 USA or visit www.sun.com if you need additional information or
 * have any questions.
 *
 * GPL HEADER END
 */
/*
 * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
 * Use is subject to license terms.
 *
 * Copyright (c) 2010, 2012, Intel Corporation.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 *
 * lustre/ldlm/ldlm_lockd.c
 *
 * Author: Peter Braam <braam@clusterfs.com>
 * Author: Phil Schwan <phil@clusterfs.com>
 */

#define DEBUG_SUBSYSTEM S_LDLM

#include "../../include/linux/libcfs/libcfs.h"
#include "../include/lustre_dlm.h"
#include "../include/obd_class.h"
#include <linux/list.h>
#include "ldlm_internal.h"

static int ldlm_num_threads;
module_param(ldlm_num_threads, int, 0444);
MODULE_PARM_DESC(ldlm_num_threads, "number of DLM service threads to start");

static char *ldlm_cpts;
module_param(ldlm_cpts, charp, 0444);
MODULE_PARM_DESC(ldlm_cpts, "CPU partitions ldlm threads should run on");

extern struct kmem_cache *ldlm_resource_slab;
extern struct kmem_cache *ldlm_lock_slab;
static struct mutex	ldlm_ref_mutex;
static int ldlm_refcount;

struct ldlm_cb_async_args {
	struct ldlm_cb_set_arg *ca_set_arg;
	struct ldlm_lock       *ca_lock;
};

/* LDLM state */

static struct ldlm_state *ldlm_state;

inline unsigned long round_timeout(unsigned long timeout)
{
	return cfs_time_seconds((int)cfs_duration_sec(cfs_time_sub(timeout, 0)) + 1);
}

/* timeout for initial callback (AST) reply (bz10399) */
static inline unsigned int ldlm_get_rq_timeout(void)
{
	/* Non-AT value */
	unsigned int timeout = min(ldlm_timeout, obd_timeout / 3);

	return timeout < 1 ? 1 : timeout;
}

#define ELT_STOPPED   0
#define ELT_READY     1
#define ELT_TERMINATE 2

struct ldlm_bl_pool {
	spinlock_t		blp_lock;

	/*
	 * blp_prio_list is used for callbacks that should be handled
	 * as a priority. It is used for LDLM_FL_DISCARD_DATA requests.
	 * see bug 13843
	 */
	struct list_head	      blp_prio_list;

	/*
	 * blp_list is used for all other callbacks which are likely
	 * to take longer to process.
	 */
	struct list_head	      blp_list;

	wait_queue_head_t	     blp_waitq;
	struct completion	blp_comp;
	atomic_t	    blp_num_threads;
	atomic_t	    blp_busy_threads;
	int		     blp_min_threads;
	int		     blp_max_threads;
};

struct ldlm_bl_work_item {
	struct list_head	      blwi_entry;
	struct ldlm_namespace  *blwi_ns;
	struct ldlm_lock_desc   blwi_ld;
	struct ldlm_lock       *blwi_lock;
	struct list_head	      blwi_head;
	int		     blwi_count;
	struct completion	blwi_comp;
	ldlm_cancel_flags_t     blwi_flags;
	int		     blwi_mem_pressure;
};


int ldlm_del_waiting_lock(struct ldlm_lock *lock)
{
	return 0;
}

int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
{
	return 0;
}



/**
 * Callback handler for receiving incoming blocking ASTs.
 *
 * This can only happen on client side.
 */
void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
			     struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
{
	int do_ast;

	LDLM_DEBUG(lock, "client blocking AST callback handler");

	lock_res_and_lock(lock);
	lock->l_flags |= LDLM_FL_CBPENDING;

	if (lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)
		lock->l_flags |= LDLM_FL_CANCEL;

	do_ast = (!lock->l_readers && !lock->l_writers);
	unlock_res_and_lock(lock);

	if (do_ast) {
		CDEBUG(D_DLMTRACE, "Lock %p already unused, calling callback (%p)\n",
		       lock, lock->l_blocking_ast);
		if (lock->l_blocking_ast != NULL)
			lock->l_blocking_ast(lock, ld, lock->l_ast_data,
					     LDLM_CB_BLOCKING);
	} else {
		CDEBUG(D_DLMTRACE, "Lock %p is referenced, will be cancelled later\n",
		       lock);
	}

	LDLM_DEBUG(lock, "client blocking callback handler END");
	LDLM_LOCK_RELEASE(lock);
}

/**
 * Callback handler for receiving incoming completion ASTs.
 *
 * This only can happen on client side.
 */
static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
				    struct ldlm_namespace *ns,
				    struct ldlm_request *dlm_req,
				    struct ldlm_lock *lock)
{
	int lvb_len;
	LIST_HEAD(ast_list);
	int rc = 0;

	LDLM_DEBUG(lock, "client completion callback handler START");

	if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE)) {
		int to = cfs_time_seconds(1);
		while (to > 0) {
			set_current_state(TASK_INTERRUPTIBLE);
			schedule_timeout(to);
			if (lock->l_granted_mode == lock->l_req_mode ||
			    lock->l_flags & LDLM_FL_DESTROYED)
				break;
		}
	}

	lvb_len = req_capsule_get_size(&req->rq_pill, &RMF_DLM_LVB, RCL_CLIENT);
	if (lvb_len < 0) {
		LDLM_ERROR(lock, "Fail to get lvb_len, rc = %d", lvb_len);
		rc = lvb_len;
		goto out;
	} else if (lvb_len > 0) {
		if (lock->l_lvb_len > 0) {
			/* for extent lock, lvb contains ost_lvb{}. */
			LASSERT(lock->l_lvb_data != NULL);

			if (unlikely(lock->l_lvb_len < lvb_len)) {
				LDLM_ERROR(lock, "Replied LVB is larger than "
					   "expectation, expected = %d, "
					   "replied = %d",
					   lock->l_lvb_len, lvb_len);
				rc = -EINVAL;
				goto out;
			}
		} else if (ldlm_has_layout(lock)) { /* for layout lock, lvb has
						     * variable length */
			void *lvb_data;

			OBD_ALLOC(lvb_data, lvb_len);
			if (lvb_data == NULL) {
				LDLM_ERROR(lock, "No memory: %d.\n", lvb_len);
				rc = -ENOMEM;
				goto out;
			}

			lock_res_and_lock(lock);
			LASSERT(lock->l_lvb_data == NULL);
			lock->l_lvb_type = LVB_T_LAYOUT;
			lock->l_lvb_data = lvb_data;
			lock->l_lvb_len = lvb_len;
			unlock_res_and_lock(lock);
		}
	}

	lock_res_and_lock(lock);
	if ((lock->l_flags & LDLM_FL_DESTROYED) ||
	    lock->l_granted_mode == lock->l_req_mode) {
		/* bug 11300: the lock has already been granted */
		unlock_res_and_lock(lock);
		LDLM_DEBUG(lock, "Double grant race happened");
		rc = 0;
		goto out;
	}

	/* If we receive the completion AST before the actual enqueue returned,
	 * then we might need to switch lock modes, resources, or extents. */
	if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
		lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
		LDLM_DEBUG(lock, "completion AST, new lock mode");
	}

	if (lock->l_resource->lr_type != LDLM_PLAIN) {
		ldlm_convert_policy_to_local(req->rq_export,
					  dlm_req->lock_desc.l_resource.lr_type,
					  &dlm_req->lock_desc.l_policy_data,
					  &lock->l_policy_data);
		LDLM_DEBUG(lock, "completion AST, new policy data");
	}

	ldlm_resource_unlink_lock(lock);
	if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
		   &lock->l_resource->lr_name,
		   sizeof(lock->l_resource->lr_name)) != 0) {
		unlock_res_and_lock(lock);
		rc = ldlm_lock_change_resource(ns, lock,
				&dlm_req->lock_desc.l_resource.lr_name);
		if (rc < 0) {
			LDLM_ERROR(lock, "Failed to allocate resource");
			goto out;
		}
		LDLM_DEBUG(lock, "completion AST, new resource");
		CERROR("change resource!\n");
		lock_res_and_lock(lock);
	}

	if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
		/* BL_AST locks are not needed in LRU.
		 * Let ldlm_cancel_lru() be fast. */
		ldlm_lock_remove_from_lru(lock);
		lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_BL_AST;
		LDLM_DEBUG(lock, "completion AST includes blocking AST");
	}

	if (lock->l_lvb_len > 0) {
		rc = ldlm_fill_lvb(lock, &req->rq_pill, RCL_CLIENT,
				   lock->l_lvb_data, lvb_len);
		if (rc < 0) {
			unlock_res_and_lock(lock);
			goto out;
		}
	}

	ldlm_grant_lock(lock, &ast_list);
	unlock_res_and_lock(lock);

	LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");

	/* Let Enqueue to call osc_lock_upcall() and initialize
	 * l_ast_data */
	OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 2);

	ldlm_run_ast_work(ns, &ast_list, LDLM_WORK_CP_AST);

	LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
			  lock);
	goto out;

out:
	if (rc < 0) {
		lock_res_and_lock(lock);
		lock->l_flags |= LDLM_FL_FAILED;
		unlock_res_and_lock(lock);
		wake_up(&lock->l_waitq);
	}
	LDLM_LOCK_RELEASE(lock);
}

/**
 * Callback handler for receiving incoming glimpse ASTs.
 *
 * This only can happen on client side.  After handling the glimpse AST
 * we also consider dropping the lock here if it is unused locally for a
 * long time.
 */
static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
				    struct ldlm_namespace *ns,
				    struct ldlm_request *dlm_req,
				    struct ldlm_lock *lock)
{
	int rc = -ENOSYS;

	LDLM_DEBUG(lock, "client glimpse AST callback handler");

	if (lock->l_glimpse_ast != NULL)
		rc = lock->l_glimpse_ast(lock, req);

	if (req->rq_repmsg != NULL) {
		ptlrpc_reply(req);
	} else {
		req->rq_status = rc;
		ptlrpc_error(req);
	}

	lock_res_and_lock(lock);
	if (lock->l_granted_mode == LCK_PW &&
	    !lock->l_readers && !lock->l_writers &&
	    cfs_time_after(cfs_time_current(),
			   cfs_time_add(lock->l_last_used,
					cfs_time_seconds(10)))) {
		unlock_res_and_lock(lock);
		if (ldlm_bl_to_thread_lock(ns, NULL, lock))
			ldlm_handle_bl_callback(ns, NULL, lock);

		return;
	}
	unlock_res_and_lock(lock);
	LDLM_LOCK_RELEASE(lock);
}

static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
{
	if (req->rq_no_reply)
		return 0;

	req->rq_status = rc;
	if (!req->rq_packed_final) {
		rc = lustre_pack_reply(req, 1, NULL, NULL);
		if (rc)
			return rc;
	}
	return ptlrpc_reply(req);
}

static int __ldlm_bl_to_thread(struct ldlm_bl_work_item *blwi,
			       ldlm_cancel_flags_t cancel_flags)
{
	struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;

	spin_lock(&blp->blp_lock);
	if (blwi->blwi_lock &&
	    blwi->blwi_lock->l_flags & LDLM_FL_DISCARD_DATA) {
		/* add LDLM_FL_DISCARD_DATA requests to the priority list */
		list_add_tail(&blwi->blwi_entry, &blp->blp_prio_list);
	} else {
		/* other blocking callbacks are added to the regular list */
		list_add_tail(&blwi->blwi_entry, &blp->blp_list);
	}
	spin_unlock(&blp->blp_lock);

	wake_up(&blp->blp_waitq);

	/* can not check blwi->blwi_flags as blwi could be already freed in
	   LCF_ASYNC mode */
	if (!(cancel_flags & LCF_ASYNC))
		wait_for_completion(&blwi->blwi_comp);

	return 0;
}

static inline void init_blwi(struct ldlm_bl_work_item *blwi,
			     struct ldlm_namespace *ns,
			     struct ldlm_lock_desc *ld,
			     struct list_head *cancels, int count,
			     struct ldlm_lock *lock,
			     ldlm_cancel_flags_t cancel_flags)
{
	init_completion(&blwi->blwi_comp);
	INIT_LIST_HEAD(&blwi->blwi_head);

	if (memory_pressure_get())
		blwi->blwi_mem_pressure = 1;

	blwi->blwi_ns = ns;
	blwi->blwi_flags = cancel_flags;
	if (ld != NULL)
		blwi->blwi_ld = *ld;
	if (count) {
		list_add(&blwi->blwi_head, cancels);
		list_del_init(cancels);
		blwi->blwi_count = count;
	} else {
		blwi->blwi_lock = lock;
	}
}

/**
 * Queues a list of locks \a cancels containing \a count locks
 * for later processing by a blocking thread.  If \a count is zero,
 * then the lock referenced as \a lock is queued instead.
 *
 * The blocking thread would then call ->l_blocking_ast callback in the lock.
 * If list addition fails an error is returned and caller is supposed to
 * call ->l_blocking_ast itself.
 */
static int ldlm_bl_to_thread(struct ldlm_namespace *ns,
			     struct ldlm_lock_desc *ld,
			     struct ldlm_lock *lock,
			     struct list_head *cancels, int count,
			     ldlm_cancel_flags_t cancel_flags)
{
	if (cancels && count == 0)
		return 0;

	if (cancel_flags & LCF_ASYNC) {
		struct ldlm_bl_work_item *blwi;

		OBD_ALLOC(blwi, sizeof(*blwi));
		if (blwi == NULL)
			return -ENOMEM;
		init_blwi(blwi, ns, ld, cancels, count, lock, cancel_flags);

		return __ldlm_bl_to_thread(blwi, cancel_flags);
	} else {
		/* if it is synchronous call do minimum mem alloc, as it could
		 * be triggered from kernel shrinker
		 */
		struct ldlm_bl_work_item blwi;

		memset(&blwi, 0, sizeof(blwi));
		init_blwi(&blwi, ns, ld, cancels, count, lock, cancel_flags);
		return __ldlm_bl_to_thread(&blwi, cancel_flags);
	}
}


int ldlm_bl_to_thread_lock(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
			   struct ldlm_lock *lock)
{
	return ldlm_bl_to_thread(ns, ld, lock, NULL, 0, LCF_ASYNC);
}

int ldlm_bl_to_thread_list(struct ldlm_namespace *ns, struct ldlm_lock_desc *ld,
			   struct list_head *cancels, int count,
			   ldlm_cancel_flags_t cancel_flags)
{
	return ldlm_bl_to_thread(ns, ld, NULL, cancels, count, cancel_flags);
}

/* Setinfo coming from Server (eg MDT) to Client (eg MDC)! */
static int ldlm_handle_setinfo(struct ptlrpc_request *req)
{
	struct obd_device *obd = req->rq_export->exp_obd;
	char *key;
	void *val;
	int keylen, vallen;
	int rc = -ENOSYS;

	DEBUG_REQ(D_HSM, req, "%s: handle setinfo\n", obd->obd_name);

	req_capsule_set(&req->rq_pill, &RQF_OBD_SET_INFO);

	key = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
	if (key == NULL) {
		DEBUG_REQ(D_IOCTL, req, "no set_info key");
		return -EFAULT;
	}
	keylen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_KEY,
				      RCL_CLIENT);
	val = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
	if (val == NULL) {
		DEBUG_REQ(D_IOCTL, req, "no set_info val");
		return -EFAULT;
	}
	vallen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_VAL,
				      RCL_CLIENT);

	/* We are responsible for swabbing contents of val */

	if (KEY_IS(KEY_HSM_COPYTOOL_SEND))
		/* Pass it on to mdc (the "export" in this case) */
		rc = obd_set_info_async(req->rq_svc_thread->t_env,
					req->rq_export,
					sizeof(KEY_HSM_COPYTOOL_SEND),
					KEY_HSM_COPYTOOL_SEND,
					vallen, val, NULL);
	else
		DEBUG_REQ(D_WARNING, req, "ignoring unknown key %s", key);

	return rc;
}

static inline void ldlm_callback_errmsg(struct ptlrpc_request *req,
					const char *msg, int rc,
					struct lustre_handle *handle)
{
	DEBUG_REQ((req->rq_no_reply || rc) ? D_WARNING : D_DLMTRACE, req,
		  "%s: [nid %s] [rc %d] [lock %#llx]",
		  msg, libcfs_id2str(req->rq_peer), rc,
		  handle ? handle->cookie : 0);
	if (req->rq_no_reply)
		CWARN("No reply was sent, maybe cause bug 21636.\n");
	else if (rc)
		CWARN("Send reply failed, maybe cause bug 21636.\n");
}

static int ldlm_handle_qc_callback(struct ptlrpc_request *req)
{
	struct obd_quotactl *oqctl;
	struct client_obd *cli = &req->rq_export->exp_obd->u.cli;

	oqctl = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
	if (oqctl == NULL) {
		CERROR("Can't unpack obd_quotactl\n");
		return -EPROTO;
	}

	oqctl->qc_stat = ptlrpc_status_ntoh(oqctl->qc_stat);

	cli->cl_qchk_stat = oqctl->qc_stat;
	return 0;
}

/* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
static int ldlm_callback_handler(struct ptlrpc_request *req)
{
	struct ldlm_namespace *ns;
	struct ldlm_request *dlm_req;
	struct ldlm_lock *lock;
	int rc;

	/* Requests arrive in sender's byte order.  The ptlrpc service
	 * handler has already checked and, if necessary, byte-swapped the
	 * incoming request message body, but I am responsible for the
	 * message buffers. */

	/* do nothing for sec context finalize */
	if (lustre_msg_get_opc(req->rq_reqmsg) == SEC_CTX_FINI)
		return 0;

	req_capsule_init(&req->rq_pill, req, RCL_SERVER);

	if (req->rq_export == NULL) {
		rc = ldlm_callback_reply(req, -ENOTCONN);
		ldlm_callback_errmsg(req, "Operate on unconnected server",
				     rc, NULL);
		return 0;
	}

	LASSERT(req->rq_export != NULL);
	LASSERT(req->rq_export->exp_obd != NULL);

	switch (lustre_msg_get_opc(req->rq_reqmsg)) {
	case LDLM_BL_CALLBACK:
		if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
			return 0;
		break;
	case LDLM_CP_CALLBACK:
		if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CP_CALLBACK_NET))
			return 0;
		break;
	case LDLM_GL_CALLBACK:
		if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_GL_CALLBACK_NET))
			return 0;
		break;
	case LDLM_SET_INFO:
		rc = ldlm_handle_setinfo(req);
		ldlm_callback_reply(req, rc);
		return 0;
	case OBD_QC_CALLBACK:
		req_capsule_set(&req->rq_pill, &RQF_QC_CALLBACK);
		if (OBD_FAIL_CHECK(OBD_FAIL_OBD_QC_CALLBACK_NET))
			return 0;
		rc = ldlm_handle_qc_callback(req);
		ldlm_callback_reply(req, rc);
		return 0;
	default:
		CERROR("unknown opcode %u\n",
		       lustre_msg_get_opc(req->rq_reqmsg));
		ldlm_callback_reply(req, -EPROTO);
		return 0;
	}

	ns = req->rq_export->exp_obd->obd_namespace;
	LASSERT(ns != NULL);

	req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);

	dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
	if (dlm_req == NULL) {
		rc = ldlm_callback_reply(req, -EPROTO);
		ldlm_callback_errmsg(req, "Operate without parameter", rc,
				     NULL);
		return 0;
	}

	/* Force a known safe race, send a cancel to the server for a lock
	 * which the server has already started a blocking callback on. */
	if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE) &&
	    lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
		rc = ldlm_cli_cancel(&dlm_req->lock_handle[0], 0);
		if (rc < 0)
			CERROR("ldlm_cli_cancel: %d\n", rc);
	}

	lock = ldlm_handle2lock_long(&dlm_req->lock_handle[0], 0);
	if (!lock) {
		CDEBUG(D_DLMTRACE, "callback on lock %#llx - lock "
		       "disappeared\n", dlm_req->lock_handle[0].cookie);
		rc = ldlm_callback_reply(req, -EINVAL);
		ldlm_callback_errmsg(req, "Operate with invalid parameter", rc,
				     &dlm_req->lock_handle[0]);
		return 0;
	}

	if ((lock->l_flags & LDLM_FL_FAIL_LOC) &&
	    lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK)
		OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);

	/* Copy hints/flags (e.g. LDLM_FL_DISCARD_DATA) from AST. */
	lock_res_and_lock(lock);
	lock->l_flags |= ldlm_flags_from_wire(dlm_req->lock_flags &
					      LDLM_AST_FLAGS);
	if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
		/* If somebody cancels lock and cache is already dropped,
		 * or lock is failed before cp_ast received on client,
		 * we can tell the server we have no lock. Otherwise, we
		 * should send cancel after dropping the cache. */
		if (((lock->l_flags & LDLM_FL_CANCELING) &&
		    (lock->l_flags & LDLM_FL_BL_DONE)) ||
		    (lock->l_flags & LDLM_FL_FAILED)) {
			LDLM_DEBUG(lock, "callback on lock "
				   "%#llx - lock disappeared\n",
				   dlm_req->lock_handle[0].cookie);
			unlock_res_and_lock(lock);
			LDLM_LOCK_RELEASE(lock);
			rc = ldlm_callback_reply(req, -EINVAL);
			ldlm_callback_errmsg(req, "Operate on stale lock", rc,
					     &dlm_req->lock_handle[0]);
			return 0;
		}
		/* BL_AST locks are not needed in LRU.
		 * Let ldlm_cancel_lru() be fast. */
		ldlm_lock_remove_from_lru(lock);
		lock->l_flags |= LDLM_FL_BL_AST;
	}
	unlock_res_and_lock(lock);

	/* We want the ost thread to get this reply so that it can respond
	 * to ost requests (write cache writeback) that might be triggered
	 * in the callback.
	 *
	 * But we'd also like to be able to indicate in the reply that we're
	 * cancelling right now, because it's unused, or have an intent result
	 * in the reply, so we might have to push the responsibility for sending
	 * the reply down into the AST handlers, alas. */

	switch (lustre_msg_get_opc(req->rq_reqmsg)) {
	case LDLM_BL_CALLBACK:
		CDEBUG(D_INODE, "blocking ast\n");
		req_capsule_extend(&req->rq_pill, &RQF_LDLM_BL_CALLBACK);
		if (!(lock->l_flags & LDLM_FL_CANCEL_ON_BLOCK)) {
			rc = ldlm_callback_reply(req, 0);
			if (req->rq_no_reply || rc)
				ldlm_callback_errmsg(req, "Normal process", rc,
						     &dlm_req->lock_handle[0]);
		}
		if (ldlm_bl_to_thread_lock(ns, &dlm_req->lock_desc, lock))
			ldlm_handle_bl_callback(ns, &dlm_req->lock_desc, lock);
		break;
	case LDLM_CP_CALLBACK:
		CDEBUG(D_INODE, "completion ast\n");
		req_capsule_extend(&req->rq_pill, &RQF_LDLM_CP_CALLBACK);
		ldlm_callback_reply(req, 0);
		ldlm_handle_cp_callback(req, ns, dlm_req, lock);
		break;
	case LDLM_GL_CALLBACK:
		CDEBUG(D_INODE, "glimpse ast\n");
		req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_CALLBACK);
		ldlm_handle_gl_callback(req, ns, dlm_req, lock);
		break;
	default:
		LBUG();			 /* checked above */
	}

	return 0;
}


static struct ldlm_bl_work_item *ldlm_bl_get_work(struct ldlm_bl_pool *blp)
{
	struct ldlm_bl_work_item *blwi = NULL;
	static unsigned int num_bl = 0;

	spin_lock(&blp->blp_lock);
	/* process a request from the blp_list at least every blp_num_threads */
	if (!list_empty(&blp->blp_list) &&
	    (list_empty(&blp->blp_prio_list) || num_bl == 0))
		blwi = list_entry(blp->blp_list.next,
				      struct ldlm_bl_work_item, blwi_entry);
	else
		if (!list_empty(&blp->blp_prio_list))
			blwi = list_entry(blp->blp_prio_list.next,
					      struct ldlm_bl_work_item,
					      blwi_entry);

	if (blwi) {
		if (++num_bl >= atomic_read(&blp->blp_num_threads))
			num_bl = 0;
		list_del(&blwi->blwi_entry);
	}
	spin_unlock(&blp->blp_lock);

	return blwi;
}

/* This only contains temporary data until the thread starts */
struct ldlm_bl_thread_data {
	char			bltd_name[CFS_CURPROC_COMM_MAX];
	struct ldlm_bl_pool	*bltd_blp;
	struct completion	bltd_comp;
	int			bltd_num;
};

static int ldlm_bl_thread_main(void *arg);

static int ldlm_bl_thread_start(struct ldlm_bl_pool *blp)
{
	struct ldlm_bl_thread_data bltd = { .bltd_blp = blp };
	struct task_struct *task;

	init_completion(&bltd.bltd_comp);
	bltd.bltd_num = atomic_read(&blp->blp_num_threads);
	snprintf(bltd.bltd_name, sizeof(bltd.bltd_name),
		"ldlm_bl_%02d", bltd.bltd_num);
	task = kthread_run(ldlm_bl_thread_main, &bltd, "%s", bltd.bltd_name);
	if (IS_ERR(task)) {
		CERROR("cannot start LDLM thread ldlm_bl_%02d: rc %ld\n",
		       atomic_read(&blp->blp_num_threads), PTR_ERR(task));
		return PTR_ERR(task);
	}
	wait_for_completion(&bltd.bltd_comp);

	return 0;
}

/**
 * Main blocking requests processing thread.
 *
 * Callers put locks into its queue by calling ldlm_bl_to_thread.
 * This thread in the end ends up doing actual call to ->l_blocking_ast
 * for queued locks.
 */
static int ldlm_bl_thread_main(void *arg)
{
	struct ldlm_bl_pool *blp;

	{
		struct ldlm_bl_thread_data *bltd = arg;

		blp = bltd->bltd_blp;

		atomic_inc(&blp->blp_num_threads);
		atomic_inc(&blp->blp_busy_threads);

		complete(&bltd->bltd_comp);
		/* cannot use bltd after this, it is only on caller's stack */
	}

	while (1) {
		struct l_wait_info lwi = { 0 };
		struct ldlm_bl_work_item *blwi = NULL;
		int busy;

		blwi = ldlm_bl_get_work(blp);

		if (blwi == NULL) {
			atomic_dec(&blp->blp_busy_threads);
			l_wait_event_exclusive(blp->blp_waitq,
					 (blwi = ldlm_bl_get_work(blp)) != NULL,
					 &lwi);
			busy = atomic_inc_return(&blp->blp_busy_threads);
		} else {
			busy = atomic_read(&blp->blp_busy_threads);
		}

		if (blwi->blwi_ns == NULL)
			/* added by ldlm_cleanup() */
			break;

		/* Not fatal if racy and have a few too many threads */
		if (unlikely(busy < blp->blp_max_threads &&
			     busy >= atomic_read(&blp->blp_num_threads) &&
			     !blwi->blwi_mem_pressure))
			/* discard the return value, we tried */
			ldlm_bl_thread_start(blp);

		if (blwi->blwi_mem_pressure)
			memory_pressure_set();

		if (blwi->blwi_count) {
			int count;
			/* The special case when we cancel locks in LRU
			 * asynchronously, we pass the list of locks here.
			 * Thus locks are marked LDLM_FL_CANCELING, but NOT
			 * canceled locally yet. */
			count = ldlm_cli_cancel_list_local(&blwi->blwi_head,
							   blwi->blwi_count,
							   LCF_BL_AST);
			ldlm_cli_cancel_list(&blwi->blwi_head, count, NULL,
					     blwi->blwi_flags);
		} else {
			ldlm_handle_bl_callback(blwi->blwi_ns, &blwi->blwi_ld,
						blwi->blwi_lock);
		}
		if (blwi->blwi_mem_pressure)
			memory_pressure_clr();

		if (blwi->blwi_flags & LCF_ASYNC)
			OBD_FREE(blwi, sizeof(*blwi));
		else
			complete(&blwi->blwi_comp);
	}

	atomic_dec(&blp->blp_busy_threads);
	atomic_dec(&blp->blp_num_threads);
	complete(&blp->blp_comp);
	return 0;
}


static int ldlm_setup(void);
static int ldlm_cleanup(void);

int ldlm_get_ref(void)
{
	int rc = 0;

	mutex_lock(&ldlm_ref_mutex);
	if (++ldlm_refcount == 1) {
		rc = ldlm_setup();
		if (rc)
			ldlm_refcount--;
	}
	mutex_unlock(&ldlm_ref_mutex);

	return rc;
}
EXPORT_SYMBOL(ldlm_get_ref);

void ldlm_put_ref(void)
{
	mutex_lock(&ldlm_ref_mutex);
	if (ldlm_refcount == 1) {
		int rc = ldlm_cleanup();
		if (rc)
			CERROR("ldlm_cleanup failed: %d\n", rc);
		else
			ldlm_refcount--;
	} else {
		ldlm_refcount--;
	}
	mutex_unlock(&ldlm_ref_mutex);
}
EXPORT_SYMBOL(ldlm_put_ref);

/*
 * Export handle<->lock hash operations.
 */
static unsigned
ldlm_export_lock_hash(struct cfs_hash *hs, const void *key, unsigned mask)
{
	return cfs_hash_u64_hash(((struct lustre_handle *)key)->cookie, mask);
}

static void *
ldlm_export_lock_key(struct hlist_node *hnode)
{
	struct ldlm_lock *lock;

	lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
	return &lock->l_remote_handle;
}

static void
ldlm_export_lock_keycpy(struct hlist_node *hnode, void *key)
{
	struct ldlm_lock     *lock;

	lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
	lock->l_remote_handle = *(struct lustre_handle *)key;
}

static int
ldlm_export_lock_keycmp(const void *key, struct hlist_node *hnode)
{
	return lustre_handle_equal(ldlm_export_lock_key(hnode), key);
}

static void *
ldlm_export_lock_object(struct hlist_node *hnode)
{
	return hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
}

static void
ldlm_export_lock_get(struct cfs_hash *hs, struct hlist_node *hnode)
{
	struct ldlm_lock *lock;

	lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
	LDLM_LOCK_GET(lock);
}

static void
ldlm_export_lock_put(struct cfs_hash *hs, struct hlist_node *hnode)
{
	struct ldlm_lock *lock;

	lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
	LDLM_LOCK_RELEASE(lock);
}

static cfs_hash_ops_t ldlm_export_lock_ops = {
	.hs_hash	= ldlm_export_lock_hash,
	.hs_key	 = ldlm_export_lock_key,
	.hs_keycmp      = ldlm_export_lock_keycmp,
	.hs_keycpy      = ldlm_export_lock_keycpy,
	.hs_object      = ldlm_export_lock_object,
	.hs_get	 = ldlm_export_lock_get,
	.hs_put	 = ldlm_export_lock_put,
	.hs_put_locked  = ldlm_export_lock_put,
};

int ldlm_init_export(struct obd_export *exp)
{
	int rc;
	exp->exp_lock_hash =
		cfs_hash_create(obd_uuid2str(&exp->exp_client_uuid),
				HASH_EXP_LOCK_CUR_BITS,
				HASH_EXP_LOCK_MAX_BITS,
				HASH_EXP_LOCK_BKT_BITS, 0,
				CFS_HASH_MIN_THETA, CFS_HASH_MAX_THETA,
				&ldlm_export_lock_ops,
				CFS_HASH_DEFAULT | CFS_HASH_REHASH_KEY |
				CFS_HASH_NBLK_CHANGE);

	if (!exp->exp_lock_hash)
		return -ENOMEM;

	rc = ldlm_init_flock_export(exp);
	if (rc)
		goto err;

	return 0;
err:
	ldlm_destroy_export(exp);
	return rc;
}
EXPORT_SYMBOL(ldlm_init_export);

void ldlm_destroy_export(struct obd_export *exp)
{
	cfs_hash_putref(exp->exp_lock_hash);
	exp->exp_lock_hash = NULL;

	ldlm_destroy_flock_export(exp);
}
EXPORT_SYMBOL(ldlm_destroy_export);

static int ldlm_setup(void)
{
	static struct ptlrpc_service_conf	conf;
	struct ldlm_bl_pool			*blp = NULL;
	int rc = 0;
	int i;

	if (ldlm_state != NULL)
		return -EALREADY;

	OBD_ALLOC(ldlm_state, sizeof(*ldlm_state));
	if (ldlm_state == NULL)
		return -ENOMEM;

	rc = ldlm_proc_setup();
	if (rc != 0)
		goto out;

	memset(&conf, 0, sizeof(conf));
	conf = (typeof(conf)) {
		.psc_name		= "ldlm_cbd",
		.psc_watchdog_factor	= 2,
		.psc_buf		= {
			.bc_nbufs		= LDLM_CLIENT_NBUFS,
			.bc_buf_size		= LDLM_BUFSIZE,
			.bc_req_max_size	= LDLM_MAXREQSIZE,
			.bc_rep_max_size	= LDLM_MAXREPSIZE,
			.bc_req_portal		= LDLM_CB_REQUEST_PORTAL,
			.bc_rep_portal		= LDLM_CB_REPLY_PORTAL,
		},
		.psc_thr		= {
			.tc_thr_name		= "ldlm_cb",
			.tc_thr_factor		= LDLM_THR_FACTOR,
			.tc_nthrs_init		= LDLM_NTHRS_INIT,
			.tc_nthrs_base		= LDLM_NTHRS_BASE,
			.tc_nthrs_max		= LDLM_NTHRS_MAX,
			.tc_nthrs_user		= ldlm_num_threads,
			.tc_cpu_affinity	= 1,
			.tc_ctx_tags		= LCT_MD_THREAD | LCT_DT_THREAD,
		},
		.psc_cpt		= {
			.cc_pattern		= ldlm_cpts,
		},
		.psc_ops		= {
			.so_req_handler		= ldlm_callback_handler,
		},
	};
	ldlm_state->ldlm_cb_service = \
			ptlrpc_register_service(&conf, ldlm_svc_proc_dir);
	if (IS_ERR(ldlm_state->ldlm_cb_service)) {
		CERROR("failed to start service\n");
		rc = PTR_ERR(ldlm_state->ldlm_cb_service);
		ldlm_state->ldlm_cb_service = NULL;
		goto out;
	}


	OBD_ALLOC(blp, sizeof(*blp));
	if (blp == NULL) {
		rc = -ENOMEM;
		goto out;
	}
	ldlm_state->ldlm_bl_pool = blp;

	spin_lock_init(&blp->blp_lock);
	INIT_LIST_HEAD(&blp->blp_list);
	INIT_LIST_HEAD(&blp->blp_prio_list);
	init_waitqueue_head(&blp->blp_waitq);
	atomic_set(&blp->blp_num_threads, 0);
	atomic_set(&blp->blp_busy_threads, 0);

	if (ldlm_num_threads == 0) {
		blp->blp_min_threads = LDLM_NTHRS_INIT;
		blp->blp_max_threads = LDLM_NTHRS_MAX;
	} else {
		blp->blp_min_threads = blp->blp_max_threads = \
			min_t(int, LDLM_NTHRS_MAX, max_t(int, LDLM_NTHRS_INIT,
							 ldlm_num_threads));
	}

	for (i = 0; i < blp->blp_min_threads; i++) {
		rc = ldlm_bl_thread_start(blp);
		if (rc < 0)
			goto out;
	}


	rc = ldlm_pools_init();
	if (rc) {
		CERROR("Failed to initialize LDLM pools: %d\n", rc);
		goto out;
	}
	return 0;

 out:
	ldlm_cleanup();
	return rc;
}

static int ldlm_cleanup(void)
{
	if (!list_empty(ldlm_namespace_list(LDLM_NAMESPACE_SERVER)) ||
	    !list_empty(ldlm_namespace_list(LDLM_NAMESPACE_CLIENT))) {
		CERROR("ldlm still has namespaces; clean these up first.\n");
		ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
		ldlm_dump_all_namespaces(LDLM_NAMESPACE_CLIENT, D_DLMTRACE);
		return -EBUSY;
	}

	ldlm_pools_fini();

	if (ldlm_state->ldlm_bl_pool != NULL) {
		struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;

		while (atomic_read(&blp->blp_num_threads) > 0) {
			struct ldlm_bl_work_item blwi = { .blwi_ns = NULL };

			init_completion(&blp->blp_comp);

			spin_lock(&blp->blp_lock);
			list_add_tail(&blwi.blwi_entry, &blp->blp_list);
			wake_up(&blp->blp_waitq);
			spin_unlock(&blp->blp_lock);

			wait_for_completion(&blp->blp_comp);
		}

		OBD_FREE(blp, sizeof(*blp));
	}

	if (ldlm_state->ldlm_cb_service != NULL)
		ptlrpc_unregister_service(ldlm_state->ldlm_cb_service);

	ldlm_proc_cleanup();


	OBD_FREE(ldlm_state, sizeof(*ldlm_state));
	ldlm_state = NULL;

	return 0;
}

int ldlm_init(void)
{
	mutex_init(&ldlm_ref_mutex);
	mutex_init(ldlm_namespace_lock(LDLM_NAMESPACE_SERVER));
	mutex_init(ldlm_namespace_lock(LDLM_NAMESPACE_CLIENT));
	ldlm_resource_slab = kmem_cache_create("ldlm_resources",
					       sizeof(struct ldlm_resource), 0,
					       SLAB_HWCACHE_ALIGN, NULL);
	if (ldlm_resource_slab == NULL)
		return -ENOMEM;

	ldlm_lock_slab = kmem_cache_create("ldlm_locks",
			      sizeof(struct ldlm_lock), 0,
			      SLAB_HWCACHE_ALIGN | SLAB_DESTROY_BY_RCU, NULL);
	if (ldlm_lock_slab == NULL) {
		kmem_cache_destroy(ldlm_resource_slab);
		return -ENOMEM;
	}

	ldlm_interval_slab = kmem_cache_create("interval_node",
					sizeof(struct ldlm_interval),
					0, SLAB_HWCACHE_ALIGN, NULL);
	if (ldlm_interval_slab == NULL) {
		kmem_cache_destroy(ldlm_resource_slab);
		kmem_cache_destroy(ldlm_lock_slab);
		return -ENOMEM;
	}
#if LUSTRE_TRACKS_LOCK_EXP_REFS
	class_export_dump_hook = ldlm_dump_export_locks;
#endif
	return 0;
}

void ldlm_exit(void)
{
	if (ldlm_refcount)
		CERROR("ldlm_refcount is %d in ldlm_exit!\n", ldlm_refcount);
	kmem_cache_destroy(ldlm_resource_slab);
	/* ldlm_lock_put() use RCU to call ldlm_lock_free, so need call
	 * synchronize_rcu() to wait a grace period elapsed, so that
	 * ldlm_lock_free() get a chance to be called. */
	synchronize_rcu();
	kmem_cache_destroy(ldlm_lock_slab);
	kmem_cache_destroy(ldlm_interval_slab);
}