cpg.c 58 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106
  1. /*
  2. * Copyright (c) 2006-2009 Red Hat, Inc.
  3. *
  4. * All rights reserved.
  5. *
  6. * Author: Christine Caulfield (ccaulfie@redhat.com)
  7. * Author: Jan Friesse (jfriesse@redhat.com)
  8. *
  9. * This software licensed under BSD license, the text of which follows:
  10. *
  11. * Redistribution and use in source and binary forms, with or without
  12. * modification, are permitted provided that the following conditions are met:
  13. *
  14. * - Redistributions of source code must retain the above copyright notice,
  15. * this list of conditions and the following disclaimer.
  16. * - Redistributions in binary form must reproduce the above copyright notice,
  17. * this list of conditions and the following disclaimer in the documentation
  18. * and/or other materials provided with the distribution.
  19. * - Neither the name of the MontaVista Software, Inc. nor the names of its
  20. * contributors may be used to endorse or promote products derived from this
  21. * software without specific prior written permission.
  22. *
  23. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTIBUTORS "AS IS"
  24. * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  25. * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  26. * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  27. * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  28. * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  29. * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  30. * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  31. * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  32. * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  33. * THE POSSIBILITY OF SUCH DAMAGE.
  34. */
  35. #include <config.h>
  36. #ifdef HAVE_ALLOCA_H
  37. #include <alloca.h>
  38. #endif
  39. #include <sys/types.h>
  40. #include <sys/socket.h>
  41. #include <sys/un.h>
  42. #include <sys/ioctl.h>
  43. #include <netinet/in.h>
  44. #include <sys/uio.h>
  45. #include <unistd.h>
  46. #include <fcntl.h>
  47. #include <stdlib.h>
  48. #include <stdio.h>
  49. #include <errno.h>
  50. #include <time.h>
  51. #include <assert.h>
  52. #include <unistd.h>
  53. #include <netinet/in.h>
  54. #include <arpa/inet.h>
  55. #include <sys/mman.h>
  56. #include <qb/qbmap.h>
  57. #include <corosync/corotypes.h>
  58. #include <qb/qbipc_common.h>
  59. #include <corosync/corodefs.h>
  60. #include <corosync/list.h>
  61. #include <corosync/jhash.h>
  62. #include <corosync/lcr/lcr_comp.h>
  63. #include <corosync/logsys.h>
  64. #include <corosync/coroapi.h>
  65. #include <corosync/cpg.h>
  66. #include <corosync/ipc_cpg.h>
  67. LOGSYS_DECLARE_SUBSYS ("CPG");
  68. #define GROUP_HASH_SIZE 32
  69. enum cpg_message_req_types {
  70. MESSAGE_REQ_EXEC_CPG_PROCJOIN = 0,
  71. MESSAGE_REQ_EXEC_CPG_PROCLEAVE = 1,
  72. MESSAGE_REQ_EXEC_CPG_JOINLIST = 2,
  73. MESSAGE_REQ_EXEC_CPG_MCAST = 3,
  74. MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD = 4,
  75. MESSAGE_REQ_EXEC_CPG_DOWNLIST = 5
  76. };
  77. struct zcb_mapped {
  78. struct list_head list;
  79. void *addr;
  80. size_t size;
  81. };
  82. /*
  83. * state` exec deliver
  84. * match group name, pid -> if matched deliver for YES:
  85. * XXX indicates impossible state
  86. *
  87. * join leave mcast
  88. * UNJOINED XXX XXX NO
  89. * LEAVE_STARTED XXX YES(unjoined_enter) YES
  90. * JOIN_STARTED YES(join_started_enter) XXX NO
  91. * JOIN_COMPLETED XXX NO YES
  92. *
  93. * join_started_enter
  94. * set JOIN_COMPLETED
  95. * add entry to process_info list
  96. * unjoined_enter
  97. * set UNJOINED
  98. * delete entry from process_info list
  99. *
  100. *
  101. * library accept join error codes
  102. * UNJOINED YES(CS_OK) set JOIN_STARTED
  103. * LEAVE_STARTED NO(CS_ERR_BUSY)
  104. * JOIN_STARTED NO(CS_ERR_EXIST)
  105. * JOIN_COMPlETED NO(CS_ERR_EXIST)
  106. *
  107. * library accept leave error codes
  108. * UNJOINED NO(CS_ERR_NOT_EXIST)
  109. * LEAVE_STARTED NO(CS_ERR_NOT_EXIST)
  110. * JOIN_STARTED NO(CS_ERR_BUSY)
  111. * JOIN_COMPLETED YES(CS_OK) set LEAVE_STARTED
  112. *
  113. * library accept mcast
  114. * UNJOINED NO(CS_ERR_NOT_EXIST)
  115. * LEAVE_STARTED NO(CS_ERR_NOT_EXIST)
  116. * JOIN_STARTED YES(CS_OK)
  117. * JOIN_COMPLETED YES(CS_OK)
  118. */
  119. enum cpd_state {
  120. CPD_STATE_UNJOINED,
  121. CPD_STATE_LEAVE_STARTED,
  122. CPD_STATE_JOIN_STARTED,
  123. CPD_STATE_JOIN_COMPLETED
  124. };
  125. enum cpg_sync_state {
  126. CPGSYNC_DOWNLIST,
  127. CPGSYNC_JOINLIST
  128. };
  129. enum cpg_downlist_state_e {
  130. CPG_DOWNLIST_NONE,
  131. CPG_DOWNLIST_WAITING_FOR_MESSAGES,
  132. CPG_DOWNLIST_APPLYING,
  133. };
  134. static enum cpg_downlist_state_e downlist_state;
  135. static struct list_head downlist_messages_head;
  136. struct cpg_pd {
  137. void *conn;
  138. mar_cpg_name_t group_name;
  139. uint32_t pid;
  140. enum cpd_state cpd_state;
  141. unsigned int flags;
  142. int initial_totem_conf_sent;
  143. struct list_head list;
  144. struct list_head iteration_instance_list_head;
  145. struct list_head zcb_mapped_list_head;
  146. };
  147. struct cpg_iteration_instance {
  148. hdb_handle_t handle;
  149. struct list_head list;
  150. struct list_head items_list_head; /* List of process_info */
  151. struct list_head *current_pointer;
  152. };
  153. DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db,NULL);
  154. DECLARE_LIST_INIT(cpg_pd_list_head);
  155. static unsigned int my_member_list[PROCESSOR_COUNT_MAX];
  156. static unsigned int my_member_list_entries;
  157. static unsigned int my_old_member_list[PROCESSOR_COUNT_MAX];
  158. static unsigned int my_old_member_list_entries = 0;
  159. static struct corosync_api_v1 *api = NULL;
  160. static enum cpg_sync_state my_sync_state = CPGSYNC_DOWNLIST;
  161. static mar_cpg_ring_id_t last_sync_ring_id;
  162. struct process_info {
  163. unsigned int nodeid;
  164. uint32_t pid;
  165. mar_cpg_name_t group;
  166. struct list_head list; /* on the group_info members list */
  167. };
  168. DECLARE_LIST_INIT(process_info_list_head);
  169. struct join_list_entry {
  170. uint32_t pid;
  171. mar_cpg_name_t group_name;
  172. };
  173. /*
  174. * Service Interfaces required by service_message_handler struct
  175. */
  176. static int cpg_exec_init_fn (struct corosync_api_v1 *);
  177. static int cpg_lib_init_fn (void *conn);
  178. static int cpg_lib_exit_fn (void *conn);
  179. static void message_handler_req_exec_cpg_procjoin (
  180. const void *message,
  181. unsigned int nodeid);
  182. static void message_handler_req_exec_cpg_procleave (
  183. const void *message,
  184. unsigned int nodeid);
  185. static void message_handler_req_exec_cpg_joinlist (
  186. const void *message,
  187. unsigned int nodeid);
  188. static void message_handler_req_exec_cpg_mcast (
  189. const void *message,
  190. unsigned int nodeid);
  191. static void message_handler_req_exec_cpg_downlist_old (
  192. const void *message,
  193. unsigned int nodeid);
  194. static void message_handler_req_exec_cpg_downlist (
  195. const void *message,
  196. unsigned int nodeid);
  197. static void exec_cpg_procjoin_endian_convert (void *msg);
  198. static void exec_cpg_joinlist_endian_convert (void *msg);
  199. static void exec_cpg_mcast_endian_convert (void *msg);
  200. static void exec_cpg_downlist_endian_convert_old (void *msg);
  201. static void exec_cpg_downlist_endian_convert (void *msg);
  202. static void message_handler_req_lib_cpg_join (void *conn, const void *message);
  203. static void message_handler_req_lib_cpg_leave (void *conn, const void *message);
  204. static void message_handler_req_lib_cpg_finalize (void *conn, const void *message);
  205. static void message_handler_req_lib_cpg_mcast (void *conn, const void *message);
  206. static void message_handler_req_lib_cpg_membership (void *conn,
  207. const void *message);
  208. static void message_handler_req_lib_cpg_local_get (void *conn,
  209. const void *message);
  210. static void message_handler_req_lib_cpg_iteration_initialize (
  211. void *conn,
  212. const void *message);
  213. static void message_handler_req_lib_cpg_iteration_next (
  214. void *conn,
  215. const void *message);
  216. static void message_handler_req_lib_cpg_iteration_finalize (
  217. void *conn,
  218. const void *message);
  219. static void message_handler_req_lib_cpg_zc_alloc (
  220. void *conn,
  221. const void *message);
  222. static void message_handler_req_lib_cpg_zc_free (
  223. void *conn,
  224. const void *message);
  225. static void message_handler_req_lib_cpg_zc_execute (
  226. void *conn,
  227. const void *message);
  228. static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason);
  229. static int cpg_exec_send_downlist(void);
  230. static int cpg_exec_send_joinlist(void);
  231. static void downlist_messages_delete (void);
  232. static void downlist_master_choose_and_send (void);
  233. static void cpg_sync_init_v2 (
  234. const unsigned int *trans_list,
  235. size_t trans_list_entries,
  236. const unsigned int *member_list,
  237. size_t member_list_entries,
  238. const struct memb_ring_id *ring_id);
  239. static int cpg_sync_process (void);
  240. static void cpg_sync_activate (void);
  241. static void cpg_sync_abort (void);
  242. static int notify_lib_totem_membership (
  243. void *conn,
  244. int member_list_entries,
  245. const unsigned int *member_list);
  246. static inline int zcb_all_free (
  247. struct cpg_pd *cpd);
  248. /*
  249. * Library Handler Definition
  250. */
  251. static struct corosync_lib_handler cpg_lib_engine[] =
  252. {
  253. { /* 0 - MESSAGE_REQ_CPG_JOIN */
  254. .lib_handler_fn = message_handler_req_lib_cpg_join,
  255. .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
  256. },
  257. { /* 1 - MESSAGE_REQ_CPG_LEAVE */
  258. .lib_handler_fn = message_handler_req_lib_cpg_leave,
  259. .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
  260. },
  261. { /* 2 - MESSAGE_REQ_CPG_MCAST */
  262. .lib_handler_fn = message_handler_req_lib_cpg_mcast,
  263. .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
  264. },
  265. { /* 3 - MESSAGE_REQ_CPG_MEMBERSHIP */
  266. .lib_handler_fn = message_handler_req_lib_cpg_membership,
  267. .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
  268. },
  269. { /* 4 - MESSAGE_REQ_CPG_LOCAL_GET */
  270. .lib_handler_fn = message_handler_req_lib_cpg_local_get,
  271. .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
  272. },
  273. { /* 5 - MESSAGE_REQ_CPG_ITERATIONINITIALIZE */
  274. .lib_handler_fn = message_handler_req_lib_cpg_iteration_initialize,
  275. .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
  276. },
  277. { /* 6 - MESSAGE_REQ_CPG_ITERATIONNEXT */
  278. .lib_handler_fn = message_handler_req_lib_cpg_iteration_next,
  279. .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
  280. },
  281. { /* 7 - MESSAGE_REQ_CPG_ITERATIONFINALIZE */
  282. .lib_handler_fn = message_handler_req_lib_cpg_iteration_finalize,
  283. .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
  284. },
  285. { /* 8 - MESSAGE_REQ_CPG_FINALIZE */
  286. .lib_handler_fn = message_handler_req_lib_cpg_finalize,
  287. .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
  288. },
  289. { /* 9 */
  290. .lib_handler_fn = message_handler_req_lib_cpg_zc_alloc,
  291. .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
  292. },
  293. { /* 10 */
  294. .lib_handler_fn = message_handler_req_lib_cpg_zc_free,
  295. .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
  296. },
  297. { /* 11 */
  298. .lib_handler_fn = message_handler_req_lib_cpg_zc_execute,
  299. .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
  300. },
  301. };
  302. static struct corosync_exec_handler cpg_exec_engine[] =
  303. {
  304. { /* 0 - MESSAGE_REQ_EXEC_CPG_PROCJOIN */
  305. .exec_handler_fn = message_handler_req_exec_cpg_procjoin,
  306. .exec_endian_convert_fn = exec_cpg_procjoin_endian_convert
  307. },
  308. { /* 1 - MESSAGE_REQ_EXEC_CPG_PROCLEAVE */
  309. .exec_handler_fn = message_handler_req_exec_cpg_procleave,
  310. .exec_endian_convert_fn = exec_cpg_procjoin_endian_convert
  311. },
  312. { /* 2 - MESSAGE_REQ_EXEC_CPG_JOINLIST */
  313. .exec_handler_fn = message_handler_req_exec_cpg_joinlist,
  314. .exec_endian_convert_fn = exec_cpg_joinlist_endian_convert
  315. },
  316. { /* 3 - MESSAGE_REQ_EXEC_CPG_MCAST */
  317. .exec_handler_fn = message_handler_req_exec_cpg_mcast,
  318. .exec_endian_convert_fn = exec_cpg_mcast_endian_convert
  319. },
  320. { /* 4 - MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD */
  321. .exec_handler_fn = message_handler_req_exec_cpg_downlist_old,
  322. .exec_endian_convert_fn = exec_cpg_downlist_endian_convert_old
  323. },
  324. { /* 5 - MESSAGE_REQ_EXEC_CPG_DOWNLIST */
  325. .exec_handler_fn = message_handler_req_exec_cpg_downlist,
  326. .exec_endian_convert_fn = exec_cpg_downlist_endian_convert
  327. },
  328. };
  329. struct corosync_service_engine cpg_service_engine = {
  330. .name = "corosync cluster closed process group service v1.01",
  331. .id = CPG_SERVICE,
  332. .priority = 1,
  333. .private_data_size = sizeof (struct cpg_pd),
  334. .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED,
  335. .allow_inquorate = CS_LIB_ALLOW_INQUORATE,
  336. .lib_init_fn = cpg_lib_init_fn,
  337. .lib_exit_fn = cpg_lib_exit_fn,
  338. .lib_engine = cpg_lib_engine,
  339. .lib_engine_count = sizeof (cpg_lib_engine) / sizeof (struct corosync_lib_handler),
  340. .exec_init_fn = cpg_exec_init_fn,
  341. .exec_dump_fn = NULL,
  342. .exec_engine = cpg_exec_engine,
  343. .exec_engine_count = sizeof (cpg_exec_engine) / sizeof (struct corosync_exec_handler),
  344. .sync_mode = CS_SYNC_V1_APIV2,
  345. .sync_init = (sync_init_v1_fn_t)cpg_sync_init_v2,
  346. .sync_process = cpg_sync_process,
  347. .sync_activate = cpg_sync_activate,
  348. .sync_abort = cpg_sync_abort
  349. };
  350. /*
  351. * Dynamic loader definition
  352. */
  353. static struct corosync_service_engine *cpg_get_service_engine_ver0 (void);
  354. static struct corosync_service_engine_iface_ver0 cpg_service_engine_iface = {
  355. .corosync_get_service_engine_ver0 = cpg_get_service_engine_ver0
  356. };
  357. static struct lcr_iface corosync_cpg_ver0[1] = {
  358. {
  359. .name = "corosync_cpg",
  360. .version = 0,
  361. .versions_replace = 0,
  362. .versions_replace_count = 0,
  363. .dependencies = 0,
  364. .dependency_count = 0,
  365. .constructor = NULL,
  366. .destructor = NULL,
  367. .interfaces = NULL
  368. }
  369. };
  370. static struct lcr_comp cpg_comp_ver0 = {
  371. .iface_count = 1,
  372. .ifaces = corosync_cpg_ver0
  373. };
  374. static struct corosync_service_engine *cpg_get_service_engine_ver0 (void)
  375. {
  376. return (&cpg_service_engine);
  377. }
  378. #ifdef COROSYNC_SOLARIS
  379. void corosync_lcr_component_register (void);
  380. void corosync_lcr_component_register (void) {
  381. #else
  382. __attribute__ ((constructor)) static void corosync_lcr_component_register (void) {
  383. #endif
  384. lcr_interfaces_set (&corosync_cpg_ver0[0], &cpg_service_engine_iface);
  385. lcr_component_register (&cpg_comp_ver0);
  386. }
  387. struct req_exec_cpg_procjoin {
  388. struct qb_ipc_request_header header __attribute__((aligned(8)));
  389. mar_cpg_name_t group_name __attribute__((aligned(8)));
  390. mar_uint32_t pid __attribute__((aligned(8)));
  391. mar_uint32_t reason __attribute__((aligned(8)));
  392. };
  393. struct req_exec_cpg_mcast {
  394. struct qb_ipc_request_header header __attribute__((aligned(8)));
  395. mar_cpg_name_t group_name __attribute__((aligned(8)));
  396. mar_uint32_t msglen __attribute__((aligned(8)));
  397. mar_uint32_t pid __attribute__((aligned(8)));
  398. mar_message_source_t source __attribute__((aligned(8)));
  399. mar_uint8_t message[] __attribute__((aligned(8)));
  400. };
  401. struct req_exec_cpg_downlist_old {
  402. struct qb_ipc_request_header header __attribute__((aligned(8)));
  403. mar_uint32_t left_nodes __attribute__((aligned(8)));
  404. mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)));
  405. };
  406. struct req_exec_cpg_downlist {
  407. struct qb_ipc_request_header header __attribute__((aligned(8)));
  408. /* merge decisions */
  409. mar_uint32_t old_members __attribute__((aligned(8)));
  410. /* downlist below */
  411. mar_uint32_t left_nodes __attribute__((aligned(8)));
  412. mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)));
  413. };
  414. struct downlist_msg {
  415. mar_uint32_t sender_nodeid;
  416. mar_uint32_t old_members __attribute__((aligned(8)));
  417. mar_uint32_t left_nodes __attribute__((aligned(8)));
  418. mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)));
  419. struct list_head list;
  420. };
  421. static struct req_exec_cpg_downlist g_req_exec_cpg_downlist;
  422. static void cpg_sync_init_v2 (
  423. const unsigned int *trans_list,
  424. size_t trans_list_entries,
  425. const unsigned int *member_list,
  426. size_t member_list_entries,
  427. const struct memb_ring_id *ring_id)
  428. {
  429. int entries;
  430. int i, j;
  431. int found;
  432. my_sync_state = CPGSYNC_DOWNLIST;
  433. memcpy (my_member_list, member_list, member_list_entries *
  434. sizeof (unsigned int));
  435. my_member_list_entries = member_list_entries;
  436. last_sync_ring_id.nodeid = ring_id->rep.nodeid;
  437. last_sync_ring_id.seq = ring_id->seq;
  438. downlist_state = CPG_DOWNLIST_WAITING_FOR_MESSAGES;
  439. entries = 0;
  440. /*
  441. * Determine list of nodeids for downlist message
  442. */
  443. for (i = 0; i < my_old_member_list_entries; i++) {
  444. found = 0;
  445. for (j = 0; j < trans_list_entries; j++) {
  446. if (my_old_member_list[i] == trans_list[j]) {
  447. found = 1;
  448. break;
  449. }
  450. }
  451. if (found == 0) {
  452. g_req_exec_cpg_downlist.nodeids[entries++] =
  453. my_old_member_list[i];
  454. }
  455. }
  456. g_req_exec_cpg_downlist.left_nodes = entries;
  457. }
  458. static int cpg_sync_process (void)
  459. {
  460. int res = -1;
  461. if (my_sync_state == CPGSYNC_DOWNLIST) {
  462. res = cpg_exec_send_downlist();
  463. if (res == -1) {
  464. return (-1);
  465. }
  466. my_sync_state = CPGSYNC_JOINLIST;
  467. }
  468. if (my_sync_state == CPGSYNC_JOINLIST) {
  469. res = cpg_exec_send_joinlist();
  470. }
  471. return (res);
  472. }
  473. static void cpg_sync_activate (void)
  474. {
  475. memcpy (my_old_member_list, my_member_list,
  476. my_member_list_entries * sizeof (unsigned int));
  477. my_old_member_list_entries = my_member_list_entries;
  478. if (downlist_state == CPG_DOWNLIST_WAITING_FOR_MESSAGES) {
  479. downlist_master_choose_and_send ();
  480. }
  481. downlist_messages_delete ();
  482. downlist_state = CPG_DOWNLIST_NONE;
  483. notify_lib_totem_membership (NULL, my_member_list_entries, my_member_list);
  484. }
  485. static void cpg_sync_abort (void)
  486. {
  487. downlist_state = CPG_DOWNLIST_NONE;
  488. downlist_messages_delete ();
  489. }
  490. static int notify_lib_totem_membership (
  491. void *conn,
  492. int member_list_entries,
  493. const unsigned int *member_list)
  494. {
  495. struct list_head *iter;
  496. char *buf;
  497. int size;
  498. struct res_lib_cpg_totem_confchg_callback *res;
  499. size = sizeof(struct res_lib_cpg_totem_confchg_callback) +
  500. sizeof(mar_uint32_t) * (member_list_entries);
  501. buf = alloca(size);
  502. if (!buf)
  503. return CS_ERR_LIBRARY;
  504. res = (struct res_lib_cpg_totem_confchg_callback *)buf;
  505. res->member_list_entries = member_list_entries;
  506. res->header.size = size;
  507. res->header.id = MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK;
  508. res->header.error = CS_OK;
  509. memcpy (&res->ring_id, &last_sync_ring_id, sizeof (mar_cpg_ring_id_t));
  510. memcpy (res->member_list, member_list, res->member_list_entries * sizeof (mar_uint32_t));
  511. if (conn == NULL) {
  512. for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) {
  513. struct cpg_pd *cpg_pd = list_entry (iter, struct cpg_pd, list);
  514. api->ipc_dispatch_send (cpg_pd->conn, buf, size);
  515. }
  516. } else {
  517. api->ipc_dispatch_send (conn, buf, size);
  518. }
  519. return CS_OK;
  520. }
  521. static int notify_lib_joinlist(
  522. const mar_cpg_name_t *group_name,
  523. void *conn,
  524. int joined_list_entries,
  525. mar_cpg_address_t *joined_list,
  526. int left_list_entries,
  527. mar_cpg_address_t *left_list,
  528. int id)
  529. {
  530. int size;
  531. char *buf;
  532. struct list_head *iter;
  533. int count;
  534. struct res_lib_cpg_confchg_callback *res;
  535. mar_cpg_address_t *retgi;
  536. count = 0;
  537. for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
  538. struct process_info *pi = list_entry (iter, struct process_info, list);
  539. if (mar_name_compare (&pi->group, group_name) == 0) {
  540. int i;
  541. int founded = 0;
  542. for (i = 0; i < left_list_entries; i++) {
  543. if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) {
  544. founded++;
  545. }
  546. }
  547. if (!founded)
  548. count++;
  549. }
  550. }
  551. size = sizeof(struct res_lib_cpg_confchg_callback) +
  552. sizeof(mar_cpg_address_t) * (count + left_list_entries + joined_list_entries);
  553. buf = alloca(size);
  554. if (!buf)
  555. return CS_ERR_LIBRARY;
  556. res = (struct res_lib_cpg_confchg_callback *)buf;
  557. res->joined_list_entries = joined_list_entries;
  558. res->left_list_entries = left_list_entries;
  559. res->member_list_entries = count;
  560. retgi = res->member_list;
  561. res->header.size = size;
  562. res->header.id = id;
  563. res->header.error = CS_OK;
  564. memcpy(&res->group_name, group_name, sizeof(mar_cpg_name_t));
  565. for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
  566. struct process_info *pi=list_entry (iter, struct process_info, list);
  567. if (mar_name_compare (&pi->group, group_name) == 0) {
  568. int i;
  569. int founded = 0;
  570. for (i = 0;i < left_list_entries; i++) {
  571. if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) {
  572. founded++;
  573. }
  574. }
  575. if (!founded) {
  576. retgi->nodeid = pi->nodeid;
  577. retgi->pid = pi->pid;
  578. retgi++;
  579. }
  580. }
  581. }
  582. if (left_list_entries) {
  583. memcpy (retgi, left_list, left_list_entries * sizeof(mar_cpg_address_t));
  584. retgi += left_list_entries;
  585. }
  586. if (joined_list_entries) {
  587. memcpy (retgi, joined_list, joined_list_entries * sizeof(mar_cpg_address_t));
  588. retgi += joined_list_entries;
  589. }
  590. if (conn) {
  591. api->ipc_dispatch_send (conn, buf, size);
  592. } else {
  593. for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) {
  594. struct cpg_pd *cpd = list_entry (iter, struct cpg_pd, list);
  595. if (mar_name_compare (&cpd->group_name, group_name) == 0) {
  596. assert (joined_list_entries <= 1);
  597. if (joined_list_entries) {
  598. if (joined_list[0].pid == cpd->pid &&
  599. joined_list[0].nodeid == api->totem_nodeid_get()) {
  600. cpd->cpd_state = CPD_STATE_JOIN_COMPLETED;
  601. }
  602. }
  603. if (cpd->cpd_state == CPD_STATE_JOIN_COMPLETED ||
  604. cpd->cpd_state == CPD_STATE_LEAVE_STARTED) {
  605. api->ipc_dispatch_send (cpd->conn, buf, size);
  606. }
  607. if (left_list_entries) {
  608. if (left_list[0].pid == cpd->pid &&
  609. left_list[0].nodeid == api->totem_nodeid_get() &&
  610. left_list[0].reason == CONFCHG_CPG_REASON_LEAVE) {
  611. cpd->pid = 0;
  612. memset (&cpd->group_name, 0, sizeof(cpd->group_name));
  613. cpd->cpd_state = CPD_STATE_UNJOINED;
  614. }
  615. }
  616. }
  617. }
  618. }
  619. /*
  620. * Traverse thru cpds and send totem membership for cpd, where it is not send yet
  621. */
  622. for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) {
  623. struct cpg_pd *cpd = list_entry (iter, struct cpg_pd, list);
  624. if ((cpd->flags & CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF) && (cpd->initial_totem_conf_sent == 0)) {
  625. cpd->initial_totem_conf_sent = 1;
  626. notify_lib_totem_membership (cpd->conn, my_old_member_list_entries, my_old_member_list);
  627. }
  628. }
  629. return CS_OK;
  630. }
  631. static void downlist_log(const char *msg, struct downlist_msg* dl)
  632. {
  633. log_printf (LOG_DEBUG,
  634. "%s: sender %s; members(old:%d left:%d)",
  635. msg,
  636. api->totem_ifaces_print(dl->sender_nodeid),
  637. dl->old_members,
  638. dl->left_nodes);
  639. }
  640. static struct downlist_msg* downlist_master_choose (void)
  641. {
  642. struct downlist_msg *cmp;
  643. struct downlist_msg *best = NULL;
  644. struct list_head *iter;
  645. uint32_t cmp_members;
  646. uint32_t best_members;
  647. for (iter = downlist_messages_head.next;
  648. iter != &downlist_messages_head;
  649. iter = iter->next) {
  650. cmp = list_entry(iter, struct downlist_msg, list);
  651. downlist_log("comparing", cmp);
  652. if (best == NULL) {
  653. best = cmp;
  654. continue;
  655. }
  656. best_members = best->old_members - best->left_nodes;
  657. cmp_members = cmp->old_members - cmp->left_nodes;
  658. if (cmp_members < best_members) {
  659. continue;
  660. }
  661. else if (cmp_members > best_members) {
  662. best = cmp;
  663. }
  664. else if (cmp->sender_nodeid < best->sender_nodeid) {
  665. best = cmp;
  666. }
  667. }
  668. return best;
  669. }
  670. static void downlist_master_choose_and_send (void)
  671. {
  672. struct downlist_msg *stored_msg;
  673. struct list_head *iter;
  674. struct process_info *left_pi;
  675. qb_map_t *group_map;
  676. struct cpg_name cpg_group;
  677. mar_cpg_name_t group;
  678. struct confchg_data{
  679. struct cpg_name cpg_group;
  680. mar_cpg_address_t left_list[CPG_MEMBERS_MAX];
  681. int left_list_entries;
  682. struct list_head list;
  683. } *pcd;
  684. qb_map_iter_t *miter;
  685. int i, size;
  686. downlist_state = CPG_DOWNLIST_APPLYING;
  687. stored_msg = downlist_master_choose ();
  688. if (!stored_msg) {
  689. log_printf (LOGSYS_LEVEL_DEBUG, "NO chosen downlist");
  690. return;
  691. }
  692. downlist_log("chosen downlist", stored_msg);
  693. group_map = qb_skiplist_create();
  694. /*
  695. * only the cpg groups included in left nodes should receive
  696. * confchg event, so we will collect these cpg groups and
  697. * relative left_lists here.
  698. */
  699. for (iter = process_info_list_head.next; iter != &process_info_list_head; ) {
  700. struct process_info *pi = list_entry(iter, struct process_info, list);
  701. iter = iter->next;
  702. left_pi = NULL;
  703. for (i = 0; i < stored_msg->left_nodes; i++) {
  704. if (pi->nodeid == stored_msg->nodeids[i]) {
  705. left_pi = pi;
  706. break;
  707. }
  708. }
  709. if (left_pi) {
  710. marshall_from_mar_cpg_name_t(&cpg_group, &left_pi->group);
  711. cpg_group.value[cpg_group.length] = 0;
  712. pcd = (struct confchg_data *)qb_map_get(group_map, cpg_group.value);
  713. if (pcd == NULL) {
  714. pcd = (struct confchg_data *)calloc(1, sizeof(struct confchg_data));
  715. memcpy(&pcd->cpg_group, &cpg_group, sizeof(struct cpg_name));
  716. qb_map_put(group_map, pcd->cpg_group.value, pcd);
  717. }
  718. size = pcd->left_list_entries;
  719. pcd->left_list[size].nodeid = left_pi->nodeid;
  720. pcd->left_list[size].pid = left_pi->pid;
  721. pcd->left_list[size].reason = CONFCHG_CPG_REASON_NODEDOWN;
  722. pcd->left_list_entries++;
  723. list_del (&left_pi->list);
  724. free (left_pi);
  725. }
  726. }
  727. /* send only one confchg event per cpg group */
  728. miter = qb_map_iter_create(group_map);
  729. while (qb_map_iter_next(miter, (void **)&pcd)) {
  730. marshall_to_mar_cpg_name_t(&group, &pcd->cpg_group);
  731. log_printf (LOG_DEBUG, "left_list_entries:%d", pcd->left_list_entries);
  732. for (i=0; i<pcd->left_list_entries; i++) {
  733. log_printf (LOG_DEBUG, "left_list[%d] group:%d, ip:%s, pid:%d",
  734. i, pcd->cpg_group.value,
  735. (char*)api->totem_ifaces_print(pcd->left_list[i].nodeid),
  736. pcd->left_list[i].pid);
  737. }
  738. /* send confchg event */
  739. notify_lib_joinlist(&group, NULL,
  740. 0, NULL,
  741. pcd->left_list_entries,
  742. pcd->left_list,
  743. MESSAGE_RES_CPG_CONFCHG_CALLBACK);
  744. free(pcd);
  745. }
  746. qb_map_iter_free(miter);
  747. qb_map_destroy(group_map);
  748. }
  749. static void downlist_messages_delete (void)
  750. {
  751. struct downlist_msg *stored_msg;
  752. struct list_head *iter, *iter_next;
  753. for (iter = downlist_messages_head.next;
  754. iter != &downlist_messages_head;
  755. iter = iter_next) {
  756. iter_next = iter->next;
  757. stored_msg = list_entry(iter, struct downlist_msg, list);
  758. list_del (&stored_msg->list);
  759. free (stored_msg);
  760. }
  761. }
  762. static int cpg_exec_init_fn (struct corosync_api_v1 *corosync_api)
  763. {
  764. #ifdef COROSYNC_SOLARIS
  765. logsys_subsys_init();
  766. #endif
  767. list_init (&downlist_messages_head);
  768. api = corosync_api;
  769. return (0);
  770. }
  771. static void cpg_iteration_instance_finalize (struct cpg_iteration_instance *cpg_iteration_instance)
  772. {
  773. struct list_head *iter, *iter_next;
  774. struct process_info *pi;
  775. for (iter = cpg_iteration_instance->items_list_head.next;
  776. iter != &cpg_iteration_instance->items_list_head;
  777. iter = iter_next) {
  778. iter_next = iter->next;
  779. pi = list_entry (iter, struct process_info, list);
  780. list_del (&pi->list);
  781. free (pi);
  782. }
  783. list_del (&cpg_iteration_instance->list);
  784. hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle);
  785. }
  786. static void cpg_pd_finalize (struct cpg_pd *cpd)
  787. {
  788. struct list_head *iter, *iter_next;
  789. struct cpg_iteration_instance *cpii;
  790. zcb_all_free(cpd);
  791. for (iter = cpd->iteration_instance_list_head.next;
  792. iter != &cpd->iteration_instance_list_head;
  793. iter = iter_next) {
  794. iter_next = iter->next;
  795. cpii = list_entry (iter, struct cpg_iteration_instance, list);
  796. cpg_iteration_instance_finalize (cpii);
  797. }
  798. list_del (&cpd->list);
  799. }
  800. static int cpg_lib_exit_fn (void *conn)
  801. {
  802. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  803. log_printf(LOGSYS_LEVEL_DEBUG, "exit_fn for conn=%p\n", conn);
  804. if (cpd->group_name.length > 0) {
  805. cpg_node_joinleave_send (cpd->pid, &cpd->group_name,
  806. MESSAGE_REQ_EXEC_CPG_PROCLEAVE, CONFCHG_CPG_REASON_PROCDOWN);
  807. }
  808. cpg_pd_finalize (cpd);
  809. api->ipc_refcnt_dec (conn);
  810. return (0);
  811. }
  812. static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason)
  813. {
  814. struct req_exec_cpg_procjoin req_exec_cpg_procjoin;
  815. struct iovec req_exec_cpg_iovec;
  816. int result;
  817. memcpy(&req_exec_cpg_procjoin.group_name, group_name, sizeof(mar_cpg_name_t));
  818. req_exec_cpg_procjoin.pid = pid;
  819. req_exec_cpg_procjoin.reason = reason;
  820. req_exec_cpg_procjoin.header.size = sizeof(req_exec_cpg_procjoin);
  821. req_exec_cpg_procjoin.header.id = SERVICE_ID_MAKE(CPG_SERVICE, fn);
  822. req_exec_cpg_iovec.iov_base = (char *)&req_exec_cpg_procjoin;
  823. req_exec_cpg_iovec.iov_len = sizeof(req_exec_cpg_procjoin);
  824. result = api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED);
  825. return (result);
  826. }
  827. /* Can byteswap join & leave messages */
  828. static void exec_cpg_procjoin_endian_convert (void *msg)
  829. {
  830. struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = msg;
  831. req_exec_cpg_procjoin->pid = swab32(req_exec_cpg_procjoin->pid);
  832. swab_mar_cpg_name_t (&req_exec_cpg_procjoin->group_name);
  833. req_exec_cpg_procjoin->reason = swab32(req_exec_cpg_procjoin->reason);
  834. }
  835. static void exec_cpg_joinlist_endian_convert (void *msg_v)
  836. {
  837. char *msg = msg_v;
  838. struct qb_ipc_response_header *res = (struct qb_ipc_response_header *)msg;
  839. struct join_list_entry *jle = (struct join_list_entry *)(msg + sizeof(struct qb_ipc_response_header));
  840. swab_mar_int32_t (&res->size);
  841. while ((const char*)jle < msg + res->size) {
  842. jle->pid = swab32(jle->pid);
  843. swab_mar_cpg_name_t (&jle->group_name);
  844. jle++;
  845. }
  846. }
  847. static void exec_cpg_downlist_endian_convert_old (void *msg)
  848. {
  849. }
  850. static void exec_cpg_downlist_endian_convert (void *msg)
  851. {
  852. struct req_exec_cpg_downlist *req_exec_cpg_downlist = msg;
  853. unsigned int i;
  854. req_exec_cpg_downlist->left_nodes = swab32(req_exec_cpg_downlist->left_nodes);
  855. req_exec_cpg_downlist->old_members = swab32(req_exec_cpg_downlist->old_members);
  856. for (i = 0; i < req_exec_cpg_downlist->left_nodes; i++) {
  857. req_exec_cpg_downlist->nodeids[i] = swab32(req_exec_cpg_downlist->nodeids[i]);
  858. }
  859. }
  860. static void exec_cpg_mcast_endian_convert (void *msg)
  861. {
  862. struct req_exec_cpg_mcast *req_exec_cpg_mcast = msg;
  863. swab_coroipc_request_header_t (&req_exec_cpg_mcast->header);
  864. swab_mar_cpg_name_t (&req_exec_cpg_mcast->group_name);
  865. req_exec_cpg_mcast->pid = swab32(req_exec_cpg_mcast->pid);
  866. req_exec_cpg_mcast->msglen = swab32(req_exec_cpg_mcast->msglen);
  867. swab_mar_message_source_t (&req_exec_cpg_mcast->source);
  868. }
  869. static struct process_info *process_info_find(const mar_cpg_name_t *group_name, uint32_t pid, unsigned int nodeid) {
  870. struct list_head *iter;
  871. for (iter = process_info_list_head.next; iter != &process_info_list_head; ) {
  872. struct process_info *pi = list_entry (iter, struct process_info, list);
  873. iter = iter->next;
  874. if (pi->pid == pid && pi->nodeid == nodeid &&
  875. mar_name_compare (&pi->group, group_name) == 0) {
  876. return pi;
  877. }
  878. }
  879. return NULL;
  880. }
  881. static void do_proc_join(
  882. const mar_cpg_name_t *name,
  883. uint32_t pid,
  884. unsigned int nodeid,
  885. int reason)
  886. {
  887. struct process_info *pi;
  888. struct process_info *pi_entry;
  889. mar_cpg_address_t notify_info;
  890. struct list_head *list;
  891. struct list_head *list_to_add = NULL;
  892. if (process_info_find (name, pid, nodeid) != NULL) {
  893. return ;
  894. }
  895. pi = malloc (sizeof (struct process_info));
  896. if (!pi) {
  897. log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct");
  898. return;
  899. }
  900. pi->nodeid = nodeid;
  901. pi->pid = pid;
  902. memcpy(&pi->group, name, sizeof(*name));
  903. list_init(&pi->list);
  904. /*
  905. * Insert new process in sorted order so synchronization works properly
  906. */
  907. list_to_add = &process_info_list_head;
  908. for (list = process_info_list_head.next; list != &process_info_list_head; list = list->next) {
  909. pi_entry = list_entry(list, struct process_info, list);
  910. if (pi_entry->nodeid > pi->nodeid ||
  911. (pi_entry->nodeid == pi->nodeid && pi_entry->pid > pi->pid)) {
  912. break;
  913. }
  914. list_to_add = list;
  915. }
  916. list_add (&pi->list, list_to_add);
  917. notify_info.pid = pi->pid;
  918. notify_info.nodeid = nodeid;
  919. notify_info.reason = reason;
  920. notify_lib_joinlist(&pi->group, NULL,
  921. 1, &notify_info,
  922. 0, NULL,
  923. MESSAGE_RES_CPG_CONFCHG_CALLBACK);
  924. }
  925. static void message_handler_req_exec_cpg_downlist_old (
  926. const void *message,
  927. unsigned int nodeid)
  928. {
  929. log_printf (LOGSYS_LEVEL_WARNING, "downlist OLD from node %d",
  930. nodeid);
  931. }
  932. static void message_handler_req_exec_cpg_downlist(
  933. const void *message,
  934. unsigned int nodeid)
  935. {
  936. const struct req_exec_cpg_downlist *req_exec_cpg_downlist = message;
  937. int i;
  938. struct list_head *iter;
  939. struct downlist_msg *stored_msg;
  940. int found;
  941. if (downlist_state != CPG_DOWNLIST_WAITING_FOR_MESSAGES) {
  942. log_printf (LOGSYS_LEVEL_WARNING, "downlist left_list: %d received in state %d",
  943. req_exec_cpg_downlist->left_nodes, downlist_state);
  944. return;
  945. }
  946. stored_msg = malloc (sizeof (struct downlist_msg));
  947. stored_msg->sender_nodeid = nodeid;
  948. stored_msg->old_members = req_exec_cpg_downlist->old_members;
  949. stored_msg->left_nodes = req_exec_cpg_downlist->left_nodes;
  950. memcpy (stored_msg->nodeids, req_exec_cpg_downlist->nodeids,
  951. req_exec_cpg_downlist->left_nodes * sizeof (mar_uint32_t));
  952. list_init (&stored_msg->list);
  953. list_add (&stored_msg->list, &downlist_messages_head);
  954. for (i = 0; i < my_member_list_entries; i++) {
  955. found = 0;
  956. for (iter = downlist_messages_head.next;
  957. iter != &downlist_messages_head;
  958. iter = iter->next) {
  959. stored_msg = list_entry(iter, struct downlist_msg, list);
  960. if (my_member_list[i] == stored_msg->sender_nodeid) {
  961. found = 1;
  962. }
  963. }
  964. if (!found) {
  965. return;
  966. }
  967. }
  968. downlist_master_choose_and_send ();
  969. }
  970. static void message_handler_req_exec_cpg_procjoin (
  971. const void *message,
  972. unsigned int nodeid)
  973. {
  974. const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message;
  975. log_printf(LOGSYS_LEVEL_DEBUG, "got procjoin message from cluster node %d\n", nodeid);
  976. do_proc_join (&req_exec_cpg_procjoin->group_name,
  977. req_exec_cpg_procjoin->pid, nodeid,
  978. CONFCHG_CPG_REASON_JOIN);
  979. }
  980. static void message_handler_req_exec_cpg_procleave (
  981. const void *message,
  982. unsigned int nodeid)
  983. {
  984. const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message;
  985. struct process_info *pi;
  986. struct list_head *iter;
  987. mar_cpg_address_t notify_info;
  988. log_printf(LOGSYS_LEVEL_DEBUG, "got procleave message from cluster node %d\n", nodeid);
  989. notify_info.pid = req_exec_cpg_procjoin->pid;
  990. notify_info.nodeid = nodeid;
  991. notify_info.reason = req_exec_cpg_procjoin->reason;
  992. notify_lib_joinlist(&req_exec_cpg_procjoin->group_name, NULL,
  993. 0, NULL,
  994. 1, &notify_info,
  995. MESSAGE_RES_CPG_CONFCHG_CALLBACK);
  996. for (iter = process_info_list_head.next; iter != &process_info_list_head; ) {
  997. pi = list_entry(iter, struct process_info, list);
  998. iter = iter->next;
  999. if (pi->pid == req_exec_cpg_procjoin->pid && pi->nodeid == nodeid &&
  1000. mar_name_compare (&pi->group, &req_exec_cpg_procjoin->group_name)==0) {
  1001. list_del (&pi->list);
  1002. free (pi);
  1003. }
  1004. }
  1005. }
  1006. /* Got a proclist from another node */
  1007. static void message_handler_req_exec_cpg_joinlist (
  1008. const void *message_v,
  1009. unsigned int nodeid)
  1010. {
  1011. const char *message = message_v;
  1012. const struct qb_ipc_response_header *res = (const struct qb_ipc_response_header *)message;
  1013. const struct join_list_entry *jle = (const struct join_list_entry *)(message + sizeof(struct qb_ipc_response_header));
  1014. log_printf(LOGSYS_LEVEL_DEBUG, "got joinlist message from node %x\n",
  1015. nodeid);
  1016. /* Ignore our own messages */
  1017. if (nodeid == api->totem_nodeid_get()) {
  1018. return;
  1019. }
  1020. while ((const char*)jle < message + res->size) {
  1021. do_proc_join (&jle->group_name, jle->pid, nodeid,
  1022. CONFCHG_CPG_REASON_NODEUP);
  1023. jle++;
  1024. }
  1025. }
  1026. static void message_handler_req_exec_cpg_mcast (
  1027. const void *message,
  1028. unsigned int nodeid)
  1029. {
  1030. const struct req_exec_cpg_mcast *req_exec_cpg_mcast = message;
  1031. struct res_lib_cpg_deliver_callback res_lib_cpg_mcast;
  1032. int msglen = req_exec_cpg_mcast->msglen;
  1033. struct list_head *iter, *pi_iter;
  1034. struct cpg_pd *cpd;
  1035. struct iovec iovec[2];
  1036. int known_node = 0;
  1037. res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_DELIVER_CALLBACK;
  1038. res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast) + msglen;
  1039. res_lib_cpg_mcast.msglen = msglen;
  1040. res_lib_cpg_mcast.pid = req_exec_cpg_mcast->pid;
  1041. res_lib_cpg_mcast.nodeid = nodeid;
  1042. memcpy(&res_lib_cpg_mcast.group_name, &req_exec_cpg_mcast->group_name,
  1043. sizeof(mar_cpg_name_t));
  1044. iovec[0].iov_base = (void *)&res_lib_cpg_mcast;
  1045. iovec[0].iov_len = sizeof (res_lib_cpg_mcast);
  1046. iovec[1].iov_base = (char*)message+sizeof(*req_exec_cpg_mcast);
  1047. iovec[1].iov_len = msglen;
  1048. for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; ) {
  1049. cpd = list_entry(iter, struct cpg_pd, list);
  1050. iter = iter->next;
  1051. if ((cpd->cpd_state == CPD_STATE_LEAVE_STARTED || cpd->cpd_state == CPD_STATE_JOIN_COMPLETED)
  1052. && (mar_name_compare (&cpd->group_name, &req_exec_cpg_mcast->group_name) == 0)) {
  1053. if (!known_node) {
  1054. /* Try to find, if we know the node */
  1055. for (pi_iter = process_info_list_head.next;
  1056. pi_iter != &process_info_list_head; pi_iter = pi_iter->next) {
  1057. struct process_info *pi = list_entry (pi_iter, struct process_info, list);
  1058. if (pi->nodeid == nodeid &&
  1059. mar_name_compare (&pi->group, &req_exec_cpg_mcast->group_name) == 0) {
  1060. known_node = 1;
  1061. break;
  1062. }
  1063. }
  1064. }
  1065. if (!known_node) {
  1066. log_printf(LOGSYS_LEVEL_WARNING, "Unknown node -> we will not deliver message");
  1067. return ;
  1068. }
  1069. api->ipc_dispatch_iov_send (cpd->conn, iovec, 2);
  1070. }
  1071. }
  1072. }
  1073. static int cpg_exec_send_downlist(void)
  1074. {
  1075. struct iovec iov;
  1076. g_req_exec_cpg_downlist.header.id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_DOWNLIST);
  1077. g_req_exec_cpg_downlist.header.size = sizeof(struct req_exec_cpg_downlist);
  1078. g_req_exec_cpg_downlist.old_members = my_old_member_list_entries;
  1079. iov.iov_base = (void *)&g_req_exec_cpg_downlist;
  1080. iov.iov_len = g_req_exec_cpg_downlist.header.size;
  1081. return (api->totem_mcast (&iov, 1, TOTEM_AGREED));
  1082. }
  1083. static int cpg_exec_send_joinlist(void)
  1084. {
  1085. int count = 0;
  1086. struct list_head *iter;
  1087. struct qb_ipc_response_header *res;
  1088. char *buf;
  1089. struct join_list_entry *jle;
  1090. struct iovec req_exec_cpg_iovec;
  1091. for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
  1092. struct process_info *pi = list_entry (iter, struct process_info, list);
  1093. if (pi->nodeid == api->totem_nodeid_get ()) {
  1094. count++;
  1095. }
  1096. }
  1097. /* Nothing to send */
  1098. if (!count)
  1099. return 0;
  1100. buf = alloca(sizeof(struct qb_ipc_response_header) + sizeof(struct join_list_entry) * count);
  1101. if (!buf) {
  1102. log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate joinlist buffer");
  1103. return -1;
  1104. }
  1105. jle = (struct join_list_entry *)(buf + sizeof(struct qb_ipc_response_header));
  1106. res = (struct qb_ipc_response_header *)buf;
  1107. for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
  1108. struct process_info *pi = list_entry (iter, struct process_info, list);
  1109. if (pi->nodeid == api->totem_nodeid_get ()) {
  1110. memcpy (&jle->group_name, &pi->group, sizeof (mar_cpg_name_t));
  1111. jle->pid = pi->pid;
  1112. jle++;
  1113. }
  1114. }
  1115. res->id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_JOINLIST);
  1116. res->size = sizeof(struct qb_ipc_response_header)+sizeof(struct join_list_entry) * count;
  1117. req_exec_cpg_iovec.iov_base = buf;
  1118. req_exec_cpg_iovec.iov_len = res->size;
  1119. return (api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED));
  1120. }
  1121. static int cpg_lib_init_fn (void *conn)
  1122. {
  1123. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1124. memset (cpd, 0, sizeof(struct cpg_pd));
  1125. cpd->conn = conn;
  1126. list_add (&cpd->list, &cpg_pd_list_head);
  1127. list_init (&cpd->iteration_instance_list_head);
  1128. list_init (&cpd->zcb_mapped_list_head);
  1129. api->ipc_refcnt_inc (conn);
  1130. log_printf(LOGSYS_LEVEL_DEBUG, "lib_init_fn: conn=%p, cpd=%p\n", conn, cpd);
  1131. return (0);
  1132. }
  1133. /* Join message from the library */
  1134. static void message_handler_req_lib_cpg_join (void *conn, const void *message)
  1135. {
  1136. const struct req_lib_cpg_join *req_lib_cpg_join = message;
  1137. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1138. struct res_lib_cpg_join res_lib_cpg_join;
  1139. cs_error_t error = CS_OK;
  1140. struct list_head *iter;
  1141. /* Test, if we don't have same pid and group name joined */
  1142. for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) {
  1143. struct cpg_pd *cpd_item = list_entry (iter, struct cpg_pd, list);
  1144. if (cpd_item->pid == req_lib_cpg_join->pid &&
  1145. mar_name_compare(&req_lib_cpg_join->group_name, &cpd_item->group_name) == 0) {
  1146. /* We have same pid and group name joined -> return error */
  1147. error = CS_ERR_EXIST;
  1148. goto response_send;
  1149. }
  1150. }
  1151. /*
  1152. * Same check must be done in process info list, because there may be not yet delivered
  1153. * leave of client.
  1154. */
  1155. for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
  1156. struct process_info *pi = list_entry (iter, struct process_info, list);
  1157. if (pi->nodeid == api->totem_nodeid_get () && pi->pid == req_lib_cpg_join->pid &&
  1158. mar_name_compare(&req_lib_cpg_join->group_name, &pi->group) == 0) {
  1159. /* We have same pid and group name joined -> return error */
  1160. error = CS_ERR_TRY_AGAIN;
  1161. goto response_send;
  1162. }
  1163. }
  1164. switch (cpd->cpd_state) {
  1165. case CPD_STATE_UNJOINED:
  1166. error = CS_OK;
  1167. cpd->cpd_state = CPD_STATE_JOIN_STARTED;
  1168. cpd->pid = req_lib_cpg_join->pid;
  1169. cpd->flags = req_lib_cpg_join->flags;
  1170. memcpy (&cpd->group_name, &req_lib_cpg_join->group_name,
  1171. sizeof (cpd->group_name));
  1172. cpg_node_joinleave_send (req_lib_cpg_join->pid,
  1173. &req_lib_cpg_join->group_name,
  1174. MESSAGE_REQ_EXEC_CPG_PROCJOIN, CONFCHG_CPG_REASON_JOIN);
  1175. break;
  1176. case CPD_STATE_LEAVE_STARTED:
  1177. error = CS_ERR_BUSY;
  1178. break;
  1179. case CPD_STATE_JOIN_STARTED:
  1180. error = CS_ERR_EXIST;
  1181. break;
  1182. case CPD_STATE_JOIN_COMPLETED:
  1183. error = CS_ERR_EXIST;
  1184. break;
  1185. }
  1186. response_send:
  1187. res_lib_cpg_join.header.size = sizeof(res_lib_cpg_join);
  1188. res_lib_cpg_join.header.id = MESSAGE_RES_CPG_JOIN;
  1189. res_lib_cpg_join.header.error = error;
  1190. api->ipc_response_send (conn, &res_lib_cpg_join, sizeof(res_lib_cpg_join));
  1191. }
  1192. /* Leave message from the library */
  1193. static void message_handler_req_lib_cpg_leave (void *conn, const void *message)
  1194. {
  1195. struct res_lib_cpg_leave res_lib_cpg_leave;
  1196. cs_error_t error = CS_OK;
  1197. struct req_lib_cpg_leave *req_lib_cpg_leave = (struct req_lib_cpg_leave *)message;
  1198. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1199. log_printf(LOGSYS_LEVEL_DEBUG, "got leave request on %p\n", conn);
  1200. switch (cpd->cpd_state) {
  1201. case CPD_STATE_UNJOINED:
  1202. error = CS_ERR_NOT_EXIST;
  1203. break;
  1204. case CPD_STATE_LEAVE_STARTED:
  1205. error = CS_ERR_NOT_EXIST;
  1206. break;
  1207. case CPD_STATE_JOIN_STARTED:
  1208. error = CS_ERR_BUSY;
  1209. break;
  1210. case CPD_STATE_JOIN_COMPLETED:
  1211. error = CS_OK;
  1212. cpd->cpd_state = CPD_STATE_LEAVE_STARTED;
  1213. cpg_node_joinleave_send (req_lib_cpg_leave->pid,
  1214. &req_lib_cpg_leave->group_name,
  1215. MESSAGE_REQ_EXEC_CPG_PROCLEAVE,
  1216. CONFCHG_CPG_REASON_LEAVE);
  1217. break;
  1218. }
  1219. /* send return */
  1220. res_lib_cpg_leave.header.size = sizeof(res_lib_cpg_leave);
  1221. res_lib_cpg_leave.header.id = MESSAGE_RES_CPG_LEAVE;
  1222. res_lib_cpg_leave.header.error = error;
  1223. api->ipc_response_send(conn, &res_lib_cpg_leave, sizeof(res_lib_cpg_leave));
  1224. }
  1225. /* Finalize message from library */
  1226. static void message_handler_req_lib_cpg_finalize (
  1227. void *conn,
  1228. const void *message)
  1229. {
  1230. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1231. struct res_lib_cpg_finalize res_lib_cpg_finalize;
  1232. cs_error_t error = CS_OK;
  1233. log_printf (LOGSYS_LEVEL_DEBUG, "cpg finalize for conn=%p\n", conn);
  1234. /*
  1235. * We will just remove cpd from list. After this call, connection will be
  1236. * closed on lib side, and cpg_lib_exit_fn will be called
  1237. */
  1238. list_del (&cpd->list);
  1239. list_init (&cpd->list);
  1240. res_lib_cpg_finalize.header.size = sizeof (res_lib_cpg_finalize);
  1241. res_lib_cpg_finalize.header.id = MESSAGE_RES_CPG_FINALIZE;
  1242. res_lib_cpg_finalize.header.error = error;
  1243. api->ipc_response_send (conn, &res_lib_cpg_finalize,
  1244. sizeof (res_lib_cpg_finalize));
  1245. }
  1246. static int
  1247. memory_map (
  1248. const char *path,
  1249. size_t bytes,
  1250. void **buf)
  1251. {
  1252. int32_t fd;
  1253. void *addr_orig;
  1254. void *addr;
  1255. int32_t res;
  1256. fd = open (path, O_RDWR, 0600);
  1257. unlink (path);
  1258. if (fd == -1) {
  1259. return (-1);
  1260. }
  1261. res = ftruncate (fd, bytes);
  1262. if (res == -1) {
  1263. goto error_close_unlink;
  1264. }
  1265. addr_orig = mmap (NULL, bytes, PROT_NONE,
  1266. MAP_ANONYMOUS | MAP_PRIVATE, -1, 0);
  1267. if (addr_orig == MAP_FAILED) {
  1268. goto error_close_unlink;
  1269. }
  1270. addr = mmap (addr_orig, bytes, PROT_READ | PROT_WRITE,
  1271. MAP_FIXED | MAP_SHARED, fd, 0);
  1272. if (addr != addr_orig) {
  1273. munmap(addr_orig, bytes);
  1274. goto error_close_unlink;
  1275. }
  1276. #ifdef COROSYNC_BSD
  1277. madvise(addr, bytes, MADV_NOSYNC);
  1278. #endif
  1279. res = close (fd);
  1280. if (res) {
  1281. return (-1);
  1282. }
  1283. *buf = addr_orig;
  1284. return (0);
  1285. error_close_unlink:
  1286. close (fd);
  1287. unlink(path);
  1288. return -1;
  1289. }
  1290. static inline int zcb_alloc (
  1291. struct cpg_pd *cpd,
  1292. const char *path_to_file,
  1293. size_t size,
  1294. void **addr)
  1295. {
  1296. struct zcb_mapped *zcb_mapped;
  1297. unsigned int res;
  1298. zcb_mapped = malloc (sizeof (struct zcb_mapped));
  1299. if (zcb_mapped == NULL) {
  1300. return (-1);
  1301. }
  1302. res = memory_map (
  1303. path_to_file,
  1304. size,
  1305. addr);
  1306. if (res == -1) {
  1307. free (zcb_mapped);
  1308. return (-1);
  1309. }
  1310. list_init (&zcb_mapped->list);
  1311. zcb_mapped->addr = *addr;
  1312. zcb_mapped->size = size;
  1313. list_add_tail (&zcb_mapped->list, &cpd->zcb_mapped_list_head);
  1314. return (0);
  1315. }
  1316. static inline int zcb_free (struct zcb_mapped *zcb_mapped)
  1317. {
  1318. unsigned int res;
  1319. res = munmap (zcb_mapped->addr, zcb_mapped->size);
  1320. list_del (&zcb_mapped->list);
  1321. free (zcb_mapped);
  1322. return (res);
  1323. }
  1324. static inline int zcb_by_addr_free (struct cpg_pd *cpd, void *addr)
  1325. {
  1326. struct list_head *list;
  1327. struct zcb_mapped *zcb_mapped;
  1328. unsigned int res = 0;
  1329. for (list = cpd->zcb_mapped_list_head.next;
  1330. list != &cpd->zcb_mapped_list_head; list = list->next) {
  1331. zcb_mapped = list_entry (list, struct zcb_mapped, list);
  1332. if (zcb_mapped->addr == addr) {
  1333. res = zcb_free (zcb_mapped);
  1334. break;
  1335. }
  1336. }
  1337. return (res);
  1338. }
  1339. static inline int zcb_all_free (
  1340. struct cpg_pd *cpd)
  1341. {
  1342. struct list_head *list;
  1343. struct zcb_mapped *zcb_mapped;
  1344. for (list = cpd->zcb_mapped_list_head.next;
  1345. list != &cpd->zcb_mapped_list_head;) {
  1346. zcb_mapped = list_entry (list, struct zcb_mapped, list);
  1347. list = list->next;
  1348. zcb_free (zcb_mapped);
  1349. }
  1350. return (0);
  1351. }
  1352. union u {
  1353. uint64_t server_addr;
  1354. void *server_ptr;
  1355. };
  1356. static uint64_t void2serveraddr (void *server_ptr)
  1357. {
  1358. union u u;
  1359. u.server_ptr = server_ptr;
  1360. return (u.server_addr);
  1361. }
  1362. static void *serveraddr2void (uint64_t server_addr)
  1363. {
  1364. union u u;
  1365. u.server_addr = server_addr;
  1366. return (u.server_ptr);
  1367. };
  1368. static void message_handler_req_lib_cpg_zc_alloc (
  1369. void *conn,
  1370. const void *message)
  1371. {
  1372. mar_req_coroipcc_zc_alloc_t *hdr = (mar_req_coroipcc_zc_alloc_t *)message;
  1373. struct qb_ipc_response_header res_header;
  1374. void *addr = NULL;
  1375. struct coroipcs_zc_header *zc_header;
  1376. unsigned int res;
  1377. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1378. log_printf(LOGSYS_LEVEL_DEBUG, "path: %s", hdr->path_to_file);
  1379. res = zcb_alloc (cpd, hdr->path_to_file, hdr->map_size,
  1380. &addr);
  1381. assert(res == 0);
  1382. zc_header = (struct coroipcs_zc_header *)addr;
  1383. zc_header->server_address = void2serveraddr(addr);
  1384. res_header.size = sizeof (struct qb_ipc_response_header);
  1385. res_header.id = 0;
  1386. api->ipc_response_send (conn,
  1387. &res_header,
  1388. res_header.size);
  1389. }
  1390. static void message_handler_req_lib_cpg_zc_free (
  1391. void *conn,
  1392. const void *message)
  1393. {
  1394. mar_req_coroipcc_zc_free_t *hdr = (mar_req_coroipcc_zc_free_t *)message;
  1395. struct qb_ipc_response_header res_header;
  1396. void *addr = NULL;
  1397. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1398. log_printf(LOGSYS_LEVEL_DEBUG, " free'ing");
  1399. addr = serveraddr2void (hdr->server_address);
  1400. zcb_by_addr_free (cpd, addr);
  1401. res_header.size = sizeof (struct qb_ipc_response_header);
  1402. res_header.id = 0;
  1403. api->ipc_response_send (
  1404. conn, &res_header,
  1405. res_header.size);
  1406. }
  1407. /* Mcast message from the library */
  1408. static void message_handler_req_lib_cpg_mcast (void *conn, const void *message)
  1409. {
  1410. const struct req_lib_cpg_mcast *req_lib_cpg_mcast = message;
  1411. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1412. mar_cpg_name_t group_name = cpd->group_name;
  1413. struct iovec req_exec_cpg_iovec[2];
  1414. struct req_exec_cpg_mcast req_exec_cpg_mcast;
  1415. int msglen = req_lib_cpg_mcast->msglen;
  1416. int result;
  1417. cs_error_t error = CS_ERR_NOT_EXIST;
  1418. log_printf(LOGSYS_LEVEL_DEBUG, "got mcast request on %p\n", conn);
  1419. switch (cpd->cpd_state) {
  1420. case CPD_STATE_UNJOINED:
  1421. error = CS_ERR_NOT_EXIST;
  1422. break;
  1423. case CPD_STATE_LEAVE_STARTED:
  1424. error = CS_ERR_NOT_EXIST;
  1425. break;
  1426. case CPD_STATE_JOIN_STARTED:
  1427. error = CS_OK;
  1428. break;
  1429. case CPD_STATE_JOIN_COMPLETED:
  1430. error = CS_OK;
  1431. break;
  1432. }
  1433. if (error == CS_OK) {
  1434. req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen;
  1435. req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
  1436. MESSAGE_REQ_EXEC_CPG_MCAST);
  1437. req_exec_cpg_mcast.pid = cpd->pid;
  1438. req_exec_cpg_mcast.msglen = msglen;
  1439. api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
  1440. memcpy(&req_exec_cpg_mcast.group_name, &group_name,
  1441. sizeof(mar_cpg_name_t));
  1442. req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
  1443. req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
  1444. req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message;
  1445. req_exec_cpg_iovec[1].iov_len = msglen;
  1446. result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
  1447. assert(result == 0);
  1448. } else {
  1449. log_printf(LOGSYS_LEVEL_ERROR, "*** %p can't mcast to group %s state:%d, error:%d\n",
  1450. conn, group_name.value, cpd->cpd_state, error);
  1451. }
  1452. }
  1453. static void message_handler_req_lib_cpg_zc_execute (
  1454. void *conn,
  1455. const void *message)
  1456. {
  1457. mar_req_coroipcc_zc_execute_t *hdr = (mar_req_coroipcc_zc_execute_t *)message;
  1458. struct qb_ipc_request_header *header;
  1459. struct res_lib_cpg_mcast res_lib_cpg_mcast;
  1460. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1461. struct iovec req_exec_cpg_iovec[2];
  1462. struct req_exec_cpg_mcast req_exec_cpg_mcast;
  1463. struct req_lib_cpg_mcast *req_lib_cpg_mcast;
  1464. int result;
  1465. cs_error_t error = CS_ERR_NOT_EXIST;
  1466. log_printf(LOGSYS_LEVEL_DEBUG, "got ZC mcast request on %p\n", conn);
  1467. header = (struct qb_ipc_request_header *)(((char *)serveraddr2void(hdr->server_address) + sizeof (struct coroipcs_zc_header)));
  1468. req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)header;
  1469. switch (cpd->cpd_state) {
  1470. case CPD_STATE_UNJOINED:
  1471. error = CS_ERR_NOT_EXIST;
  1472. break;
  1473. case CPD_STATE_LEAVE_STARTED:
  1474. error = CS_ERR_NOT_EXIST;
  1475. break;
  1476. case CPD_STATE_JOIN_STARTED:
  1477. error = CS_OK;
  1478. break;
  1479. case CPD_STATE_JOIN_COMPLETED:
  1480. error = CS_OK;
  1481. break;
  1482. }
  1483. res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast);
  1484. res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_MCAST;
  1485. if (error == CS_OK) {
  1486. req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + req_lib_cpg_mcast->msglen;
  1487. req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
  1488. MESSAGE_REQ_EXEC_CPG_MCAST);
  1489. req_exec_cpg_mcast.pid = cpd->pid;
  1490. req_exec_cpg_mcast.msglen = req_lib_cpg_mcast->msglen;
  1491. api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
  1492. memcpy(&req_exec_cpg_mcast.group_name, &cpd->group_name,
  1493. sizeof(mar_cpg_name_t));
  1494. req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
  1495. req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
  1496. req_exec_cpg_iovec[1].iov_base = (char *)header + sizeof(struct req_lib_cpg_mcast);
  1497. req_exec_cpg_iovec[1].iov_len = req_exec_cpg_mcast.msglen;
  1498. result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
  1499. if (result == 0) {
  1500. res_lib_cpg_mcast.header.error = CS_OK;
  1501. } else {
  1502. res_lib_cpg_mcast.header.error = CS_ERR_TRY_AGAIN;
  1503. }
  1504. } else {
  1505. res_lib_cpg_mcast.header.error = error;
  1506. }
  1507. api->ipc_response_send (conn, &res_lib_cpg_mcast,
  1508. sizeof (res_lib_cpg_mcast));
  1509. }
  1510. static void message_handler_req_lib_cpg_membership (void *conn,
  1511. const void *message)
  1512. {
  1513. struct req_lib_cpg_membership_get *req_lib_cpg_membership_get =
  1514. (struct req_lib_cpg_membership_get *)message;
  1515. struct res_lib_cpg_membership_get res_lib_cpg_membership_get;
  1516. struct list_head *iter;
  1517. int member_count = 0;
  1518. res_lib_cpg_membership_get.header.id = MESSAGE_RES_CPG_MEMBERSHIP;
  1519. res_lib_cpg_membership_get.header.error = CS_OK;
  1520. res_lib_cpg_membership_get.header.size =
  1521. sizeof (struct req_lib_cpg_membership_get);
  1522. for (iter = process_info_list_head.next;
  1523. iter != &process_info_list_head; iter = iter->next) {
  1524. struct process_info *pi = list_entry (iter, struct process_info, list);
  1525. if (mar_name_compare (&pi->group, &req_lib_cpg_membership_get->group_name) == 0) {
  1526. res_lib_cpg_membership_get.member_list[member_count].nodeid = pi->nodeid;
  1527. res_lib_cpg_membership_get.member_list[member_count].pid = pi->pid;
  1528. member_count += 1;
  1529. }
  1530. }
  1531. res_lib_cpg_membership_get.member_count = member_count;
  1532. api->ipc_response_send (conn, &res_lib_cpg_membership_get,
  1533. sizeof (res_lib_cpg_membership_get));
  1534. }
  1535. static void message_handler_req_lib_cpg_local_get (void *conn,
  1536. const void *message)
  1537. {
  1538. struct res_lib_cpg_local_get res_lib_cpg_local_get;
  1539. res_lib_cpg_local_get.header.size = sizeof (res_lib_cpg_local_get);
  1540. res_lib_cpg_local_get.header.id = MESSAGE_RES_CPG_LOCAL_GET;
  1541. res_lib_cpg_local_get.header.error = CS_OK;
  1542. res_lib_cpg_local_get.local_nodeid = api->totem_nodeid_get ();
  1543. api->ipc_response_send (conn, &res_lib_cpg_local_get,
  1544. sizeof (res_lib_cpg_local_get));
  1545. }
  1546. static void message_handler_req_lib_cpg_iteration_initialize (
  1547. void *conn,
  1548. const void *message)
  1549. {
  1550. const struct req_lib_cpg_iterationinitialize *req_lib_cpg_iterationinitialize = message;
  1551. struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
  1552. hdb_handle_t cpg_iteration_handle = 0;
  1553. struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize;
  1554. struct list_head *iter, *iter2;
  1555. struct cpg_iteration_instance *cpg_iteration_instance;
  1556. cs_error_t error = CS_OK;
  1557. int res;
  1558. log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration initialize\n");
  1559. /* Because between calling this function and *next can be some operations which will
  1560. * change list, we must do full copy.
  1561. */
  1562. /*
  1563. * Create new iteration instance
  1564. */
  1565. res = hdb_handle_create (&cpg_iteration_handle_t_db, sizeof (struct cpg_iteration_instance),
  1566. &cpg_iteration_handle);
  1567. if (res != 0) {
  1568. error = CS_ERR_NO_MEMORY;
  1569. goto response_send;
  1570. }
  1571. res = hdb_handle_get (&cpg_iteration_handle_t_db, cpg_iteration_handle, (void *)&cpg_iteration_instance);
  1572. if (res != 0) {
  1573. error = CS_ERR_BAD_HANDLE;
  1574. goto error_destroy;
  1575. }
  1576. list_init (&cpg_iteration_instance->items_list_head);
  1577. cpg_iteration_instance->handle = cpg_iteration_handle;
  1578. /*
  1579. * Create copy of process_info list "grouped by" group name
  1580. */
  1581. for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
  1582. struct process_info *pi = list_entry (iter, struct process_info, list);
  1583. struct process_info *new_pi;
  1584. if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) {
  1585. /*
  1586. * Try to find processed group name in our list new list
  1587. */
  1588. int found = 0;
  1589. for (iter2 = cpg_iteration_instance->items_list_head.next;
  1590. iter2 != &cpg_iteration_instance->items_list_head;
  1591. iter2 = iter2->next) {
  1592. struct process_info *pi2 = list_entry (iter2, struct process_info, list);
  1593. if (mar_name_compare (&pi2->group, &pi->group) == 0) {
  1594. found = 1;
  1595. break;
  1596. }
  1597. }
  1598. if (found) {
  1599. /*
  1600. * We have this name in list -> don't add
  1601. */
  1602. continue ;
  1603. }
  1604. } else if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_ONE_GROUP) {
  1605. /*
  1606. * Test pi group name with request
  1607. */
  1608. if (mar_name_compare (&pi->group, &req_lib_cpg_iterationinitialize->group_name) != 0)
  1609. /*
  1610. * Not same -> don't add
  1611. */
  1612. continue ;
  1613. }
  1614. new_pi = malloc (sizeof (struct process_info));
  1615. if (!new_pi) {
  1616. log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct");
  1617. error = CS_ERR_NO_MEMORY;
  1618. goto error_put_destroy;
  1619. }
  1620. memcpy (new_pi, pi, sizeof (struct process_info));
  1621. list_init (&new_pi->list);
  1622. if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) {
  1623. /*
  1624. * pid and nodeid -> undefined
  1625. */
  1626. new_pi->pid = new_pi->nodeid = 0;
  1627. }
  1628. /*
  1629. * We will return list "grouped" by "group name", so try to find right place to add
  1630. */
  1631. for (iter2 = cpg_iteration_instance->items_list_head.next;
  1632. iter2 != &cpg_iteration_instance->items_list_head;
  1633. iter2 = iter2->next) {
  1634. struct process_info *pi2 = list_entry (iter2, struct process_info, list);
  1635. if (mar_name_compare (&pi2->group, &pi->group) == 0) {
  1636. break;
  1637. }
  1638. }
  1639. list_add (&new_pi->list, iter2);
  1640. }
  1641. /*
  1642. * Now we have a full "grouped by" copy of process_info list
  1643. */
  1644. /*
  1645. * Add instance to current cpd list
  1646. */
  1647. list_init (&cpg_iteration_instance->list);
  1648. list_add (&cpg_iteration_instance->list, &cpd->iteration_instance_list_head);
  1649. cpg_iteration_instance->current_pointer = &cpg_iteration_instance->items_list_head;
  1650. error_put_destroy:
  1651. hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_handle);
  1652. error_destroy:
  1653. if (error != CS_OK) {
  1654. hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_handle);
  1655. }
  1656. response_send:
  1657. res_lib_cpg_iterationinitialize.header.size = sizeof (res_lib_cpg_iterationinitialize);
  1658. res_lib_cpg_iterationinitialize.header.id = MESSAGE_RES_CPG_ITERATIONINITIALIZE;
  1659. res_lib_cpg_iterationinitialize.header.error = error;
  1660. res_lib_cpg_iterationinitialize.iteration_handle = cpg_iteration_handle;
  1661. api->ipc_response_send (conn, &res_lib_cpg_iterationinitialize,
  1662. sizeof (res_lib_cpg_iterationinitialize));
  1663. }
  1664. static void message_handler_req_lib_cpg_iteration_next (
  1665. void *conn,
  1666. const void *message)
  1667. {
  1668. const struct req_lib_cpg_iterationnext *req_lib_cpg_iterationnext = message;
  1669. struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext;
  1670. struct cpg_iteration_instance *cpg_iteration_instance;
  1671. cs_error_t error = CS_OK;
  1672. int res;
  1673. struct process_info *pi;
  1674. log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration next\n");
  1675. res = hdb_handle_get (&cpg_iteration_handle_t_db,
  1676. req_lib_cpg_iterationnext->iteration_handle,
  1677. (void *)&cpg_iteration_instance);
  1678. if (res != 0) {
  1679. error = CS_ERR_LIBRARY;
  1680. goto error_exit;
  1681. }
  1682. assert (cpg_iteration_instance);
  1683. cpg_iteration_instance->current_pointer = cpg_iteration_instance->current_pointer->next;
  1684. if (cpg_iteration_instance->current_pointer == &cpg_iteration_instance->items_list_head) {
  1685. error = CS_ERR_NO_SECTIONS;
  1686. goto error_put;
  1687. }
  1688. pi = list_entry (cpg_iteration_instance->current_pointer, struct process_info, list);
  1689. /*
  1690. * Copy iteration data
  1691. */
  1692. res_lib_cpg_iterationnext.description.nodeid = pi->nodeid;
  1693. res_lib_cpg_iterationnext.description.pid = pi->pid;
  1694. memcpy (&res_lib_cpg_iterationnext.description.group,
  1695. &pi->group,
  1696. sizeof (mar_cpg_name_t));
  1697. error_put:
  1698. hdb_handle_put (&cpg_iteration_handle_t_db, req_lib_cpg_iterationnext->iteration_handle);
  1699. error_exit:
  1700. res_lib_cpg_iterationnext.header.size = sizeof (res_lib_cpg_iterationnext);
  1701. res_lib_cpg_iterationnext.header.id = MESSAGE_RES_CPG_ITERATIONNEXT;
  1702. res_lib_cpg_iterationnext.header.error = error;
  1703. api->ipc_response_send (conn, &res_lib_cpg_iterationnext,
  1704. sizeof (res_lib_cpg_iterationnext));
  1705. }
  1706. static void message_handler_req_lib_cpg_iteration_finalize (
  1707. void *conn,
  1708. const void *message)
  1709. {
  1710. const struct req_lib_cpg_iterationfinalize *req_lib_cpg_iterationfinalize = message;
  1711. struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize;
  1712. struct cpg_iteration_instance *cpg_iteration_instance;
  1713. cs_error_t error = CS_OK;
  1714. int res;
  1715. log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration finalize\n");
  1716. res = hdb_handle_get (&cpg_iteration_handle_t_db,
  1717. req_lib_cpg_iterationfinalize->iteration_handle,
  1718. (void *)&cpg_iteration_instance);
  1719. if (res != 0) {
  1720. error = CS_ERR_LIBRARY;
  1721. goto error_exit;
  1722. }
  1723. assert (cpg_iteration_instance);
  1724. cpg_iteration_instance_finalize (cpg_iteration_instance);
  1725. hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle);
  1726. error_exit:
  1727. res_lib_cpg_iterationfinalize.header.size = sizeof (res_lib_cpg_iterationfinalize);
  1728. res_lib_cpg_iterationfinalize.header.id = MESSAGE_RES_CPG_ITERATIONFINALIZE;
  1729. res_lib_cpg_iterationfinalize.header.error = error;
  1730. api->ipc_response_send (conn, &res_lib_cpg_iterationfinalize,
  1731. sizeof (res_lib_cpg_iterationfinalize));
  1732. }