4
0

cpg.c 31 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155
  1. /*
  2. * vi: set autoindent tabstop=4 shiftwidth=4 :
  3. *
  4. * Copyright (c) 2006-2012 Red Hat, Inc.
  5. *
  6. * All rights reserved.
  7. *
  8. * Author: Christine Caulfield (ccaulfi@redhat.com)
  9. * Author: Jan Friesse (jfriesse@redhat.com)
  10. *
  11. * This software licensed under BSD license, the text of which follows:
  12. *
  13. * Redistribution and use in source and binary forms, with or without
  14. * modification, are permitted provided that the following conditions are met:
  15. *
  16. * - Redistributions of source code must retain the above copyright notice,
  17. * this list of conditions and the following disclaimer.
  18. * - Redistributions in binary form must reproduce the above copyright notice,
  19. * this list of conditions and the following disclaimer in the documentation
  20. * and/or other materials provided with the distribution.
  21. * - Neither the name of the MontaVista Software, Inc. nor the names of its
  22. * contributors may be used to endorse or promote products derived from this
  23. * software without specific prior written permission.
  24. *
  25. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  26. * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  27. * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  28. * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  29. * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  30. * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  31. * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  32. * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  33. * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  34. * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  35. * THE POSSIBILITY OF SUCH DAMAGE.
  36. */
  37. /*
  38. * Provides a closed process group API using the coroipcc executive
  39. */
  40. #include <config.h>
  41. #include <stdlib.h>
  42. #include <stdio.h>
  43. #include <string.h>
  44. #include <unistd.h>
  45. #include <sys/types.h>
  46. #include <sys/socket.h>
  47. #include <sys/mman.h>
  48. #include <errno.h>
  49. #include <limits.h>
  50. #include <qb/qbdefs.h>
  51. #include <qb/qbipcc.h>
  52. #include <qb/qblog.h>
  53. #include <corosync/hdb.h>
  54. #include <corosync/list.h>
  55. #include <corosync/corotypes.h>
  56. #include <corosync/corodefs.h>
  57. #include <corosync/cpg.h>
  58. #include <corosync/ipc_cpg.h>
  59. #include "util.h"
  60. struct cpg_inst {
  61. qb_ipcc_connection_t *c;
  62. int finalize;
  63. void *context;
  64. union {
  65. cpg_model_data_t model_data;
  66. cpg_model_v1_data_t model_v1_data;
  67. };
  68. struct list_head iteration_list_head;
  69. };
  70. static void cpg_inst_free (void *inst);
  71. DECLARE_HDB_DATABASE(cpg_handle_t_db, cpg_inst_free);
  72. struct cpg_iteration_instance_t {
  73. cpg_iteration_handle_t cpg_iteration_handle;
  74. qb_ipcc_connection_t *conn;
  75. hdb_handle_t executive_iteration_handle;
  76. struct list_head list;
  77. };
  78. DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db,NULL);
  79. /*
  80. * Internal (not visible by API) functions
  81. */
  82. static cs_error_t
  83. coroipcc_msg_send_reply_receive (
  84. qb_ipcc_connection_t *c,
  85. const struct iovec *iov,
  86. unsigned int iov_len,
  87. void *res_msg,
  88. size_t res_len)
  89. {
  90. return qb_to_cs_error(qb_ipcc_sendv_recv(c, iov, iov_len, res_msg, res_len,
  91. CS_IPC_TIMEOUT_MS));
  92. }
  93. static void cpg_iteration_instance_finalize (struct cpg_iteration_instance_t *cpg_iteration_instance)
  94. {
  95. list_del (&cpg_iteration_instance->list);
  96. hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_instance->cpg_iteration_handle);
  97. }
  98. static void cpg_inst_free (void *inst)
  99. {
  100. struct cpg_inst *cpg_inst = (struct cpg_inst *)inst;
  101. qb_ipcc_disconnect(cpg_inst->c);
  102. }
  103. static void cpg_inst_finalize (struct cpg_inst *cpg_inst, hdb_handle_t handle)
  104. {
  105. struct list_head *iter, *iter_next;
  106. struct cpg_iteration_instance_t *cpg_iteration_instance;
  107. /*
  108. * Traverse thru iteration instances and delete them
  109. */
  110. for (iter = cpg_inst->iteration_list_head.next; iter != &cpg_inst->iteration_list_head;iter = iter_next) {
  111. iter_next = iter->next;
  112. cpg_iteration_instance = list_entry (iter, struct cpg_iteration_instance_t, list);
  113. cpg_iteration_instance_finalize (cpg_iteration_instance);
  114. }
  115. hdb_handle_destroy (&cpg_handle_t_db, handle);
  116. }
  117. /**
  118. * @defgroup cpg_coroipcc The closed process group API
  119. * @ingroup coroipcc
  120. *
  121. * @{
  122. */
  123. cs_error_t cpg_initialize (
  124. cpg_handle_t *handle,
  125. cpg_callbacks_t *callbacks)
  126. {
  127. cpg_model_v1_data_t model_v1_data;
  128. memset (&model_v1_data, 0, sizeof (cpg_model_v1_data_t));
  129. if (callbacks) {
  130. model_v1_data.cpg_deliver_fn = callbacks->cpg_deliver_fn;
  131. model_v1_data.cpg_confchg_fn = callbacks->cpg_confchg_fn;
  132. }
  133. return (cpg_model_initialize (handle, CPG_MODEL_V1, (cpg_model_data_t *)&model_v1_data, NULL));
  134. }
  135. cs_error_t cpg_model_initialize (
  136. cpg_handle_t *handle,
  137. cpg_model_t model,
  138. cpg_model_data_t *model_data,
  139. void *context)
  140. {
  141. cs_error_t error;
  142. struct cpg_inst *cpg_inst;
  143. if (model != CPG_MODEL_V1) {
  144. error = CS_ERR_INVALID_PARAM;
  145. goto error_no_destroy;
  146. }
  147. error = hdb_error_to_cs (hdb_handle_create (&cpg_handle_t_db, sizeof (struct cpg_inst), handle));
  148. if (error != CS_OK) {
  149. goto error_no_destroy;
  150. }
  151. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, *handle, (void *)&cpg_inst));
  152. if (error != CS_OK) {
  153. goto error_destroy;
  154. }
  155. cpg_inst->c = qb_ipcc_connect ("cpg", IPC_REQUEST_SIZE);
  156. if (cpg_inst->c == NULL) {
  157. error = qb_to_cs_error(-errno);
  158. goto error_put_destroy;
  159. }
  160. if (model_data != NULL) {
  161. switch (model) {
  162. case CPG_MODEL_V1:
  163. memcpy (&cpg_inst->model_v1_data, model_data, sizeof (cpg_model_v1_data_t));
  164. if ((cpg_inst->model_v1_data.flags & ~(CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF)) != 0) {
  165. error = CS_ERR_INVALID_PARAM;
  166. goto error_destroy;
  167. }
  168. break;
  169. }
  170. }
  171. cpg_inst->model_data.model = model;
  172. cpg_inst->context = context;
  173. list_init(&cpg_inst->iteration_list_head);
  174. hdb_handle_put (&cpg_handle_t_db, *handle);
  175. return (CS_OK);
  176. error_put_destroy:
  177. hdb_handle_put (&cpg_handle_t_db, *handle);
  178. error_destroy:
  179. hdb_handle_destroy (&cpg_handle_t_db, *handle);
  180. error_no_destroy:
  181. return (error);
  182. }
  183. cs_error_t cpg_finalize (
  184. cpg_handle_t handle)
  185. {
  186. struct cpg_inst *cpg_inst;
  187. struct iovec iov;
  188. struct req_lib_cpg_finalize req_lib_cpg_finalize;
  189. struct res_lib_cpg_finalize res_lib_cpg_finalize;
  190. cs_error_t error;
  191. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  192. if (error != CS_OK) {
  193. return (error);
  194. }
  195. /*
  196. * Another thread has already started finalizing
  197. */
  198. if (cpg_inst->finalize) {
  199. hdb_handle_put (&cpg_handle_t_db, handle);
  200. return (CS_ERR_BAD_HANDLE);
  201. }
  202. cpg_inst->finalize = 1;
  203. /*
  204. * Send service request
  205. */
  206. req_lib_cpg_finalize.header.size = sizeof (struct req_lib_cpg_finalize);
  207. req_lib_cpg_finalize.header.id = MESSAGE_REQ_CPG_FINALIZE;
  208. iov.iov_base = (void *)&req_lib_cpg_finalize;
  209. iov.iov_len = sizeof (struct req_lib_cpg_finalize);
  210. error = coroipcc_msg_send_reply_receive (cpg_inst->c,
  211. &iov,
  212. 1,
  213. &res_lib_cpg_finalize,
  214. sizeof (struct res_lib_cpg_finalize));
  215. cpg_inst_finalize (cpg_inst, handle);
  216. hdb_handle_put (&cpg_handle_t_db, handle);
  217. return (error);
  218. }
  219. cs_error_t cpg_fd_get (
  220. cpg_handle_t handle,
  221. int *fd)
  222. {
  223. cs_error_t error;
  224. struct cpg_inst *cpg_inst;
  225. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  226. if (error != CS_OK) {
  227. return (error);
  228. }
  229. error = qb_to_cs_error (qb_ipcc_fd_get (cpg_inst->c, fd));
  230. hdb_handle_put (&cpg_handle_t_db, handle);
  231. return (error);
  232. }
  233. cs_error_t cpg_context_get (
  234. cpg_handle_t handle,
  235. void **context)
  236. {
  237. cs_error_t error;
  238. struct cpg_inst *cpg_inst;
  239. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  240. if (error != CS_OK) {
  241. return (error);
  242. }
  243. *context = cpg_inst->context;
  244. hdb_handle_put (&cpg_handle_t_db, handle);
  245. return (CS_OK);
  246. }
  247. cs_error_t cpg_context_set (
  248. cpg_handle_t handle,
  249. void *context)
  250. {
  251. cs_error_t error;
  252. struct cpg_inst *cpg_inst;
  253. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  254. if (error != CS_OK) {
  255. return (error);
  256. }
  257. cpg_inst->context = context;
  258. hdb_handle_put (&cpg_handle_t_db, handle);
  259. return (CS_OK);
  260. }
  261. cs_error_t cpg_dispatch (
  262. cpg_handle_t handle,
  263. cs_dispatch_flags_t dispatch_types)
  264. {
  265. int timeout = -1;
  266. cs_error_t error;
  267. int cont = 1; /* always continue do loop except when set to 0 */
  268. struct cpg_inst *cpg_inst;
  269. struct res_lib_cpg_confchg_callback *res_cpg_confchg_callback;
  270. struct res_lib_cpg_deliver_callback *res_cpg_deliver_callback;
  271. struct res_lib_cpg_totem_confchg_callback *res_cpg_totem_confchg_callback;
  272. struct cpg_inst cpg_inst_copy;
  273. struct qb_ipc_response_header *dispatch_data;
  274. struct cpg_address member_list[CPG_MEMBERS_MAX];
  275. struct cpg_address left_list[CPG_MEMBERS_MAX];
  276. struct cpg_address joined_list[CPG_MEMBERS_MAX];
  277. struct cpg_name group_name;
  278. mar_cpg_address_t *left_list_start;
  279. mar_cpg_address_t *joined_list_start;
  280. unsigned int i;
  281. struct cpg_ring_id ring_id;
  282. uint32_t totem_member_list[CPG_MEMBERS_MAX];
  283. int32_t errno_res;
  284. char dispatch_buf[IPC_DISPATCH_SIZE];
  285. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  286. if (error != CS_OK) {
  287. return (error);
  288. }
  289. /*
  290. * Timeout instantly for CS_DISPATCH_ONE_NONBLOCKING or CS_DISPATCH_ALL and
  291. * wait indefinately for CS_DISPATCH_ONE or CS_DISPATCH_BLOCKING
  292. */
  293. if (dispatch_types == CS_DISPATCH_ALL || dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) {
  294. timeout = 0;
  295. }
  296. dispatch_data = (struct qb_ipc_response_header *)dispatch_buf;
  297. do {
  298. errno_res = qb_ipcc_event_recv (
  299. cpg_inst->c,
  300. dispatch_buf,
  301. IPC_DISPATCH_SIZE,
  302. timeout);
  303. error = qb_to_cs_error (errno_res);
  304. if (error == CS_ERR_BAD_HANDLE) {
  305. error = CS_OK;
  306. goto error_put;
  307. }
  308. if (error == CS_ERR_TRY_AGAIN) {
  309. if (dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) {
  310. /*
  311. * Don't mask error
  312. */
  313. goto error_put;
  314. }
  315. error = CS_OK;
  316. if (dispatch_types == CS_DISPATCH_ALL) {
  317. break; /* exit do while cont is 1 loop */
  318. } else {
  319. continue; /* next poll */
  320. }
  321. }
  322. if (error != CS_OK) {
  323. goto error_put;
  324. }
  325. /*
  326. * Make copy of callbacks, message data, unlock instance, and call callback
  327. * A risk of this dispatch method is that the callback routines may
  328. * operate at the same time that cpgFinalize has been called.
  329. */
  330. memcpy (&cpg_inst_copy, cpg_inst, sizeof (struct cpg_inst));
  331. switch (cpg_inst_copy.model_data.model) {
  332. case CPG_MODEL_V1:
  333. /*
  334. * Dispatch incoming message
  335. */
  336. switch (dispatch_data->id) {
  337. case MESSAGE_RES_CPG_DELIVER_CALLBACK:
  338. if (cpg_inst_copy.model_v1_data.cpg_deliver_fn == NULL) {
  339. break;
  340. }
  341. res_cpg_deliver_callback = (struct res_lib_cpg_deliver_callback *)dispatch_data;
  342. marshall_from_mar_cpg_name_t (
  343. &group_name,
  344. &res_cpg_deliver_callback->group_name);
  345. cpg_inst_copy.model_v1_data.cpg_deliver_fn (handle,
  346. &group_name,
  347. res_cpg_deliver_callback->nodeid,
  348. res_cpg_deliver_callback->pid,
  349. &res_cpg_deliver_callback->message,
  350. res_cpg_deliver_callback->msglen);
  351. break;
  352. case MESSAGE_RES_CPG_CONFCHG_CALLBACK:
  353. if (cpg_inst_copy.model_v1_data.cpg_confchg_fn == NULL) {
  354. break;
  355. }
  356. res_cpg_confchg_callback = (struct res_lib_cpg_confchg_callback *)dispatch_data;
  357. for (i = 0; i < res_cpg_confchg_callback->member_list_entries; i++) {
  358. marshall_from_mar_cpg_address_t (&member_list[i],
  359. &res_cpg_confchg_callback->member_list[i]);
  360. }
  361. left_list_start = res_cpg_confchg_callback->member_list +
  362. res_cpg_confchg_callback->member_list_entries;
  363. for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) {
  364. marshall_from_mar_cpg_address_t (&left_list[i],
  365. &left_list_start[i]);
  366. }
  367. joined_list_start = res_cpg_confchg_callback->member_list +
  368. res_cpg_confchg_callback->member_list_entries +
  369. res_cpg_confchg_callback->left_list_entries;
  370. for (i = 0; i < res_cpg_confchg_callback->joined_list_entries; i++) {
  371. marshall_from_mar_cpg_address_t (&joined_list[i],
  372. &joined_list_start[i]);
  373. }
  374. marshall_from_mar_cpg_name_t (
  375. &group_name,
  376. &res_cpg_confchg_callback->group_name);
  377. cpg_inst_copy.model_v1_data.cpg_confchg_fn (handle,
  378. &group_name,
  379. member_list,
  380. res_cpg_confchg_callback->member_list_entries,
  381. left_list,
  382. res_cpg_confchg_callback->left_list_entries,
  383. joined_list,
  384. res_cpg_confchg_callback->joined_list_entries);
  385. break;
  386. case MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK:
  387. if (cpg_inst_copy.model_v1_data.cpg_totem_confchg_fn == NULL) {
  388. break;
  389. }
  390. res_cpg_totem_confchg_callback = (struct res_lib_cpg_totem_confchg_callback *)dispatch_data;
  391. marshall_from_mar_cpg_ring_id_t (&ring_id, &res_cpg_totem_confchg_callback->ring_id);
  392. for (i = 0; i < res_cpg_totem_confchg_callback->member_list_entries; i++) {
  393. totem_member_list[i] = res_cpg_totem_confchg_callback->member_list[i];
  394. }
  395. cpg_inst_copy.model_v1_data.cpg_totem_confchg_fn (handle,
  396. ring_id,
  397. res_cpg_totem_confchg_callback->member_list_entries,
  398. totem_member_list);
  399. break;
  400. default:
  401. error = CS_ERR_LIBRARY;
  402. goto error_put;
  403. break;
  404. } /* - switch (dispatch_data->id) */
  405. break; /* case CPG_MODEL_V1 */
  406. } /* - switch (cpg_inst_copy.model_data.model) */
  407. if (cpg_inst_copy.finalize || cpg_inst->finalize) {
  408. /*
  409. * If the finalize has been called then get out of the dispatch.
  410. */
  411. cpg_inst->finalize = 1;
  412. error = CS_ERR_BAD_HANDLE;
  413. goto error_put;
  414. }
  415. /*
  416. * Determine if more messages should be processed
  417. */
  418. if (dispatch_types == CS_DISPATCH_ONE || dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) {
  419. cont = 0;
  420. }
  421. } while (cont);
  422. error_put:
  423. hdb_handle_put (&cpg_handle_t_db, handle);
  424. return (error);
  425. }
  426. cs_error_t cpg_join (
  427. cpg_handle_t handle,
  428. const struct cpg_name *group)
  429. {
  430. cs_error_t error;
  431. struct cpg_inst *cpg_inst;
  432. struct iovec iov[2];
  433. struct req_lib_cpg_join req_lib_cpg_join;
  434. struct res_lib_cpg_join response;
  435. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  436. if (error != CS_OK) {
  437. return (error);
  438. }
  439. /* Now join */
  440. req_lib_cpg_join.header.size = sizeof (struct req_lib_cpg_join);
  441. req_lib_cpg_join.header.id = MESSAGE_REQ_CPG_JOIN;
  442. req_lib_cpg_join.pid = getpid();
  443. req_lib_cpg_join.flags = 0;
  444. switch (cpg_inst->model_data.model) {
  445. case CPG_MODEL_V1:
  446. req_lib_cpg_join.flags = cpg_inst->model_v1_data.flags;
  447. break;
  448. }
  449. marshall_to_mar_cpg_name_t (&req_lib_cpg_join.group_name,
  450. group);
  451. iov[0].iov_base = (void *)&req_lib_cpg_join;
  452. iov[0].iov_len = sizeof (struct req_lib_cpg_join);
  453. do {
  454. error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 1,
  455. &response, sizeof (struct res_lib_cpg_join));
  456. if (error != CS_OK) {
  457. goto error_exit;
  458. }
  459. } while (response.header.error == CS_ERR_BUSY);
  460. error = response.header.error;
  461. error_exit:
  462. hdb_handle_put (&cpg_handle_t_db, handle);
  463. return (error);
  464. }
  465. cs_error_t cpg_leave (
  466. cpg_handle_t handle,
  467. const struct cpg_name *group)
  468. {
  469. cs_error_t error;
  470. struct cpg_inst *cpg_inst;
  471. struct iovec iov[2];
  472. struct req_lib_cpg_leave req_lib_cpg_leave;
  473. struct res_lib_cpg_leave res_lib_cpg_leave;
  474. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  475. if (error != CS_OK) {
  476. return (error);
  477. }
  478. req_lib_cpg_leave.header.size = sizeof (struct req_lib_cpg_leave);
  479. req_lib_cpg_leave.header.id = MESSAGE_REQ_CPG_LEAVE;
  480. req_lib_cpg_leave.pid = getpid();
  481. marshall_to_mar_cpg_name_t (&req_lib_cpg_leave.group_name,
  482. group);
  483. iov[0].iov_base = (void *)&req_lib_cpg_leave;
  484. iov[0].iov_len = sizeof (struct req_lib_cpg_leave);
  485. do {
  486. error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 1,
  487. &res_lib_cpg_leave, sizeof (struct res_lib_cpg_leave));
  488. if (error != CS_OK) {
  489. goto error_exit;
  490. }
  491. } while (res_lib_cpg_leave.header.error == CS_ERR_BUSY);
  492. error = res_lib_cpg_leave.header.error;
  493. error_exit:
  494. hdb_handle_put (&cpg_handle_t_db, handle);
  495. return (error);
  496. }
  497. cs_error_t cpg_membership_get (
  498. cpg_handle_t handle,
  499. struct cpg_name *group_name,
  500. struct cpg_address *member_list,
  501. int *member_list_entries)
  502. {
  503. cs_error_t error;
  504. struct cpg_inst *cpg_inst;
  505. struct iovec iov;
  506. struct req_lib_cpg_membership_get req_lib_cpg_membership_get;
  507. struct res_lib_cpg_membership_get res_lib_cpg_membership_get;
  508. unsigned int i;
  509. if (member_list == NULL) {
  510. return (CS_ERR_INVALID_PARAM);
  511. }
  512. if (member_list_entries == NULL) {
  513. return (CS_ERR_INVALID_PARAM);
  514. }
  515. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  516. if (error != CS_OK) {
  517. return (error);
  518. }
  519. req_lib_cpg_membership_get.header.size = sizeof (struct req_lib_cpg_membership_get);
  520. req_lib_cpg_membership_get.header.id = MESSAGE_REQ_CPG_MEMBERSHIP;
  521. marshall_to_mar_cpg_name_t (&req_lib_cpg_membership_get.group_name,
  522. group_name);
  523. iov.iov_base = (void *)&req_lib_cpg_membership_get;
  524. iov.iov_len = sizeof (struct req_lib_cpg_membership_get);
  525. error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1,
  526. &res_lib_cpg_membership_get, sizeof (res_lib_cpg_membership_get));
  527. if (error != CS_OK) {
  528. goto error_exit;
  529. }
  530. error = res_lib_cpg_membership_get.header.error;
  531. /*
  532. * Copy results to caller
  533. */
  534. *member_list_entries = res_lib_cpg_membership_get.member_count;
  535. if (member_list) {
  536. for (i = 0; i < res_lib_cpg_membership_get.member_count; i++) {
  537. marshall_from_mar_cpg_address_t (&member_list[i],
  538. &res_lib_cpg_membership_get.member_list[i]);
  539. }
  540. }
  541. error_exit:
  542. hdb_handle_put (&cpg_handle_t_db, handle);
  543. return (error);
  544. }
  545. cs_error_t cpg_local_get (
  546. cpg_handle_t handle,
  547. unsigned int *local_nodeid)
  548. {
  549. cs_error_t error;
  550. struct cpg_inst *cpg_inst;
  551. struct iovec iov;
  552. struct req_lib_cpg_local_get req_lib_cpg_local_get;
  553. struct res_lib_cpg_local_get res_lib_cpg_local_get;
  554. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  555. if (error != CS_OK) {
  556. return (error);
  557. }
  558. req_lib_cpg_local_get.header.size = sizeof (struct qb_ipc_request_header);
  559. req_lib_cpg_local_get.header.id = MESSAGE_REQ_CPG_LOCAL_GET;
  560. iov.iov_base = (void *)&req_lib_cpg_local_get;
  561. iov.iov_len = sizeof (struct req_lib_cpg_local_get);
  562. error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1,
  563. &res_lib_cpg_local_get, sizeof (res_lib_cpg_local_get));
  564. if (error != CS_OK) {
  565. goto error_exit;
  566. }
  567. error = res_lib_cpg_local_get.header.error;
  568. *local_nodeid = res_lib_cpg_local_get.local_nodeid;
  569. error_exit:
  570. hdb_handle_put (&cpg_handle_t_db, handle);
  571. return (error);
  572. }
  573. cs_error_t cpg_flow_control_state_get (
  574. cpg_handle_t handle,
  575. cpg_flow_control_state_t *flow_control_state)
  576. {
  577. cs_error_t error;
  578. struct cpg_inst *cpg_inst;
  579. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  580. if (error != CS_OK) {
  581. return (error);
  582. }
  583. *flow_control_state = CPG_FLOW_CONTROL_DISABLED;
  584. error = CS_OK;
  585. hdb_handle_put (&cpg_handle_t_db, handle);
  586. return (error);
  587. }
  588. static int
  589. memory_map (char *path, const char *file, void **buf, size_t bytes)
  590. {
  591. int32_t fd;
  592. void *addr_orig;
  593. void *addr;
  594. int32_t res;
  595. char *buffer;
  596. int32_t i;
  597. size_t written;
  598. size_t page_size;
  599. snprintf (path, PATH_MAX, "/dev/shm/%s", file);
  600. fd = mkstemp (path);
  601. if (fd == -1) {
  602. snprintf (path, PATH_MAX, LOCALSTATEDIR "/run/%s", file);
  603. fd = mkstemp (path);
  604. if (fd == -1) {
  605. return (-1);
  606. }
  607. }
  608. res = ftruncate (fd, bytes);
  609. if (res == -1) {
  610. goto error_close_unlink;
  611. }
  612. page_size = (size_t)sysconf(_SC_PAGESIZE);
  613. buffer = malloc (page_size);
  614. if (buffer == NULL) {
  615. goto error_close_unlink;
  616. }
  617. memset (buffer, 0, page_size);
  618. for (i = 0; i < (bytes / page_size); i++) {
  619. retry_write:
  620. written = write (fd, buffer, page_size);
  621. if (written == -1 && errno == EINTR) {
  622. goto retry_write;
  623. }
  624. if (written != page_size) {
  625. free (buffer);
  626. goto error_close_unlink;
  627. }
  628. }
  629. free (buffer);
  630. addr_orig = mmap (NULL, bytes, PROT_NONE,
  631. MAP_ANONYMOUS | MAP_PRIVATE, -1, 0);
  632. if (addr_orig == MAP_FAILED) {
  633. goto error_close_unlink;
  634. }
  635. addr = mmap (addr_orig, bytes, PROT_READ | PROT_WRITE,
  636. MAP_FIXED | MAP_SHARED, fd, 0);
  637. if (addr != addr_orig) {
  638. goto error_close_unlink;
  639. }
  640. #ifdef COROSYNC_BSD
  641. madvise(addr_orig, bytes, MADV_NOSYNC);
  642. #endif
  643. res = close (fd);
  644. if (res) {
  645. return (-1);
  646. }
  647. *buf = addr_orig;
  648. return 0;
  649. error_close_unlink:
  650. close (fd);
  651. unlink(path);
  652. return -1;
  653. }
  654. cs_error_t cpg_zcb_alloc (
  655. cpg_handle_t handle,
  656. size_t size,
  657. void **buffer)
  658. {
  659. void *buf = NULL;
  660. char path[PATH_MAX];
  661. mar_req_coroipcc_zc_alloc_t req_coroipcc_zc_alloc;
  662. struct qb_ipc_response_header res_coroipcs_zc_alloc;
  663. size_t map_size;
  664. struct iovec iovec;
  665. struct coroipcs_zc_header *hdr;
  666. cs_error_t error;
  667. struct cpg_inst *cpg_inst;
  668. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  669. if (error != CS_OK) {
  670. return (error);
  671. }
  672. map_size = size + sizeof (struct req_lib_cpg_mcast) + sizeof (struct coroipcs_zc_header);
  673. assert(memory_map (path, "corosync_zerocopy-XXXXXX", &buf, map_size) != -1);
  674. req_coroipcc_zc_alloc.header.size = sizeof (mar_req_coroipcc_zc_alloc_t);
  675. req_coroipcc_zc_alloc.header.id = MESSAGE_REQ_CPG_ZC_ALLOC;
  676. req_coroipcc_zc_alloc.map_size = map_size;
  677. strcpy (req_coroipcc_zc_alloc.path_to_file, path);
  678. iovec.iov_base = (void *)&req_coroipcc_zc_alloc;
  679. iovec.iov_len = sizeof (mar_req_coroipcc_zc_alloc_t);
  680. error = coroipcc_msg_send_reply_receive (
  681. cpg_inst->c,
  682. &iovec,
  683. 1,
  684. &res_coroipcs_zc_alloc,
  685. sizeof (struct qb_ipc_response_header));
  686. hdr = (struct coroipcs_zc_header *)buf;
  687. hdr->map_size = map_size;
  688. *buffer = ((char *)buf) + sizeof (struct coroipcs_zc_header);
  689. hdb_handle_put (&cpg_handle_t_db, handle);
  690. *buffer = ((char *)*buffer) + sizeof (struct req_lib_cpg_mcast);
  691. return (error);
  692. }
  693. cs_error_t cpg_zcb_free (
  694. cpg_handle_t handle,
  695. void *buffer)
  696. {
  697. cs_error_t error;
  698. struct cpg_inst *cpg_inst;
  699. mar_req_coroipcc_zc_free_t req_coroipcc_zc_free;
  700. struct qb_ipc_response_header res_coroipcs_zc_free;
  701. struct iovec iovec;
  702. struct coroipcs_zc_header *header = (struct coroipcs_zc_header *)((char *)buffer - sizeof (struct coroipcs_zc_header));
  703. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  704. if (error != CS_OK) {
  705. return (error);
  706. }
  707. req_coroipcc_zc_free.header.size = sizeof (mar_req_coroipcc_zc_free_t);
  708. req_coroipcc_zc_free.header.id = MESSAGE_REQ_CPG_ZC_FREE;
  709. req_coroipcc_zc_free.map_size = header->map_size;
  710. req_coroipcc_zc_free.server_address = header->server_address;
  711. iovec.iov_base = (void *)&req_coroipcc_zc_free;
  712. iovec.iov_len = sizeof (mar_req_coroipcc_zc_free_t);
  713. error = coroipcc_msg_send_reply_receive (
  714. cpg_inst->c,
  715. &iovec,
  716. 1,
  717. &res_coroipcs_zc_free,
  718. sizeof (struct qb_ipc_response_header));
  719. munmap ((void *)header, header->map_size);
  720. hdb_handle_put (&cpg_handle_t_db, handle);
  721. return (error);
  722. }
  723. cs_error_t cpg_zcb_mcast_joined (
  724. cpg_handle_t handle,
  725. cpg_guarantee_t guarantee,
  726. void *msg,
  727. size_t msg_len)
  728. {
  729. cs_error_t error;
  730. struct cpg_inst *cpg_inst;
  731. struct req_lib_cpg_mcast *req_lib_cpg_mcast;
  732. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  733. mar_req_coroipcc_zc_execute_t req_coroipcc_zc_execute;
  734. struct coroipcs_zc_header *hdr;
  735. struct iovec iovec;
  736. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  737. if (error != CS_OK) {
  738. return (error);
  739. }
  740. req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)(((char *)msg) - sizeof (struct req_lib_cpg_mcast));
  741. req_lib_cpg_mcast->header.size = sizeof (struct req_lib_cpg_mcast) +
  742. msg_len;
  743. req_lib_cpg_mcast->header.id = MESSAGE_REQ_CPG_MCAST;
  744. req_lib_cpg_mcast->guarantee = guarantee;
  745. req_lib_cpg_mcast->msglen = msg_len;
  746. hdr = (struct coroipcs_zc_header *)(((char *)req_lib_cpg_mcast) - sizeof (struct coroipcs_zc_header));
  747. req_coroipcc_zc_execute.header.size = sizeof (mar_req_coroipcc_zc_execute_t);
  748. req_coroipcc_zc_execute.header.id = MESSAGE_REQ_CPG_ZC_EXECUTE;
  749. req_coroipcc_zc_execute.server_address = hdr->server_address;
  750. iovec.iov_base = (void *)&req_coroipcc_zc_execute;
  751. iovec.iov_len = sizeof (mar_req_coroipcc_zc_execute_t);
  752. error = coroipcc_msg_send_reply_receive (
  753. cpg_inst->c,
  754. &iovec,
  755. 1,
  756. &res_lib_cpg_mcast,
  757. sizeof(res_lib_cpg_mcast));
  758. if (error != CS_OK) {
  759. goto error_exit;
  760. }
  761. error = res_lib_cpg_mcast.header.error;
  762. error_exit:
  763. hdb_handle_put (&cpg_handle_t_db, handle);
  764. return (error);
  765. }
  766. cs_error_t cpg_mcast_joined (
  767. cpg_handle_t handle,
  768. cpg_guarantee_t guarantee,
  769. const struct iovec *iovec,
  770. unsigned int iov_len)
  771. {
  772. int i;
  773. cs_error_t error;
  774. struct cpg_inst *cpg_inst;
  775. struct iovec iov[64];
  776. struct req_lib_cpg_mcast req_lib_cpg_mcast;
  777. size_t msg_len = 0;
  778. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  779. if (error != CS_OK) {
  780. return (error);
  781. }
  782. for (i = 0; i < iov_len; i++ ) {
  783. msg_len += iovec[i].iov_len;
  784. }
  785. req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_mcast) +
  786. msg_len;
  787. req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_MCAST;
  788. req_lib_cpg_mcast.guarantee = guarantee;
  789. req_lib_cpg_mcast.msglen = msg_len;
  790. iov[0].iov_base = (void *)&req_lib_cpg_mcast;
  791. iov[0].iov_len = sizeof (struct req_lib_cpg_mcast);
  792. memcpy (&iov[1], iovec, iov_len * sizeof (struct iovec));
  793. qb_ipcc_fc_enable_max_set(cpg_inst->c, 2);
  794. error = qb_to_cs_error(qb_ipcc_sendv(cpg_inst->c, iov, iov_len + 1));
  795. qb_ipcc_fc_enable_max_set(cpg_inst->c, 1);
  796. hdb_handle_put (&cpg_handle_t_db, handle);
  797. return (error);
  798. }
  799. cs_error_t cpg_iteration_initialize(
  800. cpg_handle_t handle,
  801. cpg_iteration_type_t iteration_type,
  802. const struct cpg_name *group,
  803. cpg_iteration_handle_t *cpg_iteration_handle)
  804. {
  805. cs_error_t error;
  806. struct iovec iov;
  807. struct cpg_inst *cpg_inst;
  808. struct cpg_iteration_instance_t *cpg_iteration_instance;
  809. struct req_lib_cpg_iterationinitialize req_lib_cpg_iterationinitialize;
  810. struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize;
  811. if (cpg_iteration_handle == NULL) {
  812. return (CS_ERR_INVALID_PARAM);
  813. }
  814. if ((iteration_type == CPG_ITERATION_ONE_GROUP && group == NULL) ||
  815. (iteration_type != CPG_ITERATION_ONE_GROUP && group != NULL)) {
  816. return (CS_ERR_INVALID_PARAM);
  817. }
  818. if (iteration_type != CPG_ITERATION_NAME_ONLY && iteration_type != CPG_ITERATION_ONE_GROUP &&
  819. iteration_type != CPG_ITERATION_ALL) {
  820. return (CS_ERR_INVALID_PARAM);
  821. }
  822. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  823. if (error != CS_OK) {
  824. return (error);
  825. }
  826. error = hdb_error_to_cs (hdb_handle_create (&cpg_iteration_handle_t_db,
  827. sizeof (struct cpg_iteration_instance_t), cpg_iteration_handle));
  828. if (error != CS_OK) {
  829. goto error_put_cpg_db;
  830. }
  831. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, *cpg_iteration_handle,
  832. (void *)&cpg_iteration_instance));
  833. if (error != CS_OK) {
  834. goto error_destroy;
  835. }
  836. cpg_iteration_instance->conn = cpg_inst->c;
  837. list_init (&cpg_iteration_instance->list);
  838. req_lib_cpg_iterationinitialize.header.size = sizeof (struct req_lib_cpg_iterationinitialize);
  839. req_lib_cpg_iterationinitialize.header.id = MESSAGE_REQ_CPG_ITERATIONINITIALIZE;
  840. req_lib_cpg_iterationinitialize.iteration_type = iteration_type;
  841. if (group) {
  842. marshall_to_mar_cpg_name_t (&req_lib_cpg_iterationinitialize.group_name, group);
  843. }
  844. iov.iov_base = (void *)&req_lib_cpg_iterationinitialize;
  845. iov.iov_len = sizeof (struct req_lib_cpg_iterationinitialize);
  846. error = coroipcc_msg_send_reply_receive (cpg_inst->c,
  847. &iov,
  848. 1,
  849. &res_lib_cpg_iterationinitialize,
  850. sizeof (struct res_lib_cpg_iterationinitialize));
  851. if (error != CS_OK) {
  852. goto error_put_destroy;
  853. }
  854. cpg_iteration_instance->executive_iteration_handle =
  855. res_lib_cpg_iterationinitialize.iteration_handle;
  856. cpg_iteration_instance->cpg_iteration_handle = *cpg_iteration_handle;
  857. list_add (&cpg_iteration_instance->list, &cpg_inst->iteration_list_head);
  858. hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  859. hdb_handle_put (&cpg_handle_t_db, handle);
  860. return (res_lib_cpg_iterationinitialize.header.error);
  861. error_put_destroy:
  862. hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  863. error_destroy:
  864. hdb_handle_destroy (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  865. error_put_cpg_db:
  866. hdb_handle_put (&cpg_handle_t_db, handle);
  867. return (error);
  868. }
  869. cs_error_t cpg_iteration_next(
  870. cpg_iteration_handle_t handle,
  871. struct cpg_iteration_description_t *description)
  872. {
  873. cs_error_t error;
  874. struct cpg_iteration_instance_t *cpg_iteration_instance;
  875. struct req_lib_cpg_iterationnext req_lib_cpg_iterationnext;
  876. struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext;
  877. if (description == NULL) {
  878. return CS_ERR_INVALID_PARAM;
  879. }
  880. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle,
  881. (void *)&cpg_iteration_instance));
  882. if (error != CS_OK) {
  883. goto error_exit;
  884. }
  885. req_lib_cpg_iterationnext.header.size = sizeof (struct req_lib_cpg_iterationnext);
  886. req_lib_cpg_iterationnext.header.id = MESSAGE_REQ_CPG_ITERATIONNEXT;
  887. req_lib_cpg_iterationnext.iteration_handle = cpg_iteration_instance->executive_iteration_handle;
  888. error = qb_to_cs_error (qb_ipcc_send (cpg_iteration_instance->conn,
  889. &req_lib_cpg_iterationnext,
  890. req_lib_cpg_iterationnext.header.size));
  891. if (error != CS_OK) {
  892. goto error_put;
  893. }
  894. error = qb_to_cs_error (qb_ipcc_recv (cpg_iteration_instance->conn,
  895. &res_lib_cpg_iterationnext,
  896. sizeof(struct res_lib_cpg_iterationnext), -1));
  897. if (error != CS_OK) {
  898. goto error_put;
  899. }
  900. marshall_from_mar_cpg_iteration_description_t(
  901. description,
  902. &res_lib_cpg_iterationnext.description);
  903. error = res_lib_cpg_iterationnext.header.error;
  904. error_put:
  905. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  906. error_exit:
  907. return (error);
  908. }
  909. cs_error_t cpg_iteration_finalize (
  910. cpg_iteration_handle_t handle)
  911. {
  912. cs_error_t error;
  913. struct iovec iov;
  914. struct cpg_iteration_instance_t *cpg_iteration_instance;
  915. struct req_lib_cpg_iterationfinalize req_lib_cpg_iterationfinalize;
  916. struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize;
  917. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle,
  918. (void *)&cpg_iteration_instance));
  919. if (error != CS_OK) {
  920. goto error_exit;
  921. }
  922. req_lib_cpg_iterationfinalize.header.size = sizeof (struct req_lib_cpg_iterationfinalize);
  923. req_lib_cpg_iterationfinalize.header.id = MESSAGE_REQ_CPG_ITERATIONFINALIZE;
  924. req_lib_cpg_iterationfinalize.iteration_handle = cpg_iteration_instance->executive_iteration_handle;
  925. iov.iov_base = (void *)&req_lib_cpg_iterationfinalize;
  926. iov.iov_len = sizeof (struct req_lib_cpg_iterationfinalize);
  927. error = coroipcc_msg_send_reply_receive (cpg_iteration_instance->conn,
  928. &iov,
  929. 1,
  930. &res_lib_cpg_iterationfinalize,
  931. sizeof (struct req_lib_cpg_iterationfinalize));
  932. if (error != CS_OK) {
  933. goto error_put;
  934. }
  935. cpg_iteration_instance_finalize (cpg_iteration_instance);
  936. hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->cpg_iteration_handle);
  937. return (res_lib_cpg_iterationfinalize.header.error);
  938. error_put:
  939. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  940. error_exit:
  941. return (error);
  942. }
  943. /** @} */