X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=ecacadfd5bae7dc2b423bdae3b5023cb835d797b;hp=87ade12fa33a32e71dd52cc685a6f3acb02e967b;hb=0c5b37182381d8350ead74f5c10f272c746e3277;hpb=ec1c466fca0cb257282dc505ddb45450610531f3 diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 87ade12fa..ecacadfd5 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -1,20 +1,10 @@ /* - * Copyright (C) 2011 - Julien Desfossez - * Mathieu Desnoyers - * Copyright (C) 2017 - Jérémie Galarneau + * 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. */ #define _LGPL_SOURCE @@ -29,6 +19,7 @@ #include #include #include +#include #include #include @@ -43,6 +34,10 @@ #include #include #include +#include +#include +#include +#include #include "kernel-consumer.h" @@ -122,33 +117,49 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream, 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; - 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; } @@ -163,6 +174,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. @@ -176,20 +200,13 @@ 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); } ret = kernctl_buffer_flush_empty(stream->wait_fd); @@ -227,25 +244,17 @@ 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"); - 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; + 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); @@ -272,8 +281,16 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, goto error_put_subbuf; } - read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len, - padded_len - len, NULL); + 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( + stream, &subbuf_view, + padded_len - len); /* * We write the padded len in local tracefiles but the data len * when using a relay. Display the error but continue processing @@ -312,6 +329,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); } @@ -333,15 +352,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); @@ -351,16 +373,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; @@ -369,28 +388,26 @@ 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 { health_code_update(); - ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx, NULL); + ret_read = lttng_consumer_read_subbuffer(metadata_stream, ctx, true); 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; @@ -411,15 +428,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; } @@ -432,19 +451,24 @@ error: int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll) { - ssize_t ret; + int ret_func; enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS; struct lttcomm_consumer_msg msg; health_code_update(); - ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg)); - if (ret != sizeof(msg)) { - if (ret > 0) { - lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD); - ret = -1; + { + ssize_t ret_recv; + + ret_recv = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg)); + if (ret_recv != sizeof(msg)) { + if (ret_recv > 0) { + lttng_consumer_send_error(ctx, + LTTCOMM_CONSUMERD_ERROR_RECV_CMD); + ret_recv = -1; + } + return ret_recv; } - return ret; } health_code_update(); @@ -470,13 +494,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, case LTTNG_CONSUMER_ADD_CHANNEL: { struct lttng_consumer_channel *new_channel; - int ret_recv; + int ret_send_status, ret_add_channel; + const uint64_t chunk_id = msg.u.channel.chunk_id.value; health_code_update(); /* First send a status message before receiving the fds. */ - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { /* Somehow, the session daemon is not responding anymore. */ goto error_fatal; } @@ -485,13 +510,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); @@ -524,16 +553,20 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); if (ctx->on_recv_channel != NULL) { - ret_recv = ctx->on_recv_channel(new_channel); - if (ret_recv == 0) { - ret = consumer_add_channel(new_channel, ctx); - } else if (ret_recv < 0) { + int ret_recv_channel = + ctx->on_recv_channel(new_channel); + if (ret_recv_channel == 0) { + ret_add_channel = consumer_add_channel( + new_channel, ctx); + } else if (ret_recv_channel < 0) { goto end_nosignal; } } else { - ret = consumer_add_channel(new_channel, ctx); + ret_add_channel = + consumer_add_channel(new_channel, ctx); } - if (msg.u.channel.type == CONSUMER_CHANNEL_TYPE_DATA && !ret) { + if (msg.u.channel.type == CONSUMER_CHANNEL_TYPE_DATA && + !ret_add_channel) { int monitor_start_ret; DBG("Consumer starting monitor timer"); @@ -546,15 +579,15 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ERR("Starting channel monitoring timer failed"); goto end_nosignal; } - } health_code_update(); /* If we received an error in add_channel, we need to report it. */ - if (ret < 0) { - ret = consumer_send_status_msg(sock, ret); - if (ret < 0) { + if (ret_add_channel < 0) { + ret_send_status = consumer_send_status_msg( + sock, ret_add_channel); + if (ret_send_status < 0) { goto error_fatal; } goto end_nosignal; @@ -569,6 +602,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, struct lttng_consumer_stream *new_stream; struct lttng_consumer_channel *channel; int alloc_ret = 0; + int ret_send_status, ret_poll, ret_get_max_subbuf_size; + ssize_t ret_pipe_write, ret_recv; /* * Get stream's channel reference. Needed when adding the stream to the @@ -587,35 +622,35 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); /* First send a status message before receiving the fds. */ - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 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 */ health_poll_entry(); - ret = lttng_consumer_poll_socket(consumer_sockpoll); + ret_poll = lttng_consumer_poll_socket(consumer_sockpoll); health_poll_exit(); - if (ret) { - goto error_fatal; + if (ret_poll) { + goto error_add_stream_fatal; } health_code_update(); /* Get stream file descriptor from socket */ - ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1); - if (ret != sizeof(fd)) { + ret_recv = lttcomm_recv_fds_unix_sock(sock, &fd, 1); + if (ret_recv != sizeof(fd)) { lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD); - rcu_read_unlock(); - return ret; + ret_func = ret_recv; + goto end; } health_code_update(); @@ -625,27 +660,27 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, * above recv() failed, the session daemon is notified through the * error socket and the teardown is eventually done. */ - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 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_stream_create( + 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, - channel->monitor, - msg.u.stream.trace_archive_id); + channel->monitor); if (new_stream == NULL) { switch (alloc_ret) { case -ENOMEM: @@ -654,28 +689,21 @@ 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_get_max_subbuf_size = kernctl_get_max_subbuf_size( + new_stream->wait_fd, &new_stream->max_sb_size); + if (ret_get_max_subbuf_size < 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; - } - break; - case CONSUMER_CHANNEL_MMAP: - new_stream->output = LTTNG_EVENT_MMAP; - break; - default: - ERR("Stream output unknown %d", channel->output); - goto end_nosignal; - } /* * We've just assigned the channel to the stream so increment the @@ -699,14 +727,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) { + int ret_recv_stream = ctx->on_recv_stream(new_stream); + if (ret_recv_stream < 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) { @@ -719,16 +749,22 @@ 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. */ if (new_stream->net_seq_idx != (uint64_t) -1ULL) { - ret = consumer_send_relayd_stream(new_stream, - new_stream->chan->pathname); - if (ret < 0) { + int ret_send_relayd_stream; + + ret_send_relayd_stream = consumer_send_relayd_stream( + new_stream, new_stream->chan->pathname); + if (ret_send_relayd_stream < 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; } /* @@ -737,13 +773,20 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, * 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) { - goto end_nosignal; + int ret_send_relayd_streams_sent; + + ret_send_relayd_streams_sent = + consumer_send_relayd_streams_sent( + new_stream->net_seq_idx); + if (ret_send_relayd_streams_sent < 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) { @@ -759,8 +802,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); - ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream)); - if (ret < 0) { + ret_pipe_write = lttng_pipe_write( + stream_pipe, &new_stream, sizeof(new_stream)); + if (ret_pipe_write < 0) { ERR("Consumer write %s stream to pipe %d", new_stream->metadata_flag ? "metadata" : "data", lttng_pipe_get_writefd(stream_pipe)); @@ -769,16 +813,22 @@ 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) %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: { struct lttng_consumer_channel *channel; + int ret_send_status; /* * Get stream's channel reference. Needed when adding the stream to the @@ -800,10 +850,11 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, /* * Send status code to session daemon. */ - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 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(); @@ -813,20 +864,25 @@ 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(); /* Send stream to relayd if the stream has an ID. */ if (msg.u.sent_streams.net_seq_idx != (uint64_t) -1ULL) { - ret = consumer_send_relayd_streams_sent( + int ret_send_relay_streams; + + ret_send_relay_streams = consumer_send_relayd_streams_sent( msg.u.sent_streams.net_seq_idx); - if (ret < 0) { - goto end_nosignal; + if (ret_send_relay_streams < 0) { + 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: { @@ -837,6 +893,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, { uint64_t index = msg.u.destroy_relayd.net_seq_idx; struct consumer_relayd_sock_pair *relayd; + int ret_send_status; DBG("Kernel consumer destroying relayd %" PRIu64, index); @@ -863,8 +920,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { /* Somehow, the session daemon is not responding anymore. */ goto error_fatal; } @@ -873,18 +930,20 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } case LTTNG_CONSUMER_DATA_PENDING: { - int32_t ret; + int32_t ret_data_pending; uint64_t id = msg.u.data_pending.session_id; + ssize_t ret_send; DBG("Kernel consumer data pending command for id %" PRIu64, id); - ret = consumer_data_pending(id); + ret_data_pending = consumer_data_pending(id); health_code_update(); /* Send back returned value to session daemon */ - ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); - if (ret < 0) { + ret_send = lttcomm_send_unix_sock(sock, &ret_data_pending, + sizeof(ret_data_pending)); + if (ret_send < 0) { PERROR("send data pending ret code"); goto error_fatal; } @@ -897,30 +956,49 @@ 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; + int ret_send_status; + + 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) { + int ret_snapshot; + + ret_snapshot = lttng_kconsumer_snapshot_metadata( + channel, key, + msg.u.snapshot_channel.pathname, + msg.u.snapshot_channel.relayd_id, + ctx); + if (ret_snapshot < 0) { + ERR("Snapshot metadata failed"); + ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED; + } + } else { + int ret_snapshot; + + ret_snapshot = 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_snapshot < 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); - if (ret < 0) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { /* Somehow, the session daemon is not responding anymore. */ goto end_nosignal; } @@ -930,6 +1008,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, { uint64_t key = msg.u.destroy_channel.key; struct lttng_consumer_channel *channel; + int ret_send_status; channel = consumer_find_channel(key); if (!channel) { @@ -939,17 +1018,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 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; } /* @@ -965,7 +1044,7 @@ 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: @@ -1033,39 +1112,43 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE: { int channel_monitor_pipe; + int ret_send_status, ret_set_channel_monitor_pipe; + ssize_t ret_recv; ret_code = LTTCOMM_CONSUMERD_SUCCESS; /* Successfully received the command's type. */ - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { goto error_fatal; } - ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe, - 1); - if (ret != sizeof(channel_monitor_pipe)) { + ret_recv = lttcomm_recv_fds_unix_sock( + sock, &channel_monitor_pipe, 1); + if (ret_recv != 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) { + ret_set_channel_monitor_pipe = + consumer_timer_thread_set_channel_monitor_pipe( + channel_monitor_pipe); + if (!ret_set_channel_monitor_pipe) { int flags; + int ret_fcntl; ret_code = LTTCOMM_CONSUMERD_SUCCESS; /* Set the pipe as non-blocking. */ - ret = fcntl(channel_monitor_pipe, F_GETFL, 0); - if (ret == -1) { + ret_fcntl = fcntl(channel_monitor_pipe, F_GETFL, 0); + if (ret_fcntl == -1) { PERROR("fcntl get flags of the channel monitoring pipe"); goto error_fatal; } - flags = ret; + flags = ret_fcntl; - ret = fcntl(channel_monitor_pipe, F_SETFL, + ret_fcntl = fcntl(channel_monitor_pipe, F_SETFL, flags | O_NONBLOCK); - if (ret == -1) { + if (ret_fcntl == -1) { PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe"); goto error_fatal; } @@ -1073,309 +1156,349 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } else { ret_code = LTTCOMM_CONSUMERD_ALREADY_SET; } - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { goto error_fatal; } break; } - case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE: + case LTTNG_CONSUMER_ROTATE_CHANNEL: { - int channel_rotate_pipe; - int flags; + struct lttng_consumer_channel *channel; + uint64_t key = msg.u.rotate_channel.key; + int ret_send_status; - 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; - } + DBG("Consumer rotate channel %" PRIu64, key); - ret = lttcomm_recv_fds_unix_sock(sock, &channel_rotate_pipe, 1); - if (ret != (ssize_t) sizeof(channel_rotate_pipe)) { - ERR("Failed to receive channel rotate pipe"); - goto error_fatal; - } + 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. + */ + int ret_rotate_channel; + + ret_rotate_channel = lttng_consumer_rotate_channel( + channel, key, + msg.u.rotate_channel.relayd_id, + msg.u.rotate_channel.metadata, ctx); + if (ret_rotate_channel < 0) { + ERR("Rotate channel failed"); + ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL; + } - 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; + health_code_update(); } - 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; + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto error_rotate_channel; } - 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; + if (channel) { + /* Rotate the streams that are ready right now. */ + int ret_rotate; + + ret_rotate = lttng_consumer_rotate_ready_streams( + channel, key, ctx); + if (ret_rotate < 0) { + ERR("Rotate ready streams failed"); + } } break; +error_rotate_channel: + goto end_nosignal; } - case LTTNG_CONSUMER_ROTATE_CHANNEL: + case LTTNG_CONSUMER_CLEAR_CHANNEL: { - DBG("Consumer rotate channel %" PRIu64, msg.u.rotate_channel.key); + struct lttng_consumer_channel *channel; + uint64_t key = msg.u.clear_channel.key; + int ret_send_status; - /* - * Sample the rotate position of all the streams in this channel. - */ - ret = lttng_consumer_rotate_channel(msg.u.rotate_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"); + channel = consumer_find_channel(key); + if (!channel) { + DBG("Channel %" PRIu64 " not found", key); ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; - } + } else { + int ret_clear_channel; - health_code_update(); + ret_clear_channel = + lttng_consumer_clear_channel(channel); + if (ret_clear_channel) { + ERR("Clear channel failed"); + ret_code = ret_clear_channel; + } - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { - /* Somehow, the session daemon is not responding anymore. */ - goto end_nosignal; + health_code_update(); } - /* Rotate the streams that are ready right now. */ - ret = lttng_consumer_rotate_ready_streams( - msg.u.rotate_channel.key, ctx); - if (ret < 0) { - ERR("Rotate ready streams failed"); + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; } break; } - case LTTNG_CONSUMER_ROTATE_RENAME: + case LTTNG_CONSUMER_INIT: { - 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_CHAN_NOT_FOUND; - } + int ret_send_status; + 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) { + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { /* Somehow, the session daemon is not responding anymore. */ goto end_nosignal; } break; } - case LTTNG_CONSUMER_ROTATE_PENDING_RELAY: + case LTTNG_CONSUMER_CREATE_TRACE_CHUNK: { - int pending; - uint32_t pending_reply; - - DBG("Consumer rotate pending on relay for session %" PRIu64, - msg.u.rotate_pending_relay.session_id); - pending = lttng_consumer_rotate_pending_relay( - msg.u.rotate_pending_relay.session_id, - msg.u.rotate_pending_relay.relayd_id, - msg.u.rotate_pending_relay.chunk_id); - if (pending < 0) { - ERR("Rotate pending relay failed"); - ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; - } else { - pending_reply = !!pending; - } + const struct lttng_credentials credentials = { + .uid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.create_trace_chunk.credentials.value.uid), + .gid = LTTNG_OPTIONAL_INIT_VALUE(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; - health_code_update(); + /* + * The session daemon will only provide a chunk directory file + * descriptor for local traces. + */ + if (is_local_trace) { + int chunk_dirfd; + int ret_send_status; + ssize_t ret_recv; + + /* Acnowledge the reception of the command. */ + ret_send_status = consumer_send_status_msg( + sock, LTTCOMM_CONSUMERD_SUCCESS); + if (ret_send_status < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { - /* Somehow, the session daemon is not responding anymore. */ - goto end_nosignal; - } + ret_recv = lttcomm_recv_fds_unix_sock( + sock, &chunk_dirfd, 1); + if (ret_recv != sizeof(chunk_dirfd)) { + ERR("Failed to receive trace chunk directory file descriptor"); + goto error_fatal; + } - if (pending < 0) { - /* - * An error occured while running the command; - * don't send the 'pending' flag as the sessiond - * will not read it. - */ - break; + 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; + } } - /* Send back returned value to session daemon */ - ret = lttcomm_send_unix_sock(sock, &pending_reply, - sizeof(pending_reply)); - if (ret < 0) { - PERROR("send data pending ret code"); + 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]; + ssize_t ret_send; + + 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_send = lttcomm_send_unix_sock(sock, &reply, sizeof(reply)); + if (ret_send != sizeof(reply)) { goto error_fatal; } - break; + ret_send = lttcomm_send_unix_sock( + sock, path, reply.path_length); + if (ret_send != reply.path_length) { + goto error_fatal; + } + goto end_nosignal; } - case LTTNG_CONSUMER_MKDIR: + case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS: { - 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"); + 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; + } + case LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS: + { + const uint64_t key = msg.u.open_channel_packets.key; + struct lttng_consumer_channel *channel = + consumer_find_channel(key); + + if (channel) { + pthread_mutex_lock(&channel->lock); + ret_code = lttng_consumer_open_channel_packets(channel); + pthread_mutex_unlock(&channel->lock); + } else { + WARN("Channel %" PRIu64 " not found", key); 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; - } - break; + 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_func = 1; + goto end; error_fatal: - rcu_read_unlock(); /* This will issue a consumer stop. */ - return -1; + ret_func = -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. + */ + { + int ret_send_status; + + ret_send_status = consumer_send_status_msg(sock, ret_code); + if (ret_send_status < 0) { + goto error_fatal; + } + } + + ret_func = 1; + +end: + health_code_update(); + rcu_read_unlock(); + return ret_func; } /* - * Populate index values of a kernel stream. Values are set in big endian order. + * Sync metadata meaning request them to the session daemon and snapshot to the + * metadata thread can consumer them. * - * Return 0 on success or else a negative value. + * Metadata stream lock MUST be acquired. */ -static int get_index_values(struct ctf_packet_index *index, int infd) +enum sync_metadata_status lttng_kconsumer_sync_metadata( + struct lttng_consumer_stream *metadata) { int ret; + enum sync_metadata_status status; - ret = kernctl_get_timestamp_begin(infd, &index->timestamp_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); - if (ret < 0) { - PERROR("kernctl_get_timestamp_end"); - goto error; - } - index->timestamp_end = htobe64(index->timestamp_end); + assert(metadata); - ret = kernctl_get_events_discarded(infd, &index->events_discarded); + ret = kernctl_buffer_flush(metadata->wait_fd); if (ret < 0) { - PERROR("kernctl_get_events_discarded"); - goto error; + ERR("Failed to flush kernel stream"); + status = SYNC_METADATA_STATUS_ERROR; + goto end; } - index->events_discarded = htobe64(index->events_discarded); - ret = kernctl_get_content_size(infd, &index->content_size); + ret = kernctl_snapshot(metadata->wait_fd); if (ret < 0) { - PERROR("kernctl_get_content_size"); - goto error; + if (errno == EAGAIN) { + /* No new metadata, exit. */ + DBG("Sync metadata, no new kernel metadata"); + status = SYNC_METADATA_STATUS_NO_DATA; + } else { + ERR("Sync metadata, taking kernel snapshot failed."); + status = SYNC_METADATA_STATUS_ERROR; + } + } else { + status = SYNC_METADATA_STATUS_NEW_DATA; } - index->content_size = htobe64(index->content_size); - ret = kernctl_get_packet_size(infd, &index->packet_size); - if (ret < 0) { - PERROR("kernctl_get_packet_size"); - goto error; - } - index->packet_size = htobe64(index->packet_size); +end: + return status; +} - ret = kernctl_get_stream_id(infd, &index->stream_id); - if (ret < 0) { - PERROR("kernctl_get_stream_id"); - goto error; - } - index->stream_id = htobe64(index->stream_id); +static +int extract_common_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) +{ + int ret; - ret = kernctl_get_instance_id(infd, &index->stream_instance_id); - if (ret < 0) { - if (ret == -ENOTTY) { - /* Command not implemented by lttng-modules. */ - index->stream_instance_id = -1ULL; - } else { - PERROR("kernctl_get_instance_id"); - goto error; - } + ret = kernctl_get_subbuf_size( + stream->wait_fd, &subbuf->info.data.subbuf_size); + if (ret) { + goto end; } - index->stream_instance_id = htobe64(index->stream_instance_id); - ret = kernctl_get_sequence_number(infd, &index->packet_seq_num); - if (ret < 0) { - if (ret == -ENOTTY) { - /* Command not implemented by lttng-modules. */ - index->packet_seq_num = -1ULL; - ret = 0; - } else { - PERROR("kernctl_get_sequence_number"); - goto error; - } + ret = kernctl_get_padded_subbuf_size( + stream->wait_fd, &subbuf->info.data.padded_subbuf_size); + if (ret) { + goto end; } - index->packet_seq_num = htobe64(index->packet_seq_num); -error: +end: return ret; } -/* - * Sync metadata meaning request them to the session daemon and snapshot to the - * metadata thread can consumer them. - * - * Metadata stream lock MUST be acquired. - * - * Return 0 if new metadatda is available, EAGAIN if the metadata stream - * is empty or a negative value on error. - */ -int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata) + +static +int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) { int ret; - assert(metadata); - - ret = kernctl_buffer_flush(metadata->wait_fd); - if (ret < 0) { - ERR("Failed to flush kernel stream"); + ret = extract_common_subbuffer_info(stream, subbuf); + if (ret) { goto end; } - ret = kernctl_snapshot(metadata->wait_fd); - if (ret < 0) { - if (ret != -EAGAIN) { - ERR("Sync metadata, taking kernel snapshot failed."); - goto end; - } - DBG("Sync metadata, no new kernel metadata"); - /* No new metadata, exit. */ - ret = ENODATA; + ret = kernctl_get_metadata_version( + stream->wait_fd, &subbuf->info.metadata.version); + if (ret) { goto end; } @@ -1384,361 +1507,258 @@ end: } static -int update_stream_stats(struct lttng_consumer_stream *stream) +int extract_data_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) { int ret; - uint64_t seq, discarded; - ret = kernctl_get_sequence_number(stream->wait_fd, &seq); + ret = extract_common_subbuffer_info(stream, subbuf); + if (ret) { + goto end; + } + + ret = kernctl_get_packet_size( + stream->wait_fd, &subbuf->info.data.packet_size); if (ret < 0) { - if (ret == -ENOTTY) { - /* Command not implemented by lttng-modules. */ - seq = -1ULL; - } else { - PERROR("kernctl_get_sequence_number"); - goto end; - } + PERROR("Failed to get sub-buffer packet size"); + goto end; } - /* - * Start the sequence when we extract the first packet in case we don't - * start at 0 (for example if a consumer is not connected to the - * session immediately after the beginning). - */ - if (stream->last_sequence_number == -1ULL) { - stream->last_sequence_number = seq; - } else if (seq > stream->last_sequence_number) { - stream->chan->lost_packets += seq - - stream->last_sequence_number - 1; - } else { - /* seq <= last_sequence_number */ - ERR("Sequence number inconsistent : prev = %" PRIu64 - ", current = %" PRIu64, - stream->last_sequence_number, seq); - ret = -1; + ret = kernctl_get_content_size( + stream->wait_fd, &subbuf->info.data.content_size); + if (ret < 0) { + PERROR("Failed to get sub-buffer content size"); goto end; } - stream->last_sequence_number = seq; - ret = kernctl_get_events_discarded(stream->wait_fd, &discarded); + ret = kernctl_get_timestamp_begin( + stream->wait_fd, &subbuf->info.data.timestamp_begin); if (ret < 0) { - PERROR("kernctl_get_events_discarded"); + PERROR("Failed to get sub-buffer begin timestamp"); goto end; } - if (discarded < stream->last_discarded_events) { - /* - * 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; + + ret = kernctl_get_timestamp_end( + stream->wait_fd, &subbuf->info.data.timestamp_end); + if (ret < 0) { + PERROR("Failed to get sub-buffer end timestamp"); + goto end; + } + + ret = kernctl_get_events_discarded( + stream->wait_fd, &subbuf->info.data.events_discarded); + if (ret) { + PERROR("Failed to get sub-buffer events discarded count"); + goto end; + } + + ret = kernctl_get_sequence_number(stream->wait_fd, + &subbuf->info.data.sequence_number.value); + if (ret) { + /* May not be supported by older LTTng-modules. */ + if (ret != -ENOTTY) { + PERROR("Failed to get sub-buffer sequence number"); + goto end; + } } else { - stream->chan->discarded_events += discarded - - stream->last_discarded_events; + subbuf->info.data.sequence_number.is_set = true; } - stream->last_discarded_events = discarded; - ret = 0; + ret = kernctl_get_stream_id( + stream->wait_fd, &subbuf->info.data.stream_id); + if (ret < 0) { + PERROR("Failed to get stream id"); + goto end; + } + + ret = kernctl_get_instance_id(stream->wait_fd, + &subbuf->info.data.stream_instance_id.value); + if (ret) { + /* May not be supported by older LTTng-modules. */ + if (ret != -ENOTTY) { + PERROR("Failed to get stream instance id"); + goto end; + } + } else { + subbuf->info.data.stream_instance_id.is_set = true; + } end: return ret; } -/* - * Check if the local version of the metadata stream matches with the version - * of the metadata stream in the kernel. If it was updated, set the reset flag - * on the stream. - */ static -int metadata_stream_check_version(int infd, struct lttng_consumer_stream *stream) +int get_subbuffer_common(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) { int ret; - uint64_t cur_version; - 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"); + ret = kernctl_get_next_subbuf(stream->wait_fd); + if (ret) { goto end; } - if (stream->metadata_version == cur_version) { - ret = 0; + ret = stream->read_subbuffer_ops.extract_subbuffer_info( + stream, subbuffer); +end: + return ret; +} + +static +int get_next_subbuffer_splice(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + int ret; + + ret = get_subbuffer_common(stream, subbuffer); + if (ret) { goto end; } - DBG("New metadata version detected"); - stream->metadata_version = cur_version; - stream->reset_metadata_flag = 1; - ret = 0; - + subbuffer->buffer.fd = stream->wait_fd; end: return ret; } -/* - * 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, bool *rotated) +static +int get_next_subbuffer_mmap(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) { - unsigned long len, subbuf_size, padding; - 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); + int ret; + const char *addr; - /* - * 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; - } + ret = get_subbuffer_common(stream, subbuffer); + if (ret) { + goto end; } - /* Get the next subbuffer */ - err = kernctl_get_next_subbuf(infd); - if (err != 0) { - /* - * This is a debug message even for single-threaded consumer, - * because poll() have more relaxed criterions than get subbuf, - * so get_subbuf may fail for short race windows where poll() - * would issue wakeups. - */ - DBG("Reserving sub buffer failed (everything is normal, " - "it is due to concurrency)"); - ret = err; - goto error; + ret = get_current_subbuf_addr(stream, &addr); + if (ret) { + goto end; } - /* Get the full subbuffer size including padding */ - err = kernctl_get_padded_subbuf_size(infd, &len); - if (err != 0) { - PERROR("Getting sub-buffer len failed."); - 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; - } - ret = err; - goto error; - } + subbuffer->buffer.buffer = lttng_buffer_view_init( + addr, 0, subbuffer->info.data.padded_subbuf_size); +end: + return ret; +} - if (!stream->metadata_flag) { - ret = get_index_values(&index, infd); - if (ret < 0) { - 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; - } - ret = update_stream_stats(stream); - if (ret < 0) { - 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) { - 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; - } +static +int get_next_subbuffer_metadata_check(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + int ret; + const char *addr; + bool coherent; + + ret = kernctl_get_next_subbuf_metadata_check(stream->wait_fd, + &coherent); + if (ret) { + goto end; } - switch (stream->chan->output) { - case CONSUMER_CHANNEL_SPLICE: - /* - * XXX: The lttng-modules splice "actor" does not handle copying - * partial pages hence only using the subbuffer size without the - * padding makes the splice fail. - */ - subbuf_size = len; - padding = 0; + ret = stream->read_subbuffer_ops.extract_subbuffer_info( + stream, subbuffer); + if (ret) { + goto end; + } - /* splice the subbuffer to the tracefile */ - ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size, - padding, &index); - /* - * XXX: Splice does not support network streaming so the return value - * is simply checked against subbuf_size and not like the mmap() op. - */ - if (ret != subbuf_size) { - /* - * display the error but continue processing to try - * to release the subbuffer - */ - ERR("Error splicing to tracefile (ret: %zd != len: %lu)", - ret, subbuf_size); - write_index = 0; - } - break; - case CONSUMER_CHANNEL_MMAP: - /* 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 (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; - } - ret = err; - goto error; - } + LTTNG_OPTIONAL_SET(&subbuffer->info.metadata.coherent, coherent); - /* Make sure the tracer is not gone mad on us! */ - assert(len >= subbuf_size); + ret = get_current_subbuf_addr(stream, &addr); + if (ret) { + goto end; + } - padding = len - subbuf_size; + subbuffer->buffer.buffer = lttng_buffer_view_init( + addr, 0, subbuffer->info.data.padded_subbuf_size); + DBG("Got metadata packet with padded_subbuf_size = %lu, coherent = %s", + subbuffer->info.metadata.padded_subbuf_size, + coherent ? "true" : "false"); +end: + return ret; +} - /* write the subbuffer to the tracefile */ - ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, - 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. - */ - if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) || - (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) { - /* - * Display the error but continue processing to try to release the - * subbuffer. This is a DBG statement since this is possible to - * happen without being a critical error. - */ - DBG("Error writing to tracefile " - "(ret: %zd != len: %lu != subbuf_size: %lu)", - ret, len, subbuf_size); - write_index = 0; - } - break; - default: - ERR("Unknown output method"); - ret = -EPERM; - } +static +int put_next_subbuffer(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + const int ret = kernctl_put_next_subbuf(stream->wait_fd); - err = kernctl_put_next_subbuf(infd); - if (err != 0) { - if (err == -EFAULT) { - PERROR("Error in unreserving sub buffer\n"); - } else if (err == -EIO) { + if (ret) { + if (ret == -EFAULT) { + PERROR("Error in unreserving sub buffer"); + } else if (ret == -EIO) { /* Should never happen with newer LTTng versions */ - PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted"); } - ret = err; - goto error; } - /* Write index if needed. */ - if (!write_index) { - goto rotate; - } + return ret; +} - if (stream->chan->live_timer_interval && !stream->metadata_flag) { - /* - * In live, block until all the metadata is sent. - */ - pthread_mutex_lock(&stream->metadata_timer_lock); - assert(!stream->missed_metadata_flush); - stream->waiting_on_metadata = true; - pthread_mutex_unlock(&stream->metadata_timer_lock); - - err = consumer_stream_sync_metadata(ctx, stream->session_id); - - pthread_mutex_lock(&stream->metadata_timer_lock); - stream->waiting_on_metadata = false; - if (stream->missed_metadata_flush) { - stream->missed_metadata_flush = false; - pthread_mutex_unlock(&stream->metadata_timer_lock); - (void) consumer_flush_kernel_index(stream); +static +bool is_get_next_check_metadata_available(int tracer_fd) +{ + return kernctl_get_next_subbuf_metadata_check(tracer_fd, NULL) != + -ENOTTY; +} + +static +int lttng_kconsumer_set_stream_ops( + struct lttng_consumer_stream *stream) +{ + int ret = 0; + + if (stream->metadata_flag && stream->chan->is_live) { + DBG("Attempting to enable metadata bucketization for live consumers"); + if (is_get_next_check_metadata_available(stream->wait_fd)) { + DBG("Kernel tracer supports get_next_subbuffer_metadata_check, metadata will be accumulated until a coherent state is reached"); + stream->read_subbuffer_ops.get_next_subbuffer = + get_next_subbuffer_metadata_check; + ret = consumer_stream_enable_metadata_bucketization( + stream); + if (ret) { + goto end; + } } else { - pthread_mutex_unlock(&stream->metadata_timer_lock); - } - if (err < 0) { - goto error; + /* + * The kernel tracer version is too old to indicate + * when the metadata stream has reached a "coherent" + * (parseable) point. + * + * This means that a live viewer may see an incoherent + * sequence of metadata and fail to parse it. + */ + WARN("Kernel tracer does not support get_next_subbuffer_metadata_check which may cause live clients to fail to parse the metadata stream"); + metadata_bucket_destroy(stream->metadata_bucket); + stream->metadata_bucket = NULL; } } - err = consumer_stream_write_index(stream, &index); - if (err < 0) { - goto error; + if (!stream->read_subbuffer_ops.get_next_subbuffer) { + if (stream->chan->output == CONSUMER_CHANNEL_MMAP) { + stream->read_subbuffer_ops.get_next_subbuffer = + get_next_subbuffer_mmap; + } else { + stream->read_subbuffer_ops.get_next_subbuffer = + get_next_subbuffer_splice; + } } -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; + if (stream->metadata_flag) { + stream->read_subbuffer_ops.extract_subbuffer_info = + extract_metadata_subbuffer_info; + } else { + stream->read_subbuffer_ops.extract_subbuffer_info = + extract_data_subbuffer_info; + if (stream->chan->is_live) { + stream->read_subbuffer_ops.send_live_beacon = + consumer_flush_kernel_index; } - } else if (rotation_ret < 0) { - ERR("Checking if stream is ready to rotate"); - ret = -1; - goto error; } -error: + stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer; +end: return ret; } @@ -1749,33 +1769,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) { - struct lttng_index_file *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; - } - assert(!stream->index_file); - stream->index_file = index_file; - } } if (stream->output == LTTNG_EVENT_MMAP) { @@ -1798,6 +1800,11 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) } } + ret = lttng_kconsumer_set_stream_ops(stream); + if (ret) { + goto error_close_fd; + } + /* we return 0 to let the library handle the FD internally */ return 0;