{
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);
}
{
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, 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);
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,
}
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;
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,
{
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;
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_rcu_unlock;
+ }
+ 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;