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

// SPDX-License-Identifier: GPL-2.0-only
/******************************************************************************
*******************************************************************************
**
**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
**  Copyright (C) 2005-2008 Red Hat, Inc.  All rights reserved.
**
**
*******************************************************************************
******************************************************************************/

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

retry:
	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 == -ETIMEDOUT)
		goto retry;
	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;

retry:
	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);
	if (error == -ETIMEDOUT)
		goto retry;
 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;
}

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;

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