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

/******************************************************************************
*******************************************************************************
**
**  Copyright (C) 2005-2011 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"
#include "lowcomms.h"

int dlm_slots_version(struct dlm_header *h)
{
	if ((h->h_version & 0x0000FFFF) < DLM_HEADER_SLOTS)
		return 0;
	return 1;
}

void dlm_slot_save(struct dlm_ls *ls, struct dlm_rcom *rc,
		   struct dlm_member *memb)
{
	struct rcom_config *rf = (struct rcom_config *)rc->rc_buf;

	if (!dlm_slots_version(&rc->rc_header))
		return;

	memb->slot = le16_to_cpu(rf->rf_our_slot);
	memb->generation = le32_to_cpu(rf->rf_generation);
}

void dlm_slots_copy_out(struct dlm_ls *ls, struct dlm_rcom *rc)
{
	struct dlm_slot *slot;
	struct rcom_slot *ro;
	int i;

	ro = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config));

	/* ls_slots array is sparse, but not rcom_slots */

	for (i = 0; i < ls->ls_slots_size; i++) {
		slot = &ls->ls_slots[i];
		if (!slot->nodeid)
			continue;
		ro->ro_nodeid = cpu_to_le32(slot->nodeid);
		ro->ro_slot = cpu_to_le16(slot->slot);
		ro++;
	}
}

#define SLOT_DEBUG_LINE 128

static void log_slots(struct dlm_ls *ls, uint32_t gen, int num_slots,
		      struct rcom_slot *ro0, struct dlm_slot *array,
		      int array_size)
{
	char line[SLOT_DEBUG_LINE];
	int len = SLOT_DEBUG_LINE - 1;
	int pos = 0;
	int ret, i;

	memset(line, 0, sizeof(line));

	if (array) {
		for (i = 0; i < array_size; i++) {
			if (!array[i].nodeid)
				continue;

			ret = snprintf(line + pos, len - pos, " %d:%d",
				       array[i].slot, array[i].nodeid);
			if (ret >= len - pos)
				break;
			pos += ret;
		}
	} else if (ro0) {
		for (i = 0; i < num_slots; i++) {
			ret = snprintf(line + pos, len - pos, " %d:%d",
				       ro0[i].ro_slot, ro0[i].ro_nodeid);
			if (ret >= len - pos)
				break;
			pos += ret;
		}
	}

	log_rinfo(ls, "generation %u slots %d%s", gen, num_slots, line);
}

int dlm_slots_copy_in(struct dlm_ls *ls)
{
	struct dlm_member *memb;
	struct dlm_rcom *rc = ls->ls_recover_buf;
	struct rcom_config *rf = (struct rcom_config *)rc->rc_buf;
	struct rcom_slot *ro0, *ro;
	int our_nodeid = dlm_our_nodeid();
	int i, num_slots;
	uint32_t gen;

	if (!dlm_slots_version(&rc->rc_header))
		return -1;

	gen = le32_to_cpu(rf->rf_generation);
	if (gen <= ls->ls_generation) {
		log_error(ls, "dlm_slots_copy_in gen %u old %u",
			  gen, ls->ls_generation);
	}
	ls->ls_generation = gen;

	num_slots = le16_to_cpu(rf->rf_num_slots);
	if (!num_slots)
		return -1;

	ro0 = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config));

	for (i = 0, ro = ro0; i < num_slots; i++, ro++) {
		ro->ro_nodeid = le32_to_cpu(ro->ro_nodeid);
		ro->ro_slot = le16_to_cpu(ro->ro_slot);
	}

	log_slots(ls, gen, num_slots, ro0, NULL, 0);

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		for (i = 0, ro = ro0; i < num_slots; i++, ro++) {
			if (ro->ro_nodeid != memb->nodeid)
				continue;
			memb->slot = ro->ro_slot;
			memb->slot_prev = memb->slot;
			break;
		}

		if (memb->nodeid == our_nodeid) {
			if (ls->ls_slot && ls->ls_slot != memb->slot) {
				log_error(ls, "dlm_slots_copy_in our slot "
					  "changed %d %d", ls->ls_slot,
					  memb->slot);
				return -1;
			}

			if (!ls->ls_slot)
				ls->ls_slot = memb->slot;
		}

		if (!memb->slot) {
			log_error(ls, "dlm_slots_copy_in nodeid %d no slot",
				   memb->nodeid);
			return -1;
		}
	}

	return 0;
}

/* for any nodes that do not support slots, we will not have set memb->slot
   in wait_status_all(), so memb->slot will remain -1, and we will not
   assign slots or set ls_num_slots here */

int dlm_slots_assign(struct dlm_ls *ls, int *num_slots, int *slots_size,
		     struct dlm_slot **slots_out, uint32_t *gen_out)
{
	struct dlm_member *memb;
	struct dlm_slot *array;
	int our_nodeid = dlm_our_nodeid();
	int array_size, max_slots, i;
	int need = 0;
	int max = 0;
	int num = 0;
	uint32_t gen = 0;

	/* our own memb struct will have slot -1 gen 0 */

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		if (memb->nodeid == our_nodeid) {
			memb->slot = ls->ls_slot;
			memb->generation = ls->ls_generation;
			break;
		}
	}

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		if (memb->generation > gen)
			gen = memb->generation;

		/* node doesn't support slots */

		if (memb->slot == -1)
			return -1;

		/* node needs a slot assigned */

		if (!memb->slot)
			need++;

		/* node has a slot assigned */

		num++;

		if (!max || max < memb->slot)
			max = memb->slot;

		/* sanity check, once slot is assigned it shouldn't change */

		if (memb->slot_prev && memb->slot && memb->slot_prev != memb->slot) {
			log_error(ls, "nodeid %d slot changed %d %d",
				  memb->nodeid, memb->slot_prev, memb->slot);
			return -1;
		}
		memb->slot_prev = memb->slot;
	}

	array_size = max + need;
	array = kcalloc(array_size, sizeof(*array), GFP_NOFS);
	if (!array)
		return -ENOMEM;

	num = 0;

	/* fill in slots (offsets) that are used */

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

		if (memb->slot > array_size) {
			log_error(ls, "invalid slot number %d", memb->slot);
			kfree(array);
			return -1;
		}

		array[memb->slot - 1].nodeid = memb->nodeid;
		array[memb->slot - 1].slot = memb->slot;
		num++;
	}

	/* assign new slots from unused offsets */

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		if (memb->slot)
			continue;

		for (i = 0; i < array_size; i++) {
			if (array[i].nodeid)
				continue;

			memb->slot = i + 1;
			memb->slot_prev = memb->slot;
			array[i].nodeid = memb->nodeid;
			array[i].slot = memb->slot;
			num++;

			if (!ls->ls_slot && memb->nodeid == our_nodeid)
				ls->ls_slot = memb->slot;
			break;
		}

		if (!memb->slot) {
			log_error(ls, "no free slot found");
			kfree(array);
			return -1;
		}
	}

	gen++;

	log_slots(ls, gen, num, NULL, array, array_size);

	max_slots = (dlm_config.ci_buffer_size - sizeof(struct dlm_rcom) -
		     sizeof(struct rcom_config)) / sizeof(struct rcom_slot);

	if (num > max_slots) {
		log_error(ls, "num_slots %d exceeds max_slots %d",
			  num, max_slots);
		kfree(array);
		return -1;
	}

	*gen_out = gen;
	*slots_out = array;
	*slots_size = array_size;
	*num_slots = num;
	return 0;
}

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, struct dlm_config_node *node)
{
	struct dlm_member *memb;
	int error;

	memb = kzalloc(sizeof(*memb), GFP_NOFS);
	if (!memb)
		return -ENOMEM;

	error = dlm_lowcomms_connect_node(node->nodeid);
	if (error < 0) {
		kfree(memb);
		return error;
	}

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

static struct dlm_member *find_memb(struct list_head *head, int nodeid)
{
	struct dlm_member *memb;

	list_for_each_entry(memb, head, list) {
		if (memb->nodeid == nodeid)
			return memb;
	}
	return NULL;
}

int dlm_is_member(struct dlm_ls *ls, int nodeid)
{
	if (find_memb(&ls->ls_nodes, nodeid))
		return 1;
	return 0;
}

int dlm_is_removed(struct dlm_ls *ls, int nodeid)
{
	if (find_memb(&ls->ls_nodes_gone, 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_array(total, sizeof(*array), GFP_NOFS);
	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, 0);
		if (error)
			break;
	}
	if (error)
		log_rinfo(ls, "ping_members aborted %d last nodeid %d",
			  error, ls->ls_recover_nodeid);
	return error;
}

static void dlm_lsop_recover_prep(struct dlm_ls *ls)
{
	if (!ls->ls_ops || !ls->ls_ops->recover_prep)
		return;
	ls->ls_ops->recover_prep(ls->ls_ops_arg);
}

static void dlm_lsop_recover_slot(struct dlm_ls *ls, struct dlm_member *memb)
{
	struct dlm_slot slot;
	uint32_t seq;
	int error;

	if (!ls->ls_ops || !ls->ls_ops->recover_slot)
		return;

	/* if there is no comms connection with this node
	   or the present comms connection is newer
	   than the one when this member was added, then
	   we consider the node to have failed (versus
	   being removed due to dlm_release_lockspace) */

	error = dlm_comm_seq(memb->nodeid, &seq);

	if (!error && seq == memb->comm_seq)
		return;

	slot.nodeid = memb->nodeid;
	slot.slot = memb->slot;

	ls->ls_ops->recover_slot(ls->ls_ops_arg, &slot);
}

void dlm_lsop_recover_done(struct dlm_ls *ls)
{
	struct dlm_member *memb;
	struct dlm_slot *slots;
	int i, num;

	if (!ls->ls_ops || !ls->ls_ops->recover_done)
		return;

	num = ls->ls_num_nodes;
	slots = kcalloc(num, sizeof(*slots), GFP_KERNEL);
	if (!slots)
		return;

	i = 0;
	list_for_each_entry(memb, &ls->ls_nodes, list) {
		if (i == num) {
			log_error(ls, "dlm_lsop_recover_done bad num %d", num);
			goto out;
		}
		slots[i].nodeid = memb->nodeid;
		slots[i].slot = memb->slot;
		i++;
	}

	ls->ls_ops->recover_done(ls->ls_ops_arg, slots, num,
				 ls->ls_slot, ls->ls_generation);
 out:
	kfree(slots);
}

static struct dlm_config_node *find_config_node(struct dlm_recover *rv,
						int nodeid)
{
	int i;

	for (i = 0; i < rv->nodes_count; i++) {
		if (rv->nodes[i].nodeid == nodeid)
			return &rv->nodes[i];
	}
	return NULL;
}

int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
{
	struct dlm_member *memb, *safe;
	struct dlm_config_node *node;
	int i, error, 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_rinfo(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) {
		node = find_config_node(rv, memb->nodeid);
		if (node && !node->new)
			continue;

		if (!node) {
			log_rinfo(ls, "remove member %d", memb->nodeid);
		} else {
			/* removed and re-added */
			log_rinfo(ls, "remove member %d comm_seq %u %u",
				  memb->nodeid, memb->comm_seq, node->comm_seq);
		}

		neg++;
		list_move(&memb->list, &ls->ls_nodes_gone);
		ls->ls_num_nodes--;
		dlm_lsop_recover_slot(ls, memb);
	}

	/* add new members to ls_nodes */

	for (i = 0; i < rv->nodes_count; i++) {
		node = &rv->nodes[i];
		if (dlm_is_member(ls, node->nodeid))
			continue;
		dlm_add_member(ls, node);
		log_rinfo(ls, "add member %d", node->nodeid);
	}

	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);
	*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);
	}

	log_rinfo(ls, "dlm_recover_members %d nodes", ls->ls_num_nodes);
	return error;
}

/* Userspace guarantees that dlm_ls_stop() has completed on all nodes before
   dlm_ls_start() is called on any of them to start the new recovery. */

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

	/*
	 * Prevent dlm_recv from being in the middle of something when we do
	 * the stop.  This includes ensuring dlm_recv isn't processing a
	 * recovery message (rcom), while dlm_recoverd is aborting and
	 * resetting things from an in-progress recovery.  i.e. we want
	 * dlm_recoverd to abort its recovery without worrying about dlm_recv
	 * processing an rcom at the same time.  Stopping dlm_recv also makes
	 * it easy for dlm_receive_message() to check locking stopped and add a
	 * message to the requestqueue without races.
	 */

	down_write(&ls->ls_recv_active);

	/*
	 * Abort any recovery that's in progress (see RECOVER_STOP,
	 * dlm_recovery_stopped()) and tell any other threads running in the
	 * dlm to quit any processing (see RUNNING, dlm_locking_stopped()).
	 */

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

	/*
	 * Let dlm_recv run again, now any normal messages will be saved on the
	 * requestqueue for later.
	 */

	up_write(&ls->ls_recv_active);

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

	if (new) {
		set_bit(LSFL_RECOVER_DOWN, &ls->ls_flags);
		wake_up_process(ls->ls_recoverd_task);
		wait_event(ls->ls_recover_lock_wait,
			   test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags));
	}

	/*
	 * The recoverd suspend/resume makes sure that dlm_recoverd (if
	 * running) has noticed RECOVER_STOP above and quit processing the
	 * previous recovery.
	 */

	dlm_recoverd_suspend(ls);

	spin_lock(&ls->ls_recover_lock);
	kfree(ls->ls_slots);
	ls->ls_slots = NULL;
	ls->ls_num_slots = 0;
	ls->ls_slots_size = 0;
	ls->ls_recover_status = 0;
	spin_unlock(&ls->ls_recover_lock);

	dlm_recoverd_resume(ls);

	if (!ls->ls_recover_begin)
		ls->ls_recover_begin = jiffies;

	dlm_lsop_recover_prep(ls);
	return 0;
}

int dlm_ls_start(struct dlm_ls *ls)
{
	struct dlm_recover *rv, *rv_old;
	struct dlm_config_node *nodes;
	int error, count;

	rv = kzalloc(sizeof(*rv), GFP_NOFS);
	if (!rv)
		return -ENOMEM;

	error = dlm_config_nodes(ls->ls_name, &nodes, &count);
	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->nodes = nodes;
	rv->nodes_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) {
		log_error(ls, "unused recovery %llx %d",
			  (unsigned long long)rv_old->seq, rv_old->nodes_count);
		kfree(rv_old->nodes);
		kfree(rv_old);
	}

	set_bit(LSFL_RECOVER_WORK, &ls->ls_flags);
	wake_up_process(ls->ls_recoverd_task);
	return 0;

 fail:
	kfree(rv);
	kfree(nodes);
	return error;
}
Commit	Line	Data
e7fd4179 DT	1	/******************************************************************************
	2	*******************************************************************************
	3	**
60f98d18	4	** Copyright (C) 2005-2011 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"
e7fd4179 DT	19	#include "config.h"
391fbdc5	20	#include "lowcomms.h"
e7fd4179	21
757a4271 DT	22	int dlm_slots_version(struct dlm_header *h)
	23	{
	24	if ((h->h_version & 0x0000FFFF) < DLM_HEADER_SLOTS)
	25	return 0;
	26	return 1;
	27	}
	28
	29	void dlm_slot_save(struct dlm_ls ls, struct dlm_rcom rc,
60f98d18	30	struct dlm_member *memb)
757a4271 DT	31	{
	32	struct rcom_config rf = (struct rcom_config )rc->rc_buf;
	33
	34	if (!dlm_slots_version(&rc->rc_header))
	35	return;
	36
	37	memb->slot = le16_to_cpu(rf->rf_our_slot);
	38	memb->generation = le32_to_cpu(rf->rf_generation);
	39	}
	40
	41	void dlm_slots_copy_out(struct dlm_ls ls, struct dlm_rcom rc)
	42	{
	43	struct dlm_slot *slot;
	44	struct rcom_slot *ro;
	45	int i;
	46
	47	ro = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config));
	48
	49	/* ls_slots array is sparse, but not rcom_slots */
	50
	51	for (i = 0; i < ls->ls_slots_size; i++) {
	52	slot = &ls->ls_slots[i];
	53	if (!slot->nodeid)
	54	continue;
	55	ro->ro_nodeid = cpu_to_le32(slot->nodeid);
	56	ro->ro_slot = cpu_to_le16(slot->slot);
	57	ro++;
	58	}
	59	}
	60
	61	#define SLOT_DEBUG_LINE 128
	62
075f0177 DT	63	static void log_slots(struct dlm_ls *ls, uint32_t gen, int num_slots,
	64	struct rcom_slot ro0, struct dlm_slot array,
	65	int array_size)
757a4271 DT	66	{
	67	char line[SLOT_DEBUG_LINE];
	68	int len = SLOT_DEBUG_LINE - 1;
	69	int pos = 0;
	70	int ret, i;
	71
757a4271 DT	72	memset(line, 0, sizeof(line));
	73
	74	if (array) {
	75	for (i = 0; i < array_size; i++) {
	76	if (!array[i].nodeid)
	77	continue;
	78
	79	ret = snprintf(line + pos, len - pos, " %d:%d",
	80	array[i].slot, array[i].nodeid);
	81	if (ret >= len - pos)
	82	break;
	83	pos += ret;
	84	}
	85	} else if (ro0) {
	86	for (i = 0; i < num_slots; i++) {
	87	ret = snprintf(line + pos, len - pos, " %d:%d",
	88	ro0[i].ro_slot, ro0[i].ro_nodeid);
	89	if (ret >= len - pos)
	90	break;
	91	pos += ret;
	92	}
	93	}
	94
075f0177	95	log_rinfo(ls, "generation %u slots %d%s", gen, num_slots, line);
757a4271 DT	96	}
	97
	98	int dlm_slots_copy_in(struct dlm_ls *ls)
	99	{
	100	struct dlm_member *memb;
	101	struct dlm_rcom *rc = ls->ls_recover_buf;
	102	struct rcom_config rf = (struct rcom_config )rc->rc_buf;
	103	struct rcom_slot ro0, ro;
	104	int our_nodeid = dlm_our_nodeid();
	105	int i, num_slots;
	106	uint32_t gen;
	107
	108	if (!dlm_slots_version(&rc->rc_header))
	109	return -1;
	110
	111	gen = le32_to_cpu(rf->rf_generation);
	112	if (gen <= ls->ls_generation) {
	113	log_error(ls, "dlm_slots_copy_in gen %u old %u",
	114	gen, ls->ls_generation);
	115	}
	116	ls->ls_generation = gen;
	117
	118	num_slots = le16_to_cpu(rf->rf_num_slots);
	119	if (!num_slots)
	120	return -1;
	121
	122	ro0 = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config));
	123
	124	for (i = 0, ro = ro0; i < num_slots; i++, ro++) {
	125	ro->ro_nodeid = le32_to_cpu(ro->ro_nodeid);
	126	ro->ro_slot = le16_to_cpu(ro->ro_slot);
	127	}
	128
075f0177	129	log_slots(ls, gen, num_slots, ro0, NULL, 0);
757a4271 DT	130
	131	list_for_each_entry(memb, &ls->ls_nodes, list) {
	132	for (i = 0, ro = ro0; i < num_slots; i++, ro++) {
	133	if (ro->ro_nodeid != memb->nodeid)
	134	continue;
	135	memb->slot = ro->ro_slot;
	136	memb->slot_prev = memb->slot;
	137	break;
	138	}
	139
	140	if (memb->nodeid == our_nodeid) {
	141	if (ls->ls_slot && ls->ls_slot != memb->slot) {
	142	log_error(ls, "dlm_slots_copy_in our slot "
	143	"changed %d %d", ls->ls_slot,
	144	memb->slot);
	145	return -1;
	146	}
	147
	148	if (!ls->ls_slot)
	149	ls->ls_slot = memb->slot;
	150	}
	151
	152	if (!memb->slot) {
	153	log_error(ls, "dlm_slots_copy_in nodeid %d no slot",
	154	memb->nodeid);
	155	return -1;
	156	}
	157	}
	158
	159	return 0;
	160	}
	161
	162	/* for any nodes that do not support slots, we will not have set memb->slot
	163	in wait_status_all(), so memb->slot will remain -1, and we will not
	164	assign slots or set ls_num_slots here */
	165
	166	int dlm_slots_assign(struct dlm_ls ls, int num_slots, int *slots_size,
	167	struct dlm_slot *slots_out, uint32_t gen_out)
	168	{
	169	struct dlm_member *memb;
	170	struct dlm_slot *array;
	171	int our_nodeid = dlm_our_nodeid();
	172	int array_size, max_slots, i;
	173	int need = 0;
	174	int max = 0;
	175	int num = 0;
	176	uint32_t gen = 0;
	177
	178	/* our own memb struct will have slot -1 gen 0 */
	179
	180	list_for_each_entry(memb, &ls->ls_nodes, list) {
	181	if (memb->nodeid == our_nodeid) {
	182	memb->slot = ls->ls_slot;
	183	memb->generation = ls->ls_generation;
	184	break;
	185	}
	186	}
	187
	188	list_for_each_entry(memb, &ls->ls_nodes, list) {
	189	if (memb->generation > gen)
	190	gen = memb->generation;
	191
	192	/* node doesn't support slots */
	193
194	if (memb->slot == -1)
195	return -1;
196
197	/* node needs a slot assigned */
198
199	if (!memb->slot)
200	need++;
201
202	/* node has a slot assigned */
203
204	num++;
205
206	if (!max \|\| max < memb->slot)
207	max = memb->slot;
208
209	/* sanity check, once slot is assigned it shouldn't change */
210
211	if (memb->slot_prev && memb->slot && memb->slot_prev != memb->slot) {
212	log_error(ls, "nodeid %d slot changed %d %d",
213	memb->nodeid, memb->slot_prev, memb->slot);
214	return -1;
215	}
216	memb->slot_prev = memb->slot;
217	}
218
219	array_size = max + need;
2f48e061	220	array = kcalloc(array_size, sizeof(*array), GFP_NOFS);
757a4271 DT	221	if (!array)
	222	return -ENOMEM;
	223
	224	num = 0;
	225
	226	/* fill in slots (offsets) that are used */
	227
	228	list_for_each_entry(memb, &ls->ls_nodes, list) {
	229	if (!memb->slot)
	230	continue;
	231
	232	if (memb->slot > array_size) {
	233	log_error(ls, "invalid slot number %d", memb->slot);
	234	kfree(array);
	235	return -1;
	236	}
	237
	238	array[memb->slot - 1].nodeid = memb->nodeid;
	239	array[memb->slot - 1].slot = memb->slot;
	240	num++;
	241	}
	242
	243	/* assign new slots from unused offsets */
	244
	245	list_for_each_entry(memb, &ls->ls_nodes, list) {
	246	if (memb->slot)
	247	continue;
	248
	249	for (i = 0; i < array_size; i++) {
	250	if (array[i].nodeid)
	251	continue;
	252
	253	memb->slot = i + 1;
	254	memb->slot_prev = memb->slot;
	255	array[i].nodeid = memb->nodeid;
	256	array[i].slot = memb->slot;
	257	num++;
	258
	259	if (!ls->ls_slot && memb->nodeid == our_nodeid)
	260	ls->ls_slot = memb->slot;
	261	break;
	262	}
	263
	264	if (!memb->slot) {
	265	log_error(ls, "no free slot found");
	266	kfree(array);
	267	return -1;
	268	}
	269	}
	270
	271	gen++;
	272
075f0177	273	log_slots(ls, gen, num, NULL, array, array_size);
757a4271 DT	274
	275	max_slots = (dlm_config.ci_buffer_size - sizeof(struct dlm_rcom) -
	276	sizeof(struct rcom_config)) / sizeof(struct rcom_slot);
	277
	278	if (num > max_slots) {
	279	log_error(ls, "num_slots %d exceeds max_slots %d",
	280	num, max_slots);
	281	kfree(array);
	282	return -1;
	283	}
	284
	285	*gen_out = gen;
	286	*slots_out = array;
	287	*slots_size = array_size;
	288	*num_slots = num;
	289	return 0;
	290	}
	291
e7fd4179 DT	292	static void add_ordered_member(struct dlm_ls ls, struct dlm_member new)
	293	{
	294	struct dlm_member *memb = NULL;
	295	struct list_head *tmp;
	296	struct list_head *newlist = &new->list;
	297	struct list_head *head = &ls->ls_nodes;
	298
	299	list_for_each(tmp, head) {
	300	memb = list_entry(tmp, struct dlm_member, list);
	301	if (new->nodeid < memb->nodeid)
	302	break;
	303	}
	304
	305	if (!memb)
	306	list_add_tail(newlist, head);
	307	else {
	308	/* FIXME: can use list macro here */
	309	newlist->prev = tmp->prev;
	310	newlist->next = tmp;
	311	tmp->prev->next = newlist;
	312	tmp->prev = newlist;
	313	}
	314	}
	315
60f98d18	316	static int dlm_add_member(struct dlm_ls ls, struct dlm_config_node node)
e7fd4179 DT	317	{
e7fd4179 DT	318	struct dlm_member *memb;
60f98d18	319	int error;
e7fd4179	320
d12ad1a9	321	memb = kzalloc(sizeof(*memb), GFP_NOFS);
e7fd4179 DT	322	if (!memb)
	323	return -ENOMEM;
	324
60f98d18	325	error = dlm_lowcomms_connect_node(node->nodeid);
391fbdc5 CC	326	if (error < 0) {
	327	kfree(memb);
	328	return error;
	329	}
	330
60f98d18 DT	331	memb->nodeid = node->nodeid;
	332	memb->weight = node->weight;
	333	memb->comm_seq = node->comm_seq;
e7fd4179 DT	334	add_ordered_member(ls, memb);
	335	ls->ls_num_nodes++;
	336	return 0;
	337	}
	338
60f98d18	339	static struct dlm_member find_memb(struct list_head head, int nodeid)
e7fd4179 DT	340	{
	341	struct dlm_member *memb;
	342
60f98d18	343	list_for_each_entry(memb, head, list) {
e7fd4179	344	if (memb->nodeid == nodeid)
60f98d18	345	return memb;
e7fd4179	346	}
60f98d18 DT	347	return NULL;
	348	}
	349
	350	int dlm_is_member(struct dlm_ls *ls, int nodeid)
	351	{
	352	if (find_memb(&ls->ls_nodes, nodeid))
	353	return 1;
90135925	354	return 0;
e7fd4179 DT	355	}
	356
	357	int dlm_is_removed(struct dlm_ls *ls, int nodeid)
	358	{
60f98d18 DT	359	if (find_memb(&ls->ls_nodes_gone, nodeid))
60f98d18 DT	360	return 1;
90135925	361	return 0;
e7fd4179 DT	362	}
	363
	364	static void clear_memb_list(struct list_head *head)
	365	{
	366	struct dlm_member *memb;
	367
	368	while (!list_empty(head)) {
	369	memb = list_entry(head->next, struct dlm_member, list);
	370	list_del(&memb->list);
	371	kfree(memb);
	372	}
	373	}
	374
	375	void dlm_clear_members(struct dlm_ls *ls)
	376	{
	377	clear_memb_list(&ls->ls_nodes);
	378	ls->ls_num_nodes = 0;
	379	}
	380
	381	void dlm_clear_members_gone(struct dlm_ls *ls)
	382	{
	383	clear_memb_list(&ls->ls_nodes_gone);
	384	}
	385
	386	static void make_member_array(struct dlm_ls *ls)
	387	{
	388	struct dlm_member *memb;
	389	int i, w, x = 0, total = 0, all_zero = 0, *array;
	390
	391	kfree(ls->ls_node_array);
	392	ls->ls_node_array = NULL;
	393
	394	list_for_each_entry(memb, &ls->ls_nodes, list) {
	395	if (memb->weight)
	396	total += memb->weight;
	397	}
	398
	399	/* all nodes revert to weight of 1 if all have weight 0 */
	400
	401	if (!total) {
	402	total = ls->ls_num_nodes;
	403	all_zero = 1;
	404	}
	405
	406	ls->ls_total_weight = total;
790854be	407	array = kmalloc_array(total, sizeof(*array), GFP_NOFS);
e7fd4179 DT	408	if (!array)
	409	return;
	410
	411	list_for_each_entry(memb, &ls->ls_nodes, list) {
	412	if (!all_zero && !memb->weight)
	413	continue;
	414
	415	if (all_zero)
	416	w = 1;
	417	else
	418	w = memb->weight;
	419
	420	DLM_ASSERT(x < total, printk("total %d x %d\n", total, x););
	421
	422	for (i = 0; i < w; i++)
	423	array[x++] = memb->nodeid;
	424	}
	425
	426	ls->ls_node_array = array;
	427	}
	428
	429	/* send a status request to all members just to establish comms connections */
	430
f6db1b8e	431	static int ping_members(struct dlm_ls *ls)
e7fd4179 DT	432	{
e7fd4179 DT	433	struct dlm_member *memb;
f6db1b8e DT	434	int error = 0;
	435
	436	list_for_each_entry(memb, &ls->ls_nodes, list) {
	437	error = dlm_recovery_stopped(ls);
	438	if (error)
	439	break;
757a4271	440	error = dlm_rcom_status(ls, memb->nodeid, 0);
f6db1b8e DT	441	if (error)
	442	break;
	443	}
	444	if (error)
075f0177	445	log_rinfo(ls, "ping_members aborted %d last nodeid %d",
faa0f267	446	error, ls->ls_recover_nodeid);
f6db1b8e	447	return error;
e7fd4179 DT	448	}
e7fd4179 DT	449
60f98d18 DT	450	static void dlm_lsop_recover_prep(struct dlm_ls *ls)
	451	{
	452	if (!ls->ls_ops \|\| !ls->ls_ops->recover_prep)
	453	return;
	454	ls->ls_ops->recover_prep(ls->ls_ops_arg);
	455	}
	456
	457	static void dlm_lsop_recover_slot(struct dlm_ls ls, struct dlm_member memb)
	458	{
	459	struct dlm_slot slot;
	460	uint32_t seq;
	461	int error;
	462
	463	if (!ls->ls_ops \|\| !ls->ls_ops->recover_slot)
	464	return;
	465
	466	/* if there is no comms connection with this node
	467	or the present comms connection is newer
	468	than the one when this member was added, then
	469	we consider the node to have failed (versus
	470	being removed due to dlm_release_lockspace) */
	471
	472	error = dlm_comm_seq(memb->nodeid, &seq);
	473
	474	if (!error && seq == memb->comm_seq)
	475	return;
	476
	477	slot.nodeid = memb->nodeid;
	478	slot.slot = memb->slot;
	479
	480	ls->ls_ops->recover_slot(ls->ls_ops_arg, &slot);
	481	}
	482
	483	void dlm_lsop_recover_done(struct dlm_ls *ls)
	484	{
	485	struct dlm_member *memb;
	486	struct dlm_slot *slots;
	487	int i, num;
	488
	489	if (!ls->ls_ops \|\| !ls->ls_ops->recover_done)
	490	return;
	491
	492	num = ls->ls_num_nodes;
2f48e061	493	slots = kcalloc(num, sizeof(*slots), GFP_KERNEL);
60f98d18 DT	494	if (!slots)
	495	return;
	496
	497	i = 0;
	498	list_for_each_entry(memb, &ls->ls_nodes, list) {
	499	if (i == num) {
	500	log_error(ls, "dlm_lsop_recover_done bad num %d", num);
	501	goto out;
	502	}
	503	slots[i].nodeid = memb->nodeid;
	504	slots[i].slot = memb->slot;
	505	i++;
	506	}
	507
	508	ls->ls_ops->recover_done(ls->ls_ops_arg, slots, num,
	509	ls->ls_slot, ls->ls_generation);
	510	out:
	511	kfree(slots);
	512	}
	513
	514	static struct dlm_config_node find_config_node(struct dlm_recover rv,
	515	int nodeid)
	516	{
	517	int i;
	518
	519	for (i = 0; i < rv->nodes_count; i++) {
	520	if (rv->nodes[i].nodeid == nodeid)
	521	return &rv->nodes[i];
	522	}
	523	return NULL;
	524	}
	525
e7fd4179 DT	526	int dlm_recover_members(struct dlm_ls ls, struct dlm_recover rv, int *neg_out)
	527	{
	528	struct dlm_member memb, safe;
60f98d18 DT	529	struct dlm_config_node *node;
60f98d18 DT	530	int i, error, neg = 0, low = -1;
e7fd4179	531
91c0dc93 DT	532	/* previously removed members that we've not finished removing need to
	533	count as a negative change so the "neg" recovery steps will happen */
	534
	535	list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
075f0177	536	log_rinfo(ls, "prev removed member %d", memb->nodeid);
91c0dc93 DT	537	neg++;
	538	}
	539
e7fd4179 DT	540	/* move departed members from ls_nodes to ls_nodes_gone */
	541
	542	list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
60f98d18 DT	543	node = find_config_node(rv, memb->nodeid);
	544	if (node && !node->new)
	545	continue;
e7fd4179	546
60f98d18	547	if (!node) {
075f0177	548	log_rinfo(ls, "remove member %d", memb->nodeid);
60f98d18 DT	549	} else {
60f98d18 DT	550	/* removed and re-added */
075f0177	551	log_rinfo(ls, "remove member %d comm_seq %u %u",
60f98d18	552	memb->nodeid, memb->comm_seq, node->comm_seq);
e7fd4179	553	}
d44e0fc7	554
d44e0fc7	555	neg++;
60f98d18 DT	556	list_move(&memb->list, &ls->ls_nodes_gone);
	557	ls->ls_num_nodes--;
	558	dlm_lsop_recover_slot(ls, memb);
d44e0fc7 DT	559	}
d44e0fc7 DT	560
e7fd4179 DT	561	/* add new members to ls_nodes */
e7fd4179 DT	562
60f98d18 DT	563	for (i = 0; i < rv->nodes_count; i++) {
	564	node = &rv->nodes[i];
	565	if (dlm_is_member(ls, node->nodeid))
e7fd4179	566	continue;
60f98d18	567	dlm_add_member(ls, node);
075f0177	568	log_rinfo(ls, "add member %d", node->nodeid);
e7fd4179 DT	569	}
	570
	571	list_for_each_entry(memb, &ls->ls_nodes, list) {
	572	if (low == -1 \|\| memb->nodeid < low)
	573	low = memb->nodeid;
	574	}
	575	ls->ls_low_nodeid = low;
	576
	577	make_member_array(ls);
e7fd4179 DT	578	*neg_out = neg;
e7fd4179 DT	579
f6db1b8e	580	error = ping_members(ls);
8b0e7b2c DT	581	if (!error \|\| error == -EPROTO) {
	582	/* new_lockspace() may be waiting to know if the config
	583	is good or bad */
	584	ls->ls_members_result = error;
	585	complete(&ls->ls_members_done);
	586	}
e7fd4179	587
075f0177	588	log_rinfo(ls, "dlm_recover_members %d nodes", ls->ls_num_nodes);
e7fd4179 DT	589	return error;
	590	}
	591
c36258b5 DT	592	/* Userspace guarantees that dlm_ls_stop() has completed on all nodes before
c36258b5 DT	593	dlm_ls_start() is called on any of them to start the new recovery. */
e7fd4179 DT	594
	595	int dlm_ls_stop(struct dlm_ls *ls)
	596	{
	597	int new;
	598
	599	/*
c36258b5 DT	600	* Prevent dlm_recv from being in the middle of something when we do
	601	* the stop. This includes ensuring dlm_recv isn't processing a
	602	* recovery message (rcom), while dlm_recoverd is aborting and
	603	* resetting things from an in-progress recovery. i.e. we want
	604	* dlm_recoverd to abort its recovery without worrying about dlm_recv
	605	* processing an rcom at the same time. Stopping dlm_recv also makes
	606	* it easy for dlm_receive_message() to check locking stopped and add a
	607	* message to the requestqueue without races.
	608	*/
	609
	610	down_write(&ls->ls_recv_active);
	611
	612	/*
475f230c	613	* Abort any recovery that's in progress (see RECOVER_STOP,
c36258b5 DT	614	* dlm_recovery_stopped()) and tell any other threads running in the
c36258b5 DT	615	* dlm to quit any processing (see RUNNING, dlm_locking_stopped()).
e7fd4179 DT	616	*/
	617
	618	spin_lock(&ls->ls_recover_lock);
475f230c	619	set_bit(LSFL_RECOVER_STOP, &ls->ls_flags);
e7fd4179 DT	620	new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
	621	ls->ls_recover_seq++;
	622	spin_unlock(&ls->ls_recover_lock);
	623
c36258b5 DT	624	/*
	625	* Let dlm_recv run again, now any normal messages will be saved on the
	626	* requestqueue for later.
	627	*/
	628
	629	up_write(&ls->ls_recv_active);
	630
e7fd4179 DT	631	/*
e7fd4179 DT	632	* This in_recovery lock does two things:
e7fd4179	633	* 1) Keeps this function from returning until all threads are out
c41b20e7	634	* of locking routines and locking is truly stopped.
e7fd4179 DT	635	* 2) Keeps any new requests from being processed until it's unlocked
	636	* when recovery is complete.
	637	*/
	638
475f230c DT	639	if (new) {
	640	set_bit(LSFL_RECOVER_DOWN, &ls->ls_flags);
	641	wake_up_process(ls->ls_recoverd_task);
	642	wait_event(ls->ls_recover_lock_wait,
	643	test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags));
	644	}
e7fd4179 DT	645
	646	/*
	647	* The recoverd suspend/resume makes sure that dlm_recoverd (if
475f230c	648	* running) has noticed RECOVER_STOP above and quit processing the
c36258b5	649	* previous recovery.
e7fd4179 DT	650	*/
	651
	652	dlm_recoverd_suspend(ls);
757a4271 DT	653
	654	spin_lock(&ls->ls_recover_lock);
	655	kfree(ls->ls_slots);
	656	ls->ls_slots = NULL;
	657	ls->ls_num_slots = 0;
	658	ls->ls_slots_size = 0;
e7fd4179	659	ls->ls_recover_status = 0;
757a4271 DT	660	spin_unlock(&ls->ls_recover_lock);
757a4271 DT	661
e7fd4179	662	dlm_recoverd_resume(ls);
3ae1acf9 DT	663
	664	if (!ls->ls_recover_begin)
	665	ls->ls_recover_begin = jiffies;
60f98d18 DT	666
60f98d18 DT	667	dlm_lsop_recover_prep(ls);
e7fd4179 DT	668	return 0;
	669	}
	670
	671	int dlm_ls_start(struct dlm_ls *ls)
	672	{
2ab93ae1	673	struct dlm_recover rv, rv_old;
60f98d18 DT	674	struct dlm_config_node *nodes;
60f98d18 DT	675	int error, count;
e7fd4179	676
d12ad1a9	677	rv = kzalloc(sizeof(*rv), GFP_NOFS);
e7fd4179 DT	678	if (!rv)
e7fd4179 DT	679	return -ENOMEM;
e7fd4179	680
60f98d18	681	error = dlm_config_nodes(ls->ls_name, &nodes, &count);
d44e0fc7	682	if (error < 0)
e7fd4179 DT	683	goto fail;
	684
	685	spin_lock(&ls->ls_recover_lock);
	686
	687	/* the lockspace needs to be stopped before it can be started */
	688
	689	if (!dlm_locking_stopped(ls)) {
	690	spin_unlock(&ls->ls_recover_lock);
	691	log_error(ls, "start ignored: lockspace running");
	692	error = -EINVAL;
	693	goto fail;
	694	}
	695
60f98d18 DT	696	rv->nodes = nodes;
60f98d18 DT	697	rv->nodes_count = count;
e7fd4179 DT	698	rv->seq = ++ls->ls_recover_seq;
	699	rv_old = ls->ls_recover_args;
	700	ls->ls_recover_args = rv;
	701	spin_unlock(&ls->ls_recover_lock);
	702
	703	if (rv_old) {
d44e0fc7	704	log_error(ls, "unused recovery %llx %d",
60f98d18 DT	705	(unsigned long long)rv_old->seq, rv_old->nodes_count);
60f98d18 DT	706	kfree(rv_old->nodes);
e7fd4179 DT	707	kfree(rv_old);
	708	}
	709
475f230c DT	710	set_bit(LSFL_RECOVER_WORK, &ls->ls_flags);
475f230c DT	711	wake_up_process(ls->ls_recoverd_task);
e7fd4179 DT	712	return 0;
	713
	714	fail:
	715	kfree(rv);
60f98d18	716	kfree(nodes);
e7fd4179 DT	717	return error;
	718	}
	719