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