[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)
{
	const uint64_t uuid_h1 = *reinterpret_cast<const uint64_t *>(&key->sessiond_uuid[0]);
	const uint64_t uuid_h2 = *reinterpret_cast<const 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 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;

	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);
	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);
	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);

	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);

	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;

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