[lttng-tools.git] / src / bin / lttng-relayd / session.c

/*
 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
 *                      David Goulet <dgoulet@efficios.com>
 *               2015 - Mathieu Desnoyers <mathieu.desnoyers@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.
 */

#define _LGPL_SOURCE
#include <common/common.h>
#include <common/utils.h>
#include <common/compat/uuid.h>
#include <urcu/rculist.h>

#include "ctf-trace.h"
#include "lttng-relayd.h"
#include "session.h"
#include "sessiond-trace-chunks.h"
#include "stream.h"
#include <common/defaults.h>

/* Global session id used in the session creation. */
static uint64_t last_relay_session_id;
static pthread_mutex_t last_relay_session_id_lock = PTHREAD_MUTEX_INITIALIZER;

static int session_set_anonymous_chunk(struct relay_session *session)
{
	int ret = 0;
	struct lttng_trace_chunk *chunk = NULL;
	enum lttng_trace_chunk_status status;
	struct lttng_directory_handle output_directory;
	char *output_path;
	bool output_path_allocated = false;

	if (!opt_output_path) {
		/* No output path defined */
		const char *home_dir = utils_get_home_dir();
		if (!home_dir) {
			ERR("Home path not found."
					" Please specify an output path using -o, --output PATH");
			ret = -1;
			goto end;
		}
		ret = asprintf(&output_path, "%s/%s", home_dir, DEFAULT_TRACE_DIR_NAME);
		if (ret < 0) {
			PERROR("asprintf trace dir name");
			ret = -1;
			goto end;
		}
		output_path_allocated = true;
	} else {
		output_path = opt_output_path;
		output_path_allocated = false;
	}

	ret = lttng_directory_handle_init(&output_directory, output_path);
	if (ret) {
		goto end;
	}

	chunk = lttng_trace_chunk_create_anonymous();
	if (!chunk) {
		goto end;
	}

	status = lttng_trace_chunk_set_credentials_current_user(chunk);
	if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
		ret = -1;
		goto end;
	}

	status = lttng_trace_chunk_set_as_owner(chunk, &output_directory);
	if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
		ret = -1;
		goto end;
	}
	session->current_trace_chunk = chunk;
	chunk = NULL;
end:
	lttng_trace_chunk_put(chunk);
	lttng_directory_handle_fini(&output_directory);
	if (output_path_allocated) {
		free(output_path);
	}
	return ret;
}

/*
 * Create a new session by assigning a new session ID.
 *
 * Return allocated session or else NULL.
 */
struct relay_session *session_create(const char *session_name,
		const char *hostname, const char *base_path,
		uint32_t live_timer,
		bool snapshot,
		const lttng_uuid sessiond_uuid,
		const uint64_t *id_sessiond,
		const uint64_t *current_chunk_id,
		const time_t *creation_time,
		uint32_t major,
		uint32_t minor,
		bool session_name_contains_creation_time)
{
	int ret;
	struct relay_session *session = NULL;

	if (session_name && strstr(session_name, ".")) {
		ERR("Illegal character in session name: \"%s\"",
				session_name);
		goto error;
	}
	if (base_path && strstr(base_path, "../")) {
		ERR("Invalid session base path walks up the path hierarchy: \"%s\"",
				base_path);
		goto error;
	}
	if (hostname && strstr(hostname, ".")) {
		ERR("Invalid character in hostname: \"%s\"",
				hostname);
		goto error;
	}

	session = zmalloc(sizeof(*session));
	if (!session) {
		PERROR("Failed to allocate session");
		goto error;
	}
	if (lttng_strncpy(session->session_name, session_name,
			sizeof(session->session_name))) {
	        WARN("Session name exceeds maximal allowed length");
		goto error;
	}
	if (lttng_strncpy(session->hostname, hostname,
			sizeof(session->hostname))) {
		WARN("Hostname exceeds maximal allowed length");
		goto error;
	}
	if (lttng_strncpy(session->base_path, base_path,
			sizeof(session->base_path))) {
		WARN("Base path exceeds maximal allowed length");
		goto error;
	}
	if (creation_time) {
		LTTNG_OPTIONAL_SET(&session->creation_time, *creation_time);
	}
	session->session_name_contains_creation_time =
			session_name_contains_creation_time;

	session->ctf_traces_ht = lttng_ht_new(0, LTTNG_HT_TYPE_STRING);
	if (!session->ctf_traces_ht) {
		goto error;
	}

	pthread_mutex_lock(&last_relay_session_id_lock);
	session->id = ++last_relay_session_id;
	pthread_mutex_unlock(&last_relay_session_id_lock);

	session->major = major;
	session->minor = minor;
	lttng_ht_node_init_u64(&session->session_n, session->id);
	urcu_ref_init(&session->ref);
	CDS_INIT_LIST_HEAD(&session->recv_list);
	pthread_mutex_init(&session->lock, NULL);
	pthread_mutex_init(&session->recv_list_lock, NULL);

	session->live_timer = live_timer;
	session->snapshot = snapshot;
	lttng_uuid_copy(session->sessiond_uuid, sessiond_uuid);

	if (id_sessiond) {
		LTTNG_OPTIONAL_SET(&session->id_sessiond, *id_sessiond);
	}

	ret = sessiond_trace_chunk_registry_session_created(
			sessiond_trace_chunk_registry, sessiond_uuid);
	if (ret) {
		goto error;
	}

	if (id_sessiond && current_chunk_id) {
		session->current_trace_chunk =
				sessiond_trace_chunk_registry_get_chunk(
					sessiond_trace_chunk_registry,
					session->sessiond_uuid,
					session->id_sessiond.value,
					*current_chunk_id);
		if (!session->current_trace_chunk) {
		        char uuid_str[UUID_STR_LEN];

			lttng_uuid_to_str(sessiond_uuid, uuid_str);
			ERR("Could not find trace chunk: sessiond = {%s}, sessiond session id = %" PRIu64 ", trace chunk id = %" PRIu64,
					uuid_str, *id_sessiond,
					*current_chunk_id);
                }
	} else if (!id_sessiond) {
		/*
		 * Pre-2.11 peers will not announce trace chunks. An
		 * anonymous trace chunk which will remain set for the
		 * duration of the session is created.
		 */
		ret = session_set_anonymous_chunk(session);
		if (ret) {
			goto error;
		}
	}

	lttng_ht_add_unique_u64(sessions_ht, &session->session_n);
	return session;

error:
	session_put(session);
	return NULL;
}

/* Should be called with RCU read-side lock held. */
bool session_get(struct relay_session *session)
{
	return urcu_ref_get_unless_zero(&session->ref);
}

/*
 * Lookup a session within the session hash table using the session id
 * as key. A session reference is taken when a session is returned.
 * session_put() must be called on that session.
 *
 * Return session or NULL if not found.
 */
struct relay_session *session_get_by_id(uint64_t id)
{
	struct relay_session *session = NULL;
	struct lttng_ht_node_u64 *node;
	struct lttng_ht_iter iter;

	rcu_read_lock();
	lttng_ht_lookup(sessions_ht, &id, &iter);
	node = lttng_ht_iter_get_node_u64(&iter);
	if (!node) {
		DBG("Session find by ID %" PRIu64 " id NOT found", id);
		goto end;
	}
	session = caa_container_of(node, struct relay_session, session_n);
	DBG("Session find by ID %" PRIu64 " id found", id);
	if (!session_get(session)) {
		session = NULL;
	}
end:
	rcu_read_unlock();
	return session;
}

static void rcu_destroy_session(struct rcu_head *rcu_head)
{
	struct relay_session *session =
			caa_container_of(rcu_head, struct relay_session,
				rcu_node);
	/*
	 * Since each trace has a reference on the session, it means
	 * that if we are at the point where we teardown the session, no
	 * trace belonging to that session exist at this point.
	 * Calling lttng_ht_destroy in call_rcu worker thread so we
	 * don't hold the RCU read-side lock while calling it.
	 */
	lttng_ht_destroy(session->ctf_traces_ht);
	free(session);
}

/*
 * Delete session from the given hash table.
 *
 * Return lttng ht del error code being 0 on success and 1 on failure.
 */
static int session_delete(struct relay_session *session)
{
	struct lttng_ht_iter iter;

	iter.iter.node = &session->session_n.node;
	return lttng_ht_del(sessions_ht, &iter);
}


static void destroy_session(struct relay_session *session)
{
	int ret;

	ret = session_delete(session);
	assert(!ret);
	lttng_trace_chunk_put(session->current_trace_chunk);
	session->current_trace_chunk = NULL;
	lttng_trace_chunk_put(session->pending_closure_trace_chunk);
	session->pending_closure_trace_chunk = NULL;
	ret = sessiond_trace_chunk_registry_session_destroyed(
			sessiond_trace_chunk_registry, session->sessiond_uuid);
	assert(!ret);
	call_rcu(&session->rcu_node, rcu_destroy_session);
}

void session_release(struct urcu_ref *ref)
{
	struct relay_session *session =
			caa_container_of(ref, struct relay_session, ref);

	destroy_session(session);
}

void session_put(struct relay_session *session)
{
	rcu_read_lock();
	urcu_ref_put(&session->ref, session_release);
	rcu_read_unlock();
}

int session_close(struct relay_session *session)
{
	int ret = 0;
	struct ctf_trace *trace;
	struct lttng_ht_iter iter;
	struct relay_stream *stream;

	pthread_mutex_lock(&session->lock);
	DBG("closing session %" PRIu64 ": is conn already closed %d",
			session->id, session->connection_closed);
	session->connection_closed = true;
	pthread_mutex_unlock(&session->lock);

	rcu_read_lock();
	cds_lfht_for_each_entry(session->ctf_traces_ht->ht,
			&iter.iter, trace, node.node) {
		ret = ctf_trace_close(trace);
		if (ret) {
			goto rcu_unlock;
		}
	}
	cds_list_for_each_entry_rcu(stream, &session->recv_list,
			recv_node) {
		/* Close streams which have not been published yet. */
		try_stream_close(stream);
	}
rcu_unlock:
	rcu_read_unlock();
	if (ret) {
		return ret;
	}
	/* Put self-reference from create. */
	session_put(session);
	return ret;
}

int session_abort(struct relay_session *session)
{
	int ret = 0;

	if (!session) {
		return 0;
	}

	pthread_mutex_lock(&session->lock);
	DBG("aborting session %" PRIu64, session->id);
	session->aborted = true;
	pthread_mutex_unlock(&session->lock);
	return ret;
}

void print_sessions(void)
{
	struct lttng_ht_iter iter;
	struct relay_session *session;

	if (!sessions_ht) {
		return;
	}

	rcu_read_lock();
	cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, session,
			session_n.node) {
		if (!session_get(session)) {
			continue;
		}
		DBG("session %p refcount %ld session %" PRIu64,
			session,
			session->ref.refcount,
			session->id);
		session_put(session);
	}
	rcu_read_unlock();
}
Commit	Line	Data
2f8f53af DG	1	/*
	2	* Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
	3	* David Goulet <dgoulet@efficios.com>
7591bab1	4	* 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
2f8f53af DG	5	*
	6	* This program is free software; you can redistribute it and/or modify it
	7	* under the terms of the GNU General Public License, version 2 only, as
	8	* published by the Free Software Foundation.
	9	*
	10	* This program is distributed in the hope that it will be useful, but WITHOUT
	11	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	12	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	13	* more details.
	14	*
	15	* You should have received a copy of the GNU General Public License along with
	16	* this program; if not, write to the Free Software Foundation, Inc., 51
	17	* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	18	*/
	19
6c1c0768	20	#define _LGPL_SOURCE
2a174661	21	#include <common/common.h>
a620c891	22	#include <common/utils.h>
1e791a74	23	#include <common/compat/uuid.h>
7591bab1	24	#include <urcu/rculist.h>
2a174661 DG	25
2a174661 DG	26	#include "ctf-trace.h"
a620c891	27	#include "lttng-relayd.h"
2f8f53af	28	#include "session.h"
23c8ff50	29	#include "sessiond-trace-chunks.h"
a620c891	30	#include "stream.h"
a0409c33	31	#include <common/defaults.h>
2a174661 DG	32
	33	/* Global session id used in the session creation. */
	34	static uint64_t last_relay_session_id;
7591bab1	35	static pthread_mutex_t last_relay_session_id_lock = PTHREAD_MUTEX_INITIALIZER;
2a174661	36
1e791a74 JG	37	static int session_set_anonymous_chunk(struct relay_session *session)
	38	{
	39	int ret = 0;
	40	struct lttng_trace_chunk *chunk = NULL;
	41	enum lttng_trace_chunk_status status;
	42	struct lttng_directory_handle output_directory;
a0409c33 MD	43	char *output_path;
a0409c33 MD	44	bool output_path_allocated = false;
a620c891	45
a0409c33	46	if (!opt_output_path) {
a620c891	47	/* No output path defined */
a0409c33 MD	48	const char *home_dir = utils_get_home_dir();
a0409c33 MD	49	if (!home_dir) {
334dfcb7 JG	50	ERR("Home path not found."
334dfcb7 JG	51	" Please specify an output path using -o, --output PATH");
a620c891 JR	52	ret = -1;
	53	goto end;
	54	}
a0409c33 MD	55	ret = asprintf(&output_path, "%s/%s", home_dir, DEFAULT_TRACE_DIR_NAME);
	56	if (ret < 0) {
	57	PERROR("asprintf trace dir name");
	58	ret = -1;
	59	goto end;
	60	}
	61	output_path_allocated = true;
	62	} else {
	63	output_path = opt_output_path;
	64	output_path_allocated = false;
a620c891	65	}
1e791a74	66
a0409c33	67	ret = lttng_directory_handle_init(&output_directory, output_path);
1e791a74 JG	68	if (ret) {
	69	goto end;
	70	}
	71
	72	chunk = lttng_trace_chunk_create_anonymous();
	73	if (!chunk) {
	74	goto end;
	75	}
	76
	77	status = lttng_trace_chunk_set_credentials_current_user(chunk);
	78	if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
	79	ret = -1;
	80	goto end;
	81	}
	82
	83	status = lttng_trace_chunk_set_as_owner(chunk, &output_directory);
	84	if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
	85	ret = -1;
	86	goto end;
	87	}
	88	session->current_trace_chunk = chunk;
	89	chunk = NULL;
	90	end:
	91	lttng_trace_chunk_put(chunk);
	92	lttng_directory_handle_fini(&output_directory);
a0409c33 MD	93	if (output_path_allocated) {
	94	free(output_path);
	95	}
1e791a74 JG	96	return ret;
	97	}
	98
2a174661 DG	99	/*
	100	* Create a new session by assigning a new session ID.
	101	*
	102	* Return allocated session or else NULL.
	103	*/
7591bab1	104	struct relay_session session_create(const char session_name,
6fa5fe7c	105	const char hostname, const char base_path,
db1da059 JG	106	uint32_t live_timer,
	107	bool snapshot,
	108	const lttng_uuid sessiond_uuid,
	109	const uint64_t *id_sessiond,
	110	const uint64_t *current_chunk_id,
	111	const time_t *creation_time,
	112	uint32_t major,
46ef2188 MD	113	uint32_t minor,
46ef2188 MD	114	bool session_name_contains_creation_time)
2a174661	115	{
23c8ff50	116	int ret;
590f0324 JG	117	struct relay_session *session = NULL;
	118
	119	if (session_name && strstr(session_name, ".")) {
	120	ERR("Illegal character in session name: \"%s\"",
	121	session_name);
	122	goto error;
	123	}
	124	if (base_path && strstr(base_path, "../")) {
	125	ERR("Invalid session base path walks up the path hierarchy: \"%s\"",
	126	base_path);
	127	goto error;
	128	}
	129	if (hostname && strstr(hostname, ".")) {
	130	ERR("Invalid character in hostname: \"%s\"",
	131	hostname);
	132	goto error;
	133	}
2a174661 DG	134
	135	session = zmalloc(sizeof(*session));
	136	if (!session) {
1e791a74	137	PERROR("Failed to allocate session");
2a174661 DG	138	goto error;
2a174661 DG	139	}
bb5d54e7 MD	140	if (lttng_strncpy(session->session_name, session_name,
bb5d54e7 MD	141	sizeof(session->session_name))) {
1e791a74	142	WARN("Session name exceeds maximal allowed length");
bb5d54e7 MD	143	goto error;
	144	}
	145	if (lttng_strncpy(session->hostname, hostname,
	146	sizeof(session->hostname))) {
1e791a74	147	WARN("Hostname exceeds maximal allowed length");
bb5d54e7 MD	148	goto error;
bb5d54e7 MD	149	}
6fa5fe7c MD	150	if (lttng_strncpy(session->base_path, base_path,
	151	sizeof(session->base_path))) {
	152	WARN("Base path exceeds maximal allowed length");
	153	goto error;
	154	}
46ef2188 MD	155	if (creation_time) {
	156	LTTNG_OPTIONAL_SET(&session->creation_time, *creation_time);
	157	}
	158	session->session_name_contains_creation_time =
	159	session_name_contains_creation_time;
6fa5fe7c	160
2a174661 DG	161	session->ctf_traces_ht = lttng_ht_new(0, LTTNG_HT_TYPE_STRING);
2a174661 DG	162	if (!session->ctf_traces_ht) {
2a174661 DG	163	goto error;
	164	}
	165
7591bab1	166	pthread_mutex_lock(&last_relay_session_id_lock);
2a174661	167	session->id = ++last_relay_session_id;
7591bab1 MD	168	pthread_mutex_unlock(&last_relay_session_id_lock);
	169
	170	session->major = major;
	171	session->minor = minor;
2a174661	172	lttng_ht_node_init_u64(&session->session_n, session->id);
7591bab1 MD	173	urcu_ref_init(&session->ref);
	174	CDS_INIT_LIST_HEAD(&session->recv_list);
	175	pthread_mutex_init(&session->lock, NULL);
7591bab1 MD	176	pthread_mutex_init(&session->recv_list_lock, NULL);
7591bab1 MD	177
7591bab1 MD	178	session->live_timer = live_timer;
7591bab1 MD	179	session->snapshot = snapshot;
23c8ff50 JG	180	lttng_uuid_copy(session->sessiond_uuid, sessiond_uuid);
23c8ff50 JG	181
1e791a74 JG	182	if (id_sessiond) {
	183	LTTNG_OPTIONAL_SET(&session->id_sessiond, *id_sessiond);
	184	}
	185
23c8ff50 JG	186	ret = sessiond_trace_chunk_registry_session_created(
	187	sessiond_trace_chunk_registry, sessiond_uuid);
	188	if (ret) {
	189	goto error;
	190	}
7591bab1	191
1e791a74 JG	192	if (id_sessiond && current_chunk_id) {
	193	session->current_trace_chunk =
	194	sessiond_trace_chunk_registry_get_chunk(
	195	sessiond_trace_chunk_registry,
	196	session->sessiond_uuid,
	197	session->id_sessiond.value,
	198	*current_chunk_id);
	199	if (!session->current_trace_chunk) {
	200	char uuid_str[UUID_STR_LEN];
	201
	202	lttng_uuid_to_str(sessiond_uuid, uuid_str);
	203	ERR("Could not find trace chunk: sessiond = {%s}, sessiond session id = %" PRIu64 ", trace chunk id = %" PRIu64,
	204	uuid_str, *id_sessiond,
	205	*current_chunk_id);
	206	}
	207	} else if (!id_sessiond) {
	208	/*
	209	* Pre-2.11 peers will not announce trace chunks. An
	210	* anonymous trace chunk which will remain set for the
	211	* duration of the session is created.
	212	*/
	213	ret = session_set_anonymous_chunk(session);
	214	if (ret) {
	215	goto error;
	216	}
	217	}
	218
7591bab1	219	lttng_ht_add_unique_u64(sessions_ht, &session->session_n);
bb5d54e7	220	return session;
2a174661 DG	221
2a174661 DG	222	error:
1e791a74	223	session_put(session);
bb5d54e7	224	return NULL;
2a174661	225	}
2f8f53af	226
7591bab1 MD	227	/* Should be called with RCU read-side lock held. */
	228	bool session_get(struct relay_session *session)
	229	{
ce4d4083	230	return urcu_ref_get_unless_zero(&session->ref);
7591bab1 MD	231	}
7591bab1 MD	232
2f8f53af	233	/*
7591bab1 MD	234	* Lookup a session within the session hash table using the session id
	235	* as key. A session reference is taken when a session is returned.
	236	* session_put() must be called on that session.
2f8f53af DG	237	*
	238	* Return session or NULL if not found.
	239	*/
7591bab1	240	struct relay_session *session_get_by_id(uint64_t id)
2f8f53af DG	241	{
2f8f53af DG	242	struct relay_session *session = NULL;
2a174661	243	struct lttng_ht_node_u64 *node;
2f8f53af DG	244	struct lttng_ht_iter iter;
2f8f53af DG	245
7591bab1 MD	246	rcu_read_lock();
7591bab1 MD	247	lttng_ht_lookup(sessions_ht, &id, &iter);
2a174661	248	node = lttng_ht_iter_get_node_u64(&iter);
2f8f53af	249	if (!node) {
2a174661	250	DBG("Session find by ID %" PRIu64 " id NOT found", id);
2f8f53af DG	251	goto end;
	252	}
	253	session = caa_container_of(node, struct relay_session, session_n);
2a174661	254	DBG("Session find by ID %" PRIu64 " id found", id);
7591bab1 MD	255	if (!session_get(session)) {
	256	session = NULL;
	257	}
2f8f53af	258	end:
7591bab1	259	rcu_read_unlock();
2f8f53af DG	260	return session;
2f8f53af DG	261	}
2a174661	262
7591bab1 MD	263	static void rcu_destroy_session(struct rcu_head *rcu_head)
	264	{
	265	struct relay_session *session =
	266	caa_container_of(rcu_head, struct relay_session,
	267	rcu_node);
49e614cb MD	268	/*
	269	* Since each trace has a reference on the session, it means
	270	* that if we are at the point where we teardown the session, no
	271	* trace belonging to that session exist at this point.
	272	* Calling lttng_ht_destroy in call_rcu worker thread so we
	273	* don't hold the RCU read-side lock while calling it.
	274	*/
	275	lttng_ht_destroy(session->ctf_traces_ht);
7591bab1 MD	276	free(session);
	277	}
	278
2a174661 DG	279	/*
	280	* Delete session from the given hash table.
	281	*
	282	* Return lttng ht del error code being 0 on success and 1 on failure.
	283	*/
7591bab1	284	static int session_delete(struct relay_session *session)
2a174661 DG	285	{
	286	struct lttng_ht_iter iter;
	287
2a174661	288	iter.iter.node = &session->session_n.node;
7591bab1	289	return lttng_ht_del(sessions_ht, &iter);
2a174661 DG	290	}
2a174661 DG	291
7591bab1 MD	292
	293	static void destroy_session(struct relay_session *session)
	294	{
	295	int ret;
	296
	297	ret = session_delete(session);
	298	assert(!ret);
639ddf68	299	lttng_trace_chunk_put(session->current_trace_chunk);
c35f9726	300	session->current_trace_chunk = NULL;
62bad3bf JG	301	lttng_trace_chunk_put(session->pending_closure_trace_chunk);
62bad3bf JG	302	session->pending_closure_trace_chunk = NULL;
23c8ff50 JG	303	ret = sessiond_trace_chunk_registry_session_destroyed(
	304	sessiond_trace_chunk_registry, session->sessiond_uuid);
	305	assert(!ret);
7591bab1 MD	306	call_rcu(&session->rcu_node, rcu_destroy_session);
	307	}
	308
	309	void session_release(struct urcu_ref *ref)
2a174661	310	{
7591bab1 MD	311	struct relay_session *session =
7591bab1 MD	312	caa_container_of(ref, struct relay_session, ref);
2a174661	313
7591bab1 MD	314	destroy_session(session);
7591bab1 MD	315	}
2a174661	316
7591bab1 MD	317	void session_put(struct relay_session *session)
	318	{
	319	rcu_read_lock();
7591bab1	320	urcu_ref_put(&session->ref, session_release);
7591bab1	321	rcu_read_unlock();
2a174661 DG	322	}
2a174661 DG	323
7591bab1	324	int session_close(struct relay_session *session)
2a174661 DG	325	{
2a174661 DG	326	int ret = 0;
7591bab1 MD	327	struct ctf_trace *trace;
	328	struct lttng_ht_iter iter;
	329	struct relay_stream *stream;
	330
	331	pthread_mutex_lock(&session->lock);
	332	DBG("closing session %" PRIu64 ": is conn already closed %d",
	333	session->id, session->connection_closed);
7591bab1	334	session->connection_closed = true;
7591bab1	335	pthread_mutex_unlock(&session->lock);
2a174661	336
7591bab1 MD	337	rcu_read_lock();
	338	cds_lfht_for_each_entry(session->ctf_traces_ht->ht,
	339	&iter.iter, trace, node.node) {
	340	ret = ctf_trace_close(trace);
	341	if (ret) {
	342	goto rcu_unlock;
2a174661 DG	343	}
2a174661 DG	344	}
7591bab1 MD	345	cds_list_for_each_entry_rcu(stream, &session->recv_list,
7591bab1 MD	346	recv_node) {
bda7c7b9 JG	347	/* Close streams which have not been published yet. */
bda7c7b9 JG	348	try_stream_close(stream);
7591bab1 MD	349	}
	350	rcu_unlock:
	351	rcu_read_unlock();
	352	if (ret) {
	353	return ret;
	354	}
	355	/* Put self-reference from create. */
	356	session_put(session);
	357	return ret;
2a174661 DG	358	}
2a174661 DG	359
98ba050e JR	360	int session_abort(struct relay_session *session)
	361	{
	362	int ret = 0;
	363
	364	if (!session) {
	365	return 0;
	366	}
	367
	368	pthread_mutex_lock(&session->lock);
	369	DBG("aborting session %" PRIu64, session->id);
98ba050e	370	session->aborted = true;
98ba050e JR	371	pthread_mutex_unlock(&session->lock);
	372	return ret;
	373	}
	374
7591bab1	375	void print_sessions(void)
2a174661	376	{
2a174661	377	struct lttng_ht_iter iter;
7591bab1	378	struct relay_session *session;
2a174661	379
ce3f3ba3 JG	380	if (!sessions_ht) {
	381	return;
	382	}
	383
2a174661	384	rcu_read_lock();
7591bab1 MD	385	cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, session,
	386	session_n.node) {
	387	if (!session_get(session)) {
	388	continue;
	389	}
	390	DBG("session %p refcount %ld session %" PRIu64,
	391	session,
	392	session->ref.refcount,
	393	session->id);
	394	session_put(session);
2a174661	395	}
2a174661	396	rcu_read_unlock();
2a174661	397	}