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