[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);

	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");

	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,
	55	uchan->name);
	56
37278a1e	57	ret = consumer_send_channel(sock, &msg);
48842b30	58	if (ret < 0) {
48842b30 DG	59	goto error;
48842b30 DG	60	}
00e2e675	61
8010679a	62	fd = uchan->obj->shm_fd;
00e2e675	63	ret = consumer_send_fds(sock, &fd, 1);
48842b30	64	if (ret < 0) {
48842b30 DG	65	goto error;
	66	}
	67
37278a1e DG	68	error:
	69	return ret;
	70	}
	71
	72	/*
	73	* Send a single stream to the consumer using ADD_STREAM command.
	74	*/
	75	static int send_channel_stream(int sock, struct ust_app_channel *uchan,
	76	struct ust_app_session usess, struct ltt_ust_stream stream,
	77	struct consumer_output consumer, const char pathname)
	78	{
	79	int ret, fds[2];
	80	struct lttcomm_consumer_msg msg;
	81
	82	/* Safety net */
	83	assert(uchan);
	84	assert(usess);
	85	assert(stream);
	86	assert(consumer);
	87
	88	DBG2("Sending stream %d of channel %s to kernel consumer",
	89	stream->obj->shm_fd, uchan->name);
	90
	91	consumer_init_stream_comm_msg(&msg,
	92	LTTNG_CONSUMER_ADD_STREAM,
	93	uchan->obj->shm_fd,
	94	stream->obj->shm_fd,
	95	LTTNG_CONSUMER_ACTIVE_STREAM,
	96	DEFAULT_UST_CHANNEL_OUTPUT,
	97	stream->obj->memory_map_size,
	98	usess->uid,
	99	usess->gid,
	100	consumer->net_seq_index,
	101	0, /* Metadata flag unset */
	102	stream->name,
	103	pathname);
	104
	105	/* Send stream and file descriptor */
	106	fds[0] = stream->obj->shm_fd;
	107	fds[1] = stream->obj->wait_fd;
	108	ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
	109	if (ret < 0) {
	110	goto error;
	111	}
	112
	113	error:
	114	return ret;
	115	}
	116
	117	/*
	118	* Send all stream fds of UST channel to the consumer.
	119	*/
173af62f	120	static int send_channel_streams(int sock,
37278a1e DG	121	struct ust_app_channel uchan, struct ust_app_session usess,
	122	struct consumer_output *consumer)
	123	{
	124	int ret;
	125	char tmp_path[PATH_MAX];
	126	const char *pathname;
	127	struct ltt_ust_stream stream, tmp;
	128
	129	DBG("Sending streams of channel %s to UST consumer", uchan->name);
	130
	131	ret = send_channel(sock, uchan);
	132	if (ret < 0) {
	133	goto error;
	134	}
	135
00e2e675 DG	136	/* Get the right path name destination */
	137	if (consumer->type == CONSUMER_DST_LOCAL) {
	138	/* Set application path to the destination path */
a4b92340 DG	139	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s/%s",
a4b92340 DG	140	consumer->dst.trace_path, consumer->subdir, usess->path);
00e2e675 DG	141	if (ret < 0) {
	142	PERROR("snprintf stream path");
	143	goto error;
	144	}
	145	pathname = tmp_path;
	146	DBG3("UST local consumer tracefile path: %s", pathname);
	147	} else {
	148	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
37278a1e	149	consumer->subdir, usess->path);
00e2e675 DG	150	if (ret < 0) {
	151	PERROR("snprintf stream path");
	152	goto error;
	153	}
	154	pathname = tmp_path;
	155	DBG3("UST network consumer subdir path: %s", pathname);
	156	}
	157
d80a6244	158	cds_list_for_each_entry_safe(stream, tmp, &uchan->streams.head, list) {
48842b30	159	if (!stream->obj->shm_fd) {
5af2f756	160	continue;
48842b30	161	}
48842b30	162
37278a1e	163	ret = send_channel_stream(sock, uchan, usess, stream, consumer,
00e2e675	164	pathname);
48842b30	165	if (ret < 0) {
48842b30 DG	166	goto error;
48842b30 DG	167	}
48842b30	168	}
48842b30	169
00e2e675	170	DBG("UST consumer channel streams sent");
48842b30 DG	171
	172	return 0;
	173
	174	error:
	175	return ret;
	176	}
	177
	178	/*
37278a1e	179	* Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
48842b30	180	*/
173af62f	181	static int send_metadata(int sock, struct ust_app_session *usess,
00e2e675	182	struct consumer_output *consumer)
48842b30	183	{
37278a1e	184	int ret, fd, fds[2];
00e2e675 DG	185	char tmp_path[PATH_MAX];
00e2e675 DG	186	const char *pathname;
37278a1e	187	struct lttcomm_consumer_msg msg;
48842b30	188
37278a1e DG	189	/* Safety net */
	190	assert(usess);
	191	assert(consumer);
48842b30	192
37278a1e DG	193	if (sock < 0) {
37278a1e DG	194	ERR("Consumer socket is negative (%d)", sock);
7753dea8 MD	195	return -EINVAL;
	196	}
	197
37278a1e DG	198	if (usess->metadata->obj->shm_fd == 0) {
	199	ERR("Metadata obj shm_fd is 0");
	200	ret = -1;
	201	goto error;
	202	}
48842b30	203
37278a1e	204	DBG("UST consumer sending metadata stream fd");
00e2e675	205
37278a1e DG	206	consumer_init_channel_comm_msg(&msg,
	207	LTTNG_CONSUMER_ADD_CHANNEL,
	208	usess->metadata->obj->shm_fd,
	209	usess->metadata->attr.subbuf_size,
	210	usess->metadata->obj->memory_map_size,
	211	"metadata");
	212
	213	ret = consumer_send_channel(sock, &msg);
	214	if (ret < 0) {
	215	goto error;
	216	}
	217
	218	/* Sending metadata shared memory fd */
	219	fd = usess->metadata->obj->shm_fd;
	220	ret = consumer_send_fds(sock, &fd, 1);
	221	if (ret < 0) {
	222	goto error;
	223	}
00e2e675	224
37278a1e DG	225	/* Get correct path name destination */
	226	if (consumer->type == CONSUMER_DST_LOCAL) {
	227	/* Set application path to the destination path */
a4b92340 DG	228	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s/%s",
a4b92340 DG	229	consumer->dst.trace_path, consumer->subdir, usess->path);
48842b30	230	if (ret < 0) {
37278a1e	231	PERROR("snprintf stream path");
48842b30 DG	232	goto error;
48842b30 DG	233	}
37278a1e	234	pathname = tmp_path;
48842b30	235
37278a1e DG	236	/* Create directory */
	237	ret = run_as_mkdir_recursive(pathname, S_IRWXU \| S_IRWXG,
	238	usess->uid, usess->gid);
	239	if (ret < 0) {
	240	if (ret != -EEXIST) {
	241	ERR("Trace directory creation error");
00e2e675 DG	242	goto error;
00e2e675 DG	243	}
48842b30	244	}
37278a1e DG	245	} else {
	246	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
	247	consumer->subdir, usess->path);
48842b30	248	if (ret < 0) {
37278a1e	249	PERROR("snprintf metadata path");
48842b30 DG	250	goto error;
48842b30 DG	251	}
37278a1e DG	252	pathname = tmp_path;
	253	}
	254
	255	consumer_init_stream_comm_msg(&msg,
	256	LTTNG_CONSUMER_ADD_STREAM,
	257	usess->metadata->obj->shm_fd,
	258	usess->metadata->stream_obj->shm_fd,
	259	LTTNG_CONSUMER_ACTIVE_STREAM,
	260	DEFAULT_UST_CHANNEL_OUTPUT,
	261	usess->metadata->stream_obj->memory_map_size,
	262	usess->uid,
	263	usess->gid,
	264	consumer->net_seq_index,
	265	1, /* Flag metadata set */
	266	"metadata",
	267	pathname);
	268
	269	/* Send stream and file descriptor */
	270	fds[0] = usess->metadata->stream_obj->shm_fd;
	271	fds[1] = usess->metadata->stream_obj->wait_fd;
	272	ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
	273	if (ret < 0) {
	274	goto error;
	275	}
	276
	277	error:
	278	return ret;
	279	}
	280
	281	/*
	282	* Send all stream fds of the UST session to the consumer.
	283	*/
173af62f DG	284	int ust_consumer_send_session(struct ust_app_session *usess,
173af62f DG	285	struct consumer_output consumer, struct consumer_socket sock)
37278a1e DG	286	{
37278a1e DG	287	int ret = 0;
37278a1e DG	288	struct lttng_ht_iter iter;
	289	struct ust_app_channel *ua_chan;
	290
173af62f	291	assert(usess);
a4b92340 DG	292
	293	if (consumer == NULL \|\| sock == NULL) {
	294	/* There is no consumer so just ignoring the command. */
	295	DBG("UST consumer does not exist. Not sending streams");
	296	return 0;
	297	}
37278a1e	298
173af62f DG	299	DBG("Sending metadata stream fd to consumer on %d", sock->fd);
	300
	301	pthread_mutex_lock(sock->lock);
37278a1e DG	302
37278a1e DG	303	/* Sending metadata information to the consumer */
173af62f	304	ret = send_metadata(sock->fd, usess, consumer);
37278a1e DG	305	if (ret < 0) {
37278a1e DG	306	goto error;
48842b30 DG	307	}
	308
	309	/* Send each channel fd streams of session */
	310	rcu_read_lock();
bec39940 DG	311	cds_lfht_for_each_entry(usess->channels->ht, &iter.iter, ua_chan,
bec39940 DG	312	node.node) {
aeb96892 DG	313	/*
	314	* Indicate that the channel was not created on the tracer side so skip
	315	* sending unexisting streams.
	316	*/
	317	if (ua_chan->obj == NULL) {
	318	continue;
	319	}
	320
173af62f	321	ret = send_channel_streams(sock->fd, ua_chan, usess, consumer);
48842b30	322	if (ret < 0) {
5485f822	323	rcu_read_unlock();
48842b30 DG	324	goto error;
48842b30 DG	325	}
48842b30 DG	326	}
	327	rcu_read_unlock();
	328
	329	DBG("consumer fds (metadata and channel streams) sent");
	330
173af62f DG	331	/* All good! */
173af62f DG	332	ret = 0;
48842b30 DG	333
48842b30 DG	334	error:
173af62f	335	pthread_mutex_unlock(sock->lock);
48842b30 DG	336	return ret;
48842b30 DG	337	}