[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
 * as published by the Free Software Foundation; only version 2
 * of the License.
 *
 * 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., 59 Temple Place - Suite 330, Boston, MA  02111-1307, 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
 */
int lttng_kconsumer_on_read_subbuffer_mmap(
		struct lttng_consumer_local_data *ctx,
		struct lttng_consumer_stream *stream, unsigned long len)
{
	unsigned long mmap_offset;
	long 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.
 */
int lttng_kconsumer_on_read_subbuffer_splice(
		struct lttng_consumer_local_data *ctx,
		struct lttng_consumer_stream *stream, unsigned long len)
{
	long 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 %ld", 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 %ld", 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:
	/* signal the poll thread */
	ret = write(ctx->consumer_poll_pipe[1], "4", 1);
	if (ret < 0) {
		perror("write consumer poll");
	}
end_nosignal:
	return 0;
}

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