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)
{
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;
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));
}
config.h
coroapi.h
SERVICE_ID_MAKE
#define SERVICE_ID_MAKE(a, b)
Definition coroapi.h:458
CS_LIB_ALLOW_INQUORATE
@ CS_LIB_ALLOW_INQUORATE
Definition coroapi.h:164
nodeid
unsigned int nodeid
Definition coroapi.h:0
addr
unsigned char addr[TOTEMIP_ADDRLEN]
Definition coroapi.h:2
CS_LIB_FLOW_CONTROL_REQUIRED
@ CS_LIB_FLOW_CONTROL_REQUIRED
Definition coroapi.h:152
CS_LIB_FLOW_CONTROL_NOT_REQUIRED
@ CS_LIB_FLOW_CONTROL_NOT_REQUIRED
Definition coroapi.h:153
TOTEM_AGREED
#define TOTEM_AGREED
Definition coroapi.h:102
PROCESSOR_COUNT_MAX
#define PROCESSOR_COUNT_MAX
Definition coroapi.h:96
corodefs.h
CPG_SERVICE
@ CPG_SERVICE
Definition corodefs.h:46
corotypes.h
CS_PRI_NODE_ID
#define CS_PRI_NODE_ID
Definition corotypes.h:59
cs_error_t
cs_error_t
The cs_error_t enum.
Definition corotypes.h:98
CS_ERR_NO_SECTIONS
@ CS_ERR_NO_SECTIONS
Definition corotypes.h:125
CS_ERR_NAME_TOO_LONG
@ CS_ERR_NAME_TOO_LONG
Definition corotypes.h:111
CS_ERR_NO_MEMORY
@ CS_ERR_NO_MEMORY
Definition corotypes.h:106
CS_ERR_BUSY
@ CS_ERR_BUSY
Definition corotypes.h:108
CS_ERR_BAD_HANDLE
@ CS_ERR_BAD_HANDLE
Definition corotypes.h:107
CS_ERR_TRY_AGAIN
@ CS_ERR_TRY_AGAIN
Definition corotypes.h:104
CS_OK
@ CS_OK
Definition corotypes.h:99
CS_ERR_LIBRARY
@ CS_ERR_LIBRARY
Definition corotypes.h:100
CS_ERR_INTERRUPT
@ CS_ERR_INTERRUPT
Definition corotypes.h:114
CS_ERR_NOT_EXIST
@ CS_ERR_NOT_EXIST
Definition corotypes.h:110
CS_ERR_EXIST
@ CS_ERR_EXIST
Definition corotypes.h:112
cpg.h
cpg_service_engine
struct corosync_service_engine cpg_service_engine
Definition exec/cpg.c:433
QB_LIST_DECLARE
QB_LIST_DECLARE(cpg_pd_list_head)
cpg_message_req_types
cpg_message_req_types
Definition exec/cpg.c:79
MESSAGE_REQ_EXEC_CPG_JOINLIST
@ MESSAGE_REQ_EXEC_CPG_JOINLIST
Definition exec/cpg.c:82
MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST
@ MESSAGE_REQ_EXEC_CPG_PARTIAL_MCAST
Definition exec/cpg.c:86
MESSAGE_REQ_EXEC_CPG_PROCLEAVE
@ MESSAGE_REQ_EXEC_CPG_PROCLEAVE
Definition exec/cpg.c:81
MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD
@ MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD
Definition exec/cpg.c:84
MESSAGE_REQ_EXEC_CPG_MCAST
@ MESSAGE_REQ_EXEC_CPG_MCAST
Definition exec/cpg.c:83
MESSAGE_REQ_EXEC_CPG_PROCJOIN
@ MESSAGE_REQ_EXEC_CPG_PROCJOIN
Definition exec/cpg.c:80
MESSAGE_REQ_EXEC_CPG_DOWNLIST
@ MESSAGE_REQ_EXEC_CPG_DOWNLIST
Definition exec/cpg.c:85
cpd_state
cpd_state
Definition exec/cpg.c:131
CPD_STATE_JOIN_STARTED
@ CPD_STATE_JOIN_STARTED
Definition exec/cpg.c:134
CPD_STATE_LEAVE_STARTED
@ CPD_STATE_LEAVE_STARTED
Definition exec/cpg.c:133
CPD_STATE_UNJOINED
@ CPD_STATE_UNJOINED
Definition exec/cpg.c:132
CPD_STATE_JOIN_COMPLETED
@ CPD_STATE_JOIN_COMPLETED
Definition exec/cpg.c:135
cpg_get_service_engine_ver0
struct corosync_service_engine * cpg_get_service_engine_ver0(void)
Definition exec/cpg.c:454
cpg_sync_state
cpg_sync_state
Definition exec/cpg.c:138
CPGSYNC_DOWNLIST
@ CPGSYNC_DOWNLIST
Definition exec/cpg.c:139
CPGSYNC_JOINLIST
@ CPGSYNC_JOINLIST
Definition exec/cpg.c:140
CPG_MAX_NAME_LENGTH
#define CPG_MAX_NAME_LENGTH
Definition cpg.h:117
CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF
#define CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF
Definition cpg.h:194
CPG_MEMBERS_MAX
#define CPG_MEMBERS_MAX
Definition cpg.h:126
CPG_ITERATION_ONE_GROUP
@ CPG_ITERATION_ONE_GROUP
Definition cpg.h:97
CPG_ITERATION_NAME_ONLY
@ CPG_ITERATION_NAME_ONLY
Definition cpg.h:96
CPG_REASON_UNDEFINED
@ CPG_REASON_UNDEFINED
Definition cpg.h:84
DECLARE_HDB_DATABASE
#define DECLARE_HDB_DATABASE
Definition hdb.h:98
hdb_handle_t
qb_handle_t hdb_handle_t
Definition hdb.h:52
ipc_cpg.h
LIBCPG_PARTIAL_FIRST
@ LIBCPG_PARTIAL_FIRST
Definition ipc_cpg.h:104
MESSAGE_RES_CPG_ITERATIONFINALIZE
@ MESSAGE_RES_CPG_ITERATIONFINALIZE
Definition ipc_cpg.h:79
MESSAGE_RES_CPG_MEMBERSHIP
@ MESSAGE_RES_CPG_MEMBERSHIP
Definition ipc_cpg.h:71
MESSAGE_RES_CPG_LEAVE
@ MESSAGE_RES_CPG_LEAVE
Definition ipc_cpg.h:69
MESSAGE_RES_CPG_MCAST
@ MESSAGE_RES_CPG_MCAST
Definition ipc_cpg.h:70
MESSAGE_RES_CPG_PARTIAL_DELIVER_CALLBACK
@ MESSAGE_RES_CPG_PARTIAL_DELIVER_CALLBACK
Definition ipc_cpg.h:85
MESSAGE_RES_CPG_FINALIZE
@ MESSAGE_RES_CPG_FINALIZE
Definition ipc_cpg.h:80
MESSAGE_RES_CPG_DELIVER_CALLBACK
@ MESSAGE_RES_CPG_DELIVER_CALLBACK
Definition ipc_cpg.h:73
MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK
@ MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK
Definition ipc_cpg.h:81
MESSAGE_RES_CPG_ITERATIONNEXT
@ MESSAGE_RES_CPG_ITERATIONNEXT
Definition ipc_cpg.h:78
MESSAGE_RES_CPG_CONFCHG_CALLBACK
@ MESSAGE_RES_CPG_CONFCHG_CALLBACK
Definition ipc_cpg.h:72
MESSAGE_RES_CPG_PARTIAL_SEND
@ MESSAGE_RES_CPG_PARTIAL_SEND
Definition ipc_cpg.h:86
MESSAGE_RES_CPG_ITERATIONINITIALIZE
@ MESSAGE_RES_CPG_ITERATIONINITIALIZE
Definition ipc_cpg.h:77
MESSAGE_RES_CPG_LOCAL_GET
@ MESSAGE_RES_CPG_LOCAL_GET
Definition ipc_cpg.h:75
MESSAGE_RES_CPG_JOIN
@ MESSAGE_RES_CPG_JOIN
Definition ipc_cpg.h:68
CONFCHG_CPG_REASON_LEAVE
@ CONFCHG_CPG_REASON_LEAVE
Definition ipc_cpg.h:94
CONFCHG_CPG_REASON_PROCDOWN
@ CONFCHG_CPG_REASON_PROCDOWN
Definition ipc_cpg.h:97
CONFCHG_CPG_REASON_NODEUP
@ CONFCHG_CPG_REASON_NODEUP
Definition ipc_cpg.h:96
CONFCHG_CPG_REASON_JOIN
@ CONFCHG_CPG_REASON_JOIN
Definition ipc_cpg.h:93
CONFCHG_CPG_REASON_NODEDOWN
@ CONFCHG_CPG_REASON_NODEDOWN
Definition ipc_cpg.h:95
logsys.h
LOGSYS_LEVEL_ERROR
#define LOGSYS_LEVEL_ERROR
Definition logsys.h:72
log_printf
#define log_printf(level, format, args...)
Definition logsys.h:332
LOGSYS_DECLARE_SUBSYS
#define LOGSYS_DECLARE_SUBSYS(subsys)
The LOGSYS_DECLARE_SUBSYS macro.
Definition logsys.h:306
LOGSYS_LEVEL_WARNING
#define LOGSYS_LEVEL_WARNING
Definition logsys.h:73
LOGSYS_LEVEL_DEBUG
#define LOGSYS_LEVEL_DEBUG
Definition logsys.h:76
LOGSYS_LEVEL_TRACE
#define LOGSYS_LEVEL_TRACE
Definition logsys.h:77
mar_uint32_t
uint32_t mar_uint32_t
Definition mar_gen.h:53
mar_uint8_t
uint8_t mar_uint8_t
Definition mar_gen.h:51
service.h
coroipcs_zc_header
coroipcs_zc_header struct
Definition ipc_cpg.h:498
coroipcs_zc_header::server_address
uint64_t server_address
Definition ipc_cpg.h:500
corosync_api_v1
The corosync_api_v1 struct.
Definition coroapi.h:225
corosync_exec_handler
The corosync_exec_handler struct.
Definition coroapi.h:475
corosync_lib_handler
The corosync_lib_handler struct.
Definition coroapi.h:467
corosync_service_engine
The corosync_service_engine struct.
Definition coroapi.h:490
cpg_iteration_instance
Definition exec/cpg.c:159
cpg_iteration_instance::handle
hdb_handle_t handle
Definition exec/cpg.c:160
cpg_iteration_instance::list
struct qb_list_head list
Definition exec/cpg.c:161
cpg_iteration_instance::items_list_head
struct qb_list_head items_list_head
Definition exec/cpg.c:162
cpg_iteration_instance::current_pointer
struct qb_list_head * current_pointer
Definition exec/cpg.c:163
cpg_name
The cpg_name struct.
Definition cpg.h:121
cpg_pd
Definition exec/cpg.c:145
cpg_pd::cpd_state
enum cpd_state cpd_state
Definition exec/cpg.c:149
cpg_pd::pid
uint32_t pid
Definition exec/cpg.c:148
cpg_pd::transition_counter
uint64_t transition_counter
Definition exec/cpg.c:152
cpg_pd::initial_totem_conf_sent
int initial_totem_conf_sent
Definition exec/cpg.c:151
cpg_pd::group_name
mar_cpg_name_t group_name
Definition exec/cpg.c:147
cpg_pd::iteration_instance_list_head
struct qb_list_head iteration_instance_list_head
Definition exec/cpg.c:155
cpg_pd::initial_transition_counter
uint64_t initial_transition_counter
Definition exec/cpg.c:153
cpg_pd::list
struct qb_list_head list
Definition exec/cpg.c:154
cpg_pd::zcb_mapped_list_head
struct qb_list_head zcb_mapped_list_head
Definition exec/cpg.c:156
cpg_pd::conn
void * conn
Definition exec/cpg.c:146
cpg_pd::flags
unsigned int flags
Definition exec/cpg.c:150
join_list_confchg_data
Definition exec/cpg.c:197
join_list_confchg_data::join_list_entries
int join_list_entries
Definition exec/cpg.c:200
join_list_confchg_data::join_list
mar_cpg_address_t join_list[CPG_MEMBERS_MAX]
Definition exec/cpg.c:199
join_list_confchg_data::cpg_group
mar_cpg_name_t cpg_group
Definition exec/cpg.c:198
join_list_entry
Definition exec/cpg.c:192
join_list_entry::pid
uint32_t pid
Definition exec/cpg.c:193
join_list_entry::group_name
mar_cpg_name_t group_name
Definition exec/cpg.c:194
joinlist_msg
Definition exec/cpg.c:501
joinlist_msg::sender_nodeid
mar_uint32_t sender_nodeid
Definition exec/cpg.c:502
joinlist_msg::pid
uint32_t pid
Definition exec/cpg.c:503
joinlist_msg::list
struct qb_list_head list
Definition exec/cpg.c:505
joinlist_msg::group_name
mar_cpg_name_t group_name
Definition exec/cpg.c:504
mar_cpg_address_t
mar_cpg_address_t struct
Definition ipc_cpg.h:155
mar_cpg_name_t
mar_cpg_name_t struct
Definition ipc_cpg.h:112
mar_cpg_ring_id_t
mar_cpg_ring_id_t struct
Definition ipc_cpg.h:230
mar_message_source_t
The mar_message_source_t struct.
Definition coroapi.h:50
mar_req_coroipcc_zc_alloc_t
mar_req_coroipcc_zc_alloc_t struct
Definition ipc_cpg.h:472
mar_req_coroipcc_zc_execute_t
mar_req_coroipcc_zc_execute_t struct
Definition ipc_cpg.h:490
mar_req_coroipcc_zc_free_t
mar_req_coroipcc_zc_free_t struct
Definition ipc_cpg.h:481
memb_ring_id
The memb_ring_id struct.
Definition coroapi.h:122
process_info
Definition exec/cpg.c:184
process_info::list
struct qb_list_head list
Definition exec/cpg.c:188
process_info::pid
uint32_t pid
Definition exec/cpg.c:186
process_info::group
mar_cpg_name_t group
Definition exec/cpg.c:187
process_info::nodeid
unsigned int nodeid
Definition exec/cpg.c:185
req_exec_cpg_downlist_old
Definition exec/cpg.c:486
req_exec_cpg_downlist_old::__attribute__
struct qb_ipc_request_header header __attribute__((aligned(8)))
req_exec_cpg_downlist_old::__attribute__
mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)))
req_exec_cpg_downlist_old::__attribute__
mar_uint32_t left_nodes __attribute__((aligned(8)))
req_exec_cpg_downlist
Definition exec/cpg.c:492
req_exec_cpg_downlist::__attribute__
mar_uint32_t old_members __attribute__((aligned(8)))
req_exec_cpg_downlist::__attribute__
mar_uint32_t left_nodes __attribute__((aligned(8)))
req_exec_cpg_downlist::__attribute__
struct qb_ipc_request_header header __attribute__((aligned(8)))
req_exec_cpg_downlist::__attribute__
mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)))
req_exec_cpg_mcast
Definition exec/cpg.c:466
req_exec_cpg_mcast::__attribute__
struct qb_ipc_request_header header __attribute__((aligned(8)))
req_exec_cpg_mcast::__attribute__
mar_cpg_name_t group_name __attribute__((aligned(8)))
req_exec_cpg_mcast::__attribute__
mar_uint32_t pid __attribute__((aligned(8)))
req_exec_cpg_mcast::__attribute__
mar_uint32_t msglen __attribute__((aligned(8)))
req_exec_cpg_mcast::__attribute__
mar_message_source_t source __attribute__((aligned(8)))
req_exec_cpg_mcast::__attribute__
mar_uint8_t message[] __attribute__((aligned(8)))
req_exec_cpg_partial_mcast
Definition exec/cpg.c:475
req_exec_cpg_partial_mcast::__attribute__
mar_uint32_t type __attribute__((aligned(8)))
req_exec_cpg_partial_mcast::__attribute__
mar_message_source_t source __attribute__((aligned(8)))
req_exec_cpg_partial_mcast::__attribute__
mar_cpg_name_t group_name __attribute__((aligned(8)))
req_exec_cpg_partial_mcast::__attribute__
mar_uint32_t pid __attribute__((aligned(8)))
req_exec_cpg_partial_mcast::__attribute__
mar_uint8_t message[] __attribute__((aligned(8)))
req_exec_cpg_partial_mcast::__attribute__
struct qb_ipc_request_header header __attribute__((aligned(8)))
req_exec_cpg_partial_mcast::__attribute__
mar_uint32_t msglen __attribute__((aligned(8)))
req_exec_cpg_partial_mcast::__attribute__
mar_uint32_t fraglen __attribute__((aligned(8)))
req_exec_cpg_procjoin
Definition exec/cpg.c:459
req_exec_cpg_procjoin::__attribute__
mar_uint32_t pid __attribute__((aligned(8)))
req_exec_cpg_procjoin::__attribute__
mar_uint32_t reason __attribute__((aligned(8)))
req_exec_cpg_procjoin::__attribute__
struct qb_ipc_request_header header __attribute__((aligned(8)))
req_exec_cpg_procjoin::__attribute__
mar_cpg_name_t group_name __attribute__((aligned(8)))
req_lib_cpg_iterationfinalize
The req_lib_cpg_iterationfinalize struct.
Definition ipc_cpg.h:457
req_lib_cpg_iterationinitialize
The req_lib_cpg_iterationinitialize struct.
Definition ipc_cpg.h:424
req_lib_cpg_iterationnext
The req_lib_cpg_iterationnext struct.
Definition ipc_cpg.h:441
req_lib_cpg_join
The req_lib_cpg_join struct.
Definition ipc_cpg.h:251
req_lib_cpg_leave
The req_lib_cpg_leave struct.
Definition ipc_cpg.h:408
req_lib_cpg_mcast
The req_lib_cpg_mcast struct.
Definition ipc_cpg.h:304
req_lib_cpg_membership_get
The req_lib_cpg_membership_get struct.
Definition ipc_cpg.h:367
req_lib_cpg_partial_mcast
The req_lib_cpg_partial_mcast struct.
Definition ipc_cpg.h:314
res_lib_cpg_confchg_callback
The res_lib_cpg_confchg_callback struct.
Definition ipc_cpg.h:384
res_lib_cpg_confchg_callback::member_list
mar_cpg_address_t member_list[]
Definition ipc_cpg.h:390
res_lib_cpg_deliver_callback
Message from another node.
Definition ipc_cpg.h:333
res_lib_cpg_finalize
The res_lib_cpg_finalize struct.
Definition ipc_cpg.h:275
res_lib_cpg_iterationfinalize
The res_lib_cpg_iterationfinalize struct.
Definition ipc_cpg.h:465
res_lib_cpg_iterationinitialize
The res_lib_cpg_iterationinitialize struct.
Definition ipc_cpg.h:433
res_lib_cpg_iterationnext
The res_lib_cpg_iterationnext struct.
Definition ipc_cpg.h:449
res_lib_cpg_join
The res_lib_cpg_join struct.
Definition ipc_cpg.h:261
res_lib_cpg_leave
The res_lib_cpg_leave struct.
Definition ipc_cpg.h:417
res_lib_cpg_local_get
The res_lib_cpg_local_get struct.
Definition ipc_cpg.h:289
res_lib_cpg_mcast
The res_lib_cpg_mcast struct.
Definition ipc_cpg.h:326
res_lib_cpg_membership_get
The res_lib_cpg_membership_get struct.
Definition ipc_cpg.h:375
res_lib_cpg_membership_get::member_list
mar_cpg_address_t member_list[PROCESSOR_COUNT_MAX]
Definition ipc_cpg.h:378
res_lib_cpg_partial_deliver_callback
The res_lib_cpg_partial_deliver_callback struct.
Definition ipc_cpg.h:345
res_lib_cpg_partial_send
The res_lib_cpg_partial_send struct.
Definition ipc_cpg.h:297
res_lib_cpg_totem_confchg_callback
The res_lib_cpg_totem_confchg_callback struct.
Definition ipc_cpg.h:398
res_lib_cpg_totem_confchg_callback::member_list
mar_uint32_t member_list[]
Definition ipc_cpg.h:402
zcb_mapped
Definition exec/cpg.c:89
zcb_mapped::size
size_t size
Definition exec/cpg.c:92
zcb_mapped::addr
void * addr
Definition exec/cpg.c:91
zcb_mapped::list
struct qb_list_head list
Definition exec/cpg.c:90
swab32
#define swab32(x)
The swab32 macro.
Definition swab.h:51
sync_init
int sync_init(int(*sync_callbacks_retrieve)(int service_id, struct sync_callbacks *callbacks), void(*synchronization_completed)(void))
Definition sync.c:158
type
char type
Definition totem.h:2
ring_id
struct memb_ring_id ring_id
Definition totemsrp.c:4
header
struct totem_message_header header
Definition totemsrp.c:0
u
Definition exec/cpg.c:1813
u::server_addr
uint64_t server_addr
Definition exec/cpg.c:1814
u::server_ptr
void * server_ptr
Definition exec/cpg.c:1815