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