X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=7032a7f7ffcc9bea224db0c6059f1928de72aa47;hp=32169034095be1b9fe8889e3c9aee0906dc1e58f;hb=a2814ea7573bf5edd5323d6f89c48ff14105db69;hpb=e57427575fc6deabed09cb5b4fae029d05e5bfd9 diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 321690340..7032a7f7f 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -1,21 +1,14 @@ /* - * Copyright (C) 2011 - Julien Desfossez - * Mathieu Desnoyers + * Copyright (C) 2011 Julien Desfossez + * Copyright (C) 2011 Mathieu Desnoyers + * Copyright (C) 2017 Jérémie Galarneau * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License, version 2 only, - * as published by the Free Software Foundation. + * SPDX-License-Identifier: GPL-2.0-only * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License along - * with this program; if not, write to the Free Software Foundation, Inc., - * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ +#include "common/buffer-view.h" +#include #define _LGPL_SOURCE #include #include @@ -42,12 +35,12 @@ #include #include #include +#include #include "kernel-consumer.h" extern struct lttng_consumer_global_data consumer_data; extern int consumer_poll_timeout; -extern volatile int consumer_quit; /* * Take a snapshot for a specific fd @@ -60,14 +53,30 @@ int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream) 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."); - ret = -errno; } return ret; } +/* + * Sample consumed and produced positions for a specific fd. + * + * Returns 0 on success, < 0 on error. + */ +int lttng_kconsumer_sample_snapshot_positions( + struct lttng_consumer_stream *stream) +{ + assert(stream); + + return kernctl_snapshot_sample_positions(stream->wait_fd); +} + /* * Get the produced position * @@ -82,7 +91,6 @@ int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream, ret = kernctl_snapshot_get_produced(infd, pos); if (ret != 0) { PERROR("kernctl_snapshot_get_produced"); - ret = -errno; } return ret; @@ -102,47 +110,60 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream, ret = kernctl_snapshot_get_consumed(infd, pos); if (ret != 0) { PERROR("kernctl_snapshot_get_consumed"); - ret = -errno; } return ret; } +static +int get_current_subbuf_addr(struct lttng_consumer_stream *stream, + const char **addr) +{ + int ret; + unsigned long mmap_offset; + const char *mmap_base = stream->mmap_base; + + ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset); + if (ret < 0) { + PERROR("Failed to get mmap read offset"); + goto error; + } + + *addr = mmap_base + mmap_offset; +error: + return ret; +} + /* * 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. The channel lock must be held by the caller. * * 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, +static 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; - unsigned long consumed_pos, produced_pos; - 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); + ERR("Unsupported output type for channel \"%s\": mmap output is required to record a snapshot", + channel->name); ret = -1; goto end; } cds_list_for_each_entry(stream, &channel->streams.head, send_node) { - /* Are we at a position _before_ the first available packet ? */ - bool before_first_packet = true; + unsigned long consumed_pos, produced_pos; health_code_update(); @@ -151,6 +172,19 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, */ pthread_mutex_lock(&stream->lock); + assert(channel->trace_chunk); + if (!lttng_trace_chunk_get(channel->trace_chunk)) { + /* + * Can't happen barring an internal error as the channel + * holds a reference to the trace chunk. + */ + ERR("Failed to acquire reference to channel's trace chunk"); + ret = -1; + goto end_unlock; + } + assert(!stream->trace_chunk); + stream->trace_chunk = channel->trace_chunk; + /* * Assign the received relayd ID so we can use it for streaming. The streams * are not visible to anyone so this is OK to change it. @@ -164,33 +198,29 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, goto end_unlock; } } else { - ret = utils_create_stream_file(path, stream->name, - stream->chan->tracefile_size, - stream->tracefile_count_current, - stream->uid, stream->gid, NULL); + ret = consumer_stream_create_output_files(stream, + false); if (ret < 0) { - ERR("utils_create_stream_file"); goto end_unlock; } - - stream->out_fd = ret; - stream->tracefile_size_current = 0; - - DBG("Kernel consumer snapshot stream %s/%s (%" PRIu64 ")", - path, stream->name, stream->key); + DBG("Kernel consumer snapshot stream (%" PRIu64 ")", + stream->key); } - if (relayd_id != -1ULL) { - ret = consumer_send_relayd_streams_sent(relayd_id); + + ret = kernctl_buffer_flush_empty(stream->wait_fd); + if (ret < 0) { + /* + * Doing a buffer flush which does not take into + * account empty packets. This is not perfect + * for stream intersection, but required as a + * fall-back when "flush_empty" is not + * implemented by lttng-modules. + */ + ret = kernctl_buffer_flush(stream->wait_fd); if (ret < 0) { - ERR("sending streams sent to relayd"); + ERR("Failed to flush kernel stream"); goto end_unlock; } - } - - ret = kernctl_buffer_flush(stream->wait_fd); - if (ret < 0) { - ERR("Failed to flush kernel stream"); - ret = -errno; goto end_unlock; } @@ -212,65 +242,52 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, goto end_unlock; } - if (stream->max_sb_size == 0) { - ret = kernctl_get_max_subbuf_size(stream->wait_fd, - &stream->max_sb_size); - if (ret < 0) { - ERR("Getting kernel max_sb_size"); - ret = -errno; - goto end_unlock; - } - } - consumed_pos = consumer_get_consume_start_pos(consumed_pos, produced_pos, nb_packets_per_stream, stream->max_sb_size); - while (consumed_pos < produced_pos) { + while ((long) (consumed_pos - produced_pos) < 0) { ssize_t read_len; unsigned long len, padded_len; - int lost_packet = 0; + const char *subbuf_addr; + struct lttng_buffer_view subbuf_view; health_code_update(); - DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos); ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos); if (ret < 0) { - if (errno != EAGAIN) { + if (ret != -EAGAIN) { PERROR("kernctl_get_subbuf snapshot"); - ret = -errno; goto end_unlock; } DBG("Kernel consumer get subbuf failed. Skipping it."); consumed_pos += stream->max_sb_size; - - /* - * Start accounting lost packets only when we - * already have extracted packets (to match the - * content of the final snapshot). - */ - if (!before_first_packet) { - lost_packet = 1; - } + stream->chan->lost_packets++; continue; } ret = kernctl_get_subbuf_size(stream->wait_fd, &len); if (ret < 0) { ERR("Snapshot kernctl_get_subbuf_size"); - ret = -errno; goto error_put_subbuf; } ret = kernctl_get_padded_subbuf_size(stream->wait_fd, &padded_len); if (ret < 0) { ERR("Snapshot kernctl_get_padded_subbuf_size"); - ret = -errno; goto error_put_subbuf; } - read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len, + ret = get_current_subbuf_addr(stream, &subbuf_addr); + if (ret) { + goto error_put_subbuf; + } + + subbuf_view = lttng_buffer_view_init( + subbuf_addr, 0, padded_len); + read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, + stream, &subbuf_view, padded_len - len, NULL); /* * We write the padded len in local tracefiles but the data len @@ -292,20 +309,9 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, ret = kernctl_put_subbuf(stream->wait_fd); if (ret < 0) { ERR("Snapshot kernctl_put_subbuf"); - ret = -errno; goto end_unlock; } consumed_pos += stream->max_sb_size; - - /* - * Only account lost packets located between - * succesfully extracted packets (do not account before - * and after since they are not visible in the - * resulting snapshot). - */ - stream->chan->lost_packets += lost_packet; - lost_packet = 0; - before_first_packet = false; } if (relayd_id == (uint64_t) -1ULL) { @@ -321,6 +327,8 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, close_relayd_stream(stream); stream->net_seq_idx = (uint64_t) -1ULL; } + lttng_trace_chunk_put(stream->trace_chunk); + stream->trace_chunk = NULL; pthread_mutex_unlock(&stream->lock); } @@ -331,7 +339,6 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, error_put_subbuf: ret = kernctl_put_subbuf(stream->wait_fd); if (ret < 0) { - ret = -errno; ERR("Snapshot kernctl_put_subbuf error path"); } end_unlock: @@ -343,15 +350,18 @@ end: /* * Read the whole metadata available for a snapshot. + * RCU read-side lock must be held across this function to ensure existence of + * metadata_channel. The channel lock must be held by the caller. * * 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); @@ -361,16 +371,13 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, 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); + assert(metadata_channel->trace_chunk); + assert(metadata_stream->trace_chunk); + /* Flag once that we have a valid relayd for the stream. */ if (relayd_id != (uint64_t) -1ULL) { use_relayd = 1; @@ -379,17 +386,14 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, 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->chan->tracefile_size, - metadata_stream->tracefile_count_current, - metadata_stream->uid, metadata_stream->gid, NULL); + ret = consumer_stream_create_output_files(metadata_stream, + false); if (ret < 0) { - goto error; + goto error_snapshot; } - metadata_stream->out_fd = ret; } do { @@ -400,7 +404,8 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, 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; @@ -421,15 +426,17 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, */ } metadata_stream->out_fd = -1; + lttng_trace_chunk_put(metadata_stream->trace_chunk); + metadata_stream->trace_chunk = NULL; } } 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; } @@ -471,16 +478,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, case LTTNG_CONSUMER_ADD_RELAYD_SOCKET: { /* Session daemon status message are handled in the following call. */ - ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index, + consumer_add_relayd_socket(msg.u.relayd_sock.net_index, msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll, &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id, - msg.u.relayd_sock.relayd_session_id); + msg.u.relayd_sock.relayd_session_id); goto end_nosignal; } case LTTNG_CONSUMER_ADD_CHANNEL: { struct lttng_consumer_channel *new_channel; int ret_recv; + const uint64_t chunk_id = msg.u.channel.chunk_id.value; health_code_update(); @@ -495,13 +503,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, DBG("consumer_add_channel %" PRIu64, msg.u.channel.channel_key); new_channel = consumer_allocate_channel(msg.u.channel.channel_key, - msg.u.channel.session_id, msg.u.channel.pathname, - msg.u.channel.name, msg.u.channel.uid, msg.u.channel.gid, + msg.u.channel.session_id, + msg.u.channel.chunk_id.is_set ? + &chunk_id : NULL, + msg.u.channel.pathname, + msg.u.channel.name, msg.u.channel.relayd_id, msg.u.channel.output, msg.u.channel.tracefile_size, msg.u.channel.tracefile_count, 0, msg.u.channel.monitor, msg.u.channel.live_timer_interval, + msg.u.channel.is_live, NULL, NULL); if (new_channel == NULL) { lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR); @@ -543,9 +555,20 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } else { ret = consumer_add_channel(new_channel, ctx); } - if (CONSUMER_CHANNEL_TYPE_DATA) { + if (msg.u.channel.type == CONSUMER_CHANNEL_TYPE_DATA && !ret) { + int monitor_start_ret; + + DBG("Consumer starting monitor timer"); consumer_timer_live_start(new_channel, msg.u.channel.live_timer_interval); + monitor_start_ret = consumer_timer_monitor_start( + new_channel, + msg.u.channel.monitor_timer_interval); + if (monitor_start_ret < 0) { + ERR("Starting channel monitoring timer failed"); + goto end_nosignal; + } + } health_code_update(); @@ -589,14 +612,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ - goto error_fatal; + goto error_add_stream_fatal; } health_code_update(); if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) { /* Channel was not found. */ - goto end_nosignal; + goto error_add_stream_nosignal; } /* Blocking call */ @@ -604,7 +627,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = lttng_consumer_poll_socket(consumer_sockpoll); health_poll_exit(); if (ret) { - goto error_fatal; + goto error_add_stream_fatal; } health_code_update(); @@ -613,8 +636,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1); if (ret != sizeof(fd)) { lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD); - rcu_read_unlock(); - return ret; + goto end; } health_code_update(); @@ -627,19 +649,20 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ - goto end_nosignal; + goto error_add_stream_nosignal; } health_code_update(); - new_stream = consumer_allocate_stream(channel->key, + pthread_mutex_lock(&channel->lock); + new_stream = consumer_allocate_stream( + channel, + channel->key, fd, - LTTNG_CONSUMER_ACTIVE_STREAM, channel->name, - channel->uid, - channel->gid, channel->relayd_id, channel->session_id, + channel->trace_chunk, msg.u.stream.cpu, &alloc_ret, channel->type, @@ -652,17 +675,28 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR); break; } - goto end_nosignal; + pthread_mutex_unlock(&channel->lock); + goto error_add_stream_nosignal; } - new_stream->chan = channel; new_stream->wait_fd = fd; + ret = kernctl_get_max_subbuf_size(new_stream->wait_fd, + &new_stream->max_sb_size); + if (ret < 0) { + pthread_mutex_unlock(&channel->lock); + ERR("Failed to get kernel maximal subbuffer size"); + goto error_add_stream_nosignal; + } + + consumer_stream_update_channel_attributes(new_stream, + channel); switch (channel->output) { case CONSUMER_CHANNEL_SPLICE: new_stream->output = LTTNG_EVENT_SPLICE; ret = utils_create_pipe(new_stream->splice_pipe); if (ret < 0) { - goto end_nosignal; + pthread_mutex_unlock(&channel->lock); + goto error_add_stream_nosignal; } break; case CONSUMER_CHANNEL_MMAP: @@ -670,7 +704,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, break; default: ERR("Stream output unknown %d", channel->output); - goto end_nosignal; + pthread_mutex_unlock(&channel->lock); + goto error_add_stream_nosignal; } /* @@ -695,14 +730,16 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); + pthread_mutex_lock(&new_stream->lock); if (ctx->on_recv_stream) { ret = ctx->on_recv_stream(new_stream); if (ret < 0) { + pthread_mutex_unlock(&new_stream->lock); + pthread_mutex_unlock(&channel->lock); consumer_stream_free(new_stream); - goto end_nosignal; + goto error_add_stream_nosignal; } } - health_code_update(); if (new_stream->metadata_flag) { @@ -715,7 +752,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, "relayd id %" PRIu64, new_stream->name, new_stream->net_seq_idx); cds_list_add(&new_stream->send_node, &channel->streams.head); - break; + pthread_mutex_unlock(&new_stream->lock); + pthread_mutex_unlock(&channel->lock); + goto end_add_stream; } /* Send stream to relayd if the stream has an ID. */ @@ -723,33 +762,40 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = consumer_send_relayd_stream(new_stream, new_stream->chan->pathname); if (ret < 0) { + pthread_mutex_unlock(&new_stream->lock); + pthread_mutex_unlock(&channel->lock); consumer_stream_free(new_stream); - goto end_nosignal; + goto error_add_stream_nosignal; + } + + /* + * If adding an extra stream to an already + * existing channel (e.g. cpu hotplug), we need + * to send the "streams_sent" command to relayd. + */ + if (channel->streams_sent_to_relayd) { + ret = consumer_send_relayd_streams_sent( + new_stream->net_seq_idx); + if (ret < 0) { + pthread_mutex_unlock(&new_stream->lock); + pthread_mutex_unlock(&channel->lock); + goto error_add_stream_nosignal; + } } } + pthread_mutex_unlock(&new_stream->lock); + pthread_mutex_unlock(&channel->lock); /* 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(); @@ -764,12 +810,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } else { consumer_del_stream_for_data(new_stream); } - goto end_nosignal; + goto error_add_stream_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); +end_add_stream: break; +error_add_stream_nosignal: + goto end_nosignal; +error_add_stream_fatal: + goto error_fatal; } case LTTNG_CONSUMER_STREAMS_SENT: { @@ -798,7 +849,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = consumer_send_status_msg(sock, ret_code); if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) { /* Somehow, the session daemon is not responding anymore. */ - goto end_nosignal; + goto error_streams_sent_nosignal; } health_code_update(); @@ -808,7 +859,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, * streams in this channel. */ if (!channel->monitor) { - break; + goto end_error_streams_sent; } health_code_update(); @@ -817,10 +868,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = consumer_send_relayd_streams_sent( msg.u.sent_streams.net_seq_idx); if (ret < 0) { - goto end_nosignal; + goto error_streams_sent_nosignal; } + channel->streams_sent_to_relayd = true; } +end_error_streams_sent: break; +error_streams_sent_nosignal: + goto end_nosignal; } case LTTNG_CONSUMER_UPDATE_STREAM: { @@ -891,26 +946,36 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } 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; + pthread_mutex_lock(&channel->lock); + 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; + } } + pthread_mutex_unlock(&channel->lock); } - health_code_update(); ret = consumer_send_status_msg(sock, ret_code); @@ -936,14 +1001,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ - goto end_nosignal; + goto end_destroy_channel; } health_code_update(); /* Stop right now if no channel was found. */ if (!channel) { - goto end_nosignal; + goto end_destroy_channel; } /* @@ -959,12 +1024,13 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, assert(!uatomic_sub_return(&channel->refcount, 1)); consumer_del_channel(channel); - +end_destroy_channel: goto end_nosignal; } case LTTNG_CONSUMER_DISCARDED_EVENTS: { - uint64_t ret; + ssize_t ret; + uint64_t count; struct lttng_consumer_channel *channel; uint64_t id = msg.u.discarded_events.session_id; uint64_t key = msg.u.discarded_events.channel_key; @@ -976,15 +1042,15 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel) { ERR("Kernel consumer discarded events channel %" PRIu64 " not found", key); - ret = 0; + count = 0; } else { - ret = channel->discarded_events; + count = channel->discarded_events; } health_code_update(); /* Send back returned value to session daemon */ - ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + ret = lttcomm_send_unix_sock(sock, &count, sizeof(count)); if (ret < 0) { PERROR("send discarded events"); goto error_fatal; @@ -994,7 +1060,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } case LTTNG_CONSUMER_LOST_PACKETS: { - uint64_t ret; + ssize_t ret; + uint64_t count; struct lttng_consumer_channel *channel; uint64_t id = msg.u.lost_packets.session_id; uint64_t key = msg.u.lost_packets.channel_key; @@ -1006,15 +1073,15 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel) { ERR("Kernel consumer lost packets channel %" PRIu64 " not found", key); - ret = 0; + count = 0; } else { - ret = channel->lost_packets; + count = channel->lost_packets; } health_code_update(); /* Send back returned value to session daemon */ - ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + ret = lttcomm_send_unix_sock(sock, &count, sizeof(count)); if (ret < 0) { PERROR("send lost packets"); goto error_fatal; @@ -1022,24 +1089,273 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, break; } + case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE: + { + int channel_monitor_pipe; + + 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_monitor_pipe, + 1); + if (ret != sizeof(channel_monitor_pipe)) { + ERR("Failed to receive channel monitor pipe"); + goto error_fatal; + } + + DBG("Received channel monitor pipe (%d)", channel_monitor_pipe); + ret = consumer_timer_thread_set_channel_monitor_pipe( + channel_monitor_pipe); + if (!ret) { + int flags; + + ret_code = LTTCOMM_CONSUMERD_SUCCESS; + /* Set the pipe as non-blocking. */ + ret = fcntl(channel_monitor_pipe, F_GETFL, 0); + if (ret == -1) { + PERROR("fcntl get flags of the channel monitoring pipe"); + goto error_fatal; + } + flags = ret; + + ret = fcntl(channel_monitor_pipe, F_SETFL, + flags | O_NONBLOCK); + if (ret == -1) { + PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe"); + goto error_fatal; + } + DBG("Channel monitor pipe set as non-blocking"); + } else { + ret_code = LTTCOMM_CONSUMERD_ALREADY_SET; + } + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + goto error_fatal; + } + 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.relayd_id, + msg.u.rotate_channel.metadata, + 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 error_rotate_channel; + } + 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; +error_rotate_channel: + goto end_nosignal; + } + case LTTNG_CONSUMER_CLEAR_CHANNEL: + { + struct lttng_consumer_channel *channel; + uint64_t key = msg.u.clear_channel.key; + + channel = consumer_find_channel(key); + if (!channel) { + DBG("Channel %" PRIu64 " not found", key); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } else { + ret = lttng_consumer_clear_channel(channel); + if (ret) { + ERR("Clear channel failed"); + ret_code = ret; + } + + 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_INIT: + { + ret_code = lttng_consumer_init_command(ctx, + msg.u.init.sessiond_uuid); + 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_CREATE_TRACE_CHUNK: + { + const struct lttng_credentials credentials = { + .uid = msg.u.create_trace_chunk.credentials.value.uid, + .gid = msg.u.create_trace_chunk.credentials.value.gid, + }; + const bool is_local_trace = + !msg.u.create_trace_chunk.relayd_id.is_set; + const uint64_t relayd_id = + msg.u.create_trace_chunk.relayd_id.value; + const char *chunk_override_name = + *msg.u.create_trace_chunk.override_name ? + msg.u.create_trace_chunk.override_name : + NULL; + struct lttng_directory_handle *chunk_directory_handle = NULL; + + /* + * The session daemon will only provide a chunk directory file + * descriptor for local traces. + */ + if (is_local_trace) { + int chunk_dirfd; + + /* Acnowledge the reception of the command. */ + ret = consumer_send_status_msg(sock, + LTTCOMM_CONSUMERD_SUCCESS); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + + ret = lttcomm_recv_fds_unix_sock(sock, &chunk_dirfd, 1); + if (ret != sizeof(chunk_dirfd)) { + ERR("Failed to receive trace chunk directory file descriptor"); + goto error_fatal; + } + + DBG("Received trace chunk directory fd (%d)", + chunk_dirfd); + chunk_directory_handle = lttng_directory_handle_create_from_dirfd( + chunk_dirfd); + if (!chunk_directory_handle) { + ERR("Failed to initialize chunk directory handle from directory file descriptor"); + if (close(chunk_dirfd)) { + PERROR("Failed to close chunk directory file descriptor"); + } + goto error_fatal; + } + } + + ret_code = lttng_consumer_create_trace_chunk( + !is_local_trace ? &relayd_id : NULL, + msg.u.create_trace_chunk.session_id, + msg.u.create_trace_chunk.chunk_id, + (time_t) msg.u.create_trace_chunk + .creation_timestamp, + chunk_override_name, + msg.u.create_trace_chunk.credentials.is_set ? + &credentials : + NULL, + chunk_directory_handle); + lttng_directory_handle_put(chunk_directory_handle); + goto end_msg_sessiond; + } + case LTTNG_CONSUMER_CLOSE_TRACE_CHUNK: + { + enum lttng_trace_chunk_command_type close_command = + msg.u.close_trace_chunk.close_command.value; + const uint64_t relayd_id = + msg.u.close_trace_chunk.relayd_id.value; + struct lttcomm_consumer_close_trace_chunk_reply reply; + char path[LTTNG_PATH_MAX]; + + ret_code = lttng_consumer_close_trace_chunk( + msg.u.close_trace_chunk.relayd_id.is_set ? + &relayd_id : + NULL, + msg.u.close_trace_chunk.session_id, + msg.u.close_trace_chunk.chunk_id, + (time_t) msg.u.close_trace_chunk.close_timestamp, + msg.u.close_trace_chunk.close_command.is_set ? + &close_command : + NULL, path); + reply.ret_code = ret_code; + reply.path_length = strlen(path) + 1; + ret = lttcomm_send_unix_sock(sock, &reply, sizeof(reply)); + if (ret != sizeof(reply)) { + goto error_fatal; + } + ret = lttcomm_send_unix_sock(sock, path, reply.path_length); + if (ret != reply.path_length) { + goto error_fatal; + } + goto end_nosignal; + } + case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS: + { + const uint64_t relayd_id = + msg.u.trace_chunk_exists.relayd_id.value; + + ret_code = lttng_consumer_trace_chunk_exists( + msg.u.trace_chunk_exists.relayd_id.is_set ? + &relayd_id : NULL, + msg.u.trace_chunk_exists.session_id, + msg.u.trace_chunk_exists.chunk_id); + goto end_msg_sessiond; + } default: goto end_nosignal; } end_nosignal: - rcu_read_unlock(); - /* * Return 1 to indicate success since the 0 value can be a socket * shutdown during the recv() or send() call. */ - health_code_update(); - return 1; - + ret = 1; + goto end; error_fatal: - rcu_read_unlock(); /* This will issue a consumer stop. */ - return -1; + ret = -1; + goto end; +end_msg_sessiond: + /* + * The returned value here is not useful since either way we'll return 1 to + * the caller because the session daemon socket management is done + * elsewhere. Returning a negative code or 0 will shutdown the consumer. + */ + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + goto error_fatal; + } + ret = 1; +end: + health_code_update(); + rcu_read_unlock(); + return ret; } /* @@ -1050,63 +1366,82 @@ error_fatal: static int get_index_values(struct ctf_packet_index *index, int infd) { int ret; + uint64_t packet_size, content_size, timestamp_begin, timestamp_end, + events_discarded, stream_id, stream_instance_id, + packet_seq_num; - ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin); + ret = kernctl_get_timestamp_begin(infd, ×tamp_begin); if (ret < 0) { PERROR("kernctl_get_timestamp_begin"); goto error; } - index->timestamp_begin = htobe64(index->timestamp_begin); - ret = kernctl_get_timestamp_end(infd, &index->timestamp_end); + ret = kernctl_get_timestamp_end(infd, ×tamp_end); if (ret < 0) { PERROR("kernctl_get_timestamp_end"); goto error; } - index->timestamp_end = htobe64(index->timestamp_end); - ret = kernctl_get_events_discarded(infd, &index->events_discarded); + ret = kernctl_get_events_discarded(infd, &events_discarded); if (ret < 0) { PERROR("kernctl_get_events_discarded"); goto error; } - index->events_discarded = htobe64(index->events_discarded); - ret = kernctl_get_content_size(infd, &index->content_size); + ret = kernctl_get_content_size(infd, &content_size); if (ret < 0) { PERROR("kernctl_get_content_size"); goto error; } - index->content_size = htobe64(index->content_size); - ret = kernctl_get_packet_size(infd, &index->packet_size); + ret = kernctl_get_packet_size(infd, &packet_size); if (ret < 0) { PERROR("kernctl_get_packet_size"); goto error; } - index->packet_size = htobe64(index->packet_size); - ret = kernctl_get_stream_id(infd, &index->stream_id); + ret = kernctl_get_stream_id(infd, &stream_id); if (ret < 0) { PERROR("kernctl_get_stream_id"); goto error; } - index->stream_id = htobe64(index->stream_id); - ret = kernctl_get_instance_id(infd, &index->stream_instance_id); + ret = kernctl_get_instance_id(infd, &stream_instance_id); if (ret < 0) { - PERROR("kernctl_get_instance_id"); - goto error; + if (ret == -ENOTTY) { + /* Command not implemented by lttng-modules. */ + stream_instance_id = -1ULL; + } else { + PERROR("kernctl_get_instance_id"); + goto error; + } } - index->stream_instance_id = htobe64(index->stream_instance_id); - ret = kernctl_get_sequence_number(infd, &index->packet_seq_num); + ret = kernctl_get_sequence_number(infd, &packet_seq_num); if (ret < 0) { - PERROR("kernctl_get_sequence_number"); - goto error; + if (ret == -ENOTTY) { + /* Command not implemented by lttng-modules. */ + packet_seq_num = -1ULL; + ret = 0; + } else { + PERROR("kernctl_get_sequence_number"); + goto error; + } } index->packet_seq_num = htobe64(index->packet_seq_num); + *index = (typeof(*index)) { + .offset = index->offset, + .packet_size = htobe64(packet_size), + .content_size = htobe64(content_size), + .timestamp_begin = htobe64(timestamp_begin), + .timestamp_end = htobe64(timestamp_end), + .events_discarded = htobe64(events_discarded), + .stream_id = htobe64(stream_id), + .stream_instance_id = htobe64(stream_instance_id), + .packet_seq_num = htobe64(packet_seq_num), + }; + error: return ret; } @@ -1133,7 +1468,7 @@ int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata) ret = kernctl_snapshot(metadata->wait_fd); if (ret < 0) { - if (errno != EAGAIN) { + if (ret != -EAGAIN) { ERR("Sync metadata, taking kernel snapshot failed."); goto end; } @@ -1155,8 +1490,14 @@ int update_stream_stats(struct lttng_consumer_stream *stream) ret = kernctl_get_sequence_number(stream->wait_fd, &seq); if (ret < 0) { - PERROR("kernctl_get_sequence_number"); - goto end; + if (ret == -ENOTTY) { + /* Command not implemented by lttng-modules. */ + seq = -1ULL; + stream->sequence_number_unavailable = true; + } else { + PERROR("kernctl_get_sequence_number"); + goto end; + } } /* @@ -1186,8 +1527,8 @@ int update_stream_stats(struct lttng_consumer_stream *stream) } if (discarded < stream->last_discarded_events) { /* - * Overflow has occured. We assume only one wrap-around - * has occured. + * Overflow has occurred. We assume only one wrap-around + * has occurred. */ stream->chan->discarded_events += (1ULL << (CAA_BITS_PER_LONG - 1)) - stream->last_discarded_events + discarded; @@ -1215,6 +1556,14 @@ int metadata_stream_check_version(int infd, struct lttng_consumer_stream *stream ret = kernctl_get_metadata_version(infd, &cur_version); if (ret < 0) { + if (ret == -ENOTTY) { + /* + * LTTng-modules does not implement this + * command. + */ + ret = 0; + goto end; + } ERR("Failed to get the metadata version"); goto end; } @@ -1235,18 +1584,33 @@ end: /* * Consume data on a file descriptor and write it on a trace file. + * The stream and channel locks must be held by the caller. */ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, struct lttng_consumer_local_data *ctx) { 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; + 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); + 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) { @@ -1258,8 +1622,8 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, */ DBG("Reserving sub buffer failed (everything is normal, " "it is due to concurrency)"); - ret = -errno; - goto end; + ret = err; + goto error; } /* Get the full subbuffer size including padding */ @@ -1268,17 +1632,17 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, PERROR("Getting sub-buffer len failed."); err = kernctl_put_subbuf(infd); if (err != 0) { - if (errno == EFAULT) { + if (err == -EFAULT) { PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { + } else if (err == -EIO) { /* Should never happen with newer LTTng versions */ PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; - goto end; + ret = err; + goto error; } - ret = -errno; - goto end; + ret = err; + goto error; } if (!stream->metadata_flag) { @@ -1286,26 +1650,48 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, if (ret < 0) { err = kernctl_put_subbuf(infd); if (err != 0) { - if (errno == EFAULT) { + if (err == -EFAULT) { PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { + } else if (err == -EIO) { /* Should never happen with newer LTTng versions */ PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; - goto end; + ret = err; + goto error; } - goto end; + goto error; } ret = update_stream_stats(stream); if (ret < 0) { - goto end; + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (err == -EFAULT) { + PERROR("Error in unreserving sub buffer\n"); + } else if (err == -EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = err; + goto error; + } + goto error; } } else { write_index = 0; ret = metadata_stream_check_version(infd, stream); if (ret < 0) { - goto end; + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (err == -EFAULT) { + PERROR("Error in unreserving sub buffer\n"); + } else if (err == -EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = err; + goto error; + } + goto error; } } @@ -1337,23 +1723,32 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, } break; case CONSUMER_CHANNEL_MMAP: + { + const char *subbuf_addr; + struct lttng_buffer_view subbuf_view; + /* Get subbuffer size without padding */ err = kernctl_get_subbuf_size(infd, &subbuf_size); if (err != 0) { PERROR("Getting sub-buffer len failed."); err = kernctl_put_subbuf(infd); if (err != 0) { - if (errno == EFAULT) { + if (err == -EFAULT) { PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { + } else if (err == -EIO) { /* Should never happen with newer LTTng versions */ PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; - goto end; + ret = err; + goto error; } - ret = -errno; - goto end; + ret = err; + goto error; + } + + ret = get_current_subbuf_addr(stream, &subbuf_addr); + if (ret) { + goto error_put_subbuf; } /* Make sure the tracer is not gone mad on us! */ @@ -1361,13 +1756,15 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, padding = len - subbuf_size; + subbuf_view = lttng_buffer_view_init(subbuf_addr, 0, len); + /* write the subbuffer to the tracefile */ - ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, - padding, &index); + ret = lttng_consumer_on_read_subbuffer_mmap( + ctx, stream, &subbuf_view, padding, &index); /* - * The mmap operation should write subbuf_size amount of data when - * network streaming or the full padding (len) size when we are _not_ - * streaming. + * The mmap operation should write subbuf_size amount of data + * when network streaming or the full padding (len) size when we + * are _not_ streaming. */ if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) || (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) { @@ -1382,26 +1779,27 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, write_index = 0; } break; + } default: ERR("Unknown output method"); ret = -EPERM; } - +error_put_subbuf: err = kernctl_put_next_subbuf(infd); if (err != 0) { - if (errno == EFAULT) { + if (err == -EFAULT) { PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { + } else if (err == -EIO) { /* Should never happen with newer LTTng versions */ PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; - goto end; + ret = err; + goto error; } /* Write index if needed. */ if (!write_index) { - goto end; + goto rotate; } if (stream->chan->live_timer_interval && !stream->metadata_flag) { @@ -1425,16 +1823,35 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, 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); + 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; } @@ -1445,29 +1862,15 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) assert(stream); /* - * Don't create anything if this is set for streaming or should not be - * monitored. + * Don't create anything if this is set for streaming or if there is + * no current trace chunk on the parent channel. */ - if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) { - 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) { + if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor && + stream->chan->trace_chunk) { + ret = consumer_stream_create_output_files(stream, true); + if (ret) { goto error; } - stream->out_fd = ret; - stream->tracefile_size_current = 0; - - if (!stream->metadata_flag) { - ret = index_create_file(stream->chan->pathname, - stream->name, stream->uid, stream->gid, - stream->chan->tracefile_size, - stream->tracefile_count_current); - if (ret < 0) { - goto error; - } - stream->index_fd = ret; - } } if (stream->output == LTTNG_EVENT_MMAP) { @@ -1477,7 +1880,6 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len); if (ret != 0) { PERROR("kernctl_get_mmap_len"); - ret = -errno; goto error_close_fd; } stream->mmap_len = (size_t) mmap_len;