X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=3c51cfa6f40221cff9c2ee8ce65b8788c3ca7afc;hp=7a7f7954be3c843df01bfea1af6d3c815e345773;hb=1f8d1c146ba9c46338d04c416a75ee6d105c5c50;hpb=d295668767ac8234e83984e1812d342d03293d88 diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index 7a7f7954b..3c51cfa6f 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -52,6 +52,7 @@ #include #include #include +#include struct lttng_consumer_global_data consumer_data = { .stream_count = 0, @@ -410,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); @@ -707,11 +709,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. @@ -936,6 +933,7 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream, data_hdr.stream_id = htobe64(stream->relayd_stream_id); data_hdr.data_size = htobe32(data_size); data_hdr.padding_size = htobe32(padding); + /* * Note that net_seq_num below is assigned with the *current* value of * next_net_seq_num and only after that the next_net_seq_num will be @@ -1011,32 +1009,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 @@ -1052,59 +1035,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; } /* @@ -1395,6 +1328,7 @@ void lttng_consumer_cleanup(void) { struct lttng_ht_iter iter; struct lttng_consumer_channel *channel; + unsigned int trace_chunks_left; rcu_read_lock(); @@ -1419,6 +1353,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); } @@ -1725,8 +1680,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( /* RCU lock for the relayd pointer */ rcu_read_lock(); - - assert(stream->chan->trace_chunk); + assert(stream->net_seq_idx != (uint64_t) -1ULL || + stream->trace_chunk); /* Flag that the current stream if set for network streaming. */ if (stream->net_seq_idx != (uint64_t) -1ULL) { @@ -1856,7 +1811,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( relayd_hang_up = 1; /* Socket operation failed. We consider the relayd dead */ - if (errno == EPIPE || errno == EINVAL || errno == EBADF) { + if (errno == EPIPE) { /* * This is possible if the fd is closed on the other side * (outfd) or any write problem. It can be verbose a bit for a @@ -2219,6 +2174,7 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } +static void lttng_consumer_close_all_metadata(void) { switch (consumer_data.type) { @@ -2251,7 +2207,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); /* @@ -2263,11 +2220,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. */ @@ -2279,28 +2242,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); @@ -4011,12 +3975,28 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, struct lttng_consumer_stream *stream; struct lttng_ht_iter iter; struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht; + struct lttng_dynamic_array stream_rotation_positions; + uint64_t next_chunk_id, stream_count = 0; + enum lttng_trace_chunk_status chunk_status; + const bool is_local_trace = relayd_id == -1ULL; + struct consumer_relayd_sock_pair *relayd = NULL; + bool rotating_to_new_chunk = true; DBG("Consumer sample rotate position for channel %" PRIu64, key); + lttng_dynamic_array_init(&stream_rotation_positions, + sizeof(struct relayd_stream_rotation_position), NULL); + rcu_read_lock(); pthread_mutex_lock(&channel->lock); + assert(channel->trace_chunk); + chunk_status = lttng_trace_chunk_get_id(channel->trace_chunk, + &next_chunk_id); + if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { + ret = -1; + goto end_unlock_channel; + } cds_lfht_for_each_entry_duplicate(ht->ht, ht->hash_fct(&channel->key, lttng_ht_seed), @@ -4031,6 +4011,10 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, */ pthread_mutex_lock(&stream->lock); + if (stream->trace_chunk == stream->chan->trace_chunk) { + rotating_to_new_chunk = false; + } + ret = lttng_consumer_sample_snapshot_positions(stream); if (ret < 0) { ERR("Failed to sample snapshot position during channel rotation"); @@ -4050,6 +4034,10 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, stream->rotate_ready = true; } + /* + * 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 flush stream %" PRIu64 " during channel rotation", @@ -4057,18 +4045,86 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, goto end_unlock_stream; } + 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'. + */ + 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), + }; + + ret = lttng_dynamic_array_add_element( + &stream_rotation_positions, + &position); + if (ret) { + ERR("Failed to allocate stream rotation position"); + goto end_unlock_stream; + } + stream_count++; + } pthread_mutex_unlock(&stream->lock); } + stream = NULL; pthread_mutex_unlock(&channel->lock); + if (is_local_trace) { + ret = 0; + goto end; + } + + relayd = consumer_find_relayd(relayd_id); + if (!relayd) { + ERR("Failed to find relayd %" PRIu64, relayd_id); + ret = -1; + goto end; + } + + pthread_mutex_lock(&relayd->ctrl_sock_mutex); + ret = relayd_rotate_streams(&relayd->control_sock, stream_count, + rotating_to_new_chunk ? &next_chunk_id : NULL, + (const struct relayd_stream_rotation_position *) + stream_rotation_positions.buffer.data); + pthread_mutex_unlock(&relayd->ctrl_sock_mutex); + if (ret < 0) { + ERR("Relayd rotate stream failed. Cleaning up relayd %" PRIu64, + relayd->net_seq_idx); + lttng_consumer_cleanup_relayd(relayd); + goto end; + } + ret = 0; goto end; end_unlock_stream: pthread_mutex_unlock(&stream->lock); +end_unlock_channel: pthread_mutex_unlock(&channel->lock); end: rcu_read_unlock(); + lttng_dynamic_array_reset(&stream_rotation_positions); return ret; } @@ -4167,52 +4223,6 @@ end: return ret; } -/* - * Perform the rotation a stream file on the relay. - */ -int rotate_relay_stream(struct lttng_consumer_local_data *ctx, - struct lttng_consumer_stream *stream) -{ - int ret; - struct consumer_relayd_sock_pair *relayd; - uint64_t chunk_id; - enum lttng_trace_chunk_status chunk_status; - - DBG("Rotate relay stream"); - relayd = consumer_find_relayd(stream->net_seq_idx); - if (!relayd) { - ERR("Failed to find relayd"); - ret = -1; - goto end; - } - - chunk_status = lttng_trace_chunk_get_id(stream->chan->trace_chunk, - &chunk_id); - if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { - ERR("Failed to retrieve the id of the current trace chunk of channel \"%s\"", - stream->chan->name); - ret = -1; - goto end; - } - - pthread_mutex_lock(&relayd->ctrl_sock_mutex); - ret = relayd_rotate_stream(&relayd->control_sock, - stream->relayd_stream_id, - chunk_id, - stream->last_sequence_number); - pthread_mutex_unlock(&relayd->ctrl_sock_mutex); - if (ret < 0) { - ERR("Relayd rotate stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx); - lttng_consumer_cleanup_relayd(relayd); - } - if (ret) { - ERR("Rotate relay stream"); - } - -end: - return ret; -} - /* * Performs the stream rotation for the rotate session feature if needed. * It must be called with the channel and stream locks held. @@ -4254,14 +4264,12 @@ int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx, stream->trace_chunk = stream->chan->trace_chunk; } - if (stream->net_seq_idx != (uint64_t) -1ULL) { - ret = rotate_relay_stream(ctx, stream); - } else { + if (stream->net_seq_idx == (uint64_t) -1ULL) { ret = rotate_local_stream(ctx, stream); - } - if (ret < 0) { - ERR("Failed to rotate stream, ret = %i", ret); - goto error; + if (ret < 0) { + ERR("Failed to rotate stream, ret = %i", ret); + goto error; + } } if (stream->metadata_flag && stream->trace_chunk) { @@ -4348,7 +4356,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; @@ -4374,7 +4382,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]; @@ -4429,7 +4437,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) { @@ -4437,7 +4445,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; } } @@ -4447,7 +4455,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 @@ -4458,7 +4466,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( 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; } } @@ -4470,7 +4478,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(); @@ -4493,11 +4501,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); + 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); @@ -4507,22 +4517,59 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( break; } } - rcu_read_unlock(); + if (relayd_id) { + struct consumer_relayd_sock_pair *relayd; + + relayd = consumer_find_relayd(*relayd_id); + if (relayd) { + pthread_mutex_lock(&relayd->ctrl_sock_mutex); + ret = relayd_create_trace_chunk( + &relayd->control_sock, published_chunk); + pthread_mutex_unlock(&relayd->ctrl_sock_mutex); + } else { + ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64, *relayd_id); + } + + 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, 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); + } + + ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED; + goto error_unlock; + } + } +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) + uint64_t chunk_id, time_t chunk_close_timestamp, + 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; char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)]; const char *relayd_id_str = "(none)"; + const char *close_command_name = "none"; struct lttng_ht_iter iter; struct lttng_consumer_channel *channel; enum lttng_trace_chunk_status chunk_status; @@ -4538,16 +4585,21 @@ enum lttcomm_return_code lttng_consumer_close_trace_chunk( } else { relayd_id_str = "(formatting error)"; } - } + } + if (close_command) { + close_command_name = lttng_trace_chunk_command_type_get_name( + *close_command); + } DBG("Consumer close trace chunk command: relayd_id = %s" - ", session_id = %" PRIu64 - ", chunk_id = %" PRIu64, relayd_id_str, - session_id, chunk_id); + ", session_id = %" PRIu64 ", chunk_id = %" PRIu64 + ", close command = %s", + relayd_id_str, session_id, chunk_id, + close_command_name); + chunk = lttng_trace_chunk_registry_find_chunk( - consumer_data.chunk_registry, session_id, - chunk_id); - if (!chunk) { + consumer_data.chunk_registry, session_id, chunk_id); + if (!chunk) { ERR("Failed to find chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64, session_id, chunk_id); @@ -4561,12 +4613,15 @@ enum lttcomm_return_code lttng_consumer_close_trace_chunk( ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED; goto end; } - /* - * Release the reference returned by the "find" operation and - * the session daemon's implicit reference to the chunk. - */ - lttng_trace_chunk_put(chunk); - lttng_trace_chunk_put(chunk); + + if (close_command) { + chunk_status = lttng_trace_chunk_set_close_command( + chunk, *close_command); + if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { + ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED; + goto end; + } + } /* * chunk is now invalid to access as we no longer hold a reference to @@ -4597,8 +4652,38 @@ enum lttcomm_return_code lttng_consumer_close_trace_chunk( ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED; } } + + if (relayd_id) { + int ret; + struct consumer_relayd_sock_pair *relayd; + + relayd = consumer_find_relayd(*relayd_id); + if (relayd) { + pthread_mutex_lock(&relayd->ctrl_sock_mutex); + ret = relayd_close_trace_chunk( + &relayd->control_sock, chunk, + path); + pthread_mutex_unlock(&relayd->ctrl_sock_mutex); + } else { + ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64, + *relayd_id); + } + + if (!relayd || ret) { + ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED; + goto error_unlock; + } + } +error_unlock: rcu_read_unlock(); end: + /* + * Release the reference returned by the "find" operation and + * the session daemon's implicit reference to the chunk. + */ + lttng_trace_chunk_put(chunk); + lttng_trace_chunk_put(chunk); + return ret_code; } @@ -4606,10 +4691,13 @@ enum lttcomm_return_code lttng_consumer_trace_chunk_exists( const uint64_t *relayd_id, uint64_t session_id, uint64_t chunk_id) { + 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_local, chunk_exists_remote; if (relayd_id) { int ret; @@ -4625,16 +4713,53 @@ enum lttcomm_return_code lttng_consumer_trace_chunk_exists( } DBG("Consumer trace chunk exists command: relayd_id = %s" - ", session_id = %" PRIu64 ", chunk_id = %" PRIu64, relayd_id_str, - session_id, chunk_id); - chunk = lttng_trace_chunk_registry_find_chunk( - consumer_data.chunk_registry, session_id, chunk_id); - DBG("Trace chunk %s locally", chunk ? "exists" : "does not exist"); - ret_code = chunk ? LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL : + ret = lttng_trace_chunk_registry_chunk_exists( + consumer_data.chunk_registry, session_id, + 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; + goto end; + } else if (is_local_trace) { + ret_code = LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK; + goto end; + } + + rcu_read_lock(); + relayd = consumer_find_relayd(*relayd_id); + if (!relayd) { + ERR("Failed to find relayd %" PRIu64, *relayd_id); + ret_code = LTTCOMM_CONSUMERD_INVALID_PARAMETERS; + goto end_rcu_unlock; + } + DBG("Looking up existence of trace chunk on relay daemon"); + pthread_mutex_lock(&relayd->ctrl_sock_mutex); + ret = relayd_trace_chunk_exists(&relayd->control_sock, chunk_id, + &chunk_exists_remote); + pthread_mutex_unlock(&relayd->ctrl_sock_mutex); + if (ret < 0) { + ERR("Failed to look-up the existence of trace chunk on relay daemon"); + ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL; + goto end_rcu_unlock; + } + + ret_code = chunk_exists_remote ? + LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE : LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK; + DBG("Trace chunk %s on relay daemon", + chunk_exists_remote ? "exists" : "does not exist"); - lttng_trace_chunk_put(chunk); +end_rcu_unlock: + rcu_read_unlock(); +end: return ret_code; }