#include <common/config/session-config.h>
#include <common/dynamic-buffer.h>
#include <common/buffer-view.h>
+#include <common/fd-tracker/utils.h>
#include <urcu/rculist.h>
#include "version.h"
if (tracing_group_name_override) {
free((void *) tracing_group_name);
}
- fd_tracker_log(the_fd_tracker);
}
/*
"Quit pipe", thread_quit_pipe);
}
+/*
+ * Init health quit pipe.
+ *
+ * Return -1 on error or 0 if all pipes are created.
+ */
+static int init_health_quit_pipe(void)
+{
+ return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
+ "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_thread_poll_set(struct lttng_poll_event *events, int size)
+static int create_named_thread_poll_set(struct lttng_poll_event *events,
+ int size, const char *name)
{
int ret;
goto error;
}
- ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
- if (ret < 0) {
- goto error;
- }
+ ret = fd_tracker_util_poll_create(the_fd_tracker,
+ name, events, 1, LTTNG_CLOEXEC);
/* Add quit pipe */
ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
return 0;
}
+static int create_sock(void *data, int *out_fd)
+{
+ int ret;
+ struct lttcomm_sock *sock = data;
+
+ ret = lttcomm_create_sock(sock);
+ if (ret < 0) {
+ goto end;
+ }
+
+ *out_fd = sock->fd;
+end:
+ return ret;
+}
+
+static int close_sock(void *data, int *in_fd)
+{
+ struct lttcomm_sock *sock = data;
+
+ return sock->ops->close(sock);
+}
+
+static int accept_sock(void *data, int *out_fd)
+{
+ int ret = 0;
+ /* Socks is an array of in_sock, out_sock. */
+ struct lttcomm_sock **socks = data;
+ struct lttcomm_sock *in_sock = socks[0];
+
+ socks[1] = in_sock->ops->accept(in_sock);
+ if (!socks[1]) {
+ ret = -1;
+ goto end;
+ }
+ *out_fd = socks[1]->fd;
+end:
+ return ret;
+}
+
/*
* Create and init socket from uri.
*/
-static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri)
+static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri,
+ const char *name)
{
- int ret;
+ int ret, sock_fd;
struct lttcomm_sock *sock = NULL;
+ char uri_str[PATH_MAX];
+ char *formated_name = NULL;
sock = lttcomm_alloc_sock_from_uri(uri);
if (sock == NULL) {
goto error;
}
- ret = lttcomm_create_sock(sock);
- if (ret < 0) {
- goto error;
+ /*
+ * Don't fail to create the socket if the name can't be built as it is
+ * only used for debugging purposes.
+ */
+ ret = uri_to_str_url(uri, uri_str, sizeof(uri_str));
+ uri_str[sizeof(uri_str) - 1] = '\0';
+ if (ret >= 0) {
+ ret = asprintf(&formated_name, "%s socket @ %s", name,
+ uri_str);
+ if (ret < 0) {
+ formated_name = NULL;
+ }
}
- DBG("Listening on sock %d", sock->fd);
+
+ ret = fd_tracker_open_unsuspendable_fd(the_fd_tracker, &sock_fd,
+ (const char **) (formated_name ? &formated_name : NULL),
+ 1, create_sock, sock);
+ free(formated_name);
+ DBG("Listening on %s socket %d", name, sock->fd);
ret = sock->ops->bind(sock);
if (ret < 0) {
return NULL;
}
+static
+struct lttcomm_sock *accept_relayd_sock(struct lttcomm_sock *listening_sock,
+ const char *name)
+{
+ int out_fd, ret;
+ struct lttcomm_sock *socks[2] = { listening_sock, NULL };
+ struct lttcomm_sock *new_sock = NULL;
+
+ ret = fd_tracker_open_unsuspendable_fd(
+ the_fd_tracker, &out_fd,
+ (const char **) &name,
+ 1, accept_sock, &socks);
+ if (ret) {
+ goto end;
+ }
+ new_sock = socks[1];
+ DBG("%s accepted, socket %d", name, new_sock->fd);
+end:
+ return new_sock;
+}
+
/*
* This thread manages the listening for new connections on the network
*/
health_code_update();
- control_sock = relay_socket_create(control_uri);
+ control_sock = relay_socket_create(control_uri, "Control listener");
if (!control_sock) {
goto error_sock_control;
}
- data_sock = relay_socket_create(data_uri);
+ data_sock = relay_socket_create(data_uri, "Data listener");
if (!data_sock) {
goto error_sock_relay;
}
* Pass 3 as size here for the thread quit pipe, control and
* data socket.
*/
- ret = create_thread_poll_set(&events, 3);
+ ret = create_named_thread_poll_set(&events, 3, "Listener thread epoll");
if (ret < 0) {
goto error_create_poll;
}
*/
int val = 1;
struct relay_connection *new_conn;
- struct lttcomm_sock *newsock;
+ struct lttcomm_sock *newsock = NULL;
enum connection_type type;
if (pollfd == data_sock->fd) {
type = RELAY_DATA;
- newsock = data_sock->ops->accept(data_sock);
- DBG("Relay data connection accepted, socket %d",
- newsock->fd);
+ newsock = accept_relayd_sock(data_sock,
+ "Data socket to relayd");
} else {
assert(pollfd == control_sock->fd);
type = RELAY_CONTROL;
- newsock = control_sock->ops->accept(control_sock);
- DBG("Relay control connection accepted, socket %d",
- newsock->fd);
+ newsock = accept_relayd_sock(control_sock,
+ "Control socket to relayd");
}
if (!newsock) {
PERROR("accepting sock");
error:
error_poll_add:
error_testpoint:
- lttng_poll_clean(&events);
+ (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
error_create_poll:
if (data_sock->fd >= 0) {
- ret = data_sock->ops->close(data_sock);
+ ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
+ &data_sock->fd, 1, close_sock,
+ data_sock);
if (ret) {
PERROR("close");
}
lttcomm_destroy_sock(data_sock);
error_sock_relay:
if (control_sock->fd >= 0) {
- ret = control_sock->ops->close(control_sock);
+ ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
+ &control_sock->fd, 1, close_sock,
+ control_sock);
if (ret) {
PERROR("close");
}
goto end_unlock;
}
- ret = utils_rotate_stream_file(stream->path_name, stream->channel_name,
- 0, 0, -1, -1, stream->stream_fd->fd, NULL,
- &stream->stream_fd->fd);
+ ret = stream_fd_rotate(stream->stream_fd,
+ stream->path_name, stream->channel_name, 0, 0, NULL);
if (ret < 0) {
ERR("Failed to rotate metadata file %s of channel %s",
stream->path_name, stream->channel_name);
struct lttcomm_relayd_metadata_payload metadata_payload_header;
struct relay_stream *metadata_stream;
uint64_t metadata_payload_size;
+ int metadata_fd = -1;
if (!session) {
ERR("Metadata sent before version check");
pthread_mutex_lock(&metadata_stream->lock);
- size_ret = lttng_write(metadata_stream->stream_fd->fd,
+ metadata_fd = stream_fd_get_fd(metadata_stream->stream_fd);
+ if (metadata_fd < 0) {
+ goto end_put;
+ }
+ size_ret = lttng_write(metadata_fd,
payload->data + sizeof(metadata_payload_header),
metadata_payload_size);
if (size_ret < metadata_payload_size) {
ERR("Relay error writing metadata on file");
ret = -1;
- goto end_put;
+ goto end_put_fd;
}
- size_ret = write_padding_to_file(metadata_stream->stream_fd->fd,
+ size_ret = write_padding_to_file(metadata_fd,
metadata_payload_header.padding_size);
if (size_ret < (int64_t) metadata_payload_header.padding_size) {
ret = -1;
- goto end_put;
+ goto end_put_fd;
}
metadata_stream->metadata_received +=
DBG2("Relay metadata written. Updated metadata_received %" PRIu64,
metadata_stream->metadata_received);
+end_put_fd:
+ stream_fd_put_fd(metadata_stream->stream_fd);
end_put:
pthread_mutex_unlock(&metadata_stream->lock);
stream_put(metadata_stream);
/* Put ref on previous index_file. */
if (stream->index_file) {
- lttng_index_file_put(stream->index_file);
+ relay_index_file_put(stream->index_file);
stream->index_file = NULL;
}
major = stream->trace->session->major;
minor = stream->trace->session->minor;
- stream->index_file = lttng_index_file_create(stream->path_name,
+ stream->index_file = relay_index_file_create(stream->path_name,
stream->channel_name,
- -1, -1, stream->tracefile_size,
+ stream->tracefile_size,
tracefile_array_get_file_index_head(stream->tfa),
lttng_to_index_major(major, minor),
lttng_to_index_minor(major, minor));
/* new_id is updated by utils_rotate_stream_file. */
new_id = old_id;
- ret = utils_rotate_stream_file(stream->path_name,
+ ret = stream_fd_rotate(stream->stream_fd, stream->path_name,
stream->channel_name, stream->tracefile_size,
- stream->tracefile_count, -1,
- -1, stream->stream_fd->fd,
- &new_id, &stream->stream_fd->fd);
+ stream->tracefile_count, &new_id);
if (ret < 0) {
ERR("Failed to rotate stream output file");
status = RELAY_CONNECTION_STATUS_ERROR;
bool new_stream = false, close_requested = false;
uint64_t left_to_receive = state->left_to_receive;
struct relay_session *session;
+ int stream_fd = -1;
DBG3("Receiving data for stream id %" PRIu64 " seqnum %" PRIu64 ", %" PRIu64" bytes received, %" PRIu64 " bytes left to receive",
state->header.stream_id, state->header.net_seq_num,
}
}
+ stream_fd = stream_fd_get_fd(stream->stream_fd);
+ if (stream_fd < 0) {
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end_stream_unlock;
+ }
+
/*
* The size of the "chunk" received on any iteration is bounded by:
* - the data left to receive,
PERROR("Socket %d error", conn->sock->fd);
status = RELAY_CONNECTION_STATUS_ERROR;
}
- goto end_stream_unlock;
+ goto end_put_fd;
} else if (ret == 0) {
/* No more data ready to be consumed on socket. */
DBG3("No more data ready for consumption on data socket of stream id %" PRIu64,
recv_size = ret;
/* Write data to stream output fd. */
- write_ret = lttng_write(stream->stream_fd->fd, data_buffer,
+ write_ret = lttng_write(stream_fd, data_buffer,
recv_size);
if (write_ret < (ssize_t) recv_size) {
ERR("Relay error writing data to file");
status = RELAY_CONNECTION_STATUS_ERROR;
- goto end_stream_unlock;
+ goto end_put_fd;
}
left_to_receive -= recv_size;
DBG3("Partial receive on data connection of stream id %" PRIu64 ", %" PRIu64 " bytes received, %" PRIu64 " bytes left to receive",
state->header.stream_id, state->received,
state->left_to_receive);
- goto end_stream_unlock;
+ goto end_put_fd;
}
- ret = write_padding_to_file(stream->stream_fd->fd,
+ ret = write_padding_to_file(stream_fd,
state->header.padding_size);
if ((int64_t) ret < (int64_t) state->header.padding_size) {
ERR("write_padding_to_file: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d",
stream->stream_handle,
state->header.net_seq_num, ret);
status = RELAY_CONNECTION_STATUS_ERROR;
- goto end_stream_unlock;
+ goto end_put_fd;
}
stream->stream_handle,
state->header.net_seq_num, ret);
status = RELAY_CONNECTION_STATUS_ERROR;
- goto end_stream_unlock;
+ goto end_put_fd;
}
}
connection_reset_protocol_state(conn);
state = NULL;
+end_put_fd:
+ stream_fd_put_fd(stream->stream_fd);
end_stream_unlock:
close_requested = stream->close_requested;
pthread_mutex_unlock(&stream->lock);
(void) lttng_poll_del(events, pollfd);
- ret = close(pollfd);
+ ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, &pollfd, 1,
+ fd_tracker_util_close_fd, NULL);
if (ret < 0) {
ERR("Closing pollfd %d", pollfd);
}
goto relay_connections_ht_error;
}
- ret = create_thread_poll_set(&events, 2);
+ ret = create_named_thread_poll_set(&events, 2, "Worker thread epoll");
if (ret < 0) {
goto error_poll_create;
}
}
rcu_read_unlock();
- lttng_poll_clean(&events);
+ (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
error_poll_create:
lttng_ht_destroy(relay_connections_ht);
relay_connections_ht_error:
"Relayd connection pipe", relay_conn_pipe);
}
+static
+int stdio_open(void *data, int *fds)
+{
+ fds[0] = fileno(stdout);
+ fds[1] = fileno(stderr);
+ return 0;
+}
+
+static
+int noop_close(void *data, int *fds)
+{
+ return 0;
+}
+
+static
+int track_stdio(void)
+{
+ int fds[2];
+ const char *names[] = { "stdout", "stderr" };
+
+ return fd_tracker_open_unsuspendable_fd(the_fd_tracker, fds,
+ names, 2, stdio_open, NULL);
+}
+
+static
+void untrack_stdio(void)
+{
+ int fds[] = { fileno(stdout), fileno(stderr) };
+
+ /*
+ * noop_close is used since we don't really want to close
+ * the stdio output fds; we merely want to stop tracking them.
+ */
+ (void) fd_tracker_close_unsuspendable_fd(the_fd_tracker,
+ fds, 2, noop_close, NULL);
+}
+
/*
* main
*/
goto exit_options;
}
+ ret = track_stdio();
+ if (ret) {
+ retval = -1;
+ goto exit_options;
+ }
+
/* Initialize thread health monitoring */
health_relayd = health_app_create(NR_HEALTH_RELAYD_TYPES);
if (!health_relayd) {
goto exit_init_data;
}
- ret = utils_create_pipe(health_quit_pipe);
+ ret = init_health_quit_pipe();
if (ret) {
retval = -1;
goto exit_health_quit_pipe;
}
exit_health_thread:
- utils_close_pipe(health_quit_pipe);
+ (void) fd_tracker_util_pipe_close(the_fd_tracker, health_quit_pipe);
exit_health_quit_pipe:
exit_init_data:
/* Ensure all prior call_rcu are done. */
rcu_barrier();
+ untrack_stdio();
+ /*
+ * fd_tracker_destroy() will log the contents of the fd-tracker
+ * if a leak is detected.
+ */
fd_tracker_destroy(the_fd_tracker);
rcu_unregister_thread();
projects / lttng-tools.git / blobdiff