[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(struct consumer_socket *sock,
		struct ust_app_channel *uchan)
{
	int ret, fd;
	struct lttcomm_consumer_msg msg;

	/* Safety net */
	assert(uchan);
	assert(sock);

	if (sock->fd < 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(struct consumer_socket *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);
	assert(sock);

	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,
			usess->id);

	/* 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(struct consumer_socket *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;

	assert(sock);

	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(struct consumer_socket *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);
	assert(sock);

	if (sock->fd < 0) {
		ERR("Consumer socket is negative (%d)", sock->fd);
		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,
			usess->id);

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