[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:
	{
		uint64_t index = msg.u.destroy_relayd.net_seq_idx;
		struct consumer_relayd_sock_pair *relayd;

		DBG("Kernel consumer destroying relayd %" PRIu64, index);

		/* Get relayd reference if exists. */
		relayd = consumer_find_relayd(index);
		if (relayd == NULL) {
			ERR("Unable to find relayd %" PRIu64, index);
			goto end_nosignal;
		}

		/*
		 * Each relayd socket pair has a refcount of stream attached to it
		 * which tells if the relayd is still active or not depending on the
		 * refcount value.
		 *
		 * This will set the destroy flag of the relayd object and destroy it
		 * if the refcount reaches zero when called.
		 *
		 * The destroy can happen either here or when a stream fd hangs up.
		 */
		consumer_flag_relayd_for_destroy(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	{
a6ba4fe1	214	uint64_t index = msg.u.destroy_relayd.net_seq_idx;
3f8e211f DG	215	struct consumer_relayd_sock_pair *relayd;
3f8e211f DG	216
a6ba4fe1	217	DBG("Kernel consumer destroying relayd %" PRIu64, index);
3f8e211f DG	218
3f8e211f DG	219	/* Get relayd reference if exists. */
a6ba4fe1	220	relayd = consumer_find_relayd(index);
3f8e211f	221	if (relayd == NULL) {
a6ba4fe1	222	ERR("Unable to find relayd %" PRIu64, index);
3f8e211f	223	goto end_nosignal;
3bd1e081	224	}
3f8e211f	225
a6ba4fe1 DG	226	/*
	227	* Each relayd socket pair has a refcount of stream attached to it
	228	* which tells if the relayd is still active or not depending on the
	229	* refcount value.
	230	*
	231	* This will set the destroy flag of the relayd object and destroy it
	232	* if the refcount reaches zero when called.
	233	*
	234	* The destroy can happen either here or when a stream fd hangs up.
	235	*/
	236	consumer_flag_relayd_for_destroy(relayd);
3f8e211f	237
3f8e211f	238	goto end_nosignal;
3bd1e081 MD	239	}
3bd1e081 MD	240	default:
3f8e211f	241	goto end_nosignal;
3bd1e081	242	}
3f8e211f	243
04fdd819	244	/*
3f8e211f DG	245	* Wake-up the other end by writing a null byte in the pipe (non-blocking).
	246	* Important note: Because writing into the pipe is non-blocking (and
	247	* therefore we allow dropping wakeup data, as long as there is wakeup data
	248	* present in the pipe buffer to wake up the other end), the other end
	249	* should perform the following sequence for waiting:
	250	*
04fdd819 MD	251	* 1) empty the pipe (reads).
	252	* 2) perform update operation.
	253	* 3) wait on the pipe (poll).
	254	*/
	255	do {
	256	ret = write(ctx->consumer_poll_pipe[1], "", 1);
6f94560a	257	} while (ret < 0 && errno == EINTR);
3bd1e081	258	end_nosignal:
b0b335c8	259	rcu_read_unlock();
3bd1e081 MD	260	return 0;
3bd1e081 MD	261	}
d41f73b7 MD	262
	263	/*
	264	* Consume data on a file descriptor and write it on a trace file.
	265	*/
4078b776	266	ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7 MD	267	struct lttng_consumer_local_data *ctx)
	268	{
	269	unsigned long len;
	270	int err;
4078b776	271	ssize_t ret = 0;
d41f73b7 MD	272	int infd = stream->wait_fd;
	273
	274	DBG("In read_subbuffer (infd : %d)", infd);
	275	/* Get the next subbuffer */
	276	err = kernctl_get_next_subbuf(infd);
	277	if (err != 0) {
d41f73b7 MD	278	/*
	279	* This is a debug message even for single-threaded consumer,
	280	* because poll() have more relaxed criterions than get subbuf,
	281	* so get_subbuf may fail for short race windows where poll()
	282	* would issue wakeups.
	283	*/
	284	DBG("Reserving sub buffer failed (everything is normal, "
	285	"it is due to concurrency)");
	286	goto end;
	287	}
	288
	289	switch (stream->output) {
	290	case LTTNG_EVENT_SPLICE:
	291	/* read the whole subbuffer */
	292	err = kernctl_get_padded_subbuf_size(infd, &len);
	293	if (err != 0) {
87dc6a9c	294	errno = -ret;
d41f73b7 MD	295	perror("Getting sub-buffer len failed.");
	296	goto end;
	297	}
	298
	299	/* splice the subbuffer to the tracefile */
	300	ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
47e81c02	301	if (ret != len) {
d41f73b7 MD	302	/*
	303	* display the error but continue processing to try
	304	* to release the subbuffer
	305	*/
77c7c900	306	ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
00e2e675	307	ret, len);
d41f73b7	308	}
47e81c02	309
d41f73b7 MD	310	break;
	311	case LTTNG_EVENT_MMAP:
	312	/* read the used subbuffer size */
	313	err = kernctl_get_padded_subbuf_size(infd, &len);
	314	if (err != 0) {
87dc6a9c	315	errno = -ret;
d41f73b7 MD	316	perror("Getting sub-buffer len failed.");
	317	goto end;
	318	}
	319	/* write the subbuffer to the tracefile */
	320	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
47e81c02	321	if (ret != len) {
d41f73b7 MD	322	/*
	323	* display the error but continue processing to try
	324	* to release the subbuffer
	325	*/
	326	ERR("Error writing to tracefile");
	327	}
	328	break;
	329	default:
	330	ERR("Unknown output method");
	331	ret = -1;
	332	}
	333
	334	err = kernctl_put_next_subbuf(infd);
	335	if (err != 0) {
87dc6a9c	336	errno = -ret;
d41f73b7 MD	337	if (errno == EFAULT) {
	338	perror("Error in unreserving sub buffer\n");
	339	} else if (errno == EIO) {
	340	/* Should never happen with newer LTTng versions */
	341	perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
	342	}
	343	goto end;
	344	}
	345
	346	end:
	347	return ret;
	348	}
	349
	350	int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
	351	{
	352	int ret;
	353
	354	/* Opening the tracefile in write mode */
00e2e675	355	if (strlen(stream->path_name) > 0 && stream->net_seq_idx == -1) {
e11d277b	356	ret = run_as_open(stream->path_name,
60b6c79c MD	357	O_WRONLY\|O_CREAT\|O_TRUNC,
	358	S_IRWXU\|S_IRWXG\|S_IRWXO,
	359	stream->uid, stream->gid);
d41f73b7 MD	360	if (ret < 0) {
	361	ERR("Opening %s", stream->path_name);
	362	perror("open");
	363	goto error;
	364	}
	365	stream->out_fd = ret;
	366	}
	367
	368	if (stream->output == LTTNG_EVENT_MMAP) {
	369	/* get the len of the mmap region */
	370	unsigned long mmap_len;
	371
	372	ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
	373	if (ret != 0) {
87dc6a9c	374	errno = -ret;
d41f73b7 MD	375	perror("kernctl_get_mmap_len");
	376	goto error_close_fd;
	377	}
	378	stream->mmap_len = (size_t) mmap_len;
	379
	380	stream->mmap_base = mmap(NULL, stream->mmap_len,
	381	PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
	382	if (stream->mmap_base == MAP_FAILED) {
	383	perror("Error mmaping");
	384	ret = -1;
	385	goto error_close_fd;
	386	}
	387	}
	388
	389	/* we return 0 to let the library handle the FD internally */
	390	return 0;
	391
	392	error_close_fd:
	393	{
	394	int err;
	395
	396	err = close(stream->out_fd);
	397	assert(!err);
	398	}
	399	error:
	400	return ret;
	401	}
	402