rcom.c 16 KB

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