int infd = stream->wait_fd;
ret = kernctl_snapshot(infd);
- if (ret != 0) {
+ /*
+ * -EAGAIN is not an error, it just means that there is no data to
+ * be read.
+ */
+ if (ret != 0 && ret != -EAGAIN) {
PERROR("Getting sub-buffer snapshot.");
}
/*
* Take a snapshot of all the stream of a channel
+ * RCU read-side lock must be held across this function to ensure existence of
+ * channel.
*
* Returns 0 on success, < 0 on error
*/
-int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
- uint64_t relayd_id, uint64_t nb_packets_per_stream,
+int lttng_kconsumer_snapshot_channel(struct lttng_consumer_channel *channel,
+ uint64_t key, char *path, uint64_t relayd_id, uint64_t nb_packets_per_stream,
struct lttng_consumer_local_data *ctx)
{
int ret;
- struct lttng_consumer_channel *channel;
struct lttng_consumer_stream *stream;
DBG("Kernel consumer snapshot 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;
- }
-
/* Splice is not supported yet for channel snapshot. */
if (channel->output != CONSUMER_CHANNEL_MMAP) {
ERR("Unsupported output %d", channel->output);
DBG("Kernel consumer snapshot stream %s/%s (%" PRIu64 ")",
path, stream->name, stream->key);
}
- if (relayd_id != -1ULL) {
- ret = consumer_send_relayd_streams_sent(relayd_id);
- if (ret < 0) {
- ERR("sending streams sent to relayd");
- goto end_unlock;
- }
- channel->streams_sent_to_relayd = true;
- }
ret = kernctl_buffer_flush_empty(stream->wait_fd);
if (ret < 0) {
/*
* Read the whole metadata available for a snapshot.
+ * RCU read-side lock must be held across this function to ensure existence of
+ * metadata_channel.
*
* Returns 0 on success, < 0 on error
*/
-int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
- uint64_t relayd_id, struct lttng_consumer_local_data *ctx)
+static int lttng_kconsumer_snapshot_metadata(struct lttng_consumer_channel *metadata_channel,
+ uint64_t key, char *path, uint64_t relayd_id,
+ struct lttng_consumer_local_data *ctx)
{
int ret, use_relayd = 0;
ssize_t ret_read;
- struct lttng_consumer_channel *metadata_channel;
struct lttng_consumer_stream *metadata_stream;
assert(ctx);
rcu_read_lock();
- metadata_channel = consumer_find_channel(key);
- if (!metadata_channel) {
- ERR("Kernel snapshot metadata not found for key %" PRIu64, key);
- ret = -1;
- goto error;
- }
-
metadata_stream = metadata_channel->metadata_stream;
assert(metadata_stream);
+ pthread_mutex_lock(&metadata_stream->lock);
/* Flag once that we have a valid relayd for the stream. */
if (relayd_id != (uint64_t) -1ULL) {
if (use_relayd) {
ret = consumer_send_relayd_stream(metadata_stream, path);
if (ret < 0) {
- goto error;
+ goto error_snapshot;
}
} else {
ret = utils_create_stream_file(path, metadata_stream->name,
metadata_stream->tracefile_count_current,
metadata_stream->uid, metadata_stream->gid, NULL);
if (ret < 0) {
- goto error;
+ goto error_snapshot;
}
metadata_stream->out_fd = ret;
}
do {
health_code_update();
- ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx);
+ ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx, NULL);
if (ret_read < 0) {
if (ret_read != -EAGAIN) {
ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)",
ret_read);
- goto error;
+ ret = ret_read;
+ goto error_snapshot;
}
/* ret_read is negative at this point so we will exit the loop. */
continue;
}
ret = 0;
-
+error_snapshot:
+ pthread_mutex_unlock(&metadata_stream->lock);
cds_list_del(&metadata_stream->send_node);
consumer_stream_destroy(metadata_stream, NULL);
metadata_channel->metadata_stream = NULL;
-error:
rcu_read_unlock();
return ret;
}
msg.u.stream.cpu,
&alloc_ret,
channel->type,
- channel->monitor);
+ channel->monitor,
+ msg.u.stream.trace_archive_id);
if (new_stream == NULL) {
switch (alloc_ret) {
case -ENOMEM:
new_stream->chan = channel;
new_stream->wait_fd = fd;
+ consumer_stream_update_channel_attributes(new_stream,
+ channel);
switch (channel->output) {
case CONSUMER_CHANNEL_SPLICE:
new_stream->output = LTTNG_EVENT_SPLICE;
/* Get the right pipe where the stream will be sent. */
if (new_stream->metadata_flag) {
- ret = consumer_add_metadata_stream(new_stream);
- if (ret) {
- ERR("Consumer add metadata stream %" PRIu64 " failed. Continuing",
- new_stream->key);
- consumer_stream_free(new_stream);
- goto end_nosignal;
- }
+ consumer_add_metadata_stream(new_stream);
stream_pipe = ctx->consumer_metadata_pipe;
} else {
- ret = consumer_add_data_stream(new_stream);
- if (ret) {
- ERR("Consumer add stream %" PRIu64 " failed. Continuing",
- new_stream->key);
- consumer_stream_free(new_stream);
- goto end_nosignal;
- }
+ consumer_add_data_stream(new_stream);
stream_pipe = ctx->consumer_data_pipe;
}
- /* Vitible to other threads */
+ /* Visible to other threads */
new_stream->globally_visible = 1;
health_code_update();
goto end_nosignal;
}
- DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
- new_stream->name, fd, new_stream->relayd_stream_id);
+ DBG("Kernel consumer ADD_STREAM %s (fd: %d) %s with relayd id %" PRIu64,
+ new_stream->name, fd, new_stream->chan->pathname, new_stream->relayd_stream_id);
break;
}
case LTTNG_CONSUMER_STREAMS_SENT:
}
case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
{
- if (msg.u.snapshot_channel.metadata == 1) {
- ret = lttng_kconsumer_snapshot_metadata(msg.u.snapshot_channel.key,
- msg.u.snapshot_channel.pathname,
- msg.u.snapshot_channel.relayd_id, ctx);
- if (ret < 0) {
- ERR("Snapshot metadata failed");
- ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
- }
+ struct lttng_consumer_channel *channel;
+ uint64_t key = msg.u.snapshot_channel.key;
+
+ channel = consumer_find_channel(key);
+ if (!channel) {
+ ERR("Channel %" PRIu64 " not found", key);
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
} else {
- ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
- msg.u.snapshot_channel.pathname,
- msg.u.snapshot_channel.relayd_id,
- msg.u.snapshot_channel.nb_packets_per_stream,
- ctx);
- if (ret < 0) {
- ERR("Snapshot channel failed");
- ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ if (msg.u.snapshot_channel.metadata == 1) {
+ ret = lttng_kconsumer_snapshot_metadata(channel, key,
+ msg.u.snapshot_channel.pathname,
+ msg.u.snapshot_channel.relayd_id, ctx);
+ if (ret < 0) {
+ ERR("Snapshot metadata failed");
+ ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
+ }
+ } else {
+ ret = lttng_kconsumer_snapshot_channel(channel, key,
+ msg.u.snapshot_channel.pathname,
+ msg.u.snapshot_channel.relayd_id,
+ msg.u.snapshot_channel.nb_packets_per_stream,
+ ctx);
+ if (ret < 0) {
+ ERR("Snapshot channel failed");
+ ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
+ }
}
}
-
health_code_update();
ret = consumer_send_status_msg(sock, ret_code);
}
break;
}
+ case LTTNG_CONSUMER_ROTATE_CHANNEL:
+ {
+ struct lttng_consumer_channel *channel;
+ uint64_t key = msg.u.rotate_channel.key;
+
+ DBG("Consumer rotate channel %" PRIu64, key);
+
+ channel = consumer_find_channel(key);
+ if (!channel) {
+ ERR("Channel %" PRIu64 " not found", key);
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ } else {
+ /*
+ * Sample the rotate position of all the streams in this channel.
+ */
+ ret = lttng_consumer_rotate_channel(channel, key,
+ msg.u.rotate_channel.pathname,
+ msg.u.rotate_channel.relayd_id,
+ msg.u.rotate_channel.metadata,
+ msg.u.rotate_channel.new_chunk_id,
+ ctx);
+ if (ret < 0) {
+ ERR("Rotate channel failed");
+ ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL;
+ }
+
+ 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;
+ }
+ if (channel) {
+ /* Rotate the streams that are ready right now. */
+ ret = lttng_consumer_rotate_ready_streams(
+ channel, key, ctx);
+ if (ret < 0) {
+ ERR("Rotate ready streams failed");
+ }
+ }
+
+ break;
+ }
+ case LTTNG_CONSUMER_ROTATE_RENAME:
+ {
+ DBG("Consumer rename session %" PRIu64 " after rotation, old path = \"%s\", new path = \"%s\"",
+ msg.u.rotate_rename.session_id,
+ msg.u.rotate_rename.old_path,
+ msg.u.rotate_rename.new_path);
+ ret = lttng_consumer_rotate_rename(msg.u.rotate_rename.old_path,
+ msg.u.rotate_rename.new_path,
+ msg.u.rotate_rename.uid,
+ msg.u.rotate_rename.gid,
+ msg.u.rotate_rename.relayd_id);
+ if (ret < 0) {
+ ERR("Rotate rename failed");
+ ret_code = LTTCOMM_CONSUMERD_ROTATE_RENAME_FAILED;
+ }
+
+ 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;
+ }
+ break;
+ }
+ case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_LOCAL:
+ {
+ int pending;
+ uint32_t pending_reply;
+
+ DBG("Perform local check of pending rotation for session id %" PRIu64,
+ msg.u.check_rotation_pending_local.session_id);
+ pending = lttng_consumer_check_rotation_pending_local(
+ msg.u.check_rotation_pending_local.session_id,
+ msg.u.check_rotation_pending_local.chunk_id);
+ if (pending < 0) {
+ ERR("Local rotation pending check failed with code %i", pending);
+ ret_code = LTTCOMM_CONSUMERD_ROTATION_PENDING_LOCAL_FAILED;
+ } else {
+ pending_reply = !!pending;
+ }
+
+ 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;
+ }
+
+ if (pending < 0) {
+ /*
+ * An error occured while running the command;
+ * don't send the 'pending' flag as the sessiond
+ * will not read it.
+ */
+ break;
+ }
+
+ /* Send back returned value to session daemon */
+ ret = lttcomm_send_unix_sock(sock, &pending_reply,
+ sizeof(pending_reply));
+ if (ret < 0) {
+ PERROR("Failed to send rotation pending return code");
+ goto error_fatal;
+ }
+ break;
+ }
+ case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_RELAY:
+ {
+ int pending;
+ uint32_t pending_reply;
+
+ DBG("Perform relayd check of pending rotation for session id %" PRIu64,
+ msg.u.check_rotation_pending_relay.session_id);
+ pending = lttng_consumer_check_rotation_pending_relay(
+ msg.u.check_rotation_pending_relay.session_id,
+ msg.u.check_rotation_pending_relay.relayd_id,
+ msg.u.check_rotation_pending_relay.chunk_id);
+ if (pending < 0) {
+ ERR("Relayd rotation pending check failed with code %i", pending);
+ ret_code = LTTCOMM_CONSUMERD_ROTATION_PENDING_RELAY_FAILED;
+ } else {
+ pending_reply = !!pending;
+ }
+
+ 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;
+ }
+
+ if (pending < 0) {
+ /*
+ * An error occured while running the command;
+ * don't send the 'pending' flag as the sessiond
+ * will not read it.
+ */
+ break;
+ }
+
+ /* Send back returned value to session daemon */
+ ret = lttcomm_send_unix_sock(sock, &pending_reply,
+ sizeof(pending_reply));
+ if (ret < 0) {
+ PERROR("Failed to send rotation pending return code");
+ goto error_fatal;
+ }
+ break;
+ }
+ case LTTNG_CONSUMER_MKDIR:
+ {
+ DBG("Consumer mkdir %s in session %" PRIu64,
+ msg.u.mkdir.path,
+ msg.u.mkdir.session_id);
+ ret = lttng_consumer_mkdir(msg.u.mkdir.path,
+ msg.u.mkdir.uid,
+ msg.u.mkdir.gid,
+ msg.u.mkdir.relayd_id);
+ if (ret < 0) {
+ ERR("consumer mkdir failed");
+ ret_code = LTTCOMM_CONSUMERD_MKDIR_FAILED;
+ }
+
+ 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;
+ }
+ break;
+ }
default:
goto end_nosignal;
}
* Consume data on a file descriptor and write it on a trace file.
*/
ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
- struct lttng_consumer_local_data *ctx)
+ struct lttng_consumer_local_data *ctx, bool *rotated)
{
unsigned long len, subbuf_size, padding;
- int err, write_index = 1;
+ int err, write_index = 1, rotation_ret;
ssize_t ret = 0;
int infd = stream->wait_fd;
struct ctf_packet_index index;
DBG("In read_subbuffer (infd : %d)", infd);
+ /*
+ * If the stream was flagged to be ready for rotation before we extract the
+ * next packet, rotate it now.
+ */
+ if (stream->rotate_ready) {
+ DBG("Rotate stream before extracting data");
+ rotation_ret = lttng_consumer_rotate_stream(ctx, stream, rotated);
+ if (rotation_ret < 0) {
+ ERR("Stream rotation error");
+ ret = -1;
+ goto error;
+ }
+ }
+
/* Get the next subbuffer */
err = kernctl_get_next_subbuf(infd);
if (err != 0) {
DBG("Reserving sub buffer failed (everything is normal, "
"it is due to concurrency)");
ret = err;
- goto end;
+ goto error;
}
/* Get the full subbuffer size including padding */
PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
}
ret = err;
- goto end;
+ goto error;
}
ret = err;
- goto end;
+ goto error;
}
if (!stream->metadata_flag) {
PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
}
ret = err;
- goto end;
+ goto error;
}
- goto end;
+ goto error;
}
ret = update_stream_stats(stream);
if (ret < 0) {
PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
}
ret = err;
- goto end;
+ goto error;
}
- goto end;
+ goto error;
}
} else {
write_index = 0;
PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
}
ret = err;
- goto end;
+ goto error;
}
- goto end;
+ goto error;
}
}
PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
}
ret = err;
- goto end;
+ goto error;
}
ret = err;
- goto end;
+ goto error;
}
/* Make sure the tracer is not gone mad on us! */
PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
}
ret = err;
- goto end;
+ goto error;
}
/* Write index if needed. */
if (!write_index) {
- goto end;
+ goto rotate;
}
if (stream->chan->live_timer_interval && !stream->metadata_flag) {
pthread_mutex_unlock(&stream->metadata_timer_lock);
}
if (err < 0) {
- goto end;
+ goto error;
}
}
err = consumer_stream_write_index(stream, &index);
if (err < 0) {
- goto end;
+ goto error;
}
-end:
+rotate:
+ /*
+ * After extracting the packet, we check if the stream is now ready to be
+ * rotated and perform the action immediately.
+ */
+ rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
+ if (rotation_ret == 1) {
+ rotation_ret = lttng_consumer_rotate_stream(ctx, stream, rotated);
+ if (rotation_ret < 0) {
+ ERR("Stream rotation error");
+ ret = -1;
+ goto error;
+ }
+ } else if (rotation_ret < 0) {
+ ERR("Checking if stream is ready to rotate");
+ ret = -1;
+ goto error;
+ }
+
+error:
return ret;
}
projects / lttng-tools.git / blobdiff