Proxmox-Port/libqb
1
0
Fork 0
mirror of https://salsa.debian.org/ha-team/libqb synced 2026-08-08 20:07:21 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

IPC: rewrite (simpler API & more structured layout).

Browse Source 
- implement using posix message queues
- implement using sys-v message queues
- implement shared memory ringbuffers
- add auth via unix sockets
- add items to the TODO

This is still a bit rough, more work to follow...

Signed-off-by: Angus Salkeld <asalkeld@redhat.com>
This commit is contained in:
Angus Salkeld 2010-08-23 12:46:36 +10:00
parent 5221880227
commit 8a6b8d78db
21 changed files with 2487 additions and 3201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
TODO
View File

@ -6,3 +6,12 @@ Generic Items
* use doxygen to generate the man pages
* write unit test cases
IPC
---
* flow control
* client or server to specify the message size
* 1 or per session request queues/ringbuffers
* glib integration
* bmcpt fails - fix
* should we hide the message header?
* we probably need a qb_ipc_msg_alloc()

7
configure.ac
View File

@ -131,15 +131,14 @@ AC_FUNC_STRERROR_R
AC_FUNC_SELECT_ARGTYPES
AC_TYPE_SIGNAL
AC_FUNC_VPRINTF
AC_CHECK_FUNCS([alarm alphasort atexit bzero dup2 endgrent endpwent fcntl \
getcwd getpeerucred getpeereid gettimeofday inet_ntoa memmove \
memset mkdir scandir select socket strcasecmp strchr strdup \
AC_CHECK_FUNCS([alphasort atexit endgrent endpwent fcntl \
getcwd getpeerucred getpeereid gettimeofday inet_ntoa \
memset scandir socket strcasecmp strchr strdup \
strerror strrchr strspn strstr pthread_spin_lock \
clock_gettime localeconv localtime_r munmap pathconf putenv setenv \
pthread_spin_unlock pututxline lgammal setkey \
crypt encrypt getdate strsignal])
## local defines
PACKAGE_FEATURES=""

31
include/qb/qbipc_common.h
View File

@ -23,15 +23,34 @@
#ifndef QB_IPC_COMMON_H_DEFINED
#define QB_IPC_COMMON_H_DEFINED
typedef struct {
int32_t size __attribute__ ((aligned(8)));
struct qb_ipc_request_header {
int32_t id __attribute__ ((aligned(8)));
} qb_ipc_request_header_t __attribute__ ((aligned(8)));
int32_t size __attribute__ ((aligned(8)));
uint64_t session_id __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
typedef struct {
int32_t size __attribute__ ((aligned(8)));
struct qb_ipc_response_header {
int32_t id __attribute__ ((aligned(8)));
int32_t size __attribute__ ((aligned(8)));
int32_t error __attribute__ ((aligned(8)));
} qb_ipc_response_header_t __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
enum qb_ipc_type {
QB_IPC_SOCKET,
QB_IPC_SHM,
QB_IPC_POSIX_MQ,
QB_IPC_SYSV_MQ,
};
enum qb_ipc_msg_ids {
QB_IPC_MSG_UNUSED,
QB_IPC_MSG_AUTHENTICATE,
QB_IPC_MSG_CONNECT,
QB_IPC_MSG_DISCONNECT,
QB_IPC_MSG_NEW_MESSAGE,
QB_IPC_MSG_USER_START,
};
#endif /* QB_IPC_COMMON_H_DEFINED */

47
include/qb/qbipcc.h
View File

@ -34,49 +34,18 @@ extern "C" {
#endif
/* *INDENT-ON* */
int32_t
qb_ipcc_service_connect(const char *socket_name,
uint32_t service,
size_t request_size,
size_t respnse__size,
size_t dispatch_size, qb_handle_t * handle);
typedef struct qb_ipcc_connection qb_ipcc_connection_t;
int32_t qb_ipcc_service_disconnect(qb_handle_t handle);
qb_ipcc_connection_t*
qb_ipcc_connect(const char *name, enum qb_ipc_type type);
int32_t qb_ipcc_fd_get(qb_handle_t handle, int32_t * fd);
int32_t qb_ipcc_send(qb_ipcc_connection_t* c, const void *msg_ptr,
size_t msg_len);
ssize_t qb_ipcc_recv(qb_ipcc_connection_t* c, const void *msg_ptr,
size_t msg_len);
int32_t qb_ipcc_dispatch_get(qb_handle_t handle, void **buf, int32_t timeout);
void qb_ipcc_disconnect(qb_ipcc_connection_t* c);
int32_t qb_ipcc_dispatch_put(qb_handle_t handle);
int32_t
qb_ipcc_dispatch_flow_control_get(qb_handle_t handle,
uint32_t * flow_control_state);
int32_t
qb_ipcc_msg_send(qb_handle_t handle, const struct iovec *iov, uint32_t iov_len);
int32_t
qb_ipcc_msg_send_reply_receive(qb_handle_t handle,
const struct iovec *iov,
uint32_t iov_len, void *res_msg, size_t res_len);
int32_t
qb_ipcc_msg_send_reply_receive_in_buf_get(qb_handle_t handle,
const struct iovec *iov,
uint32_t iov_len, void **res_msg);
int32_t qb_ipcc_msg_send_reply_receive_in_buf_put(qb_handle_t handle);
int32_t
qb_ipcc_zcb_alloc(qb_handle_t handle,
void **buffer, size_t size, size_t header_size);
int32_t qb_ipcc_zcb_free(qb_handle_t handle, void *buffer);
int32_t
qb_ipcc_zcb_msg_send_reply_receive(qb_handle_t handle,
void *msg, void *res_msg, size_t res_len);
/* *INDENT-OFF* */
#ifdef __cplusplus

91
include/qb/qbipcs.h
View File

@ -1,7 +1,8 @@
/*
* Copyright (C) 2006-2009 Red Hat, Inc.
*
* Author: Steven Dake <sdake@redhat.com>
* Author: Steven Dake <sdake@redhat.com>,
* Angus Salkeld <asalkeld@redhat.com>
*
* This file is part of libqb.
*
@ -24,6 +25,7 @@
#include <stdlib.h>
#include <qb/qbipc_common.h>
#include <qb/qbhdb.h>
/* *INDENT-OFF* */
#ifdef __cplusplus
@ -31,71 +33,50 @@ extern "C" {
#endif
/* *INDENT-ON* */
struct iovec;
typedef qb_handle_t qb_ipcs_connection_pt;
typedef qb_handle_t qb_ipcs_service_pt;
typedef int32_t(*qb_ipcs_init_fn_lvalue) (void *conn);
typedef int32_t(*qb_ipcs_exit_fn_lvalue) (void *conn);
typedef void (*qb_ipcs_handler_fn_lvalue) (void *conn, const void *msg);
typedef int32_t (*qb_ipcs_dispatch_fn_t) (qb_ipcs_service_pt s, int32_t fd, int32_t revents,
void *data);
typedef int32_t (*qb_ipcs_dispatch_add_fn)(qb_ipcs_service_pt s, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn);
typedef int32_t (*qb_ipcs_dispatch_rm_fn)(qb_ipcs_service_pt s, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn);
struct qb_ipcs_init_state {
const char *socket_name;
int32_t sched_policy;
const struct sched_param *sched_param;
void *(*malloc) (size_t size);
void (*free) (void *ptr);
int32_t(*service_available) (uint32_t service);
int32_t(*private_data_size_get) (uint32_t service);
int32_t(*security_valid) (int32_t uid, int32_t gid);
void (*serialize_lock) (void);
void (*serialize_unlock) (void);
int32_t(*sending_allowed) (uint32_t service, uint32_t id,
const void *msg,
void *sending_allowed_private_data);
void (*sending_allowed_release) (void *sending_allowed_private_data);
void (*poll_accept_add) (int32_t fd);
void (*poll_dispatch_add) (int32_t fd, void *context);
void (*poll_dispatch_modify) (int32_t fd, int32_t events);
void (*poll_dispatch_destroy) (int32_t fd, void *context);
void (*fatal_error) (const char *error_msg);
qb_ipcs_init_fn_lvalue(*init_fn_get) (uint32_t service);
qb_ipcs_exit_fn_lvalue(*exit_fn_get) (uint32_t service);
qb_ipcs_handler_fn_lvalue(*handler_fn_get) (uint32_t service,
uint32_t id);
qb_handle_t(*stats_create_connection) (const char *name, pid_t pid,
int32_t fd);
void (*stats_destroy_connection) (qb_handle_t handle);
void (*stats_update_value) (qb_handle_t handle,
const char *name, const void *value,
size_t value_len);
void (*stats_increment_value) (qb_handle_t handle, const char *name);
void (*stats_decrement_value) (qb_handle_t handle, const char *name);
struct qb_ipcs_poll_handlers {
qb_ipcs_dispatch_add_fn dispatch_add;
qb_ipcs_dispatch_rm_fn dispatch_rm;
};
void qb_ipcs_ipc_init(struct qb_ipcs_init_state *init_state);
typedef int32_t (*qb_ipcs_connection_authenticate_fn) (qb_ipcs_connection_pt c, uid_t uid, gid_t gid);
typedef void (*qb_ipcs_connection_created_fn) (qb_ipcs_connection_pt c);
typedef void (*qb_ipcs_connection_destroyed_fn) (qb_ipcs_connection_pt c);
typedef void (*qb_ipcs_msg_process_fn) (qb_ipcs_connection_pt c,
void *data, size_t size);
void *qb_ipcs_private_data_get(void *conn);
struct qb_ipcs_service_handlers {
qb_ipcs_connection_authenticate_fn connection_authenticate;
qb_ipcs_connection_created_fn connection_created;
qb_ipcs_msg_process_fn msg_process;
qb_ipcs_connection_destroyed_fn connection_destroyed;
};
int32_t qb_ipcs_response_send(void *conn, const void *msg, size_t mlen);
qb_ipcs_service_pt qb_ipcs_create(const char* name,
enum qb_ipc_type type,
size_t max_msg_size);
int32_t qb_ipcs_response_iov_send(void *conn,
const struct iovec *iov, uint32_t iov_len);
void qb_ipcs_service_handlers_set(qb_ipcs_service_pt s,
struct qb_ipcs_service_handlers *handlers);
int32_t qb_ipcs_dispatch_send(void *conn, const void *msg, size_t mlen);
void qb_ipcs_poll_handlers_set(qb_ipcs_service_pt s,
struct qb_ipcs_poll_handlers *handlers);
int32_t qb_ipcs_dispatch_iov_send(void *conn,
const struct iovec *iov, uint32_t iov_len);
int32_t qb_ipcs_run(qb_ipcs_service_pt s, qb_handle_t poll);
void qb_ipcs_refcount_inc(void *conn);
void qb_ipcs_destroy(qb_ipcs_service_pt s);
void qb_ipcs_refcount_dec(void *conn);
void qb_ipcs_ipc_exit(void);
int32_t qb_ipcs_ipc_service_exit(uint32_t service);
int32_t qb_ipcs_handler_accept(int32_t fd, int32_t revent, void *context);
int32_t qb_ipcs_handler_dispatch(int32_t fd, int32_t revent, void *context);
ssize_t qb_ipcs_response_send(qb_ipcs_connection_pt c, void *data, size_t size);
/* *INDENT-OFF* */
#ifdef __cplusplus

2
include/qb/qblist.h
View File

@ -74,6 +74,8 @@ static void inline qb_list_add(struct qb_list_head *element,
/**
* Add to the list (but at the end of the list).
*
* @param element pointer to the element to add
* @param head pointer to the list head
* @see qb_list_add()
*/

1
include/qb/qbrb.h
View File

@ -180,6 +180,7 @@ int32_t qb_rb_chunk_commit(qb_ringbuffer_t * rb, size_t len);
* qb_rb_chunk_reclaim().
* @param rb ringbuffer instance
* @param data_out (out) a pointer to the next chunk to read (not copied).
* @param ms_timeout (in) time to wait for new data.
*
* @return the size of the chunk (0 if buffer empty).
*/

4
lib/Makefile.am
View File

@ -44,7 +44,9 @@ lib_LTLIBRARIES = libqb.la
libqb_la_CPPFLAGS = -I$(top_builddir)/include -I$(top_srcdir)/include
libqb_la_LDFLAGS = -version-info 0:0:0
libqb_la_SOURCES = util.c tsafe.c hash.c poll.c timer.c wthread.c \
ipcc.c ipcs.c logsys.c ringbuffer.c ringbuffer_helper.c \
ipcc.c ipcs.c \
ipc_posix_mq.c ipc_sysv_mq.c ipc_shm.c ipc_us.c \
logsys.c ringbuffer.c ringbuffer_helper.c \
hdb.c
pkgconfigdir = $(libdir)/pkgconfig

1
lib/hash.c
View File

@ -23,7 +23,6 @@
#include <pthread.h>
#include "util_int.h"
#include <qb/qbhdb.h>
#include <qb/qbhdb.h>
#include <qb/qblist.h>
#include <qb/qbhash.h>

261
lib/ipc_int.h
View File

@ -2,6 +2,7 @@
* Copyright (C) 2009 Red Hat, Inc.
*
* Author: Steven Dake <sdake@redhat.com>
* Angus Salkeld <asalkeld@redhat.com>
*
* This file is part of libqb.
*
@ -18,11 +19,18 @@
* You should have received a copy of the GNU Lesser General Public License
* along with libqb. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef QB_IPC_IPC_H_DEFINED
#define QB_IPC_IPC_H_DEFINED
#ifndef QB_IPC_INT_H_DEFINED
#define QB_IPC_INT_H_DEFINED
#include <unistd.h>
#include "config.h"
#include <dirent.h>
#include <mqueue.h>
#include <qb/qblist.h>
#include <qb/qbipcc.h>
#include <qb/qbipcs.h>
#include <qb/qbipc_common.h>
#include <qb/qbrb.h>
/*
* Darwin claims to support process shared synchronization
@ -41,94 +49,189 @@
#include <semaphore.h>
#endif
enum req_init_types {
MESSAGE_REQ_RESPONSE_INIT = 0,
MESSAGE_REQ_DISPATCH_INIT = 1
struct qb_ipcc_connection;
struct qb_ipcc_funcs {
int32_t (*send)(struct qb_ipcc_connection* c, const void *msg_ptr,
size_t msg_len);
ssize_t (*recv)(struct qb_ipcc_connection* c, const void *msg_ptr,
size_t msg_len);
void (*disconnect)(struct qb_ipcc_connection* c);
};
#define MESSAGE_REQ_CHANGE_EUID 1
#define MESSAGE_REQ_OUTQ_FLUSH 2
#define MESSAGE_RES_OUTQ_EMPTY 0
#define MESSAGE_RES_OUTQ_NOT_EMPTY 1
#define MESSAGE_RES_ENABLE_FLOWCONTROL 2
#define MESSAGE_RES_OUTQ_FLUSH_NR 3
struct control_buffer {
uint32_t read;
uint32_t write;
#if _POSIX_THREAD_PROCESS_SHARED > 0
sem_t sem1;
sem_t sem2;
#endif
struct qb_ipcc_pmq_one_way {
mqd_t q;
char name[NAME_MAX];
};
enum res_init_types {
MESSAGE_RES_INIT
struct qb_ipcc_smq_one_way {
int32_t q;
int32_t key;
};
typedef struct {
int32_t service __attribute__ ((aligned(8)));
uint64_t semkey __attribute__ ((aligned(8)));
char control_file[64] __attribute__ ((aligned(8)));
char request_file[64] __attribute__ ((aligned(8)));
char response_file[64] __attribute__ ((aligned(8)));
char dispatch_file[64] __attribute__ ((aligned(8)));
size_t control_size __attribute__ ((aligned(8)));
size_t request_size __attribute__ ((aligned(8)));
size_t response_size __attribute__ ((aligned(8)));
size_t dispatch_size __attribute__ ((aligned(8)));
} mar_req_setup_t __attribute__ ((aligned(8)));
struct qb_ipcc_shm_one_way {
qb_ringbuffer_t *rb;
char name[NAME_MAX];
};
typedef struct {
int32_t error __attribute__ ((aligned(8)));
} mar_res_setup_t __attribute__ ((aligned(8)));
struct qb_ipcc_pmq_connection {
struct qb_ipcc_pmq_one_way request;
struct qb_ipcc_pmq_one_way response;
struct qb_ipcc_pmq_one_way dispatch;
};
struct qb_ipcc_smq_connection {
struct qb_ipcc_smq_one_way request;
struct qb_ipcc_smq_one_way response;
struct qb_ipcc_smq_one_way dispatch;
};
struct qb_ipcc_shm_connection {
struct qb_ipcc_shm_one_way request;
struct qb_ipcc_shm_one_way response;
struct qb_ipcc_shm_one_way dispatch;
};
struct qb_ipcc_connection {
enum qb_ipc_type type;
char name[NAME_MAX];
uint64_t session_id;
int32_t needs_sock_for_poll;
int32_t sock;
union {
struct qb_ipcc_pmq_connection pmq;
struct qb_ipcc_smq_connection smq;
struct qb_ipcc_shm_connection shm;
} u;
struct qb_ipcc_funcs funcs;
size_t max_msg_size;
char *receive_buf;
};
int32_t qb_ipc_us_send(int32_t s, const void *msg, size_t len);
int32_t qb_ipc_us_recv (int32_t s, void *msg, size_t len);
int32_t qb_ipcc_us_connect(const char *socket_name, int32_t *sock_pt);
void qb_ipcc_us_disconnect (int32_t sock);
int32_t qb_ipcc_pmq_connect(struct qb_ipcc_connection *c);
int32_t qb_ipcc_soc_connect(struct qb_ipcc_connection *c);
int32_t qb_ipcc_smq_connect(struct qb_ipcc_connection *c);
int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection *c);
struct qb_ipcs_service;
struct qb_ipcs_connection;
struct qb_ipcs_funcs {
void (*destroy)(struct qb_ipcs_service *s);
int32_t (*connect)(struct qb_ipcs_service *s, struct qb_ipcs_connection *c,
void *data, size_t size);
void (*disconnect)(struct qb_ipcs_connection *c);
ssize_t (*request_recv)(struct qb_ipcs_service *s, void *buf, size_t buf_size);
ssize_t (*response_send)(struct qb_ipcs_connection *c, void *data, size_t size);
};
struct qb_ipcs_service {
enum qb_ipc_type type;
char name[NAME_MAX];
pid_t pid;
int32_t needs_sock_for_poll;
int32_t server_sock;
qb_handle_t poll_handle;
struct qb_ipcs_service_handlers serv_fns;
struct qb_ipcs_poll_handlers poll_fns;
struct qb_ipcs_funcs funcs;
struct qb_list_head connections;
union {
mqd_t q;
qb_ringbuffer_t *rb;
struct qb_ipcc_smq_one_way smq;
} u;
size_t max_msg_size;
char *receive_buf;
};
struct qb_ipcs_connection {
qb_ipcs_connection_pt handle;
pid_t pid;
uid_t euid;
gid_t egid;
int32_t sock;
union {
struct qb_ipcc_pmq_connection pmq;
struct qb_ipcc_smq_connection smq;
struct qb_ipcc_shm_connection shm;
} u;
struct qb_ipcs_service *service;
struct qb_list_head list;
};
int32_t qb_ipcs_pmq_create(struct qb_ipcs_service *s);
int32_t qb_ipcs_soc_create(struct qb_ipcs_service *s);
int32_t qb_ipcs_smq_create(struct qb_ipcs_service *s);
int32_t qb_ipcs_shm_create(struct qb_ipcs_service *s);
int32_t qb_ipcs_us_publish(struct qb_ipcs_service *s);
int32_t qb_ipcs_us_withdraw(struct qb_ipcs_service *s);
int32_t qb_ipcs_dispatch_connection_request(qb_handle_t hdb_handle_t,
int32_t fd, int32_t revents, void *data);
int32_t qb_ipcs_dispatch_service_request(qb_handle_t hdb_handle_t,
int32_t fd, int32_t revents, void *data);
struct qb_ipcs_connection* qb_ipcs_connection_alloc(struct qb_ipcs_service *s);
int32_t qb_ipcs_process_request(struct qb_ipcs_service *s,
struct qb_ipc_request_header *hdr);
void qb_ipcs_disconnect(struct qb_ipcs_connection *c);
struct mar_req_initial_setup {
struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
struct mar_res_initial_setup {
struct qb_ipc_response_header hdr __attribute__ ((aligned(8)));
int32_t connection_type __attribute__ ((aligned(8)));
uint64_t session_id __attribute__ ((aligned(8)));
uint32_t max_msg_size __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
struct mar_req_shm_setup {
struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
uint32_t pid __attribute__ ((aligned(8)));
char request[PATH_MAX] __attribute__ ((aligned(8)));
char response[PATH_MAX] __attribute__ ((aligned(8)));
char dispatch[PATH_MAX] __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
struct mar_req_pmq_setup {
struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
uint32_t pid __attribute__ ((aligned(8)));
char response_mq[NAME_MAX] __attribute__ ((aligned(8)));
char dispatch_mq[NAME_MAX] __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
struct mar_req_smq_setup {
struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
uint32_t pid __attribute__ ((aligned(8)));
int32_t response_key __attribute__ ((aligned(8)));
int32_t dispatch_key __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
struct mar_res_setup {
struct qb_ipc_response_header hdr __attribute__ ((aligned(8)));
size_t max_msg_size __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
typedef struct {
uid_t euid __attribute__ ((aligned(8)));
gid_t egid __attribute__ ((aligned(8)));
} mar_req_priv_change __attribute__ ((aligned(8)));
typedef struct {
qb_ipc_response_header_t header __attribute__ ((aligned(8)));
uint64_t conn_info __attribute__ ((aligned(8)));
} mar_res_lib_response_init_t __attribute__ ((aligned(8)));
typedef struct {
qb_ipc_response_header_t header __attribute__ ((aligned(8)));
} mar_res_lib_dispatch_init_t __attribute__ ((aligned(8)));
typedef struct {
uint32_t nodeid __attribute__ ((aligned(8)));
void *conn __attribute__ ((aligned(8)));
} mar_message_source_t __attribute__ ((aligned(8)));
typedef struct {
qb_ipc_request_header_t header __attribute__ ((aligned(8)));
size_t map_size __attribute__ ((aligned(8)));
char path_to_file[128] __attribute__ ((aligned(8)));
} mar_req_qb_ipcc_zc_alloc_t __attribute__ ((aligned(8)));
typedef struct {
qb_ipc_request_header_t header __attribute__ ((aligned(8)));
size_t map_size __attribute__ ((aligned(8)));
uint64_t server_address __attribute__ ((aligned(8)));
} mar_req_qb_ipcc_zc_free_t __attribute__ ((aligned(8)));
typedef struct {
qb_ipc_request_header_t header __attribute__ ((aligned(8)));
uint64_t server_address __attribute__ ((aligned(8)));
} mar_req_qb_ipcc_zc_execute_t __attribute__ ((aligned(8)));
struct qb_ipcs_zc_header {
int32_t map_size;
uint64_t server_address;
};
#define SOCKET_SERVICE_INIT 0xFFFFFFFF
#define ZC_ALLOC_HEADER 0xFFFFFFFF
#define ZC_FREE_HEADER 0xFFFFFFFE
#define ZC_EXECUTE_HEADER 0xFFFFFFFD
#endif /* QB_IPC_IPC_H_DEFINED */
#endif /* QB_IPC_INT_H_DEFINED */

436
lib/ipc_posix_mq.c Normal file
View File

@ -0,0 +1,436 @@
/*
* Copyright (C) 2010 Red Hat, Inc.
*
* Author: Angus Salkeld <asalkeld@redhat.com>
*
* This file is part of libqb.
*
* libqb is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation, either version 2.1 of the License, or
* (at your option) any later version.
*
* libqb is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with libqb. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os_base.h"
#include <mqueue.h>
#include <sys/resource.h>
#include "ipc_int.h"
#include "util_int.h"
#include <qb/qbpoll.h>
static ssize_t qb_ipcs_pmq_dispatch_send(struct qb_ipcs_connection *c,
void *data, size_t size);
/*
* utility functions
* --------------------------------------------------------
*/
static int32_t posix_mq_increase_limits(size_t max_msg_size,
int32_t q_len)
{
FILE* proc_fd;
int32_t msgsize_max;
char size_str[10];
int32_t res = 0;
struct rlimit rlim;
int32_t q_limit;
proc_fd = fopen("/proc/sys/fs/mqueue/msgsize_max", "r+");
if (proc_fd > 0) {
res = fscanf(proc_fd, "%d", &msgsize_max);
} else {
qb_util_log(LOG_ERR, "fopen failed");
res = -1;
}
if (res == 1) {
if (msgsize_max <= max_msg_size) {
/* we need to increase the size */
snprintf(size_str, 10, "%zd", (max_msg_size + 1));
fwrite(size_str, 1, strlen(size_str), proc_fd);
}
} else {
qb_util_log(LOG_ERR, "fscanf failed");
return -1;
}
fclose(proc_fd);
if (getrlimit(RLIMIT_MSGQUEUE, &rlim) != 0) {
qb_util_log(LOG_ERR, "getrlimit failed");
return -1;
}
q_limit = (max_msg_size * q_len * 4) / 3;
rlim.rlim_cur += q_limit;
rlim.rlim_max += q_limit;
if (setrlimit(RLIMIT_MSGQUEUE, &rlim) != 0) {
qb_util_log(LOG_ERR, "setrlimit failed");
return -1;
}
return 0;
}
static int32_t posix_mq_create(const char* mq_name, size_t max_msg_size,
int32_t flags)
{
struct mq_attr attr;
int32_t res = 0;
int32_t q_len = 10;
attr.mq_flags = O_NONBLOCK;
attr.mq_maxmsg = q_len;
attr.mq_msgsize = max_msg_size;
mq_unlink(mq_name);
res = mq_open(mq_name, flags, 0600, &attr);
if (res == -1) {
perror(mq_name);
}
printf("%s(%s, %zd, %d) == %d\n",
__func__, mq_name, max_msg_size, flags, res);
return res;
}
/*
* client functions
* --------------------------------------------------------
*/
static int32_t qb_ipcc_pmq_send(struct qb_ipcc_connection *c,
const void *msg_ptr, size_t msg_len)
{
return mq_send(c->u.pmq.request.q, msg_ptr, msg_len, 1);
}
static ssize_t qb_ipcc_pmq_recv(struct qb_ipcc_connection *c,
const void *msg_ptr, size_t msg_len)
{
uint32_t msg_prio;
return mq_receive(c->u.pmq.response.q, (char *)msg_ptr, c->max_msg_size,
&msg_prio);
}
static void qb_ipcc_pmq_disconnect(struct qb_ipcc_connection *c)
{
struct qb_ipc_request_header hdr;
printf("%s()\n", __func__);
if (c->needs_sock_for_poll) {
return;
}
hdr.id = QB_IPC_MSG_DISCONNECT;
hdr.session_id = c->session_id;
hdr.size = sizeof(hdr);
mq_send(c->u.pmq.request.q, (const char *)&hdr, hdr.size, 30);
mq_close(c->u.pmq.dispatch.q);
mq_unlink(c->u.pmq.dispatch.name);
mq_close(c->u.pmq.response.q);
mq_unlink(c->u.pmq.response.name);
mq_close(c->u.pmq.request.q);
}
static int32_t _ipcc_pmq_connect_to_service_(struct qb_ipcc_connection *c)
{
int32_t res;
ssize_t size;
uint32_t priority;
struct mar_req_pmq_setup start;
struct mar_res_setup *msg_res;
start.hdr.id = QB_IPC_MSG_CONNECT;
start.hdr.session_id = c->session_id;
start.pid = getpid();
start.hdr.size = sizeof(struct mar_req_pmq_setup);
strcpy(start.response_mq, c->u.pmq.response.name);
strcpy(start.dispatch_mq, c->u.pmq.dispatch.name);
res =
mq_send(c->u.pmq.request.q, (const char *)&start, start.hdr.size,
30);
if (res == -1) {
res = errno;
perror("mq_send");
return res;
}
printf("sent request to server %d\n", res);
printf("mq_receive'ing on %d\n", c->u.pmq.response.q);
mq_recv_again:
size = mq_receive(c->u.pmq.response.q, c->receive_buf,
c->max_msg_size, &priority);
if (size == -1 && errno == EAGAIN) {
usleep(100000);
goto mq_recv_again;
}
if (size == -1) {
res = errno;
perror("_ipcc_pmq_connect_to_service_:mq_receive");
goto cleanup;
}
printf("received response from server %zd\n", size);
msg_res = (struct mar_res_setup *)c->receive_buf;
res = msg_res->hdr.error;
if (res == 0) {
c->max_msg_size = msg_res->max_msg_size;
}
cleanup:
return res;
}
int32_t qb_ipcc_pmq_connect(struct qb_ipcc_connection * c)
{
int32_t res = 0;
c->funcs.send = qb_ipcc_pmq_send;
c->funcs.recv = qb_ipcc_pmq_recv;
c->funcs.disconnect = qb_ipcc_pmq_disconnect;
if (strlen(c->name) > (NAME_MAX - 20)) {
errno = EINVAL;
return -1;
}
/* Connect to the service's request message queue.
*/
posix_mq_increase_limits(c->max_msg_size, 10);
snprintf(c->u.pmq.request.name, NAME_MAX, "/%s", c->name);
c->u.pmq.request.q = mq_open(c->u.pmq.request.name,
O_WRONLY | O_NONBLOCK);
if (c->u.pmq.request.q == -1) {
perror("mq_open:REQUEST");
return -1;
}
/* Create the response message queue.
*/
res = snprintf(c->u.pmq.response.name,
NAME_MAX, "/%s-response-%d",
c->name, getpid());
posix_mq_increase_limits(c->max_msg_size, 10);
c->u.pmq.response.q = posix_mq_create(c->u.pmq.response.name,
c->max_msg_size,
O_RDONLY | O_CREAT | O_EXCL | O_NONBLOCK);
if (c->u.pmq.response.q == -1) {
perror("mq_open:RESPONSE");
goto cleanup_request;
}
res =
snprintf(c->u.pmq.dispatch.name, NAME_MAX, "/%s-dispatch-%d",
c->name, getpid());
posix_mq_increase_limits(c->max_msg_size, 10);
c->u.pmq.dispatch.q = posix_mq_create(c->u.pmq.dispatch.name,
c->max_msg_size,
O_RDONLY | O_CREAT | O_EXCL | O_NONBLOCK);
if (c->u.pmq.dispatch.q == -1) {
perror("mq_open:DISPATCH");
goto cleanup_request_response;
}
res = _ipcc_pmq_connect_to_service_(c);
if (res == 0) {
return 0;
}
printf("%s:%d\n", __FILE__, __LINE__);
mq_close(c->u.pmq.dispatch.q);
mq_unlink(c->u.pmq.dispatch.name);
cleanup_request_response:
mq_close(c->u.pmq.response.q);
mq_unlink(c->u.pmq.response.name);
cleanup_request:
mq_close(c->u.pmq.request.q);
return -1;
}
/*
* service functions
* --------------------------------------------------------
*/
static void qb_ipcs_pmq_disconnect(struct qb_ipcs_connection *c)
{
struct qb_ipc_response_header msg;
msg.id = QB_IPC_MSG_DISCONNECT;
msg.size = sizeof(msg);
msg.error = 0;
qb_ipcs_pmq_dispatch_send(c, &msg, msg.size);
}
static void qb_ipcs_pmq_destroy(struct qb_ipcs_service *s)
{
struct qb_ipcs_connection *c = NULL;
struct qb_list_head *iter;
struct qb_list_head *iter_next;
printf("%s\n", __func__);
for (iter = s->connections.next;
iter != &s->connections; iter = iter_next) {
iter_next = iter->next;
c = qb_list_entry(iter, struct qb_ipcs_connection, list);
if (c == NULL) {
continue;
}
qb_ipcs_disconnect(c);
}
if (mq_close(s->u.q) == -1)
perror("mq_close");
if (mq_unlink(s->name) == -1)
perror("mq_unlink");
}
static int32_t qb_ipcs_pmq_connect(struct qb_ipcs_service *s,
struct qb_ipcs_connection *c, void *data,
size_t size)
{
int32_t res;
struct mar_req_pmq_setup *init = (struct mar_req_pmq_setup *)data;
struct mar_res_setup accept_msg;
c->pid = init->pid;
c->service = s;
/* setup the response message queue
*/
posix_mq_increase_limits(c->service->max_msg_size, 10);
strcpy(c->u.pmq.response.name, init->response_mq);
c->u.pmq.response.q = mq_open(c->u.pmq.response.name,
O_WRONLY | O_NONBLOCK);
if (c->u.pmq.response.q == -1) {
res = errno;
perror("mq_open:RESPONSE");
return res;
}
qb_util_log(LOG_DEBUG, "%s:%s (fd==%d)",
__func__, c->u.pmq.response.name, c->u.pmq.response.q);
/* setup the dispatch message queue
*/
posix_mq_increase_limits(c->service->max_msg_size, 10);
strcpy(c->u.pmq.dispatch.name, init->dispatch_mq);
qb_util_log(LOG_DEBUG, "%s:%s", __func__, c->u.pmq.dispatch.name);
c->u.pmq.dispatch.q = mq_open(c->u.pmq.dispatch.name,
O_WRONLY | O_NONBLOCK);
if (c->u.pmq.dispatch.q == -1) {
res = errno;
perror("mq_open:DISPATCH");
goto cleanup_response;
}
/* send the "connection accepted" mesage back.
*/
accept_msg.hdr.id = QB_IPC_MSG_CONNECT;
accept_msg.hdr.size = sizeof(struct mar_res_setup);
accept_msg.hdr.error = 0;
accept_msg.max_msg_size = s->max_msg_size;
res =
mq_send(c->u.pmq.response.q, (const char *)&accept_msg,
sizeof(struct mar_res_setup), 30);
if (res == -1) {
res = errno;
perror("mq_send:RESPONSE");
goto cleanup_response;
}
return 0;
cleanup_response:
accept_msg.hdr.error = res;
mq_send(c->u.pmq.response.q, (const char *)&accept_msg,
sizeof(struct mar_res_setup), 30);
mq_close(c->u.pmq.response.q);
return res;
}
static ssize_t qb_ipcs_pmq_request_recv(struct qb_ipcs_service *s, void *buf,
size_t buf_size)
{
uint32_t msg_prio;
return mq_receive(s->u.q, buf, buf_size, &msg_prio);
}
#if 0
static int32_t qb_ipcs_pmq_fd_get(struct qb_ipcs_service *s)
{
return s->u.q;
}
#endif
static ssize_t qb_ipcs_pmq_response_send(struct qb_ipcs_connection *c,
void *data, size_t size)
{
return mq_send(c->u.pmq.response.q, (const char *)data, size, 1);
}
static ssize_t qb_ipcs_pmq_dispatch_send(struct qb_ipcs_connection *c,
void *data, size_t size)
{
return mq_send(c->u.pmq.dispatch.q, (const char *)data, size, 1);
}
int32_t qb_ipcs_pmq_create(struct qb_ipcs_service *s)
{
char mq_name[NAME_MAX];
snprintf(mq_name, NAME_MAX, "/%s", s->name);
s->funcs.destroy = qb_ipcs_pmq_destroy;
s->funcs.request_recv = qb_ipcs_pmq_request_recv;
s->funcs.response_send = qb_ipcs_pmq_response_send;
s->funcs.connect = qb_ipcs_pmq_connect;
s->funcs.disconnect = qb_ipcs_pmq_disconnect;
posix_mq_increase_limits(s->max_msg_size, 10);
s->u.q = posix_mq_create(mq_name, s->max_msg_size,
(O_RDONLY | O_CREAT | O_EXCL | O_NONBLOCK));
if (s->u.q == -1) {
perror("posix_mq_create:REQUEST");
return -1;
}
qb_util_log(LOG_DEBUG, "%s() %d", __func__, s->u.q);
if (!s->needs_sock_for_poll) {
qb_poll_dispatch_add(s->poll_handle, s->u.q,
POLLIN | POLLPRI | POLLNVAL,
s, qb_ipcs_dispatch_service_request);
}
return 0;
}

326
lib/ipc_shm.c Normal file
View File

@ -0,0 +1,326 @@
/*
* Copyright (C) 2010 Red Hat, Inc.
*
* Author: Angus Salkeld <asalkeld@redhat.com>
*
* This file is part of libqb.
*
* libqb is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation, either version 2.1 of the License, or
* (at your option) any later version.
*
* libqb is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with libqb. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os_base.h"
#include "ipc_int.h"
#include "util_int.h"
#include <qb/qbpoll.h>
#include <qb/qbrb.h>
static ssize_t qb_ipcs_shm_dispatch_send(struct qb_ipcs_connection *c,
void *data, size_t size);
/*
* utility functions
* --------------------------------------------------------
*/
/*
* client functions
* --------------------------------------------------------
*/
static void qb_ipcc_shm_disconnect(struct qb_ipcc_connection* c)
{
qb_rb_close(c->u.shm.request.rb);
qb_rb_close(c->u.shm.response.rb);
qb_rb_close(c->u.shm.dispatch.rb);
}
static int32_t qb_ipcc_shm_send(struct qb_ipcc_connection *c,
const void *msg_ptr, size_t msg_len)
{
return qb_rb_chunk_write(c->u.shm.request.rb, msg_ptr, msg_len);
}
static ssize_t qb_ipcc_shm_recv(struct qb_ipcc_connection *c,
const void *msg_ptr, size_t msg_len)
{
int32_t res = qb_rb_chunk_read(c->u.shm.response.rb, (void*)msg_ptr, msg_len, 0);
if (res == -1 && errno == ETIMEDOUT) {
errno = EAGAIN;
}
return res;
}
static int32_t _ipcc_shm_connect_to_service_(struct qb_ipcc_connection *c)
{
int32_t res;
ssize_t size;
struct mar_req_shm_setup start;
struct mar_res_setup *msg_res;
start.hdr.id = QB_IPC_MSG_CONNECT;
start.hdr.session_id = c->session_id;
start.pid = getpid();
start.hdr.size = sizeof(struct mar_req_shm_setup);
strcpy(start.response, qb_rb_name_get(c->u.shm.response.rb));
strcpy(start.dispatch, qb_rb_name_get(c->u.shm.dispatch.rb));
res = qb_rb_chunk_write(c->u.shm.request.rb, (const char *)&start, start.hdr.size);
if (res == -1) {
res = errno;
perror("mq_send");
return res;
}
if (c->needs_sock_for_poll) {
qb_ipc_us_send(c->sock, &start, 1);
}
printf("sent request to server %d\n", res);
size = qb_rb_chunk_read(c->u.shm.response.rb, c->receive_buf,
c->max_msg_size, 100000);
if (size == -1) {
res = errno;
perror("_ipcc_shm_connect_to_service_:qb_rb_chunk_read");
goto cleanup;
}
printf("received response from server %zd\n", size);
msg_res = (struct mar_res_setup *)c->receive_buf;
res = msg_res->hdr.error;
if (res == 0) {
c->max_msg_size = msg_res->max_msg_size;
}
cleanup:
return res;
}
int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection * c)
{
int32_t res = 0;
c->funcs.send = qb_ipcc_shm_send;
c->funcs.recv = qb_ipcc_shm_recv;
c->funcs.disconnect = qb_ipcc_shm_disconnect;
if (strlen(c->name) > (NAME_MAX - 20)) {
errno = EINVAL;
return -1;
}
/* Connect to the service's request message queue.
*/
c->u.shm.request.rb = qb_rb_open(c->name, c->max_msg_size,
QB_RB_FLAG_SHARED_PROCESS);
if (c->u.shm.request.rb == NULL) {
perror("qb_rb_open:REQUEST");
return -1;
}
/* Create the response message queue.
*/
res = snprintf(c->u.shm.response.name,
NAME_MAX, "%s-response-%d",
c->name, getpid());
c->u.shm.response.rb = qb_rb_open(c->u.shm.response.name,
c->max_msg_size,
QB_RB_FLAG_CREATE | QB_RB_FLAG_SHARED_PROCESS);
if (c->u.shm.response.rb == NULL) {
perror("qb_rb_open:RESPONSE");
goto cleanup_request;
}
res =
snprintf(c->u.shm.dispatch.name, NAME_MAX, "%s-dispatch-%d",
c->name, getpid());
c->u.shm.dispatch.rb = qb_rb_open(c->u.shm.dispatch.name,
c->max_msg_size,
QB_RB_FLAG_CREATE | QB_RB_FLAG_SHARED_PROCESS);
if (c->u.shm.dispatch.rb == NULL) {
perror("qb_rb_open:DISPATCH");
goto cleanup_request_response;
}
res = _ipcc_shm_connect_to_service_(c);
if (res == 0) {
return 0;
}
printf("%s:%d\n", __FILE__, __LINE__);
qb_rb_close(c->u.shm.dispatch.rb);
cleanup_request_response:
qb_rb_close(c->u.shm.response.rb);
cleanup_request:
qb_rb_close(c->u.shm.request.rb);
return -1;
}
/*
* service functions
* --------------------------------------------------------
*/
static void qb_ipcs_shm_disconnect(struct qb_ipcs_connection *c)
{
struct qb_ipc_response_header msg;
msg.id = QB_IPC_MSG_DISCONNECT;
msg.size = sizeof(msg);
msg.error = 0;
qb_ipcs_shm_dispatch_send(c, &msg, msg.size);
if (c->u.shm.response.rb)
qb_rb_close(c->u.shm.response.rb);
if (c->u.shm.dispatch.rb)
qb_rb_close(c->u.shm.dispatch.rb);
}
static void qb_ipcs_shm_destroy(struct qb_ipcs_service *s)
{
struct qb_ipcs_connection *c = NULL;
struct qb_list_head *iter;
struct qb_list_head *iter_next;
printf("%s\n", __func__);
for (iter = s->connections.next;
iter != &s->connections; iter = iter_next) {
iter_next = iter->next;
c = qb_list_entry(iter, struct qb_ipcs_connection, list);
if (c == NULL) {
continue;
}
qb_ipcs_disconnect(c);
}
qb_rb_close(s->u.rb);
}
static int32_t qb_ipcs_shm_connect(struct qb_ipcs_service *s,
struct qb_ipcs_connection *c, void *data,
size_t size)
{
int32_t res;
struct mar_req_shm_setup *init = (struct mar_req_shm_setup *)data;
struct mar_res_setup accept_msg;
c->pid = init->pid;
c->service = s;
printf("connecting to client [%d]\n", c->pid);
/* setup the response message queue
*/
strcpy(c->u.shm.response.name, init->response);
qb_util_log(LOG_DEBUG, "%s:%s", __func__, c->u.shm.response.name);
c->u.shm.response.rb = qb_rb_open(c->u.shm.response.name,
s->max_msg_size,
QB_RB_FLAG_SHARED_PROCESS);
if (c->u.shm.response.rb == NULL) {
res = errno;
perror("qb_rb_open:RESPONSE");
return res;
}
/* setup the dispatch message queue
*/
strcpy(c->u.shm.dispatch.name, init->dispatch);
qb_util_log(LOG_DEBUG, "%s:%s", __func__, c->u.shm.dispatch.name);
c->u.shm.dispatch.rb = qb_rb_open(c->u.shm.dispatch.name,
s->max_msg_size,
QB_RB_FLAG_SHARED_PROCESS);
if (c->u.shm.dispatch.rb == NULL) {
res = errno;
perror("mq_open:DISPATCH");
goto cleanup_response;
}
/* send the "connection accepted" message back.
*/
accept_msg.hdr.id = QB_IPC_MSG_CONNECT;
accept_msg.hdr.size = sizeof(struct mar_res_setup);
accept_msg.hdr.error = 0;
accept_msg.max_msg_size = s->max_msg_size;
qb_util_log(LOG_DEBUG, "%s:sending response", __func__);
res = qb_rb_chunk_write(c->u.shm.response.rb, (const char *)&accept_msg,
sizeof(struct mar_res_setup));
if (res == -1) {
res = errno;
perror("qb_rb_chunk_write:RESPONSE");
goto cleanup_response;
}
return 0;
cleanup_response:
accept_msg.hdr.error = res;
qb_rb_chunk_write(c->u.shm.response.rb, (const char *)&accept_msg,
sizeof(struct mar_res_setup));
qb_rb_close(c->u.shm.response.rb);
return res;
}
static ssize_t qb_ipcs_shm_request_recv(struct qb_ipcs_service *s, void *buf,
size_t buf_size)
{
int32_t res = qb_rb_chunk_read(s->u.rb, buf, buf_size, 0);
if (res == -1 && errno == ETIMEDOUT) {
errno = EAGAIN;
}
return res;
}
static ssize_t qb_ipcs_shm_response_send(struct qb_ipcs_connection *c,
void *data, size_t size)
{
return qb_rb_chunk_write(c->u.shm.response.rb, (const char *)data, size);
}
static ssize_t qb_ipcs_shm_dispatch_send(struct qb_ipcs_connection *c,
void *data, size_t size)
{
return qb_rb_chunk_write(c->u.shm.dispatch.rb, (const char *)data, size);
}
int32_t qb_ipcs_shm_create(struct qb_ipcs_service *s)
{
s->funcs.destroy = qb_ipcs_shm_destroy;
s->funcs.request_recv = qb_ipcs_shm_request_recv;
s->funcs.response_send = qb_ipcs_shm_response_send;
s->funcs.connect = qb_ipcs_shm_connect;
s->funcs.disconnect = qb_ipcs_shm_disconnect;
s->u.rb = qb_rb_open(s->name, s->max_msg_size,
QB_RB_FLAG_CREATE | QB_RB_FLAG_SHARED_PROCESS);
if (s->u.rb == NULL) {
perror("qb_rb_open:REQUEST");
return -1;
}
qb_util_log(LOG_DEBUG, "%s() %d", __func__, s->u.q);
return 0;
}

416
lib/ipc_sysv_mq.c Normal file
View File

@ -0,0 +1,416 @@
/*
* Copyright (C) 2010 Red Hat, Inc.
*
* Author: Angus Salkeld <asalkeld@redhat.com>
*
* This file is part of libqb.
*
* libqb is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation, either version 2.1 of the License, or
* (at your option) any later version.
*
* libqb is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with libqb. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os_base.h"
#include <sys/ipc.h>
#include <sys/msg.h>
#include <qb/qbpoll.h>
#include "ipc_int.h"
#include "util_int.h"
#ifndef MSGMAX
#define MSGMAX 8192
#endif
static ssize_t qb_ipcs_smq_dispatch_send(struct qb_ipcs_connection *c,
void *data, size_t size);
/*
* utility functions
* --------------------------------------------------------
*/
static int32_t sysv_mq_create(struct qb_ipcs_service *s,
struct qb_ipcc_smq_one_way *mq)
{
struct msqid_ds info;
int32_t res = 0;
mq->q = msgget(mq->key, IPC_CREAT | IPC_NOWAIT);
if (mq->q == -1) {
perror("msgget:REQUEST");
return -1;
}
res = msgctl(s->u.smq.q, IPC_STAT, &info);
if (res != 0) {
perror("msgctl");
qb_util_log(LOG_ERR, "error getting mq info");
}
info.msg_qbytes = 10 * MSGMAX;
res = msgctl(s->u.smq.q, IPC_SET, &info);
if (res != 0) {
perror("msgctl");
qb_util_log(LOG_ERR, "error changing msg_qbytes to %d",
10 * MSGMAX);
}
return 0;
}
static int32_t sysv_mq_unnamed_create(struct qb_ipcc_smq_one_way *queue)
{
retry_creating_the_q:
queue->key = random();
queue->q = msgget(queue->key, IPC_CREAT | IPC_EXCL | IPC_NOWAIT);
if (queue->q == -1 && errno == EEXIST) {
goto retry_creating_the_q;
} else if (queue->q == -1) {
return -1;
}
return 0;
}
static key_t sysv_key_from_name(const char *name)
{
char key_location[PATH_MAX];
snprintf(key_location, PATH_MAX, "/tmp/qb_%s.smq", name);
return ftok(key_location, 0);
}
/*
* client functions
* --------------------------------------------------------
*/
static int32_t qb_ipcc_smq_send(struct qb_ipcc_connection *c,
const void *msg_ptr, size_t msg_len)
{
//printf("%s()\n", __func__);
return msgsnd(c->u.smq.request.q, msg_ptr, msg_len, 0);
}
static ssize_t qb_ipcc_smq_recv(struct qb_ipcc_connection *c,
const void *msg_ptr, size_t msg_len)
{
int32_t res;
res = msgrcv(c->u.smq.response.q, (char *)msg_ptr,
c->max_msg_size, 0, IPC_NOWAIT);
//printf("%s() %d\n", __func__, res);
if (res == -1 && errno == ENOMSG) {
/* just to be consistent with other IPC types.
*/
errno = EAGAIN;
}
return res;
}
static void qb_ipcc_smq_disconnect(struct qb_ipcc_connection *c)
{
struct qb_ipc_request_header hdr;
printf("%s()\n", __func__);
if (c->needs_sock_for_poll) {
return;
}
hdr.id = QB_IPC_MSG_DISCONNECT;
hdr.session_id = c->session_id;
hdr.size = sizeof(hdr);
msgsnd(c->u.smq.request.q, (const char *)&hdr, hdr.size, 0);
msgctl(c->u.smq.dispatch.q, IPC_RMID, NULL);
msgctl(c->u.smq.response.q, IPC_RMID, NULL);
}
static int32_t _smq_connect_to_service_(struct qb_ipcc_connection *c)
{
int32_t res;
ssize_t size;
struct mar_req_smq_setup start;
struct mar_res_setup *msg_res;
start.hdr.id = QB_IPC_MSG_CONNECT;
start.hdr.session_id = c->session_id;
start.pid = getpid();
start.hdr.size = sizeof(struct mar_req_smq_setup);
start.response_key = c->u.smq.response.key;
start.dispatch_key = c->u.smq.dispatch.key;
if (c->needs_sock_for_poll) {
qb_ipc_us_send(c->sock, &start, 1);
}
res = msgsnd(c->u.smq.request.q, (const char *)&start,
start.hdr.size, 0);
if (res == -1) {
res = errno;
perror("msgsnd");
return res;
}
printf("sent request to server %d\n", res);
mq_recv_again:
size = msgrcv(c->u.smq.response.q, c->receive_buf,
c->max_msg_size, 0, IPC_NOWAIT);
if (size == -1 && (errno == EAGAIN || errno == ENOMSG)) {
usleep(100000);
goto mq_recv_again;
}
if (size == -1) {
res = errno;
perror("msgrcv");
goto cleanup;
}
printf("received response from server %zd\n", size);
msg_res = (struct mar_res_setup *)c->receive_buf;
res = msg_res->hdr.error;
if (res == 0) {
c->max_msg_size = msg_res->max_msg_size;
}
cleanup:
return res;
}
int32_t qb_ipcc_smq_connect(struct qb_ipcc_connection * c)
{
int32_t res;
c->funcs.send = qb_ipcc_smq_send;
c->funcs.recv = qb_ipcc_smq_recv;
c->funcs.disconnect = qb_ipcc_smq_disconnect;
c->type = QB_IPC_SYSV_MQ;
if (strlen(c->name) > (NAME_MAX - 20)) {
free(c);
errno = EINVAL;
return -1;
}
/* Connect to the service's request message queue.
*/
c->u.smq.request.key = sysv_key_from_name(c->name);
if (c->u.smq.request.key == -1) {
perror("ftok:REQUEST");
free(c);
return -1;
}
c->u.smq.request.q = msgget(c->u.smq.request.key, IPC_NOWAIT);
if (c->u.smq.request.q == -1) {
perror("msgget:REQUEST");
free(c);
return -1;
}
/* Create the response message queue.
*/
if (sysv_mq_unnamed_create(&c->u.smq.response) == -1) {
perror("msgget:RESPONSE");
goto cleanup_request;
}
/* Create the dispatch message queue.
*/
if (sysv_mq_unnamed_create(&c->u.smq.dispatch) == -1) {
perror("msgget:DISPATCH");
goto cleanup_request_response;
}
res = _smq_connect_to_service_(c);
if (res == 0) {
return 0;
}
printf("%s:%d\n", __FILE__, __LINE__);
msgctl(c->u.smq.dispatch.q, IPC_RMID, NULL);
cleanup_request_response:
msgctl(c->u.smq.response.q, IPC_RMID, NULL);
cleanup_request:
free(c);
return -1;
}
/*
* service functions
* --------------------------------------------------------
*/
static void qb_ipcs_smq_disconnect(struct qb_ipcs_connection *c)
{
struct qb_ipc_response_header msg;
if (c->service->needs_sock_for_poll) {
return;
}
msg.id = QB_IPC_MSG_DISCONNECT;
msg.size = sizeof(msg);
msg.error = 0;
qb_ipcs_smq_dispatch_send(c, &msg, msg.size);
}
static void qb_ipcs_smq_destroy(struct qb_ipcs_service *s)
{
struct qb_ipcs_connection *c = NULL;
struct qb_list_head *iter;
struct qb_list_head *iter_next;
printf("%s\n", __func__);
for (iter = s->connections.next;
iter != &s->connections; iter = iter_next) {
iter_next = iter->next;
c = qb_list_entry(iter, struct qb_ipcs_connection, list);
if (c == NULL) {
continue;
}
qb_ipcs_disconnect(c);
}
if (msgctl(s->u.smq.q, IPC_RMID, NULL) == -1)
perror("msgctl:RMID");
}
static int32_t qb_ipcs_smq_connect(struct qb_ipcs_service *s,
struct qb_ipcs_connection *c, void *data,
size_t size)
{
int32_t res;
struct mar_req_smq_setup *init = (struct mar_req_smq_setup *)data;
struct mar_res_setup accept_msg;
c->pid = init->pid;
c->service = s;
accept_msg.hdr.id = QB_IPC_MSG_CONNECT;
accept_msg.hdr.size = sizeof(struct mar_res_setup);
/* setup the response message queue
*/
c->u.smq.response.key = init->response_key;
c->u.smq.response.q = msgget(c->u.smq.response.key, IPC_NOWAIT);
if (c->u.smq.response.q == -1) {
res = errno;
perror("msgget:RESPONSE");
goto cleanup;
}
/* setup the dispatch message queue
*/
c->u.smq.dispatch.key = init->dispatch_key;
c->u.smq.dispatch.q = msgget(c->u.smq.dispatch.key, IPC_NOWAIT);
if (c->u.smq.dispatch.q == -1) {
res = errno;
perror("msgget:DISPATCH");
goto cleanup_response;
}
/* send the "connection accepted" message back.
*/
accept_msg.hdr.error = 0;
accept_msg.max_msg_size = s->max_msg_size;
res = msgsnd(c->u.smq.response.q, (const char *)&accept_msg,
sizeof(struct mar_res_setup), 0);
if (res == -1) {
res = errno;
perror("msgsnd:RESPONSE");
goto cleanup_response;
}
return 0;
cleanup_response:
accept_msg.hdr.error = res;
msgsnd(c->u.smq.response.q, (const char *)&accept_msg,
sizeof(struct mar_res_setup), 0);
cleanup:
free(c);
return res;
}
#if 0
static int32_t qb_ipcs_smq_is_msg_ready(struct qb_ipcs_service *s)
{
struct msqid_ds info;
int32_t res = msgctl(s->u.smq.q, IPC_STAT, &info);
if (res == 0) {
return info.msg_qnum;
} else {
perror("is_msg_ready");
}
return -1;
}
#endif
static ssize_t qb_ipcs_smq_request_recv(struct qb_ipcs_service *s, void *buf,
size_t buf_size)
{
ssize_t res = msgrcv(s->u.q, buf, buf_size, 0, 0);
//qb_util_log(LOG_INFO, "%s() %d", __func__, res);
if (res == -1 && errno == ENOMSG) {
return 0;
}
return res;
}
static ssize_t qb_ipcs_smq_response_send(struct qb_ipcs_connection *c,
void *data, size_t size)
{
//qb_util_log(LOG_INFO, "%s()", __func__);
return msgsnd(c->u.smq.response.q, (const char *)data, size, 0);
}
static ssize_t qb_ipcs_smq_dispatch_send(struct qb_ipcs_connection *c,
void *data, size_t size)
{
return msgsnd(c->u.smq.dispatch.q, (const char *)data, size, 0);
}
int32_t qb_ipcs_smq_create(struct qb_ipcs_service * s)
{
int32_t fd;
char key_location[PATH_MAX];
snprintf(key_location, PATH_MAX, "/tmp/qb_%s.smq", s->name);
fd = creat(key_location, 0600);
s->u.smq.key = ftok(key_location, 0);
s->funcs.destroy = qb_ipcs_smq_destroy;
s->funcs.connect = qb_ipcs_smq_connect;
s->funcs.disconnect = qb_ipcs_smq_disconnect;
s->funcs.response_send = qb_ipcs_smq_response_send;
s->funcs.request_recv = qb_ipcs_smq_request_recv;
s->max_msg_size = MSGMAX;
return sysv_mq_create(s, &s->u.smq);
}

568
lib/ipc_us.c Normal file
View File

@ -0,0 +1,568 @@
/*
* Copyright (C) 2010 Red Hat, Inc.
*
* Author: Angus Salkeld <asalkeld@redhat.com>
*
* This file is part of libqb.
*
* libqb is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation, either version 2.1 of the License, or
* (at your option) any later version.
*
* libqb is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with libqb. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os_base.h"
#if defined(HAVE_GETPEERUCRED)
#include <ucred.h>
#endif
#ifdef HAVE_SYS_UN_H
#include <sys/un.h>
#endif /* HAVE_SYS_UN_H */
#include <qb/qbipcs.h>
#include <qb/qbpoll.h>
#include "util_int.h"
#include "ipc_int.h"
#define SERVER_BACKLOG 5
#if defined(QB_LINUX) || defined(QB_SOLARIS)
#define QB_SUN_LEN(a) sizeof(*(a))
#else
#define QB_SUN_LEN(a) SUN_LEN(a)
#endif
static int32_t qb_ipcs_us_connection_acceptor(qb_handle_t handle,
int fd, int revent, void *data);
#ifdef SO_NOSIGPIPE
static void socket_nosigpipe(int32_t s)
{
int32_t on = 1;
setsockopt(s, SOL_SOCKET, SO_NOSIGPIPE, (void *)&on, sizeof(on));
}
#endif
#ifndef MSG_NOSIGNAL
#define MSG_NOSIGNAL 0
#endif
int32_t qb_ipc_us_send(int32_t s, const void *msg, size_t len)
{
int32_t result;
struct msghdr msg_send;
struct iovec iov_send;
char *rbuf = (char *)msg;
int processed = 0;
msg_send.msg_iov = &iov_send;
msg_send.msg_iovlen = 1;
msg_send.msg_name = 0;
msg_send.msg_namelen = 0;
#if !defined(QB_SOLARIS)
msg_send.msg_control = 0;
msg_send.msg_controllen = 0;
msg_send.msg_flags = 0;
#else
msg_send.msg_accrights = NULL;
msg_send.msg_accrightslen = 0;
#endif
retry_send:
iov_send.iov_base = &rbuf[processed];
iov_send.iov_len = len - processed;
result = sendmsg(s, &msg_send, MSG_NOSIGNAL);
if (result == -1 && errno == EAGAIN) {
goto retry_send;
}
if (result == -1) {
return errno;
}
processed += result;
if (processed != len) {
goto retry_send;
}
return 0;
}
static int32_t qb_ipc_us_recv_msghdr(int32_t s,
struct msghdr *hdr, const char *msg,
size_t len)
{
int result;
int processed = 0;
retry_recv:
hdr->msg_iov->iov_base = (void *)&msg[processed];
hdr->msg_iov->iov_len = len - processed;
result = recvmsg(s, hdr, MSG_NOSIGNAL | MSG_WAITALL);
if (result == -1 && errno == EAGAIN) {
goto retry_recv;
}
if (result == -1) {
return errno;
}
#if defined(QB_SOLARIS) || defined(QB_BSD) || defined(QB_DARWIN)
/* On many OS poll never return POLLHUP or POLLERR.
* EOF is detected when recvmsg return 0.
*/
if (result == 0) {
return errno; //ENOTCONN
}
#endif
processed += result;
if (processed != len) {
goto retry_recv;
}
assert(processed == len);
return 0;
}
int32_t qb_ipc_us_recv(int32_t s, void *msg, size_t len)
{
struct msghdr msg_recv;
struct iovec iov_recv;
msg_recv.msg_iov = &iov_recv;
msg_recv.msg_iovlen = 1;
msg_recv.msg_name = 0;
msg_recv.msg_namelen = 0;
#if !defined (QB_SOLARIS)
msg_recv.msg_control = 0;
msg_recv.msg_controllen = 0;
msg_recv.msg_flags = 0;
#else
msg_recv.msg_accrights = NULL;
msg_recv.msg_accrightslen = 0;
#endif
return qb_ipc_us_recv_msghdr(s, &msg_recv, msg, len);
}
static int32_t qb_ipcs_uc_recv_and_auth(struct qb_ipcs_connection *c)
{
int32_t res = 0;
int32_t recv_res = 0;
struct msghdr msg_recv;
struct iovec iov_recv;
int32_t authenticated = QB_FALSE;
char setup_msg[sizeof(struct mar_req_initial_setup)];
#ifdef QB_LINUX
struct cmsghdr *cmsg;
char cmsg_cred[CMSG_SPACE(sizeof(struct ucred))];
int off = 0;
int on = 1;
struct ucred *cred;
#endif
msg_recv.msg_flags = 0;
msg_recv.msg_iov = &iov_recv;
msg_recv.msg_iovlen = 1;
msg_recv.msg_name = 0;
msg_recv.msg_namelen = 0;
#ifdef QB_LINUX
msg_recv.msg_control = (void *)cmsg_cred;
msg_recv.msg_controllen = sizeof(cmsg_cred);
#endif
#ifdef QB_SOLARIS
msg_recv.msg_accrights = 0;
msg_recv.msg_accrightslen = 0;
#endif /* QB_SOLARIS */
iov_recv.iov_base = &setup_msg;
iov_recv.iov_len = sizeof(struct mar_req_initial_setup);
#ifdef QB_LINUX
setsockopt(c->sock, SOL_SOCKET, SO_PASSCRED, &on, sizeof(on));
#endif
recv_res = qb_ipc_us_recv_msghdr(c->sock, &msg_recv, setup_msg,
sizeof(struct mar_req_initial_setup));
if (recv_res != 0) {
authenticated = QB_FALSE;
res = recv_res;
goto cleanup_and_return;
}
res = -1;
/*
* currently support getpeerucred, getpeereid, and SO_PASSCRED credential
* retrieval mechanisms for various Platforms
*/
#ifdef HAVE_GETPEERUCRED
/*
* Solaris and some BSD systems
*/
{
ucred_t *uc = NULL;
if (getpeerucred(c->sock, &uc) == 0) {
res = 0;
c->euid = ucred_geteuid(uc);
c->egid = ucred_getegid(uc);
c->pid = ucred_getpid(uc);
ucred_free(uc);
}
}
#elif HAVE_GETPEEREID
/*
* Usually MacOSX systems
*/
{
/*
* TODO get the peer's pid.
* c->pid = ?;
*/
res = getpeereid(c->sock, &c->euid, &c->egid);
}
#elif SO_PASSCRED
/*
* Usually Linux systems
*/
cmsg = CMSG_FIRSTHDR(&msg_recv);
assert(cmsg);
cred = (struct ucred *)CMSG_DATA(cmsg);
if (cred) {
res = 0;
c->pid = cred->pid;
c->euid = cred->uid;
c->egid = cred->gid;
}
#else /* no credentials */
authenticated = QB_TRUE;
#endif /* no credentials */
cleanup_and_return:
#ifdef QB_LINUX
setsockopt(c->sock, SOL_SOCKET, SO_PASSCRED, &off, sizeof(off));
#endif
if (res == 0) {
if (c->service->serv_fns.connection_authenticate &&
c->service->serv_fns.connection_authenticate(c->handle,
c->euid,
c->egid)) {
authenticated = QB_TRUE;
} else if (c->service->serv_fns.connection_authenticate == NULL) {
authenticated = QB_TRUE;
}
}
if (!authenticated) {
return -1;
}
return 1;
}
int32_t qb_ipcc_us_connect(const char *socket_name, int32_t * sock_pt)
{
int32_t request_fd;
struct sockaddr_un address;
#if defined(QB_SOLARIS)
request_fd = socket(PF_UNIX, SOCK_STREAM, 0);
#else
request_fd = socket(PF_LOCAL, SOCK_STREAM, 0);
#endif
if (request_fd == -1) {
return errno;
}
#ifdef SO_NOSIGPIPE
socket_nosigpipe(request_fd);
#endif
memset(&address, 0, sizeof(struct sockaddr_un));
address.sun_family = AF_UNIX;
#if defined(QB_BSD) || defined(QB_DARWIN)
address.sun_len = SUN_LEN(&address);
#endif
#if defined(QB_LINUX)
sprintf(address.sun_path + 1, "%s", socket_name);
#else
sprintf(address.sun_path, "%s/%s", SOCKETDIR, socket_name);
#endif
if (connect(request_fd, (struct sockaddr *)&address,
QB_SUN_LEN(&address)) == -1) {
goto error_connect;
}
*sock_pt = request_fd;
return 0;
error_connect:
close(request_fd);
*sock_pt = -1;
return errno;
}
void qb_ipcc_us_disconnect(int32_t sock)
{
shutdown(sock, SHUT_RDWR);
close(sock);
}
#if 0
cs_error_t coroipcc_dispatch_get(hdb_handle_t handle, void **data, int timeout)
{
struct pollfd ufds;
int poll_events;
char buf;
struct ipc_instance *ipc_instance;
char *data_addr;
cs_error_t error = CS_OK;
int res;
error =
hdb_error_to_cs(hdb_handle_get
(&ipc_hdb, handle, (void **)&ipc_instance));
if (error != CS_OK) {
return (error);
}
*data = NULL;
ufds.fd = ipc_instance->fd;
ufds.events = POLLIN;
ufds.revents = 0;
poll_events = poll(&ufds, 1, timeout);
if (poll_events == -1 && errno == EINTR) {
error = CS_ERR_TRY_AGAIN;
goto error_put;
} else if (poll_events == -1) {
error = CS_ERR_LIBRARY;
goto error_put;
} else if (poll_events == 0) {
error = CS_ERR_TRY_AGAIN;
goto error_put;
}
if (poll_events == 1 && (ufds.revents & (POLLERR | POLLHUP))) {
error = CS_ERR_LIBRARY;
goto error_put;
}
error = socket_recv(ipc_instance->fd, &buf, 1);
assert(error == CS_OK);
if (shared_mem_dispatch_bytes_left(ipc_instance) > 500000) {
/*
* Notify coroipcs to flush any pending dispatch messages
*/
res =
ipc_sem_post(ipc_instance->control_buffer,
SEMAPHORE_REQUEST_OR_FLUSH_OR_EXIT);
if (res != CS_OK) {
error = CS_ERR_LIBRARY;
goto error_put;
}
}
data_addr = ipc_instance->dispatch_buffer;
data_addr = &data_addr[ipc_instance->control_buffer->read];
*data = (void *)data_addr;
return (CS_OK);
error_put:
hdb_handle_put(&ipc_hdb, handle);
return (error);
}
#endif
/*
**************************************************************************
* SERVER
*/
int32_t qb_ipcs_us_publish(struct qb_ipcs_service * s)
{
struct sockaddr_un un_addr;
int32_t res;
/*
* Create socket for IPC clients, name socket, listen for connections
*/
#if defined(QB_SOLARIS)
s->server_sock = socket(PF_UNIX, SOCK_STREAM, 0);
#else
s->server_sock = socket(PF_LOCAL, SOCK_STREAM, 0);
#endif
if (s->server_sock == -1) {
char error_str[100];
strerror_r(errno, error_str, 100);
qb_util_log(LOG_ERR,
"Cannot create server socket: %s\n", error_str);
return -1;
}
res = fcntl(s->server_sock, F_SETFL, O_NONBLOCK);
if (res == -1) {
char error_str[100];
strerror_r(errno, error_str, 100);
qb_util_log(LOG_CRIT,
"Could not set non-blocking operation on server socket: %s\n",
error_str);
goto error_close;
}
memset(&un_addr, 0, sizeof(struct sockaddr_un));
un_addr.sun_family = AF_UNIX;
#if defined(QB_BSD) || defined(QB_DARWIN)
un_addr.sun_len = SUN_LEN(&un_addr);
#endif
#if defined(QB_LINUX)
sprintf(un_addr.sun_path + 1, "%s", s->name);
#else
{
struct stat stat_out;
res = stat(SOCKETDIR, &stat_out);
if (res == -1 || (res == 0 && !S_ISDIR(stat_out.st_mode))) {
qb_util_log(LOG_CRIT,
"Required directory not present %s\n",
SOCKETDIR);
goto error_close;
}
sprintf(un_addr.sun_path, "%s/%s", SOCKETDIR, name);
unlink(un_addr.sun_path);
}
#endif
res =
bind(s->server_sock, (struct sockaddr *)&un_addr,
QB_SUN_LEN(&un_addr));
if (res) {
char error_str[100];
strerror_r(errno, error_str, 100);
qb_util_log(LOG_CRIT,
"Could not bind AF_UNIX (%s): %s.\n",
un_addr.sun_path, error_str);
goto error_close;
}
/*
* Allow eveyrone to write to the socket since the IPC layer handles
* security automatically
*/
#if !defined(QB_LINUX)
res = chmod(un_addr.sun_path, S_IRWXU | S_IRWXG | S_IRWXO);
#endif
listen(s->server_sock, SERVER_BACKLOG);
qb_poll_dispatch_add(s->poll_handle, s->server_sock,
POLLIN | POLLPRI | POLLNVAL,
s, qb_ipcs_us_connection_acceptor);
return 0;
error_close:
close(s->server_sock);
return -1;
}
int32_t qb_ipcs_us_withdraw(struct qb_ipcs_service * s)
{
shutdown(s->server_sock, SHUT_RDWR);
close(s->server_sock);
return 0;
}
static int32_t qb_ipcs_us_connection_acceptor(qb_handle_t handle,
int fd, int revent, void *data)
{
struct sockaddr_un un_addr;
int32_t new_fd;
struct qb_ipcs_connection *c;
struct qb_ipcs_service *s = (struct qb_ipcs_service *)data;
struct mar_res_initial_setup init_res;
int32_t res;
socklen_t addrlen = sizeof(struct sockaddr_un);
retry_accept:
new_fd = accept(fd, (struct sockaddr *)&un_addr, &addrlen);
if (new_fd == -1 && errno == EINTR) {
goto retry_accept;
}
if (new_fd == -1) {
char error_str[100];
strerror_r(errno, error_str, 100);
qb_util_log(LOG_ERR,
"Could not accept Library connection: %s\n",
error_str);
return (0); /* This is an error, but -1 would indicate disconnect from poll loop */
}
res = fcntl(new_fd, F_SETFL, O_NONBLOCK);
if (res == -1) {
char error_str[100];
strerror_r(errno, error_str, 100);
qb_util_log(LOG_ERR,
"Could not set non-blocking operation on library connection: %s\n",
error_str);
close(new_fd);
return (0); /* This is an error, but -1 would indicate disconnect from poll loop */
}
/*
* Valid accept
*/
c = qb_ipcs_connection_alloc(s);
c->sock = new_fd;
if (qb_ipcs_uc_recv_and_auth(c) == 1) {
qb_util_log(LOG_INFO, "IPC credentials authenticated");
qb_list_add(&c->list, &s->connections);
init_res.hdr.id = QB_IPC_MSG_AUTHENTICATE;
init_res.hdr.size = sizeof(init_res);
init_res.hdr.error = 0;
init_res.connection_type = s->type;
init_res.session_id = c->handle;
init_res.max_msg_size = s->max_msg_size;
qb_ipc_us_send(c->sock, &init_res, init_res.hdr.size);
if (s->needs_sock_for_poll) {
qb_poll_dispatch_add(s->poll_handle, c->sock,
POLLIN | POLLPRI | POLLNVAL,
c, qb_ipcs_dispatch_connection_request);
}
} else {
qb_util_log(LOG_ERR, "Invalid IPC credentials.");
init_res.hdr.id = QB_IPC_MSG_AUTHENTICATE;
init_res.hdr.size = sizeof(init_res);
init_res.hdr.error = EACCES;
init_res.session_id = 0;
qb_ipc_us_send(c->sock, &init_res, init_res.hdr.size);
qb_ipcs_disconnect(c);
}
return (0);
}

1166
lib/ipcc.c
View File

File diff suppressed because it is too large Load Diff

1841
lib/ipcs.c
View File

File diff suppressed because it is too large Load Diff

11
lib/poll.c
View File

@ -393,8 +393,7 @@ static int32_t _qb_poll_job_run(struct qb_poll_instance *poll_instance)
{
struct qb_poll_job *job = NULL;
struct qb_list_head *iter;
int32_t this_job_executed;
int32_t job_executed = QB_FALSE;
size_t jobs_run = 0;
for (iter = poll_instance->job_list.next;
iter != &poll_instance->job_list;
@ -403,12 +402,12 @@ static int32_t _qb_poll_job_run(struct qb_poll_instance *poll_instance)
if (job == NULL) {
continue;
}
this_job_executed = job->execute_fn(job->data);
if (this_job_executed > 0) {
job_executed = QB_TRUE;
jobs_run += job->execute_fn(job->data);
if (jobs_run > 10) {
break;
}
}
return job_executed;
return (jobs_run > 0);
}
int32_t qb_poll_run(qb_handle_t handle)

110
tests/bmc.c
View File

@ -18,22 +18,25 @@
* You should have received a copy of the GNU Lesser General Public License
* along with libqb. If not, see <http://www.gnu.org/licenses/>.
*/
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <qb/qbipcc.h>
#include <errno.h>
#include <assert.h>
#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <signal.h>
#define ITERATIONS 10000
pid_t mypid;
int32_t blocking = 1;
int32_t verbose = 0;
#define ITERATIONS 10000
static qb_ipcc_connection_t *conn;
#define MAX_MSG_SIZE (8192*128)
static struct timeval tv1, tv2, tv_elapsed;
@ -47,10 +50,6 @@ do { \
} \
} while (0)
FILE *mbs_fp;
FILE *ops_fp;
static void bm_start(void)
{
gettimeofday(&tv1, NULL);
@ -73,64 +72,54 @@ static void bm_finish(const char *operation, int32_t size)
(((float)tv_elapsed.tv_sec) +
(((float)tv_elapsed.tv_usec) / 1000000.0))) / (1024.0 * 1024.0);
fprintf(ops_fp, "%d %9.3f\n", size, ops_per_sec);
fflush(ops_fp);
fprintf(mbs_fp, "%d %9.3f\n", size, mbs_per_sec);
fflush(mbs_fp);
printf("write size %d OPs/sec %9.3f ", size, ops_per_sec);
printf("MB/sec %9.3f\n", mbs_per_sec);
}
qb_handle_t bmc_ipc_handle;
static void bmc_connect(void)
{
uint32_t res;
res = qb_ipcc_service_connect("qb_ipcs_bm",
0,
8192 * 128,
8192 * 128, 8192 * 128, &bmc_ipc_handle);
}
static char buffer[1024 * 1024];
static void bmc_send_nozc(uint32_t size)
static int32_t bmc_send_nozc(uint32_t size)
{
struct iovec iov[2];
qb_ipc_request_header_t req_header;
qb_ipc_response_header_t res_header;
struct qb_ipc_request_header *req_header = (struct qb_ipc_request_header *)buffer;
struct qb_ipc_response_header res_header;
int32_t res;
req_header.id = 0;
req_header.size = sizeof(qb_ipc_request_header_t) + size;
iov[0].iov_base = &req_header;
iov[0].iov_len = sizeof(qb_ipc_request_header_t);
iov[1].iov_base = buffer;
iov[1].iov_len = size;
req_header->id = QB_IPC_MSG_USER_START + 3;
req_header->size = sizeof(struct qb_ipc_request_header) + size;
repeat_send:
if (blocking) {
res = qb_ipcc_msg_send_reply_receive(bmc_ipc_handle,
iov, 2,
&res_header,
sizeof
(qb_ipc_response_header_t));
} else {
res = qb_ipcc_msg_send(bmc_ipc_handle, iov, 2);
}
res = qb_ipcc_send(conn, req_header, req_header->size);
if (res == -1) {
if (errno == ENOMEM) {
if (errno == EAGAIN || errno == ENOMEM) {
goto repeat_send;
} else if (errno == EINVAL || errno == EINTR) {
perror("qb_ipcc_send");
return -1;
} else {
printf("qb_ipcc_msg_send: %d(%s)\n", res, strerror(res));
perror("qb_ipcc_send");
goto repeat_send;
}
}
if (blocking) {
repeat_recv:
res = qb_ipcc_recv(conn,
&res_header,
sizeof(struct qb_ipc_response_header));
if (res == -1 && errno == EAGAIN) {
goto repeat_recv;
}
if (res == -1 && errno == EINTR) {
return -1;
}
assert(res == sizeof(struct qb_ipc_response_header));
assert(res_header.id == 13);
assert(res_header.size == sizeof(struct qb_ipc_response_header));
}
return 0;
}
qb_ipc_request_header_t *global_zcb_buffer;
struct qb_ipc_request_header *global_zcb_buffer;
static void show_usage(const char *name)
{
@ -147,8 +136,8 @@ static void show_usage(const char *name)
static void sigterm_handler(int32_t num)
{
printf("writer: %s(%d)\n", __func__, num);
qb_ipcc_service_disconnect(bmc_ipc_handle);
printf("bmc: %s(%d)\n", __func__, num);
qb_ipcc_disconnect(conn);
exit(0);
}
@ -166,6 +155,8 @@ int32_t main(int32_t argc, char *argv[])
int32_t i, j;
size_t size;
mypid = getpid();
qb_util_set_log_function(libqb_log_writer);
while ((opt = getopt(argc, argv, options)) != -1) {
@ -187,19 +178,28 @@ int32_t main(int32_t argc, char *argv[])
signal(SIGINT, sigterm_handler);
signal(SIGILL, sigterm_handler);
signal(SIGTERM, sigterm_handler);
bmc_connect();
ops_fp = fopen("opsec", "w");
mbs_fp = fopen("mbsec", "w");
conn = qb_ipcc_connect("bm1", QB_IPC_SHM);
// conn = qb_ipcc_connect("bm1", QB_IPC_POSIX_MQ);
// conn = qb_ipcc_connect("bm1", QB_IPC_SYSV_MQ);
if (conn == NULL) {
perror("qb_ipcc_connect");
exit(1);
}
for (j = 1; j < 49; j++) {
size = 10 * j * j;
size = (10 * j * j * j) + sizeof(struct qb_ipc_request_header);
if (size >= MAX_MSG_SIZE)
break;
bm_start();
for (i = 0; i < ITERATIONS; i++) {
bmc_send_nozc(size);
if (bmc_send_nozc(size) == -1) {
break;
}
}
bm_finish("send_nozc", size);
}
qb_ipcc_service_disconnect(bmc_ipc_handle);
qb_ipcc_disconnect(conn);
return EXIT_SUCCESS;
}

68
tests/bmcpt.c
View File

@ -22,6 +22,8 @@
#include <sys/types.h>
#include <sys/socket.h>
#include <qb/qbipcc.h>
#include <errno.h>
#include <assert.h>
#include <stdio.h>
#include <stdint.h>
#include <string.h>
@ -32,7 +34,7 @@
#define ITERATIONS 10000000
struct bm_ctx {
qb_handle_t bmc_ipc_handle;
qb_ipcc_connection_t *conn;
struct timeval tv1;
struct timeval tv2;
struct timeval tv_elapsed;
@ -80,45 +82,52 @@ static void bm_finish(struct bm_ctx *ctx, const char *operation, int32_t size)
static void bmc_connect(struct bm_ctx *ctx)
{
uint32_t res;
res = qb_ipcc_service_connect("qb_ipcs_bm",
0,
8192 * 128,
8192 * 128,
8192 * 128, &ctx->bmc_ipc_handle);
ctx->conn = qb_ipcc_connect("bm1", QB_IPC_SHM);
}
static void bmc_disconnect(struct bm_ctx *ctx)
{
qb_ipcc_service_disconnect(ctx->bmc_ipc_handle);
qb_ipcc_disconnect(ctx->conn);
}
static char buffer[1024 * 1024];
static void bmc_send_nozc(struct bm_ctx *ctx, uint32_t size)
static int32_t bmc_send_nozc(struct bm_ctx *ctx, uint32_t size)
{
struct iovec iov[2];
qb_ipc_request_header_t req_header;
qb_ipc_response_header_t res_header;
struct qb_ipc_request_header *req_header = (struct qb_ipc_request_header *)buffer;
struct qb_ipc_response_header res_header;
int32_t res;
req_header.id = 0;
req_header.size = sizeof(qb_ipc_request_header_t) + size;
iov[0].iov_base = &req_header;
iov[0].iov_len = sizeof(qb_ipc_request_header_t);
iov[1].iov_base = buffer;
iov[1].iov_len = size;
req_header->id = QB_IPC_MSG_USER_START + 3;
req_header->size = sizeof(struct qb_ipc_request_header) + size;
repeat_send:
res = qb_ipcc_msg_send_reply_receive(ctx->bmc_ipc_handle,
iov,
2,
&res_header,
sizeof(qb_ipc_response_header_t));
if (res != 0) {
goto repeat_send;
res = qb_ipcc_send(ctx->conn, req_header, req_header->size);
if (res == -1) {
if (errno == EAGAIN || errno == ENOMEM) {
goto repeat_send;
} else if (errno == EINVAL || errno == EINTR) {
perror("qb_ipcc_send");
return -1;
} else {
perror("qb_ipcc_send");
goto repeat_send;
}
}
repeat_recv:
res = qb_ipcc_recv(ctx->conn,
&res_header,
sizeof(struct qb_ipc_response_header));
if (res == -1 && errno == EAGAIN) {
goto repeat_recv;
}
if (res == -1 && errno == EINTR) {
return -1;
}
assert(res == sizeof(struct qb_ipc_response_header));
assert(res_header.id == 13);
assert(res_header.size == sizeof(struct qb_ipc_response_header));
return 0;
}
uint32_t alarm_notice = 0;
@ -130,14 +139,15 @@ static void sigalrm_handler(int32_t num)
static void *benchmark(void *ctx)
{
struct bm_ctx *bm_ctx = (struct bm_ctx *)ctx;
int32_t res;
bmc_connect(bm_ctx);
bm_start(bm_ctx);
for (;;) {
bm_ctx->counter++;
bmc_send_nozc(bm_ctx, 1000 * bm_ctx->multi);
if (alarm_notice) {
res = bmc_send_nozc(bm_ctx, 1000 * bm_ctx->multi);
if (alarm_notice || res == -1) {
bm_finish(bm_ctx, "send_nozc", 1000 * bm_ctx->multi);
bmc_disconnect(bm_ctx);
return (NULL);

278
tests/bms.c
View File

@ -50,240 +50,69 @@ int32_t blocking = 1;
int32_t verbose = 0;
static qb_handle_t bms_poll_handle;
static qb_ipcs_service_pt s1;
struct lib_handler {
void (*lib_handler_fn) (void *conn, const void *msg);
};
struct service_engine {
uint32_t private_data_size;
int32_t (*lib_init_fn) (void *conn);
int32_t (*lib_exit_fn) (void *conn);
struct lib_handler *lib_engine;
};
static void bms_benchmark_one_fn(void *conn, const void *msg)
static int32_t s1_connection_authenticate_fn(qb_ipcs_connection_pt conn, uid_t uid, gid_t gid)
{
qb_ipc_response_header_t res;
if (uid == 0 && gid == 0) {
if (verbose) {
printf("%s:%d %s authenticated connection\n",
__FILE__, __LINE__, __func__);
}
return 1;
}
printf("%s:%d %s() BAD user!\n", __FILE__, __LINE__, __func__);
return 0;
}
static void s1_connection_created_fn(qb_ipcs_connection_pt conn)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
res.size = sizeof(qb_ipc_response_header_t);
res.id = 0;
res.error = 0;
}
static void s1_connection_destroyed_fn(qb_ipcs_connection_pt conn)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
}
static void s1_msg_process_fn(qb_ipcs_connection_pt conn,
void *data, size_t size)
{
struct qb_ipc_request_header *req_pt = (struct qb_ipc_request_header *)data;
struct qb_ipc_response_header response;
ssize_t res;
if (verbose > 2) {
printf("%s:%d %s > msg:%d, size:%d\n",
__FILE__, __LINE__, __func__,
req_pt->id, req_pt->size);
}
response.size = sizeof(struct qb_ipc_response_header);
response.id = 13;
response.error = 0;
if (blocking == 1) {
qb_ipcs_response_send(conn, &res, sizeof(res));
res = qb_ipcs_response_send(conn, &response,
sizeof(response));
if (res == -1) {
perror("qb_ipcs_response_send");
}
}
}
int32_t ii = 0;
static void bms_benchmark_two_fn(void *conn, const void *msg)
{
const qb_ipc_request_header_t *req = msg;
const char *req_buf = (char *)msg + sizeof(qb_ipc_request_header_t);
qb_ipc_response_header_t res_done;
qb_ipc_response_header_t res;
struct iovec iovec[2];
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
res.size =
req->size - sizeof(qb_ipc_request_header_t) +
sizeof(qb_ipc_response_header_t);
res.error = 0;
iovec[0].iov_base = &res;
iovec[0].iov_len = sizeof(res);
iovec[1].iov_base = (void *)req_buf;
iovec[1].iov_len = req->size - sizeof(qb_ipc_request_header_t);
qb_ipcs_dispatch_iov_send(conn, iovec, 2);
res_done.size = sizeof(qb_ipc_response_header_t);
res_done.error = 0;
qb_ipcs_response_send(conn, &res_done, sizeof(res_done));
}
static int32_t bms_lib_init_fn(void *conn)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
return (0);
}
static int32_t bms_lib_exit_fn(void *conn)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
return (0);
}
static struct lib_handler bms_lib_engine_one[] = {
{ /* entry 0 */
.lib_handler_fn = bms_benchmark_one_fn,
},
{ /* entry 1 */
.lib_handler_fn = bms_benchmark_two_fn,
}
};
static struct service_engine services[1] = {
{
.private_data_size = 0,
.lib_init_fn = bms_lib_init_fn,
.lib_exit_fn = bms_lib_exit_fn,
.lib_engine = bms_lib_engine_one,
}
};
static void bms_serialize_lock(void)
{
}
static void bms_serialize_unlock(void)
{
}
/*
* Provides the glue from bms to the IPC Service
*/
static int32_t bms_private_data_size_get(uint32_t service)
{
return (services[service].private_data_size);
}
static qb_ipcs_init_fn_lvalue bms_init_fn_get(uint32_t service)
{
return (services[service].lib_init_fn);
}
static qb_ipcs_exit_fn_lvalue bms_exit_fn_get(uint32_t service)
{
return (services[service].lib_exit_fn);
}
static qb_ipcs_handler_fn_lvalue bms_handler_fn_get(uint32_t service,
uint32_t id)
{
return (services[service].lib_engine[id].lib_handler_fn);
}
static int32_t bms_security_valid(int32_t euid, int32_t egid)
{
if (euid == 0 || egid == 0) {
return (1);
}
printf("%s:%d %s NOT VALID!\n", __FILE__, __LINE__, __func__);
return (0);
}
static int32_t bms_service_available(uint32_t service)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
return (service < 1);
}
static int32_t bms_sending_allowed(uint32_t service,
uint32_t id,
const void *msg,
void *sending_allowed_private_data)
{
return (1);
}
static void bms_sending_allowed_release(void *sending_allowed_private_data)
{
}
static void ipc_log_fn(const char *file_name,
int32_t file_line, int32_t severity, const char *msg)
{
fprintf(stderr, "%s:%d [%d] %s\n", file_name, file_line, severity, msg);
}
static void ipc_fatal_error(const char *error_msg)
{
printf("FATAL Error: %s\n", error_msg);
exit(1);
}
static int32_t bms_poll_handler_accept(qb_handle_t handle,
int32_t fd, int32_t revent, void *context)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
return (qb_ipcs_handler_accept(fd, revent, context));
}
static int32_t bms_poll_handler_dispatch(qb_handle_t handle,
int32_t fd, int32_t revent, void *context)
{
return (qb_ipcs_handler_dispatch(fd, revent, context));
}
static void bms_poll_accept_add(int32_t fd)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
qb_poll_dispatch_add(bms_poll_handle, fd, POLLIN | POLLNVAL, 0,
bms_poll_handler_accept);
}
static void bms_poll_dispatch_add(int32_t fd, void *context)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
qb_poll_dispatch_add(bms_poll_handle, fd, POLLIN | POLLNVAL, context,
bms_poll_handler_dispatch);
}
static void bms_poll_dispatch_modify(int32_t fd, int32_t events)
{
if (verbose) {
printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
}
qb_poll_dispatch_modify(bms_poll_handle, fd, events,
bms_poll_handler_dispatch);
}
struct sched_param sched_param = {
.sched_priority = 99,
};
struct qb_ipcs_init_state ipc_init_state = {
.socket_name = "qb_ipcs_bm",
.sched_policy = 0,
.sched_param = NULL,
.malloc = malloc,
.free = free,
.fatal_error = ipc_fatal_error,
.security_valid = bms_security_valid,
.service_available = bms_service_available,
.private_data_size_get = bms_private_data_size_get,
.serialize_lock = bms_serialize_lock,
.serialize_unlock = bms_serialize_unlock,
.sending_allowed = bms_sending_allowed,
.sending_allowed_release = bms_sending_allowed_release,
.poll_accept_add = bms_poll_accept_add,
.poll_dispatch_add = bms_poll_dispatch_add,
.poll_dispatch_modify = bms_poll_dispatch_modify,
.init_fn_get = bms_init_fn_get,
.exit_fn_get = bms_exit_fn_get,
.handler_fn_get = bms_handler_fn_get
};
static void sigusr1_handler(int32_t num)
{
printf("%s(%d)\n", __func__, num);
qb_ipcs_ipc_exit();
qb_ipcs_destroy(s1);
exit(0);
}
@ -304,6 +133,12 @@ int32_t main(int32_t argc, char *argv[])
{
const char *options = "nvh";
int32_t opt;
struct qb_ipcs_service_handlers sh = {
.connection_authenticate = s1_connection_authenticate_fn,
.connection_created = s1_connection_created_fn,
.msg_process = s1_msg_process_fn,
.connection_destroyed = s1_connection_destroyed_fn,
};
while ((opt = getopt(argc, argv, options)) != -1) {
switch (opt) {
@ -311,7 +146,7 @@ int32_t main(int32_t argc, char *argv[])
blocking = 0;
break;
case 'v':
verbose = 1;
verbose++;
break;
case 'h':
default:
@ -328,8 +163,15 @@ int32_t main(int32_t argc, char *argv[])
bms_poll_handle = qb_poll_create();
qb_ipcs_ipc_init(&ipc_init_state);
s1 = qb_ipcs_create("bm1", QB_IPC_SHM, 8192*64);
// s1 = qb_ipcs_create("bm1", QB_IPC_POSIX_MQ, 8192*64);
// s1 = qb_ipcs_create("bm1", QB_IPC_SYSV_MQ, 8192*64);
if (s1 == 0) {
perror("qb_ipcs_create");
exit(1);
}
qb_ipcs_service_handlers_set(s1, &sh);
qb_ipcs_run(s1, bms_poll_handle);
qb_poll_run(bms_poll_handle);
return EXIT_SUCCESS;

14
tests/check_rb.c
View File

@ -32,7 +32,7 @@
START_TEST(test_ring_buffer1)
{
char my_buf[512];
qb_ipc_request_header_t *hdr;
struct qb_ipc_request_header *hdr;
char *str;
qb_ringbuffer_t *rb;
int32_t i;
@ -44,8 +44,8 @@ START_TEST(test_ring_buffer1)
fail_if(rb == NULL);
for (b = 0; b < 3; b++) {
hdr = (qb_ipc_request_header_t *) my_buf;
str = my_buf + sizeof(qb_ipc_request_header_t);
hdr = (struct qb_ipc_request_header *) my_buf;
str = my_buf + sizeof(struct qb_ipc_request_header);
for (i = 0; i < 900; i++) {
hdr->id = __LINE__ + i;
@ -53,7 +53,7 @@ START_TEST(test_ring_buffer1)
sprintf(str, "ID: %d (%s + i(%d)) -- %s-%s!",
hdr->id, "actually the line number", i,
__func__, __FILE__) + 1;
hdr->size += sizeof(qb_ipc_request_header_t);
hdr->size += sizeof(struct qb_ipc_request_header);
avail = qb_rb_space_free(rb);
actual = qb_rb_chunk_write(rb, hdr, hdr->size);
if (avail < (hdr->size + (2 * sizeof(uint32_t)))) {
@ -65,8 +65,8 @@ START_TEST(test_ring_buffer1)
memset(my_buf, 0, sizeof(my_buf));
hdr = (qb_ipc_request_header_t *) my_buf;
str = my_buf + sizeof(qb_ipc_request_header_t);
hdr = (struct qb_ipc_request_header *) my_buf;
str = my_buf + sizeof(struct qb_ipc_request_header);
for (i = 0; i < 15; i++) {
actual = qb_rb_chunk_read(rb, hdr, 512, 0);
@ -74,7 +74,7 @@ START_TEST(test_ring_buffer1)
ck_assert_int_eq(0, qb_rb_chunks_used(rb));
break;
}
str[actual - sizeof(qb_ipc_request_header_t)] = '\0';
str[actual - sizeof(struct qb_ipc_request_header)] = '\0';
ck_assert_int_eq(actual, hdr->size);
}