fs/dlm/rcom.c

   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