diff options
Diffstat (limited to 'exec/cpg.c')
| -rw-r--r-- | exec/cpg.c | 2344 |
1 files changed, 2344 insertions, 0 deletions
diff --git a/exec/cpg.c b/exec/cpg.c new file mode 100644 index 0000000..0439d14 --- /dev/null +++ b/exec/cpg.c @@ -0,0 +1,2344 @@ +/* + * 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, ¬ify_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, ¬ify_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)); +} |