cpg.c 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746
  1. /*
  2. * vi: set autoindent tabstop=4 shiftwidth=4 :
  3. *
  4. * Copyright (c) 2006-2009 Red Hat, Inc.
  5. *
  6. * All rights reserved.
  7. *
  8. * Author: Patrick Caulfield (pcaulfie@redhat.com)
  9. *
  10. * This software licensed under BSD license, the text of which follows:
  11. *
  12. * Redistribution and use in source and binary forms, with or without
  13. * modification, are permitted provided that the following conditions are met:
  14. *
  15. * - Redistributions of source code must retain the above copyright notice,
  16. * this list of conditions and the following disclaimer.
  17. * - Redistributions in binary form must reproduce the above copyright notice,
  18. * this list of conditions and the following disclaimer in the documentation
  19. * and/or other materials provided with the distribution.
  20. * - Neither the name of the MontaVista Software, Inc. nor the names of its
  21. * contributors may be used to endorse or promote products derived from this
  22. * software without specific prior written permission.
  23. *
  24. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  25. * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  26. * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  27. * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  28. * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  29. * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  30. * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  31. * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  32. * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  33. * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  34. * THE POSSIBILITY OF SUCH DAMAGE.
  35. */
  36. /*
  37. * Provides a closed process group API using the coroipcc executive
  38. */
  39. #include <config.h>
  40. #include <stdlib.h>
  41. #include <string.h>
  42. #include <unistd.h>
  43. #include <pthread.h>
  44. #include <sys/types.h>
  45. #include <sys/socket.h>
  46. #include <errno.h>
  47. #include <corosync/corotypes.h>
  48. #include <corosync/coroipc_types.h>
  49. #include <corosync/coroipcc.h>
  50. #include <corosync/corodefs.h>
  51. #include <corosync/cpg.h>
  52. #include <corosync/mar_gen.h>
  53. #include <corosync/mar_cpg.h>
  54. #include <corosync/ipc_cpg.h>
  55. struct cpg_inst {
  56. void *ipc_ctx;
  57. int finalize;
  58. cpg_callbacks_t callbacks;
  59. void *context;
  60. pthread_mutex_t response_mutex;
  61. pthread_mutex_t dispatch_mutex;
  62. };
  63. static void cpg_instance_destructor (void *instance);
  64. DECLARE_SAHDB_DATABASE(cpg_handle_t_db,cpg_instance_destructor);
  65. /*
  66. * Clean up function for a cpg instance (cpg_nitialize) handle
  67. */
  68. static void cpg_instance_destructor (void *instance)
  69. {
  70. struct cpg_inst *cpg_inst = instance;
  71. pthread_mutex_destroy (&cpg_inst->response_mutex);
  72. pthread_mutex_destroy (&cpg_inst->dispatch_mutex);
  73. }
  74. /**
  75. * @defgroup cpg_coroipcc The closed process group API
  76. * @ingroup coroipcc
  77. *
  78. * @{
  79. */
  80. cs_error_t cpg_initialize (
  81. cpg_handle_t *handle,
  82. cpg_callbacks_t *callbacks)
  83. {
  84. cs_error_t error;
  85. struct cpg_inst *cpg_inst;
  86. error = saHandleCreate (&cpg_handle_t_db, sizeof (struct cpg_inst), handle);
  87. if (error != CS_OK) {
  88. goto error_no_destroy;
  89. }
  90. error = saHandleInstanceGet (&cpg_handle_t_db, *handle, (void *)&cpg_inst);
  91. if (error != CS_OK) {
  92. goto error_destroy;
  93. }
  94. error = coroipcc_service_connect (
  95. COROSYNC_SOCKET_NAME,
  96. CPG_SERVICE,
  97. IPC_REQUEST_SIZE,
  98. IPC_RESPONSE_SIZE,
  99. IPC_DISPATCH_SIZE,
  100. &cpg_inst->ipc_ctx);
  101. if (error != CS_OK) {
  102. goto error_put_destroy;
  103. }
  104. memcpy (&cpg_inst->callbacks, callbacks, sizeof (cpg_callbacks_t));
  105. pthread_mutex_init (&cpg_inst->response_mutex, NULL);
  106. pthread_mutex_init (&cpg_inst->dispatch_mutex, NULL);
  107. saHandleInstancePut (&cpg_handle_t_db, *handle);
  108. return (CS_OK);
  109. error_put_destroy:
  110. saHandleInstancePut (&cpg_handle_t_db, *handle);
  111. error_destroy:
  112. saHandleDestroy (&cpg_handle_t_db, *handle);
  113. error_no_destroy:
  114. return (error);
  115. }
  116. cs_error_t cpg_finalize (
  117. cpg_handle_t handle)
  118. {
  119. struct cpg_inst *cpg_inst;
  120. cs_error_t error;
  121. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  122. if (error != CS_OK) {
  123. return (error);
  124. }
  125. pthread_mutex_lock (&cpg_inst->response_mutex);
  126. /*
  127. * Another thread has already started finalizing
  128. */
  129. if (cpg_inst->finalize) {
  130. pthread_mutex_unlock (&cpg_inst->response_mutex);
  131. saHandleInstancePut (&cpg_handle_t_db, handle);
  132. return (CPG_ERR_BAD_HANDLE);
  133. }
  134. cpg_inst->finalize = 1;
  135. coroipcc_service_disconnect (cpg_inst->ipc_ctx);
  136. pthread_mutex_unlock (&cpg_inst->response_mutex);
  137. saHandleDestroy (&cpg_handle_t_db, handle);
  138. saHandleInstancePut (&cpg_handle_t_db, handle);
  139. return (CPG_OK);
  140. }
  141. cs_error_t cpg_fd_get (
  142. cpg_handle_t handle,
  143. int *fd)
  144. {
  145. cs_error_t error;
  146. struct cpg_inst *cpg_inst;
  147. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  148. if (error != CS_OK) {
  149. return (error);
  150. }
  151. *fd = coroipcc_fd_get (cpg_inst->ipc_ctx);
  152. saHandleInstancePut (&cpg_handle_t_db, handle);
  153. return (CS_OK);
  154. }
  155. cs_error_t cpg_context_get (
  156. cpg_handle_t handle,
  157. void **context)
  158. {
  159. cs_error_t error;
  160. struct cpg_inst *cpg_inst;
  161. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  162. if (error != CS_OK) {
  163. return (error);
  164. }
  165. *context = cpg_inst->context;
  166. saHandleInstancePut (&cpg_handle_t_db, handle);
  167. return (CS_OK);
  168. }
  169. cs_error_t cpg_context_set (
  170. cpg_handle_t handle,
  171. void *context)
  172. {
  173. cs_error_t error;
  174. struct cpg_inst *cpg_inst;
  175. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  176. if (error != CS_OK) {
  177. return (error);
  178. }
  179. cpg_inst->context = context;
  180. saHandleInstancePut (&cpg_handle_t_db, handle);
  181. return (CS_OK);
  182. }
  183. cs_error_t cpg_dispatch (
  184. cpg_handle_t handle,
  185. cs_dispatch_flags_t dispatch_types)
  186. {
  187. int timeout = -1;
  188. cs_error_t error;
  189. int cont = 1; /* always continue do loop except when set to 0 */
  190. int dispatch_avail;
  191. struct cpg_inst *cpg_inst;
  192. struct res_lib_cpg_confchg_callback *res_cpg_confchg_callback;
  193. struct res_lib_cpg_deliver_callback *res_cpg_deliver_callback;
  194. cpg_callbacks_t callbacks;
  195. coroipc_response_header_t *dispatch_data;
  196. int ignore_dispatch = 0;
  197. struct cpg_address member_list[CPG_MEMBERS_MAX];
  198. struct cpg_address left_list[CPG_MEMBERS_MAX];
  199. struct cpg_address joined_list[CPG_MEMBERS_MAX];
  200. struct cpg_name group_name;
  201. mar_cpg_address_t *left_list_start;
  202. mar_cpg_address_t *joined_list_start;
  203. unsigned int i;
  204. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  205. if (error != CS_OK) {
  206. return (error);
  207. }
  208. /*
  209. * Timeout instantly for SA_DISPATCH_ONE or SA_DISPATCH_ALL and
  210. * wait indefinately for SA_DISPATCH_BLOCKING
  211. */
  212. if (dispatch_types == CPG_DISPATCH_ALL) {
  213. timeout = 0;
  214. }
  215. do {
  216. pthread_mutex_lock (&cpg_inst->dispatch_mutex);
  217. dispatch_avail = coroipcc_dispatch_get (
  218. cpg_inst->ipc_ctx,
  219. (void **)&dispatch_data,
  220. timeout);
  221. pthread_mutex_unlock (&cpg_inst->dispatch_mutex);
  222. if (dispatch_avail == 0 && dispatch_types == CPG_DISPATCH_ALL) {
  223. pthread_mutex_unlock (&cpg_inst->dispatch_mutex);
  224. break; /* exit do while cont is 1 loop */
  225. } else
  226. if (dispatch_avail == 0) {
  227. pthread_mutex_unlock (&cpg_inst->dispatch_mutex);
  228. continue; /* next poll */
  229. }
  230. if (dispatch_avail == -1) {
  231. if (cpg_inst->finalize == 1) {
  232. error = CS_OK;
  233. } else {
  234. error = CS_ERR_LIBRARY;
  235. }
  236. goto error_put;
  237. }
  238. /*
  239. * Make copy of callbacks, message data, unlock instance, and call callback
  240. * A risk of this dispatch method is that the callback routines may
  241. * operate at the same time that cpgFinalize has been called.
  242. */
  243. memcpy (&callbacks, &cpg_inst->callbacks, sizeof (cpg_callbacks_t));
  244. /*
  245. * Dispatch incoming message
  246. */
  247. switch (dispatch_data->id) {
  248. case MESSAGE_RES_CPG_DELIVER_CALLBACK:
  249. res_cpg_deliver_callback = (struct res_lib_cpg_deliver_callback *)dispatch_data;
  250. marshall_from_mar_cpg_name_t (
  251. &group_name,
  252. &res_cpg_deliver_callback->group_name);
  253. callbacks.cpg_deliver_fn (handle,
  254. &group_name,
  255. res_cpg_deliver_callback->nodeid,
  256. res_cpg_deliver_callback->pid,
  257. &res_cpg_deliver_callback->message,
  258. res_cpg_deliver_callback->msglen);
  259. break;
  260. case MESSAGE_RES_CPG_CONFCHG_CALLBACK:
  261. res_cpg_confchg_callback = (struct res_lib_cpg_confchg_callback *)dispatch_data;
  262. for (i = 0; i < res_cpg_confchg_callback->member_list_entries; i++) {
  263. marshall_from_mar_cpg_address_t (&member_list[i],
  264. &res_cpg_confchg_callback->member_list[i]);
  265. }
  266. left_list_start = res_cpg_confchg_callback->member_list +
  267. res_cpg_confchg_callback->member_list_entries;
  268. for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) {
  269. marshall_from_mar_cpg_address_t (&left_list[i],
  270. &left_list_start[i]);
  271. }
  272. joined_list_start = res_cpg_confchg_callback->member_list +
  273. res_cpg_confchg_callback->member_list_entries +
  274. res_cpg_confchg_callback->left_list_entries;
  275. for (i = 0; i < res_cpg_confchg_callback->joined_list_entries; i++) {
  276. marshall_from_mar_cpg_address_t (&joined_list[i],
  277. &joined_list_start[i]);
  278. }
  279. marshall_from_mar_cpg_name_t (
  280. &group_name,
  281. &res_cpg_confchg_callback->group_name);
  282. callbacks.cpg_confchg_fn (handle,
  283. &group_name,
  284. member_list,
  285. res_cpg_confchg_callback->member_list_entries,
  286. left_list,
  287. res_cpg_confchg_callback->left_list_entries,
  288. joined_list,
  289. res_cpg_confchg_callback->joined_list_entries);
  290. break;
  291. default:
  292. coroipcc_dispatch_put (cpg_inst->ipc_ctx);
  293. error = CS_ERR_LIBRARY;
  294. goto error_put;
  295. break;
  296. }
  297. coroipcc_dispatch_put (cpg_inst->ipc_ctx);
  298. /*
  299. * Determine if more messages should be processed
  300. * */
  301. switch (dispatch_types) {
  302. case CPG_DISPATCH_ONE:
  303. if (ignore_dispatch) {
  304. ignore_dispatch = 0;
  305. } else {
  306. cont = 0;
  307. }
  308. break;
  309. case CPG_DISPATCH_ALL:
  310. if (ignore_dispatch) {
  311. ignore_dispatch = 0;
  312. }
  313. break;
  314. case CPG_DISPATCH_BLOCKING:
  315. break;
  316. }
  317. } while (cont);
  318. error_put:
  319. saHandleInstancePut (&cpg_handle_t_db, handle);
  320. return (error);
  321. }
  322. cs_error_t cpg_join (
  323. cpg_handle_t handle,
  324. const struct cpg_name *group)
  325. {
  326. cs_error_t error;
  327. struct cpg_inst *cpg_inst;
  328. struct iovec iov[2];
  329. struct req_lib_cpg_join req_lib_cpg_join;
  330. struct res_lib_cpg_join res_lib_cpg_join;
  331. struct req_lib_cpg_trackstart req_lib_cpg_trackstart;
  332. struct res_lib_cpg_trackstart res_lib_cpg_trackstart;
  333. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  334. if (error != CS_OK) {
  335. return (error);
  336. }
  337. pthread_mutex_lock (&cpg_inst->response_mutex);
  338. /* Automatically add a tracker */
  339. req_lib_cpg_trackstart.header.size = sizeof (struct req_lib_cpg_trackstart);
  340. req_lib_cpg_trackstart.header.id = MESSAGE_REQ_CPG_TRACKSTART;
  341. marshall_to_mar_cpg_name_t (&req_lib_cpg_trackstart.group_name,
  342. group);
  343. iov[0].iov_base = &req_lib_cpg_trackstart;
  344. iov[0].iov_len = sizeof (struct req_lib_cpg_trackstart);
  345. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, iov, 1,
  346. &res_lib_cpg_trackstart, sizeof (struct res_lib_cpg_trackstart));
  347. if (error != CS_OK) {
  348. pthread_mutex_unlock (&cpg_inst->response_mutex);
  349. goto error_exit;
  350. }
  351. /* Now join */
  352. req_lib_cpg_join.header.size = sizeof (struct req_lib_cpg_join);
  353. req_lib_cpg_join.header.id = MESSAGE_REQ_CPG_JOIN;
  354. req_lib_cpg_join.pid = getpid();
  355. marshall_to_mar_cpg_name_t (&req_lib_cpg_join.group_name,
  356. group);
  357. iov[0].iov_base = &req_lib_cpg_join;
  358. iov[0].iov_len = sizeof (struct req_lib_cpg_join);
  359. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, iov, 1,
  360. &res_lib_cpg_join, sizeof (struct res_lib_cpg_join));
  361. pthread_mutex_unlock (&cpg_inst->response_mutex);
  362. if (error != CS_OK) {
  363. goto error_exit;
  364. }
  365. error = res_lib_cpg_join.header.error;
  366. error_exit:
  367. saHandleInstancePut (&cpg_handle_t_db, handle);
  368. return (error);
  369. }
  370. cs_error_t cpg_leave (
  371. cpg_handle_t handle,
  372. const struct cpg_name *group)
  373. {
  374. cs_error_t error;
  375. struct cpg_inst *cpg_inst;
  376. struct iovec iov[2];
  377. struct req_lib_cpg_leave req_lib_cpg_leave;
  378. struct res_lib_cpg_leave res_lib_cpg_leave;
  379. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  380. if (error != CS_OK) {
  381. return (error);
  382. }
  383. req_lib_cpg_leave.header.size = sizeof (struct req_lib_cpg_leave);
  384. req_lib_cpg_leave.header.id = MESSAGE_REQ_CPG_LEAVE;
  385. req_lib_cpg_leave.pid = getpid();
  386. marshall_to_mar_cpg_name_t (&req_lib_cpg_leave.group_name,
  387. group);
  388. iov[0].iov_base = &req_lib_cpg_leave;
  389. iov[0].iov_len = sizeof (struct req_lib_cpg_leave);
  390. pthread_mutex_lock (&cpg_inst->response_mutex);
  391. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, iov, 1,
  392. &res_lib_cpg_leave, sizeof (struct res_lib_cpg_leave));
  393. pthread_mutex_unlock (&cpg_inst->response_mutex);
  394. if (error != CS_OK) {
  395. goto error_exit;
  396. }
  397. error = res_lib_cpg_leave.header.error;
  398. error_exit:
  399. saHandleInstancePut (&cpg_handle_t_db, handle);
  400. return (error);
  401. }
  402. cs_error_t cpg_membership_get (
  403. cpg_handle_t handle,
  404. struct cpg_name *group_name,
  405. struct cpg_address *member_list,
  406. int *member_list_entries)
  407. {
  408. cs_error_t error;
  409. struct cpg_inst *cpg_inst;
  410. struct iovec iov;
  411. struct req_lib_cpg_membership req_lib_cpg_membership_get;
  412. struct res_lib_cpg_confchg_callback res_lib_cpg_membership_get;
  413. unsigned int i;
  414. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  415. if (error != CS_OK) {
  416. return (error);
  417. }
  418. req_lib_cpg_membership_get.header.size = sizeof (coroipc_request_header_t);
  419. req_lib_cpg_membership_get.header.id = MESSAGE_REQ_CPG_MEMBERSHIP;
  420. iov.iov_base = &req_lib_cpg_membership_get;
  421. iov.iov_len = sizeof (coroipc_request_header_t);
  422. pthread_mutex_lock (&cpg_inst->response_mutex);
  423. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, &iov, 1,
  424. &res_lib_cpg_membership_get, sizeof (coroipc_response_header_t));
  425. pthread_mutex_unlock (&cpg_inst->response_mutex);
  426. if (error != CS_OK) {
  427. goto error_exit;
  428. }
  429. error = res_lib_cpg_membership_get.header.error;
  430. /*
  431. * Copy results to caller
  432. */
  433. *member_list_entries = res_lib_cpg_membership_get.member_list_entries;
  434. if (member_list) {
  435. for (i = 0; i < res_lib_cpg_membership_get.member_list_entries; i++) {
  436. marshall_from_mar_cpg_address_t (&member_list[i],
  437. &res_lib_cpg_membership_get.member_list[i]);
  438. }
  439. }
  440. error_exit:
  441. saHandleInstancePut (&cpg_handle_t_db, handle);
  442. return (error);
  443. }
  444. cs_error_t cpg_local_get (
  445. cpg_handle_t handle,
  446. unsigned int *local_nodeid)
  447. {
  448. cs_error_t error;
  449. struct cpg_inst *cpg_inst;
  450. struct iovec iov;
  451. struct req_lib_cpg_local_get req_lib_cpg_local_get;
  452. struct res_lib_cpg_local_get res_lib_cpg_local_get;
  453. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  454. if (error != CS_OK) {
  455. return (error);
  456. }
  457. req_lib_cpg_local_get.header.size = sizeof (coroipc_request_header_t);
  458. req_lib_cpg_local_get.header.id = MESSAGE_REQ_CPG_LOCAL_GET;
  459. iov.iov_base = &req_lib_cpg_local_get;
  460. iov.iov_len = sizeof (struct req_lib_cpg_local_get);
  461. pthread_mutex_lock (&cpg_inst->response_mutex);
  462. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, &iov, 1,
  463. &res_lib_cpg_local_get, sizeof (res_lib_cpg_local_get));
  464. pthread_mutex_unlock (&cpg_inst->response_mutex);
  465. if (error != CS_OK) {
  466. goto error_exit;
  467. }
  468. error = res_lib_cpg_local_get.header.error;
  469. *local_nodeid = res_lib_cpg_local_get.local_nodeid;
  470. error_exit:
  471. saHandleInstancePut (&cpg_handle_t_db, handle);
  472. return (error);
  473. }
  474. cs_error_t cpg_flow_control_state_get (
  475. cpg_handle_t handle,
  476. cpg_flow_control_state_t *flow_control_state)
  477. {
  478. cs_error_t error;
  479. struct cpg_inst *cpg_inst;
  480. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  481. if (error != CS_OK) {
  482. return (error);
  483. }
  484. *flow_control_state = coroipcc_dispatch_flow_control_get (cpg_inst->ipc_ctx);
  485. saHandleInstancePut (&cpg_handle_t_db, handle);
  486. return (error);
  487. }
  488. cs_error_t cpg_zcb_alloc (
  489. cpg_handle_t handle,
  490. size_t size,
  491. void **buffer)
  492. {
  493. cs_error_t error;
  494. struct cpg_inst *cpg_inst;
  495. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  496. if (error != CS_OK) {
  497. return (error);
  498. }
  499. error = coroipcc_zcb_alloc (cpg_inst->ipc_ctx,
  500. buffer,
  501. size,
  502. sizeof (struct req_lib_cpg_mcast));
  503. saHandleInstancePut (&cpg_handle_t_db, handle);
  504. *buffer = ((char *)*buffer) + sizeof (struct req_lib_cpg_mcast);
  505. return (error);
  506. }
  507. cs_error_t cpg_zcb_free (
  508. cpg_handle_t handle,
  509. void *buffer)
  510. {
  511. cs_error_t error;
  512. struct cpg_inst *cpg_inst;
  513. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  514. if (error != CS_OK) {
  515. return (error);
  516. }
  517. coroipcc_zcb_free (cpg_inst->ipc_ctx, ((char *)buffer) - sizeof (struct req_lib_cpg_mcast));
  518. saHandleInstancePut (&cpg_handle_t_db, handle);
  519. return (error);
  520. }
  521. cs_error_t cpg_zcb_mcast_joined (
  522. cpg_handle_t handle,
  523. cpg_guarantee_t guarantee,
  524. void *msg,
  525. size_t msg_len)
  526. {
  527. cs_error_t error;
  528. struct cpg_inst *cpg_inst;
  529. struct req_lib_cpg_mcast *req_lib_cpg_mcast;
  530. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  531. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  532. if (error != CS_OK) {
  533. return (error);
  534. }
  535. req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)(((char *)msg) - sizeof (struct req_lib_cpg_mcast));
  536. req_lib_cpg_mcast->header.size = sizeof (struct req_lib_cpg_mcast) +
  537. msg_len;
  538. req_lib_cpg_mcast->header.id = MESSAGE_REQ_CPG_MCAST;
  539. req_lib_cpg_mcast->guarantee = guarantee;
  540. req_lib_cpg_mcast->msglen = msg_len;
  541. pthread_mutex_lock (&cpg_inst->response_mutex);
  542. error = coroipcc_zcb_msg_send_reply_receive (
  543. cpg_inst->ipc_ctx,
  544. req_lib_cpg_mcast,
  545. &res_lib_cpg_mcast,
  546. sizeof (res_lib_cpg_mcast));
  547. pthread_mutex_unlock (&cpg_inst->response_mutex);
  548. if (error != CS_OK) {
  549. goto error_exit;
  550. }
  551. error = res_lib_cpg_mcast.header.error;
  552. error_exit:
  553. saHandleInstancePut (&cpg_handle_t_db, handle);
  554. return (error);
  555. }
  556. cs_error_t cpg_mcast_joined (
  557. cpg_handle_t handle,
  558. cpg_guarantee_t guarantee,
  559. const struct iovec *iovec,
  560. unsigned int iov_len)
  561. {
  562. int i;
  563. cs_error_t error;
  564. struct cpg_inst *cpg_inst;
  565. struct iovec iov[64];
  566. struct req_lib_cpg_mcast req_lib_cpg_mcast;
  567. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  568. size_t msg_len = 0;
  569. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  570. if (error != CS_OK) {
  571. return (error);
  572. }
  573. for (i = 0; i < iov_len; i++ ) {
  574. msg_len += iovec[i].iov_len;
  575. }
  576. req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_mcast) +
  577. msg_len;
  578. req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_MCAST;
  579. req_lib_cpg_mcast.guarantee = guarantee;
  580. req_lib_cpg_mcast.msglen = msg_len;
  581. iov[0].iov_base = &req_lib_cpg_mcast;
  582. iov[0].iov_len = sizeof (struct req_lib_cpg_mcast);
  583. memcpy (&iov[1], iovec, iov_len * sizeof (struct iovec));
  584. pthread_mutex_lock (&cpg_inst->response_mutex);
  585. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, iov,
  586. iov_len + 1, &res_lib_cpg_mcast, sizeof (res_lib_cpg_mcast));
  587. pthread_mutex_unlock (&cpg_inst->response_mutex);
  588. if (error != CS_OK) {
  589. goto error_exit;
  590. }
  591. error = res_lib_cpg_mcast.header.error;
  592. error_exit:
  593. saHandleInstancePut (&cpg_handle_t_db, handle);
  594. return (error);
  595. }
  596. /** @} */