[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 <inttypes.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"
#include "notification-thread-commands.h"
#include "session.h"
#include "lttng-sessiond.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%s",
				consumer->dst.session_root_path,
				consumer->chunk_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 *ksession,
		unsigned int monitor)
{
	int ret;
	char *pathname;
	struct lttcomm_consumer_msg lkm;
	struct consumer_output *consumer;
	enum lttng_error_code status;
	struct ltt_session *session;
	struct lttng_channel_extended *channel_attr_extended;

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

	consumer = ksession->consumer;
	channel_attr_extended = (struct lttng_channel_extended *)
			channel->channel->attr.extended.ptr;

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

	if (monitor) {
		pathname = create_channel_path(consumer, ksession->uid,
				ksession->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->key,
			ksession->id,
			pathname,
			ksession->uid,
			ksession->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,
			channel_attr_extended->monitor_timer_interval);

	health_code_update();

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

	health_code_update();
	rcu_read_lock();
	session = session_find_by_id(ksession->id);
	assert(session);

	status = notification_thread_command_add_channel(
			notification_thread_handle, session->name,
			ksession->uid, ksession->gid,
			channel->channel->name, channel->key,
			LTTNG_DOMAIN_KERNEL,
			channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf);
	rcu_read_unlock();
	if (status != LTTNG_OK) {
		ret = -1;
		goto error;
	}

	channel->published_to_notification_thread = true;

error:
	free(pathname);
	return ret;
}

/*
 * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 *
 * The consumer socket lock must be held by the caller.
 */
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, 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->key,
			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.
 *
 * The consumer socket lock must be held by the caller.
 */
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 = LTTNG_OK;
	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);

	if (!channel->sent_to_consumer) {
		ret = kernel_consumer_add_channel(sock, channel, session, monitor);
		if (ret < 0) {
			goto error;
		}
		channel->sent_to_consumer = true;
	}

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

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

error:
	return ret;
}

/*
 * Send all stream fds of the kernel session to the consumer.
 *
 * The consumer socket lock must be held by the caller.
 */
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->key);
			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 %" PRIu64, channel->key);

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

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