X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=4613284a8d5d6c29afd0c0fa17a725037c0b2f66;hp=e99dc54ed0a98897eb866a19bceb5ecf7087f163;hb=96393977613193a0e28343ea32a60c819a00e726;hpb=c70636a7342f34e3be68fcf411cf3e3718b8e73f diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index e99dc54ed..4613284a8 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -1,20 +1,10 @@ /* - * Copyright (C) 2011 - Julien Desfossez - * Mathieu Desnoyers - * 2012 - David Goulet + * Copyright (C) 2011 Julien Desfossez + * Copyright (C) 2011 Mathieu Desnoyers + * Copyright (C) 2012 David Goulet * - * 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 @@ -91,6 +81,20 @@ int consumer_quit; static struct lttng_ht *metadata_ht; static struct lttng_ht *data_ht; +static const char *get_consumer_domain(void) +{ + switch (consumer_data.type) { + case LTTNG_CONSUMER_KERNEL: + return DEFAULT_KERNEL_TRACE_DIR; + case LTTNG_CONSUMER64_UST: + /* Fall-through. */ + case LTTNG_CONSUMER32_UST: + return DEFAULT_UST_TRACE_DIR; + default: + abort(); + } +} + /* * Notify a thread lttng pipe to poll back again. This usually means that some * global state has changed so we just send back the thread in a poll wait @@ -603,6 +607,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE; stream->index_file = NULL; stream->last_sequence_number = -1ULL; + stream->rotate_position = -1ULL; pthread_mutex_init(&stream->lock, NULL); pthread_mutex_init(&stream->metadata_timer_lock, NULL); @@ -815,7 +820,7 @@ int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, /* Add stream on the relayd */ pthread_mutex_lock(&relayd->ctrl_sock_mutex); ret = relayd_add_stream(&relayd->control_sock, stream->name, - path, &stream->relayd_stream_id, + get_consumer_domain(), path, &stream->relayd_stream_id, stream->chan->tracefile_size, stream->chan->tracefile_count, stream->trace_chunk); @@ -2174,6 +2179,7 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } +static void lttng_consumer_close_all_metadata(void) { switch (consumer_data.type) { @@ -3938,15 +3944,23 @@ int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_act switch (consumer_data.type) { case LTTNG_CONSUMER_KERNEL: - ret = kernctl_buffer_flush(stream->wait_fd); - if (ret < 0) { - ERR("Failed to flush kernel stream"); - goto end; + if (producer_active) { + ret = kernctl_buffer_flush(stream->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end; + } + } else { + ret = kernctl_buffer_flush_empty(stream->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end; + } } break; case LTTNG_CONSUMER32_UST: case LTTNG_CONSUMER64_UST: - lttng_ustctl_flush_buffer(stream, producer_active); + lttng_ustconsumer_flush_buffer(stream, producer_active); break; default: ERR("Unknown consumer_data type"); @@ -4001,7 +4015,7 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, ht->hash_fct(&channel->key, lttng_ht_seed), ht->match_fct, &channel->key, &iter.iter, stream, node_channel_id.node) { - unsigned long consumed_pos; + unsigned long produced_pos = 0, consumed_pos = 0; health_code_update(); @@ -4014,65 +4028,95 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, rotating_to_new_chunk = false; } - ret = lttng_consumer_sample_snapshot_positions(stream); - if (ret < 0) { - ERR("Failed to sample snapshot position during channel rotation"); - goto end_unlock_stream; + /* + * Do not flush an empty packet when rotating from a NULL trace + * chunk. The stream has no means to output data, and the prior + * rotation which rotated to NULL performed that side-effect already. + */ + if (stream->trace_chunk) { + /* + * For metadata stream, do an active flush, which does not + * produce empty packets. For data streams, empty-flush; + * ensures we have at least one packet in each stream per trace + * chunk, even if no data was produced. + */ + ret = consumer_flush_buffer(stream, stream->metadata_flag ? 1 : 0); + if (ret < 0) { + ERR("Failed to flush stream %" PRIu64 " during channel rotation", + stream->key); + goto end_unlock_stream; + } } - ret = lttng_consumer_get_produced_snapshot(stream, - &stream->rotate_position); - if (ret < 0) { - ERR("Failed to sample produced position during channel rotation"); + ret = lttng_consumer_take_snapshot(stream); + if (ret < 0 && ret != -ENODATA && ret != -EAGAIN) { + ERR("Failed to sample snapshot position during channel rotation"); goto end_unlock_stream; } + if (!ret) { + ret = lttng_consumer_get_produced_snapshot(stream, + &produced_pos); + if (ret < 0) { + ERR("Failed to sample produced position during channel rotation"); + goto end_unlock_stream; + } - lttng_consumer_get_consumed_snapshot(stream, - &consumed_pos); - if (consumed_pos == stream->rotate_position) { + ret = lttng_consumer_get_consumed_snapshot(stream, + &consumed_pos); + if (ret < 0) { + ERR("Failed to sample consumed position during channel rotation"); + goto end_unlock_stream; + } + } + /* + * Align produced position on the start-of-packet boundary of the first + * packet going into the next trace chunk. + */ + produced_pos = ALIGN_FLOOR(produced_pos, stream->max_sb_size); + if (consumed_pos == produced_pos) { + DBG("Set rotate ready for stream %" PRIu64 " produced = %lu consumed = %lu", + stream->key, produced_pos, consumed_pos); stream->rotate_ready = true; + } else { + DBG("Different consumed and produced positions " + "for stream %" PRIu64 " produced = %lu consumed = %lu", + stream->key, produced_pos, consumed_pos); } - /* - * Active flush; has no effect if the production position - * is at a packet boundary. + * The rotation position is based on the packet_seq_num of the + * packet following the last packet that was consumed for this + * stream, incremented by the offset between produced and + * consumed positions. This rotation position is a lower bound + * (inclusive) at which the next trace chunk starts. Since it + * is a lower bound, it is OK if the packet_seq_num does not + * correspond exactly to the same packet identified by the + * consumed_pos, which can happen in overwrite mode. */ - ret = consumer_flush_buffer(stream, 1); - if (ret < 0) { - ERR("Failed to flush stream %" PRIu64 " during channel rotation", + if (stream->sequence_number_unavailable) { + /* + * Rotation should never be performed on a session which + * interacts with a pre-2.8 lttng-modules, which does + * not implement packet sequence number. + */ + ERR("Failure to rotate stream %" PRIu64 ": sequence number unavailable", stream->key); + ret = -1; goto end_unlock_stream; } + stream->rotate_position = stream->last_sequence_number + 1 + + ((produced_pos - consumed_pos) / stream->max_sb_size); + DBG("Set rotation position for stream %" PRIu64 " at position %" PRIu64, + stream->key, stream->rotate_position); if (!is_local_trace) { /* * The relay daemon control protocol expects a rotation * position as "the sequence number of the first packet - * _after_ the current trace chunk. - * - * At the moment when the positions of the buffers are - * sampled, the production position does not necessarily - * sit at a packet boundary. The 'active' flush - * operation above will push the production position to - * the next packet boundary _if_ it is not already - * sitting at such a boundary. - * - * Assuming a current production position that is not - * on the bound of a packet, the 'target' sequence - * number is - * (consumed_pos / subbuffer_size) + 1 - * Note the '+ 1' to ensure the current packet is - * part of the current trace chunk. - * - * However, if the production position is already at - * a packet boundary, the '+ 1' is not necessary as the - * last packet of the current chunk is already - * 'complete'. + * _after_ the current trace chunk". */ const struct relayd_stream_rotation_position position = { .stream_id = stream->relayd_stream_id, - .rotate_at_seq_num = (stream->rotate_position / stream->max_sb_size) + - !!(stream->rotate_position % stream->max_sb_size), + .rotate_at_seq_num = stream->rotate_position, }; ret = lttng_dynamic_array_add_element( @@ -4127,60 +4171,172 @@ end: return ret; } -/* - * Check if a stream is ready to be rotated after extracting it. - * - * Return 1 if it is ready for rotation, 0 if it is not, a negative value on - * error. Stream lock must be held. - */ -int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream) +static +int consumer_clear_buffer(struct lttng_consumer_stream *stream) { - int ret; - unsigned long consumed_pos; + int ret = 0; + unsigned long consumed_pos_before, consumed_pos_after; - if (!stream->rotate_position && !stream->rotate_ready) { - ret = 0; + ret = lttng_consumer_sample_snapshot_positions(stream); + if (ret < 0) { + ERR("Taking snapshot positions"); goto end; } - if (stream->rotate_ready) { - ret = 1; + ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_before); + if (ret < 0) { + ERR("Consumed snapshot position"); goto end; } - /* - * If we don't have the rotate_ready flag, check the consumed position - * to determine if we need to rotate. - */ + switch (consumer_data.type) { + case LTTNG_CONSUMER_KERNEL: + ret = kernctl_buffer_clear(stream->wait_fd); + if (ret < 0) { + ERR("Failed to clear kernel stream (ret = %d)", ret); + goto end; + } + break; + case LTTNG_CONSUMER32_UST: + case LTTNG_CONSUMER64_UST: + lttng_ustconsumer_clear_buffer(stream); + break; + default: + ERR("Unknown consumer_data type"); + abort(); + } + ret = lttng_consumer_sample_snapshot_positions(stream); if (ret < 0) { ERR("Taking snapshot positions"); goto end; } - - ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos); + ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_after); if (ret < 0) { ERR("Consumed snapshot position"); goto end; } + DBG("clear: before: %lu after: %lu", consumed_pos_before, consumed_pos_after); +end: + return ret; +} - /* Rotate position not reached yet (with check for overflow). */ - if ((long) (consumed_pos - stream->rotate_position) < 0) { - ret = 0; - goto end; +static +int consumer_clear_stream(struct lttng_consumer_stream *stream) +{ + int ret; + + ret = consumer_flush_buffer(stream, 1); + if (ret < 0) { + ERR("Failed to flush stream %" PRIu64 " during channel clear", + stream->key); + ret = LTTCOMM_CONSUMERD_FATAL; + goto error; } - ret = 1; -end: + ret = consumer_clear_buffer(stream); + if (ret < 0) { + ERR("Failed to clear stream %" PRIu64 " during channel clear", + stream->key); + ret = LTTCOMM_CONSUMERD_FATAL; + goto error; + } + + ret = LTTCOMM_CONSUMERD_SUCCESS; +error: return ret; } +static +int consumer_clear_unmonitored_channel(struct lttng_consumer_channel *channel) +{ + int ret; + struct lttng_consumer_stream *stream; + + rcu_read_lock(); + pthread_mutex_lock(&channel->lock); + cds_list_for_each_entry(stream, &channel->streams.head, send_node) { + health_code_update(); + pthread_mutex_lock(&stream->lock); + ret = consumer_clear_stream(stream); + if (ret) { + goto error_unlock; + } + pthread_mutex_unlock(&stream->lock); + } + pthread_mutex_unlock(&channel->lock); + rcu_read_unlock(); + return 0; + +error_unlock: + pthread_mutex_unlock(&stream->lock); + pthread_mutex_unlock(&channel->lock); + rcu_read_unlock(); + return ret; +} + +/* + * Check if a stream is ready to be rotated after extracting it. + * + * Return 1 if it is ready for rotation, 0 if it is not, a negative value on + * error. Stream lock must be held. + */ +int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream) +{ + DBG("Check is rotate ready for stream %" PRIu64 + " ready %u rotate_position %" PRIu64 + " last_sequence_number %" PRIu64, + stream->key, stream->rotate_ready, + stream->rotate_position, stream->last_sequence_number); + if (stream->rotate_ready) { + return 1; + } + + /* + * If packet seq num is unavailable, it means we are interacting + * with a pre-2.8 lttng-modules which does not implement the + * sequence number. Rotation should never be used by sessiond in this + * scenario. + */ + if (stream->sequence_number_unavailable) { + ERR("Internal error: rotation used on stream %" PRIu64 + " with unavailable sequence number", + stream->key); + return -1; + } + + if (stream->rotate_position == -1ULL || + stream->last_sequence_number == -1ULL) { + return 0; + } + + /* + * Rotate position not reached yet. The stream rotate position is + * the position of the next packet belonging to the next trace chunk, + * but consumerd considers rotation ready when reaching the last + * packet of the current chunk, hence the "rotate_position - 1". + */ + + DBG("Check is rotate ready for stream %" PRIu64 + " last_sequence_number %" PRIu64 + " rotate_position %" PRIu64, + stream->key, stream->last_sequence_number, + stream->rotate_position); + if (stream->last_sequence_number >= stream->rotate_position - 1) { + return 1; + } + + return 0; +} + /* * Reset the state for a stream after a rotation occurred. */ void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream) { - stream->rotate_position = 0; + DBG("lttng_consumer_reset_stream_rotate_state for stream %" PRIu64, + stream->key); + stream->rotate_position = -1ULL; stream->rotate_ready = false; } @@ -4432,7 +4588,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( * and LTTNG_CONSUMER_DESTROY_TRACE_CHUNK commands. */ created_chunk = lttng_trace_chunk_create(chunk_id, - chunk_creation_timestamp); + chunk_creation_timestamp, NULL); if (!created_chunk) { ERR("Failed to create trace chunk"); ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED; @@ -4462,6 +4618,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( */ chunk_status = lttng_trace_chunk_set_as_user(created_chunk, chunk_directory_handle); + chunk_directory_handle = NULL; if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { ERR("Failed to set trace chunk's directory handle"); ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED; @@ -4762,3 +4919,67 @@ end_rcu_unlock: end: return ret_code; } + +static +int consumer_clear_monitored_channel(struct lttng_consumer_channel *channel) +{ + struct lttng_ht *ht; + struct lttng_consumer_stream *stream; + struct lttng_ht_iter iter; + int ret; + + ht = consumer_data.stream_per_chan_id_ht; + + rcu_read_lock(); + cds_lfht_for_each_entry_duplicate(ht->ht, + ht->hash_fct(&channel->key, lttng_ht_seed), + ht->match_fct, &channel->key, + &iter.iter, stream, node_channel_id.node) { + /* + * Protect against teardown with mutex. + */ + pthread_mutex_lock(&stream->lock); + if (cds_lfht_is_node_deleted(&stream->node.node)) { + goto next; + } + ret = consumer_clear_stream(stream); + if (ret) { + goto error_unlock; + } + next: + pthread_mutex_unlock(&stream->lock); + } + rcu_read_unlock(); + return LTTCOMM_CONSUMERD_SUCCESS; + +error_unlock: + pthread_mutex_unlock(&stream->lock); + rcu_read_unlock(); + return ret; +} + +int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel) +{ + int ret; + + DBG("Consumer clear channel %" PRIu64, channel->key); + + if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) { + /* + * Nothing to do for the metadata channel/stream. + * Snapshot mechanism already take care of the metadata + * handling/generation, and monitored channels only need to + * have their data stream cleared.. + */ + ret = LTTCOMM_CONSUMERD_SUCCESS; + goto end; + } + + if (!channel->monitor) { + ret = consumer_clear_unmonitored_channel(channel); + } else { + ret = consumer_clear_monitored_channel(channel); + } +end: + return ret; +}