cpg.c 18 KB

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