+
+/*
+ * Check if a stream is ready to be rotated after extracting it.
+ *
+ * Return 1 if it is ready for rotation, 0 if it is not, a negative value on
+ * error. Stream lock must be held.
+ */
+int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
+{
+ int ret;
+ unsigned long consumed_pos;
+
+ if (!stream->rotate_position && !stream->rotate_ready) {
+ ret = 0;
+ goto end;
+ }
+
+ if (stream->rotate_ready) {
+ ret = 1;
+ goto end;
+ }
+
+ /*
+ * If we don't have the rotate_ready flag, check the consumed position
+ * to determine if we need to rotate.
+ */
+ ret = lttng_consumer_sample_snapshot_positions(stream);
+ if (ret < 0) {
+ ERR("Taking snapshot positions");
+ goto end;
+ }
+
+ ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos);
+ if (ret < 0) {
+ ERR("Consumed snapshot position");
+ goto end;
+ }
+
+ /* Rotate position not reached yet (with check for overflow). */
+ if ((long) (consumed_pos - stream->rotate_position) < 0) {
+ ret = 0;
+ goto end;
+ }
+ ret = 1;
+
+end:
+ return ret;
+}
+
+/*
+ * Reset the state for a stream after a rotation occurred.
+ */
+void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream)
+{
+ stream->rotate_position = 0;
+ stream->rotate_ready = false;
+}
+
+/*
+ * Perform the rotation a local stream file.
+ */
+int rotate_local_stream(struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *stream)
+{
+ int ret;
+
+ DBG("Rotate local stream: stream key %" PRIu64 ", channel key %" PRIu64 " at path %s",
+ stream->key,
+ stream->chan->key,
+ stream->channel_read_only_attributes.path);
+
+ ret = close(stream->out_fd);
+ if (ret < 0) {
+ PERROR("Closing trace file (fd %d), stream %" PRIu64,
+ stream->out_fd, stream->key);
+ assert(0);
+ goto error;
+ }
+
+ ret = utils_create_stream_file(
+ stream->channel_read_only_attributes.path,
+ stream->name,
+ stream->channel_read_only_attributes.tracefile_size,
+ stream->tracefile_count_current,
+ stream->uid, stream->gid, NULL);
+ if (ret < 0) {
+ ERR("Rotate create stream file");
+ goto error;
+ }
+ stream->out_fd = ret;
+ stream->tracefile_size_current = 0;
+
+ if (!stream->metadata_flag) {
+ struct lttng_index_file *index_file;
+
+ lttng_index_file_put(stream->index_file);
+
+ index_file = lttng_index_file_create(
+ stream->channel_read_only_attributes.path,
+ stream->name, stream->uid, stream->gid,
+ stream->channel_read_only_attributes.tracefile_size,
+ stream->tracefile_count_current,
+ CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
+ if (!index_file) {
+ ERR("Create index file during rotation");
+ goto error;
+ }
+ stream->index_file = index_file;
+ stream->out_fd_offset = 0;
+ }
+
+ ret = 0;
+ goto end;
+
+error:
+ ret = -1;
+end:
+ return ret;
+
+}
+
+/*
+ * Perform the rotation a stream file on the relay.
+ */
+int rotate_relay_stream(struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *stream)
+{
+ int ret;
+ struct consumer_relayd_sock_pair *relayd;
+
+ DBG("Rotate relay stream");
+ relayd = consumer_find_relayd(stream->net_seq_idx);
+ if (!relayd) {
+ ERR("Failed to find relayd");
+ ret = -1;
+ goto end;
+ }
+
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ ret = relayd_rotate_stream(&relayd->control_sock,
+ stream->relayd_stream_id,
+ stream->channel_read_only_attributes.path,
+ stream->chan->current_chunk_id,
+ stream->last_sequence_number);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+ if (ret) {
+ ERR("Rotate relay stream");
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * Performs the stream rotation for the rotate session feature if needed.
+ * It must be called with the stream lock held.
+ *
+ * Return 0 on success, a negative number of error.
+ */
+int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *stream, bool *rotated)
+{
+ int ret;
+
+ DBG("Consumer rotate stream %" PRIu64, stream->key);
+
+ if (stream->net_seq_idx != (uint64_t) -1ULL) {
+ ret = rotate_relay_stream(ctx, stream);
+ } else {
+ ret = rotate_local_stream(ctx, stream);
+ }
+ if (ret < 0) {
+ ERR("Rotate stream");
+ goto error;
+ }
+
+ if (stream->metadata_flag) {
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ /*
+ * Reset the position of what has been read from the metadata
+ * cache to 0 so we can dump it again.
+ */
+ ret = kernctl_metadata_cache_dump(stream->wait_fd);
+ if (ret < 0) {
+ ERR("Failed to dump the kernel metadata cache after rotation");
+ goto error;
+ }
+ break;
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ /*
+ * Reset the position pushed from the metadata cache so it
+ * will write from the beginning on the next push.
+ */
+ stream->ust_metadata_pushed = 0;
+ break;
+ default:
+ ERR("Unknown consumer_data type");
+ abort();
+ }
+ }
+ lttng_consumer_reset_stream_rotate_state(stream);
+
+ if (rotated) {
+ *rotated = true;
+ }
+
+ ret = 0;
+
+error:
+ return ret;
+}
+
+static
+int rotate_rename_local(const char *old_path, const char *new_path,
+ uid_t uid, gid_t gid)
+{
+ int ret;
+
+ assert(old_path);
+ assert(new_path);
+
+ ret = utils_mkdir_recursive(new_path, S_IRWXU | S_IRWXG, uid, gid);
+ if (ret < 0) {
+ ERR("Create directory on rotate");
+ goto end;
+ }
+
+ ret = rename(old_path, new_path);
+ if (ret < 0 && errno != ENOENT) {
+ PERROR("Rename completed rotation chunk");
+ goto end;
+ }
+
+ ret = 0;
+end:
+ return ret;
+}
+
+static
+int rotate_rename_relay(const char *old_path, const char *new_path,
+ uint64_t relayd_id)
+{
+ int ret;
+ struct consumer_relayd_sock_pair *relayd;
+
+ relayd = consumer_find_relayd(relayd_id);
+ if (!relayd) {
+ ERR("Failed to find relayd while running rotate_rename_relay command");
+ ret = -1;
+ goto end;
+ }
+
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ ret = relayd_rotate_rename(&relayd->control_sock, old_path, new_path);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+end:
+ return ret;
+}
+
+int lttng_consumer_rotate_rename(const char *old_path, const char *new_path,
+ uid_t uid, gid_t gid, uint64_t relayd_id)
+{
+ if (relayd_id != -1ULL) {
+ return rotate_rename_relay(old_path, new_path, relayd_id);
+ } else {
+ return rotate_rename_local(old_path, new_path, uid, gid);
+ }
+}
+
+static
+int mkdir_local(const char *path, uid_t uid, gid_t gid)
+{
+ int ret;
+
+ ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, uid, gid);
+ if (ret < 0) {
+ /* utils_mkdir_recursive logs an error. */
+ goto end;
+ }
+
+ ret = 0;
+end:
+ return ret;
+}
+
+static
+int mkdir_relay(const char *path, uint64_t relayd_id)
+{
+ int ret;
+ struct consumer_relayd_sock_pair *relayd;
+
+ relayd = consumer_find_relayd(relayd_id);
+ if (!relayd) {
+ ERR("Failed to find relayd");
+ ret = -1;
+ goto end;
+ }
+
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ ret = relayd_mkdir(&relayd->control_sock, path);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+
+end:
+ return ret;
+
+}
+
+int lttng_consumer_mkdir(const char *path, uid_t uid, gid_t gid,
+ uint64_t relayd_id)
+{
+ if (relayd_id != -1ULL) {
+ return mkdir_relay(path, relayd_id);
+ } else {
+ return mkdir_local(path, uid, gid);
+ }
+}