dlmglue.c

ocfs1.2.7 源码

/* -*- mode: c; c-basic-offset: 8; -*-
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * dlmglue.c
 *
 * Code which implements an OCFS2 specific interface to our DLM.
 *
 * Copyright (C) 2003, 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/types.h>
#include <linux/slab.h>
#include <linux/highmem.h>
#include <linux/mm.h>
#include <linux/smp_lock.h>
#include <linux/crc32.h>
#include <linux/kthread.h>
#include <linux/debugfs.h>
#include <linux/seq_file.h>

#include <cluster/heartbeat.h>
#include <cluster/nodemanager.h>
#include <cluster/tcp.h>

#include <dlm/dlmapi.h>

#define MLOG_MASK_PREFIX ML_DLM_GLUE
#include <cluster/masklog.h>

#include "ocfs2.h"

#include "alloc.h"
#include "dlmglue.h"
#include "extent_map.h"
#include "heartbeat.h"
#include "inode.h"
#include "journal.h"
#include "slot_map.h"
#include "super.h"
#include "uptodate.h"
#include "vote.h"

#include "buffer_head_io.h"

static void ocfs2_inode_ast_func(void *opaque);
static void ocfs2_inode_bast_func(void *opaque,
				  int level);
static void ocfs2_super_ast_func(void *opaque);
static void ocfs2_super_bast_func(void *opaque,
				  int level);
static void ocfs2_rename_ast_func(void *opaque);
static void ocfs2_rename_bast_func(void *opaque,
				   int level);

/* so far, all locks have gotten along with the same unlock ast */
static void ocfs2_unlock_ast_func(void *opaque,
				  enum dlm_status status);
static int ocfs2_do_unblock_meta(struct inode *inode,
				 int *requeue);
static int ocfs2_unblock_meta(struct ocfs2_lock_res *lockres,
			      int *requeue);
static int ocfs2_unblock_data(struct ocfs2_lock_res *lockres,
			      int *requeue);
static int ocfs2_unblock_osb_lock(struct ocfs2_lock_res *lockres,
				  int *requeue);
typedef void (ocfs2_convert_worker_t)(struct ocfs2_lock_res *, int);
static int ocfs2_generic_unblock_lock(struct ocfs2_super *osb,
				      struct ocfs2_lock_res *lockres,
				      int *requeue,
				      ocfs2_convert_worker_t *worker);

struct ocfs2_lock_res_ops {
	void (*ast)(void *);
	void (*bast)(void *, int);
	void (*unlock_ast)(void *, enum dlm_status);
	int  (*unblock)(struct ocfs2_lock_res *, int *);
};

static struct ocfs2_lock_res_ops ocfs2_inode_meta_lops = {
	.ast		= ocfs2_inode_ast_func,
	.bast		= ocfs2_inode_bast_func,
	.unlock_ast	= ocfs2_unlock_ast_func,
	.unblock	= ocfs2_unblock_meta,
};

static void ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
				      int blocking);

static struct ocfs2_lock_res_ops ocfs2_inode_data_lops = {
	.ast		= ocfs2_inode_ast_func,
	.bast		= ocfs2_inode_bast_func,
	.unlock_ast	= ocfs2_unlock_ast_func,
	.unblock	= ocfs2_unblock_data,
};

static struct ocfs2_lock_res_ops ocfs2_super_lops = {
	.ast		= ocfs2_super_ast_func,
	.bast		= ocfs2_super_bast_func,
	.unlock_ast	= ocfs2_unlock_ast_func,
	.unblock	= ocfs2_unblock_osb_lock,
};

static struct ocfs2_lock_res_ops ocfs2_rename_lops = {
	.ast		= ocfs2_rename_ast_func,
	.bast		= ocfs2_rename_bast_func,
	.unlock_ast	= ocfs2_unlock_ast_func,
	.unblock	= ocfs2_unblock_osb_lock,
};

static inline int ocfs2_is_inode_lock(struct ocfs2_lock_res *lockres)
{
	return lockres->l_type == OCFS2_LOCK_TYPE_META ||
		lockres->l_type == OCFS2_LOCK_TYPE_DATA;
}

static inline int ocfs2_is_super_lock(struct ocfs2_lock_res *lockres)
{
	return lockres->l_type == OCFS2_LOCK_TYPE_SUPER;
}

static inline int ocfs2_is_rename_lock(struct ocfs2_lock_res *lockres)
{
	return lockres->l_type == OCFS2_LOCK_TYPE_RENAME;
}

static inline struct ocfs2_super *ocfs2_lock_res_super(struct ocfs2_lock_res *lockres)
{
	BUG_ON(!ocfs2_is_super_lock(lockres)
	       && !ocfs2_is_rename_lock(lockres));

	return (struct ocfs2_super *) lockres->l_priv;
}

static inline struct inode *ocfs2_lock_res_inode(struct ocfs2_lock_res *lockres)
{
	BUG_ON(!ocfs2_is_inode_lock(lockres));

	return (struct inode *) lockres->l_priv;
}

static int ocfs2_lock_create(struct ocfs2_super *osb,
			     struct ocfs2_lock_res *lockres,
			     int level,
			     int flags);
static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lockres,
						     int wanted);
static int ocfs2_cluster_lock(struct ocfs2_super *osb,
			      struct ocfs2_lock_res *lockres,
			      int level,
			      int lkm_flags,
			      ocfs2_lock_callback cb,
			      unsigned long cb_data);
static void ocfs2_cluster_unlock(struct ocfs2_super *osb,
				 struct ocfs2_lock_res *lockres,
				 int level);
static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres);
static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres);
static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres);
static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres, int level);
static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
					struct ocfs2_lock_res *lockres);
static inline void ocfs2_recover_from_dlm_error(struct ocfs2_lock_res *lockres,
						int convert);
#define ocfs2_log_dlm_error(_func, _stat, _lockres) do {	\
	mlog(ML_ERROR, "Dlm error \"%s\" while calling %s on "	\
		"resource %s: %s\n", dlm_errname(_stat), _func,	\
		_lockres->l_name, dlm_errmsg(_stat));		\
} while (0)
static void ocfs2_vote_on_unlock(struct ocfs2_super *osb,
				 struct ocfs2_lock_res *lockres);
static int ocfs2_meta_lock_update(struct inode *inode,
				  struct buffer_head **bh);
static void ocfs2_drop_osb_locks(struct ocfs2_super *osb);
static inline int ocfs2_highest_compat_lock_level(int level);
static int __ocfs2_downconvert_lock(struct ocfs2_super *osb,
				    struct ocfs2_lock_res *lockres,
				    int new_level,
				    int lvb);
static int __ocfs2_cancel_convert(struct ocfs2_super *osb,
				  struct ocfs2_lock_res *lockres);
static inline int ocfs2_can_downconvert_meta_lock(struct inode *inode,
						  struct ocfs2_lock_res *lockres,
						  int new_level);

static void ocfs2_build_lock_name(enum ocfs2_lock_type type,
				  u64 blkno,
				  u32 generation,
				  char *name)
{
	int len;

	mlog_entry_void();

	BUG_ON(type >= OCFS2_NUM_LOCK_TYPES);

	len = snprintf(name, OCFS2_LOCK_ID_MAX_LEN, "%c%s%016"MLFx64"%08x",
		       ocfs2_lock_type_char(type), OCFS2_LOCK_ID_PAD, blkno,
		       generation);

	BUG_ON(len != (OCFS2_LOCK_ID_MAX_LEN - 1));

	mlog(0, "built lock resource with name: %s\n", name);

	mlog_exit_void();
}

static spinlock_t ocfs2_dlm_tracking_lock = SPIN_LOCK_UNLOCKED;

static void ocfs2_add_lockres_tracking(struct ocfs2_lock_res *res,
				       struct ocfs2_dlm_debug *dlm_debug)
{
	mlog(0, "Add tracking for lockres %s\n", res->l_name);

	spin_lock(&ocfs2_dlm_tracking_lock);
	list_add(&res->l_debug_list, &dlm_debug->d_lockres_tracking);
	spin_unlock(&ocfs2_dlm_tracking_lock);
}

static void ocfs2_remove_lockres_tracking(struct ocfs2_lock_res *res)
{
	spin_lock(&ocfs2_dlm_tracking_lock);
	if (!list_empty(&res->l_debug_list))
		list_del_init(&res->l_debug_list);
	spin_unlock(&ocfs2_dlm_tracking_lock);
}

static void ocfs2_lock_res_init_common(struct ocfs2_super *osb,
				       struct ocfs2_lock_res *res,
				       enum ocfs2_lock_type type,
				       u64 blkno,
				       u32 generation,
				       struct ocfs2_lock_res_ops *ops,
				       void *priv)
{
	ocfs2_build_lock_name(type, blkno, generation, res->l_name);

	res->l_type          = type;
	res->l_ops           = ops;
	res->l_priv          = priv;

	res->l_level         = LKM_IVMODE;
	res->l_requested     = LKM_IVMODE;
	res->l_blocking      = LKM_IVMODE;
	res->l_action        = OCFS2_AST_INVALID;
	res->l_unlock_action = OCFS2_UNLOCK_INVALID;

	res->l_flags         = OCFS2_LOCK_INITIALIZED;

	ocfs2_add_lockres_tracking(res, osb->osb_dlm_debug);
}

void ocfs2_lock_res_init_once(struct ocfs2_lock_res *res)
{
	/* This also clears out the lock status block */
	memset(res, 0, sizeof(struct ocfs2_lock_res));
	spin_lock_init(&res->l_lock);
	init_waitqueue_head(&res->l_event);
	INIT_LIST_HEAD(&res->l_blocked_list);
	INIT_LIST_HEAD(&res->l_flag_cb_list);
}

void ocfs2_inode_lock_res_init(struct ocfs2_lock_res *res,
			       enum ocfs2_lock_type type,
			       struct inode *inode)
{
	struct ocfs2_lock_res_ops *ops;

	BUG_ON(type != OCFS2_LOCK_TYPE_META &&
	       type != OCFS2_LOCK_TYPE_DATA);

	if (type == OCFS2_LOCK_TYPE_META)
		ops = &ocfs2_inode_meta_lops;
	else
		ops = &ocfs2_inode_data_lops;

	ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), res, type,
				   OCFS2_I(inode)->ip_blkno,
				   inode->i_generation, ops, inode);
}

static void ocfs2_super_lock_res_init(struct ocfs2_lock_res *res,
				      struct ocfs2_super *osb)
{
	/* Superblock lockres doesn't come from a slab so we call init
	 * once on it manually.  */
	ocfs2_lock_res_init_once(res);
	ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_SUPER,
				   OCFS2_SUPER_BLOCK_BLKNO, 0,
				   &ocfs2_super_lops, osb);
}

static void ocfs2_rename_lock_res_init(struct ocfs2_lock_res *res,
				       struct ocfs2_super *osb)
{
	/* Rename lockres doesn't come from a slab so we call init
	 * once on it manually.  */
	ocfs2_lock_res_init_once(res);
	ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_RENAME, 0, 0,
				   &ocfs2_rename_lops, osb);
}

void ocfs2_lock_res_free(struct ocfs2_lock_res *res)
{
	mlog_entry_void();

	if (!(res->l_flags & OCFS2_LOCK_INITIALIZED))
		return;

	ocfs2_remove_lockres_tracking(res);

	mlog_bug_on_msg(!list_empty(&res->l_blocked_list),
			"Lockres %s is on the blocked list\n",
			res->l_name);
	mlog_bug_on_msg(!list_empty(&res->l_flag_cb_list),
			"Lockres %s has flag callbacks pending\n",
			res->l_name);
	mlog_bug_on_msg(spin_is_locked(&res->l_lock),
			"Lockres %s is locked\n",
			res->l_name);
	mlog_bug_on_msg(res->l_ro_holders,
			"Lockres %s has %u ro holders\n",
			res->l_name, res->l_ro_holders);
	mlog_bug_on_msg(res->l_ex_holders,
			"Lockres %s has %u ex holders\n",
			res->l_name, res->l_ex_holders);

	/* Need to clear out the lock status block for the dlm */
	memset(&res->l_lksb, 0, sizeof(res->l_lksb));

	res->l_flags = 0UL;
	mlog_exit_void();
}

static inline void ocfs2_inc_holders(struct ocfs2_lock_res *lockres,
				     int level)
{
	mlog_entry_void();

	BUG_ON(!lockres);

	switch(level) {
	case LKM_EXMODE:
		lockres->l_ex_holders++;
		break;
	case LKM_PRMODE:
		lockres->l_ro_holders++;
		break;
	default:
		BUG();
	}

	mlog_exit_void();
}

static inline void ocfs2_dec_holders(struct ocfs2_lock_res *lockres,
				     int level)
{
	mlog_entry_void();

	BUG_ON(!lockres);

	switch(level) {
	case LKM_EXMODE:
		BUG_ON(!lockres->l_ex_holders);
		lockres->l_ex_holders--;
		break;
	case LKM_PRMODE:
		BUG_ON(!lockres->l_ro_holders);
		lockres->l_ro_holders--;
		break;
	default:
		BUG();
	}
	mlog_exit_void();
}

/* WARNING: This function lives in a world where the only three lock
 * levels are EX, PR, and NL. It *will* have to be adjusted when more
 * lock types are added. */
static inline int ocfs2_highest_compat_lock_level(int level)
{
	int new_level = LKM_EXMODE;

	if (level == LKM_EXMODE)
		new_level = LKM_NLMODE;
	else if (level == LKM_PRMODE)
		new_level = LKM_PRMODE;
	return new_level;
}

/* XXX must be called with lockres->l_lock held */
static void lockres_set_flags(struct ocfs2_lock_res *lockres, unsigned long newflags)
{
	struct list_head *pos, *tmp;
	struct ocfs2_lockres_flag_callback *fcb;

	assert_spin_locked(&lockres->l_lock);

	lockres->l_flags = newflags;

	list_for_each_safe(pos, tmp, &lockres->l_flag_cb_list) {
		fcb = list_entry(pos, struct ocfs2_lockres_flag_callback,
				 fc_lockres_item);
		if ((lockres->l_flags & fcb->fc_flag_mask) !=
		    fcb->fc_flag_goal)
			continue;

		list_del_init(&fcb->fc_lockres_item);
		fcb->fc_cb(0, fcb->fc_data);
		if (fcb->fc_free_once_called)
			kfree(fcb);
	}
}

static void lockres_or_flags(struct ocfs2_lock_res *lockres, unsigned long or)
{
	lockres_set_flags(lockres, lockres->l_flags | or);
}
static void lockres_clear_flags(struct ocfs2_lock_res *lockres, unsigned long clear)
{
	lockres_set_flags(lockres, lockres->l_flags & ~clear);
}

static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres)
{
	mlog_entry_void();

	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_ATTACHED));
	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BLOCKED));
	BUG_ON(lockres->l_blocking <= LKM_NLMODE);

	lockres->l_level = lockres->l_requested;
	if (lockres->l_level <=
	    ocfs2_highest_compat_lock_level(lockres->l_blocking)) {
		lockres->l_blocking = LKM_NLMODE;
		lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
	}
	lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);

	mlog_exit_void();
}

static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres)
{
	mlog_entry_void();

	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_ATTACHED));

	/* Convert from RO to EX doesn't really need anything as our
	 * information is already up to data. Convert from NL to
	 * *anything* however should mark ourselves as needing an
	 * update */
	if (lockres->l_level == LKM_NLMODE)
		lockres_or_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);

	lockres->l_level = lockres->l_requested;
	lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);

	mlog_exit_void();
}

static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres)
{
	mlog_entry_void();

	BUG_ON((!lockres->l_flags & OCFS2_LOCK_BUSY));
	BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);

	if (lockres->l_requested > LKM_NLMODE &&
	    !(lockres->l_flags & OCFS2_LOCK_LOCAL))
		lockres_or_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);

	lockres->l_level = lockres->l_requested;
	lockres_or_flags(lockres, OCFS2_LOCK_ATTACHED);
	lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);

	mlog_exit_void();
}

static void ocfs2_inode_ast_func(void *opaque)
{
	struct ocfs2_lock_res *lockres = opaque;
	struct inode *inode;
	struct dlm_lockstatus *lksb;