ipc_glue.c
Go to the documentation of this file.
/*
* Copyright (c) 2010-2017 Red Hat, Inc.
*
* All rights reserved.
*
* Author: Angus Salkeld <asalkeld@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 Red Hat, 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 CONTRIBUTORS "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>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <assert.h>
#include <sys/uio.h>
#include <string.h>
#include <qb/qbdefs.h>
#include <qb/qblist.h>
#include <qb/qbutil.h>
#include <qb/qbloop.h>
#include <qb/qbipcs.h>
#include <corosync/swab.h>
#include <corosync/corotypes.h>
#include <corosync/corodefs.h>
#include <corosync/totem/totempg.h>
#include <corosync/logsys.h>
#include <corosync/icmap.h>
#include "sync.h"
#include "timer.h"
#include "main.h"
#include "util.h"
#include "apidef.h"
#include "service.h"
#include "ipcs_stats.h"
#include "stats.h"
LOGSYS_DECLARE_SUBSYS ("MAIN");
static struct corosync_api_v1 *api = NULL;
static int32_t ipc_not_enough_fds_left = 0;
static int32_t ipc_fc_is_quorate; /* boolean */
static int32_t ipc_fc_totem_queue_level; /* percentage used */
static int32_t ipc_fc_sync_in_process; /* boolean */
static int32_t ipc_allow_connections = 0; /* boolean */
#define CS_IPCS_MAPPER_SERV_NAME 256
struct cs_ipcs_mapper {
int32_t id;
qb_ipcs_service_t *inst;
char name[CS_IPCS_MAPPER_SERV_NAME];
};
struct outq_item {
void *msg;
size_t mlen;
struct qb_list_head list;
};
static struct cs_ipcs_mapper ipcs_mapper[SERVICES_COUNT_MAX];
static int32_t cs_ipcs_job_add(enum qb_loop_priority p, void *data, qb_loop_job_dispatch_fn fn);
static int32_t cs_ipcs_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn);
static int32_t cs_ipcs_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn);
static int32_t cs_ipcs_dispatch_del(int32_t fd);
static void outq_flush (void *data);
static struct qb_ipcs_poll_handlers corosync_poll_funcs = {
.job_add = cs_ipcs_job_add,
.dispatch_add = cs_ipcs_dispatch_add,
.dispatch_mod = cs_ipcs_dispatch_mod,
.dispatch_del = cs_ipcs_dispatch_del,
};
static int32_t cs_ipcs_connection_accept (qb_ipcs_connection_t *c, uid_t euid, gid_t egid);
static void cs_ipcs_connection_created(qb_ipcs_connection_t *c);
static int32_t cs_ipcs_msg_process(qb_ipcs_connection_t *c,
void *data, size_t size);
static int32_t cs_ipcs_connection_closed (qb_ipcs_connection_t *c);
static void cs_ipcs_connection_destroyed (qb_ipcs_connection_t *c);
static struct qb_ipcs_service_handlers corosync_service_funcs = {
.connection_accept = cs_ipcs_connection_accept,
.connection_created = cs_ipcs_connection_created,
.msg_process = cs_ipcs_msg_process,
.connection_closed = cs_ipcs_connection_closed,
.connection_destroyed = cs_ipcs_connection_destroyed,
};
static struct ipcs_global_stats global_stats;
static const char* cs_ipcs_serv_short_name(int32_t service_id)
{
const char *name;
switch (service_id) {
case CFG_SERVICE:
name = "cfg";
break;
case CPG_SERVICE:
name = "cpg";
break;
case QUORUM_SERVICE:
name = "quorum";
break;
case PLOAD_SERVICE:
name = "pload";
break;
case VOTEQUORUM_SERVICE:
name = "votequorum";
break;
case MON_SERVICE:
name = "mon";
break;
case WD_SERVICE:
name = "wd";
break;
case CMAP_SERVICE:
name = "cmap";
break;
default:
name = NULL;
break;
}
return name;
}
void cs_ipc_allow_connections(int32_t allow)
{
ipc_allow_connections = allow;
}
int32_t cs_ipcs_service_destroy(int32_t service_id)
{
if (ipcs_mapper[service_id].inst) {
qb_ipcs_destroy(ipcs_mapper[service_id].inst);
ipcs_mapper[service_id].inst = NULL;
}
return 0;
}
static int32_t cs_ipcs_connection_accept (qb_ipcs_connection_t *c, uid_t euid, gid_t egid)
{
int32_t service = qb_ipcs_service_id_get(c);
uint8_t u8;
char key_name[ICMAP_KEYNAME_MAXLEN];
if (!ipc_allow_connections) {
log_printf(LOGSYS_LEVEL_DEBUG, "Denied connection, corosync is not ready");
return -EAGAIN;
}
if (corosync_service[service] == NULL ||
ipcs_mapper[service].inst == NULL) {
return -ENOSYS;
}
if (ipc_not_enough_fds_left) {
return -EMFILE;
}
if (euid == 0 || egid == 0) {
return 0;
}
snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.uid.%u", euid);
if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
return 0;
snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.config.uid.%u", euid);
if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
return 0;
snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.gid.%u", egid);
if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
return 0;
snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.config.gid.%u", egid);
if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
return 0;
log_printf(LOGSYS_LEVEL_ERROR, "Denied connection attempt from %d:%d", euid, egid);
return -EACCES;
}
static char * pid_to_name (pid_t pid, char *out_name, size_t name_len)
{
char *name;
char *rest;
FILE *fp;
char fname[32];
char buf[256];
snprintf (fname, 32, "/proc/%d/stat", pid);
fp = fopen (fname, "r");
if (!fp) {
return NULL;
}
if (fgets (buf, sizeof (buf), fp) == NULL) {
fclose (fp);
return NULL;
}
fclose (fp);
name = strrchr (buf, '(');
if (!name) {
return NULL;
}
/* move past the bracket */
name++;
rest = strrchr (buf, ')');
if (rest == NULL || rest[1] != ' ') {
return NULL;
}
*rest = '\0';
/* move past the NULL and space */
rest += 2;
/* copy the name */
strncpy (out_name, name, name_len - 1);
out_name[name_len - 1] = '\0';
return out_name;
}
static void cs_ipcs_connection_created(qb_ipcs_connection_t *c)
{
int32_t service = 0;
struct cs_ipcs_conn_context *context;
struct qb_ipcs_connection_stats stats;
size_t size = sizeof(struct cs_ipcs_conn_context);
log_printf(LOG_DEBUG, "connection created");
service = qb_ipcs_service_id_get(c);
size += corosync_service[service]->private_data_size;
context = calloc(1, size);
if (context == NULL) {
qb_ipcs_disconnect(c);
return;
}
qb_list_init(&context->outq_head);
context->queuing = QB_FALSE;
context->queued = 0;
context->sent = 0;
qb_ipcs_context_set(c, context);
if (corosync_service[service]->lib_init_fn(c) != 0) {
log_printf(LOG_ERR, "lib_init_fn failed, disconnecting");
qb_ipcs_disconnect(c);
return;
}
qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
if (!pid_to_name (stats.client_pid, context->proc_name, sizeof(context->proc_name))) {
context->proc_name[0] = '\0';
}
stats_ipcs_add_connection(service, stats.client_pid, c);
global_stats.active++;
}
void cs_ipc_refcnt_inc(void *conn)
{
qb_ipcs_connection_ref(conn);
}
void cs_ipc_refcnt_dec(void *conn)
{
qb_ipcs_connection_unref(conn);
}
void *cs_ipcs_private_data_get(void *conn)
{
struct cs_ipcs_conn_context *cnx;
cnx = qb_ipcs_context_get(conn);
return &cnx->data[0];
}
static void cs_ipcs_connection_destroyed (qb_ipcs_connection_t *c)
{
struct cs_ipcs_conn_context *context;
struct qb_list_head *list, *tmp_iter;
struct outq_item *outq_item;
log_printf(LOG_DEBUG, "%s() ", __func__);
context = qb_ipcs_context_get(c);
if (context) {
qb_list_for_each_safe(list, tmp_iter, &(context->outq_head)) {
outq_item = qb_list_entry (list, struct outq_item, list);
qb_list_del (list);
free (outq_item->msg);
free (outq_item);
}
free(context);
}
}
static int32_t cs_ipcs_connection_closed (qb_ipcs_connection_t *c)
{
int32_t res = 0;
int32_t service = qb_ipcs_service_id_get(c);
struct qb_ipcs_connection_stats stats;
log_printf(LOG_DEBUG, "%s() ", __func__);
res = corosync_service[service]->lib_exit_fn(c);
if (res != 0) {
return res;
}
qb_loop_job_del(cs_poll_handle_get(), QB_LOOP_HIGH, c, outq_flush);
qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
stats_ipcs_del_connection(service, stats.client_pid, c);
global_stats.active--;
global_stats.closed++;
return 0;
}
int cs_ipcs_response_iov_send (void *conn,
const struct iovec *iov,
unsigned int iov_len)
{
int32_t rc = qb_ipcs_response_sendv(conn, iov, iov_len);
if (rc >= 0) {
return 0;
}
return rc;
}
int cs_ipcs_response_send(void *conn, const void *msg, size_t mlen)
{
int32_t rc = qb_ipcs_response_send(conn, msg, mlen);
if (rc >= 0) {
return 0;
}
return rc;
}
static void outq_flush (void *data)
{
qb_ipcs_connection_t *conn = data;
struct qb_list_head *list, *tmp_iter;
struct outq_item *outq_item;
int32_t rc;
struct cs_ipcs_conn_context *context = qb_ipcs_context_get(conn);
qb_list_for_each_safe(list, tmp_iter, &(context->outq_head)) {
outq_item = qb_list_entry (list, struct outq_item, list);
rc = qb_ipcs_event_send(conn, outq_item->msg, outq_item->mlen);
if (rc < 0 && rc != -EAGAIN) {
errno = -rc;
qb_perror(LOG_ERR, "qb_ipcs_event_send");
return;
} else if (rc == -EAGAIN) {
break;
}
assert(rc == outq_item->mlen);
context->sent++;
context->queued--;
qb_list_del (list);
free (outq_item->msg);
free (outq_item);
}
if (qb_list_empty (&context->outq_head)) {
context->queuing = QB_FALSE;
log_printf(LOGSYS_LEVEL_INFO, "Q empty, queued:%d sent:%d.",
context->queued, context->sent);
context->queued = 0;
context->sent = 0;
} else {
qb_loop_job_add(cs_poll_handle_get(), QB_LOOP_HIGH, conn, outq_flush);
}
}
static void msg_send_or_queue(qb_ipcs_connection_t *conn, const struct iovec *iov, uint32_t iov_len)
{
int32_t rc = 0;
int32_t i;
int32_t bytes_msg = 0;
struct outq_item *outq_item;
char *write_buf = 0;
struct cs_ipcs_conn_context *context = qb_ipcs_context_get(conn);
for (i = 0; i < iov_len; i++) {
bytes_msg += iov[i].iov_len;
}
if (!context->queuing) {
assert(qb_list_empty (&context->outq_head));
rc = qb_ipcs_event_sendv(conn, iov, iov_len);
if (rc == bytes_msg) {
context->sent++;
return;
}
if (rc == -EAGAIN) {
context->queued = 0;
context->sent = 0;
context->queuing = QB_TRUE;
qb_loop_job_add(cs_poll_handle_get(), QB_LOOP_HIGH, conn, outq_flush);
} else {
log_printf(LOGSYS_LEVEL_ERROR, "event_send retuned %d, expected %d!", rc, bytes_msg);
return;
}
}
outq_item = malloc (sizeof (struct outq_item));
if (outq_item == NULL) {
qb_ipcs_disconnect(conn);
return;
}
outq_item->msg = malloc (bytes_msg);
if (outq_item->msg == NULL) {
free (outq_item);
qb_ipcs_disconnect(conn);
return;
}
write_buf = outq_item->msg;
for (i = 0; i < iov_len; i++) {
memcpy (write_buf, iov[i].iov_base, iov[i].iov_len);
write_buf += iov[i].iov_len;
}
outq_item->mlen = bytes_msg;
qb_list_init (&outq_item->list);
qb_list_add_tail (&outq_item->list, &context->outq_head);
context->queued++;
}
int cs_ipcs_dispatch_send(void *conn, const void *msg, size_t mlen)
{
struct iovec iov;
iov.iov_base = (void *)msg;
iov.iov_len = mlen;
msg_send_or_queue (conn, &iov, 1);
return 0;
}
int cs_ipcs_dispatch_iov_send (void *conn,
const struct iovec *iov,
unsigned int iov_len)
{
msg_send_or_queue(conn, iov, iov_len);
return 0;
}
static int32_t cs_ipcs_msg_process(qb_ipcs_connection_t *c,
void *data, size_t size)
{
struct qb_ipc_response_header response;
struct qb_ipc_request_header *request_pt = (struct qb_ipc_request_header *)data;
int32_t service = qb_ipcs_service_id_get(c);
int32_t send_ok = 0;
int32_t is_async_call = QB_FALSE;
ssize_t res = -1;
int sending_allowed_private_data;
struct cs_ipcs_conn_context *cnx;
send_ok = corosync_sending_allowed (service,
request_pt->id,
request_pt,
&sending_allowed_private_data);
is_async_call = (service == CPG_SERVICE && request_pt->id == 2);
/*
* This happens when the message contains some kind of invalid
* parameter, such as an invalid size
*/
if (send_ok == -EINVAL) {
response.size = sizeof (response);
response.id = 0;
response.error = CS_ERR_INVALID_PARAM;
cnx = qb_ipcs_context_get(c);
if (cnx) {
cnx->invalid_request++;
}
if (is_async_call) {
log_printf(LOGSYS_LEVEL_INFO, "*** %s() invalid message! size:%d error:%d",
__func__, response.size, response.error);
} else {
qb_ipcs_response_send (c,
&response,
sizeof (response));
}
res = -EINVAL;
} else if (send_ok < 0) {
cnx = qb_ipcs_context_get(c);
if (cnx) {
cnx->overload++;
}
if (!is_async_call) {
/*
* Overload, tell library to retry
*/
response.size = sizeof (response);
response.id = 0;
response.error = CS_ERR_TRY_AGAIN;
qb_ipcs_response_send (c,
&response,
sizeof (response));
} else {
log_printf(LOGSYS_LEVEL_WARNING,
"*** %s() (%d:%d - %d) %s!",
__func__, service, request_pt->id,
is_async_call, strerror(-send_ok));
}
res = -ENOBUFS;
}
if (send_ok >= 0) {
corosync_service[service]->lib_engine[request_pt->id].lib_handler_fn(c, request_pt);
res = 0;
}
corosync_sending_allowed_release (&sending_allowed_private_data);
return res;
}
static int32_t cs_ipcs_job_add(enum qb_loop_priority p, void *data, qb_loop_job_dispatch_fn fn)
{
return qb_loop_job_add(cs_poll_handle_get(), p, data, fn);
}
static int32_t cs_ipcs_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn)
{
return qb_loop_poll_add(cs_poll_handle_get(), p, fd, events, data, fn);
}
static int32_t cs_ipcs_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn)
{
return qb_loop_poll_mod(cs_poll_handle_get(), p, fd, events, data, fn);
}
static int32_t cs_ipcs_dispatch_del(int32_t fd)
{
return qb_loop_poll_del(cs_poll_handle_get(), fd);
}
static void cs_ipcs_low_fds_event(int32_t not_enough, int32_t fds_available)
{
ipc_not_enough_fds_left = not_enough;
if (not_enough) {
log_printf(LOGSYS_LEVEL_WARNING, "refusing new connections (fds_available:%d)",
fds_available);
} else {
log_printf(LOGSYS_LEVEL_NOTICE, "allowing new connections (fds_available:%d)",
fds_available);
}
}
int32_t cs_ipcs_q_level_get(void)
{
return ipc_fc_totem_queue_level;
}
static qb_loop_timer_handle ipcs_check_for_flow_control_timer;
static void cs_ipcs_check_for_flow_control(void)
{
int32_t i;
int32_t fc_enabled;
for (i = 0; i < SERVICES_COUNT_MAX; i++) {
if (corosync_service[i] == NULL || ipcs_mapper[i].inst == NULL) {
continue;
}
fc_enabled = QB_IPCS_RATE_OFF;
if (ipc_fc_is_quorate == 1 ||
corosync_service[i]->allow_inquorate == CS_LIB_ALLOW_INQUORATE) {
/*
* we are quorate
* now check flow control
*/
if (ipc_fc_totem_queue_level != TOTEM_Q_LEVEL_CRITICAL &&
ipc_fc_sync_in_process == 0) {
fc_enabled = QB_FALSE;
} else if (ipc_fc_totem_queue_level != TOTEM_Q_LEVEL_CRITICAL &&
i == VOTEQUORUM_SERVICE) {
/*
* Allow message processing for votequorum service even
* in sync phase
*/
fc_enabled = QB_FALSE;
} else {
fc_enabled = QB_IPCS_RATE_OFF_2;
}
}
if (fc_enabled) {
qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, fc_enabled);
qb_loop_timer_add(cs_poll_handle_get(), QB_LOOP_MED, 1*QB_TIME_NS_IN_MSEC,
NULL, corosync_recheck_the_q_level, &ipcs_check_for_flow_control_timer);
} else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_LOW) {
qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_FAST);
} else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_GOOD) {
qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_NORMAL);
} else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_HIGH) {
qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_SLOW);
}
}
}
static void cs_ipcs_fc_quorum_changed(int quorate, void *context)
{
ipc_fc_is_quorate = quorate;
cs_ipcs_check_for_flow_control();
}
static void cs_ipcs_totem_queue_level_changed(enum totem_q_level level)
{
ipc_fc_totem_queue_level = level;
cs_ipcs_check_for_flow_control();
}
void cs_ipcs_sync_state_changed(int32_t sync_in_process)
{
ipc_fc_sync_in_process = sync_in_process;
cs_ipcs_check_for_flow_control();
}
void cs_ipcs_get_global_stats(struct ipcs_global_stats *ipcs_stats)
{
memcpy(ipcs_stats, &global_stats, sizeof(global_stats));
}
cs_error_t cs_ipcs_get_conn_stats(int service_id, uint32_t pid, void *conn_ptr, struct ipcs_conn_stats *ipcs_stats)
{
struct cs_ipcs_conn_context *cnx;
qb_ipcs_connection_t *c, *prev;
int found = 0;
if (corosync_service[service_id] == NULL || ipcs_mapper[service_id].inst == NULL) {
return CS_ERR_NOT_EXIST;
}
qb_ipcs_stats_get(ipcs_mapper[service_id].inst, &ipcs_stats->srv, QB_FALSE);
for (c = qb_ipcs_connection_first_get(ipcs_mapper[service_id].inst);
c;
prev = c, c = qb_ipcs_connection_next_get(ipcs_mapper[service_id].inst, prev), qb_ipcs_connection_unref(prev)) {
cnx = qb_ipcs_context_get(c);
if (cnx == NULL) continue;
if (c != conn_ptr) continue;
qb_ipcs_connection_stats_get(c, &ipcs_stats->conn, QB_FALSE);
if (ipcs_stats->conn.client_pid != pid) {
continue;
}
found = 1;
memcpy(&ipcs_stats->cnx, cnx, sizeof(struct cs_ipcs_conn_context));
}
if (!found) {
return CS_ERR_NOT_EXIST;
}
return CS_OK;
}
void cs_ipcs_clear_stats()
{
struct cs_ipcs_conn_context *cnx;
struct ipcs_conn_stats ipcs_stats;
qb_ipcs_connection_t *c, *prev;
int service_id;
/* Global stats are easy */
memset(&global_stats, 0, sizeof(global_stats));
for (service_id = 0; service_id < SERVICES_COUNT_MAX; service_id++) {
if (!ipcs_mapper[service_id].inst) {
continue;
}
for (c = qb_ipcs_connection_first_get(ipcs_mapper[service_id].inst);
c;
prev = c, c = qb_ipcs_connection_next_get(ipcs_mapper[service_id].inst, prev), qb_ipcs_connection_unref(prev)) {
/* Get stats with 'clear_after_read' set */
qb_ipcs_connection_stats_get(c, &ipcs_stats.conn, QB_TRUE);
/* Our own stats */
cnx = qb_ipcs_context_get(c);
if (cnx == NULL) continue;
cnx->invalid_request = 0;
cnx->overload = 0;
cnx->sent = 0;
}
}
}
static enum qb_ipc_type cs_get_ipc_type (void)
{
char *str;
int found = 0;
enum qb_ipc_type ret = QB_IPC_NATIVE;
if (icmap_get_string("system.qb_ipc_type", &str) != CS_OK) {
log_printf(LOGSYS_LEVEL_DEBUG, "No configured system.qb_ipc_type. Using native ipc");
return QB_IPC_NATIVE;
}
if (strcmp(str, "native") == 0) {
ret = QB_IPC_NATIVE;
found = 1;
}
if (strcmp(str, "shm") == 0) {
ret = QB_IPC_SHM;
found = 1;
}
if (strcmp(str, "socket") == 0) {
ret = QB_IPC_SOCKET;
found = 1;
}
if (found) {
log_printf(LOGSYS_LEVEL_DEBUG, "Using %s ipc", str);
} else {
log_printf(LOGSYS_LEVEL_DEBUG, "Unknown ipc type %s", str);
}
free(str);
return ret;
}
const char *cs_ipcs_service_init(struct corosync_service_engine *service)
{
const char *serv_short_name;
serv_short_name = cs_ipcs_serv_short_name(service->id);
if (service->lib_engine_count == 0) {
log_printf (LOGSYS_LEVEL_DEBUG,
"NOT Initializing IPC on %s [%d]",
serv_short_name,
service->id);
return NULL;
}
if (strlen(serv_short_name) >= CS_IPCS_MAPPER_SERV_NAME) {
log_printf (LOGSYS_LEVEL_ERROR, "service name %s is too long", serv_short_name);
return "qb_ipcs_run error";
}
ipcs_mapper[service->id].id = service->id;
strcpy(ipcs_mapper[service->id].name, serv_short_name);
log_printf (LOGSYS_LEVEL_DEBUG,
"Initializing IPC on %s [%d]",
ipcs_mapper[service->id].name,
ipcs_mapper[service->id].id);
ipcs_mapper[service->id].inst = qb_ipcs_create(ipcs_mapper[service->id].name,
ipcs_mapper[service->id].id,
cs_get_ipc_type(),
&corosync_service_funcs);
assert(ipcs_mapper[service->id].inst);
qb_ipcs_poll_handlers_set(ipcs_mapper[service->id].inst,
&corosync_poll_funcs);
if (qb_ipcs_run(ipcs_mapper[service->id].inst) != 0) {
log_printf (LOGSYS_LEVEL_ERROR, "Can't initialize IPC");
return "qb_ipcs_run error";
}
return NULL;
}
void cs_ipcs_init(void)
{
api = apidef_get ();
qb_loop_poll_low_fds_event_set(cs_poll_handle_get(), cs_ipcs_low_fds_event);
api->quorum_register_callback (cs_ipcs_fc_quorum_changed, NULL);
totempg_queue_level_register_callback (cs_ipcs_totem_queue_level_changed);
global_stats.active = 0;
global_stats.closed = 0;
}
apidef_get
struct corosync_api_v1 * apidef_get(void)
Definition apidef.c:146
apidef.h
config.h
CS_LIB_ALLOW_INQUORATE
@ CS_LIB_ALLOW_INQUORATE
Definition coroapi.h:164
SERVICES_COUNT_MAX
#define SERVICES_COUNT_MAX
Definition coroapi.h:462
corodefs.h
CPG_SERVICE
@ CPG_SERVICE
Definition corodefs.h:46
MON_SERVICE
@ MON_SERVICE
Definition corodefs.h:50
WD_SERVICE
@ WD_SERVICE
Definition corodefs.h:51
PLOAD_SERVICE
@ PLOAD_SERVICE
Definition corodefs.h:48
CMAP_SERVICE
@ CMAP_SERVICE
Definition corodefs.h:44
CFG_SERVICE
@ CFG_SERVICE
Definition corodefs.h:45
QUORUM_SERVICE
@ QUORUM_SERVICE
Definition corodefs.h:47
VOTEQUORUM_SERVICE
@ VOTEQUORUM_SERVICE
Definition corodefs.h:49
corotypes.h
cs_error_t
cs_error_t
The cs_error_t enum.
Definition corotypes.h:98
CS_ERR_TRY_AGAIN
@ CS_ERR_TRY_AGAIN
Definition corotypes.h:104
CS_OK
@ CS_OK
Definition corotypes.h:99
CS_ERR_INVALID_PARAM
@ CS_ERR_INVALID_PARAM
Definition corotypes.h:105
CS_ERR_NOT_EXIST
@ CS_ERR_NOT_EXIST
Definition corotypes.h:110
icmap.h
icmap_get_uint8
cs_error_t icmap_get_uint8(const char *key_name, uint8_t *u8)
Definition icmap.c:872
ICMAP_KEYNAME_MAXLEN
#define ICMAP_KEYNAME_MAXLEN
Maximum length of key in icmap.
Definition icmap.h:48
icmap_get_string
cs_error_t icmap_get_string(const char *key_name, char **str)
Shortcut for icmap_get for string type.
Definition icmap.c:860
cs_ipcs_init
void cs_ipcs_init(void)
Definition ipc_glue.c:818
cs_ipcs_private_data_get
void * cs_ipcs_private_data_get(void *conn)
Definition ipc_glue.c:312
cs_ipcs_sync_state_changed
void cs_ipcs_sync_state_changed(int32_t sync_in_process)
Definition ipc_glue.c:664
cs_ipcs_service_destroy
int32_t cs_ipcs_service_destroy(int32_t service_id)
Definition ipc_glue.c:164
CS_IPCS_MAPPER_SERV_NAME
#define CS_IPCS_MAPPER_SERV_NAME
Definition ipc_glue.c:75
cs_ipc_refcnt_inc
void cs_ipc_refcnt_inc(void *conn)
Definition ipc_glue.c:302
cs_ipcs_get_conn_stats
cs_error_t cs_ipcs_get_conn_stats(int service_id, uint32_t pid, void *conn_ptr, struct ipcs_conn_stats *ipcs_stats)
Definition ipc_glue.c:675
cs_ipc_refcnt_dec
void cs_ipc_refcnt_dec(void *conn)
Definition ipc_glue.c:307
cs_ipcs_service_init
const char * cs_ipcs_service_init(struct corosync_service_engine *service)
Definition ipc_glue.c:778
cs_ipcs_dispatch_iov_send
int cs_ipcs_dispatch_iov_send(void *conn, const struct iovec *iov, unsigned int iov_len)
Definition ipc_glue.c:483
cs_ipcs_get_global_stats
void cs_ipcs_get_global_stats(struct ipcs_global_stats *ipcs_stats)
Definition ipc_glue.c:670
cs_ipcs_response_send
int cs_ipcs_response_send(void *conn, const void *msg, size_t mlen)
Definition ipc_glue.c:374
cs_ipcs_response_iov_send
int cs_ipcs_response_iov_send(void *conn, const struct iovec *iov, unsigned int iov_len)
Definition ipc_glue.c:363
cs_ipcs_q_level_get
int32_t cs_ipcs_q_level_get(void)
Definition ipc_glue.c:601
cs_ipcs_clear_stats
void cs_ipcs_clear_stats()
Definition ipc_glue.c:709
cs_ipc_allow_connections
void cs_ipc_allow_connections(int32_t allow)
Definition ipc_glue.c:159
cs_ipcs_dispatch_send
int cs_ipcs_dispatch_send(void *conn, const void *msg, size_t mlen)
Definition ipc_glue.c:474
ipcs_stats.h
util.h
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_LEVEL_INFO
#define LOGSYS_LEVEL_INFO
Definition logsys.h:75
LOGSYS_LEVEL_NOTICE
#define LOGSYS_LEVEL_NOTICE
Definition logsys.h:74
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
corosync_sending_allowed_release
void corosync_sending_allowed_release(void *sending_allowed_private_data)
Definition main.c:785
cs_poll_handle_get
qb_loop_t * cs_poll_handle_get(void)
Definition main.c:178
corosync_recheck_the_q_level
void corosync_recheck_the_q_level(void *data)
Definition main.c:723
corosync_sending_allowed
int corosync_sending_allowed(unsigned int service, unsigned int id, const void *msg, void *sending_allowed_private_data)
Definition main.c:737
main.h
quorate
uint32_t quorate
Definition sam.c:134
corosync_service
struct corosync_service_engine * corosync_service[SERVICES_COUNT_MAX]
Definition service.c:110
service.h
stats_ipcs_add_connection
void stats_ipcs_add_connection(int service_id, uint32_t pid, void *ptr)
Definition stats.c:765
stats_ipcs_del_connection
void stats_ipcs_del_connection(int service_id, uint32_t pid, void *ptr)
Definition stats.c:775
stats.h
corosync_api_v1
The corosync_api_v1 struct.
Definition coroapi.h:225
corosync_api_v1::quorum_register_callback
int(* quorum_register_callback)(quorum_callback_fn_t callback_fn, void *context)
Definition coroapi.h:386
corosync_lib_handler::lib_handler_fn
void(* lib_handler_fn)(void *conn, const void *msg)
Definition coroapi.h:468
corosync_service_engine
The corosync_service_engine struct.
Definition coroapi.h:490
corosync_service_engine::id
unsigned short id
Definition coroapi.h:492
corosync_service_engine::lib_exit_fn
int(* lib_exit_fn)(void *conn)
Definition coroapi.h:503
corosync_service_engine::private_data_size
size_t private_data_size
Definition coroapi.h:496
corosync_service_engine::lib_engine
struct corosync_lib_handler * lib_engine
Definition coroapi.h:504
corosync_service_engine::lib_engine_count
int lib_engine_count
Definition coroapi.h:505
cs_ipcs_conn_context
Definition ipcs_stats.h:35
cs_ipcs_conn_context::queued
uint32_t queued
Definition ipcs_stats.h:38
cs_ipcs_conn_context::queuing
int32_t queuing
Definition ipcs_stats.h:37
cs_ipcs_conn_context::overload
uint64_t overload
Definition ipcs_stats.h:40
cs_ipcs_conn_context::data
char data[1]
Definition ipcs_stats.h:43
cs_ipcs_conn_context::sent
uint32_t sent
Definition ipcs_stats.h:41
cs_ipcs_conn_context::proc_name
char proc_name[32]
Definition ipcs_stats.h:42
cs_ipcs_conn_context::invalid_request
uint64_t invalid_request
Definition ipcs_stats.h:39
cs_ipcs_conn_context::outq_head
struct qb_list_head outq_head
Definition ipcs_stats.h:36
cs_ipcs_mapper
Definition ipc_glue.c:77
cs_ipcs_mapper::id
int32_t id
Definition ipc_glue.c:78
cs_ipcs_mapper::inst
qb_ipcs_service_t * inst
Definition ipc_glue.c:79
cs_ipcs_mapper::name
char name[CS_IPCS_MAPPER_SERV_NAME]
Definition ipc_glue.c:80
ipcs_conn_stats
Definition ipcs_stats.h:53
ipcs_conn_stats::conn
struct qb_ipcs_connection_stats conn
Definition ipcs_stats.h:55
ipcs_conn_stats::cnx
struct cs_ipcs_conn_context cnx
Definition ipcs_stats.h:56
ipcs_conn_stats::srv
struct qb_ipcs_stats srv
Definition ipcs_stats.h:54
ipcs_global_stats
Definition ipcs_stats.h:47
ipcs_global_stats::active
uint64_t active
Definition ipcs_stats.h:48
ipcs_global_stats::closed
uint64_t closed
Definition ipcs_stats.h:49
outq_item
Definition ipc_glue.c:83
outq_item::mlen
size_t mlen
Definition ipc_glue.c:85
outq_item::list
struct qb_list_head list
Definition ipc_glue.c:86
outq_item::msg
void * msg
Definition ipc_glue.c:84
swab.h
sync.h
timer.h
totempg.h
Totem Single Ring Protocol.
totempg_queue_level_register_callback
void totempg_queue_level_register_callback(totem_queue_level_changed_fn)
Definition totempg.c:1547
totem_q_level
totem_q_level
Definition totempg.h:180
TOTEM_Q_LEVEL_GOOD
@ TOTEM_Q_LEVEL_GOOD
Definition totempg.h:182
TOTEM_Q_LEVEL_HIGH
@ TOTEM_Q_LEVEL_HIGH
Definition totempg.h:183
TOTEM_Q_LEVEL_LOW
@ TOTEM_Q_LEVEL_LOW
Definition totempg.h:181
TOTEM_Q_LEVEL_CRITICAL
@ TOTEM_Q_LEVEL_CRITICAL
Definition totempg.h:184