[linux.git] / fs / dlm / member.c

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