corosync  3.0.2
exec/cpg.c
Go to the documentation of this file.
1 /*
2  * Copyright (c) 2006-2019 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 
36 #include <config.h>
37 
38 #ifdef HAVE_ALLOCA_H
39 #include <alloca.h>
40 #endif
41 #include <sys/types.h>
42 #include <sys/socket.h>
43 #include <sys/un.h>
44 #include <sys/ioctl.h>
45 #include <netinet/in.h>
46 #include <sys/uio.h>
47 #include <unistd.h>
48 #include <fcntl.h>
49 #include <stdlib.h>
50 #include <stdio.h>
51 #include <errno.h>
52 #include <time.h>
53 #include <assert.h>
54 #include <arpa/inet.h>
55 #include <sys/mman.h>
56 
57 #include <qb/qblist.h>
58 #include <qb/qbmap.h>
59 
60 #include <corosync/corotypes.h>
61 #include <qb/qbipc_common.h>
62 #include <corosync/corodefs.h>
63 #include <corosync/logsys.h>
64 #include <corosync/coroapi.h>
65 
66 #include <corosync/cpg.h>
67 #include <corosync/ipc_cpg.h>
68 
69 #ifndef MAP_ANONYMOUS
70 #define MAP_ANONYMOUS MAP_ANON
71 #endif
72 
73 #include "service.h"
74 
75 LOGSYS_DECLARE_SUBSYS ("CPG");
76 
77 #define GROUP_HASH_SIZE 32
78 
87 };
88 
89 struct zcb_mapped {
90  struct qb_list_head list;
91  void *addr;
92  size_t size;
93 };
94 /*
95  * state` exec deliver
96  * match group name, pid -> if matched deliver for YES:
97  * XXX indicates impossible state
98  *
99  * join leave mcast
100  * UNJOINED XXX XXX NO
101  * LEAVE_STARTED XXX YES(unjoined_enter) YES
102  * JOIN_STARTED YES(join_started_enter) XXX NO
103  * JOIN_COMPLETED XXX NO YES
104  *
105  * join_started_enter
106  * set JOIN_COMPLETED
107  * add entry to process_info list
108  * unjoined_enter
109  * set UNJOINED
110  * delete entry from process_info list
111  *
112  *
113  * library accept join error codes
114  * UNJOINED YES(CS_OK) set JOIN_STARTED
115  * LEAVE_STARTED NO(CS_ERR_BUSY)
116  * JOIN_STARTED NO(CS_ERR_EXIST)
117  * JOIN_COMPlETED NO(CS_ERR_EXIST)
118  *
119  * library accept leave error codes
120  * UNJOINED NO(CS_ERR_NOT_EXIST)
121  * LEAVE_STARTED NO(CS_ERR_NOT_EXIST)
122  * JOIN_STARTED NO(CS_ERR_BUSY)
123  * JOIN_COMPLETED YES(CS_OK) set LEAVE_STARTED
124  *
125  * library accept mcast
126  * UNJOINED NO(CS_ERR_NOT_EXIST)
127  * LEAVE_STARTED NO(CS_ERR_NOT_EXIST)
128  * JOIN_STARTED YES(CS_OK)
129  * JOIN_COMPLETED YES(CS_OK)
130  */
131 enum cpd_state {
136 };
137 
141 };
142 
143 static struct qb_list_head joinlist_messages_head;
144 
145 struct cpg_pd {
146  void *conn;
148  uint32_t pid;
150  unsigned int flags;
152  uint64_t transition_counter; /* These two are used when sending fragmented messages */
154  struct qb_list_head list;
155  struct qb_list_head iteration_instance_list_head;
156  struct qb_list_head zcb_mapped_list_head;
157 };
158 
161  struct qb_list_head list;
162  struct qb_list_head items_list_head; /* List of process_info */
163  struct qb_list_head *current_pointer;
164 };
165 
166 DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db,NULL);
167 
168 QB_LIST_DECLARE (cpg_pd_list_head);
169 
170 static unsigned int my_member_list[PROCESSOR_COUNT_MAX];
171 
172 static unsigned int my_member_list_entries;
173 
174 static unsigned int my_old_member_list[PROCESSOR_COUNT_MAX];
175 
176 static unsigned int my_old_member_list_entries = 0;
177 
178 static struct corosync_api_v1 *api = NULL;
179 
180 static enum cpg_sync_state my_sync_state = CPGSYNC_DOWNLIST;
181 
182 static mar_cpg_ring_id_t last_sync_ring_id;
183 
184 struct process_info {
185  unsigned int nodeid;
186  uint32_t pid;
188  struct qb_list_head list; /* on the group_info members list */
189 };
190 QB_LIST_DECLARE (process_info_list_head);
191 
193  uint32_t pid;
195 };
196 
197 /*
198  * Service Interfaces required by service_message_handler struct
199  */
200 static char *cpg_exec_init_fn (struct corosync_api_v1 *);
201 
202 static int cpg_lib_init_fn (void *conn);
203 
204 static int cpg_lib_exit_fn (void *conn);
205 
206 static void message_handler_req_exec_cpg_procjoin (
207  const void *message,
208  unsigned int nodeid);
209 
210 static void message_handler_req_exec_cpg_procleave (
211  const void *message,
212  unsigned int nodeid);
213 
214 static void message_handler_req_exec_cpg_joinlist (
215  const void *message,
216  unsigned int nodeid);
217 
218 static void message_handler_req_exec_cpg_mcast (
219  const void *message,
220  unsigned int nodeid);
221 
222 static void message_handler_req_exec_cpg_partial_mcast (
223  const void *message,
224  unsigned int nodeid);
225 
226 static void message_handler_req_exec_cpg_downlist_old (
227  const void *message,
228  unsigned int nodeid);
229 
230 static void message_handler_req_exec_cpg_downlist (
231  const void *message,
232  unsigned int nodeid);
233 
234 static void exec_cpg_procjoin_endian_convert (void *msg);
235 
236 static void exec_cpg_joinlist_endian_convert (void *msg);
237 
238 static void exec_cpg_mcast_endian_convert (void *msg);
239 
240 static void exec_cpg_partial_mcast_endian_convert (void *msg);
241 
242 static void exec_cpg_downlist_endian_convert_old (void *msg);
243 
244 static void exec_cpg_downlist_endian_convert (void *msg);
245 
246 static void message_handler_req_lib_cpg_join (void *conn, const void *message);
247 
248 static void message_handler_req_lib_cpg_leave (void *conn, const void *message);
249 
250 static void message_handler_req_lib_cpg_finalize (void *conn, const void *message);
251 
252 static void message_handler_req_lib_cpg_mcast (void *conn, const void *message);
253 
254 static void message_handler_req_lib_cpg_partial_mcast (void *conn, const void *message);
255 
256 static void message_handler_req_lib_cpg_membership (void *conn,
257  const void *message);
258 
259 static void message_handler_req_lib_cpg_local_get (void *conn,
260  const void *message);
261 
262 static void message_handler_req_lib_cpg_iteration_initialize (
263  void *conn,
264  const void *message);
265 
266 static void message_handler_req_lib_cpg_iteration_next (
267  void *conn,
268  const void *message);
269 
270 static void message_handler_req_lib_cpg_iteration_finalize (
271  void *conn,
272  const void *message);
273 
274 static void message_handler_req_lib_cpg_zc_alloc (
275  void *conn,
276  const void *message);
277 
278 static void message_handler_req_lib_cpg_zc_free (
279  void *conn,
280  const void *message);
281 
282 static void message_handler_req_lib_cpg_zc_execute (
283  void *conn,
284  const void *message);
285 
286 static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason);
287 
288 static int cpg_exec_send_downlist(void);
289 
290 static int cpg_exec_send_joinlist(void);
291 
292 static void downlist_inform_clients (void);
293 
294 static void joinlist_inform_clients (void);
295 
296 static void joinlist_messages_delete (void);
297 
298 static void cpg_sync_init (
299  const unsigned int *trans_list,
300  size_t trans_list_entries,
301  const unsigned int *member_list,
302  size_t member_list_entries,
303  const struct memb_ring_id *ring_id);
304 
305 static int cpg_sync_process (void);
306 
307 static void cpg_sync_activate (void);
308 
309 static void cpg_sync_abort (void);
310 
311 static void do_proc_join(
312  const mar_cpg_name_t *name,
313  uint32_t pid,
314  unsigned int nodeid,
315  int reason);
316 
317 static void do_proc_leave(
318  const mar_cpg_name_t *name,
319  uint32_t pid,
320  unsigned int nodeid,
321  int reason);
322 
323 static int notify_lib_totem_membership (
324  void *conn,
325  int member_list_entries,
326  const unsigned int *member_list);
327 
328 static inline int zcb_all_free (
329  struct cpg_pd *cpd);
330 
331 static char *cpg_print_group_name (
332  const mar_cpg_name_t *group);
333 
334 /*
335  * Library Handler Definition
336  */
337 static struct corosync_lib_handler cpg_lib_engine[] =
338 {
339  { /* 0 - MESSAGE_REQ_CPG_JOIN */
340  .lib_handler_fn = message_handler_req_lib_cpg_join,
341  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
342  },
343  { /* 1 - MESSAGE_REQ_CPG_LEAVE */
344  .lib_handler_fn = message_handler_req_lib_cpg_leave,
345  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
346  },
347  { /* 2 - MESSAGE_REQ_CPG_MCAST */
348  .lib_handler_fn = message_handler_req_lib_cpg_mcast,
349  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
350  },
351  { /* 3 - MESSAGE_REQ_CPG_MEMBERSHIP */
352  .lib_handler_fn = message_handler_req_lib_cpg_membership,
353  .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
354  },
355  { /* 4 - MESSAGE_REQ_CPG_LOCAL_GET */
356  .lib_handler_fn = message_handler_req_lib_cpg_local_get,
357  .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
358  },
359  { /* 5 - MESSAGE_REQ_CPG_ITERATIONINITIALIZE */
360  .lib_handler_fn = message_handler_req_lib_cpg_iteration_initialize,
361  .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
362  },
363  { /* 6 - MESSAGE_REQ_CPG_ITERATIONNEXT */
364  .lib_handler_fn = message_handler_req_lib_cpg_iteration_next,
365  .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
366  },
367  { /* 7 - MESSAGE_REQ_CPG_ITERATIONFINALIZE */
368  .lib_handler_fn = message_handler_req_lib_cpg_iteration_finalize,
369  .flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
370  },
371  { /* 8 - MESSAGE_REQ_CPG_FINALIZE */
372  .lib_handler_fn = message_handler_req_lib_cpg_finalize,
373  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
374  },
375  { /* 9 */
376  .lib_handler_fn = message_handler_req_lib_cpg_zc_alloc,
377  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
378  },
379  { /* 10 */
380  .lib_handler_fn = message_handler_req_lib_cpg_zc_free,
381  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
382  },
383  { /* 11 */
384  .lib_handler_fn = message_handler_req_lib_cpg_zc_execute,
385  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
386  },
387  { /* 12 */
388  .lib_handler_fn = message_handler_req_lib_cpg_partial_mcast,
389  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
390  },
391 
392 };
393 
394 static struct corosync_exec_handler cpg_exec_engine[] =
395 {
396  { /* 0 - MESSAGE_REQ_EXEC_CPG_PROCJOIN */
397  .exec_handler_fn = message_handler_req_exec_cpg_procjoin,
398  .exec_endian_convert_fn = exec_cpg_procjoin_endian_convert
399  },
400  { /* 1 - MESSAGE_REQ_EXEC_CPG_PROCLEAVE */
401  .exec_handler_fn = message_handler_req_exec_cpg_procleave,
402  .exec_endian_convert_fn = exec_cpg_procjoin_endian_convert
403  },
404  { /* 2 - MESSAGE_REQ_EXEC_CPG_JOINLIST */
405  .exec_handler_fn = message_handler_req_exec_cpg_joinlist,
406  .exec_endian_convert_fn = exec_cpg_joinlist_endian_convert
407  },
408  { /* 3 - MESSAGE_REQ_EXEC_CPG_MCAST */
409  .exec_handler_fn = message_handler_req_exec_cpg_mcast,
410  .exec_endian_convert_fn = exec_cpg_mcast_endian_convert
411  },
412  { /* 4 - MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD */
413  .exec_handler_fn = message_handler_req_exec_cpg_downlist_old,
414  .exec_endian_convert_fn = exec_cpg_downlist_endian_convert_old
415  },
416  { /* 5 - MESSAGE_REQ_EXEC_CPG_DOWNLIST */
417  .exec_handler_fn = message_handler_req_exec_cpg_downlist,
418  .exec_endian_convert_fn = exec_cpg_downlist_endian_convert
419  },
420  { /* 6 - MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST */
421  .exec_handler_fn = message_handler_req_exec_cpg_partial_mcast,
422  .exec_endian_convert_fn = exec_cpg_partial_mcast_endian_convert
423  },
424 };
425 
427  .name = "corosync cluster closed process group service v1.01",
428  .id = CPG_SERVICE,
429  .priority = 1,
430  .private_data_size = sizeof (struct cpg_pd),
431  .flow_control = CS_LIB_FLOW_CONTROL_REQUIRED,
432  .allow_inquorate = CS_LIB_ALLOW_INQUORATE,
433  .lib_init_fn = cpg_lib_init_fn,
434  .lib_exit_fn = cpg_lib_exit_fn,
435  .lib_engine = cpg_lib_engine,
436  .lib_engine_count = sizeof (cpg_lib_engine) / sizeof (struct corosync_lib_handler),
437  .exec_init_fn = cpg_exec_init_fn,
438  .exec_dump_fn = NULL,
439  .exec_engine = cpg_exec_engine,
440  .exec_engine_count = sizeof (cpg_exec_engine) / sizeof (struct corosync_exec_handler),
441  .sync_init = cpg_sync_init,
442  .sync_process = cpg_sync_process,
443  .sync_activate = cpg_sync_activate,
444  .sync_abort = cpg_sync_abort
445 };
446 
448 {
449  return (&cpg_service_engine);
450 }
451 
453  struct qb_ipc_request_header header __attribute__((aligned(8)));
454  mar_cpg_name_t group_name __attribute__((aligned(8)));
455  mar_uint32_t pid __attribute__((aligned(8)));
456  mar_uint32_t reason __attribute__((aligned(8)));
457 };
458 
460  struct qb_ipc_request_header header __attribute__((aligned(8)));
461  mar_cpg_name_t group_name __attribute__((aligned(8)));
462  mar_uint32_t msglen __attribute__((aligned(8)));
463  mar_uint32_t pid __attribute__((aligned(8)));
464  mar_message_source_t source __attribute__((aligned(8)));
465  mar_uint8_t message[] __attribute__((aligned(8)));
466 };
467 
469  struct qb_ipc_request_header header __attribute__((aligned(8)));
470  mar_cpg_name_t group_name __attribute__((aligned(8)));
471  mar_uint32_t msglen __attribute__((aligned(8)));
472  mar_uint32_t fraglen __attribute__((aligned(8)));
473  mar_uint32_t pid __attribute__((aligned(8)));
474  mar_uint32_t type __attribute__((aligned(8)));
475  mar_message_source_t source __attribute__((aligned(8)));
476  mar_uint8_t message[] __attribute__((aligned(8)));
477 };
478 
480  struct qb_ipc_request_header header __attribute__((aligned(8)));
481  mar_uint32_t left_nodes __attribute__((aligned(8)));
482  mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)));
483 };
484 
486  struct qb_ipc_request_header header __attribute__((aligned(8)));
487  /* merge decisions */
488  mar_uint32_t old_members __attribute__((aligned(8)));
489  /* downlist below */
490  mar_uint32_t left_nodes __attribute__((aligned(8)));
491  mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)));
492 };
493 
494 struct joinlist_msg {
496  uint32_t pid;
498  struct qb_list_head list;
499 };
500 
501 static struct req_exec_cpg_downlist g_req_exec_cpg_downlist;
502 
503 /*
504  * Function print group name. It's not reentrant
505  */
506 static char *cpg_print_group_name(const mar_cpg_name_t *group)
507 {
508  static char res[CPG_MAX_NAME_LENGTH * 4 + 1];
509  int dest_pos = 0;
510  char c;
511  int i;
512 
513  for (i = 0; i < group->length; i++) {
514  c = group->value[i];
515 
516  if (c >= ' ' && c < 0x7f && c != '\\') {
517  res[dest_pos++] = c;
518  } else {
519  if (c == '\\') {
520  res[dest_pos++] = '\\';
521  res[dest_pos++] = '\\';
522  } else {
523  snprintf(res + dest_pos, sizeof(res) - dest_pos, "\\x%02X", c);
524  dest_pos += 4;
525  }
526  }
527  }
528  res[dest_pos] = 0;
529 
530  return (res);
531 }
532 
533 static void cpg_sync_init (
534  const unsigned int *trans_list,
535  size_t trans_list_entries,
536  const unsigned int *member_list,
537  size_t member_list_entries,
538  const struct memb_ring_id *ring_id)
539 {
540  int entries;
541  int i, j;
542  int found;
543 
544  my_sync_state = CPGSYNC_DOWNLIST;
545 
546  memcpy (my_member_list, member_list, member_list_entries *
547  sizeof (unsigned int));
548  my_member_list_entries = member_list_entries;
549 
550  last_sync_ring_id.nodeid = ring_id->nodeid;
551  last_sync_ring_id.seq = ring_id->seq;
552 
553  entries = 0;
554  /*
555  * Determine list of nodeids for downlist message
556  */
557  for (i = 0; i < my_old_member_list_entries; i++) {
558  found = 0;
559  for (j = 0; j < trans_list_entries; j++) {
560  if (my_old_member_list[i] == trans_list[j]) {
561  found = 1;
562  break;
563  }
564  }
565  if (found == 0) {
566  g_req_exec_cpg_downlist.nodeids[entries++] =
567  my_old_member_list[i];
568  }
569  }
570  g_req_exec_cpg_downlist.left_nodes = entries;
571 }
572 
573 static int cpg_sync_process (void)
574 {
575  int res = -1;
576 
577  if (my_sync_state == CPGSYNC_DOWNLIST) {
578  res = cpg_exec_send_downlist();
579  if (res == -1) {
580  return (-1);
581  }
582  my_sync_state = CPGSYNC_JOINLIST;
583  }
584  if (my_sync_state == CPGSYNC_JOINLIST) {
585  res = cpg_exec_send_joinlist();
586  }
587  return (res);
588 }
589 
590 static void cpg_sync_activate (void)
591 {
592  memcpy (my_old_member_list, my_member_list,
593  my_member_list_entries * sizeof (unsigned int));
594  my_old_member_list_entries = my_member_list_entries;
595 
596  downlist_inform_clients ();
597 
598  joinlist_inform_clients ();
599 
600  joinlist_messages_delete ();
601 
602  notify_lib_totem_membership (NULL, my_member_list_entries, my_member_list);
603 }
604 
605 static void cpg_sync_abort (void)
606 {
607 
608  joinlist_messages_delete ();
609 }
610 
611 static int notify_lib_totem_membership (
612  void *conn,
613  int member_list_entries,
614  const unsigned int *member_list)
615 {
616  struct qb_list_head *iter;
617  char *buf;
618  int size;
620 
621  size = sizeof(struct res_lib_cpg_totem_confchg_callback) +
622  sizeof(mar_uint32_t) * (member_list_entries);
623  buf = alloca(size);
624  if (!buf)
625  return CS_ERR_LIBRARY;
626 
627  res = (struct res_lib_cpg_totem_confchg_callback *)buf;
628  res->member_list_entries = member_list_entries;
629  res->header.size = size;
631  res->header.error = CS_OK;
632 
633  memcpy (&res->ring_id, &last_sync_ring_id, sizeof (mar_cpg_ring_id_t));
634  memcpy (res->member_list, member_list, res->member_list_entries * sizeof (mar_uint32_t));
635 
636  if (conn == NULL) {
637  qb_list_for_each(iter, &cpg_pd_list_head) {
638  struct cpg_pd *cpg_pd = qb_list_entry (iter, struct cpg_pd, list);
639  api->ipc_dispatch_send (cpg_pd->conn, buf, size);
640  }
641  } else {
642  api->ipc_dispatch_send (conn, buf, size);
643  }
644 
645  return CS_OK;
646 }
647 
648 static int notify_lib_joinlist(
649  const mar_cpg_name_t *group_name,
650  void *conn,
651  int joined_list_entries,
652  mar_cpg_address_t *joined_list,
653  int left_list_entries,
654  mar_cpg_address_t *left_list,
655  int id)
656 {
657  int size;
658  char *buf;
659  struct qb_list_head *iter;
660  int count;
661  struct res_lib_cpg_confchg_callback *res;
662  mar_cpg_address_t *retgi;
663 
664  count = 0;
665 
666  qb_list_for_each(iter, &process_info_list_head) {
667  struct process_info *pi = qb_list_entry (iter, struct process_info, list);
668  if (mar_name_compare (&pi->group, group_name) == 0) {
669  int i;
670  int founded = 0;
671 
672  for (i = 0; i < left_list_entries; i++) {
673  if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) {
674  founded++;
675  }
676  }
677 
678  if (!founded)
679  count++;
680  }
681  }
682 
683  size = sizeof(struct res_lib_cpg_confchg_callback) +
684  sizeof(mar_cpg_address_t) * (count + left_list_entries + joined_list_entries);
685  buf = alloca(size);
686  if (!buf)
687  return CS_ERR_LIBRARY;
688 
689  res = (struct res_lib_cpg_confchg_callback *)buf;
690  res->joined_list_entries = joined_list_entries;
691  res->left_list_entries = left_list_entries;
692  res->member_list_entries = count;
693  retgi = res->member_list;
694  res->header.size = size;
695  res->header.id = id;
696  res->header.error = CS_OK;
697  memcpy(&res->group_name, group_name, sizeof(mar_cpg_name_t));
698 
699  qb_list_for_each(iter, &process_info_list_head) {
700  struct process_info *pi=qb_list_entry (iter, struct process_info, list);
701 
702  if (mar_name_compare (&pi->group, group_name) == 0) {
703  int i;
704  int founded = 0;
705 
706  for (i = 0;i < left_list_entries; i++) {
707  if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) {
708  founded++;
709  }
710  }
711 
712  if (!founded) {
713  retgi->nodeid = pi->nodeid;
714  retgi->pid = pi->pid;
715  retgi->reason = CPG_REASON_UNDEFINED;
716  retgi++;
717  }
718  }
719  }
720 
721  if (left_list_entries) {
722  memcpy (retgi, left_list, left_list_entries * sizeof(mar_cpg_address_t));
723  retgi += left_list_entries;
724  }
725 
726  if (joined_list_entries) {
727  memcpy (retgi, joined_list, joined_list_entries * sizeof(mar_cpg_address_t));
728  retgi += joined_list_entries;
729  }
730 
731  if (conn) {
732  api->ipc_dispatch_send (conn, buf, size);
733  } else {
734  qb_list_for_each(iter, &cpg_pd_list_head) {
735  struct cpg_pd *cpd = qb_list_entry (iter, struct cpg_pd, list);
736  if (mar_name_compare (&cpd->group_name, group_name) == 0) {
737  assert (joined_list_entries <= 1);
738  if (joined_list_entries) {
739  if (joined_list[0].pid == cpd->pid &&
740  joined_list[0].nodeid == api->totem_nodeid_get()) {
742  }
743  }
744  if (cpd->cpd_state == CPD_STATE_JOIN_COMPLETED ||
746 
747  api->ipc_dispatch_send (cpd->conn, buf, size);
748  cpd->transition_counter++;
749  }
750  if (left_list_entries) {
751  if (left_list[0].pid == cpd->pid &&
752  left_list[0].nodeid == api->totem_nodeid_get() &&
753  left_list[0].reason == CONFCHG_CPG_REASON_LEAVE) {
754 
755  cpd->pid = 0;
756  memset (&cpd->group_name, 0, sizeof(cpd->group_name));
758  }
759  }
760  }
761  }
762  }
763 
764 
765  /*
766  * Traverse thru cpds and send totem membership for cpd, where it is not send yet
767  */
768  qb_list_for_each(iter, &cpg_pd_list_head) {
769  struct cpg_pd *cpd = qb_list_entry (iter, struct cpg_pd, list);
770 
772  cpd->initial_totem_conf_sent = 1;
773 
774  notify_lib_totem_membership (cpd->conn, my_old_member_list_entries, my_old_member_list);
775  }
776  }
777 
778  return CS_OK;
779 }
780 
781 static void downlist_log(const char *msg, struct req_exec_cpg_downlist *dl)
782 {
783  log_printf (LOG_DEBUG,
784  "%s: members(old:%d left:%d)",
785  msg,
786  dl->old_members,
787  dl->left_nodes);
788 }
789 
790 static void downlist_inform_clients (void)
791 {
792  struct qb_list_head *iter, *tmp_iter;
793  struct process_info *left_pi;
794  qb_map_t *group_map;
795  struct cpg_name cpg_group;
796  mar_cpg_name_t group;
797  struct confchg_data{
798  struct cpg_name cpg_group;
800  int left_list_entries;
801  struct qb_list_head list;
802  } *pcd;
803  qb_map_iter_t *miter;
804  int i, size;
805 
806  downlist_log("my downlist", &g_req_exec_cpg_downlist);
807 
808  group_map = qb_skiplist_create();
809 
810  /*
811  * only the cpg groups included in left nodes should receive
812  * confchg event, so we will collect these cpg groups and
813  * relative left_lists here.
814  */
815  qb_list_for_each_safe(iter, tmp_iter, &process_info_list_head) {
816  struct process_info *pi = qb_list_entry(iter, struct process_info, list);
817 
818  left_pi = NULL;
819  for (i = 0; i < g_req_exec_cpg_downlist.left_nodes; i++) {
820 
821  if (pi->nodeid == g_req_exec_cpg_downlist.nodeids[i]) {
822  left_pi = pi;
823  break;
824  }
825  }
826 
827  if (left_pi) {
828  marshall_from_mar_cpg_name_t(&cpg_group, &left_pi->group);
829  cpg_group.value[cpg_group.length] = 0;
830 
831  pcd = (struct confchg_data *)qb_map_get(group_map, cpg_group.value);
832  if (pcd == NULL) {
833  pcd = (struct confchg_data *)calloc(1, sizeof(struct confchg_data));
834  memcpy(&pcd->cpg_group, &cpg_group, sizeof(struct cpg_name));
835  qb_map_put(group_map, pcd->cpg_group.value, pcd);
836  }
837  size = pcd->left_list_entries;
838  pcd->left_list[size].nodeid = left_pi->nodeid;
839  pcd->left_list[size].pid = left_pi->pid;
840  pcd->left_list[size].reason = CONFCHG_CPG_REASON_NODEDOWN;
841  pcd->left_list_entries++;
842  qb_list_del (&left_pi->list);
843  free (left_pi);
844  }
845  }
846 
847  /* send only one confchg event per cpg group */
848  miter = qb_map_iter_create(group_map);
849  while (qb_map_iter_next(miter, (void **)&pcd)) {
850  marshall_to_mar_cpg_name_t(&group, &pcd->cpg_group);
851 
852  log_printf (LOG_DEBUG, "left_list_entries:%d", pcd->left_list_entries);
853  for (i=0; i<pcd->left_list_entries; i++) {
854  log_printf (LOG_DEBUG, "left_list[%d] group:%s, ip:%s, pid:%d",
855  i, cpg_print_group_name(&group),
856  (char*)api->totem_ifaces_print(pcd->left_list[i].nodeid),
857  pcd->left_list[i].pid);
858  }
859 
860  /* send confchg event */
861  notify_lib_joinlist(&group, NULL,
862  0, NULL,
863  pcd->left_list_entries,
864  pcd->left_list,
866 
867  free(pcd);
868  }
869  qb_map_iter_free(miter);
870  qb_map_destroy(group_map);
871 }
872 
873 /*
874  * Remove processes that might have left the group while we were suspended.
875  */
876 static void joinlist_remove_zombie_pi_entries (void)
877 {
878  struct qb_list_head *pi_iter, *tmp_iter;
879  struct qb_list_head *jl_iter;
880  struct process_info *pi;
881  struct joinlist_msg *stored_msg;
882  int found;
883 
884  qb_list_for_each_safe(pi_iter, tmp_iter, &process_info_list_head) {
885  pi = qb_list_entry (pi_iter, struct process_info, list);
886 
887  /*
888  * Ignore local node
889  */
890  if (pi->nodeid == api->totem_nodeid_get()) {
891  continue ;
892  }
893 
894  /*
895  * Try to find message in joinlist messages
896  */
897  found = 0;
898  qb_list_for_each(jl_iter, &joinlist_messages_head) {
899  stored_msg = qb_list_entry(jl_iter, struct joinlist_msg, list);
900 
901  if (stored_msg->sender_nodeid == api->totem_nodeid_get()) {
902  continue ;
903  }
904 
905  if (pi->nodeid == stored_msg->sender_nodeid &&
906  pi->pid == stored_msg->pid &&
907  mar_name_compare (&pi->group, &stored_msg->group_name) == 0) {
908  found = 1;
909  break ;
910  }
911  }
912 
913  if (!found) {
914  do_proc_leave(&pi->group, pi->pid, pi->nodeid, CONFCHG_CPG_REASON_PROCDOWN);
915  }
916  }
917 }
918 
919 static void joinlist_inform_clients (void)
920 {
921  struct joinlist_msg *stored_msg;
922  struct qb_list_head *iter;
923  unsigned int i;
924 
925  i = 0;
926  qb_list_for_each(iter, &joinlist_messages_head) {
927  stored_msg = qb_list_entry(iter, struct joinlist_msg, list);
928 
929  log_printf (LOG_DEBUG, "joinlist_messages[%u] group:%s, ip:%s, pid:%d",
930  i++, cpg_print_group_name(&stored_msg->group_name),
931  (char*)api->totem_ifaces_print(stored_msg->sender_nodeid),
932  stored_msg->pid);
933 
934  /* Ignore our own messages */
935  if (stored_msg->sender_nodeid == api->totem_nodeid_get()) {
936  continue ;
937  }
938 
939  do_proc_join (&stored_msg->group_name, stored_msg->pid, stored_msg->sender_nodeid,
941  }
942 
943  joinlist_remove_zombie_pi_entries ();
944 }
945 
946 static void joinlist_messages_delete (void)
947 {
948  struct joinlist_msg *stored_msg;
949  struct qb_list_head *iter, *tmp_iter;
950 
951  qb_list_for_each_safe(iter, tmp_iter, &joinlist_messages_head) {
952  stored_msg = qb_list_entry(iter, struct joinlist_msg, list);
953  qb_list_del (&stored_msg->list);
954  free (stored_msg);
955  }
956  qb_list_init (&joinlist_messages_head);
957 }
958 
959 static char *cpg_exec_init_fn (struct corosync_api_v1 *corosync_api)
960 {
961  qb_list_init (&joinlist_messages_head);
962  api = corosync_api;
963  return (NULL);
964 }
965 
966 static void cpg_iteration_instance_finalize (struct cpg_iteration_instance *cpg_iteration_instance)
967 {
968  struct qb_list_head *iter, *tmp_iter;
969  struct process_info *pi;
970 
971  qb_list_for_each_safe(iter, tmp_iter, &(cpg_iteration_instance->items_list_head)) {
972  pi = qb_list_entry (iter, struct process_info, list);
973  qb_list_del (&pi->list);
974  free (pi);
975  }
976 
977  qb_list_del (&cpg_iteration_instance->list);
978  hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle);
979 }
980 
981 static void cpg_pd_finalize (struct cpg_pd *cpd)
982 {
983  struct qb_list_head *iter, *tmp_iter;
984  struct cpg_iteration_instance *cpii;
985 
986  zcb_all_free(cpd);
987  qb_list_for_each_safe(iter, tmp_iter, &(cpd->iteration_instance_list_head)) {
988  cpii = qb_list_entry (iter, struct cpg_iteration_instance, list);
989 
990  cpg_iteration_instance_finalize (cpii);
991  }
992 
993  qb_list_del (&cpd->list);
994 }
995 
996 static int cpg_lib_exit_fn (void *conn)
997 {
998  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
999 
1000  log_printf(LOGSYS_LEVEL_DEBUG, "exit_fn for conn=%p", conn);
1001 
1002  if (cpd->group_name.length > 0 && cpd->cpd_state != CPD_STATE_LEAVE_STARTED) {
1003  cpg_node_joinleave_send (cpd->pid, &cpd->group_name,
1005  }
1006 
1007  cpg_pd_finalize (cpd);
1008 
1009  api->ipc_refcnt_dec (conn);
1010  return (0);
1011 }
1012 
1013 static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason)
1014 {
1016  struct iovec req_exec_cpg_iovec;
1017  int result;
1018 
1019  memcpy(&req_exec_cpg_procjoin.group_name, group_name, sizeof(mar_cpg_name_t));
1020  req_exec_cpg_procjoin.pid = pid;
1021  req_exec_cpg_procjoin.reason = reason;
1022 
1023  req_exec_cpg_procjoin.header.size = sizeof(req_exec_cpg_procjoin);
1025 
1026  req_exec_cpg_iovec.iov_base = (char *)&req_exec_cpg_procjoin;
1027  req_exec_cpg_iovec.iov_len = sizeof(req_exec_cpg_procjoin);
1028 
1029  result = api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED);
1030 
1031  return (result);
1032 }
1033 
1034 /* Can byteswap join & leave messages */
1035 static void exec_cpg_procjoin_endian_convert (void *msg)
1036 {
1038 
1039  req_exec_cpg_procjoin->pid = swab32(req_exec_cpg_procjoin->pid);
1040  swab_mar_cpg_name_t (&req_exec_cpg_procjoin->group_name);
1041  req_exec_cpg_procjoin->reason = swab32(req_exec_cpg_procjoin->reason);
1042 }
1043 
1044 static void exec_cpg_joinlist_endian_convert (void *msg_v)
1045 {
1046  char *msg = msg_v;
1047  struct qb_ipc_response_header *res = (struct qb_ipc_response_header *)msg;
1048  struct join_list_entry *jle = (struct join_list_entry *)(msg + sizeof(struct qb_ipc_response_header));
1049 
1050  swab_mar_int32_t (&res->size);
1051 
1052  while ((const char*)jle < msg + res->size) {
1053  jle->pid = swab32(jle->pid);
1054  swab_mar_cpg_name_t (&jle->group_name);
1055  jle++;
1056  }
1057 }
1058 
1059 static void exec_cpg_downlist_endian_convert_old (void *msg)
1060 {
1061 }
1062 
1063 static void exec_cpg_downlist_endian_convert (void *msg)
1064 {
1066  unsigned int i;
1067 
1068  req_exec_cpg_downlist->left_nodes = swab32(req_exec_cpg_downlist->left_nodes);
1069  req_exec_cpg_downlist->old_members = swab32(req_exec_cpg_downlist->old_members);
1070 
1071  for (i = 0; i < req_exec_cpg_downlist->left_nodes; i++) {
1072  req_exec_cpg_downlist->nodeids[i] = swab32(req_exec_cpg_downlist->nodeids[i]);
1073  }
1074 }
1075 
1076 
1077 static void exec_cpg_mcast_endian_convert (void *msg)
1078 {
1079  struct req_exec_cpg_mcast *req_exec_cpg_mcast = msg;
1080 
1081  swab_coroipc_request_header_t (&req_exec_cpg_mcast->header);
1082  swab_mar_cpg_name_t (&req_exec_cpg_mcast->group_name);
1083  req_exec_cpg_mcast->pid = swab32(req_exec_cpg_mcast->pid);
1084  req_exec_cpg_mcast->msglen = swab32(req_exec_cpg_mcast->msglen);
1085  swab_mar_message_source_t (&req_exec_cpg_mcast->source);
1086 }
1087 
1088 static void exec_cpg_partial_mcast_endian_convert (void *msg)
1089 {
1091 
1092  swab_coroipc_request_header_t (&req_exec_cpg_mcast->header);
1093  swab_mar_cpg_name_t (&req_exec_cpg_mcast->group_name);
1094  req_exec_cpg_mcast->pid = swab32(req_exec_cpg_mcast->pid);
1095  req_exec_cpg_mcast->msglen = swab32(req_exec_cpg_mcast->msglen);
1096  req_exec_cpg_mcast->fraglen = swab32(req_exec_cpg_mcast->fraglen);
1097  req_exec_cpg_mcast->type = swab32(req_exec_cpg_mcast->type);
1098  swab_mar_message_source_t (&req_exec_cpg_mcast->source);
1099 }
1100 
1101 static struct process_info *process_info_find(const mar_cpg_name_t *group_name, uint32_t pid, unsigned int nodeid) {
1102  struct qb_list_head *iter;
1103 
1104  qb_list_for_each(iter, &process_info_list_head) {
1105  struct process_info *pi = qb_list_entry (iter, struct process_info, list);
1106 
1107  if (pi->pid == pid && pi->nodeid == nodeid &&
1108  mar_name_compare (&pi->group, group_name) == 0) {
1109  return pi;
1110  }
1111  }
1112 
1113  return NULL;
1114 }
1115 
1116 static void do_proc_join(
1117  const mar_cpg_name_t *name,
1118  uint32_t pid,
1119  unsigned int nodeid,
1120  int reason)
1121 {
1122  struct process_info *pi;
1123  struct process_info *pi_entry;
1124  mar_cpg_address_t notify_info;
1125  struct qb_list_head *list;
1126  struct qb_list_head *list_to_add = NULL;
1127 
1128  if (process_info_find (name, pid, nodeid) != NULL) {
1129  return ;
1130  }
1131  pi = malloc (sizeof (struct process_info));
1132  if (!pi) {
1133  log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct");
1134  return;
1135  }
1136  pi->nodeid = nodeid;
1137  pi->pid = pid;
1138  memcpy(&pi->group, name, sizeof(*name));
1139  qb_list_init(&pi->list);
1140 
1141  /*
1142  * Insert new process in sorted order so synchronization works properly
1143  */
1144  list_to_add = &process_info_list_head;
1145  qb_list_for_each(list, &process_info_list_head) {
1146  pi_entry = qb_list_entry(list, struct process_info, list);
1147  if (pi_entry->nodeid > pi->nodeid ||
1148  (pi_entry->nodeid == pi->nodeid && pi_entry->pid > pi->pid)) {
1149 
1150  break;
1151  }
1152  list_to_add = list;
1153  }
1154  qb_list_add (&pi->list, list_to_add);
1155 
1156  notify_info.pid = pi->pid;
1157  notify_info.nodeid = nodeid;
1158  notify_info.reason = reason;
1159 
1160  notify_lib_joinlist(&pi->group, NULL,
1161  1, &notify_info,
1162  0, NULL,
1164 }
1165 
1166 static void do_proc_leave(
1167  const mar_cpg_name_t *name,
1168  uint32_t pid,
1169  unsigned int nodeid,
1170  int reason)
1171 {
1172  struct process_info *pi;
1173  struct qb_list_head *iter, *tmp_iter;
1174  mar_cpg_address_t notify_info;
1175 
1176  notify_info.pid = pid;
1177  notify_info.nodeid = nodeid;
1178  notify_info.reason = reason;
1179 
1180  notify_lib_joinlist(name, NULL,
1181  0, NULL,
1182  1, &notify_info,
1184 
1185  qb_list_for_each_safe(iter, tmp_iter, &process_info_list_head) {
1186  pi = qb_list_entry(iter, struct process_info, list);
1187 
1188  if (pi->pid == pid && pi->nodeid == nodeid &&
1189  mar_name_compare (&pi->group, name)==0) {
1190  qb_list_del (&pi->list);
1191  free (pi);
1192  }
1193  }
1194 }
1195 
1196 static void message_handler_req_exec_cpg_downlist_old (
1197  const void *message,
1198  unsigned int nodeid)
1199 {
1200  log_printf (LOGSYS_LEVEL_WARNING, "downlist OLD from node 0x%x",
1201  nodeid);
1202 }
1203 
1204 static void message_handler_req_exec_cpg_downlist(
1205  const void *message,
1206  unsigned int nodeid)
1207 {
1208  const struct req_exec_cpg_downlist *req_exec_cpg_downlist = message;
1209 
1210  log_printf (LOGSYS_LEVEL_WARNING, "downlist left_list: %d received",
1211  req_exec_cpg_downlist->left_nodes);
1212 }
1213 
1214 
1215 static void message_handler_req_exec_cpg_procjoin (
1216  const void *message,
1217  unsigned int nodeid)
1218 {
1219  const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message;
1220 
1221  log_printf(LOGSYS_LEVEL_DEBUG, "got procjoin message from cluster node 0x%x (%s) for pid %u",
1222  nodeid,
1223  api->totem_ifaces_print(nodeid),
1224  (unsigned int)req_exec_cpg_procjoin->pid);
1225 
1226  do_proc_join (&req_exec_cpg_procjoin->group_name,
1227  req_exec_cpg_procjoin->pid, nodeid,
1229 }
1230 
1231 static void message_handler_req_exec_cpg_procleave (
1232  const void *message,
1233  unsigned int nodeid)
1234 {
1235  const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message;
1236 
1237  log_printf(LOGSYS_LEVEL_DEBUG, "got procleave message from cluster node 0x%x (%s) for pid %u",
1238  nodeid,
1239  api->totem_ifaces_print(nodeid),
1240  (unsigned int)req_exec_cpg_procjoin->pid);
1241 
1242  do_proc_leave (&req_exec_cpg_procjoin->group_name,
1243  req_exec_cpg_procjoin->pid, nodeid,
1244  req_exec_cpg_procjoin->reason);
1245 }
1246 
1247 
1248 /* Got a proclist from another node */
1249 static void message_handler_req_exec_cpg_joinlist (
1250  const void *message_v,
1251  unsigned int nodeid)
1252 {
1253  const char *message = message_v;
1254  const struct qb_ipc_response_header *res = (const struct qb_ipc_response_header *)message;
1255  const struct join_list_entry *jle = (const struct join_list_entry *)(message + sizeof(struct qb_ipc_response_header));
1256  struct joinlist_msg *stored_msg;
1257 
1258  log_printf(LOGSYS_LEVEL_DEBUG, "got joinlist message from node 0x%x",
1259  nodeid);
1260 
1261  while ((const char*)jle < message + res->size) {
1262  stored_msg = malloc (sizeof (struct joinlist_msg));
1263  memset(stored_msg, 0, sizeof (struct joinlist_msg));
1264  stored_msg->sender_nodeid = nodeid;
1265  stored_msg->pid = jle->pid;
1266  memcpy(&stored_msg->group_name, &jle->group_name, sizeof(mar_cpg_name_t));
1267  qb_list_init (&stored_msg->list);
1268  qb_list_add (&stored_msg->list, &joinlist_messages_head);
1269  jle++;
1270  }
1271 }
1272 
1273 static void message_handler_req_exec_cpg_mcast (
1274  const void *message,
1275  unsigned int nodeid)
1276 {
1277  const struct req_exec_cpg_mcast *req_exec_cpg_mcast = message;
1279  int msglen = req_exec_cpg_mcast->msglen;
1280  struct qb_list_head *iter, *pi_iter, *tmp_iter;
1281  struct cpg_pd *cpd;
1282  struct iovec iovec[2];
1283  int known_node = 0;
1284 
1286  res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast) + msglen;
1287  res_lib_cpg_mcast.msglen = msglen;
1288  res_lib_cpg_mcast.pid = req_exec_cpg_mcast->pid;
1289  res_lib_cpg_mcast.nodeid = nodeid;
1290 
1291  memcpy(&res_lib_cpg_mcast.group_name, &req_exec_cpg_mcast->group_name,
1292  sizeof(mar_cpg_name_t));
1293  iovec[0].iov_base = (void *)&res_lib_cpg_mcast;
1294  iovec[0].iov_len = sizeof (res_lib_cpg_mcast);
1295 
1296  iovec[1].iov_base = (char*)message+sizeof(*req_exec_cpg_mcast);
1297  iovec[1].iov_len = msglen;
1298 
1299  qb_list_for_each_safe(iter, tmp_iter, &cpg_pd_list_head) {
1300  cpd = qb_list_entry(iter, struct cpg_pd, list);
1302  && (mar_name_compare (&cpd->group_name, &req_exec_cpg_mcast->group_name) == 0)) {
1303 
1304  if (!known_node) {
1305  /* Try to find, if we know the node */
1306  qb_list_for_each(pi_iter, &process_info_list_head) {
1307  struct process_info *pi = qb_list_entry (pi_iter, struct process_info, list);
1308 
1309  if (pi->nodeid == nodeid &&
1310  mar_name_compare (&pi->group, &req_exec_cpg_mcast->group_name) == 0) {
1311  known_node = 1;
1312  break;
1313  }
1314  }
1315  }
1316 
1317  if (!known_node) {
1318  log_printf(LOGSYS_LEVEL_WARNING, "Unknown node -> we will not deliver message");
1319  return ;
1320  }
1321 
1322  api->ipc_dispatch_iov_send (cpd->conn, iovec, 2);
1323  }
1324  }
1325 }
1326 
1327 static void message_handler_req_exec_cpg_partial_mcast (
1328  const void *message,
1329  unsigned int nodeid)
1330 {
1331  const struct req_exec_cpg_partial_mcast *req_exec_cpg_mcast = message;
1333  int msglen = req_exec_cpg_mcast->fraglen;
1334  struct qb_list_head *iter, *pi_iter, *tmp_iter;
1335  struct cpg_pd *cpd;
1336  struct iovec iovec[2];
1337  int known_node = 0;
1338 
1339  log_printf(LOGSYS_LEVEL_DEBUG, "Got fragmented message from node %d, size = %d bytes\n", nodeid, msglen);
1340 
1342  res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast) + msglen;
1343  res_lib_cpg_mcast.fraglen = msglen;
1344  res_lib_cpg_mcast.msglen = req_exec_cpg_mcast->msglen;
1345  res_lib_cpg_mcast.pid = req_exec_cpg_mcast->pid;
1346  res_lib_cpg_mcast.type = req_exec_cpg_mcast->type;
1347  res_lib_cpg_mcast.nodeid = nodeid;
1348 
1349  memcpy(&res_lib_cpg_mcast.group_name, &req_exec_cpg_mcast->group_name,
1350  sizeof(mar_cpg_name_t));
1351  iovec[0].iov_base = (void *)&res_lib_cpg_mcast;
1352  iovec[0].iov_len = sizeof (res_lib_cpg_mcast);
1353 
1354  iovec[1].iov_base = (char*)message+sizeof(*req_exec_cpg_mcast);
1355  iovec[1].iov_len = msglen;
1356 
1357  qb_list_for_each_safe(iter, tmp_iter, &cpg_pd_list_head) {
1358  cpd = qb_list_entry(iter, struct cpg_pd, list);
1359 
1361  && (mar_name_compare (&cpd->group_name, &req_exec_cpg_mcast->group_name) == 0)) {
1362 
1363  if (!known_node) {
1364  /* Try to find, if we know the node */
1365  qb_list_for_each(pi_iter, &process_info_list_head) {
1366  struct process_info *pi = qb_list_entry (pi_iter, struct process_info, list);
1367 
1368  if (pi->nodeid == nodeid &&
1369  mar_name_compare (&pi->group, &req_exec_cpg_mcast->group_name) == 0) {
1370  known_node = 1;
1371  break;
1372  }
1373  }
1374  }
1375 
1376  if (!known_node) {
1377  log_printf(LOGSYS_LEVEL_WARNING, "Unknown node -> we will not deliver message");
1378  return ;
1379  }
1380 
1381  api->ipc_dispatch_iov_send (cpd->conn, iovec, 2);
1382  }
1383  }
1384 }
1385 
1386 
1387 static int cpg_exec_send_downlist(void)
1388 {
1389  struct iovec iov;
1390 
1391  g_req_exec_cpg_downlist.header.id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_DOWNLIST);
1392  g_req_exec_cpg_downlist.header.size = sizeof(struct req_exec_cpg_downlist);
1393 
1394  g_req_exec_cpg_downlist.old_members = my_old_member_list_entries;
1395 
1396  iov.iov_base = (void *)&g_req_exec_cpg_downlist;
1397  iov.iov_len = g_req_exec_cpg_downlist.header.size;
1398 
1399  return (api->totem_mcast (&iov, 1, TOTEM_AGREED));
1400 }
1401 
1402 static int cpg_exec_send_joinlist(void)
1403 {
1404  int count = 0;
1405  struct qb_list_head *iter;
1406  struct qb_ipc_response_header *res;
1407  char *buf;
1408  struct join_list_entry *jle;
1409  struct iovec req_exec_cpg_iovec;
1410 
1411  qb_list_for_each(iter, &process_info_list_head) {
1412  struct process_info *pi = qb_list_entry (iter, struct process_info, list);
1413 
1414  if (pi->nodeid == api->totem_nodeid_get ()) {
1415  count++;
1416  }
1417  }
1418 
1419  /* Nothing to send */
1420  if (!count)
1421  return 0;
1422 
1423  buf = alloca(sizeof(struct qb_ipc_response_header) + sizeof(struct join_list_entry) * count);
1424  if (!buf) {
1425  log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate joinlist buffer");
1426  return -1;
1427  }
1428 
1429  jle = (struct join_list_entry *)(buf + sizeof(struct qb_ipc_response_header));
1430  res = (struct qb_ipc_response_header *)buf;
1431 
1432  qb_list_for_each(iter, &process_info_list_head) {
1433  struct process_info *pi = qb_list_entry (iter, struct process_info, list);
1434 
1435  if (pi->nodeid == api->totem_nodeid_get ()) {
1436  memcpy (&jle->group_name, &pi->group, sizeof (mar_cpg_name_t));
1437  jle->pid = pi->pid;
1438  jle++;
1439  }
1440  }
1441 
1443  res->size = sizeof(struct qb_ipc_response_header)+sizeof(struct join_list_entry) * count;
1444 
1445  req_exec_cpg_iovec.iov_base = buf;
1446  req_exec_cpg_iovec.iov_len = res->size;
1447 
1448  return (api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED));
1449 }
1450 
1451 static int cpg_lib_init_fn (void *conn)
1452 {
1453  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1454  memset (cpd, 0, sizeof(struct cpg_pd));
1455  cpd->conn = conn;
1456  qb_list_add (&cpd->list, &cpg_pd_list_head);
1457 
1458  qb_list_init (&cpd->iteration_instance_list_head);
1459  qb_list_init (&cpd->zcb_mapped_list_head);
1460 
1461  api->ipc_refcnt_inc (conn);
1462  log_printf(LOGSYS_LEVEL_DEBUG, "lib_init_fn: conn=%p, cpd=%p", conn, cpd);
1463  return (0);
1464 }
1465 
1466 /* Join message from the library */
1467 static void message_handler_req_lib_cpg_join (void *conn, const void *message)
1468 {
1469  const struct req_lib_cpg_join *req_lib_cpg_join = message;
1470  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1472  cs_error_t error = CS_OK;
1473  struct qb_list_head *iter;
1474 
1475  /* Test, if we don't have same pid and group name joined */
1476  qb_list_for_each(iter, &cpg_pd_list_head) {
1477  struct cpg_pd *cpd_item = qb_list_entry (iter, struct cpg_pd, list);
1478 
1479  if (cpd_item->pid == req_lib_cpg_join->pid &&
1480  mar_name_compare(&req_lib_cpg_join->group_name, &cpd_item->group_name) == 0) {
1481 
1482  /* We have same pid and group name joined -> return error */
1483  error = CS_ERR_EXIST;
1484  goto response_send;
1485  }
1486  }
1487 
1488  /*
1489  * Same check must be done in process info list, because there may be not yet delivered
1490  * leave of client.
1491  */
1492  qb_list_for_each(iter, &process_info_list_head) {
1493  struct process_info *pi = qb_list_entry (iter, struct process_info, list);
1494 
1495  if (pi->nodeid == api->totem_nodeid_get () && pi->pid == req_lib_cpg_join->pid &&
1496  mar_name_compare(&req_lib_cpg_join->group_name, &pi->group) == 0) {
1497  /* We have same pid and group name joined -> return error */
1498  error = CS_ERR_TRY_AGAIN;
1499  goto response_send;
1500  }
1501  }
1502 
1503  if (req_lib_cpg_join->group_name.length > CPG_MAX_NAME_LENGTH) {
1504  error = CS_ERR_NAME_TOO_LONG;
1505  goto response_send;
1506  }
1507 
1508  switch (cpd->cpd_state) {
1509  case CPD_STATE_UNJOINED:
1510  error = CS_OK;
1512  cpd->pid = req_lib_cpg_join->pid;
1513  cpd->flags = req_lib_cpg_join->flags;
1514  memcpy (&cpd->group_name, &req_lib_cpg_join->group_name,
1515  sizeof (cpd->group_name));
1516 
1517  cpg_node_joinleave_send (req_lib_cpg_join->pid,
1518  &req_lib_cpg_join->group_name,
1520  break;
1522  error = CS_ERR_BUSY;
1523  break;
1525  error = CS_ERR_EXIST;
1526  break;
1528  error = CS_ERR_EXIST;
1529  break;
1530  }
1531 
1532 response_send:
1533  res_lib_cpg_join.header.size = sizeof(res_lib_cpg_join);
1535  res_lib_cpg_join.header.error = error;
1536  api->ipc_response_send (conn, &res_lib_cpg_join, sizeof(res_lib_cpg_join));
1537 }
1538 
1539 /* Leave message from the library */
1540 static void message_handler_req_lib_cpg_leave (void *conn, const void *message)
1541 {
1543  cs_error_t error = CS_OK;
1544  struct req_lib_cpg_leave *req_lib_cpg_leave = (struct req_lib_cpg_leave *)message;
1545  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1546 
1547  log_printf(LOGSYS_LEVEL_DEBUG, "got leave request on %p", conn);
1548 
1549  switch (cpd->cpd_state) {
1550  case CPD_STATE_UNJOINED:
1551  error = CS_ERR_NOT_EXIST;
1552  break;
1554  error = CS_ERR_NOT_EXIST;
1555  break;
1557  error = CS_ERR_BUSY;
1558  break;
1560  error = CS_OK;
1562  cpg_node_joinleave_send (req_lib_cpg_leave->pid,
1563  &req_lib_cpg_leave->group_name,
1566  break;
1567  }
1568 
1569  /* send return */
1570  res_lib_cpg_leave.header.size = sizeof(res_lib_cpg_leave);
1572  res_lib_cpg_leave.header.error = error;
1574 }
1575 
1576 /* Finalize message from library */
1577 static void message_handler_req_lib_cpg_finalize (
1578  void *conn,
1579  const void *message)
1580 {
1581  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1583  cs_error_t error = CS_OK;
1584 
1585  log_printf (LOGSYS_LEVEL_DEBUG, "cpg finalize for conn=%p", conn);
1586 
1587  /*
1588  * We will just remove cpd from list. After this call, connection will be
1589  * closed on lib side, and cpg_lib_exit_fn will be called
1590  */
1591  qb_list_del (&cpd->list);
1592  qb_list_init (&cpd->list);
1593 
1594  res_lib_cpg_finalize.header.size = sizeof (res_lib_cpg_finalize);
1596  res_lib_cpg_finalize.header.error = error;
1597 
1599  sizeof (res_lib_cpg_finalize));
1600 }
1601 
1602 static int
1603 memory_map (
1604  const char *path,
1605  size_t bytes,
1606  void **buf)
1607 {
1608  int32_t fd;
1609  void *addr;
1610  int32_t res;
1611 
1612  fd = open (path, O_RDWR, 0600);
1613 
1614  unlink (path);
1615 
1616  if (fd == -1) {
1617  return (-1);
1618  }
1619 
1620  res = ftruncate (fd, bytes);
1621  if (res == -1) {
1622  goto error_close_unlink;
1623  }
1624 
1625  addr = mmap (NULL, bytes, PROT_READ | PROT_WRITE,
1626  MAP_SHARED, fd, 0);
1627 
1628  if (addr == MAP_FAILED) {
1629  goto error_close_unlink;
1630  }
1631 #ifdef MADV_NOSYNC
1632  madvise(addr, bytes, MADV_NOSYNC);
1633 #endif
1634 
1635  res = close (fd);
1636  if (res) {
1637  munmap (addr, bytes);
1638  return (-1);
1639  }
1640  *buf = addr;
1641  return (0);
1642 
1643 error_close_unlink:
1644  close (fd);
1645  unlink(path);
1646  return -1;
1647 }
1648 
1649 static inline int zcb_alloc (
1650  struct cpg_pd *cpd,
1651  const char *path_to_file,
1652  size_t size,
1653  void **addr)
1654 {
1655  struct zcb_mapped *zcb_mapped;
1656  unsigned int res;
1657 
1658  zcb_mapped = malloc (sizeof (struct zcb_mapped));
1659  if (zcb_mapped == NULL) {
1660  return (-1);
1661  }
1662 
1663  res = memory_map (
1664  path_to_file,
1665  size,
1666  addr);
1667  if (res == -1) {
1668  free (zcb_mapped);
1669  return (-1);
1670  }
1671 
1672  qb_list_init (&zcb_mapped->list);
1673  zcb_mapped->addr = *addr;
1674  zcb_mapped->size = size;
1675  qb_list_add_tail (&zcb_mapped->list, &cpd->zcb_mapped_list_head);
1676  return (0);
1677 }
1678 
1679 
1680 static inline int zcb_free (struct zcb_mapped *zcb_mapped)
1681 {
1682  unsigned int res;
1683 
1684  res = munmap (zcb_mapped->addr, zcb_mapped->size);
1685  qb_list_del (&zcb_mapped->list);
1686  free (zcb_mapped);
1687  return (res);
1688 }
1689 
1690 static inline int zcb_by_addr_free (struct cpg_pd *cpd, void *addr)
1691 {
1692  struct qb_list_head *list, *tmp_iter;
1693  struct zcb_mapped *zcb_mapped;
1694  unsigned int res = 0;
1695 
1696  qb_list_for_each_safe(list, tmp_iter, &(cpd->zcb_mapped_list_head)) {
1697  zcb_mapped = qb_list_entry (list, struct zcb_mapped, list);
1698 
1699  if (zcb_mapped->addr == addr) {
1700  res = zcb_free (zcb_mapped);
1701  break;
1702  }
1703 
1704  }
1705  return (res);
1706 }
1707 
1708 static inline int zcb_all_free (
1709  struct cpg_pd *cpd)
1710 {
1711  struct qb_list_head *list, *tmp_iter;
1712  struct zcb_mapped *zcb_mapped;
1713 
1714  qb_list_for_each_safe(list, tmp_iter, &(cpd->zcb_mapped_list_head)) {
1715  zcb_mapped = qb_list_entry (list, struct zcb_mapped, list);
1716 
1717  zcb_free (zcb_mapped);
1718  }
1719  return (0);
1720 }
1721 
1722 union u {
1723  uint64_t server_addr;
1724  void *server_ptr;
1725 };
1726 
1727 static uint64_t void2serveraddr (void *server_ptr)
1728 {
1729  union u u;
1730 
1731  u.server_ptr = server_ptr;
1732  return (u.server_addr);
1733 }
1734 
1735 static void *serveraddr2void (uint64_t server_addr)
1736 {
1737  union u u;
1738 
1740  return (u.server_ptr);
1741 };
1742 
1743 static void message_handler_req_lib_cpg_zc_alloc (
1744  void *conn,
1745  const void *message)
1746 {
1748  struct qb_ipc_response_header res_header;
1749  void *addr = NULL;
1750  struct coroipcs_zc_header *zc_header;
1751  unsigned int res;
1752  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1753 
1754  log_printf(LOGSYS_LEVEL_DEBUG, "path: %s", hdr->path_to_file);
1755 
1756  res = zcb_alloc (cpd, hdr->path_to_file, hdr->map_size,
1757  &addr);
1758  assert(res == 0);
1759 
1760  zc_header = (struct coroipcs_zc_header *)addr;
1761  zc_header->server_address = void2serveraddr(addr);
1762 
1763  res_header.size = sizeof (struct qb_ipc_response_header);
1764  res_header.id = 0;
1765  api->ipc_response_send (conn,
1766  &res_header,
1767  res_header.size);
1768 }
1769 
1770 static void message_handler_req_lib_cpg_zc_free (
1771  void *conn,
1772  const void *message)
1773 {
1775  struct qb_ipc_response_header res_header;
1776  void *addr = NULL;
1777  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1778 
1779  log_printf(LOGSYS_LEVEL_DEBUG, " free'ing");
1780 
1781  addr = serveraddr2void (hdr->server_address);
1782 
1783  zcb_by_addr_free (cpd, addr);
1784 
1785  res_header.size = sizeof (struct qb_ipc_response_header);
1786  res_header.id = 0;
1787  api->ipc_response_send (
1788  conn, &res_header,
1789  res_header.size);
1790 }
1791 
1792 /* Fragmented mcast message from the library */
1793 static void message_handler_req_lib_cpg_partial_mcast (void *conn, const void *message)
1794 {
1795  const struct req_lib_cpg_partial_mcast *req_lib_cpg_mcast = message;
1796  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1797  mar_cpg_name_t group_name = cpd->group_name;
1798 
1799  struct iovec req_exec_cpg_iovec[2];
1800  struct req_exec_cpg_partial_mcast req_exec_cpg_mcast;
1801  struct res_lib_cpg_partial_send res_lib_cpg_partial_send;
1802  int msglen = req_lib_cpg_mcast->fraglen;
1803  int result;
1804  cs_error_t error = CS_ERR_NOT_EXIST;
1805 
1806  log_printf(LOGSYS_LEVEL_TRACE, "got fragmented mcast request on %p", conn);
1807  log_printf(LOGSYS_LEVEL_DEBUG, "Sending fragmented message size = %d bytes\n", msglen);
1808 
1809  switch (cpd->cpd_state) {
1810  case CPD_STATE_UNJOINED:
1811  error = CS_ERR_NOT_EXIST;
1812  break;
1814  error = CS_ERR_NOT_EXIST;
1815  break;
1817  error = CS_OK;
1818  break;
1820  error = CS_OK;
1821  break;
1822  }
1823 
1824  res_lib_cpg_partial_send.header.size = sizeof(res_lib_cpg_partial_send);
1825  res_lib_cpg_partial_send.header.id = MESSAGE_RES_CPG_PARTIAL_SEND;
1826 
1827  if (req_lib_cpg_mcast->type == LIBCPG_PARTIAL_FIRST) {
1829  }
1831  error = CS_ERR_INTERRUPT;
1832  }
1833 
1834  if (error == CS_OK) {
1835  req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen;
1836  req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
1838  req_exec_cpg_mcast.pid = cpd->pid;
1839  req_exec_cpg_mcast.msglen = req_lib_cpg_mcast->msglen;
1840  req_exec_cpg_mcast.type = req_lib_cpg_mcast->type;
1841  req_exec_cpg_mcast.fraglen = req_lib_cpg_mcast->fraglen;
1842  api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
1843  memcpy(&req_exec_cpg_mcast.group_name, &group_name,
1844  sizeof(mar_cpg_name_t));
1845 
1846  req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
1847  req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
1848  req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message;
1849  req_exec_cpg_iovec[1].iov_len = msglen;
1850 
1851  result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
1852  assert(result == 0);
1853  } else {
1854  log_printf(LOGSYS_LEVEL_ERROR, "*** %p can't mcast to group %s state:%d, error:%d",
1855  conn, group_name.value, cpd->cpd_state, error);
1856  }
1857 
1858  res_lib_cpg_partial_send.header.error = error;
1859  api->ipc_response_send (conn, &res_lib_cpg_partial_send,
1860  sizeof (res_lib_cpg_partial_send));
1861 }
1862 
1863 /* Mcast message from the library */
1864 static void message_handler_req_lib_cpg_mcast (void *conn, const void *message)
1865 {
1866  const struct req_lib_cpg_mcast *req_lib_cpg_mcast = message;
1867  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1868  mar_cpg_name_t group_name = cpd->group_name;
1869 
1870  struct iovec req_exec_cpg_iovec[2];
1871  struct req_exec_cpg_mcast req_exec_cpg_mcast;
1872  int msglen = req_lib_cpg_mcast->msglen;
1873  int result;
1874  cs_error_t error = CS_ERR_NOT_EXIST;
1875 
1876  log_printf(LOGSYS_LEVEL_TRACE, "got mcast request on %p", conn);
1877 
1878  switch (cpd->cpd_state) {
1879  case CPD_STATE_UNJOINED:
1880  error = CS_ERR_NOT_EXIST;
1881  break;
1883  error = CS_ERR_NOT_EXIST;
1884  break;
1886  error = CS_OK;
1887  break;
1889  error = CS_OK;
1890  break;
1891  }
1892 
1893  if (error == CS_OK) {
1894  req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen;
1895  req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
1897  req_exec_cpg_mcast.pid = cpd->pid;
1898  req_exec_cpg_mcast.msglen = msglen;
1899  api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
1900  memcpy(&req_exec_cpg_mcast.group_name, &group_name,
1901  sizeof(mar_cpg_name_t));
1902 
1903  req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
1904  req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
1905  req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message;
1906  req_exec_cpg_iovec[1].iov_len = msglen;
1907 
1908  result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
1909  assert(result == 0);
1910  } else {
1911  log_printf(LOGSYS_LEVEL_ERROR, "*** %p can't mcast to group %s state:%d, error:%d",
1912  conn, group_name.value, cpd->cpd_state, error);
1913  }
1914 }
1915 
1916 static void message_handler_req_lib_cpg_zc_execute (
1917  void *conn,
1918  const void *message)
1919 {
1921  struct qb_ipc_request_header *header;
1922  struct res_lib_cpg_mcast res_lib_cpg_mcast;
1923  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
1924  struct iovec req_exec_cpg_iovec[2];
1925  struct req_exec_cpg_mcast req_exec_cpg_mcast;
1927  int result;
1928  cs_error_t error = CS_ERR_NOT_EXIST;
1929 
1930  log_printf(LOGSYS_LEVEL_TRACE, "got ZC mcast request on %p", conn);
1931 
1932  header = (struct qb_ipc_request_header *)(((char *)serveraddr2void(hdr->server_address) + sizeof (struct coroipcs_zc_header)));
1933  req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)header;
1934 
1935  switch (cpd->cpd_state) {
1936  case CPD_STATE_UNJOINED:
1937  error = CS_ERR_NOT_EXIST;
1938  break;
1940  error = CS_ERR_NOT_EXIST;
1941  break;
1943  error = CS_OK;
1944  break;
1946  error = CS_OK;
1947  break;
1948  }
1949 
1950  res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast);
1951  res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_MCAST;
1952  if (error == CS_OK) {
1953  req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + req_lib_cpg_mcast->msglen;
1954  req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
1956  req_exec_cpg_mcast.pid = cpd->pid;
1957  req_exec_cpg_mcast.msglen = req_lib_cpg_mcast->msglen;
1958  api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
1959  memcpy(&req_exec_cpg_mcast.group_name, &cpd->group_name,
1960  sizeof(mar_cpg_name_t));
1961 
1962  req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
1963  req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
1964  req_exec_cpg_iovec[1].iov_base = (char *)header + sizeof(struct req_lib_cpg_mcast);
1965  req_exec_cpg_iovec[1].iov_len = req_exec_cpg_mcast.msglen;
1966 
1967  result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
1968  if (result == 0) {
1969  res_lib_cpg_mcast.header.error = CS_OK;
1970  } else {
1971  res_lib_cpg_mcast.header.error = CS_ERR_TRY_AGAIN;
1972  }
1973  } else {
1974  res_lib_cpg_mcast.header.error = error;
1975  }
1976 
1977  api->ipc_response_send (conn, &res_lib_cpg_mcast,
1978  sizeof (res_lib_cpg_mcast));
1979 
1980 }
1981 
1982 static void message_handler_req_lib_cpg_membership (void *conn,
1983  const void *message)
1984 {
1986  (struct req_lib_cpg_membership_get *)message;
1987  struct res_lib_cpg_membership_get res_lib_cpg_membership_get;
1988  struct qb_list_head *iter;
1989  int member_count = 0;
1990 
1991  res_lib_cpg_membership_get.header.id = MESSAGE_RES_CPG_MEMBERSHIP;
1992  res_lib_cpg_membership_get.header.error = CS_OK;
1993  res_lib_cpg_membership_get.header.size =
1994  sizeof (struct res_lib_cpg_membership_get);
1995 
1996  qb_list_for_each(iter, &process_info_list_head) {
1997  struct process_info *pi = qb_list_entry (iter, struct process_info, list);
1998  if (mar_name_compare (&pi->group, &req_lib_cpg_membership_get->group_name) == 0) {
1999  res_lib_cpg_membership_get.member_list[member_count].nodeid = pi->nodeid;
2000  res_lib_cpg_membership_get.member_list[member_count].pid = pi->pid;
2001  member_count += 1;
2002  }
2003  }
2004  res_lib_cpg_membership_get.member_count = member_count;
2005 
2006  api->ipc_response_send (conn, &res_lib_cpg_membership_get,
2007  sizeof (res_lib_cpg_membership_get));
2008 }
2009 
2010 static void message_handler_req_lib_cpg_local_get (void *conn,
2011  const void *message)
2012 {
2013  struct res_lib_cpg_local_get res_lib_cpg_local_get;
2014 
2015  res_lib_cpg_local_get.header.size = sizeof (res_lib_cpg_local_get);
2016  res_lib_cpg_local_get.header.id = MESSAGE_RES_CPG_LOCAL_GET;
2017  res_lib_cpg_local_get.header.error = CS_OK;
2018  res_lib_cpg_local_get.local_nodeid = api->totem_nodeid_get ();
2019 
2020  api->ipc_response_send (conn, &res_lib_cpg_local_get,
2021  sizeof (res_lib_cpg_local_get));
2022 }
2023 
2024 static void message_handler_req_lib_cpg_iteration_initialize (
2025  void *conn,
2026  const void *message)
2027 {
2029  struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
2030  hdb_handle_t cpg_iteration_handle = 0;
2031  struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize;
2032  struct qb_list_head *iter, *iter2;
2033  struct cpg_iteration_instance *cpg_iteration_instance;
2034  cs_error_t error = CS_OK;
2035  int res;
2036 
2037  log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration initialize");
2038 
2039  /* Because between calling this function and *next can be some operations which will
2040  * change list, we must do full copy.
2041  */
2042 
2043  /*
2044  * Create new iteration instance
2045  */
2046  res = hdb_handle_create (&cpg_iteration_handle_t_db, sizeof (struct cpg_iteration_instance),
2047  &cpg_iteration_handle);
2048 
2049  if (res != 0) {
2050  error = CS_ERR_NO_MEMORY;
2051  goto response_send;
2052  }
2053 
2054  res = hdb_handle_get (&cpg_iteration_handle_t_db, cpg_iteration_handle, (void *)&cpg_iteration_instance);
2055 
2056  if (res != 0) {
2057  error = CS_ERR_BAD_HANDLE;
2058  goto error_destroy;
2059  }
2060 
2061  qb_list_init (&cpg_iteration_instance->items_list_head);
2062  cpg_iteration_instance->handle = cpg_iteration_handle;
2063 
2064  /*
2065  * Create copy of process_info list "grouped by" group name
2066  */
2067  qb_list_for_each(iter, &process_info_list_head) {
2068  struct process_info *pi = qb_list_entry (iter, struct process_info, list);
2069  struct process_info *new_pi;
2070 
2071  if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) {
2072  /*
2073  * Try to find processed group name in our list new list
2074  */
2075  int found = 0;
2076 
2077  qb_list_for_each(iter2, &(cpg_iteration_instance->items_list_head)) {
2078  struct process_info *pi2 = qb_list_entry (iter2, struct process_info, list);
2079 
2080  if (mar_name_compare (&pi2->group, &pi->group) == 0) {
2081  found = 1;
2082  break;
2083  }
2084  }
2085 
2086  if (found) {
2087  /*
2088  * We have this name in list -> don't add
2089  */
2090  continue ;
2091  }
2092  } else if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_ONE_GROUP) {
2093  /*
2094  * Test pi group name with request
2095  */
2096  if (mar_name_compare (&pi->group, &req_lib_cpg_iterationinitialize->group_name) != 0)
2097  /*
2098  * Not same -> don't add
2099  */
2100  continue ;
2101  }
2102 
2103  new_pi = malloc (sizeof (struct process_info));
2104  if (!new_pi) {
2105  log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct");
2106 
2107  error = CS_ERR_NO_MEMORY;
2108 
2109  goto error_put_destroy;
2110  }
2111 
2112  memcpy (new_pi, pi, sizeof (struct process_info));
2113  qb_list_init (&new_pi->list);
2114 
2115  if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) {
2116  /*
2117  * pid and nodeid -> undefined
2118  */
2119  new_pi->pid = new_pi->nodeid = 0;
2120  }
2121 
2122  /*
2123  * We will return list "grouped" by "group name", so try to find right place to add
2124  */
2125  qb_list_for_each(iter2, &(cpg_iteration_instance->items_list_head)) {
2126  struct process_info *pi2 = qb_list_entry (iter2, struct process_info, list);
2127 
2128  if (mar_name_compare (&pi2->group, &pi->group) == 0) {
2129  break;
2130  }
2131  }
2132 
2133  qb_list_add (&new_pi->list, iter2);
2134  }
2135 
2136  /*
2137  * Now we have a full "grouped by" copy of process_info list
2138  */
2139 
2140  /*
2141  * Add instance to current cpd list
2142  */
2143  qb_list_init (&cpg_iteration_instance->list);
2144  qb_list_add (&cpg_iteration_instance->list, &cpd->iteration_instance_list_head);
2145 
2146  cpg_iteration_instance->current_pointer = &cpg_iteration_instance->items_list_head;
2147 
2148 error_put_destroy:
2149  hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_handle);
2150 error_destroy:
2151  if (error != CS_OK) {
2152  hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_handle);
2153  }
2154 
2155 response_send:
2156  res_lib_cpg_iterationinitialize.header.size = sizeof (res_lib_cpg_iterationinitialize);
2157  res_lib_cpg_iterationinitialize.header.id = MESSAGE_RES_CPG_ITERATIONINITIALIZE;
2158  res_lib_cpg_iterationinitialize.header.error = error;
2159  res_lib_cpg_iterationinitialize.iteration_handle = cpg_iteration_handle;
2160 
2161  api->ipc_response_send (conn, &res_lib_cpg_iterationinitialize,
2162  sizeof (res_lib_cpg_iterationinitialize));
2163 }
2164 
2165 static void message_handler_req_lib_cpg_iteration_next (
2166  void *conn,
2167  const void *message)
2168 {
2169  const struct req_lib_cpg_iterationnext *req_lib_cpg_iterationnext = message;
2170  struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext;
2171  struct cpg_iteration_instance *cpg_iteration_instance;
2172  cs_error_t error = CS_OK;
2173  int res;
2174  struct process_info *pi;
2175 
2176  log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration next");
2177 
2178  res = hdb_handle_get (&cpg_iteration_handle_t_db,
2179  req_lib_cpg_iterationnext->iteration_handle,
2180  (void *)&cpg_iteration_instance);
2181 
2182  if (res != 0) {
2183  error = CS_ERR_LIBRARY;
2184  goto error_exit;
2185  }
2186 
2187  assert (cpg_iteration_instance);
2188 
2189  cpg_iteration_instance->current_pointer = cpg_iteration_instance->current_pointer->next;
2190 
2191  if (cpg_iteration_instance->current_pointer == &cpg_iteration_instance->items_list_head) {
2192  error = CS_ERR_NO_SECTIONS;
2193  goto error_put;
2194  }
2195 
2196  pi = qb_list_entry (cpg_iteration_instance->current_pointer, struct process_info, list);
2197 
2198  /*
2199  * Copy iteration data
2200  */
2201  res_lib_cpg_iterationnext.description.nodeid = pi->nodeid;
2202  res_lib_cpg_iterationnext.description.pid = pi->pid;
2203  memcpy (&res_lib_cpg_iterationnext.description.group,
2204  &pi->group,
2205  sizeof (mar_cpg_name_t));
2206 
2207 error_put:
2208  hdb_handle_put (&cpg_iteration_handle_t_db, req_lib_cpg_iterationnext->iteration_handle);
2209 error_exit:
2210  res_lib_cpg_iterationnext.header.size = sizeof (res_lib_cpg_iterationnext);
2211  res_lib_cpg_iterationnext.header.id = MESSAGE_RES_CPG_ITERATIONNEXT;
2212  res_lib_cpg_iterationnext.header.error = error;
2213 
2214  api->ipc_response_send (conn, &res_lib_cpg_iterationnext,
2215  sizeof (res_lib_cpg_iterationnext));
2216 }
2217 
2218 static void message_handler_req_lib_cpg_iteration_finalize (
2219  void *conn,
2220  const void *message)
2221 {
2223  struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize;
2224  struct cpg_iteration_instance *cpg_iteration_instance;
2225  cs_error_t error = CS_OK;
2226  int res;
2227 
2228  log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration finalize");
2229 
2230  res = hdb_handle_get (&cpg_iteration_handle_t_db,
2231  req_lib_cpg_iterationfinalize->iteration_handle,
2232  (void *)&cpg_iteration_instance);
2233 
2234  if (res != 0) {
2235  error = CS_ERR_LIBRARY;
2236  goto error_exit;
2237  }
2238 
2239  assert (cpg_iteration_instance);
2240 
2241  cpg_iteration_instance_finalize (cpg_iteration_instance);
2242  hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle);
2243 
2244 error_exit:
2245  res_lib_cpg_iterationfinalize.header.size = sizeof (res_lib_cpg_iterationfinalize);
2246  res_lib_cpg_iterationfinalize.header.id = MESSAGE_RES_CPG_ITERATIONFINALIZE;
2247  res_lib_cpg_iterationfinalize.header.error = error;
2248 
2249  api->ipc_response_send (conn, &res_lib_cpg_iterationfinalize,
2250  sizeof (res_lib_cpg_iterationfinalize));
2251 }
void *(* ipc_private_data_get)(void *conn)
Definition: coroapi.h:256
#define TOTEM_AGREED
Definition: coroapi.h:102
int initial_totem_conf_sent
Definition: exec/cpg.c:151
const char * name
Definition: coroapi.h:491
Definition: exec/cpg.c:1722
mar_cpg_address_t member_list[]
Definition: ipc_cpg.h:390
mar_req_coroipcc_zc_free_t struct
Definition: ipc_cpg.h:481
#define CPG_MAX_NAME_LENGTH
Definition: cpg.h:116
mar_cpg_address_t struct
Definition: ipc_cpg.h:155
uint64_t initial_transition_counter
Definition: exec/cpg.c:153
#define LOGSYS_LEVEL_TRACE
Definition: logsys.h:77
mar_uint32_t sender_nodeid
Definition: exec/cpg.c:495
#define CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF
Definition: cpg.h:193
The req_lib_cpg_join struct.
Definition: ipc_cpg.h:251
mar_req_coroipcc_zc_alloc_t struct
Definition: ipc_cpg.h:472
The cpg_name struct.
Definition: cpg.h:120
The corosync_service_engine struct.
Definition: coroapi.h:490
int(* ipc_dispatch_iov_send)(void *conn, const struct iovec *iov, unsigned int iov_len)
Definition: coroapi.h:265
int(* ipc_response_send)(void *conn, const void *msg, size_t mlen)
Definition: coroapi.h:258
struct corosync_service_engine * cpg_get_service_engine_ver0(void)
Definition: exec/cpg.c:447
cpg_sync_state
Definition: exec/cpg.c:138
The res_lib_cpg_partial_deliver_callback struct.
Definition: ipc_cpg.h:345
The req_lib_cpg_mcast struct.
Definition: ipc_cpg.h:304
mar_cpg_name_t group
Definition: exec/cpg.c:187
The corosync_lib_handler struct.
Definition: coroapi.h:467
The res_lib_cpg_membership_get struct.
Definition: ipc_cpg.h:375
The res_lib_cpg_iterationnext struct.
Definition: ipc_cpg.h:449
hdb_handle_t handle
Definition: exec/cpg.c:160
uint32_t pid
Definition: exec/cpg.c:186
#define CPG_MEMBERS_MAX
Definition: cpg.h:125
The res_lib_cpg_iterationinitialize struct.
Definition: ipc_cpg.h:433
The corosync_exec_handler struct.
Definition: coroapi.h:475
struct qb_list_head list
Definition: exec/cpg.c:498
int(* totem_mcast)(const struct iovec *iovec, unsigned int iov_len, unsigned int guarantee)
Definition: coroapi.h:279
coroipcs_zc_header struct
Definition: ipc_cpg.h:498
uint64_t transition_counter
Definition: exec/cpg.c:152
#define log_printf(level, format, args...)
Definition: logsys.h:323
The res_lib_cpg_partial_send struct.
Definition: ipc_cpg.h:297
void(* exec_handler_fn)(const void *msg, unsigned int nodeid)
Definition: coroapi.h:476
uint64_t server_address
Definition: ipc_cpg.h:500
void * conn
Definition: exec/cpg.c:146
#define SERVICE_ID_MAKE(a, b)
Definition: coroapi.h:458
The req_lib_cpg_iterationinitialize struct.
Definition: ipc_cpg.h:424
struct qb_list_head list
Definition: exec/cpg.c:90
#define LOGSYS_LEVEL_WARNING
Definition: logsys.h:73
The res_lib_cpg_join struct.
Definition: ipc_cpg.h:261
unsigned int flags
Definition: exec/cpg.c:150
unsigned int(* totem_nodeid_get)(void)
Definition: coroapi.h:275
uint32_t pid
Definition: exec/cpg.c:496
void(* ipc_refcnt_dec)(void *conn)
Definition: coroapi.h:270
struct qb_list_head * current_pointer
Definition: exec/cpg.c:163
struct qb_list_head list
Definition: exec/cpg.c:161
mar_req_coroipcc_zc_execute_t struct
Definition: ipc_cpg.h:490
The res_lib_cpg_mcast struct.
Definition: ipc_cpg.h:326
#define LOGSYS_LEVEL_ERROR
Definition: logsys.h:72
size_t size
Definition: exec/cpg.c:92
mar_cpg_name_t struct
Definition: ipc_cpg.h:112
void * server_ptr
Definition: exec/cpg.c:1724
uint32_t pid
Definition: exec/cpg.c:148
cs_error_t
The cs_error_t enum.
Definition: corotypes.h:94
The req_lib_cpg_leave struct.
Definition: ipc_cpg.h:408
#define LOGSYS_LEVEL_DEBUG
Definition: logsys.h:76
mar_cpg_address_t member_list[PROCESSOR_COUNT_MAX]
Definition: ipc_cpg.h:378
mar_cpg_name_t group_name
Definition: exec/cpg.c:194
The req_lib_cpg_iterationfinalize struct.
Definition: ipc_cpg.h:457
uint8_t mar_uint8_t
Definition: mar_gen.h:51
struct qb_list_head list
Definition: exec/cpg.c:154
mar_cpg_name_t group_name
Definition: exec/cpg.c:147
The corosync_api_v1 struct.
Definition: coroapi.h:225
typedef __attribute__
cpg_message_req_types
Definition: exec/cpg.c:79
LOGSYS_DECLARE_SUBSYS("CPG")
DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db, NULL)
struct totem_message_header header
Definition: totemsrp.c:260
#define swab32(x)
The swab32 macro.
Definition: swab.h:51
enum cpd_state cpd_state
Definition: exec/cpg.c:149
The res_lib_cpg_finalize struct.
Definition: ipc_cpg.h:275
uint32_t mar_uint32_t
Definition: mar_gen.h:53
unsigned int nodeid
Definition: exec/cpg.c:185
struct qb_list_head items_list_head
Definition: exec/cpg.c:162
struct qb_list_head zcb_mapped_list_head
Definition: exec/cpg.c:156
struct qb_list_head list
Definition: exec/cpg.c:188
The res_lib_cpg_local_get struct.
Definition: ipc_cpg.h:289
#define PROCESSOR_COUNT_MAX
Definition: coroapi.h:96
Definition: exec/cpg.c:192
qb_handle_t hdb_handle_t
Definition: hdb.h:52
The memb_ring_id struct.
Definition: coroapi.h:122
The res_lib_cpg_iterationfinalize struct.
Definition: ipc_cpg.h:465
struct corosync_service_engine cpg_service_engine
Definition: exec/cpg.c:426
The req_lib_cpg_partial_mcast struct.
Definition: ipc_cpg.h:314
unsigned int nodeid
Definition: coroapi.h:123
The req_lib_cpg_iterationnext struct.
Definition: ipc_cpg.h:441
const char *(* totem_ifaces_print)(unsigned int nodeid)
Definition: coroapi.h:290
uint32_t pid
Definition: exec/cpg.c:193
The res_lib_cpg_confchg_callback struct.
Definition: ipc_cpg.h:384
void * addr
Definition: exec/cpg.c:91
mar_cpg_name_t group_name
Definition: exec/cpg.c:497
unsigned long long seq
Definition: coroapi.h:124
struct qb_list_head iteration_instance_list_head
Definition: exec/cpg.c:155
void(* lib_handler_fn)(void *conn, const void *msg)
Definition: coroapi.h:468
char type
Definition: totem.h:55
The req_lib_cpg_membership_get struct.
Definition: ipc_cpg.h:367
QB_LIST_DECLARE(cpg_pd_list_head)
int(* ipc_dispatch_send)(void *conn, const void *msg, size_t mlen)
Definition: coroapi.h:263
uint64_t server_addr
Definition: exec/cpg.c:1723
The res_lib_cpg_leave struct.
Definition: ipc_cpg.h:417
unsigned int nodeid
Definition: coroapi.h:75
struct memb_ring_id ring_id
Definition: totemsrp.c:264
mar_cpg_ring_id_t struct
Definition: ipc_cpg.h:230
void(* ipc_source_set)(mar_message_source_t *source, void *conn)
Definition: coroapi.h:252
cpd_state
Definition: exec/cpg.c:131
The res_lib_cpg_totem_confchg_callback struct.
Definition: ipc_cpg.h:398
Message from another node.
Definition: ipc_cpg.h:333
The mar_message_source_t struct.
Definition: coroapi.h:50
void(* ipc_refcnt_inc)(void *conn)
Definition: coroapi.h:268