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

/*
 * Copyright (C) 2011 - David Goulet <david.goulet@polymtl.ca>
 *
 * 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
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

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

#include "consumer.h"
#include "ust-consumer.h"

/*
 * Send a single channel to the consumer using command ADD_CHANNEL.
 */
static int send_channel(int sock, struct ust_app_channel *uchan)
{
	int ret, fd;
	struct lttcomm_consumer_msg msg;

	/* Safety net */
	assert(uchan);

	if (sock < 0) {
		ret = -EINVAL;
		goto error;
	}

	DBG2("Sending channel %s to UST consumer", uchan->name);

	consumer_init_channel_comm_msg(&msg,
			LTTNG_CONSUMER_ADD_CHANNEL,
			uchan->obj->shm_fd,
			uchan->attr.subbuf_size,
			uchan->obj->memory_map_size,
			uchan->name,
			uchan->streams.count);

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

	fd = uchan->obj->shm_fd;
	ret = consumer_send_fds(sock, &fd, 1);
	if (ret < 0) {
		goto error;
	}

error:
	return ret;
}

/*
 * Send a single stream to the consumer using ADD_STREAM command.
 */
static int send_channel_stream(int sock, struct ust_app_channel *uchan,
		struct ust_app_session *usess, struct ltt_ust_stream *stream,
		struct consumer_output *consumer, const char *pathname)
{
	int ret, fds[2];
	struct lttcomm_consumer_msg msg;

	/* Safety net */
	assert(uchan);
	assert(usess);
	assert(stream);
	assert(consumer);

	DBG2("Sending stream %d of channel %s to kernel consumer",
			stream->obj->shm_fd, uchan->name);

	consumer_init_stream_comm_msg(&msg,
			LTTNG_CONSUMER_ADD_STREAM,
			uchan->obj->shm_fd,
			stream->obj->shm_fd,
			LTTNG_CONSUMER_ACTIVE_STREAM,
			DEFAULT_UST_CHANNEL_OUTPUT,
			stream->obj->memory_map_size,
			usess->uid,
			usess->gid,
			consumer->net_seq_index,
			0, /* Metadata flag unset */
			stream->name,
			pathname);

	/* Send stream and file descriptor */
	fds[0] = stream->obj->shm_fd;
	fds[1] = stream->obj->wait_fd;
	ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
	if (ret < 0) {
		goto error;
	}

error:
	return ret;
}

/*
 * Send all stream fds of UST channel to the consumer.
 */
static int send_channel_streams(int sock,
		struct ust_app_channel *uchan, struct ust_app_session *usess,
		struct consumer_output *consumer)
{
	int ret;
	char tmp_path[PATH_MAX];
	const char *pathname;
	struct ltt_ust_stream *stream, *tmp;

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

	ret = send_channel(sock, uchan);
	if (ret < 0) {
		goto error;
	}

	/* 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.trace_path, consumer->subdir, usess->path);
		if (ret < 0) {
			PERROR("snprintf stream path");
			goto error;
		}
		pathname = tmp_path;
		DBG3("UST local consumer tracefile path: %s", pathname);
	} else {
		ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
				consumer->subdir, usess->path);
		if (ret < 0) {
			PERROR("snprintf stream path");
			goto error;
		}
		pathname = tmp_path;
		DBG3("UST network consumer subdir path: %s", pathname);
	}

	cds_list_for_each_entry_safe(stream, tmp, &uchan->streams.head, list) {
		if (!stream->obj->shm_fd) {
			continue;
		}

		ret = send_channel_stream(sock, uchan, usess, stream, consumer,
				pathname);
		if (ret < 0) {
			goto error;
		}
	}

	DBG("UST consumer channel streams sent");

	return 0;

error:
	return ret;
}

/*
 * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 */
static int send_metadata(int sock, struct ust_app_session *usess,
		struct consumer_output *consumer)
{
	int ret, fd, fds[2];
	char tmp_path[PATH_MAX];
	const char *pathname;
	struct lttcomm_consumer_msg msg;

	/* Safety net */
	assert(usess);
	assert(consumer);

	if (sock < 0) {
		ERR("Consumer socket is negative (%d)", sock);
		return -EINVAL;
	}

	if (usess->metadata->obj->shm_fd == 0) {
		ERR("Metadata obj shm_fd is 0");
		ret = -1;
		goto error;
	}

	DBG("UST consumer sending metadata stream fd");

	consumer_init_channel_comm_msg(&msg,
			LTTNG_CONSUMER_ADD_CHANNEL,
			usess->metadata->obj->shm_fd,
			usess->metadata->attr.subbuf_size,
			usess->metadata->obj->memory_map_size,
			"metadata",
			1);

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

	/* Sending metadata shared memory fd */
	fd = usess->metadata->obj->shm_fd;
	ret = consumer_send_fds(sock, &fd, 1);
	if (ret < 0) {
		goto error;
	}

	/* Get correct 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.trace_path, consumer->subdir, usess->path);
		if (ret < 0) {
			PERROR("snprintf stream path");
			goto error;
		}
		pathname = tmp_path;

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

	consumer_init_stream_comm_msg(&msg,
			LTTNG_CONSUMER_ADD_STREAM,
			usess->metadata->obj->shm_fd,
			usess->metadata->stream_obj->shm_fd,
			LTTNG_CONSUMER_ACTIVE_STREAM,
			DEFAULT_UST_CHANNEL_OUTPUT,
			usess->metadata->stream_obj->memory_map_size,
			usess->uid,
			usess->gid,
			consumer->net_seq_index,
			1, /* Flag metadata set */
			"metadata",
			pathname);

	/* Send stream and file descriptor */
	fds[0] = usess->metadata->stream_obj->shm_fd;
	fds[1] = usess->metadata->stream_obj->wait_fd;
	ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
	if (ret < 0) {
		goto error;
	}

error:
	return ret;
}

/*
 * Send all stream fds of the UST session to the consumer.
 */
int ust_consumer_send_session(struct ust_app_session *usess,
		struct consumer_output *consumer, struct consumer_socket *sock)
{
	int ret = 0;
	struct lttng_ht_iter iter;
	struct ust_app_channel *ua_chan;

	assert(usess);

	if (consumer == NULL || sock == NULL) {
		/* There is no consumer so just ignoring the command. */
		DBG("UST consumer does not exist. Not sending streams");
		return 0;
	}

	DBG("Sending metadata stream fd to consumer on %d", sock->fd);

	pthread_mutex_lock(sock->lock);

	/* Sending metadata information to the consumer */
	ret = send_metadata(sock->fd, usess, consumer);
	if (ret < 0) {
		goto error;
	}

	/* Send each channel fd streams of session */
	rcu_read_lock();
	cds_lfht_for_each_entry(usess->channels->ht, &iter.iter, ua_chan,
			node.node) {
		/*
		 * Indicate that the channel was not created on the tracer side so skip
		 * sending unexisting streams.
		 */
		if (ua_chan->obj == NULL) {
			continue;
		}

		ret = send_channel_streams(sock->fd, ua_chan, usess, consumer);
		if (ret < 0) {
			rcu_read_unlock();
			goto error;
		}
	}
	rcu_read_unlock();

	DBG("consumer fds (metadata and channel streams) sent");

	/* All good! */
	ret = 0;

error:
	pthread_mutex_unlock(sock->lock);
	return ret;
}
Commit	Line	Data
48842b30 DG	1	/*
	2	* Copyright (C) 2011 - David Goulet <david.goulet@polymtl.ca>
	3	*
d14d33bf AM	4	* This program is free software; you can redistribute it and/or modify
	5	* it under the terms of the GNU General Public License, version 2 only,
	6	* as published by the Free Software Foundation.
48842b30	7	*
d14d33bf AM	8	* This program is distributed in the hope that it will be useful,
	9	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	10	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	11	* GNU General Public License for more details.
48842b30	12	*
d14d33bf AM	13	* You should have received a copy of the GNU General Public License along
	14	* with this program; if not, write to the Free Software Foundation, Inc.,
	15	* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
48842b30 DG	16	*/
	17
	18	#define _GNU_SOURCE
	19	#include <errno.h>
	20	#include <stdio.h>
	21	#include <stdlib.h>
	22	#include <string.h>
	23	#include <unistd.h>
	24
990570ed	25	#include <common/common.h>
db758600	26	#include <common/consumer.h>
990570ed	27	#include <common/defaults.h>
48842b30	28
00e2e675	29	#include "consumer.h"
48842b30 DG	30	#include "ust-consumer.h"
	31
	32	/*
37278a1e	33	* Send a single channel to the consumer using command ADD_CHANNEL.
48842b30	34	*/
37278a1e	35	static int send_channel(int sock, struct ust_app_channel *uchan)
48842b30	36	{
8010679a	37	int ret, fd;
37278a1e	38	struct lttcomm_consumer_msg msg;
48842b30	39
37278a1e DG	40	/* Safety net */
	41	assert(uchan);
	42
	43	if (sock < 0) {
	44	ret = -EINVAL;
	45	goto error;
	46	}
48842b30	47
37278a1e DG	48	DBG2("Sending channel %s to UST consumer", uchan->name);
	49
	50	consumer_init_channel_comm_msg(&msg,
00e2e675 DG	51	LTTNG_CONSUMER_ADD_CHANNEL,
	52	uchan->obj->shm_fd,
	53	uchan->attr.subbuf_size,
	54	uchan->obj->memory_map_size,
c30aaa51 MD	55	uchan->name,
c30aaa51 MD	56	uchan->streams.count);
00e2e675	57
37278a1e	58	ret = consumer_send_channel(sock, &msg);
48842b30	59	if (ret < 0) {
48842b30 DG	60	goto error;
48842b30 DG	61	}
00e2e675	62
8010679a	63	fd = uchan->obj->shm_fd;
00e2e675	64	ret = consumer_send_fds(sock, &fd, 1);
48842b30	65	if (ret < 0) {
48842b30 DG	66	goto error;
	67	}
	68
37278a1e DG	69	error:
	70	return ret;
	71	}
	72
	73	/*
	74	* Send a single stream to the consumer using ADD_STREAM command.
	75	*/
	76	static int send_channel_stream(int sock, struct ust_app_channel *uchan,
	77	struct ust_app_session usess, struct ltt_ust_stream stream,
	78	struct consumer_output consumer, const char pathname)
	79	{
	80	int ret, fds[2];
	81	struct lttcomm_consumer_msg msg;
	82
	83	/* Safety net */
	84	assert(uchan);
	85	assert(usess);
	86	assert(stream);
	87	assert(consumer);
	88
	89	DBG2("Sending stream %d of channel %s to kernel consumer",
	90	stream->obj->shm_fd, uchan->name);
	91
	92	consumer_init_stream_comm_msg(&msg,
	93	LTTNG_CONSUMER_ADD_STREAM,
	94	uchan->obj->shm_fd,
	95	stream->obj->shm_fd,
	96	LTTNG_CONSUMER_ACTIVE_STREAM,
	97	DEFAULT_UST_CHANNEL_OUTPUT,
	98	stream->obj->memory_map_size,
	99	usess->uid,
	100	usess->gid,
	101	consumer->net_seq_index,
	102	0, /* Metadata flag unset */
	103	stream->name,
	104	pathname);
	105
	106	/* Send stream and file descriptor */
	107	fds[0] = stream->obj->shm_fd;
	108	fds[1] = stream->obj->wait_fd;
	109	ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
	110	if (ret < 0) {
	111	goto error;
	112	}
	113
	114	error:
	115	return ret;
	116	}
	117
	118	/*
	119	* Send all stream fds of UST channel to the consumer.
	120	*/
173af62f	121	static int send_channel_streams(int sock,
37278a1e DG	122	struct ust_app_channel uchan, struct ust_app_session usess,
	123	struct consumer_output *consumer)
	124	{
	125	int ret;
	126	char tmp_path[PATH_MAX];
	127	const char *pathname;
	128	struct ltt_ust_stream stream, tmp;
	129
	130	DBG("Sending streams of channel %s to UST consumer", uchan->name);
	131
	132	ret = send_channel(sock, uchan);
	133	if (ret < 0) {
	134	goto error;
	135	}
	136
00e2e675 DG	137	/* Get the right path name destination */
	138	if (consumer->type == CONSUMER_DST_LOCAL) {
	139	/* Set application path to the destination path */
a4b92340 DG	140	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s/%s",
a4b92340 DG	141	consumer->dst.trace_path, consumer->subdir, usess->path);
00e2e675 DG	142	if (ret < 0) {
	143	PERROR("snprintf stream path");
	144	goto error;
	145	}
	146	pathname = tmp_path;
	147	DBG3("UST local consumer tracefile path: %s", pathname);
	148	} else {
	149	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
37278a1e	150	consumer->subdir, usess->path);
00e2e675 DG	151	if (ret < 0) {
	152	PERROR("snprintf stream path");
	153	goto error;
	154	}
	155	pathname = tmp_path;
	156	DBG3("UST network consumer subdir path: %s", pathname);
	157	}
	158
d80a6244	159	cds_list_for_each_entry_safe(stream, tmp, &uchan->streams.head, list) {
48842b30	160	if (!stream->obj->shm_fd) {
5af2f756	161	continue;
48842b30	162	}
48842b30	163
37278a1e	164	ret = send_channel_stream(sock, uchan, usess, stream, consumer,
00e2e675	165	pathname);
48842b30	166	if (ret < 0) {
48842b30 DG	167	goto error;
48842b30 DG	168	}
48842b30	169	}
48842b30	170
00e2e675	171	DBG("UST consumer channel streams sent");
48842b30 DG	172
	173	return 0;
	174
	175	error:
	176	return ret;
	177	}
	178
	179	/*
37278a1e	180	* Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
48842b30	181	*/
173af62f	182	static int send_metadata(int sock, struct ust_app_session *usess,
00e2e675	183	struct consumer_output *consumer)
48842b30	184	{
37278a1e	185	int ret, fd, fds[2];
00e2e675 DG	186	char tmp_path[PATH_MAX];
00e2e675 DG	187	const char *pathname;
37278a1e	188	struct lttcomm_consumer_msg msg;
48842b30	189
37278a1e DG	190	/* Safety net */
	191	assert(usess);
	192	assert(consumer);
48842b30	193
37278a1e DG	194	if (sock < 0) {
37278a1e DG	195	ERR("Consumer socket is negative (%d)", sock);
7753dea8 MD	196	return -EINVAL;
	197	}
	198
37278a1e DG	199	if (usess->metadata->obj->shm_fd == 0) {
	200	ERR("Metadata obj shm_fd is 0");
	201	ret = -1;
	202	goto error;
	203	}
48842b30	204
37278a1e	205	DBG("UST consumer sending metadata stream fd");
00e2e675	206
37278a1e DG	207	consumer_init_channel_comm_msg(&msg,
	208	LTTNG_CONSUMER_ADD_CHANNEL,
	209	usess->metadata->obj->shm_fd,
	210	usess->metadata->attr.subbuf_size,
	211	usess->metadata->obj->memory_map_size,
c30aaa51 MD	212	"metadata",
c30aaa51 MD	213	1);
37278a1e DG	214
	215	ret = consumer_send_channel(sock, &msg);
	216	if (ret < 0) {
	217	goto error;
	218	}
	219
	220	/* Sending metadata shared memory fd */
	221	fd = usess->metadata->obj->shm_fd;
	222	ret = consumer_send_fds(sock, &fd, 1);
	223	if (ret < 0) {
	224	goto error;
	225	}
00e2e675	226
37278a1e DG	227	/* Get correct path name destination */
	228	if (consumer->type == CONSUMER_DST_LOCAL) {
	229	/* Set application path to the destination path */
a4b92340 DG	230	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s/%s",
a4b92340 DG	231	consumer->dst.trace_path, consumer->subdir, usess->path);
48842b30	232	if (ret < 0) {
37278a1e	233	PERROR("snprintf stream path");
48842b30 DG	234	goto error;
48842b30 DG	235	}
37278a1e	236	pathname = tmp_path;
48842b30	237
37278a1e DG	238	/* Create directory */
	239	ret = run_as_mkdir_recursive(pathname, S_IRWXU \| S_IRWXG,
	240	usess->uid, usess->gid);
	241	if (ret < 0) {
	242	if (ret != -EEXIST) {
	243	ERR("Trace directory creation error");
00e2e675 DG	244	goto error;
00e2e675 DG	245	}
48842b30	246	}
37278a1e DG	247	} else {
	248	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
	249	consumer->subdir, usess->path);
48842b30	250	if (ret < 0) {
37278a1e	251	PERROR("snprintf metadata path");
48842b30 DG	252	goto error;
48842b30 DG	253	}
37278a1e DG	254	pathname = tmp_path;
	255	}
	256
	257	consumer_init_stream_comm_msg(&msg,
	258	LTTNG_CONSUMER_ADD_STREAM,
	259	usess->metadata->obj->shm_fd,
	260	usess->metadata->stream_obj->shm_fd,
	261	LTTNG_CONSUMER_ACTIVE_STREAM,
	262	DEFAULT_UST_CHANNEL_OUTPUT,
	263	usess->metadata->stream_obj->memory_map_size,
	264	usess->uid,
	265	usess->gid,
	266	consumer->net_seq_index,
	267	1, /* Flag metadata set */
	268	"metadata",
	269	pathname);
	270
	271	/* Send stream and file descriptor */
	272	fds[0] = usess->metadata->stream_obj->shm_fd;
	273	fds[1] = usess->metadata->stream_obj->wait_fd;
	274	ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
	275	if (ret < 0) {
	276	goto error;
	277	}
	278
	279	error:
	280	return ret;
	281	}
	282
	283	/*
	284	* Send all stream fds of the UST session to the consumer.
	285	*/
173af62f DG	286	int ust_consumer_send_session(struct ust_app_session *usess,
173af62f DG	287	struct consumer_output consumer, struct consumer_socket sock)
37278a1e DG	288	{
37278a1e DG	289	int ret = 0;
37278a1e DG	290	struct lttng_ht_iter iter;
	291	struct ust_app_channel *ua_chan;
	292
173af62f	293	assert(usess);
a4b92340 DG	294
	295	if (consumer == NULL \|\| sock == NULL) {
	296	/* There is no consumer so just ignoring the command. */
	297	DBG("UST consumer does not exist. Not sending streams");
	298	return 0;
	299	}
37278a1e	300
173af62f DG	301	DBG("Sending metadata stream fd to consumer on %d", sock->fd);
	302
	303	pthread_mutex_lock(sock->lock);
37278a1e DG	304
37278a1e DG	305	/* Sending metadata information to the consumer */
173af62f	306	ret = send_metadata(sock->fd, usess, consumer);
37278a1e DG	307	if (ret < 0) {
37278a1e DG	308	goto error;
48842b30 DG	309	}
	310
	311	/* Send each channel fd streams of session */
	312	rcu_read_lock();
bec39940 DG	313	cds_lfht_for_each_entry(usess->channels->ht, &iter.iter, ua_chan,
bec39940 DG	314	node.node) {
aeb96892 DG	315	/*
	316	* Indicate that the channel was not created on the tracer side so skip
	317	* sending unexisting streams.
	318	*/
	319	if (ua_chan->obj == NULL) {
	320	continue;
	321	}
	322
173af62f	323	ret = send_channel_streams(sock->fd, ua_chan, usess, consumer);
48842b30	324	if (ret < 0) {
5485f822	325	rcu_read_unlock();
48842b30 DG	326	goto error;
48842b30 DG	327	}
48842b30 DG	328	}
	329	rcu_read_unlock();
	330
	331	DBG("consumer fds (metadata and channel streams) sent");
	332
173af62f DG	333	/* All good! */
173af62f DG	334	ret = 0;
48842b30 DG	335
48842b30 DG	336	error:
173af62f	337	pthread_mutex_unlock(sock->lock);
48842b30 DG	338	return ret;
48842b30 DG	339	}