cpg.c 38 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427
  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. */
  416. if (assembly_data) {
  417. error = CS_ERR_MESSAGE_ERROR;
  418. goto error_put;
  419. }
  420. assembly_data = malloc(sizeof(struct cpg_assembly_data));
  421. if (!assembly_data) {
  422. error = CS_ERR_NO_MEMORY;
  423. goto error_put;
  424. }
  425. assembly_data->nodeid = res_cpg_partial_deliver_callback->nodeid;
  426. assembly_data->pid = res_cpg_partial_deliver_callback->pid;
  427. assembly_data->assembly_buf = malloc(res_cpg_partial_deliver_callback->msglen);
  428. if (!assembly_data->assembly_buf) {
  429. free(assembly_data);
  430. error = CS_ERR_NO_MEMORY;
  431. goto error_put;
  432. }
  433. assembly_data->assembly_buf_ptr = 0;
  434. list_init (&assembly_data->list);
  435. list_add (&assembly_data->list, &cpg_inst->assembly_list_head);
  436. }
  437. if (assembly_data) {
  438. memcpy(assembly_data->assembly_buf + assembly_data->assembly_buf_ptr,
  439. res_cpg_partial_deliver_callback->message, res_cpg_partial_deliver_callback->fraglen);
  440. assembly_data->assembly_buf_ptr += res_cpg_partial_deliver_callback->fraglen;
  441. if (res_cpg_partial_deliver_callback->type == LIBCPG_PARTIAL_LAST) {
  442. cpg_inst_copy.model_v1_data.cpg_deliver_fn (handle,
  443. &group_name,
  444. res_cpg_partial_deliver_callback->nodeid,
  445. res_cpg_partial_deliver_callback->pid,
  446. assembly_data->assembly_buf,
  447. res_cpg_partial_deliver_callback->msglen);
  448. list_del (&assembly_data->list);
  449. free(assembly_data->assembly_buf);
  450. free(assembly_data);
  451. }
  452. }
  453. break;
  454. case MESSAGE_RES_CPG_CONFCHG_CALLBACK:
  455. if (cpg_inst_copy.model_v1_data.cpg_confchg_fn == NULL) {
  456. break;
  457. }
  458. res_cpg_confchg_callback = (struct res_lib_cpg_confchg_callback *)dispatch_data;
  459. for (i = 0; i < res_cpg_confchg_callback->member_list_entries; i++) {
  460. marshall_from_mar_cpg_address_t (&member_list[i],
  461. &res_cpg_confchg_callback->member_list[i]);
  462. }
  463. left_list_start = res_cpg_confchg_callback->member_list +
  464. res_cpg_confchg_callback->member_list_entries;
  465. for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) {
  466. marshall_from_mar_cpg_address_t (&left_list[i],
  467. &left_list_start[i]);
  468. }
  469. joined_list_start = res_cpg_confchg_callback->member_list +
  470. res_cpg_confchg_callback->member_list_entries +
  471. res_cpg_confchg_callback->left_list_entries;
  472. for (i = 0; i < res_cpg_confchg_callback->joined_list_entries; i++) {
  473. marshall_from_mar_cpg_address_t (&joined_list[i],
  474. &joined_list_start[i]);
  475. }
  476. marshall_from_mar_cpg_name_t (
  477. &group_name,
  478. &res_cpg_confchg_callback->group_name);
  479. cpg_inst_copy.model_v1_data.cpg_confchg_fn (handle,
  480. &group_name,
  481. member_list,
  482. res_cpg_confchg_callback->member_list_entries,
  483. left_list,
  484. res_cpg_confchg_callback->left_list_entries,
  485. joined_list,
  486. res_cpg_confchg_callback->joined_list_entries);
  487. /*
  488. * If member left while his partial packet was being assembled, assembly data must be removed from list
  489. */
  490. for (i = 0; i < res_cpg_confchg_callback->left_list_entries; i++) {
  491. for (iter = cpg_inst->assembly_list_head.next; iter != &cpg_inst->assembly_list_head;iter = tmp_iter) {
  492. struct cpg_assembly_data *current_assembly_data = list_entry (iter, struct cpg_assembly_data, list);
  493. tmp_iter = iter->next;
  494. if (current_assembly_data->nodeid != left_list[i].nodeid || current_assembly_data->pid != left_list[i].pid)
  495. continue;
  496. list_del (&current_assembly_data->list);
  497. free(current_assembly_data->assembly_buf);
  498. free(current_assembly_data);
  499. }
  500. }
  501. break;
  502. case MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK:
  503. if (cpg_inst_copy.model_v1_data.cpg_totem_confchg_fn == NULL) {
  504. break;
  505. }
  506. res_cpg_totem_confchg_callback = (struct res_lib_cpg_totem_confchg_callback *)dispatch_data;
  507. marshall_from_mar_cpg_ring_id_t (&ring_id, &res_cpg_totem_confchg_callback->ring_id);
  508. for (i = 0; i < res_cpg_totem_confchg_callback->member_list_entries; i++) {
  509. totem_member_list[i] = res_cpg_totem_confchg_callback->member_list[i];
  510. }
  511. cpg_inst_copy.model_v1_data.cpg_totem_confchg_fn (handle,
  512. ring_id,
  513. res_cpg_totem_confchg_callback->member_list_entries,
  514. totem_member_list);
  515. break;
  516. default:
  517. error = CS_ERR_LIBRARY;
  518. goto error_put;
  519. break;
  520. } /* - switch (dispatch_data->id) */
  521. break; /* case CPG_MODEL_V1 */
  522. } /* - switch (cpg_inst_copy.model_data.model) */
  523. if (cpg_inst_copy.finalize || cpg_inst->finalize) {
  524. /*
  525. * If the finalize has been called then get out of the dispatch.
  526. */
  527. cpg_inst->finalize = 1;
  528. error = CS_ERR_BAD_HANDLE;
  529. goto error_put;
  530. }
  531. /*
  532. * Determine if more messages should be processed
  533. */
  534. if (dispatch_types == CS_DISPATCH_ONE || dispatch_types == CS_DISPATCH_ONE_NONBLOCKING) {
  535. cont = 0;
  536. }
  537. } while (cont);
  538. error_put:
  539. hdb_handle_put (&cpg_handle_t_db, handle);
  540. return (error);
  541. }
  542. cs_error_t cpg_join (
  543. cpg_handle_t handle,
  544. const struct cpg_name *group)
  545. {
  546. cs_error_t error;
  547. struct cpg_inst *cpg_inst;
  548. struct iovec iov[2];
  549. struct req_lib_cpg_join req_lib_cpg_join;
  550. struct res_lib_cpg_join response;
  551. if (group->length > CPG_MAX_NAME_LENGTH) {
  552. return (CS_ERR_NAME_TOO_LONG);
  553. }
  554. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  555. if (error != CS_OK) {
  556. return (error);
  557. }
  558. /* Now join */
  559. req_lib_cpg_join.header.size = sizeof (struct req_lib_cpg_join);
  560. req_lib_cpg_join.header.id = MESSAGE_REQ_CPG_JOIN;
  561. req_lib_cpg_join.pid = getpid();
  562. req_lib_cpg_join.flags = 0;
  563. switch (cpg_inst->model_data.model) {
  564. case CPG_MODEL_V1:
  565. req_lib_cpg_join.flags = cpg_inst->model_v1_data.flags;
  566. break;
  567. }
  568. marshall_to_mar_cpg_name_t (&req_lib_cpg_join.group_name,
  569. group);
  570. iov[0].iov_base = (void *)&req_lib_cpg_join;
  571. iov[0].iov_len = sizeof (struct req_lib_cpg_join);
  572. do {
  573. error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 1,
  574. &response, sizeof (struct res_lib_cpg_join));
  575. if (error != CS_OK) {
  576. goto error_exit;
  577. }
  578. } while (response.header.error == CS_ERR_BUSY);
  579. error = response.header.error;
  580. error_exit:
  581. hdb_handle_put (&cpg_handle_t_db, handle);
  582. return (error);
  583. }
  584. cs_error_t cpg_leave (
  585. cpg_handle_t handle,
  586. const struct cpg_name *group)
  587. {
  588. cs_error_t error;
  589. struct cpg_inst *cpg_inst;
  590. struct iovec iov[2];
  591. struct req_lib_cpg_leave req_lib_cpg_leave;
  592. struct res_lib_cpg_leave res_lib_cpg_leave;
  593. if (group->length > CPG_MAX_NAME_LENGTH) {
  594. return (CS_ERR_NAME_TOO_LONG);
  595. }
  596. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  597. if (error != CS_OK) {
  598. return (error);
  599. }
  600. req_lib_cpg_leave.header.size = sizeof (struct req_lib_cpg_leave);
  601. req_lib_cpg_leave.header.id = MESSAGE_REQ_CPG_LEAVE;
  602. req_lib_cpg_leave.pid = getpid();
  603. marshall_to_mar_cpg_name_t (&req_lib_cpg_leave.group_name,
  604. group);
  605. iov[0].iov_base = (void *)&req_lib_cpg_leave;
  606. iov[0].iov_len = sizeof (struct req_lib_cpg_leave);
  607. do {
  608. error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 1,
  609. &res_lib_cpg_leave, sizeof (struct res_lib_cpg_leave));
  610. if (error != CS_OK) {
  611. goto error_exit;
  612. }
  613. } while (res_lib_cpg_leave.header.error == CS_ERR_BUSY);
  614. error = res_lib_cpg_leave.header.error;
  615. error_exit:
  616. hdb_handle_put (&cpg_handle_t_db, handle);
  617. return (error);
  618. }
  619. cs_error_t cpg_membership_get (
  620. cpg_handle_t handle,
  621. struct cpg_name *group_name,
  622. struct cpg_address *member_list,
  623. int *member_list_entries)
  624. {
  625. cs_error_t error;
  626. struct cpg_inst *cpg_inst;
  627. struct iovec iov;
  628. struct req_lib_cpg_membership_get req_lib_cpg_membership_get;
  629. struct res_lib_cpg_membership_get res_lib_cpg_membership_get;
  630. unsigned int i;
  631. if (group_name->length > CPG_MAX_NAME_LENGTH) {
  632. return (CS_ERR_NAME_TOO_LONG);
  633. }
  634. if (member_list == NULL) {
  635. return (CS_ERR_INVALID_PARAM);
  636. }
  637. if (member_list_entries == NULL) {
  638. return (CS_ERR_INVALID_PARAM);
  639. }
  640. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  641. if (error != CS_OK) {
  642. return (error);
  643. }
  644. req_lib_cpg_membership_get.header.size = sizeof (struct req_lib_cpg_membership_get);
  645. req_lib_cpg_membership_get.header.id = MESSAGE_REQ_CPG_MEMBERSHIP;
  646. marshall_to_mar_cpg_name_t (&req_lib_cpg_membership_get.group_name,
  647. group_name);
  648. iov.iov_base = (void *)&req_lib_cpg_membership_get;
  649. iov.iov_len = sizeof (struct req_lib_cpg_membership_get);
  650. error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1,
  651. &res_lib_cpg_membership_get, sizeof (res_lib_cpg_membership_get));
  652. if (error != CS_OK) {
  653. goto error_exit;
  654. }
  655. error = res_lib_cpg_membership_get.header.error;
  656. /*
  657. * Copy results to caller
  658. */
  659. *member_list_entries = res_lib_cpg_membership_get.member_count;
  660. if (member_list) {
  661. for (i = 0; i < res_lib_cpg_membership_get.member_count; i++) {
  662. marshall_from_mar_cpg_address_t (&member_list[i],
  663. &res_lib_cpg_membership_get.member_list[i]);
  664. }
  665. }
  666. error_exit:
  667. hdb_handle_put (&cpg_handle_t_db, handle);
  668. return (error);
  669. }
  670. cs_error_t cpg_local_get (
  671. cpg_handle_t handle,
  672. unsigned int *local_nodeid)
  673. {
  674. cs_error_t error;
  675. struct cpg_inst *cpg_inst;
  676. struct iovec iov;
  677. struct req_lib_cpg_local_get req_lib_cpg_local_get;
  678. struct res_lib_cpg_local_get res_lib_cpg_local_get;
  679. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  680. if (error != CS_OK) {
  681. return (error);
  682. }
  683. req_lib_cpg_local_get.header.size = sizeof (struct qb_ipc_request_header);
  684. req_lib_cpg_local_get.header.id = MESSAGE_REQ_CPG_LOCAL_GET;
  685. iov.iov_base = (void *)&req_lib_cpg_local_get;
  686. iov.iov_len = sizeof (struct req_lib_cpg_local_get);
  687. error = coroipcc_msg_send_reply_receive (cpg_inst->c, &iov, 1,
  688. &res_lib_cpg_local_get, sizeof (res_lib_cpg_local_get));
  689. if (error != CS_OK) {
  690. goto error_exit;
  691. }
  692. error = res_lib_cpg_local_get.header.error;
  693. *local_nodeid = res_lib_cpg_local_get.local_nodeid;
  694. error_exit:
  695. hdb_handle_put (&cpg_handle_t_db, handle);
  696. return (error);
  697. }
  698. cs_error_t cpg_flow_control_state_get (
  699. cpg_handle_t handle,
  700. cpg_flow_control_state_t *flow_control_state)
  701. {
  702. cs_error_t error;
  703. struct cpg_inst *cpg_inst;
  704. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  705. if (error != CS_OK) {
  706. return (error);
  707. }
  708. *flow_control_state = CPG_FLOW_CONTROL_DISABLED;
  709. error = CS_OK;
  710. hdb_handle_put (&cpg_handle_t_db, handle);
  711. return (error);
  712. }
  713. static int
  714. memory_map (char *path, const char *file, void **buf, size_t bytes)
  715. {
  716. int32_t fd;
  717. void *addr;
  718. int32_t res;
  719. char *buffer;
  720. int32_t i;
  721. size_t written;
  722. size_t page_size;
  723. long int sysconf_page_size;
  724. mode_t old_umask;
  725. snprintf (path, PATH_MAX, "/dev/shm/%s", file);
  726. old_umask = umask(CPG_MEMORY_MAP_UMASK);
  727. fd = mkstemp (path);
  728. (void)umask(old_umask);
  729. if (fd == -1) {
  730. snprintf (path, PATH_MAX, LOCALSTATEDIR "/run/%s", file);
  731. old_umask = umask(CPG_MEMORY_MAP_UMASK);
  732. fd = mkstemp (path);
  733. (void)umask(old_umask);
  734. if (fd == -1) {
  735. return (-1);
  736. }
  737. }
  738. res = ftruncate (fd, bytes);
  739. if (res == -1) {
  740. goto error_close_unlink;
  741. }
  742. sysconf_page_size = sysconf(_SC_PAGESIZE);
  743. if (sysconf_page_size <= 0) {
  744. goto error_close_unlink;
  745. }
  746. page_size = sysconf_page_size;
  747. buffer = malloc (page_size);
  748. if (buffer == NULL) {
  749. goto error_close_unlink;
  750. }
  751. memset (buffer, 0, page_size);
  752. for (i = 0; i < (bytes / page_size); i++) {
  753. retry_write:
  754. written = write (fd, buffer, page_size);
  755. if (written == -1 && errno == EINTR) {
  756. goto retry_write;
  757. }
  758. if (written != page_size) {
  759. free (buffer);
  760. goto error_close_unlink;
  761. }
  762. }
  763. free (buffer);
  764. addr = mmap (NULL, bytes, PROT_READ | PROT_WRITE,
  765. MAP_SHARED, fd, 0);
  766. if (addr == MAP_FAILED) {
  767. goto error_close_unlink;
  768. }
  769. #ifdef MADV_NOSYNC
  770. madvise(addr, bytes, MADV_NOSYNC);
  771. #endif
  772. res = close (fd);
  773. if (res) {
  774. munmap(addr, bytes);
  775. return (-1);
  776. }
  777. *buf = addr;
  778. return 0;
  779. error_close_unlink:
  780. close (fd);
  781. unlink(path);
  782. return -1;
  783. }
  784. cs_error_t cpg_zcb_alloc (
  785. cpg_handle_t handle,
  786. size_t size,
  787. void **buffer)
  788. {
  789. void *buf = NULL;
  790. char path[PATH_MAX];
  791. mar_req_coroipcc_zc_alloc_t req_coroipcc_zc_alloc;
  792. struct qb_ipc_response_header res_coroipcs_zc_alloc;
  793. size_t map_size;
  794. struct iovec iovec;
  795. struct coroipcs_zc_header *hdr;
  796. cs_error_t error;
  797. struct cpg_inst *cpg_inst;
  798. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  799. if (error != CS_OK) {
  800. return (error);
  801. }
  802. map_size = size + sizeof (struct req_lib_cpg_mcast) + sizeof (struct coroipcs_zc_header);
  803. assert(memory_map (path, "corosync_zerocopy-XXXXXX", &buf, map_size) != -1);
  804. if (strlen(path) >= CPG_ZC_PATH_LEN) {
  805. unlink(path);
  806. munmap (buf, map_size);
  807. return (CS_ERR_NAME_TOO_LONG);
  808. }
  809. req_coroipcc_zc_alloc.header.size = sizeof (mar_req_coroipcc_zc_alloc_t);
  810. req_coroipcc_zc_alloc.header.id = MESSAGE_REQ_CPG_ZC_ALLOC;
  811. req_coroipcc_zc_alloc.map_size = map_size;
  812. strcpy (req_coroipcc_zc_alloc.path_to_file, path);
  813. iovec.iov_base = (void *)&req_coroipcc_zc_alloc;
  814. iovec.iov_len = sizeof (mar_req_coroipcc_zc_alloc_t);
  815. error = coroipcc_msg_send_reply_receive (
  816. cpg_inst->c,
  817. &iovec,
  818. 1,
  819. &res_coroipcs_zc_alloc,
  820. sizeof (struct qb_ipc_response_header));
  821. if (error != CS_OK) {
  822. goto error_exit;
  823. }
  824. hdr = (struct coroipcs_zc_header *)buf;
  825. hdr->map_size = map_size;
  826. *buffer = ((char *)buf) + sizeof (struct coroipcs_zc_header) + sizeof (struct req_lib_cpg_mcast);
  827. error_exit:
  828. hdb_handle_put (&cpg_handle_t_db, handle);
  829. return (error);
  830. }
  831. cs_error_t cpg_zcb_free (
  832. cpg_handle_t handle,
  833. void *buffer)
  834. {
  835. cs_error_t error;
  836. unsigned int res;
  837. struct cpg_inst *cpg_inst;
  838. mar_req_coroipcc_zc_free_t req_coroipcc_zc_free;
  839. struct qb_ipc_response_header res_coroipcs_zc_free;
  840. struct iovec iovec;
  841. struct coroipcs_zc_header *header = (struct coroipcs_zc_header *)((char *)buffer - sizeof (struct coroipcs_zc_header) - sizeof (struct req_lib_cpg_mcast));
  842. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  843. if (error != CS_OK) {
  844. return (error);
  845. }
  846. req_coroipcc_zc_free.header.size = sizeof (mar_req_coroipcc_zc_free_t);
  847. req_coroipcc_zc_free.header.id = MESSAGE_REQ_CPG_ZC_FREE;
  848. req_coroipcc_zc_free.map_size = header->map_size;
  849. req_coroipcc_zc_free.server_address = header->server_address;
  850. iovec.iov_base = (void *)&req_coroipcc_zc_free;
  851. iovec.iov_len = sizeof (mar_req_coroipcc_zc_free_t);
  852. error = coroipcc_msg_send_reply_receive (
  853. cpg_inst->c,
  854. &iovec,
  855. 1,
  856. &res_coroipcs_zc_free,
  857. sizeof (struct qb_ipc_response_header));
  858. if (error != CS_OK) {
  859. goto error_exit;
  860. }
  861. res = munmap ((void *)header, header->map_size);
  862. if (res == -1) {
  863. error = qb_to_cs_error(-errno);
  864. goto error_exit;
  865. }
  866. error_exit:
  867. hdb_handle_put (&cpg_handle_t_db, handle);
  868. return (error);
  869. }
  870. cs_error_t cpg_zcb_mcast_joined (
  871. cpg_handle_t handle,
  872. cpg_guarantee_t guarantee,
  873. void *msg,
  874. size_t msg_len)
  875. {
  876. cs_error_t error;
  877. struct cpg_inst *cpg_inst;
  878. struct req_lib_cpg_mcast *req_lib_cpg_mcast;
  879. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  880. mar_req_coroipcc_zc_execute_t req_coroipcc_zc_execute;
  881. struct coroipcs_zc_header *hdr;
  882. struct iovec iovec;
  883. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  884. if (error != CS_OK) {
  885. return (error);
  886. }
  887. if (msg_len > IPC_REQUEST_SIZE) {
  888. error = CS_ERR_TOO_BIG;
  889. goto error_exit;
  890. }
  891. req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)(((char *)msg) - sizeof (struct req_lib_cpg_mcast));
  892. req_lib_cpg_mcast->header.size = sizeof (struct req_lib_cpg_mcast) +
  893. msg_len;
  894. req_lib_cpg_mcast->header.id = MESSAGE_REQ_CPG_MCAST;
  895. req_lib_cpg_mcast->guarantee = guarantee;
  896. req_lib_cpg_mcast->msglen = msg_len;
  897. hdr = (struct coroipcs_zc_header *)(((char *)req_lib_cpg_mcast) - sizeof (struct coroipcs_zc_header));
  898. req_coroipcc_zc_execute.header.size = sizeof (mar_req_coroipcc_zc_execute_t);
  899. req_coroipcc_zc_execute.header.id = MESSAGE_REQ_CPG_ZC_EXECUTE;
  900. req_coroipcc_zc_execute.server_address = hdr->server_address;
  901. iovec.iov_base = (void *)&req_coroipcc_zc_execute;
  902. iovec.iov_len = sizeof (mar_req_coroipcc_zc_execute_t);
  903. error = coroipcc_msg_send_reply_receive (
  904. cpg_inst->c,
  905. &iovec,
  906. 1,
  907. &res_lib_cpg_mcast,
  908. sizeof(res_lib_cpg_mcast));
  909. if (error != CS_OK) {
  910. goto error_exit;
  911. }
  912. error = res_lib_cpg_mcast.header.error;
  913. error_exit:
  914. hdb_handle_put (&cpg_handle_t_db, handle);
  915. return (error);
  916. }
  917. static cs_error_t send_fragments (
  918. struct cpg_inst *cpg_inst,
  919. cpg_guarantee_t guarantee,
  920. size_t msg_len,
  921. const struct iovec *iovec,
  922. unsigned int iov_len)
  923. {
  924. int i;
  925. cs_error_t error = CS_OK;
  926. struct iovec iov[2];
  927. struct req_lib_cpg_partial_mcast req_lib_cpg_mcast;
  928. struct res_lib_cpg_partial_send res_lib_cpg_partial_send;
  929. size_t sent = 0;
  930. size_t iov_sent = 0;
  931. int retry_count;
  932. req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_PARTIAL_MCAST;
  933. req_lib_cpg_mcast.guarantee = guarantee;
  934. req_lib_cpg_mcast.msglen = msg_len;
  935. iov[0].iov_base = (void *)&req_lib_cpg_mcast;
  936. iov[0].iov_len = sizeof (struct req_lib_cpg_partial_mcast);
  937. i=0;
  938. iov_sent = 0 ;
  939. qb_ipcc_fc_enable_max_set(cpg_inst->c, 2);
  940. while (error == CS_OK && sent < msg_len) {
  941. retry_count = 0;
  942. if ( (iovec[i].iov_len - iov_sent) > cpg_inst->max_msg_size) {
  943. iov[1].iov_len = cpg_inst->max_msg_size;
  944. }
  945. else {
  946. iov[1].iov_len = iovec[i].iov_len - iov_sent;
  947. }
  948. if (sent == 0) {
  949. req_lib_cpg_mcast.type = LIBCPG_PARTIAL_FIRST;
  950. }
  951. else if ((sent + iov[1].iov_len) == msg_len) {
  952. req_lib_cpg_mcast.type = LIBCPG_PARTIAL_LAST;
  953. }
  954. else {
  955. req_lib_cpg_mcast.type = LIBCPG_PARTIAL_CONTINUED;
  956. }
  957. req_lib_cpg_mcast.fraglen = iov[1].iov_len;
  958. req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_partial_mcast) + iov[1].iov_len;
  959. iov[1].iov_base = (char *)iovec[i].iov_base + iov_sent;
  960. resend:
  961. error = coroipcc_msg_send_reply_receive (cpg_inst->c, iov, 2,
  962. &res_lib_cpg_partial_send,
  963. sizeof (res_lib_cpg_partial_send));
  964. if (error == CS_ERR_TRY_AGAIN) {
  965. fprintf(stderr, "sleep. counter=%d\n", retry_count);
  966. if (++retry_count > MAX_RETRIES) {
  967. goto error_exit;
  968. }
  969. usleep(10000);
  970. goto resend;
  971. }
  972. iov_sent += iov[1].iov_len;
  973. sent += iov[1].iov_len;
  974. /* Next iovec */
  975. if (iov_sent >= iovec[i].iov_len) {
  976. i++;
  977. iov_sent = 0;
  978. }
  979. error = res_lib_cpg_partial_send.header.error;
  980. }
  981. error_exit:
  982. qb_ipcc_fc_enable_max_set(cpg_inst->c, 1);
  983. return error;
  984. }
  985. cs_error_t cpg_mcast_joined (
  986. cpg_handle_t handle,
  987. cpg_guarantee_t guarantee,
  988. const struct iovec *iovec,
  989. unsigned int iov_len)
  990. {
  991. int i;
  992. cs_error_t error;
  993. struct cpg_inst *cpg_inst;
  994. struct iovec iov[64];
  995. struct req_lib_cpg_mcast req_lib_cpg_mcast;
  996. size_t msg_len = 0;
  997. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  998. if (error != CS_OK) {
  999. return (error);
  1000. }
  1001. for (i = 0; i < iov_len; i++ ) {
  1002. msg_len += iovec[i].iov_len;
  1003. }
  1004. if (msg_len > cpg_inst->max_msg_size) {
  1005. error = send_fragments(cpg_inst, guarantee, msg_len, iovec, iov_len);
  1006. goto error_exit;
  1007. }
  1008. req_lib_cpg_mcast.header.size = sizeof (struct req_lib_cpg_mcast) +
  1009. msg_len;
  1010. req_lib_cpg_mcast.header.id = MESSAGE_REQ_CPG_MCAST;
  1011. req_lib_cpg_mcast.guarantee = guarantee;
  1012. req_lib_cpg_mcast.msglen = msg_len;
  1013. iov[0].iov_base = (void *)&req_lib_cpg_mcast;
  1014. iov[0].iov_len = sizeof (struct req_lib_cpg_mcast);
  1015. memcpy (&iov[1], iovec, iov_len * sizeof (struct iovec));
  1016. qb_ipcc_fc_enable_max_set(cpg_inst->c, 2);
  1017. error = qb_to_cs_error(qb_ipcc_sendv(cpg_inst->c, iov, iov_len + 1));
  1018. qb_ipcc_fc_enable_max_set(cpg_inst->c, 1);
  1019. error_exit:
  1020. hdb_handle_put (&cpg_handle_t_db, handle);
  1021. return (error);
  1022. }
  1023. cs_error_t cpg_iteration_initialize(
  1024. cpg_handle_t handle,
  1025. cpg_iteration_type_t iteration_type,
  1026. const struct cpg_name *group,
  1027. cpg_iteration_handle_t *cpg_iteration_handle)
  1028. {
  1029. cs_error_t error;
  1030. struct iovec iov;
  1031. struct cpg_inst *cpg_inst;
  1032. struct cpg_iteration_instance_t *cpg_iteration_instance;
  1033. struct req_lib_cpg_iterationinitialize req_lib_cpg_iterationinitialize;
  1034. struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize;
  1035. if (group && group->length > CPG_MAX_NAME_LENGTH) {
  1036. return (CS_ERR_NAME_TOO_LONG);
  1037. }
  1038. if (cpg_iteration_handle == NULL) {
  1039. return (CS_ERR_INVALID_PARAM);
  1040. }
  1041. if ((iteration_type == CPG_ITERATION_ONE_GROUP && group == NULL) ||
  1042. (iteration_type != CPG_ITERATION_ONE_GROUP && group != NULL)) {
  1043. return (CS_ERR_INVALID_PARAM);
  1044. }
  1045. if (iteration_type != CPG_ITERATION_NAME_ONLY && iteration_type != CPG_ITERATION_ONE_GROUP &&
  1046. iteration_type != CPG_ITERATION_ALL) {
  1047. return (CS_ERR_INVALID_PARAM);
  1048. }
  1049. error = hdb_error_to_cs (hdb_handle_get (&cpg_handle_t_db, handle, (void *)&cpg_inst));
  1050. if (error != CS_OK) {
  1051. return (error);
  1052. }
  1053. error = hdb_error_to_cs (hdb_handle_create (&cpg_iteration_handle_t_db,
  1054. sizeof (struct cpg_iteration_instance_t), cpg_iteration_handle));
  1055. if (error != CS_OK) {
  1056. goto error_put_cpg_db;
  1057. }
  1058. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, *cpg_iteration_handle,
  1059. (void *)&cpg_iteration_instance));
  1060. if (error != CS_OK) {
  1061. goto error_destroy;
  1062. }
  1063. cpg_iteration_instance->conn = cpg_inst->c;
  1064. list_init (&cpg_iteration_instance->list);
  1065. req_lib_cpg_iterationinitialize.header.size = sizeof (struct req_lib_cpg_iterationinitialize);
  1066. req_lib_cpg_iterationinitialize.header.id = MESSAGE_REQ_CPG_ITERATIONINITIALIZE;
  1067. req_lib_cpg_iterationinitialize.iteration_type = iteration_type;
  1068. if (group) {
  1069. marshall_to_mar_cpg_name_t (&req_lib_cpg_iterationinitialize.group_name, group);
  1070. }
  1071. iov.iov_base = (void *)&req_lib_cpg_iterationinitialize;
  1072. iov.iov_len = sizeof (struct req_lib_cpg_iterationinitialize);
  1073. error = coroipcc_msg_send_reply_receive (cpg_inst->c,
  1074. &iov,
  1075. 1,
  1076. &res_lib_cpg_iterationinitialize,
  1077. sizeof (struct res_lib_cpg_iterationinitialize));
  1078. if (error != CS_OK) {
  1079. goto error_put_destroy;
  1080. }
  1081. cpg_iteration_instance->executive_iteration_handle =
  1082. res_lib_cpg_iterationinitialize.iteration_handle;
  1083. cpg_iteration_instance->cpg_iteration_handle = *cpg_iteration_handle;
  1084. list_add (&cpg_iteration_instance->list, &cpg_inst->iteration_list_head);
  1085. hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  1086. hdb_handle_put (&cpg_handle_t_db, handle);
  1087. return (res_lib_cpg_iterationinitialize.header.error);
  1088. error_put_destroy:
  1089. hdb_handle_put (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  1090. error_destroy:
  1091. hdb_handle_destroy (&cpg_iteration_handle_t_db, *cpg_iteration_handle);
  1092. error_put_cpg_db:
  1093. hdb_handle_put (&cpg_handle_t_db, handle);
  1094. return (error);
  1095. }
  1096. cs_error_t cpg_iteration_next(
  1097. cpg_iteration_handle_t handle,
  1098. struct cpg_iteration_description_t *description)
  1099. {
  1100. cs_error_t error;
  1101. struct cpg_iteration_instance_t *cpg_iteration_instance;
  1102. struct req_lib_cpg_iterationnext req_lib_cpg_iterationnext;
  1103. struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext;
  1104. if (description == NULL) {
  1105. return CS_ERR_INVALID_PARAM;
  1106. }
  1107. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle,
  1108. (void *)&cpg_iteration_instance));
  1109. if (error != CS_OK) {
  1110. goto error_exit;
  1111. }
  1112. req_lib_cpg_iterationnext.header.size = sizeof (struct req_lib_cpg_iterationnext);
  1113. req_lib_cpg_iterationnext.header.id = MESSAGE_REQ_CPG_ITERATIONNEXT;
  1114. req_lib_cpg_iterationnext.iteration_handle = cpg_iteration_instance->executive_iteration_handle;
  1115. error = qb_to_cs_error (qb_ipcc_send (cpg_iteration_instance->conn,
  1116. &req_lib_cpg_iterationnext,
  1117. req_lib_cpg_iterationnext.header.size));
  1118. if (error != CS_OK) {
  1119. goto error_put;
  1120. }
  1121. error = qb_to_cs_error (qb_ipcc_recv (cpg_iteration_instance->conn,
  1122. &res_lib_cpg_iterationnext,
  1123. sizeof(struct res_lib_cpg_iterationnext), -1));
  1124. if (error != CS_OK) {
  1125. goto error_put;
  1126. }
  1127. marshall_from_mar_cpg_iteration_description_t(
  1128. description,
  1129. &res_lib_cpg_iterationnext.description);
  1130. error = res_lib_cpg_iterationnext.header.error;
  1131. error_put:
  1132. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  1133. error_exit:
  1134. return (error);
  1135. }
  1136. cs_error_t cpg_iteration_finalize (
  1137. cpg_iteration_handle_t handle)
  1138. {
  1139. cs_error_t error;
  1140. struct iovec iov;
  1141. struct cpg_iteration_instance_t *cpg_iteration_instance;
  1142. struct req_lib_cpg_iterationfinalize req_lib_cpg_iterationfinalize;
  1143. struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize;
  1144. error = hdb_error_to_cs (hdb_handle_get (&cpg_iteration_handle_t_db, handle,
  1145. (void *)&cpg_iteration_instance));
  1146. if (error != CS_OK) {
  1147. goto error_exit;
  1148. }
  1149. req_lib_cpg_iterationfinalize.header.size = sizeof (struct req_lib_cpg_iterationfinalize);
  1150. req_lib_cpg_iterationfinalize.header.id = MESSAGE_REQ_CPG_ITERATIONFINALIZE;
  1151. req_lib_cpg_iterationfinalize.iteration_handle = cpg_iteration_instance->executive_iteration_handle;
  1152. iov.iov_base = (void *)&req_lib_cpg_iterationfinalize;
  1153. iov.iov_len = sizeof (struct req_lib_cpg_iterationfinalize);
  1154. error = coroipcc_msg_send_reply_receive (cpg_iteration_instance->conn,
  1155. &iov,
  1156. 1,
  1157. &res_lib_cpg_iterationfinalize,
  1158. sizeof (struct req_lib_cpg_iterationfinalize));
  1159. if (error != CS_OK) {
  1160. goto error_put;
  1161. }
  1162. cpg_iteration_instance_finalize (cpg_iteration_instance);
  1163. hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->cpg_iteration_handle);
  1164. return (res_lib_cpg_iterationfinalize.header.error);
  1165. error_put:
  1166. hdb_handle_put (&cpg_iteration_handle_t_db, handle);
  1167. error_exit:
  1168. return (error);
  1169. }
  1170. /** @} */