[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 _LGPL_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <unistd.h>

#include <common/common.h>
#include <common/defaults.h>
#include <common/compat/string.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 = lttng_strndup(tmp_path, sizeof(tmp_path));
		if (!pathname) {
			PERROR("lttng_strndup");
			goto error;
		}

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