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

/*
 * Copyright (C) 2019 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License, version 2 only,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#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[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)
{
	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[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(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;

	/* 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);
	}
	return ret;
}

struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry_create(void)
{
	struct sessiond_trace_chunk_registry *sessiond_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);

	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[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[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[UUID_STR_LEN];
	char chunk_id_str[MAX_INT_DEC_LEN(typeof(chunk_id))] = "-1";
	struct lttng_trace_chunk *published_chunk = NULL;

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