[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/compat/fcntl.h>

#include "kernel-consumer.h"

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

/*
 * Mmap the ring buffer, read it and write the data to the tracefile.
 *
 * Returns the number of bytes written
 */
ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
		struct lttng_consumer_local_data *ctx,
		struct lttng_consumer_stream *stream, unsigned long len)
{
	unsigned long mmap_offset;
	ssize_t ret = 0;
	off_t orig_offset = stream->out_fd_offset;
	int fd = stream->wait_fd;
	int outfd = stream->out_fd;

	/* get the offset inside the fd to mmap */
	ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
	if (ret != 0) {
		errno = -ret;
		perror("kernctl_get_mmap_read_offset");
		goto end;
	}

	while (len > 0) {
		ret = write(outfd, stream->mmap_base + mmap_offset, len);
		if (ret >= len) {
			len = 0;
		} else if (ret < 0) {
			errno = -ret;
			perror("Error in file write");
			goto end;
		}
		/* This won't block, but will start writeout asynchronously */
		lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
				SYNC_FILE_RANGE_WRITE);
		stream->out_fd_offset += ret;
	}

	lttng_consumer_sync_trace_file(stream, orig_offset);

	goto end;

end:
	return ret;
}

/*
 * Splice the data from the ring buffer to the tracefile.
 *
 * Returns the number of bytes spliced.
 */
ssize_t lttng_kconsumer_on_read_subbuffer_splice(
		struct lttng_consumer_local_data *ctx,
		struct lttng_consumer_stream *stream, unsigned long len)
{
	ssize_t ret = 0;
	loff_t offset = 0;
	off_t orig_offset = stream->out_fd_offset;
	int fd = stream->wait_fd;
	int outfd = stream->out_fd;

	while (len > 0) {
		DBG("splice chan to pipe offset %lu (fd : %d)",
				(unsigned long)offset, fd);
		ret = splice(fd, &offset, ctx->consumer_thread_pipe[1], NULL, len,
				SPLICE_F_MOVE | SPLICE_F_MORE);
		DBG("splice chan to pipe ret %zd", ret);
		if (ret < 0) {
			errno = -ret;
			perror("Error in relay splice");
			goto splice_error;
		}

		ret = splice(ctx->consumer_thread_pipe[0], NULL, outfd, NULL, ret,
				SPLICE_F_MOVE | SPLICE_F_MORE);
		DBG("splice pipe to file %zd", ret);
		if (ret < 0) {
			errno = -ret;
			perror("Error in file splice");
			goto splice_error;
		}
		len -= ret;
		/* This won't block, but will start writeout asynchronously */
		lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
				SYNC_FILE_RANGE_WRITE);
		stream->out_fd_offset += ret;
	}
	lttng_consumer_sync_trace_file(stream, orig_offset);

	goto end;

splice_error:
	/* send the appropriate error description to sessiond */
	switch(ret) {
	case EBADF:
		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EBADF);
		break;
	case EINVAL:
		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EINVAL);
		break;
	case ENOMEM:
		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ENOMEM);
		break;
	case ESPIPE:
		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ESPIPE);
		break;
	}

end:
	return ret;
}

/*
 * 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;
	}

	switch (msg.cmd_type) {
	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:
	{
		struct lttng_consumer_stream *new_stream;
		int fd;

		/* block */
		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
			return -EINTR;
		}
		ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
		if (ret != sizeof(fd)) {
			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
			return ret;
		}

		DBG("consumer_add_stream %s (%d)", msg.u.stream.path_name,
			fd);
		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);
		if (new_stream == NULL) {
			lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
			goto end;
		}
		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;
			}
		} else {
			consumer_add_stream(new_stream);
		}
		break;
	}
	case LTTNG_CONSUMER_UPDATE_STREAM:
	{
		if (ctx->on_update_stream != NULL) {
			ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
			if (ret == 0) {
				consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
			} else if (ret < 0) {
				goto end;
			}
		} else {
			consumer_change_stream_state(msg.u.stream.stream_key,
				msg.u.stream.state);
		}
		break;
	}
	default:
		break;
	}
end:
	/*
	 * 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 == -1UL && errno == EINTR);
end_nosignal:
	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 < 0) {
				/*
				 * display the error but continue processing to try
				 * to release the subbuffer
				 */
				ERR("Error splicing to tracefile");
			}
			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 < 0) {
				/*
				 * 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 (stream->path_name != NULL) {
		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>
dbb5dfe6	34	#include <common/compat/fcntl.h>
0857097f	35
10a8a223	36	#include "kernel-consumer.h"
3bd1e081 MD	37
	38	extern struct lttng_consumer_global_data consumer_data;
	39	extern int consumer_poll_timeout;
	40	extern volatile int consumer_quit;
	41
	42	/*
	43	* Mmap the ring buffer, read it and write the data to the tracefile.
	44	*
	45	* Returns the number of bytes written
	46	*/
4078b776	47	ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
3bd1e081 MD	48	struct lttng_consumer_local_data *ctx,
	49	struct lttng_consumer_stream *stream, unsigned long len)
	50	{
	51	unsigned long mmap_offset;
4078b776	52	ssize_t ret = 0;
3bd1e081 MD	53	off_t orig_offset = stream->out_fd_offset;
	54	int fd = stream->wait_fd;
	55	int outfd = stream->out_fd;
	56
	57	/* get the offset inside the fd to mmap */
	58	ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
	59	if (ret != 0) {
87dc6a9c	60	errno = -ret;
3bd1e081 MD	61	perror("kernctl_get_mmap_read_offset");
	62	goto end;
	63	}
	64
	65	while (len > 0) {
	66	ret = write(outfd, stream->mmap_base + mmap_offset, len);
	67	if (ret >= len) {
	68	len = 0;
	69	} else if (ret < 0) {
87dc6a9c	70	errno = -ret;
3bd1e081 MD	71	perror("Error in file write");
	72	goto end;
	73	}
	74	/* This won't block, but will start writeout asynchronously */
dbb5dfe6	75	lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
3bd1e081 MD	76	SYNC_FILE_RANGE_WRITE);
	77	stream->out_fd_offset += ret;
	78	}
	79
	80	lttng_consumer_sync_trace_file(stream, orig_offset);
	81
	82	goto end;
	83
	84	end:
	85	return ret;
	86	}
	87
	88	/*
	89	* Splice the data from the ring buffer to the tracefile.
	90	*
	91	* Returns the number of bytes spliced.
	92	*/
4078b776	93	ssize_t lttng_kconsumer_on_read_subbuffer_splice(
3bd1e081 MD	94	struct lttng_consumer_local_data *ctx,
	95	struct lttng_consumer_stream *stream, unsigned long len)
	96	{
4078b776	97	ssize_t ret = 0;
3bd1e081 MD	98	loff_t offset = 0;
	99	off_t orig_offset = stream->out_fd_offset;
	100	int fd = stream->wait_fd;
	101	int outfd = stream->out_fd;
	102
	103	while (len > 0) {
	104	DBG("splice chan to pipe offset %lu (fd : %d)",
	105	(unsigned long)offset, fd);
	106	ret = splice(fd, &offset, ctx->consumer_thread_pipe[1], NULL, len,
	107	SPLICE_F_MOVE \| SPLICE_F_MORE);
630543f2	108	DBG("splice chan to pipe ret %zd", ret);
3bd1e081	109	if (ret < 0) {
87dc6a9c	110	errno = -ret;
3bd1e081 MD	111	perror("Error in relay splice");
	112	goto splice_error;
	113	}
	114
	115	ret = splice(ctx->consumer_thread_pipe[0], NULL, outfd, NULL, ret,
	116	SPLICE_F_MOVE \| SPLICE_F_MORE);
630543f2	117	DBG("splice pipe to file %zd", ret);
3bd1e081	118	if (ret < 0) {
87dc6a9c	119	errno = -ret;
3bd1e081 MD	120	perror("Error in file splice");
	121	goto splice_error;
	122	}
	123	len -= ret;
	124	/* This won't block, but will start writeout asynchronously */
dbb5dfe6	125	lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
3bd1e081 MD	126	SYNC_FILE_RANGE_WRITE);
	127	stream->out_fd_offset += ret;
	128	}
	129	lttng_consumer_sync_trace_file(stream, orig_offset);
	130
	131	goto end;
	132
	133	splice_error:
	134	/* send the appropriate error description to sessiond */
	135	switch(ret) {
	136	case EBADF:
	137	lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EBADF);
	138	break;
	139	case EINVAL:
	140	lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EINVAL);
	141	break;
	142	case ENOMEM:
	143	lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ENOMEM);
	144	break;
	145	case ESPIPE:
	146	lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ESPIPE);
	147	break;
	148	}
	149
	150	end:
	151	return ret;
	152	}
	153
	154	/*
	155	* Take a snapshot for a specific fd
	156	*
	157	* Returns 0 on success, < 0 on error
	158	*/
	159	int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
	160	struct lttng_consumer_stream *stream)
	161	{
	162	int ret = 0;
	163	int infd = stream->wait_fd;
	164
	165	ret = kernctl_snapshot(infd);
	166	if (ret != 0) {
87dc6a9c	167	errno = -ret;
3bd1e081 MD	168	perror("Getting sub-buffer snapshot.");
	169	}
	170
	171	return ret;
	172	}
	173
	174	/*
	175	* Get the produced position
	176	*
	177	* Returns 0 on success, < 0 on error
	178	*/
	179	int lttng_kconsumer_get_produced_snapshot(
	180	struct lttng_consumer_local_data *ctx,
	181	struct lttng_consumer_stream *stream,
	182	unsigned long *pos)
	183	{
	184	int ret;
	185	int infd = stream->wait_fd;
	186
	187	ret = kernctl_snapshot_get_produced(infd, pos);
	188	if (ret != 0) {
87dc6a9c	189	errno = -ret;
3bd1e081 MD	190	perror("kernctl_snapshot_get_produced");
	191	}
	192
	193	return ret;
	194	}
	195
	196	int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
	197	int sock, struct pollfd *consumer_sockpoll)
	198	{
	199	ssize_t ret;
	200	struct lttcomm_consumer_msg msg;
	201
	202	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
	203	if (ret != sizeof(msg)) {
f2fc6720	204	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
3bd1e081 MD	205	return ret;
	206	}
	207	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
	208	return -ENOENT;
	209	}
	210
	211	switch (msg.cmd_type) {
	212	case LTTNG_CONSUMER_ADD_CHANNEL:
	213	{
	214	struct lttng_consumer_channel *new_channel;
	215
	216	DBG("consumer_add_channel %d", msg.u.channel.channel_key);
	217	new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
	218	-1, -1,
	219	msg.u.channel.mmap_len,
	220	msg.u.channel.max_sb_size);
	221	if (new_channel == NULL) {
	222	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	223	goto end_nosignal;
	224	}
	225	if (ctx->on_recv_channel != NULL) {
	226	ret = ctx->on_recv_channel(new_channel);
	227	if (ret == 0) {
	228	consumer_add_channel(new_channel);
	229	} else if (ret < 0) {
	230	goto end_nosignal;
	231	}
	232	} else {
	233	consumer_add_channel(new_channel);
	234	}
	235	goto end_nosignal;
	236	}
	237	case LTTNG_CONSUMER_ADD_STREAM:
	238	{
	239	struct lttng_consumer_stream *new_stream;
f2fc6720	240	int fd;
3bd1e081 MD	241
	242	/* block */
	243	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
	244	return -EINTR;
	245	}
f2fc6720 MD	246	ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
f2fc6720 MD	247	if (ret != sizeof(fd)) {
3bd1e081 MD	248	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
	249	return ret;
	250	}
3bd1e081	251
f2fc6720 MD	252	DBG("consumer_add_stream %s (%d)", msg.u.stream.path_name,
f2fc6720 MD	253	fd);
3bd1e081 MD	254	new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
3bd1e081 MD	255	msg.u.stream.stream_key,
f2fc6720	256	fd, fd,
3bd1e081 MD	257	msg.u.stream.state,
	258	msg.u.stream.mmap_len,
	259	msg.u.stream.output,
6df2e2c9 MD	260	msg.u.stream.path_name,
	261	msg.u.stream.uid,
	262	msg.u.stream.gid);
3bd1e081 MD	263	if (new_stream == NULL) {
	264	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	265	goto end;
	266	}
	267	if (ctx->on_recv_stream != NULL) {
	268	ret = ctx->on_recv_stream(new_stream);
	269	if (ret == 0) {
	270	consumer_add_stream(new_stream);
	271	} else if (ret < 0) {
	272	goto end;
	273	}
	274	} else {
	275	consumer_add_stream(new_stream);
	276	}
	277	break;
	278	}
	279	case LTTNG_CONSUMER_UPDATE_STREAM:
	280	{
	281	if (ctx->on_update_stream != NULL) {
	282	ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
	283	if (ret == 0) {
	284	consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
	285	} else if (ret < 0) {
	286	goto end;
	287	}
	288	} else {
	289	consumer_change_stream_state(msg.u.stream.stream_key,
	290	msg.u.stream.state);
	291	}
	292	break;
	293	}
	294	default:
	295	break;
	296	}
	297	end:
04fdd819 MD	298	/*
	299	* Wake-up the other end by writing a null byte in the pipe
	300	* (non-blocking). Important note: Because writing into the
	301	* pipe is non-blocking (and therefore we allow dropping wakeup
	302	* data, as long as there is wakeup data present in the pipe
	303	* buffer to wake up the other end), the other end should
	304	* perform the following sequence for waiting:
	305	* 1) empty the pipe (reads).
	306	* 2) perform update operation.
	307	* 3) wait on the pipe (poll).
	308	*/
	309	do {
	310	ret = write(ctx->consumer_poll_pipe[1], "", 1);
	311	} while (ret == -1UL && errno == EINTR);
3bd1e081 MD	312	end_nosignal:
	313	return 0;
	314	}
d41f73b7 MD	315
	316	/*
	317	* Consume data on a file descriptor and write it on a trace file.
	318	*/
4078b776	319	ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7 MD	320	struct lttng_consumer_local_data *ctx)
	321	{
	322	unsigned long len;
	323	int err;
4078b776	324	ssize_t ret = 0;
d41f73b7 MD	325	int infd = stream->wait_fd;
	326
	327	DBG("In read_subbuffer (infd : %d)", infd);
	328	/* Get the next subbuffer */
	329	err = kernctl_get_next_subbuf(infd);
	330	if (err != 0) {
d41f73b7 MD	331	/*
	332	* This is a debug message even for single-threaded consumer,
	333	* because poll() have more relaxed criterions than get subbuf,
	334	* so get_subbuf may fail for short race windows where poll()
	335	* would issue wakeups.
	336	*/
	337	DBG("Reserving sub buffer failed (everything is normal, "
	338	"it is due to concurrency)");
	339	goto end;
	340	}
	341
	342	switch (stream->output) {
	343	case LTTNG_EVENT_SPLICE:
	344	/* read the whole subbuffer */
	345	err = kernctl_get_padded_subbuf_size(infd, &len);
	346	if (err != 0) {
87dc6a9c	347	errno = -ret;
d41f73b7 MD	348	perror("Getting sub-buffer len failed.");
	349	goto end;
	350	}
	351
	352	/* splice the subbuffer to the tracefile */
	353	ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
	354	if (ret < 0) {
	355	/*
	356	* display the error but continue processing to try
	357	* to release the subbuffer
	358	*/
	359	ERR("Error splicing to tracefile");
	360	}
	361	break;
	362	case LTTNG_EVENT_MMAP:
	363	/* read the used subbuffer size */
	364	err = kernctl_get_padded_subbuf_size(infd, &len);
	365	if (err != 0) {
87dc6a9c	366	errno = -ret;
d41f73b7 MD	367	perror("Getting sub-buffer len failed.");
	368	goto end;
	369	}
	370	/* write the subbuffer to the tracefile */
	371	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
	372	if (ret < 0) {
	373	/*
	374	* display the error but continue processing to try
	375	* to release the subbuffer
	376	*/
	377	ERR("Error writing to tracefile");
	378	}
	379	break;
	380	default:
	381	ERR("Unknown output method");
	382	ret = -1;
	383	}
	384
	385	err = kernctl_put_next_subbuf(infd);
	386	if (err != 0) {
87dc6a9c	387	errno = -ret;
d41f73b7 MD	388	if (errno == EFAULT) {
	389	perror("Error in unreserving sub buffer\n");
	390	} else if (errno == EIO) {
	391	/* Should never happen with newer LTTng versions */
	392	perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
	393	}
	394	goto end;
	395	}
	396
	397	end:
	398	return ret;
	399	}
	400
	401	int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
	402	{
	403	int ret;
	404
	405	/* Opening the tracefile in write mode */
	406	if (stream->path_name != NULL) {
e11d277b	407	ret = run_as_open(stream->path_name,
60b6c79c MD	408	O_WRONLY\|O_CREAT\|O_TRUNC,
	409	S_IRWXU\|S_IRWXG\|S_IRWXO,
	410	stream->uid, stream->gid);
d41f73b7 MD	411	if (ret < 0) {
	412	ERR("Opening %s", stream->path_name);
	413	perror("open");
	414	goto error;
	415	}
	416	stream->out_fd = ret;
	417	}
	418
	419	if (stream->output == LTTNG_EVENT_MMAP) {
	420	/* get the len of the mmap region */
	421	unsigned long mmap_len;
	422
	423	ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
	424	if (ret != 0) {
87dc6a9c	425	errno = -ret;
d41f73b7 MD	426	perror("kernctl_get_mmap_len");
	427	goto error_close_fd;
	428	}
	429	stream->mmap_len = (size_t) mmap_len;
	430
	431	stream->mmap_base = mmap(NULL, stream->mmap_len,
	432	PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
	433	if (stream->mmap_base == MAP_FAILED) {
	434	perror("Error mmaping");
	435	ret = -1;
	436	goto error_close_fd;
	437	}
	438	}
	439
	440	/* we return 0 to let the library handle the FD internally */
	441	return 0;
	442
	443	error_close_fd:
	444	{
	445	int err;
	446
	447	err = close(stream->out_fd);
	448	assert(!err);
	449	}
	450	error:
	451	return ret;
	452	}
	453