[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.h"
#include <urcu.h>
#include <urcu/rculfhash.h>
#include <urcu/ref.h>
#include <common/macros.h>
#include <common/hashtable/hashtable.h>
#include <common/hashtable/utils.h>
#include <common/trace-chunk-registry.h>
#include <common/defaults.h>
#include <common/error.h>
#include <common/string-utils/format.h>
#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;
};

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

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