X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=5a39a0cc7e22822154ed2c6da334df0c1693355f;hp=b2f4c2686d2f7f81b5bc21c4249a681e92a5bad3;hb=5416a504a1ff9f2444e594084836dc8b59739526;hpb=c35f9726a22f1d93e14589688d830efccda196f3 diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index b2f4c2686..5a39a0cc7 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -411,7 +411,8 @@ void consumer_del_channel(struct lttng_consumer_channel *channel) rcu_read_unlock(); } - call_rcu(&channel->node.head, free_channel_rcu); + channel->is_deleted = true; + call_rcu(&channel->node.head, free_channel_rcu); end: pthread_mutex_unlock(&channel->lock); pthread_mutex_unlock(&consumer_data.lock); @@ -602,6 +603,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); @@ -708,11 +710,6 @@ void consumer_add_data_stream(struct lttng_consumer_stream *stream) pthread_mutex_unlock(&consumer_data.lock); } -void consumer_del_data_stream(struct lttng_consumer_stream *stream) -{ - consumer_del_stream(stream, data_ht); -} - /* * Add relayd socket to global consumer data hashtable. RCU read side lock MUST * be acquired before calling this. @@ -1013,32 +1010,17 @@ int lttng_consumer_channel_set_trace_chunk( struct lttng_consumer_channel *channel, struct lttng_trace_chunk *new_trace_chunk) { - int ret = 0; - const bool is_local_trace = channel->relayd_id == -1ULL; - bool update_stream_trace_chunk; - struct cds_lfht_iter iter; - struct lttng_consumer_stream *stream; - unsigned long channel_hash; - pthread_mutex_lock(&channel->lock); - /* - * A stream can transition to a state where it and its channel - * no longer belong to a trace chunk. For instance, this happens when - * a session is rotated while it is inactive. After the rotation - * of an inactive session completes, the channel and its streams no - * longer belong to a trace chunk. - * - * However, if a session is stopped, rotated, and started again, - * the session daemon will create a new chunk and send it to its peers. - * In that case, the streams' transition to a new chunk can be performed - * immediately. - * - * This trace chunk transition could also be performed lazily when - * a buffer is consumed. However, creating the files here allows the - * consumer daemon to report any creation error to the session daemon - * and cause the start of the tracing session to fail. - */ - update_stream_trace_chunk = !channel->trace_chunk && new_trace_chunk; + if (channel->is_deleted) { + /* + * The channel has been logically deleted and should no longer + * be used. It has released its reference to its current trace + * chunk and should not acquire a new one. + * + * Return success as there is nothing for the caller to do. + */ + goto end; + } /* * The acquisition of the reference cannot fail (barring @@ -1054,59 +1036,9 @@ int lttng_consumer_channel_set_trace_chunk( lttng_trace_chunk_put(channel->trace_chunk); channel->trace_chunk = new_trace_chunk; - if (!is_local_trace || !new_trace_chunk) { - /* Not an error. */ - goto end; - } - - if (!update_stream_trace_chunk) { - goto end; - } - - channel_hash = consumer_data.stream_per_chan_id_ht->hash_fct( - &channel->key, lttng_ht_seed); - rcu_read_lock(); - cds_lfht_for_each_entry_duplicate(consumer_data.stream_per_chan_id_ht->ht, - channel_hash, - consumer_data.stream_per_chan_id_ht->match_fct, - &channel->key, &iter, stream, node_channel_id.node) { - bool acquired_reference, should_regenerate_metadata = false; - - acquired_reference = lttng_trace_chunk_get(channel->trace_chunk); - assert(acquired_reference); - - pthread_mutex_lock(&stream->lock); - - /* - * On a transition from "no-chunk" to a new chunk, a metadata - * stream's content must be entirely dumped. This must occcur - * _after_ the creation of the metadata stream's output files - * as the consumption thread (not necessarily the one executing - * this) may start to consume during the call to - * consumer_metadata_stream_dump(). - */ - should_regenerate_metadata = - stream->metadata_flag && - !stream->trace_chunk && channel->trace_chunk; - stream->trace_chunk = channel->trace_chunk; - ret = consumer_stream_create_output_files(stream, true); - if (ret) { - pthread_mutex_unlock(&stream->lock); - goto end_rcu_unlock; - } - if (should_regenerate_metadata) { - ret = consumer_metadata_stream_dump(stream); - } - pthread_mutex_unlock(&stream->lock); - if (ret) { - goto end_rcu_unlock; - } - } -end_rcu_unlock: - rcu_read_unlock(); end: pthread_mutex_unlock(&channel->lock); - return ret; + return 0; } /* @@ -1397,6 +1329,7 @@ void lttng_consumer_cleanup(void) { struct lttng_ht_iter iter; struct lttng_consumer_channel *channel; + unsigned int trace_chunks_left; rcu_read_lock(); @@ -1421,6 +1354,27 @@ void lttng_consumer_cleanup(void) */ lttng_ht_destroy(consumer_data.stream_list_ht); + /* + * Trace chunks in the registry may still exist if the session + * daemon has encountered an internal error and could not + * tear down its sessions and/or trace chunks properly. + * + * Release the session daemon's implicit reference to any remaining + * trace chunk and print an error if any trace chunk was found. Note + * that there are _no_ legitimate cases for trace chunks to be left, + * it is a leak. However, it can happen following a crash of the + * session daemon and not emptying the registry would cause an assertion + * to hit. + */ + trace_chunks_left = lttng_trace_chunk_registry_put_each_chunk( + consumer_data.chunk_registry); + if (trace_chunks_left) { + ERR("%u trace chunks are leaked by lttng-consumerd. " + "This can be caused by an internal error of the session daemon.", + trace_chunks_left); + } + /* Run all callbacks freeing each chunk. */ + rcu_barrier(); lttng_trace_chunk_registry_destroy(consumer_data.chunk_registry); } @@ -1727,9 +1681,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( /* RCU lock for the relayd pointer */ rcu_read_lock(); - assert(stream->net_seq_idx != (uint64_t) -1ULL || - stream->chan->trace_chunk); + stream->trace_chunk); /* Flag that the current stream if set for network streaming. */ if (stream->net_seq_idx != (uint64_t) -1ULL) { @@ -2222,6 +2175,7 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } +static void lttng_consumer_close_all_metadata(void) { switch (consumer_data.type) { @@ -2254,7 +2208,8 @@ void lttng_consumer_close_all_metadata(void) void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, struct lttng_ht *ht) { - struct lttng_consumer_channel *free_chan = NULL; + struct lttng_consumer_channel *channel = NULL; + bool free_channel = false; assert(stream); /* @@ -2266,11 +2221,17 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, DBG3("Consumer delete metadata stream %d", stream->wait_fd); pthread_mutex_lock(&consumer_data.lock); - pthread_mutex_lock(&stream->chan->lock); + /* + * Note that this assumes that a stream's channel is never changed and + * that the stream's lock doesn't need to be taken to sample its + * channel. + */ + channel = stream->chan; + pthread_mutex_lock(&channel->lock); pthread_mutex_lock(&stream->lock); - if (stream->chan->metadata_cache) { + if (channel->metadata_cache) { /* Only applicable to userspace consumers. */ - pthread_mutex_lock(&stream->chan->metadata_cache->lock); + pthread_mutex_lock(&channel->metadata_cache->lock); } /* Remove any reference to that stream. */ @@ -2282,28 +2243,29 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, consumer_stream_destroy_buffers(stream); /* Atomically decrement channel refcount since other threads can use it. */ - if (!uatomic_sub_return(&stream->chan->refcount, 1) - && !uatomic_read(&stream->chan->nb_init_stream_left)) { + if (!uatomic_sub_return(&channel->refcount, 1) + && !uatomic_read(&channel->nb_init_stream_left)) { /* Go for channel deletion! */ - free_chan = stream->chan; + free_channel = true; } + stream->chan = NULL; /* * Nullify the stream reference so it is not used after deletion. The * channel lock MUST be acquired before being able to check for a NULL * pointer value. */ - stream->chan->metadata_stream = NULL; + channel->metadata_stream = NULL; - if (stream->chan->metadata_cache) { - pthread_mutex_unlock(&stream->chan->metadata_cache->lock); + if (channel->metadata_cache) { + pthread_mutex_unlock(&channel->metadata_cache->lock); } pthread_mutex_unlock(&stream->lock); - pthread_mutex_unlock(&stream->chan->lock); + pthread_mutex_unlock(&channel->lock); pthread_mutex_unlock(&consumer_data.lock); - if (free_chan) { - consumer_del_channel(free_chan); + if (free_channel) { + consumer_del_channel(channel); } lttng_trace_chunk_put(stream->trace_chunk); @@ -3978,15 +3940,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"); @@ -4041,7 +4011,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(); @@ -4054,37 +4024,78 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, rotating_to_new_chunk = false; } - ret = lttng_consumer_sample_snapshot_positions(stream); + /* + * Active flush; has no effect if the production position + * is at a packet boundary. + */ + ret = consumer_flush_buffer(stream, 1); if (ret < 0) { - ERR("Failed to sample snapshot position during channel rotation"); + 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) { stream->rotate_ready = true; } - - ret = consumer_flush_buffer(stream, 1); - if (ret < 0) { - ERR("Failed to flush stream %" PRIu64 " during channel rotation", + /* + * 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. + */ + 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); 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". + */ const struct relayd_stream_rotation_position position = { .stream_id = stream->relayd_stream_id, - .rotate_at_seq_num = (stream->rotate_position / - stream->max_sb_size) + 1, + .rotate_at_seq_num = stream->rotate_position, }; ret = lttng_dynamic_array_add_element( @@ -4147,44 +4158,39 @@ end: */ int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream) { - int ret; - unsigned long consumed_pos; - - if (!stream->rotate_position && !stream->rotate_ready) { - ret = 0; - goto end; - } - if (stream->rotate_ready) { - ret = 1; - goto end; + return 1; } /* - * If we don't have the rotate_ready flag, check the consumed position - * to determine if we need to rotate. + * 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. */ - ret = lttng_consumer_sample_snapshot_positions(stream); - if (ret < 0) { - ERR("Taking snapshot positions"); - goto end; + if (stream->sequence_number_unavailable) { + ERR("Internal error: rotation used on stream %" PRIu64 + " with unavailable sequence number", + stream->key); + return -1; } - ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos); - if (ret < 0) { - ERR("Consumed snapshot position"); - goto end; + if (stream->rotate_position == -1ULL || + stream->last_sequence_number == -1ULL) { + return 0; } - /* Rotate position not reached yet (with check for overflow). */ - if ((long) (consumed_pos - stream->rotate_position) < 0) { - ret = 0; - goto end; + /* + * 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". + */ + if (stream->last_sequence_number >= stream->rotate_position - 1) { + return 1; } - ret = 1; -end: - return ret; + return 0; } /* @@ -4192,7 +4198,7 @@ end: */ void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream) { - stream->rotate_position = 0; + stream->rotate_position = -1ULL; stream->rotate_ready = false; } @@ -4367,7 +4373,7 @@ enum lttcomm_return_code lttng_consumer_init_command( const lttng_uuid sessiond_uuid) { enum lttcomm_return_code ret; - char uuid_str[UUID_STR_LEN]; + char uuid_str[LTTNG_UUID_STR_LEN]; if (ctx->sessiond_uuid.is_set) { ret = LTTCOMM_CONSUMERD_ALREADY_SET; @@ -4393,7 +4399,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( { int ret; enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS; - struct lttng_trace_chunk *created_chunk, *published_chunk; + struct lttng_trace_chunk *created_chunk = NULL, *published_chunk = NULL; enum lttng_trace_chunk_status chunk_status; char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)]; char creation_timestamp_buffer[ISO8601_STR_LEN]; @@ -4448,7 +4454,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( if (!created_chunk) { ERR("Failed to create trace chunk"); ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED; - goto end; + goto error; } if (chunk_override_name) { @@ -4456,7 +4462,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( chunk_override_name); if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED; - goto end; + goto error; } } @@ -4466,7 +4472,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { ERR("Failed to set trace chunk credentials"); ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED; - goto end; + goto error; } /* * The consumer daemon has no ownership of the chunk output @@ -4474,10 +4480,11 @@ 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; - goto end; + goto error; } } @@ -4489,7 +4496,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( if (!published_chunk) { ERR("Failed to publish trace chunk"); ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED; - goto end; + goto error; } rcu_read_lock(); @@ -4512,11 +4519,13 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( * channels. */ enum lttcomm_return_code close_ret; + char path[LTTNG_PATH_MAX]; DBG("Failed to set new trace chunk on existing channels, rolling back"); close_ret = lttng_consumer_close_trace_chunk(relayd_id, session_id, chunk_id, - chunk_creation_timestamp, NULL); + chunk_creation_timestamp, NULL, + path); if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) { ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64, session_id, chunk_id); @@ -4542,12 +4551,13 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( if (!relayd || ret) { enum lttcomm_return_code close_ret; + char path[LTTNG_PATH_MAX]; close_ret = lttng_consumer_close_trace_chunk(relayd_id, session_id, chunk_id, chunk_creation_timestamp, - NULL); + NULL, path); if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) { ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64, session_id, @@ -4555,21 +4565,23 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( } ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED; - goto error; + goto error_unlock; } } -error: +error_unlock: rcu_read_unlock(); +error: /* Release the reference returned by the "publish" operation. */ lttng_trace_chunk_put(published_chunk); -end: + lttng_trace_chunk_put(created_chunk); return ret_code; } enum lttcomm_return_code lttng_consumer_close_trace_chunk( const uint64_t *relayd_id, uint64_t session_id, uint64_t chunk_id, time_t chunk_close_timestamp, - const enum lttng_trace_chunk_command_type *close_command) + const enum lttng_trace_chunk_command_type *close_command, + char *path) { enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS; struct lttng_trace_chunk *chunk; @@ -4667,7 +4679,8 @@ enum lttcomm_return_code lttng_consumer_close_trace_chunk( if (relayd) { pthread_mutex_lock(&relayd->ctrl_sock_mutex); ret = relayd_close_trace_chunk( - &relayd->control_sock, chunk); + &relayd->control_sock, chunk, + path); pthread_mutex_unlock(&relayd->ctrl_sock_mutex); } else { ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64, @@ -4698,12 +4711,11 @@ enum lttcomm_return_code lttng_consumer_trace_chunk_exists( { int ret; enum lttcomm_return_code ret_code; - struct lttng_trace_chunk *chunk; char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)]; const char *relayd_id_str = "(none)"; const bool is_local_trace = !relayd_id; struct consumer_relayd_sock_pair *relayd = NULL; - bool chunk_exists_remote; + bool chunk_exists_local, chunk_exists_remote; if (relayd_id) { int ret; @@ -4721,13 +4733,19 @@ enum lttcomm_return_code lttng_consumer_trace_chunk_exists( DBG("Consumer trace chunk exists command: relayd_id = %s" ", chunk_id = %" PRIu64, relayd_id_str, chunk_id); - chunk = lttng_trace_chunk_registry_find_chunk( + ret = lttng_trace_chunk_registry_chunk_exists( consumer_data.chunk_registry, session_id, - chunk_id); - DBG("Trace chunk %s locally", chunk ? "exists" : "does not exist"); - if (chunk) { + chunk_id, &chunk_exists_local); + if (ret) { + /* Internal error. */ + ERR("Failed to query the existence of a trace chunk"); + ret_code = LTTCOMM_CONSUMERD_FATAL; + goto end; + } + DBG("Trace chunk %s locally", + chunk_exists_local ? "exists" : "does not exist"); + if (chunk_exists_local) { ret_code = LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL; - lttng_trace_chunk_put(chunk); goto end; } else if (is_local_trace) { ret_code = LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;