#include <common/trace-chunk.h>
#include <common/trace-chunk-registry.h>
#include <common/string-utils/format.h>
+#include <common/dynamic-array.h>
struct lttng_consumer_global_data consumer_data = {
.stream_count = 0,
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);
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.
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
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
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;
}
/*
{
struct lttng_ht_iter iter;
struct lttng_consumer_channel *channel;
+ unsigned int trace_chunks_left;
rcu_read_lock();
*/
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);
}
/* 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) {
}
}
+static
void lttng_consumer_close_all_metadata(void)
{
switch (consumer_data.type) {
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);
/*
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. */
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);
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),
*/
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");
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",
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;
}
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.
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) {
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;
{
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];
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) {
chunk_override_name);
if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
- goto end;
+ goto error;
}
}
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
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;
}
}
if (!published_chunk) {
ERR("Failed to publish trace chunk");
ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
- goto end;
+ goto error;
}
rcu_read_lock();
* 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);
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);
+ 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,
}
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)
+ 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;
} 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);
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
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;
}
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;
}
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;
}
projects / lttng-tools.git / blobdiff