diff --git a/lib/ipc_setup.c b/lib/ipc_setup.c index 0de7115..73b4e37 100644 --- a/lib/ipc_setup.c +++ b/lib/ipc_setup.c @@ -1,954 +1,953 @@ /* - * Copyright (C) 2010,2013 Red Hat, Inc. + * Copyright 2010-2024 Red Hat, Inc. * * Author: Angus Salkeld * * 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 . */ #include "os_base.h" #include #if defined(HAVE_GETPEERUCRED) #include #endif #ifdef HAVE_SYS_UN_H #include #endif /* HAVE_SYS_UN_H */ #ifdef HAVE_SYS_STAT_H #include #endif #ifdef HAVE_SYS_MMAN_H #include #endif #include #include #include #include #include "util_int.h" #include "ipc_int.h" struct ipc_auth_ugp { uid_t uid; gid_t gid; pid_t pid; }; struct ipc_auth_data { int32_t sock; struct qb_ipcs_service *s; union { struct qb_ipc_connection_request req; struct qb_ipc_connection_response res; } msg; struct msghdr msg_recv; struct iovec iov_recv; struct ipc_auth_ugp ugp; size_t processed; size_t len; #ifdef SO_PASSCRED char *cmsg_cred; #endif }; static int32_t qb_ipcs_us_connection_acceptor(int fd, int revent, void *data); ssize_t qb_ipc_us_send(struct qb_ipc_one_way *one_way, const void *msg, size_t len) { int32_t result; int32_t processed = 0; char *rbuf = (char *)msg; qb_sigpipe_ctl(QB_SIGPIPE_IGNORE); retry_send: result = send(one_way->u.us.sock, &rbuf[processed], len - processed, MSG_NOSIGNAL); if (result == -1) { if (errno == EAGAIN && processed > 0) { goto retry_send; } else { qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); return -errno; } } processed += result; if (processed != len) { goto retry_send; } qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); return processed; } static ssize_t qb_ipc_us_recv_msghdr(struct ipc_auth_data *data) { char *msg = (char *) &data->msg; int32_t result; qb_sigpipe_ctl(QB_SIGPIPE_IGNORE); retry_recv: data->msg_recv.msg_iov->iov_base = &msg[data->processed]; data->msg_recv.msg_iov->iov_len = data->len - data->processed; result = recvmsg(data->sock, &data->msg_recv, MSG_NOSIGNAL | MSG_WAITALL); if (result == -1 && errno == EAGAIN) { qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); return -EAGAIN; } if (result == -1) { qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); return -errno; } if (result == 0) { qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); qb_util_log(LOG_DEBUG, "recv(fd %d) got 0 bytes assuming ENOTCONN", data->sock); return -ENOTCONN; } data->processed += result; if (data->processed != data->len) { goto retry_recv; } qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); assert(data->processed == data->len); return data->processed; } int32_t qb_ipc_us_sock_error_is_disconnected(int err) { if (err >= 0) { return QB_FALSE; } else if (err == -EAGAIN || err == -ETIMEDOUT || err == -EINTR || #ifdef EWOULDBLOCK err == -EWOULDBLOCK || #endif err == -EMSGSIZE || err == -ENOMSG || err == -EINVAL) { return QB_FALSE; } return QB_TRUE; } int32_t qb_ipc_us_ready(struct qb_ipc_one_way * ow_data, struct qb_ipc_one_way * ow_conn, int32_t ms_timeout, int32_t events) { struct pollfd ufds[2]; int32_t poll_events; int numfds = 1; int i; ufds[0].fd = ow_data->u.us.sock; ufds[0].events = events; ufds[0].revents = 0; if (ow_conn && ow_data != ow_conn) { numfds++; ufds[1].fd = ow_conn->u.us.sock; ufds[1].events = POLLIN; ufds[1].revents = 0; } poll_events = poll(ufds, numfds, ms_timeout); if ((poll_events == -1 && errno == EINTR) || poll_events == 0) { return -EAGAIN; } else if (poll_events == -1) { return -errno; } for (i = 0; i < poll_events; i++) { if (ufds[i].revents & POLLERR) { qb_util_log(LOG_DEBUG, "poll(fd %d) got POLLERR", ufds[i].fd); return -ENOTCONN; } else if (ufds[i].revents & POLLHUP) { qb_util_log(LOG_DEBUG, "poll(fd %d) got POLLHUP", ufds[i].fd); return -ENOTCONN; } else if (ufds[i].revents & POLLNVAL) { qb_util_log(LOG_DEBUG, "poll(fd %d) got POLLNVAL", ufds[i].fd); return -ENOTCONN; } else if (ufds[i].revents == 0) { qb_util_log(LOG_DEBUG, "poll(fd %d) zero revents", ufds[i].fd); return -ENOTCONN; } } return 0; } /* * recv an entire message - and try hard to get all of it. */ ssize_t qb_ipc_us_recv(struct qb_ipc_one_way * one_way, void *msg, size_t len, int32_t timeout) { int32_t result; int32_t final_rc = 0; int32_t processed = 0; int32_t to_recv = len; char *data = msg; qb_sigpipe_ctl(QB_SIGPIPE_IGNORE); retry_recv: result = recv(one_way->u.us.sock, &data[processed], to_recv, MSG_NOSIGNAL | MSG_WAITALL); if (result == -1) { if (errno == EAGAIN && (processed > 0 || timeout == -1)) { result = qb_ipc_us_ready(one_way, NULL, timeout, POLLIN); if (result == 0 || result == -EAGAIN) { goto retry_recv; } final_rc = result; goto cleanup_sigpipe; } else if (errno == ECONNRESET || errno == EPIPE) { final_rc = -ENOTCONN; goto cleanup_sigpipe; } else { final_rc = -errno; goto cleanup_sigpipe; } } if (result == 0) { final_rc = -ENOTCONN; goto cleanup_sigpipe; } processed += result; to_recv -= result; if (processed != len) { goto retry_recv; } final_rc = processed; cleanup_sigpipe: qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); return final_rc; } static int32_t qb_ipcc_stream_sock_connect(const char *socket_name, int32_t * sock_pt) { int32_t request_fd; struct sockaddr_un address; int32_t res = 0; request_fd = socket(PF_UNIX, SOCK_STREAM, 0); if (request_fd == -1) { return -errno; } qb_socket_nosigpipe(request_fd); res = qb_sys_fd_nonblock_cloexec_set(request_fd); if (res < 0) { goto error_connect; } memset(&address, 0, sizeof(struct sockaddr_un)); address.sun_family = AF_UNIX; #ifdef HAVE_STRUCT_SOCKADDR_UN_SUN_LEN address.sun_len = QB_SUN_LEN(&address); #endif if (!use_filesystem_sockets()) { snprintf(address.sun_path + 1, UNIX_PATH_MAX - 1, "%s", socket_name); } else { snprintf(address.sun_path, sizeof(address.sun_path), "%s/%s", SOCKETDIR, socket_name); } if (connect(request_fd, (struct sockaddr *)&address, QB_SUN_LEN(&address)) == -1) { res = -errno; goto error_connect; } *sock_pt = request_fd; return 0; error_connect: close(request_fd); *sock_pt = -1; return res; } void qb_ipcc_us_sock_close(int32_t sock) { shutdown(sock, SHUT_RDWR); close(sock); } static int32_t qb_ipc_auth_creds(struct ipc_auth_data *data) { int32_t res = 0; /* * 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(data->sock, &uc) == 0) { res = 0; data->ugp.uid = ucred_geteuid(uc); data->ugp.gid = ucred_getegid(uc); data->ugp.pid = ucred_getpid(uc); ucred_free(uc); } else { res = -errno; } } #elif defined(HAVE_GETPEEREID) /* * Usually MacOSX systems */ { /* * TODO get the peer's pid. * c->pid = ?; */ if (getpeereid(data->sock, &data->ugp.uid, &data->ugp.gid) == 0) { res = 0; } else { res = -errno; } } #elif defined(SO_PASSCRED) /* * Usually Linux systems */ { struct ucred cred; struct cmsghdr *cmsg; res = -EINVAL; for (cmsg = CMSG_FIRSTHDR(&data->msg_recv); cmsg != NULL; cmsg = CMSG_NXTHDR(&data->msg_recv, cmsg)) { if (cmsg->cmsg_type != SCM_CREDENTIALS) continue; memcpy(&cred, CMSG_DATA(cmsg), sizeof(struct ucred)); res = 0; data->ugp.pid = cred.pid; data->ugp.uid = cred.uid; data->ugp.gid = cred.gid; break; } } #else /* no credentials */ data->ugp.pid = 0; data->ugp.uid = 0; data->ugp.gid = 0; res = -ENOTSUP; #endif /* no credentials */ return res; } static void destroy_ipc_auth_data(struct ipc_auth_data *data) { if (data->s) { qb_ipcs_unref(data->s); } #ifdef SO_PASSCRED free(data->cmsg_cred); #endif free(data); } static struct ipc_auth_data * init_ipc_auth_data(int sock, size_t len) { struct ipc_auth_data *data = calloc(1, sizeof(struct ipc_auth_data)); if (data == NULL) { return NULL; } data->msg_recv.msg_iov = &data->iov_recv; data->msg_recv.msg_iovlen = 1; data->msg_recv.msg_name = 0; data->msg_recv.msg_namelen = 0; #ifdef SO_PASSCRED data->cmsg_cred = calloc(1, CMSG_SPACE(sizeof(struct ucred))); if (data->cmsg_cred == NULL) { destroy_ipc_auth_data(data); return NULL; } data->msg_recv.msg_control = (void *)data->cmsg_cred; data->msg_recv.msg_controllen = CMSG_SPACE(sizeof(struct ucred)); #endif #ifdef QB_SOLARIS data->msg_recv.msg_accrights = 0; data->msg_recv.msg_accrightslen = 0; #else data->msg_recv.msg_flags = 0; #endif /* QB_SOLARIS */ data->len = len; data->iov_recv.iov_base = &data->msg; data->iov_recv.iov_len = data->len; data->sock = sock; return data; } 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; #ifdef QB_LINUX int on = 1; #endif res = qb_ipcc_stream_sock_connect(c->name, &c->setup.u.us.sock); if (res != 0) { return res; } #ifdef QB_LINUX setsockopt(c->setup.u.us.sock, SOL_SOCKET, SO_PASSCRED, &on, sizeof(on)); #endif memset(&request, 0, sizeof(request)); 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; } /* ... To be continued ... (when the FD is active) */ return 0; } #define AUTH_RECV_MAX_RETRIES 10 #define AUTH_RECV_SLEEP_TIME_US 100 /* Called from ipcc_connect_continue() when async connect socket is active */ int qb_ipcc_setup_connect_continue(struct qb_ipcc_connection *c, struct qb_ipc_connection_response *r) { struct ipc_auth_data *data; int32_t res; int retry_count = 0; #ifdef QB_LINUX int off = 0; #endif data = init_ipc_auth_data(c->setup.u.us.sock, sizeof(struct qb_ipc_connection_response)); if (data == NULL) { qb_ipcc_us_sock_close(c->setup.u.us.sock); return -ENOMEM; } retry: res = qb_ipc_us_recv_msghdr(data); if (res == -EAGAIN && ++retry_count < AUTH_RECV_MAX_RETRIES) { struct timespec ts = {0, AUTH_RECV_SLEEP_TIME_US*QB_TIME_NS_IN_USEC}; struct timespec ts_left = {0, 0}; nanosleep(&ts, &ts_left); goto retry; } #ifdef QB_LINUX setsockopt(c->setup.u.us.sock, SOL_SOCKET, SO_PASSCRED, &off, sizeof(off)); #endif if (res != data->len) { destroy_ipc_auth_data(data); return res; } memcpy(r, &data->msg.res, sizeof(struct qb_ipc_connection_response)); qb_ipc_auth_creds(data); c->egid = data->ugp.gid; c->euid = data->ugp.uid; c->server_pid = data->ugp.pid; destroy_ipc_auth_data(data); return r->hdr.error; } /* ************************************************************************** * SERVER */ int32_t qb_ipcs_us_publish(struct qb_ipcs_service * s) { struct sockaddr_un un_addr; int32_t res; #ifdef SO_PASSCRED int on = 1; #endif /* * Create socket for IPC clients, name socket, listen for connections */ s->server_sock = socket(PF_UNIX, SOCK_STREAM, 0); if (s->server_sock == -1) { res = -errno; qb_util_perror(LOG_ERR, "Cannot create server socket"); return res; } res = qb_sys_fd_nonblock_cloexec_set(s->server_sock); if (res < 0) { 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 qb_util_log(LOG_INFO, "server name: %s", s->name); if (!use_filesystem_sockets()) { snprintf(un_addr.sun_path + 1, UNIX_PATH_MAX - 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))) { res = -errno; qb_util_log(LOG_CRIT, "Required directory not present %s", SOCKETDIR); goto error_close; } snprintf(un_addr.sun_path, sizeof(un_addr.sun_path), "%s/%s", SOCKETDIR, s->name); unlink(un_addr.sun_path); } res = bind(s->server_sock, (struct sockaddr *)&un_addr, QB_SUN_LEN(&un_addr)); if (res) { res = -errno; qb_util_perror(LOG_ERR, "Could not bind AF_UNIX (%s)", un_addr.sun_path); goto error_close; } /* * Allow everyone to write to the socket since the IPC layer handles * security automatically */ if (use_filesystem_sockets()) { (void)chmod(un_addr.sun_path, S_IRWXU | S_IRWXG | S_IRWXO); } #ifdef SO_PASSCRED (void)setsockopt(s->server_sock, SOL_SOCKET, SO_PASSCRED, &on, sizeof(on)); #endif if (listen(s->server_sock, SERVER_BACKLOG) == -1) { qb_util_perror(LOG_ERR, "socket listen failed"); } res = s->poll_fns.dispatch_add(s->poll_priority, s->server_sock, POLLIN | POLLPRI | POLLNVAL, s, qb_ipcs_us_connection_acceptor); return res; error_close: close(s->server_sock); return res; } int32_t qb_ipcs_us_withdraw(struct qb_ipcs_service * s) { qb_util_log(LOG_INFO, "withdrawing server sockets"); (void)s->poll_fns.dispatch_del(s->server_sock); shutdown(s->server_sock, SHUT_RDWR); if (use_filesystem_sockets()) { struct sockaddr_un sockname; socklen_t socklen = sizeof(sockname); if ((getsockname(s->server_sock, (struct sockaddr *)&sockname, &socklen) == 0) && sockname.sun_family == AF_UNIX) { #ifdef HAVE_STRUCT_SOCKADDR_UN_SUN_LEN /* * Terminating NUL on FreeBSD is not part of the sun_path. * Add it to use sun_path as a parameter of unlink */ sockname.sun_path[sockname.sun_len - offsetof(struct sockaddr_un, sun_path)] = '\0'; #endif unlink(sockname.sun_path); } } close(s->server_sock); s->server_sock = -1; 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; int32_t res2 = 0; uint32_t max_buffer_size = QB_MAX(req->max_msg_size, s->max_buffer_size); struct qb_ipc_connection_response response; const char suffix[] = "/qb"; int desc_len; c = qb_ipcs_connection_alloc(s); if (c == NULL) { qb_ipcc_us_sock_close(sock); return -ENOMEM; } c->receive_buf = calloc(1, max_buffer_size); if (c->receive_buf == NULL) { free(c); qb_ipcc_us_sock_close(sock); return -ENOMEM; } c->setup.u.us.sock = sock; c->request.max_msg_size = max_buffer_size; c->response.max_msg_size = max_buffer_size; c->event.max_msg_size = max_buffer_size; c->pid = ugp->pid; c->auth.uid = c->euid = ugp->uid; c->auth.gid = c->egid = ugp->gid; c->auth.mode = 0600; c->stats.client_pid = ugp->pid; memset(&response, 0, sizeof(response)); #if defined(QB_LINUX) || defined(QB_CYGWIN) desc_len = snprintf(c->description, CONNECTION_DESCRIPTION - sizeof suffix, "/dev/shm/qb-%d-%d-%d-XXXXXX", s->pid, ugp->pid, c->setup.u.us.sock); if (desc_len < 0) { res = -errno; goto send_response; } if (desc_len >= CONNECTION_DESCRIPTION - sizeof suffix) { res = -ENAMETOOLONG; goto send_response; } if (mkdtemp(c->description) == NULL) { res = -errno; goto send_response; } if (chmod(c->description, 0770)) { res = -errno; goto send_response; } /* chown can fail because we might not be root */ (void)chown(c->description, c->auth.uid, c->auth.gid); /* We can't pass just a directory spec to the clients */ memcpy(c->description + desc_len, suffix, sizeof suffix); #else desc_len = snprintf(c->description, CONNECTION_DESCRIPTION, "%d-%d-%d", s->pid, ugp->pid, c->setup.u.us.sock); if (desc_len < 0) { res = -errno; goto send_response; } if (desc_len >= CONNECTION_DESCRIPTION) { res = -ENAMETOOLONG; goto send_response; } #endif if (auth_result == 0 && 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_DEBUG, "IPC credentials authenticated (%s)", c->description); if (s->funcs.connect) { res = s->funcs.connect(s, c, &response); if (res != 0) { goto send_response; } } /* * The connection is good, add it to the active connection list */ c->state = QB_IPCS_CONNECTION_ACTIVE; qb_list_add(&c->list, &s->connections); 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; s->stats.active_connections++; } res2 = qb_ipc_us_send(&c->setup, &response, response.hdr.size); if (res == 0 && res2 != response.hdr.size) { res = res2; } if (res == 0) { qb_ipcs_connection_ref(c); if (s->serv_fns.connection_created) { s->serv_fns.connection_created(c); } if (c->state == QB_IPCS_CONNECTION_ACTIVE) { c->state = QB_IPCS_CONNECTION_ESTABLISHED; } qb_ipcs_connection_unref(c); } else { if (res == -EACCES) { - qb_util_log(LOG_ERR, "Invalid IPC credentials (%s).", + qb_util_log(LOG_INFO, "IPC connection credentials rejected (%s)", c->description); } else if (res == -EAGAIN) { - qb_util_log(LOG_WARNING, "Denied connection, is not ready (%s)", + qb_util_log(LOG_INFO, "IPC connection not ready (%s)", c->description); } else { - errno = -res; - qb_util_perror(LOG_ERR, - "Error in connection setup (%s)", + qb_util_perror(LOG_INFO, "IPC connection setup failed (%s)", c->description); + errno = -res; } if (c->state == QB_IPCS_CONNECTION_INACTIVE) { /* This removes the initial alloc ref */ qb_ipcs_connection_unref(c); qb_ipcc_us_sock_close(sock); } else { qb_ipcs_disconnect(c); } } return res; } static int32_t process_auth(int32_t fd, int32_t revents, void *d) { struct ipc_auth_data *data = (struct ipc_auth_data *) d; int32_t res = 0; #ifdef SO_PASSCRED int off = 0; #endif if (data->s->server_sock == -1) { qb_util_log(LOG_DEBUG, "Closing fd (%d) for server shutdown", fd); res = -ESHUTDOWN; goto cleanup_and_return; } if (revents & POLLNVAL) { qb_util_log(LOG_DEBUG, "NVAL conn fd (%d)", fd); res = -EINVAL; goto cleanup_and_return; } if (revents & POLLHUP) { qb_util_log(LOG_DEBUG, "HUP conn fd (%d)", fd); res = -ESHUTDOWN; goto cleanup_and_return; } if ((revents & POLLIN) == 0) { return 0; } res = qb_ipc_us_recv_msghdr(data); if (res == -EAGAIN) { /* yield to mainloop, Let mainloop call us again */ return 0; } if (res != data->len) { res = -EIO; goto cleanup_and_return; } res = qb_ipc_auth_creds(data); cleanup_and_return: #ifdef SO_PASSCRED setsockopt(data->sock, SOL_SOCKET, SO_PASSCRED, &off, sizeof(off)); #endif (void)data->s->poll_fns.dispatch_del(data->sock); if (res < 0) { close(data->sock); } else if (data->msg.req.hdr.id == QB_IPC_MSG_AUTHENTICATE) { (void)handle_new_connection(data->s, res, data->sock, &data->msg, data->len, &data->ugp); } else { close(data->sock); } destroy_ipc_auth_data(data); return 1; } static void qb_ipcs_uc_recv_and_auth(int32_t sock, struct qb_ipcs_service *s) { int res = 0; struct ipc_auth_data *data = NULL; #ifdef SO_PASSCRED int on = 1; #endif data = init_ipc_auth_data(sock, sizeof(struct qb_ipc_connection_request)); if (data == NULL) { close(sock); /* -ENOMEM */ return; } data->s = s; qb_ipcs_ref(data->s); #ifdef SO_PASSCRED setsockopt(sock, SOL_SOCKET, SO_PASSCRED, &on, sizeof(on)); #endif res = s->poll_fns.dispatch_add(s->poll_priority, data->sock, POLLIN | POLLPRI | POLLNVAL, data, process_auth); if (res < 0) { qb_util_log(LOG_DEBUG, "Failed to arrange for AUTH for fd (%d)", data->sock); close(sock); destroy_ipc_auth_data(data); } } 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_service *s = (struct qb_ipcs_service *)data; int32_t res; socklen_t addrlen = sizeof(struct sockaddr_un); if (revent & (POLLNVAL | POLLHUP | POLLERR)) { /* * handle shutdown more cleanly. */ return -1; } retry_accept: errno = 0; new_fd = accept(fd, (struct sockaddr *)&un_addr, &addrlen); if (new_fd == -1 && errno == EINTR) { goto retry_accept; } if (new_fd == -1 && errno == EBADF) { qb_util_perror(LOG_ERR, "Could not accept client connection from fd:%d", fd); return -1; } if (new_fd == -1) { qb_util_perror(LOG_ERR, "Could not accept client connection"); /* This is an error, but -1 would indicate disconnect * from the poll loop */ return 0; } res = qb_sys_fd_nonblock_cloexec_set(new_fd); if (res < 0) { close(new_fd); /* This is an error, but -1 would indicate disconnect * from the poll loop */ return 0; } qb_ipcs_uc_recv_and_auth(new_fd, s); return 0; } void remove_tempdir(const char *name) { #if defined(QB_LINUX) || defined(QB_CYGWIN) char dirname[PATH_MAX]; char *slash = strrchr(name, '/'); if (slash && slash - name < sizeof dirname) { memcpy(dirname, name, slash - name); dirname[slash - name] = '\0'; /* This gets called more than it needs to be really, so we don't check * the return code. It's more of a desperate attempt to clean up after ourself * in either the server or client. */ (void)rmdir(dirname); } #endif } diff --git a/lib/ipcc.c b/lib/ipcc.c index 4f35ea2..4471f1f 100644 --- a/lib/ipcc.c +++ b/lib/ipcc.c @@ -1,539 +1,539 @@ /* - * Copyright (C) 2010 Red Hat, Inc. + * Copyright 2010-2024 Red Hat, Inc. * * Author: Angus Salkeld * * 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 . */ #include "os_base.h" #include #include "ipc_int.h" #include "util_int.h" #include #include qb_ipcc_connection_t * qb_ipcc_connect(const char *name, size_t max_msg_size) { int32_t res; qb_ipcc_connection_t *c = NULL; struct qb_ipc_connection_response response; c = calloc(1, sizeof(struct qb_ipcc_connection)); if (c == NULL) { return NULL; } c->setup.max_msg_size = QB_MAX(max_msg_size, sizeof(struct qb_ipc_connection_response)); (void)strlcpy(c->name, name, NAME_MAX); res = qb_ipcc_us_setup_connect(c, &response); if (res < 0) { goto disconnect_and_cleanup; } qb_ipc_us_ready(&c->setup, NULL, -1, POLLIN); res = qb_ipcc_connect_continue(c); if (res != 0) { /* qb_ipcc_connect_continue() has cleaned up for us */ errno = -res; return NULL; } return c; disconnect_and_cleanup: if (c->setup.u.us.sock >= 0) { qb_ipcc_us_sock_close(c->setup.u.us.sock); } free(c->receive_buf); free(c); errno = -res; return NULL; } qb_ipcc_connection_t * qb_ipcc_connect_async(const char *name, size_t max_msg_size, int *connect_fd) { int32_t res; qb_ipcc_connection_t *c = NULL; struct qb_ipc_connection_response response; c = calloc(1, sizeof(struct qb_ipcc_connection)); if (c == NULL) { return NULL; } c->setup.max_msg_size = QB_MAX(max_msg_size, sizeof(struct qb_ipc_connection_response)); (void)strlcpy(c->name, name, NAME_MAX); res = qb_ipcc_us_setup_connect(c, &response); if (res < 0) { goto disconnect_and_cleanup; } *connect_fd = c->setup.u.us.sock; return c; disconnect_and_cleanup: if (c->setup.u.us.sock >= 0) { qb_ipcc_us_sock_close(c->setup.u.us.sock); } free(c->receive_buf); free(c); errno = -res; return NULL; } int qb_ipcc_connect_continue(struct qb_ipcc_connection * c) { struct qb_ipc_connection_response response; int32_t res; /* Finish up the authentication part */ res = qb_ipcc_setup_connect_continue(c, &response); if (res != 0) { goto disconnect_and_cleanup; } c->response.type = response.connection_type; c->request.type = response.connection_type; c->event.type = response.connection_type; c->setup.type = response.connection_type; 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; c->receive_buf = calloc(1, response.max_msg_size); c->fc_enable_max = 1; if (c->receive_buf == NULL) { res = -ENOMEM; goto disconnect_and_cleanup; } switch (c->request.type) { case QB_IPC_SHM: res = qb_ipcc_shm_connect(c, &response); break; case QB_IPC_SOCKET: res = qb_ipcc_us_connect(c, &response); break; case QB_IPC_POSIX_MQ: case QB_IPC_SYSV_MQ: res = -ENOTSUP; break; default: res = -EINVAL; break; } if (res != 0) { goto disconnect_and_cleanup; } c->is_connected = QB_TRUE; return 0; disconnect_and_cleanup: if (c->setup.u.us.sock >= 0) { qb_ipcc_us_sock_close(c->setup.u.us.sock); } free(c->receive_buf); free(c); errno = -res; return res; } static int32_t _check_connection_state_with(struct qb_ipcc_connection * c, int32_t res, struct qb_ipc_one_way * one_way, int32_t ms_timeout, int32_t events) { if (res >= 0) return res; if (qb_ipc_us_sock_error_is_disconnected(res)) { errno = -res; qb_util_perror(LOG_DEBUG, "interpreting result %d as a disconnect", res); c->is_connected = QB_FALSE; } if (res == -EAGAIN || res == -ETIMEDOUT) { int32_t res2; int32_t poll_ms = ms_timeout; if (res == -ETIMEDOUT) { poll_ms = 0; } res2 = qb_ipc_us_ready(one_way, &c->setup, poll_ms, events); if (qb_ipc_us_sock_error_is_disconnected(res2)) { errno = -res2; qb_util_perror(LOG_DEBUG, "%s %d %s", "interpreting result", res2, "(from socket) as a disconnect"); c->is_connected = QB_FALSE; res = res2; } else if (res != -ETIMEDOUT) { /* if the result we're checking against is a TIMEOUT error. * don't override that result with another error that does * not imply a disconnect */ res = res2; } } return res; } static int32_t _check_connection_state(struct qb_ipcc_connection * c, int32_t res) { if (res >= 0) return res; if (qb_ipc_us_sock_error_is_disconnected(res)) { errno = -res; qb_util_perror(LOG_DEBUG, "interpreting result %d as a disconnect", res); c->is_connected = QB_FALSE; } return res; } static struct qb_ipc_one_way * _event_sock_one_way_get(struct qb_ipcc_connection * c) { if (c->needs_sock_for_poll) { return &c->setup; } return &c->event; } static struct qb_ipc_one_way * _response_sock_one_way_get(struct qb_ipcc_connection * c) { if (c->needs_sock_for_poll) { return &c->setup; } return &c->response; } ssize_t qb_ipcc_send(struct qb_ipcc_connection * c, const void *msg_ptr, size_t msg_len) { ssize_t res; ssize_t res2; if (c == NULL) { return -EINVAL; } if (msg_len > c->request.max_msg_size) { return -EMSGSIZE; } if (c->funcs.fc_get) { res = c->funcs.fc_get(&c->request); if (res < 0) { return res; } else if (res > 0 && res <= c->fc_enable_max) { return -EAGAIN; } else { /* * we can transmit */ } } res = c->funcs.send(&c->request, msg_ptr, msg_len); if (res == msg_len && c->needs_sock_for_poll) { do { res2 = qb_ipc_us_send(&c->setup, msg_ptr, 1); } while (res2 == -EAGAIN); if (res2 == -EPIPE) { res2 = -ENOTCONN; } if (res2 != 1) { res = res2; } } return _check_connection_state(c, res); } int32_t qb_ipcc_fc_enable_max_set(struct qb_ipcc_connection * c, uint32_t max) { if (c == NULL || max > 2) { return -EINVAL; } c->fc_enable_max = max; return 0; } ssize_t qb_ipcc_sendv(struct qb_ipcc_connection * c, const struct iovec * iov, size_t iov_len) { int32_t total_size = 0; int32_t i; int32_t res; int32_t res2; for (i = 0; i < iov_len; i++) { total_size += iov[i].iov_len; } if (c == NULL) { return -EINVAL; } if (total_size > c->request.max_msg_size) { return -EMSGSIZE; } if (c->funcs.fc_get) { res = c->funcs.fc_get(&c->request); if (res < 0) { return res; } else if (res > 0 && res <= c->fc_enable_max) { return -EAGAIN; } else { /* * we can transmit */ } } res = c->funcs.sendv(&c->request, iov, iov_len); if (res > 0 && c->needs_sock_for_poll) { do { res2 = qb_ipc_us_send(&c->setup, &res, 1); } while (res2 == -EAGAIN); if (res2 == -EPIPE) { res2 = -ENOTCONN; } if (res2 != 1) { res = res2; } } return _check_connection_state(c, res); } ssize_t qb_ipcc_recv(struct qb_ipcc_connection * c, void *msg_ptr, size_t msg_len, int32_t ms_timeout) { int32_t res = 0; int32_t connect_res = 0; if (c == NULL) { return -EINVAL; } res = c->funcs.recv(&c->response, msg_ptr, msg_len, ms_timeout); if (res >= 0) { return res; } /* if we didn't get a msg, check connection state */ connect_res = _check_connection_state_with(c, res, _response_sock_one_way_get(c), ms_timeout, POLLIN); /* only report the connection state check result if an error is returned. */ if (connect_res < 0) { return connect_res; } return res; } ssize_t qb_ipcc_sendv_recv(qb_ipcc_connection_t * c, const struct iovec * iov, uint32_t iov_len, void *res_msg, size_t res_len, int32_t ms_timeout) { ssize_t res = 0; int32_t timeout_now; int32_t timeout_rem = ms_timeout; if (c == NULL) { return -EINVAL; } if (c->funcs.fc_get) { res = c->funcs.fc_get(&c->request); if (res < 0) { return res; } else if (res > 0 && res <= c->fc_enable_max) { return -EAGAIN; } else { /* * we can transmit */ } } res = qb_ipcc_sendv(c, iov, iov_len); if (res < 0) { return res; } do { /* following is a liveness-driven interleaving (for cases the server side failed/exited) */ if (timeout_rem > QB_IPC_MAX_WAIT_MS || ms_timeout == -1) { timeout_now = QB_IPC_MAX_WAIT_MS; } else { timeout_now = timeout_rem; } res = qb_ipcc_recv(c, res_msg, res_len, timeout_now); if (res == -ETIMEDOUT) { if (ms_timeout < 0) { res = -EAGAIN; } else { timeout_rem -= timeout_now; if (timeout_rem > 0) { res = -EAGAIN; } } } else if (res < 0 && res != -EAGAIN) { errno = -res; qb_util_perror(LOG_DEBUG, "qb_ipcc_recv %d timeout:(%d/%d)", res, timeout_now, timeout_rem); } } while (res == -EAGAIN && c->is_connected); return res; } int32_t qb_ipcc_fd_get(struct qb_ipcc_connection * c, int32_t * fd) { if (c == NULL) { return -EINVAL; } if (c->event.type == QB_IPC_SOCKET) { *fd = c->event.u.us.sock; } else { *fd = c->setup.u.us.sock; } return 0; } int32_t qb_ipcc_auth_get(struct qb_ipcc_connection * c, pid_t *pid, uid_t *uid, gid_t *gid) { if (c == NULL) { return -EINVAL; } if (pid) { *pid = c->server_pid; } if (uid) { *uid = c->euid; } if (gid) { *gid = c->egid; } return 0; } ssize_t qb_ipcc_event_recv(struct qb_ipcc_connection * c, void *msg_pt, size_t msg_len, int32_t ms_timeout) { char one_byte = 1; int32_t res; ssize_t size; if (c == NULL) { return -EINVAL; } res = _check_connection_state_with(c, -EAGAIN, _event_sock_one_way_get(c), ms_timeout, POLLIN); if (res < 0) { return res; } size = c->funcs.recv(&c->event, msg_pt, msg_len, ms_timeout); if (size > 0 && c->needs_sock_for_poll) { res = qb_ipc_us_recv(&c->setup, &one_byte, 1, -1); if (res != 1) { size = res; } } return _check_connection_state(c, size); } void qb_ipcc_disconnect(struct qb_ipcc_connection *c) { struct qb_ipc_one_way *ow = NULL; - qb_util_log(LOG_DEBUG, "%s()", __func__); + qb_util_log(LOG_TRACE, "%s(%s)", __func__, (c == NULL)? "NULL" : ""); if (c == NULL) { return; } ow = _event_sock_one_way_get(c); (void)_check_connection_state_with(c, -EAGAIN, ow, 0, POLLIN); if (c->funcs.disconnect) { c->funcs.disconnect(c); } free(c->receive_buf); free(c); } void qb_ipcc_context_set(struct qb_ipcc_connection *c, void *context) { if (c == NULL) { return; } c->context = context; } void *qb_ipcc_context_get(struct qb_ipcc_connection *c) { if (c == NULL) { return NULL; } return c->context; } int32_t qb_ipcc_is_connected(qb_ipcc_connection_t *c) { struct qb_ipc_one_way *ow; if (c == NULL) { return QB_FALSE; } ow = _response_sock_one_way_get(c); (void)_check_connection_state_with(c, -EAGAIN, ow, 0, POLLIN); return c->is_connected; } int32_t qb_ipcc_get_buffer_size(qb_ipcc_connection_t * c) { if (c == NULL) { return -EINVAL; } return c->event.max_msg_size; } diff --git a/lib/ringbuffer.c b/lib/ringbuffer.c index b769edf..96a05f1 100644 --- a/lib/ringbuffer.c +++ b/lib/ringbuffer.c @@ -1,950 +1,950 @@ /* - * Copyright (C) 2010-2011 Red Hat, Inc. + * Copyright 2010-2024 Red Hat, Inc. * * Author: Angus Salkeld * * 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 . */ #include "ringbuffer_int.h" #include #include "atomic_int.h" #define QB_RB_FILE_HEADER_VERSION 1 /* * #define CRAZY_DEBUG_PRINTFS 1 */ #ifdef CRAZY_DEBUG_PRINTFS #define DEBUG_PRINTF(format, args...) \ do { \ printf(format, ##args); \ } while(0) #else #define DEBUG_PRINTF(format, args...) #endif /* CRAZY_DEBUG_PRINTFS */ /* * move the write pointer to the next 128 byte boundary * write_pt goes in 4 bytes (sizeof(uint32_t)) * #define USE_CACHE_LINE_ALIGNMENT 1 */ #ifdef USE_CACHE_LINE_ALIGNMENT #define QB_CACHE_LINE_SIZE 128 #define QB_CACHE_LINE_WORDS (QB_CACHE_LINE_SIZE/sizeof(uint32_t)) #define idx_cache_line_step(idx) \ do { \ if (idx % QB_CACHE_LINE_WORDS) { \ idx += (QB_CACHE_LINE_WORDS - (idx % QB_CACHE_LINE_WORDS)); \ } \ if (idx > (rb->shared_hdr->word_size - 1)) { \ idx = ((idx) % (rb->shared_hdr->word_size)); \ } \ } while (0) #else #define QB_CACHE_LINE_SIZE 0 #define QB_CACHE_LINE_WORDS 0 #define idx_cache_line_step(idx) \ do { \ if (idx > (rb->shared_hdr->word_size - 1)) { \ idx = ((idx) % (rb->shared_hdr->word_size)); \ } \ } while (0) #endif /* the chunk header is two words * 1) the chunk data size * 2) the magic number */ #define QB_RB_CHUNK_HEADER_WORDS 2 #define QB_RB_CHUNK_HEADER_SIZE (sizeof(uint32_t) * QB_RB_CHUNK_HEADER_WORDS) /* * margin is the gap we leave when checking to see if we have enough * space for a new chunk. * So: * qb_rb_space_free() >= QB_RB_CHUNK_MARGIN + new data chunk * The extra word size is to allow for non word sized data chunks. * QB_CACHE_LINE_WORDS is to make sure we have space to align the * chunk. */ #define QB_RB_WORD_ALIGN 1 #define QB_RB_CHUNK_MARGIN (sizeof(uint32_t) * (QB_RB_CHUNK_HEADER_WORDS +\ QB_RB_WORD_ALIGN +\ QB_CACHE_LINE_WORDS)) #define QB_RB_CHUNK_MAGIC 0xA1A1A1A1 #define QB_RB_CHUNK_MAGIC_DEAD 0xD0D0D0D0 #define QB_RB_CHUNK_MAGIC_ALLOC 0xA110CED0 #define QB_RB_CHUNK_SIZE_GET(rb, pointer) rb->shared_data[pointer] #define QB_RB_CHUNK_MAGIC_GET(rb, pointer) \ qb_atomic_int_get_ex((int32_t*)&rb->shared_data[(pointer + 1) % rb->shared_hdr->word_size], \ QB_ATOMIC_ACQUIRE) #define QB_RB_CHUNK_MAGIC_SET(rb, pointer, new_val) \ qb_atomic_int_set_ex((int32_t*)&rb->shared_data[(pointer + 1) % rb->shared_hdr->word_size], \ new_val, QB_ATOMIC_RELEASE) #define QB_RB_CHUNK_DATA_GET(rb, pointer) \ &rb->shared_data[(pointer + QB_RB_CHUNK_HEADER_WORDS) % rb->shared_hdr->word_size] #define QB_MAGIC_ASSERT(_ptr_) \ do { \ uint32_t chunk_magic = QB_RB_CHUNK_MAGIC_GET(rb, _ptr_); \ if (chunk_magic != QB_RB_CHUNK_MAGIC) print_header(rb); \ assert(chunk_magic == QB_RB_CHUNK_MAGIC); \ } while (0) #define idx_step(idx) \ do { \ if (idx > (rb->shared_hdr->word_size - 1)) { \ idx = ((idx) % (rb->shared_hdr->word_size)); \ } \ } while (0) static void print_header(struct qb_ringbuffer_s * rb); static int _rb_chunk_reclaim(struct qb_ringbuffer_s * rb); qb_ringbuffer_t * qb_rb_open(const char *name, size_t size, uint32_t flags, size_t shared_user_data_size) { return qb_rb_open_2(name, size, flags, shared_user_data_size, NULL); } qb_ringbuffer_t * qb_rb_open_2(const char *name, size_t size, uint32_t flags, size_t shared_user_data_size, struct qb_rb_notifier *notifiers) { struct qb_ringbuffer_s *rb; size_t real_size; size_t shared_size; char path[PATH_MAX]; int32_t fd_hdr; int32_t fd_data; uint32_t file_flags = O_RDWR; char filename[PATH_MAX]; int32_t error = 0; void *shm_addr; long page_size = sysconf(_SC_PAGESIZE); #ifdef QB_ARCH_HPPA page_size = QB_MAX(page_size, 0x00400000); /* align to page colour */ #elif defined(QB_FORCE_SHM_ALIGN) page_size = QB_MAX(page_size, 16 * 1024); #endif /* QB_FORCE_SHM_ALIGN */ /* The user of this api expects the 'size' parameter passed into this function * to be reflective of the max size single write we can do to the * ringbuffer. This means we have to add both the 'margin' space used * to calculate if there is enough space for a new chunk as well as the '+1' that * prevents overlap of the read/write pointers */ size += QB_RB_CHUNK_MARGIN + 1; real_size = QB_ROUNDUP(size, page_size); shared_size = sizeof(struct qb_ringbuffer_shared_s) + shared_user_data_size; if (flags & QB_RB_FLAG_CREATE) { file_flags |= O_CREAT | O_TRUNC | O_EXCL; } rb = calloc(1, sizeof(struct qb_ringbuffer_s)); if (rb == NULL) { return NULL; } /* * Create a shared_hdr memory segment for the header. */ snprintf(filename, PATH_MAX, "%s-header", name); fd_hdr = qb_sys_mmap_file_open(path, filename, shared_size, file_flags); if (fd_hdr < 0) { error = fd_hdr; qb_util_log(LOG_ERR, "couldn't create file for mmap"); goto cleanup_hdr; } rb->shared_hdr = mmap(0, shared_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd_hdr, 0); if (rb->shared_hdr == MAP_FAILED) { error = -errno; qb_util_log(LOG_ERR, "couldn't create mmap for header"); goto cleanup_hdr; } qb_atomic_init(); rb->flags = flags; /* * create the semaphore */ if (flags & QB_RB_FLAG_CREATE) { rb->shared_data = NULL; /* rb->shared_hdr->word_size tracks data by ints and not bytes/chars. */ rb->shared_hdr->word_size = real_size / sizeof(uint32_t); rb->shared_hdr->write_pt = 0; rb->shared_hdr->read_pt = 0; (void)strlcpy(rb->shared_hdr->hdr_path, path, PATH_MAX); } if (notifiers && notifiers->post_fn) { error = 0; memcpy(&rb->notifier, notifiers, sizeof(struct qb_rb_notifier)); } else { error = qb_rb_sem_create(rb, flags); } if (error < 0) { errno = -error; qb_util_perror(LOG_ERR, "couldn't create a semaphore"); goto cleanup_hdr; } /* Create the shared_data memory segment for the actual ringbuffer. * They have to be separate. */ if (flags & QB_RB_FLAG_CREATE) { snprintf(filename, PATH_MAX, "%s-data", name); fd_data = qb_sys_mmap_file_open(path, filename, real_size, file_flags); (void)strlcpy(rb->shared_hdr->data_path, path, PATH_MAX); } else { fd_data = qb_sys_mmap_file_open(path, rb->shared_hdr->data_path, real_size, file_flags); } if (fd_data < 0) { error = fd_data; qb_util_log(LOG_ERR, "couldn't create file for mmap"); goto cleanup_hdr; } - qb_util_log(LOG_DEBUG, + qb_util_log(LOG_TRACE, "shm size:%ld; real_size:%ld; rb->word_size:%d", size, real_size, rb->shared_hdr->word_size); /* this function closes fd_data */ error = qb_sys_circular_mmap(fd_data, &shm_addr, real_size); rb->shared_data = shm_addr; if (error != 0) { qb_util_log(LOG_ERR, "couldn't create circular mmap on %s", rb->shared_hdr->data_path); goto cleanup_data; } if (flags & QB_RB_FLAG_CREATE) { memset(rb->shared_data, 0, real_size); rb->shared_data[rb->shared_hdr->word_size] = 5; rb->shared_hdr->ref_count = 1; } else { qb_atomic_int_inc(&rb->shared_hdr->ref_count); } close(fd_hdr); return rb; cleanup_data: if (flags & QB_RB_FLAG_CREATE) { unlink(rb->shared_hdr->data_path); } cleanup_hdr: if (fd_hdr >= 0) { close(fd_hdr); } if (rb && (rb->shared_hdr != MAP_FAILED) && (flags & QB_RB_FLAG_CREATE)) { unlink(rb->shared_hdr->hdr_path); if (rb->notifier.destroy_fn) { (void)rb->notifier.destroy_fn(rb->notifier.instance); } } if (rb && (rb->shared_hdr != MAP_FAILED)) { munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s)); } free(rb); errno = -error; return NULL; } void qb_rb_close(struct qb_ringbuffer_s * rb) { if (rb == NULL) { return; } qb_enter(); (void)qb_atomic_int_dec_and_test(&rb->shared_hdr->ref_count); (void)qb_rb_close_helper(rb, rb->flags & QB_RB_FLAG_CREATE, QB_FALSE); } void qb_rb_force_close(struct qb_ringbuffer_s * rb) { if (rb == NULL) { return; } qb_enter(); qb_atomic_int_set(&rb->shared_hdr->ref_count, -1); (void)qb_rb_close_helper(rb, QB_TRUE, QB_TRUE); } char * qb_rb_name_get(struct qb_ringbuffer_s * rb) { if (rb == NULL) { return NULL; } return rb->shared_hdr->hdr_path; } void * qb_rb_shared_user_data_get(struct qb_ringbuffer_s * rb) { if (rb == NULL) { return NULL; } return rb->shared_hdr->user_data; } int32_t qb_rb_refcount_get(struct qb_ringbuffer_s * rb) { if (rb == NULL) { return -EINVAL; } return qb_atomic_int_get(&rb->shared_hdr->ref_count); } ssize_t qb_rb_space_free(struct qb_ringbuffer_s * rb) { uint32_t write_size; uint32_t read_size; size_t space_free = 0; if (rb == NULL) { return -EINVAL; } if (rb->notifier.space_used_fn) { return (rb->shared_hdr->word_size * sizeof(uint32_t)) - rb->notifier.space_used_fn(rb->notifier.instance); } write_size = rb->shared_hdr->write_pt; read_size = rb->shared_hdr->read_pt; if (write_size > read_size) { space_free = (read_size - write_size + rb->shared_hdr->word_size) - 1; } else if (write_size < read_size) { space_free = (read_size - write_size) - 1; } else { if (rb->notifier.q_len_fn && rb->notifier.q_len_fn(rb->notifier.instance) > 0) { space_free = 0; } else { space_free = rb->shared_hdr->word_size; } } /* word -> bytes */ return (space_free * sizeof(uint32_t)); } ssize_t qb_rb_space_used(struct qb_ringbuffer_s * rb) { uint32_t write_size; uint32_t read_size; size_t space_used; if (rb == NULL) { return -EINVAL; } if (rb->notifier.space_used_fn) { return rb->notifier.space_used_fn(rb->notifier.instance); } write_size = rb->shared_hdr->write_pt; read_size = rb->shared_hdr->read_pt; if (write_size > read_size) { space_used = write_size - read_size; } else if (write_size < read_size) { space_used = (write_size - read_size + rb->shared_hdr->word_size) - 1; } else { space_used = 0; } /* word -> bytes */ return (space_used * sizeof(uint32_t)); } ssize_t qb_rb_chunks_used(struct qb_ringbuffer_s *rb) { if (rb == NULL) { return -EINVAL; } if (rb->notifier.q_len_fn) { return rb->notifier.q_len_fn(rb->notifier.instance); } return -ENOTSUP; } void * qb_rb_chunk_alloc(struct qb_ringbuffer_s * rb, size_t len) { uint32_t write_pt; if (rb == NULL) { errno = EINVAL; return NULL; } /* * Reclaim data if we are over writing and we need space */ if (rb->flags & QB_RB_FLAG_OVERWRITE) { while (qb_rb_space_free(rb) < (len + QB_RB_CHUNK_MARGIN)) { int rc = _rb_chunk_reclaim(rb); if (rc != 0) { return NULL; /* errno already set */ } } } else { if (qb_rb_space_free(rb) < (len + QB_RB_CHUNK_MARGIN)) { errno = EAGAIN; return NULL; } } write_pt = rb->shared_hdr->write_pt; /* * insert the chunk header */ rb->shared_data[write_pt] = 0; QB_RB_CHUNK_MAGIC_SET(rb, write_pt, QB_RB_CHUNK_MAGIC_ALLOC); /* * return a pointer to the beginning of the chunk data */ return (void *)QB_RB_CHUNK_DATA_GET(rb, write_pt); } static uint32_t qb_rb_chunk_step(struct qb_ringbuffer_s * rb, uint32_t pointer) { uint32_t chunk_size = QB_RB_CHUNK_SIZE_GET(rb, pointer); /* * skip over the chunk header */ pointer += QB_RB_CHUNK_HEADER_WORDS; /* * skip over the user's data. */ pointer += (chunk_size / sizeof(uint32_t)); /* make allowance for non-word sizes */ if ((chunk_size % (sizeof(uint32_t) * QB_RB_WORD_ALIGN)) != 0) { pointer++; } idx_cache_line_step(pointer); return pointer; } int32_t qb_rb_chunk_commit(struct qb_ringbuffer_s * rb, size_t len) { uint32_t old_write_pt; if (rb == NULL) { return -EINVAL; } /* * commit the magic & chunk_size */ old_write_pt = rb->shared_hdr->write_pt; rb->shared_data[old_write_pt] = len; /* * commit the new write pointer */ rb->shared_hdr->write_pt = qb_rb_chunk_step(rb, old_write_pt); QB_RB_CHUNK_MAGIC_SET(rb, old_write_pt, QB_RB_CHUNK_MAGIC); DEBUG_PRINTF("commit [%zd] read: %u, write: %u -> %u (%u)\n", (rb->notifier.q_len_fn ? rb->notifier.q_len_fn(rb->notifier.instance) : 0), rb->shared_hdr->read_pt, old_write_pt, rb->shared_hdr->write_pt, rb->shared_hdr->word_size); /* * post the notification to the reader */ if (rb->notifier.post_fn) { return rb->notifier.post_fn(rb->notifier.instance, len); } return 0; } ssize_t qb_rb_chunk_write(struct qb_ringbuffer_s * rb, const void *data, size_t len) { char *dest = qb_rb_chunk_alloc(rb, len); int32_t res = 0; if (rb == NULL) { return -EINVAL; } if (dest == NULL) { return -errno; } memcpy(dest, data, len); res = qb_rb_chunk_commit(rb, len); if (res < 0) { return res; } return len; } static int _rb_chunk_reclaim(struct qb_ringbuffer_s * rb) { uint32_t old_read_pt; uint32_t new_read_pt; uint32_t old_chunk_size; uint32_t chunk_magic; int rc = 0; old_read_pt = rb->shared_hdr->read_pt; chunk_magic = QB_RB_CHUNK_MAGIC_GET(rb, old_read_pt); if (chunk_magic != QB_RB_CHUNK_MAGIC) { errno = EINVAL; return -errno; } old_chunk_size = QB_RB_CHUNK_SIZE_GET(rb, old_read_pt); new_read_pt = qb_rb_chunk_step(rb, old_read_pt); /* * clear the header */ rb->shared_data[old_read_pt] = 0; QB_RB_CHUNK_MAGIC_SET(rb, old_read_pt, QB_RB_CHUNK_MAGIC_DEAD); /* * set the new read pointer after clearing the header * to prevent a situation where a fast writer will write their * new chunk between setting the new read pointer and clearing the * header. */ rb->shared_hdr->read_pt = new_read_pt; if (rb->notifier.reclaim_fn) { rc = rb->notifier.reclaim_fn(rb->notifier.instance, old_chunk_size); if (rc < 0) { errno = -rc; qb_util_perror(LOG_WARNING, "reclaim_fn"); } } DEBUG_PRINTF("reclaim [%zd]: read: %u -> %u, write: %u\n", (rb->notifier.q_len_fn ? rb->notifier.q_len_fn(rb->notifier.instance) : 0), old_read_pt, rb->shared_hdr->read_pt, rb->shared_hdr->write_pt); return rc; } void qb_rb_chunk_reclaim(struct qb_ringbuffer_s * rb) { if (rb == NULL) { return; } _rb_chunk_reclaim(rb); } ssize_t qb_rb_chunk_peek(struct qb_ringbuffer_s * rb, void **data_out, int32_t timeout) { uint32_t read_pt; uint32_t chunk_size; uint32_t chunk_magic; int32_t res = 0; if (rb == NULL) { return -EINVAL; } if (rb->notifier.timedwait_fn) { res = rb->notifier.timedwait_fn(rb->notifier.instance, timeout); } if (res < 0 && res != -EIDRM) { if (res == -ETIMEDOUT) { return 0; } else { errno = -res; qb_util_perror(LOG_ERR, "sem_timedwait"); } return res; } read_pt = rb->shared_hdr->read_pt; chunk_magic = QB_RB_CHUNK_MAGIC_GET(rb, read_pt); if (chunk_magic != QB_RB_CHUNK_MAGIC) { if (rb->notifier.post_fn) { (void)rb->notifier.post_fn(rb->notifier.instance, res); } #ifdef EBADMSG return -EBADMSG; #else return -EINVAL; #endif } chunk_size = QB_RB_CHUNK_SIZE_GET(rb, read_pt); *data_out = QB_RB_CHUNK_DATA_GET(rb, read_pt); return chunk_size; } ssize_t qb_rb_chunk_read(struct qb_ringbuffer_s * rb, void *data_out, size_t len, int32_t timeout) { uint32_t read_pt; uint32_t chunk_size; uint32_t chunk_magic; int32_t res = 0; if (rb == NULL) { return -EINVAL; } if (rb->notifier.timedwait_fn) { res = rb->notifier.timedwait_fn(rb->notifier.instance, timeout); } if (res < 0 && res != -EIDRM) { if (res != -ETIMEDOUT) { errno = -res; qb_util_perror(LOG_ERR, "sem_timedwait"); } return res; } read_pt = rb->shared_hdr->read_pt; chunk_magic = QB_RB_CHUNK_MAGIC_GET(rb, read_pt); if (chunk_magic != QB_RB_CHUNK_MAGIC) { if (rb->notifier.timedwait_fn == NULL) { return -ETIMEDOUT; } else { (void)rb->notifier.post_fn(rb->notifier.instance, res); #ifdef EBADMSG return -EBADMSG; #else return -EINVAL; #endif } } chunk_size = QB_RB_CHUNK_SIZE_GET(rb, read_pt); if (len < chunk_size) { qb_util_log(LOG_ERR, "trying to recv chunk of size %d but %d available", len, chunk_size); if (rb->notifier.post_fn) { (void)rb->notifier.post_fn(rb->notifier.instance, chunk_size); } return -ENOBUFS; } memcpy(data_out, QB_RB_CHUNK_DATA_GET(rb, read_pt), chunk_size); _rb_chunk_reclaim(rb); return chunk_size; } static void print_header(struct qb_ringbuffer_s * rb) { printf("Ringbuffer: \n"); if (rb->flags & QB_RB_FLAG_OVERWRITE) { printf(" ->OVERWRITE\n"); } else { printf(" ->NORMAL\n"); } #ifndef S_SPLINT_S printf(" ->write_pt [%" PRIu32 "]\n", rb->shared_hdr->write_pt); printf(" ->read_pt [%" PRIu32 "]\n", rb->shared_hdr->read_pt); printf(" ->size [%" PRIu32 " words]\n", rb->shared_hdr->word_size); printf(" =>free [%zd bytes]\n", qb_rb_space_free(rb)); printf(" =>used [%zd bytes]\n", qb_rb_space_used(rb)); #endif /* S_SPLINT_S */ } /* * FILE HEADER ORDER * 1. word_size * 2. write_pt * 3. read_pt * 4. version * 5. header_hash * * 6. data */ ssize_t qb_rb_write_to_file(struct qb_ringbuffer_s * rb, int32_t fd) { ssize_t result; ssize_t written_size = 0; uint32_t hash = 0; uint32_t version = QB_RB_FILE_HEADER_VERSION; if (rb == NULL) { return -EINVAL; } print_header(rb); /* * 1. word_size */ result = write(fd, &rb->shared_hdr->word_size, sizeof(uint32_t)); if (result != sizeof(uint32_t)) { return -errno; } written_size += result; /* * 2. 3. store the read & write pointers */ result = write(fd, (void *)&rb->shared_hdr->write_pt, sizeof(uint32_t)); if (result != sizeof(uint32_t)) { return -errno; } written_size += result; result = write(fd, (void *)&rb->shared_hdr->read_pt, sizeof(uint32_t)); if (result != sizeof(uint32_t)) { return -errno; } written_size += result; /* * 4. version used */ result = write(fd, &version, sizeof(uint32_t)); if (result != sizeof(uint32_t)) { return -errno; } written_size += result; /* * 5. hash helps us verify header is not corrupted on file read */ hash = rb->shared_hdr->word_size + rb->shared_hdr->write_pt + rb->shared_hdr->read_pt + QB_RB_FILE_HEADER_VERSION; result = write(fd, &hash, sizeof(uint32_t)); if (result != sizeof(uint32_t)) { return -errno; } written_size += result; result = write(fd, rb->shared_data, rb->shared_hdr->word_size * sizeof(uint32_t)); if (result != rb->shared_hdr->word_size * sizeof(uint32_t)) { return -errno; } written_size += result; qb_util_log(LOG_DEBUG, " writing total of: %zd\n", written_size); return written_size; } qb_ringbuffer_t * qb_rb_create_from_file(int32_t fd, uint32_t flags) { ssize_t n_read; size_t n_required; size_t total_read = 0; uint32_t read_pt; uint32_t write_pt; struct qb_ringbuffer_s *rb; uint32_t word_size = 0; uint32_t version = 0; uint32_t hash = 0; uint32_t calculated_hash = 0; struct stat st; if (fd < 0) { return NULL; } if (fstat(fd, &st)) { qb_util_perror(LOG_ERR, "Unable to stat blackbox file"); return NULL; } /* * 1. word size */ n_required = sizeof(uint32_t); n_read = read(fd, &word_size, n_required); if (n_read != n_required) { qb_util_perror(LOG_ERR, "Unable to read blackbox file header"); return NULL; } total_read += n_read; if (word_size > (st.st_size / sizeof(uint32_t))) { qb_util_perror(LOG_ERR, "Invalid word size read from blackbox header"); return NULL; } /* * 2. 3. read & write pointers */ n_read = read(fd, &write_pt, sizeof(uint32_t)); assert(n_read == sizeof(uint32_t)); total_read += n_read; n_read = read(fd, &read_pt, sizeof(uint32_t)); assert(n_read == sizeof(uint32_t)); total_read += n_read; if (write_pt > st.st_size || read_pt > st.st_size) { qb_util_perror(LOG_ERR, "Invalid pointers read from blackbox header"); return NULL; } /* * 4. version */ n_required = sizeof(uint32_t); n_read = read(fd, &version, n_required); if (n_read != n_required) { qb_util_perror(LOG_ERR, "Unable to read blackbox file header"); return NULL; } total_read += n_read; /* * 5. Hash */ n_required = sizeof(uint32_t); n_read = read(fd, &hash, n_required); if (n_read != n_required) { qb_util_perror(LOG_ERR, "Unable to read blackbox file header"); return NULL; } total_read += n_read; calculated_hash = word_size + write_pt + read_pt + version; if (hash != calculated_hash) { qb_util_log(LOG_ERR, "Corrupt blackbox: File header hash (%d) does not match calculated hash (%d)", hash, calculated_hash); return NULL; } else if (version != QB_RB_FILE_HEADER_VERSION) { qb_util_log(LOG_ERR, "Wrong file header version. Expected %d got %d", QB_RB_FILE_HEADER_VERSION, version); return NULL; } /* * 6. data */ n_required = (word_size * sizeof(uint32_t)); /* * qb_rb_open adds QB_RB_CHUNK_MARGIN + 1 to the requested size. */ rb = qb_rb_open("create_from_file", n_required - (QB_RB_CHUNK_MARGIN + 1), QB_RB_FLAG_CREATE | QB_RB_FLAG_NO_SEMAPHORE, 0); if (rb == NULL) { return NULL; } rb->shared_hdr->read_pt = read_pt; rb->shared_hdr->write_pt = write_pt; n_read = read(fd, rb->shared_data, n_required); if (n_read < 0) { qb_util_perror(LOG_ERR, "Unable to read blackbox file data"); goto cleanup_fail; } total_read += n_read; if (n_read != n_required) { qb_util_log(LOG_WARNING, "read %zd bytes, but expected %zu", n_read, n_required); goto cleanup_fail; } qb_util_log(LOG_DEBUG, "read total of: %zd", total_read); print_header(rb); return rb; cleanup_fail: qb_rb_close(rb); return NULL; } int32_t qb_rb_chown(struct qb_ringbuffer_s * rb, uid_t owner, gid_t group) { int32_t res; if (rb == NULL) { return -EINVAL; } res = chown(rb->shared_hdr->data_path, owner, group); if (res < 0 && errno != EPERM) { return -errno; } res = chown(rb->shared_hdr->hdr_path, owner, group); if (res < 0 && errno != EPERM) { return -errno; } return 0; } int32_t qb_rb_chmod(qb_ringbuffer_t * rb, mode_t mode) { int32_t res; if (rb == NULL) { return -EINVAL; } res = chmod(rb->shared_hdr->data_path, mode); if (res < 0) { return -errno; } res = chmod(rb->shared_hdr->hdr_path, mode); if (res < 0) { return -errno; } return 0; } diff --git a/lib/ringbuffer_helper.c b/lib/ringbuffer_helper.c index 11795ff..e47b7d5 100644 --- a/lib/ringbuffer_helper.c +++ b/lib/ringbuffer_helper.c @@ -1,402 +1,402 @@ /* * Copyright (C) 2010 Red Hat, Inc. * * Author: Angus Salkeld * * 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 . */ #include "ringbuffer_int.h" #include static int32_t my_posix_sem_timedwait(void * instance, int32_t ms_timeout) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; struct timespec ts_timeout; int32_t res; if (ms_timeout > 0) { qb_util_timespec_from_epoch_get(&ts_timeout); qb_timespec_add_ms(&ts_timeout, ms_timeout); } sem_wait_again: if (ms_timeout > 0) { res = rpl_sem_timedwait(&rb->shared_hdr->posix_sem, &ts_timeout); } else if (ms_timeout == 0) { res = rpl_sem_trywait(&rb->shared_hdr->posix_sem); } else { res = rpl_sem_wait(&rb->shared_hdr->posix_sem); } if (res == -1) { switch (errno) { case EINTR: goto sem_wait_again; break; case EAGAIN: res = -ETIMEDOUT; break; case ETIMEDOUT: res = -errno; break; default: res = -errno; qb_util_perror(LOG_ERR, "error waiting for semaphore"); break; } } return res; } static int32_t my_posix_sem_post(void * instance, size_t msg_size) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; if (rpl_sem_post(&rb->shared_hdr->posix_sem) < 0) { return -errno; } else { return 0; } } static ssize_t my_posix_getvalue_fn(void * instance) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; int val; if (rpl_sem_getvalue(&rb->shared_hdr->posix_sem, &val) < 0) { return -errno; } else { return val; } } static int32_t my_posix_sem_destroy(void * instance) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; qb_enter(); if (rpl_sem_destroy(&rb->shared_hdr->posix_sem) == -1) { return -errno; } else { return 0; } } static int32_t my_posix_sem_create(void * instance, uint32_t flags) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; int32_t pshared = QB_FALSE; if (flags & QB_RB_FLAG_SHARED_PROCESS) { if ((flags & QB_RB_FLAG_CREATE) == 0) { return 0; } pshared = QB_TRUE; } if (rpl_sem_init(&rb->shared_hdr->posix_sem, pshared, 0) == -1) { return -errno; } else { return 0; } } static int32_t my_sysv_sem_timedwait(void * instance, int32_t ms_timeout) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; struct sembuf sops[1]; int32_t res = 0; #ifdef HAVE_SEMTIMEDOP struct timespec ts_timeout; struct timespec *ts_pt; if (ms_timeout >= 0) { /* * Note: sem_timedwait takes an absolute time where as semtimedop * takes a relative time. */ ts_timeout.tv_sec = 0; ts_timeout.tv_nsec = 0; qb_timespec_add_ms(&ts_timeout, ms_timeout); ts_pt = &ts_timeout; } else { ts_pt = NULL; } #endif /* HAVE_SEMTIMEDOP */ /* * wait for sem post. */ sops[0].sem_num = 0; sops[0].sem_op = -1; #ifdef HAVE_SEMTIMEDOP sops[0].sem_flg = 0; #else sops[0].sem_flg = IPC_NOWAIT; #endif /* HAVE_SEMTIMEDOP */ semop_again: #ifdef HAVE_SEMTIMEDOP if (semtimedop(rb->sem_id, sops, 1, ts_pt) == -1) #else if (semop(rb->sem_id, sops, 1) == -1) #endif /* HAVE_SEMTIMEDOP */ { if (errno == EINTR) { goto semop_again; } else if (errno == EAGAIN) { /* make consistent with sem_timedwait */ res = -ETIMEDOUT; } else { res = -errno; qb_util_perror(LOG_ERR, "error waiting for semaphore"); } return res; } return 0; } static int32_t my_sysv_sem_post(void * instance, size_t msg_size) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; struct sembuf sops[1]; if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) { return 0; } sops[0].sem_num = 0; sops[0].sem_op = 1; sops[0].sem_flg = 0; semop_again: if (semop(rb->sem_id, sops, 1) == -1) { if (errno == EINTR) { goto semop_again; } else { qb_util_perror(LOG_ERR, "could not increment semaphore"); } return -errno; } return 0; } static ssize_t my_sysv_getvalue_fn(void * instance) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; ssize_t res = semctl(rb->sem_id, 0, GETVAL, 0); if (res == -1) { return -errno; } return res; } static int32_t my_sysv_sem_destroy(void * instance) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; if (semctl(rb->sem_id, 0, IPC_RMID, 0) == -1) { return -errno; } else { return 0; } } static int32_t my_sysv_sem_create(void * instance, uint32_t flags) { struct qb_ringbuffer_s *rb = (struct qb_ringbuffer_s *)instance; union semun options; int32_t res; key_t sem_key; sem_key = ftok(rb->shared_hdr->hdr_path, (rb->shared_hdr->word_size + 1)); if (sem_key == -1) { res = -errno; qb_util_perror(LOG_ERR, "couldn't get a sem id"); return res; } if (flags & QB_RB_FLAG_CREATE) { rb->sem_id = semget(sem_key, 1, IPC_CREAT | IPC_EXCL | 0600); if (rb->sem_id == -1) { res = -errno; qb_util_perror(LOG_ERR, "couldn't create a semaphore"); return res; } options.val = 0; res = semctl(rb->sem_id, 0, SETVAL, options); } else { rb->sem_id = semget(sem_key, 0, 0600); if (rb->sem_id == -1) { res = -errno; qb_util_perror(LOG_ERR, "couldn't get a sem id"); return res; } res = 0; } qb_util_log(LOG_DEBUG, "sem key:%d, id:%d, value:%d", (int)sem_key, rb->sem_id, semctl(rb->sem_id, 0, GETVAL, 0)); return res; } int32_t qb_rb_sem_create(struct qb_ringbuffer_s * rb, uint32_t flags) { int32_t rc; int32_t use_posix = QB_TRUE; if ((flags & QB_RB_FLAG_SHARED_PROCESS) && !(flags & QB_RB_FLAG_NO_SEMAPHORE)) { #if defined(HAVE_POSIX_PSHARED_SEMAPHORE) || \ defined(HAVE_RPL_PSHARED_SEMAPHORE) use_posix = QB_TRUE; #else #ifdef HAVE_SYSV_PSHARED_SEMAPHORE use_posix = QB_FALSE; #else return -ENOTSUP; #endif /* HAVE_SYSV_PSHARED_SEMAPHORE */ #endif /* HAVE_POSIX_PSHARED_SEMAPHORE */ } if (flags & QB_RB_FLAG_NO_SEMAPHORE) { rc = 0; rb->notifier.instance = NULL; rb->notifier.timedwait_fn = NULL; rb->notifier.post_fn = NULL; rb->notifier.q_len_fn = NULL; rb->notifier.space_used_fn = NULL; rb->notifier.destroy_fn = NULL; } else if (use_posix) { rc = my_posix_sem_create(rb, flags); rb->notifier.instance = rb; rb->notifier.timedwait_fn = my_posix_sem_timedwait; rb->notifier.post_fn = my_posix_sem_post; rb->notifier.q_len_fn = my_posix_getvalue_fn; rb->notifier.space_used_fn = NULL; rb->notifier.destroy_fn = my_posix_sem_destroy; } else { rc = my_sysv_sem_create(rb, flags); rb->notifier.instance = rb; rb->notifier.timedwait_fn = my_sysv_sem_timedwait; rb->notifier.post_fn = my_sysv_sem_post; rb->notifier.q_len_fn = my_sysv_getvalue_fn; rb->notifier.space_used_fn = NULL; rb->notifier.destroy_fn = my_sysv_sem_destroy; } return rc; } /* For qb_rb_close_helper, we need to open directory in read-only mode and with as lightweight + strict flags as available at given platform (O_PATH for the former, O_DIRECTORY for the latter); end result is available as RB_DIR_RO_FLAGS. */ #if defined(HAVE_OPENAT) && defined(HAVE_UNLINKAT) # ifndef O_DIRECTORY # define RB_DIR_RO_FLAGS1 O_RDONLY # else # define RB_DIR_RO_FLAGS1 O_RDONLY|O_DIRECTORY # endif # ifndef O_PATH # define RB_DIR_RO_FLAGS RB_DIR_RO_FLAGS1 # else # define RB_DIR_RO_FLAGS RB_DIR_RO_FLAGS1|O_PATH # endif int32_t qb_rb_close_helper(struct qb_ringbuffer_s * rb, int32_t unlink_it, int32_t truncate_fallback) { int32_t res = 0, res2 = 0; uint32_t word_size = rb->shared_hdr->word_size; char *hdr_path = rb->shared_hdr->hdr_path; if (unlink_it) { - qb_util_log(LOG_DEBUG, "Free'ing ringbuffer: %s", hdr_path); + qb_util_log(LOG_TRACE, "Free'ing ringbuffer: %s", hdr_path); if (rb->notifier.destroy_fn) { (void)rb->notifier.destroy_fn(rb->notifier.instance); } } else { - qb_util_log(LOG_DEBUG, "Closing ringbuffer: %s", hdr_path); + qb_util_log(LOG_TRACE, "Closing ringbuffer: %s", hdr_path); hdr_path = NULL; } if (unlink_it) { char *data_path = rb->shared_hdr->data_path; char *sep = strrchr(data_path, '/'); /* we could modify data_path in-situ, but that would segfault if we hadn't write permissions to the underlying mmap'd file */ char dir_path[PATH_MAX]; int dirfd; if (sep != NULL) { strncpy(dir_path, data_path, sep - data_path); dir_path[sep - data_path] = '\0'; if ((dirfd = open(dir_path, RB_DIR_RO_FLAGS)) != -1) { res = qb_sys_unlink_or_truncate_at(dirfd, sep + 1, truncate_fallback); /* the dirname part is assumed to be the same */ if (strncmp(dir_path, hdr_path, sep - data_path)) { qb_util_perror(LOG_DEBUG, "header path is corrupted: %s", hdr_path); res = -ENXIO; } sep = hdr_path + (sep - data_path); /* now, don't touch neither data_path nor hdr_path */ res2 = qb_sys_unlink_or_truncate_at(dirfd, sep + 1, truncate_fallback); close(dirfd); } else { res = -errno; qb_util_perror(LOG_DEBUG, "Cannot open dir: %s", hdr_path); } } else { res = -EINVAL; qb_util_perror(LOG_DEBUG, "Not dir-separable path: %s", hdr_path); } #else res = qb_sys_unlink_or_truncate(data_path, truncate_fallback); res2 = qb_sys_unlink_or_truncate(hdr_path, truncate_fallback); #endif /* defined(HAVE_OPENAT) && defined(HAVE_UNLINKAT) */ res = res ? res : res2; hdr_path = NULL; } /* if (unlink_it) */ if (munmap(rb->shared_data, (word_size * sizeof(uint32_t)) << 1) == -1) { res = res ? res : -errno; qb_util_perror(LOG_DEBUG, "Cannot munmap shared_data"); } if (munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s)) == -1) { res = res ? res : -errno; qb_util_perror(LOG_DEBUG, "Cannot munmap shared_hdr"); } free(rb); return res; }