cpg.c 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742
  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/cpg.h>
  49. #include <corosync/ipc_cpg.h>
  50. #include <corosync/mar_cpg.h>
  51. #include <corosync/coroipcc.h>
  52. struct cpg_inst {
  53. void *ipc_ctx;
  54. int finalize;
  55. cpg_callbacks_t callbacks;
  56. void *context;
  57. pthread_mutex_t response_mutex;
  58. pthread_mutex_t dispatch_mutex;
  59. };
  60. static void cpg_instance_destructor (void *instance);
  61. DECLARE_SAHDB_DATABASE(cpg_handle_t_db,cpg_instance_destructor);
  62. /*
  63. * Clean up function for a cpg instance (cpg_nitialize) handle
  64. */
  65. static void cpg_instance_destructor (void *instance)
  66. {
  67. struct cpg_inst *cpg_inst = instance;
  68. pthread_mutex_destroy (&cpg_inst->response_mutex);
  69. pthread_mutex_destroy (&cpg_inst->dispatch_mutex);
  70. }
  71. /**
  72. * @defgroup cpg_coroipcc The closed process group API
  73. * @ingroup coroipcc
  74. *
  75. * @{
  76. */
  77. cs_error_t cpg_initialize (
  78. cpg_handle_t *handle,
  79. cpg_callbacks_t *callbacks)
  80. {
  81. cs_error_t error;
  82. struct cpg_inst *cpg_inst;
  83. error = saHandleCreate (&cpg_handle_t_db, sizeof (struct cpg_inst), handle);
  84. if (error != CS_OK) {
  85. goto error_no_destroy;
  86. }
  87. error = saHandleInstanceGet (&cpg_handle_t_db, *handle, (void *)&cpg_inst);
  88. if (error != CS_OK) {
  89. goto error_destroy;
  90. }
  91. error = coroipcc_service_connect (
  92. IPC_SOCKET_NAME,
  93. CPG_SERVICE,
  94. IPC_REQUEST_SIZE,
  95. IPC_RESPONSE_SIZE,
  96. IPC_DISPATCH_SIZE,
  97. &cpg_inst->ipc_ctx);
  98. if (error != CS_OK) {
  99. goto error_put_destroy;
  100. }
  101. memcpy (&cpg_inst->callbacks, callbacks, sizeof (cpg_callbacks_t));
  102. pthread_mutex_init (&cpg_inst->response_mutex, NULL);
  103. pthread_mutex_init (&cpg_inst->dispatch_mutex, NULL);
  104. saHandleInstancePut (&cpg_handle_t_db, *handle);
  105. return (CS_OK);
  106. error_put_destroy:
  107. saHandleInstancePut (&cpg_handle_t_db, *handle);
  108. error_destroy:
  109. saHandleDestroy (&cpg_handle_t_db, *handle);
  110. error_no_destroy:
  111. return (error);
  112. }
  113. cs_error_t cpg_finalize (
  114. cpg_handle_t handle)
  115. {
  116. struct cpg_inst *cpg_inst;
  117. cs_error_t error;
  118. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  119. if (error != CS_OK) {
  120. return (error);
  121. }
  122. pthread_mutex_lock (&cpg_inst->response_mutex);
  123. /*
  124. * Another thread has already started finalizing
  125. */
  126. if (cpg_inst->finalize) {
  127. pthread_mutex_unlock (&cpg_inst->response_mutex);
  128. saHandleInstancePut (&cpg_handle_t_db, handle);
  129. return (CPG_ERR_BAD_HANDLE);
  130. }
  131. cpg_inst->finalize = 1;
  132. coroipcc_service_disconnect (cpg_inst->ipc_ctx);
  133. pthread_mutex_unlock (&cpg_inst->response_mutex);
  134. saHandleDestroy (&cpg_handle_t_db, handle);
  135. saHandleInstancePut (&cpg_handle_t_db, handle);
  136. return (CPG_OK);
  137. }
  138. cs_error_t cpg_fd_get (
  139. cpg_handle_t handle,
  140. int *fd)
  141. {
  142. cs_error_t error;
  143. struct cpg_inst *cpg_inst;
  144. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  145. if (error != CS_OK) {
  146. return (error);
  147. }
  148. *fd = coroipcc_fd_get (cpg_inst->ipc_ctx);
  149. saHandleInstancePut (&cpg_handle_t_db, handle);
  150. return (CS_OK);
  151. }
  152. cs_error_t cpg_context_get (
  153. cpg_handle_t handle,
  154. void **context)
  155. {
  156. cs_error_t error;
  157. struct cpg_inst *cpg_inst;
  158. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  159. if (error != CS_OK) {
  160. return (error);
  161. }
  162. *context = cpg_inst->context;
  163. saHandleInstancePut (&cpg_handle_t_db, handle);
  164. return (CS_OK);
  165. }
  166. cs_error_t cpg_context_set (
  167. cpg_handle_t handle,
  168. void *context)
  169. {
  170. cs_error_t error;
  171. struct cpg_inst *cpg_inst;
  172. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  173. if (error != CS_OK) {
  174. return (error);
  175. }
  176. cpg_inst->context = context;
  177. saHandleInstancePut (&cpg_handle_t_db, handle);
  178. return (CS_OK);
  179. }
  180. cs_error_t cpg_dispatch (
  181. cpg_handle_t handle,
  182. cs_dispatch_flags_t dispatch_types)
  183. {
  184. int timeout = -1;
  185. cs_error_t error;
  186. int cont = 1; /* always continue do loop except when set to 0 */
  187. int dispatch_avail;
  188. struct cpg_inst *cpg_inst;
  189. struct res_lib_cpg_confchg_callback *res_cpg_confchg_callback;
  190. struct res_lib_cpg_deliver_callback *res_cpg_deliver_callback;
  191. cpg_callbacks_t callbacks;
  192. mar_res_header_t *dispatch_data;
  193. int ignore_dispatch = 0;
  194. struct cpg_address member_list[CPG_MEMBERS_MAX];
  195. struct cpg_address left_list[CPG_MEMBERS_MAX];
  196. struct cpg_address joined_list[CPG_MEMBERS_MAX];
  197. struct cpg_name group_name;
  198. mar_cpg_address_t *left_list_start;
  199. mar_cpg_address_t *joined_list_start;
  200. unsigned int i;
  201. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  202. if (error != CS_OK) {
  203. return (error);
  204. }
  205. /*
  206. * Timeout instantly for SA_DISPATCH_ONE or SA_DISPATCH_ALL and
  207. * wait indefinately for SA_DISPATCH_BLOCKING
  208. */
  209. if (dispatch_types == CPG_DISPATCH_ALL) {
  210. timeout = 0;
  211. }
  212. do {
  213. pthread_mutex_lock (&cpg_inst->dispatch_mutex);
  214. dispatch_avail = coroipcc_dispatch_get (
  215. cpg_inst->ipc_ctx,
  216. (void **)&dispatch_data,
  217. timeout);
  218. pthread_mutex_unlock (&cpg_inst->dispatch_mutex);
  219. if (dispatch_avail == 0 && dispatch_types == CPG_DISPATCH_ALL) {
  220. pthread_mutex_unlock (&cpg_inst->dispatch_mutex);
  221. break; /* exit do while cont is 1 loop */
  222. } else
  223. if (dispatch_avail == 0) {
  224. pthread_mutex_unlock (&cpg_inst->dispatch_mutex);
  225. continue; /* next poll */
  226. }
  227. if (dispatch_avail == -1) {
  228. if (cpg_inst->finalize == 1) {
  229. error = CS_OK;
  230. } else {
  231. error = CS_ERR_LIBRARY;
  232. }
  233. goto error_put;
  234. }
  235. /*
  236. * Make copy of callbacks, message data, unlock instance, and call callback
  237. * A risk of this dispatch method is that the callback routines may
  238. * operate at the same time that cpgFinalize has been called.
  239. */
  240. memcpy (&callbacks, &cpg_inst->callbacks, sizeof (cpg_callbacks_t));
  241. /*
  242. * Dispatch incoming message
  243. */
  244. switch (dispatch_data->id) {
  245. case MESSAGE_RES_CPG_DELIVER_CALLBACK:
  246. res_cpg_deliver_callback = (struct res_lib_cpg_deliver_callback *)dispatch_data;
  247. marshall_from_mar_cpg_name_t (
  248. &group_name,
  249. &res_cpg_deliver_callback->group_name);
  250. callbacks.cpg_deliver_fn (handle,
  251. &group_name,
  252. res_cpg_deliver_callback->nodeid,
  253. res_cpg_deliver_callback->pid,
  254. &res_cpg_deliver_callback->message,
  255. res_cpg_deliver_callback->msglen);
  256. break;
  257. case MESSAGE_RES_CPG_CONFCHG_CALLBACK:
  258. res_cpg_confchg_callback = (struct res_lib_cpg_confchg_callback *)dispatch_data;
  259. for (i = 0; i < res_cpg_confchg_callback->member_list_entries; i++) {
  260. marshall_from_mar_cpg_address_t (&member_list[i],
  261. &res_cpg_confchg_callback->member_list[i]);
  262. }
  263. left_list_start = res_cpg_confchg_callback->member_list +
  264. res_cpg_confchg_callback->member_list_entries;
  265. for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) {
  266. marshall_from_mar_cpg_address_t (&left_list[i],
  267. &left_list_start[i]);
  268. }
  269. joined_list_start = res_cpg_confchg_callback->member_list +
  270. res_cpg_confchg_callback->member_list_entries +
  271. res_cpg_confchg_callback->left_list_entries;
  272. for (i = 0; i < res_cpg_confchg_callback->joined_list_entries; i++) {
  273. marshall_from_mar_cpg_address_t (&joined_list[i],
  274. &joined_list_start[i]);
  275. }
  276. marshall_from_mar_cpg_name_t (
  277. &group_name,
  278. &res_cpg_confchg_callback->group_name);
  279. callbacks.cpg_confchg_fn (handle,
  280. &group_name,
  281. member_list,
  282. res_cpg_confchg_callback->member_list_entries,
  283. left_list,
  284. res_cpg_confchg_callback->left_list_entries,
  285. joined_list,
  286. res_cpg_confchg_callback->joined_list_entries);
  287. break;
  288. default:
  289. coroipcc_dispatch_put (cpg_inst->ipc_ctx);
  290. error = CS_ERR_LIBRARY;
  291. goto error_put;
  292. break;
  293. }
  294. coroipcc_dispatch_put (cpg_inst->ipc_ctx);
  295. /*
  296. * Determine if more messages should be processed
  297. * */
  298. switch (dispatch_types) {
  299. case CPG_DISPATCH_ONE:
  300. if (ignore_dispatch) {
  301. ignore_dispatch = 0;
  302. } else {
  303. cont = 0;
  304. }
  305. break;
  306. case CPG_DISPATCH_ALL:
  307. if (ignore_dispatch) {
  308. ignore_dispatch = 0;
  309. }
  310. break;
  311. case CPG_DISPATCH_BLOCKING:
  312. break;
  313. }
  314. } while (cont);
  315. error_put:
  316. saHandleInstancePut (&cpg_handle_t_db, handle);
  317. return (error);
  318. }
  319. cs_error_t cpg_join (
  320. cpg_handle_t handle,
  321. const struct cpg_name *group)
  322. {
  323. cs_error_t error;
  324. struct cpg_inst *cpg_inst;
  325. struct iovec iov[2];
  326. struct req_lib_cpg_join req_lib_cpg_join;
  327. struct res_lib_cpg_join res_lib_cpg_join;
  328. struct req_lib_cpg_trackstart req_lib_cpg_trackstart;
  329. struct res_lib_cpg_trackstart res_lib_cpg_trackstart;
  330. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  331. if (error != CS_OK) {
  332. return (error);
  333. }
  334. pthread_mutex_lock (&cpg_inst->response_mutex);
  335. /* Automatically add a tracker */
  336. req_lib_cpg_trackstart.header.size = sizeof (struct req_lib_cpg_trackstart);
  337. req_lib_cpg_trackstart.header.id = MESSAGE_REQ_CPG_TRACKSTART;
  338. marshall_to_mar_cpg_name_t (&req_lib_cpg_trackstart.group_name,
  339. group);
  340. iov[0].iov_base = &req_lib_cpg_trackstart;
  341. iov[0].iov_len = sizeof (struct req_lib_cpg_trackstart);
  342. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, iov, 1,
  343. &res_lib_cpg_trackstart, sizeof (struct res_lib_cpg_trackstart));
  344. if (error != CS_OK) {
  345. pthread_mutex_unlock (&cpg_inst->response_mutex);
  346. goto error_exit;
  347. }
  348. /* Now join */
  349. req_lib_cpg_join.header.size = sizeof (struct req_lib_cpg_join);
  350. req_lib_cpg_join.header.id = MESSAGE_REQ_CPG_JOIN;
  351. req_lib_cpg_join.pid = getpid();
  352. marshall_to_mar_cpg_name_t (&req_lib_cpg_join.group_name,
  353. group);
  354. iov[0].iov_base = &req_lib_cpg_join;
  355. iov[0].iov_len = sizeof (struct req_lib_cpg_join);
  356. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, iov, 1,
  357. &res_lib_cpg_join, sizeof (struct res_lib_cpg_join));
  358. pthread_mutex_unlock (&cpg_inst->response_mutex);
  359. if (error != CS_OK) {
  360. goto error_exit;
  361. }
  362. error = res_lib_cpg_join.header.error;
  363. error_exit:
  364. saHandleInstancePut (&cpg_handle_t_db, handle);
  365. return (error);
  366. }
  367. cs_error_t cpg_leave (
  368. cpg_handle_t handle,
  369. const struct cpg_name *group)
  370. {
  371. cs_error_t error;
  372. struct cpg_inst *cpg_inst;
  373. struct iovec iov[2];
  374. struct req_lib_cpg_leave req_lib_cpg_leave;
  375. struct res_lib_cpg_leave res_lib_cpg_leave;
  376. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  377. if (error != CS_OK) {
  378. return (error);
  379. }
  380. req_lib_cpg_leave.header.size = sizeof (struct req_lib_cpg_leave);
  381. req_lib_cpg_leave.header.id = MESSAGE_REQ_CPG_LEAVE;
  382. req_lib_cpg_leave.pid = getpid();
  383. marshall_to_mar_cpg_name_t (&req_lib_cpg_leave.group_name,
  384. group);
  385. iov[0].iov_base = &req_lib_cpg_leave;
  386. iov[0].iov_len = sizeof (struct req_lib_cpg_leave);
  387. pthread_mutex_lock (&cpg_inst->response_mutex);
  388. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, iov, 1,
  389. &res_lib_cpg_leave, sizeof (struct res_lib_cpg_leave));
  390. pthread_mutex_unlock (&cpg_inst->response_mutex);
  391. if (error != CS_OK) {
  392. goto error_exit;
  393. }
  394. error = res_lib_cpg_leave.header.error;
  395. error_exit:
  396. saHandleInstancePut (&cpg_handle_t_db, handle);
  397. return (error);
  398. }
  399. cs_error_t cpg_membership_get (
  400. cpg_handle_t handle,
  401. struct cpg_name *group_name,
  402. struct cpg_address *member_list,
  403. int *member_list_entries)
  404. {
  405. cs_error_t error;
  406. struct cpg_inst *cpg_inst;
  407. struct iovec iov;
  408. struct req_lib_cpg_membership req_lib_cpg_membership_get;
  409. struct res_lib_cpg_confchg_callback res_lib_cpg_membership_get;
  410. unsigned int i;
  411. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  412. if (error != CS_OK) {
  413. return (error);
  414. }
  415. req_lib_cpg_membership_get.header.size = sizeof (mar_req_header_t);
  416. req_lib_cpg_membership_get.header.id = MESSAGE_REQ_CPG_MEMBERSHIP;
  417. iov.iov_base = &req_lib_cpg_membership_get;
  418. iov.iov_len = sizeof (mar_req_header_t);
  419. pthread_mutex_lock (&cpg_inst->response_mutex);
  420. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, &iov, 1,
  421. &res_lib_cpg_membership_get, sizeof (mar_res_header_t));
  422. pthread_mutex_unlock (&cpg_inst->response_mutex);
  423. if (error != CS_OK) {
  424. goto error_exit;
  425. }
  426. error = res_lib_cpg_membership_get.header.error;
  427. /*
  428. * Copy results to caller
  429. */
  430. *member_list_entries = res_lib_cpg_membership_get.member_list_entries;
  431. if (member_list) {
  432. for (i = 0; i < res_lib_cpg_membership_get.member_list_entries; i++) {
  433. marshall_from_mar_cpg_address_t (&member_list[i],
  434. &res_lib_cpg_membership_get.member_list[i]);
  435. }
  436. }
  437. error_exit:
  438. saHandleInstancePut (&cpg_handle_t_db, handle);
  439. return (error);
  440. }
  441. cs_error_t cpg_local_get (
  442. cpg_handle_t handle,
  443. unsigned int *local_nodeid)
  444. {
  445. cs_error_t error;
  446. struct cpg_inst *cpg_inst;
  447. struct iovec iov;
  448. struct req_lib_cpg_local_get req_lib_cpg_local_get;
  449. struct res_lib_cpg_local_get res_lib_cpg_local_get;
  450. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  451. if (error != CS_OK) {
  452. return (error);
  453. }
  454. req_lib_cpg_local_get.header.size = sizeof (mar_req_header_t);
  455. req_lib_cpg_local_get.header.id = MESSAGE_REQ_CPG_LOCAL_GET;
  456. iov.iov_base = &req_lib_cpg_local_get;
  457. iov.iov_len = sizeof (struct req_lib_cpg_local_get);
  458. pthread_mutex_lock (&cpg_inst->response_mutex);
  459. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, &iov, 1,
  460. &res_lib_cpg_local_get, sizeof (res_lib_cpg_local_get));
  461. pthread_mutex_unlock (&cpg_inst->response_mutex);
  462. if (error != CS_OK) {
  463. goto error_exit;
  464. }
  465. error = res_lib_cpg_local_get.header.error;
  466. *local_nodeid = res_lib_cpg_local_get.local_nodeid;
  467. error_exit:
  468. saHandleInstancePut (&cpg_handle_t_db, handle);
  469. return (error);
  470. }
  471. cs_error_t cpg_flow_control_state_get (
  472. cpg_handle_t handle,
  473. cpg_flow_control_state_t *flow_control_state)
  474. {
  475. cs_error_t error;
  476. struct cpg_inst *cpg_inst;
  477. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  478. if (error != CS_OK) {
  479. return (error);
  480. }
  481. *flow_control_state = coroipcc_dispatch_flow_control_get (cpg_inst->ipc_ctx);
  482. saHandleInstancePut (&cpg_handle_t_db, handle);
  483. return (error);
  484. }
  485. cs_error_t cpg_zcb_alloc (
  486. cpg_handle_t handle,
  487. size_t size,
  488. void **buffer)
  489. {
  490. cs_error_t error;
  491. struct cpg_inst *cpg_inst;
  492. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  493. if (error != CS_OK) {
  494. return (error);
  495. }
  496. error = coroipcc_zcb_alloc (cpg_inst->ipc_ctx,
  497. buffer,
  498. size,
  499. sizeof (struct req_lib_cpg_mcast));
  500. saHandleInstancePut (&cpg_handle_t_db, handle);
  501. *buffer = ((char *)*buffer) + sizeof (struct req_lib_cpg_mcast);
  502. return (error);
  503. }
  504. cs_error_t cpg_zcb_free (
  505. cpg_handle_t handle,
  506. void *buffer)
  507. {
  508. cs_error_t error;
  509. struct cpg_inst *cpg_inst;
  510. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  511. if (error != CS_OK) {
  512. return (error);
  513. }
  514. coroipcc_zcb_free (cpg_inst->ipc_ctx, ((char *)buffer) - sizeof (struct req_lib_cpg_mcast));
  515. saHandleInstancePut (&cpg_handle_t_db, handle);
  516. return (error);
  517. }
  518. cs_error_t cpg_zcb_mcast_joined (
  519. cpg_handle_t handle,
  520. cpg_guarantee_t guarantee,
  521. void *msg,
  522. size_t msg_len)
  523. {
  524. cs_error_t error;
  525. struct cpg_inst *cpg_inst;
  526. struct req_lib_cpg_mcast *req_lib_cpg_mcast;
  527. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  528. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  529. if (error != CS_OK) {
  530. return (error);
  531. }
  532. req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)(((char *)msg) - sizeof (struct req_lib_cpg_mcast));
  533. req_lib_cpg_mcast->header.size = sizeof (struct req_lib_cpg_mcast) +
  534. msg_len;
  535. req_lib_cpg_mcast->header.id = MESSAGE_REQ_CPG_MCAST;
  536. req_lib_cpg_mcast->guarantee = guarantee;
  537. req_lib_cpg_mcast->msglen = msg_len;
  538. pthread_mutex_lock (&cpg_inst->response_mutex);
  539. error = coroipcc_zcb_msg_send_reply_receive (
  540. cpg_inst->ipc_ctx,
  541. req_lib_cpg_mcast,
  542. &res_lib_cpg_mcast,
  543. sizeof (res_lib_cpg_mcast));
  544. pthread_mutex_unlock (&cpg_inst->response_mutex);
  545. if (error != CS_OK) {
  546. goto error_exit;
  547. }
  548. error = res_lib_cpg_mcast.header.error;
  549. error_exit:
  550. saHandleInstancePut (&cpg_handle_t_db, handle);
  551. return (error);
  552. }
  553. cs_error_t cpg_mcast_joined (
  554. cpg_handle_t handle,
  555. cpg_guarantee_t guarantee,
  556. const struct iovec *iovec,
  557. unsigned int iov_len)
  558. {
  559. int i;
  560. cs_error_t error;
  561. struct cpg_inst *cpg_inst;
  562. struct iovec iov[64];
  563. struct req_lib_cpg_mcast req_lib_cpg_mcast;
  564. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  565. size_t msg_len = 0;
  566. error = saHandleInstanceGet (&cpg_handle_t_db, handle, (void *)&cpg_inst);
  567. if (error != CS_OK) {
  568. return (error);
  569. }
  570. for (i = 0; i < iov_len; i++ ) {
  571. msg_len += iovec[i].iov_len;
  572. }
  573. req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_mcast) +
  574. msg_len;
  575. req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_MCAST;
  576. req_lib_cpg_mcast.guarantee = guarantee;
  577. req_lib_cpg_mcast.msglen = msg_len;
  578. iov[0].iov_base = &req_lib_cpg_mcast;
  579. iov[0].iov_len = sizeof (struct req_lib_cpg_mcast);
  580. memcpy (&iov[1], iovec, iov_len * sizeof (struct iovec));
  581. pthread_mutex_lock (&cpg_inst->response_mutex);
  582. error = coroipcc_msg_send_reply_receive (cpg_inst->ipc_ctx, iov,
  583. iov_len + 1, &res_lib_cpg_mcast, sizeof (res_lib_cpg_mcast));
  584. pthread_mutex_unlock (&cpg_inst->response_mutex);
  585. if (error != CS_OK) {
  586. goto error_exit;
  587. }
  588. error = res_lib_cpg_mcast.header.error;
  589. error_exit:
  590. saHandleInstancePut (&cpg_handle_t_db, handle);
  591. return (error);
  592. }
  593. /** @} */