[lttng-tools.git] / src / bin / lttng-relayd / sessiond-trace-chunks.cpp

/*
 * Copyright (C) 2019 Jérémie Galarneau <jeremie.galarneau@efficios.com>
 *
 * SPDX-License-Identifier: GPL-2.0-only
 *
 */

#include "sessiond-trace-chunks.hpp"
#include <urcu.h>
#include <urcu/rculfhash.h>
#include <urcu/ref.h>
#include <common/macros.hpp>
#include <common/hashtable/hashtable.hpp>
#include <common/hashtable/utils.hpp>
#include <common/trace-chunk-registry.hpp>
#include <common/defaults.hpp>
#include <common/error.hpp>
#include <common/string-utils/format.hpp>
#include <stdio.h>
#include <inttypes.h>

/*
 * Lifetime of trace chunks within the relay daemon.
 *
 * Trace chunks are shared accross connections initiated from a given
 * session daemon. When a session is created by a consumer daemon, the
 * UUID of its associated session daemon is transmitted (in the case of
 * 2.11+ consumer daemons).
 *
 * The sessiond_trace_chunk_registry_new_session() and
 * sessiond_trace_chunk_registry_session_closed() methods create and
 * manage the reference count of lttng_trace_chunk_registry objects
 * associated to the various sessiond instances served by the relay daemon.
 *
 * When all sessions associated with a given sessiond instance are
 * destroyed, its registry is destroyed.
 *
 * lttng_trace_chunk objects are uniquely identified by the
 * (sessiond_uuid, sessiond_session_id, chunk_id) tuple. If a trace chunk
 * matching that tuple already exists, a new reference to the trace chunk
 * is acquired and it is returned to the caller. Otherwise, a new trace
 * chunk is created. This is how trace chunks are de-duplicated across
 * multiple consumer daemons managed by the same session daemon.
 *
 * Note that trace chunks are always added to their matching
 * lttng_trace_chunk_registry. They are automatically removed from the
 * trace chunk registry when their reference count reaches zero.
 */

/*
 * It is assumed that the sessiond_trace_chunk_registry is created and
 * destroyed by the same thread.
 */
struct sessiond_trace_chunk_registry {
	/* Maps an lttng_uuid to an lttng_trace_chunk_registry. */
	struct cds_lfht *ht;
};

namespace {
struct trace_chunk_registry_ht_key {
	lttng_uuid sessiond_uuid;
};

struct trace_chunk_registry_ht_element {
	struct trace_chunk_registry_ht_key key;
	struct urcu_ref ref;
	/* Node into the sessiond_trace_chunk_registry's hash table. */
	struct cds_lfht_node ht_node;
	/* Used for defered call_rcu reclaim. */
	struct rcu_head rcu_node;
	struct lttng_trace_chunk_registry *trace_chunk_registry;
	struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry;
};
} /* namespace */

static
unsigned long trace_chunk_registry_ht_key_hash(
		const struct trace_chunk_registry_ht_key *key)
{
	uint64_t uuid_h1 = ((uint64_t *) key->sessiond_uuid)[0];
	uint64_t uuid_h2 = ((uint64_t *) key->sessiond_uuid)[1];

	return hash_key_u64(&uuid_h1, lttng_ht_seed) ^
			hash_key_u64(&uuid_h2, lttng_ht_seed);
}

/* cds_lfht match function */
static
int trace_chunk_registry_ht_key_match(struct cds_lfht_node *node,
		const void *_key)
{
	const struct trace_chunk_registry_ht_key *key =
			(struct trace_chunk_registry_ht_key *) _key;
	struct trace_chunk_registry_ht_element *registry;

	registry = container_of(node, typeof(*registry), ht_node);
	return lttng_uuid_is_equal(key->sessiond_uuid,
			registry->key.sessiond_uuid);
}

static
void trace_chunk_registry_ht_element_free(struct rcu_head *node)
{
	struct trace_chunk_registry_ht_element *element =
			container_of(node, typeof(*element), rcu_node);

	free(element);
}

static
void trace_chunk_registry_ht_element_release(struct urcu_ref *ref)
{
	struct trace_chunk_registry_ht_element *element =
			container_of(ref, typeof(*element), ref);
	char uuid_str[LTTNG_UUID_STR_LEN];

	lttng_uuid_to_str(element->key.sessiond_uuid, uuid_str);

	DBG("Destroying trace chunk registry associated to sessiond {%s}",
			uuid_str);
	if (element->sessiond_trace_chunk_registry) {
		/* Unpublish. */
		rcu_read_lock();
		cds_lfht_del(element->sessiond_trace_chunk_registry->ht,
				&element->ht_node);
		rcu_read_unlock();
		element->sessiond_trace_chunk_registry = NULL;
	}

	lttng_trace_chunk_registry_destroy(element->trace_chunk_registry);
	/* Defered reclaim of the object */
	call_rcu(&element->rcu_node, trace_chunk_registry_ht_element_free);
}

static
bool trace_chunk_registry_ht_element_get(
		struct trace_chunk_registry_ht_element *element)
{
	return urcu_ref_get_unless_zero(&element->ref);
}

static
void trace_chunk_registry_ht_element_put(
		struct trace_chunk_registry_ht_element *element)
{
	if (!element) {
		return;
	}

	urcu_ref_put(&element->ref, trace_chunk_registry_ht_element_release);
}

/* Acquires a reference to the returned element on behalf of the caller. */
static
struct trace_chunk_registry_ht_element *trace_chunk_registry_ht_element_find(
		struct sessiond_trace_chunk_registry *sessiond_registry,
		const struct trace_chunk_registry_ht_key *key)
{
	struct trace_chunk_registry_ht_element *element = NULL;
	struct cds_lfht_node *node;
	struct cds_lfht_iter iter;

	rcu_read_lock();
	cds_lfht_lookup(sessiond_registry->ht,
			trace_chunk_registry_ht_key_hash(key),
			trace_chunk_registry_ht_key_match,
			key,
			&iter);
	node = cds_lfht_iter_get_node(&iter);
	if (node) {
		element = container_of(node, typeof(*element), ht_node);
		/*
		 * Only consider the look-up as successful if a reference
		 * could be acquired.
		 */
		if (!trace_chunk_registry_ht_element_get(element)) {
			element = NULL;
		}
	}
	rcu_read_unlock();
	return element;
}

static
int trace_chunk_registry_ht_element_create(
		struct sessiond_trace_chunk_registry *sessiond_registry,
		const struct trace_chunk_registry_ht_key *key)
{
	int ret = 0;
	struct trace_chunk_registry_ht_element *new_element;
	struct lttng_trace_chunk_registry *trace_chunk_registry;
	char uuid_str[LTTNG_UUID_STR_LEN];

	lttng_uuid_to_str(key->sessiond_uuid, uuid_str);

	trace_chunk_registry = lttng_trace_chunk_registry_create();
	if (!trace_chunk_registry) {
		ret = -1;
		goto end;
	}

	new_element = zmalloc<trace_chunk_registry_ht_element>();
	if (!new_element) {
		ret = -1;
		goto end;
	}

	memcpy(&new_element->key, key, sizeof(new_element->key));
	urcu_ref_init(&new_element->ref);
	cds_lfht_node_init(&new_element->ht_node);
	new_element->trace_chunk_registry = trace_chunk_registry;
	trace_chunk_registry = NULL;

	/* Attempt to publish the new element. */
	rcu_read_lock();
	while (1) {
		struct cds_lfht_node *published_node;
		struct trace_chunk_registry_ht_element *published_element;

		published_node = cds_lfht_add_unique(sessiond_registry->ht,
				trace_chunk_registry_ht_key_hash(&new_element->key),
				trace_chunk_registry_ht_key_match,
				&new_element->key,
				&new_element->ht_node);
		if (published_node == &new_element->ht_node) {
			/* New element published successfully. */
			DBG("Created trace chunk registry for sessiond {%s}",
					uuid_str);
			new_element->sessiond_trace_chunk_registry =
					sessiond_registry;
			break;
		}

		/*
		 * An equivalent element was published during the creation of
		 * this element. Attempt to acquire a reference to the one that
		 * was already published and release the reference to the copy
		 * we created if successful.
		 */
		published_element = container_of(published_node,
				typeof(*published_element), ht_node);
		if (trace_chunk_registry_ht_element_get(published_element)) {
			DBG("Acquired reference to trace chunk registry of sessiond {%s}",
					uuid_str);
			trace_chunk_registry_ht_element_put(new_element);
			new_element = NULL;
			break;
		}
		/*
		 * A reference to the previously published element could not
		 * be acquired. Hence, retry to publish our copy of the
		 * element.
		 */
	}
	rcu_read_unlock();
end:
	if (ret < 0) {
		ERR("Failed to create trace chunk registry for session daemon {%s}",
				uuid_str);
	}
	lttng_trace_chunk_registry_destroy(trace_chunk_registry);
	return ret;
}

struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry_create(void)
{
	struct sessiond_trace_chunk_registry *sessiond_registry =
			zmalloc<sessiond_trace_chunk_registry>();

	if (!sessiond_registry) {
		goto end;
	}

	sessiond_registry->ht = cds_lfht_new(DEFAULT_HT_SIZE,
			1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!sessiond_registry->ht) {
		goto error;
	}

end:
	return sessiond_registry;
error:
	sessiond_trace_chunk_registry_destroy(sessiond_registry);
	return NULL;
}

void sessiond_trace_chunk_registry_destroy(
		struct sessiond_trace_chunk_registry *sessiond_registry)
{
	int ret = cds_lfht_destroy(sessiond_registry->ht, NULL);

	LTTNG_ASSERT(!ret);
	free(sessiond_registry);
}

int sessiond_trace_chunk_registry_session_created(
		struct sessiond_trace_chunk_registry *sessiond_registry,
		const lttng_uuid sessiond_uuid)
{
	int ret = 0;
	struct trace_chunk_registry_ht_key key;
	struct trace_chunk_registry_ht_element *element;

	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);

	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	if (element) {
		char uuid_str[LTTNG_UUID_STR_LEN];

		lttng_uuid_to_str(sessiond_uuid, uuid_str);
		DBG("Acquired reference to trace chunk registry of sessiond {%s}",
				uuid_str);
		goto end;
	} else {
		ret = trace_chunk_registry_ht_element_create(
				sessiond_registry, &key);
	}
end:
	return ret;
}

int sessiond_trace_chunk_registry_session_destroyed(
		struct sessiond_trace_chunk_registry *sessiond_registry,
		const lttng_uuid sessiond_uuid)
{
	int ret = 0;
	struct trace_chunk_registry_ht_key key;
	struct trace_chunk_registry_ht_element *element;
	char uuid_str[LTTNG_UUID_STR_LEN];

	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);

	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	if (element) {
		DBG("Releasing reference to trace chunk registry of sessiond {%s}",
				uuid_str);
		/*
		 * Release the reference held by the session and the reference
		 * acquired through the "find" operation.
		 */
		trace_chunk_registry_ht_element_put(element);
		trace_chunk_registry_ht_element_put(element);
	} else {
		ERR("Failed to find trace chunk registry of sessiond {%s}",
				uuid_str);
		ret = -1;
	}
	return ret;
}

struct lttng_trace_chunk *sessiond_trace_chunk_registry_publish_chunk(
		struct sessiond_trace_chunk_registry *sessiond_registry,
		const lttng_uuid sessiond_uuid, uint64_t session_id,
		struct lttng_trace_chunk *new_chunk)
{
	enum lttng_trace_chunk_status status;
	uint64_t chunk_id;
	bool is_anonymous_chunk;
	struct trace_chunk_registry_ht_key key;
	struct trace_chunk_registry_ht_element *element = NULL;
	char uuid_str[LTTNG_UUID_STR_LEN];
	char chunk_id_str[MAX_INT_DEC_LEN(typeof(chunk_id))] = "-1";
	struct lttng_trace_chunk *published_chunk = NULL;
	bool trace_chunk_already_published;

	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);

	status = lttng_trace_chunk_get_id(new_chunk, &chunk_id);
	if (status == LTTNG_TRACE_CHUNK_STATUS_OK) {
		int ret;

		ret = snprintf(chunk_id_str, sizeof(chunk_id_str), "%" PRIu64,
				chunk_id);
		if (ret < 0) {
			lttng_strncpy(chunk_id_str, "-1", sizeof(chunk_id_str));
			WARN("Failed to format trace chunk id");
		}
		is_anonymous_chunk = false;
	} else if (status == LTTNG_TRACE_CHUNK_STATUS_NONE) {
		is_anonymous_chunk = true;
	} else {
		ERR("Failed to get trace chunk id");
		goto end;
	}

	DBG("Attempting to publish trace chunk: sessiond {%s}, session_id = "
			"%" PRIu64 ", chunk_id = %s",
			uuid_str, session_id,
			is_anonymous_chunk ? "anonymous" : chunk_id_str);

	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	if (!element) {
		ERR("Failed to find registry of sessiond {%s}", uuid_str);
		goto end;
	}

	published_chunk = lttng_trace_chunk_registry_publish_chunk(
			element->trace_chunk_registry, session_id, new_chunk,
			&trace_chunk_already_published);
	/*
	 * When the trace chunk is first published, two references to the
	 * published chunks exist. One is taken by the registry while the other
	 * is being returned to the caller. In the use case of the relay daemon,
	 * the reference held by the registry itself is undesirable.
	 *
	 * We want the trace chunk to be removed from the registry as soon
	 * as it is not being used by the relay daemon (through a session
	 * or a stream). This differs from the behaviour of the consumer
	 * daemon which relies on an explicit command from the session
	 * daemon to release the registry's reference.
	 *
	 * In cases where the trace chunk had already been published,
	 * the reference belonging to the sessiond trace chunk
	 * registry instance has already been 'put'. We simply return
	 * the published trace chunk with a reference taken on behalf of the
	 * caller.
	 */
	if (!trace_chunk_already_published) {
		lttng_trace_chunk_put(published_chunk);
	}
end:
	trace_chunk_registry_ht_element_put(element);
	return published_chunk;
}

struct lttng_trace_chunk *
sessiond_trace_chunk_registry_get_anonymous_chunk(
		struct sessiond_trace_chunk_registry *sessiond_registry,
		const lttng_uuid sessiond_uuid,
		uint64_t session_id)
{
	struct lttng_trace_chunk *chunk = NULL;
	struct trace_chunk_registry_ht_element *element;
	struct trace_chunk_registry_ht_key key;
	char uuid_str[LTTNG_UUID_STR_LEN];

	lttng_uuid_to_str(sessiond_uuid, uuid_str);

	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	if (!element) {
		ERR("Failed to find trace chunk registry of sessiond {%s}",
				uuid_str);
		goto end;
	}

	chunk = lttng_trace_chunk_registry_find_anonymous_chunk(
			element->trace_chunk_registry,
			session_id);
	trace_chunk_registry_ht_element_put(element);
end:
	return chunk;
}

struct lttng_trace_chunk *
sessiond_trace_chunk_registry_get_chunk(
		struct sessiond_trace_chunk_registry *sessiond_registry,
		const lttng_uuid sessiond_uuid,
		uint64_t session_id, uint64_t chunk_id)
{
	struct lttng_trace_chunk *chunk = NULL;
	struct trace_chunk_registry_ht_element *element;
	struct trace_chunk_registry_ht_key key;
	char uuid_str[LTTNG_UUID_STR_LEN];

	lttng_uuid_to_str(sessiond_uuid, uuid_str);

	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	if (!element) {
		ERR("Failed to find trace chunk registry of sessiond {%s}",
				uuid_str);
		goto end;
	}

	chunk = lttng_trace_chunk_registry_find_chunk(
			element->trace_chunk_registry,
			session_id, chunk_id);
	trace_chunk_registry_ht_element_put(element);
end:
	return chunk;
}

int sessiond_trace_chunk_registry_chunk_exists(
		struct sessiond_trace_chunk_registry *sessiond_registry,
		const lttng_uuid sessiond_uuid,
		uint64_t session_id, uint64_t chunk_id, bool *chunk_exists)
{
	int ret;
	struct trace_chunk_registry_ht_element *element;
	struct trace_chunk_registry_ht_key key;

	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	if (!element) {
		char uuid_str[LTTNG_UUID_STR_LEN];

		lttng_uuid_to_str(sessiond_uuid, uuid_str);
		/*
		 * While this certainly means that the chunk does not exist,
		 * it is unexpected for a chunk existence query to target a
		 * session daemon that does not have an active
		 * connection/registry. This would indicate a protocol
		 * (or internal) error.
		 */
		ERR("Failed to find trace chunk registry of sessiond {%s}",
				uuid_str);
		ret = -1;
		goto end;
	}

	ret = lttng_trace_chunk_registry_chunk_exists(
			element->trace_chunk_registry,
			session_id, chunk_id, chunk_exists);
	trace_chunk_registry_ht_element_put(element);
end:
	return ret;
}
Commit	Line	Data
0c1b0f77	1	/*
ab5be9fa	2	* Copyright (C) 2019 Jérémie Galarneau <jeremie.galarneau@efficios.com>
0c1b0f77	3	*
ab5be9fa	4	* SPDX-License-Identifier: GPL-2.0-only
0c1b0f77	5	*
0c1b0f77 JG	6	*/
0c1b0f77 JG	7
c9e313bc	8	#include "sessiond-trace-chunks.hpp"
0c1b0f77 JG	9	#include <urcu.h>
	10	#include <urcu/rculfhash.h>
	11	#include <urcu/ref.h>
c9e313bc SM	12	#include <common/macros.hpp>
	13	#include <common/hashtable/hashtable.hpp>
	14	#include <common/hashtable/utils.hpp>
	15	#include <common/trace-chunk-registry.hpp>
	16	#include <common/defaults.hpp>
	17	#include <common/error.hpp>
	18	#include <common/string-utils/format.hpp>
0c1b0f77 JG	19	#include <stdio.h>
	20	#include <inttypes.h>
	21
	22	/*
	23	* Lifetime of trace chunks within the relay daemon.
	24	*
	25	* Trace chunks are shared accross connections initiated from a given
	26	* session daemon. When a session is created by a consumer daemon, the
	27	* UUID of its associated session daemon is transmitted (in the case of
	28	* 2.11+ consumer daemons).
	29	*
	30	* The sessiond_trace_chunk_registry_new_session() and
	31	* sessiond_trace_chunk_registry_session_closed() methods create and
	32	* manage the reference count of lttng_trace_chunk_registry objects
	33	* associated to the various sessiond instances served by the relay daemon.
	34	*
	35	* When all sessions associated with a given sessiond instance are
	36	* destroyed, its registry is destroyed.
	37	*
	38	* lttng_trace_chunk objects are uniquely identified by the
	39	* (sessiond_uuid, sessiond_session_id, chunk_id) tuple. If a trace chunk
	40	* matching that tuple already exists, a new reference to the trace chunk
	41	* is acquired and it is returned to the caller. Otherwise, a new trace
	42	* chunk is created. This is how trace chunks are de-duplicated across
	43	* multiple consumer daemons managed by the same session daemon.
	44	*
	45	* Note that trace chunks are always added to their matching
	46	* lttng_trace_chunk_registry. They are automatically removed from the
	47	* trace chunk registry when their reference count reaches zero.
	48	*/
	49
	50	/*
	51	* It is assumed that the sessiond_trace_chunk_registry is created and
	52	* destroyed by the same thread.
	53	*/
	54	struct sessiond_trace_chunk_registry {
	55	/* Maps an lttng_uuid to an lttng_trace_chunk_registry. */
	56	struct cds_lfht *ht;
	57	};
	58
f1494934	59	namespace {
0c1b0f77 JG	60	struct trace_chunk_registry_ht_key {
	61	lttng_uuid sessiond_uuid;
	62	};
	63
	64	struct trace_chunk_registry_ht_element {
	65	struct trace_chunk_registry_ht_key key;
	66	struct urcu_ref ref;
	67	/* Node into the sessiond_trace_chunk_registry's hash table. */
	68	struct cds_lfht_node ht_node;
	69	/* Used for defered call_rcu reclaim. */
	70	struct rcu_head rcu_node;
	71	struct lttng_trace_chunk_registry *trace_chunk_registry;
	72	struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry;
	73	};
f1494934	74	} /* namespace */
0c1b0f77 JG	75
	76	static
	77	unsigned long trace_chunk_registry_ht_key_hash(
	78	const struct trace_chunk_registry_ht_key *key)
	79	{
	80	uint64_t uuid_h1 = ((uint64_t *) key->sessiond_uuid)[0];
	81	uint64_t uuid_h2 = ((uint64_t *) key->sessiond_uuid)[1];
	82
	83	return hash_key_u64(&uuid_h1, lttng_ht_seed) ^
	84	hash_key_u64(&uuid_h2, lttng_ht_seed);
	85	}
	86
	87	/* cds_lfht match function */
	88	static
	89	int trace_chunk_registry_ht_key_match(struct cds_lfht_node *node,
	90	const void *_key)
	91	{
	92	const struct trace_chunk_registry_ht_key *key =
	93	(struct trace_chunk_registry_ht_key *) _key;
	94	struct trace_chunk_registry_ht_element *registry;
	95
	96	registry = container_of(node, typeof(*registry), ht_node);
	97	return lttng_uuid_is_equal(key->sessiond_uuid,
	98	registry->key.sessiond_uuid);
	99	}
	100
	101	static
	102	void trace_chunk_registry_ht_element_free(struct rcu_head *node)
	103	{
	104	struct trace_chunk_registry_ht_element *element =
	105	container_of(node, typeof(*element), rcu_node);
	106
0c1b0f77 JG	107	free(element);
	108	}
	109
	110	static
	111	void trace_chunk_registry_ht_element_release(struct urcu_ref *ref)
	112	{
	113	struct trace_chunk_registry_ht_element *element =
	114	container_of(ref, typeof(*element), ref);
c70636a7	115	char uuid_str[LTTNG_UUID_STR_LEN];
0c1b0f77 JG	116
	117	lttng_uuid_to_str(element->key.sessiond_uuid, uuid_str);
	118
	119	DBG("Destroying trace chunk registry associated to sessiond {%s}",
	120	uuid_str);
	121	if (element->sessiond_trace_chunk_registry) {
	122	/* Unpublish. */
	123	rcu_read_lock();
	124	cds_lfht_del(element->sessiond_trace_chunk_registry->ht,
	125	&element->ht_node);
	126	rcu_read_unlock();
	127	element->sessiond_trace_chunk_registry = NULL;
	128	}
	129
c35f9726	130	lttng_trace_chunk_registry_destroy(element->trace_chunk_registry);
0c1b0f77 JG	131	/* Defered reclaim of the object */
	132	call_rcu(&element->rcu_node, trace_chunk_registry_ht_element_free);
	133	}
	134
	135	static
	136	bool trace_chunk_registry_ht_element_get(
	137	struct trace_chunk_registry_ht_element *element)
	138	{
	139	return urcu_ref_get_unless_zero(&element->ref);
	140	}
	141
	142	static
	143	void trace_chunk_registry_ht_element_put(
	144	struct trace_chunk_registry_ht_element *element)
	145	{
cd65fb86 FD	146	if (!element) {
	147	return;
	148	}
	149
0c1b0f77 JG	150	urcu_ref_put(&element->ref, trace_chunk_registry_ht_element_release);
	151	}
	152
	153	/* Acquires a reference to the returned element on behalf of the caller. */
	154	static
	155	struct trace_chunk_registry_ht_element *trace_chunk_registry_ht_element_find(
	156	struct sessiond_trace_chunk_registry *sessiond_registry,
	157	const struct trace_chunk_registry_ht_key *key)
	158	{
	159	struct trace_chunk_registry_ht_element *element = NULL;
	160	struct cds_lfht_node *node;
	161	struct cds_lfht_iter iter;
	162
	163	rcu_read_lock();
	164	cds_lfht_lookup(sessiond_registry->ht,
	165	trace_chunk_registry_ht_key_hash(key),
	166	trace_chunk_registry_ht_key_match,
	167	key,
	168	&iter);
	169	node = cds_lfht_iter_get_node(&iter);
	170	if (node) {
	171	element = container_of(node, typeof(*element), ht_node);
	172	/*
	173	* Only consider the look-up as successful if a reference
	174	* could be acquired.
	175	*/
	176	if (!trace_chunk_registry_ht_element_get(element)) {
	177	element = NULL;
	178	}
	179	}
	180	rcu_read_unlock();
	181	return element;
	182	}
	183
	184	static
	185	int trace_chunk_registry_ht_element_create(
	186	struct sessiond_trace_chunk_registry *sessiond_registry,
	187	const struct trace_chunk_registry_ht_key *key)
	188	{
	189	int ret = 0;
	190	struct trace_chunk_registry_ht_element *new_element;
	191	struct lttng_trace_chunk_registry *trace_chunk_registry;
c70636a7	192	char uuid_str[LTTNG_UUID_STR_LEN];
0c1b0f77 JG	193
	194	lttng_uuid_to_str(key->sessiond_uuid, uuid_str);
	195
	196	trace_chunk_registry = lttng_trace_chunk_registry_create();
	197	if (!trace_chunk_registry) {
	198	ret = -1;
	199	goto end;
	200	}
	201
64803277	202	new_element = zmalloc<trace_chunk_registry_ht_element>();
0c1b0f77 JG	203	if (!new_element) {
	204	ret = -1;
	205	goto end;
	206	}
	207
	208	memcpy(&new_element->key, key, sizeof(new_element->key));
	209	urcu_ref_init(&new_element->ref);
	210	cds_lfht_node_init(&new_element->ht_node);
	211	new_element->trace_chunk_registry = trace_chunk_registry;
e441f4e9	212	trace_chunk_registry = NULL;
0c1b0f77 JG	213
	214	/* Attempt to publish the new element. */
	215	rcu_read_lock();
	216	while (1) {
	217	struct cds_lfht_node *published_node;
	218	struct trace_chunk_registry_ht_element *published_element;
	219
	220	published_node = cds_lfht_add_unique(sessiond_registry->ht,
	221	trace_chunk_registry_ht_key_hash(&new_element->key),
	222	trace_chunk_registry_ht_key_match,
	223	&new_element->key,
	224	&new_element->ht_node);
	225	if (published_node == &new_element->ht_node) {
	226	/* New element published successfully. */
	227	DBG("Created trace chunk registry for sessiond {%s}",
	228	uuid_str);
	229	new_element->sessiond_trace_chunk_registry =
	230	sessiond_registry;
	231	break;
	232	}
	233
	234	/*
	235	* An equivalent element was published during the creation of
	236	* this element. Attempt to acquire a reference to the one that
	237	* was already published and release the reference to the copy
	238	* we created if successful.
	239	*/
	240	published_element = container_of(published_node,
	241	typeof(*published_element), ht_node);
	242	if (trace_chunk_registry_ht_element_get(published_element)) {
	243	DBG("Acquired reference to trace chunk registry of sessiond {%s}",
	244	uuid_str);
	245	trace_chunk_registry_ht_element_put(new_element);
	246	new_element = NULL;
	247	break;
	248	}
	249	/*
	250	* A reference to the previously published element could not
	251	* be acquired. Hence, retry to publish our copy of the
	252	* element.
	253	*/
	254	}
	255	rcu_read_unlock();
	256	end:
	257	if (ret < 0) {
	258	ERR("Failed to create trace chunk registry for session daemon {%s}",
	259	uuid_str);
	260	}
e441f4e9	261	lttng_trace_chunk_registry_destroy(trace_chunk_registry);
0c1b0f77 JG	262	return ret;
	263	}
	264
	265	struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry_create(void)
	266	{
	267	struct sessiond_trace_chunk_registry *sessiond_registry =
64803277	268	zmalloc<sessiond_trace_chunk_registry>();
0c1b0f77 JG	269
	270	if (!sessiond_registry) {
	271	goto end;
	272	}
	273
	274	sessiond_registry->ht = cds_lfht_new(DEFAULT_HT_SIZE,
	275	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	276	if (!sessiond_registry->ht) {
	277	goto error;
	278	}
	279
	280	end:
	281	return sessiond_registry;
	282	error:
	283	sessiond_trace_chunk_registry_destroy(sessiond_registry);
	284	return NULL;
	285	}
	286
	287	void sessiond_trace_chunk_registry_destroy(
	288	struct sessiond_trace_chunk_registry *sessiond_registry)
	289	{
	290	int ret = cds_lfht_destroy(sessiond_registry->ht, NULL);
	291
a0377dfe	292	LTTNG_ASSERT(!ret);
0c1b0f77 JG	293	free(sessiond_registry);
	294	}
	295
	296	int sessiond_trace_chunk_registry_session_created(
	297	struct sessiond_trace_chunk_registry *sessiond_registry,
	298	const lttng_uuid sessiond_uuid)
	299	{
	300	int ret = 0;
	301	struct trace_chunk_registry_ht_key key;
	302	struct trace_chunk_registry_ht_element *element;
	303
	304	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	305
	306	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	307	if (element) {
c70636a7	308	char uuid_str[LTTNG_UUID_STR_LEN];
0c1b0f77 JG	309
	310	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	311	DBG("Acquired reference to trace chunk registry of sessiond {%s}",
	312	uuid_str);
	313	goto end;
	314	} else {
	315	ret = trace_chunk_registry_ht_element_create(
	316	sessiond_registry, &key);
	317	}
	318	end:
	319	return ret;
	320	}
	321
	322	int sessiond_trace_chunk_registry_session_destroyed(
	323	struct sessiond_trace_chunk_registry *sessiond_registry,
	324	const lttng_uuid sessiond_uuid)
	325	{
	326	int ret = 0;
	327	struct trace_chunk_registry_ht_key key;
	328	struct trace_chunk_registry_ht_element *element;
c70636a7	329	char uuid_str[LTTNG_UUID_STR_LEN];
0c1b0f77 JG	330
	331	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	332	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	333
	334	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	335	if (element) {
	336	DBG("Releasing reference to trace chunk registry of sessiond {%s}",
	337	uuid_str);
	338	/*
	339	* Release the reference held by the session and the reference
	340	* acquired through the "find" operation.
	341	*/
	342	trace_chunk_registry_ht_element_put(element);
	343	trace_chunk_registry_ht_element_put(element);
	344	} else {
	345	ERR("Failed to find trace chunk registry of sessiond {%s}",
	346	uuid_str);
	347	ret = -1;
	348	}
	349	return ret;
	350	}
	351
	352	struct lttng_trace_chunk *sessiond_trace_chunk_registry_publish_chunk(
	353	struct sessiond_trace_chunk_registry *sessiond_registry,
	354	const lttng_uuid sessiond_uuid, uint64_t session_id,
	355	struct lttng_trace_chunk *new_chunk)
	356	{
	357	enum lttng_trace_chunk_status status;
	358	uint64_t chunk_id;
	359	bool is_anonymous_chunk;
	360	struct trace_chunk_registry_ht_key key;
	361	struct trace_chunk_registry_ht_element *element = NULL;
c70636a7	362	char uuid_str[LTTNG_UUID_STR_LEN];
0c1b0f77 JG	363	char chunk_id_str[MAX_INT_DEC_LEN(typeof(chunk_id))] = "-1";
0c1b0f77 JG	364	struct lttng_trace_chunk *published_chunk = NULL;
c5c79321	365	bool trace_chunk_already_published;
0c1b0f77 JG	366
	367	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	368	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	369
	370	status = lttng_trace_chunk_get_id(new_chunk, &chunk_id);
22df7435	371	if (status == LTTNG_TRACE_CHUNK_STATUS_OK) {
0c1b0f77 JG	372	int ret;
	373
	374	ret = snprintf(chunk_id_str, sizeof(chunk_id_str), "%" PRIu64,
	375	chunk_id);
	376	if (ret < 0) {
	377	lttng_strncpy(chunk_id_str, "-1", sizeof(chunk_id_str));
	378	WARN("Failed to format trace chunk id");
	379	}
	380	is_anonymous_chunk = false;
22df7435	381	} else if (status == LTTNG_TRACE_CHUNK_STATUS_NONE) {
0c1b0f77 JG	382	is_anonymous_chunk = true;
	383	} else {
	384	ERR("Failed to get trace chunk id");
	385	goto end;
	386	}
	387
22df7435	388	DBG("Attempting to publish trace chunk: sessiond {%s}, session_id = "
0c1b0f77 JG	389	"%" PRIu64 ", chunk_id = %s",
	390	uuid_str, session_id,
	391	is_anonymous_chunk ? "anonymous" : chunk_id_str);
	392
22df7435	393	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
0c1b0f77 JG	394	if (!element) {
	395	ERR("Failed to find registry of sessiond {%s}", uuid_str);
	396	goto end;
	397	}
	398
22df7435	399	published_chunk = lttng_trace_chunk_registry_publish_chunk(
c5c79321 JG	400	element->trace_chunk_registry, session_id, new_chunk,
c5c79321 JG	401	&trace_chunk_already_published);
c35f9726	402	/*
c5c79321 JG	403	* When the trace chunk is first published, two references to the
	404	* published chunks exist. One is taken by the registry while the other
	405	* is being returned to the caller. In the use case of the relay daemon,
	406	* the reference held by the registry itself is undesirable.
c35f9726 JG	407	*
	408	* We want the trace chunk to be removed from the registry as soon
	409	* as it is not being used by the relay daemon (through a session
	410	* or a stream). This differs from the behaviour of the consumer
	411	* daemon which relies on an explicit command from the session
	412	* daemon to release the registry's reference.
c5c79321 JG	413	*
	414	* In cases where the trace chunk had already been published,
	415	* the reference belonging to the sessiond trace chunk
	416	* registry instance has already been 'put'. We simply return
	417	* the published trace chunk with a reference taken on behalf of the
	418	* caller.
c35f9726	419	*/
c5c79321 JG	420	if (!trace_chunk_already_published) {
	421	lttng_trace_chunk_put(published_chunk);
	422	}
0c1b0f77 JG	423	end:
	424	trace_chunk_registry_ht_element_put(element);
	425	return published_chunk;
	426	}
	427
	428	struct lttng_trace_chunk *
	429	sessiond_trace_chunk_registry_get_anonymous_chunk(
	430	struct sessiond_trace_chunk_registry *sessiond_registry,
	431	const lttng_uuid sessiond_uuid,
	432	uint64_t session_id)
	433	{
	434	struct lttng_trace_chunk *chunk = NULL;
	435	struct trace_chunk_registry_ht_element *element;
	436	struct trace_chunk_registry_ht_key key;
c70636a7	437	char uuid_str[LTTNG_UUID_STR_LEN];
0c1b0f77 JG	438
	439	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	440
	441	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	442	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	443	if (!element) {
	444	ERR("Failed to find trace chunk registry of sessiond {%s}",
	445	uuid_str);
	446	goto end;
	447	}
	448
	449	chunk = lttng_trace_chunk_registry_find_anonymous_chunk(
	450	element->trace_chunk_registry,
	451	session_id);
	452	trace_chunk_registry_ht_element_put(element);
	453	end:
	454	return chunk;
	455	}
	456
	457	struct lttng_trace_chunk *
	458	sessiond_trace_chunk_registry_get_chunk(
	459	struct sessiond_trace_chunk_registry *sessiond_registry,
	460	const lttng_uuid sessiond_uuid,
	461	uint64_t session_id, uint64_t chunk_id)
	462	{
	463	struct lttng_trace_chunk *chunk = NULL;
	464	struct trace_chunk_registry_ht_element *element;
	465	struct trace_chunk_registry_ht_key key;
c70636a7	466	char uuid_str[LTTNG_UUID_STR_LEN];
0c1b0f77 JG	467
	468	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	469
	470	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	471	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	472	if (!element) {
	473	ERR("Failed to find trace chunk registry of sessiond {%s}",
	474	uuid_str);
	475	goto end;
	476	}
	477
	478	chunk = lttng_trace_chunk_registry_find_chunk(
	479	element->trace_chunk_registry,
	480	session_id, chunk_id);
	481	trace_chunk_registry_ht_element_put(element);
	482	end:
	483	return chunk;
	484	}
6b584c2e JG	485
	486	int sessiond_trace_chunk_registry_chunk_exists(
	487	struct sessiond_trace_chunk_registry *sessiond_registry,
	488	const lttng_uuid sessiond_uuid,
	489	uint64_t session_id, uint64_t chunk_id, bool *chunk_exists)
	490	{
	491	int ret;
	492	struct trace_chunk_registry_ht_element *element;
	493	struct trace_chunk_registry_ht_key key;
	494
	495	lttng_uuid_copy(key.sessiond_uuid, sessiond_uuid);
	496	element = trace_chunk_registry_ht_element_find(sessiond_registry, &key);
	497	if (!element) {
c70636a7	498	char uuid_str[LTTNG_UUID_STR_LEN];
6b584c2e JG	499
	500	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	501	/*
	502	* While this certainly means that the chunk does not exist,
	503	* it is unexpected for a chunk existence query to target a
	504	* session daemon that does not have an active
	505	* connection/registry. This would indicate a protocol
	506	* (or internal) error.
	507	*/
	508	ERR("Failed to find trace chunk registry of sessiond {%s}",
	509	uuid_str);
	510	ret = -1;
	511	goto end;
	512	}
	513
	514	ret = lttng_trace_chunk_registry_chunk_exists(
	515	element->trace_chunk_registry,
	516	session_id, chunk_id, chunk_exists);
	517	trace_chunk_registry_ht_element_put(element);
	518	end:
	519	return ret;
	520	}