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