[deliverable/linux.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);
}

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 (!(le32_to_cpu(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;

	ls->ls_recover_nodeid = nodeid;

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

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

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

	error = create_rcom(ls, to_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 = 0xFFFFFFFF;

	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;

	if (rc_in->rc_id == 0xFFFFFFFF) {
		log_error(ls, "receive_rcom_lookup dump from %d", nodeid);
		dlm_dump_rsb_name(ls, rc_in->rc_buf, len);
		return;
	}

	error = dlm_master_lookup(ls, nodeid, rc_in->rc_buf, len,
				  DLM_LU_RECOVER_MASTER, &ret_nodeid, NULL);
	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;
}

/*
 * Ignore messages for stage Y before we set
 * recover_status bit for stage X:
 *
 * recover_status = 0
 *
 * dlm_recover_members()
 * - send nothing
 * - recv nothing
 * - ignore NAMES, NAMES_REPLY
 * - ignore LOOKUP, LOOKUP_REPLY
 * - ignore LOCK, LOCK_REPLY
 *
 * recover_status |= NODES
 *
 * dlm_recover_members_wait()
 *
 * dlm_recover_directory()
 * - send NAMES
 * - recv NAMES_REPLY
 * - ignore LOOKUP, LOOKUP_REPLY
 * - ignore LOCK, LOCK_REPLY
 *
 * recover_status |= DIR
 *
 * dlm_recover_directory_wait()
 *
 * dlm_recover_masters()
 * - send LOOKUP
 * - recv LOOKUP_REPLY
 *
 * dlm_recover_locks()
 * - send LOCKS
 * - recv LOCKS_REPLY
 *
 * recover_status |= LOCKS
 *
 * dlm_recover_locks_wait()
 *
 * recover_status |= DONE
 */

/* 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, names = 0, lookup = 0, lock = 0;
	uint32_t status;
	uint64_t seq;

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

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

	if (stop && (rc->rc_type != DLM_RCOM_STATUS))
		goto ignore;

	if (reply && (rc->rc_seq_reply != seq))
		goto ignore;

	if (!(status & DLM_RS_NODES) && (names || lookup || lock))
		goto ignore;

	if (!(status & DLM_RS_DIR) && (lookup || lock))
		goto ignore;

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

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