[deliverable/linux.git] / fs / dlm / rcom.c

/******************************************************************************
*******************************************************************************
**
**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
**  Copyright (C) 2005 Red Hat, Inc.  All rights reserved.
**
**  This copyrighted material is made available to anyone wishing to use,
**  modify, copy, or redistribute it subject to the terms and conditions
**  of the GNU General Public License v.2.
**
*******************************************************************************
******************************************************************************/

#include "dlm_internal.h"
#include "lockspace.h"
#include "member.h"
#include "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_KERNEL, &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 = ls->ls_lvblen;
	rf->rf_lsflags = ls->ls_exflags;
}

static int check_config(struct dlm_ls *ls, struct rcom_config *rf, int nodeid)
{
	if (rf->rf_lvblen != ls->ls_lvblen ||
	    rf->rf_lsflags != ls->ls_exflags) {
		log_error(ls, "config mismatch: %d,%x nodeid %d: %d,%x",
			  ls->ls_lvblen, ls->ls_exflags,
			  nodeid, rf->rf_lvblen, rf->rf_lsflags);
		return -EINVAL;
	}
	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 = (struct dlm_rcom *) 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.buffer_size);

	send_rcom(ls, mh, rc);

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

	rc = (struct dlm_rcom *) 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, (struct rcom_config *) rc->rc_buf,
				     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);
}

static void receive_rcom_status_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	receive_sync_reply(ls, rc_in);
}

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, len = sizeof(struct dlm_rcom);

	ls->ls_recover_nodeid = nodeid;

	if (nodeid == dlm_our_nodeid()) {
		dlm_copy_master_names(ls, last_name, last_len,
		                      ls->ls_recover_buf + len,
		                      dlm_config.buffer_size - len, 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.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.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);
}

static void receive_rcom_names_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	receive_sync_reply(ls, rc_in);
}

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 = lkb->lkb_ownpid;
	rl->rl_lkid = lkb->lkb_id;
	rl->rl_exflags = lkb->lkb_exflags;
	rl->rl_flags = lkb->lkb_flags;
	rl->rl_lvbseq = 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 = lkb->lkb_wait_type;

	if (lkb->lkb_bastaddr)
		rl->rl_asts |= AST_BAST;
	if (lkb->lkb_astaddr)
		rl->rl_asts |= AST_COMP;

	rl->rl_namelen = 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;
}

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

static void receive_rcom_lock_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	dlm_recover_process_copy(ls, rc_in);
}

static int 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_KERNEL, &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 = -1;

	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_error(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_recvd; corresponds to dlm_receive_message() but special
   recovery-only comms are sent through here. */

void dlm_receive_rcom(struct dlm_header *hd, int nodeid)
{
	struct dlm_rcom *rc = (struct dlm_rcom *) hd;
	struct dlm_ls *ls;

	dlm_rcom_in(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. */

	ls = dlm_find_lockspace_global(hd->h_lockspace);
	if (!ls) {
		log_print("lockspace %x from %d type %x not found",
			  hd->h_lockspace, nodeid, rc->rc_type);
		if (rc->rc_type == DLM_RCOM_STATUS)
			send_ls_not_ready(nodeid, rc);
		return;
	}

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

	if (is_old_reply(ls, rc))
		goto out;

	if (nodeid != rc->rc_header.h_nodeid) {
		log_error(ls, "bad rcom nodeid %d from %d",
			  rc->rc_header.h_nodeid, nodeid);
		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:
		receive_rcom_lock(ls, rc);
		break;

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

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

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

	case DLM_RCOM_LOCK_REPLY:
		receive_rcom_lock_reply(ls, rc);
		break;

	default:
		DLM_ASSERT(0, printk("rc_type=%x\n", rc->rc_type););
	}
 out:
	dlm_put_lockspace(ls);
}
Commit	Line	Data
e7fd4179 DT	1	/******************************************************************************
	2	*******************************************************************************
	3	**
	4	** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
	5	** Copyright (C) 2005 Red Hat, Inc. All rights reserved.
	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
	41	mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_KERNEL, &mb);
	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	{
	81	rf->rf_lvblen = ls->ls_lvblen;
	82	rf->rf_lsflags = ls->ls_exflags;
	83	}
	84
	85	static int check_config(struct dlm_ls ls, struct rcom_config rf, int nodeid)
	86	{
	87	if (rf->rf_lvblen != ls->ls_lvblen \|\|
	88	rf->rf_lsflags != ls->ls_exflags) {
	89	log_error(ls, "config mismatch: %d,%x nodeid %d: %d,%x",
	90	ls->ls_lvblen, ls->ls_exflags,
	91	nodeid, rf->rf_lvblen, rf->rf_lsflags);
	92	return -EINVAL;
	93	}
	94	return 0;
	95	}
	96
98f176fb DT	97	static void allow_sync_reply(struct dlm_ls ls, uint64_t new_seq)
	98	{
	99	spin_lock(&ls->ls_rcom_spin);
	100	*new_seq = ++ls->ls_rcom_seq;
	101	set_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	102	spin_unlock(&ls->ls_rcom_spin);
	103	}
	104
	105	static void disallow_sync_reply(struct dlm_ls *ls)
	106	{
	107	spin_lock(&ls->ls_rcom_spin);
	108	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
	109	clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
	110	spin_unlock(&ls->ls_rcom_spin);
	111	}
	112
e7fd4179 DT	113	int dlm_rcom_status(struct dlm_ls *ls, int nodeid)
	114	{
	115	struct dlm_rcom *rc;
	116	struct dlm_mhandle *mh;
	117	int error = 0;
	118
faa0f267	119	ls->ls_recover_nodeid = nodeid;
e7fd4179 DT	120
	121	if (nodeid == dlm_our_nodeid()) {
	122	rc = (struct dlm_rcom *) ls->ls_recover_buf;
	123	rc->rc_result = dlm_recover_status(ls);
	124	goto out;
	125	}
	126
	127	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS, 0, &rc, &mh);
	128	if (error)
	129	goto out;
98f176fb DT	130
	131	allow_sync_reply(ls, &rc->rc_id);
	132	memset(ls->ls_recover_buf, 0, dlm_config.buffer_size);
e7fd4179 DT	133
	134	send_rcom(ls, mh, rc);
	135
	136	error = dlm_wait_function(ls, &rcom_response);
98f176fb	137	disallow_sync_reply(ls);
e7fd4179 DT	138	if (error)
	139	goto out;
	140
	141	rc = (struct dlm_rcom *) ls->ls_recover_buf;
	142
	143	if (rc->rc_result == -ESRCH) {
	144	/* we pretend the remote lockspace exists with 0 status */
	145	log_debug(ls, "remote node %d not ready", nodeid);
	146	rc->rc_result = 0;
	147	} else
	148	error = check_config(ls, (struct rcom_config *) rc->rc_buf,
	149	nodeid);
	150	/* the caller looks at rc_result for the remote recovery status */
	151	out:
	152	return error;
	153	}
	154
	155	static void receive_rcom_status(struct dlm_ls ls, struct dlm_rcom rc_in)
	156	{
	157	struct dlm_rcom *rc;
	158	struct dlm_mhandle *mh;
	159	int error, nodeid = rc_in->rc_header.h_nodeid;
	160
	161	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY,
	162	sizeof(struct rcom_config), &rc, &mh);
	163	if (error)
	164	return;
4a99c3d9	165	rc->rc_id = rc_in->rc_id;
38aa8b0c	166	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	167	rc->rc_result = dlm_recover_status(ls);
	168	make_config(ls, (struct rcom_config *) rc->rc_buf);
	169
	170	send_rcom(ls, mh, rc);
	171	}
	172
4a99c3d9	173	static void receive_sync_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
e7fd4179	174	{
98f176fb DT	175	spin_lock(&ls->ls_rcom_spin);
	176	if (!test_bit(LSFL_RCOM_WAIT, &ls->ls_flags) \|\|
	177	rc_in->rc_id != ls->ls_rcom_seq) {
	178	log_debug(ls, "reject reply %d from %d seq %llx expect %llx",
	179	rc_in->rc_type, rc_in->rc_header.h_nodeid,
57adf7ee RK	180	(unsigned long long)rc_in->rc_id,
57adf7ee RK	181	(unsigned long long)ls->ls_rcom_seq);
98f176fb	182	goto out;
4a99c3d9	183	}
e7fd4179 DT	184	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
e7fd4179 DT	185	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
98f176fb	186	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
e7fd4179	187	wake_up(&ls->ls_wait_general);
98f176fb DT	188	out:
98f176fb DT	189	spin_unlock(&ls->ls_rcom_spin);
e7fd4179 DT	190	}
e7fd4179 DT	191
4a99c3d9 DT	192	static void receive_rcom_status_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	193	{
	194	receive_sync_reply(ls, rc_in);
	195	}
	196
e7fd4179 DT	197	int dlm_rcom_names(struct dlm_ls ls, int nodeid, char last_name, int last_len)
	198	{
	199	struct dlm_rcom *rc;
	200	struct dlm_mhandle *mh;
	201	int error = 0, len = sizeof(struct dlm_rcom);
	202
faa0f267	203	ls->ls_recover_nodeid = nodeid;
e7fd4179 DT	204
	205	if (nodeid == dlm_our_nodeid()) {
	206	dlm_copy_master_names(ls, last_name, last_len,
	207	ls->ls_recover_buf + len,
	208	dlm_config.buffer_size - len, nodeid);
	209	goto out;
	210	}
	211
	212	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES, last_len, &rc, &mh);
	213	if (error)
	214	goto out;
	215	memcpy(rc->rc_buf, last_name, last_len);
98f176fb DT	216
	217	allow_sync_reply(ls, &rc->rc_id);
	218	memset(ls->ls_recover_buf, 0, dlm_config.buffer_size);
e7fd4179 DT	219
	220	send_rcom(ls, mh, rc);
	221
	222	error = dlm_wait_function(ls, &rcom_response);
98f176fb	223	disallow_sync_reply(ls);
e7fd4179 DT	224	out:
	225	return error;
	226	}
	227
	228	static void receive_rcom_names(struct dlm_ls ls, struct dlm_rcom rc_in)
	229	{
	230	struct dlm_rcom *rc;
	231	struct dlm_mhandle *mh;
38aa8b0c	232	int error, inlen, outlen, nodeid;
e7fd4179 DT	233
	234	nodeid = rc_in->rc_header.h_nodeid;
	235	inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
	236	outlen = dlm_config.buffer_size - sizeof(struct dlm_rcom);
	237
	238	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh);
	239	if (error)
	240	return;
4a99c3d9	241	rc->rc_id = rc_in->rc_id;
38aa8b0c	242	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	243
	244	dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen,
	245	nodeid);
	246	send_rcom(ls, mh, rc);
	247	}
	248
	249	static void receive_rcom_names_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	250	{
4a99c3d9	251	receive_sync_reply(ls, rc_in);
e7fd4179 DT	252	}
	253
	254	int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid)
	255	{
	256	struct dlm_rcom *rc;
	257	struct dlm_mhandle *mh;
	258	struct dlm_ls *ls = r->res_ls;
	259	int error;
	260
	261	error = create_rcom(ls, dir_nodeid, DLM_RCOM_LOOKUP, r->res_length,
	262	&rc, &mh);
	263	if (error)
	264	goto out;
	265	memcpy(rc->rc_buf, r->res_name, r->res_length);
	266	rc->rc_id = (unsigned long) r;
	267
	268	send_rcom(ls, mh, rc);
	269	out:
	270	return error;
	271	}
	272
	273	static void receive_rcom_lookup(struct dlm_ls ls, struct dlm_rcom rc_in)
	274	{
	275	struct dlm_rcom *rc;
	276	struct dlm_mhandle *mh;
	277	int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid;
	278	int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
	279
	280	error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh);
	281	if (error)
	282	return;
	283
	284	error = dlm_dir_lookup(ls, nodeid, rc_in->rc_buf, len, &ret_nodeid);
	285	if (error)
	286	ret_nodeid = error;
	287	rc->rc_result = ret_nodeid;
	288	rc->rc_id = rc_in->rc_id;
38aa8b0c	289	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	290
	291	send_rcom(ls, mh, rc);
	292	}
	293
	294	static void receive_rcom_lookup_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	295	{
	296	dlm_recover_master_reply(ls, rc_in);
	297	}
	298
	299	static void pack_rcom_lock(struct dlm_rsb r, struct dlm_lkb lkb,
	300	struct rcom_lock *rl)
	301	{
	302	memset(rl, 0, sizeof(*rl));
	303
	304	rl->rl_ownpid = lkb->lkb_ownpid;
	305	rl->rl_lkid = lkb->lkb_id;
	306	rl->rl_exflags = lkb->lkb_exflags;
	307	rl->rl_flags = lkb->lkb_flags;
	308	rl->rl_lvbseq = lkb->lkb_lvbseq;
	309	rl->rl_rqmode = lkb->lkb_rqmode;
	310	rl->rl_grmode = lkb->lkb_grmode;
	311	rl->rl_status = lkb->lkb_status;
	312	rl->rl_wait_type = lkb->lkb_wait_type;
	313
	314	if (lkb->lkb_bastaddr)
	315	rl->rl_asts \|= AST_BAST;
	316	if (lkb->lkb_astaddr)
	317	rl->rl_asts \|= AST_COMP;
	318
e7fd4179 DT	319	rl->rl_namelen = r->res_length;
	320	memcpy(rl->rl_name, r->res_name, r->res_length);
	321
	322	/* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
	323	If so, receive_rcom_lock_args() won't take this copy. */
	324
	325	if (lkb->lkb_lvbptr)
	326	memcpy(rl->rl_lvb, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
	327	}
	328
	329	int dlm_send_rcom_lock(struct dlm_rsb r, struct dlm_lkb lkb)
	330	{
	331	struct dlm_ls *ls = r->res_ls;
	332	struct dlm_rcom *rc;
	333	struct dlm_mhandle *mh;
	334	struct rcom_lock *rl;
	335	int error, len = sizeof(struct rcom_lock);
	336
	337	if (lkb->lkb_lvbptr)
	338	len += ls->ls_lvblen;
	339
	340	error = create_rcom(ls, r->res_nodeid, DLM_RCOM_LOCK, len, &rc, &mh);
	341	if (error)
	342	goto out;
	343
	344	rl = (struct rcom_lock *) rc->rc_buf;
	345	pack_rcom_lock(r, lkb, rl);
	346	rc->rc_id = (unsigned long) r;
	347
	348	send_rcom(ls, mh, rc);
	349	out:
	350	return error;
	351	}
	352
	353	static void receive_rcom_lock(struct dlm_ls ls, struct dlm_rcom rc_in)
	354	{
	355	struct dlm_rcom *rc;
	356	struct dlm_mhandle *mh;
	357	int error, nodeid = rc_in->rc_header.h_nodeid;
	358
	359	dlm_recover_master_copy(ls, rc_in);
	360
	361	error = create_rcom(ls, nodeid, DLM_RCOM_LOCK_REPLY,
	362	sizeof(struct rcom_lock), &rc, &mh);
	363	if (error)
	364	return;
	365
	366	/* We send back the same rcom_lock struct we received, but
	367	dlm_recover_master_copy() has filled in rl_remid and rl_result */
	368
	369	memcpy(rc->rc_buf, rc_in->rc_buf, sizeof(struct rcom_lock));
	370	rc->rc_id = rc_in->rc_id;
38aa8b0c	371	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	372
	373	send_rcom(ls, mh, rc);
	374	}
	375
	376	static void receive_rcom_lock_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	377	{
e7fd4179 DT	378	dlm_recover_process_copy(ls, rc_in);
	379	}
	380
	381	static int send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
	382	{
	383	struct dlm_rcom *rc;
1babdb45	384	struct rcom_config *rf;
e7fd4179 DT	385	struct dlm_mhandle *mh;
e7fd4179 DT	386	char *mb;
1babdb45	387	int mb_len = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);
e7fd4179 DT	388
	389	mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_KERNEL, &mb);
	390	if (!mh)
	391	return -ENOBUFS;
	392	memset(mb, 0, mb_len);
	393
	394	rc = (struct dlm_rcom *) mb;
	395
	396	rc->rc_header.h_version = (DLM_HEADER_MAJOR \| DLM_HEADER_MINOR);
	397	rc->rc_header.h_lockspace = rc_in->rc_header.h_lockspace;
	398	rc->rc_header.h_nodeid = dlm_our_nodeid();
	399	rc->rc_header.h_length = mb_len;
	400	rc->rc_header.h_cmd = DLM_RCOM;
	401
	402	rc->rc_type = DLM_RCOM_STATUS_REPLY;
f5888750	403	rc->rc_id = rc_in->rc_id;
38aa8b0c	404	rc->rc_seq_reply = rc_in->rc_seq;
e7fd4179 DT	405	rc->rc_result = -ESRCH;
e7fd4179 DT	406
1babdb45 DT	407	rf = (struct rcom_config *) rc->rc_buf;
	408	rf->rf_lvblen = -1;
	409
e7fd4179 DT	410	dlm_rcom_out(rc);
	411	dlm_lowcomms_commit_buffer(mh);
	412
	413	return 0;
	414	}
	415
38aa8b0c DT	416	static int is_old_reply(struct dlm_ls ls, struct dlm_rcom rc)
	417	{
	418	uint64_t seq;
	419	int rv = 0;
	420
	421	switch (rc->rc_type) {
	422	case DLM_RCOM_STATUS_REPLY:
	423	case DLM_RCOM_NAMES_REPLY:
	424	case DLM_RCOM_LOOKUP_REPLY:
	425	case DLM_RCOM_LOCK_REPLY:
	426	spin_lock(&ls->ls_recover_lock);
	427	seq = ls->ls_recover_seq;
	428	spin_unlock(&ls->ls_recover_lock);
	429	if (rc->rc_seq_reply != seq) {
	430	log_error(ls, "ignoring old reply %x from %d "
	431	"seq_reply %llx expect %llx",
	432	rc->rc_type, rc->rc_header.h_nodeid,
	433	(unsigned long long)rc->rc_seq_reply,
	434	(unsigned long long)seq);
	435	rv = 1;
	436	}
	437	}
	438	return rv;
	439	}
	440
e7fd4179 DT	441	/* Called by dlm_recvd; corresponds to dlm_receive_message() but special
	442	recovery-only comms are sent through here. */
	443
	444	void dlm_receive_rcom(struct dlm_header *hd, int nodeid)
	445	{
	446	struct dlm_rcom rc = (struct dlm_rcom ) hd;
	447	struct dlm_ls *ls;
	448
	449	dlm_rcom_in(rc);
	450
	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	ls = dlm_find_lockspace_global(hd->h_lockspace);
	455	if (!ls) {
435618b7 DT	456	log_print("lockspace %x from %d type %x not found",
	457	hd->h_lockspace, nodeid, rc->rc_type);
	458	if (rc->rc_type == DLM_RCOM_STATUS)
	459	send_ls_not_ready(nodeid, rc);
e7fd4179 DT	460	return;
	461	}
	462
	463	if (dlm_recovery_stopped(ls) && (rc->rc_type != DLM_RCOM_STATUS)) {
	464	log_error(ls, "ignoring recovery message %x from %d",
	465	rc->rc_type, nodeid);
	466	goto out;
	467	}
	468
38aa8b0c DT	469	if (is_old_reply(ls, rc))
	470	goto out;
	471
e7fd4179 DT	472	if (nodeid != rc->rc_header.h_nodeid) {
	473	log_error(ls, "bad rcom nodeid %d from %d",
	474	rc->rc_header.h_nodeid, nodeid);
	475	goto out;
	476	}
	477
	478	switch (rc->rc_type) {
	479	case DLM_RCOM_STATUS:
	480	receive_rcom_status(ls, rc);
	481	break;
	482
	483	case DLM_RCOM_NAMES:
	484	receive_rcom_names(ls, rc);
	485	break;
	486
	487	case DLM_RCOM_LOOKUP:
	488	receive_rcom_lookup(ls, rc);
	489	break;
	490
	491	case DLM_RCOM_LOCK:
	492	receive_rcom_lock(ls, rc);
	493	break;
	494
	495	case DLM_RCOM_STATUS_REPLY:
	496	receive_rcom_status_reply(ls, rc);
	497	break;
	498
	499	case DLM_RCOM_NAMES_REPLY:
	500	receive_rcom_names_reply(ls, rc);
	501	break;
	502
	503	case DLM_RCOM_LOOKUP_REPLY:
	504	receive_rcom_lookup_reply(ls, rc);
	505	break;
	506
	507	case DLM_RCOM_LOCK_REPLY:
	508	receive_rcom_lock_reply(ls, rc);
	509	break;
	510
	511	default:
	512	DLM_ASSERT(0, printk("rc_type=%x\n", rc->rc_type););
	513	}
	514	out:
	515	dlm_put_lockspace(ls);
	516	}
	517