rcom.c 13.1 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40
/******************************************************************************
*******************************************************************************
**
**  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;

P
Patrick Caulfield 已提交
41
	mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, ls->ls_allocation, &mb);
42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58
	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;

D
David Teigland 已提交
59 60 61 62
	spin_lock(&ls->ls_recover_lock);
	rc->rc_seq = ls->ls_recover_seq;
	spin_unlock(&ls->ls_recover_lock);

63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84
	*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;
}

D
David Teigland 已提交
85
static int check_config(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid)
86
{
D
David Teigland 已提交
87 88 89 90 91 92
	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);
93
		return -EPROTO;
D
David Teigland 已提交
94 95
	}

96 97 98 99 100
	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);
101
		return -EPROTO;
102 103 104 105
	}
	return 0;
}

106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121
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);
}

122 123 124 125 126 127
int dlm_rcom_status(struct dlm_ls *ls, int nodeid)
{
	struct dlm_rcom *rc;
	struct dlm_mhandle *mh;
	int error = 0;

128
	ls->ls_recover_nodeid = nodeid;
129 130 131 132 133 134 135 136 137 138

	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;
139 140

	allow_sync_reply(ls, &rc->rc_id);
141
	memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
142 143 144 145

	send_rcom(ls, mh, rc);

	error = dlm_wait_function(ls, &rcom_response);
146
	disallow_sync_reply(ls);
147 148 149 150 151 152 153 154 155 156
	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
D
David Teigland 已提交
157
		error = check_config(ls, rc, nodeid);
158 159 160 161 162 163 164 165 166 167 168 169 170 171 172
	/* 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;
173
	rc->rc_id = rc_in->rc_id;
D
David Teigland 已提交
174
	rc->rc_seq_reply = rc_in->rc_seq;
175 176 177 178 179 180
	rc->rc_result = dlm_recover_status(ls);
	make_config(ls, (struct rcom_config *) rc->rc_buf);

	send_rcom(ls, mh, rc);
}

181
static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
182
{
183 184 185 186 187
	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,
188 189
			  (unsigned long long)rc_in->rc_id,
			  (unsigned long long)ls->ls_rcom_seq);
190
		goto out;
191
	}
192 193
	memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
	set_bit(LSFL_RCOM_READY, &ls->ls_flags);
194
	clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
195
	wake_up(&ls->ls_wait_general);
196 197
 out:
	spin_unlock(&ls->ls_rcom_spin);
198 199
}

200 201 202 203 204
static void receive_rcom_status_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
{
	receive_sync_reply(ls, rc_in);
}

205 206 207 208 209 210
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);

211
	ls->ls_recover_nodeid = nodeid;
212 213 214 215

	if (nodeid == dlm_our_nodeid()) {
		dlm_copy_master_names(ls, last_name, last_len,
		                      ls->ls_recover_buf + len,
216
		                      dlm_config.ci_buffer_size - len, nodeid);
217 218 219 220 221 222 223
		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);
224 225

	allow_sync_reply(ls, &rc->rc_id);
226
	memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
227 228 229 230

	send_rcom(ls, mh, rc);

	error = dlm_wait_function(ls, &rcom_response);
231
	disallow_sync_reply(ls);
232 233 234 235 236 237 238 239
 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;
D
David Teigland 已提交
240
	int error, inlen, outlen, nodeid;
241 242 243

	nodeid = rc_in->rc_header.h_nodeid;
	inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
244
	outlen = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);
245 246 247 248

	error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh);
	if (error)
		return;
249
	rc->rc_id = rc_in->rc_id;
D
David Teigland 已提交
250
	rc->rc_seq_reply = rc_in->rc_seq;
251 252 253 254 255 256 257 258

	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)
{
259
	receive_sync_reply(ls, rc_in);
260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296
}

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;
D
David Teigland 已提交
297
	rc->rc_seq_reply = rc_in->rc_seq;
298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378

	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;
D
David Teigland 已提交
379
	rc->rc_seq_reply = rc_in->rc_seq;
380 381 382 383 384 385 386 387 388

	send_rcom(ls, mh, rc);
}

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

D
David Teigland 已提交
389
static int send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
390 391
{
	struct dlm_rcom *rc;
392
	struct rcom_config *rf;
393 394
	struct dlm_mhandle *mh;
	char *mb;
395
	int mb_len = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);
396

D
David Teigland 已提交
397
	mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_NOFS, &mb);
398 399 400 401 402 403 404 405 406 407 408 409 410
	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;
411
	rc->rc_id = rc_in->rc_id;
D
David Teigland 已提交
412
	rc->rc_seq_reply = rc_in->rc_seq;
413 414
	rc->rc_result = -ESRCH;

415 416 417
	rf = (struct rcom_config *) rc->rc_buf;
	rf->rf_lvblen = -1;

418 419 420 421 422 423
	dlm_rcom_out(rc);
	dlm_lowcomms_commit_buffer(mh);

	return 0;
}

D
David Teigland 已提交
424 425 426 427 428 429 430 431 432 433 434 435 436 437
static int is_old_reply(struct dlm_ls *ls, struct dlm_rcom *rc)
{
	uint64_t seq;
	int rv = 0;

	switch (rc->rc_type) {
	case DLM_RCOM_STATUS_REPLY:
	case DLM_RCOM_NAMES_REPLY:
	case DLM_RCOM_LOOKUP_REPLY:
	case DLM_RCOM_LOCK_REPLY:
		spin_lock(&ls->ls_recover_lock);
		seq = ls->ls_recover_seq;
		spin_unlock(&ls->ls_recover_lock);
		if (rc->rc_seq_reply != seq) {
438
			log_debug(ls, "ignoring old reply %x from %d "
D
David Teigland 已提交
439 440 441 442 443 444 445 446 447 448
				      "seq_reply %llx expect %llx",
				      rc->rc_type, rc->rc_header.h_nodeid,
				      (unsigned long long)rc->rc_seq_reply,
				      (unsigned long long)seq);
			rv = 1;
		}
	}
	return rv;
}

449 450 451 452 453 454 455 456 457 458 459 460 461 462 463
/* 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) {
464 465 466
		log_print("lockspace %x from %d type %x not found",
			  hd->h_lockspace, nodeid, rc->rc_type);
		if (rc->rc_type == DLM_RCOM_STATUS)
D
David Teigland 已提交
467
			send_ls_not_ready(nodeid, rc);
468 469 470 471
		return;
	}

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

D
David Teigland 已提交
477 478 479
	if (is_old_reply(ls, rc))
		goto out;

480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525
	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);
}