4
0

cpg.c 39 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438
  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/qblist.h>
  53. #include <qb/qbdefs.h>
  54. #include <qb/qbipcc.h>
  55. #include <qb/qblog.h>
  56. #include <corosync/hdb.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 qb_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 qb_list_head iteration_list_head;
  91. uint32_t max_msg_size;
  92. struct qb_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 qb_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. qb_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 qb_list_head *iter, *tmp_iter;
  130. struct cpg_iteration_instance_t *cpg_iteration_instance;
  131. /*
  132. * Traverse thru iteration instances and delete them
  133. */
  134. qb_list_for_each_safe(iter, tmp_iter, &(cpg_inst->iteration_list_head)) {
  135. cpg_iteration_instance = qb_list_entry (iter, struct cpg_iteration_instance_t, list);
  136. cpg_iteration_instance_finalize (cpg_iteration_instance);
  137. }
  138. hdb_handle_destroy (&cpg_handle_t_db, handle);
  139. }
  140. /**
  141. * @defgroup cpg_coroipcc The closed process group API
  142. * @ingroup coroipcc
  143. *
  144. * @{
  145. */
  146. cs_error_t cpg_initialize (
  147. cpg_handle_t *handle,
  148. cpg_callbacks_t *callbacks)
  149. {
  150. cpg_model_v1_data_t model_v1_data;
  151. memset (&model_v1_data, 0, sizeof (cpg_model_v1_data_t));
  152. if (callbacks) {
  153. model_v1_data.cpg_deliver_fn = callbacks->cpg_deliver_fn;
  154. model_v1_data.cpg_confchg_fn = callbacks->cpg_confchg_fn;
  155. }
  156. return (cpg_model_initialize (handle, CPG_MODEL_V1, (cpg_model_data_t *)&model_v1_data, NULL));
  157. }
  158. cs_error_t cpg_model_initialize (
  159. cpg_handle_t *handle,
  160. cpg_model_t model,
  161. cpg_model_data_t *model_data,
  162. void *context)
  163. {
  164. cs_error_t error;
  165. struct cpg_inst *cpg_inst;
  166. if (model != CPG_MODEL_V1) {
  167. error = CS_ERR_INVALID_PARAM;
  168. goto error_no_destroy;
  169. }
  170. error = hdb_error_to_cs (hdb_handle_create (&cpg_handle_t_db, sizeof (struct cpg_inst), handle));
  171. if (error != CS_OK) {
  172. goto error_no_destroy;
  173. }
  174. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, *handle, (void *)&cpg_inst));
  175. if (error != CS_OK) {
  176. goto error_destroy;
  177. }
  178. cpg_inst->c = qb_ipcc_connect ("cpg", IPC_REQUEST_SIZE);
  179. if (cpg_inst->c == NULL) {
  180. error = qb_to_cs_error(-errno);
  181. goto error_put_destroy;
  182. }
  183. if (model_data != NULL) {
  184. switch (model) {
  185. case CPG_MODEL_V1:
  186. memcpy (&cpg_inst->model_v1_data, model_data, sizeof (cpg_model_v1_data_t));
  187. if ((cpg_inst->model_v1_data.flags & ~(CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF)) != 0) {
  188. error = CS_ERR_INVALID_PARAM;
  189. goto error_destroy;
  190. }
  191. break;
  192. }
  193. }
  194. /* Allow space for corosync internal headers */
  195. cpg_inst->max_msg_size = IPC_REQUEST_SIZE - 1024;
  196. cpg_inst->model_data.model = model;
  197. cpg_inst->context = context;
  198. qb_list_init(&cpg_inst->iteration_list_head);
  199. qb_list_init(&cpg_inst->assembly_list_head);
  200. hdb_handle_put (&cpg_handle_t_db, *handle);
  201. return (CS_OK);
  202. error_put_destroy:
  203. hdb_handle_put (&cpg_handle_t_db, *handle);
  204. error_destroy:
  205. hdb_handle_destroy (&cpg_handle_t_db, *handle);
  206. error_no_destroy:
  207. return (error);
  208. }
  209. cs_error_t cpg_finalize (
  210. cpg_handle_t handle)
  211. {
  212. struct cpg_inst *cpg_inst;
  213. struct iovec iov;
  214. struct req_lib_cpg_finalize req_lib_cpg_finalize;
  215. struct res_lib_cpg_finalize res_lib_cpg_finalize;
  216. cs_error_t error;
  217. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  218. if (error != CS_OK) {
  219. return (error);
  220. }
  221. /*
  222. * Another thread has already started finalizing
  223. */
  224. if (cpg_inst->finalize) {
  225. hdb_handle_put (&cpg_handle_t_db, handle);
  226. return (CS_ERR_BAD_HANDLE);
  227. }
  228. cpg_inst->finalize = 1;
  229. /*
  230. * Send service request
  231. */
  232. req_lib_cpg_finalize.header.size = sizeof (struct req_lib_cpg_finalize);
  233. req_lib_cpg_finalize.header.id = MESSAGE_REQ_CPG_FINALIZE;
  234. iov.iov_base = (void *)&req_lib_cpg_finalize;
  235. iov.iov_len = sizeof (struct req_lib_cpg_finalize);
  236. error = coroipcc_msg_send_reply_receive (cpg_inst->c,
  237. &iov,
  238. 1,
  239. &res_lib_cpg_finalize,
  240. sizeof (struct res_lib_cpg_finalize));
  241. cpg_inst_finalize (cpg_inst, handle);
  242. hdb_handle_put (&cpg_handle_t_db, handle);
  243. return (error);
  244. }
  245. cs_error_t cpg_fd_get (
  246. cpg_handle_t handle,
  247. int *fd)
  248. {
  249. cs_error_t error;
  250. struct cpg_inst *cpg_inst;
  251. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  252. if (error != CS_OK) {
  253. return (error);
  254. }
  255. error = qb_to_cs_error (qb_ipcc_fd_get (cpg_inst->c, fd));
  256. hdb_handle_put (&cpg_handle_t_db, handle);
  257. return (error);
  258. }
  259. cs_error_t cpg_max_atomic_msgsize_get (
  260. cpg_handle_t handle,
  261. uint32_t *size)
  262. {
  263. cs_error_t error;
  264. struct cpg_inst *cpg_inst;
  265. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  266. if (error != CS_OK) {
  267. return (error);
  268. }
  269. *size = cpg_inst->max_msg_size;
  270. hdb_handle_put (&cpg_handle_t_db, handle);
  271. return (error);
  272. }
  273. cs_error_t cpg_context_get (
  274. cpg_handle_t handle,
  275. void **context)
  276. {
  277. cs_error_t error;
  278. struct cpg_inst *cpg_inst;
  279. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  280. if (error != CS_OK) {
  281. return (error);
  282. }
  283. *context = cpg_inst->context;
  284. hdb_handle_put (&cpg_handle_t_db, handle);
  285. return (CS_OK);
  286. }
  287. cs_error_t cpg_context_set (
  288. cpg_handle_t handle,
  289. void *context)
  290. {
  291. cs_error_t error;
  292. struct cpg_inst *cpg_inst;
  293. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  294. if (error != CS_OK) {
  295. return (error);
  296. }
  297. cpg_inst->context = context;
  298. hdb_handle_put (&cpg_handle_t_db, handle);
  299. return (CS_OK);
  300. }
  301. cs_error_t cpg_dispatch (
  302. cpg_handle_t handle,
  303. cs_dispatch_flags_t dispatch_types)
  304. {
  305. int timeout = -1;
  306. cs_error_t error;
  307. int cont = 1; /* always continue do loop except when set to 0 */
  308. struct cpg_inst *cpg_inst;
  309. struct res_lib_cpg_confchg_callback *res_cpg_confchg_callback;
  310. struct res_lib_cpg_deliver_callback *res_cpg_deliver_callback;
  311. struct res_lib_cpg_partial_deliver_callback *res_cpg_partial_deliver_callback;
  312. struct res_lib_cpg_totem_confchg_callback *res_cpg_totem_confchg_callback;
  313. struct cpg_inst cpg_inst_copy;
  314. struct qb_ipc_response_header *dispatch_data;
  315. struct cpg_address member_list[CPG_MEMBERS_MAX];
  316. struct cpg_address left_list[CPG_MEMBERS_MAX];
  317. struct cpg_address joined_list[CPG_MEMBERS_MAX];
  318. struct cpg_name group_name;
  319. struct cpg_assembly_data *assembly_data;
  320. struct qb_list_head *iter, *tmp_iter;
  321. mar_cpg_address_t *left_list_start;
  322. mar_cpg_address_t *joined_list_start;
  323. unsigned int i;
  324. struct cpg_ring_id ring_id;
  325. uint32_t totem_member_list[CPG_MEMBERS_MAX];
  326. int32_t errno_res;
  327. char dispatch_buf[IPC_DISPATCH_SIZE];
  328. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  329. if (error != CS_OK) {
  330. return (error);
  331. }
  332. /*
  333. * Timeout instantly for CS_DISPATCH_ONE_NONBLOCKING or CS_DISPATCH_ALL and
  334. * wait indefinitely for CS_DISPATCH_ONE or CS_DISPATCH_BLOCKING
  335. */
  336. if (dispatch_types == CS_DISPATCH_ALL || dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) {
  337. timeout = 0;
  338. }
  339. dispatch_data = (struct qb_ipc_response_header *)dispatch_buf;
  340. do {
  341. errno_res = qb_ipcc_event_recv (
  342. cpg_inst->c,
  343. dispatch_buf,
  344. IPC_DISPATCH_SIZE,
  345. timeout);
  346. error = qb_to_cs_error (errno_res);
  347. if (error == CS_ERR_BAD_HANDLE) {
  348. error = CS_OK;
  349. goto error_put;
  350. }
  351. if (error == CS_ERR_TRY_AGAIN) {
  352. if (dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) {
  353. /*
  354. * Don't mask error
  355. */
  356. goto error_put;
  357. }
  358. error = CS_OK;
  359. if (dispatch_types == CS_DISPATCH_ALL) {
  360. break; /* exit do while cont is 1 loop */
  361. } else {
  362. continue; /* next poll */
  363. }
  364. }
  365. if (error != CS_OK) {
  366. goto error_put;
  367. }
  368. /*
  369. * Make copy of callbacks, message data, unlock instance, and call callback
  370. * A risk of this dispatch method is that the callback routines may
  371. * operate at the same time that cpgFinalize has been called.
  372. */
  373. memcpy (&cpg_inst_copy, cpg_inst, sizeof (struct cpg_inst));
  374. switch (cpg_inst_copy.model_data.model) {
  375. case CPG_MODEL_V1:
  376. /*
  377. * Dispatch incoming message
  378. */
  379. switch (dispatch_data->id) {
  380. case MESSAGE_RES_CPG_DELIVER_CALLBACK:
  381. if (cpg_inst_copy.model_v1_data.cpg_deliver_fn == NULL) {
  382. break;
  383. }
  384. res_cpg_deliver_callback = (struct res_lib_cpg_deliver_callback *)dispatch_data;
  385. marshall_from_mar_cpg_name_t (
  386. &group_name,
  387. &res_cpg_deliver_callback->group_name);
  388. cpg_inst_copy.model_v1_data.cpg_deliver_fn (handle,
  389. &group_name,
  390. res_cpg_deliver_callback->nodeid,
  391. res_cpg_deliver_callback->pid,
  392. &res_cpg_deliver_callback->message,
  393. res_cpg_deliver_callback->msglen);
  394. break;
  395. case MESSAGE_RES_CPG_PARTIAL_DELIVER_CALLBACK:
  396. res_cpg_partial_deliver_callback = (struct res_lib_cpg_partial_deliver_callback *)dispatch_data;
  397. marshall_from_mar_cpg_name_t (
  398. &group_name,
  399. &res_cpg_partial_deliver_callback->group_name);
  400. /*
  401. * Search for assembly data for current messages (nodeid, pid) pair in list of assemblies
  402. */
  403. assembly_data = NULL;
  404. qb_list_for_each(iter, &(cpg_inst->assembly_list_head)) {
  405. struct cpg_assembly_data *current_assembly_data = qb_list_entry (iter, struct cpg_assembly_data, list);
  406. if (current_assembly_data->nodeid == res_cpg_partial_deliver_callback->nodeid && current_assembly_data->pid == res_cpg_partial_deliver_callback->pid) {
  407. assembly_data = current_assembly_data;
  408. break;
  409. }
  410. }
  411. if (res_cpg_partial_deliver_callback->type == LIBCPG_PARTIAL_FIRST) {
  412. /*
  413. * As this is LIBCPG_PARTIAL_FIRST packet, check that there is no ongoing assembly.
  414. * Otherwise the sending of packet must have been interrupted and error should have
  415. * been reported to sending client. Therefore here last assembly will be dropped.
  416. */
  417. if (assembly_data) {
  418. qb_list_del (&assembly_data->list);
  419. free(assembly_data->assembly_buf);
  420. free(assembly_data);
  421. // coverity[UNUSED_VALUE:SUPPRESS] defensive programming
  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. qb_list_init (&assembly_data->list);
  439. qb_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. if (cpg_inst_copy.model_v1_data.cpg_deliver_fn != NULL) {
  447. cpg_inst_copy.model_v1_data.cpg_deliver_fn (handle,
  448. &group_name,
  449. res_cpg_partial_deliver_callback->nodeid,
  450. res_cpg_partial_deliver_callback->pid,
  451. assembly_data->assembly_buf,
  452. res_cpg_partial_deliver_callback->msglen);
  453. }
  454. qb_list_del (&assembly_data->list);
  455. free(assembly_data->assembly_buf);
  456. free(assembly_data);
  457. }
  458. }
  459. break;
  460. case MESSAGE_RES_CPG_CONFCHG_CALLBACK:
  461. if (cpg_inst_copy.model_v1_data.cpg_confchg_fn == NULL) {
  462. break;
  463. }
  464. res_cpg_confchg_callback = (struct res_lib_cpg_confchg_callback *)dispatch_data;
  465. for (i = 0; i < res_cpg_confchg_callback->member_list_entries; i++) {
  466. marshall_from_mar_cpg_address_t (&member_list[i],
  467. &res_cpg_confchg_callback->member_list[i]);
  468. }
  469. left_list_start = res_cpg_confchg_callback->member_list +
  470. res_cpg_confchg_callback->member_list_entries;
  471. for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) {
  472. marshall_from_mar_cpg_address_t (&left_list[i],
  473. &left_list_start[i]);
  474. }
  475. joined_list_start = res_cpg_confchg_callback->member_list +
  476. res_cpg_confchg_callback->member_list_entries +
  477. res_cpg_confchg_callback->left_list_entries;
  478. for (i = 0; i < res_cpg_confchg_callback->joined_list_entries; i++) {
  479. marshall_from_mar_cpg_address_t (&joined_list[i],
  480. &joined_list_start[i]);
  481. }
  482. marshall_from_mar_cpg_name_t (
  483. &group_name,
  484. &res_cpg_confchg_callback->group_name);
  485. cpg_inst_copy.model_v1_data.cpg_confchg_fn (handle,
  486. &group_name,
  487. member_list,
  488. res_cpg_confchg_callback->member_list_entries,
  489. left_list,
  490. res_cpg_confchg_callback->left_list_entries,
  491. joined_list,
  492. res_cpg_confchg_callback->joined_list_entries);
  493. /*
  494. * If member left while his partial packet was being assembled, assembly data must be removed from list
  495. */
  496. for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) {
  497. qb_list_for_each_safe(iter, tmp_iter, &(cpg_inst->assembly_list_head)) {
  498. struct cpg_assembly_data *current_assembly_data = qb_list_entry (iter, struct cpg_assembly_data, list);
  499. if (current_assembly_data->nodeid != left_list[i].nodeid || current_assembly_data->pid != left_list[i].pid)
  500. continue;
  501. qb_list_del (&current_assembly_data->list);
  502. free(current_assembly_data->assembly_buf);
  503. free(current_assembly_data);
  504. }
  505. }
  506. break;
  507. case MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK:
  508. if (cpg_inst_copy.model_v1_data.cpg_totem_confchg_fn == NULL) {
  509. break;
  510. }
  511. res_cpg_totem_confchg_callback = (struct res_lib_cpg_totem_confchg_callback *)dispatch_data;
  512. marshall_from_mar_cpg_ring_id_t (&ring_id, &res_cpg_totem_confchg_callback->ring_id);
  513. for (i = 0; i < res_cpg_totem_confchg_callback->member_list_entries; i++) {
  514. totem_member_list[i] = res_cpg_totem_confchg_callback->member_list[i];
  515. }
  516. cpg_inst_copy.model_v1_data.cpg_totem_confchg_fn (handle,
  517. ring_id,
  518. res_cpg_totem_confchg_callback->member_list_entries,
  519. totem_member_list);
  520. break;
  521. default:
  522. error = CS_ERR_LIBRARY;
  523. goto error_put;
  524. break;
  525. } /* - switch (dispatch_data->id) */
  526. break; /* case CPG_MODEL_V1 */
  527. } /* - switch (cpg_inst_copy.model_data.model) */
  528. if (cpg_inst_copy.finalize || cpg_inst->finalize) {
  529. /*
  530. * If the finalize has been called then get out of the dispatch.
  531. */
  532. cpg_inst->finalize = 1;
  533. error = CS_ERR_BAD_HANDLE;
  534. goto error_put;
  535. }
  536. /*
  537. * Determine if more messages should be processed
  538. */
  539. if (dispatch_types == CS_DISPATCH_ONE || dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) {
  540. cont = 0;
  541. }
  542. } while (cont);
  543. error_put:
  544. hdb_handle_put (&cpg_handle_t_db, handle);
  545. return (error);
  546. }
  547. cs_error_t cpg_join (
  548. cpg_handle_t handle,
  549. const struct cpg_name *group)
  550. {
  551. cs_error_t error;
  552. struct cpg_inst *cpg_inst;
  553. struct iovec iov[2];
  554. struct req_lib_cpg_join req_lib_cpg_join;
  555. struct res_lib_cpg_join response;
  556. if (group->length > CPG_MAX_NAME_LENGTH) {
  557. return (CS_ERR_NAME_TOO_LONG);
  558. }
  559. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  560. if (error != CS_OK) {
  561. return (error);
  562. }
  563. /* Now join */
  564. req_lib_cpg_join.header.size = sizeof (struct req_lib_cpg_join);
  565. req_lib_cpg_join.header.id = MESSAGE_REQ_CPG_JOIN;
  566. req_lib_cpg_join.pid = getpid();
  567. req_lib_cpg_join.flags = 0;
  568. switch (cpg_inst->model_data.model) {
  569. case CPG_MODEL_V1:
  570. req_lib_cpg_join.flags = cpg_inst->model_v1_data.flags;
  571. break;
  572. }
  573. marshall_to_mar_cpg_name_t (&req_lib_cpg_join.group_name,
  574. group);
  575. iov[0].iov_base = (void *)&req_lib_cpg_join;
  576. iov[0].iov_len = sizeof (struct req_lib_cpg_join);
  577. do {
  578. error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 1,
  579. &response, sizeof (struct res_lib_cpg_join));
  580. if (error != CS_OK) {
  581. goto error_exit;
  582. }
  583. } while (response.header.error == CS_ERR_BUSY);
  584. error = response.header.error;
  585. error_exit:
  586. hdb_handle_put (&cpg_handle_t_db, handle);
  587. return (error);
  588. }
  589. cs_error_t cpg_leave (
  590. cpg_handle_t handle,
  591. const struct cpg_name *group)
  592. {
  593. cs_error_t error;
  594. struct cpg_inst *cpg_inst;
  595. struct iovec iov[2];
  596. struct req_lib_cpg_leave req_lib_cpg_leave;
  597. struct res_lib_cpg_leave res_lib_cpg_leave;
  598. if (group->length > CPG_MAX_NAME_LENGTH) {
  599. return (CS_ERR_NAME_TOO_LONG);
  600. }
  601. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  602. if (error != CS_OK) {
  603. return (error);
  604. }
  605. req_lib_cpg_leave.header.size = sizeof (struct req_lib_cpg_leave);
  606. req_lib_cpg_leave.header.id = MESSAGE_REQ_CPG_LEAVE;
  607. req_lib_cpg_leave.pid = getpid();
  608. marshall_to_mar_cpg_name_t (&req_lib_cpg_leave.group_name,
  609. group);
  610. iov[0].iov_base = (void *)&req_lib_cpg_leave;
  611. iov[0].iov_len = sizeof (struct req_lib_cpg_leave);
  612. do {
  613. error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 1,
  614. &res_lib_cpg_leave, sizeof (struct res_lib_cpg_leave));
  615. if (error != CS_OK) {
  616. goto error_exit;
  617. }
  618. } while (res_lib_cpg_leave.header.error == CS_ERR_BUSY);
  619. error = res_lib_cpg_leave.header.error;
  620. error_exit:
  621. hdb_handle_put (&cpg_handle_t_db, handle);
  622. return (error);
  623. }
  624. cs_error_t cpg_membership_get (
  625. cpg_handle_t handle,
  626. struct cpg_name *group_name,
  627. struct cpg_address *member_list,
  628. int *member_list_entries)
  629. {
  630. cs_error_t error;
  631. struct cpg_inst *cpg_inst;
  632. struct iovec iov;
  633. struct req_lib_cpg_membership_get req_lib_cpg_membership_get;
  634. struct res_lib_cpg_membership_get res_lib_cpg_membership_get;
  635. unsigned int i;
  636. if (group_name->length > CPG_MAX_NAME_LENGTH) {
  637. return (CS_ERR_NAME_TOO_LONG);
  638. }
  639. if (member_list == NULL) {
  640. return (CS_ERR_INVALID_PARAM);
  641. }
  642. if (member_list_entries == NULL) {
  643. return (CS_ERR_INVALID_PARAM);
  644. }
  645. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  646. if (error != CS_OK) {
  647. return (error);
  648. }
  649. req_lib_cpg_membership_get.header.size = sizeof (struct req_lib_cpg_membership_get);
  650. req_lib_cpg_membership_get.header.id = MESSAGE_REQ_CPG_MEMBERSHIP;
  651. marshall_to_mar_cpg_name_t (&req_lib_cpg_membership_get.group_name,
  652. group_name);
  653. iov.iov_base = (void *)&req_lib_cpg_membership_get;
  654. iov.iov_len = sizeof (struct req_lib_cpg_membership_get);
  655. error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1,
  656. &res_lib_cpg_membership_get, sizeof (res_lib_cpg_membership_get));
  657. if (error != CS_OK) {
  658. goto error_exit;
  659. }
  660. error = res_lib_cpg_membership_get.header.error;
  661. /*
  662. * Copy results to caller
  663. */
  664. *member_list_entries = res_lib_cpg_membership_get.member_count;
  665. if (member_list) {
  666. for (i = 0; i < res_lib_cpg_membership_get.member_count; i++) {
  667. marshall_from_mar_cpg_address_t (&member_list[i],
  668. &res_lib_cpg_membership_get.member_list[i]);
  669. }
  670. }
  671. error_exit:
  672. hdb_handle_put (&cpg_handle_t_db, handle);
  673. return (error);
  674. }
  675. cs_error_t cpg_local_get (
  676. cpg_handle_t handle,
  677. unsigned int *local_nodeid)
  678. {
  679. cs_error_t error;
  680. struct cpg_inst *cpg_inst;
  681. struct iovec iov;
  682. struct req_lib_cpg_local_get req_lib_cpg_local_get;
  683. struct res_lib_cpg_local_get res_lib_cpg_local_get;
  684. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  685. if (error != CS_OK) {
  686. return (error);
  687. }
  688. req_lib_cpg_local_get.header.size = sizeof (struct qb_ipc_request_header);
  689. req_lib_cpg_local_get.header.id = MESSAGE_REQ_CPG_LOCAL_GET;
  690. iov.iov_base = (void *)&req_lib_cpg_local_get;
  691. iov.iov_len = sizeof (struct req_lib_cpg_local_get);
  692. error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1,
  693. &res_lib_cpg_local_get, sizeof (res_lib_cpg_local_get));
  694. if (error != CS_OK) {
  695. goto error_exit;
  696. }
  697. error = res_lib_cpg_local_get.header.error;
  698. *local_nodeid = res_lib_cpg_local_get.local_nodeid;
  699. error_exit:
  700. hdb_handle_put (&cpg_handle_t_db, handle);
  701. return (error);
  702. }
  703. cs_error_t cpg_flow_control_state_get (
  704. cpg_handle_t handle,
  705. cpg_flow_control_state_t *flow_control_state)
  706. {
  707. cs_error_t error;
  708. struct cpg_inst *cpg_inst;
  709. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  710. if (error != CS_OK) {
  711. return (error);
  712. }
  713. *flow_control_state = CPG_FLOW_CONTROL_DISABLED;
  714. error = CS_OK;
  715. hdb_handle_put (&cpg_handle_t_db, handle);
  716. return (error);
  717. }
  718. static int
  719. memory_map (char *path, const char *file, void **buf, size_t bytes)
  720. {
  721. int32_t fd;
  722. void *addr;
  723. int32_t res;
  724. char *buffer;
  725. int32_t i;
  726. size_t written;
  727. size_t page_size;
  728. long int sysconf_page_size;
  729. mode_t old_umask;
  730. snprintf (path, PATH_MAX, "/dev/shm/%s", file);
  731. old_umask = umask(CPG_MEMORY_MAP_UMASK);
  732. fd = mkstemp (path);
  733. (void)umask(old_umask);
  734. if (fd == -1) {
  735. snprintf (path, PATH_MAX, LOCALSTATEDIR "/run/%s", file);
  736. old_umask = umask(CPG_MEMORY_MAP_UMASK);
  737. fd = mkstemp (path);
  738. (void)umask(old_umask);
  739. if (fd == -1) {
  740. return (-1);
  741. }
  742. }
  743. res = ftruncate (fd, bytes);
  744. if (res == -1) {
  745. goto error_close_unlink;
  746. }
  747. sysconf_page_size = sysconf(_SC_PAGESIZE);
  748. if (sysconf_page_size <= 0) {
  749. goto error_close_unlink;
  750. }
  751. page_size = sysconf_page_size;
  752. buffer = malloc (page_size);
  753. if (buffer == NULL) {
  754. goto error_close_unlink;
  755. }
  756. memset (buffer, 0, page_size);
  757. for (i = 0; i < (bytes / page_size); i++) {
  758. retry_write:
  759. written = write (fd, buffer, page_size);
  760. if (written == -1 && errno == EINTR) {
  761. goto retry_write;
  762. }
  763. if (written != page_size) {
  764. free (buffer);
  765. goto error_close_unlink;
  766. }
  767. }
  768. free (buffer);
  769. addr = mmap (NULL, bytes, PROT_READ | PROT_WRITE,
  770. MAP_SHARED, fd, 0);
  771. if (addr == MAP_FAILED) {
  772. goto error_close_unlink;
  773. }
  774. #ifdef MADV_NOSYNC
  775. madvise(addr, bytes, MADV_NOSYNC);
  776. #endif
  777. res = close (fd);
  778. if (res) {
  779. munmap(addr, bytes);
  780. return (-1);
  781. }
  782. *buf = addr;
  783. return 0;
  784. error_close_unlink:
  785. close (fd);
  786. unlink(path);
  787. return -1;
  788. }
  789. cs_error_t cpg_zcb_alloc (
  790. cpg_handle_t handle,
  791. size_t size,
  792. void **buffer)
  793. {
  794. void *buf = NULL;
  795. char path[PATH_MAX];
  796. mar_req_coroipcc_zc_alloc_t req_coroipcc_zc_alloc;
  797. struct qb_ipc_response_header res_coroipcs_zc_alloc;
  798. size_t map_size;
  799. struct iovec iovec;
  800. struct coroipcs_zc_header *hdr;
  801. cs_error_t error;
  802. struct cpg_inst *cpg_inst;
  803. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  804. if (error != CS_OK) {
  805. return (error);
  806. }
  807. map_size = size + sizeof (struct req_lib_cpg_mcast) + sizeof (struct coroipcs_zc_header);
  808. assert(memory_map (path, "corosync_zerocopy-XXXXXX", &buf, map_size) != -1);
  809. if (strlen(path) >= CPG_ZC_PATH_LEN) {
  810. unlink(path);
  811. munmap (buf, map_size);
  812. return (CS_ERR_NAME_TOO_LONG);
  813. }
  814. req_coroipcc_zc_alloc.header.size = sizeof (mar_req_coroipcc_zc_alloc_t);
  815. req_coroipcc_zc_alloc.header.id = MESSAGE_REQ_CPG_ZC_ALLOC;
  816. req_coroipcc_zc_alloc.map_size = map_size;
  817. strcpy (req_coroipcc_zc_alloc.path_to_file, path);
  818. iovec.iov_base = (void *)&req_coroipcc_zc_alloc;
  819. iovec.iov_len = sizeof (mar_req_coroipcc_zc_alloc_t);
  820. error = coroipcc_msg_send_reply_receive (
  821. cpg_inst->c,
  822. &iovec,
  823. 1,
  824. &res_coroipcs_zc_alloc,
  825. sizeof (struct qb_ipc_response_header));
  826. if (error != CS_OK) {
  827. goto error_exit;
  828. }
  829. hdr = (struct coroipcs_zc_header *)buf;
  830. hdr->map_size = map_size;
  831. *buffer = ((char *)buf) + sizeof (struct coroipcs_zc_header) + sizeof (struct req_lib_cpg_mcast);
  832. error_exit:
  833. hdb_handle_put (&cpg_handle_t_db, handle);
  834. /*
  835. * Coverity correctly reports an error here. We cannot safely munmap and unlink the file, because
  836. * the timing of the failure is the key issue: if a failure occurs before the IPC reply,
  837. * the file should be deleted.
  838. * However, if the failure happens during the IPC reply, Corosync has already deleted the file.
  839. * This means the cpg library could attempt to delete a non-existing file (not a problem) or,
  840. * in a theoretical race condition, delete a new file created by another application.
  841. * There are multiple possible solutions, but none of them are ready to be implemented yet.
  842. */
  843. return (error);
  844. }
  845. cs_error_t cpg_zcb_free (
  846. cpg_handle_t handle,
  847. void *buffer)
  848. {
  849. cs_error_t error;
  850. unsigned int res;
  851. struct cpg_inst *cpg_inst;
  852. mar_req_coroipcc_zc_free_t req_coroipcc_zc_free;
  853. struct qb_ipc_response_header res_coroipcs_zc_free;
  854. struct iovec iovec;
  855. struct coroipcs_zc_header *header = (struct coroipcs_zc_header *)((char *)buffer - sizeof (struct coroipcs_zc_header) - sizeof (struct req_lib_cpg_mcast));
  856. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  857. if (error != CS_OK) {
  858. return (error);
  859. }
  860. req_coroipcc_zc_free.header.size = sizeof (mar_req_coroipcc_zc_free_t);
  861. req_coroipcc_zc_free.header.id = MESSAGE_REQ_CPG_ZC_FREE;
  862. req_coroipcc_zc_free.map_size = header->map_size;
  863. req_coroipcc_zc_free.server_address = header->server_address;
  864. iovec.iov_base = (void *)&req_coroipcc_zc_free;
  865. iovec.iov_len = sizeof (mar_req_coroipcc_zc_free_t);
  866. error = coroipcc_msg_send_reply_receive (
  867. cpg_inst->c,
  868. &iovec,
  869. 1,
  870. &res_coroipcs_zc_free,
  871. sizeof (struct qb_ipc_response_header));
  872. if (error != CS_OK) {
  873. goto error_exit;
  874. }
  875. res = munmap ((void *)header, header->map_size);
  876. if (res == -1) {
  877. error = qb_to_cs_error(-errno);
  878. goto error_exit;
  879. }
  880. error_exit:
  881. hdb_handle_put (&cpg_handle_t_db, handle);
  882. return (error);
  883. }
  884. cs_error_t cpg_zcb_mcast_joined (
  885. cpg_handle_t handle,
  886. cpg_guarantee_t guarantee,
  887. void *msg,
  888. size_t msg_len)
  889. {
  890. cs_error_t error;
  891. struct cpg_inst *cpg_inst;
  892. struct req_lib_cpg_mcast *req_lib_cpg_mcast;
  893. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  894. mar_req_coroipcc_zc_execute_t req_coroipcc_zc_execute;
  895. struct coroipcs_zc_header *hdr;
  896. struct iovec iovec;
  897. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  898. if (error != CS_OK) {
  899. return (error);
  900. }
  901. if (msg_len > IPC_REQUEST_SIZE) {
  902. error = CS_ERR_TOO_BIG;
  903. goto error_exit;
  904. }
  905. req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)(((char *)msg) - sizeof (struct req_lib_cpg_mcast));
  906. req_lib_cpg_mcast->header.size = sizeof (struct req_lib_cpg_mcast) +
  907. msg_len;
  908. req_lib_cpg_mcast->header.id = MESSAGE_REQ_CPG_MCAST;
  909. req_lib_cpg_mcast->guarantee = guarantee;
  910. req_lib_cpg_mcast->msglen = msg_len;
  911. hdr = (struct coroipcs_zc_header *)(((char *)req_lib_cpg_mcast) - sizeof (struct coroipcs_zc_header));
  912. req_coroipcc_zc_execute.header.size = sizeof (mar_req_coroipcc_zc_execute_t);
  913. req_coroipcc_zc_execute.header.id = MESSAGE_REQ_CPG_ZC_EXECUTE;
  914. req_coroipcc_zc_execute.server_address = hdr->server_address;
  915. iovec.iov_base = (void *)&req_coroipcc_zc_execute;
  916. iovec.iov_len = sizeof (mar_req_coroipcc_zc_execute_t);
  917. error = coroipcc_msg_send_reply_receive (
  918. cpg_inst->c,
  919. &iovec,
  920. 1,
  921. &res_lib_cpg_mcast,
  922. sizeof(res_lib_cpg_mcast));
  923. if (error != CS_OK) {
  924. goto error_exit;
  925. }
  926. error = res_lib_cpg_mcast.header.error;
  927. error_exit:
  928. hdb_handle_put (&cpg_handle_t_db, handle);
  929. return (error);
  930. }
  931. static cs_error_t send_fragments (
  932. struct cpg_inst *cpg_inst,
  933. cpg_guarantee_t guarantee,
  934. size_t msg_len,
  935. const struct iovec *iovec,
  936. unsigned int iov_len)
  937. {
  938. int i;
  939. cs_error_t error = CS_OK;
  940. struct iovec iov[2];
  941. struct req_lib_cpg_partial_mcast req_lib_cpg_mcast;
  942. struct res_lib_cpg_partial_send res_lib_cpg_partial_send;
  943. size_t sent = 0;
  944. size_t iov_sent = 0;
  945. int retry_count;
  946. req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_PARTIAL_MCAST;
  947. req_lib_cpg_mcast.guarantee = guarantee;
  948. req_lib_cpg_mcast.msglen = msg_len;
  949. iov[0].iov_base = (void *)&req_lib_cpg_mcast;
  950. iov[0].iov_len = sizeof (struct req_lib_cpg_partial_mcast);
  951. i=0;
  952. iov_sent = 0 ;
  953. qb_ipcc_fc_enable_max_set(cpg_inst->c, 2);
  954. while (error == CS_OK && sent < msg_len) {
  955. retry_count = 0;
  956. if ( (iovec[i].iov_len - iov_sent) > cpg_inst->max_msg_size) {
  957. iov[1].iov_len = cpg_inst->max_msg_size;
  958. }
  959. else {
  960. iov[1].iov_len = iovec[i].iov_len - iov_sent;
  961. }
  962. if (sent == 0) {
  963. req_lib_cpg_mcast.type = LIBCPG_PARTIAL_FIRST;
  964. }
  965. else if ((sent + iov[1].iov_len) == msg_len) {
  966. req_lib_cpg_mcast.type = LIBCPG_PARTIAL_LAST;
  967. }
  968. else {
  969. req_lib_cpg_mcast.type = LIBCPG_PARTIAL_CONTINUED;
  970. }
  971. req_lib_cpg_mcast.fraglen = iov[1].iov_len;
  972. req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_partial_mcast) + iov[1].iov_len;
  973. iov[1].iov_base = (char *)iovec[i].iov_base + iov_sent;
  974. resend:
  975. error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 2,
  976. &res_lib_cpg_partial_send,
  977. sizeof (res_lib_cpg_partial_send));
  978. if (error == CS_ERR_TRY_AGAIN) {
  979. fprintf(stderr, "sleep. counter=%d\n", retry_count);
  980. if (++retry_count > MAX_RETRIES) {
  981. goto error_exit;
  982. }
  983. usleep(10000);
  984. goto resend;
  985. }
  986. iov_sent += iov[1].iov_len;
  987. sent += iov[1].iov_len;
  988. /* Next iovec */
  989. if (iov_sent >= iovec[i].iov_len) {
  990. i++;
  991. iov_sent = 0;
  992. }
  993. error = res_lib_cpg_partial_send.header.error;
  994. }
  995. error_exit:
  996. qb_ipcc_fc_enable_max_set(cpg_inst->c, 1);
  997. return error;
  998. }
  999. cs_error_t cpg_mcast_joined (
  1000. cpg_handle_t handle,
  1001. cpg_guarantee_t guarantee,
  1002. const struct iovec *iovec,
  1003. unsigned int iov_len)
  1004. {
  1005. int i;
  1006. cs_error_t error;
  1007. struct cpg_inst *cpg_inst;
  1008. struct iovec iov[64];
  1009. struct req_lib_cpg_mcast req_lib_cpg_mcast;
  1010. size_t msg_len = 0;
  1011. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  1012. if (error != CS_OK) {
  1013. return (error);
  1014. }
  1015. for (i = 0; i < iov_len; i++ ) {
  1016. msg_len += iovec[i].iov_len;
  1017. }
  1018. if (msg_len > cpg_inst->max_msg_size) {
  1019. error = send_fragments(cpg_inst, guarantee, msg_len, iovec, iov_len);
  1020. goto error_exit;
  1021. }
  1022. req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_mcast) +
  1023. msg_len;
  1024. req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_MCAST;
  1025. req_lib_cpg_mcast.guarantee = guarantee;
  1026. req_lib_cpg_mcast.msglen = msg_len;
  1027. iov[0].iov_base = (void *)&req_lib_cpg_mcast;
  1028. iov[0].iov_len = sizeof (struct req_lib_cpg_mcast);
  1029. memcpy (&iov[1], iovec, iov_len * sizeof (struct iovec));
  1030. qb_ipcc_fc_enable_max_set(cpg_inst->c, 2);
  1031. error = qb_to_cs_error(qb_ipcc_sendv(cpg_inst->c, iov, iov_len + 1));
  1032. qb_ipcc_fc_enable_max_set(cpg_inst->c, 1);
  1033. error_exit:
  1034. hdb_handle_put (&cpg_handle_t_db, handle);
  1035. return (error);
  1036. }
  1037. cs_error_t cpg_iteration_initialize(
  1038. cpg_handle_t handle,
  1039. cpg_iteration_type_t iteration_type,
  1040. const struct cpg_name *group,
  1041. cpg_iteration_handle_t *cpg_iteration_handle)
  1042. {
  1043. cs_error_t error;
  1044. struct iovec iov;
  1045. struct cpg_inst *cpg_inst;
  1046. struct cpg_iteration_instance_t *cpg_iteration_instance;
  1047. struct req_lib_cpg_iterationinitialize req_lib_cpg_iterationinitialize;
  1048. struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize;
  1049. if (group && group->length > CPG_MAX_NAME_LENGTH) {
  1050. return (CS_ERR_NAME_TOO_LONG);
  1051. }
  1052. if (cpg_iteration_handle == NULL) {
  1053. return (CS_ERR_INVALID_PARAM);
  1054. }
  1055. if ((iteration_type == CPG_ITERATION_ONE_GROUP && group == NULL) ||
  1056. (iteration_type != CPG_ITERATION_ONE_GROUP && group != NULL)) {
  1057. return (CS_ERR_INVALID_PARAM);
  1058. }
  1059. if (iteration_type != CPG_ITERATION_NAME_ONLY && iteration_type != CPG_ITERATION_ONE_GROUP &&
  1060. iteration_type != CPG_ITERATION_ALL) {
  1061. return (CS_ERR_INVALID_PARAM);
  1062. }
  1063. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  1064. if (error != CS_OK) {
  1065. return (error);
  1066. }
  1067. error = hdb_error_to_cs (hdb_handle_create (&cpg_iteration_handle_t_db,
  1068. sizeof (struct cpg_iteration_instance_t), cpg_iteration_handle));
  1069. if (error != CS_OK) {
  1070. goto error_put_cpg_db;
  1071. }
  1072. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, *cpg_iteration_handle,
  1073. (void *)&cpg_iteration_instance));
  1074. if (error != CS_OK) {
  1075. goto error_destroy;
  1076. }
  1077. cpg_iteration_instance->conn = cpg_inst->c;
  1078. qb_list_init (&cpg_iteration_instance->list);
  1079. req_lib_cpg_iterationinitialize.header.size = sizeof (struct req_lib_cpg_iterationinitialize);
  1080. req_lib_cpg_iterationinitialize.header.id = MESSAGE_REQ_CPG_ITERATIONINITIALIZE;
  1081. req_lib_cpg_iterationinitialize.iteration_type = iteration_type;
  1082. if (group) {
  1083. marshall_to_mar_cpg_name_t (&req_lib_cpg_iterationinitialize.group_name, group);
  1084. }
  1085. iov.iov_base = (void *)&req_lib_cpg_iterationinitialize;
  1086. iov.iov_len = sizeof (struct req_lib_cpg_iterationinitialize);
  1087. error = coroipcc_msg_send_reply_receive (cpg_inst->c,
  1088. &iov,
  1089. 1,
  1090. &res_lib_cpg_iterationinitialize,
  1091. sizeof (struct res_lib_cpg_iterationinitialize));
  1092. if (error != CS_OK) {
  1093. goto error_put_destroy;
  1094. }
  1095. cpg_iteration_instance->executive_iteration_handle =
  1096. res_lib_cpg_iterationinitialize.iteration_handle;
  1097. cpg_iteration_instance->cpg_iteration_handle = *cpg_iteration_handle;
  1098. qb_list_add (&cpg_iteration_instance->list, &cpg_inst->iteration_list_head);
  1099. hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  1100. hdb_handle_put (&cpg_handle_t_db, handle);
  1101. return (res_lib_cpg_iterationinitialize.header.error);
  1102. error_put_destroy:
  1103. hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  1104. error_destroy:
  1105. hdb_handle_destroy (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  1106. error_put_cpg_db:
  1107. hdb_handle_put (&cpg_handle_t_db, handle);
  1108. return (error);
  1109. }
  1110. cs_error_t cpg_iteration_next(
  1111. cpg_iteration_handle_t handle,
  1112. struct cpg_iteration_description_t *description)
  1113. {
  1114. cs_error_t error;
  1115. struct cpg_iteration_instance_t *cpg_iteration_instance;
  1116. struct req_lib_cpg_iterationnext req_lib_cpg_iterationnext;
  1117. struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext;
  1118. if (description == NULL) {
  1119. return CS_ERR_INVALID_PARAM;
  1120. }
  1121. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle,
  1122. (void *)&cpg_iteration_instance));
  1123. if (error != CS_OK) {
  1124. goto error_exit;
  1125. }
  1126. req_lib_cpg_iterationnext.header.size = sizeof (struct req_lib_cpg_iterationnext);
  1127. req_lib_cpg_iterationnext.header.id = MESSAGE_REQ_CPG_ITERATIONNEXT;
  1128. req_lib_cpg_iterationnext.iteration_handle = cpg_iteration_instance->executive_iteration_handle;
  1129. error = qb_to_cs_error (qb_ipcc_send (cpg_iteration_instance->conn,
  1130. &req_lib_cpg_iterationnext,
  1131. req_lib_cpg_iterationnext.header.size));
  1132. if (error != CS_OK) {
  1133. goto error_put;
  1134. }
  1135. error = qb_to_cs_error (qb_ipcc_recv (cpg_iteration_instance->conn,
  1136. &res_lib_cpg_iterationnext,
  1137. sizeof(struct res_lib_cpg_iterationnext), -1));
  1138. if (error != CS_OK) {
  1139. goto error_put;
  1140. }
  1141. marshall_from_mar_cpg_iteration_description_t(
  1142. description,
  1143. &res_lib_cpg_iterationnext.description);
  1144. error = res_lib_cpg_iterationnext.header.error;
  1145. error_put:
  1146. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  1147. error_exit:
  1148. return (error);
  1149. }
  1150. cs_error_t cpg_iteration_finalize (
  1151. cpg_iteration_handle_t handle)
  1152. {
  1153. cs_error_t error;
  1154. struct iovec iov;
  1155. struct cpg_iteration_instance_t *cpg_iteration_instance;
  1156. struct req_lib_cpg_iterationfinalize req_lib_cpg_iterationfinalize;
  1157. struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize;
  1158. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle,
  1159. (void *)&cpg_iteration_instance));
  1160. if (error != CS_OK) {
  1161. goto error_exit;
  1162. }
  1163. req_lib_cpg_iterationfinalize.header.size = sizeof (struct req_lib_cpg_iterationfinalize);
  1164. req_lib_cpg_iterationfinalize.header.id = MESSAGE_REQ_CPG_ITERATIONFINALIZE;
  1165. req_lib_cpg_iterationfinalize.iteration_handle = cpg_iteration_instance->executive_iteration_handle;
  1166. iov.iov_base = (void *)&req_lib_cpg_iterationfinalize;
  1167. iov.iov_len = sizeof (struct req_lib_cpg_iterationfinalize);
  1168. error = coroipcc_msg_send_reply_receive (cpg_iteration_instance->conn,
  1169. &iov,
  1170. 1,
  1171. &res_lib_cpg_iterationfinalize,
  1172. sizeof (struct req_lib_cpg_iterationfinalize));
  1173. if (error != CS_OK) {
  1174. goto error_put;
  1175. }
  1176. cpg_iteration_instance_finalize (cpg_iteration_instance);
  1177. hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->cpg_iteration_handle);
  1178. return (res_lib_cpg_iterationfinalize.header.error);
  1179. error_put:
  1180. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  1181. error_exit:
  1182. return (error);
  1183. }
  1184. /** @} */