+ * Unexpected, protocol error/bug.
+ * It could mean that we received a rotation position
+ * that is in the past.
+ */
+ ERR("Stream %" PRIu64 " index is in an inconsistent state (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
+ stream->stream_handle,
+ stream->rotate_at_seq_num,
+ stream->prev_data_seq,
+ stream->prev_index_seq);
+ ret = -1;
+ goto end;
+ } else {
+ DBG("Rotating stream %" PRIu64 " index file",
+ stream->stream_handle);
+ ret = create_rotate_index_file(stream, stream->path_name);
+ stream->index_rotated = true;
+
+ if (stream->data_rotated && stream->index_rotated) {
+ /* Rotation completed; reset its state. */
+ DBG("Rotation completed for stream %" PRIu64,
+ stream->stream_handle);
+ stream->rotate_at_seq_num = -1ULL;
+ stream->data_rotated = false;
+ stream->index_rotated = false;
+ }
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * Check if a stream's data file (as opposed to index) should be rotated
+ * (for session rotation).
+ * Must be called with the stream lock held.
+ *
+ * Return 0 on success, a negative value on error.
+ */
+static
+int try_rotate_stream_data(struct relay_stream *stream)
+{
+ int ret = 0;
+
+ if (stream->rotate_at_seq_num == -1ULL) {
+ /* No rotation expected. */
+ goto end;
+ }
+
+ if (stream->data_rotated) {
+ /* Rotation of the data file has already occurred. */
+ goto end;
+ }
+
+ if (stream->prev_data_seq == -1ULL ||
+ stream->prev_data_seq < stream->rotate_at_seq_num) {
+ DBG("Stream %" PRIu64 " not yet ready for rotation (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ")",
+ stream->stream_handle,
+ stream->rotate_at_seq_num,
+ stream->prev_data_seq);
+ goto end;
+ } else if (stream->prev_data_seq > stream->rotate_at_seq_num) {
+ /*
+ * prev_data_seq is checked here since indexes and rotation
+ * commands are serialized with respect to each other.
+ */
+ DBG("Rotation after too much data has been written in tracefile "
+ "for stream %" PRIu64 ", need to truncate before "
+ "rotating", stream->stream_handle);
+ ret = rotate_truncate_stream(stream);
+ if (ret) {
+ ERR("Failed to truncate stream");
+ goto end;
+ }
+ } else if (stream->prev_data_seq != stream->rotate_at_seq_num) {
+ /*
+ * Unexpected, protocol error/bug.
+ * It could mean that we received a rotation position
+ * that is in the past.
+ */
+ ERR("Stream %" PRIu64 " data is in an inconsistent state (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ")",
+ stream->stream_handle,
+ stream->rotate_at_seq_num,
+ stream->prev_data_seq);
+ ret = -1;
+ goto end;
+ } else {
+ ret = do_rotate_stream_data(stream);
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * relay_recv_metadata: receive the metadata for the session.
+ */
+static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret = 0;
+ ssize_t size_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_metadata_payload metadata_payload_header;
+ struct relay_stream *metadata_stream;
+ uint64_t metadata_payload_size;
+
+ if (!session) {
+ ERR("Metadata sent before version check");
+ ret = -1;
+ goto end;
+ }
+
+ if (recv_hdr->data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
+ ERR("Incorrect data size");
+ ret = -1;
+ goto end;
+ }
+ metadata_payload_size = recv_hdr->data_size -
+ sizeof(struct lttcomm_relayd_metadata_payload);
+
+ memcpy(&metadata_payload_header, payload->data,
+ sizeof(metadata_payload_header));
+ metadata_payload_header.stream_id = be64toh(
+ metadata_payload_header.stream_id);
+ metadata_payload_header.padding_size = be32toh(
+ metadata_payload_header.padding_size);
+
+ metadata_stream = stream_get_by_id(metadata_payload_header.stream_id);
+ if (!metadata_stream) {
+ ret = -1;
+ goto end;
+ }
+
+ pthread_mutex_lock(&metadata_stream->lock);
+
+ size_ret = lttng_write(metadata_stream->stream_fd->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;
+ }
+
+ size_ret = write_padding_to_file(metadata_stream->stream_fd->fd,
+ metadata_payload_header.padding_size);
+ if (size_ret < (int64_t) metadata_payload_header.padding_size) {
+ ret = -1;
+ goto end_put;
+ }
+
+ metadata_stream->metadata_received +=
+ metadata_payload_size + metadata_payload_header.padding_size;
+ DBG2("Relay metadata written. Updated metadata_received %" PRIu64,
+ metadata_stream->metadata_received);
+
+ ret = try_rotate_stream_data(metadata_stream);
+ if (ret < 0) {
+ goto end_put;
+ }
+
+end_put:
+ pthread_mutex_unlock(&metadata_stream->lock);
+ stream_put(metadata_stream);
+end:
+ return ret;
+}
+
+/*
+ * relay_send_version: send relayd version number
+ */
+static int relay_send_version(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct lttcomm_relayd_version reply, msg;
+ bool compatible = true;
+
+ conn->version_check_done = true;
+
+ /* Get version from the other side. */
+ if (payload->size < sizeof(msg)) {
+ ERR("Unexpected payload size in \"relay_send_version\": expected >= %zu bytes, got %zu bytes",
+ sizeof(msg), payload->size);
+ ret = -1;
+ goto end;
+ }
+
+ memcpy(&msg, payload->data, sizeof(msg));
+ msg.major = be32toh(msg.major);
+ msg.minor = be32toh(msg.minor);
+
+ memset(&reply, 0, sizeof(reply));
+ reply.major = RELAYD_VERSION_COMM_MAJOR;
+ reply.minor = RELAYD_VERSION_COMM_MINOR;
+
+ /* Major versions must be the same */
+ if (reply.major != msg.major) {
+ DBG("Incompatible major versions (%u vs %u), deleting session",
+ reply.major, msg.major);
+ compatible = false;
+ }
+
+ conn->major = reply.major;
+ /* We adapt to the lowest compatible version */
+ if (reply.minor <= msg.minor) {
+ conn->minor = reply.minor;
+ } else {
+ conn->minor = msg.minor;
+ }
+
+ reply.major = htobe32(reply.major);
+ reply.minor = htobe32(reply.minor);
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"send version\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ goto end;
+ } else {
+ ret = 0;
+ }
+
+ if (!compatible) {
+ ret = -1;
+ goto end;
+ }
+
+ DBG("Version check done using protocol %u.%u", conn->major,
+ conn->minor);
+
+end:
+ return ret;
+}
+
+/*
+ * 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,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_data_pending msg;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ ssize_t send_ret;
+ int ret;
+ uint64_t stream_seq;
+
+ DBG("Data pending command received");
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to check for data before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (payload->size < sizeof(msg)) {
+ ERR("Unexpected payload size in \"relay_data_pending\": expected >= %zu bytes, got %zu bytes",
+ sizeof(msg), payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+ memcpy(&msg, payload->data, sizeof(msg));
+ msg.stream_id = be64toh(msg.stream_id);
+ msg.last_net_seq_num = be64toh(msg.last_net_seq_num);
+
+ stream = stream_get_by_id(msg.stream_id);
+ if (stream == NULL) {
+ ret = -1;
+ goto end;
+ }
+
+ pthread_mutex_lock(&stream->lock);
+
+ if (session_streams_have_index(session)) {
+ /*
+ * Ensure that both the index and stream data have been
+ * flushed up to the requested point.
+ */
+ stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
+ } else {
+ stream_seq = stream->prev_data_seq;
+ }
+ DBG("Data pending for stream id %" PRIu64 ": prev_data_seq %" PRIu64
+ ", prev_index_seq %" PRIu64
+ ", and last_seq %" PRIu64, msg.stream_id,
+ stream->prev_data_seq, stream->prev_index_seq,
+ msg.last_net_seq_num);
+
+ /* Avoid wrapping issue */
+ if (((int64_t) (stream_seq - msg.last_net_seq_num)) >= 0) {
+ /* Data has in fact been written and is NOT pending */
+ ret = 0;
+ } else {
+ /* Data still being streamed thus pending */
+ ret = 1;
+ }
+
+ stream->data_pending_check_done = true;
+ pthread_mutex_unlock(&stream->lock);
+
+ stream_put(stream);
+end:
+
+ memset(&reply, 0, sizeof(reply));
+ reply.ret_code = htobe32(ret);
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"data pending\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Wait for the control socket to reach a quiescent state.
+ *
+ * Note that for now, when receiving this command from the session
+ * daemon, this means that every subsequent commands or data received on
+ * 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,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct relay_stream *stream;
+ struct lttcomm_relayd_quiescent_control msg;
+ struct lttcomm_relayd_generic_reply reply;
+
+ DBG("Checking quiescent state on control socket");
+
+ if (!conn->session || !conn->version_check_done) {
+ ERR("Trying to check for data before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (payload->size < sizeof(msg)) {
+ ERR("Unexpected payload size in \"relay_quiescent_control\": expected >= %zu bytes, got %zu bytes",
+ sizeof(msg), payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+ memcpy(&msg, payload->data, sizeof(msg));
+ msg.stream_id = be64toh(msg.stream_id);
+
+ stream = stream_get_by_id(msg.stream_id);
+ if (!stream) {
+ goto reply;
+ }
+ pthread_mutex_lock(&stream->lock);
+ stream->data_pending_check_done = true;
+ pthread_mutex_unlock(&stream->lock);
+
+ DBG("Relay quiescent control pending flag set to %" PRIu64, msg.stream_id);
+ stream_put(stream);
+reply:
+ memset(&reply, 0, sizeof(reply));
+ reply.ret_code = htobe32(LTTNG_OK);
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"quiescent control\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ } else {
+ ret = 0;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Initialize a data pending command. This means that a consumer is about
+ * to ask for data pending for each stream it holds. Simply iterate over
+ * all streams of a session and set the data_pending_check_done flag.
+ *
+ * This command returns to the client a LTTNG_OK code.
+ */
+static int relay_begin_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct lttng_ht_iter iter;
+ struct lttcomm_relayd_begin_data_pending msg;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+
+ assert(recv_hdr);
+ assert(conn);
+
+ DBG("Init streams for data pending");
+
+ if (!conn->session || !conn->version_check_done) {
+ ERR("Trying to check for data before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (payload->size < sizeof(msg)) {
+ ERR("Unexpected payload size in \"relay_begin_data_pending\": expected >= %zu bytes, got %zu bytes",
+ sizeof(msg), payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+ memcpy(&msg, payload->data, sizeof(msg));
+ msg.session_id = be64toh(msg.session_id);
+
+ /*
+ * Iterate over all streams to set the begin data pending flag.
+ * For now, the streams are indexed by stream handle so we have
+ * to iterate over all streams to find the one associated with
+ * the right session_id.
+ */
+ rcu_read_lock();
+ cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
+ node.node) {
+ if (!stream_get(stream)) {
+ continue;
+ }
+ if (stream->trace->session->id == msg.session_id) {
+ pthread_mutex_lock(&stream->lock);
+ stream->data_pending_check_done = false;
+ pthread_mutex_unlock(&stream->lock);
+ DBG("Set begin data pending flag to stream %" PRIu64,
+ stream->stream_handle);
+ }
+ stream_put(stream);
+ }
+ rcu_read_unlock();
+
+ memset(&reply, 0, sizeof(reply));
+ /* All good, send back reply. */
+ reply.ret_code = htobe32(LTTNG_OK);
+
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"begin data pending\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ } else {
+ ret = 0;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * End data pending command. This will check, for a given session id, if
+ * each stream associated with it has its data_pending_check_done flag
+ * set. If not, this means that the client lost track of the stream but
+ * the data is still being streamed on our side. In this case, we inform
+ * the client that data is in flight.
+ *
+ * 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,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct lttng_ht_iter iter;
+ struct lttcomm_relayd_end_data_pending msg;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ uint32_t is_data_inflight = 0;
+
+ DBG("End data pending command");
+
+ if (!conn->session || !conn->version_check_done) {
+ ERR("Trying to check for data before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (payload->size < sizeof(msg)) {
+ ERR("Unexpected payload size in \"relay_end_data_pending\": expected >= %zu bytes, got %zu bytes",
+ sizeof(msg), payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+ memcpy(&msg, payload->data, sizeof(msg));
+ msg.session_id = be64toh(msg.session_id);
+
+ /*
+ * Iterate over all streams to see if the begin data pending
+ * flag is set.
+ */
+ rcu_read_lock();
+ cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
+ node.node) {
+ if (!stream_get(stream)) {
+ continue;
+ }
+ if (stream->trace->session->id != msg.session_id) {
+ stream_put(stream);
+ continue;
+ }
+ pthread_mutex_lock(&stream->lock);
+ if (!stream->data_pending_check_done) {
+ uint64_t stream_seq;
+
+ if (session_streams_have_index(conn->session)) {
+ /*
+ * Ensure that both the index and stream data have been
+ * flushed up to the requested point.
+ */
+ stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
+ } else {
+ stream_seq = stream->prev_data_seq;
+ }
+ if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
+ is_data_inflight = 1;
+ DBG("Data is still in flight for stream %" PRIu64,
+ stream->stream_handle);
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+ break;
+ }
+ }
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+ }
+ rcu_read_unlock();
+
+ memset(&reply, 0, sizeof(reply));
+ /* All good, send back reply. */
+ reply.ret_code = htobe32(is_data_inflight);
+
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"end data pending\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ } else {
+ ret = 0;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Receive an index for a specific stream.
+ *
+ * Return 0 on success else a negative value.
+ */
+static int relay_recv_index(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_index index_info;
+ struct relay_index *index;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ size_t msg_len;
+
+ assert(conn);
+
+ DBG("Relay receiving index");
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to close a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ msg_len = lttcomm_relayd_index_len(
+ lttng_to_index_major(conn->major, conn->minor),
+ lttng_to_index_minor(conn->major, conn->minor));
+ if (payload->size < msg_len) {
+ ERR("Unexpected payload size in \"relay_recv_index\": expected >= %zu bytes, got %zu bytes",
+ msg_len, payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+ memcpy(&index_info, payload->data, msg_len);
+ index_info.relay_stream_id = be64toh(index_info.relay_stream_id);
+ index_info.net_seq_num = be64toh(index_info.net_seq_num);
+ index_info.packet_size = be64toh(index_info.packet_size);
+ index_info.content_size = be64toh(index_info.content_size);
+ index_info.timestamp_begin = be64toh(index_info.timestamp_begin);
+ index_info.timestamp_end = be64toh(index_info.timestamp_end);
+ index_info.events_discarded = be64toh(index_info.events_discarded);
+ index_info.stream_id = be64toh(index_info.stream_id);
+
+ if (conn->minor >= 8) {
+ index_info.stream_instance_id =
+ be64toh(index_info.stream_instance_id);
+ index_info.packet_seq_num = be64toh(index_info.packet_seq_num);
+ }
+
+ stream = stream_get_by_id(index_info.relay_stream_id);
+ if (!stream) {
+ ERR("stream_get_by_id not found");
+ ret = -1;
+ goto end;
+ }
+ pthread_mutex_lock(&stream->lock);
+
+ /* Live beacon handling */
+ if (index_info.packet_size == 0) {
+ DBG("Received live beacon for stream %" PRIu64,
+ stream->stream_handle);
+
+ /*
+ * Only flag a stream inactive when it has already
+ * received data and no indexes are in flight.
+ */
+ if (stream->index_received_seqcount > 0
+ && stream->indexes_in_flight == 0) {
+ stream->beacon_ts_end = index_info.timestamp_end;
+ }
+ ret = 0;
+ goto end_stream_put;
+ } else {
+ stream->beacon_ts_end = -1ULL;
+ }
+
+ if (stream->ctf_stream_id == -1ULL) {
+ stream->ctf_stream_id = index_info.stream_id;
+ }
+ index = relay_index_get_by_id_or_create(stream, index_info.net_seq_num);
+ if (!index) {
+ ret = -1;
+ ERR("relay_index_get_by_id_or_create index NULL");
+ goto end_stream_put;
+ }
+ if (set_index_control_data(index, &index_info, conn)) {
+ ERR("set_index_control_data error");
+ relay_index_put(index);
+ ret = -1;
+ goto end_stream_put;
+ }
+ ret = relay_index_try_flush(index);
+ if (ret == 0) {
+ tracefile_array_commit_seq(stream->tfa);
+ stream->index_received_seqcount++;
+ stream->pos_after_last_complete_data_index += index->total_size;
+ stream->prev_index_seq = index_info.net_seq_num;
+
+ ret = try_rotate_stream_index(stream);
+ if (ret < 0) {
+ goto end_stream_put;
+ }
+ } else if (ret > 0) {
+ /* no flush. */
+ ret = 0;
+ } else {
+ /*
+ * ret < 0
+ *
+ * relay_index_try_flush is responsible for the self-reference
+ * put of the index object on error.
+ */
+ ERR("relay_index_try_flush error %d", ret);
+ ret = -1;
+ }
+
+end_stream_put:
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+
+end:
+
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"recv index\" command reply (ret = %zd)", send_ret);
+ ret = -1;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Receive the streams_sent message.
+ *
+ * Return 0 on success else a negative value.
+ */
+static int relay_streams_sent(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct lttcomm_relayd_generic_reply reply;
+
+ assert(conn);
+
+ DBG("Relay receiving streams_sent");
+
+ if (!conn->session || !conn->version_check_done) {
+ ERR("Trying to close a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ /*
+ * Publish every pending stream in the connection recv list which are
+ * now ready to be used by the viewer.
+ */
+ publish_connection_local_streams(conn);
+
+ memset(&reply, 0, sizeof(reply));
+ reply.ret_code = htobe32(LTTNG_OK);
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"streams sent\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ } else {
+ /* Success. */
+ ret = 0;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * relay_rotate_session_stream: rotate a stream to a new tracefile for the session
+ * rotation feature (not the tracefile rotation feature).
+ */
+static int relay_rotate_session_stream(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_rotate_stream stream_info;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ size_t header_len;
+ size_t path_len;
+ struct lttng_buffer_view new_path_view;
+
+ DBG("Rotate stream received");
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to rotate a stream before version check");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ if (session->major == 2 && session->minor < 11) {
+ ERR("Unsupported feature before 2.11");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ header_len = sizeof(struct lttcomm_relayd_rotate_stream);
+
+ if (payload->size < header_len) {
+ ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected >= %zu bytes, got %zu bytes",
+ header_len, payload->size);
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ memcpy(&stream_info, payload->data, header_len);
+
+ /* Convert to host */
+ stream_info.pathname_length = be32toh(stream_info.pathname_length);
+ stream_info.stream_id = be64toh(stream_info.stream_id);
+ stream_info.new_chunk_id = be64toh(stream_info.new_chunk_id);
+ stream_info.rotate_at_seq_num = be64toh(stream_info.rotate_at_seq_num);
+
+ path_len = stream_info.pathname_length;
+ if (payload->size < header_len + path_len) {
+ ERR("Unexpected payload size in \"relay_rotate_session_stream\" including path: expected >= %zu bytes, got %zu bytes",
+ header_len + path_len, payload->size);
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ /* Ensure it fits in local filename length. */
+ if (path_len >= LTTNG_PATH_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("Length of relay_rotate_session_stream command's path name (%zu bytes) exceeds the maximal allowed length of %i bytes",
+ path_len, LTTNG_PATH_MAX);
+ goto end;
+ }
+
+ new_path_view = lttng_buffer_view_from_view(payload, header_len,
+ stream_info.pathname_length);
+
+ stream = stream_get_by_id(stream_info.stream_id);
+ if (!stream) {
+ ret = -1;
+ goto end;
+ }
+
+ pthread_mutex_lock(&stream->lock);
+
+ /*
+ * Update the trace path (just the folder, the stream name does not
+ * change).
+ */
+ free(stream->prev_path_name);
+ stream->prev_path_name = stream->path_name;
+ stream->path_name = create_output_path(new_path_view.data);
+ if (!stream->path_name) {
+ ERR("Failed to create a new output path");
+ ret = -1;
+ goto end_stream_unlock;
+ }
+ ret = utils_mkdir_recursive(stream->path_name, S_IRWXU | S_IRWXG,
+ -1, -1);
+ if (ret < 0) {
+ ERR("relay creating output directory");
+ ret = -1;
+ goto end_stream_unlock;
+ }
+
+ assert(stream->current_chunk_id.is_set);
+ stream->current_chunk_id.value = stream_info.new_chunk_id;
+
+ if (stream->is_metadata) {
+ /*
+ * Metadata streams have no index; consider its rotation
+ * complete.
+ */
+ stream->index_rotated = true;
+ /*
+ * The metadata stream is sent only over the control connection
+ * so we know we have all the data to perform the stream
+ * rotation.
+ */
+ ret = do_rotate_stream_data(stream);
+ } else {
+ stream->rotate_at_seq_num = stream_info.rotate_at_seq_num;
+ ret = try_rotate_stream_data(stream);
+ if (ret < 0) {
+ goto end_stream_unlock;
+ }
+
+ ret = try_rotate_stream_index(stream);
+ if (ret < 0) {
+ goto end_stream_unlock;
+ }
+ }
+
+end_stream_unlock:
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+end:
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(struct lttcomm_relayd_generic_reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"rotate session stream\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ }
+
+end_no_reply:
+ return ret;
+}
+
+/*
+ * relay_mkdir: Create a folder on the disk.
+ */
+static int relay_mkdir(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_mkdir path_info_header;
+ struct lttcomm_relayd_generic_reply reply;
+ char *path = NULL;
+ size_t header_len;
+ ssize_t send_ret;
+ struct lttng_buffer_view path_view;
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to create a directory before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (session->major == 2 && session->minor < 11) {
+ /*
+ * This client is not supposed to use this command since
+ * it predates its introduction.
+ */
+ ERR("relay_mkdir command is unsupported before LTTng 2.11");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ header_len = sizeof(path_info_header);
+ if (payload->size < header_len) {
+ ERR("Unexpected payload size in \"relay_mkdir\": expected >= %zu bytes, got %zu bytes",
+ header_len, payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+
+ memcpy(&path_info_header, payload->data, header_len);
+
+ path_info_header.length = be32toh(path_info_header.length);
+
+ if (payload->size < header_len + path_info_header.length) {
+ ERR("Unexpected payload size in \"relay_mkdir\" including path: expected >= %zu bytes, got %zu bytes",
+ header_len + path_info_header.length, payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+
+ /* Ensure that it fits in local path length. */
+ if (path_info_header.length >= LTTNG_PATH_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("Path name argument of mkdir command (%" PRIu32 " bytes) exceeds the maximal length allowed (%d bytes)",
+ path_info_header.length, LTTNG_PATH_MAX);
+ goto end;
+ }
+
+ path_view = lttng_buffer_view_from_view(payload, header_len,
+ path_info_header.length);
+
+ path = create_output_path(path_view.data);
+ if (!path) {
+ ERR("Failed to create output path");
+ ret = -1;
+ goto end;
+ }
+
+ DBG("MKDIR command has path \"%s\", changed to \"%s\"", path_view.data, path);
+ ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, -1, -1);
+ if (ret < 0) {
+ ERR("relay creating output directory");
+ goto end;
+ }
+
+ ret = 0;
+
+end:
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"mkdir\" command reply (ret = %zd)", send_ret);
+ ret = -1;
+ }
+
+end_no_session:
+ free(path);
+ return ret;
+}
+
+static int validate_rotate_rename_path_length(const char *path_type,
+ uint32_t path_length)
+{
+ int ret = 0;
+
+ if (path_length > LTTNG_PATH_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("rotate rename \"%s\" path name length (%" PRIu32 " bytes) exceeds the allowed size of %i bytes",
+ path_type, path_length, LTTNG_PATH_MAX);
+ } else if (path_length == 0) {
+ ret = -EINVAL;
+ ERR("rotate rename \"%s\" path name has an illegal length of 0", path_type);
+ }
+ return ret;
+}
+
+/*
+ * relay_rotate_rename: rename the trace folder after a rotation is
+ * completed. We are not closing any fd here, just moving the folder, so it
+ * works even if data is still in-flight.
+ */
+static int relay_rotate_rename(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_generic_reply reply;
+ struct lttcomm_relayd_rotate_rename header;
+ size_t header_len;
+ size_t received_paths_size;
+ char *complete_old_path = NULL, *complete_new_path = NULL;
+ struct lttng_buffer_view old_path_view;
+ struct lttng_buffer_view new_path_view;
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to rename a trace folder before version check");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ if (session->major == 2 && session->minor < 11) {
+ ERR("relay_rotate_rename command is unsupported before LTTng 2.11");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ header_len = sizeof(header);
+ if (payload->size < header_len) {
+ ERR("Unexpected payload size in \"relay_rotate_rename\": expected >= %zu bytes, got %zu bytes",
+ header_len, payload->size);
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ memcpy(&header, payload->data, header_len);
+
+ header.old_path_length = be32toh(header.old_path_length);
+ header.new_path_length = be32toh(header.new_path_length);
+ received_paths_size = header.old_path_length + header.new_path_length;
+
+ if (payload->size < header_len + received_paths_size) {
+ ERR("Unexpected payload size in \"relay_rotate_rename\" including paths: expected >= %zu bytes, got %zu bytes",
+ header_len, payload->size);
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ /* Ensure the paths don't exceed their allowed size. */
+ ret = validate_rotate_rename_path_length("old", header.old_path_length);
+ if (ret) {
+ goto end;
+ }
+ ret = validate_rotate_rename_path_length("new", header.new_path_length);
+ if (ret) {
+ goto end;
+ }
+
+ old_path_view = lttng_buffer_view_from_view(payload, header_len,
+ header.old_path_length);
+ new_path_view = lttng_buffer_view_from_view(payload,
+ header_len + header.old_path_length,
+ header.new_path_length);
+
+ /* Validate that both paths received are NULL terminated. */
+ if (old_path_view.data[old_path_view.size - 1] != '\0') {
+ ERR("relay_rotate_rename command's \"old\" path is invalid (not NULL terminated)");
+ ret = -1;
+ goto end;
+ }
+ if (new_path_view.data[new_path_view.size - 1] != '\0') {
+ ERR("relay_rotate_rename command's \"new\" path is invalid (not NULL terminated)");
+ ret = -1;
+ goto end;
+ }
+
+ DBG("ROTATE_RENAME command has argument old path = \"%s\", new_path = \"%s\"",
+ old_path_view.data, new_path_view.data);
+ complete_old_path = create_output_path(old_path_view.data);
+ if (!complete_old_path) {
+ ERR("Failed to build old output path in rotate_rename command");
+ ret = -1;
+ goto end;
+ }
+
+ complete_new_path = create_output_path(new_path_view.data);
+ if (!complete_new_path) {
+ ERR("Failed to build new output path in rotate_rename command");
+ ret = -1;
+ goto end;
+ }
+ DBG("Expanded ROTATE_RENAME arguments to old path = \"%s\", new_path = \"%s\"",
+ complete_old_path, complete_new_path);
+
+ ret = utils_mkdir_recursive(complete_new_path, S_IRWXU | S_IRWXG,
+ -1, -1);
+ if (ret < 0) {
+ ERR("Failed to mkdir() rotate_rename's \"new\" output directory at \"%s\"",
+ complete_new_path);
+ goto end;
+ }
+
+ /*
+ * If a domain has not yet created its channel, the domain-specific
+ * folder might not exist, but this is not an error.
+ */
+ ret = rename(complete_old_path, complete_new_path);
+ if (ret < 0 && errno != ENOENT) {
+ PERROR("Renaming chunk in rotate_rename command from \"%s\" to \"%s\"",
+ complete_old_path, complete_new_path);
+ goto end;
+ }
+ ret = 0;
+
+end:
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(reply), 0);
+ if (send_ret < sizeof(reply)) {
+ ERR("Failed to send \"rotate rename\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ }
+
+end_no_reply:
+ free(complete_old_path);
+ free(complete_new_path);
+ return ret;
+}
+
+/*
+ * Check if all the streams in the session have completed the last rotation.
+ * The chunk_id value is used to distinguish the cases where a stream was
+ * closed on the consumerd before the rotation started but it still active on
+ * the relayd, and the case where a stream appeared on the consumerd/relayd
+ * after the last rotation started (in that case, it is already writing in the
+ * new chunk folder).
+ */
+static
+int relay_rotate_pending(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_rotate_pending msg;
+ struct lttcomm_relayd_rotate_pending_reply reply;
+ struct lttng_ht_iter iter;
+ struct relay_stream *stream;
+ int ret = 0;
+ ssize_t send_ret;
+ uint64_t chunk_id;
+ bool rotate_pending = false;
+
+ DBG("Rotate pending command received");
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to check for data before version check");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ if (session->major == 2 && session->minor < 11) {
+ ERR("Unsupported feature before 2.11");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ if (payload->size < sizeof(msg)) {
+ ERR("Unexpected payload size in \"relay_rotate_pending\": expected >= %zu bytes, got %zu bytes",
+ sizeof(msg), payload->size);
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ memcpy(&msg, payload->data, sizeof(msg));
+
+ chunk_id = be64toh(msg.chunk_id);
+
+ DBG("Evaluating rotate pending for session \"%s\" and chunk id %" PRIu64,
+ session->session_name, chunk_id);
+
+ /*
+ * Iterate over all the streams in the session and check if they are
+ * still waiting for data to perform their rotation.
+ */
+ rcu_read_lock();
+ cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
+ node.node) {
+ if (!stream_get(stream)) {
+ continue;
+ }
+ if (stream->trace->session != session) {
+ stream_put(stream);
+ continue;
+ }
+ pthread_mutex_lock(&stream->lock);
+ if (stream->rotate_at_seq_num != -1ULL) {
+ /* We have not yet performed the rotation. */
+ rotate_pending = true;
+ DBG("Stream %" PRIu64 " is still rotating",
+ stream->stream_handle);
+ } else if (stream->current_chunk_id.value <= chunk_id) {
+ /*
+ * Stream closed on the consumer but still active on the
+ * relay.
+ */
+ rotate_pending = true;
+ DBG("Stream %" PRIu64 " did not exist on the consumer "
+ "when the last rotation started, but is"
+ "still waiting for data before getting"
+ "closed",
+ stream->stream_handle);
+ }
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+ if (rotate_pending) {
+ goto send_reply;
+ }
+ }
+
+send_reply:
+ rcu_read_unlock();
+ memset(&reply, 0, sizeof(reply));
+ reply.generic.ret_code = htobe32((uint32_t) LTTNG_OK);
+ reply.is_pending = (uint8_t) !!rotate_pending;
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"rotate pending\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ }
+
+end_no_reply:
+ return ret;
+}
+
+#define DBG_CMD(cmd_name, conn) \
+ DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd);
+
+static int relay_process_control_command(struct relay_connection *conn,
+ const struct lttcomm_relayd_hdr *header,
+ const struct lttng_buffer_view *payload)
+{
+ int ret = 0;
+
+ switch (header->cmd) {
+ case RELAYD_CREATE_SESSION:
+ DBG_CMD("RELAYD_CREATE_SESSION", conn);
+ ret = relay_create_session(header, conn, payload);
+ break;