X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.cpp;h=528d1451cf769feea1a694550b5220d20f7d852d;hb=8a00688e1d58cc5a2e77eba206ff23bd6105130c;hp=33684addda431e1f70f7a9d767b26bd9dfab8ea5;hpb=4971b7f0243bd3a7a661bcf1cfe95f0f8014b59b;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.cpp b/src/bin/lttng-relayd/main.cpp index 33684addd..528d1451c 100644 --- a/src/bin/lttng-relayd/main.cpp +++ b/src/bin/lttng-relayd/main.cpp @@ -36,45 +36,45 @@ #include #include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include "backward-compatibility-group-by.h" -#include "cmd.h" -#include "connection.h" -#include "ctf-trace.h" -#include "health-relayd.h" -#include "index.h" -#include "live.h" -#include "lttng-relayd.h" -#include "session.h" -#include "sessiond-trace-chunks.h" -#include "stream.h" -#include "tcp_keep_alive.h" -#include "testpoint.h" -#include "tracefile-array.h" -#include "utils.h" -#include "version.h" -#include "viewer-stream.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "backward-compatibility-group-by.hpp" +#include "cmd.hpp" +#include "connection.hpp" +#include "ctf-trace.hpp" +#include "health-relayd.hpp" +#include "index.hpp" +#include "live.hpp" +#include "lttng-relayd.hpp" +#include "session.hpp" +#include "sessiond-trace-chunks.hpp" +#include "stream.hpp" +#include "tcp_keep_alive.hpp" +#include "testpoint.hpp" +#include "tracefile-array.hpp" +#include "utils.hpp" +#include "version.hpp" +#include "viewer-stream.hpp" static const char *help_msg = #ifdef LTTNG_EMBED_HELP @@ -126,12 +126,6 @@ static int tracing_group_name_override; const char * const config_section_name = "relayd"; -/* - * Quit pipe for all threads. This permits a single cancellation point - * for all threads when receiving an event on the pipe. - */ -int thread_quit_pipe[2] = { -1, -1 }; - /* * This pipe is used to inform the worker thread that a command is queued and * ready to be processed. @@ -405,7 +399,8 @@ end: * See config_entry_handler_cb comment in common/config/session-config.h for the * return value conventions. */ -static int config_entry_handler(const struct config_entry *entry, void *unused) +static int config_entry_handler(const struct config_entry *entry, + void *unused __attribute__((unused))) { int ret = 0, i; @@ -676,7 +671,8 @@ static void print_global_objects(void) print_sessions(); } -static int noop_close(void *data, int *fds) +static int noop_close(void *data __attribute__((unused)), + int *fds __attribute__((unused))) { return 0; } @@ -720,10 +716,7 @@ static void relayd_cleanup(void) (void) fd_tracker_util_pipe_close( the_fd_tracker, health_quit_pipe); } - if (thread_quit_pipe[0] != -1) { - (void) fd_tracker_util_pipe_close( - the_fd_tracker, thread_quit_pipe); - } + relayd_close_thread_quit_pipe(); if (sessiond_trace_chunk_registry) { sessiond_trace_chunk_registry_destroy( sessiond_trace_chunk_registry); @@ -746,23 +739,6 @@ static void relayd_cleanup(void) } } -/* - * Write to writable pipe used to notify a thread. - */ -static int notify_thread_pipe(int wpipe) -{ - ssize_t ret; - - ret = lttng_write(wpipe, "!", 1); - if (ret < 1) { - PERROR("write poll pipe"); - goto end; - } - ret = 0; -end: - return ret; -} - static int notify_health_quit_pipe(int *pipe) { ssize_t ret; @@ -786,7 +762,7 @@ int lttng_relay_stop_threads(void) /* Stopping all threads */ DBG("Terminating all threads"); - if (notify_thread_pipe(thread_quit_pipe[1])) { + if (relayd_notify_thread_quit_pipe()) { ERR("write error on thread quit pipe"); retval = -1; } @@ -890,17 +866,6 @@ void lttng_relay_notify_ready(void) } } -/* - * Init thread quit pipe. - * - * Return -1 on error or 0 if all pipes are created. - */ -static int init_thread_quit_pipe(void) -{ - return fd_tracker_util_pipe_open_cloexec( - the_fd_tracker, "Quit pipe", thread_quit_pipe); -} - /* * Init health quit pipe. * @@ -912,52 +877,6 @@ static int init_health_quit_pipe(void) "Health quit pipe", health_quit_pipe); } -/* - * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set. - */ -static int create_named_thread_poll_set(struct lttng_poll_event *events, - int size, const char *name) -{ - int ret; - - if (events == NULL || size == 0) { - ret = -1; - goto error; - } - - ret = fd_tracker_util_poll_create(the_fd_tracker, - name, events, 1, LTTNG_CLOEXEC); - if (ret) { - PERROR("Failed to create \"%s\" poll file descriptor", name); - goto error; - } - - /* Add quit pipe */ - ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR); - if (ret < 0) { - goto error; - } - - return 0; - -error: - return ret; -} - -/* - * Check if the thread quit pipe was triggered. - * - * Return 1 if it was triggered else 0; - */ -static int check_thread_quit_pipe(int fd, uint32_t events) -{ - if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) { - return 1; - } - - return 0; -} - static int create_sock(void *data, int *out_fd) { int ret; @@ -973,7 +892,7 @@ end: return ret; } -static int close_sock(void *data, int *in_fd) +static int close_sock(void *data, int *in_fd __attribute__((unused))) { struct lttcomm_sock *sock = (lttcomm_sock *) data; @@ -1085,10 +1004,10 @@ end: /* * This thread manages the listening for new connections on the network */ -static void *relay_thread_listener(void *data) +static void *relay_thread_listener(void *data __attribute__((unused))) { - int i, ret, pollfd, err = -1; - uint32_t revents, nb_fd; + int i, ret, err = -1; + uint32_t nb_fd; struct lttng_poll_event events; struct lttcomm_sock *control_sock, *data_sock; @@ -1159,15 +1078,15 @@ restart: DBG("Relay new connection received"); for (i = 0; i < nb_fd; i++) { - health_code_update(); - /* Fetch once the poll data */ - revents = LTTNG_POLL_GETEV(&events, i); - pollfd = LTTNG_POLL_GETFD(&events, i); + const auto revents = LTTNG_POLL_GETEV(&events, i); + const auto pollfd = LTTNG_POLL_GETFD(&events, i); - /* Thread quit pipe has been closed. Killing thread. */ - ret = check_thread_quit_pipe(pollfd, revents); - if (ret) { + health_code_update(); + + /* Activity on thread quit pipe, exiting. */ + if (relayd_is_thread_quit_pipe(pollfd)) { + DBG("Activity on thread quit pipe"); err = 0; goto exit; } @@ -1289,7 +1208,7 @@ error_sock_control: /* * This thread manages the dispatching of the requests to worker threads */ -static void *relay_thread_dispatcher(void *data) +static void *relay_thread_dispatcher(void *data __attribute__((unused))) { int err = -1; ssize_t ret; @@ -1327,7 +1246,7 @@ static void *relay_thread_dispatcher(void *data) /* Continue thread execution */ break; } - new_conn = caa_container_of(node, struct relay_connection, qnode); + new_conn = lttng::utils::container_of(node, &relay_connection::qnode); DBG("Dispatching request waiting on sock %d", new_conn->sock->fd); @@ -1376,7 +1295,8 @@ static bool session_streams_have_index(const struct relay_session *session) * * On success, send back the session id or else return a negative value. */ -static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_create_session( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -1568,7 +1488,8 @@ end: /* * relay_add_stream: allocate a new stream for a session */ -static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_add_stream( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -1707,7 +1628,8 @@ end_no_session: /* * relay_close_stream: close a specific stream */ -static int relay_close_stream(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_close_stream( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -1785,7 +1707,8 @@ end_no_session: * relay_reset_metadata: reset a metadata stream */ static -int relay_reset_metadata(const struct lttcomm_relayd_hdr *recv_hdr, +int relay_reset_metadata( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -1884,9 +1807,10 @@ static void relay_unknown_command(struct relay_connection *conn) * relay_start: send an acknowledgment to the client to tell if we are * ready to receive data. We are ready if a session is established. */ -static int relay_start(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_start( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, - const struct lttng_buffer_view *payload) + const struct lttng_buffer_view *payload __attribute__((unused))) { int ret = 0; ssize_t send_ret; @@ -1977,7 +1901,8 @@ end: /* * relay_send_version: send relayd version number */ -static int relay_send_version(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_send_version( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -2047,7 +1972,8 @@ end: /* * Check for data pending for a given stream id from the session daemon. */ -static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_data_pending( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -2136,7 +2062,8 @@ end_no_session: * the control socket has been handled. So, this is why we simply return * OK here. */ -static int relay_quiescent_control(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_quiescent_control( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -2276,7 +2203,8 @@ end_no_session: * * Return to the client if there is data in flight or not with a ret_code. */ -static int relay_end_data_pending(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_end_data_pending( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -2368,7 +2296,8 @@ end_no_session: * * Return 0 on success else a negative value. */ -static int relay_recv_index(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_recv_index( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -2456,9 +2385,10 @@ end_no_session: * * Return 0 on success else a negative value. */ -static int relay_streams_sent(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_streams_sent( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, - const struct lttng_buffer_view *payload) + const struct lttng_buffer_view *payload __attribute__((unused))) { int ret; ssize_t send_ret; @@ -2526,11 +2456,14 @@ static ssize_t relay_unpack_rotate_streams_header( * We start by "unpacking" `stream_count` to figure out the padding length * emited by our peer. */ - memcpy(&rotate_streams.stream_count, payload->data, - sizeof(rotate_streams.stream_count)); - rotate_streams = (typeof(rotate_streams)) { - .stream_count = be32toh(rotate_streams.stream_count), - }; + { + decltype(rotate_streams.stream_count) stream_count; + + memcpy(&stream_count, payload->data, sizeof(stream_count)); + rotate_streams.stream_count = be32toh(stream_count); + } + + rotate_streams.new_chunk_id = LTTNG_OPTIONAL_INIT_UNSET; /* * Payload size expected given the possible padding lengths in @@ -2560,13 +2493,11 @@ static ssize_t relay_unpack_rotate_streams_header( memcpy(&packed_rotate_streams, payload->data, header_len); /* Unpack the packed structure to the natively-packed version. */ - *_rotate_streams = (typeof(*_rotate_streams)) { - .stream_count = be32toh(packed_rotate_streams.stream_count), - .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) { - .is_set = !!packed_rotate_streams.new_chunk_id.is_set, - .value = be64toh(packed_rotate_streams.new_chunk_id.value), - } + _rotate_streams->new_chunk_id = (typeof(_rotate_streams->new_chunk_id)){ + .is_set = !!packed_rotate_streams.new_chunk_id.is_set, + .value = be64toh(packed_rotate_streams.new_chunk_id.value), }; + _rotate_streams->stream_count = be32toh(packed_rotate_streams.stream_count); } else if (payload->size == expected_payload_size_3_bytes_padding) { struct lttcomm_relayd_rotate_streams_3_bytes_padding padded_rotate_streams; @@ -2576,13 +2507,11 @@ static ssize_t relay_unpack_rotate_streams_header( memcpy(&padded_rotate_streams, payload->data, header_len); /* Unpack the 3-byte padded structure to the natively-packed version. */ - *_rotate_streams = (typeof(*_rotate_streams)) { - .stream_count = be32toh(padded_rotate_streams.stream_count), - .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) { - .is_set = !!padded_rotate_streams.new_chunk_id.is_set, - .value = be64toh(padded_rotate_streams.new_chunk_id.value), - } + _rotate_streams->new_chunk_id = (typeof(_rotate_streams->new_chunk_id)){ + .is_set = !!padded_rotate_streams.new_chunk_id.is_set, + .value = be64toh(padded_rotate_streams.new_chunk_id.value), }; + _rotate_streams->stream_count = be32toh(padded_rotate_streams.stream_count); } else if (payload->size == expected_payload_size_7_bytes_padding) { struct lttcomm_relayd_rotate_streams_7_bytes_padding padded_rotate_streams; @@ -2592,13 +2521,11 @@ static ssize_t relay_unpack_rotate_streams_header( memcpy(&padded_rotate_streams, payload->data, header_len); /* Unpack the 7-byte padded structure to the natively-packed version. */ - *_rotate_streams = (typeof(*_rotate_streams)) { - .stream_count = be32toh(padded_rotate_streams.stream_count), - .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) { - .is_set = !!padded_rotate_streams.new_chunk_id.is_set, - .value = be64toh(padded_rotate_streams.new_chunk_id.value), - } + _rotate_streams->new_chunk_id = (typeof(_rotate_streams->new_chunk_id)){ + .is_set = !!padded_rotate_streams.new_chunk_id.is_set, + .value = be64toh(padded_rotate_streams.new_chunk_id.value), }; + _rotate_streams->stream_count = be32toh(padded_rotate_streams.stream_count); header_len = sizeof(padded_rotate_streams); } else { @@ -2620,7 +2547,7 @@ error: * session rotation feature (not the tracefile rotation feature). */ static int relay_rotate_session_streams( - const struct lttcomm_relayd_hdr *recv_hdr, + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -2758,7 +2685,8 @@ end_no_reply: /* * relay_create_trace_chunk: create a new trace chunk */ -static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_create_trace_chunk( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -2944,7 +2872,8 @@ end_no_reply: /* * relay_close_trace_chunk: close a trace chunk */ -static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_close_trace_chunk( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -3200,7 +3129,8 @@ end_no_reply: /* * relay_trace_chunk_exists: check if a trace chunk exists */ -static int relay_trace_chunk_exists(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_trace_chunk_exists( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -3263,7 +3193,8 @@ end_no_reply: /* * relay_get_configuration: query whether feature is available */ -static int relay_get_configuration(const struct lttcomm_relayd_hdr *recv_hdr, +static int relay_get_configuration( + const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)), struct relay_connection *conn, const struct lttng_buffer_view *payload) { @@ -3405,7 +3336,10 @@ static enum relay_connection_status relay_process_control_receive_payload( reception_buffer->data + state->received, state->left_to_receive, MSG_DONTWAIT); if (ret < 0) { + DIAGNOSTIC_PUSH + DIAGNOSTIC_IGNORE_LOGICAL_OP if (errno != EAGAIN && errno != EWOULDBLOCK) { + DIAGNOSTIC_POP PERROR("Unable to receive command payload on sock %d", conn->sock->fd); status = RELAY_CONNECTION_STATUS_ERROR; @@ -3478,7 +3412,10 @@ static enum relay_connection_status relay_process_control_receive_header( reception_buffer->data + state->received, state->left_to_receive, MSG_DONTWAIT); if (ret < 0) { + DIAGNOSTIC_PUSH + DIAGNOSTIC_IGNORE_LOGICAL_OP if (errno != EAGAIN && errno != EWOULDBLOCK) { + DIAGNOSTIC_POP PERROR("Unable to receive control command header on sock %d", conn->sock->fd); status = RELAY_CONNECTION_STATUS_ERROR; @@ -3589,7 +3526,10 @@ static enum relay_connection_status relay_process_data_receive_header( state->header_reception_buffer + state->received, state->left_to_receive, MSG_DONTWAIT); if (ret < 0) { + DIAGNOSTIC_PUSH + DIAGNOSTIC_IGNORE_LOGICAL_OP if (errno != EAGAIN && errno != EWOULDBLOCK) { + DIAGNOSTIC_POP PERROR("Unable to receive data header on sock %d", conn->sock->fd); status = RELAY_CONNECTION_STATUS_ERROR; } @@ -3716,7 +3656,10 @@ static enum relay_connection_status relay_process_data_receive_payload( ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, recv_size, MSG_DONTWAIT); if (ret < 0) { + DIAGNOSTIC_PUSH + DIAGNOSTIC_IGNORE_LOGICAL_OP if (errno != EAGAIN && errno != EWOULDBLOCK) { + DIAGNOSTIC_POP PERROR("Socket %d error", conn->sock->fd); status = RELAY_CONNECTION_STATUS_ERROR; } @@ -3885,7 +3828,7 @@ static void relay_thread_close_connection(struct lttng_poll_event *events, /* * This thread does the actual work */ -static void *relay_thread_worker(void *data) +static void *relay_thread_worker(void *data __attribute__((unused))) { int ret, err = -1, last_seen_data_fd = -1; uint32_t nb_fd; @@ -3952,14 +3895,14 @@ restart: */ for (i = 0; i < nb_fd; i++) { /* Fetch once the poll data */ - uint32_t revents = LTTNG_POLL_GETEV(&events, i); - int pollfd = LTTNG_POLL_GETFD(&events, i); + const auto revents = LTTNG_POLL_GETEV(&events, i); + const auto pollfd = LTTNG_POLL_GETFD(&events, i); health_code_update(); - /* Thread quit pipe has been closed. Killing thread. */ - ret = check_thread_quit_pipe(pollfd, revents); - if (ret) { + /* Activity on thread quit pipe, exiting. */ + if (relayd_is_thread_quit_pipe(pollfd)) { + DBG("Activity on thread quit pipe"); err = 0; goto exit; } @@ -4213,7 +4156,7 @@ static int create_relay_conn_pipe(void) "Relayd connection pipe", relay_conn_pipe); } -static int stdio_open(void *data, int *fds) +static int stdio_open(void *data __attribute__((unused)), int *fds) { fds[0] = fileno(stdout); fds[1] = fileno(stderr); @@ -4365,7 +4308,7 @@ int main(int argc, char **argv) } /* Create thread quit pipe */ - if (init_thread_quit_pipe()) { + if (relayd_init_thread_quit_pipe()) { retval = -1; goto exit_options; } @@ -4412,7 +4355,7 @@ int main(int argc, char **argv) /* Create thread to manage the client socket */ ret = pthread_create(&health_thread, default_pthread_attr(), - thread_manage_health, (void *) NULL); + thread_manage_health_relayd, (void *) NULL); if (ret) { errno = ret; PERROR("pthread_create health");