[lttng-tools.git] / src / bin / lttng-sessiond / kernel-consumer.c

/*
 * Copyright (C) 2012 - David Goulet <dgoulet@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 _GNU_SOURCE
#define _LGPL_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>

#include <common/common.h>
#include <common/defaults.h>

#include "consumer.h"
#include "health-sessiond.h"
#include "kernel-consumer.h"

static char *create_channel_path(struct consumer_output *consumer,
		uid_t uid, gid_t gid)
{
	int ret;
	char tmp_path[PATH_MAX];
	char *pathname = NULL;

	assert(consumer);

	/* Get the right path name destination */
	if (consumer->type == CONSUMER_DST_LOCAL) {
		/* Set application path to the destination path */
		ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s",
				consumer->dst.trace_path, consumer->subdir);
		if (ret < 0) {
			PERROR("snprintf kernel channel path");
			goto error;
		}
		pathname = strndup(tmp_path, sizeof(tmp_path));
		if (!pathname) {
			PERROR("strndup");
			goto error;
		}

		/* Create directory */
		ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, uid, gid);
		if (ret < 0) {
			if (ret != -EEXIST) {
				ERR("Trace directory creation error");
				goto error;
			}
		}
		DBG3("Kernel local consumer tracefile path: %s", pathname);
	} else {
		ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
		if (ret < 0) {
			PERROR("snprintf kernel metadata path");
			goto error;
		}
		pathname = strndup(tmp_path, sizeof(tmp_path));
		if (!pathname) {
			PERROR("strndup");
			goto error;
		}
		DBG3("Kernel network consumer subdir path: %s", pathname);
	}

	return pathname;

error:
	free(pathname);
	return NULL;
}

/*
 * Sending a single channel to the consumer with command ADD_CHANNEL.
 */
int kernel_consumer_add_channel(struct consumer_socket *sock,
		struct ltt_kernel_channel *channel, struct ltt_kernel_session *session,
		unsigned int monitor)
{
	int ret;
	char *pathname;
	struct lttcomm_consumer_msg lkm;
	struct consumer_output *consumer;

	/* Safety net */
	assert(channel);
	assert(session);
	assert(session->consumer);

	consumer = session->consumer;

	DBG("Kernel consumer adding channel %s to kernel consumer",
			channel->channel->name);

	if (monitor) {
		pathname = create_channel_path(consumer, session->uid, session->gid);
	} else {
		/* Empty path. */
		pathname = strdup("");
	}
	if (!pathname) {
		ret = -1;
		goto error;
	}

	/* Prep channel message structure */
	consumer_init_channel_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_CHANNEL,
			channel->fd,
			session->id,
			pathname,
			session->uid,
			session->gid,
			consumer->net_seq_index,
			channel->channel->name,
			channel->stream_count,
			channel->channel->attr.output,
			CONSUMER_CHANNEL_TYPE_DATA,
			channel->channel->attr.tracefile_size,
			channel->channel->attr.tracefile_count,
			monitor,
			channel->channel->attr.live_timer_interval);

	health_code_update();

	ret = consumer_send_channel(sock, &lkm);
	if (ret < 0) {
		goto error;
	}

	health_code_update();

error:
	free(pathname);
	return ret;
}

/*
 * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 */
int kernel_consumer_add_metadata(struct consumer_socket *sock,
		struct ltt_kernel_session *session, unsigned int monitor)
{
	int ret;
	char *pathname;
	struct lttcomm_consumer_msg lkm;
	struct consumer_output *consumer;

	/* Safety net */
	assert(session);
	assert(session->consumer);
	assert(sock);

	DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd);

	/* Get consumer output pointer */
	consumer = session->consumer;

	if (monitor) {
		pathname = create_channel_path(consumer, session->uid, session->gid);
	} else {
		/* Empty path. */
		pathname = strdup("");
	}
	if (!pathname) {
		ret = -1;
		goto error;
	}

	/* Prep channel message structure */
	consumer_init_channel_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_CHANNEL,
			session->metadata->fd,
			session->id,
			pathname,
			session->uid,
			session->gid,
			consumer->net_seq_index,
			DEFAULT_METADATA_NAME,
			1,
			DEFAULT_KERNEL_CHANNEL_OUTPUT,
			CONSUMER_CHANNEL_TYPE_METADATA,
			0, 0,
			monitor, 0);

	health_code_update();

	ret = consumer_send_channel(sock, &lkm);
	if (ret < 0) {
		goto error;
	}

	health_code_update();

	/* Prep stream message structure */
	consumer_init_stream_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_STREAM,
			session->metadata->fd,
			session->metadata_stream_fd,
			0); /* CPU: 0 for metadata. */

	health_code_update();

	/* Send stream and file descriptor */
	ret = consumer_send_stream(sock, consumer, &lkm,
			&session->metadata_stream_fd, 1);
	if (ret < 0) {
		goto error;
	}

	health_code_update();

error:
	free(pathname);
	return ret;
}

/*
 * Sending a single stream to the consumer with command ADD_STREAM.
 */
int kernel_consumer_add_stream(struct consumer_socket *sock,
		struct ltt_kernel_channel *channel, struct ltt_kernel_stream *stream,
		struct ltt_kernel_session *session, unsigned int monitor)
{
	int ret;
	struct lttcomm_consumer_msg lkm;
	struct consumer_output *consumer;

	assert(channel);
	assert(stream);
	assert(session);
	assert(session->consumer);
	assert(sock);

	DBG("Sending stream %d of channel %s to kernel consumer",
			stream->fd, channel->channel->name);

	/* Get consumer output pointer */
	consumer = session->consumer;

	/* Prep stream consumer message */
	consumer_init_stream_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_STREAM,
			channel->fd,
			stream->fd,
			stream->cpu);

	health_code_update();

	/* Send stream and file descriptor */
	ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
	if (ret < 0) {
		goto error;
	}

	health_code_update();

error:
	return ret;
}

/*
 * Sending the notification that all streams were sent with STREAMS_SENT.
 */
int kernel_consumer_streams_sent(struct consumer_socket *sock,
		struct ltt_kernel_session *session, uint64_t channel_key)
{
	int ret;
	struct lttcomm_consumer_msg lkm;
	struct consumer_output *consumer;

	assert(sock);
	assert(session);

	DBG("Sending streams_sent");
	/* Get consumer output pointer */
	consumer = session->consumer;

	/* Prep stream consumer message */
	consumer_init_streams_sent_comm_msg(&lkm,
			LTTNG_CONSUMER_STREAMS_SENT,
			channel_key, consumer->net_seq_index);

	health_code_update();

	/* Send stream and file descriptor */
	ret = consumer_send_msg(sock, &lkm);
	if (ret < 0) {
		goto error;
	}

error:
	return ret;
}

/*
 * Send all stream fds of kernel channel to the consumer.
 */
int kernel_consumer_send_channel_stream(struct consumer_socket *sock,
		struct ltt_kernel_channel *channel, struct ltt_kernel_session *session,
		unsigned int monitor)
{
	int ret;
	struct ltt_kernel_stream *stream;

	/* Safety net */
	assert(channel);
	assert(session);
	assert(session->consumer);
	assert(sock);

	/* Bail out if consumer is disabled */
	if (!session->consumer->enabled) {
		ret = LTTNG_OK;
		goto error;
	}

	DBG("Sending streams of channel %s to kernel consumer",
			channel->channel->name);

	ret = kernel_consumer_add_channel(sock, channel, session, monitor);
	if (ret < 0) {
		goto error;
	}

	/* Send streams */
	cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
		if (!stream->fd) {
			continue;
		}

		/* Add stream on the kernel consumer side. */
		ret = kernel_consumer_add_stream(sock, channel, stream, session,
				monitor);
		if (ret < 0) {
			goto error;
		}
	}

error:
	return ret;
}

/*
 * Send all stream fds of the kernel session to the consumer.
 */
int kernel_consumer_send_session(struct consumer_socket *sock,
		struct ltt_kernel_session *session)
{
	int ret, monitor = 0;
	struct ltt_kernel_channel *chan;

	/* Safety net */
	assert(session);
	assert(session->consumer);
	assert(sock);

	/* Bail out if consumer is disabled */
	if (!session->consumer->enabled) {
		ret = LTTNG_OK;
		goto error;
	}

	/* Don't monitor the streams on the consumer if in flight recorder. */
	if (session->output_traces) {
		monitor = 1;
	}

	DBG("Sending session stream to kernel consumer");

	if (session->metadata_stream_fd >= 0 && session->metadata) {
		ret = kernel_consumer_add_metadata(sock, session, monitor);
		if (ret < 0) {
			goto error;
		}
	}

	/* Send channel and streams of it */
	cds_list_for_each_entry(chan, &session->channel_list.head, list) {
		ret = kernel_consumer_send_channel_stream(sock, chan, session,
				monitor);
		if (ret < 0) {
			goto error;
		}
		if (monitor) {
			/*
			 * Inform the relay that all the streams for the
			 * channel were sent.
			 */
			ret = kernel_consumer_streams_sent(sock, session, chan->fd);
			if (ret < 0) {
				goto error;
			}
		}
	}

	DBG("Kernel consumer FDs of metadata and channel streams sent");

	session->consumer_fds_sent = 1;
	return 0;

error:
	return ret;
}

int kernel_consumer_destroy_channel(struct consumer_socket *socket,
		struct ltt_kernel_channel *channel)
{
	int ret;
	struct lttcomm_consumer_msg msg;

	assert(channel);
	assert(socket);

	DBG("Sending kernel consumer destroy channel key %d", channel->fd);

	memset(&msg, 0, sizeof(msg));
	msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
	msg.u.destroy_channel.key = channel->fd;

	pthread_mutex_lock(socket->lock);
	health_code_update();

	ret = consumer_send_msg(socket, &msg);
	if (ret < 0) {
		goto error;
	}

error:
	health_code_update();
	pthread_mutex_unlock(socket->lock);
	return ret;
}

int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
		struct ltt_kernel_metadata *metadata)
{
	int ret;
	struct lttcomm_consumer_msg msg;

	assert(metadata);
	assert(socket);

	DBG("Sending kernel consumer destroy channel key %d", metadata->fd);

	memset(&msg, 0, sizeof(msg));
	msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
	msg.u.destroy_channel.key = metadata->fd;

	pthread_mutex_lock(socket->lock);
	health_code_update();

	ret = consumer_send_msg(socket, &msg);
	if (ret < 0) {
		goto error;
	}

error:
	health_code_update();
	pthread_mutex_unlock(socket->lock);
	return ret;
}
Commit	Line	Data
f1e16794 DG	1	/*
	2	* Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
	3	*
	4	* This program is free software; you can redistribute it and/or modify it
	5	* under the terms of the GNU General Public License, version 2 only, as
	6	* 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 with
	14	* this program; if not, write to the Free Software Foundation, Inc., 51
	15	* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	16	*/
	17
	18	#define _GNU_SOURCE
6c1c0768	19	#define _LGPL_SOURCE
f1e16794 DG	20	#include <stdio.h>
	21	#include <stdlib.h>
	22	#include <string.h>
	23	#include <sys/stat.h>
	24	#include <unistd.h>
	25
	26	#include <common/common.h>
	27	#include <common/defaults.h>
f1e16794	28
00e2e675	29	#include "consumer.h"
8782cc74	30	#include "health-sessiond.h"
f1e16794 DG	31	#include "kernel-consumer.h"
f1e16794 DG	32
2bba9e53 DG	33	static char create_channel_path(struct consumer_output consumer,
2bba9e53 DG	34	uid_t uid, gid_t gid)
00e2e675 DG	35	{
00e2e675 DG	36	int ret;
ffe60014	37	char tmp_path[PATH_MAX];
2bba9e53	38	char *pathname = NULL;
00e2e675	39
2bba9e53	40	assert(consumer);
00e2e675	41
ffe60014 DG	42	/* Get the right path name destination */
	43	if (consumer->type == CONSUMER_DST_LOCAL) {
	44	/* Set application path to the destination path */
dec56f6c	45	ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s",
ffe60014 DG	46	consumer->dst.trace_path, consumer->subdir);
ffe60014 DG	47	if (ret < 0) {
2bba9e53	48	PERROR("snprintf kernel channel path");
ffe60014 DG	49	goto error;
ffe60014 DG	50	}
2bba9e53	51	pathname = strndup(tmp_path, sizeof(tmp_path));
bb3c4e70 MD	52	if (!pathname) {
	53	PERROR("strndup");
	54	goto error;
	55	}
ffe60014 DG	56
ffe60014 DG	57	/* Create directory */
2bba9e53	58	ret = run_as_mkdir_recursive(pathname, S_IRWXU \| S_IRWXG, uid, gid);
ffe60014 DG	59	if (ret < 0) {
	60	if (ret != -EEXIST) {
	61	ERR("Trace directory creation error");
	62	goto error;
	63	}
	64	}
	65	DBG3("Kernel local consumer tracefile path: %s", pathname);
	66	} else {
	67	ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
	68	if (ret < 0) {
2bba9e53	69	PERROR("snprintf kernel metadata path");
ffe60014 DG	70	goto error;
ffe60014 DG	71	}
2bba9e53	72	pathname = strndup(tmp_path, sizeof(tmp_path));
bb3c4e70 MD	73	if (!pathname) {
	74	PERROR("strndup");
	75	goto error;
	76	}
ffe60014 DG	77	DBG3("Kernel network consumer subdir path: %s", pathname);
	78	}
	79
2bba9e53 DG	80	return pathname;
	81
	82	error:
	83	free(pathname);
	84	return NULL;
	85	}
	86
	87	/*
	88	* Sending a single channel to the consumer with command ADD_CHANNEL.
	89	*/
	90	int kernel_consumer_add_channel(struct consumer_socket *sock,
	91	struct ltt_kernel_channel channel, struct ltt_kernel_session session,
	92	unsigned int monitor)
	93	{
	94	int ret;
	95	char *pathname;
	96	struct lttcomm_consumer_msg lkm;
	97	struct consumer_output *consumer;
	98
	99	/* Safety net */
	100	assert(channel);
	101	assert(session);
	102	assert(session->consumer);
	103
	104	consumer = session->consumer;
	105
	106	DBG("Kernel consumer adding channel %s to kernel consumer",
	107	channel->channel->name);
	108
	109	if (monitor) {
	110	pathname = create_channel_path(consumer, session->uid, session->gid);
2bba9e53 DG	111	} else {
2bba9e53 DG	112	/* Empty path. */
53efb85a	113	pathname = strdup("");
2bba9e53	114	}
bb3c4e70 MD	115	if (!pathname) {
	116	ret = -1;
	117	goto error;
	118	}
2bba9e53	119
00e2e675 DG	120	/* Prep channel message structure */
	121	consumer_init_channel_comm_msg(&lkm,
	122	LTTNG_CONSUMER_ADD_CHANNEL,
	123	channel->fd,
ffe60014 DG	124	session->id,
	125	pathname,
	126	session->uid,
	127	session->gid,
	128	consumer->net_seq_index,
c30aaa51	129	channel->channel->name,
ffe60014 DG	130	channel->stream_count,
ffe60014 DG	131	channel->channel->attr.output,
1624d5b7 JD	132	CONSUMER_CHANNEL_TYPE_DATA,
1624d5b7 JD	133	channel->channel->attr.tracefile_size,
2bba9e53	134	channel->channel->attr.tracefile_count,
ecc48a90 JD	135	monitor,
ecc48a90 JD	136	channel->channel->attr.live_timer_interval);
00e2e675	137
840cb59c	138	health_code_update();
ca03de58	139
00e2e675 DG	140	ret = consumer_send_channel(sock, &lkm);
	141	if (ret < 0) {
	142	goto error;
	143	}
	144
840cb59c	145	health_code_update();
ca03de58	146
00e2e675	147	error:
53efb85a	148	free(pathname);
00e2e675 DG	149	return ret;
	150	}
	151
	152	/*
	153	* Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
	154	*/
f50f23d9	155	int kernel_consumer_add_metadata(struct consumer_socket *sock,
2bba9e53	156	struct ltt_kernel_session *session, unsigned int monitor)
00e2e675 DG	157	{
00e2e675 DG	158	int ret;
2bba9e53	159	char *pathname;
00e2e675	160	struct lttcomm_consumer_msg lkm;
a7d9a3e7	161	struct consumer_output *consumer;
00e2e675 DG	162
	163	/* Safety net */
	164	assert(session);
	165	assert(session->consumer);
f50f23d9	166	assert(sock);
00e2e675 DG	167
	168	DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd);
	169
	170	/* Get consumer output pointer */
a7d9a3e7	171	consumer = session->consumer;
00e2e675	172
2bba9e53 DG	173	if (monitor) {
2bba9e53 DG	174	pathname = create_channel_path(consumer, session->uid, session->gid);
00e2e675	175	} else {
2bba9e53	176	/* Empty path. */
53efb85a	177	pathname = strdup("");
00e2e675	178	}
bb3c4e70 MD	179	if (!pathname) {
	180	ret = -1;
	181	goto error;
	182	}
00e2e675 DG	183
	184	/* Prep channel message structure */
	185	consumer_init_channel_comm_msg(&lkm,
	186	LTTNG_CONSUMER_ADD_CHANNEL,
	187	session->metadata->fd,
ffe60014 DG	188	session->id,
	189	pathname,
	190	session->uid,
	191	session->gid,
	192	consumer->net_seq_index,
30079b6b	193	DEFAULT_METADATA_NAME,
ffe60014 DG	194	1,
ffe60014 DG	195	DEFAULT_KERNEL_CHANNEL_OUTPUT,
1624d5b7	196	CONSUMER_CHANNEL_TYPE_METADATA,
2bba9e53	197	0, 0,
ecc48a90	198	monitor, 0);
00e2e675	199
840cb59c	200	health_code_update();
ca03de58	201
00e2e675 DG	202	ret = consumer_send_channel(sock, &lkm);
	203	if (ret < 0) {
	204	goto error;
	205	}
	206
840cb59c	207	health_code_update();
ca03de58	208
00e2e675 DG	209	/* Prep stream message structure */
	210	consumer_init_stream_comm_msg(&lkm,
	211	LTTNG_CONSUMER_ADD_STREAM,
	212	session->metadata->fd,
	213	session->metadata_stream_fd,
1624d5b7	214	0); /* CPU: 0 for metadata. */
00e2e675	215
840cb59c	216	health_code_update();
ca03de58	217
00e2e675	218	/* Send stream and file descriptor */
a7d9a3e7	219	ret = consumer_send_stream(sock, consumer, &lkm,
00e2e675 DG	220	&session->metadata_stream_fd, 1);
	221	if (ret < 0) {
	222	goto error;
	223	}
	224
840cb59c	225	health_code_update();
ca03de58	226
00e2e675	227	error:
53efb85a	228	free(pathname);
00e2e675 DG	229	return ret;
	230	}
	231
	232	/*
	233	* Sending a single stream to the consumer with command ADD_STREAM.
	234	*/
f50f23d9 DG	235	int kernel_consumer_add_stream(struct consumer_socket *sock,
f50f23d9 DG	236	struct ltt_kernel_channel channel, struct ltt_kernel_stream stream,
2bba9e53	237	struct ltt_kernel_session *session, unsigned int monitor)
00e2e675 DG	238	{
00e2e675 DG	239	int ret;
00e2e675	240	struct lttcomm_consumer_msg lkm;
a7d9a3e7	241	struct consumer_output *consumer;
00e2e675 DG	242
	243	assert(channel);
	244	assert(stream);
	245	assert(session);
	246	assert(session->consumer);
f50f23d9	247	assert(sock);
00e2e675 DG	248
	249	DBG("Sending stream %d of channel %s to kernel consumer",
	250	stream->fd, channel->channel->name);
	251
	252	/* Get consumer output pointer */
a7d9a3e7	253	consumer = session->consumer;
00e2e675	254
00e2e675	255	/* Prep stream consumer message */
ffe60014 DG	256	consumer_init_stream_comm_msg(&lkm,
ffe60014 DG	257	LTTNG_CONSUMER_ADD_STREAM,
00e2e675 DG	258	channel->fd,
00e2e675 DG	259	stream->fd,
ffe60014	260	stream->cpu);
00e2e675	261
840cb59c	262	health_code_update();
ca03de58	263
00e2e675	264	/* Send stream and file descriptor */
a7d9a3e7	265	ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
00e2e675 DG	266	if (ret < 0) {
	267	goto error;
	268	}
	269
840cb59c	270	health_code_update();
ca03de58	271
00e2e675 DG	272	error:
	273	return ret;
	274	}
	275
a4baae1b JD	276	/*
	277	* Sending the notification that all streams were sent with STREAMS_SENT.
	278	*/
	279	int kernel_consumer_streams_sent(struct consumer_socket *sock,
	280	struct ltt_kernel_session *session, uint64_t channel_key)
	281	{
	282	int ret;
	283	struct lttcomm_consumer_msg lkm;
	284	struct consumer_output *consumer;
	285
	286	assert(sock);
	287	assert(session);
	288
	289	DBG("Sending streams_sent");
	290	/* Get consumer output pointer */
	291	consumer = session->consumer;
	292
	293	/* Prep stream consumer message */
	294	consumer_init_streams_sent_comm_msg(&lkm,
	295	LTTNG_CONSUMER_STREAMS_SENT,
	296	channel_key, consumer->net_seq_index);
	297
	298	health_code_update();
	299
	300	/* Send stream and file descriptor */
	301	ret = consumer_send_msg(sock, &lkm);
	302	if (ret < 0) {
	303	goto error;
	304	}
	305
	306	error:
	307	return ret;
	308	}
	309
f1e16794 DG	310	/*
	311	* Send all stream fds of kernel channel to the consumer.
	312	*/
f50f23d9	313	int kernel_consumer_send_channel_stream(struct consumer_socket *sock,
2bba9e53 DG	314	struct ltt_kernel_channel channel, struct ltt_kernel_session session,
2bba9e53 DG	315	unsigned int monitor)
f1e16794	316	{
00e2e675	317	int ret;
f1e16794	318	struct ltt_kernel_stream *stream;
00e2e675 DG	319
	320	/* Safety net */
	321	assert(channel);
	322	assert(session);
	323	assert(session->consumer);
f50f23d9	324	assert(sock);
00e2e675 DG	325
	326	/* Bail out if consumer is disabled */
	327	if (!session->consumer->enabled) {
f73fabfd	328	ret = LTTNG_OK;
00e2e675 DG	329	goto error;
00e2e675 DG	330	}
f1e16794 DG	331
	332	DBG("Sending streams of channel %s to kernel consumer",
	333	channel->channel->name);
	334
2bba9e53	335	ret = kernel_consumer_add_channel(sock, channel, session, monitor);
f1e16794	336	if (ret < 0) {
f1e16794 DG	337	goto error;
	338	}
	339
	340	/* Send streams */
	341	cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
	342	if (!stream->fd) {
	343	continue;
	344	}
00e2e675 DG	345
00e2e675 DG	346	/* Add stream on the kernel consumer side. */
2bba9e53 DG	347	ret = kernel_consumer_add_stream(sock, channel, stream, session,
2bba9e53 DG	348	monitor);
f1e16794	349	if (ret < 0) {
f1e16794 DG	350	goto error;
	351	}
	352	}
	353
f1e16794 DG	354	error:
	355	return ret;
	356	}
	357
	358	/*
	359	* Send all stream fds of the kernel session to the consumer.
	360	*/
f50f23d9 DG	361	int kernel_consumer_send_session(struct consumer_socket *sock,
f50f23d9 DG	362	struct ltt_kernel_session *session)
f1e16794	363	{
2bba9e53	364	int ret, monitor = 0;
f1e16794	365	struct ltt_kernel_channel *chan;
f1e16794	366
00e2e675 DG	367	/* Safety net */
	368	assert(session);
	369	assert(session->consumer);
f50f23d9	370	assert(sock);
f1e16794	371
00e2e675 DG	372	/* Bail out if consumer is disabled */
00e2e675 DG	373	if (!session->consumer->enabled) {
f73fabfd	374	ret = LTTNG_OK;
00e2e675	375	goto error;
f1e16794 DG	376	}
f1e16794 DG	377
2bba9e53 DG	378	/* Don't monitor the streams on the consumer if in flight recorder. */
	379	if (session->output_traces) {
	380	monitor = 1;
	381	}
	382
00e2e675 DG	383	DBG("Sending session stream to kernel consumer");
00e2e675 DG	384
609af759	385	if (session->metadata_stream_fd >= 0 && session->metadata) {
2bba9e53	386	ret = kernel_consumer_add_metadata(sock, session, monitor);
f1e16794	387	if (ret < 0) {
f1e16794 DG	388	goto error;
f1e16794 DG	389	}
f1e16794 DG	390	}
f1e16794 DG	391
00e2e675	392	/* Send channel and streams of it */
f1e16794	393	cds_list_for_each_entry(chan, &session->channel_list.head, list) {
2bba9e53 DG	394	ret = kernel_consumer_send_channel_stream(sock, chan, session,
2bba9e53 DG	395	monitor);
f1e16794 DG	396	if (ret < 0) {
	397	goto error;
	398	}
601262d6 JD	399	if (monitor) {
	400	/*
	401	* Inform the relay that all the streams for the
	402	* channel were sent.
	403	*/
	404	ret = kernel_consumer_streams_sent(sock, session, chan->fd);
	405	if (ret < 0) {
	406	goto error;
	407	}
	408	}
f1e16794 DG	409	}
f1e16794 DG	410
00e2e675	411	DBG("Kernel consumer FDs of metadata and channel streams sent");
f1e16794	412
4ce9ff51	413	session->consumer_fds_sent = 1;
f1e16794 DG	414	return 0;
	415
	416	error:
	417	return ret;
	418	}
07b86b52 JD	419
	420	int kernel_consumer_destroy_channel(struct consumer_socket *socket,
	421	struct ltt_kernel_channel *channel)
	422	{
	423	int ret;
	424	struct lttcomm_consumer_msg msg;
	425
	426	assert(channel);
	427	assert(socket);
07b86b52 JD	428
	429	DBG("Sending kernel consumer destroy channel key %d", channel->fd);
	430
53efb85a	431	memset(&msg, 0, sizeof(msg));
07b86b52 JD	432	msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
	433	msg.u.destroy_channel.key = channel->fd;
	434
	435	pthread_mutex_lock(socket->lock);
	436	health_code_update();
	437
	438	ret = consumer_send_msg(socket, &msg);
	439	if (ret < 0) {
	440	goto error;
	441	}
	442
	443	error:
	444	health_code_update();
	445	pthread_mutex_unlock(socket->lock);
	446	return ret;
	447	}
	448
	449	int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
	450	struct ltt_kernel_metadata *metadata)
	451	{
	452	int ret;
	453	struct lttcomm_consumer_msg msg;
	454
	455	assert(metadata);
	456	assert(socket);
07b86b52 JD	457
	458	DBG("Sending kernel consumer destroy channel key %d", metadata->fd);
	459
53efb85a	460	memset(&msg, 0, sizeof(msg));
07b86b52 JD	461	msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
	462	msg.u.destroy_channel.key = metadata->fd;
	463
	464	pthread_mutex_lock(socket->lock);
	465	health_code_update();
	466
	467	ret = consumer_send_msg(socket, &msg);
	468	if (ret < 0) {
	469	goto error;
	470	}
	471
	472	error:
	473	health_code_update();
	474	pthread_mutex_unlock(socket->lock);
	475	return ret;
	476	}