[lttng-tools.git] / src / common / kernel-consumer / kernel-consumer.c

/*
 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
 *                      Mathieu Desnoyers <mathieu.desnoyers@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 _GNU_SOURCE
#include <assert.h>
#include <poll.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/stat.h>

#include <common/common.h>
#include <common/kernel-ctl/kernel-ctl.h>
#include <common/sessiond-comm/sessiond-comm.h>
#include <common/sessiond-comm/relayd.h>
#include <common/compat/fcntl.h>
#include <common/relayd/relayd.h>

#include "kernel-consumer.h"

extern struct lttng_consumer_global_data consumer_data;
extern int consumer_poll_timeout;
extern volatile int consumer_quit;

/*
 * Take a snapshot for a specific fd
 *
 * Returns 0 on success, < 0 on error
 */
int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
		struct lttng_consumer_stream *stream)
{
	int ret = 0;
	int infd = stream->wait_fd;

	ret = kernctl_snapshot(infd);
	if (ret != 0) {
		errno = -ret;
		perror("Getting sub-buffer snapshot.");
	}

	return ret;
}

/*
 * Get the produced position
 *
 * Returns 0 on success, < 0 on error
 */
int lttng_kconsumer_get_produced_snapshot(
		struct lttng_consumer_local_data *ctx,
		struct lttng_consumer_stream *stream,
		unsigned long *pos)
{
	int ret;
	int infd = stream->wait_fd;

	ret = kernctl_snapshot_get_produced(infd, pos);
	if (ret != 0) {
		errno = -ret;
		perror("kernctl_snapshot_get_produced");
	}

	return ret;
}

int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
		int sock, struct pollfd *consumer_sockpoll)
{
	ssize_t ret;
	struct lttcomm_consumer_msg msg;

	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
	if (ret != sizeof(msg)) {
		lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
		return ret;
	}
	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
		return -ENOENT;
	}

	/* relayd needs RCU read-side protection */
	rcu_read_lock();

	switch (msg.cmd_type) {
	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
	{
		int fd;
		struct consumer_relayd_sock_pair *relayd;

		DBG("Consumer adding relayd socket");

		/* Get relayd reference if exists. */
		relayd = consumer_find_relayd(msg.u.relayd_sock.net_index);
		if (relayd == NULL) {
			/* Not found. Allocate one. */
			relayd = consumer_allocate_relayd_sock_pair(
					msg.u.relayd_sock.net_index);
			if (relayd == NULL) {
				lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
				goto end_nosignal;
			}
		}

		/* Poll on consumer socket. */
		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
			rcu_read_unlock();
			return -EINTR;
		}

		/* Get relayd socket from session daemon */
		ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
		if (ret != sizeof(fd)) {
			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
			goto end_nosignal;
		}

		/* Copy socket information and received FD */
		switch (msg.u.relayd_sock.type) {
		case LTTNG_STREAM_CONTROL:
			/* Copy received lttcomm socket */
			lttcomm_copy_sock(&relayd->control_sock, &msg.u.relayd_sock.sock);

			ret = lttcomm_create_sock(&relayd->control_sock);
			if (ret < 0) {
				goto end_nosignal;
			}

			/* Close the created socket fd which is useless */
			close(relayd->control_sock.fd);

			/* Assign new file descriptor */
			relayd->control_sock.fd = fd;
			break;
		case LTTNG_STREAM_DATA:
			/* Copy received lttcomm socket */
			lttcomm_copy_sock(&relayd->data_sock, &msg.u.relayd_sock.sock);
			ret = lttcomm_create_sock(&relayd->data_sock);
			if (ret < 0) {
				goto end_nosignal;
			}

			/* Close the created socket fd which is useless */
			close(relayd->data_sock.fd);

			/* Assign new file descriptor */
			relayd->data_sock.fd = fd;
			break;
		default:
			ERR("Unknown relayd socket type");
			goto end_nosignal;
		}

		DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
				msg.u.relayd_sock.type == LTTNG_STREAM_CONTROL ? "control" : "data",
				relayd->net_seq_idx, fd);

		/*
		 * Add relayd socket pair to consumer data hashtable. If object already
		 * exists or on error, the function gracefully returns.
		 */
		consumer_add_relayd(relayd);

		goto end_nosignal;
	}
	case LTTNG_CONSUMER_ADD_CHANNEL:
	{
		struct lttng_consumer_channel *new_channel;

		DBG("consumer_add_channel %d", msg.u.channel.channel_key);
		new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
				-1, -1,
				msg.u.channel.mmap_len,
				msg.u.channel.max_sb_size);
		if (new_channel == NULL) {
			lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
			goto end_nosignal;
		}
		if (ctx->on_recv_channel != NULL) {
			ret = ctx->on_recv_channel(new_channel);
			if (ret == 0) {
				consumer_add_channel(new_channel);
			} else if (ret < 0) {
				goto end_nosignal;
			}
		} else {
			consumer_add_channel(new_channel);
		}
		goto end_nosignal;
	}
	case LTTNG_CONSUMER_ADD_STREAM:
	{
		int fd;
		struct consumer_relayd_sock_pair *relayd = NULL;
		struct lttng_consumer_stream *new_stream;

		/* block */
		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
			rcu_read_unlock();
			return -EINTR;
		}

		/* Get stream file descriptor from socket */
		ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
		if (ret != sizeof(fd)) {
			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
			rcu_read_unlock();
			return ret;
		}

		new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
				msg.u.stream.stream_key,
				fd, fd,
				msg.u.stream.state,
				msg.u.stream.mmap_len,
				msg.u.stream.output,
				msg.u.stream.path_name,
				msg.u.stream.uid,
				msg.u.stream.gid,
				msg.u.stream.net_index,
				msg.u.stream.metadata_flag);
		if (new_stream == NULL) {
			lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
			goto end_nosignal;
		}

		/* The stream is not metadata. Get relayd reference if exists. */
		relayd = consumer_find_relayd(msg.u.stream.net_index);
		if (relayd != NULL) {
			/* Add stream on the relayd */
			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
			ret = relayd_add_stream(&relayd->control_sock,
					msg.u.stream.name, msg.u.stream.path_name,
					&new_stream->relayd_stream_id);
			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
			if (ret < 0) {
				goto end_nosignal;
			}
		} else if (msg.u.stream.net_index != -1) {
			ERR("Network sequence index %d unknown. Not adding stream.",
					msg.u.stream.net_index);
			free(new_stream);
			goto end_nosignal;
		}

		if (ctx->on_recv_stream != NULL) {
			ret = ctx->on_recv_stream(new_stream);
			if (ret == 0) {
				consumer_add_stream(new_stream);
			} else if (ret < 0) {
				goto end_nosignal;
			}
		} else {
			consumer_add_stream(new_stream);
		}

		DBG("Kernel consumer_add_stream (%d)", fd);
		break;
	}
	case LTTNG_CONSUMER_UPDATE_STREAM:
	{
		rcu_read_unlock();
		return -ENOSYS;
	}
	case LTTNG_CONSUMER_DESTROY_RELAYD:
	{
		struct consumer_relayd_sock_pair *relayd;

		DBG("Kernel consumer destroying relayd %zu",
				msg.u.destroy_relayd.net_seq_idx);

		/* Get relayd reference if exists. */
		relayd = consumer_find_relayd(msg.u.destroy_relayd.net_seq_idx);
		if (relayd == NULL) {
			ERR("Unable to find relayd %zu", msg.u.destroy_relayd.net_seq_idx);
			goto end_nosignal;
		}

		/* Set destroy flag for this object */
		uatomic_set(&relayd->destroy_flag, 1);

		/* Destroy the relayd if refcount is 0 else set the destroy flag. */
		if (uatomic_read(&relayd->refcount) == 0) {
			consumer_destroy_relayd(relayd);
		}
		goto end_nosignal;
	}
	default:
		goto end_nosignal;
	}

	/*
	 * Wake-up the other end by writing a null byte in the pipe (non-blocking).
	 * Important note: Because writing into the pipe is non-blocking (and
	 * therefore we allow dropping wakeup data, as long as there is wakeup data
	 * present in the pipe buffer to wake up the other end), the other end
	 * should perform the following sequence for waiting:
	 *
	 * 1) empty the pipe (reads).
	 * 2) perform update operation.
	 * 3) wait on the pipe (poll).
	 */
	do {
		ret = write(ctx->consumer_poll_pipe[1], "", 1);
	} while (ret < 0 && errno == EINTR);
end_nosignal:
	rcu_read_unlock();
	return 0;
}

/*
 * Consume data on a file descriptor and write it on a trace file.
 */
ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
		struct lttng_consumer_local_data *ctx)
{
	unsigned long len;
	int err;
	ssize_t ret = 0;
	int infd = stream->wait_fd;

	DBG("In read_subbuffer (infd : %d)", infd);
	/* Get the next subbuffer */
	err = kernctl_get_next_subbuf(infd);
	if (err != 0) {
		/*
		 * This is a debug message even for single-threaded consumer,
		 * because poll() have more relaxed criterions than get subbuf,
		 * so get_subbuf may fail for short race windows where poll()
		 * would issue wakeups.
		 */
		DBG("Reserving sub buffer failed (everything is normal, "
				"it is due to concurrency)");
		goto end;
	}

	switch (stream->output) {
		case LTTNG_EVENT_SPLICE:
			/* read the whole subbuffer */
			err = kernctl_get_padded_subbuf_size(infd, &len);
			if (err != 0) {
				errno = -ret;
				perror("Getting sub-buffer len failed.");
				goto end;
			}

			/* splice the subbuffer to the tracefile */
			ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
			if (ret != len) {
				/*
				 * display the error but continue processing to try
				 * to release the subbuffer
				 */
				ERR("Error splicing to tracefile (ret: %ld != len: %ld)",
						ret, len);
			}

			break;
		case LTTNG_EVENT_MMAP:
			/* read the used subbuffer size */
			err = kernctl_get_padded_subbuf_size(infd, &len);
			if (err != 0) {
				errno = -ret;
				perror("Getting sub-buffer len failed.");
				goto end;
			}
			/* write the subbuffer to the tracefile */
			ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
			if (ret != len) {
				/*
				 * display the error but continue processing to try
				 * to release the subbuffer
				 */
				ERR("Error writing to tracefile");
			}
			break;
		default:
			ERR("Unknown output method");
			ret = -1;
	}

	err = kernctl_put_next_subbuf(infd);
	if (err != 0) {
		errno = -ret;
		if (errno == EFAULT) {
			perror("Error in unreserving sub buffer\n");
		} else if (errno == EIO) {
			/* Should never happen with newer LTTng versions */
			perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
		}
		goto end;
	}

end:
	return ret;
}

int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
{
	int ret;

	/* Opening the tracefile in write mode */
	if (strlen(stream->path_name) > 0 && stream->net_seq_idx == -1) {
		ret = run_as_open(stream->path_name,
				O_WRONLY|O_CREAT|O_TRUNC,
				S_IRWXU|S_IRWXG|S_IRWXO,
				stream->uid, stream->gid);
		if (ret < 0) {
			ERR("Opening %s", stream->path_name);
			perror("open");
			goto error;
		}
		stream->out_fd = ret;
	}

	if (stream->output == LTTNG_EVENT_MMAP) {
		/* get the len of the mmap region */
		unsigned long mmap_len;

		ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
		if (ret != 0) {
			errno = -ret;
			perror("kernctl_get_mmap_len");
			goto error_close_fd;
		}
		stream->mmap_len = (size_t) mmap_len;

		stream->mmap_base = mmap(NULL, stream->mmap_len,
				PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
		if (stream->mmap_base == MAP_FAILED) {
			perror("Error mmaping");
			ret = -1;
			goto error_close_fd;
		}
	}

	/* we return 0 to let the library handle the FD internally */
	return 0;

error_close_fd:
	{
		int err;

		err = close(stream->out_fd);
		assert(!err);
	}
error:
	return ret;
}
Commit	Line	Data
3bd1e081 MD	1	/*
	2	* Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
	3	* Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	4	*
d14d33bf AM	5	* This program is free software; you can redistribute it and/or modify
	6	* it under the terms of the GNU General Public License, version 2 only,
	7	* as published by the Free Software Foundation.
3bd1e081 MD	8	*
	9	* This program is distributed in the hope that it will be useful,
	10	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	11	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	12	* GNU General Public License for more details.
	13	*
d14d33bf AM	14	* You should have received a copy of the GNU General Public License along
	15	* with this program; if not, write to the Free Software Foundation, Inc.,
	16	* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
3bd1e081 MD	17	*/
	18
	19	#define _GNU_SOURCE
	20	#include <assert.h>
3bd1e081 MD	21	#include <poll.h>
	22	#include <pthread.h>
	23	#include <stdlib.h>
	24	#include <string.h>
	25	#include <sys/mman.h>
	26	#include <sys/socket.h>
	27	#include <sys/types.h>
	28	#include <unistd.h>
dbb5dfe6	29	#include <sys/stat.h>
3bd1e081	30
990570ed	31	#include <common/common.h>
10a8a223	32	#include <common/kernel-ctl/kernel-ctl.h>
10a8a223	33	#include <common/sessiond-comm/sessiond-comm.h>
00e2e675	34	#include <common/sessiond-comm/relayd.h>
dbb5dfe6	35	#include <common/compat/fcntl.h>
00e2e675	36	#include <common/relayd/relayd.h>
0857097f	37
10a8a223	38	#include "kernel-consumer.h"
3bd1e081 MD	39
	40	extern struct lttng_consumer_global_data consumer_data;
	41	extern int consumer_poll_timeout;
	42	extern volatile int consumer_quit;
	43
3bd1e081 MD	44	/*
	45	* Take a snapshot for a specific fd
	46	*
	47	* Returns 0 on success, < 0 on error
	48	*/
	49	int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
	50	struct lttng_consumer_stream *stream)
	51	{
	52	int ret = 0;
	53	int infd = stream->wait_fd;
	54
	55	ret = kernctl_snapshot(infd);
	56	if (ret != 0) {
87dc6a9c	57	errno = -ret;
3bd1e081 MD	58	perror("Getting sub-buffer snapshot.");
	59	}
	60
	61	return ret;
	62	}
	63
	64	/*
	65	* Get the produced position
	66	*
	67	* Returns 0 on success, < 0 on error
	68	*/
	69	int lttng_kconsumer_get_produced_snapshot(
	70	struct lttng_consumer_local_data *ctx,
	71	struct lttng_consumer_stream *stream,
	72	unsigned long *pos)
	73	{
	74	int ret;
	75	int infd = stream->wait_fd;
	76
	77	ret = kernctl_snapshot_get_produced(infd, pos);
	78	if (ret != 0) {
87dc6a9c	79	errno = -ret;
3bd1e081 MD	80	perror("kernctl_snapshot_get_produced");
	81	}
	82
	83	return ret;
	84	}
	85
	86	int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
	87	int sock, struct pollfd *consumer_sockpoll)
	88	{
	89	ssize_t ret;
	90	struct lttcomm_consumer_msg msg;
	91
	92	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
	93	if (ret != sizeof(msg)) {
f2fc6720	94	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
3bd1e081 MD	95	return ret;
	96	}
	97	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
	98	return -ENOENT;
	99	}
	100
b0b335c8 MD	101	/* relayd needs RCU read-side protection */
	102	rcu_read_lock();
	103
3bd1e081	104	switch (msg.cmd_type) {
00e2e675 DG	105	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
	106	{
	107	int fd;
	108	struct consumer_relayd_sock_pair *relayd;
	109
	110	DBG("Consumer adding relayd socket");
	111
	112	/* Get relayd reference if exists. */
	113	relayd = consumer_find_relayd(msg.u.relayd_sock.net_index);
	114	if (relayd == NULL) {
	115	/* Not found. Allocate one. */
	116	relayd = consumer_allocate_relayd_sock_pair(
	117	msg.u.relayd_sock.net_index);
	118	if (relayd == NULL) {
	119	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	120	goto end_nosignal;
	121	}
	122	}
	123
	124	/* Poll on consumer socket. */
	125	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
3f8e211f	126	rcu_read_unlock();
00e2e675 DG	127	return -EINTR;
	128	}
	129
	130	/* Get relayd socket from session daemon */
	131	ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
	132	if (ret != sizeof(fd)) {
	133	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
	134	goto end_nosignal;
	135	}
	136
	137	/* Copy socket information and received FD */
	138	switch (msg.u.relayd_sock.type) {
	139	case LTTNG_STREAM_CONTROL:
	140	/* Copy received lttcomm socket */
	141	lttcomm_copy_sock(&relayd->control_sock, &msg.u.relayd_sock.sock);
	142
	143	ret = lttcomm_create_sock(&relayd->control_sock);
	144	if (ret < 0) {
	145	goto end_nosignal;
	146	}
	147
	148	/* Close the created socket fd which is useless */
	149	close(relayd->control_sock.fd);
	150
	151	/* Assign new file descriptor */
	152	relayd->control_sock.fd = fd;
	153	break;
	154	case LTTNG_STREAM_DATA:
	155	/* Copy received lttcomm socket */
	156	lttcomm_copy_sock(&relayd->data_sock, &msg.u.relayd_sock.sock);
	157	ret = lttcomm_create_sock(&relayd->data_sock);
	158	if (ret < 0) {
	159	goto end_nosignal;
	160	}
	161
	162	/* Close the created socket fd which is useless */
	163	close(relayd->data_sock.fd);
	164
	165	/* Assign new file descriptor */
	166	relayd->data_sock.fd = fd;
	167	break;
	168	default:
	169	ERR("Unknown relayd socket type");
	170	goto end_nosignal;
	171	}
	172
	173	DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
	174	msg.u.relayd_sock.type == LTTNG_STREAM_CONTROL ? "control" : "data",
	175	relayd->net_seq_idx, fd);
	176
	177	/*
	178	* Add relayd socket pair to consumer data hashtable. If object already
	179	* exists or on error, the function gracefully returns.
	180	*/
	181	consumer_add_relayd(relayd);
	182
	183	goto end_nosignal;
	184	}
3bd1e081 MD	185	case LTTNG_CONSUMER_ADD_CHANNEL:
	186	{
	187	struct lttng_consumer_channel *new_channel;
	188
	189	DBG("consumer_add_channel %d", msg.u.channel.channel_key);
	190	new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
	191	-1, -1,
	192	msg.u.channel.mmap_len,
	193	msg.u.channel.max_sb_size);
	194	if (new_channel == NULL) {
	195	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	196	goto end_nosignal;
	197	}
	198	if (ctx->on_recv_channel != NULL) {
	199	ret = ctx->on_recv_channel(new_channel);
	200	if (ret == 0) {
	201	consumer_add_channel(new_channel);
	202	} else if (ret < 0) {
	203	goto end_nosignal;
	204	}
	205	} else {
	206	consumer_add_channel(new_channel);
	207	}
	208	goto end_nosignal;
	209	}
	210	case LTTNG_CONSUMER_ADD_STREAM:
	211	{
f2fc6720	212	int fd;
00e2e675 DG	213	struct consumer_relayd_sock_pair *relayd = NULL;
00e2e675 DG	214	struct lttng_consumer_stream *new_stream;
3bd1e081 MD	215
	216	/* block */
	217	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
3f8e211f	218	rcu_read_unlock();
3bd1e081 MD	219	return -EINTR;
3bd1e081 MD	220	}
00e2e675 DG	221
00e2e675 DG	222	/* Get stream file descriptor from socket */
f2fc6720 MD	223	ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
f2fc6720 MD	224	if (ret != sizeof(fd)) {
3bd1e081	225	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
3f8e211f	226	rcu_read_unlock();
3bd1e081 MD	227	return ret;
3bd1e081 MD	228	}
3bd1e081	229
3bd1e081 MD	230	new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
3bd1e081 MD	231	msg.u.stream.stream_key,
f2fc6720	232	fd, fd,
3bd1e081 MD	233	msg.u.stream.state,
	234	msg.u.stream.mmap_len,
	235	msg.u.stream.output,
6df2e2c9 MD	236	msg.u.stream.path_name,
6df2e2c9 MD	237	msg.u.stream.uid,
00e2e675 DG	238	msg.u.stream.gid,
	239	msg.u.stream.net_index,
	240	msg.u.stream.metadata_flag);
3bd1e081 MD	241	if (new_stream == NULL) {
3bd1e081 MD	242	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
3f8e211f	243	goto end_nosignal;
3bd1e081	244	}
00e2e675 DG	245
	246	/* The stream is not metadata. Get relayd reference if exists. */
	247	relayd = consumer_find_relayd(msg.u.stream.net_index);
	248	if (relayd != NULL) {
	249	/* Add stream on the relayd */
	250	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
	251	ret = relayd_add_stream(&relayd->control_sock,
	252	msg.u.stream.name, msg.u.stream.path_name,
	253	&new_stream->relayd_stream_id);
	254	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
	255	if (ret < 0) {
3f8e211f	256	goto end_nosignal;
00e2e675 DG	257	}
	258	} else if (msg.u.stream.net_index != -1) {
	259	ERR("Network sequence index %d unknown. Not adding stream.",
	260	msg.u.stream.net_index);
	261	free(new_stream);
3f8e211f	262	goto end_nosignal;
00e2e675 DG	263	}
00e2e675 DG	264
3bd1e081 MD	265	if (ctx->on_recv_stream != NULL) {
	266	ret = ctx->on_recv_stream(new_stream);
	267	if (ret == 0) {
	268	consumer_add_stream(new_stream);
	269	} else if (ret < 0) {
3f8e211f	270	goto end_nosignal;
3bd1e081 MD	271	}
	272	} else {
	273	consumer_add_stream(new_stream);
	274	}
00e2e675 DG	275
00e2e675 DG	276	DBG("Kernel consumer_add_stream (%d)", fd);
3bd1e081 MD	277	break;
	278	}
	279	case LTTNG_CONSUMER_UPDATE_STREAM:
	280	{
3f8e211f DG	281	rcu_read_unlock();
	282	return -ENOSYS;
	283	}
	284	case LTTNG_CONSUMER_DESTROY_RELAYD:
	285	{
	286	struct consumer_relayd_sock_pair *relayd;
	287
	288	DBG("Kernel consumer destroying relayd %zu",
	289	msg.u.destroy_relayd.net_seq_idx);
	290
	291	/* Get relayd reference if exists. */
	292	relayd = consumer_find_relayd(msg.u.destroy_relayd.net_seq_idx);
	293	if (relayd == NULL) {
	294	ERR("Unable to find relayd %zu", msg.u.destroy_relayd.net_seq_idx);
	295	goto end_nosignal;
3bd1e081	296	}
3f8e211f DG	297
	298	/* Set destroy flag for this object */
	299	uatomic_set(&relayd->destroy_flag, 1);
	300
	301	/* Destroy the relayd if refcount is 0 else set the destroy flag. */
	302	if (uatomic_read(&relayd->refcount) == 0) {
	303	consumer_destroy_relayd(relayd);
	304	}
	305	goto end_nosignal;
3bd1e081 MD	306	}
3bd1e081 MD	307	default:
3f8e211f	308	goto end_nosignal;
3bd1e081	309	}
3f8e211f	310
04fdd819	311	/*
3f8e211f DG	312	* Wake-up the other end by writing a null byte in the pipe (non-blocking).
	313	* Important note: Because writing into the pipe is non-blocking (and
	314	* therefore we allow dropping wakeup data, as long as there is wakeup data
	315	* present in the pipe buffer to wake up the other end), the other end
	316	* should perform the following sequence for waiting:
	317	*
04fdd819 MD	318	* 1) empty the pipe (reads).
	319	* 2) perform update operation.
	320	* 3) wait on the pipe (poll).
	321	*/
	322	do {
	323	ret = write(ctx->consumer_poll_pipe[1], "", 1);
6f94560a	324	} while (ret < 0 && errno == EINTR);
3bd1e081	325	end_nosignal:
b0b335c8	326	rcu_read_unlock();
3bd1e081 MD	327	return 0;
3bd1e081 MD	328	}
d41f73b7 MD	329
	330	/*
	331	* Consume data on a file descriptor and write it on a trace file.
	332	*/
4078b776	333	ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7 MD	334	struct lttng_consumer_local_data *ctx)
	335	{
	336	unsigned long len;
	337	int err;
4078b776	338	ssize_t ret = 0;
d41f73b7 MD	339	int infd = stream->wait_fd;
	340
	341	DBG("In read_subbuffer (infd : %d)", infd);
	342	/* Get the next subbuffer */
	343	err = kernctl_get_next_subbuf(infd);
	344	if (err != 0) {
d41f73b7 MD	345	/*
	346	* This is a debug message even for single-threaded consumer,
	347	* because poll() have more relaxed criterions than get subbuf,
	348	* so get_subbuf may fail for short race windows where poll()
	349	* would issue wakeups.
	350	*/
	351	DBG("Reserving sub buffer failed (everything is normal, "
	352	"it is due to concurrency)");
	353	goto end;
	354	}
	355
	356	switch (stream->output) {
	357	case LTTNG_EVENT_SPLICE:
	358	/* read the whole subbuffer */
	359	err = kernctl_get_padded_subbuf_size(infd, &len);
	360	if (err != 0) {
87dc6a9c	361	errno = -ret;
d41f73b7 MD	362	perror("Getting sub-buffer len failed.");
	363	goto end;
	364	}
	365
	366	/* splice the subbuffer to the tracefile */
	367	ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
47e81c02	368	if (ret != len) {
d41f73b7 MD	369	/*
	370	* display the error but continue processing to try
	371	* to release the subbuffer
	372	*/
00e2e675 DG	373	ERR("Error splicing to tracefile (ret: %ld != len: %ld)",
00e2e675 DG	374	ret, len);
d41f73b7	375	}
47e81c02	376
d41f73b7 MD	377	break;
	378	case LTTNG_EVENT_MMAP:
	379	/* read the used subbuffer size */
	380	err = kernctl_get_padded_subbuf_size(infd, &len);
	381	if (err != 0) {
87dc6a9c	382	errno = -ret;
d41f73b7 MD	383	perror("Getting sub-buffer len failed.");
	384	goto end;
	385	}
	386	/* write the subbuffer to the tracefile */
	387	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
47e81c02	388	if (ret != len) {
d41f73b7 MD	389	/*
	390	* display the error but continue processing to try
	391	* to release the subbuffer
	392	*/
	393	ERR("Error writing to tracefile");
	394	}
	395	break;
	396	default:
	397	ERR("Unknown output method");
	398	ret = -1;
	399	}
	400
	401	err = kernctl_put_next_subbuf(infd);
	402	if (err != 0) {
87dc6a9c	403	errno = -ret;
d41f73b7 MD	404	if (errno == EFAULT) {
	405	perror("Error in unreserving sub buffer\n");
	406	} else if (errno == EIO) {
	407	/* Should never happen with newer LTTng versions */
	408	perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
	409	}
	410	goto end;
	411	}
	412
	413	end:
	414	return ret;
	415	}
	416
	417	int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
	418	{
	419	int ret;
	420
	421	/* Opening the tracefile in write mode */
00e2e675	422	if (strlen(stream->path_name) > 0 && stream->net_seq_idx == -1) {
e11d277b	423	ret = run_as_open(stream->path_name,
60b6c79c MD	424	O_WRONLY\|O_CREAT\|O_TRUNC,
	425	S_IRWXU\|S_IRWXG\|S_IRWXO,
	426	stream->uid, stream->gid);
d41f73b7 MD	427	if (ret < 0) {
	428	ERR("Opening %s", stream->path_name);
	429	perror("open");
	430	goto error;
	431	}
	432	stream->out_fd = ret;
	433	}
	434
	435	if (stream->output == LTTNG_EVENT_MMAP) {
	436	/* get the len of the mmap region */
	437	unsigned long mmap_len;
	438
	439	ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
	440	if (ret != 0) {
87dc6a9c	441	errno = -ret;
d41f73b7 MD	442	perror("kernctl_get_mmap_len");
	443	goto error_close_fd;
	444	}
	445	stream->mmap_len = (size_t) mmap_len;
	446
	447	stream->mmap_base = mmap(NULL, stream->mmap_len,
	448	PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
	449	if (stream->mmap_base == MAP_FAILED) {
	450	perror("Error mmaping");
	451	ret = -1;
	452	goto error_close_fd;
	453	}
	454	}
	455
	456	/* we return 0 to let the library handle the FD internally */
	457	return 0;
	458
	459	error_close_fd:
	460	{
	461	int err;
	462
	463	err = close(stream->out_fd);
	464	assert(!err);
	465	}
	466	error:
	467	return ret;
	468	}
	469