[deliverable/linux.git] / fs / gfs2 / locking / dlm / lock.c

/*
 * Copyright (C) Sistina Software, Inc.  1997-2003 All rights reserved.
 * Copyright (C) 2004-2005 Red Hat, Inc.  All rights reserved.
 *
 * This copyrighted material is made available to anyone wishing to use,
 * modify, copy, or redistribute it subject to the terms and conditions
 * of the GNU General Public License version 2.
 */

#include "lock_dlm.h"

static char junk_lvb[GDLM_LVB_SIZE];

static void queue_complete(struct gdlm_lock *lp)
{
	struct gdlm_ls *ls = lp->ls;

	clear_bit(LFL_ACTIVE, &lp->flags);

	spin_lock(&ls->async_lock);
	list_add_tail(&lp->clist, &ls->complete);
	spin_unlock(&ls->async_lock);
	wake_up(&ls->thread_wait);
}

static inline void gdlm_ast(void *astarg)
{
	queue_complete(astarg);
}

static inline void gdlm_bast(void *astarg, int mode)
{
	struct gdlm_lock *lp = astarg;
	struct gdlm_ls *ls = lp->ls;

	if (!mode) {
		printk(KERN_INFO "lock_dlm: bast mode zero %x,%llx\n",
			lp->lockname.ln_type,
			(unsigned long long)lp->lockname.ln_number);
		return;
	}

	spin_lock(&ls->async_lock);
	if (!lp->bast_mode) {
		list_add_tail(&lp->blist, &ls->blocking);
		lp->bast_mode = mode;
	} else if (lp->bast_mode < mode)
		lp->bast_mode = mode;
	spin_unlock(&ls->async_lock);
	wake_up(&ls->thread_wait);
}

void gdlm_queue_delayed(struct gdlm_lock *lp)
{
	struct gdlm_ls *ls = lp->ls;

	spin_lock(&ls->async_lock);
	list_add_tail(&lp->delay_list, &ls->delayed);
	spin_unlock(&ls->async_lock);
}

/* convert gfs lock-state to dlm lock-mode */

static int16_t make_mode(int16_t lmstate)
{
	switch (lmstate) {
	case LM_ST_UNLOCKED:
		return DLM_LOCK_NL;
	case LM_ST_EXCLUSIVE:
		return DLM_LOCK_EX;
	case LM_ST_DEFERRED:
		return DLM_LOCK_CW;
	case LM_ST_SHARED:
		return DLM_LOCK_PR;
	}
	gdlm_assert(0, "unknown LM state %d", lmstate);
	return -1;
}

/* convert dlm lock-mode to gfs lock-state */

int16_t gdlm_make_lmstate(int16_t dlmmode)
{
	switch (dlmmode) {
	case DLM_LOCK_IV:
	case DLM_LOCK_NL:
		return LM_ST_UNLOCKED;
	case DLM_LOCK_EX:
		return LM_ST_EXCLUSIVE;
	case DLM_LOCK_CW:
		return LM_ST_DEFERRED;
	case DLM_LOCK_PR:
		return LM_ST_SHARED;
	}
	gdlm_assert(0, "unknown DLM mode %d", dlmmode);
	return -1;
}

/* verify agreement with GFS on the current lock state, NB: DLM_LOCK_NL and
   DLM_LOCK_IV are both considered LM_ST_UNLOCKED by GFS. */

static void check_cur_state(struct gdlm_lock *lp, unsigned int cur_state)
{
	int16_t cur = make_mode(cur_state);
	if (lp->cur != DLM_LOCK_IV)
		gdlm_assert(lp->cur == cur, "%d, %d", lp->cur, cur);
}

static inline unsigned int make_flags(struct gdlm_lock *lp,
				      unsigned int gfs_flags,
				      int16_t cur, int16_t req)
{
	unsigned int lkf = 0;

	if (gfs_flags & LM_FLAG_TRY)
		lkf |= DLM_LKF_NOQUEUE;

	if (gfs_flags & LM_FLAG_TRY_1CB) {
		lkf |= DLM_LKF_NOQUEUE;
		lkf |= DLM_LKF_NOQUEUEBAST;
	}

	if (gfs_flags & LM_FLAG_PRIORITY) {
		lkf |= DLM_LKF_NOORDER;
		lkf |= DLM_LKF_HEADQUE;
	}

	if (gfs_flags & LM_FLAG_ANY) {
		if (req == DLM_LOCK_PR)
			lkf |= DLM_LKF_ALTCW;
		else if (req == DLM_LOCK_CW)
			lkf |= DLM_LKF_ALTPR;
	}

	if (lp->lksb.sb_lkid != 0) {
		lkf |= DLM_LKF_CONVERT;

		/* Conversion deadlock avoidance by DLM */

		if (!test_bit(LFL_FORCE_PROMOTE, &lp->flags) &&
		    !(lkf & DLM_LKF_NOQUEUE) &&
		    cur > DLM_LOCK_NL && req > DLM_LOCK_NL && cur != req)
			lkf |= DLM_LKF_CONVDEADLK;
	}

	if (lp->lvb)
		lkf |= DLM_LKF_VALBLK;

	return lkf;
}

/* make_strname - convert GFS lock numbers to a string */

static inline void make_strname(struct lm_lockname *lockname,
				struct gdlm_strname *str)
{
	sprintf(str->name, "%8x%16llx", lockname->ln_type,
		(unsigned long long)lockname->ln_number);
	str->namelen = GDLM_STRNAME_BYTES;
}

static int gdlm_create_lp(struct gdlm_ls *ls, struct lm_lockname *name,
			  struct gdlm_lock **lpp)
{
	struct gdlm_lock *lp;

	lp = kzalloc(sizeof(struct gdlm_lock), GFP_KERNEL);
	if (!lp)
		return -ENOMEM;

	lp->lockname = *name;
	lp->ls = ls;
	lp->cur = DLM_LOCK_IV;
	lp->lvb = NULL;
	lp->hold_null = NULL;
	init_completion(&lp->ast_wait);
	INIT_LIST_HEAD(&lp->clist);
	INIT_LIST_HEAD(&lp->blist);
	INIT_LIST_HEAD(&lp->delay_list);

	spin_lock(&ls->async_lock);
	list_add(&lp->all_list, &ls->all_locks);
	ls->all_locks_count++;
	spin_unlock(&ls->async_lock);

	*lpp = lp;
	return 0;
}

void gdlm_delete_lp(struct gdlm_lock *lp)
{
	struct gdlm_ls *ls = lp->ls;

	spin_lock(&ls->async_lock);
	if (!list_empty(&lp->clist))
		list_del_init(&lp->clist);
	if (!list_empty(&lp->blist))
		list_del_init(&lp->blist);
	if (!list_empty(&lp->delay_list))
		list_del_init(&lp->delay_list);
	gdlm_assert(!list_empty(&lp->all_list), "%x,%llx", lp->lockname.ln_type,
		    (unsigned long long)lp->lockname.ln_number);
	list_del_init(&lp->all_list);
	ls->all_locks_count--;
	spin_unlock(&ls->async_lock);

	kfree(lp);
}

int gdlm_get_lock(lm_lockspace_t *lockspace, struct lm_lockname *name,
		  lm_lock_t **lockp)
{
	struct gdlm_lock *lp;
	int error;

	error = gdlm_create_lp((struct gdlm_ls *) lockspace, name, &lp);

	*lockp = (lm_lock_t *) lp;
	return error;
}

void gdlm_put_lock(lm_lock_t *lock)
{
	gdlm_delete_lp((struct gdlm_lock *) lock);
}

unsigned int gdlm_do_lock(struct gdlm_lock *lp)
{
	struct gdlm_ls *ls = lp->ls;
	struct gdlm_strname str;
	int error, bast = 1;

	/*
	 * When recovery is in progress, delay lock requests for submission
	 * once recovery is done.  Requests for recovery (NOEXP) and unlocks
	 * can pass.
	 */

	if (test_bit(DFL_BLOCK_LOCKS, &ls->flags) &&
	    !test_bit(LFL_NOBLOCK, &lp->flags) && lp->req != DLM_LOCK_NL) {
		gdlm_queue_delayed(lp);
		return LM_OUT_ASYNC;
	}

	/*
	 * Submit the actual lock request.
	 */

	if (test_bit(LFL_NOBAST, &lp->flags))
		bast = 0;

	make_strname(&lp->lockname, &str);

	set_bit(LFL_ACTIVE, &lp->flags);

	log_debug("lk %x,%llx id %x %d,%d %x", lp->lockname.ln_type,
		  (unsigned long long)lp->lockname.ln_number, lp->lksb.sb_lkid,
		  lp->cur, lp->req, lp->lkf);

	error = dlm_lock(ls->dlm_lockspace, lp->req, &lp->lksb, lp->lkf,
			 str.name, str.namelen, 0, gdlm_ast, lp,
			 bast ? gdlm_bast : NULL);

	if ((error == -EAGAIN) && (lp->lkf & DLM_LKF_NOQUEUE)) {
		lp->lksb.sb_status = -EAGAIN;
		queue_complete(lp);
		error = 0;
	}

	if (error) {
		log_debug("%s: gdlm_lock %x,%llx err=%d cur=%d req=%d lkf=%x "
			  "flags=%lx", ls->fsname, lp->lockname.ln_type,
			  (unsigned long long)lp->lockname.ln_number, error,
			  lp->cur, lp->req, lp->lkf, lp->flags);
		return LM_OUT_ERROR;
	}
	return LM_OUT_ASYNC;
}

static unsigned int gdlm_do_unlock(struct gdlm_lock *lp)
{
	struct gdlm_ls *ls = lp->ls;
	unsigned int lkf = 0;
	int error;

	set_bit(LFL_DLM_UNLOCK, &lp->flags);
	set_bit(LFL_ACTIVE, &lp->flags);

	if (lp->lvb)
		lkf = DLM_LKF_VALBLK;

	log_debug("un %x,%llx %x %d %x", lp->lockname.ln_type,
		  (unsigned long long)lp->lockname.ln_number,
		  lp->lksb.sb_lkid, lp->cur, lkf);

	error = dlm_unlock(ls->dlm_lockspace, lp->lksb.sb_lkid, lkf, NULL, lp);

	if (error) {
		log_debug("%s: gdlm_unlock %x,%llx err=%d cur=%d req=%d lkf=%x "
			  "flags=%lx", ls->fsname, lp->lockname.ln_type,
			  (unsigned long long)lp->lockname.ln_number, error,
			  lp->cur, lp->req, lp->lkf, lp->flags);
		return LM_OUT_ERROR;
	}
	return LM_OUT_ASYNC;
}

unsigned int gdlm_lock(lm_lock_t *lock, unsigned int cur_state,
		       unsigned int req_state, unsigned int flags)
{
	struct gdlm_lock *lp = (struct gdlm_lock *) lock;

	clear_bit(LFL_DLM_CANCEL, &lp->flags);
	if (flags & LM_FLAG_NOEXP)
		set_bit(LFL_NOBLOCK, &lp->flags);

	check_cur_state(lp, cur_state);
	lp->req = make_mode(req_state);
	lp->lkf = make_flags(lp, flags, lp->cur, lp->req);

	return gdlm_do_lock(lp);
}

unsigned int gdlm_unlock(lm_lock_t *lock, unsigned int cur_state)
{
	struct gdlm_lock *lp = (struct gdlm_lock *) lock;

	clear_bit(LFL_DLM_CANCEL, &lp->flags);
	if (lp->cur == DLM_LOCK_IV)
		return 0;
	return gdlm_do_unlock(lp);
}

void gdlm_cancel(lm_lock_t *lock)
{
	struct gdlm_lock *lp = (struct gdlm_lock *) lock;
	struct gdlm_ls *ls = lp->ls;
	int error, delay_list = 0;

	if (test_bit(LFL_DLM_CANCEL, &lp->flags))
		return;

	log_info("gdlm_cancel %x,%llx flags %lx", lp->lockname.ln_type,
		 (unsigned long long)lp->lockname.ln_number, lp->flags);

	spin_lock(&ls->async_lock);
	if (!list_empty(&lp->delay_list)) {
		list_del_init(&lp->delay_list);
		delay_list = 1;
	}
	spin_unlock(&ls->async_lock);

	if (delay_list) {
		set_bit(LFL_CANCEL, &lp->flags);
		set_bit(LFL_ACTIVE, &lp->flags);
		queue_complete(lp);
		return;
	}

	if (!test_bit(LFL_ACTIVE, &lp->flags) ||
	    test_bit(LFL_DLM_UNLOCK, &lp->flags)) {
		log_info("gdlm_cancel skip %x,%llx flags %lx",
		 	 lp->lockname.ln_type,
			 (unsigned long long)lp->lockname.ln_number, lp->flags);
		return;
	}

	/* the lock is blocked in the dlm */

	set_bit(LFL_DLM_CANCEL, &lp->flags);
	set_bit(LFL_ACTIVE, &lp->flags);

	error = dlm_unlock(ls->dlm_lockspace, lp->lksb.sb_lkid, DLM_LKF_CANCEL,
			   NULL, lp);

	log_info("gdlm_cancel rv %d %x,%llx flags %lx", error,
		 lp->lockname.ln_type,
		 (unsigned long long)lp->lockname.ln_number, lp->flags);

	if (error == -EBUSY)
		clear_bit(LFL_DLM_CANCEL, &lp->flags);
}

static int gdlm_add_lvb(struct gdlm_lock *lp)
{
	char *lvb;

	lvb = kzalloc(GDLM_LVB_SIZE, GFP_KERNEL);
	if (!lvb)
		return -ENOMEM;

	lp->lksb.sb_lvbptr = lvb;
	lp->lvb = lvb;
	return 0;
}

static void gdlm_del_lvb(struct gdlm_lock *lp)
{
	kfree(lp->lvb);
	lp->lvb = NULL;
	lp->lksb.sb_lvbptr = NULL;
}

/* This can do a synchronous dlm request (requiring a lock_dlm thread to get
   the completion) because gfs won't call hold_lvb() during a callback (from
   the context of a lock_dlm thread). */

static int hold_null_lock(struct gdlm_lock *lp)
{
	struct gdlm_lock *lpn = NULL;
	int error;

	if (lp->hold_null) {
		printk(KERN_INFO "lock_dlm: lvb already held\n");
		return 0;
	}

	error = gdlm_create_lp(lp->ls, &lp->lockname, &lpn);
	if (error)
		goto out;

	lpn->lksb.sb_lvbptr = junk_lvb;
	lpn->lvb = junk_lvb;

	lpn->req = DLM_LOCK_NL;
	lpn->lkf = DLM_LKF_VALBLK | DLM_LKF_EXPEDITE;
	set_bit(LFL_NOBAST, &lpn->flags);
	set_bit(LFL_INLOCK, &lpn->flags);

	init_completion(&lpn->ast_wait);
	gdlm_do_lock(lpn);
	wait_for_completion(&lpn->ast_wait);
	error = lpn->lksb.sb_status;
	if (error) {
		printk(KERN_INFO "lock_dlm: hold_null_lock dlm error %d\n",
		       error);
		gdlm_delete_lp(lpn);
		lpn = NULL;
	}
 out:
	lp->hold_null = lpn;
	return error;
}

/* This cannot do a synchronous dlm request (requiring a lock_dlm thread to get
   the completion) because gfs may call unhold_lvb() during a callback (from
   the context of a lock_dlm thread) which could cause a deadlock since the
   other lock_dlm thread could be engaged in recovery. */

static void unhold_null_lock(struct gdlm_lock *lp)
{
	struct gdlm_lock *lpn = lp->hold_null;

	gdlm_assert(lpn, "%x,%llx", lp->lockname.ln_type,
		    (unsigned long long)lp->lockname.ln_number);
	lpn->lksb.sb_lvbptr = NULL;
	lpn->lvb = NULL;
	set_bit(LFL_UNLOCK_DELETE, &lpn->flags);
	gdlm_do_unlock(lpn);
	lp->hold_null = NULL;
}

/* Acquire a NL lock because gfs requires the value block to remain
   intact on the resource while the lvb is "held" even if it's holding no locks
   on the resource. */

int gdlm_hold_lvb(lm_lock_t *lock, char **lvbp)
{
	struct gdlm_lock *lp = (struct gdlm_lock *) lock;
	int error;

	error = gdlm_add_lvb(lp);
	if (error)
		return error;

	*lvbp = lp->lvb;

	error = hold_null_lock(lp);
	if (error)
		gdlm_del_lvb(lp);

	return error;
}

void gdlm_unhold_lvb(lm_lock_t *lock, char *lvb)
{
	struct gdlm_lock *lp = (struct gdlm_lock *) lock;

	unhold_null_lock(lp);
	gdlm_del_lvb(lp);
}

void gdlm_sync_lvb(lm_lock_t *lock, char *lvb)
{
	struct gdlm_lock *lp = (struct gdlm_lock *) lock;

	if (lp->cur != DLM_LOCK_EX)
		return;

	init_completion(&lp->ast_wait);
	set_bit(LFL_SYNC_LVB, &lp->flags);

	lp->req = DLM_LOCK_EX;
	lp->lkf = make_flags(lp, 0, lp->cur, lp->req);

	gdlm_do_lock(lp);
	wait_for_completion(&lp->ast_wait);
}

void gdlm_submit_delayed(struct gdlm_ls *ls)
{
	struct gdlm_lock *lp, *safe;

	spin_lock(&ls->async_lock);
	list_for_each_entry_safe(lp, safe, &ls->delayed, delay_list) {
		list_del_init(&lp->delay_list);
		list_add_tail(&lp->delay_list, &ls->submit);
	}
	spin_unlock(&ls->async_lock);
	wake_up(&ls->thread_wait);
}

int gdlm_release_all_locks(struct gdlm_ls *ls)
{
	struct gdlm_lock *lp, *safe;
	int count = 0;

	spin_lock(&ls->async_lock);
	list_for_each_entry_safe(lp, safe, &ls->all_locks, all_list) {
		list_del_init(&lp->all_list);

		if (lp->lvb && lp->lvb != junk_lvb)
			kfree(lp->lvb);
		kfree(lp);
		count++;
	}
	spin_unlock(&ls->async_lock);

	return count;
}
Commit	Line	Data
869d81df DT	1	/*
	2	* Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
	3	* Copyright (C) 2004-2005 Red Hat, Inc. All rights reserved.
	4	*
	5	* This copyrighted material is made available to anyone wishing to use,
	6	* modify, copy, or redistribute it subject to the terms and conditions
e9fc2aa0	7	* of the GNU General Public License version 2.
869d81df	8	*/
29b7998d DT	9
	10	#include "lock_dlm.h"
	11
	12	static char junk_lvb[GDLM_LVB_SIZE];
	13
	14	static void queue_complete(struct gdlm_lock *lp)
	15	{
	16	struct gdlm_ls *ls = lp->ls;
	17
	18	clear_bit(LFL_ACTIVE, &lp->flags);
	19
	20	spin_lock(&ls->async_lock);
	21	list_add_tail(&lp->clist, &ls->complete);
	22	spin_unlock(&ls->async_lock);
	23	wake_up(&ls->thread_wait);
	24	}
	25
	26	static inline void gdlm_ast(void *astarg)
	27	{
e7f5c01c	28	queue_complete(astarg);
29b7998d DT	29	}
	30
	31	static inline void gdlm_bast(void *astarg, int mode)
	32	{
	33	struct gdlm_lock *lp = astarg;
	34	struct gdlm_ls *ls = lp->ls;
	35
	36	if (!mode) {
d92a8d48	37	printk(KERN_INFO "lock_dlm: bast mode zero %x,%llx\n",
9229f013 DT	38	lp->lockname.ln_type,
9229f013 DT	39	(unsigned long long)lp->lockname.ln_number);
29b7998d DT	40	return;
	41	}
	42
	43	spin_lock(&ls->async_lock);
	44	if (!lp->bast_mode) {
	45	list_add_tail(&lp->blist, &ls->blocking);
	46	lp->bast_mode = mode;
	47	} else if (lp->bast_mode < mode)
	48	lp->bast_mode = mode;
	49	spin_unlock(&ls->async_lock);
	50	wake_up(&ls->thread_wait);
	51	}
	52
	53	void gdlm_queue_delayed(struct gdlm_lock *lp)
	54	{
	55	struct gdlm_ls *ls = lp->ls;
	56
	57	spin_lock(&ls->async_lock);
	58	list_add_tail(&lp->delay_list, &ls->delayed);
	59	spin_unlock(&ls->async_lock);
	60	}
	61
	62	/* convert gfs lock-state to dlm lock-mode */
	63
	64	static int16_t make_mode(int16_t lmstate)
	65	{
	66	switch (lmstate) {
	67	case LM_ST_UNLOCKED:
	68	return DLM_LOCK_NL;
	69	case LM_ST_EXCLUSIVE:
	70	return DLM_LOCK_EX;
	71	case LM_ST_DEFERRED:
	72	return DLM_LOCK_CW;
	73	case LM_ST_SHARED:
	74	return DLM_LOCK_PR;
29b7998d	75	}
869d81df DT	76	gdlm_assert(0, "unknown LM state %d", lmstate);
869d81df DT	77	return -1;
29b7998d DT	78	}
	79
	80	/* convert dlm lock-mode to gfs lock-state */
	81
	82	int16_t gdlm_make_lmstate(int16_t dlmmode)
	83	{
	84	switch (dlmmode) {
	85	case DLM_LOCK_IV:
	86	case DLM_LOCK_NL:
	87	return LM_ST_UNLOCKED;
	88	case DLM_LOCK_EX:
	89	return LM_ST_EXCLUSIVE;
	90	case DLM_LOCK_CW:
	91	return LM_ST_DEFERRED;
	92	case DLM_LOCK_PR:
	93	return LM_ST_SHARED;
29b7998d	94	}
869d81df DT	95	gdlm_assert(0, "unknown DLM mode %d", dlmmode);
869d81df DT	96	return -1;
29b7998d DT	97	}
	98
	99	/* verify agreement with GFS on the current lock state, NB: DLM_LOCK_NL and
	100	DLM_LOCK_IV are both considered LM_ST_UNLOCKED by GFS. */
	101
	102	static void check_cur_state(struct gdlm_lock *lp, unsigned int cur_state)
	103	{
	104	int16_t cur = make_mode(cur_state);
	105	if (lp->cur != DLM_LOCK_IV)
869d81df	106	gdlm_assert(lp->cur == cur, "%d, %d", lp->cur, cur);
29b7998d DT	107	}
	108
	109	static inline unsigned int make_flags(struct gdlm_lock *lp,
	110	unsigned int gfs_flags,
	111	int16_t cur, int16_t req)
	112	{
	113	unsigned int lkf = 0;
	114
	115	if (gfs_flags & LM_FLAG_TRY)
	116	lkf \|= DLM_LKF_NOQUEUE;
	117
	118	if (gfs_flags & LM_FLAG_TRY_1CB) {
	119	lkf \|= DLM_LKF_NOQUEUE;
	120	lkf \|= DLM_LKF_NOQUEUEBAST;
	121	}
	122
	123	if (gfs_flags & LM_FLAG_PRIORITY) {
	124	lkf \|= DLM_LKF_NOORDER;
	125	lkf \|= DLM_LKF_HEADQUE;
	126	}
	127
	128	if (gfs_flags & LM_FLAG_ANY) {
	129	if (req == DLM_LOCK_PR)
	130	lkf \|= DLM_LKF_ALTCW;
	131	else if (req == DLM_LOCK_CW)
	132	lkf \|= DLM_LKF_ALTPR;
	133	}
	134
	135	if (lp->lksb.sb_lkid != 0) {
	136	lkf \|= DLM_LKF_CONVERT;
	137
	138	/* Conversion deadlock avoidance by DLM */
	139
	140	if (!test_bit(LFL_FORCE_PROMOTE, &lp->flags) &&
	141	!(lkf & DLM_LKF_NOQUEUE) &&
	142	cur > DLM_LOCK_NL && req > DLM_LOCK_NL && cur != req)
	143	lkf \|= DLM_LKF_CONVDEADLK;
	144	}
	145
	146	if (lp->lvb)
	147	lkf \|= DLM_LKF_VALBLK;
	148
	149	return lkf;
	150	}
	151
	152	/* make_strname - convert GFS lock numbers to a string */
	153
	154	static inline void make_strname(struct lm_lockname *lockname,
	155	struct gdlm_strname *str)
	156	{
869d81df	157	sprintf(str->name, "%8x%16llx", lockname->ln_type,
9229f013	158	(unsigned long long)lockname->ln_number);
29b7998d DT	159	str->namelen = GDLM_STRNAME_BYTES;
	160	}
	161
08bc2dbc AB	162	static int gdlm_create_lp(struct gdlm_ls ls, struct lm_lockname name,
08bc2dbc AB	163	struct gdlm_lock **lpp)
29b7998d DT	164	{
	165	struct gdlm_lock *lp;
	166
869d81df	167	lp = kzalloc(sizeof(struct gdlm_lock), GFP_KERNEL);
29b7998d DT	168	if (!lp)
	169	return -ENOMEM;
	170
29b7998d DT	171	lp->lockname = *name;
	172	lp->ls = ls;
	173	lp->cur = DLM_LOCK_IV;
	174	lp->lvb = NULL;
	175	lp->hold_null = NULL;
	176	init_completion(&lp->ast_wait);
	177	INIT_LIST_HEAD(&lp->clist);
	178	INIT_LIST_HEAD(&lp->blist);
	179	INIT_LIST_HEAD(&lp->delay_list);
	180
	181	spin_lock(&ls->async_lock);
	182	list_add(&lp->all_list, &ls->all_locks);
	183	ls->all_locks_count++;
	184	spin_unlock(&ls->async_lock);
	185
	186	*lpp = lp;
	187	return 0;
	188	}
	189
	190	void gdlm_delete_lp(struct gdlm_lock *lp)
	191	{
	192	struct gdlm_ls *ls = lp->ls;
	193
	194	spin_lock(&ls->async_lock);
	195	if (!list_empty(&lp->clist))
	196	list_del_init(&lp->clist);
	197	if (!list_empty(&lp->blist))
	198	list_del_init(&lp->blist);
	199	if (!list_empty(&lp->delay_list))
	200	list_del_init(&lp->delay_list);
9229f013 DT	201	gdlm_assert(!list_empty(&lp->all_list), "%x,%llx", lp->lockname.ln_type,
9229f013 DT	202	(unsigned long long)lp->lockname.ln_number);
29b7998d DT	203	list_del_init(&lp->all_list);
	204	ls->all_locks_count--;
	205	spin_unlock(&ls->async_lock);
	206
	207	kfree(lp);
	208	}
	209
	210	int gdlm_get_lock(lm_lockspace_t lockspace, struct lm_lockname name,
	211	lm_lock_t **lockp)
	212	{
	213	struct gdlm_lock *lp;
	214	int error;
	215
	216	error = gdlm_create_lp((struct gdlm_ls *) lockspace, name, &lp);
	217
	218	lockp = (lm_lock_t ) lp;
	219	return error;
	220	}
	221
	222	void gdlm_put_lock(lm_lock_t *lock)
	223	{
	224	gdlm_delete_lp((struct gdlm_lock *) lock);
	225	}
	226
8d3b35a4	227	unsigned int gdlm_do_lock(struct gdlm_lock *lp)
29b7998d DT	228	{
	229	struct gdlm_ls *ls = lp->ls;
	230	struct gdlm_strname str;
	231	int error, bast = 1;
	232
	233	/*
	234	* When recovery is in progress, delay lock requests for submission
	235	* once recovery is done. Requests for recovery (NOEXP) and unlocks
	236	* can pass.
	237	*/
	238
	239	if (test_bit(DFL_BLOCK_LOCKS, &ls->flags) &&
	240	!test_bit(LFL_NOBLOCK, &lp->flags) && lp->req != DLM_LOCK_NL) {
	241	gdlm_queue_delayed(lp);
869d81df	242	return LM_OUT_ASYNC;
29b7998d DT	243	}
	244
	245	/*
	246	* Submit the actual lock request.
	247	*/
	248
	249	if (test_bit(LFL_NOBAST, &lp->flags))
	250	bast = 0;
	251
	252	make_strname(&lp->lockname, &str);
	253
	254	set_bit(LFL_ACTIVE, &lp->flags);
	255
869d81df	256	log_debug("lk %x,%llx id %x %d,%d %x", lp->lockname.ln_type,
9229f013	257	(unsigned long long)lp->lockname.ln_number, lp->lksb.sb_lkid,
29b7998d DT	258	lp->cur, lp->req, lp->lkf);
	259
	260	error = dlm_lock(ls->dlm_lockspace, lp->req, &lp->lksb, lp->lkf,
e7f5c01c	261	str.name, str.namelen, 0, gdlm_ast, lp,
8d3b35a4	262	bast ? gdlm_bast : NULL);
29b7998d DT	263
	264	if ((error == -EAGAIN) && (lp->lkf & DLM_LKF_NOQUEUE)) {
	265	lp->lksb.sb_status = -EAGAIN;
	266	queue_complete(lp);
	267	error = 0;
	268	}
	269
869d81df DT	270	if (error) {
	271	log_debug("%s: gdlm_lock %x,%llx err=%d cur=%d req=%d lkf=%x "
	272	"flags=%lx", ls->fsname, lp->lockname.ln_type,
9229f013 DT	273	(unsigned long long)lp->lockname.ln_number, error,
9229f013 DT	274	lp->cur, lp->req, lp->lkf, lp->flags);
869d81df DT	275	return LM_OUT_ERROR;
	276	}
	277	return LM_OUT_ASYNC;
29b7998d DT	278	}
29b7998d DT	279
08bc2dbc	280	static unsigned int gdlm_do_unlock(struct gdlm_lock *lp)
29b7998d	281	{
869d81df	282	struct gdlm_ls *ls = lp->ls;
29b7998d DT	283	unsigned int lkf = 0;
	284	int error;
	285
	286	set_bit(LFL_DLM_UNLOCK, &lp->flags);
	287	set_bit(LFL_ACTIVE, &lp->flags);
	288
	289	if (lp->lvb)
	290	lkf = DLM_LKF_VALBLK;
	291
869d81df	292	log_debug("un %x,%llx %x %d %x", lp->lockname.ln_type,
9229f013 DT	293	(unsigned long long)lp->lockname.ln_number,
9229f013 DT	294	lp->lksb.sb_lkid, lp->cur, lkf);
29b7998d	295
869d81df	296	error = dlm_unlock(ls->dlm_lockspace, lp->lksb.sb_lkid, lkf, NULL, lp);
29b7998d	297
869d81df DT	298	if (error) {
	299	log_debug("%s: gdlm_unlock %x,%llx err=%d cur=%d req=%d lkf=%x "
	300	"flags=%lx", ls->fsname, lp->lockname.ln_type,
9229f013 DT	301	(unsigned long long)lp->lockname.ln_number, error,
9229f013 DT	302	lp->cur, lp->req, lp->lkf, lp->flags);
869d81df DT	303	return LM_OUT_ERROR;
	304	}
	305	return LM_OUT_ASYNC;
29b7998d DT	306	}
	307
	308	unsigned int gdlm_lock(lm_lock_t *lock, unsigned int cur_state,
	309	unsigned int req_state, unsigned int flags)
	310	{
	311	struct gdlm_lock lp = (struct gdlm_lock ) lock;
	312
	313	clear_bit(LFL_DLM_CANCEL, &lp->flags);
	314	if (flags & LM_FLAG_NOEXP)
	315	set_bit(LFL_NOBLOCK, &lp->flags);
	316
	317	check_cur_state(lp, cur_state);
	318	lp->req = make_mode(req_state);
	319	lp->lkf = make_flags(lp, flags, lp->cur, lp->req);
	320
8d3b35a4	321	return gdlm_do_lock(lp);
29b7998d DT	322	}
	323
	324	unsigned int gdlm_unlock(lm_lock_t *lock, unsigned int cur_state)
	325	{
	326	struct gdlm_lock lp = (struct gdlm_lock ) lock;
	327
	328	clear_bit(LFL_DLM_CANCEL, &lp->flags);
	329	if (lp->cur == DLM_LOCK_IV)
	330	return 0;
869d81df	331	return gdlm_do_unlock(lp);
29b7998d DT	332	}
	333
	334	void gdlm_cancel(lm_lock_t *lock)
	335	{
	336	struct gdlm_lock lp = (struct gdlm_lock ) lock;
	337	struct gdlm_ls *ls = lp->ls;
	338	int error, delay_list = 0;
	339
	340	if (test_bit(LFL_DLM_CANCEL, &lp->flags))
	341	return;
	342
9229f013 DT	343	log_info("gdlm_cancel %x,%llx flags %lx", lp->lockname.ln_type,
9229f013 DT	344	(unsigned long long)lp->lockname.ln_number, lp->flags);
29b7998d DT	345
	346	spin_lock(&ls->async_lock);
	347	if (!list_empty(&lp->delay_list)) {
	348	list_del_init(&lp->delay_list);
	349	delay_list = 1;
	350	}
	351	spin_unlock(&ls->async_lock);
	352
	353	if (delay_list) {
	354	set_bit(LFL_CANCEL, &lp->flags);
	355	set_bit(LFL_ACTIVE, &lp->flags);
	356	queue_complete(lp);
	357	return;
	358	}
	359
	360	if (!test_bit(LFL_ACTIVE, &lp->flags) \|\|
9229f013	361	test_bit(LFL_DLM_UNLOCK, &lp->flags)) {
869d81df	362	log_info("gdlm_cancel skip %x,%llx flags %lx",
9229f013 DT	363	lp->lockname.ln_type,
9229f013 DT	364	(unsigned long long)lp->lockname.ln_number, lp->flags);
29b7998d DT	365	return;
	366	}
	367
	368	/* the lock is blocked in the dlm */
	369
	370	set_bit(LFL_DLM_CANCEL, &lp->flags);
	371	set_bit(LFL_ACTIVE, &lp->flags);
	372
	373	error = dlm_unlock(ls->dlm_lockspace, lp->lksb.sb_lkid, DLM_LKF_CANCEL,
	374	NULL, lp);
	375
869d81df	376	log_info("gdlm_cancel rv %d %x,%llx flags %lx", error,
9229f013 DT	377	lp->lockname.ln_type,
9229f013 DT	378	(unsigned long long)lp->lockname.ln_number, lp->flags);
29b7998d DT	379
	380	if (error == -EBUSY)
	381	clear_bit(LFL_DLM_CANCEL, &lp->flags);
	382	}
	383
08bc2dbc	384	static int gdlm_add_lvb(struct gdlm_lock *lp)
29b7998d DT	385	{
	386	char *lvb;
	387
869d81df	388	lvb = kzalloc(GDLM_LVB_SIZE, GFP_KERNEL);
29b7998d DT	389	if (!lvb)
	390	return -ENOMEM;
	391
29b7998d DT	392	lp->lksb.sb_lvbptr = lvb;
	393	lp->lvb = lvb;
	394	return 0;
	395	}
	396
08bc2dbc	397	static void gdlm_del_lvb(struct gdlm_lock *lp)
29b7998d DT	398	{
	399	kfree(lp->lvb);
	400	lp->lvb = NULL;
	401	lp->lksb.sb_lvbptr = NULL;
	402	}
	403
	404	/* This can do a synchronous dlm request (requiring a lock_dlm thread to get
	405	the completion) because gfs won't call hold_lvb() during a callback (from
	406	the context of a lock_dlm thread). */
	407
	408	static int hold_null_lock(struct gdlm_lock *lp)
	409	{
	410	struct gdlm_lock *lpn = NULL;
	411	int error;
	412
	413	if (lp->hold_null) {
d92a8d48	414	printk(KERN_INFO "lock_dlm: lvb already held\n");
29b7998d DT	415	return 0;
	416	}
	417
	418	error = gdlm_create_lp(lp->ls, &lp->lockname, &lpn);
	419	if (error)
	420	goto out;
	421
	422	lpn->lksb.sb_lvbptr = junk_lvb;
	423	lpn->lvb = junk_lvb;
	424
	425	lpn->req = DLM_LOCK_NL;
	426	lpn->lkf = DLM_LKF_VALBLK \| DLM_LKF_EXPEDITE;
	427	set_bit(LFL_NOBAST, &lpn->flags);
	428	set_bit(LFL_INLOCK, &lpn->flags);
	429
	430	init_completion(&lpn->ast_wait);
8d3b35a4	431	gdlm_do_lock(lpn);
29b7998d	432	wait_for_completion(&lpn->ast_wait);
c5921fd0	433	error = lpn->lksb.sb_status;
29b7998d	434	if (error) {
d92a8d48 SW	435	printk(KERN_INFO "lock_dlm: hold_null_lock dlm error %d\n",
d92a8d48 SW	436	error);
29b7998d DT	437	gdlm_delete_lp(lpn);
	438	lpn = NULL;
	439	}
	440	out:
	441	lp->hold_null = lpn;
	442	return error;
	443	}
	444
	445	/* This cannot do a synchronous dlm request (requiring a lock_dlm thread to get
	446	the completion) because gfs may call unhold_lvb() during a callback (from
	447	the context of a lock_dlm thread) which could cause a deadlock since the
	448	other lock_dlm thread could be engaged in recovery. */
	449
	450	static void unhold_null_lock(struct gdlm_lock *lp)
	451	{
	452	struct gdlm_lock *lpn = lp->hold_null;
	453
9229f013 DT	454	gdlm_assert(lpn, "%x,%llx", lp->lockname.ln_type,
9229f013 DT	455	(unsigned long long)lp->lockname.ln_number);
29b7998d DT	456	lpn->lksb.sb_lvbptr = NULL;
	457	lpn->lvb = NULL;
	458	set_bit(LFL_UNLOCK_DELETE, &lpn->flags);
	459	gdlm_do_unlock(lpn);
	460	lp->hold_null = NULL;
	461	}
	462
	463	/* Acquire a NL lock because gfs requires the value block to remain
	464	intact on the resource while the lvb is "held" even if it's holding no locks
	465	on the resource. */
	466
	467	int gdlm_hold_lvb(lm_lock_t lock, char *lvbp)
	468	{
	469	struct gdlm_lock lp = (struct gdlm_lock ) lock;
	470	int error;
	471
	472	error = gdlm_add_lvb(lp);
	473	if (error)
	474	return error;
	475
	476	*lvbp = lp->lvb;
	477
	478	error = hold_null_lock(lp);
	479	if (error)
	480	gdlm_del_lvb(lp);
	481
	482	return error;
	483	}
	484
	485	void gdlm_unhold_lvb(lm_lock_t lock, char lvb)
	486	{
	487	struct gdlm_lock lp = (struct gdlm_lock ) lock;
	488
	489	unhold_null_lock(lp);
	490	gdlm_del_lvb(lp);
	491	}
	492
	493	void gdlm_sync_lvb(lm_lock_t lock, char lvb)
	494	{
	495	struct gdlm_lock lp = (struct gdlm_lock ) lock;
	496
	497	if (lp->cur != DLM_LOCK_EX)
	498	return;
	499
	500	init_completion(&lp->ast_wait);
	501	set_bit(LFL_SYNC_LVB, &lp->flags);
	502
	503	lp->req = DLM_LOCK_EX;
	504	lp->lkf = make_flags(lp, 0, lp->cur, lp->req);
	505
8d3b35a4	506	gdlm_do_lock(lp);
29b7998d DT	507	wait_for_completion(&lp->ast_wait);
	508	}
	509
	510	void gdlm_submit_delayed(struct gdlm_ls *ls)
	511	{
	512	struct gdlm_lock lp, safe;
	513
	514	spin_lock(&ls->async_lock);
	515	list_for_each_entry_safe(lp, safe, &ls->delayed, delay_list) {
	516	list_del_init(&lp->delay_list);
	517	list_add_tail(&lp->delay_list, &ls->submit);
	518	}
	519	spin_unlock(&ls->async_lock);
	520	wake_up(&ls->thread_wait);
	521	}
	522
	523	int gdlm_release_all_locks(struct gdlm_ls *ls)
	524	{
	525	struct gdlm_lock lp, safe;
	526	int count = 0;
	527
	528	spin_lock(&ls->async_lock);
	529	list_for_each_entry_safe(lp, safe, &ls->all_locks, all_list) {
	530	list_del_init(&lp->all_list);
	531
	532	if (lp->lvb && lp->lvb != junk_lvb)
	533	kfree(lp->lvb);
	534	kfree(lp);
	535	count++;
	536	}
	537	spin_unlock(&ls->async_lock);
	538
	539	return count;
	540	}
	541