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=29a27134f0e372c8c8295114048537a07dcc6453;hb=a2814ea7573bf5edd5323d6f89c48ff14105db69;hpb=ecd1a12fac39784bded85c0f06e47ace2dc98cde diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 29a27134f..7032a7f7f 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -1,22 +1,14 @@ /* - * 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. */ +#include "common/buffer-view.h" +#include #define _LGPL_SOURCE #include #include @@ -123,6 +115,25 @@ 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 @@ -238,9 +249,10 @@ static int lttng_kconsumer_snapshot_channel( 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); @@ -267,7 +279,15 @@ static int lttng_kconsumer_snapshot_channel( 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 @@ -493,6 +513,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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); @@ -634,7 +655,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); pthread_mutex_lock(&channel->lock); - new_stream = consumer_allocate_stream(channel->key, + new_stream = consumer_allocate_stream( + channel, + channel->key, fd, channel->name, channel->relayd_id, @@ -656,7 +679,6 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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); @@ -1159,6 +1181,32 @@ end_destroy_channel: 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, @@ -1185,8 +1233,7 @@ error_rotate_channel: *msg.u.create_trace_chunk.override_name ? msg.u.create_trace_chunk.override_name : NULL; - LTTNG_OPTIONAL(struct lttng_directory_handle) chunk_directory_handle = - LTTNG_OPTIONAL_INIT; + struct lttng_directory_handle *chunk_directory_handle = NULL; /* * The session daemon will only provide a chunk directory file @@ -1211,17 +1258,15 @@ error_rotate_channel: DBG("Received trace chunk directory fd (%d)", chunk_dirfd); - ret = lttng_directory_handle_init_from_dirfd( - &chunk_directory_handle.value, + chunk_directory_handle = lttng_directory_handle_create_from_dirfd( chunk_dirfd); - if (ret) { + 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; } - chunk_directory_handle.is_set = true; } ret_code = lttng_consumer_create_trace_chunk( @@ -1234,14 +1279,8 @@ error_rotate_channel: msg.u.create_trace_chunk.credentials.is_set ? &credentials : NULL, - chunk_directory_handle.is_set ? - &chunk_directory_handle.value : - NULL); - - if (chunk_directory_handle.is_set) { - lttng_directory_handle_fini( - &chunk_directory_handle.value); - } + chunk_directory_handle); + lttng_directory_handle_put(chunk_directory_handle); goto end_msg_sessiond; } case LTTNG_CONSUMER_CLOSE_TRACE_CHUNK: @@ -1327,66 +1366,62 @@ end: 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) { if (ret == -ENOTTY) { /* Command not implemented by lttng-modules. */ - index->stream_instance_id = -1ULL; + 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) { if (ret == -ENOTTY) { /* Command not implemented by lttng-modules. */ - index->packet_seq_num = -1ULL; + packet_seq_num = -1ULL; ret = 0; } else { PERROR("kernctl_get_sequence_number"); @@ -1395,6 +1430,18 @@ static int get_index_values(struct ctf_packet_index *index, int infd) } 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; } @@ -1446,6 +1493,7 @@ int update_stream_stats(struct lttng_consumer_stream *stream) if (ret == -ENOTTY) { /* Command not implemented by lttng-modules. */ seq = -1ULL; + stream->sequence_number_unavailable = true; } else { PERROR("kernctl_get_sequence_number"); goto end; @@ -1545,7 +1593,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, 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); @@ -1675,6 +1723,10 @@ 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) { @@ -1694,18 +1746,25 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, 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! */ assert(len >= subbuf_size); 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)) { @@ -1720,11 +1779,12 @@ 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 (err == -EFAULT) {