dlmconvert.c

linux 内核源代码

/* -*- mode: c; c-basic-offset: 8; -*-
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * dlmconvert.c
 *
 * underlying calls for lock conversion
 *
 * Copyright (C) 2004 Oracle.  All rights reserved.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * 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 for more details.
 *
 * You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 021110-1307, USA.
 *
 */


#include <linux/module.h>
#include <linux/fs.h>
#include <linux/types.h>
#include <linux/slab.h>
#include <linux/highmem.h>
#include <linux/utsname.h>
#include <linux/init.h>
#include <linux/sysctl.h>
#include <linux/random.h>
#include <linux/blkdev.h>
#include <linux/socket.h>
#include <linux/inet.h>
#include <linux/spinlock.h>


#include "cluster/heartbeat.h"
#include "cluster/nodemanager.h"
#include "cluster/tcp.h"

#include "dlmapi.h"
#include "dlmcommon.h"

#include "dlmconvert.h"

#define MLOG_MASK_PREFIX ML_DLM
#include "cluster/masklog.h"

/* NOTE: __dlmconvert_master is the only function in here that
 * needs a spinlock held on entry (res->spinlock) and it is the
 * only one that holds a lock on exit (res->spinlock).
 * All other functions in here need no locks and drop all of
 * the locks that they acquire. */
static enum dlm_status __dlmconvert_master(struct dlm_ctxt *dlm,
					   struct dlm_lock_resource *res,
					   struct dlm_lock *lock, int flags,
					   int type, int *call_ast,
					   int *kick_thread);
static enum dlm_status dlm_send_remote_convert_request(struct dlm_ctxt *dlm,
					   struct dlm_lock_resource *res,
					   struct dlm_lock *lock, int flags, int type);

/*
 * this is only called directly by dlmlock(), and only when the
 * local node is the owner of the lockres
 * locking:
 *   caller needs:  none
 *   taken:         takes and drops res->spinlock
 *   held on exit:  none
 * returns: see __dlmconvert_master
 */
enum dlm_status dlmconvert_master(struct dlm_ctxt *dlm,
				  struct dlm_lock_resource *res,
				  struct dlm_lock *lock, int flags, int type)
{
	int call_ast = 0, kick_thread = 0;
	enum dlm_status status;

	spin_lock(&res->spinlock);
	/* we are not in a network handler, this is fine */
	__dlm_wait_on_lockres(res);
	__dlm_lockres_reserve_ast(res);
	res->state |= DLM_LOCK_RES_IN_PROGRESS;

	status = __dlmconvert_master(dlm, res, lock, flags, type,
				     &call_ast, &kick_thread);

	res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
	spin_unlock(&res->spinlock);
	wake_up(&res->wq);
	if (status != DLM_NORMAL && status != DLM_NOTQUEUED)
		dlm_error(status);

	/* either queue the ast or release it */
	if (call_ast)
		dlm_queue_ast(dlm, lock);
	else
		dlm_lockres_release_ast(dlm, res);

	if (kick_thread)
		dlm_kick_thread(dlm, res);

	return status;
}

/* performs lock conversion at the lockres master site
 * locking:
 *   caller needs:  res->spinlock
 *   taken:         takes and drops lock->spinlock
 *   held on exit:  res->spinlock
 * returns: DLM_NORMAL, DLM_NOTQUEUED, DLM_DENIED
 *   call_ast: whether ast should be called for this lock
 *   kick_thread: whether dlm_kick_thread should be called
 */
static enum dlm_status __dlmconvert_master(struct dlm_ctxt *dlm,
					   struct dlm_lock_resource *res,
					   struct dlm_lock *lock, int flags,
					   int type, int *call_ast,
					   int *kick_thread)
{
	enum dlm_status status = DLM_NORMAL;
	struct list_head *iter;
	struct dlm_lock *tmplock=NULL;

	assert_spin_locked(&res->spinlock);

	mlog_entry("type=%d, convert_type=%d, new convert_type=%d\n",
		   lock->ml.type, lock->ml.convert_type, type);

	spin_lock(&lock->spinlock);

	/* already converting? */
	if (lock->ml.convert_type != LKM_IVMODE) {
		mlog(ML_ERROR, "attempted to convert a lock with a lock "
		     "conversion pending\n");
		status = DLM_DENIED;
		goto unlock_exit;
	}

	/* must be on grant queue to convert */
	if (!dlm_lock_on_list(&res->granted, lock)) {
		mlog(ML_ERROR, "attempted to convert a lock not on grant "
		     "queue\n");
		status = DLM_DENIED;
		goto unlock_exit;
	}

	if (flags & LKM_VALBLK) {
		switch (lock->ml.type) {
			case LKM_EXMODE:
				/* EX + LKM_VALBLK + convert == set lvb */
				mlog(0, "will set lvb: converting %s->%s\n",
				     dlm_lock_mode_name(lock->ml.type),
				     dlm_lock_mode_name(type));
				lock->lksb->flags |= DLM_LKSB_PUT_LVB;
				break;
			case LKM_PRMODE:
			case LKM_NLMODE:
				/* refetch if new level is not NL */
				if (type > LKM_NLMODE) {
					mlog(0, "will fetch new value into "
					     "lvb: converting %s->%s\n",
					     dlm_lock_mode_name(lock->ml.type),
					     dlm_lock_mode_name(type));
					lock->lksb->flags |= DLM_LKSB_GET_LVB;
				} else {
					mlog(0, "will NOT fetch new value "
					     "into lvb: converting %s->%s\n",
					     dlm_lock_mode_name(lock->ml.type),
					     dlm_lock_mode_name(type));
					flags &= ~(LKM_VALBLK);
				}
				break;
		}
	}


	/* in-place downconvert? */
	if (type <= lock->ml.type)
		goto grant;

	/* upconvert from here on */
	status = DLM_NORMAL;
	list_for_each(iter, &res->granted) {
		tmplock = list_entry(iter, struct dlm_lock, list);
		if (tmplock == lock)
			continue;
		if (!dlm_lock_compatible(tmplock->ml.type, type))
			goto switch_queues;
	}

	list_for_each(iter, &res->converting) {
		tmplock = list_entry(iter, struct dlm_lock, list);
		if (!dlm_lock_compatible(tmplock->ml.type, type))
			goto switch_queues;
		/* existing conversion requests take precedence */
		if (!dlm_lock_compatible(tmplock->ml.convert_type, type))
			goto switch_queues;
	}

	/* fall thru to grant */

grant:
	mlog(0, "res %.*s, granting %s lock\n", res->lockname.len,
	     res->lockname.name, dlm_lock_mode_name(type));
	/* immediately grant the new lock type */
	lock->lksb->status = DLM_NORMAL;
	if (lock->ml.node == dlm->node_num)
		mlog(0, "doing in-place convert for nonlocal lock\n");
	lock->ml.type = type;
	if (lock->lksb->flags & DLM_LKSB_PUT_LVB)
		memcpy(res->lvb, lock->lksb->lvb, DLM_LVB_LEN);

	status = DLM_NORMAL;
	*call_ast = 1;
	goto unlock_exit;

switch_queues:
	if (flags & LKM_NOQUEUE) {
		mlog(0, "failed to convert NOQUEUE lock %.*s from "
		     "%d to %d...\n", res->lockname.len, res->lockname.name,
		     lock->ml.type, type);
		status = DLM_NOTQUEUED;
		goto unlock_exit;
	}
	mlog(0, "res %.*s, queueing...\n", res->lockname.len,
	     res->lockname.name);

	lock->ml.convert_type = type;
	/* do not alter lock refcount.  switching lists. */
	list_move_tail(&lock->list, &res->converting);

unlock_exit:
	spin_unlock(&lock->spinlock);
	if (status == DLM_DENIED) {
		__dlm_print_one_lock_resource(res);
	}
	if (status == DLM_NORMAL)
		*kick_thread = 1;
	return status;
}

void dlm_revert_pending_convert(struct dlm_lock_resource *res,
				struct dlm_lock *lock)
{
	/* do not alter lock refcount.  switching lists. */
	list_move_tail(&lock->list, &res->granted);
	lock->ml.convert_type = LKM_IVMODE;
	lock->lksb->flags &= ~(DLM_LKSB_GET_LVB|DLM_LKSB_PUT_LVB);
}

/* messages the master site to do lock conversion
 * locking:
 *   caller needs:  none
 *   taken:         takes and drops res->spinlock, uses DLM_LOCK_RES_IN_PROGRESS
 *   held on exit:  none
 * returns: DLM_NORMAL, DLM_RECOVERING, status from remote node
 */
enum dlm_status dlmconvert_remote(struct dlm_ctxt *dlm,
				  struct dlm_lock_resource *res,
				  struct dlm_lock *lock, int flags, int type)
{
	enum dlm_status status;

	mlog(0, "type=%d, convert_type=%d, busy=%d\n", lock->ml.type,
	     lock->ml.convert_type, res->state & DLM_LOCK_RES_IN_PROGRESS);

	spin_lock(&res->spinlock);
	if (res->state & DLM_LOCK_RES_RECOVERING) {