exec/cpg.c

Go to the documentation of this file.

/*
 * Copyright (c) 2006-2019 Red Hat, Inc.
 *
 * All rights reserved.
 *
 * Author: Christine Caulfield (ccaulfie@redhat.com)
 * Author: Jan Friesse (jfriesse@redhat.com)
 *
 * This software licensed under BSD license, the text of which follows:
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * - Redistributions of source code must retain the above copyright notice,
 *   this list of conditions and the following disclaimer.
 * - Redistributions in binary form must reproduce the above copyright notice,
 *   this list of conditions and the following disclaimer in the documentation
 *   and/or other materials provided with the distribution.
 * - Neither the name of the MontaVista Software, Inc. nor the names of its
 *   contributors may be used to endorse or promote products derived from this
 *   software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
 * THE POSSIBILITY OF SUCH DAMAGE.
 */
 
#include <config.h>
 
#ifdef HAVE_ALLOCA_H
#include <alloca.h>
#endif
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/ioctl.h>
#include <netinet/in.h>
#include <sys/uio.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <time.h>
#include <assert.h>
#include <arpa/inet.h>
#include <sys/mman.h>
 
#include <qb/qblist.h>
#include <qb/qbmap.h>
 
#include <corosync/corotypes.h>
#include <qb/qbipc_common.h>
#include <corosync/corodefs.h>
#include <corosync/logsys.h>
#include <corosync/coroapi.h>
 
#include <corosync/cpg.h>
#include <corosync/ipc_cpg.h>
 
#ifndef MAP_ANONYMOUS
#define MAP_ANONYMOUS MAP_ANON
#endif
 
#include "service.h"
 
LOGSYS_DECLARE_SUBSYS ("CPG");
 
#define GROUP_HASH_SIZE 32
 
enum cpg_message_req_types {
        MESSAGE_REQ_EXEC_CPG_PROCJOIN = 0,
        MESSAGE_REQ_EXEC_CPG_PROCLEAVE = 1,
        MESSAGE_REQ_EXEC_CPG_JOINLIST = 2,
        MESSAGE_REQ_EXEC_CPG_MCAST = 3,
        MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD = 4,
        MESSAGE_REQ_EXEC_CPG_DOWNLIST = 5,
        MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST = 6,
};
 
struct zcb_mapped {
        struct qb_list_head list;
        void *addr;
        size_t size;
};
/*
 * state`               exec deliver
 * match group name, pid -> if matched deliver for YES:
 * XXX indicates impossible state
 *
 *                      join                    leave                   mcast
 * UNJOINED             XXX                     XXX                     NO
 * LEAVE_STARTED        XXX                     YES(unjoined_enter)     YES
 * JOIN_STARTED         YES(join_started_enter) XXX                     NO
 * JOIN_COMPLETED       XXX                     NO                      YES
 *
 * join_started_enter
 *      set JOIN_COMPLETED
 *      add entry to process_info list
 * unjoined_enter
 *      set UNJOINED
 *      delete entry from process_info list
 *
 *
 *                      library accept join error codes
 * UNJOINED             YES(CS_OK)                      set JOIN_STARTED
 * LEAVE_STARTED        NO(CS_ERR_BUSY)
 * JOIN_STARTED         NO(CS_ERR_EXIST)
 * JOIN_COMPlETED       NO(CS_ERR_EXIST)
 *
 *                      library accept leave error codes
 * UNJOINED             NO(CS_ERR_NOT_EXIST)
 * LEAVE_STARTED        NO(CS_ERR_NOT_EXIST)
 * JOIN_STARTED         NO(CS_ERR_BUSY)
 * JOIN_COMPLETED       YES(CS_OK)                      set LEAVE_STARTED
 *
 *                      library accept mcast
 * UNJOINED             NO(CS_ERR_NOT_EXIST)
 * LEAVE_STARTED        NO(CS_ERR_NOT_EXIST)
 * JOIN_STARTED         YES(CS_OK)
 * JOIN_COMPLETED       YES(CS_OK)
 */
enum cpd_state {
        CPD_STATE_UNJOINED,
        CPD_STATE_LEAVE_STARTED,
        CPD_STATE_JOIN_STARTED,
        CPD_STATE_JOIN_COMPLETED
};
 
enum cpg_sync_state {
        CPGSYNC_DOWNLIST,
        CPGSYNC_JOINLIST
};
 
static struct qb_list_head joinlist_messages_head;
 
struct cpg_pd {
        void *conn;
        mar_cpg_name_t group_name;
        uint32_t pid;
        enum cpd_state cpd_state;
        unsigned int flags;
        int initial_totem_conf_sent;
        uint64_t transition_counter; /* These two are used when sending fragmented messages */
        uint64_t initial_transition_counter;
        struct qb_list_head list;
        struct qb_list_head iteration_instance_list_head;
        struct qb_list_head zcb_mapped_list_head;
};
 
struct cpg_iteration_instance {
        hdb_handle_t handle;
        struct qb_list_head list;
        struct qb_list_head items_list_head; /* List of process_info */
        struct qb_list_head *current_pointer;
};
 
DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db,NULL);
 
QB_LIST_DECLARE (cpg_pd_list_head);
 
static unsigned int my_member_list[PROCESSOR_COUNT_MAX];
 
static unsigned int my_member_list_entries;
 
static unsigned int my_old_member_list[PROCESSOR_COUNT_MAX];
 
static unsigned int my_old_member_list_entries = 0;
 
static struct corosync_api_v1 *api = NULL;
 
static enum cpg_sync_state my_sync_state = CPGSYNC_DOWNLIST;
 
static mar_cpg_ring_id_t last_sync_ring_id;
 
struct process_info {
        unsigned int nodeid;
        uint32_t pid;
        mar_cpg_name_t group;
        struct qb_list_head list; /* on the group_info members list */
};
QB_LIST_DECLARE (process_info_list_head);
 
struct join_list_entry {
        uint32_t pid;
        mar_cpg_name_t group_name;
};
 
struct join_list_confchg_data {
        mar_cpg_name_t cpg_group;
        mar_cpg_address_t join_list[CPG_MEMBERS_MAX];
        int join_list_entries;
};
 
/*
 * Service Interfaces required by service_message_handler struct
 */
static char *cpg_exec_init_fn (struct corosync_api_v1 *);
 
static int cpg_lib_init_fn (void *conn);
 
static int cpg_lib_exit_fn (void *conn);
 
static void message_handler_req_exec_cpg_procjoin (
        const void *message,
        unsigned int nodeid);
 
static void message_handler_req_exec_cpg_procleave (
        const void *message,
        unsigned int nodeid);
 
static void message_handler_req_exec_cpg_joinlist (
        const void *message,
        unsigned int nodeid);
 
static void message_handler_req_exec_cpg_mcast (
        const void *message,
        unsigned int nodeid);
 
static void message_handler_req_exec_cpg_partial_mcast (
        const void *message,
        unsigned int nodeid);
 
static void message_handler_req_exec_cpg_downlist_old (
        const void *message,
        unsigned int nodeid);
 
static void message_handler_req_exec_cpg_downlist (
        const void *message,
        unsigned int nodeid);
 
static void exec_cpg_procjoin_endian_convert (void *msg);
 
static void exec_cpg_joinlist_endian_convert (void *msg);
 
static void exec_cpg_mcast_endian_convert (void *msg);
 
static void exec_cpg_partial_mcast_endian_convert (void *msg);
 
static void exec_cpg_downlist_endian_convert_old (void *msg);
 
static void exec_cpg_downlist_endian_convert (void *msg);
 
static void message_handler_req_lib_cpg_join (void *conn, const void *message);
 
static void message_handler_req_lib_cpg_leave (void *conn, const void *message);
 
static void message_handler_req_lib_cpg_finalize (void *conn, const void *message);
 
static void message_handler_req_lib_cpg_mcast (void *conn, const void *message);
 
static void message_handler_req_lib_cpg_partial_mcast (void *conn, const void *message);
 
static void message_handler_req_lib_cpg_membership (void *conn,
                                                    const void *message);
 
static void message_handler_req_lib_cpg_local_get (void *conn,
                                                   const void *message);
 
static void message_handler_req_lib_cpg_iteration_initialize (
        void *conn,
        const void *message);
 
static void message_handler_req_lib_cpg_iteration_next (
        void *conn,
        const void *message);
 
static void message_handler_req_lib_cpg_iteration_finalize (
        void *conn,
        const void *message);
 
static void message_handler_req_lib_cpg_zc_alloc (
        void *conn,
        const void *message);
 
static void message_handler_req_lib_cpg_zc_free (
        void *conn,
        const void *message);
 
static void message_handler_req_lib_cpg_zc_execute (
        void *conn,
        const void *message);
 
static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason);
 
static int cpg_exec_send_downlist(void);
 
static int cpg_exec_send_joinlist(void);
 
static void downlist_inform_clients (void);
 
static void joinlist_inform_clients (void);
 
static void joinlist_messages_delete (void);
 
static void cpg_sync_init (
        const unsigned int *trans_list,
        size_t trans_list_entries,
        const unsigned int *member_list,
        size_t member_list_entries,
        const struct memb_ring_id *ring_id);
 
static int  cpg_sync_process (void);
 
static void cpg_sync_activate (void);
 
static void cpg_sync_abort (void);
 
static void do_proc_join(
        const mar_cpg_name_t *name,
        uint32_t pid,
        unsigned int nodeid,
        int reason,
        qb_map_t *group_notify_map);
 
static void do_proc_leave(
        const mar_cpg_name_t *name,
        uint32_t pid,
        unsigned int nodeid,
        int reason);
 
static int notify_lib_totem_membership (
        void *conn,
        int member_list_entries,
        const unsigned int *member_list);
 
static inline int zcb_all_free (
        struct cpg_pd *cpd);
 
static char *cpg_print_group_name (
        const mar_cpg_name_t *group);
 
/*
 * Library Handler Definition
 */
static struct corosync_lib_handler cpg_lib_engine[] =
{
        { /* 0 - MESSAGE_REQ_CPG_JOIN */
                .lib_handler_fn                         = message_handler_req_lib_cpg_join,
                .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED
        },
        { /* 1 - MESSAGE_REQ_CPG_LEAVE */
                .lib_handler_fn                         = message_handler_req_lib_cpg_leave,
                .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED
        },
        { /* 2 - MESSAGE_REQ_CPG_MCAST */
                .lib_handler_fn                         = message_handler_req_lib_cpg_mcast,
                .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED
        },
        { /* 3 - MESSAGE_REQ_CPG_MEMBERSHIP */
                .lib_handler_fn                         = message_handler_req_lib_cpg_membership,
                .flow_control                           = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
        },
        { /* 4 - MESSAGE_REQ_CPG_LOCAL_GET */
                .lib_handler_fn                         = message_handler_req_lib_cpg_local_get,
                .flow_control                           = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
        },
        { /* 5 - MESSAGE_REQ_CPG_ITERATIONINITIALIZE */
                .lib_handler_fn                         = message_handler_req_lib_cpg_iteration_initialize,
                .flow_control                           = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
        },
        { /* 6 - MESSAGE_REQ_CPG_ITERATIONNEXT */
                .lib_handler_fn                         = message_handler_req_lib_cpg_iteration_next,
                .flow_control                           = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
        },
        { /* 7 - MESSAGE_REQ_CPG_ITERATIONFINALIZE */
                .lib_handler_fn                         = message_handler_req_lib_cpg_iteration_finalize,
                .flow_control                           = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
        },
        { /* 8 - MESSAGE_REQ_CPG_FINALIZE */
                .lib_handler_fn                         = message_handler_req_lib_cpg_finalize,
                .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED
        },
        { /* 9 */
                .lib_handler_fn                         = message_handler_req_lib_cpg_zc_alloc,
                .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED
        },
        { /* 10 */
                .lib_handler_fn                         = message_handler_req_lib_cpg_zc_free,
                .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED
        },
        { /* 11 */
                .lib_handler_fn                         = message_handler_req_lib_cpg_zc_execute,
                .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED
        },
        { /* 12 */
                .lib_handler_fn                         = message_handler_req_lib_cpg_partial_mcast,
                .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED
        },
 
};
 
static struct corosync_exec_handler cpg_exec_engine[] =
{
        { /* 0 - MESSAGE_REQ_EXEC_CPG_PROCJOIN */
                .exec_handler_fn        = message_handler_req_exec_cpg_procjoin,
                .exec_endian_convert_fn = exec_cpg_procjoin_endian_convert
        },
        { /* 1 - MESSAGE_REQ_EXEC_CPG_PROCLEAVE */
                .exec_handler_fn        = message_handler_req_exec_cpg_procleave,
                .exec_endian_convert_fn = exec_cpg_procjoin_endian_convert
        },
        { /* 2 - MESSAGE_REQ_EXEC_CPG_JOINLIST */
                .exec_handler_fn        = message_handler_req_exec_cpg_joinlist,
                .exec_endian_convert_fn = exec_cpg_joinlist_endian_convert
        },
        { /* 3 - MESSAGE_REQ_EXEC_CPG_MCAST */
                .exec_handler_fn        = message_handler_req_exec_cpg_mcast,
                .exec_endian_convert_fn = exec_cpg_mcast_endian_convert
        },
        { /* 4 - MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD */
                .exec_handler_fn        = message_handler_req_exec_cpg_downlist_old,
                .exec_endian_convert_fn = exec_cpg_downlist_endian_convert_old
        },
        { /* 5 - MESSAGE_REQ_EXEC_CPG_DOWNLIST */
                .exec_handler_fn        = message_handler_req_exec_cpg_downlist,
                .exec_endian_convert_fn = exec_cpg_downlist_endian_convert
        },
        { /* 6 - MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST */
                .exec_handler_fn        = message_handler_req_exec_cpg_partial_mcast,
                .exec_endian_convert_fn = exec_cpg_partial_mcast_endian_convert
        },
};
 
struct corosync_service_engine cpg_service_engine = {
        .name                                   = "corosync cluster closed process group service v1.01",
        .id                                     = CPG_SERVICE,
        .priority                               = 1,
        .private_data_size                      = sizeof (struct cpg_pd),
        .flow_control                           = CS_LIB_FLOW_CONTROL_REQUIRED,
        .allow_inquorate                        = CS_LIB_ALLOW_INQUORATE,
        .lib_init_fn                            = cpg_lib_init_fn,
        .lib_exit_fn                            = cpg_lib_exit_fn,
        .lib_engine                             = cpg_lib_engine,
        .lib_engine_count                       = sizeof (cpg_lib_engine) / sizeof (struct corosync_lib_handler),
        .exec_init_fn                           = cpg_exec_init_fn,
        .exec_dump_fn                           = NULL,
        .exec_engine                            = cpg_exec_engine,
        .exec_engine_count                      = sizeof (cpg_exec_engine) / sizeof (struct corosync_exec_handler),
        .sync_init                              = cpg_sync_init,
        .sync_process                           = cpg_sync_process,
        .sync_activate                          = cpg_sync_activate,
        .sync_abort                             = cpg_sync_abort
};
 
struct corosync_service_engine *cpg_get_service_engine_ver0 (void)
{
        return (&cpg_service_engine);
}
 
struct req_exec_cpg_procjoin {
        struct qb_ipc_request_header header __attribute__((aligned(8)));
        mar_cpg_name_t group_name __attribute__((aligned(8)));
        mar_uint32_t pid __attribute__((aligned(8)));
        mar_uint32_t reason __attribute__((aligned(8)));
};
 
struct req_exec_cpg_mcast {
        struct qb_ipc_request_header header __attribute__((aligned(8)));
        mar_cpg_name_t group_name __attribute__((aligned(8)));
        mar_uint32_t msglen __attribute__((aligned(8)));
        mar_uint32_t pid __attribute__((aligned(8)));
        mar_message_source_t source __attribute__((aligned(8)));
        mar_uint8_t message[] __attribute__((aligned(8)));
};
 
struct req_exec_cpg_partial_mcast {
        struct qb_ipc_request_header header __attribute__((aligned(8)));
        mar_cpg_name_t group_name __attribute__((aligned(8)));
        mar_uint32_t msglen __attribute__((aligned(8)));
        mar_uint32_t fraglen __attribute__((aligned(8)));
        mar_uint32_t pid __attribute__((aligned(8)));
        mar_uint32_t type __attribute__((aligned(8)));
        mar_message_source_t source __attribute__((aligned(8)));
        mar_uint8_t message[] __attribute__((aligned(8)));
};
 
struct req_exec_cpg_downlist_old {
        struct qb_ipc_request_header header __attribute__((aligned(8)));
        mar_uint32_t left_nodes __attribute__((aligned(8)));
        mar_uint32_t nodeids[PROCESSOR_COUNT_MAX]  __attribute__((aligned(8)));
};
 
struct req_exec_cpg_downlist {
        struct qb_ipc_request_header header __attribute__((aligned(8)));
        /* merge decisions */
        mar_uint32_t old_members __attribute__((aligned(8)));
        /* downlist below */
        mar_uint32_t left_nodes __attribute__((aligned(8)));
        mar_uint32_t nodeids[PROCESSOR_COUNT_MAX]  __attribute__((aligned(8)));
};
 
struct joinlist_msg {
        mar_uint32_t sender_nodeid;
        uint32_t pid;
        mar_cpg_name_t group_name;
        struct qb_list_head list;
};
 
static struct req_exec_cpg_downlist g_req_exec_cpg_downlist;
 
/*
 * Function print group name. It's not reentrant
 */
static char *cpg_print_group_name(const mar_cpg_name_t *group)
{
        static char res[CPG_MAX_NAME_LENGTH * 4 + 1];
        int dest_pos = 0;
        char c;
        int i;
 
        for (i = 0; i < group->length; i++) {
                c = group->value[i];
 
                if (c >= ' ' && c < 0x7f && c != '\\') {
                        res[dest_pos++] = c;
                } else {
                        if (c == '\\') {
                                res[dest_pos++] = '\\';
                                res[dest_pos++] = '\\';
                        } else {
                                snprintf(res + dest_pos, sizeof(res) - dest_pos, "\\x%02X", c);
                                dest_pos += 4;
                        }
                }
        }
        res[dest_pos] = 0;
 
        return (res);
}
 
static void cpg_sync_init (
        const unsigned int *trans_list,
        size_t trans_list_entries,
        const unsigned int *member_list,
        size_t member_list_entries,
        const struct memb_ring_id *ring_id)
{
        int entries;
        int i, j;
        int found;
 
        my_sync_state = CPGSYNC_DOWNLIST;
 
        memcpy (my_member_list, member_list, member_list_entries *
                sizeof (unsigned int));
        my_member_list_entries = member_list_entries;
 
        last_sync_ring_id.nodeid = ring_id->nodeid;
        last_sync_ring_id.seq = ring_id->seq;
 
        entries = 0;
        /*
         * Determine list of nodeids for downlist message
         */
        for (i = 0; i < my_old_member_list_entries; i++) {
                found = 0;
                for (j = 0; j < trans_list_entries; j++) {
                        if (my_old_member_list[i] == trans_list[j]) {
                                found = 1;
                                break;
                        }
                }
                if (found == 0) {
                        g_req_exec_cpg_downlist.nodeids[entries++] =
                                my_old_member_list[i];
                }
        }
        g_req_exec_cpg_downlist.left_nodes = entries;
}
 
static int cpg_sync_process (void)
{
        int res = -1;
 
        if (my_sync_state == CPGSYNC_DOWNLIST) {
                res = cpg_exec_send_downlist();
                if (res == -1) {
                        return (-1);
                }
                my_sync_state = CPGSYNC_JOINLIST;
        }
        if (my_sync_state == CPGSYNC_JOINLIST) {
                res = cpg_exec_send_joinlist();
        }
        return (res);
}
 
static void cpg_sync_activate (void)
{
        memcpy (my_old_member_list, my_member_list,
                my_member_list_entries * sizeof (unsigned int));
        my_old_member_list_entries = my_member_list_entries;
 
        downlist_inform_clients ();
 
        joinlist_inform_clients ();
 
        joinlist_messages_delete ();
 
        notify_lib_totem_membership (NULL, my_member_list_entries, my_member_list);
}
 
static void cpg_sync_abort (void)
{
 
        joinlist_messages_delete ();
}
 
static int notify_lib_totem_membership (
        void *conn,
        int member_list_entries,
        const unsigned int *member_list)
{
        struct qb_list_head *iter;
        char *buf;
        int size;
        struct res_lib_cpg_totem_confchg_callback *res;
 
        size = sizeof(struct res_lib_cpg_totem_confchg_callback) +
                sizeof(mar_uint32_t) * (member_list_entries);
        buf = alloca(size);
        if (!buf)
                return CS_ERR_LIBRARY;
 
        res = (struct res_lib_cpg_totem_confchg_callback *)buf;
        res->member_list_entries = member_list_entries;
        res->header.size = size;
        res->header.id = MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK;
        res->header.error = CS_OK;
 
        memcpy (&res->ring_id, &last_sync_ring_id, sizeof (mar_cpg_ring_id_t));
        memcpy (res->member_list, member_list, res->member_list_entries * sizeof (mar_uint32_t));
 
        if (conn == NULL) {
                qb_list_for_each(iter, &cpg_pd_list_head) {
                        struct cpg_pd *cpg_pd = qb_list_entry (iter, struct cpg_pd, list);
                        api->ipc_dispatch_send (cpg_pd->conn, buf, size);
                }
        } else {
                api->ipc_dispatch_send (conn, buf, size);
        }
 
        return CS_OK;
}
 
/*
 * Helper function for notify_lib_joinlist which prepares member_list using
 * process_info_list with removed left_list items.
 * member_list_entries - When not NULL it contains number of member_list entries
 * member_list - When not NULL it is used as pointer to start of preallocated
 *               array of members. Pointer is adjusted to the end of array on
 *               exit.
 */
static void notify_lib_joinlist_fill_member_list(
        const mar_cpg_name_t *group_name,
        int left_list_entries,
        const mar_cpg_address_t *left_list,
        int *member_list_entries,
        mar_cpg_address_t **member_list)
{
        struct qb_list_head *iter;
        int i;
 
        if (member_list_entries != NULL) {
                *member_list_entries = 0;
        }
 
        qb_list_for_each(iter, &process_info_list_head) {
                struct process_info *pi = qb_list_entry (iter, struct process_info, list);
 
                if (mar_name_compare (&pi->group, group_name) == 0) {
                        int in_left_list = 0;
 
                        for (i = 0; i < left_list_entries; i++) {
                                if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) {
                                        in_left_list = 1;
                                        break ;
                                }
                        }
 
                        if (!in_left_list) {
                                if (member_list_entries != NULL) {
                                        (*member_list_entries)++;
                                }
 
                                if (member_list != NULL) {
                                        (*member_list)->nodeid = pi->nodeid;
                                        (*member_list)->pid = pi->pid;
                                        (*member_list)->reason = CPG_REASON_UNDEFINED;
                                        (*member_list)++;
                                }
                        }
                }
        }
}
 
static int notify_lib_joinlist(
        const mar_cpg_name_t *group_name,
        int joined_list_entries,
        mar_cpg_address_t *joined_list,
        int left_list_entries,
        mar_cpg_address_t *left_list,
        int id)
{
        int size;
        char *buf;
        struct qb_list_head *iter;
        int member_list_entries;
        struct res_lib_cpg_confchg_callback *res;
        mar_cpg_address_t *retgi;
        int i;
 
        /*
         * Find size of member_list (use process_info_list but remove items in left_list)
         */
        notify_lib_joinlist_fill_member_list(group_name, left_list_entries, left_list,
            &member_list_entries, NULL);
 
        size = sizeof(struct res_lib_cpg_confchg_callback) +
                sizeof(mar_cpg_address_t) * (member_list_entries + left_list_entries + joined_list_entries);
        buf = alloca(size);
        if (!buf)
                return CS_ERR_LIBRARY;
 
        res = (struct res_lib_cpg_confchg_callback *)buf;
        res->joined_list_entries = joined_list_entries;
        res->left_list_entries = left_list_entries;
        res->member_list_entries = member_list_entries;
        retgi = res->member_list;
        res->header.size = size;
        res->header.id = id;
        res->header.error = CS_OK;
        memcpy(&res->group_name, group_name, sizeof(mar_cpg_name_t));
 
        /*
         * Fill res->memberlist. Use process_info_list but remove items in left_list.
         */
        notify_lib_joinlist_fill_member_list(group_name, left_list_entries, left_list,
            NULL, &retgi);
 
        /*
         * Fill res->left_list
         */
        if (left_list_entries) {
                memcpy (retgi, left_list, left_list_entries * sizeof(mar_cpg_address_t));
                retgi += left_list_entries;
        }
 
        if (joined_list_entries) {
                /*
                 * Fill res->joined_list
                 */
                memcpy (retgi, joined_list, joined_list_entries * sizeof(mar_cpg_address_t));
                retgi += joined_list_entries;
 
                /*
                 * Update cpd_state for all local joined processes in group
                 */
                for (i = 0; i < joined_list_entries; i++) {
                        if (joined_list[i].nodeid == api->totem_nodeid_get()) {
                                qb_list_for_each(iter, &cpg_pd_list_head) {
                                        struct cpg_pd *cpd = qb_list_entry (iter, struct cpg_pd, list);
                                        if (joined_list[i].pid == cpd->pid &&
                                            mar_name_compare (&cpd->group_name, group_name) == 0) {
                                                cpd->cpd_state = CPD_STATE_JOIN_COMPLETED;
                                        }
                                }
                        }
                }
        }
 
        /*
         * Send notification to all ipc clients joined in group_name
         */
        qb_list_for_each(iter, &cpg_pd_list_head) {
                struct cpg_pd *cpd = qb_list_entry (iter, struct cpg_pd, list);
                if (mar_name_compare (&cpd->group_name, group_name) == 0) {
                        if (cpd->cpd_state == CPD_STATE_JOIN_COMPLETED ||
                                cpd->cpd_state == CPD_STATE_LEAVE_STARTED) {
 
                                api->ipc_dispatch_send (cpd->conn, buf, size);
                                cpd->transition_counter++;
                        }
                }
        }
 
        if (left_list_entries) {
                /*
                 * Zero internal cpd state for all local processes leaving group
                 * (this loop is not strictly needed because left_list always either
                 *  contains exactly one process running on local node or more items
                 *  but none of them is running on local node)
                 */
                for (i = 0; i < joined_list_entries; i++) {
                        if (left_list[i].nodeid == api->totem_nodeid_get() &&
                            left_list[i].reason == CONFCHG_CPG_REASON_LEAVE) {
                                qb_list_for_each(iter, &cpg_pd_list_head) {
                                        struct cpg_pd *cpd = qb_list_entry (iter, struct cpg_pd, list);
                                        if (left_list[i].pid == cpd->pid &&
                                            mar_name_compare (&cpd->group_name, group_name) == 0) {
                                                cpd->pid = 0;
                                                memset (&cpd->group_name, 0, sizeof(cpd->group_name));
                                                cpd->cpd_state = CPD_STATE_UNJOINED;
                                        }
                                }
                        }
                }
        }
 
        /*
         * Traverse thru cpds and send totem membership for cpd, where it is not send yet
         */
        qb_list_for_each(iter, &cpg_pd_list_head) {
                struct cpg_pd *cpd = qb_list_entry (iter, struct cpg_pd, list);
 
                if ((cpd->flags & CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF) && (cpd->initial_totem_conf_sent == 0)) {
                        cpd->initial_totem_conf_sent = 1;
 
                        notify_lib_totem_membership (cpd->conn, my_old_member_list_entries, my_old_member_list);
                }
        }
 
        return CS_OK;
}
 
static void downlist_log(const char *msg, struct req_exec_cpg_downlist *dl)
{
        log_printf (LOG_DEBUG,
                    "%s: members(old:%d left:%d)",
                    msg,
                    dl->old_members,
                    dl->left_nodes);
}
 
static void downlist_inform_clients (void)
{
        struct qb_list_head *iter, *tmp_iter;
        struct process_info *left_pi;
        qb_map_t *group_map;
        struct cpg_name cpg_group;
        mar_cpg_name_t group;
        struct confchg_data{
                struct cpg_name cpg_group;
                mar_cpg_address_t left_list[CPG_MEMBERS_MAX];
                int left_list_entries;
                struct qb_list_head  list;
        } *pcd;
        qb_map_iter_t *miter;
        int i, size;
 
        downlist_log("my downlist", &g_req_exec_cpg_downlist);
 
        group_map = qb_skiplist_create();
 
        /*
         * only the cpg groups included in left nodes should receive
         * confchg event, so we will collect these cpg groups and
         * relative left_lists here.
         */
        qb_list_for_each_safe(iter, tmp_iter, &process_info_list_head) {
                struct process_info *pi = qb_list_entry(iter, struct process_info, list);
 
                left_pi = NULL;
                for (i = 0; i < g_req_exec_cpg_downlist.left_nodes; i++) {
 
                        if (pi->nodeid == g_req_exec_cpg_downlist.nodeids[i]) {
                                left_pi = pi;
                                break;
                        }
                }
 
                if (left_pi) {
                        marshall_from_mar_cpg_name_t(&cpg_group, &left_pi->group);
                        cpg_group.value[cpg_group.length] = 0;
 
                        pcd = (struct confchg_data *)qb_map_get(group_map, cpg_group.value);
                        if (pcd == NULL) {
                                pcd = (struct confchg_data *)calloc(1, sizeof(struct confchg_data));
                                memcpy(&pcd->cpg_group, &cpg_group, sizeof(struct cpg_name));
                                qb_map_put(group_map, pcd->cpg_group.value, pcd);
                        }
                        size = pcd->left_list_entries;
                        pcd->left_list[size].nodeid = left_pi->nodeid;
                        pcd->left_list[size].pid = left_pi->pid;
                        pcd->left_list[size].reason = CONFCHG_CPG_REASON_NODEDOWN;
                        pcd->left_list_entries++;
                        qb_list_del (&left_pi->list);
                        free (left_pi);
                }
        }
 
        /* send only one confchg event per cpg group */
        miter = qb_map_iter_create(group_map);
        while (qb_map_iter_next(miter, (void **)&pcd)) {
                marshall_to_mar_cpg_name_t(&group, &pcd->cpg_group);
 
                log_printf (LOG_DEBUG, "left_list_entries:%d", pcd->left_list_entries);
                for (i=0; i<pcd->left_list_entries; i++) {
                        log_printf (LOG_DEBUG, "left_list[%d] group:%s, ip:%s, pid:%d",
                                i, cpg_print_group_name(&group),
                                (char*)api->totem_ifaces_print(pcd->left_list[i].nodeid),
                                pcd->left_list[i].pid);
                }
 
                /* send confchg event */
                notify_lib_joinlist(&group,
                        0, NULL,
                        pcd->left_list_entries,
                        pcd->left_list,
                        MESSAGE_RES_CPG_CONFCHG_CALLBACK);
 
                free(pcd);
        }
        qb_map_iter_free(miter);
        qb_map_destroy(group_map);
}
 
/*
 * Remove processes that might have left the group while we were suspended.
 */
static void joinlist_remove_zombie_pi_entries (void)
{
        struct qb_list_head *pi_iter, *tmp_iter;
        struct qb_list_head *jl_iter;
        struct process_info *pi;
        struct joinlist_msg *stored_msg;
        int found;
 
        qb_list_for_each_safe(pi_iter, tmp_iter, &process_info_list_head) {
                pi = qb_list_entry (pi_iter, struct process_info, list);
 
                /*
                 * Ignore local node
                 */
                if (pi->nodeid == api->totem_nodeid_get()) {
                        continue ;
                }
 
                /*
                 * Try to find message in joinlist messages
                 */
                found = 0;
                qb_list_for_each(jl_iter, &joinlist_messages_head) {
                        stored_msg = qb_list_entry(jl_iter, struct joinlist_msg, list);
 
                        if (stored_msg->sender_nodeid == api->totem_nodeid_get()) {
                                continue ;
                        }
 
                        if (pi->nodeid == stored_msg->sender_nodeid &&
                            pi->pid == stored_msg->pid &&
                            mar_name_compare (&pi->group, &stored_msg->group_name) == 0) {
                                found = 1;
                                break ;
                        }
                }
 
                if (!found) {
                        do_proc_leave(&pi->group, pi->pid, pi->nodeid, CONFCHG_CPG_REASON_PROCDOWN);
                }
        }
}
 
static void joinlist_inform_clients (void)
{
        struct joinlist_msg *stored_msg;
        struct qb_list_head *iter;
        unsigned int i;
        qb_map_t *group_notify_map;
        qb_map_iter_t *miter;
        struct join_list_confchg_data *jld;
 
        group_notify_map = qb_skiplist_create();
 
        i = 0;
        qb_list_for_each(iter, &joinlist_messages_head) {
                stored_msg = qb_list_entry(iter, struct joinlist_msg, list);
 
                log_printf (LOG_DEBUG, "joinlist_messages[%u] group:%s, ip:%s, pid:%d",
                        i++, cpg_print_group_name(&stored_msg->group_name),
                        (char*)api->totem_ifaces_print(stored_msg->sender_nodeid),
                        stored_msg->pid);
 
                /* Ignore our own messages */
                if (stored_msg->sender_nodeid == api->totem_nodeid_get()) {
                        continue ;
                }
 
                do_proc_join (&stored_msg->group_name, stored_msg->pid, stored_msg->sender_nodeid,
                        CONFCHG_CPG_REASON_NODEUP, group_notify_map);
        }
 
        miter = qb_map_iter_create(group_notify_map);
        while (qb_map_iter_next(miter, (void **)&jld)) {
                notify_lib_joinlist(&jld->cpg_group,
                                    jld->join_list_entries, jld->join_list,
                                    0, NULL,
                                    MESSAGE_RES_CPG_CONFCHG_CALLBACK);
                free(jld);
        }
        qb_map_iter_free(miter);
        qb_map_destroy(group_notify_map);
 
        joinlist_remove_zombie_pi_entries ();
}
 
static void joinlist_messages_delete (void)
{
        struct joinlist_msg *stored_msg;
        struct qb_list_head *iter, *tmp_iter;
 
        qb_list_for_each_safe(iter, tmp_iter, &joinlist_messages_head) {
                stored_msg = qb_list_entry(iter, struct joinlist_msg, list);
                qb_list_del (&stored_msg->list);
                free (stored_msg);
        }
        qb_list_init (&joinlist_messages_head);
}
 
static char *cpg_exec_init_fn (struct corosync_api_v1 *corosync_api)
{
        qb_list_init (&joinlist_messages_head);
        api = corosync_api;
        return (NULL);
}
 
static void cpg_iteration_instance_finalize (struct cpg_iteration_instance *cpg_iteration_instance)
{
        struct qb_list_head *iter, *tmp_iter;
        struct process_info *pi;
 
        qb_list_for_each_safe(iter, tmp_iter, &(cpg_iteration_instance->items_list_head)) {
                pi = qb_list_entry (iter, struct process_info, list);
                qb_list_del (&pi->list);
                free (pi);
        }
 
        qb_list_del (&cpg_iteration_instance->list);
        hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle);
}
 
static void cpg_pd_finalize (struct cpg_pd *cpd)
{
        struct qb_list_head *iter, *tmp_iter;
        struct cpg_iteration_instance *cpii;
 
        zcb_all_free(cpd);
        qb_list_for_each_safe(iter, tmp_iter, &(cpd->iteration_instance_list_head)) {
                cpii = qb_list_entry (iter, struct cpg_iteration_instance, list);
 
                cpg_iteration_instance_finalize (cpii);
        }
 
        qb_list_del (&cpd->list);
}
 
static int cpg_lib_exit_fn (void *conn)
{
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
 
        log_printf(LOGSYS_LEVEL_DEBUG, "exit_fn for conn=%p", conn);
 
        if (cpd->group_name.length > 0 && cpd->cpd_state != CPD_STATE_LEAVE_STARTED) {
                cpg_node_joinleave_send (cpd->pid, &cpd->group_name,
                                MESSAGE_REQ_EXEC_CPG_PROCLEAVE, CONFCHG_CPG_REASON_PROCDOWN);
        }
 
        cpg_pd_finalize (cpd);
 
        api->ipc_refcnt_dec (conn);
        return (0);
}
 
static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason)
{
        struct req_exec_cpg_procjoin req_exec_cpg_procjoin;
        struct iovec req_exec_cpg_iovec;
        int result;
 
        memset(&req_exec_cpg_procjoin, 0, sizeof(req_exec_cpg_procjoin));
 
        memcpy(&req_exec_cpg_procjoin.group_name, group_name, sizeof(mar_cpg_name_t));
        req_exec_cpg_procjoin.pid = pid;
        req_exec_cpg_procjoin.reason = reason;
 
        req_exec_cpg_procjoin.header.size = sizeof(req_exec_cpg_procjoin);
        req_exec_cpg_procjoin.header.id = SERVICE_ID_MAKE(CPG_SERVICE, fn);
 
        req_exec_cpg_iovec.iov_base = (char *)&req_exec_cpg_procjoin;
        req_exec_cpg_iovec.iov_len = sizeof(req_exec_cpg_procjoin);
 
        result = api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED);
 
        return (result);
}
 
/* Can byteswap join & leave messages */
static void exec_cpg_procjoin_endian_convert (void *msg)
{
        struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = msg;
 
        req_exec_cpg_procjoin->pid = swab32(req_exec_cpg_procjoin->pid);
        swab_mar_cpg_name_t (&req_exec_cpg_procjoin->group_name);
        req_exec_cpg_procjoin->reason = swab32(req_exec_cpg_procjoin->reason);
}
 
static void exec_cpg_joinlist_endian_convert (void *msg_v)
{
        char *msg = msg_v;
        struct qb_ipc_response_header *res = (struct qb_ipc_response_header *)msg;
        struct join_list_entry *jle = (struct join_list_entry *)(msg + sizeof(struct qb_ipc_response_header));
 
        swab_mar_int32_t (&res->size);
 
        while ((const char*)jle < msg + res->size) {
                jle->pid = swab32(jle->pid);
                swab_mar_cpg_name_t (&jle->group_name);
                jle++;
        }
}
 
static void exec_cpg_downlist_endian_convert_old (void *msg)
{
}
 
static void exec_cpg_downlist_endian_convert (void *msg)
{
        struct req_exec_cpg_downlist *req_exec_cpg_downlist = msg;
        unsigned int i;
 
        req_exec_cpg_downlist->left_nodes = swab32(req_exec_cpg_downlist->left_nodes);
        req_exec_cpg_downlist->old_members = swab32(req_exec_cpg_downlist->old_members);
 
        for (i = 0; i < req_exec_cpg_downlist->left_nodes; i++) {
                req_exec_cpg_downlist->nodeids[i] = swab32(req_exec_cpg_downlist->nodeids[i]);
        }
}
 
 
static void exec_cpg_mcast_endian_convert (void *msg)
{
        struct req_exec_cpg_mcast *req_exec_cpg_mcast = msg;
 
        swab_coroipc_request_header_t (&req_exec_cpg_mcast->header);
        swab_mar_cpg_name_t (&req_exec_cpg_mcast->group_name);
        req_exec_cpg_mcast->pid = swab32(req_exec_cpg_mcast->pid);
        req_exec_cpg_mcast->msglen = swab32(req_exec_cpg_mcast->msglen);
        swab_mar_message_source_t (&req_exec_cpg_mcast->source);
}
 
static void exec_cpg_partial_mcast_endian_convert (void *msg)
{
        struct req_exec_cpg_partial_mcast *req_exec_cpg_mcast = msg;
 
        swab_coroipc_request_header_t (&req_exec_cpg_mcast->header);
        swab_mar_cpg_name_t (&req_exec_cpg_mcast->group_name);
        req_exec_cpg_mcast->pid = swab32(req_exec_cpg_mcast->pid);
        req_exec_cpg_mcast->msglen = swab32(req_exec_cpg_mcast->msglen);
        req_exec_cpg_mcast->fraglen = swab32(req_exec_cpg_mcast->fraglen);
        req_exec_cpg_mcast->type = swab32(req_exec_cpg_mcast->type);
        swab_mar_message_source_t (&req_exec_cpg_mcast->source);
}
 
static struct process_info *process_info_find(const mar_cpg_name_t *group_name, uint32_t pid, unsigned int nodeid) {
        struct qb_list_head *iter;
 
        qb_list_for_each(iter, &process_info_list_head) {
                struct process_info *pi = qb_list_entry (iter, struct process_info, list);
 
                if (pi->pid == pid && pi->nodeid == nodeid &&
                        mar_name_compare (&pi->group, group_name) == 0) {
                                return pi;
                }
        }
 
        return NULL;
}
 
static void do_proc_join(
        const mar_cpg_name_t *name,
        uint32_t pid,
        unsigned int nodeid,
        int reason,
        qb_map_t *group_notify_map)
{
        struct process_info *pi;
        struct process_info *pi_entry;
        mar_cpg_address_t notify_info;
        struct qb_list_head *list;
        struct qb_list_head *list_to_add = NULL;
        int size;
 
        if (process_info_find (name, pid, nodeid) != NULL) {
                return ;
        }
        pi = malloc (sizeof (struct process_info));
        if (!pi) {
                log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct");
                return;
        }
        pi->nodeid = nodeid;
        pi->pid = pid;
        memcpy(&pi->group, name, sizeof(*name));
        qb_list_init(&pi->list);
 
        /*
         * Insert new process in sorted order so synchronization works properly
         */
        list_to_add = &process_info_list_head;
        qb_list_for_each(list, &process_info_list_head) {
                pi_entry = qb_list_entry(list, struct process_info, list);
                if (pi_entry->nodeid > pi->nodeid ||
                        (pi_entry->nodeid == pi->nodeid && pi_entry->pid > pi->pid)) {
 
                        break;
                }
                list_to_add = list;
        }
        qb_list_add (&pi->list, list_to_add);
 
        notify_info.pid = pi->pid;
        notify_info.nodeid = nodeid;
        notify_info.reason = reason;
 
        if (group_notify_map == NULL) {
                notify_lib_joinlist(&pi->group,
                                    1, &notify_info,
                                    0, NULL,
                                    MESSAGE_RES_CPG_CONFCHG_CALLBACK);
        } else {
                struct join_list_confchg_data *jld = qb_map_get(group_notify_map, pi->group.value);
                if (jld == NULL) {
                        jld = (struct join_list_confchg_data *)calloc(1, sizeof(struct join_list_confchg_data));
                        memcpy(&jld->cpg_group, &pi->group, sizeof(mar_cpg_name_t));
                        qb_map_put(group_notify_map, jld->cpg_group.value, jld);
                }
                size = jld->join_list_entries;
                jld->join_list[size].nodeid = notify_info.nodeid;
                jld->join_list[size].pid = notify_info.pid;
                jld->join_list[size].reason = notify_info.reason;
                jld->join_list_entries++;
        }
}
 
static void do_proc_leave(
        const mar_cpg_name_t *name,
        uint32_t pid,
        unsigned int nodeid,
        int reason)
{
        struct process_info *pi;
        struct qb_list_head *iter, *tmp_iter;
        mar_cpg_address_t notify_info;
 
        notify_info.pid = pid;
        notify_info.nodeid = nodeid;
        notify_info.reason = reason;
 
        notify_lib_joinlist(name,
                0, NULL,
                1, &notify_info,
                MESSAGE_RES_CPG_CONFCHG_CALLBACK);
 
        qb_list_for_each_safe(iter, tmp_iter, &process_info_list_head) {
                pi = qb_list_entry(iter, struct process_info, list);
 
                if (pi->pid == pid && pi->nodeid == nodeid &&
                        mar_name_compare (&pi->group, name)==0) {
                        qb_list_del (&pi->list);
                        free (pi);
                }
        }
}
 
static void message_handler_req_exec_cpg_downlist_old (
        const void *message,
        unsigned int nodeid)
{
        log_printf (LOGSYS_LEVEL_DEBUG, "downlist OLD from node " CS_PRI_NODE_ID,
                nodeid);
}
 
static void message_handler_req_exec_cpg_downlist(
        const void *message,
        unsigned int nodeid)
{
        const struct req_exec_cpg_downlist *req_exec_cpg_downlist = message;
 
        log_printf (LOGSYS_LEVEL_DEBUG, "downlist left_list: %d received",
                        req_exec_cpg_downlist->left_nodes);
}
 
 
static void message_handler_req_exec_cpg_procjoin (
        const void *message,
        unsigned int nodeid)
{
        const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message;
 
        log_printf(LOGSYS_LEVEL_DEBUG, "got procjoin message from cluster node " CS_PRI_NODE_ID " (%s) for pid %u",
                nodeid,
                api->totem_ifaces_print(nodeid),
                (unsigned int)req_exec_cpg_procjoin->pid);
 
        do_proc_join (&req_exec_cpg_procjoin->group_name,
                req_exec_cpg_procjoin->pid, nodeid,
                CONFCHG_CPG_REASON_JOIN, NULL);
}
 
static void message_handler_req_exec_cpg_procleave (
        const void *message,
        unsigned int nodeid)
{
        const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message;
 
        log_printf(LOGSYS_LEVEL_DEBUG, "got procleave message from cluster node " CS_PRI_NODE_ID " (%s) for pid %u",
                nodeid,
                api->totem_ifaces_print(nodeid),
                (unsigned int)req_exec_cpg_procjoin->pid);
 
        do_proc_leave (&req_exec_cpg_procjoin->group_name,
                req_exec_cpg_procjoin->pid, nodeid,
                req_exec_cpg_procjoin->reason);
}
 
 
/* Got a proclist from another node */
static void message_handler_req_exec_cpg_joinlist (
        const void *message_v,
        unsigned int nodeid)
{
        const char *message = message_v;
        const struct qb_ipc_response_header *res = (const struct qb_ipc_response_header *)message;
        const struct join_list_entry *jle = (const struct join_list_entry *)(message + sizeof(struct qb_ipc_response_header));
        struct joinlist_msg *stored_msg;
 
        log_printf(LOGSYS_LEVEL_DEBUG, "got joinlist message from node " CS_PRI_NODE_ID,
                nodeid);
 
        while ((const char*)jle < message + res->size) {
                stored_msg = malloc (sizeof (struct joinlist_msg));
                memset(stored_msg, 0, sizeof (struct joinlist_msg));
                stored_msg->sender_nodeid = nodeid;
                stored_msg->pid = jle->pid;
                memcpy(&stored_msg->group_name, &jle->group_name, sizeof(mar_cpg_name_t));
                qb_list_init (&stored_msg->list);
                qb_list_add (&stored_msg->list, &joinlist_messages_head);
                jle++;
        }
}
 
static void message_handler_req_exec_cpg_mcast (
        const void *message,
        unsigned int nodeid)
{
        const struct req_exec_cpg_mcast *req_exec_cpg_mcast = message;
        struct res_lib_cpg_deliver_callback res_lib_cpg_mcast;
        int msglen = req_exec_cpg_mcast->msglen;
        struct qb_list_head *iter, *pi_iter, *tmp_iter;
        struct cpg_pd *cpd;
        struct iovec iovec[2];
        int known_node = 0;
 
        res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_DELIVER_CALLBACK;
        res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast) + msglen;
        res_lib_cpg_mcast.msglen = msglen;
        res_lib_cpg_mcast.pid = req_exec_cpg_mcast->pid;
        res_lib_cpg_mcast.nodeid = nodeid;
 
        memcpy(&res_lib_cpg_mcast.group_name, &req_exec_cpg_mcast->group_name,
                sizeof(mar_cpg_name_t));
        iovec[0].iov_base = (void *)&res_lib_cpg_mcast;
        iovec[0].iov_len = sizeof (res_lib_cpg_mcast);
 
        iovec[1].iov_base = (char*)message+sizeof(*req_exec_cpg_mcast);
        iovec[1].iov_len = msglen;
 
        qb_list_for_each_safe(iter, tmp_iter, &cpg_pd_list_head) {
                cpd = qb_list_entry(iter, struct cpg_pd, list);
                if ((cpd->cpd_state == CPD_STATE_LEAVE_STARTED || cpd->cpd_state == CPD_STATE_JOIN_COMPLETED)
                        && (mar_name_compare (&cpd->group_name, &req_exec_cpg_mcast->group_name) == 0)) {
 
                        if (!known_node) {
                                /* Try to find, if we know the node */
                                qb_list_for_each(pi_iter, &process_info_list_head) {
                                        struct process_info *pi = qb_list_entry (pi_iter, struct process_info, list);
 
                                        if (pi->nodeid == nodeid &&
                                                mar_name_compare (&pi->group, &req_exec_cpg_mcast->group_name) == 0) {
                                                known_node = 1;
                                                break;
                                        }
                                }
                        }
 
                        if (!known_node) {
                                log_printf(LOGSYS_LEVEL_WARNING, "Unknown node -> we will not deliver message");
                                return ;
                        }
 
                        api->ipc_dispatch_iov_send (cpd->conn, iovec, 2);
                }
        }
}
 
static void message_handler_req_exec_cpg_partial_mcast (
        const void *message,
        unsigned int nodeid)
{
        const struct req_exec_cpg_partial_mcast *req_exec_cpg_mcast = message;
        struct res_lib_cpg_partial_deliver_callback res_lib_cpg_mcast;
        int msglen = req_exec_cpg_mcast->fraglen;
        struct qb_list_head *iter, *pi_iter, *tmp_iter;
        struct cpg_pd *cpd;
        struct iovec iovec[2];
        int known_node = 0;
 
        log_printf(LOGSYS_LEVEL_DEBUG, "Got fragmented message from node " CS_PRI_NODE_ID ", size = %d bytes\n", nodeid, msglen);
 
        res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_PARTIAL_DELIVER_CALLBACK;
        res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast) + msglen;
        res_lib_cpg_mcast.fraglen = msglen;
        res_lib_cpg_mcast.msglen = req_exec_cpg_mcast->msglen;
        res_lib_cpg_mcast.pid = req_exec_cpg_mcast->pid;
        res_lib_cpg_mcast.type = req_exec_cpg_mcast->type;
        res_lib_cpg_mcast.nodeid = nodeid;
 
        memcpy(&res_lib_cpg_mcast.group_name, &req_exec_cpg_mcast->group_name,
               sizeof(mar_cpg_name_t));
        iovec[0].iov_base = (void *)&res_lib_cpg_mcast;
        iovec[0].iov_len = sizeof (res_lib_cpg_mcast);
 
        iovec[1].iov_base = (char*)message+sizeof(*req_exec_cpg_mcast);
        iovec[1].iov_len = msglen;
 
        qb_list_for_each_safe(iter, tmp_iter, &cpg_pd_list_head) {
                cpd = qb_list_entry(iter, struct cpg_pd, list);
 
                if ((cpd->cpd_state == CPD_STATE_LEAVE_STARTED || cpd->cpd_state == CPD_STATE_JOIN_COMPLETED)
                    && (mar_name_compare (&cpd->group_name, &req_exec_cpg_mcast->group_name) == 0)) {
 
                        if (!known_node) {
                                /* Try to find, if we know the node */
                                qb_list_for_each(pi_iter, &process_info_list_head) {
                                        struct process_info *pi = qb_list_entry (pi_iter, struct process_info, list);
 
                                        if (pi->nodeid == nodeid &&
                                            mar_name_compare (&pi->group, &req_exec_cpg_mcast->group_name) == 0) {
                                                known_node = 1;
                                                break;
                                        }
                                }
                        }
 
                        if (!known_node) {
                                log_printf(LOGSYS_LEVEL_WARNING, "Unknown node -> we will not deliver message");
                                return ;
                        }
 
                        api->ipc_dispatch_iov_send (cpd->conn, iovec, 2);
                }
        }
}
 
 
static int cpg_exec_send_downlist(void)
{
        struct iovec iov;
 
        g_req_exec_cpg_downlist.header.id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_DOWNLIST);
        g_req_exec_cpg_downlist.header.size = sizeof(struct req_exec_cpg_downlist);
 
        g_req_exec_cpg_downlist.old_members = my_old_member_list_entries;
 
        iov.iov_base = (void *)&g_req_exec_cpg_downlist;
        iov.iov_len = g_req_exec_cpg_downlist.header.size;
 
        return (api->totem_mcast (&iov, 1, TOTEM_AGREED));
}
 
static int cpg_exec_send_joinlist(void)
{
        int count = 0;
        struct qb_list_head *iter;
        struct qb_ipc_response_header *res;
        char *buf;
        size_t buf_size;
        struct join_list_entry *jle;
        struct iovec req_exec_cpg_iovec;
 
        qb_list_for_each(iter, &process_info_list_head) {
                struct process_info *pi = qb_list_entry (iter, struct process_info, list);
 
                if (pi->nodeid == api->totem_nodeid_get ()) {
                        count++;
                }
        }
 
        /* Nothing to send */
        if (!count)
                return 0;
 
        buf_size = sizeof(struct qb_ipc_response_header) + sizeof(struct join_list_entry) * count;
        buf = alloca(buf_size);
        if (!buf) {
                log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate joinlist buffer");
                return -1;
        }
        memset(buf, 0, buf_size);
 
        jle = (struct join_list_entry *)(buf + sizeof(struct qb_ipc_response_header));
        res = (struct qb_ipc_response_header *)buf;
 
        qb_list_for_each(iter, &process_info_list_head) {
                struct process_info *pi = qb_list_entry (iter, struct process_info, list);
 
                if (pi->nodeid == api->totem_nodeid_get ()) {
                        memcpy (&jle->group_name, &pi->group, sizeof (mar_cpg_name_t));
                        jle->pid = pi->pid;
                        jle++;
                }
        }
 
        res->id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_JOINLIST);
        res->size = sizeof(struct qb_ipc_response_header)+sizeof(struct join_list_entry) * count;
 
        req_exec_cpg_iovec.iov_base = buf;
        req_exec_cpg_iovec.iov_len = res->size;
 
        return (api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED));
}
 
static int cpg_lib_init_fn (void *conn)
{
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
        memset (cpd, 0, sizeof(struct cpg_pd));
        cpd->conn = conn;
        qb_list_add (&cpd->list, &cpg_pd_list_head);
 
        qb_list_init (&cpd->iteration_instance_list_head);
        qb_list_init (&cpd->zcb_mapped_list_head);
 
        api->ipc_refcnt_inc (conn);
        log_printf(LOGSYS_LEVEL_DEBUG, "lib_init_fn: conn=%p, cpd=%p", conn, cpd);
        return (0);
}
 
/* Join message from the library */
static void message_handler_req_lib_cpg_join (void *conn, const void *message)
{
        const struct req_lib_cpg_join *req_lib_cpg_join = message;
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
        struct res_lib_cpg_join res_lib_cpg_join;
        cs_error_t error = CS_OK;
        struct qb_list_head *iter;
 
        /* Test, if we don't have same pid and group name joined */
        qb_list_for_each(iter, &cpg_pd_list_head) {
                struct cpg_pd *cpd_item = qb_list_entry (iter, struct cpg_pd, list);
 
                if (cpd_item->pid == req_lib_cpg_join->pid &&
                        mar_name_compare(&req_lib_cpg_join->group_name, &cpd_item->group_name) == 0) {
 
                        /* We have same pid and group name joined -> return error */
                        error = CS_ERR_EXIST;
                        goto response_send;
                }
        }
 
        /*
         * Same check must be done in process info list, because there may be not yet delivered
         * leave of client.
         */
        qb_list_for_each(iter, &process_info_list_head) {
                struct process_info *pi = qb_list_entry (iter, struct process_info, list);
 
                if (pi->nodeid == api->totem_nodeid_get () && pi->pid == req_lib_cpg_join->pid &&
                    mar_name_compare(&req_lib_cpg_join->group_name, &pi->group) == 0) {
                        /* We have same pid and group name joined -> return error */
                        error = CS_ERR_TRY_AGAIN;
                        goto response_send;
                }
        }
 
        if (req_lib_cpg_join->group_name.length > CPG_MAX_NAME_LENGTH) {
                error = CS_ERR_NAME_TOO_LONG;
                goto response_send;
        }
 
        switch (cpd->cpd_state) {
        case CPD_STATE_UNJOINED:
                error = CS_OK;
                cpd->cpd_state = CPD_STATE_JOIN_STARTED;
                cpd->pid = req_lib_cpg_join->pid;
                cpd->flags = req_lib_cpg_join->flags;
                memcpy (&cpd->group_name, &req_lib_cpg_join->group_name,
                        sizeof (cpd->group_name));
 
                cpg_node_joinleave_send (req_lib_cpg_join->pid,
                        &req_lib_cpg_join->group_name,
                        MESSAGE_REQ_EXEC_CPG_PROCJOIN, CONFCHG_CPG_REASON_JOIN);
                break;
        case CPD_STATE_LEAVE_STARTED:
                error = CS_ERR_BUSY;
                break;
        case CPD_STATE_JOIN_STARTED:
                error = CS_ERR_EXIST;
                break;
        case CPD_STATE_JOIN_COMPLETED:
                error = CS_ERR_EXIST;
                break;
        }
 
response_send:
        res_lib_cpg_join.header.size = sizeof(res_lib_cpg_join);
        res_lib_cpg_join.header.id = MESSAGE_RES_CPG_JOIN;
        res_lib_cpg_join.header.error = error;
        api->ipc_response_send (conn, &res_lib_cpg_join, sizeof(res_lib_cpg_join));
}
 
/* Leave message from the library */
static void message_handler_req_lib_cpg_leave (void *conn, const void *message)
{
        struct res_lib_cpg_leave res_lib_cpg_leave;
        cs_error_t error = CS_OK;
        struct req_lib_cpg_leave  *req_lib_cpg_leave = (struct req_lib_cpg_leave *)message;
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
 
        log_printf(LOGSYS_LEVEL_DEBUG, "got leave request on %p", conn);
 
        switch (cpd->cpd_state) {
        case CPD_STATE_UNJOINED:
                error = CS_ERR_NOT_EXIST;
                break;
        case CPD_STATE_LEAVE_STARTED:
                error = CS_ERR_NOT_EXIST;
                break;
        case CPD_STATE_JOIN_STARTED:
                error = CS_ERR_BUSY;
                break;
        case CPD_STATE_JOIN_COMPLETED:
                error = CS_OK;
                cpd->cpd_state = CPD_STATE_LEAVE_STARTED;
                cpg_node_joinleave_send (req_lib_cpg_leave->pid,
                        &req_lib_cpg_leave->group_name,
                        MESSAGE_REQ_EXEC_CPG_PROCLEAVE,
                        CONFCHG_CPG_REASON_LEAVE);
                break;
        }
 
        /* send return */
        res_lib_cpg_leave.header.size = sizeof(res_lib_cpg_leave);
        res_lib_cpg_leave.header.id = MESSAGE_RES_CPG_LEAVE;
        res_lib_cpg_leave.header.error = error;
        api->ipc_response_send(conn, &res_lib_cpg_leave, sizeof(res_lib_cpg_leave));
}
 
/* Finalize message from library */
static void message_handler_req_lib_cpg_finalize (
        void *conn,
        const void *message)
{
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
        struct res_lib_cpg_finalize res_lib_cpg_finalize;
        cs_error_t error = CS_OK;
 
        log_printf (LOGSYS_LEVEL_DEBUG, "cpg finalize for conn=%p", conn);
 
        /*
         * We will just remove cpd from list. After this call, connection will be
         * closed on lib side, and cpg_lib_exit_fn will be called
         */
        qb_list_del (&cpd->list);
        qb_list_init (&cpd->list);
 
        res_lib_cpg_finalize.header.size = sizeof (res_lib_cpg_finalize);
        res_lib_cpg_finalize.header.id = MESSAGE_RES_CPG_FINALIZE;
        res_lib_cpg_finalize.header.error = error;
 
        api->ipc_response_send (conn, &res_lib_cpg_finalize,
                sizeof (res_lib_cpg_finalize));
}
 
static int
memory_map (
        const char *path,
        size_t bytes,
        void **buf)
{
        int32_t fd;
        void *addr;
        int32_t res;
 
        fd = open (path, O_RDWR, 0600);
 
        unlink (path);
 
        if (fd == -1) {
                return (-1);
        }
 
        res = ftruncate (fd, bytes);
        if (res == -1) {
                goto error_close_unlink;
        }
 
        addr = mmap (NULL, bytes, PROT_READ | PROT_WRITE,
                MAP_SHARED, fd, 0);
 
        if (addr == MAP_FAILED) {
                goto error_close_unlink;
        }
#ifdef MADV_NOSYNC
        madvise(addr, bytes, MADV_NOSYNC);
#endif
 
        res = close (fd);
        if (res) {
                munmap (addr, bytes);
                return (-1);
        }
        *buf = addr;
        return (0);
 
error_close_unlink:
        close (fd);
        unlink(path);
        return -1;
}
 
static inline int zcb_alloc (
        struct cpg_pd *cpd,
        const char *path_to_file,
        size_t size,
        void **addr)
{
        struct zcb_mapped *zcb_mapped;
        unsigned int res;
 
        zcb_mapped = malloc (sizeof (struct zcb_mapped));
        if (zcb_mapped == NULL) {
                return (-1);
        }
 
        res = memory_map (
                path_to_file,
                size,
                addr);
        if (res == -1) {
                free (zcb_mapped);
                return (-1);
        }
 
        qb_list_init (&zcb_mapped->list);
        zcb_mapped->addr = *addr;
        zcb_mapped->size = size;
        qb_list_add_tail (&zcb_mapped->list, &cpd->zcb_mapped_list_head);
        return (0);
}
 
 
static inline int zcb_free (struct zcb_mapped *zcb_mapped)
{
        unsigned int res;
 
        res = munmap (zcb_mapped->addr, zcb_mapped->size);
        qb_list_del (&zcb_mapped->list);
        free (zcb_mapped);
        return (res);
}
 
static inline int zcb_by_addr_free (struct cpg_pd *cpd, void *addr)
{
        struct qb_list_head *list, *tmp_iter;
        struct zcb_mapped *zcb_mapped;
        unsigned int res = 0;
 
        qb_list_for_each_safe(list, tmp_iter, &(cpd->zcb_mapped_list_head)) {
                zcb_mapped = qb_list_entry (list, struct zcb_mapped, list);
 
                if (zcb_mapped->addr == addr) {
                        res = zcb_free (zcb_mapped);
                        break;
                }
 
        }
        return (res);
}
 
static inline int zcb_all_free (
        struct cpg_pd *cpd)
{
        struct qb_list_head *list, *tmp_iter;
        struct zcb_mapped *zcb_mapped;
 
        qb_list_for_each_safe(list, tmp_iter, &(cpd->zcb_mapped_list_head)) {
                zcb_mapped = qb_list_entry (list, struct zcb_mapped, list);
 
                zcb_free (zcb_mapped);
        }
        return (0);
}
 
union u {
        uint64_t server_addr;
        void *server_ptr;
};
 
static uint64_t void2serveraddr (void *server_ptr)
{
        union u u;
 
        u.server_ptr = server_ptr;
        return (u.server_addr);
}
 
static void *serveraddr2void (uint64_t server_addr)
{
        union u u;
 
        u.server_addr = server_addr;
        return (u.server_ptr);
};
 
static void message_handler_req_lib_cpg_zc_alloc (
        void *conn,
        const void *message)
{
        mar_req_coroipcc_zc_alloc_t *hdr = (mar_req_coroipcc_zc_alloc_t *)message;
        struct qb_ipc_response_header res_header;
        void *addr = NULL;
        struct coroipcs_zc_header *zc_header;
        unsigned int res;
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
 
        log_printf(LOGSYS_LEVEL_DEBUG, "path: %s", hdr->path_to_file);
 
        res = zcb_alloc (cpd, hdr->path_to_file, hdr->map_size,
                &addr);
        assert(res == 0);
 
        zc_header = (struct coroipcs_zc_header *)addr;
        zc_header->server_address = void2serveraddr(addr);
 
        res_header.size = sizeof (struct qb_ipc_response_header);
        res_header.id = 0;
        api->ipc_response_send (conn,
                &res_header,
                res_header.size);
}
 
static void message_handler_req_lib_cpg_zc_free (
        void *conn,
        const void *message)
{
        mar_req_coroipcc_zc_free_t *hdr = (mar_req_coroipcc_zc_free_t *)message;
        struct qb_ipc_response_header res_header;
        void *addr = NULL;
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
 
        log_printf(LOGSYS_LEVEL_DEBUG, " free'ing");
 
        addr = serveraddr2void (hdr->server_address);
 
        zcb_by_addr_free (cpd, addr);
 
        res_header.size = sizeof (struct qb_ipc_response_header);
        res_header.id = 0;
        api->ipc_response_send (
                conn, &res_header,
                res_header.size);
}
 
/* Fragmented mcast message from the library */
static void message_handler_req_lib_cpg_partial_mcast (void *conn, const void *message)
{
        const struct req_lib_cpg_partial_mcast *req_lib_cpg_mcast = message;
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
        mar_cpg_name_t group_name = cpd->group_name;
 
        struct iovec req_exec_cpg_iovec[2];
        struct req_exec_cpg_partial_mcast req_exec_cpg_mcast;
        struct res_lib_cpg_partial_send res_lib_cpg_partial_send;
        int msglen = req_lib_cpg_mcast->fraglen;
        int result;
        cs_error_t error = CS_ERR_NOT_EXIST;
 
        log_printf(LOGSYS_LEVEL_TRACE, "got fragmented mcast request on %p", conn);
        log_printf(LOGSYS_LEVEL_DEBUG, "Sending fragmented message size = %d bytes\n", msglen);
 
        switch (cpd->cpd_state) {
        case CPD_STATE_UNJOINED:
                error = CS_ERR_NOT_EXIST;
                break;
        case CPD_STATE_LEAVE_STARTED:
                error = CS_ERR_NOT_EXIST;
                break;
        case CPD_STATE_JOIN_STARTED:
                error = CS_OK;
                break;
        case CPD_STATE_JOIN_COMPLETED:
                error = CS_OK;
                break;
        }
 
        res_lib_cpg_partial_send.header.size = sizeof(res_lib_cpg_partial_send);
        res_lib_cpg_partial_send.header.id = MESSAGE_RES_CPG_PARTIAL_SEND;
 
        if (req_lib_cpg_mcast->type == LIBCPG_PARTIAL_FIRST) {
                cpd->initial_transition_counter = cpd->transition_counter;
        }
        if (cpd->transition_counter != cpd->initial_transition_counter) {
                error = CS_ERR_INTERRUPT;
        }
 
        if (error == CS_OK) {
                req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen;
                req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
                                                               MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST);
                req_exec_cpg_mcast.pid = cpd->pid;
                req_exec_cpg_mcast.msglen = req_lib_cpg_mcast->msglen;
                req_exec_cpg_mcast.type = req_lib_cpg_mcast->type;
                req_exec_cpg_mcast.fraglen = req_lib_cpg_mcast->fraglen;
                api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
                memcpy(&req_exec_cpg_mcast.group_name, &group_name,
                       sizeof(mar_cpg_name_t));
 
                req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
                req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
                req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message;
                req_exec_cpg_iovec[1].iov_len = msglen;
 
                result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
                assert(result == 0);
        } else {
                log_printf(LOGSYS_LEVEL_ERROR, "*** %p can't mcast to group %s state:%d, error:%d",
                           conn, group_name.value, cpd->cpd_state, error);
        }
 
        res_lib_cpg_partial_send.header.error = error;
        api->ipc_response_send (conn, &res_lib_cpg_partial_send,
                                sizeof (res_lib_cpg_partial_send));
}
 
/* Mcast message from the library */
static void message_handler_req_lib_cpg_mcast (void *conn, const void *message)
{
        const struct req_lib_cpg_mcast *req_lib_cpg_mcast = message;
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
        mar_cpg_name_t group_name = cpd->group_name;
 
        struct iovec req_exec_cpg_iovec[2];
        struct req_exec_cpg_mcast req_exec_cpg_mcast;
        int msglen = req_lib_cpg_mcast->msglen;
        int result;
        cs_error_t error = CS_ERR_NOT_EXIST;
 
        log_printf(LOGSYS_LEVEL_TRACE, "got mcast request on %p", conn);
 
        switch (cpd->cpd_state) {
        case CPD_STATE_UNJOINED:
                error = CS_ERR_NOT_EXIST;
                break;
        case CPD_STATE_LEAVE_STARTED:
                error = CS_ERR_NOT_EXIST;
                break;
        case CPD_STATE_JOIN_STARTED:
                error = CS_OK;
                break;
        case CPD_STATE_JOIN_COMPLETED:
                error = CS_OK;
                break;
        }
 
        if (error == CS_OK) {
                memset(&req_exec_cpg_mcast, 0, sizeof(req_exec_cpg_mcast));
 
                req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen;
                req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
                        MESSAGE_REQ_EXEC_CPG_MCAST);
                req_exec_cpg_mcast.pid = cpd->pid;
                req_exec_cpg_mcast.msglen = msglen;
                api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
                memcpy(&req_exec_cpg_mcast.group_name, &group_name,
                        sizeof(mar_cpg_name_t));
 
                req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
                req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
                req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message;
                req_exec_cpg_iovec[1].iov_len = msglen;
 
                result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
                assert(result == 0);
        } else {
                log_printf(LOGSYS_LEVEL_ERROR, "*** %p can't mcast to group %s state:%d, error:%d",
                        conn, group_name.value, cpd->cpd_state, error);
        }
}
 
static void message_handler_req_lib_cpg_zc_execute (
        void *conn,
        const void *message)
{
        mar_req_coroipcc_zc_execute_t *hdr = (mar_req_coroipcc_zc_execute_t *)message;
        struct qb_ipc_request_header *header;
        struct res_lib_cpg_mcast res_lib_cpg_mcast;
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
        struct iovec req_exec_cpg_iovec[2];
        struct req_exec_cpg_mcast req_exec_cpg_mcast;
        struct req_lib_cpg_mcast *req_lib_cpg_mcast;
        int result;
        cs_error_t error = CS_ERR_NOT_EXIST;
 
        log_printf(LOGSYS_LEVEL_TRACE, "got ZC mcast request on %p", conn);
 
        header = (struct qb_ipc_request_header *)(((char *)serveraddr2void(hdr->server_address) + sizeof (struct coroipcs_zc_header)));
        req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)header;
 
        switch (cpd->cpd_state) {
        case CPD_STATE_UNJOINED:
                error = CS_ERR_NOT_EXIST;
                break;
        case CPD_STATE_LEAVE_STARTED:
                error = CS_ERR_NOT_EXIST;
                break;
        case CPD_STATE_JOIN_STARTED:
                error = CS_OK;
                break;
        case CPD_STATE_JOIN_COMPLETED:
                error = CS_OK;
                break;
        }
 
        res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast);
        res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_MCAST;
        if (error == CS_OK) {
                req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + req_lib_cpg_mcast->msglen;
                req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
                        MESSAGE_REQ_EXEC_CPG_MCAST);
                req_exec_cpg_mcast.pid = cpd->pid;
                req_exec_cpg_mcast.msglen = req_lib_cpg_mcast->msglen;
                api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
                memcpy(&req_exec_cpg_mcast.group_name, &cpd->group_name,
                        sizeof(mar_cpg_name_t));
 
                req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
                req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
                req_exec_cpg_iovec[1].iov_base = (char *)header + sizeof(struct req_lib_cpg_mcast);
                req_exec_cpg_iovec[1].iov_len = req_exec_cpg_mcast.msglen;
 
                result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
                if (result == 0) {
                        res_lib_cpg_mcast.header.error = CS_OK;
                } else {
                        res_lib_cpg_mcast.header.error = CS_ERR_TRY_AGAIN;
                }
        } else {
                res_lib_cpg_mcast.header.error = error;
        }
 
        api->ipc_response_send (conn, &res_lib_cpg_mcast,
                sizeof (res_lib_cpg_mcast));
 
}
 
static void message_handler_req_lib_cpg_membership (void *conn,
                                                    const void *message)
{
        struct req_lib_cpg_membership_get *req_lib_cpg_membership_get =
                (struct req_lib_cpg_membership_get *)message;
        struct res_lib_cpg_membership_get res_lib_cpg_membership_get;
        struct qb_list_head *iter;
        int member_count = 0;
 
        res_lib_cpg_membership_get.header.id = MESSAGE_RES_CPG_MEMBERSHIP;
        res_lib_cpg_membership_get.header.error = CS_OK;
        res_lib_cpg_membership_get.header.size =
                sizeof (struct res_lib_cpg_membership_get);
 
        qb_list_for_each(iter, &process_info_list_head) {
                struct process_info *pi = qb_list_entry (iter, struct process_info, list);
                if (mar_name_compare (&pi->group, &req_lib_cpg_membership_get->group_name) == 0) {
                        res_lib_cpg_membership_get.member_list[member_count].nodeid = pi->nodeid;
                        res_lib_cpg_membership_get.member_list[member_count].pid = pi->pid;
                        member_count += 1;
                }
        }
        res_lib_cpg_membership_get.member_count = member_count;
 
        api->ipc_response_send (conn, &res_lib_cpg_membership_get,
                sizeof (res_lib_cpg_membership_get));
}
 
static void message_handler_req_lib_cpg_local_get (void *conn,
                                                   const void *message)
{
        struct res_lib_cpg_local_get res_lib_cpg_local_get;
 
        res_lib_cpg_local_get.header.size = sizeof (res_lib_cpg_local_get);
        res_lib_cpg_local_get.header.id = MESSAGE_RES_CPG_LOCAL_GET;
        res_lib_cpg_local_get.header.error = CS_OK;
        res_lib_cpg_local_get.local_nodeid = api->totem_nodeid_get ();
 
        api->ipc_response_send (conn, &res_lib_cpg_local_get,
                sizeof (res_lib_cpg_local_get));
}
 
static void message_handler_req_lib_cpg_iteration_initialize (
        void *conn,
        const void *message)
{
        const struct req_lib_cpg_iterationinitialize *req_lib_cpg_iterationinitialize = message;
        struct cpg_pd *cpd = (struct cpg_pd *)api->ipc_private_data_get (conn);
        hdb_handle_t cpg_iteration_handle = 0;
        struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize;
        struct qb_list_head *iter, *iter2;
        struct cpg_iteration_instance *cpg_iteration_instance;
        cs_error_t error = CS_OK;
        int res;
 
        log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration initialize");
 
        /* Because between calling this function and *next can be some operations which will
         * change list, we must do full copy.
         */
 
        /*
         * Create new iteration instance
         */
        res = hdb_handle_create (&cpg_iteration_handle_t_db, sizeof (struct cpg_iteration_instance),
                        &cpg_iteration_handle);
 
        if (res != 0) {
                error = CS_ERR_NO_MEMORY;
                goto response_send;
        }
 
        res = hdb_handle_get (&cpg_iteration_handle_t_db, cpg_iteration_handle, (void *)&cpg_iteration_instance);
 
        if (res != 0) {
                error = CS_ERR_BAD_HANDLE;
                goto error_destroy;
        }
 
        qb_list_init (&cpg_iteration_instance->items_list_head);
        cpg_iteration_instance->handle = cpg_iteration_handle;
 
        /*
         * Create copy of process_info list "grouped by" group name
         */
        qb_list_for_each(iter, &process_info_list_head) {
                struct process_info *pi = qb_list_entry (iter, struct process_info, list);
                struct process_info *new_pi;
 
                if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) {
                        /*
                         * Try to find processed group name in our list new list
                         */
                        int found = 0;
 
                        qb_list_for_each(iter2, &(cpg_iteration_instance->items_list_head)) {
                                 struct process_info *pi2 = qb_list_entry (iter2, struct process_info, list);
 
                                 if (mar_name_compare (&pi2->group, &pi->group) == 0) {
                                        found = 1;
                                        break;
                                 }
                        }
 
                        if (found) {
                                /*
                                 * We have this name in list -> don't add
                                 */
                                continue ;
                        }
                } else if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_ONE_GROUP) {
                        /*
                         * Test pi group name with request
                         */
                        if (mar_name_compare (&pi->group, &req_lib_cpg_iterationinitialize->group_name) != 0)
                                /*
                                 * Not same -> don't add
                                 */
                                continue ;
                }
 
                new_pi = malloc (sizeof (struct process_info));
                if (!new_pi) {
                        log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct");
 
                        error = CS_ERR_NO_MEMORY;
 
                        goto error_put_destroy;
                }
 
                memcpy (new_pi, pi, sizeof (struct process_info));
                qb_list_init (&new_pi->list);
 
                if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) {
                        /*
                         * pid and nodeid -> undefined
                         */
                        new_pi->pid = new_pi->nodeid = 0;
                }
 
                /*
                 * We will return list "grouped" by "group name", so try to find right place to add
                 */
                qb_list_for_each(iter2, &(cpg_iteration_instance->items_list_head)) {
                        struct process_info *pi2 = qb_list_entry (iter2, struct process_info, list);
 
                        if (mar_name_compare (&pi2->group, &pi->group) == 0) {
                                break;
                        }
                }
 
                qb_list_add (&new_pi->list, iter2);
        }
 
        /*
         * Now we have a full "grouped by" copy of process_info list
         */
 
        /*
         * Add instance to current cpd list
         */
        qb_list_init (&cpg_iteration_instance->list);
        qb_list_add (&cpg_iteration_instance->list, &cpd->iteration_instance_list_head);
 
        cpg_iteration_instance->current_pointer = &cpg_iteration_instance->items_list_head;
 
error_put_destroy:
        hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_handle);
error_destroy:
        if (error != CS_OK) {
                hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_handle);
        }
 
response_send:
        res_lib_cpg_iterationinitialize.header.size = sizeof (res_lib_cpg_iterationinitialize);
        res_lib_cpg_iterationinitialize.header.id = MESSAGE_RES_CPG_ITERATIONINITIALIZE;
        res_lib_cpg_iterationinitialize.header.error = error;
        res_lib_cpg_iterationinitialize.iteration_handle = cpg_iteration_handle;
 
        api->ipc_response_send (conn, &res_lib_cpg_iterationinitialize,
                sizeof (res_lib_cpg_iterationinitialize));
}
 
static void message_handler_req_lib_cpg_iteration_next (
        void *conn,
        const void *message)
{
        const struct req_lib_cpg_iterationnext *req_lib_cpg_iterationnext = message;
        struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext;
        struct cpg_iteration_instance *cpg_iteration_instance;
        cs_error_t error = CS_OK;
        int res;
        struct process_info *pi;
 
        log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration next");
 
        res = hdb_handle_get (&cpg_iteration_handle_t_db,
                        req_lib_cpg_iterationnext->iteration_handle,
                        (void *)&cpg_iteration_instance);
 
        if (res != 0) {
                error = CS_ERR_LIBRARY;
                goto error_exit;
        }
 
        assert (cpg_iteration_instance);
 
        cpg_iteration_instance->current_pointer = cpg_iteration_instance->current_pointer->next;
 
        if (cpg_iteration_instance->current_pointer == &cpg_iteration_instance->items_list_head) {
                error = CS_ERR_NO_SECTIONS;
                goto error_put;
        }
 
        pi = qb_list_entry (cpg_iteration_instance->current_pointer, struct process_info, list);
 
        /*
         * Copy iteration data
         */
        res_lib_cpg_iterationnext.description.nodeid = pi->nodeid;
        res_lib_cpg_iterationnext.description.pid = pi->pid;
        memcpy (&res_lib_cpg_iterationnext.description.group,
                        &pi->group,
                        sizeof (mar_cpg_name_t));
 
error_put:
        hdb_handle_put (&cpg_iteration_handle_t_db, req_lib_cpg_iterationnext->iteration_handle);
error_exit:
        res_lib_cpg_iterationnext.header.size = sizeof (res_lib_cpg_iterationnext);
        res_lib_cpg_iterationnext.header.id = MESSAGE_RES_CPG_ITERATIONNEXT;
        res_lib_cpg_iterationnext.header.error = error;
 
        api->ipc_response_send (conn, &res_lib_cpg_iterationnext,
                sizeof (res_lib_cpg_iterationnext));
}
 
static void message_handler_req_lib_cpg_iteration_finalize (
        void *conn,
        const void *message)
{
        const struct req_lib_cpg_iterationfinalize *req_lib_cpg_iterationfinalize = message;
        struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize;
        struct cpg_iteration_instance *cpg_iteration_instance;
        cs_error_t error = CS_OK;
        int res;
 
        log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration finalize");
 
        res = hdb_handle_get (&cpg_iteration_handle_t_db,
                        req_lib_cpg_iterationfinalize->iteration_handle,
                        (void *)&cpg_iteration_instance);
 
        if (res != 0) {
                error = CS_ERR_LIBRARY;
                goto error_exit;
        }
 
        assert (cpg_iteration_instance);
 
        cpg_iteration_instance_finalize (cpg_iteration_instance);
        hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle);
 
error_exit:
        res_lib_cpg_iterationfinalize.header.size = sizeof (res_lib_cpg_iterationfinalize);
        res_lib_cpg_iterationfinalize.header.id = MESSAGE_RES_CPG_ITERATIONFINALIZE;
        res_lib_cpg_iterationfinalize.header.error = error;
 
        api->ipc_response_send (conn, &res_lib_cpg_iterationfinalize,
                sizeof (res_lib_cpg_iterationfinalize));
}