[net-next-2.6.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"


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

/* 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 make_config(struct dlm_ls *ls, struct rcom_config *rf)
{
	rf->rf_lvblen = cpu_to_le32(ls->ls_lvblen);
	rf->rf_lsflags = cpu_to_le32(ls->ls_exflags);
}

static int check_config(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid)
{
	struct rcom_config *rf = (struct rcom_config *) rc->rc_buf;
	size_t conf_size = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);

	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 (rc->rc_header.h_length < conf_size) {
		log_error(ls, "config too short: %d nodeid %d",
			  rc->rc_header.h_length, nodeid);
		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);
}

int dlm_rcom_status(struct dlm_ls *ls, int nodeid)
{
	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, 0, &rc, &mh);
	if (error)
		goto out;

	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;
	} else
		error = check_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;
	int error, nodeid = rc_in->rc_header.h_nodeid;

	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY,
			    sizeof(struct rcom_config), &rc, &mh);
	if (error)
		return;
	rc->rc_id = rc_in->rc_id;
	rc->rc_seq_reply = rc_in->rc_seq;
	rc->rc_result = dlm_recover_status(ls);
	make_config(ls, (struct rcom_config *) rc->rc_buf);

	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 |= AST_BAST;
	if (lkb->lkb_astfn)
		rl->rl_asts |= AST_COMP;

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

static int is_old_reply(struct dlm_ls *ls, struct dlm_rcom *rc)
{
	uint64_t seq;
	int rv = 0;

	switch (rc->rc_type) {
	case DLM_RCOM_STATUS_REPLY:
	case DLM_RCOM_NAMES_REPLY:
	case DLM_RCOM_LOOKUP_REPLY:
	case DLM_RCOM_LOCK_REPLY:
		spin_lock(&ls->ls_recover_lock);
		seq = ls->ls_recover_seq;
		spin_unlock(&ls->ls_recover_lock);
		if (rc->rc_seq_reply != seq) {
			log_debug(ls, "ignoring old reply %x from %d "
				      "seq_reply %llx expect %llx",
				      rc->rc_type, rc->rc_header.h_nodeid,
				      (unsigned long long)rc->rc_seq_reply,
				      (unsigned long long)seq);
			rv = 1;
		}
	}
	return rv;
}

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

	if (dlm_recovery_stopped(ls) && (rc->rc_type != DLM_RCOM_STATUS)) {
		log_debug(ls, "ignoring recovery message %x from %d",
			  rc->rc_type, nodeid);
		goto out;
	}

	if (is_old_reply(ls, rc))
		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"
	26
	27
	28	static int rcom_response(struct dlm_ls *ls)
	29	{
	30	return test_bit(LSFL_RCOM_READY, &ls->ls_flags);
	31	}
	32
	33	static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len,
	34	struct dlm_rcom rc_ret, struct dlm_mhandle mh_ret)
	35	{
	36	struct dlm_rcom *rc;
	37	struct dlm_mhandle *mh;
	38	char *mb;
	39	int mb_len = sizeof(struct dlm_rcom) + len;
	40
573c24c4	41	mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
e7fd4179 DT	42	if (!mh) {
	43	log_print("create_rcom to %d type %d len %d ENOBUFS",
	44	to_nodeid, type, len);
	45	return -ENOBUFS;
	46	}
	47	memset(mb, 0, mb_len);
	48
	49	rc = (struct dlm_rcom *) mb;
	50
	51	rc->rc_header.h_version = (DLM_HEADER_MAJOR \| DLM_HEADER_MINOR);
	52	rc->rc_header.h_lockspace = ls->ls_global_id;
	53	rc->rc_header.h_nodeid = dlm_our_nodeid();
	54	rc->rc_header.h_length = mb_len;
	55	rc->rc_header.h_cmd = DLM_RCOM;
	56
	57	rc->rc_type = type;
	58
38aa8b0c DT	59	spin_lock(&ls->ls_recover_lock);
	60	rc->rc_seq = ls->ls_recover_seq;
	61	spin_unlock(&ls->ls_recover_lock);
	62
e7fd4179 DT	63	*mh_ret = mh;
	64	*rc_ret = rc;
	65	return 0;
	66	}
	67
	68	static void send_rcom(struct dlm_ls ls, struct dlm_mhandle mh,
	69	struct dlm_rcom *rc)
	70	{
	71	dlm_rcom_out(rc);
	72	dlm_lowcomms_commit_buffer(mh);
	73	}
	74
	75	/* When replying to a status request, a node also sends back its
	76	configuration values. The requesting node then checks that the remote
	77	node is configured the same way as itself. */
	78
	79	static void make_config(struct dlm_ls ls, struct rcom_config rf)
	80	{
93ff2971 AV	81	rf->rf_lvblen = cpu_to_le32(ls->ls_lvblen);
93ff2971 AV	82	rf->rf_lsflags = cpu_to_le32(ls->ls_exflags);
e7fd4179 DT	83	}
e7fd4179 DT	84
9e971b71	85	static int check_config(struct dlm_ls ls, struct dlm_rcom rc, int nodeid)
e7fd4179	86	{
9e971b71	87	struct rcom_config rf = (struct rcom_config ) rc->rc_buf;
02ed16b6	88	size_t conf_size = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);
9e971b71 DT	89
	90	if ((rc->rc_header.h_version & 0xFFFF0000) != DLM_HEADER_MAJOR) {
	91	log_error(ls, "version mismatch: %x nodeid %d: %x",
	92	DLM_HEADER_MAJOR \| DLM_HEADER_MINOR, nodeid,
	93	rc->rc_header.h_version);
8b0e7b2c	94	return -EPROTO;
9e971b71 DT	95	}
9e971b71 DT	96
02ed16b6 AV	97	if (rc->rc_header.h_length < conf_size) {
	98	log_error(ls, "config too short: %d nodeid %d",
	99	rc->rc_header.h_length, nodeid);
	100	return -EPROTO;
	101	}
	102
93ff2971 AV	103	if (le32_to_cpu(rf->rf_lvblen) != ls->ls_lvblen \|\|
93ff2971 AV	104	le32_to_cpu(rf->rf_lsflags) != ls->ls_exflags) {
e7fd4179	105	log_error(ls, "config mismatch: %d,%x nodeid %d: %d,%x",
93ff2971 AV	106	ls->ls_lvblen, ls->ls_exflags, nodeid,
	107	le32_to_cpu(rf->rf_lvblen),
	108	le32_to_cpu(rf->rf_lsflags));
8b0e7b2c	109	return -EPROTO;
e7fd4179 DT	110	}
	111	return 0;
	112	}
	113
98f176fb DT	114	static void allow_sync_reply(struct dlm_ls ls, uint64_t new_seq)
	115	{
	116	spin_lock(&ls->ls_rcom_spin);
	117	*new_seq = ++ls->ls_rcom_seq;
	118	set_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	119	spin_unlock(&ls->ls_rcom_spin);
	120	}
	121
	122	static void disallow_sync_reply(struct dlm_ls *ls)
	123	{
	124	spin_lock(&ls->ls_rcom_spin);
	125	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	126	clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
	127	spin_unlock(&ls->ls_rcom_spin);
	128	}
	129
e7fd4179 DT	130	int dlm_rcom_status(struct dlm_ls *ls, int nodeid)
	131	{
	132	struct dlm_rcom *rc;
	133	struct dlm_mhandle *mh;
	134	int error = 0;
	135
faa0f267	136	ls->ls_recover_nodeid = nodeid;
e7fd4179 DT	137
e7fd4179 DT	138	if (nodeid == dlm_our_nodeid()) {
4007685c	139	rc = ls->ls_recover_buf;
e7fd4179 DT	140	rc->rc_result = dlm_recover_status(ls);
	141	goto out;
	142	}
	143
	144	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS, 0, &rc, &mh);
	145	if (error)
	146	goto out;
98f176fb DT	147
98f176fb DT	148	allow_sync_reply(ls, &rc->rc_id);
68c817a1	149	memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
e7fd4179 DT	150
	151	send_rcom(ls, mh, rc);
	152
	153	error = dlm_wait_function(ls, &rcom_response);
98f176fb	154	disallow_sync_reply(ls);
e7fd4179 DT	155	if (error)
	156	goto out;
	157
4007685c	158	rc = ls->ls_recover_buf;
e7fd4179 DT	159
	160	if (rc->rc_result == -ESRCH) {
	161	/* we pretend the remote lockspace exists with 0 status */
	162	log_debug(ls, "remote node %d not ready", nodeid);
	163	rc->rc_result = 0;
	164	} else
9e971b71	165	error = check_config(ls, rc, nodeid);
e7fd4179 DT	166	/* the caller looks at rc_result for the remote recovery status */
	167	out:
	168	return error;
	169	}
	170
	171	static void receive_rcom_status(struct dlm_ls ls, struct dlm_rcom rc_in)
	172	{
	173	struct dlm_rcom *rc;
	174	struct dlm_mhandle *mh;
	175	int error, nodeid = rc_in->rc_header.h_nodeid;
	176
	177	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY,
	178	sizeof(struct rcom_config), &rc, &mh);
	179	if (error)
	180	return;
4a99c3d9	181	rc->rc_id = rc_in->rc_id;
38aa8b0c	182	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	183	rc->rc_result = dlm_recover_status(ls);
	184	make_config(ls, (struct rcom_config *) rc->rc_buf);
	185
	186	send_rcom(ls, mh, rc);
	187	}
	188
4a99c3d9	189	static void receive_sync_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
e7fd4179	190	{
98f176fb DT	191	spin_lock(&ls->ls_rcom_spin);
	192	if (!test_bit(LSFL_RCOM_WAIT, &ls->ls_flags) \|\|
	193	rc_in->rc_id != ls->ls_rcom_seq) {
	194	log_debug(ls, "reject reply %d from %d seq %llx expect %llx",
	195	rc_in->rc_type, rc_in->rc_header.h_nodeid,
57adf7ee RK	196	(unsigned long long)rc_in->rc_id,
57adf7ee RK	197	(unsigned long long)ls->ls_rcom_seq);
98f176fb	198	goto out;
4a99c3d9	199	}
e7fd4179 DT	200	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
e7fd4179 DT	201	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
98f176fb	202	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
e7fd4179	203	wake_up(&ls->ls_wait_general);
98f176fb DT	204	out:
98f176fb DT	205	spin_unlock(&ls->ls_rcom_spin);
e7fd4179 DT	206	}
	207
	208	int dlm_rcom_names(struct dlm_ls ls, int nodeid, char last_name, int last_len)
	209	{
	210	struct dlm_rcom *rc;
	211	struct dlm_mhandle *mh;
4007685c AV	212	int error = 0;
4007685c AV	213	int max_size = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);
e7fd4179	214
faa0f267	215	ls->ls_recover_nodeid = nodeid;
e7fd4179 DT	216
e7fd4179 DT	217	if (nodeid == dlm_our_nodeid()) {
599e0f58 DT	218	ls->ls_recover_buf->rc_header.h_length =
599e0f58 DT	219	dlm_config.ci_buffer_size;
e7fd4179	220	dlm_copy_master_names(ls, last_name, last_len,
4007685c AV	221	ls->ls_recover_buf->rc_buf,
4007685c AV	222	max_size, nodeid);
e7fd4179 DT	223	goto out;
	224	}
	225
	226	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES, last_len, &rc, &mh);
	227	if (error)
	228	goto out;
	229	memcpy(rc->rc_buf, last_name, last_len);
98f176fb DT	230
98f176fb DT	231	allow_sync_reply(ls, &rc->rc_id);
68c817a1	232	memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
e7fd4179 DT	233
	234	send_rcom(ls, mh, rc);
	235
	236	error = dlm_wait_function(ls, &rcom_response);
98f176fb	237	disallow_sync_reply(ls);
e7fd4179 DT	238	out:
	239	return error;
	240	}
	241
	242	static void receive_rcom_names(struct dlm_ls ls, struct dlm_rcom rc_in)
	243	{
	244	struct dlm_rcom *rc;
	245	struct dlm_mhandle *mh;
38aa8b0c	246	int error, inlen, outlen, nodeid;
e7fd4179 DT	247
	248	nodeid = rc_in->rc_header.h_nodeid;
	249	inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
68c817a1	250	outlen = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);
e7fd4179 DT	251
	252	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh);
	253	if (error)
	254	return;
4a99c3d9	255	rc->rc_id = rc_in->rc_id;
38aa8b0c	256	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	257
	258	dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen,
	259	nodeid);
	260	send_rcom(ls, mh, rc);
	261	}
	262
e7fd4179 DT	263	int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid)
	264	{
	265	struct dlm_rcom *rc;
	266	struct dlm_mhandle *mh;
	267	struct dlm_ls *ls = r->res_ls;
	268	int error;
	269
	270	error = create_rcom(ls, dir_nodeid, DLM_RCOM_LOOKUP, r->res_length,
	271	&rc, &mh);
	272	if (error)
	273	goto out;
	274	memcpy(rc->rc_buf, r->res_name, r->res_length);
	275	rc->rc_id = (unsigned long) r;
	276
	277	send_rcom(ls, mh, rc);
	278	out:
	279	return error;
	280	}
	281
	282	static void receive_rcom_lookup(struct dlm_ls ls, struct dlm_rcom rc_in)
	283	{
	284	struct dlm_rcom *rc;
	285	struct dlm_mhandle *mh;
	286	int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid;
	287	int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
	288
	289	error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh);
	290	if (error)
	291	return;
	292
	293	error = dlm_dir_lookup(ls, nodeid, rc_in->rc_buf, len, &ret_nodeid);
	294	if (error)
	295	ret_nodeid = error;
	296	rc->rc_result = ret_nodeid;
	297	rc->rc_id = rc_in->rc_id;
38aa8b0c	298	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	299
	300	send_rcom(ls, mh, rc);
	301	}
	302
	303	static void receive_rcom_lookup_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	304	{
	305	dlm_recover_master_reply(ls, rc_in);
	306	}
	307
	308	static void pack_rcom_lock(struct dlm_rsb r, struct dlm_lkb lkb,
	309	struct rcom_lock *rl)
	310	{
	311	memset(rl, 0, sizeof(*rl));
	312
163a1859 AV	313	rl->rl_ownpid = cpu_to_le32(lkb->lkb_ownpid);
	314	rl->rl_lkid = cpu_to_le32(lkb->lkb_id);
	315	rl->rl_exflags = cpu_to_le32(lkb->lkb_exflags);
	316	rl->rl_flags = cpu_to_le32(lkb->lkb_flags);
	317	rl->rl_lvbseq = cpu_to_le32(lkb->lkb_lvbseq);
e7fd4179 DT	318	rl->rl_rqmode = lkb->lkb_rqmode;
	319	rl->rl_grmode = lkb->lkb_grmode;
	320	rl->rl_status = lkb->lkb_status;
163a1859	321	rl->rl_wait_type = cpu_to_le16(lkb->lkb_wait_type);
e7fd4179	322
e5dae548	323	if (lkb->lkb_bastfn)
e7fd4179	324	rl->rl_asts \|= AST_BAST;
e5dae548	325	if (lkb->lkb_astfn)
e7fd4179 DT	326	rl->rl_asts \|= AST_COMP;
e7fd4179 DT	327
163a1859	328	rl->rl_namelen = cpu_to_le16(r->res_length);
e7fd4179 DT	329	memcpy(rl->rl_name, r->res_name, r->res_length);
	330
	331	/* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
	332	If so, receive_rcom_lock_args() won't take this copy. */
	333
	334	if (lkb->lkb_lvbptr)
	335	memcpy(rl->rl_lvb, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
	336	}
	337
	338	int dlm_send_rcom_lock(struct dlm_rsb r, struct dlm_lkb lkb)
	339	{
	340	struct dlm_ls *ls = r->res_ls;
	341	struct dlm_rcom *rc;
	342	struct dlm_mhandle *mh;
	343	struct rcom_lock *rl;
	344	int error, len = sizeof(struct rcom_lock);
	345
	346	if (lkb->lkb_lvbptr)
	347	len += ls->ls_lvblen;
	348
	349	error = create_rcom(ls, r->res_nodeid, DLM_RCOM_LOCK, len, &rc, &mh);
	350	if (error)
	351	goto out;
	352
	353	rl = (struct rcom_lock *) rc->rc_buf;
	354	pack_rcom_lock(r, lkb, rl);
	355	rc->rc_id = (unsigned long) r;
	356
	357	send_rcom(ls, mh, rc);
	358	out:
	359	return error;
	360	}
	361
ae773d0b	362	/* needs at least dlm_rcom + rcom_lock */
e7fd4179 DT	363	static void receive_rcom_lock(struct dlm_ls ls, struct dlm_rcom rc_in)
	364	{
	365	struct dlm_rcom *rc;
	366	struct dlm_mhandle *mh;
	367	int error, nodeid = rc_in->rc_header.h_nodeid;
	368
	369	dlm_recover_master_copy(ls, rc_in);
	370
	371	error = create_rcom(ls, nodeid, DLM_RCOM_LOCK_REPLY,
	372	sizeof(struct rcom_lock), &rc, &mh);
	373	if (error)
	374	return;
	375
	376	/* We send back the same rcom_lock struct we received, but
	377	dlm_recover_master_copy() has filled in rl_remid and rl_result */
	378
	379	memcpy(rc->rc_buf, rc_in->rc_buf, sizeof(struct rcom_lock));
	380	rc->rc_id = rc_in->rc_id;
38aa8b0c	381	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	382
	383	send_rcom(ls, mh, rc);
	384	}
	385
c36258b5 DT	386	/* If the lockspace doesn't exist then still send a status message
	387	back; it's possible that it just doesn't have its global_id yet. */
	388
	389	int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
e7fd4179 DT	390	{
e7fd4179 DT	391	struct dlm_rcom *rc;
1babdb45	392	struct rcom_config *rf;
e7fd4179 DT	393	struct dlm_mhandle *mh;
e7fd4179 DT	394	char *mb;
1babdb45	395	int mb_len = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);
e7fd4179	396
41684f95	397	mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_NOFS, &mb);
e7fd4179 DT	398	if (!mh)
	399	return -ENOBUFS;
	400	memset(mb, 0, mb_len);
	401
	402	rc = (struct dlm_rcom *) mb;
	403
	404	rc->rc_header.h_version = (DLM_HEADER_MAJOR \| DLM_HEADER_MINOR);
	405	rc->rc_header.h_lockspace = rc_in->rc_header.h_lockspace;
	406	rc->rc_header.h_nodeid = dlm_our_nodeid();
	407	rc->rc_header.h_length = mb_len;
	408	rc->rc_header.h_cmd = DLM_RCOM;
	409
	410	rc->rc_type = DLM_RCOM_STATUS_REPLY;
f5888750	411	rc->rc_id = rc_in->rc_id;
38aa8b0c	412	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	413	rc->rc_result = -ESRCH;
e7fd4179 DT	414
1babdb45	415	rf = (struct rcom_config *) rc->rc_buf;
93ff2971	416	rf->rf_lvblen = cpu_to_le32(~0U);
1babdb45	417
e7fd4179 DT	418	dlm_rcom_out(rc);
	419	dlm_lowcomms_commit_buffer(mh);
	420
	421	return 0;
	422	}
	423
38aa8b0c DT	424	static int is_old_reply(struct dlm_ls ls, struct dlm_rcom rc)
	425	{
	426	uint64_t seq;
	427	int rv = 0;
	428
	429	switch (rc->rc_type) {
	430	case DLM_RCOM_STATUS_REPLY:
	431	case DLM_RCOM_NAMES_REPLY:
	432	case DLM_RCOM_LOOKUP_REPLY:
	433	case DLM_RCOM_LOCK_REPLY:
	434	spin_lock(&ls->ls_recover_lock);
	435	seq = ls->ls_recover_seq;
	436	spin_unlock(&ls->ls_recover_lock);
	437	if (rc->rc_seq_reply != seq) {
8ec68867	438	log_debug(ls, "ignoring old reply %x from %d "
38aa8b0c DT	439	"seq_reply %llx expect %llx",
	440	rc->rc_type, rc->rc_header.h_nodeid,
	441	(unsigned long long)rc->rc_seq_reply,
	442	(unsigned long long)seq);
	443	rv = 1;
	444	}
	445	}
	446	return rv;
	447	}
	448
c36258b5	449	/* Called by dlm_recv; corresponds to dlm_receive_message() but special
e7fd4179 DT	450	recovery-only comms are sent through here. */
e7fd4179 DT	451
c36258b5	452	void dlm_receive_rcom(struct dlm_ls ls, struct dlm_rcom rc, int nodeid)
e7fd4179	453	{
ae773d0b AV	454	int lock_size = sizeof(struct dlm_rcom) + sizeof(struct rcom_lock);
ae773d0b AV	455
e7fd4179	456	if (dlm_recovery_stopped(ls) && (rc->rc_type != DLM_RCOM_STATUS)) {
8ec68867	457	log_debug(ls, "ignoring recovery message %x from %d",
e7fd4179 DT	458	rc->rc_type, nodeid);
	459	goto out;
	460	}
	461
38aa8b0c DT	462	if (is_old_reply(ls, rc))
	463	goto out;
	464
e7fd4179 DT	465	switch (rc->rc_type) {
	466	case DLM_RCOM_STATUS:
	467	receive_rcom_status(ls, rc);
	468	break;
	469
	470	case DLM_RCOM_NAMES:
	471	receive_rcom_names(ls, rc);
	472	break;
	473
	474	case DLM_RCOM_LOOKUP:
	475	receive_rcom_lookup(ls, rc);
	476	break;
	477
	478	case DLM_RCOM_LOCK:
ae773d0b AV	479	if (rc->rc_header.h_length < lock_size)
ae773d0b AV	480	goto Eshort;
e7fd4179 DT	481	receive_rcom_lock(ls, rc);
	482	break;
	483
	484	case DLM_RCOM_STATUS_REPLY:
dbcfc347	485	receive_sync_reply(ls, rc);
e7fd4179 DT	486	break;
	487
	488	case DLM_RCOM_NAMES_REPLY:
dbcfc347	489	receive_sync_reply(ls, rc);
e7fd4179 DT	490	break;
	491
	492	case DLM_RCOM_LOOKUP_REPLY:
	493	receive_rcom_lookup_reply(ls, rc);
	494	break;
	495
	496	case DLM_RCOM_LOCK_REPLY:
ae773d0b AV	497	if (rc->rc_header.h_length < lock_size)
ae773d0b AV	498	goto Eshort;
dbcfc347	499	dlm_recover_process_copy(ls, rc);
e7fd4179 DT	500	break;
	501
	502	default:
dbcfc347	503	log_error(ls, "receive_rcom bad type %d", rc->rc_type);
e7fd4179	504	}
ae773d0b	505	out:
c36258b5	506	return;
ae773d0b AV	507	Eshort:
	508	log_error(ls, "recovery message %x from %d is too short",
	509	rc->rc_type, nodeid);
e7fd4179 DT	510	}
e7fd4179 DT	511