return ret;
}
+/*
+ * When a channel has finished the rotation of all its streams, inform the
+ * session daemon.
+ */
+static
+int rotate_notify_sessiond(struct lttng_consumer_local_data *ctx,
+ uint64_t key)
+{
+ int ret;
+
+ do {
+ ret = write(ctx->channel_rotate_pipe, &key, sizeof(key));
+ } while (ret == -1 && errno == EINTR);
+ if (ret == -1) {
+ PERROR("write to the channel rotate pipe");
+ } else {
+ DBG("Sent channel rotation notification for channel key %"
+ PRIu64, key);
+ }
+
+ return ret;
+}
+
+/*
+ * Performs the stream rotation for the rotate session feature if needed.
+ * It must be called with the stream and channel locks held.
+ *
+ * FIXME: find a way to lock the chan without deadlock. same for kernel.
+ *
+ * Return 0 on success, a negative number of error.
+ */
+static
+int stream_rotation(struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *stream)
+{
+ int ret;
+ unsigned long consumed_pos;
+
+ if (!stream->rotate_position && !stream->rotate_ready) {
+ ret = 0;
+ goto end;
+ }
+
+ /*
+ * If we don't have the rotate_ready flag, check the consumed position
+ * to determine if we need to rotate.
+ */
+ if (!stream->rotate_ready) {
+ ret = lttng_ustconsumer_sample_snapshot_positions(stream);
+ if (ret < 0) {
+ ERR("Taking UST snapshot positions");
+ goto error;
+ }
+
+ ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
+ if (ret < 0) {
+ ERR("Produced UST snapshot position");
+ goto error;
+ }
+
+ fprintf(stderr, "packet %lu, pos %lu\n", stream->key, consumed_pos);
+ /* Rotate position not reached yet. */
+ if (consumed_pos < stream->rotate_position) {
+ ret = 0;
+ goto end;
+ }
+ fprintf(stderr, "Rotate position %lu (expected %lu) reached for stream %lu\n",
+ consumed_pos, stream->rotate_position, stream->key);
+ } else {
+ fprintf(stderr, "Rotate position reached for stream %lu\n",
+ stream->key);
+ }
+
+ ret = close(stream->out_fd);
+ if (ret < 0) {
+ PERROR("Closing tracefile");
+ goto error;
+ }
+
+ fprintf(stderr, "Rotating stream %lu to %s/%s\n", stream->key,
+ stream->chan->pathname, stream->name);
+ ret = utils_create_stream_file(stream->chan->pathname, stream->name,
+ stream->chan->tracefile_size, stream->tracefile_count_current,
+ stream->uid, stream->gid, NULL);
+ if (ret < 0) {
+ 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->chan->pathname,
+ stream->name, stream->uid, stream->gid,
+ stream->chan->tracefile_size,
+ stream->tracefile_count_current,
+ CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
+ if (!index_file) {
+ goto error;
+ }
+ stream->index_file = index_file;
+ stream->out_fd_offset = 0;
+ } else {
+ /*
+ * Reset the position pushed from the metadata cache so it
+ * will write from the beginning on the next push.
+ */
+ stream->ust_metadata_pushed = 0;
+ /*
+ * Wakeup the metadata thread so it dumps the metadata cache
+ * to file again.
+ */
+ consumer_metadata_wakeup_pipe(stream->chan);
+ }
+
+ stream->rotate_position = 0;
+ stream->rotate_ready = 0;
+
+ if (--stream->chan->nr_stream_rotate_pending == 0) {
+ rotate_notify_sessiond(ctx, stream->chan->key);
+ fprintf(stderr, "SENT %lu\n", stream->chan->key);
+ }
+
+ ret = 0;
+ goto end;
+
+error:
+ ret = -1;
+end:
+ return ret;
+}
+
+/*
+ * Sample the rotate position for all the streams of a channel.
+ *
+ * Returns 0 on success, < 0 on error
+ */
+static
+int lttng_ustconsumer_rotate_channel(uint64_t key, char *path,
+ uint64_t relayd_id, uint32_t metadata,
+ struct lttng_consumer_local_data *ctx)
+{
+ int ret;
+ struct lttng_consumer_channel *channel;
+ struct lttng_consumer_stream *stream;
+ struct lttng_ht_iter iter;
+ struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
+
+ /* FIXME: metadata param is useless */
+
+ DBG("UST consumer sample rotate position for channel %" PRIu64, key);
+
+ rcu_read_lock();
+
+ channel = consumer_find_channel(key);
+ if (!channel) {
+ ERR("No channel found for key %" PRIu64, key);
+ ret = -1;
+ goto end;
+ }
+ pthread_mutex_lock(&channel->lock);
+ snprintf(channel->pathname, PATH_MAX, "%s", path);
+
+ cds_lfht_for_each_entry_duplicate(ht->ht,
+ ht->hash_fct(&channel->key, lttng_ht_seed),
+ ht->match_fct, &channel->key, &iter.iter,
+ stream, node_channel_id.node) {
+ uint64_t consumed_pos;
+ health_code_update();
+
+ /*
+ * Lock stream because we are about to change its state.
+ */
+ pthread_mutex_lock(&stream->lock);
+ ret = lttng_ustconsumer_sample_snapshot_positions(stream);
+ if (ret < 0) {
+ ERR("Taking UST snapshot positions");
+ goto end_unlock;
+ }
+
+ ret = lttng_ustconsumer_get_produced_snapshot(stream,
+ &stream->rotate_position);
+ if (ret < 0) {
+ ERR("Produced UST snapshot position");
+ goto end_unlock;
+ }
+ fprintf(stderr, "Stream %lu should rotate after %lu to %s\n",
+ stream->key, stream->rotate_position,
+ channel->pathname);
+ lttng_ustconsumer_get_consumed_snapshot(stream,
+ &consumed_pos);
+ fprintf(stderr, "consumed %lu\n", consumed_pos);
+ if (consumed_pos == stream->rotate_position) {
+ stream->rotate_ready = 1;
+ fprintf(stderr, "Stream %lu ready to rotate to %s\n",
+ stream->key, channel->pathname);
+ }
+ channel->nr_stream_rotate_pending++;
+
+ ustctl_flush_buffer(stream->ustream, 1);
+
+ pthread_mutex_unlock(&stream->lock);
+ }
+
+ ret = 0;
+ goto end_unlock_channel;
+
+end_unlock:
+ pthread_mutex_unlock(&stream->lock);
+end_unlock_channel:
+ pthread_mutex_unlock(&channel->lock);
+end:
+ rcu_read_unlock();
+ return ret;
+}
+
+/*
+ * Rotate all the ready streams.
+ *
+ * This is especially important for low throughput streams that have already
+ * been consumed, we cannot wait for their next packet to perform the
+ * rotation.
+ *
+ * Returns 0 on success, < 0 on error
+ */
+static
+int lttng_ustconsumer_rotate_ready_streams(uint64_t key,
+ struct lttng_consumer_local_data *ctx)
+{
+ int ret;
+ struct lttng_consumer_channel *channel;
+ struct lttng_consumer_stream *stream;
+ struct lttng_ht_iter iter;
+ struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
+
+ rcu_read_lock();
+
+ channel = consumer_find_channel(key);
+ if (!channel) {
+ ERR("No channel found for key %" PRIu64, key);
+ ret = -1;
+ goto end;
+ }
+ cds_lfht_for_each_entry_duplicate(ht->ht,
+ ht->hash_fct(&channel->key, lttng_ht_seed),
+ ht->match_fct, &channel->key, &iter.iter,
+ stream, node_channel_id.node) {
+ health_code_update();
+
+ /*
+ * Lock stream because we are about to change its state.
+ */
+ pthread_mutex_lock(&stream->lock);
+ if (stream->rotate_ready == 0) {
+ pthread_mutex_unlock(&stream->lock);
+ continue;
+ }
+ ret = stream_rotation(ctx, stream);
+ if (ret < 0) {
+ pthread_mutex_unlock(&stream->lock);
+ ERR("Stream rotation error");
+ goto end;
+ }
+
+ pthread_mutex_unlock(&stream->lock);
+ }
+
+ ret = 0;
+
+end:
+ rcu_read_unlock();
+ return ret;
+}
+
/*
* Receive the metadata updates from the sessiond. Supports receiving
* overlapping metadata, but is needs to always belong to a contiguous
}
goto end_msg_sessiond;
}
+ case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE:
+ {
+ int channel_rotate_pipe;
+ int flags;
+
+ ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ /* Successfully received the command's type. */
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ goto error_fatal;
+ }
+
+ ret = lttcomm_recv_fds_unix_sock(sock, &channel_rotate_pipe,
+ 1);
+ if (ret != sizeof(channel_rotate_pipe)) {
+ ERR("Failed to receive channel rotate pipe");
+ goto error_fatal;
+ }
+
+ DBG("Received channel rotate pipe (%d)", channel_rotate_pipe);
+ ctx->channel_rotate_pipe = channel_rotate_pipe;
+ /* Set the pipe as non-blocking. */
+ ret = fcntl(channel_rotate_pipe, F_GETFL, 0);
+ if (ret == -1) {
+ PERROR("fcntl get flags of the channel rotate pipe");
+ goto error_fatal;
+ }
+ flags = ret;
+
+ ret = fcntl(channel_rotate_pipe, F_SETFL,
+ flags | O_NONBLOCK);
+ if (ret == -1) {
+ PERROR("fcntl set O_NONBLOCK flag of the channel rotate pipe");
+ goto error_fatal;
+ }
+ DBG("Channel rotate pipe set as non-blocking");
+ ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ goto error_fatal;
+ }
+ break;
+ }
+ case LTTNG_CONSUMER_ROTATE_CHANNEL:
+ {
+ ret = lttng_ustconsumer_rotate_channel(msg.u.rotate_channel.key,
+ msg.u.rotate_channel.pathname,
+ msg.u.rotate_channel.relayd_id,
+ msg.u.rotate_channel.metadata,
+ ctx);
+ if (ret < 0) {
+ ERR("Rotate channel failed");
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ }
+
+ health_code_update();
+
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ /* Somehow, the session daemon is not responding anymore. */
+ goto end_nosignal;
+ }
+
+ /*
+ * Rotate the streams that are ready right now.
+ * FIXME: this is a second consecutive iteration over the
+ * streams in a channel, there is probably a better way to
+ * handle this, but it needs to be after the
+ * consumer_send_status_msg() call.
+ */
+ ret = lttng_ustconsumer_rotate_ready_streams(
+ msg.u.rotate_channel.key, ctx);
+ if (ret < 0) {
+ ERR("Rotate channel failed");
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ }
+ break;
+ }
default:
break;
}
struct lttng_consumer_local_data *ctx)
{
unsigned long len, subbuf_size, padding;
- int err, write_index = 1;
+ int err, write_index = 1, rotation_ret;
long ret = 0;
struct ustctl_consumer_stream *ustream;
struct ctf_packet_index index;
}
end:
+ /* FIXME: do we need this lock, it causes deadlocks when called
+ * at the same time with lttng_ustconsumer_rotate_channel ? */
+// pthread_mutex_lock(&stream->chan->lock);
+ rotation_ret = stream_rotation(ctx, stream);
+ if (rotation_ret < 0) {
+// pthread_mutex_unlock(&stream->chan->lock);
+ ret = -1;
+ ERR("Stream rotation error");
+ goto end;
+ }
+// pthread_mutex_unlock(&stream->chan->lock);
return ret;
}
projects / deliverable / lttng-tools.git / blobdiff