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

IPC: add support for unix sockets

Browse Source 
Signed-off-by: Angus Salkeld <asalkeld@redhat.com>
This commit is contained in:
Angus Salkeld 2010-10-17 14:23:53 +11:00
parent 8585b958cc
commit 31ca215188
8 changed files with 574 additions and 291 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/qb/qbipc_common.h
View File

@ -45,7 +45,7 @@ enum qb_ipc_type {
#define QB_IPC_MSG_NEW_MESSAGE 0
#define QB_IPC_MSG_USER_START QB_IPC_MSG_NEW_MESSAGE
#define QB_IPC_MSG_AUTHENTICATE -1
#define QB_IPC_MSG_CONNECT -2
#define QB_IPC_MSG_NEW_EVENT_SOCK -2
#define QB_IPC_MSG_DISCONNECT -3

29
lib/ipc_int.h
View File

@ -64,10 +64,16 @@ struct qb_ipc_connection_request {
uint32_t max_msg_size __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
struct qb_ipc_event_connection_request {
struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
intptr_t connection __attribute__ ((aligned(8)));
} __attribute__ ((aligned(8)));
struct qb_ipc_connection_response {
struct qb_ipc_response_header hdr __attribute__ ((aligned(8)));
int32_t connection_type __attribute__ ((aligned(8)));
uint32_t max_msg_size __attribute__ ((aligned(8)));
intptr_t connection __attribute__ ((aligned(8)));
char request[PATH_MAX] __attribute__ ((aligned(8)));
char response[PATH_MAX] __attribute__ ((aligned(8)));
char event[PATH_MAX] __attribute__ ((aligned(8)));
@ -78,6 +84,9 @@ struct qb_ipcc_connection;
struct qb_ipc_one_way {
size_t max_msg_size;
union {
struct {
int32_t sock;
} us;
struct {
mqd_t q;
char name[NAME_MAX];
@ -104,7 +113,7 @@ struct qb_ipcc_connection {
char name[NAME_MAX];
enum qb_ipc_type type;
int32_t needs_sock_for_poll;
int32_t sock;
struct qb_ipc_one_way setup;
struct qb_ipc_one_way request;
struct qb_ipc_one_way response;
struct qb_ipc_one_way event;
@ -112,18 +121,16 @@ struct qb_ipcc_connection {
char *receive_buf;
};
int32_t qb_ipcc_us_setup_connect(struct qb_ipcc_connection *c,
struct qb_ipc_connection_response *r);
ssize_t qb_ipc_us_send(struct qb_ipc_one_way *one_way, const void *msg, size_t len);
ssize_t qb_ipc_us_recv(struct qb_ipc_one_way *one_way, void *msg, size_t len, int32_t timeout);
int32_t qb_ipc_us_recv_ready(struct qb_ipc_one_way *one_way, int32_t ms_timeout);
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_ipc_us_recv_ready(int32_t s, int32_t ms_timeout);
int32_t qb_ipcc_us_connect(const char *socket_name, int32_t *sock_pt);
void qb_ipcc_us_disconnect (int32_t sock);
void qb_ipcc_us_sock_close(int32_t sock);
int32_t qb_ipcc_pmq_connect(struct qb_ipcc_connection *c, struct qb_ipc_connection_response * response);
int32_t qb_ipcc_soc_connect(struct qb_ipcc_connection *c, struct qb_ipc_connection_response * response);
int32_t qb_ipcc_us_connect(struct qb_ipcc_connection *c, struct qb_ipc_connection_response * response);
int32_t qb_ipcc_smq_connect(struct qb_ipcc_connection *c, struct qb_ipc_connection_response * response);
int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection *c, struct qb_ipc_connection_response * response);
@ -164,7 +171,7 @@ struct qb_ipcs_connection {
pid_t pid;
uid_t euid;
gid_t egid;
int32_t sock;
struct qb_ipc_one_way setup;
struct qb_ipc_one_way request;
struct qb_ipc_one_way response;
struct qb_ipc_one_way event;

2
lib/ipc_shm.c
View File

@ -192,7 +192,7 @@ static void qb_ipcs_shm_disconnect(struct qb_ipcs_connection *c)
struct qb_ipc_response_header msg;
int32_t peer_alive = QB_TRUE;
if (c->sock == -1) {
if (c->setup.u.us.sock == -1) {
peer_alive = QB_FALSE;
}

2
lib/ipc_sysv_mq.c
View File

@ -297,7 +297,7 @@ static void qb_ipcs_smq_disconnect(struct qb_ipcs_connection *c)
{
struct qb_ipc_response_header msg;
if (c->sock != -1) {
if (c->setup.u.us.sock != -1) {
msg.id = QB_IPC_MSG_DISCONNECT;
msg.size = sizeof(msg);
msg.error = 0;

628
lib/ipc_us.c
View File

@ -45,6 +45,12 @@
#define QB_SUN_LEN(a) SUN_LEN(a)
#endif
struct ipc_auth_ugp {
uid_t uid;
gid_t gid;
pid_t pid;
};
static int32_t qb_ipcs_us_connection_acceptor(int fd, int revent, void *data);
#ifdef SO_NOSIGPIPE
@ -55,21 +61,48 @@ static void socket_nosigpipe(int32_t s)
}
#endif
static void set_cloexec_flag(int32_t fd)
static int32_t set_cloexec_flag(int32_t fd)
{
int32_t res;
char error_str[100];
int32_t oldflags = fcntl(fd, F_GETFD, 0);
if (oldflags < 0) {
oldflags = 0;
}
oldflags |= FD_CLOEXEC;
fcntl(fd, F_SETFD, oldflags);
res = fcntl(fd, F_SETFD, oldflags);
if (res == -1) {
res = -errno;
strerror_r(errno, error_str, 100);
qb_util_log(LOG_CRIT,
"Could not set close-on-exit operation on socket: %s\n",
error_str);
}
return res;
}
static int32_t set_nonblock_flag(int32_t fd)
{
int32_t res;
char error_str[100];
res = fcntl(fd, F_SETFL, O_NONBLOCK);
if (res == -1) {
res = -errno;
strerror_r(errno, error_str, 100);
qb_util_log(LOG_CRIT,
"Could not set non-blocking operation on socket: %s\n",
error_str);
}
return res;
}
#ifndef MSG_NOSIGNAL
#define MSG_NOSIGNAL 0
#endif
int32_t qb_ipc_us_send(int32_t s, const void *msg, size_t len)
ssize_t qb_ipc_us_send(struct qb_ipc_one_way *one_way, const void *msg, size_t len)
{
int32_t result;
struct msghdr msg_send;
@ -95,10 +128,45 @@ 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) {
result = sendmsg(one_way->u.us.sock, &msg_send, MSG_NOSIGNAL);
if (result == -1) {
return -errno;
}
processed += result;
if (processed != len) {
goto retry_send;
}
return processed;
}
static ssize_t qb_ipc_us_sendv(struct qb_ipc_one_way *one_way, const struct iovec *iov, size_t iov_len)
{
int32_t result;
struct msghdr msg_send;
int32_t processed = 0;
size_t len = 0;
int32_t i;
for (i = 0; i < iov_len; i++) {
len += iov[i].iov_len;
}
msg_send.msg_iov = (struct iovec*)iov;
msg_send.msg_iovlen = iov_len;
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:
result = sendmsg(one_way->u.us.sock, &msg_send, MSG_NOSIGNAL);
if (result == -1) {
return -errno;
}
@ -111,7 +179,8 @@ retry_send:
}
static ssize_t qb_ipc_us_recv_msghdr(int32_t s,
struct msghdr *hdr, char *msg, size_t len)
struct msghdr *hdr,
char *msg, size_t len)
{
int32_t result;
int32_t processed = 0;
@ -146,12 +215,12 @@ retry_recv:
}
int32_t qb_ipc_us_recv_ready(int32_t s, int32_t ms_timeout)
int32_t qb_ipc_us_recv_ready(struct qb_ipc_one_way *one_way, int32_t ms_timeout)
{
struct pollfd ufds;
int32_t poll_events;
ufds.fd = s;
ufds.fd = one_way->u.us.sock;
ufds.events = POLLIN;
ufds.revents = 0;
@ -167,12 +236,13 @@ int32_t qb_ipc_us_recv_ready(int32_t s, int32_t ms_timeout)
return 0;
}
int32_t qb_ipc_us_recv(int32_t s, void *msg, size_t len)
ssize_t qb_ipc_us_recv(struct qb_ipc_one_way *one_way,
void *msg, size_t len, int32_t timeout)
{
int32_t result;
retry_recv:
result = recv(s, msg, len, MSG_NOSIGNAL | MSG_WAITALL);
result = recv(one_way->u.us.sock, msg, len, MSG_NOSIGNAL | MSG_WAITALL);
if (result == -1 && errno == EAGAIN) {
goto retry_recv;
}
@ -191,133 +261,11 @@ int32_t qb_ipc_us_recv(int32_t s, void *msg, size_t len)
}
static int32_t qb_ipcs_uc_recv_and_auth(struct qb_ipcs_connection *c)
{
int32_t res = 0;
struct msghdr msg_recv;
struct iovec iov_recv;
struct qb_ipc_connection_request setup_msg;
#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 qb_ipc_connection_request);
#ifdef QB_LINUX
setsockopt(c->sock, SOL_SOCKET, SO_PASSCRED, &on, sizeof(on));
#endif
res = qb_ipc_us_recv_msghdr(c->sock, &msg_recv, (char *)&setup_msg,
sizeof(struct qb_ipc_connection_request));
if (res < 0) {
goto cleanup_and_return;
}
if (res != sizeof(struct qb_ipc_connection_request)) {
res = -EIO;
goto cleanup_and_return;
}
c->request.max_msg_size = setup_msg.max_msg_size;
c->response.max_msg_size = setup_msg.max_msg_size;
c->event.max_msg_size = setup_msg.max_msg_size;
res = -EBADMSG;
/*
* 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);
} else {
res = -errno;
}
}
#elif HAVE_GETPEEREID
/*
* Usually MacOSX systems
*/
{
/*
* TODO get the peer's pid.
* c->pid = ?;
*/
if (getpeereid(c->sock, &c->euid, &c->egid) == 0) {
res = 0;
} else {
res = -errno;
}
}
#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 {
res = -EBADMSG;
}
#else /* no credentials */
res = -ENOTSUP;
#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_accept) {
res = c->service->serv_fns.connection_accept(c,
c->euid,
c->egid);
} else {
res = 0;
}
}
return res;
}
int32_t qb_ipcc_us_connect(const char *socket_name, int32_t * sock_pt)
static int32_t qb_ipcc_us_sock_connect(const char *socket_name, int32_t * sock_pt)
{
int32_t request_fd;
struct sockaddr_un address;
int32_t res = 0;
#if defined(QB_SOLARIS)
request_fd = socket(PF_UNIX, SOCK_STREAM, 0);
@ -330,7 +278,14 @@ int32_t qb_ipcc_us_connect(const char *socket_name, int32_t * sock_pt)
#ifdef SO_NOSIGPIPE
socket_nosigpipe(request_fd);
#endif /* SO_NOSIGPIPE */
set_cloexec_flag(request_fd);
res = set_cloexec_flag(request_fd);
if (res < 0) {
goto error_connect;
}
res = set_nonblock_flag(request_fd);
if (res < 0) {
goto error_connect;
}
memset(&address, 0, sizeof(struct sockaddr_un));
address.sun_family = AF_UNIX;
@ -345,6 +300,7 @@ int32_t qb_ipcc_us_connect(const char *socket_name, int32_t * sock_pt)
#endif
if (connect(request_fd, (struct sockaddr *)&address,
QB_SUN_LEN(&address)) == -1) {
res = -errno;
goto error_connect;
}
@ -355,15 +311,86 @@ error_connect:
close(request_fd);
*sock_pt = -1;
return -errno;
return res;
}
void qb_ipcc_us_disconnect(int32_t sock)
void qb_ipcc_us_sock_close(int32_t sock)
{
shutdown(sock, SHUT_RDWR);
close(sock);
}
int32_t qb_ipcc_us_setup_connect(struct qb_ipcc_connection *c,
struct qb_ipc_connection_response *r)
{
int32_t res;
struct qb_ipc_connection_request request;
res = qb_ipcc_us_sock_connect(c->name, &c->setup.u.us.sock);
if (res != 0) {
return res;
}
request.hdr.id = QB_IPC_MSG_AUTHENTICATE;
request.hdr.size = sizeof(request);
request.max_msg_size = c->setup.max_msg_size;
res = qb_ipc_us_send(&c->setup, &request, request.hdr.size);
if (res < 0) {
qb_ipcc_us_sock_close(c->setup.u.us.sock);
return res;
}
res = qb_ipc_us_recv(&c->setup, r, sizeof(struct qb_ipc_connection_response), 0);
if (res < 0) {
return res;
}
if (r->hdr.error != 0) {
return r->hdr.error;
}
return 0;
}
static void qb_ipcc_us_disconnect(struct qb_ipcc_connection* c)
{
close(c->request.u.us.sock);
close(c->event.u.us.sock);
}
int32_t qb_ipcc_us_connect(struct qb_ipcc_connection *c,
struct qb_ipc_connection_response *r)
{
int32_t res;
struct qb_ipc_event_connection_request request;
c->needs_sock_for_poll = QB_FALSE;
c->funcs.send = qb_ipc_us_send;
c->funcs.sendv = qb_ipc_us_sendv;
c->funcs.recv = qb_ipc_us_recv;
c->funcs.fc_get = NULL;
c->funcs.disconnect = qb_ipcc_us_disconnect;
c->request.u.us.sock = c->setup.u.us.sock;
c->response.u.us.sock = c->setup.u.us.sock;
c->setup.u.us.sock = -1;
res = qb_ipcc_us_sock_connect(c->name, &c->event.u.us.sock);
if (res != 0) {
return res;
}
request.hdr.id = QB_IPC_MSG_NEW_EVENT_SOCK;
request.hdr.size = sizeof(request);
request.connection = r->connection;
res = qb_ipc_us_send(&c->event, &request, request.hdr.size);
if (res < 0) {
qb_ipcc_us_sock_close(c->event.u.us.sock);
return res;
}
return 0;
}
/*
**************************************************************************
@ -392,14 +419,12 @@ int32_t qb_ipcs_us_publish(struct qb_ipcs_service * s)
return res;
}
set_cloexec_flag(s->server_sock);
res = fcntl(s->server_sock, F_SETFL, O_NONBLOCK);
if (res == -1) {
res = -errno;
strerror_r(errno, error_str, 100);
qb_util_log(LOG_CRIT,
"Could not set non-blocking operation on server socket: %s\n",
error_str);
res = set_cloexec_flag(s->server_sock);
if (res < 0) {
goto error_close;
}
res = set_nonblock_flag(s->server_sock);
if (res < 0) {
goto error_close;
}
@ -409,6 +434,7 @@ int32_t qb_ipcs_us_publish(struct qb_ipcs_service * s)
un_addr.sun_len = SUN_LEN(&un_addr);
#endif
qb_util_log(LOG_INFO, "server name: %s", s->name);
#if defined(QB_LINUX)
sprintf(un_addr.sun_path + 1, "%s", s->name);
#else
@ -469,14 +495,227 @@ int32_t qb_ipcs_us_withdraw(struct qb_ipcs_service * s)
return 0;
}
static int32_t handle_new_connection(struct qb_ipcs_service *s,
int32_t auth_result,
int32_t sock,
void *msg, size_t len,
struct ipc_auth_ugp *ugp)
{
struct qb_ipcs_connection *c = NULL;
struct qb_ipc_connection_request *req = msg;
int32_t res = auth_result;
struct qb_ipc_connection_response response;
char error_str[100];
if (res != 0) {
goto send_response;
}
c = qb_ipcs_connection_alloc(s);
c->setup.u.us.sock = sock;
c->request.max_msg_size = req->max_msg_size;
c->response.max_msg_size = req->max_msg_size;
c->event.max_msg_size = req->max_msg_size;
c->pid = ugp->pid;
c->euid = ugp->uid;
c->egid = ugp->gid;
if (c->service->serv_fns.connection_accept) {
res = c->service->serv_fns.connection_accept(c,
c->euid,
c->egid);
}
if (res != 0) {
goto send_response;
}
qb_util_log(LOG_INFO, "IPC credentials authenticated");
memset(&response, 0, sizeof(response));
if (s->funcs.connect) {
res = s->funcs.connect(s, c, &response);
if (res != 0) {
goto send_response;
}
}
qb_list_add(&c->list, &s->connections);
c->receive_buf = malloc(c->request.max_msg_size);
if (s->needs_sock_for_poll) {
s->poll_fns.dispatch_add(s->poll_priority, c->setup.u.us.sock,
POLLIN | POLLPRI | POLLNVAL,
c,
qb_ipcs_dispatch_connection_request);
}
if (s->type == QB_IPC_SOCKET) {
c->request.u.us.sock = c->setup.u.us.sock;
c->response.u.us.sock = c->setup.u.us.sock;
s->poll_fns.dispatch_add(s->poll_priority, c->request.u.us.sock,
POLLIN | POLLPRI | POLLNVAL,
c,
qb_ipcs_dispatch_connection_request);
}
send_response:
response.hdr.id = QB_IPC_MSG_AUTHENTICATE;
response.hdr.size = sizeof(response);
response.hdr.error = res;
if (res == 0) {
response.connection = (intptr_t)c;
response.connection_type = s->type;
response.max_msg_size = c->request.max_msg_size;
}
qb_ipc_us_send(&c->setup, &response, response.hdr.size);
if (res == 0) {
if (s->serv_fns.connection_created) {
s->serv_fns.connection_created(c);
}
} else if (res == -EACCES) {
qb_util_log(LOG_ERR, "Invalid IPC credentials.");
} else {
strerror_r(-response.hdr.error, error_str, 100);
qb_util_log(LOG_ERR, "Error in connection setup: %s.",
error_str);
}
if (res != 0 && c) {
qb_ipcs_disconnect(c);
} else if (res != 0) {
qb_ipcc_us_sock_close(sock);
}
return res;
}
static void handle_connection_new_sock(struct qb_ipcs_service *s,
int32_t sock, void *msg)
{
struct qb_ipcs_connection *c = NULL;
struct qb_ipc_event_connection_request *req = msg;
c = (struct qb_ipcs_connection *)req->connection;
c->event.u.us.sock = sock;
}
static int32_t qb_ipcs_uc_recv_and_auth(int32_t sock, void *msg, size_t len,
struct ipc_auth_ugp *ugp)
{
int32_t res = 0;
struct msghdr msg_recv;
struct iovec iov_recv;
#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 = msg;
iov_recv.iov_len = len;
#ifdef QB_LINUX
setsockopt(sock, SOL_SOCKET, SO_PASSCRED, &on, sizeof(on));
#endif
res = qb_ipc_us_recv_msghdr(sock, &msg_recv, msg, len);
if (res < 0) {
goto cleanup_and_return;
}
if (res != len) {
res = -EIO;
goto cleanup_and_return;
}
res = -EBADMSG;
/*
* 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(sock, &uc) == 0) {
res = 0;
ugp->uid = ucred_geteuid(uc);
ugp->gid = ucred_getegid(uc);
ugp->pid = ucred_getpid(uc);
ucred_free(uc);
} else {
res = -errno;
}
}
#elif HAVE_GETPEEREID
/*
* Usually MacOSX systems
*/
{
/*
* TODO get the peer's pid.
* c->pid = ?;
*/
if (getpeereid(sock, &ugp->uid, &ugp->gid) == 0) {
res = 0;
} else {
res = -errno;
}
}
#elif SO_PASSCRED
/*
* Usually Linux systems
*/
cmsg = CMSG_FIRSTHDR(&msg_recv);
assert(cmsg);
cred = (struct ucred *)CMSG_DATA(cmsg);
if (cred) {
res = 0;
ugp->pid = cred->pid;
ugp->uid = cred->uid;
ugp->gid = cred->gid;
} else {
res = -EBADMSG;
}
#else /* no credentials */
res = -ENOTSUP;
#endif /* no credentials */
cleanup_and_return:
#ifdef QB_LINUX
setsockopt(sock, SOL_SOCKET, SO_PASSCRED, &off, sizeof(off));
#endif
return res;
}
static int32_t qb_ipcs_us_connection_acceptor(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 qb_ipc_connection_response response;
int32_t res;
struct qb_ipc_connection_request setup_msg;
struct ipc_auth_ugp ugp;
socklen_t addrlen = sizeof(struct sockaddr_un);
char error_str[100];
@ -502,64 +741,59 @@ retry_accept:
return 0; /* This is an error, but -1 would indicate disconnect from poll loop */
}
set_cloexec_flag(new_fd);
res = fcntl(new_fd, F_SETFL, O_NONBLOCK);
if (res == -1) {
strerror_r(errno, error_str, 100);
qb_util_log(LOG_ERR,
"Could not set non-blocking operation on library connection: %s\n",
error_str);
res = set_cloexec_flag(new_fd);
if (res < 0) {
close(new_fd);
return 0; /* This is an error, but -1 would indicate disconnect from poll loop */
}
res = set_nonblock_flag(new_fd);
if (res < 0) {
close(new_fd);
return 0; /* This is an error, but -1 would indicate disconnect from poll loop */
}
c = qb_ipcs_connection_alloc(s);
c->sock = new_fd;
res = qb_ipcs_uc_recv_and_auth(new_fd, &setup_msg, sizeof(setup_msg),
&ugp);
res = qb_ipcs_uc_recv_and_auth(c);
if (res == 0) {
qb_util_log(LOG_INFO, "IPC credentials authenticated");
memset(&response, 0, sizeof(response));
res = s->funcs.connect(s, c, &response);
if (res != 0) {
goto send_response;
if (setup_msg.hdr.id == QB_IPC_MSG_AUTHENTICATE) {
handle_new_connection(s, res, new_fd, &setup_msg, sizeof(setup_msg),
&ugp);
} else if (setup_msg.hdr.id == QB_IPC_MSG_NEW_EVENT_SOCK) {
if (res == 0) {
handle_connection_new_sock(s, new_fd, &setup_msg);
} else {
close(new_fd);
}
qb_list_add(&c->list, &s->connections);
c->receive_buf = malloc(c->request.max_msg_size);
if (s->needs_sock_for_poll) {
s->poll_fns.dispatch_add(s->poll_priority, c->sock,
POLLIN | POLLPRI | POLLNVAL,
c,
qb_ipcs_dispatch_connection_request);
}
}
send_response:
response.hdr.id = QB_IPC_MSG_AUTHENTICATE;
response.hdr.size = sizeof(response);
response.hdr.error = res;
response.connection_type = s->type;
response.max_msg_size = c->request.max_msg_size;
qb_ipc_us_send(c->sock, &response, response.hdr.size);
if (res == 0) {
if (s->serv_fns.connection_created) {
s->serv_fns.connection_created(c);
}
} else if (res == -EACCES) {
qb_util_log(LOG_ERR, "Invalid IPC credentials.");
} else {
strerror_r(-response.hdr.error, error_str, 100);
qb_util_log(LOG_ERR, "Error in connection setup: %s.",
error_str);
}
if (res != 0) {
qb_ipcs_disconnect(c);
close(new_fd);
}
return 0;
}
static void qb_ipcs_us_disconnect(struct qb_ipcs_connection *c)
{
// close(c->setup.u.us.sock);
close(c->request.u.us.sock);
// close(c->response.u.us.sock);
close(c->event.u.us.sock);
}
void qb_ipcs_us_init(struct qb_ipcs_service *s)
{
s->funcs.connect = NULL;
s->funcs.disconnect = qb_ipcs_us_disconnect;
s->funcs.recv = qb_ipc_us_recv;
s->funcs.peek = NULL;
s->funcs.reclaim = NULL;
s->funcs.send = qb_ipc_us_send;
s->funcs.sendv = qb_ipc_us_sendv;
s->funcs.fc_set = NULL;
s->funcs.q_len_get = NULL;
s->needs_sock_for_poll = QB_FALSE;
}

97
lib/ipcc.c
View File

@ -29,54 +29,21 @@
qb_ipcc_connection_t *qb_ipcc_connect(const char *name, size_t max_msg_size)
{
int32_t res;
int32_t usock;
qb_ipcc_connection_t *c = NULL;
struct qb_ipc_connection_request request;
struct qb_ipc_connection_response response;
res = qb_ipcc_us_connect(name, &usock);
if (res != 0) {
errno = -res;
perror("qb_ipcc_us_connect");
return NULL;
}
request.hdr.id = QB_IPC_MSG_AUTHENTICATE;
request.hdr.size = sizeof(request);
request.max_msg_size = max_msg_size;
res = qb_ipc_us_send(usock, &request, request.hdr.size);
if (res < 0) {
perror("qb_ipc_us_send");
qb_ipcc_us_disconnect(usock);
errno = -res;
return NULL;
}
res = qb_ipc_us_recv(usock, &response, sizeof(response));
if (res < 0) {
perror("qb_ipc_us_recv");
qb_ipcc_us_disconnect(usock);
errno = -res;
return NULL;
}
if (response.hdr.error != 0) {
errno = -response.hdr.error;
perror("recv:message");
return NULL;
}
c = malloc(sizeof(struct qb_ipcc_connection));
if (c == NULL) {
perror("malloc:connection");
return NULL;
}
c->setup.max_msg_size = max_msg_size;
strcpy(c->name, name);
res = qb_ipcc_us_setup_connect(c, &response);
if (res < 0) {
goto disconnect_and_cleanup;
}
c->type = response.connection_type;
c->sock = usock;
qb_util_log(LOG_DEBUG, "%s() max_msg_size:%zu actual:%u", __func__,
max_msg_size, response.max_msg_size);
c->response.max_msg_size = response.max_msg_size;
c->request.max_msg_size = response.max_msg_size;
c->event.max_msg_size = response.max_msg_size;
@ -93,19 +60,22 @@ qb_ipcc_connection_t *qb_ipcc_connect(const char *name, size_t max_msg_size)
res = qb_ipcc_smq_connect(c, &response);
break;
case QB_IPC_SOCKET:
c->needs_sock_for_poll = QB_FALSE;
res = qb_ipcc_us_connect(c, &response);
break;
default:
res = -EINVAL;
break;
}
if (res != 0) {
qb_ipcc_us_disconnect(usock);
free(c);
c = NULL;
errno = -res;
goto disconnect_and_cleanup;
}
return c;
disconnect_and_cleanup:
qb_ipcc_us_sock_close(c->setup.u.us.sock);
free(c);
errno = -res;
return NULL;
}
ssize_t qb_ipcc_send(struct qb_ipcc_connection * c, const void *msg_ptr,
@ -122,7 +92,9 @@ ssize_t qb_ipcc_send(struct qb_ipcc_connection * c, const void *msg_ptr,
res = c->funcs.send(&c->request, msg_ptr, msg_len);
if (res > 0 && c->needs_sock_for_poll) {
qb_ipc_us_send(c->sock, msg_ptr, 1);
do {
res = qb_ipc_us_send(&c->setup, msg_ptr, 1);
} while (res == -EAGAIN);
}
return res;
}
@ -147,7 +119,9 @@ ssize_t qb_ipcc_sendv(struct qb_ipcc_connection* c, const struct iovec* iov,
res = c->funcs.sendv(&c->request, iov, iov_len);
if (res > 0 && c->needs_sock_for_poll) {
qb_ipc_us_send(c->sock, &res, 1);
do {
res = qb_ipc_us_send(&c->setup, &res, 1);
} while (res == -EAGAIN);
}
return res;
}
@ -190,7 +164,11 @@ repeat_recv:
int32_t qb_ipcc_fd_get(struct qb_ipcc_connection * c, int32_t * fd)
{
*fd = c->sock;
if (c->type == QB_IPC_SOCKET) {
*fd = c->event.u.us.sock;
} else {
*fd = c->setup.u.us.sock;
}
return 0;
}
@ -200,18 +178,29 @@ ssize_t qb_ipcc_event_recv(struct qb_ipcc_connection * c, void *msg_pt,
char one_byte = 1;
int32_t res;
ssize_t size;
struct qb_ipc_one_way *ow = NULL;
res = qb_ipc_us_recv_ready(c->sock, ms_timeout);
if (res < 0) {
return res;
if (c->needs_sock_for_poll) {
ow = &c->setup;
}
if (c->type == QB_IPC_SOCKET) {
ow = &c->event;
}
if (ow) {
res = qb_ipc_us_recv_ready(ow, ms_timeout);
if (res < 0) {
return res;
}
}
size = c->funcs.recv(&c->event, msg_pt, msg_len, ms_timeout);
if (size < 0) {
return size;
}
res = qb_ipc_us_recv(c->sock, &one_byte, 1);
if (res < 0) {
return res;
if (c->needs_sock_for_poll) {
res = qb_ipc_us_recv(&c->setup, &one_byte, 1, 0);
if (res < 0) {
return res;
}
}
return size;
}
@ -220,7 +209,7 @@ void qb_ipcc_disconnect(struct qb_ipcc_connection *c)
{
qb_util_log(LOG_DEBUG, "%s()", __func__);
qb_ipcc_us_disconnect(c->sock);
qb_ipcc_us_sock_close(c->setup.u.us.sock);
if (c->funcs.disconnect) {
c->funcs.disconnect(c);
}

79
lib/ipcs.c
View File

@ -86,7 +86,7 @@ int32_t qb_ipcs_run(qb_ipcs_service_pt pt)
switch (s->type) {
case QB_IPC_SOCKET:
/* qb_ipcs_us_init((struct qb_ipcs_service *)s); */
qb_ipcs_us_init((struct qb_ipcs_service *)s);
break;
case QB_IPC_SHM:
qb_ipcs_shm_init((struct qb_ipcs_service *)s);
@ -150,8 +150,13 @@ void qb_ipcs_request_rate_limit(qb_ipcs_service_pt pt, enum qb_ipcs_rate_limit r
s->poll_fns.dispatch_mod(p, c->request.u.pmq.q,
POLLIN | POLLPRI | POLLNVAL,
c, qb_ipcs_dispatch_service_request);
} else if (s->type == QB_IPC_SOCKET) {
s->poll_fns.dispatch_mod(p, c->event.u.us.sock,
POLLIN | POLLPRI | POLLNVAL,
c,
qb_ipcs_dispatch_connection_request);
} else {
s->poll_fns.dispatch_mod(p, c->sock,
s->poll_fns.dispatch_mod(p, c->setup.u.us.sock,
POLLIN | POLLPRI | POLLNVAL,
c,
qb_ipcs_dispatch_connection_request);
@ -206,6 +211,7 @@ ssize_t qb_ipcs_event_send(struct qb_ipcs_connection *c, const void *data,
size_t size)
{
ssize_t res;
ssize_t res2 = 0;
int32_t try_count = 0;
qb_ipcs_connection_ref_inc(c);
@ -215,7 +221,11 @@ ssize_t qb_ipcs_event_send(struct qb_ipcs_connection *c, const void *data,
res = c->service->funcs.send(&c->event, data, size);
} while (res == -EAGAIN && try_count < 20);
if (res > 0) {
qb_ipc_us_send(c->sock, data, 1);
if (c->service->needs_sock_for_poll) {
do {
res2 = qb_ipc_us_send(&c->setup, &res, 1);
} while (res2 == -EAGAIN);
}
} else {
qb_util_log(LOG_ERR,
"failed to send event : %s",
@ -230,6 +240,7 @@ ssize_t qb_ipcs_event_send(struct qb_ipcs_connection *c, const void *data,
ssize_t qb_ipcs_event_sendv(struct qb_ipcs_connection *c, const struct iovec * iov, size_t iov_len)
{
ssize_t res;
ssize_t res2;
int32_t try_count = 0;
qb_ipcs_connection_ref_inc(c);
@ -239,7 +250,11 @@ ssize_t qb_ipcs_event_sendv(struct qb_ipcs_connection *c, const struct iovec * i
res = c->service->funcs.sendv(&c->event, iov, iov_len);
} while (res == -EAGAIN && try_count < 20);
if (res > 0) {
qb_ipc_us_send(c->sock, &res, 1);
if (c->service->needs_sock_for_poll) {
do {
res2 = qb_ipc_us_send(&c->setup, &res, 1);
} while (res2 == -EAGAIN);
}
} else {
qb_util_log(LOG_ERR,
"failed to send event : %s",
@ -260,7 +275,6 @@ struct qb_ipcs_connection *qb_ipcs_connection_alloc(struct qb_ipcs_service *s)
c->pid = 0;
c->euid = -1;
c->egid = -1;
c->sock = -1;
qb_list_init(&c->list);
c->receive_buf = NULL;
c->fc_enabled = QB_FALSE;
@ -289,7 +303,9 @@ void qb_ipcs_connection_ref_dec(struct qb_ipcs_connection *c)
c->service->serv_fns.connection_destroyed(c);
}
c->service->funcs.disconnect(c);
qb_ipcc_us_disconnect(c->sock);
if (c->service->needs_sock_for_poll) {
qb_ipcc_us_sock_close(c->setup.u.us.sock);
}
if (c->receive_buf) {
free(c->receive_buf);
}
@ -366,6 +382,7 @@ cleanup:
}
#define IPC_REQUEST_TIMEOUT 10
#define MAX_RECV_MSGS 50
int32_t qb_ipcs_dispatch_service_request(int32_t fd, int32_t revents,
void *data)
@ -378,7 +395,26 @@ int32_t qb_ipcs_dispatch_service_request(int32_t fd, int32_t revents,
return res;
}
#define MAX_RECV_MSGS 50
static ssize_t _request_q_len_get(struct qb_ipcs_connection *c)
{
ssize_t q_len;
if (c->service->funcs.q_len_get) {
q_len = c->service->funcs.q_len_get(&c->request);
if (q_len < 0) {
q_len = 1;
}
q_len = QB_MIN(q_len, MAX_RECV_MSGS);
if (c->service->poll_priority == QB_LOOP_MED)
q_len = QB_MIN(q_len, 5);
if (c->service->poll_priority == QB_LOOP_LOW)
q_len = 1;
} else {
q_len = 1;
}
return q_len;
}
int32_t qb_ipcs_dispatch_connection_request(int32_t fd, int32_t revents,
void *data)
{
@ -390,40 +426,33 @@ int32_t qb_ipcs_dispatch_connection_request(int32_t fd, int32_t revents,
if (revents & POLLHUP) {
qb_util_log(LOG_DEBUG, "%s HUP", __func__);
qb_ipcc_us_disconnect(c->sock);
c->sock = -1;
if (c->service->needs_sock_for_poll) {
qb_ipcc_us_sock_close(c->setup.u.us.sock);
c->setup.u.us.sock = -1;
}
qb_ipcs_connection_ref_dec(c);
qb_ipcs_disconnect(c);
return -ESHUTDOWN;
}
if (c->service->funcs.q_len_get) {
avail = c->service->funcs.q_len_get(&c->request);
if (avail < 0) {
avail = 1;
}
if (avail > MAX_RECV_MSGS) {
avail = MAX_RECV_MSGS;
}
} else {
avail = 1;
}
avail = _request_q_len_get(c);
do {
res = _process_request_(c, IPC_REQUEST_TIMEOUT);
if (res > 0 || res == -ENOBUFS || res == -EINVAL) {
recvd++;
}
avail--;
} while (avail > 0 && c->service->poll_priority == QB_LOOP_HIGH);
if (res > 0) {
avail--;
}
} while (avail > 0 && res > 0);
if (c->service->needs_sock_for_poll && recvd > 0) {
qb_ipc_us_recv(c->sock, bytes, recvd);
qb_ipc_us_recv(&c->setup, bytes, recvd, 0);
}
res = QB_MIN(0, res);
if (res == -EAGAIN || res == -ENOBUFS) {
res = 0;
}
if (res != 0) {
qb_util_log(LOG_INFO, "%s returning %d : %s",
__func__, res, strerror(-res));

26
tests/check_ipc.c
View File

@ -104,7 +104,7 @@ static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c,
res = qb_ipcs_event_send(c, &response,
sizeof(response));
if (res < 0) {
perror("qb_ipcs_dispatch_send");
perror("qb_ipcs_event_send");
}
}
return 0;
@ -279,6 +279,13 @@ START_TEST(test_ipc_txrx_shm)
}
END_TEST
START_TEST(test_ipc_txrx_us)
{
ipc_type = QB_IPC_SOCKET;
test_ipc_txrx();
}
END_TEST
START_TEST(test_ipc_fc_shm)
{
turn_on_fc = QB_TRUE;
@ -367,6 +374,13 @@ START_TEST(test_ipc_disp_shm)
}
END_TEST
START_TEST(test_ipc_disp_us)
{
ipc_type = QB_IPC_SOCKET;
test_ipc_dispatch();
}
END_TEST
static Suite *ipc_suite(void)
{
TCase *tc;
@ -378,6 +392,11 @@ static Suite *ipc_suite(void)
tcase_set_timeout(tc, 6);
suite_add_tcase(s, tc);
tc = tcase_create("ipc_txrx_us");
tcase_add_test(tc, test_ipc_txrx_us);
tcase_set_timeout(tc, 6);
suite_add_tcase(s, tc);
tc = tcase_create("ipc_fc_shm");
tcase_add_test(tc, test_ipc_fc_shm);
tcase_set_timeout(tc, 6);
@ -400,6 +419,11 @@ static Suite *ipc_suite(void)
tcase_set_timeout(tc, 16);
suite_add_tcase(s, tc);
tc = tcase_create("ipc_dispatch_us");
tcase_add_test(tc, test_ipc_disp_us);
tcase_set_timeout(tc, 16);
suite_add_tcase(s, tc);
return s;
}