static struct relay_conn_queue relay_conn_queue;
/* Cap of file desriptors to be in simultaneous use by the relay daemon. */
-static unsigned int lttng_opt_fd_cap;
+static unsigned int lttng_opt_fd_pool_size = -1;
/* Global relay stream hash table. */
struct lttng_ht *relay_streams_ht;
{ "daemonize", 0, 0, 'd', },
{ "background", 0, 0, 'b', },
{ "group", 1, 0, 'g', },
- { "fd-cap", 1, 0, '\0', },
+ { "fd-pool-size", 1, 0, '\0', },
{ "help", 0, 0, 'h', },
{ "output", 1, 0, 'o', },
{ "verbose", 0, 0, 'v', },
switch (opt) {
case 0:
- if (!strcmp(optname, "fd-cap")) {
+ if (!strcmp(optname, "fd-pool-size")) {
unsigned long v;
errno = 0;
v = strtoul(arg, NULL, 0);
if (errno != 0 || !isdigit(arg[0])) {
- ERR("Wrong value in --fd-cap parameter: %s", arg);
+ ERR("Wrong value in --fd-pool-size parameter: %s", arg);
ret = -1;
goto end;
}
- if (v < DEFAULT_RELAYD_MINIMAL_FD_CAP) {
- ERR("File descriptor cap must be set to at least %d",
- DEFAULT_RELAYD_MINIMAL_FD_CAP);
- }
if (v >= UINT_MAX) {
- ERR("File descriptor cap overflow in --fd-cap parameter: %s", arg);
+ ERR("File descriptor cap overflow in --fd-pool-size parameter: %s", arg);
ret = -1;
goto end;
}
- lttng_opt_fd_cap = (unsigned int) v;
- DBG3("File descriptor cap set to %u", lttng_opt_fd_cap);
-
+ lttng_opt_fd_pool_size = (unsigned int) v;
} else {
fprintf(stderr, "unknown option %s", optname);
if (arg) {
return ret;
}
+static int set_fd_pool_size(void)
+{
+ int ret = 0;
+ struct rlimit rlimit;
+
+ ret = getrlimit(RLIMIT_NOFILE, &rlimit);
+ if (ret) {
+ PERROR("Failed to get file descriptor limit");
+ ret = -1;
+ goto end;
+ }
+
+ DBG("File descriptor count limits are %lu (soft) and %lu (hard)",
+ rlimit.rlim_cur, rlimit.rlim_max);
+ if (lttng_opt_fd_pool_size == -1) {
+ /* Use default value (soft limit - reserve). */
+ if (rlimit.rlim_cur < DEFAULT_RELAYD_MIN_FD_POOL_SIZE) {
+ ERR("The process' file number limit is too low (%lu). The process' file number limit must be set to at least %i.",
+ rlimit.rlim_cur, DEFAULT_RELAYD_MIN_FD_POOL_SIZE);
+ ret = -1;
+ goto end;
+ }
+ lttng_opt_fd_pool_size = rlimit.rlim_cur -
+ DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE;
+ goto end;
+ }
+
+ if (lttng_opt_fd_pool_size < DEFAULT_RELAYD_MIN_FD_POOL_SIZE) {
+ ERR("File descriptor pool size must be set to at least %d",
+ DEFAULT_RELAYD_MIN_FD_POOL_SIZE);
+ ret = -1;
+ goto end;
+ }
+
+ if (lttng_opt_fd_pool_size > rlimit.rlim_cur) {
+ ERR("File descriptor pool size argument (%u) exceeds the process' soft limit (%lu).",
+ lttng_opt_fd_pool_size, rlimit.rlim_cur);
+ ret = -1;
+ goto end;
+ }
+
+
+ DBG("File descriptor pool size argument (%u) adjusted to %u to accomodate transient fd uses",
+ lttng_opt_fd_pool_size,
+ lttng_opt_fd_pool_size - DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE);
+ lttng_opt_fd_pool_size -= DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE;
+end:
+ return ret;
+}
+
static int set_options(int argc, char **argv)
{
int c, ret = 0, option_index = 0, retval = 0;
goto exit;
}
}
- if (lttng_opt_fd_cap == 0) {
- int ret;
- struct rlimit rlimit;
-
- ret = getrlimit(RLIMIT_NOFILE, &rlimit);
- if (ret) {
- PERROR("Failed to get file descriptor limit");
- retval = -1;
- }
-
- lttng_opt_fd_cap = rlimit.rlim_cur;
+ ret = set_fd_pool_size();
+ if (ret) {
+ retval = -1;
+ goto exit;
}
exit:
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.
*/
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
*/
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;
}
*/
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");
(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");
}
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);
/* 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));
int ret;
/* Perform the stream rotation. */
- 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,
- NULL, &stream->stream_fd->fd);
+ stream->tracefile_count, NULL);
if (ret < 0) {
ERR("Rotating stream output file");
goto end;
static
int rotate_truncate_stream(struct relay_stream *stream)
{
- int ret, new_fd;
+ int ret, old_fd = -1, new_fd = -1;
off_t lseek_ret;
uint64_t diff, pos = 0;
char buf[FILE_COPY_BUFFER_SIZE];
+ struct stream_fd *new_stream_fd = NULL;
assert(!stream->is_metadata);
stream->pos_after_last_complete_data_index;
/* Create the new tracefile. */
- new_fd = utils_create_stream_file(stream->path_name,
- stream->channel_name,
- stream->tracefile_size, stream->tracefile_count,
- /* uid */ -1, /* gid */ -1, /* suffix */ NULL);
- if (new_fd < 0) {
+ new_stream_fd = stream_fd_create(stream->path_name, stream->channel_name,
+ stream->tracefile_size, stream->tracefile_count, NULL);
+ if (new_stream_fd < 0) {
ERR("Failed to create new stream file at path %s for channel %s",
stream->path_name, stream->channel_name);
ret = -1;
goto end;
}
+ old_fd = stream_fd_get_fd(stream->stream_fd);
+ if (old_fd < 0) {
+ ret = -1;
+ goto end;
+ }
+
+ new_fd = stream_fd_get_fd(new_stream_fd);
+ if (new_fd < 0) {
+ ret = -1;
+ goto end;
+ }
+
/*
* Rewind the current tracefile to the position at which the rotation
* should have occured.
*/
- lseek_ret = lseek(stream->stream_fd->fd,
- stream->pos_after_last_complete_data_index, SEEK_SET);
+ lseek_ret = lseek(old_fd, stream->pos_after_last_complete_data_index,
+ SEEK_SET);
if (lseek_ret < 0) {
PERROR("seek truncate stream");
ret = -1;
count = bytes_left > sizeof(buf) ? sizeof(buf) : bytes_left;
assert(count <= SIZE_MAX);
- io_ret = lttng_read(stream->stream_fd->fd, buf, count);
+ io_ret = lttng_read(old_fd, buf, count);
if (io_ret < (ssize_t) count) {
char error_string[256];
snprintf(error_string, sizeof(error_string),
- "Failed to read %" PRIu64 " bytes from fd %i in rotate_truncate_stream(), returned %zi",
- count, stream->stream_fd->fd, io_ret);
+ "Failed to read %" PRIu64 " bytes from in rotate_truncate_stream(), returned %zi",
+ count, io_ret);
if (io_ret == -1) {
PERROR("%s", error_string);
} else {
pos += count;
}
+ stream_fd_put_fd(new_stream_fd);
+ new_fd = -1;
+
/* Truncate the file to get rid of the excess data. */
- ret = ftruncate(stream->stream_fd->fd,
- stream->pos_after_last_complete_data_index);
+ ret = ftruncate(old_fd, stream->pos_after_last_complete_data_index);
if (ret) {
PERROR("ftruncate");
goto end;
}
- ret = close(stream->stream_fd->fd);
- if (ret < 0) {
- PERROR("Closing tracefile");
- goto end;
- }
+ stream_fd_put_fd(stream->stream_fd);
+ old_fd = -1;
+
+ /*
+ * Swap the old stream_fd with the new, releasing the stream's
+ * reference.
+ */
+ stream_fd_put(stream->stream_fd);
+ stream->stream_fd = new_stream_fd;
ret = create_rotate_index_file(stream);
if (ret < 0) {
goto end;
}
- stream->stream_fd->fd = new_fd;
stream->tracefile_size_current = diff;
stream->pos_after_last_complete_data_index = 0;
stream->rotate_at_seq_num = -1ULL;
ret = 0;
end:
+ if (old_fd != -1) {
+ stream_fd_put_fd(stream->stream_fd);
+ }
+ if (new_fd != -1) {
+ stream_fd_put_fd(new_stream_fd);
+ }
+
return ret;
}
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 +=
ret = try_rotate_stream(metadata_stream);
if (ret < 0) {
- goto end_put;
+ goto end_put_fd;
}
+end_put_fd:
+ stream_fd_put_fd(metadata_stream->stream_fd);
end_put:
pthread_mutex_unlock(&metadata_stream->lock);
stream_put(metadata_stream);
/* 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, index_flushed = 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;
}
}
ret = try_rotate_stream(stream);
if (ret < 0) {
status = RELAY_CONNECTION_STATUS_ERROR;
- goto end_stream_unlock;
+ goto end_put_fd;
}
-
+end_put_fd:
+ (void) 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);
}
*/
rcu_register_thread();
- the_fd_tracker = fd_tracker_create(lttng_opt_fd_cap);
+ the_fd_tracker = fd_tracker_create(lttng_opt_fd_pool_size);
if (!the_fd_tracker) {
retval = -1;
goto exit_options;
ret = track_stdio();
if (ret) {
retval = -1;
- goto exit_options;
+ goto exit_tracker;
}
/* Initialize thread health monitoring */
exit_init_data:
health_app_destroy(health_relayd);
exit_health_app_create:
-exit_options:
+
/*
* Wait for all pending call_rcu work to complete before tearing
* down data structures. call_rcu worker may be trying to
/* Ensure all prior call_rcu are done. */
rcu_barrier();
-
+exit_tracker:
untrack_stdio();
/*
* fd_tracker_destroy() will log the contents of the fd-tracker
*/
fd_tracker_destroy(the_fd_tracker);
rcu_unregister_thread();
-
+exit_options:
if (!retval) {
exit(EXIT_SUCCESS);
} else {
projects / deliverable / lttng-tools.git / blobdiff