[linux-2.6-block.git] / fs / dlm / member.c

/******************************************************************************
*******************************************************************************
**
**  Copyright (C) 2005-2007 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 v.2.
**
*******************************************************************************
******************************************************************************/

#include "dlm_internal.h"
#include "lockspace.h"
#include "member.h"
#include "recoverd.h"
#include "recover.h"
#include "rcom.h"
#include "config.h"

/*
 * Following called by dlm_recoverd thread
 */

static void add_ordered_member(struct dlm_ls *ls, struct dlm_member *new)
{
	struct dlm_member *memb = NULL;
	struct list_head *tmp;
	struct list_head *newlist = &new->list;
	struct list_head *head = &ls->ls_nodes;

	list_for_each(tmp, head) {
		memb = list_entry(tmp, struct dlm_member, list);
		if (new->nodeid < memb->nodeid)
			break;
	}

	if (!memb)
		list_add_tail(newlist, head);
	else {
		/* FIXME: can use list macro here */
		newlist->prev = tmp->prev;
		newlist->next = tmp;
		tmp->prev->next = newlist;
		tmp->prev = newlist;
	}
}

static int dlm_add_member(struct dlm_ls *ls, int nodeid)
{
	struct dlm_member *memb;
	int w;

	memb = kzalloc(sizeof(struct dlm_member), GFP_KERNEL);
	if (!memb)
		return -ENOMEM;

	w = dlm_node_weight(ls->ls_name, nodeid);
	if (w < 0) {
		kfree(memb);
		return w;
	}

	memb->nodeid = nodeid;
	memb->weight = w;
	add_ordered_member(ls, memb);
	ls->ls_num_nodes++;
	return 0;
}

static void dlm_remove_member(struct dlm_ls *ls, struct dlm_member *memb)
{
	list_move(&memb->list, &ls->ls_nodes_gone);
	ls->ls_num_nodes--;
}

static int dlm_is_member(struct dlm_ls *ls, int nodeid)
{
	struct dlm_member *memb;

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		if (memb->nodeid == nodeid)
			return 1;
	}
	return 0;
}

int dlm_is_removed(struct dlm_ls *ls, int nodeid)
{
	struct dlm_member *memb;

	list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
		if (memb->nodeid == nodeid)
			return 1;
	}
	return 0;
}

static void clear_memb_list(struct list_head *head)
{
	struct dlm_member *memb;

	while (!list_empty(head)) {
		memb = list_entry(head->next, struct dlm_member, list);
		list_del(&memb->list);
		kfree(memb);
	}
}

void dlm_clear_members(struct dlm_ls *ls)
{
	clear_memb_list(&ls->ls_nodes);
	ls->ls_num_nodes = 0;
}

void dlm_clear_members_gone(struct dlm_ls *ls)
{
	clear_memb_list(&ls->ls_nodes_gone);
}

static void make_member_array(struct dlm_ls *ls)
{
	struct dlm_member *memb;
	int i, w, x = 0, total = 0, all_zero = 0, *array;

	kfree(ls->ls_node_array);
	ls->ls_node_array = NULL;

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		if (memb->weight)
			total += memb->weight;
	}

	/* all nodes revert to weight of 1 if all have weight 0 */

	if (!total) {
		total = ls->ls_num_nodes;
		all_zero = 1;
	}

	ls->ls_total_weight = total;

	array = kmalloc(sizeof(int) * total, GFP_KERNEL);
	if (!array)
		return;

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		if (!all_zero && !memb->weight)
			continue;

		if (all_zero)
			w = 1;
		else
			w = memb->weight;

		DLM_ASSERT(x < total, printk("total %d x %d\n", total, x););

		for (i = 0; i < w; i++)
			array[x++] = memb->nodeid;
	}

	ls->ls_node_array = array;
}

/* send a status request to all members just to establish comms connections */

static int ping_members(struct dlm_ls *ls)
{
	struct dlm_member *memb;
	int error = 0;

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		error = dlm_recovery_stopped(ls);
		if (error)
			break;
		error = dlm_rcom_status(ls, memb->nodeid);
		if (error)
			break;
	}
	if (error)
		log_debug(ls, "ping_members aborted %d last nodeid %d",
			  error, ls->ls_recover_nodeid);
	return error;
}

int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
{
	struct dlm_member *memb, *safe;
	int i, error, found, pos = 0, neg = 0, low = -1;

	/* previously removed members that we've not finished removing need to
	   count as a negative change so the "neg" recovery steps will happen */

	list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
		log_debug(ls, "prev removed member %d", memb->nodeid);
		neg++;
	}

	/* move departed members from ls_nodes to ls_nodes_gone */

	list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
		found = 0;
		for (i = 0; i < rv->node_count; i++) {
			if (memb->nodeid == rv->nodeids[i]) {
				found = 1;
				break;
			}
		}

		if (!found) {
			neg++;
			dlm_remove_member(ls, memb);
			log_debug(ls, "remove member %d", memb->nodeid);
		}
	}

	/* add new members to ls_nodes */

	for (i = 0; i < rv->node_count; i++) {
		if (dlm_is_member(ls, rv->nodeids[i]))
			continue;
		dlm_add_member(ls, rv->nodeids[i]);
		pos++;
		log_debug(ls, "add member %d", rv->nodeids[i]);
	}

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		if (low == -1 || memb->nodeid < low)
			low = memb->nodeid;
	}
	ls->ls_low_nodeid = low;

	make_member_array(ls);
	dlm_set_recover_status(ls, DLM_RS_NODES);
	*neg_out = neg;

	error = ping_members(ls);
	if (!error || error == -EPROTO) {
		/* new_lockspace() may be waiting to know if the config
		   is good or bad */
		ls->ls_members_result = error;
		complete(&ls->ls_members_done);
	}
	if (error)
		goto out;

	error = dlm_recover_members_wait(ls);
 out:
	log_debug(ls, "total members %d error %d", ls->ls_num_nodes, error);
	return error;
}

/*
 * Following called from lockspace.c
 */

int dlm_ls_stop(struct dlm_ls *ls)
{
	int new;

	/*
	 * A stop cancels any recovery that's in progress (see RECOVERY_STOP,
	 * dlm_recovery_stopped()) and prevents any new locks from being
	 * processed (see RUNNING, dlm_locking_stopped()).
	 */

	spin_lock(&ls->ls_recover_lock);
	set_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
	new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
	ls->ls_recover_seq++;
	spin_unlock(&ls->ls_recover_lock);

	/*
	 * This in_recovery lock does two things:
	 *
	 * 1) Keeps this function from returning until all threads are out
	 *    of locking routines and locking is truely stopped.
	 * 2) Keeps any new requests from being processed until it's unlocked
	 *    when recovery is complete.
	 */

	if (new)
		down_write(&ls->ls_in_recovery);

	/*
	 * The recoverd suspend/resume makes sure that dlm_recoverd (if
	 * running) has noticed the clearing of RUNNING above and quit
	 * processing the previous recovery.  This will be true for all nodes
	 * before any nodes start the new recovery.
	 */

	dlm_recoverd_suspend(ls);
	ls->ls_recover_status = 0;
	dlm_recoverd_resume(ls);

	if (!ls->ls_recover_begin)
		ls->ls_recover_begin = jiffies;
	return 0;
}

int dlm_ls_start(struct dlm_ls *ls)
{
	struct dlm_recover *rv = NULL, *rv_old;
	int *ids = NULL;
	int error, count;

	rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
	if (!rv)
		return -ENOMEM;

	error = count = dlm_nodeid_list(ls->ls_name, &ids);
	if (error <= 0)
		goto fail;

	spin_lock(&ls->ls_recover_lock);

	/* the lockspace needs to be stopped before it can be started */

	if (!dlm_locking_stopped(ls)) {
		spin_unlock(&ls->ls_recover_lock);
		log_error(ls, "start ignored: lockspace running");
		error = -EINVAL;
		goto fail;
	}

	rv->nodeids = ids;
	rv->node_count = count;
	rv->seq = ++ls->ls_recover_seq;
	rv_old = ls->ls_recover_args;
	ls->ls_recover_args = rv;
	spin_unlock(&ls->ls_recover_lock);

	if (rv_old) {
		kfree(rv_old->nodeids);
		kfree(rv_old);
	}

	dlm_recoverd_kick(ls);
	return 0;

 fail:
	kfree(rv);
	kfree(ids);
	return error;
}
Commit	Line	Data
e7fd4179 DT	1	/******************************************************************************
	2	*******************************************************************************
	3	**
3ae1acf9	4	** Copyright (C) 2005-2007 Red Hat, Inc. All rights reserved.
e7fd4179 DT	5	**
	6	** This copyrighted material is made available to anyone wishing to use,
	7	** modify, copy, or redistribute it subject to the terms and conditions
	8	** of the GNU General Public License v.2.
	9	**
	10	*******************************************************************************
	11	******************************************************************************/
	12
	13	#include "dlm_internal.h"
	14	#include "lockspace.h"
	15	#include "member.h"
	16	#include "recoverd.h"
	17	#include "recover.h"
e7fd4179 DT	18	#include "rcom.h"
	19	#include "config.h"
	20
	21	/*
	22	* Following called by dlm_recoverd thread
	23	*/
	24
	25	static void add_ordered_member(struct dlm_ls ls, struct dlm_member new)
	26	{
	27	struct dlm_member *memb = NULL;
	28	struct list_head *tmp;
	29	struct list_head *newlist = &new->list;
	30	struct list_head *head = &ls->ls_nodes;
	31
	32	list_for_each(tmp, head) {
	33	memb = list_entry(tmp, struct dlm_member, list);
	34	if (new->nodeid < memb->nodeid)
	35	break;
	36	}
	37
	38	if (!memb)
	39	list_add_tail(newlist, head);
	40	else {
	41	/* FIXME: can use list macro here */
	42	newlist->prev = tmp->prev;
	43	newlist->next = tmp;
	44	tmp->prev->next = newlist;
	45	tmp->prev = newlist;
	46	}
	47	}
	48
	49	static int dlm_add_member(struct dlm_ls *ls, int nodeid)
	50	{
	51	struct dlm_member *memb;
	52	int w;
	53
90135925	54	memb = kzalloc(sizeof(struct dlm_member), GFP_KERNEL);
e7fd4179 DT	55	if (!memb)
	56	return -ENOMEM;
	57
	58	w = dlm_node_weight(ls->ls_name, nodeid);
1a2bf2ee JJ	59	if (w < 0) {
1a2bf2ee JJ	60	kfree(memb);
e7fd4179	61	return w;
1a2bf2ee	62	}
e7fd4179 DT	63
	64	memb->nodeid = nodeid;
	65	memb->weight = w;
	66	add_ordered_member(ls, memb);
	67	ls->ls_num_nodes++;
	68	return 0;
	69	}
	70
	71	static void dlm_remove_member(struct dlm_ls ls, struct dlm_member memb)
	72	{
	73	list_move(&memb->list, &ls->ls_nodes_gone);
	74	ls->ls_num_nodes--;
	75	}
	76
	77	static int dlm_is_member(struct dlm_ls *ls, int nodeid)
	78	{
	79	struct dlm_member *memb;
	80
	81	list_for_each_entry(memb, &ls->ls_nodes, list) {
	82	if (memb->nodeid == nodeid)
90135925	83	return 1;
e7fd4179	84	}
90135925	85	return 0;
e7fd4179 DT	86	}
	87
	88	int dlm_is_removed(struct dlm_ls *ls, int nodeid)
	89	{
	90	struct dlm_member *memb;
	91
	92	list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
	93	if (memb->nodeid == nodeid)
90135925	94	return 1;
e7fd4179	95	}
90135925	96	return 0;
e7fd4179 DT	97	}
	98
	99	static void clear_memb_list(struct list_head *head)
	100	{
	101	struct dlm_member *memb;
	102
	103	while (!list_empty(head)) {
	104	memb = list_entry(head->next, struct dlm_member, list);
	105	list_del(&memb->list);
	106	kfree(memb);
	107	}
	108	}
	109
	110	void dlm_clear_members(struct dlm_ls *ls)
	111	{
	112	clear_memb_list(&ls->ls_nodes);
	113	ls->ls_num_nodes = 0;
	114	}
	115
	116	void dlm_clear_members_gone(struct dlm_ls *ls)
	117	{
	118	clear_memb_list(&ls->ls_nodes_gone);
	119	}
	120
	121	static void make_member_array(struct dlm_ls *ls)
	122	{
	123	struct dlm_member *memb;
	124	int i, w, x = 0, total = 0, all_zero = 0, *array;
	125
	126	kfree(ls->ls_node_array);
	127	ls->ls_node_array = NULL;
	128
	129	list_for_each_entry(memb, &ls->ls_nodes, list) {
	130	if (memb->weight)
	131	total += memb->weight;
	132	}
	133
	134	/* all nodes revert to weight of 1 if all have weight 0 */
	135
	136	if (!total) {
	137	total = ls->ls_num_nodes;
	138	all_zero = 1;
	139	}
	140
	141	ls->ls_total_weight = total;
	142
	143	array = kmalloc(sizeof(int) * total, GFP_KERNEL);
	144	if (!array)
	145	return;
	146
	147	list_for_each_entry(memb, &ls->ls_nodes, list) {
	148	if (!all_zero && !memb->weight)
	149	continue;
	150
	151	if (all_zero)
	152	w = 1;
	153	else
	154	w = memb->weight;
	155
	156	DLM_ASSERT(x < total, printk("total %d x %d\n", total, x););
	157
	158	for (i = 0; i < w; i++)
	159	array[x++] = memb->nodeid;
	160	}
161
162	ls->ls_node_array = array;
163	}
164
165	/* send a status request to all members just to establish comms connections */
166
f6db1b8e	167	static int ping_members(struct dlm_ls *ls)
e7fd4179 DT	168	{
e7fd4179 DT	169	struct dlm_member *memb;
f6db1b8e DT	170	int error = 0;
	171
	172	list_for_each_entry(memb, &ls->ls_nodes, list) {
	173	error = dlm_recovery_stopped(ls);
	174	if (error)
	175	break;
	176	error = dlm_rcom_status(ls, memb->nodeid);
	177	if (error)
	178	break;
	179	}
	180	if (error)
faa0f267 DT	181	log_debug(ls, "ping_members aborted %d last nodeid %d",
faa0f267 DT	182	error, ls->ls_recover_nodeid);
f6db1b8e	183	return error;
e7fd4179 DT	184	}
	185
	186	int dlm_recover_members(struct dlm_ls ls, struct dlm_recover rv, int *neg_out)
	187	{
	188	struct dlm_member memb, safe;
	189	int i, error, found, pos = 0, neg = 0, low = -1;
	190
91c0dc93 DT	191	/* previously removed members that we've not finished removing need to
	192	count as a negative change so the "neg" recovery steps will happen */
	193
	194	list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
	195	log_debug(ls, "prev removed member %d", memb->nodeid);
	196	neg++;
	197	}
	198
e7fd4179 DT	199	/* move departed members from ls_nodes to ls_nodes_gone */
	200
	201	list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
90135925	202	found = 0;
e7fd4179 DT	203	for (i = 0; i < rv->node_count; i++) {
e7fd4179 DT	204	if (memb->nodeid == rv->nodeids[i]) {
90135925	205	found = 1;
e7fd4179 DT	206	break;
	207	}
	208	}
	209
	210	if (!found) {
	211	neg++;
	212	dlm_remove_member(ls, memb);
	213	log_debug(ls, "remove member %d", memb->nodeid);
	214	}
	215	}
	216
	217	/* add new members to ls_nodes */
	218
	219	for (i = 0; i < rv->node_count; i++) {
	220	if (dlm_is_member(ls, rv->nodeids[i]))
	221	continue;
	222	dlm_add_member(ls, rv->nodeids[i]);
	223	pos++;
	224	log_debug(ls, "add member %d", rv->nodeids[i]);
	225	}
	226
	227	list_for_each_entry(memb, &ls->ls_nodes, list) {
	228	if (low == -1 \|\| memb->nodeid < low)
	229	low = memb->nodeid;
	230	}
	231	ls->ls_low_nodeid = low;
	232
	233	make_member_array(ls);
	234	dlm_set_recover_status(ls, DLM_RS_NODES);
	235	*neg_out = neg;
	236
f6db1b8e	237	error = ping_members(ls);
8b0e7b2c DT	238	if (!error \|\| error == -EPROTO) {
	239	/* new_lockspace() may be waiting to know if the config
	240	is good or bad */
	241	ls->ls_members_result = error;
	242	complete(&ls->ls_members_done);
	243	}
f6db1b8e DT	244	if (error)
f6db1b8e DT	245	goto out;
e7fd4179 DT	246
e7fd4179 DT	247	error = dlm_recover_members_wait(ls);
f6db1b8e DT	248	out:
f6db1b8e DT	249	log_debug(ls, "total members %d error %d", ls->ls_num_nodes, error);
e7fd4179 DT	250	return error;
	251	}
	252
	253	/*
	254	* Following called from lockspace.c
	255	*/
	256
	257	int dlm_ls_stop(struct dlm_ls *ls)
	258	{
	259	int new;
	260
	261	/*
	262	* A stop cancels any recovery that's in progress (see RECOVERY_STOP,
	263	* dlm_recovery_stopped()) and prevents any new locks from being
	264	* processed (see RUNNING, dlm_locking_stopped()).
	265	*/
	266
	267	spin_lock(&ls->ls_recover_lock);
	268	set_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
	269	new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
	270	ls->ls_recover_seq++;
	271	spin_unlock(&ls->ls_recover_lock);
	272
	273	/*
	274	* This in_recovery lock does two things:
	275	*
	276	* 1) Keeps this function from returning until all threads are out
	277	* of locking routines and locking is truely stopped.
	278	* 2) Keeps any new requests from being processed until it's unlocked
	279	* when recovery is complete.
	280	*/
	281
	282	if (new)
	283	down_write(&ls->ls_in_recovery);
	284
	285	/*
	286	* The recoverd suspend/resume makes sure that dlm_recoverd (if
	287	* running) has noticed the clearing of RUNNING above and quit
	288	* processing the previous recovery. This will be true for all nodes
	289	* before any nodes start the new recovery.
	290	*/
	291
	292	dlm_recoverd_suspend(ls);
	293	ls->ls_recover_status = 0;
	294	dlm_recoverd_resume(ls);
3ae1acf9 DT	295
	296	if (!ls->ls_recover_begin)
	297	ls->ls_recover_begin = jiffies;
e7fd4179 DT	298	return 0;
	299	}
	300
	301	int dlm_ls_start(struct dlm_ls *ls)
	302	{
	303	struct dlm_recover rv = NULL, rv_old;
	304	int *ids = NULL;
	305	int error, count;
	306
90135925	307	rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
e7fd4179 DT	308	if (!rv)
e7fd4179 DT	309	return -ENOMEM;
e7fd4179 DT	310
	311	error = count = dlm_nodeid_list(ls->ls_name, &ids);
	312	if (error <= 0)
	313	goto fail;
	314
	315	spin_lock(&ls->ls_recover_lock);
	316
	317	/* the lockspace needs to be stopped before it can be started */
	318
	319	if (!dlm_locking_stopped(ls)) {
	320	spin_unlock(&ls->ls_recover_lock);
	321	log_error(ls, "start ignored: lockspace running");
	322	error = -EINVAL;
	323	goto fail;
	324	}
	325
	326	rv->nodeids = ids;
	327	rv->node_count = count;
	328	rv->seq = ++ls->ls_recover_seq;
	329	rv_old = ls->ls_recover_args;
	330	ls->ls_recover_args = rv;
	331	spin_unlock(&ls->ls_recover_lock);
	332
	333	if (rv_old) {
	334	kfree(rv_old->nodeids);
	335	kfree(rv_old);
	336	}
	337
	338	dlm_recoverd_kick(ls);
	339	return 0;
	340
	341	fail:
	342	kfree(rv);
	343	kfree(ids);
	344	return error;
	345	}
	346