diff --git a/examples/ipcclient.c b/examples/ipcclient.c index a949e4b..f0be5c9 100644 --- a/examples/ipcclient.c +++ b/examples/ipcclient.c @@ -1,237 +1,237 @@ /* * Copyright (c) 2011 Red Hat, Inc. * * All rights reserved. * * Author: Angus Salkeld * * 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 #include #include #include static int32_t do_benchmark = QB_FALSE; static int32_t use_events = QB_FALSE; static int alarm_notice; static qb_util_stopwatch_t *sw; #define ONE_MEG 1048576 #define MAX_MSG_SIZE ONE_MEG static char data[ONE_MEG]; struct my_req { struct qb_ipc_request_header hdr; char message[256]; }; struct my_res { struct qb_ipc_response_header hdr; char message[256]; }; static void sigalrm_handler (int num) { alarm_notice = 1; } static void _benchmark(qb_ipcc_connection_t *conn, int write_size) { struct iovec iov[2]; - unsigned int res; + ssize_t res; struct qb_ipc_request_header hdr; int write_count = 0; float secs; alarm_notice = 0; hdr.size = write_size; hdr.id = QB_IPC_MSG_USER_START + 1; iov[0].iov_base = &hdr; iov[0].iov_len = sizeof(struct qb_ipc_request_header); iov[1].iov_base = data; iov[1].iov_len = write_size - sizeof(struct qb_ipc_request_header); alarm (10); qb_util_stopwatch_start(sw); do { res = qb_ipcc_sendv(conn, iov, 2); if (res == write_size) { write_count++; } } while (alarm_notice == 0 && (res == write_size || res == -EAGAIN)); if (res < 0) { perror("qb_ipcc_sendv"); } qb_util_stopwatch_stop(sw); secs = qb_util_stopwatch_sec_elapsed_get(sw); printf ("%5d messages sent ", write_count); printf ("%5d bytes per write ", write_size); printf ("%7.3f Seconds runtime ", secs); printf ("%9.3f TP/s ", ((float)write_count) / secs); printf ("%7.3f MB/s.\n", ((float)write_count) * ((float)write_size) / secs); } static void do_throughput_benchmark(qb_ipcc_connection_t *conn) { ssize_t size = 64; int i; signal (SIGALRM, sigalrm_handler); sw = qb_util_stopwatch_create(); for (i = 0; i < 10; i++) { /* number of repetitions - up to 50k */ _benchmark (conn, size); signal (SIGALRM, sigalrm_handler); size *= 5; if (size >= (ONE_MEG - 100)) { break; } } } static void do_echo(qb_ipcc_connection_t *conn) { struct my_req req; struct my_res res; char *newline; int32_t rc; int32_t send_ten_events; while (1) { printf("SEND (q or Q to quit) : "); if (fgets(req.message, 256, stdin) == NULL) { continue; } newline = strrchr(req.message, '\n'); if (newline) { *newline = '\0'; } if (strcasecmp(req.message, "q") == 0) { break; } else { req.hdr.id = QB_IPC_MSG_USER_START + 3; req.hdr.size = sizeof(struct my_req); rc = qb_ipcc_send(conn, &req, req.hdr.size); if (rc < 0) { perror("qb_ipcc_send"); exit(0); } } send_ten_events = (strcasecmp(req.message, "events") == 0); if (rc > 0) { if (use_events && !send_ten_events) { printf("waiting for event recv\n"); rc = qb_ipcc_event_recv(conn, &res, sizeof(res), -1); } else { printf("waiting for recv\n"); rc = qb_ipcc_recv(conn, &res, sizeof(res), -1); } printf("recv %d\n", rc); if (rc < 0) { perror("qb_ipcc_recv"); exit(0); } if (send_ten_events) { int32_t i; printf("waiting for 10 events\n"); for (i = 0; i < 10; i++) { rc = qb_ipcc_event_recv(conn, &res, sizeof(res), -1); if (rc < 0) { perror("qb_ipcc_event_recv"); } else { printf("got event %d rc:%d\n", i, rc); } } } printf("Response[%d]: %s \n", res.hdr.id, res.message); } } } static void show_usage(const char *name) { printf("usage: \n"); printf("%s \n", name); printf("\n"); printf(" options:\n"); printf("\n"); printf(" -h show this help text\n"); printf(" -b benchmark\n"); printf(" -e use events instead of responses\n"); printf("\n"); } int main(int argc, char *argv[]) { qb_ipcc_connection_t *conn; const char *options = "eb"; int32_t opt; while ((opt = getopt(argc, argv, options)) != -1) { switch (opt) { case 'b': do_benchmark = QB_TRUE; break; case 'e': use_events = QB_TRUE; break; case 'h': default: show_usage(argv[0]); exit(0); break; } } qb_log_init("ipcclient", LOG_USER, LOG_TRACE); qb_log_ctl(QB_LOG_SYSLOG, QB_LOG_CONF_ENABLED, QB_FALSE); qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD, QB_LOG_FILTER_FILE, "*", LOG_TRACE); qb_log_format_set(QB_LOG_STDERR, "%f:%l [%p] %b"); qb_log_ctl(QB_LOG_STDERR, QB_LOG_CONF_ENABLED, QB_TRUE); conn = qb_ipcc_connect("ipcserver", MAX_MSG_SIZE); if (conn == NULL) { perror("qb_ipcc_connect"); exit(1); } if (do_benchmark) { do_throughput_benchmark(conn); } else { do_echo(conn); } qb_ipcc_disconnect(conn); return EXIT_SUCCESS; } diff --git a/examples/tcpserver.c b/examples/tcpserver.c index 4878d55..88210e9 100644 --- a/examples/tcpserver.c +++ b/examples/tcpserver.c @@ -1,151 +1,154 @@ /* * Copyright (c) 2011 Red Hat, Inc. * * All rights reserved. * * Author: Angus Salkeld * * 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" #ifdef HAVE_NETINET_IN_H #include #endif /* HAVE_NETINET_IN_H */ #ifdef HAVE_ARPA_INET_H #include #endif /* HAVE_ARPA_INET_H */ #ifdef HAVE_NETDB_H #include #endif /* HAVE_NETDB_H */ #ifdef HAVE_SYS_SOCKET_H #include #endif /* HAVE_SYS_SOCKET_H */ #ifdef HAVE_SYS_POLL_H #include #endif /* HAVE_SYS_POLL_H */ #include #include static int32_t sock_read_fn(int32_t fd, int32_t revents, void *data) { char recv_data[1024]; char send_data[1024]; int bytes_recieved; if (revents & POLLHUP) { printf("Socket %d peer closed\n", fd); close(fd); return QB_FALSE; } bytes_recieved = recv(fd, recv_data, 1024, 0); if (bytes_recieved < 0) { perror("recv"); return QB_TRUE; } recv_data[bytes_recieved] = '\0'; if (strcmp(recv_data, "q") == 0 || strcmp(recv_data, "Q") == 0) { printf("Quiting connection from socket %d\n", fd); close(fd); return QB_FALSE; } else { printf("Recieved: %s\n", recv_data); snprintf(send_data, 1024, "ACK %d bytes", bytes_recieved); - send(fd, send_data, strlen(send_data), 0); + if (send(fd, send_data, strlen(send_data), 0) < 0) { + close(fd); + return QB_FALSE; + } } return QB_TRUE; } static int32_t sock_accept_fn(int32_t fd, int32_t revents, void *data) { struct sockaddr_in client_addr; qb_loop_t *ml = (qb_loop_t *) data; socklen_t sin_size = sizeof(struct sockaddr_in); int connected = accept(fd, (struct sockaddr *)&client_addr, &sin_size); if (connected < 0) { perror("accept"); return QB_TRUE; } printf("I got a connection from (%s , %d)\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port)); qb_loop_poll_add(ml, QB_LOOP_MED, connected, POLLIN, ml, sock_read_fn); return QB_TRUE; } static int32_t please_exit_fn(int32_t rsignal, void *data) { qb_loop_t *ml = (qb_loop_t *) data; printf("Shutting down at you request...\n"); qb_loop_stop(ml); return QB_FALSE; } int main(int argc, char *argv[]) { int sock; int true_opt = 1; struct sockaddr_in server_addr; qb_loop_t *ml = qb_loop_create(); if ((sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) { perror("Socket"); exit(1); } if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &true_opt, sizeof(int)) == -1) { perror("Setsockopt"); exit(1); } server_addr.sin_family = AF_INET; server_addr.sin_port = htons(5000); server_addr.sin_addr.s_addr = INADDR_ANY; bzero(&(server_addr.sin_zero), 8); printf("TCPServer binding to port 5000\n"); if (bind(sock, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) == -1) { perror("Unable to bind"); exit(1); } printf("TCPServer Waiting for client on port 5000\n"); if (listen(sock, 5) == -1) { perror("Listen"); exit(1); } qb_loop_poll_add(ml, QB_LOOP_MED, sock, POLLIN, ml, sock_accept_fn); qb_loop_signal_add(ml, QB_LOOP_HIGH, SIGINT, ml, please_exit_fn, NULL); qb_loop_run(ml); close(sock); return 0; } diff --git a/lib/ipc_int.h b/lib/ipc_int.h index 45489f5..6c0fd22 100644 --- a/lib/ipc_int.h +++ b/lib/ipc_int.h @@ -1,207 +1,206 @@ /* * Copyright (C) 2009 Red Hat, Inc. * * Author: Steven Dake * 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 . */ #ifndef QB_IPC_INT_H_DEFINED #define QB_IPC_INT_H_DEFINED #include "os_base.h" #include #include #include #include #include #include #include #define QB_IPC_MAX_WAIT_MS 2000 /* Client Server SEND CONN REQ -> ACCEPT & CREATE queues or DENY <- SEND ACCEPT(with details)/DENY */ struct qb_ipc_connection_request { struct qb_ipc_request_header hdr; uint32_t max_msg_size; } __attribute__ ((aligned(8))); struct qb_ipc_event_connection_request { struct qb_ipc_request_header hdr; intptr_t connection; } __attribute__ ((aligned(8))); struct qb_ipc_connection_response { struct qb_ipc_response_header hdr; int32_t connection_type; uint32_t max_msg_size; intptr_t connection; char request[PATH_MAX]; char response[PATH_MAX]; char event[PATH_MAX]; } __attribute__ ((aligned(8))); struct qb_ipcc_connection; struct qb_ipc_one_way { size_t max_msg_size; enum qb_ipc_type type; union { struct { int32_t sock; char *sock_name; void* shared_data; char shared_file_name[NAME_MAX]; } us; struct { qb_ringbuffer_t *rb; } shm; } u; }; struct qb_ipcc_funcs { ssize_t (*recv)(struct qb_ipc_one_way *one_way, void *buf, size_t buf_size, int32_t timeout); ssize_t (*send)(struct qb_ipc_one_way *one_way, const void *data, size_t size); ssize_t (*sendv)(struct qb_ipc_one_way *one_way, const struct iovec *iov, size_t iov_len); void (*disconnect)(struct qb_ipcc_connection* c); int32_t (*fc_get)(struct qb_ipc_one_way *one_way); }; struct qb_ipcc_connection { char name[NAME_MAX]; int32_t needs_sock_for_poll; struct qb_ipc_one_way setup; struct qb_ipc_one_way request; struct qb_ipc_one_way response; struct qb_ipc_one_way event; struct qb_ipcc_funcs funcs; struct qb_ipc_request_header *receive_buf; uint32_t fc_enable_max; int32_t is_connected; void * context; }; 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_ready(struct qb_ipc_one_way *ow_data, struct qb_ipc_one_way *ow_conn, int32_t ms_timeout, int32_t events); void qb_ipcc_us_sock_close(int32_t sock); int32_t qb_ipcc_us_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); struct qb_ipcs_service; struct qb_ipcs_connection; struct qb_ipcs_funcs { int32_t (*connect)(struct qb_ipcs_service *s, struct qb_ipcs_connection *c, struct qb_ipc_connection_response *r); void (*disconnect)(struct qb_ipcs_connection *c); ssize_t (*recv)(struct qb_ipc_one_way *one_way, void *buf, size_t buf_size, int32_t timeout); ssize_t (*peek)(struct qb_ipc_one_way *one_way, void **data_out, int32_t timeout); void (*reclaim)(struct qb_ipc_one_way *one_way); ssize_t (*send)(struct qb_ipc_one_way *one_way, const void *data, size_t size); ssize_t (*sendv)(struct qb_ipc_one_way *one_way, const struct iovec* iov, size_t iov_len); void (*fc_set)(struct qb_ipc_one_way *one_way, int32_t fc_enable); ssize_t (*q_len_get)(struct qb_ipc_one_way *one_way); }; struct qb_ipcs_service { enum qb_ipc_type type; char name[NAME_MAX]; int32_t service_id; int32_t ref_count; pid_t pid; int32_t needs_sock_for_poll; int32_t server_sock; struct qb_ipcs_service_handlers serv_fns; struct qb_ipcs_poll_handlers poll_fns; struct qb_ipcs_funcs funcs; enum qb_loop_priority poll_priority; struct qb_list_head connections; struct qb_list_head list; struct qb_ipcs_stats stats; void *context; }; enum qb_ipcs_connection_state { QB_IPCS_CONNECTION_INACTIVE, QB_IPCS_CONNECTION_ACTIVE, QB_IPCS_CONNECTION_ESTABLISHED, QB_IPCS_CONNECTION_SHUTTING_DOWN, }; #define CONNECTION_DESCRIPTION (16) struct qb_ipcs_connection_auth { uid_t uid; gid_t gid; mode_t mode; }; struct qb_ipcs_connection { enum qb_ipcs_connection_state state; int32_t refcount; pid_t pid; uid_t euid; gid_t egid; struct qb_ipcs_connection_auth auth; struct qb_ipc_one_way setup; struct qb_ipc_one_way request; struct qb_ipc_one_way response; struct qb_ipc_one_way event; struct qb_ipcs_service *service; struct qb_list_head list; struct qb_ipc_request_header *receive_buf; void *context; int32_t fc_enabled; int32_t poll_events; int32_t outstanding_notifiers; char description[CONNECTION_DESCRIPTION]; struct qb_ipcs_connection_stats_2 stats; }; void qb_ipcs_us_init(struct qb_ipcs_service *s); void qb_ipcs_shm_init(struct qb_ipcs_service *s); int32_t qb_ipcs_us_publish(struct qb_ipcs_service *s); int32_t qb_ipcs_us_withdraw(struct qb_ipcs_service *s); int32_t qb_ipcc_us_sock_connect(const char *socket_name, int32_t * sock_pt); int32_t qb_ipcs_dispatch_connection_request(int32_t fd, int32_t revents, void *data); -int32_t qb_ipcs_dispatch_service_request(int32_t fd, int32_t revents, void *data); struct qb_ipcs_connection* qb_ipcs_connection_alloc(struct qb_ipcs_service *s); int32_t qb_ipcs_process_request(struct qb_ipcs_service *s, struct qb_ipc_request_header *hdr); int32_t qb_ipc_us_sock_error_is_disconnected(int err); #endif /* QB_IPC_INT_H_DEFINED */ diff --git a/lib/ipc_shm.c b/lib/ipc_shm.c index 79fbf80..3f1922f 100644 --- a/lib/ipc_shm.c +++ b/lib/ipc_shm.c @@ -1,367 +1,367 @@ /* * 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 "os_base.h" #include "ipc_int.h" #include "util_int.h" #include "ringbuffer_int.h" #include #include #include #include /* * utility functions * -------------------------------------------------------- */ /* * client functions * -------------------------------------------------------- */ static void qb_ipcc_shm_disconnect(struct qb_ipcc_connection *c) { if (c->is_connected) { qb_rb_close(c->request.u.shm.rb); qb_rb_close(c->response.u.shm.rb); qb_rb_close(c->event.u.shm.rb); } else { qb_rb_force_close(c->request.u.shm.rb); qb_rb_force_close(c->response.u.shm.rb); qb_rb_force_close(c->event.u.shm.rb); } } static ssize_t qb_ipc_shm_send(struct qb_ipc_one_way *one_way, const void *msg_ptr, size_t msg_len) { return qb_rb_chunk_write(one_way->u.shm.rb, msg_ptr, msg_len); } static ssize_t qb_ipc_shm_sendv(struct qb_ipc_one_way *one_way, const struct iovec *iov, size_t iov_len) { char *dest; int32_t res = 0; int32_t total_size = 0; int32_t i; char *pt = NULL; if (one_way->u.shm.rb == NULL) { return -ENOTCONN; } for (i = 0; i < iov_len; i++) { total_size += iov[i].iov_len; } dest = qb_rb_chunk_alloc(one_way->u.shm.rb, total_size); if (dest == NULL) { return -errno; } pt = dest; for (i = 0; i < iov_len; i++) { memcpy(pt, iov[i].iov_base, iov[i].iov_len); pt += iov[i].iov_len; } res = qb_rb_chunk_commit(one_way->u.shm.rb, total_size); if (res < 0) { return res; } return total_size; } static ssize_t qb_ipc_shm_recv(struct qb_ipc_one_way *one_way, void *msg_ptr, size_t msg_len, int32_t ms_timeout) { if (one_way->u.shm.rb == NULL) { return -ENOTCONN; } return qb_rb_chunk_read(one_way->u.shm.rb, (void *)msg_ptr, msg_len, ms_timeout); } static ssize_t qb_ipc_shm_peek(struct qb_ipc_one_way *one_way, void **data_out, int32_t ms_timeout) { ssize_t rc; if (one_way->u.shm.rb == NULL) { return -ENOTCONN; } rc = qb_rb_chunk_peek(one_way->u.shm.rb, data_out, ms_timeout); if (rc == 0) { return -EAGAIN; } return rc; } static void qb_ipc_shm_reclaim(struct qb_ipc_one_way *one_way) { if (one_way->u.shm.rb != NULL) { qb_rb_chunk_reclaim(one_way->u.shm.rb); } } static void qb_ipc_shm_fc_set(struct qb_ipc_one_way *one_way, int32_t fc_enable) { int32_t *fc; fc = qb_rb_shared_user_data_get(one_way->u.shm.rb); qb_util_log(LOG_TRACE, "setting fc to %d", fc_enable); qb_atomic_int_set(fc, fc_enable); } static int32_t qb_ipc_shm_fc_get(struct qb_ipc_one_way *one_way) { int32_t *fc; int32_t rc = qb_rb_refcount_get(one_way->u.shm.rb); if (rc != 2) { return -ENOTCONN; } fc = qb_rb_shared_user_data_get(one_way->u.shm.rb); return qb_atomic_int_get(fc); } static ssize_t qb_ipc_shm_q_len_get(struct qb_ipc_one_way *one_way) { if (one_way->u.shm.rb == NULL) { return -ENOTCONN; } return qb_rb_chunks_used(one_way->u.shm.rb); } int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection * c, struct qb_ipc_connection_response * response) { int32_t res = 0; c->funcs.send = qb_ipc_shm_send; c->funcs.sendv = qb_ipc_shm_sendv; c->funcs.recv = qb_ipc_shm_recv; c->funcs.fc_get = qb_ipc_shm_fc_get; c->funcs.disconnect = qb_ipcc_shm_disconnect; c->needs_sock_for_poll = QB_TRUE; if (strlen(c->name) > (NAME_MAX - 20)) { errno = EINVAL; return -errno; } c->request.u.shm.rb = qb_rb_open(response->request, c->request.max_msg_size, QB_RB_FLAG_SHARED_PROCESS, sizeof(int32_t)); if (c->request.u.shm.rb == NULL) { res = -errno; qb_util_perror(LOG_ERR, "qb_rb_open:REQUEST"); goto return_error; } c->response.u.shm.rb = qb_rb_open(response->response, c->response.max_msg_size, QB_RB_FLAG_SHARED_PROCESS, 0); if (c->response.u.shm.rb == NULL) { res = -errno; qb_util_perror(LOG_ERR, "qb_rb_open:RESPONSE"); goto cleanup_request; } c->event.u.shm.rb = qb_rb_open(response->event, c->response.max_msg_size, QB_RB_FLAG_SHARED_PROCESS, 0); if (c->event.u.shm.rb == NULL) { res = -errno; qb_util_perror(LOG_ERR, "qb_rb_open:EVENT"); goto cleanup_request_response; } return 0; cleanup_request_response: qb_rb_close(c->response.u.shm.rb); cleanup_request: qb_rb_close(c->request.u.shm.rb); return_error: errno = -res; qb_util_perror(LOG_ERR, "connection failed"); return res; } /* * service functions * -------------------------------------------------------- */ static void qb_ipcs_shm_disconnect(struct qb_ipcs_connection *c) { if (c->state == QB_IPCS_CONNECTION_ESTABLISHED || c->state == QB_IPCS_CONNECTION_ACTIVE) { if (c->setup.u.us.sock > 0) { qb_ipcc_us_sock_close(c->setup.u.us.sock); (void)c->service->poll_fns.dispatch_del(c->setup.u.us.sock); - qb_ipcs_connection_unref(c); c->setup.u.us.sock = -1; + qb_ipcs_connection_unref(c); } } if (c->state == QB_IPCS_CONNECTION_SHUTTING_DOWN || c->state == QB_IPCS_CONNECTION_ACTIVE) { if (c->response.u.shm.rb) { qb_rb_close(c->response.u.shm.rb); c->response.u.shm.rb = NULL; } if (c->event.u.shm.rb) { qb_rb_close(c->event.u.shm.rb); c->event.u.shm.rb = NULL; } if (c->request.u.shm.rb) { qb_rb_close(c->request.u.shm.rb); c->request.u.shm.rb = NULL; } } } static int32_t qb_ipcs_shm_rb_open(struct qb_ipcs_connection *c, struct qb_ipc_one_way *ow, const char *rb_name) { int32_t res = 0; ow->u.shm.rb = qb_rb_open(rb_name, ow->max_msg_size, QB_RB_FLAG_CREATE | QB_RB_FLAG_SHARED_PROCESS, sizeof(int32_t)); if (ow->u.shm.rb == NULL) { res = -errno; qb_util_perror(LOG_ERR, "qb_rb_open:%s", rb_name); return res; } res = qb_rb_chown(ow->u.shm.rb, c->auth.uid, c->auth.gid); if (res != 0) { qb_util_perror(LOG_ERR, "qb_rb_chown:%s", rb_name); goto cleanup; } res = qb_rb_chmod(ow->u.shm.rb, c->auth.mode); if (res != 0) { qb_util_perror(LOG_ERR, "qb_rb_chmod:%s", rb_name); goto cleanup; } return res; cleanup: qb_rb_close(ow->u.shm.rb); return res; } static int32_t qb_ipcs_shm_connect(struct qb_ipcs_service *s, struct qb_ipcs_connection *c, struct qb_ipc_connection_response *r) { int32_t res; qb_util_log(LOG_DEBUG, "connecting to client [%d]", c->pid); snprintf(r->request, NAME_MAX, "%s-request-%s", s->name, c->description); snprintf(r->response, NAME_MAX, "%s-response-%s", s->name, c->description); snprintf(r->event, NAME_MAX, "%s-event-%s", s->name, c->description); res = qb_ipcs_shm_rb_open(c, &c->request, r->request); if (res != 0) { goto cleanup; } res = qb_ipcs_shm_rb_open(c, &c->response, r->response); if (res != 0) { goto cleanup_request; } res = qb_ipcs_shm_rb_open(c, &c->event, r->event); if (res != 0) { goto cleanup_request_response; } res = s->poll_fns.dispatch_add(s->poll_priority, c->setup.u.us.sock, POLLIN | POLLPRI | POLLNVAL, c, qb_ipcs_dispatch_connection_request); if (res == 0) { qb_ipcs_connection_ref(c); } else { qb_util_log(LOG_ERR, "Error adding socket to mainloop (%s).", c->description); goto cleanup_request_response; } r->hdr.error = 0; return 0; cleanup_request_response: qb_rb_close(c->request.u.shm.rb); cleanup_request: qb_rb_close(c->response.u.shm.rb); cleanup: r->hdr.error = res; errno = -res; qb_util_perror(LOG_ERR, "shm connection FAILED"); return res; } void qb_ipcs_shm_init(struct qb_ipcs_service *s) { s->funcs.connect = qb_ipcs_shm_connect; s->funcs.disconnect = qb_ipcs_shm_disconnect; s->funcs.recv = qb_ipc_shm_recv; s->funcs.peek = qb_ipc_shm_peek; s->funcs.reclaim = qb_ipc_shm_reclaim; s->funcs.send = qb_ipc_shm_send; s->funcs.sendv = qb_ipc_shm_sendv; s->funcs.fc_set = qb_ipc_shm_fc_set; s->funcs.q_len_get = qb_ipc_shm_q_len_get; s->needs_sock_for_poll = QB_TRUE; } diff --git a/lib/ipc_socket.c b/lib/ipc_socket.c index 6a22e75..8f1e22c 100644 --- a/lib/ipc_socket.c +++ b/lib/ipc_socket.c @@ -1,645 +1,651 @@ /* * Copyright (C) 2010,2013 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" #ifdef HAVE_SYS_UN_H #include #endif /* HAVE_SYS_UN_H */ #ifdef HAVE_SYS_MMAN_H #include #endif #include #include #include #include #include "util_int.h" #include "ipc_int.h" struct ipc_us_control { int32_t sent; int32_t flow_control; }; #define SHM_CONTROL_SIZE (3 * sizeof(struct ipc_us_control)) static void set_sock_addr(struct sockaddr_un *address, const char *socket_name) { 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 defined(QB_LINUX) || defined(QB_CYGWIN) snprintf(address->sun_path + 1, UNIX_PATH_MAX - 1, "%s", socket_name); #else snprintf(address->sun_path, UNIX_PATH_MAX, "%s/%s", SOCKETDIR, socket_name); #endif } static int32_t qb_ipc_dgram_sock_setup(const char *base_name, const char *service_name, int32_t * sock_pt) { int32_t request_fd; struct sockaddr_un local_address; int32_t res = 0; char sock_path[PATH_MAX]; request_fd = socket(PF_UNIX, SOCK_DGRAM, 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; } snprintf(sock_path, PATH_MAX, "%s-%s", base_name, service_name); set_sock_addr(&local_address, sock_path); res = bind(request_fd, (struct sockaddr *)&local_address, sizeof(local_address)); if (res < 0) { goto error_connect; } *sock_pt = request_fd; return 0; error_connect: close(request_fd); *sock_pt = -1; return res; } static int32_t set_sock_size(int sockfd, size_t max_msg_size) { int32_t rc; unsigned int optval; socklen_t optlen = sizeof(optval); rc = getsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, &optval, &optlen); qb_util_log(LOG_DEBUG, "%d: getsockopt(%d, needed:%d) actual:%d", rc, sockfd, max_msg_size, optval); if (rc == 0 && optval < max_msg_size) { optval = max_msg_size; optlen = sizeof(optval); rc = setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, &optval, optlen); } return rc; } /* * bind to "base_name-local_name" * connect to "base_name-remote_name" * output sock_pt */ static int32_t qb_ipc_dgram_sock_connect(const char *base_name, const char *local_name, const char *remote_name, int32_t max_msg_size, int32_t * sock_pt) { char sock_path[PATH_MAX]; struct sockaddr_un remote_address; int32_t res = qb_ipc_dgram_sock_setup(base_name, local_name, sock_pt); if (res < 0) { return res; } snprintf(sock_path, PATH_MAX, "%s-%s", base_name, remote_name); set_sock_addr(&remote_address, sock_path); if (connect(*sock_pt, (struct sockaddr *)&remote_address, QB_SUN_LEN(&remote_address)) == -1) { res = -errno; goto error_connect; } return set_sock_size(*sock_pt, max_msg_size); error_connect: close(*sock_pt); *sock_pt = -1; return res; } static int32_t _finish_connecting(struct qb_ipc_one_way *one_way) { struct sockaddr_un remote_address; int res; int error; int retry = 0; set_sock_addr(&remote_address, one_way->u.us.sock_name); /* this retry loop is here to help connecting when trying to send * an event right after connection setup. */ do { errno = 0; res = connect(one_way->u.us.sock, (struct sockaddr *)&remote_address, QB_SUN_LEN(&remote_address)); if (res == -1) { error = -errno; qb_util_perror(LOG_DEBUG, "error calling connect()"); retry++; usleep(100000); } } while (res == -1 && retry < 10); if (res == -1) { return error; } free(one_way->u.us.sock_name); one_way->u.us.sock_name = NULL; return set_sock_size(one_way->u.us.sock, one_way->max_msg_size); } /* * client functions * -------------------------------------------------------- */ static void qb_ipcc_us_disconnect(struct qb_ipcc_connection *c) { munmap(c->request.u.us.shared_data, SHM_CONTROL_SIZE); unlink(c->request.u.us.shared_file_name); close(c->request.u.us.sock); close(c->event.u.us.sock); } static ssize_t qb_ipc_socket_send(struct qb_ipc_one_way *one_way, const void *msg_ptr, size_t msg_len) { ssize_t rc = 0; struct ipc_us_control *ctl; ctl = (struct ipc_us_control *)one_way->u.us.shared_data; if (one_way->u.us.sock_name) { rc = _finish_connecting(one_way); if (rc < 0) { qb_util_log(LOG_ERR, "socket connect-on-send"); return rc; } } qb_sigpipe_ctl(QB_SIGPIPE_IGNORE); rc = send(one_way->u.us.sock, msg_ptr, msg_len, MSG_NOSIGNAL); if (rc == -1) { rc = -errno; if (errno != EAGAIN && errno != ENOBUFS) { qb_util_perror(LOG_ERR, "socket_send:send"); } } qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); if (ctl && rc == msg_len) { qb_atomic_int_inc(&ctl->sent); } return rc; } static ssize_t qb_ipc_socket_sendv(struct qb_ipc_one_way *one_way, const struct iovec *iov, size_t iov_len) { int32_t rc; struct ipc_us_control *ctl; ctl = (struct ipc_us_control *)one_way->u.us.shared_data; qb_sigpipe_ctl(QB_SIGPIPE_IGNORE); if (one_way->u.us.sock_name) { rc = _finish_connecting(one_way); if (rc < 0) { qb_util_perror(LOG_ERR, "socket connect-on-sendv"); return rc; } } rc = writev(one_way->u.us.sock, iov, iov_len); if (rc == -1) { rc = -errno; if (errno != EAGAIN && errno != ENOBUFS) { qb_util_perror(LOG_ERR, "socket_sendv:writev %d", one_way->u.us.sock); } } qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); if (ctl && rc > 0) { qb_atomic_int_inc(&ctl->sent); } return rc; } /* * recv a message of unknown size. */ static ssize_t qb_ipc_us_recv_at_most(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 to_recv = 0; char *data = msg; struct ipc_us_control *ctl = NULL; int32_t time_waited = 0; int32_t time_to_wait = timeout; if (timeout == -1) { time_to_wait = 1000; } qb_sigpipe_ctl(QB_SIGPIPE_IGNORE); retry_peek: result = recv(one_way->u.us.sock, data, sizeof(struct qb_ipc_request_header), MSG_NOSIGNAL | MSG_PEEK); if (result == -1) { if (errno == EAGAIN && (time_waited < timeout || timeout == -1)) { result = qb_ipc_us_ready(one_way, NULL, time_to_wait, POLLIN); time_waited += time_to_wait; goto retry_peek; } else { return -errno; } } if (result >= sizeof(struct qb_ipc_request_header)) { struct qb_ipc_request_header *hdr = NULL; hdr = (struct qb_ipc_request_header *)msg; to_recv = hdr->size; } result = recv(one_way->u.us.sock, data, to_recv, MSG_NOSIGNAL | MSG_WAITALL); if (result == -1) { final_rc = -errno; goto cleanup_sigpipe; } else if (result == 0) { qb_util_log(LOG_DEBUG, "recv == 0 -> ENOTCONN"); final_rc = -ENOTCONN; goto cleanup_sigpipe; } final_rc = result; ctl = (struct ipc_us_control *)one_way->u.us.shared_data; if (ctl) { (void)qb_atomic_int_dec_and_test(&ctl->sent); } cleanup_sigpipe: qb_sigpipe_ctl(QB_SIGPIPE_DEFAULT); return final_rc; } static void qb_ipc_us_fc_set(struct qb_ipc_one_way *one_way, int32_t fc_enable) { struct ipc_us_control *ctl = (struct ipc_us_control *)one_way->u.us.shared_data; qb_util_log(LOG_TRACE, "setting fc to %d", fc_enable); qb_atomic_int_set(&ctl->flow_control, fc_enable); } static int32_t qb_ipc_us_fc_get(struct qb_ipc_one_way *one_way) { struct ipc_us_control *ctl = (struct ipc_us_control *)one_way->u.us.shared_data; return qb_atomic_int_get(&ctl->flow_control); } static ssize_t qb_ipc_us_q_len_get(struct qb_ipc_one_way *one_way) { struct ipc_us_control *ctl = (struct ipc_us_control *)one_way->u.us.shared_data; return qb_atomic_int_get(&ctl->sent); } int32_t qb_ipcc_us_connect(struct qb_ipcc_connection * c, struct qb_ipc_connection_response * r) { int32_t res; char path[PATH_MAX]; int32_t fd_hdr; char *shm_ptr; qb_atomic_init(); c->needs_sock_for_poll = QB_FALSE; c->funcs.send = qb_ipc_socket_send; c->funcs.sendv = qb_ipc_socket_sendv; c->funcs.recv = qb_ipc_us_recv_at_most; c->funcs.fc_get = qb_ipc_us_fc_get; c->funcs.disconnect = qb_ipcc_us_disconnect; fd_hdr = qb_sys_mmap_file_open(path, r->request, SHM_CONTROL_SIZE, O_RDWR); if (fd_hdr < 0) { res = fd_hdr; errno = -fd_hdr; qb_util_perror(LOG_ERR, "couldn't open file for mmap"); return res; } (void)strlcpy(c->request.u.us.shared_file_name, r->request, NAME_MAX); shm_ptr = mmap(0, SHM_CONTROL_SIZE, PROT_READ | PROT_WRITE, MAP_SHARED, fd_hdr, 0); if (shm_ptr == MAP_FAILED) { res = -errno; qb_util_perror(LOG_ERR, "couldn't create mmap for header"); goto cleanup_hdr; } c->request.u.us.shared_data = shm_ptr; c->response.u.us.shared_data = shm_ptr + sizeof(struct ipc_us_control); c->event.u.us.shared_data = shm_ptr + (2 * sizeof(struct ipc_us_control)); close(fd_hdr); + fd_hdr = -1; res = qb_ipc_dgram_sock_connect(r->response, "response", "request", r->max_msg_size, &c->request.u.us.sock); if (res != 0) { goto cleanup_hdr; } c->response.u.us.sock = c->request.u.us.sock; res = qb_ipc_dgram_sock_connect(r->response, "event", "event-tx", r->max_msg_size, &c->event.u.us.sock); if (res != 0) { goto cleanup_hdr; } return 0; cleanup_hdr: - close(fd_hdr); + if (fd_hdr >= 0) { + close(fd_hdr); + } close(c->event.u.us.sock); close(c->request.u.us.sock); unlink(r->request); munmap(c->request.u.us.shared_data, SHM_CONTROL_SIZE); return res; } /* * service functions * -------------------------------------------------------- */ static int32_t _sock_connection_liveliness(int32_t fd, int32_t revents, void *data) { struct qb_ipcs_connection *c = (struct qb_ipcs_connection *)data; qb_util_log(LOG_DEBUG, "LIVENESS: fd %d event %d conn (%s)", fd, revents, c->description); if (revents & POLLNVAL) { qb_util_log(LOG_DEBUG, "NVAL conn (%s)", c->description); qb_ipcs_disconnect(c); return -EINVAL; } if (revents & POLLHUP) { qb_util_log(LOG_DEBUG, "HUP conn (%s)", c->description); qb_ipcs_disconnect(c); return -ESHUTDOWN; } return 0; } static int32_t _sock_add_to_mainloop(struct qb_ipcs_connection *c) { int res; res = c->service->poll_fns.dispatch_add(c->service->poll_priority, c->request.u.us.sock, POLLIN | POLLPRI | POLLNVAL, c, qb_ipcs_dispatch_connection_request); if (res < 0) { qb_util_log(LOG_ERR, "Error adding socket to mainloop (%s).", c->description); return res; } qb_ipcs_connection_ref(c); res = c->service->poll_fns.dispatch_add(c->service->poll_priority, c->setup.u.us.sock, POLLIN | POLLPRI | POLLNVAL, c, _sock_connection_liveliness); qb_util_log(LOG_DEBUG, "added %d to poll loop (liveness)", c->setup.u.us.sock); if (res < 0) { qb_util_perror(LOG_ERR, "Error adding setupfd to mainloop"); (void)c->service->poll_fns.dispatch_del(c->request.u.us.sock); return res; } qb_ipcs_connection_ref(c); return res; } static void _sock_rm_from_mainloop(struct qb_ipcs_connection *c) { (void)c->service->poll_fns.dispatch_del(c->request.u.us.sock); qb_ipcs_connection_unref(c); (void)c->service->poll_fns.dispatch_del(c->setup.u.us.sock); qb_ipcs_connection_unref(c); } static void qb_ipcs_us_disconnect(struct qb_ipcs_connection *c) { qb_enter(); if (c->state == QB_IPCS_CONNECTION_ESTABLISHED || c->state == QB_IPCS_CONNECTION_ACTIVE) { _sock_rm_from_mainloop(c); qb_ipcc_us_sock_close(c->setup.u.us.sock); qb_ipcc_us_sock_close(c->request.u.us.sock); qb_ipcc_us_sock_close(c->event.u.us.sock); } if (c->state == QB_IPCS_CONNECTION_SHUTTING_DOWN || c->state == QB_IPCS_CONNECTION_ACTIVE) { munmap(c->request.u.us.shared_data, SHM_CONTROL_SIZE); unlink(c->request.u.us.shared_file_name); } } static int32_t qb_ipcs_us_connect(struct qb_ipcs_service *s, struct qb_ipcs_connection *c, struct qb_ipc_connection_response *r) { char path[PATH_MAX]; int32_t fd_hdr; int32_t res = 0; struct ipc_us_control *ctl; char *shm_ptr; qb_util_log(LOG_DEBUG, "connecting to client (%s)", c->description); c->request.u.us.sock = c->setup.u.us.sock; c->response.u.us.sock = c->setup.u.us.sock; snprintf(r->request, NAME_MAX, "qb-%s-control-%s", s->name, c->description); snprintf(r->response, NAME_MAX, "qb-%s-%s", s->name, c->description); fd_hdr = qb_sys_mmap_file_open(path, r->request, SHM_CONTROL_SIZE, O_CREAT | O_TRUNC | O_RDWR); if (fd_hdr < 0) { res = fd_hdr; errno = -fd_hdr; qb_util_perror(LOG_ERR, "couldn't create file for mmap (%s)", c->description); return res; } (void)strlcpy(r->request, path, PATH_MAX); (void)strlcpy(c->request.u.us.shared_file_name, r->request, NAME_MAX); res = chown(r->request, c->auth.uid, c->auth.gid); if (res != 0) { /* ignore res, this is just for the compiler warnings. */ res = 0; } res = chmod(r->request, c->auth.mode); if (res != 0) { /* ignore res, this is just for the compiler warnings. */ res = 0; } shm_ptr = mmap(0, SHM_CONTROL_SIZE, PROT_READ | PROT_WRITE, MAP_SHARED, fd_hdr, 0); if (shm_ptr == MAP_FAILED) { res = -errno; qb_util_perror(LOG_ERR, "couldn't create mmap for header (%s)", c->description); goto cleanup_hdr; } c->request.u.us.shared_data = shm_ptr; c->response.u.us.shared_data = shm_ptr + sizeof(struct ipc_us_control); c->event.u.us.shared_data = shm_ptr + (2 * sizeof(struct ipc_us_control)); ctl = (struct ipc_us_control *)c->request.u.us.shared_data; ctl->sent = 0; ctl->flow_control = 0; ctl = (struct ipc_us_control *)c->response.u.us.shared_data; ctl->sent = 0; ctl->flow_control = 0; ctl = (struct ipc_us_control *)c->event.u.us.shared_data; ctl->sent = 0; ctl->flow_control = 0; close(fd_hdr); + fd_hdr = -1; /* request channel */ res = qb_ipc_dgram_sock_setup(r->response, "request", &c->request.u.us.sock); if (res < 0) { goto cleanup_hdr; } c->setup.u.us.sock_name = NULL; c->request.u.us.sock_name = NULL; /* response channel */ c->response.u.us.sock = c->request.u.us.sock; snprintf(path, PATH_MAX, "%s-%s", r->response, "response"); c->response.u.us.sock_name = strdup(path); /* event channel */ res = qb_ipc_dgram_sock_setup(r->response, "event-tx", &c->event.u.us.sock); if (res < 0) { goto cleanup_hdr; } snprintf(path, PATH_MAX, "%s-%s", r->response, "event"); c->event.u.us.sock_name = strdup(path); res = _sock_add_to_mainloop(c); if (res < 0) { goto cleanup_hdr; } return res; cleanup_hdr: free(c->response.u.us.sock_name); free(c->event.u.us.sock_name); - close(fd_hdr); + if (fd_hdr >= 0) { + close(fd_hdr); + } unlink(r->request); munmap(c->request.u.us.shared_data, SHM_CONTROL_SIZE); return res; } void qb_ipcs_us_init(struct qb_ipcs_service *s) { s->funcs.connect = qb_ipcs_us_connect; s->funcs.disconnect = qb_ipcs_us_disconnect; s->funcs.recv = qb_ipc_us_recv_at_most; s->funcs.peek = NULL; s->funcs.reclaim = NULL; s->funcs.send = qb_ipc_socket_send; s->funcs.sendv = qb_ipc_socket_sendv; s->funcs.fc_set = qb_ipc_us_fc_set; s->funcs.q_len_get = qb_ipc_us_q_len_get; s->needs_sock_for_poll = QB_FALSE; qb_atomic_init(); } diff --git a/lib/ipcs.c b/lib/ipcs.c index a9ee7b3..d3baaab 100644 --- a/lib/ipcs.c +++ b/lib/ipcs.c @@ -1,950 +1,943 @@ /* * 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 "os_base.h" #include "util_int.h" #include "ipc_int.h" #include #include #include static void qb_ipcs_flowcontrol_set(struct qb_ipcs_connection *c, int32_t fc_enable); static int32_t new_event_notification(struct qb_ipcs_connection * c); static QB_LIST_DECLARE(qb_ipc_services); qb_ipcs_service_t * qb_ipcs_create(const char *name, int32_t service_id, enum qb_ipc_type type, struct qb_ipcs_service_handlers *handlers) { struct qb_ipcs_service *s; s = calloc(1, sizeof(struct qb_ipcs_service)); if (s == NULL) { return NULL; } if (type == QB_IPC_NATIVE) { #ifdef DISABLE_IPC_SHM s->type = QB_IPC_SOCKET; #else s->type = QB_IPC_SHM; #endif /* DISABLE_IPC_SHM */ } else { s->type = type; } s->pid = getpid(); s->needs_sock_for_poll = QB_FALSE; s->poll_priority = QB_LOOP_MED; /* Initial alloc ref */ qb_ipcs_ref(s); s->service_id = service_id; (void)strlcpy(s->name, name, NAME_MAX); s->serv_fns.connection_accept = handlers->connection_accept; s->serv_fns.connection_created = handlers->connection_created; s->serv_fns.msg_process = handlers->msg_process; s->serv_fns.connection_closed = handlers->connection_closed; s->serv_fns.connection_destroyed = handlers->connection_destroyed; qb_list_init(&s->connections); qb_list_init(&s->list); qb_list_add(&s->list, &qb_ipc_services); return s; } void qb_ipcs_poll_handlers_set(struct qb_ipcs_service *s, struct qb_ipcs_poll_handlers *handlers) { s->poll_fns.job_add = handlers->job_add; s->poll_fns.dispatch_add = handlers->dispatch_add; s->poll_fns.dispatch_mod = handlers->dispatch_mod; s->poll_fns.dispatch_del = handlers->dispatch_del; } void qb_ipcs_service_context_set(qb_ipcs_service_t* s, void *context) { s->context = context; } void * qb_ipcs_service_context_get(qb_ipcs_service_t* s) { return s->context; } int32_t qb_ipcs_run(struct qb_ipcs_service *s) { int32_t res = 0; if (s->poll_fns.dispatch_add == NULL || s->poll_fns.dispatch_mod == NULL || s->poll_fns.dispatch_del == NULL) { res = -EINVAL; goto run_cleanup; } switch (s->type) { case QB_IPC_SOCKET: qb_ipcs_us_init((struct qb_ipcs_service *)s); break; case QB_IPC_SHM: #ifdef DISABLE_IPC_SHM res = -ENOTSUP; #else qb_ipcs_shm_init((struct qb_ipcs_service *)s); #endif /* DISABLE_IPC_SHM */ break; case QB_IPC_POSIX_MQ: case QB_IPC_SYSV_MQ: res = -ENOTSUP; break; default: res = -EINVAL; break; } if (res == 0) { res = qb_ipcs_us_publish(s); if (res < 0) { (void)qb_ipcs_us_withdraw(s); goto run_cleanup; } } run_cleanup: if (res < 0) { /* Failed to run services, removing initial alloc reference. */ qb_ipcs_unref(s); } return res; } static int32_t _modify_dispatch_descriptor_(struct qb_ipcs_connection *c) { qb_ipcs_dispatch_mod_fn disp_mod = c->service->poll_fns.dispatch_mod; if (c->service->type == QB_IPC_SOCKET) { return disp_mod(c->service->poll_priority, c->event.u.us.sock, c->poll_events, c, qb_ipcs_dispatch_connection_request); } else { return disp_mod(c->service->poll_priority, c->setup.u.us.sock, c->poll_events, c, qb_ipcs_dispatch_connection_request); } return -EINVAL; } void qb_ipcs_request_rate_limit(struct qb_ipcs_service *s, enum qb_ipcs_rate_limit rl) { struct qb_ipcs_connection *c; enum qb_loop_priority old_p = s->poll_priority; struct qb_list_head *pos; struct qb_list_head *n; switch (rl) { case QB_IPCS_RATE_FAST: s->poll_priority = QB_LOOP_HIGH; break; case QB_IPCS_RATE_SLOW: case QB_IPCS_RATE_OFF: case QB_IPCS_RATE_OFF_2: s->poll_priority = QB_LOOP_LOW; break; default: case QB_IPCS_RATE_NORMAL: s->poll_priority = QB_LOOP_MED; break; } qb_list_for_each_safe(pos, n, &s->connections) { c = qb_list_entry(pos, struct qb_ipcs_connection, list); qb_ipcs_connection_ref(c); if (rl == QB_IPCS_RATE_OFF) { qb_ipcs_flowcontrol_set(c, 1); } else if (rl == QB_IPCS_RATE_OFF_2) { qb_ipcs_flowcontrol_set(c, 2); } else { qb_ipcs_flowcontrol_set(c, QB_FALSE); } if (old_p != s->poll_priority) { (void)_modify_dispatch_descriptor_(c); } qb_ipcs_connection_unref(c); } } void qb_ipcs_ref(struct qb_ipcs_service *s) { qb_atomic_int_inc(&s->ref_count); } void qb_ipcs_unref(struct qb_ipcs_service *s) { int32_t free_it; assert(s->ref_count > 0); free_it = qb_atomic_int_dec_and_test(&s->ref_count); if (free_it) { qb_util_log(LOG_DEBUG, "%s() - destroying", __func__); free(s); } } void qb_ipcs_destroy(struct qb_ipcs_service *s) { struct qb_ipcs_connection *c = NULL; struct qb_list_head *pos; struct qb_list_head *n; if (s == NULL) { return; } qb_list_for_each_safe(pos, n, &s->connections) { c = qb_list_entry(pos, struct qb_ipcs_connection, list); if (c == NULL) { continue; } qb_ipcs_disconnect(c); } (void)qb_ipcs_us_withdraw(s); /* service destroyed, remove initial alloc ref */ qb_ipcs_unref(s); } /* * connection API */ static struct qb_ipc_one_way * _event_sock_one_way_get(struct qb_ipcs_connection * c) { if (c->service->needs_sock_for_poll) { return &c->setup; } if (c->event.type == QB_IPC_SOCKET) { return &c->event; } return NULL; } static struct qb_ipc_one_way * _response_sock_one_way_get(struct qb_ipcs_connection * c) { if (c->service->needs_sock_for_poll) { return &c->setup; } if (c->response.type == QB_IPC_SOCKET) { return &c->response; } return NULL; } ssize_t qb_ipcs_response_send(struct qb_ipcs_connection *c, const void *data, size_t size) { ssize_t res; if (c == NULL) { return -EINVAL; } qb_ipcs_connection_ref(c); res = c->service->funcs.send(&c->response, data, size); if (res == size) { c->stats.responses++; } else if (res == -EAGAIN || res == -ETIMEDOUT) { struct qb_ipc_one_way *ow = _response_sock_one_way_get(c); if (ow) { ssize_t res2 = qb_ipc_us_ready(ow, &c->setup, 0, POLLOUT); if (res2 < 0) { res = res2; } } c->stats.send_retries++; } qb_ipcs_connection_unref(c); return res; } ssize_t qb_ipcs_response_sendv(struct qb_ipcs_connection * c, const struct iovec * iov, size_t iov_len) { ssize_t res; if (c == NULL) { return -EINVAL; } qb_ipcs_connection_ref(c); res = c->service->funcs.sendv(&c->response, iov, iov_len); if (res > 0) { c->stats.responses++; } else if (res == -EAGAIN || res == -ETIMEDOUT) { struct qb_ipc_one_way *ow = _response_sock_one_way_get(c); if (ow) { ssize_t res2 = qb_ipc_us_ready(ow, &c->setup, 0, POLLOUT); if (res2 < 0) { res = res2; } } c->stats.send_retries++; } qb_ipcs_connection_unref(c); return res; } static int32_t resend_event_notifications(struct qb_ipcs_connection *c) { ssize_t res = 0; if (!c->service->needs_sock_for_poll) { return res; } if (c->outstanding_notifiers > 0) { res = qb_ipc_us_send(&c->setup, c->receive_buf, c->outstanding_notifiers); } if (res > 0) { c->outstanding_notifiers -= res; } assert(c->outstanding_notifiers >= 0); if (c->outstanding_notifiers == 0) { c->poll_events = POLLIN | POLLPRI | POLLNVAL; (void)_modify_dispatch_descriptor_(c); } return res; } static int32_t new_event_notification(struct qb_ipcs_connection * c) { ssize_t res = 0; if (!c->service->needs_sock_for_poll) { return res; } assert(c->outstanding_notifiers >= 0); if (c->outstanding_notifiers > 0) { c->outstanding_notifiers++; res = resend_event_notifications(c); } else { res = qb_ipc_us_send(&c->setup, &c->outstanding_notifiers, 1); if (res == -EAGAIN) { /* * notify the client later, when we can. */ c->outstanding_notifiers++; c->poll_events = POLLOUT | POLLIN | POLLPRI | POLLNVAL; (void)_modify_dispatch_descriptor_(c); } } return res; } ssize_t qb_ipcs_event_send(struct qb_ipcs_connection * c, const void *data, size_t size) { ssize_t res; ssize_t resn; if (c == NULL) { return -EINVAL; } else if (size > c->event.max_msg_size) { return -EMSGSIZE; } qb_ipcs_connection_ref(c); res = c->service->funcs.send(&c->event, data, size); if (res == size) { c->stats.events++; resn = new_event_notification(c); if (resn < 0 && resn != -EAGAIN && resn != -ENOBUFS) { errno = -resn; qb_util_perror(LOG_WARNING, "new_event_notification (%s)", c->description); res = resn; } } else if (res == -EAGAIN || res == -ETIMEDOUT) { struct qb_ipc_one_way *ow = _event_sock_one_way_get(c); if (c->outstanding_notifiers > 0) { resn = resend_event_notifications(c); } if (ow) { resn = qb_ipc_us_ready(ow, &c->setup, 0, POLLOUT); if (resn < 0) { res = resn; } } c->stats.send_retries++; } qb_ipcs_connection_unref(c); return res; } ssize_t qb_ipcs_event_sendv(struct qb_ipcs_connection * c, const struct iovec * iov, size_t iov_len) { ssize_t res; ssize_t resn; if (c == NULL) { return -EINVAL; } qb_ipcs_connection_ref(c); res = c->service->funcs.sendv(&c->event, iov, iov_len); if (res > 0) { c->stats.events++; resn = new_event_notification(c); if (resn < 0 && resn != -EAGAIN) { errno = -resn; qb_util_perror(LOG_WARNING, "new_event_notification (%s)", c->description); res = resn; } } else if (res == -EAGAIN || res == -ETIMEDOUT) { struct qb_ipc_one_way *ow = _event_sock_one_way_get(c); if (c->outstanding_notifiers > 0) { resn = resend_event_notifications(c); } if (ow) { resn = qb_ipc_us_ready(ow, &c->setup, 0, POLLOUT); if (resn < 0) { res = resn; } } c->stats.send_retries++; } qb_ipcs_connection_unref(c); return res; } qb_ipcs_connection_t * qb_ipcs_connection_first_get(struct qb_ipcs_service * s) { struct qb_ipcs_connection *c; if (qb_list_empty(&s->connections)) { return NULL; } c = qb_list_first_entry(&s->connections, struct qb_ipcs_connection, list); qb_ipcs_connection_ref(c); return c; } qb_ipcs_connection_t * qb_ipcs_connection_next_get(struct qb_ipcs_service * s, struct qb_ipcs_connection * current) { struct qb_ipcs_connection *c; if (current == NULL || qb_list_is_last(¤t->list, &s->connections)) { return NULL; } c = qb_list_first_entry(¤t->list, struct qb_ipcs_connection, list); qb_ipcs_connection_ref(c); return c; } int32_t qb_ipcs_service_id_get(struct qb_ipcs_connection * c) { if (c == NULL) { return -EINVAL; } return c->service->service_id; } struct qb_ipcs_connection * qb_ipcs_connection_alloc(struct qb_ipcs_service *s) { struct qb_ipcs_connection *c = calloc(1, sizeof(struct qb_ipcs_connection)); if (c == NULL) { return NULL; } c->pid = 0; c->euid = -1; c->egid = -1; c->receive_buf = NULL; c->context = NULL; c->fc_enabled = QB_FALSE; c->state = QB_IPCS_CONNECTION_INACTIVE; c->poll_events = POLLIN | POLLPRI | POLLNVAL; c->setup.type = s->type; c->request.type = s->type; c->response.type = s->type; c->event.type = s->type; (void)strlcpy(c->description, "not set yet", CONNECTION_DESCRIPTION); /* initial alloc ref */ qb_ipcs_connection_ref(c); /* * The connection makes use of the service object. Give the connection * a reference to the service so we know the service can never be destroyed * until the connection is done with it. */ qb_ipcs_ref(s); c->service = s; qb_list_init(&c->list); return c; } void qb_ipcs_connection_ref(struct qb_ipcs_connection *c) { if (c) { qb_atomic_int_inc(&c->refcount); } } void qb_ipcs_connection_unref(struct qb_ipcs_connection *c) { int32_t free_it; if (c == NULL) { return; } if (c->refcount < 1) { qb_util_log(LOG_ERR, "ref:%d state:%d (%s)", c->refcount, c->state, c->description); assert(0); } free_it = qb_atomic_int_dec_and_test(&c->refcount); if (free_it) { qb_list_del(&c->list); if (c->service->serv_fns.connection_destroyed) { c->service->serv_fns.connection_destroyed(c); } c->service->funcs.disconnect(c); /* Let go of the connection's reference to the service */ qb_ipcs_unref(c->service); free(c->receive_buf); free(c); } } void qb_ipcs_disconnect(struct qb_ipcs_connection *c) { int32_t res = 0; qb_loop_job_dispatch_fn rerun_job; if (c == NULL) { return; } qb_util_log(LOG_DEBUG, "%s(%s) state:%d", __func__, c->description, c->state); if (c->state == QB_IPCS_CONNECTION_ACTIVE) { c->service->funcs.disconnect(c); c->state = QB_IPCS_CONNECTION_INACTIVE; c->service->stats.closed_connections++; /* return early as it's an incomplete connection. */ return; } if (c->state == QB_IPCS_CONNECTION_ESTABLISHED) { c->service->funcs.disconnect(c); c->state = QB_IPCS_CONNECTION_SHUTTING_DOWN; c->service->stats.active_connections--; c->service->stats.closed_connections++; } if (c->state == QB_IPCS_CONNECTION_SHUTTING_DOWN) { int scheduled_retry = 0; res = 0; if (c->service->serv_fns.connection_closed) { res = c->service->serv_fns.connection_closed(c); } if (res != 0) { /* OK, so they want the connection_closed * function re-run */ rerun_job = (qb_loop_job_dispatch_fn) qb_ipcs_disconnect; res = c->service->poll_fns.job_add(QB_LOOP_LOW, c, rerun_job); if (res == 0) { /* this function is going to be called again. * so hold off on the unref */ scheduled_retry = 1; } } if (scheduled_retry == 0) { /* This removes the initial alloc ref */ qb_ipcs_connection_unref(c); } } } static void qb_ipcs_flowcontrol_set(struct qb_ipcs_connection *c, int32_t fc_enable) { if (c == NULL) { return; } if (c->fc_enabled != fc_enable) { c->service->funcs.fc_set(&c->request, fc_enable); c->fc_enabled = fc_enable; c->stats.flow_control_state = fc_enable; c->stats.flow_control_count++; } } static int32_t _process_request_(struct qb_ipcs_connection *c, int32_t ms_timeout) { int32_t res = 0; ssize_t size; struct qb_ipc_request_header *hdr; qb_ipcs_connection_ref(c); if (c->service->funcs.peek && c->service->funcs.reclaim) { size = c->service->funcs.peek(&c->request, (void **)&hdr, ms_timeout); } else { hdr = c->receive_buf; size = c->service->funcs.recv(&c->request, hdr, c->request.max_msg_size, ms_timeout); } if (size < 0) { if (size != -EAGAIN && size != -ETIMEDOUT) { qb_util_perror(LOG_DEBUG, "recv from client connection failed (%s)", c->description); } else { c->stats.recv_retries++; } res = size; goto cleanup; } else if (size == 0 || hdr->id == QB_IPC_MSG_DISCONNECT) { qb_util_log(LOG_DEBUG, "client requesting a disconnect (%s)", c->description); - qb_ipcs_disconnect(c); - c = NULL; res = -ESHUTDOWN; + goto cleanup; } else { c->stats.requests++; res = c->service->serv_fns.msg_process(c, hdr, hdr->size); /* 0 == good, negative == backoff */ if (res < 0) { res = -ENOBUFS; } else { res = size; } } if (c && c->service->funcs.peek && c->service->funcs.reclaim) { c->service->funcs.reclaim(&c->request); } cleanup: qb_ipcs_connection_unref(c); return res; } #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) -{ - int32_t res = _process_request_((struct qb_ipcs_connection *)data, - IPC_REQUEST_TIMEOUT); - if (res > 0) { - return 0; - } - return res; -} - 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) { return q_len; } if (c->service->poll_priority == QB_LOOP_MED) { q_len = QB_MIN(q_len, 5); } else if (c->service->poll_priority == QB_LOOP_LOW) { q_len = 1; } else { q_len = QB_MIN(q_len, MAX_RECV_MSGS); } } else { q_len = 1; } return q_len; } int32_t qb_ipcs_dispatch_connection_request(int32_t fd, int32_t revents, void *data) { struct qb_ipcs_connection *c = (struct qb_ipcs_connection *)data; char bytes[MAX_RECV_MSGS]; int32_t res = 0; int32_t res2; int32_t recvd = 0; ssize_t avail; if (revents & POLLNVAL) { qb_util_log(LOG_DEBUG, "NVAL conn (%s)", c->description); res = -EINVAL; goto dispatch_cleanup; } if (revents & POLLHUP) { qb_util_log(LOG_DEBUG, "HUP conn (%s)", c->description); res = -ESHUTDOWN; goto dispatch_cleanup; } if (revents & POLLOUT) { /* try resend events now that fd can write */ res = resend_event_notifications(c); if (res < 0 && res != -EAGAIN) { errno = -res; qb_util_perror(LOG_WARNING, "resend_event_notifications (%s)", c->description); } /* nothing to read */ if ((revents & POLLIN) == 0) { res = 0; goto dispatch_cleanup; } } if (c->fc_enabled) { res = 0; goto dispatch_cleanup; } avail = _request_q_len_get(c); if (c->service->needs_sock_for_poll && avail == 0) { res2 = qb_ipc_us_recv(&c->setup, bytes, 1, 0); if (qb_ipc_us_sock_error_is_disconnected(res2)) { errno = -res2; qb_util_perror(LOG_WARNING, "conn (%s) disconnected", c->description); res = -ESHUTDOWN; goto dispatch_cleanup; } else { qb_util_log(LOG_WARNING, "conn (%s) Nothing in q but got POLLIN on fd:%d (res2:%d)", c->description, fd, res2); res = 0; goto dispatch_cleanup; } } do { res = _process_request_(c, IPC_REQUEST_TIMEOUT); + + if (res == -ESHUTDOWN) { + goto dispatch_cleanup; + } + if (res > 0 || res == -ENOBUFS || res == -EINVAL) { recvd++; } if (res > 0) { avail--; } } while (avail > 0 && res > 0 && !c->fc_enabled); if (c->service->needs_sock_for_poll && recvd > 0) { res2 = qb_ipc_us_recv(&c->setup, bytes, recvd, -1); if (qb_ipc_us_sock_error_is_disconnected(res2)) { errno = -res2; qb_util_perror(LOG_ERR, "error receiving from setup sock (%s)", c->description); res = -ESHUTDOWN; goto dispatch_cleanup; } } res = QB_MIN(0, res); if (res == -EAGAIN || res == -ETIMEDOUT || res == -ENOBUFS) { res = 0; } if (res != 0) { if (res != -ENOTCONN) { /* * Abnormal state (ENOTCONN is normal shutdown). */ errno = -res; qb_util_perror(LOG_ERR, "request returned error (%s)", c->description); } } dispatch_cleanup: if (res != 0) { qb_ipcs_disconnect(c); } return res; } void qb_ipcs_context_set(struct qb_ipcs_connection *c, void *context) { if (c == NULL) { return; } c->context = context; } void * qb_ipcs_context_get(struct qb_ipcs_connection *c) { if (c == NULL) { return NULL; } return c->context; } void * qb_ipcs_connection_service_context_get(qb_ipcs_connection_t *c) { if (c == NULL || c->service == NULL) { return NULL; } return c->service->context; } int32_t qb_ipcs_connection_stats_get(qb_ipcs_connection_t * c, struct qb_ipcs_connection_stats * stats, int32_t clear_after_read) { if (c == NULL) { return -EINVAL; } memcpy(stats, &c->stats, sizeof(struct qb_ipcs_connection_stats)); if (clear_after_read) { memset(&c->stats, 0, sizeof(struct qb_ipcs_connection_stats_2)); c->stats.client_pid = c->pid; } return 0; } struct qb_ipcs_connection_stats_2* qb_ipcs_connection_stats_get_2(qb_ipcs_connection_t *c, int32_t clear_after_read) { struct qb_ipcs_connection_stats_2 * stats; if (c == NULL) { errno = EINVAL; return NULL; } stats = calloc(1, sizeof(struct qb_ipcs_connection_stats_2)); if (stats == NULL) { return NULL; } memcpy(stats, &c->stats, sizeof(struct qb_ipcs_connection_stats_2)); if (c->service->funcs.q_len_get) { stats->event_q_length = c->service->funcs.q_len_get(&c->event); } else { stats->event_q_length = 0; } if (clear_after_read) { memset(&c->stats, 0, sizeof(struct qb_ipcs_connection_stats_2)); c->stats.client_pid = c->pid; } return stats; } int32_t qb_ipcs_stats_get(struct qb_ipcs_service * s, struct qb_ipcs_stats * stats, int32_t clear_after_read) { if (s == NULL) { return -EINVAL; } memcpy(stats, &s->stats, sizeof(struct qb_ipcs_stats)); if (clear_after_read) { memset(&s->stats, 0, sizeof(struct qb_ipcs_stats)); } return 0; } void qb_ipcs_connection_auth_set(qb_ipcs_connection_t *c, uid_t uid, gid_t gid, mode_t mode) { if (c) { c->auth.uid = uid; c->auth.gid = gid; c->auth.mode = mode; } } diff --git a/lib/log_thread.c b/lib/log_thread.c index 0a97c6e..56008f8 100644 --- a/lib/log_thread.c +++ b/lib/log_thread.c @@ -1,279 +1,279 @@ /* * Copyright (C) 2011 Red Hat, Inc. * * All rights reserved. * * Author: Angus Salkeld * * 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 #include #include #include #include "log_int.h" static int wthread_active = QB_FALSE; static int wthread_should_exit = QB_FALSE; static qb_thread_lock_t *logt_wthread_lock = NULL; static QB_LIST_DECLARE(logt_print_finished_records); static int logt_memory_used = 0; static int logt_dropped_messages = 0; static sem_t logt_thread_start; static sem_t logt_print_finished; static int logt_sched_param_queued = QB_FALSE; static int logt_sched_policy; #if defined(HAVE_PTHREAD_SETSCHEDPARAM) && defined(HAVE_SCHED_GET_PRIORITY_MAX) static struct sched_param logt_sched_param; #endif /* HAVE_PTHREAD_SETSCHEDPARAM && HAVE_SCHED_GET_PRIORITY_MAX */ static pthread_t logt_thread_id = 0; static void *qb_logt_worker_thread(void *data) __attribute__ ((noreturn)); static void * qb_logt_worker_thread(void *data) { struct qb_log_record *rec; int dropped = 0; int res; /* * Signal wthread_create that the initialization process may continue */ sem_post(&logt_thread_start); for (;;) { retry_sem_wait: res = sem_wait(&logt_print_finished); if (res == -1 && errno == EINTR) { goto retry_sem_wait; } else if (res == -1) { /* * This case shouldn't happen */ pthread_exit(NULL); } (void)qb_thread_lock(logt_wthread_lock); if (wthread_should_exit) { int value = -1; (void)sem_getvalue(&logt_print_finished, &value); if (value == 0) { (void)qb_thread_unlock(logt_wthread_lock); pthread_exit(NULL); } } rec = qb_list_first_entry(&logt_print_finished_records, struct qb_log_record, list); qb_list_del(&rec->list); logt_memory_used = logt_memory_used - strlen(rec->buffer) - sizeof(struct qb_log_record) - 1; dropped = logt_dropped_messages; logt_dropped_messages = 0; (void)qb_thread_unlock(logt_wthread_lock); if (dropped) { printf("%d messages lost\n", dropped); } qb_log_thread_log_write(rec->cs, rec->timestamp, rec->buffer); free(rec->buffer); free(rec); } } int32_t qb_log_thread_priority_set(int32_t policy, int32_t priority) { int res = 0; #if defined(HAVE_PTHREAD_SETSCHEDPARAM) && defined(HAVE_SCHED_GET_PRIORITY_MAX) logt_sched_policy = policy; if (policy == SCHED_OTHER #ifdef SCHED_IDLE || policy == SCHED_IDLE #endif #if defined(SCHED_BATCH) && !defined(QB_DARWIN) || policy == SCHED_BATCH #endif ) { logt_sched_param.sched_priority = 0; } else { logt_sched_param.sched_priority = priority; } if (wthread_active == QB_FALSE) { logt_sched_param_queued = QB_TRUE; } else { res = pthread_setschedparam(logt_thread_id, policy, &logt_sched_param); if (res != 0) { res = -res; } } #endif return res; } int32_t qb_log_thread_start(void) { int res; if (wthread_active) { return 0; } wthread_active = QB_TRUE; sem_init(&logt_thread_start, 0, 0); sem_init(&logt_print_finished, 0, 0); res = pthread_create(&logt_thread_id, NULL, qb_logt_worker_thread, NULL); if (res != 0) { wthread_active = QB_FALSE; return -res; } sem_wait(&logt_thread_start); if (logt_sched_param_queued) { res = qb_log_thread_priority_set(logt_sched_policy, logt_sched_param.sched_priority); if (res != 0) { goto cleanup_pthread; } logt_sched_param_queued = QB_FALSE; } logt_wthread_lock = qb_thread_lock_create(QB_THREAD_LOCK_SHORT); if (logt_wthread_lock == NULL) { goto cleanup_pthread; } return 0; cleanup_pthread: wthread_should_exit = QB_TRUE; sem_post(&logt_print_finished); pthread_join(logt_thread_id, NULL); sem_destroy(&logt_print_finished); sem_destroy(&logt_thread_start); return res; } void qb_log_thread_log_post(struct qb_log_callsite *cs, time_t timestamp, const char *buffer) { struct qb_log_record *rec; size_t buf_size; size_t total_size; rec = malloc(sizeof(struct qb_log_record)); if (rec == NULL) { return; } buf_size = strlen(buffer) + 1; total_size = sizeof(struct qb_log_record) + buf_size; rec->cs = cs; rec->buffer = malloc(buf_size); if (rec->buffer == NULL) { goto free_record; } memcpy(rec->buffer, buffer, buf_size); rec->timestamp = timestamp; qb_list_init(&rec->list); (void)qb_thread_lock(logt_wthread_lock); logt_memory_used += total_size; if (logt_memory_used > 512000) { free(rec->buffer); free(rec); logt_memory_used = logt_memory_used - total_size; logt_dropped_messages += 1; (void)qb_thread_unlock(logt_wthread_lock); return; } else { qb_list_add_tail(&rec->list, &logt_print_finished_records); } (void)qb_thread_unlock(logt_wthread_lock); sem_post(&logt_print_finished); return; free_record: free(rec); } void qb_log_thread_stop(void) { int res; int value; struct qb_log_record *rec; if (wthread_active == QB_FALSE && logt_wthread_lock == NULL) { return; } if (wthread_active == QB_FALSE) { for (;;) { res = sem_getvalue(&logt_print_finished, &value); if (res != 0 || value == 0) { - return; + break; } sem_wait(&logt_print_finished); (void)qb_thread_lock(logt_wthread_lock); rec = qb_list_first_entry(&logt_print_finished_records, struct qb_log_record, list); qb_list_del(&rec->list); logt_memory_used = logt_memory_used - strlen(rec->buffer) - sizeof(struct qb_log_record) - 1; (void)qb_thread_unlock(logt_wthread_lock); qb_log_thread_log_write(rec->cs, rec->timestamp, rec->buffer); free(rec->buffer); free(rec); } } else { wthread_should_exit = QB_TRUE; sem_post(&logt_print_finished); pthread_join(logt_thread_id, NULL); } (void)qb_thread_lock_destroy(logt_wthread_lock); sem_destroy(&logt_print_finished); sem_destroy(&logt_thread_start); } diff --git a/lib/ringbuffer.c b/lib/ringbuffer.c index 21e7821..524b0db 100644 --- a/lib/ringbuffer.c +++ b/lib/ringbuffer.c @@ -1,950 +1,949 @@ /* * Copyright (C) 2010-2011 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 void _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_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; } 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, "qb-%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, "qb-%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, "shm size:%zd; real_size:%zd; 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); - close(fd_data); return rb; cleanup_data: - close(fd_data); if (flags & QB_RB_FLAG_CREATE) { unlink(rb->shared_hdr->data_path); } cleanup_hdr: if (fd_hdr >= 0) { close(fd_hdr); } if (rb && (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 && rb->shared_hdr != NULL)) { 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); if (rb->flags & QB_RB_FLAG_CREATE) { if (rb->notifier.destroy_fn) { (void)rb->notifier.destroy_fn(rb->notifier.instance); } unlink(rb->shared_hdr->data_path); unlink(rb->shared_hdr->hdr_path); qb_util_log(LOG_DEBUG, "Free'ing ringbuffer: %s", rb->shared_hdr->hdr_path); } else { qb_util_log(LOG_DEBUG, "Closing ringbuffer: %s", rb->shared_hdr->hdr_path); } munmap(rb->shared_data, (rb->shared_hdr->word_size * sizeof(uint32_t)) << 1); munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s)); free(rb); } void qb_rb_force_close(struct qb_ringbuffer_s * rb) { if (rb == NULL) { return; } qb_enter(); if (rb->notifier.destroy_fn) { (void)rb->notifier.destroy_fn(rb->notifier.instance); } errno = 0; unlink(rb->shared_hdr->data_path); qb_util_perror(LOG_DEBUG, "Force free'ing ringbuffer: %s", rb->shared_hdr->data_path); errno = 0; unlink(rb->shared_hdr->hdr_path); qb_util_perror(LOG_DEBUG, "Force free'ing ringbuffer: %s", rb->shared_hdr->hdr_path); munmap(rb->shared_data, (rb->shared_hdr->word_size * sizeof(uint32_t)) << 1); munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s)); free(rb); } 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)) { _rb_chunk_reclaim(rb); } } 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 void _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; 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) { return; } 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) { int 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); } 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); } return 0; } 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"); } printf(" ->write_pt [%d]\n", rb->shared_hdr->write_pt); printf(" ->read_pt [%d]\n", rb->shared_hdr->read_pt); printf(" ->size [%d words]\n", rb->shared_hdr->word_size); #ifndef S_SPLINT_S printf(" =>free [%zu bytes]\n", qb_rb_space_free(rb)); printf(" =>used [%zu 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; if (fd < 0) { 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; /* * 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; /* * 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)); rb = qb_rb_open("create_from_file", n_required, 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/unix.c b/lib/unix.c index 50fe6fe..79ed538 100644 --- a/lib/unix.c +++ b/lib/unix.c @@ -1,494 +1,494 @@ /* * Copyright (C) 2011 Red Hat, Inc. * * All rights reserved. * * Author: Angus Salkeld * * 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" #ifdef HAVE_SYS_SHM_H #include #endif #ifdef HAVE_SYS_MMAN_H #include #endif #include "util_int.h" #include #include #include #if defined(MAP_ANON) && ! defined(MAP_ANONYMOUS) /* * BSD derivatives usually have MAP_ANON, not MAP_ANONYMOUS **/ #define MAP_ANONYMOUS MAP_ANON #endif char * qb_strerror_r(int errnum, char *buf, size_t buflen) { #ifdef STRERROR_R_CHAR_P return strerror_r(errnum, buf, buflen); #else if (strerror_r(errnum, buf, buflen) != 0) { buf[0] = '\0'; } return buf; #endif /* STRERROR_R_CHAR_P */ } static int32_t open_mmap_file(char *path, uint32_t file_flags) { if (strstr(path, "XXXXXX") != NULL) { mode_t old_mode = umask(077); int32_t temp_fd = mkstemp(path); (void)umask(old_mode); return temp_fd; } return open(path, file_flags, 0600); } int32_t qb_sys_mmap_file_open(char *path, const char *file, size_t bytes, uint32_t file_flags) { int32_t fd; int32_t i; int32_t res = 0; ssize_t written; char *buffer = NULL; char *is_absolute = strchr(file, '/'); if (is_absolute) { (void)strlcpy(path, file, PATH_MAX); } else { #if defined(QB_LINUX) || defined(QB_CYGWIN) snprintf(path, PATH_MAX, "/dev/shm/%s", file); #else snprintf(path, PATH_MAX, LOCALSTATEDIR "/run/%s", file); is_absolute = path; #endif } fd = open_mmap_file(path, file_flags); if (fd < 0 && !is_absolute) { qb_util_perror(LOG_ERR, "couldn't open file %s", path); snprintf(path, PATH_MAX, LOCALSTATEDIR "/run/%s", file); fd = open_mmap_file(path, file_flags); if (fd < 0) { res = -errno; qb_util_perror(LOG_ERR, "couldn't open file %s", path); return res; } } else if (fd < 0 && is_absolute) { res = -errno; qb_util_perror(LOG_ERR, "couldn't open file %s", path); return res; } if (ftruncate(fd, bytes) == -1) { res = -errno; qb_util_perror(LOG_ERR, "couldn't truncate file %s", path); goto unlink_exit; } if (file_flags & O_CREAT) { long page_size = sysconf(_SC_PAGESIZE); long write_size = QB_MIN(page_size, bytes); if (page_size < 0) { res = -errno; goto unlink_exit; } buffer = calloc(1, write_size); if (buffer == NULL) { res = -ENOMEM; goto unlink_exit; } for (i = 0; i < (bytes / write_size); i++) { retry_write: written = write(fd, buffer, write_size); if (written == -1 && errno == EINTR) { goto retry_write; } if (written != write_size) { res = -ENOSPC; free(buffer); goto unlink_exit; } } free(buffer); } return fd; unlink_exit: unlink(path); - if (fd > 0) { + if (fd >= 0) { close(fd); } return res; } int32_t qb_sys_circular_mmap(int32_t fd, void **buf, size_t bytes) { void *addr_orig = NULL; void *addr; void *addr_next; int32_t res; int flags = MAP_ANONYMOUS; #ifdef QB_FORCE_SHM_ALIGN /* On a number of arches any fixed and shared mmap() mapping address * must be aligned to 16k. If the first mmap() below is not shared then * the first mmap() will succeed because these restrictions do not apply to * private mappings. The second mmap() wants a shared memory mapping but * the address returned by the first one is only page-aligned and not * aligned to 16k. */ flags |= MAP_SHARED; #else flags |= MAP_PRIVATE; #endif /* QB_FORCE_SHM_ALIGN */ addr_orig = mmap(NULL, bytes << 1, PROT_NONE, flags, -1, 0); if (addr_orig == MAP_FAILED) { return -errno; } addr = mmap(addr_orig, bytes, PROT_READ | PROT_WRITE, MAP_FIXED | MAP_SHARED, fd, 0); if (addr != addr_orig) { res = -errno; goto cleanup_fail; } #if defined(QB_BSD) && defined(MADV_NOSYNC) madvise(addr_orig, bytes, MADV_NOSYNC); #endif addr_next = ((char *)addr_orig) + bytes; addr = mmap(addr_next, bytes, PROT_READ | PROT_WRITE, MAP_FIXED | MAP_SHARED, fd, 0); if (addr != addr_next) { res = -errno; goto cleanup_fail; } #if defined(QB_BSD) && defined(MADV_NOSYNC) madvise(((char *)addr_orig) + bytes, bytes, MADV_NOSYNC); #endif res = close(fd); if (res) { goto cleanup_fail; } *buf = addr_orig; return 0; cleanup_fail: if (addr_orig) { munmap(addr_orig, bytes << 1); } close(fd); return res; } int32_t qb_sys_fd_nonblock_cloexec_set(int32_t fd) { int32_t res = 0; int32_t oldflags = fcntl(fd, F_GETFD, 0); if (oldflags < 0) { oldflags = 0; } oldflags |= FD_CLOEXEC; res = fcntl(fd, F_SETFD, oldflags); if (res == -1) { res = -errno; qb_util_perror(LOG_ERR, "Could not set close-on-exit on fd:%d", fd); return res; } res = fcntl(fd, F_SETFL, O_NONBLOCK); if (res == -1) { res = -errno; qb_util_log(LOG_ERR, "Could not set non-blocking on fd:%d", fd); } return res; } void qb_sigpipe_ctl(enum qb_sigpipe_ctl ctl) { #if !defined(HAVE_MSG_NOSIGNAL) && !defined(HAVE_SO_NOSIGPIPE) struct sigaction act; struct sigaction oact; act.sa_handler = SIG_IGN; if (ctl == QB_SIGPIPE_IGNORE) { sigaction(SIGPIPE, &act, &oact); } else { sigaction(SIGPIPE, &oact, NULL); } #endif /* !MSG_NOSIGNAL && !SO_NOSIGPIPE */ } void qb_socket_nosigpipe(int32_t s) { #if !defined(HAVE_MSG_NOSIGNAL) && defined(HAVE_SO_NOSIGPIPE) int32_t on = 1; setsockopt(s, SOL_SOCKET, SO_NOSIGPIPE, (void *)&on, sizeof(on)); #endif /* !MSG_NOSIGNAL && SO_NOSIGPIPE */ } /* * atomic operations * -------------------------------------------------------------------------- */ #ifndef HAVE_GCC_BUILTINS_FOR_SYNC_OPERATIONS /* * We have to use the slow, but safe locking method */ static qb_thread_lock_t *qb_atomic_mutex = NULL; void qb_atomic_init(void) { if (qb_atomic_mutex == NULL) { qb_atomic_mutex = qb_thread_lock_create(QB_THREAD_LOCK_SHORT); } assert(qb_atomic_mutex); } int32_t qb_atomic_int_exchange_and_add(volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t val) { int32_t result; qb_thread_lock(qb_atomic_mutex); result = *atomic; *atomic += val; qb_thread_unlock(qb_atomic_mutex); return result; } void qb_atomic_int_add(volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t val) { qb_thread_lock(qb_atomic_mutex); *atomic += val; qb_thread_unlock(qb_atomic_mutex); } int32_t qb_atomic_int_compare_and_exchange(volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t oldval, int32_t newval) { int32_t result; qb_thread_lock(qb_atomic_mutex); if (*atomic == oldval) { result = QB_TRUE; *atomic = newval; } else { result = QB_FALSE; } qb_thread_unlock(qb_atomic_mutex); return result; } int32_t qb_atomic_pointer_compare_and_exchange(volatile void *QB_GNUC_MAY_ALIAS * atomic, void *oldval, void *newval) { int32_t result; qb_thread_lock(qb_atomic_mutex); if (*atomic == oldval) { result = QB_TRUE; *atomic = newval; } else { result = QB_FALSE; } qb_thread_unlock(qb_atomic_mutex); return result; } #ifdef QB_ATOMIC_OP_MEMORY_BARRIER_NEEDED int32_t (qb_atomic_int_get) (volatile int32_t QB_GNUC_MAY_ALIAS * atomic) { int32_t result; qb_thread_lock(qb_atomic_mutex); result = *atomic; qb_thread_unlock(qb_atomic_mutex); return result; } void (qb_atomic_int_set) (volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t newval) { qb_thread_lock(qb_atomic_mutex); *atomic = newval; qb_thread_unlock(qb_atomic_mutex); } void * (qb_atomic_pointer_get) (volatile void *QB_GNUC_MAY_ALIAS * atomic) { void *result; qb_thread_lock(qb_atomic_mutex); result = (void*)*atomic; qb_thread_unlock(qb_atomic_mutex); return result; } void (qb_atomic_pointer_set) (volatile void *QB_GNUC_MAY_ALIAS * atomic, void *newval) { qb_thread_lock(qb_atomic_mutex); *atomic = newval; qb_thread_unlock(qb_atomic_mutex); } #endif /* QB_ATOMIC_OP_MEMORY_BARRIER_NEEDED */ #else /* * gcc built-ins */ void qb_atomic_init(void) { } int32_t qb_atomic_int_exchange_and_add(volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t val) { return __sync_fetch_and_add(atomic, val); } void qb_atomic_int_add(volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t val) { __sync_fetch_and_add(atomic, val); } int32_t qb_atomic_int_compare_and_exchange(volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t oldval, int32_t newval) { return __sync_bool_compare_and_swap(atomic, oldval, newval); } int32_t qb_atomic_pointer_compare_and_exchange(volatile void *QB_GNUC_MAY_ALIAS * atomic, void *oldval, void *newval) { return __sync_bool_compare_and_swap(atomic, oldval, newval); } #ifdef QB_ATOMIC_OP_MEMORY_BARRIER_NEEDED #define QB_ATOMIC_MEMORY_BARRIER __sync_synchronize () int32_t (qb_atomic_int_get) (volatile int32_t QB_GNUC_MAY_ALIAS * atomic) { QB_ATOMIC_MEMORY_BARRIER; return *atomic; } void (qb_atomic_int_set) (volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t newval) { *atomic = newval; QB_ATOMIC_MEMORY_BARRIER; } void * (qb_atomic_pointer_get) (volatile void *QB_GNUC_MAY_ALIAS * atomic) { QB_ATOMIC_MEMORY_BARRIER; return (void*)*atomic; } void (qb_atomic_pointer_set) (volatile void *QB_GNUC_MAY_ALIAS * atomic, void *newval) { *atomic = newval; QB_ATOMIC_MEMORY_BARRIER; } #endif /* QB_ATOMIC_OP_MEMORY_BARRIER_NEEDED */ #endif /* HAVE_GCC_BUILTINS_FOR_SYNC_OPERATIONS */ #ifndef QB_ATOMIC_OP_MEMORY_BARRIER_NEEDED int32_t (qb_atomic_int_get) (volatile int32_t QB_GNUC_MAY_ALIAS * atomic) { return qb_atomic_int_get(atomic); } void (qb_atomic_int_set) (volatile int32_t QB_GNUC_MAY_ALIAS * atomic, int32_t newval) { qb_atomic_int_set(atomic, newval); } void * (qb_atomic_pointer_get) (volatile void *QB_GNUC_MAY_ALIAS * atomic) { return qb_atomic_pointer_get(atomic); } void (qb_atomic_pointer_set) (volatile void *QB_GNUC_MAY_ALIAS * atomic, void *newval) { qb_atomic_pointer_set(atomic, newval); } #endif /* !QB_ATOMIC_OP_MEMORY_BARRIER_NEEDED */ diff --git a/tests/bms.c b/tests/bms.c index fa8c7da..a965599 100644 --- a/tests/bms.c +++ b/tests/bms.c @@ -1,359 +1,361 @@ /* * Copyright (c) 2006-2009 Red Hat, Inc. * * All rights reserved. * * Author: Steven Dake * * 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 #include #include #include #include #ifdef HAVE_GLIB #include #endif int32_t blocking = QB_TRUE; int32_t events = QB_FALSE; int32_t use_glib = QB_FALSE; int32_t verbose = 0; static qb_loop_t *bms_loop; #ifdef HAVE_GLIB static GMainLoop *glib_loop; static qb_array_t *gio_map; #endif static qb_ipcs_service_t* s1; static int32_t s1_connection_accept_fn(qb_ipcs_connection_t *c, uid_t uid, gid_t gid) { #if 0 if (uid == 0 && gid == 0) { if (verbose) { qb_log(LOG_INFO, "%s:%d %s authenticated connection\n", __FILE__, __LINE__, __func__); } return 1; } qb_log(LOG_INFO, "%s:%d %s() BAD user!\n", __FILE__, __LINE__, __func__); return 0; #else return 0; #endif } static void s1_connection_created_fn(qb_ipcs_connection_t *c) { struct qb_ipcs_stats srv_stats; qb_ipcs_stats_get(s1, &srv_stats, QB_FALSE); qb_log(LOG_NOTICE, "Connection created > active:%d > closed:%d", srv_stats.active_connections, srv_stats.closed_connections); } static void s1_connection_destroyed_fn(qb_ipcs_connection_t *c) { qb_log(LOG_INFO, "connection about to be freed\n"); } static int32_t s1_connection_closed_fn(qb_ipcs_connection_t *c) { struct qb_ipcs_connection_stats stats; struct qb_ipcs_stats srv_stats; qb_ipcs_stats_get(s1, &srv_stats, QB_FALSE); qb_ipcs_connection_stats_get(c, &stats, QB_FALSE); qb_log(LOG_INFO, "Connection to pid:%d destroyed > active:%d > closed:%d", stats.client_pid, srv_stats.active_connections, srv_stats.closed_connections); qb_log(LOG_INFO, " Requests %"PRIu64"\n", stats.requests); qb_log(LOG_INFO, " Responses %"PRIu64"\n", stats.responses); qb_log(LOG_INFO, " Events %"PRIu64"\n", stats.events); qb_log(LOG_INFO, " Send retries %"PRIu64"\n", stats.send_retries); qb_log(LOG_INFO, " Recv retries %"PRIu64"\n", stats.recv_retries); qb_log(LOG_INFO, " FC state %d\n", stats.flow_control_state); qb_log(LOG_INFO, " FC count %"PRIu64"\n\n", stats.flow_control_count); return 0; } static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c, void *data, size_t size) { struct qb_ipc_request_header *req_pt = (struct qb_ipc_request_header *)data; struct qb_ipc_response_header response; ssize_t res; qb_log(LOG_TRACE, "msg:%d, size:%d", req_pt->id, req_pt->size); response.size = sizeof(struct qb_ipc_response_header); response.id = 13; response.error = 0; if (blocking) { res = qb_ipcs_response_send(c, &response, sizeof(response)); if (res < 0) { qb_perror(LOG_ERR, "qb_ipcs_response_send"); + return res; } } if (events) { res = qb_ipcs_event_send(c, &response, sizeof(response)); if (res < 0) { qb_perror(LOG_ERR, "qb_ipcs_event_send"); + return res; } } return 0; } static void sigusr1_handler(int32_t num) { qb_log(LOG_INFO, "%s(%d)\n", __func__, num); qb_ipcs_destroy(s1); exit(0); } static void show_usage(const char *name) { qb_log(LOG_INFO, "usage: \n"); qb_log(LOG_INFO, "%s \n", name); qb_log(LOG_INFO, "\n"); qb_log(LOG_INFO, " options:\n"); qb_log(LOG_INFO, "\n"); qb_log(LOG_INFO, " -n non-blocking ipc (default blocking)\n"); qb_log(LOG_INFO, " -e send events back instead for responses\n"); qb_log(LOG_INFO, " -v verbose\n"); qb_log(LOG_INFO, " -h show this help text\n"); qb_log(LOG_INFO, " -m use shared memory\n"); qb_log(LOG_INFO, " -p use posix message queues\n"); qb_log(LOG_INFO, " -s use sysv message queues\n"); qb_log(LOG_INFO, " -u use unix sockets\n"); qb_log(LOG_INFO, " -g use glib mainloop\n"); qb_log(LOG_INFO, "\n"); } #ifdef HAVE_GLIB struct gio_to_qb_poll { gboolean is_used; GIOChannel *channel; int32_t events; void * data; qb_ipcs_dispatch_fn_t fn; enum qb_loop_priority p; }; static gboolean gio_read_socket (GIOChannel *gio, GIOCondition condition, gpointer data) { struct gio_to_qb_poll *adaptor = (struct gio_to_qb_poll *)data; gint fd = g_io_channel_unix_get_fd(gio); return (adaptor->fn(fd, condition, adaptor->data) == 0); } static int32_t my_g_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t evts, void *data, qb_ipcs_dispatch_fn_t fn) { struct gio_to_qb_poll *adaptor; GIOChannel *channel; int32_t res = 0; res = qb_array_grow(gio_map, fd + 1); if (res < 0) { return res; } res = qb_array_index(gio_map, fd, (void**)&adaptor); if (res < 0) { return res; } if (adaptor->is_used) { return -EEXIST; } channel = g_io_channel_unix_new(fd); if (!channel) { return -ENOMEM; } adaptor->channel = channel; adaptor->fn = fn; adaptor->events = evts; adaptor->data = data; adaptor->p = p; adaptor->is_used = TRUE; g_io_add_watch(channel, evts, gio_read_socket, adaptor); return 0; } static int32_t my_g_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t evts, void *data, qb_ipcs_dispatch_fn_t fn) { return 0; } static int32_t my_g_dispatch_del(int32_t fd) { struct gio_to_qb_poll *adaptor; if (qb_array_index(gio_map, fd, (void**)&adaptor) == 0) { g_io_channel_unref(adaptor->channel); adaptor->is_used = FALSE; } return 0; } #endif /* HAVE_GLIB */ static int32_t my_job_add(enum qb_loop_priority p, void *data, qb_loop_job_dispatch_fn fn) { return qb_loop_job_add(bms_loop, p, data, fn); } static int32_t my_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t evts, void *data, qb_ipcs_dispatch_fn_t fn) { return qb_loop_poll_add(bms_loop, p, fd, evts, data, fn); } static int32_t my_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t evts, void *data, qb_ipcs_dispatch_fn_t fn) { return qb_loop_poll_mod(bms_loop, p, fd, evts, data, fn); } static int32_t my_dispatch_del(int32_t fd) { return qb_loop_poll_del(bms_loop, fd); } int32_t main(int32_t argc, char *argv[]) { const char *options = "nevhmpsug"; int32_t opt; int32_t rc; enum qb_ipc_type ipc_type = QB_IPC_SHM; struct qb_ipcs_service_handlers sh = { .connection_accept = s1_connection_accept_fn, .connection_created = s1_connection_created_fn, .msg_process = s1_msg_process_fn, .connection_destroyed = s1_connection_destroyed_fn, .connection_closed = s1_connection_closed_fn, }; struct qb_ipcs_poll_handlers ph = { .job_add = my_job_add, .dispatch_add = my_dispatch_add, .dispatch_mod = my_dispatch_mod, .dispatch_del = my_dispatch_del, }; #ifdef HAVE_GLIB struct qb_ipcs_poll_handlers glib_ph = { .job_add = NULL, /* FIXME */ .dispatch_add = my_g_dispatch_add, .dispatch_mod = my_g_dispatch_mod, .dispatch_del = my_g_dispatch_del, }; #endif /* HAVE_GLIB */ while ((opt = getopt(argc, argv, options)) != -1) { switch (opt) { case 'm': ipc_type = QB_IPC_SHM; break; case 'u': ipc_type = QB_IPC_SOCKET; break; case 'n': /* non-blocking */ blocking = QB_FALSE; break; case 'e': /* events */ events = QB_TRUE; break; case 'g': use_glib = QB_TRUE; break; case 'v': verbose++; break; case 'h': default: show_usage(argv[0]); exit(0); break; } } signal(SIGINT, sigusr1_handler); signal(SIGILL, sigusr1_handler); signal(SIGTERM, sigusr1_handler); qb_log_init("bms", LOG_USER, LOG_EMERG); qb_log_ctl(QB_LOG_SYSLOG, QB_LOG_CONF_ENABLED, QB_FALSE); qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD, QB_LOG_FILTER_FILE, "*", LOG_INFO + verbose); qb_log_ctl(QB_LOG_STDERR, QB_LOG_CONF_ENABLED, QB_TRUE); if (!use_glib) { bms_loop = qb_loop_create(); s1 = qb_ipcs_create("bm1", 0, ipc_type, &sh); if (s1 == 0) { qb_perror(LOG_ERR, "qb_ipcs_create"); exit(1); } qb_ipcs_poll_handlers_set(s1, &ph); rc = qb_ipcs_run(s1); if (rc != 0) { errno = -rc; qb_perror(LOG_ERR, "qb_ipcs_run"); exit(1); } qb_loop_run(bms_loop); } else { #ifdef HAVE_GLIB glib_loop = g_main_loop_new(NULL, FALSE); gio_map = qb_array_create(64, sizeof(struct gio_to_qb_poll)); s1 = qb_ipcs_create("bm1", 0, ipc_type, &sh); if (s1 == 0) { qb_perror(LOG_ERR, "qb_ipcs_create"); exit(1); } qb_ipcs_poll_handlers_set(s1, &glib_ph); rc = qb_ipcs_run(s1); if (rc != 0) { errno = -rc; qb_perror(LOG_ERR, "qb_ipcs_run"); exit(1); } g_main_loop_run(glib_loop); #else qb_log(LOG_ERR, "You don't seem to have glib-devel installed.\n"); #endif } return EXIT_SUCCESS; }