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);