rcom.c 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525
  1. /******************************************************************************
  2. *******************************************************************************
  3. **
  4. ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
  5. ** Copyright (C) 2005 Red Hat, Inc. All rights reserved.
  6. **
  7. ** This copyrighted material is made available to anyone wishing to use,
  8. ** modify, copy, or redistribute it subject to the terms and conditions
  9. ** of the GNU General Public License v.2.
  10. **
  11. *******************************************************************************
  12. ******************************************************************************/
  13. #include "dlm_internal.h"
  14. #include "lockspace.h"
  15. #include "member.h"
  16. #include "lowcomms.h"
  17. #include "midcomms.h"
  18. #include "rcom.h"
  19. #include "recover.h"
  20. #include "dir.h"
  21. #include "config.h"
  22. #include "memory.h"
  23. #include "lock.h"
  24. #include "util.h"
  25. static int rcom_response(struct dlm_ls *ls)
  26. {
  27. return test_bit(LSFL_RCOM_READY, &ls->ls_flags);
  28. }
  29. static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len,
  30. struct dlm_rcom **rc_ret, struct dlm_mhandle **mh_ret)
  31. {
  32. struct dlm_rcom *rc;
  33. struct dlm_mhandle *mh;
  34. char *mb;
  35. int mb_len = sizeof(struct dlm_rcom) + len;
  36. mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_KERNEL, &mb);
  37. if (!mh) {
  38. log_print("create_rcom to %d type %d len %d ENOBUFS",
  39. to_nodeid, type, len);
  40. return -ENOBUFS;
  41. }
  42. memset(mb, 0, mb_len);
  43. rc = (struct dlm_rcom *) mb;
  44. rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
  45. rc->rc_header.h_lockspace = ls->ls_global_id;
  46. rc->rc_header.h_nodeid = dlm_our_nodeid();
  47. rc->rc_header.h_length = mb_len;
  48. rc->rc_header.h_cmd = DLM_RCOM;
  49. rc->rc_type = type;
  50. spin_lock(&ls->ls_recover_lock);
  51. rc->rc_seq = ls->ls_recover_seq;
  52. spin_unlock(&ls->ls_recover_lock);
  53. *mh_ret = mh;
  54. *rc_ret = rc;
  55. return 0;
  56. }
  57. static void send_rcom(struct dlm_ls *ls, struct dlm_mhandle *mh,
  58. struct dlm_rcom *rc)
  59. {
  60. dlm_rcom_out(rc);
  61. dlm_lowcomms_commit_buffer(mh);
  62. }
  63. /* When replying to a status request, a node also sends back its
  64. configuration values. The requesting node then checks that the remote
  65. node is configured the same way as itself. */
  66. static void make_config(struct dlm_ls *ls, struct rcom_config *rf)
  67. {
  68. rf->rf_lvblen = ls->ls_lvblen;
  69. rf->rf_lsflags = ls->ls_exflags;
  70. }
  71. static int check_config(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid)
  72. {
  73. struct rcom_config *rf = (struct rcom_config *) rc->rc_buf;
  74. if ((rc->rc_header.h_version & 0xFFFF0000) != DLM_HEADER_MAJOR) {
  75. log_error(ls, "version mismatch: %x nodeid %d: %x",
  76. DLM_HEADER_MAJOR | DLM_HEADER_MINOR, nodeid,
  77. rc->rc_header.h_version);
  78. return -EINVAL;
  79. }
  80. if (rf->rf_lvblen != ls->ls_lvblen ||
  81. rf->rf_lsflags != ls->ls_exflags) {
  82. log_error(ls, "config mismatch: %d,%x nodeid %d: %d,%x",
  83. ls->ls_lvblen, ls->ls_exflags,
  84. nodeid, rf->rf_lvblen, rf->rf_lsflags);
  85. return -EINVAL;
  86. }
  87. return 0;
  88. }
  89. static void allow_sync_reply(struct dlm_ls *ls, uint64_t *new_seq)
  90. {
  91. spin_lock(&ls->ls_rcom_spin);
  92. *new_seq = ++ls->ls_rcom_seq;
  93. set_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
  94. spin_unlock(&ls->ls_rcom_spin);
  95. }
  96. static void disallow_sync_reply(struct dlm_ls *ls)
  97. {
  98. spin_lock(&ls->ls_rcom_spin);
  99. clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
  100. clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
  101. spin_unlock(&ls->ls_rcom_spin);
  102. }
  103. int dlm_rcom_status(struct dlm_ls *ls, int nodeid)
  104. {
  105. struct dlm_rcom *rc;
  106. struct dlm_mhandle *mh;
  107. int error = 0;
  108. ls->ls_recover_nodeid = nodeid;
  109. if (nodeid == dlm_our_nodeid()) {
  110. rc = (struct dlm_rcom *) ls->ls_recover_buf;
  111. rc->rc_result = dlm_recover_status(ls);
  112. goto out;
  113. }
  114. error = create_rcom(ls, nodeid, DLM_RCOM_STATUS, 0, &rc, &mh);
  115. if (error)
  116. goto out;
  117. allow_sync_reply(ls, &rc->rc_id);
  118. memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
  119. send_rcom(ls, mh, rc);
  120. error = dlm_wait_function(ls, &rcom_response);
  121. disallow_sync_reply(ls);
  122. if (error)
  123. goto out;
  124. rc = (struct dlm_rcom *) ls->ls_recover_buf;
  125. if (rc->rc_result == -ESRCH) {
  126. /* we pretend the remote lockspace exists with 0 status */
  127. log_debug(ls, "remote node %d not ready", nodeid);
  128. rc->rc_result = 0;
  129. } else
  130. error = check_config(ls, rc, nodeid);
  131. /* the caller looks at rc_result for the remote recovery status */
  132. out:
  133. return error;
  134. }
  135. static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  136. {
  137. struct dlm_rcom *rc;
  138. struct dlm_mhandle *mh;
  139. int error, nodeid = rc_in->rc_header.h_nodeid;
  140. error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY,
  141. sizeof(struct rcom_config), &rc, &mh);
  142. if (error)
  143. return;
  144. rc->rc_id = rc_in->rc_id;
  145. rc->rc_seq_reply = rc_in->rc_seq;
  146. rc->rc_result = dlm_recover_status(ls);
  147. make_config(ls, (struct rcom_config *) rc->rc_buf);
  148. send_rcom(ls, mh, rc);
  149. }
  150. static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  151. {
  152. spin_lock(&ls->ls_rcom_spin);
  153. if (!test_bit(LSFL_RCOM_WAIT, &ls->ls_flags) ||
  154. rc_in->rc_id != ls->ls_rcom_seq) {
  155. log_debug(ls, "reject reply %d from %d seq %llx expect %llx",
  156. rc_in->rc_type, rc_in->rc_header.h_nodeid,
  157. (unsigned long long)rc_in->rc_id,
  158. (unsigned long long)ls->ls_rcom_seq);
  159. goto out;
  160. }
  161. memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
  162. set_bit(LSFL_RCOM_READY, &ls->ls_flags);
  163. clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
  164. wake_up(&ls->ls_wait_general);
  165. out:
  166. spin_unlock(&ls->ls_rcom_spin);
  167. }
  168. static void receive_rcom_status_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  169. {
  170. receive_sync_reply(ls, rc_in);
  171. }
  172. int dlm_rcom_names(struct dlm_ls *ls, int nodeid, char *last_name, int last_len)
  173. {
  174. struct dlm_rcom *rc;
  175. struct dlm_mhandle *mh;
  176. int error = 0, len = sizeof(struct dlm_rcom);
  177. ls->ls_recover_nodeid = nodeid;
  178. if (nodeid == dlm_our_nodeid()) {
  179. dlm_copy_master_names(ls, last_name, last_len,
  180. ls->ls_recover_buf + len,
  181. dlm_config.ci_buffer_size - len, nodeid);
  182. goto out;
  183. }
  184. error = create_rcom(ls, nodeid, DLM_RCOM_NAMES, last_len, &rc, &mh);
  185. if (error)
  186. goto out;
  187. memcpy(rc->rc_buf, last_name, last_len);
  188. allow_sync_reply(ls, &rc->rc_id);
  189. memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
  190. send_rcom(ls, mh, rc);
  191. error = dlm_wait_function(ls, &rcom_response);
  192. disallow_sync_reply(ls);
  193. out:
  194. return error;
  195. }
  196. static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  197. {
  198. struct dlm_rcom *rc;
  199. struct dlm_mhandle *mh;
  200. int error, inlen, outlen, nodeid;
  201. nodeid = rc_in->rc_header.h_nodeid;
  202. inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
  203. outlen = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);
  204. error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh);
  205. if (error)
  206. return;
  207. rc->rc_id = rc_in->rc_id;
  208. rc->rc_seq_reply = rc_in->rc_seq;
  209. dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen,
  210. nodeid);
  211. send_rcom(ls, mh, rc);
  212. }
  213. static void receive_rcom_names_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  214. {
  215. receive_sync_reply(ls, rc_in);
  216. }
  217. int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid)
  218. {
  219. struct dlm_rcom *rc;
  220. struct dlm_mhandle *mh;
  221. struct dlm_ls *ls = r->res_ls;
  222. int error;
  223. error = create_rcom(ls, dir_nodeid, DLM_RCOM_LOOKUP, r->res_length,
  224. &rc, &mh);
  225. if (error)
  226. goto out;
  227. memcpy(rc->rc_buf, r->res_name, r->res_length);
  228. rc->rc_id = (unsigned long) r;
  229. send_rcom(ls, mh, rc);
  230. out:
  231. return error;
  232. }
  233. static void receive_rcom_lookup(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  234. {
  235. struct dlm_rcom *rc;
  236. struct dlm_mhandle *mh;
  237. int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid;
  238. int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
  239. error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh);
  240. if (error)
  241. return;
  242. error = dlm_dir_lookup(ls, nodeid, rc_in->rc_buf, len, &ret_nodeid);
  243. if (error)
  244. ret_nodeid = error;
  245. rc->rc_result = ret_nodeid;
  246. rc->rc_id = rc_in->rc_id;
  247. rc->rc_seq_reply = rc_in->rc_seq;
  248. send_rcom(ls, mh, rc);
  249. }
  250. static void receive_rcom_lookup_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  251. {
  252. dlm_recover_master_reply(ls, rc_in);
  253. }
  254. static void pack_rcom_lock(struct dlm_rsb *r, struct dlm_lkb *lkb,
  255. struct rcom_lock *rl)
  256. {
  257. memset(rl, 0, sizeof(*rl));
  258. rl->rl_ownpid = lkb->lkb_ownpid;
  259. rl->rl_lkid = lkb->lkb_id;
  260. rl->rl_exflags = lkb->lkb_exflags;
  261. rl->rl_flags = lkb->lkb_flags;
  262. rl->rl_lvbseq = lkb->lkb_lvbseq;
  263. rl->rl_rqmode = lkb->lkb_rqmode;
  264. rl->rl_grmode = lkb->lkb_grmode;
  265. rl->rl_status = lkb->lkb_status;
  266. rl->rl_wait_type = lkb->lkb_wait_type;
  267. if (lkb->lkb_bastaddr)
  268. rl->rl_asts |= AST_BAST;
  269. if (lkb->lkb_astaddr)
  270. rl->rl_asts |= AST_COMP;
  271. rl->rl_namelen = r->res_length;
  272. memcpy(rl->rl_name, r->res_name, r->res_length);
  273. /* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
  274. If so, receive_rcom_lock_args() won't take this copy. */
  275. if (lkb->lkb_lvbptr)
  276. memcpy(rl->rl_lvb, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
  277. }
  278. int dlm_send_rcom_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
  279. {
  280. struct dlm_ls *ls = r->res_ls;
  281. struct dlm_rcom *rc;
  282. struct dlm_mhandle *mh;
  283. struct rcom_lock *rl;
  284. int error, len = sizeof(struct rcom_lock);
  285. if (lkb->lkb_lvbptr)
  286. len += ls->ls_lvblen;
  287. error = create_rcom(ls, r->res_nodeid, DLM_RCOM_LOCK, len, &rc, &mh);
  288. if (error)
  289. goto out;
  290. rl = (struct rcom_lock *) rc->rc_buf;
  291. pack_rcom_lock(r, lkb, rl);
  292. rc->rc_id = (unsigned long) r;
  293. send_rcom(ls, mh, rc);
  294. out:
  295. return error;
  296. }
  297. static void receive_rcom_lock(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  298. {
  299. struct dlm_rcom *rc;
  300. struct dlm_mhandle *mh;
  301. int error, nodeid = rc_in->rc_header.h_nodeid;
  302. dlm_recover_master_copy(ls, rc_in);
  303. error = create_rcom(ls, nodeid, DLM_RCOM_LOCK_REPLY,
  304. sizeof(struct rcom_lock), &rc, &mh);
  305. if (error)
  306. return;
  307. /* We send back the same rcom_lock struct we received, but
  308. dlm_recover_master_copy() has filled in rl_remid and rl_result */
  309. memcpy(rc->rc_buf, rc_in->rc_buf, sizeof(struct rcom_lock));
  310. rc->rc_id = rc_in->rc_id;
  311. rc->rc_seq_reply = rc_in->rc_seq;
  312. send_rcom(ls, mh, rc);
  313. }
  314. static void receive_rcom_lock_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
  315. {
  316. dlm_recover_process_copy(ls, rc_in);
  317. }
  318. static int send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
  319. {
  320. struct dlm_rcom *rc;
  321. struct rcom_config *rf;
  322. struct dlm_mhandle *mh;
  323. char *mb;
  324. int mb_len = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);
  325. mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_KERNEL, &mb);
  326. if (!mh)
  327. return -ENOBUFS;
  328. memset(mb, 0, mb_len);
  329. rc = (struct dlm_rcom *) mb;
  330. rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
  331. rc->rc_header.h_lockspace = rc_in->rc_header.h_lockspace;
  332. rc->rc_header.h_nodeid = dlm_our_nodeid();
  333. rc->rc_header.h_length = mb_len;
  334. rc->rc_header.h_cmd = DLM_RCOM;
  335. rc->rc_type = DLM_RCOM_STATUS_REPLY;
  336. rc->rc_id = rc_in->rc_id;
  337. rc->rc_seq_reply = rc_in->rc_seq;
  338. rc->rc_result = -ESRCH;
  339. rf = (struct rcom_config *) rc->rc_buf;
  340. rf->rf_lvblen = -1;
  341. dlm_rcom_out(rc);
  342. dlm_lowcomms_commit_buffer(mh);
  343. return 0;
  344. }
  345. static int is_old_reply(struct dlm_ls *ls, struct dlm_rcom *rc)
  346. {
  347. uint64_t seq;
  348. int rv = 0;
  349. switch (rc->rc_type) {
  350. case DLM_RCOM_STATUS_REPLY:
  351. case DLM_RCOM_NAMES_REPLY:
  352. case DLM_RCOM_LOOKUP_REPLY:
  353. case DLM_RCOM_LOCK_REPLY:
  354. spin_lock(&ls->ls_recover_lock);
  355. seq = ls->ls_recover_seq;
  356. spin_unlock(&ls->ls_recover_lock);
  357. if (rc->rc_seq_reply != seq) {
  358. log_debug(ls, "ignoring old reply %x from %d "
  359. "seq_reply %llx expect %llx",
  360. rc->rc_type, rc->rc_header.h_nodeid,
  361. (unsigned long long)rc->rc_seq_reply,
  362. (unsigned long long)seq);
  363. rv = 1;
  364. }
  365. }
  366. return rv;
  367. }
  368. /* Called by dlm_recvd; corresponds to dlm_receive_message() but special
  369. recovery-only comms are sent through here. */
  370. void dlm_receive_rcom(struct dlm_header *hd, int nodeid)
  371. {
  372. struct dlm_rcom *rc = (struct dlm_rcom *) hd;
  373. struct dlm_ls *ls;
  374. dlm_rcom_in(rc);
  375. /* If the lockspace doesn't exist then still send a status message
  376. back; it's possible that it just doesn't have its global_id yet. */
  377. ls = dlm_find_lockspace_global(hd->h_lockspace);
  378. if (!ls) {
  379. log_print("lockspace %x from %d type %x not found",
  380. hd->h_lockspace, nodeid, rc->rc_type);
  381. if (rc->rc_type == DLM_RCOM_STATUS)
  382. send_ls_not_ready(nodeid, rc);
  383. return;
  384. }
  385. if (dlm_recovery_stopped(ls) && (rc->rc_type != DLM_RCOM_STATUS)) {
  386. log_debug(ls, "ignoring recovery message %x from %d",
  387. rc->rc_type, nodeid);
  388. goto out;
  389. }
  390. if (is_old_reply(ls, rc))
  391. goto out;
  392. if (nodeid != rc->rc_header.h_nodeid) {
  393. log_error(ls, "bad rcom nodeid %d from %d",
  394. rc->rc_header.h_nodeid, nodeid);
  395. goto out;
  396. }
  397. switch (rc->rc_type) {
  398. case DLM_RCOM_STATUS:
  399. receive_rcom_status(ls, rc);
  400. break;
  401. case DLM_RCOM_NAMES:
  402. receive_rcom_names(ls, rc);
  403. break;
  404. case DLM_RCOM_LOOKUP:
  405. receive_rcom_lookup(ls, rc);
  406. break;
  407. case DLM_RCOM_LOCK:
  408. receive_rcom_lock(ls, rc);
  409. break;
  410. case DLM_RCOM_STATUS_REPLY:
  411. receive_rcom_status_reply(ls, rc);
  412. break;
  413. case DLM_RCOM_NAMES_REPLY:
  414. receive_rcom_names_reply(ls, rc);
  415. break;
  416. case DLM_RCOM_LOOKUP_REPLY:
  417. receive_rcom_lookup_reply(ls, rc);
  418. break;
  419. case DLM_RCOM_LOCK_REPLY:
  420. receive_rcom_lock_reply(ls, rc);
  421. break;
  422. default:
  423. DLM_ASSERT(0, printk("rc_type=%x\n", rc->rc_type););
  424. }
  425. out:
  426. dlm_put_lockspace(ls);
  427. }