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

/******************************************************************************
*******************************************************************************
**
**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
**  Copyright (C) 2005-2008 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 "lowcomms.h"
#include "midcomms.h"
#include "rcom.h"
#include "recover.h"
#include "dir.h"
#include "config.h"
#include "memory.h"
#include "lock.h"
#include "util.h"
#include "member.h"


static int rcom_response(struct dlm_ls *ls)
{
	return test_bit(LSFL_RCOM_READY, &ls->ls_flags);
}

static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len,
		       struct dlm_rcom **rc_ret, struct dlm_mhandle **mh_ret)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	char *mb;
	int mb_len = sizeof(struct dlm_rcom) + len;

	mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
	if (!mh) {
		log_print("create_rcom to %d type %d len %d ENOBUFS",
			  to_nodeid, type, len);
		return -ENOBUFS;
	}
	memset(mb, 0, mb_len);

	rc = (struct dlm_rcom *) mb;

	rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
	rc->rc_header.h_lockspace = ls->ls_global_id;
	rc->rc_header.h_nodeid = dlm_our_nodeid();
	rc->rc_header.h_length = mb_len;
	rc->rc_header.h_cmd = DLM_RCOM;

	rc->rc_type = type;

	spin_lock(&ls->ls_recover_lock);
	rc->rc_seq = ls->ls_recover_seq;
	spin_unlock(&ls->ls_recover_lock);

	*mh_ret = mh;
	*rc_ret = rc;
	return 0;
}

static void send_rcom(struct dlm_ls *ls, struct dlm_mhandle *mh,
		      struct dlm_rcom *rc)
{
	dlm_rcom_out(rc);
	dlm_lowcomms_commit_buffer(mh);
}

static void set_rcom_status(struct dlm_ls *ls, struct rcom_status *rs,
			    uint32_t flags)
{
	rs->rs_flags = cpu_to_le32(flags);
}

/* When replying to a status request, a node also sends back its
   configuration values.  The requesting node then checks that the remote
   node is configured the same way as itself. */

static void set_rcom_config(struct dlm_ls *ls, struct rcom_config *rf,
			    uint32_t num_slots)
{
	rf->rf_lvblen = cpu_to_le32(ls->ls_lvblen);
	rf->rf_lsflags = cpu_to_le32(ls->ls_exflags);

	rf->rf_our_slot = cpu_to_le16(ls->ls_slot);
	rf->rf_num_slots = cpu_to_le16(num_slots);
	rf->rf_generation =  cpu_to_le32(ls->ls_generation);
}

static int check_rcom_config(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid)
{
	struct rcom_config *rf = (struct rcom_config *) rc->rc_buf;

	if ((rc->rc_header.h_version & 0xFFFF0000) != DLM_HEADER_MAJOR) {
		log_error(ls, "version mismatch: %x nodeid %d: %x",
			  DLM_HEADER_MAJOR | DLM_HEADER_MINOR, nodeid,
			  rc->rc_header.h_version);
		return -EPROTO;
	}

	if (le32_to_cpu(rf->rf_lvblen) != ls->ls_lvblen ||
	    le32_to_cpu(rf->rf_lsflags) != ls->ls_exflags) {
		log_error(ls, "config mismatch: %d,%x nodeid %d: %d,%x",
			  ls->ls_lvblen, ls->ls_exflags, nodeid,
			  le32_to_cpu(rf->rf_lvblen),
			  le32_to_cpu(rf->rf_lsflags));
		return -EPROTO;
	}
	return 0;
}

static void allow_sync_reply(struct dlm_ls *ls, uint64_t *new_seq)
{
	spin_lock(&ls->ls_rcom_spin);
	*new_seq = ++ls->ls_rcom_seq;
	set_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	spin_unlock(&ls->ls_rcom_spin);
}

static void disallow_sync_reply(struct dlm_ls *ls)
{
	spin_lock(&ls->ls_rcom_spin);
	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
	spin_unlock(&ls->ls_rcom_spin);
}

/*
 * low nodeid gathers one slot value at a time from each node.
 * it sets need_slots=0, and saves rf_our_slot returned from each
 * rcom_config.
 *
 * other nodes gather all slot values at once from the low nodeid.
 * they set need_slots=1, and ignore the rf_our_slot returned from each
 * rcom_config.  they use the rf_num_slots returned from the low
 * node's rcom_config.
 */

int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	int error = 0;

	ls->ls_recover_nodeid = nodeid;

	if (nodeid == dlm_our_nodeid()) {
		rc = ls->ls_recover_buf;
		rc->rc_result = dlm_recover_status(ls);
		goto out;
	}

	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS,
			    sizeof(struct rcom_status), &rc, &mh);
	if (error)
		goto out;

	set_rcom_status(ls, (struct rcom_status *)rc->rc_buf, status_flags);

	allow_sync_reply(ls, &rc->rc_id);
	memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);

	send_rcom(ls, mh, rc);

	error = dlm_wait_function(ls, &rcom_response);
	disallow_sync_reply(ls);
	if (error)
		goto out;

	rc = ls->ls_recover_buf;

	if (rc->rc_result == -ESRCH) {
		/* we pretend the remote lockspace exists with 0 status */
		log_debug(ls, "remote node %d not ready", nodeid);
		rc->rc_result = 0;
		error = 0;
	} else {
		error = check_rcom_config(ls, rc, nodeid);
	}

	/* the caller looks at rc_result for the remote recovery status */
 out:
	return error;
}

static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	struct rcom_status *rs;
	uint32_t status;
	int nodeid = rc_in->rc_header.h_nodeid;
	int len = sizeof(struct rcom_config);
	int num_slots = 0;
	int error;

	if (!dlm_slots_version(&rc_in->rc_header)) {
		status = dlm_recover_status(ls);
		goto do_create;
	}

	rs = (struct rcom_status *)rc_in->rc_buf;

	if (!(rs->rs_flags & DLM_RSF_NEED_SLOTS)) {
		status = dlm_recover_status(ls);
		goto do_create;
	}

	spin_lock(&ls->ls_recover_lock);
	status = ls->ls_recover_status;
	num_slots = ls->ls_num_slots;
	spin_unlock(&ls->ls_recover_lock);
	len += num_slots * sizeof(struct rcom_slot);

 do_create:
	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY,
			    len, &rc, &mh);
	if (error)
		return;

	rc->rc_id = rc_in->rc_id;
	rc->rc_seq_reply = rc_in->rc_seq;
	rc->rc_result = status;

	set_rcom_config(ls, (struct rcom_config *)rc->rc_buf, num_slots);

	if (!num_slots)
		goto do_send;

	spin_lock(&ls->ls_recover_lock);
	if (ls->ls_num_slots != num_slots) {
		spin_unlock(&ls->ls_recover_lock);
		log_debug(ls, "receive_rcom_status num_slots %d to %d",
			  num_slots, ls->ls_num_slots);
		rc->rc_result = 0;
		set_rcom_config(ls, (struct rcom_config *)rc->rc_buf, 0);
		goto do_send;
	}

	dlm_slots_copy_out(ls, rc);
	spin_unlock(&ls->ls_recover_lock);

 do_send:
	send_rcom(ls, mh, rc);
}

static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	spin_lock(&ls->ls_rcom_spin);
	if (!test_bit(LSFL_RCOM_WAIT, &ls->ls_flags) ||
	    rc_in->rc_id != ls->ls_rcom_seq) {
		log_debug(ls, "reject reply %d from %d seq %llx expect %llx",
			  rc_in->rc_type, rc_in->rc_header.h_nodeid,
			  (unsigned long long)rc_in->rc_id,
			  (unsigned long long)ls->ls_rcom_seq);
		goto out;
	}
	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	wake_up(&ls->ls_wait_general);
 out:
	spin_unlock(&ls->ls_rcom_spin);
}

int dlm_rcom_names(struct dlm_ls *ls, int nodeid, char *last_name, int last_len)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	int error = 0;
	int max_size = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);

	ls->ls_recover_nodeid = nodeid;

	if (nodeid == dlm_our_nodeid()) {
		ls->ls_recover_buf->rc_header.h_length =
			dlm_config.ci_buffer_size;
		dlm_copy_master_names(ls, last_name, last_len,
		                      ls->ls_recover_buf->rc_buf,
		                      max_size, nodeid);
		goto out;
	}

	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES, last_len, &rc, &mh);
	if (error)
		goto out;
	memcpy(rc->rc_buf, last_name, last_len);

	allow_sync_reply(ls, &rc->rc_id);
	memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);

	send_rcom(ls, mh, rc);

	error = dlm_wait_function(ls, &rcom_response);
	disallow_sync_reply(ls);
 out:
	return error;
}

static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	int error, inlen, outlen, nodeid;

	nodeid = rc_in->rc_header.h_nodeid;
	inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
	outlen = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);

	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh);
	if (error)
		return;
	rc->rc_id = rc_in->rc_id;
	rc->rc_seq_reply = rc_in->rc_seq;

	dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen,
			      nodeid);
	send_rcom(ls, mh, rc);
}

int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	struct dlm_ls *ls = r->res_ls;
	int error;

	error = create_rcom(ls, dir_nodeid, DLM_RCOM_LOOKUP, r->res_length,
			    &rc, &mh);
	if (error)
		goto out;
	memcpy(rc->rc_buf, r->res_name, r->res_length);
	rc->rc_id = (unsigned long) r;

	send_rcom(ls, mh, rc);
 out:
	return error;
}

static void receive_rcom_lookup(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid;
	int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);

	error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh);
	if (error)
		return;

	error = dlm_dir_lookup(ls, nodeid, rc_in->rc_buf, len, &ret_nodeid);
	if (error)
		ret_nodeid = error;
	rc->rc_result = ret_nodeid;
	rc->rc_id = rc_in->rc_id;
	rc->rc_seq_reply = rc_in->rc_seq;

	send_rcom(ls, mh, rc);
}

static void receive_rcom_lookup_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	dlm_recover_master_reply(ls, rc_in);
}

static void pack_rcom_lock(struct dlm_rsb *r, struct dlm_lkb *lkb,
			   struct rcom_lock *rl)
{
	memset(rl, 0, sizeof(*rl));

	rl->rl_ownpid = cpu_to_le32(lkb->lkb_ownpid);
	rl->rl_lkid = cpu_to_le32(lkb->lkb_id);
	rl->rl_exflags = cpu_to_le32(lkb->lkb_exflags);
	rl->rl_flags = cpu_to_le32(lkb->lkb_flags);
	rl->rl_lvbseq = cpu_to_le32(lkb->lkb_lvbseq);
	rl->rl_rqmode = lkb->lkb_rqmode;
	rl->rl_grmode = lkb->lkb_grmode;
	rl->rl_status = lkb->lkb_status;
	rl->rl_wait_type = cpu_to_le16(lkb->lkb_wait_type);

	if (lkb->lkb_bastfn)
		rl->rl_asts |= DLM_CB_BAST;
	if (lkb->lkb_astfn)
		rl->rl_asts |= DLM_CB_CAST;

	rl->rl_namelen = cpu_to_le16(r->res_length);
	memcpy(rl->rl_name, r->res_name, r->res_length);

	/* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
	   If so, receive_rcom_lock_args() won't take this copy. */

	if (lkb->lkb_lvbptr)
		memcpy(rl->rl_lvb, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
}

int dlm_send_rcom_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
{
	struct dlm_ls *ls = r->res_ls;
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	struct rcom_lock *rl;
	int error, len = sizeof(struct rcom_lock);

	if (lkb->lkb_lvbptr)
		len += ls->ls_lvblen;

	error = create_rcom(ls, r->res_nodeid, DLM_RCOM_LOCK, len, &rc, &mh);
	if (error)
		goto out;

	rl = (struct rcom_lock *) rc->rc_buf;
	pack_rcom_lock(r, lkb, rl);
	rc->rc_id = (unsigned long) r;

	send_rcom(ls, mh, rc);
 out:
	return error;
}

/* needs at least dlm_rcom + rcom_lock */
static void receive_rcom_lock(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	int error, nodeid = rc_in->rc_header.h_nodeid;

	dlm_recover_master_copy(ls, rc_in);

	error = create_rcom(ls, nodeid, DLM_RCOM_LOCK_REPLY,
			    sizeof(struct rcom_lock), &rc, &mh);
	if (error)
		return;

	/* We send back the same rcom_lock struct we received, but
	   dlm_recover_master_copy() has filled in rl_remid and rl_result */

	memcpy(rc->rc_buf, rc_in->rc_buf, sizeof(struct rcom_lock));
	rc->rc_id = rc_in->rc_id;
	rc->rc_seq_reply = rc_in->rc_seq;

	send_rcom(ls, mh, rc);
}

/* If the lockspace doesn't exist then still send a status message
   back; it's possible that it just doesn't have its global_id yet. */

int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
{
	struct dlm_rcom *rc;
	struct rcom_config *rf;
	struct dlm_mhandle *mh;
	char *mb;
	int mb_len = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);

	mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_NOFS, &mb);
	if (!mh)
		return -ENOBUFS;
	memset(mb, 0, mb_len);

	rc = (struct dlm_rcom *) mb;

	rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
	rc->rc_header.h_lockspace = rc_in->rc_header.h_lockspace;
	rc->rc_header.h_nodeid = dlm_our_nodeid();
	rc->rc_header.h_length = mb_len;
	rc->rc_header.h_cmd = DLM_RCOM;

	rc->rc_type = DLM_RCOM_STATUS_REPLY;
	rc->rc_id = rc_in->rc_id;
	rc->rc_seq_reply = rc_in->rc_seq;
	rc->rc_result = -ESRCH;

	rf = (struct rcom_config *) rc->rc_buf;
	rf->rf_lvblen = cpu_to_le32(~0U);

	dlm_rcom_out(rc);
	dlm_lowcomms_commit_buffer(mh);

	return 0;
}

/* Called by dlm_recv; corresponds to dlm_receive_message() but special
   recovery-only comms are sent through here. */

void dlm_receive_rcom(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid)
{
	int lock_size = sizeof(struct dlm_rcom) + sizeof(struct rcom_lock);
	int stop, reply = 0, lock = 0;
	uint32_t status;
	uint64_t seq;

	switch (rc->rc_type) {
	case DLM_RCOM_LOCK:
		lock = 1;
		break;
	case DLM_RCOM_LOCK_REPLY:
		lock = 1;
		reply = 1;
		break;
	case DLM_RCOM_STATUS_REPLY:
	case DLM_RCOM_NAMES_REPLY:
	case DLM_RCOM_LOOKUP_REPLY:
		reply = 1;
	};

	spin_lock(&ls->ls_recover_lock);
	status = ls->ls_recover_status;
	stop = test_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
	seq = ls->ls_recover_seq;
	spin_unlock(&ls->ls_recover_lock);

	if ((stop && (rc->rc_type != DLM_RCOM_STATUS)) ||
	    (reply && (rc->rc_seq_reply != seq)) ||
	    (lock && !(status & DLM_RS_DIR))) {
		log_limit(ls, "dlm_receive_rcom ignore msg %d "
			  "from %d %llu %llu recover seq %llu sts %x gen %u",
			   rc->rc_type,
			   nodeid,
			   (unsigned long long)rc->rc_seq,
			   (unsigned long long)rc->rc_seq_reply,
			   (unsigned long long)seq,
			   status, ls->ls_generation);
		goto out;
	}

	switch (rc->rc_type) {
	case DLM_RCOM_STATUS:
		receive_rcom_status(ls, rc);
		break;

	case DLM_RCOM_NAMES:
		receive_rcom_names(ls, rc);
		break;

	case DLM_RCOM_LOOKUP:
		receive_rcom_lookup(ls, rc);
		break;

	case DLM_RCOM_LOCK:
		if (rc->rc_header.h_length < lock_size)
			goto Eshort;
		receive_rcom_lock(ls, rc);
		break;

	case DLM_RCOM_STATUS_REPLY:
		receive_sync_reply(ls, rc);
		break;

	case DLM_RCOM_NAMES_REPLY:
		receive_sync_reply(ls, rc);
		break;

	case DLM_RCOM_LOOKUP_REPLY:
		receive_rcom_lookup_reply(ls, rc);
		break;

	case DLM_RCOM_LOCK_REPLY:
		if (rc->rc_header.h_length < lock_size)
			goto Eshort;
		dlm_recover_process_copy(ls, rc);
		break;

	default:
		log_error(ls, "receive_rcom bad type %d", rc->rc_type);
	}
out:
	return;
Eshort:
	log_error(ls, "recovery message %x from %d is too short",
			  rc->rc_type, nodeid);
}
Commit	Line	Data
e7fd4179 DT	1	/******************************************************************************
	2	*******************************************************************************
	3	**
	4	** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
dbcfc347	5	** Copyright (C) 2005-2008 Red Hat, Inc. All rights reserved.
e7fd4179 DT	6	**
	7	** This copyrighted material is made available to anyone wishing to use,
	8	** modify, copy, or redistribute it subject to the terms and conditions
	9	** of the GNU General Public License v.2.
	10	**
	11	*******************************************************************************
	12	******************************************************************************/
	13
	14	#include "dlm_internal.h"
	15	#include "lockspace.h"
	16	#include "member.h"
	17	#include "lowcomms.h"
	18	#include "midcomms.h"
	19	#include "rcom.h"
	20	#include "recover.h"
	21	#include "dir.h"
	22	#include "config.h"
	23	#include "memory.h"
	24	#include "lock.h"
	25	#include "util.h"
757a4271	26	#include "member.h"
e7fd4179 DT	27
	28
	29	static int rcom_response(struct dlm_ls *ls)
	30	{
	31	return test_bit(LSFL_RCOM_READY, &ls->ls_flags);
	32	}
	33
	34	static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len,
	35	struct dlm_rcom rc_ret, struct dlm_mhandle mh_ret)
	36	{
	37	struct dlm_rcom *rc;
	38	struct dlm_mhandle *mh;
	39	char *mb;
	40	int mb_len = sizeof(struct dlm_rcom) + len;
	41
573c24c4	42	mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
e7fd4179 DT	43	if (!mh) {
	44	log_print("create_rcom to %d type %d len %d ENOBUFS",
	45	to_nodeid, type, len);
	46	return -ENOBUFS;
	47	}
	48	memset(mb, 0, mb_len);
	49
	50	rc = (struct dlm_rcom *) mb;
	51
	52	rc->rc_header.h_version = (DLM_HEADER_MAJOR \| DLM_HEADER_MINOR);
	53	rc->rc_header.h_lockspace = ls->ls_global_id;
	54	rc->rc_header.h_nodeid = dlm_our_nodeid();
	55	rc->rc_header.h_length = mb_len;
	56	rc->rc_header.h_cmd = DLM_RCOM;
	57
	58	rc->rc_type = type;
	59
38aa8b0c DT	60	spin_lock(&ls->ls_recover_lock);
	61	rc->rc_seq = ls->ls_recover_seq;
	62	spin_unlock(&ls->ls_recover_lock);
	63
e7fd4179 DT	64	*mh_ret = mh;
	65	*rc_ret = rc;
	66	return 0;
	67	}
	68
	69	static void send_rcom(struct dlm_ls ls, struct dlm_mhandle mh,
	70	struct dlm_rcom *rc)
	71	{
	72	dlm_rcom_out(rc);
	73	dlm_lowcomms_commit_buffer(mh);
	74	}
	75
757a4271 DT	76	static void set_rcom_status(struct dlm_ls ls, struct rcom_status rs,
	77	uint32_t flags)
	78	{
	79	rs->rs_flags = cpu_to_le32(flags);
	80	}
	81
e7fd4179 DT	82	/* When replying to a status request, a node also sends back its
	83	configuration values. The requesting node then checks that the remote
	84	node is configured the same way as itself. */
	85
757a4271 DT	86	static void set_rcom_config(struct dlm_ls ls, struct rcom_config rf,
757a4271 DT	87	uint32_t num_slots)
e7fd4179	88	{
93ff2971 AV	89	rf->rf_lvblen = cpu_to_le32(ls->ls_lvblen);
93ff2971 AV	90	rf->rf_lsflags = cpu_to_le32(ls->ls_exflags);
757a4271 DT	91
	92	rf->rf_our_slot = cpu_to_le16(ls->ls_slot);
	93	rf->rf_num_slots = cpu_to_le16(num_slots);
	94	rf->rf_generation = cpu_to_le32(ls->ls_generation);
e7fd4179 DT	95	}
e7fd4179 DT	96
757a4271	97	static int check_rcom_config(struct dlm_ls ls, struct dlm_rcom rc, int nodeid)
e7fd4179	98	{
9e971b71 DT	99	struct rcom_config rf = (struct rcom_config ) rc->rc_buf;
	100
	101	if ((rc->rc_header.h_version & 0xFFFF0000) != DLM_HEADER_MAJOR) {
	102	log_error(ls, "version mismatch: %x nodeid %d: %x",
	103	DLM_HEADER_MAJOR \| DLM_HEADER_MINOR, nodeid,
	104	rc->rc_header.h_version);
8b0e7b2c	105	return -EPROTO;
9e971b71 DT	106	}
9e971b71 DT	107
93ff2971 AV	108	if (le32_to_cpu(rf->rf_lvblen) != ls->ls_lvblen \|\|
93ff2971 AV	109	le32_to_cpu(rf->rf_lsflags) != ls->ls_exflags) {
e7fd4179	110	log_error(ls, "config mismatch: %d,%x nodeid %d: %d,%x",
93ff2971 AV	111	ls->ls_lvblen, ls->ls_exflags, nodeid,
	112	le32_to_cpu(rf->rf_lvblen),
	113	le32_to_cpu(rf->rf_lsflags));
8b0e7b2c	114	return -EPROTO;
e7fd4179 DT	115	}
	116	return 0;
	117	}
	118
98f176fb DT	119	static void allow_sync_reply(struct dlm_ls ls, uint64_t new_seq)
	120	{
	121	spin_lock(&ls->ls_rcom_spin);
	122	*new_seq = ++ls->ls_rcom_seq;
	123	set_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	124	spin_unlock(&ls->ls_rcom_spin);
	125	}
	126
	127	static void disallow_sync_reply(struct dlm_ls *ls)
	128	{
	129	spin_lock(&ls->ls_rcom_spin);
	130	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	131	clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
	132	spin_unlock(&ls->ls_rcom_spin);
	133	}
	134
757a4271 DT	135	/*
	136	* low nodeid gathers one slot value at a time from each node.
	137	* it sets need_slots=0, and saves rf_our_slot returned from each
	138	* rcom_config.
	139	*
	140	* other nodes gather all slot values at once from the low nodeid.
	141	* they set need_slots=1, and ignore the rf_our_slot returned from each
	142	* rcom_config. they use the rf_num_slots returned from the low
	143	* node's rcom_config.
	144	*/
	145
	146	int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags)
e7fd4179 DT	147	{
	148	struct dlm_rcom *rc;
	149	struct dlm_mhandle *mh;
	150	int error = 0;
	151
faa0f267	152	ls->ls_recover_nodeid = nodeid;
e7fd4179 DT	153
e7fd4179 DT	154	if (nodeid == dlm_our_nodeid()) {
4007685c	155	rc = ls->ls_recover_buf;
e7fd4179 DT	156	rc->rc_result = dlm_recover_status(ls);
	157	goto out;
	158	}
	159
757a4271 DT	160	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS,
757a4271 DT	161	sizeof(struct rcom_status), &rc, &mh);
e7fd4179 DT	162	if (error)
e7fd4179 DT	163	goto out;
98f176fb	164
757a4271 DT	165	set_rcom_status(ls, (struct rcom_status *)rc->rc_buf, status_flags);
757a4271 DT	166
98f176fb	167	allow_sync_reply(ls, &rc->rc_id);
68c817a1	168	memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
e7fd4179 DT	169
	170	send_rcom(ls, mh, rc);
	171
	172	error = dlm_wait_function(ls, &rcom_response);
98f176fb	173	disallow_sync_reply(ls);
e7fd4179 DT	174	if (error)
	175	goto out;
	176
4007685c	177	rc = ls->ls_recover_buf;
e7fd4179 DT	178
	179	if (rc->rc_result == -ESRCH) {
	180	/* we pretend the remote lockspace exists with 0 status */
	181	log_debug(ls, "remote node %d not ready", nodeid);
	182	rc->rc_result = 0;
757a4271 DT	183	error = 0;
	184	} else {
	185	error = check_rcom_config(ls, rc, nodeid);
	186	}
	187
e7fd4179 DT	188	/* the caller looks at rc_result for the remote recovery status */
	189	out:
	190	return error;
	191	}
	192
	193	static void receive_rcom_status(struct dlm_ls ls, struct dlm_rcom rc_in)
	194	{
	195	struct dlm_rcom *rc;
	196	struct dlm_mhandle *mh;
757a4271 DT	197	struct rcom_status *rs;
	198	uint32_t status;
	199	int nodeid = rc_in->rc_header.h_nodeid;
	200	int len = sizeof(struct rcom_config);
	201	int num_slots = 0;
	202	int error;
	203
	204	if (!dlm_slots_version(&rc_in->rc_header)) {
	205	status = dlm_recover_status(ls);
	206	goto do_create;
	207	}
	208
	209	rs = (struct rcom_status *)rc_in->rc_buf;
e7fd4179	210
757a4271 DT	211	if (!(rs->rs_flags & DLM_RSF_NEED_SLOTS)) {
	212	status = dlm_recover_status(ls);
	213	goto do_create;
	214	}
	215
	216	spin_lock(&ls->ls_recover_lock);
	217	status = ls->ls_recover_status;
	218	num_slots = ls->ls_num_slots;
	219	spin_unlock(&ls->ls_recover_lock);
	220	len += num_slots * sizeof(struct rcom_slot);
	221
	222	do_create:
e7fd4179	223	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY,
757a4271	224	len, &rc, &mh);
e7fd4179 DT	225	if (error)
e7fd4179 DT	226	return;
757a4271	227
4a99c3d9	228	rc->rc_id = rc_in->rc_id;
38aa8b0c	229	rc->rc_seq_reply = rc_in->rc_seq;
757a4271 DT	230	rc->rc_result = status;
	231
	232	set_rcom_config(ls, (struct rcom_config *)rc->rc_buf, num_slots);
	233
	234	if (!num_slots)
	235	goto do_send;
	236
	237	spin_lock(&ls->ls_recover_lock);
	238	if (ls->ls_num_slots != num_slots) {
	239	spin_unlock(&ls->ls_recover_lock);
	240	log_debug(ls, "receive_rcom_status num_slots %d to %d",
	241	num_slots, ls->ls_num_slots);
	242	rc->rc_result = 0;
	243	set_rcom_config(ls, (struct rcom_config *)rc->rc_buf, 0);
	244	goto do_send;
	245	}
	246
	247	dlm_slots_copy_out(ls, rc);
	248	spin_unlock(&ls->ls_recover_lock);
e7fd4179	249
757a4271	250	do_send:
e7fd4179 DT	251	send_rcom(ls, mh, rc);
	252	}
	253
4a99c3d9	254	static void receive_sync_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
e7fd4179	255	{
98f176fb DT	256	spin_lock(&ls->ls_rcom_spin);
	257	if (!test_bit(LSFL_RCOM_WAIT, &ls->ls_flags) \|\|
	258	rc_in->rc_id != ls->ls_rcom_seq) {
	259	log_debug(ls, "reject reply %d from %d seq %llx expect %llx",
	260	rc_in->rc_type, rc_in->rc_header.h_nodeid,
57adf7ee RK	261	(unsigned long long)rc_in->rc_id,
57adf7ee RK	262	(unsigned long long)ls->ls_rcom_seq);
98f176fb	263	goto out;
4a99c3d9	264	}
e7fd4179 DT	265	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
e7fd4179 DT	266	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
98f176fb	267	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
e7fd4179	268	wake_up(&ls->ls_wait_general);
98f176fb DT	269	out:
98f176fb DT	270	spin_unlock(&ls->ls_rcom_spin);
e7fd4179 DT	271	}
	272
	273	int dlm_rcom_names(struct dlm_ls ls, int nodeid, char last_name, int last_len)
	274	{
	275	struct dlm_rcom *rc;
	276	struct dlm_mhandle *mh;
4007685c AV	277	int error = 0;
4007685c AV	278	int max_size = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);
e7fd4179	279
faa0f267	280	ls->ls_recover_nodeid = nodeid;
e7fd4179 DT	281
e7fd4179 DT	282	if (nodeid == dlm_our_nodeid()) {
599e0f58 DT	283	ls->ls_recover_buf->rc_header.h_length =
599e0f58 DT	284	dlm_config.ci_buffer_size;
e7fd4179	285	dlm_copy_master_names(ls, last_name, last_len,
4007685c AV	286	ls->ls_recover_buf->rc_buf,
4007685c AV	287	max_size, nodeid);
e7fd4179 DT	288	goto out;
	289	}
	290
	291	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES, last_len, &rc, &mh);
	292	if (error)
	293	goto out;
	294	memcpy(rc->rc_buf, last_name, last_len);
98f176fb DT	295
98f176fb DT	296	allow_sync_reply(ls, &rc->rc_id);
68c817a1	297	memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
e7fd4179 DT	298
	299	send_rcom(ls, mh, rc);
	300
	301	error = dlm_wait_function(ls, &rcom_response);
98f176fb	302	disallow_sync_reply(ls);
e7fd4179 DT	303	out:
	304	return error;
	305	}
	306
	307	static void receive_rcom_names(struct dlm_ls ls, struct dlm_rcom rc_in)
	308	{
	309	struct dlm_rcom *rc;
	310	struct dlm_mhandle *mh;
38aa8b0c	311	int error, inlen, outlen, nodeid;
e7fd4179 DT	312
	313	nodeid = rc_in->rc_header.h_nodeid;
	314	inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
68c817a1	315	outlen = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);
e7fd4179 DT	316
	317	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh);
	318	if (error)
	319	return;
4a99c3d9	320	rc->rc_id = rc_in->rc_id;
38aa8b0c	321	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	322
	323	dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen,
	324	nodeid);
	325	send_rcom(ls, mh, rc);
	326	}
	327
e7fd4179 DT	328	int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid)
	329	{
	330	struct dlm_rcom *rc;
	331	struct dlm_mhandle *mh;
	332	struct dlm_ls *ls = r->res_ls;
	333	int error;
	334
	335	error = create_rcom(ls, dir_nodeid, DLM_RCOM_LOOKUP, r->res_length,
	336	&rc, &mh);
	337	if (error)
	338	goto out;
	339	memcpy(rc->rc_buf, r->res_name, r->res_length);
	340	rc->rc_id = (unsigned long) r;
	341
	342	send_rcom(ls, mh, rc);
	343	out:
	344	return error;
	345	}
	346
	347	static void receive_rcom_lookup(struct dlm_ls ls, struct dlm_rcom rc_in)
	348	{
	349	struct dlm_rcom *rc;
	350	struct dlm_mhandle *mh;
	351	int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid;
	352	int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
	353
	354	error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh);
	355	if (error)
	356	return;
	357
	358	error = dlm_dir_lookup(ls, nodeid, rc_in->rc_buf, len, &ret_nodeid);
	359	if (error)
	360	ret_nodeid = error;
	361	rc->rc_result = ret_nodeid;
	362	rc->rc_id = rc_in->rc_id;
38aa8b0c	363	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	364
	365	send_rcom(ls, mh, rc);
	366	}
	367
	368	static void receive_rcom_lookup_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	369	{
	370	dlm_recover_master_reply(ls, rc_in);
	371	}
	372
	373	static void pack_rcom_lock(struct dlm_rsb r, struct dlm_lkb lkb,
	374	struct rcom_lock *rl)
	375	{
	376	memset(rl, 0, sizeof(*rl));
	377
163a1859 AV	378	rl->rl_ownpid = cpu_to_le32(lkb->lkb_ownpid);
	379	rl->rl_lkid = cpu_to_le32(lkb->lkb_id);
	380	rl->rl_exflags = cpu_to_le32(lkb->lkb_exflags);
	381	rl->rl_flags = cpu_to_le32(lkb->lkb_flags);
	382	rl->rl_lvbseq = cpu_to_le32(lkb->lkb_lvbseq);
e7fd4179 DT	383	rl->rl_rqmode = lkb->lkb_rqmode;
	384	rl->rl_grmode = lkb->lkb_grmode;
	385	rl->rl_status = lkb->lkb_status;
163a1859	386	rl->rl_wait_type = cpu_to_le16(lkb->lkb_wait_type);
e7fd4179	387
e5dae548	388	if (lkb->lkb_bastfn)
8304d6f2	389	rl->rl_asts \|= DLM_CB_BAST;
e5dae548	390	if (lkb->lkb_astfn)
8304d6f2	391	rl->rl_asts \|= DLM_CB_CAST;
e7fd4179	392
163a1859	393	rl->rl_namelen = cpu_to_le16(r->res_length);
e7fd4179 DT	394	memcpy(rl->rl_name, r->res_name, r->res_length);
	395
	396	/* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
	397	If so, receive_rcom_lock_args() won't take this copy. */
	398
	399	if (lkb->lkb_lvbptr)
	400	memcpy(rl->rl_lvb, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
	401	}
	402
	403	int dlm_send_rcom_lock(struct dlm_rsb r, struct dlm_lkb lkb)
	404	{
	405	struct dlm_ls *ls = r->res_ls;
	406	struct dlm_rcom *rc;
	407	struct dlm_mhandle *mh;
	408	struct rcom_lock *rl;
	409	int error, len = sizeof(struct rcom_lock);
	410
	411	if (lkb->lkb_lvbptr)
	412	len += ls->ls_lvblen;
	413
	414	error = create_rcom(ls, r->res_nodeid, DLM_RCOM_LOCK, len, &rc, &mh);
	415	if (error)
	416	goto out;
	417
	418	rl = (struct rcom_lock *) rc->rc_buf;
	419	pack_rcom_lock(r, lkb, rl);
	420	rc->rc_id = (unsigned long) r;
	421
	422	send_rcom(ls, mh, rc);
	423	out:
	424	return error;
	425	}
	426
ae773d0b	427	/* needs at least dlm_rcom + rcom_lock */
e7fd4179 DT	428	static void receive_rcom_lock(struct dlm_ls ls, struct dlm_rcom rc_in)
	429	{
	430	struct dlm_rcom *rc;
	431	struct dlm_mhandle *mh;
	432	int error, nodeid = rc_in->rc_header.h_nodeid;
	433
	434	dlm_recover_master_copy(ls, rc_in);
	435
	436	error = create_rcom(ls, nodeid, DLM_RCOM_LOCK_REPLY,
	437	sizeof(struct rcom_lock), &rc, &mh);
	438	if (error)
	439	return;
	440
	441	/* We send back the same rcom_lock struct we received, but
	442	dlm_recover_master_copy() has filled in rl_remid and rl_result */
	443
	444	memcpy(rc->rc_buf, rc_in->rc_buf, sizeof(struct rcom_lock));
	445	rc->rc_id = rc_in->rc_id;
38aa8b0c	446	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	447
	448	send_rcom(ls, mh, rc);
	449	}
	450
c36258b5 DT	451	/* If the lockspace doesn't exist then still send a status message
	452	back; it's possible that it just doesn't have its global_id yet. */
	453
	454	int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
e7fd4179 DT	455	{
e7fd4179 DT	456	struct dlm_rcom *rc;
1babdb45	457	struct rcom_config *rf;
e7fd4179 DT	458	struct dlm_mhandle *mh;
e7fd4179 DT	459	char *mb;
1babdb45	460	int mb_len = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);
e7fd4179	461
41684f95	462	mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_NOFS, &mb);
e7fd4179 DT	463	if (!mh)
	464	return -ENOBUFS;
	465	memset(mb, 0, mb_len);
	466
	467	rc = (struct dlm_rcom *) mb;
	468
	469	rc->rc_header.h_version = (DLM_HEADER_MAJOR \| DLM_HEADER_MINOR);
	470	rc->rc_header.h_lockspace = rc_in->rc_header.h_lockspace;
	471	rc->rc_header.h_nodeid = dlm_our_nodeid();
	472	rc->rc_header.h_length = mb_len;
	473	rc->rc_header.h_cmd = DLM_RCOM;
	474
	475	rc->rc_type = DLM_RCOM_STATUS_REPLY;
f5888750	476	rc->rc_id = rc_in->rc_id;
38aa8b0c	477	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	478	rc->rc_result = -ESRCH;
e7fd4179 DT	479
1babdb45	480	rf = (struct rcom_config *) rc->rc_buf;
93ff2971	481	rf->rf_lvblen = cpu_to_le32(~0U);
1babdb45	482
e7fd4179 DT	483	dlm_rcom_out(rc);
	484	dlm_lowcomms_commit_buffer(mh);
	485
	486	return 0;
	487	}
	488
d6e24788 DT	489	/* Called by dlm_recv; corresponds to dlm_receive_message() but special
	490	recovery-only comms are sent through here. */
	491
	492	void dlm_receive_rcom(struct dlm_ls ls, struct dlm_rcom rc, int nodeid)
38aa8b0c	493	{
d6e24788	494	int lock_size = sizeof(struct dlm_rcom) + sizeof(struct rcom_lock);
4875647a DT	495	int stop, reply = 0, lock = 0;
4875647a DT	496	uint32_t status;
38aa8b0c	497	uint64_t seq;
38aa8b0c DT	498
38aa8b0c DT	499	switch (rc->rc_type) {
4875647a DT	500	case DLM_RCOM_LOCK:
	501	lock = 1;
	502	break;
	503	case DLM_RCOM_LOCK_REPLY:
	504	lock = 1;
	505	reply = 1;
	506	break;
38aa8b0c DT	507	case DLM_RCOM_STATUS_REPLY:
	508	case DLM_RCOM_NAMES_REPLY:
	509	case DLM_RCOM_LOOKUP_REPLY:
d6e24788 DT	510	reply = 1;
d6e24788 DT	511	};
38aa8b0c	512
d6e24788	513	spin_lock(&ls->ls_recover_lock);
4875647a	514	status = ls->ls_recover_status;
d6e24788 DT	515	stop = test_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
	516	seq = ls->ls_recover_seq;
	517	spin_unlock(&ls->ls_recover_lock);
ae773d0b	518
d6e24788	519	if ((stop && (rc->rc_type != DLM_RCOM_STATUS)) \|\|
4875647a DT	520	(reply && (rc->rc_seq_reply != seq)) \|\|
4875647a DT	521	(lock && !(status & DLM_RS_DIR))) {
d6e24788	522	log_limit(ls, "dlm_receive_rcom ignore msg %d "
4875647a DT	523	"from %d %llu %llu recover seq %llu sts %x gen %u",
	524	rc->rc_type,
	525	nodeid,
d6e24788 DT	526	(unsigned long long)rc->rc_seq,
d6e24788 DT	527	(unsigned long long)rc->rc_seq_reply,
4875647a DT	528	(unsigned long long)seq,
4875647a DT	529	status, ls->ls_generation);
e7fd4179 DT	530	goto out;
	531	}
	532
e7fd4179 DT	533	switch (rc->rc_type) {
	534	case DLM_RCOM_STATUS:
	535	receive_rcom_status(ls, rc);
	536	break;
	537
	538	case DLM_RCOM_NAMES:
	539	receive_rcom_names(ls, rc);
	540	break;
	541
	542	case DLM_RCOM_LOOKUP:
	543	receive_rcom_lookup(ls, rc);
	544	break;
	545
	546	case DLM_RCOM_LOCK:
ae773d0b AV	547	if (rc->rc_header.h_length < lock_size)
ae773d0b AV	548	goto Eshort;
e7fd4179 DT	549	receive_rcom_lock(ls, rc);
	550	break;
	551
	552	case DLM_RCOM_STATUS_REPLY:
dbcfc347	553	receive_sync_reply(ls, rc);
e7fd4179 DT	554	break;
	555
	556	case DLM_RCOM_NAMES_REPLY:
dbcfc347	557	receive_sync_reply(ls, rc);
e7fd4179 DT	558	break;
	559
	560	case DLM_RCOM_LOOKUP_REPLY:
	561	receive_rcom_lookup_reply(ls, rc);
	562	break;
	563
	564	case DLM_RCOM_LOCK_REPLY:
ae773d0b AV	565	if (rc->rc_header.h_length < lock_size)
ae773d0b AV	566	goto Eshort;
dbcfc347	567	dlm_recover_process_copy(ls, rc);
e7fd4179 DT	568	break;
	569
	570	default:
dbcfc347	571	log_error(ls, "receive_rcom bad type %d", rc->rc_type);
e7fd4179	572	}
ae773d0b	573	out:
c36258b5	574	return;
ae773d0b AV	575	Eshort:
	576	log_error(ls, "recovery message %x from %d is too short",
	577	rc->rc_type, nodeid);
e7fd4179 DT	578	}
e7fd4179 DT	579