cpg.c 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931
  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: 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 <string.h>
  43. #include <unistd.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/list.h>
  53. #include <corosync/cpg.h>
  54. #include <corosync/ipc_cpg.h>
  55. #include "util.h"
  56. struct cpg_inst {
  57. hdb_handle_t handle;
  58. int finalize;
  59. cpg_callbacks_t callbacks;
  60. void *context;
  61. struct list_head iteration_list_head;
  62. };
  63. DECLARE_HDB_DATABASE(cpg_handle_t_db,NULL);
  64. struct cpg_iteration_instance_t {
  65. cpg_iteration_handle_t cpg_iteration_handle;
  66. hdb_handle_t conn_handle;
  67. hdb_handle_t executive_iteration_handle;
  68. struct list_head list;
  69. };
  70. DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db,NULL);
  71. /*
  72. * Internal (not visible by API) functions
  73. */
  74. static void cpg_iteration_instance_finalize (struct cpg_iteration_instance_t *cpg_iteration_instance)
  75. {
  76. list_del (&cpg_iteration_instance->list);
  77. hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_instance->cpg_iteration_handle);
  78. }
  79. static void cpg_inst_finalize (struct cpg_inst *cpg_inst, hdb_handle_t handle)
  80. {
  81. struct list_head *iter, *iter_next;
  82. struct cpg_iteration_instance_t *cpg_iteration_instance;
  83. /*
  84. * Traverse thru iteration instances and delete them
  85. */
  86. for (iter = cpg_inst->iteration_list_head.next; iter != &cpg_inst->iteration_list_head;iter = iter_next) {
  87. iter_next = iter->next;
  88. cpg_iteration_instance = list_entry (iter, struct cpg_iteration_instance_t, list);
  89. cpg_iteration_instance_finalize (cpg_iteration_instance);
  90. }
  91. hdb_handle_destroy (&cpg_handle_t_db, handle);
  92. }
  93. /**
  94. * @defgroup cpg_coroipcc The closed process group API
  95. * @ingroup coroipcc
  96. *
  97. * @{
  98. */
  99. cs_error_t cpg_initialize (
  100. cpg_handle_t *handle,
  101. cpg_callbacks_t *callbacks)
  102. {
  103. cs_error_t error;
  104. struct cpg_inst *cpg_inst;
  105. error = hdb_error_to_cs (hdb_handle_create (&cpg_handle_t_db, sizeof (struct cpg_inst), handle));
  106. if (error != CS_OK) {
  107. goto error_no_destroy;
  108. }
  109. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, *handle, (void *)&cpg_inst));
  110. if (error != CS_OK) {
  111. goto error_destroy;
  112. }
  113. error = coroipcc_service_connect (
  114. COROSYNC_SOCKET_NAME,
  115. CPG_SERVICE,
  116. IPC_REQUEST_SIZE,
  117. IPC_RESPONSE_SIZE,
  118. IPC_DISPATCH_SIZE,
  119. &cpg_inst->handle);
  120. if (error != CS_OK) {
  121. goto error_put_destroy;
  122. }
  123. if (callbacks) {
  124. memcpy (&cpg_inst->callbacks, callbacks, sizeof (cpg_callbacks_t));
  125. }
  126. list_init(&cpg_inst->iteration_list_head);
  127. hdb_handle_put (&cpg_handle_t_db, *handle);
  128. return (CS_OK);
  129. error_put_destroy:
  130. hdb_handle_put (&cpg_handle_t_db, *handle);
  131. error_destroy:
  132. hdb_handle_destroy (&cpg_handle_t_db, *handle);
  133. error_no_destroy:
  134. return (error);
  135. }
  136. cs_error_t cpg_finalize (
  137. cpg_handle_t handle)
  138. {
  139. struct cpg_inst *cpg_inst;
  140. struct iovec iov;
  141. struct req_lib_cpg_finalize req_lib_cpg_finalize;
  142. struct res_lib_cpg_finalize res_lib_cpg_finalize;
  143. cs_error_t error;
  144. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  145. if (error != CS_OK) {
  146. return (error);
  147. }
  148. /*
  149. * Another thread has already started finalizing
  150. */
  151. if (cpg_inst->finalize) {
  152. hdb_handle_put (&cpg_handle_t_db, handle);
  153. return (CPG_ERR_BAD_HANDLE);
  154. }
  155. cpg_inst->finalize = 1;
  156. /*
  157. * Send service request
  158. */
  159. req_lib_cpg_finalize.header.size = sizeof (struct req_lib_cpg_finalize);
  160. req_lib_cpg_finalize.header.id = MESSAGE_REQ_CPG_FINALIZE;
  161. iov.iov_base = (void *)&req_lib_cpg_finalize;
  162. iov.iov_len = sizeof (struct req_lib_cpg_finalize);
  163. error = coroipcc_msg_send_reply_receive (cpg_inst->handle,
  164. &iov,
  165. 1,
  166. &res_lib_cpg_finalize,
  167. sizeof (struct req_lib_cpg_finalize));
  168. if (error != CS_OK) {
  169. goto error_put;
  170. }
  171. coroipcc_service_disconnect (cpg_inst->handle);
  172. cpg_inst_finalize (cpg_inst, handle);
  173. hdb_handle_put (&cpg_handle_t_db, handle);
  174. return (CPG_OK);
  175. error_put:
  176. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  177. cpg_inst->finalize = 0;
  178. return (error);
  179. }
  180. cs_error_t cpg_fd_get (
  181. cpg_handle_t handle,
  182. int *fd)
  183. {
  184. cs_error_t error;
  185. struct cpg_inst *cpg_inst;
  186. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  187. if (error != CS_OK) {
  188. return (error);
  189. }
  190. error = coroipcc_fd_get (cpg_inst->handle, fd);
  191. hdb_handle_put (&cpg_handle_t_db, handle);
  192. return (error);
  193. }
  194. cs_error_t cpg_context_get (
  195. cpg_handle_t handle,
  196. void **context)
  197. {
  198. cs_error_t error;
  199. struct cpg_inst *cpg_inst;
  200. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  201. if (error != CS_OK) {
  202. return (error);
  203. }
  204. *context = cpg_inst->context;
  205. hdb_handle_put (&cpg_handle_t_db, handle);
  206. return (CS_OK);
  207. }
  208. cs_error_t cpg_context_set (
  209. cpg_handle_t handle,
  210. void *context)
  211. {
  212. cs_error_t error;
  213. struct cpg_inst *cpg_inst;
  214. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  215. if (error != CS_OK) {
  216. return (error);
  217. }
  218. cpg_inst->context = context;
  219. hdb_handle_put (&cpg_handle_t_db, handle);
  220. return (CS_OK);
  221. }
  222. cs_error_t cpg_dispatch (
  223. cpg_handle_t handle,
  224. cs_dispatch_flags_t dispatch_types)
  225. {
  226. int timeout = -1;
  227. cs_error_t error;
  228. int cont = 1; /* always continue do loop except when set to 0 */
  229. struct cpg_inst *cpg_inst;
  230. struct res_lib_cpg_confchg_callback *res_cpg_confchg_callback;
  231. struct res_lib_cpg_deliver_callback *res_cpg_deliver_callback;
  232. cpg_callbacks_t callbacks;
  233. coroipc_response_header_t *dispatch_data;
  234. struct cpg_address member_list[CPG_MEMBERS_MAX];
  235. struct cpg_address left_list[CPG_MEMBERS_MAX];
  236. struct cpg_address joined_list[CPG_MEMBERS_MAX];
  237. struct cpg_name group_name;
  238. mar_cpg_address_t *left_list_start;
  239. mar_cpg_address_t *joined_list_start;
  240. unsigned int i;
  241. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  242. if (error != CS_OK) {
  243. return (error);
  244. }
  245. /*
  246. * Timeout instantly for CS_DISPATCH_ONE or CS_DISPATCH_ALL and
  247. * wait indefinately for CS_DISPATCH_BLOCKING
  248. */
  249. if (dispatch_types == CPG_DISPATCH_ALL) {
  250. timeout = 0;
  251. }
  252. do {
  253. error = coroipcc_dispatch_get (
  254. cpg_inst->handle,
  255. (void **)&dispatch_data,
  256. timeout);
  257. if (error == CS_ERR_BAD_HANDLE) {
  258. error = CS_OK;
  259. goto error_put;
  260. }
  261. if (error == CS_ERR_TRY_AGAIN) {
  262. error = CS_OK;
  263. if (dispatch_types == CPG_DISPATCH_ALL) {
  264. break; /* exit do while cont is 1 loop */
  265. } else {
  266. continue; /* next poll */
  267. }
  268. }
  269. if (error != CS_OK) {
  270. goto error_put;
  271. }
  272. /*
  273. * Make copy of callbacks, message data, unlock instance, and call callback
  274. * A risk of this dispatch method is that the callback routines may
  275. * operate at the same time that cpgFinalize has been called.
  276. */
  277. memcpy (&callbacks, &cpg_inst->callbacks, sizeof (cpg_callbacks_t));
  278. /*
  279. * Dispatch incoming message
  280. */
  281. switch (dispatch_data->id) {
  282. case MESSAGE_RES_CPG_DELIVER_CALLBACK:
  283. if (callbacks.cpg_deliver_fn == NULL) {
  284. continue;
  285. }
  286. res_cpg_deliver_callback = (struct res_lib_cpg_deliver_callback *)dispatch_data;
  287. marshall_from_mar_cpg_name_t (
  288. &group_name,
  289. &res_cpg_deliver_callback->group_name);
  290. callbacks.cpg_deliver_fn (handle,
  291. &group_name,
  292. res_cpg_deliver_callback->nodeid,
  293. res_cpg_deliver_callback->pid,
  294. &res_cpg_deliver_callback->message,
  295. res_cpg_deliver_callback->msglen);
  296. break;
  297. case MESSAGE_RES_CPG_CONFCHG_CALLBACK:
  298. if (callbacks.cpg_confchg_fn == NULL) {
  299. continue;
  300. }
  301. res_cpg_confchg_callback = (struct res_lib_cpg_confchg_callback *)dispatch_data;
  302. for (i = 0; i < res_cpg_confchg_callback->member_list_entries; i++) {
  303. marshall_from_mar_cpg_address_t (&member_list[i],
  304. &res_cpg_confchg_callback->member_list[i]);
  305. }
  306. left_list_start = res_cpg_confchg_callback->member_list +
  307. res_cpg_confchg_callback->member_list_entries;
  308. for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) {
  309. marshall_from_mar_cpg_address_t (&left_list[i],
  310. &left_list_start[i]);
  311. }
  312. joined_list_start = res_cpg_confchg_callback->member_list +
  313. res_cpg_confchg_callback->member_list_entries +
  314. res_cpg_confchg_callback->left_list_entries;
  315. for (i = 0; i < res_cpg_confchg_callback->joined_list_entries; i++) {
  316. marshall_from_mar_cpg_address_t (&joined_list[i],
  317. &joined_list_start[i]);
  318. }
  319. marshall_from_mar_cpg_name_t (
  320. &group_name,
  321. &res_cpg_confchg_callback->group_name);
  322. callbacks.cpg_confchg_fn (handle,
  323. &group_name,
  324. member_list,
  325. res_cpg_confchg_callback->member_list_entries,
  326. left_list,
  327. res_cpg_confchg_callback->left_list_entries,
  328. joined_list,
  329. res_cpg_confchg_callback->joined_list_entries);
  330. break;
  331. default:
  332. coroipcc_dispatch_put (cpg_inst->handle);
  333. error = CS_ERR_LIBRARY;
  334. goto error_put;
  335. break;
  336. }
  337. coroipcc_dispatch_put (cpg_inst->handle);
  338. /*
  339. * Determine if more messages should be processed
  340. */
  341. if (dispatch_types == CS_DISPATCH_ONE) {
  342. cont = 0;
  343. }
  344. } while (cont);
  345. error_put:
  346. hdb_handle_put (&cpg_handle_t_db, handle);
  347. return (error);
  348. }
  349. cs_error_t cpg_join (
  350. cpg_handle_t handle,
  351. const struct cpg_name *group)
  352. {
  353. cs_error_t error;
  354. struct cpg_inst *cpg_inst;
  355. struct iovec iov[2];
  356. struct req_lib_cpg_join req_lib_cpg_join;
  357. struct res_lib_cpg_join res_lib_cpg_join;
  358. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  359. if (error != CS_OK) {
  360. return (error);
  361. }
  362. /* Now join */
  363. req_lib_cpg_join.header.size = sizeof (struct req_lib_cpg_join);
  364. req_lib_cpg_join.header.id = MESSAGE_REQ_CPG_JOIN;
  365. req_lib_cpg_join.pid = getpid();
  366. marshall_to_mar_cpg_name_t (&req_lib_cpg_join.group_name,
  367. group);
  368. iov[0].iov_base = (void *)&req_lib_cpg_join;
  369. iov[0].iov_len = sizeof (struct req_lib_cpg_join);
  370. do {
  371. error = coroipcc_msg_send_reply_receive (cpg_inst->handle, iov, 1,
  372. &res_lib_cpg_join, sizeof (struct res_lib_cpg_join));
  373. if (error != CS_OK) {
  374. goto error_exit;
  375. }
  376. } while (res_lib_cpg_join.header.error == CPG_ERR_BUSY);
  377. error = res_lib_cpg_join.header.error;
  378. error_exit:
  379. hdb_handle_put (&cpg_handle_t_db, handle);
  380. return (error);
  381. }
  382. cs_error_t cpg_leave (
  383. cpg_handle_t handle,
  384. const struct cpg_name *group)
  385. {
  386. cs_error_t error;
  387. struct cpg_inst *cpg_inst;
  388. struct iovec iov[2];
  389. struct req_lib_cpg_leave req_lib_cpg_leave;
  390. struct res_lib_cpg_leave res_lib_cpg_leave;
  391. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  392. if (error != CS_OK) {
  393. return (error);
  394. }
  395. req_lib_cpg_leave.header.size = sizeof (struct req_lib_cpg_leave);
  396. req_lib_cpg_leave.header.id = MESSAGE_REQ_CPG_LEAVE;
  397. req_lib_cpg_leave.pid = getpid();
  398. marshall_to_mar_cpg_name_t (&req_lib_cpg_leave.group_name,
  399. group);
  400. iov[0].iov_base = (void *)&req_lib_cpg_leave;
  401. iov[0].iov_len = sizeof (struct req_lib_cpg_leave);
  402. do {
  403. error = coroipcc_msg_send_reply_receive (cpg_inst->handle, iov, 1,
  404. &res_lib_cpg_leave, sizeof (struct res_lib_cpg_leave));
  405. if (error != CS_OK) {
  406. goto error_exit;
  407. }
  408. } while (res_lib_cpg_leave.header.error == CPG_ERR_BUSY);
  409. error = res_lib_cpg_leave.header.error;
  410. error_exit:
  411. hdb_handle_put (&cpg_handle_t_db, handle);
  412. return (error);
  413. }
  414. cs_error_t cpg_membership_get (
  415. cpg_handle_t handle,
  416. struct cpg_name *group_name,
  417. struct cpg_address *member_list,
  418. int *member_list_entries)
  419. {
  420. cs_error_t error;
  421. struct cpg_inst *cpg_inst;
  422. struct iovec iov;
  423. struct req_lib_cpg_membership req_lib_cpg_membership_get;
  424. struct res_lib_cpg_confchg_callback res_lib_cpg_membership_get;
  425. unsigned int i;
  426. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  427. if (error != CS_OK) {
  428. return (error);
  429. }
  430. req_lib_cpg_membership_get.header.size = sizeof (coroipc_request_header_t);
  431. req_lib_cpg_membership_get.header.id = MESSAGE_REQ_CPG_MEMBERSHIP;
  432. iov.iov_base = (void *)&req_lib_cpg_membership_get;
  433. iov.iov_len = sizeof (coroipc_request_header_t);
  434. do {
  435. error = coroipcc_msg_send_reply_receive (cpg_inst->handle, &iov, 1,
  436. &res_lib_cpg_membership_get, sizeof (coroipc_response_header_t));
  437. if (error != CS_OK) {
  438. goto error_exit;
  439. }
  440. } while (res_lib_cpg_membership_get.header.error == CPG_ERR_BUSY);
  441. error = res_lib_cpg_membership_get.header.error;
  442. /*
  443. * Copy results to caller
  444. */
  445. *member_list_entries = res_lib_cpg_membership_get.member_list_entries;
  446. if (member_list) {
  447. for (i = 0; i < res_lib_cpg_membership_get.member_list_entries; i++) {
  448. marshall_from_mar_cpg_address_t (&member_list[i],
  449. &res_lib_cpg_membership_get.member_list[i]);
  450. }
  451. }
  452. error_exit:
  453. hdb_handle_put (&cpg_handle_t_db, handle);
  454. return (error);
  455. }
  456. cs_error_t cpg_local_get (
  457. cpg_handle_t handle,
  458. unsigned int *local_nodeid)
  459. {
  460. cs_error_t error;
  461. struct cpg_inst *cpg_inst;
  462. struct iovec iov;
  463. struct req_lib_cpg_local_get req_lib_cpg_local_get;
  464. struct res_lib_cpg_local_get res_lib_cpg_local_get;
  465. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  466. if (error != CS_OK) {
  467. return (error);
  468. }
  469. req_lib_cpg_local_get.header.size = sizeof (coroipc_request_header_t);
  470. req_lib_cpg_local_get.header.id = MESSAGE_REQ_CPG_LOCAL_GET;
  471. iov.iov_base = (void *)&req_lib_cpg_local_get;
  472. iov.iov_len = sizeof (struct req_lib_cpg_local_get);
  473. error = coroipcc_msg_send_reply_receive (cpg_inst->handle, &iov, 1,
  474. &res_lib_cpg_local_get, sizeof (res_lib_cpg_local_get));
  475. if (error != CS_OK) {
  476. goto error_exit;
  477. }
  478. error = res_lib_cpg_local_get.header.error;
  479. *local_nodeid = res_lib_cpg_local_get.local_nodeid;
  480. error_exit:
  481. hdb_handle_put (&cpg_handle_t_db, handle);
  482. return (error);
  483. }
  484. cs_error_t cpg_flow_control_state_get (
  485. cpg_handle_t handle,
  486. cpg_flow_control_state_t *flow_control_state)
  487. {
  488. cs_error_t error;
  489. struct cpg_inst *cpg_inst;
  490. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  491. if (error != CS_OK) {
  492. return (error);
  493. }
  494. error = coroipcc_dispatch_flow_control_get (cpg_inst->handle, (unsigned int *)flow_control_state);
  495. hdb_handle_put (&cpg_handle_t_db, handle);
  496. return (error);
  497. }
  498. cs_error_t cpg_zcb_alloc (
  499. cpg_handle_t handle,
  500. size_t size,
  501. void **buffer)
  502. {
  503. cs_error_t error;
  504. struct cpg_inst *cpg_inst;
  505. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  506. if (error != CS_OK) {
  507. return (error);
  508. }
  509. error = coroipcc_zcb_alloc (cpg_inst->handle,
  510. buffer,
  511. size,
  512. sizeof (struct req_lib_cpg_mcast));
  513. hdb_handle_put (&cpg_handle_t_db, handle);
  514. *buffer = ((char *)*buffer) + sizeof (struct req_lib_cpg_mcast);
  515. return (error);
  516. }
  517. cs_error_t cpg_zcb_free (
  518. cpg_handle_t handle,
  519. void *buffer)
  520. {
  521. cs_error_t error;
  522. struct cpg_inst *cpg_inst;
  523. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  524. if (error != CS_OK) {
  525. return (error);
  526. }
  527. coroipcc_zcb_free (cpg_inst->handle, ((char *)buffer) - sizeof (struct req_lib_cpg_mcast));
  528. hdb_handle_put (&cpg_handle_t_db, handle);
  529. return (error);
  530. }
  531. cs_error_t cpg_zcb_mcast_joined (
  532. cpg_handle_t handle,
  533. cpg_guarantee_t guarantee,
  534. void *msg,
  535. size_t msg_len)
  536. {
  537. cs_error_t error;
  538. struct cpg_inst *cpg_inst;
  539. struct req_lib_cpg_mcast *req_lib_cpg_mcast;
  540. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  541. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  542. if (error != CS_OK) {
  543. return (error);
  544. }
  545. req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)(((char *)msg) - sizeof (struct req_lib_cpg_mcast));
  546. req_lib_cpg_mcast->header.size = sizeof (struct req_lib_cpg_mcast) +
  547. msg_len;
  548. req_lib_cpg_mcast->header.id = MESSAGE_REQ_CPG_MCAST;
  549. req_lib_cpg_mcast->guarantee = guarantee;
  550. req_lib_cpg_mcast->msglen = msg_len;
  551. error = coroipcc_zcb_msg_send_reply_receive (
  552. cpg_inst->handle,
  553. req_lib_cpg_mcast,
  554. &res_lib_cpg_mcast,
  555. sizeof (res_lib_cpg_mcast));
  556. if (error != CS_OK) {
  557. goto error_exit;
  558. }
  559. error = res_lib_cpg_mcast.header.error;
  560. error_exit:
  561. hdb_handle_put (&cpg_handle_t_db, handle);
  562. return (error);
  563. }
  564. cs_error_t cpg_mcast_joined (
  565. cpg_handle_t handle,
  566. cpg_guarantee_t guarantee,
  567. const struct iovec *iovec,
  568. unsigned int iov_len)
  569. {
  570. int i;
  571. cs_error_t error;
  572. struct cpg_inst *cpg_inst;
  573. struct iovec iov[64];
  574. struct req_lib_cpg_mcast req_lib_cpg_mcast;
  575. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  576. size_t msg_len = 0;
  577. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  578. if (error != CS_OK) {
  579. return (error);
  580. }
  581. for (i = 0; i < iov_len; i++ ) {
  582. msg_len += iovec[i].iov_len;
  583. }
  584. req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_mcast) +
  585. msg_len;
  586. req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_MCAST;
  587. req_lib_cpg_mcast.guarantee = guarantee;
  588. req_lib_cpg_mcast.msglen = msg_len;
  589. iov[0].iov_base = (void *)&req_lib_cpg_mcast;
  590. iov[0].iov_len = sizeof (struct req_lib_cpg_mcast);
  591. memcpy (&iov[1], iovec, iov_len * sizeof (struct iovec));
  592. error = coroipcc_msg_send_reply_receive (cpg_inst->handle, iov,
  593. iov_len + 1, &res_lib_cpg_mcast, sizeof (res_lib_cpg_mcast));
  594. if (error != CS_OK) {
  595. goto error_exit;
  596. }
  597. error = res_lib_cpg_mcast.header.error;
  598. error_exit:
  599. hdb_handle_put (&cpg_handle_t_db, handle);
  600. return (error);
  601. }
  602. cs_error_t cpg_iteration_initialize(
  603. cpg_handle_t handle,
  604. cpg_iteration_type_t iteration_type,
  605. const struct cpg_name *group,
  606. cpg_iteration_handle_t *cpg_iteration_handle)
  607. {
  608. cs_error_t error;
  609. struct iovec iov;
  610. struct cpg_inst *cpg_inst;
  611. struct cpg_iteration_instance_t *cpg_iteration_instance;
  612. struct req_lib_cpg_iterationinitialize req_lib_cpg_iterationinitialize;
  613. struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize;
  614. if (cpg_iteration_handle == NULL) {
  615. return (CS_ERR_INVALID_PARAM);
  616. }
  617. if ((iteration_type == CPG_ITERATION_ONE_GROUP && group == NULL) ||
  618. (iteration_type != CPG_ITERATION_ONE_GROUP && group != NULL)) {
  619. return (CS_ERR_INVALID_PARAM);
  620. }
  621. if (iteration_type != CPG_ITERATION_NAME_ONLY && iteration_type != CPG_ITERATION_ONE_GROUP &&
  622. iteration_type != CPG_ITERATION_ALL) {
  623. return (CS_ERR_INVALID_PARAM);
  624. }
  625. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  626. if (error != CS_OK) {
  627. return (error);
  628. }
  629. error = hdb_error_to_cs (hdb_handle_create (&cpg_iteration_handle_t_db,
  630. sizeof (struct cpg_iteration_instance_t), cpg_iteration_handle));
  631. if (error != CS_OK) {
  632. goto error_put_cpg_db;
  633. }
  634. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, *cpg_iteration_handle,
  635. (void *)&cpg_iteration_instance));
  636. if (error != CS_OK) {
  637. goto error_destroy;
  638. }
  639. cpg_iteration_instance->conn_handle = cpg_inst->handle;
  640. list_init (&cpg_iteration_instance->list);
  641. req_lib_cpg_iterationinitialize.header.size = sizeof (struct req_lib_cpg_iterationinitialize);
  642. req_lib_cpg_iterationinitialize.header.id = MESSAGE_REQ_CPG_ITERATIONINITIALIZE;
  643. req_lib_cpg_iterationinitialize.iteration_type = iteration_type;
  644. if (group) {
  645. marshall_to_mar_cpg_name_t (&req_lib_cpg_iterationinitialize.group_name, group);
  646. }
  647. iov.iov_base = (void *)&req_lib_cpg_iterationinitialize;
  648. iov.iov_len = sizeof (struct req_lib_cpg_iterationinitialize);
  649. error = coroipcc_msg_send_reply_receive (cpg_inst->handle,
  650. &iov,
  651. 1,
  652. &res_lib_cpg_iterationinitialize,
  653. sizeof (struct res_lib_cpg_iterationinitialize));
  654. if (error != CS_OK) {
  655. goto error_put_destroy;
  656. }
  657. cpg_iteration_instance->executive_iteration_handle =
  658. res_lib_cpg_iterationinitialize.iteration_handle;
  659. cpg_iteration_instance->cpg_iteration_handle = *cpg_iteration_handle;
  660. list_add (&cpg_iteration_instance->list, &cpg_inst->iteration_list_head);
  661. hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  662. hdb_handle_put (&cpg_handle_t_db, handle);
  663. return (res_lib_cpg_iterationinitialize.header.error);
  664. error_put_destroy:
  665. hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  666. error_destroy:
  667. hdb_handle_destroy (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  668. error_put_cpg_db:
  669. hdb_handle_put (&cpg_handle_t_db, handle);
  670. return (error);
  671. }
  672. cs_error_t cpg_iteration_next(
  673. cpg_iteration_handle_t handle,
  674. struct cpg_iteration_description_t *description)
  675. {
  676. cs_error_t error;
  677. struct cpg_iteration_instance_t *cpg_iteration_instance;
  678. struct req_lib_cpg_iterationnext req_lib_cpg_iterationnext;
  679. struct res_lib_cpg_iterationnext *res_lib_cpg_iterationnext;
  680. struct iovec iov;
  681. void *return_address;
  682. if (description == NULL) {
  683. return CS_ERR_INVALID_PARAM;
  684. }
  685. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle,
  686. (void *)&cpg_iteration_instance));
  687. if (error != CS_OK) {
  688. goto error_exit;
  689. }
  690. req_lib_cpg_iterationnext.header.size = sizeof (struct req_lib_cpg_iterationnext);
  691. req_lib_cpg_iterationnext.header.id = MESSAGE_REQ_CPG_ITERATIONNEXT;
  692. req_lib_cpg_iterationnext.iteration_handle = cpg_iteration_instance->executive_iteration_handle;
  693. iov.iov_base = (void *)&req_lib_cpg_iterationnext;
  694. iov.iov_len = sizeof (struct req_lib_cpg_iterationnext);
  695. error = coroipcc_msg_send_reply_receive_in_buf_get (cpg_iteration_instance->conn_handle,
  696. &iov,
  697. 1,
  698. &return_address);
  699. res_lib_cpg_iterationnext = return_address;
  700. if (error != CS_OK) {
  701. goto error_put;
  702. }
  703. marshall_from_mar_cpg_iteration_description_t(
  704. description,
  705. &res_lib_cpg_iterationnext->description);
  706. error = res_lib_cpg_iterationnext->header.error;
  707. coroipcc_msg_send_reply_receive_in_buf_put(
  708. cpg_iteration_instance->conn_handle);
  709. error_put:
  710. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  711. error_exit:
  712. return (error);
  713. }
  714. cs_error_t cpg_iteration_finalize (
  715. cpg_iteration_handle_t handle)
  716. {
  717. cs_error_t error;
  718. struct iovec iov;
  719. struct cpg_iteration_instance_t *cpg_iteration_instance;
  720. struct req_lib_cpg_iterationfinalize req_lib_cpg_iterationfinalize;
  721. struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize;
  722. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle,
  723. (void *)&cpg_iteration_instance));
  724. if (error != CS_OK) {
  725. goto error_exit;
  726. }
  727. req_lib_cpg_iterationfinalize.header.size = sizeof (struct req_lib_cpg_iterationfinalize);
  728. req_lib_cpg_iterationfinalize.header.id = MESSAGE_REQ_CPG_ITERATIONFINALIZE;
  729. req_lib_cpg_iterationfinalize.iteration_handle = cpg_iteration_instance->executive_iteration_handle;
  730. iov.iov_base = (void *)&req_lib_cpg_iterationfinalize;
  731. iov.iov_len = sizeof (struct req_lib_cpg_iterationfinalize);
  732. error = coroipcc_msg_send_reply_receive (cpg_iteration_instance->conn_handle,
  733. &iov,
  734. 1,
  735. &res_lib_cpg_iterationfinalize,
  736. sizeof (struct req_lib_cpg_iterationfinalize));
  737. if (error != CS_OK) {
  738. goto error_put;
  739. }
  740. cpg_iteration_instance_finalize (cpg_iteration_instance);
  741. hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->cpg_iteration_handle);
  742. return (res_lib_cpg_iterationfinalize.header.error);
  743. error_put:
  744. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  745. error_exit:
  746. return (error);
  747. }
  748. /** @} */