cpg.c 38 KB

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