[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 <inttypes.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:
	{
		ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
				msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
				&msg.u.relayd_sock.sock);
		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 %" PRIu64,
				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 %" PRIu64,
					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: %zd != len: %lu)",
						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>
77c7c900	28	#include <inttypes.h>
3bd1e081	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>
00e2e675	35	#include <common/sessiond-comm/relayd.h>
dbb5dfe6	36	#include <common/compat/fcntl.h>
00e2e675	37	#include <common/relayd/relayd.h>
0857097f	38
10a8a223	39	#include "kernel-consumer.h"
3bd1e081 MD	40
	41	extern struct lttng_consumer_global_data consumer_data;
	42	extern int consumer_poll_timeout;
	43	extern volatile int consumer_quit;
	44
3bd1e081 MD	45	/*
	46	* Take a snapshot for a specific fd
	47	*
	48	* Returns 0 on success, < 0 on error
	49	*/
	50	int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
	51	struct lttng_consumer_stream *stream)
	52	{
	53	int ret = 0;
	54	int infd = stream->wait_fd;
	55
	56	ret = kernctl_snapshot(infd);
	57	if (ret != 0) {
87dc6a9c	58	errno = -ret;
3bd1e081 MD	59	perror("Getting sub-buffer snapshot.");
	60	}
	61
	62	return ret;
	63	}
	64
	65	/*
	66	* Get the produced position
	67	*
	68	* Returns 0 on success, < 0 on error
	69	*/
	70	int lttng_kconsumer_get_produced_snapshot(
	71	struct lttng_consumer_local_data *ctx,
	72	struct lttng_consumer_stream *stream,
	73	unsigned long *pos)
	74	{
	75	int ret;
	76	int infd = stream->wait_fd;
	77
	78	ret = kernctl_snapshot_get_produced(infd, pos);
	79	if (ret != 0) {
87dc6a9c	80	errno = -ret;
3bd1e081 MD	81	perror("kernctl_snapshot_get_produced");
	82	}
	83
	84	return ret;
	85	}
	86
	87	int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
	88	int sock, struct pollfd *consumer_sockpoll)
	89	{
	90	ssize_t ret;
	91	struct lttcomm_consumer_msg msg;
	92
	93	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
	94	if (ret != sizeof(msg)) {
f2fc6720	95	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
3bd1e081 MD	96	return ret;
	97	}
	98	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
	99	return -ENOENT;
	100	}
	101
b0b335c8 MD	102	/* relayd needs RCU read-side protection */
	103	rcu_read_lock();
	104
3bd1e081	105	switch (msg.cmd_type) {
00e2e675 DG	106	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
00e2e675 DG	107	{
7735ef9e DG	108	ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
	109	msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
	110	&msg.u.relayd_sock.sock);
00e2e675 DG	111	goto end_nosignal;
00e2e675 DG	112	}
3bd1e081 MD	113	case LTTNG_CONSUMER_ADD_CHANNEL:
	114	{
	115	struct lttng_consumer_channel *new_channel;
	116
	117	DBG("consumer_add_channel %d", msg.u.channel.channel_key);
	118	new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
	119	-1, -1,
	120	msg.u.channel.mmap_len,
	121	msg.u.channel.max_sb_size);
	122	if (new_channel == NULL) {
	123	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	124	goto end_nosignal;
	125	}
	126	if (ctx->on_recv_channel != NULL) {
	127	ret = ctx->on_recv_channel(new_channel);
	128	if (ret == 0) {
	129	consumer_add_channel(new_channel);
	130	} else if (ret < 0) {
	131	goto end_nosignal;
	132	}
	133	} else {
	134	consumer_add_channel(new_channel);
	135	}
	136	goto end_nosignal;
	137	}
	138	case LTTNG_CONSUMER_ADD_STREAM:
	139	{
f2fc6720	140	int fd;
00e2e675 DG	141	struct consumer_relayd_sock_pair *relayd = NULL;
00e2e675 DG	142	struct lttng_consumer_stream *new_stream;
3bd1e081 MD	143
	144	/* block */
	145	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
3f8e211f	146	rcu_read_unlock();
3bd1e081 MD	147	return -EINTR;
3bd1e081 MD	148	}
00e2e675 DG	149
00e2e675 DG	150	/* Get stream file descriptor from socket */
f2fc6720 MD	151	ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
f2fc6720 MD	152	if (ret != sizeof(fd)) {
3bd1e081	153	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
3f8e211f	154	rcu_read_unlock();
3bd1e081 MD	155	return ret;
3bd1e081 MD	156	}
3bd1e081	157
3bd1e081 MD	158	new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
3bd1e081 MD	159	msg.u.stream.stream_key,
f2fc6720	160	fd, fd,
3bd1e081 MD	161	msg.u.stream.state,
	162	msg.u.stream.mmap_len,
	163	msg.u.stream.output,
6df2e2c9 MD	164	msg.u.stream.path_name,
6df2e2c9 MD	165	msg.u.stream.uid,
00e2e675 DG	166	msg.u.stream.gid,
	167	msg.u.stream.net_index,
	168	msg.u.stream.metadata_flag);
3bd1e081 MD	169	if (new_stream == NULL) {
3bd1e081 MD	170	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
3f8e211f	171	goto end_nosignal;
3bd1e081	172	}
00e2e675 DG	173
	174	/* The stream is not metadata. Get relayd reference if exists. */
	175	relayd = consumer_find_relayd(msg.u.stream.net_index);
	176	if (relayd != NULL) {
	177	/* Add stream on the relayd */
	178	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
	179	ret = relayd_add_stream(&relayd->control_sock,
	180	msg.u.stream.name, msg.u.stream.path_name,
	181	&new_stream->relayd_stream_id);
	182	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
	183	if (ret < 0) {
3f8e211f	184	goto end_nosignal;
00e2e675 DG	185	}
	186	} else if (msg.u.stream.net_index != -1) {
	187	ERR("Network sequence index %d unknown. Not adding stream.",
	188	msg.u.stream.net_index);
	189	free(new_stream);
3f8e211f	190	goto end_nosignal;
00e2e675 DG	191	}
00e2e675 DG	192
3bd1e081 MD	193	if (ctx->on_recv_stream != NULL) {
	194	ret = ctx->on_recv_stream(new_stream);
	195	if (ret == 0) {
	196	consumer_add_stream(new_stream);
	197	} else if (ret < 0) {
3f8e211f	198	goto end_nosignal;
3bd1e081 MD	199	}
	200	} else {
	201	consumer_add_stream(new_stream);
	202	}
00e2e675 DG	203
00e2e675 DG	204	DBG("Kernel consumer_add_stream (%d)", fd);
3bd1e081 MD	205	break;
	206	}
	207	case LTTNG_CONSUMER_UPDATE_STREAM:
	208	{
3f8e211f DG	209	rcu_read_unlock();
	210	return -ENOSYS;
	211	}
	212	case LTTNG_CONSUMER_DESTROY_RELAYD:
	213	{
	214	struct consumer_relayd_sock_pair *relayd;
	215
77c7c900	216	DBG("Kernel consumer destroying relayd %" PRIu64,
3f8e211f DG	217	msg.u.destroy_relayd.net_seq_idx);
	218
	219	/* Get relayd reference if exists. */
	220	relayd = consumer_find_relayd(msg.u.destroy_relayd.net_seq_idx);
	221	if (relayd == NULL) {
77c7c900 MD	222	ERR("Unable to find relayd %" PRIu64,
77c7c900 MD	223	msg.u.destroy_relayd.net_seq_idx);
3f8e211f	224	goto end_nosignal;
3bd1e081	225	}
3f8e211f DG	226
	227	/* Set destroy flag for this object */
	228	uatomic_set(&relayd->destroy_flag, 1);
	229
	230	/* Destroy the relayd if refcount is 0 else set the destroy flag. */
	231	if (uatomic_read(&relayd->refcount) == 0) {
	232	consumer_destroy_relayd(relayd);
	233	}
	234	goto end_nosignal;
3bd1e081 MD	235	}
3bd1e081 MD	236	default:
3f8e211f	237	goto end_nosignal;
3bd1e081	238	}
3f8e211f	239
04fdd819	240	/*
3f8e211f DG	241	* Wake-up the other end by writing a null byte in the pipe (non-blocking).
	242	* Important note: Because writing into the pipe is non-blocking (and
	243	* therefore we allow dropping wakeup data, as long as there is wakeup data
	244	* present in the pipe buffer to wake up the other end), the other end
	245	* should perform the following sequence for waiting:
	246	*
04fdd819 MD	247	* 1) empty the pipe (reads).
	248	* 2) perform update operation.
	249	* 3) wait on the pipe (poll).
	250	*/
	251	do {
	252	ret = write(ctx->consumer_poll_pipe[1], "", 1);
6f94560a	253	} while (ret < 0 && errno == EINTR);
3bd1e081	254	end_nosignal:
b0b335c8	255	rcu_read_unlock();
3bd1e081 MD	256	return 0;
3bd1e081 MD	257	}
d41f73b7 MD	258
	259	/*
	260	* Consume data on a file descriptor and write it on a trace file.
	261	*/
4078b776	262	ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7 MD	263	struct lttng_consumer_local_data *ctx)
	264	{
	265	unsigned long len;
	266	int err;
4078b776	267	ssize_t ret = 0;
d41f73b7 MD	268	int infd = stream->wait_fd;
	269
	270	DBG("In read_subbuffer (infd : %d)", infd);
	271	/* Get the next subbuffer */
	272	err = kernctl_get_next_subbuf(infd);
	273	if (err != 0) {
d41f73b7 MD	274	/*
	275	* This is a debug message even for single-threaded consumer,
	276	* because poll() have more relaxed criterions than get subbuf,
	277	* so get_subbuf may fail for short race windows where poll()
	278	* would issue wakeups.
	279	*/
	280	DBG("Reserving sub buffer failed (everything is normal, "
	281	"it is due to concurrency)");
	282	goto end;
	283	}
	284
	285	switch (stream->output) {
	286	case LTTNG_EVENT_SPLICE:
	287	/* read the whole subbuffer */
	288	err = kernctl_get_padded_subbuf_size(infd, &len);
	289	if (err != 0) {
87dc6a9c	290	errno = -ret;
d41f73b7 MD	291	perror("Getting sub-buffer len failed.");
	292	goto end;
	293	}
	294
	295	/* splice the subbuffer to the tracefile */
	296	ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
47e81c02	297	if (ret != len) {
d41f73b7 MD	298	/*
	299	* display the error but continue processing to try
	300	* to release the subbuffer
	301	*/
77c7c900	302	ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
00e2e675	303	ret, len);
d41f73b7	304	}
47e81c02	305
d41f73b7 MD	306	break;
	307	case LTTNG_EVENT_MMAP:
	308	/* read the used subbuffer size */
	309	err = kernctl_get_padded_subbuf_size(infd, &len);
	310	if (err != 0) {
87dc6a9c	311	errno = -ret;
d41f73b7 MD	312	perror("Getting sub-buffer len failed.");
	313	goto end;
	314	}
	315	/* write the subbuffer to the tracefile */
	316	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
47e81c02	317	if (ret != len) {
d41f73b7 MD	318	/*
	319	* display the error but continue processing to try
	320	* to release the subbuffer
	321	*/
	322	ERR("Error writing to tracefile");
	323	}
	324	break;
	325	default:
	326	ERR("Unknown output method");
	327	ret = -1;
	328	}
	329
	330	err = kernctl_put_next_subbuf(infd);
	331	if (err != 0) {
87dc6a9c	332	errno = -ret;
d41f73b7 MD	333	if (errno == EFAULT) {
	334	perror("Error in unreserving sub buffer\n");
	335	} else if (errno == EIO) {
	336	/* Should never happen with newer LTTng versions */
	337	perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
	338	}
	339	goto end;
	340	}
	341
	342	end:
	343	return ret;
	344	}
	345
	346	int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
	347	{
	348	int ret;
	349
	350	/* Opening the tracefile in write mode */
00e2e675	351	if (strlen(stream->path_name) > 0 && stream->net_seq_idx == -1) {
e11d277b	352	ret = run_as_open(stream->path_name,
60b6c79c MD	353	O_WRONLY\|O_CREAT\|O_TRUNC,
	354	S_IRWXU\|S_IRWXG\|S_IRWXO,
	355	stream->uid, stream->gid);
d41f73b7 MD	356	if (ret < 0) {
	357	ERR("Opening %s", stream->path_name);
	358	perror("open");
	359	goto error;
	360	}
	361	stream->out_fd = ret;
	362	}
	363
	364	if (stream->output == LTTNG_EVENT_MMAP) {
	365	/* get the len of the mmap region */
	366	unsigned long mmap_len;
	367
	368	ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
	369	if (ret != 0) {
87dc6a9c	370	errno = -ret;
d41f73b7 MD	371	perror("kernctl_get_mmap_len");
	372	goto error_close_fd;
	373	}
	374	stream->mmap_len = (size_t) mmap_len;
	375
	376	stream->mmap_base = mmap(NULL, stream->mmap_len,
	377	PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
	378	if (stream->mmap_base == MAP_FAILED) {
	379	perror("Error mmaping");
	380	ret = -1;
	381	goto error_close_fd;
	382	}
	383	}
	384
	385	/* we return 0 to let the library handle the FD internally */
	386	return 0;
	387
	388	error_close_fd:
	389	{
	390	int err;
	391
	392	err = close(stream->out_fd);
	393	assert(!err);
	394	}
	395	error:
	396	return ret;
	397	}
	398