#include <algorithm>
#include <lttng/lttng.h>
-#include <common/common.h>
-#include <common/compat/poll.h>
-#include <common/compat/socket.h>
-#include <common/compat/endian.h>
-#include <common/compat/getenv.h>
-#include <common/defaults.h>
-#include <common/daemonize.h>
-#include <common/futex.h>
-#include <common/sessiond-comm/sessiond-comm.h>
-#include <common/sessiond-comm/inet.h>
-#include <common/sessiond-comm/relayd.h>
-#include <common/uri.h>
-#include <common/utils.h>
-#include <common/path.h>
-#include <common/align.h>
-#include <common/ini-config/ini-config.h>
-#include <common/dynamic-buffer.h>
-#include <common/buffer-view.h>
-#include <common/string-utils/format.h>
-#include <common/fd-tracker/fd-tracker.h>
-#include <common/fd-tracker/utils.h>
-
-#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 <common/common.hpp>
+#include <common/compat/poll.hpp>
+#include <common/compat/socket.hpp>
+#include <common/compat/endian.hpp>
+#include <common/compat/getenv.hpp>
+#include <common/defaults.hpp>
+#include <common/daemonize.hpp>
+#include <common/futex.hpp>
+#include <common/sessiond-comm/sessiond-comm.hpp>
+#include <common/sessiond-comm/inet.hpp>
+#include <common/sessiond-comm/relayd.hpp>
+#include <common/uri.hpp>
+#include <common/utils.hpp>
+#include <common/path.hpp>
+#include <common/align.hpp>
+#include <common/ini-config/ini-config.hpp>
+#include <common/dynamic-buffer.hpp>
+#include <common/buffer-view.hpp>
+#include <common/string-utils/format.hpp>
+#include <common/fd-tracker/fd-tracker.hpp>
+#include <common/fd-tracker/utils.hpp>
+
+#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
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.
(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);
}
}
-/*
- * 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;
/* 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;
}
}
}
-/*
- * 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.
*
"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;
*/
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;
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;
}
/* 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);
* 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
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;
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;
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 {
*/
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;
}
}
/* Create thread quit pipe */
- if (init_thread_quit_pipe()) {
+ if (relayd_init_thread_quit_pipe()) {
retval = -1;
goto exit_options;
}
projects / lttng-tools.git / blobdiff