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

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

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

	memset(ls->ls_recover_buf, 0, dlm_config.buffer_size);
	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;

	send_rcom(ls, mh, rc);

	error = dlm_wait_function(ls, &rcom_response);
	clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
	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_result = dlm_recover_status(ls);
	make_config(ls, (struct rcom_config *) rc->rc_buf);

	send_rcom(ls, mh, rc);
}

static void receive_rcom_status_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
	wake_up(&ls->ls_wait_general);
}

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

	memset(ls->ls_recover_buf, 0, dlm_config.buffer_size);
	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);

	send_rcom(ls, mh, rc);

	error = dlm_wait_function(ls, &rcom_response);
	clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
 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;
	int nodeid = rc_in->rc_header.h_nodeid;
	uint32_t status = dlm_recover_status(ls);

	/*
	 * We can't run dlm_dir_rebuild_send (which uses ls_nodes) while
	 * dlm_recoverd is running ls_nodes_reconfig (which changes ls_nodes).
	 * It could only happen in rare cases where we get a late NAMES
	 * message from a previous instance of recovery.
	 */

	if (!(status & DLM_RS_NODES)) {
		log_debug(ls, "ignoring RCOM_NAMES from %u", nodeid);
		return;
	}

	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;

	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)
{
	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
	wake_up(&ls->ls_wait_general);
}

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;

	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;

	send_rcom(ls, mh, rc);
}

static void receive_rcom_lock_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	uint32_t status = dlm_recover_status(ls);

	if (!(status & DLM_RS_DIR)) {
		log_debug(ls, "ignoring RCOM_LOCK_REPLY from %u",
			  rc_in->rc_header.h_nodeid);
		return;
	}

	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 dlm_mhandle *mh;
	char *mb;
	int mb_len = sizeof(struct dlm_rcom);

	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_result = -ESRCH;

	dlm_rcom_out(rc);
	dlm_lowcomms_commit_buffer(mh);

	return 0;
}

/* 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 not found",
			  hd->h_lockspace, nodeid);
		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 (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
	59	*mh_ret = mh;
	60	*rc_ret = rc;
	61	return 0;
	62	}
	63
	64	static void send_rcom(struct dlm_ls ls, struct dlm_mhandle mh,
65	struct dlm_rcom *rc)
66	{
67	dlm_rcom_out(rc);
68	dlm_lowcomms_commit_buffer(mh);
69	}
70
71	/* When replying to a status request, a node also sends back its
72	configuration values. The requesting node then checks that the remote
73	node is configured the same way as itself. */
74
75	static void make_config(struct dlm_ls ls, struct rcom_config rf)
76	{
77	rf->rf_lvblen = ls->ls_lvblen;
78	rf->rf_lsflags = ls->ls_exflags;
79	}
80
81	static int check_config(struct dlm_ls ls, struct rcom_config rf, int nodeid)
82	{
83	if (rf->rf_lvblen != ls->ls_lvblen \|\|
84	rf->rf_lsflags != ls->ls_exflags) {
85	log_error(ls, "config mismatch: %d,%x nodeid %d: %d,%x",
86	ls->ls_lvblen, ls->ls_exflags,
87	nodeid, rf->rf_lvblen, rf->rf_lsflags);
88	return -EINVAL;
89	}
90	return 0;
91	}
92
93	int dlm_rcom_status(struct dlm_ls *ls, int nodeid)
94	{
95	struct dlm_rcom *rc;
96	struct dlm_mhandle *mh;
97	int error = 0;
98
99	memset(ls->ls_recover_buf, 0, dlm_config.buffer_size);
faa0f267	100	ls->ls_recover_nodeid = nodeid;
e7fd4179 DT	101
	102	if (nodeid == dlm_our_nodeid()) {
	103	rc = (struct dlm_rcom *) ls->ls_recover_buf;
	104	rc->rc_result = dlm_recover_status(ls);
	105	goto out;
	106	}
	107
	108	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS, 0, &rc, &mh);
	109	if (error)
	110	goto out;
	111
	112	send_rcom(ls, mh, rc);
	113
	114	error = dlm_wait_function(ls, &rcom_response);
	115	clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
	116	if (error)
	117	goto out;
	118
	119	rc = (struct dlm_rcom *) ls->ls_recover_buf;
	120
	121	if (rc->rc_result == -ESRCH) {
	122	/* we pretend the remote lockspace exists with 0 status */
	123	log_debug(ls, "remote node %d not ready", nodeid);
	124	rc->rc_result = 0;
	125	} else
	126	error = check_config(ls, (struct rcom_config *) rc->rc_buf,
	127	nodeid);
	128	/* the caller looks at rc_result for the remote recovery status */
	129	out:
	130	return error;
	131	}
	132
	133	static void receive_rcom_status(struct dlm_ls ls, struct dlm_rcom rc_in)
	134	{
	135	struct dlm_rcom *rc;
	136	struct dlm_mhandle *mh;
	137	int error, nodeid = rc_in->rc_header.h_nodeid;
	138
	139	error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY,
	140	sizeof(struct rcom_config), &rc, &mh);
	141	if (error)
	142	return;
	143	rc->rc_result = dlm_recover_status(ls);
	144	make_config(ls, (struct rcom_config *) rc->rc_buf);
	145
	146	send_rcom(ls, mh, rc);
	147	}
	148
	149	static void receive_rcom_status_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	150	{
	151	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
	152	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
	153	wake_up(&ls->ls_wait_general);
	154	}
	155
	156	int dlm_rcom_names(struct dlm_ls ls, int nodeid, char last_name, int last_len)
	157	{
	158	struct dlm_rcom *rc;
	159	struct dlm_mhandle *mh;
	160	int error = 0, len = sizeof(struct dlm_rcom);
	161
	162	memset(ls->ls_recover_buf, 0, dlm_config.buffer_size);
faa0f267	163	ls->ls_recover_nodeid = nodeid;
e7fd4179 DT	164
	165	if (nodeid == dlm_our_nodeid()) {
	166	dlm_copy_master_names(ls, last_name, last_len,
	167	ls->ls_recover_buf + len,
	168	dlm_config.buffer_size - len, nodeid);
	169	goto out;
	170	}
	171
	172	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES, last_len, &rc, &mh);
	173	if (error)
	174	goto out;
	175	memcpy(rc->rc_buf, last_name, last_len);
	176
	177	send_rcom(ls, mh, rc);
	178
	179	error = dlm_wait_function(ls, &rcom_response);
	180	clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
	181	out:
	182	return error;
	183	}
	184
	185	static void receive_rcom_names(struct dlm_ls ls, struct dlm_rcom rc_in)
	186	{
	187	struct dlm_rcom *rc;
	188	struct dlm_mhandle *mh;
	189	int error, inlen, outlen;
	190	int nodeid = rc_in->rc_header.h_nodeid;
	191	uint32_t status = dlm_recover_status(ls);
	192
	193	/*
	194	* We can't run dlm_dir_rebuild_send (which uses ls_nodes) while
	195	* dlm_recoverd is running ls_nodes_reconfig (which changes ls_nodes).
	196	* It could only happen in rare cases where we get a late NAMES
	197	* message from a previous instance of recovery.
	198	*/
	199
	200	if (!(status & DLM_RS_NODES)) {
	201	log_debug(ls, "ignoring RCOM_NAMES from %u", nodeid);
	202	return;
	203	}
	204
	205	nodeid = rc_in->rc_header.h_nodeid;
	206	inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
	207	outlen = dlm_config.buffer_size - sizeof(struct dlm_rcom);
	208
	209	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh);
	210	if (error)
	211	return;
	212
	213	dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen,
	214	nodeid);
	215	send_rcom(ls, mh, rc);
	216	}
	217
	218	static void receive_rcom_names_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	219	{
	220	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
	221	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
	222	wake_up(&ls->ls_wait_general);
	223	}
	224
	225	int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid)
	226	{
	227	struct dlm_rcom *rc;
228	struct dlm_mhandle *mh;
229	struct dlm_ls *ls = r->res_ls;
230	int error;
231
232	error = create_rcom(ls, dir_nodeid, DLM_RCOM_LOOKUP, r->res_length,
233	&rc, &mh);
234	if (error)
235	goto out;
236	memcpy(rc->rc_buf, r->res_name, r->res_length);
237	rc->rc_id = (unsigned long) r;
238
239	send_rcom(ls, mh, rc);
240	out:
241	return error;
242	}
243
244	static void receive_rcom_lookup(struct dlm_ls ls, struct dlm_rcom rc_in)
245	{
246	struct dlm_rcom *rc;
247	struct dlm_mhandle *mh;
248	int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid;
249	int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
250
251	error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh);
252	if (error)
253	return;
254
255	error = dlm_dir_lookup(ls, nodeid, rc_in->rc_buf, len, &ret_nodeid);
256	if (error)
257	ret_nodeid = error;
258	rc->rc_result = ret_nodeid;
259	rc->rc_id = rc_in->rc_id;
260
261	send_rcom(ls, mh, rc);
262	}
263
264	static void receive_rcom_lookup_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
265	{
266	dlm_recover_master_reply(ls, rc_in);
267	}
268
269	static void pack_rcom_lock(struct dlm_rsb r, struct dlm_lkb lkb,
270	struct rcom_lock *rl)
271	{
272	memset(rl, 0, sizeof(*rl));
273
274	rl->rl_ownpid = lkb->lkb_ownpid;
275	rl->rl_lkid = lkb->lkb_id;
276	rl->rl_exflags = lkb->lkb_exflags;
277	rl->rl_flags = lkb->lkb_flags;
278	rl->rl_lvbseq = lkb->lkb_lvbseq;
279	rl->rl_rqmode = lkb->lkb_rqmode;
280	rl->rl_grmode = lkb->lkb_grmode;
281	rl->rl_status = lkb->lkb_status;
282	rl->rl_wait_type = lkb->lkb_wait_type;
283
284	if (lkb->lkb_bastaddr)
285	rl->rl_asts \|= AST_BAST;
286	if (lkb->lkb_astaddr)
287	rl->rl_asts \|= AST_COMP;
288
e7fd4179 DT	289	rl->rl_namelen = r->res_length;
	290	memcpy(rl->rl_name, r->res_name, r->res_length);
	291
	292	/* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
	293	If so, receive_rcom_lock_args() won't take this copy. */
	294
	295	if (lkb->lkb_lvbptr)
	296	memcpy(rl->rl_lvb, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
	297	}
	298
	299	int dlm_send_rcom_lock(struct dlm_rsb r, struct dlm_lkb lkb)
	300	{
	301	struct dlm_ls *ls = r->res_ls;
	302	struct dlm_rcom *rc;
	303	struct dlm_mhandle *mh;
	304	struct rcom_lock *rl;
	305	int error, len = sizeof(struct rcom_lock);
	306
	307	if (lkb->lkb_lvbptr)
	308	len += ls->ls_lvblen;
	309
	310	error = create_rcom(ls, r->res_nodeid, DLM_RCOM_LOCK, len, &rc, &mh);
	311	if (error)
	312	goto out;
	313
	314	rl = (struct rcom_lock *) rc->rc_buf;
	315	pack_rcom_lock(r, lkb, rl);
	316	rc->rc_id = (unsigned long) r;
	317
	318	send_rcom(ls, mh, rc);
	319	out:
	320	return error;
	321	}
	322
	323	static void receive_rcom_lock(struct dlm_ls ls, struct dlm_rcom rc_in)
	324	{
	325	struct dlm_rcom *rc;
	326	struct dlm_mhandle *mh;
	327	int error, nodeid = rc_in->rc_header.h_nodeid;
	328
	329	dlm_recover_master_copy(ls, rc_in);
	330
	331	error = create_rcom(ls, nodeid, DLM_RCOM_LOCK_REPLY,
	332	sizeof(struct rcom_lock), &rc, &mh);
	333	if (error)
	334	return;
	335
	336	/* We send back the same rcom_lock struct we received, but
	337	dlm_recover_master_copy() has filled in rl_remid and rl_result */
	338
	339	memcpy(rc->rc_buf, rc_in->rc_buf, sizeof(struct rcom_lock));
	340	rc->rc_id = rc_in->rc_id;
	341
	342	send_rcom(ls, mh, rc);
	343	}
	344
	345	static void receive_rcom_lock_reply(struct dlm_ls ls, struct dlm_rcom rc_in)
	346	{
	347	uint32_t status = dlm_recover_status(ls);
	348
	349	if (!(status & DLM_RS_DIR)) {
	350	log_debug(ls, "ignoring RCOM_LOCK_REPLY from %u",
	351	rc_in->rc_header.h_nodeid);
	352	return;
353	}
354
355	dlm_recover_process_copy(ls, rc_in);
356	}
357
358	static int send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
359	{
360	struct dlm_rcom *rc;
361	struct dlm_mhandle *mh;
362	char *mb;
363	int mb_len = sizeof(struct dlm_rcom);
364
365	mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_KERNEL, &mb);
366	if (!mh)
367	return -ENOBUFS;
368	memset(mb, 0, mb_len);
369
370	rc = (struct dlm_rcom *) mb;
371
372	rc->rc_header.h_version = (DLM_HEADER_MAJOR \| DLM_HEADER_MINOR);
373	rc->rc_header.h_lockspace = rc_in->rc_header.h_lockspace;
374	rc->rc_header.h_nodeid = dlm_our_nodeid();
375	rc->rc_header.h_length = mb_len;
376	rc->rc_header.h_cmd = DLM_RCOM;
377
378	rc->rc_type = DLM_RCOM_STATUS_REPLY;
379	rc->rc_result = -ESRCH;
380
381	dlm_rcom_out(rc);
382	dlm_lowcomms_commit_buffer(mh);
383
384	return 0;
385	}
386
387	/* Called by dlm_recvd; corresponds to dlm_receive_message() but special
388	recovery-only comms are sent through here. */
389
390	void dlm_receive_rcom(struct dlm_header *hd, int nodeid)
391	{
392	struct dlm_rcom rc = (struct dlm_rcom ) hd;
393	struct dlm_ls *ls;
394
395	dlm_rcom_in(rc);
396
397	/* If the lockspace doesn't exist then still send a status message
398	back; it's possible that it just doesn't have its global_id yet. */
399
400	ls = dlm_find_lockspace_global(hd->h_lockspace);
401	if (!ls) {
402	log_print("lockspace %x from %d not found",
403	hd->h_lockspace, nodeid);
404	send_ls_not_ready(nodeid, rc);
405	return;
406	}
407
408	if (dlm_recovery_stopped(ls) && (rc->rc_type != DLM_RCOM_STATUS)) {
409	log_error(ls, "ignoring recovery message %x from %d",
410	rc->rc_type, nodeid);
411	goto out;
412	}
413
414	if (nodeid != rc->rc_header.h_nodeid) {
415	log_error(ls, "bad rcom nodeid %d from %d",
416	rc->rc_header.h_nodeid, nodeid);
417	goto out;
418	}
419
420	switch (rc->rc_type) {
421	case DLM_RCOM_STATUS:
422	receive_rcom_status(ls, rc);
423	break;
424
425	case DLM_RCOM_NAMES:
426	receive_rcom_names(ls, rc);
427	break;
428
429	case DLM_RCOM_LOOKUP:
430	receive_rcom_lookup(ls, rc);
431	break;
432
433	case DLM_RCOM_LOCK:
434	receive_rcom_lock(ls, rc);
435	break;
436
437	case DLM_RCOM_STATUS_REPLY:
438	receive_rcom_status_reply(ls, rc);
439	break;
440
441	case DLM_RCOM_NAMES_REPLY:
442	receive_rcom_names_reply(ls, rc);
443	break;
444
445	case DLM_RCOM_LOOKUP_REPLY:
446	receive_rcom_lookup_reply(ls, rc);
447	break;
448
449	case DLM_RCOM_LOCK_REPLY:
450	receive_rcom_lock_reply(ls, rc);
451	break;
452
453	default:
454	DLM_ASSERT(0, printk("rc_type=%x\n", rc->rc_type););
455	}
456	out:
457	dlm_put_lockspace(ls);
458	}
459