[lttng-tools.git] / src / common / ust-consumer / ust-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/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <lttng/ust-ctl.h>

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

#include "ust-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, else negative value on error.
 */
ssize_t lttng_ustconsumer_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, written = 0;
	off_t orig_offset = stream->out_fd_offset;
	int outfd = stream->out_fd;

	/* get the offset inside the fd to mmap */
	ret = ustctl_get_mmap_read_offset(stream->chan->handle,
		stream->buf, &mmap_offset);
	if (ret != 0) {
		errno = -ret;
		PERROR("ustctl_get_mmap_read_offset");
		written = ret;
		goto end;
	}
	while (len > 0) {
		ret = write(outfd, stream->mmap_base + mmap_offset, len);
		if (ret < 0) {
			if (errno == EINTR) {
				/* restart the interrupted system call */
				continue;
			} else {
				PERROR("Error in file write");
				if (written == 0) {
					written = ret;
				}
				goto end;
			}
		} else if (ret > len) {
			PERROR("Error in file write");
			written += ret;
			goto end;
		} else {
			len -= ret;
			mmap_offset += 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;
		written += ret;
	}
	lttng_consumer_sync_trace_file(stream, orig_offset);
end:
	return written;
}

/*
 * Splice the data from the ring buffer to the tracefile.
 *
 * Returns the number of bytes spliced.
 */
ssize_t lttng_ustconsumer_on_read_subbuffer_splice(
		struct lttng_consumer_local_data *ctx,
		struct lttng_consumer_stream *stream, unsigned long len)
{
	return -ENOSYS;
}

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

	ret = ustctl_snapshot(stream->chan->handle, stream->buf);
	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_ustconsumer_get_produced_snapshot(
		struct lttng_consumer_local_data *ctx,
		struct lttng_consumer_stream *stream,
		unsigned long *pos)
{
	int ret;

	ret = ustctl_snapshot_get_produced(stream->chan->handle,
			stream->buf, pos);
	if (ret != 0) {
		errno = -ret;
		PERROR("kernctl_snapshot_get_produced");
	}

	return ret;
}

int lttng_ustconsumer_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_FD);
		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;
		int fds[1];
		char *path;

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

		path = lttcomm_recv_string(sock);

		if (!path) {
			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
			return -1;
		}

		DBG("consumer_add_channel received path %s", path);

		fds[0] = open(path, O_RDWR);

		if (fds[0] < 0) {
			DBG("consumer_add_channel open error on path %s", path);
			free(path);
			return -1;
		}

		if (fcntl(fds[0], F_SETFD, FD_CLOEXEC) < 0) {
			DBG("consumer_add_channel fcntl error");
			free(path);
			return -1;
		}

		free(path);

		DBG("consumer_add_channel %d", msg.u.channel.channel_key);

		new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
				fds[0], -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 fds[2], i;
		char *shm_path, *wait_pipe_path;

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

		shm_path = lttcomm_recv_string(sock);

		if (!shm_path) {
			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
			return -1;
		}

		wait_pipe_path = lttcomm_recv_string(sock);

		if (!wait_pipe_path) {
			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
			free(shm_path);
			return -1;
		}

		DBG("consumer_add_stream received path %s", shm_path);
		DBG("consumer_add_stream received path %s", wait_pipe_path);

		fds[0] = open(shm_path, O_RDWR);

		if (fds[0] < 0) {
			DBG("consumer_add_stream open error on path %s", shm_path);
			free(shm_path);
			free(wait_pipe_path);
			return -1;
		}

		fds[1] = open(wait_pipe_path, O_RDONLY);

		if (fds[1] < 0) {
			DBG("consumer_add_stream open error on path %s", wait_pipe_path);
			PERROR("open");
			free(shm_path);
			free(wait_pipe_path);
			return -1;
		}

		free(shm_path);
		free(wait_pipe_path);

		for (i = 0; i < 2; ++i) {
			if (fcntl(fds[i], F_SETFD, FD_CLOEXEC) < 0) {
				DBG("consumer_add_stream fcntl error");
				return -1;
			}
		}

		DBG("consumer_add_stream %s (%d,%d)", msg.u.stream.path_name,
			fds[0], fds[1]);
		assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
		new_stream = consumer_allocate_stream(msg.u.channel.channel_key,
				msg.u.stream.stream_key,
				fds[0], fds[1],
				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:
	{
		return -ENOSYS;
#if 0
		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);
		}
#endif
		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 1 to indicate success since the 0 value can be a socket shutdown
	 * during the recv() or send() call.
	 */
	return 1;
}

int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
{
	struct lttng_ust_object_data obj;

	obj.handle = -1;
	obj.shm_fd = chan->shm_fd;
	obj.wait_fd = chan->wait_fd;
	obj.memory_map_size = chan->mmap_len;
	chan->handle = ustctl_map_channel(&obj);
	if (!chan->handle) {
		return -ENOMEM;
	}
	chan->wait_fd_is_copy = 1;
	chan->shm_fd = -1;

	return 0;
}

void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
{
	ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
	stream->hangup_flush_done = 1;
}

void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
{
	ustctl_unmap_channel(chan->handle);
}

int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
{
	struct lttng_ust_object_data obj;
	int ret;

	obj.handle = -1;
	obj.shm_fd = stream->shm_fd;
	obj.wait_fd = stream->wait_fd;
	obj.memory_map_size = stream->mmap_len;
	ret = ustctl_add_stream(stream->chan->handle, &obj);
	if (ret)
		return ret;
	stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
	if (!stream->buf)
		return -EBUSY;
	/* ustctl_open_stream_read has closed the shm fd. */
	stream->wait_fd_is_copy = 1;
	stream->shm_fd = -1;

	stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
	if (!stream->mmap_base) {
		return -EINVAL;
	}

	return 0;
}

void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
{
	ustctl_close_stream_read(stream->chan->handle, stream->buf);
}


int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
		struct lttng_consumer_local_data *ctx)
{
	unsigned long len;
	int err;
	long ret = 0;
	struct lttng_ust_shm_handle *handle;
	struct lttng_ust_lib_ring_buffer *buf;
	char dummy;
	ssize_t readlen;

	DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
		stream->wait_fd, stream->key);

	/* We can consume the 1 byte written into the wait_fd by UST */
	if (!stream->hangup_flush_done) {
		do {
			readlen = read(stream->wait_fd, &dummy, 1);
		} while (readlen == -1 && errno == EINTR);
		if (readlen == -1) {
			ret = readlen;
			goto end;
		}
	}

	buf = stream->buf;
	handle = stream->chan->handle;
	/* Get the next subbuffer */
	err = ustctl_get_next_subbuf(handle, buf);
	if (err != 0) {
		ret = -ret;	/* ustctl_get_next_subbuf returns negative, caller expect positive. */
		/*
		 * 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;
	}
	assert(stream->output == LTTNG_EVENT_MMAP);
	/* read the used subbuffer size */
	err = ustctl_get_padded_subbuf_size(handle, buf, &len);
	assert(err == 0);
	/* 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");
	}
	err = ustctl_put_next_subbuf(handle, buf);
	assert(err == 0);
end:
	return ret;
}

int lttng_ustconsumer_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;
	}

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

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>
dbb5dfe6	27	#include <sys/stat.h>
3bd1e081 MD	28	#include <sys/types.h>
3bd1e081 MD	29	#include <unistd.h>
9df8df5e	30	#include <lttng/ust-ctl.h>
0857097f	31
990570ed	32	#include <common/common.h>
10a8a223	33	#include <common/sessiond-comm/sessiond-comm.h>
dbb5dfe6	34	#include <common/compat/fcntl.h>
10a8a223 DG	35
10a8a223 DG	36	#include "ust-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	*
4078b776	45	* Returns the number of bytes written, else negative value on error.
3bd1e081	46	*/
4078b776	47	ssize_t lttng_ustconsumer_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;
325933ef	52	long ret = 0, written = 0;
3bd1e081 MD	53	off_t orig_offset = stream->out_fd_offset;
	54	int outfd = stream->out_fd;
	55
	56	/* get the offset inside the fd to mmap */
	57	ret = ustctl_get_mmap_read_offset(stream->chan->handle,
	58	stream->buf, &mmap_offset);
	59	if (ret != 0) {
87dc6a9c	60	errno = -ret;
4c462e79	61	PERROR("ustctl_get_mmap_read_offset");
325933ef	62	written = ret;
3bd1e081 MD	63	goto end;
	64	}
	65	while (len > 0) {
	66	ret = write(outfd, stream->mmap_base + mmap_offset, len);
325933ef MD	67	if (ret < 0) {
	68	if (errno == EINTR) {
	69	/* restart the interrupted system call */
	70	continue;
	71	} else {
	72	PERROR("Error in file write");
	73	if (written == 0) {
	74	written = ret;
	75	}
	76	goto end;
	77	}
	78	} else if (ret > len) {
4c462e79	79	PERROR("Error in file write");
325933ef	80	written += ret;
3bd1e081	81	goto end;
325933ef MD	82	} else {
	83	len -= ret;
	84	mmap_offset += ret;
3bd1e081 MD	85	}
3bd1e081 MD	86	/* This won't block, but will start writeout asynchronously */
dbb5dfe6	87	lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
3bd1e081 MD	88	SYNC_FILE_RANGE_WRITE);
3bd1e081 MD	89	stream->out_fd_offset += ret;
325933ef	90	written += ret;
3bd1e081	91	}
3bd1e081	92	lttng_consumer_sync_trace_file(stream, orig_offset);
3bd1e081	93	end:
325933ef	94	return written;
3bd1e081 MD	95	}
	96
	97	/*
	98	* Splice the data from the ring buffer to the tracefile.
	99	*
	100	* Returns the number of bytes spliced.
	101	*/
4078b776	102	ssize_t lttng_ustconsumer_on_read_subbuffer_splice(
3bd1e081 MD	103	struct lttng_consumer_local_data *ctx,
	104	struct lttng_consumer_stream *stream, unsigned long len)
	105	{
	106	return -ENOSYS;
	107	}
	108
	109	/*
	110	* Take a snapshot for a specific fd
	111	*
	112	* Returns 0 on success, < 0 on error
	113	*/
	114	int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
	115	struct lttng_consumer_stream *stream)
	116	{
	117	int ret = 0;
	118
	119	ret = ustctl_snapshot(stream->chan->handle, stream->buf);
	120	if (ret != 0) {
87dc6a9c	121	errno = -ret;
4c462e79	122	PERROR("Getting sub-buffer snapshot.");
3bd1e081 MD	123	}
	124
	125	return ret;
	126	}
	127
	128	/*
	129	* Get the produced position
	130	*
	131	* Returns 0 on success, < 0 on error
	132	*/
	133	int lttng_ustconsumer_get_produced_snapshot(
	134	struct lttng_consumer_local_data *ctx,
	135	struct lttng_consumer_stream *stream,
	136	unsigned long *pos)
	137	{
	138	int ret;
	139
	140	ret = ustctl_snapshot_get_produced(stream->chan->handle,
	141	stream->buf, pos);
	142	if (ret != 0) {
87dc6a9c	143	errno = -ret;
4c462e79	144	PERROR("kernctl_snapshot_get_produced");
3bd1e081 MD	145	}
	146
	147	return ret;
	148	}
	149
	150	int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
	151	int sock, struct pollfd *consumer_sockpoll)
	152	{
	153	ssize_t ret;
	154	struct lttcomm_consumer_msg msg;
	155
	156	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
	157	if (ret != sizeof(msg)) {
	158	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
	159	return ret;
	160	}
	161	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
	162	return -ENOENT;
	163	}
	164
	165	switch (msg.cmd_type) {
	166	case LTTNG_CONSUMER_ADD_CHANNEL:
	167	{
	168	struct lttng_consumer_channel *new_channel;
	169	int fds[1];
6684bfa4	170	char *path;
3bd1e081 MD	171
	172	/* block */
	173	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
	174	return -EINTR;
	175	}
6684bfa4 CB	176
	177	path = lttcomm_recv_string(sock);
	178
	179	if (!path) {
3bd1e081	180	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
6684bfa4	181	return -1;
3bd1e081 MD	182	}
3bd1e081 MD	183
6684bfa4 CB	184	DBG("consumer_add_channel received path %s", path);
	185
	186	fds[0] = open(path, O_RDWR);
	187
	188	if (fds[0] < 0) {
	189	DBG("consumer_add_channel open error on path %s", path);
	190	free(path);
	191	return -1;
	192	}
	193
	194	if (fcntl(fds[0], F_SETFD, FD_CLOEXEC) < 0) {
	195	DBG("consumer_add_channel fcntl error");
	196	free(path);
	197	return -1;
	198	}
	199
	200	free(path);
	201
3bd1e081 MD	202	DBG("consumer_add_channel %d", msg.u.channel.channel_key);
	203
	204	new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
	205	fds[0], -1,
	206	msg.u.channel.mmap_len,
	207	msg.u.channel.max_sb_size);
	208	if (new_channel == NULL) {
	209	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	210	goto end_nosignal;
	211	}
	212	if (ctx->on_recv_channel != NULL) {
	213	ret = ctx->on_recv_channel(new_channel);
	214	if (ret == 0) {
	215	consumer_add_channel(new_channel);
	216	} else if (ret < 0) {
	217	goto end_nosignal;
	218	}
	219	} else {
	220	consumer_add_channel(new_channel);
	221	}
	222	goto end_nosignal;
	223	}
	224	case LTTNG_CONSUMER_ADD_STREAM:
	225	{
	226	struct lttng_consumer_stream *new_stream;
6684bfa4 CB	227	int fds[2], i;
6684bfa4 CB	228	char shm_path, wait_pipe_path;
3bd1e081 MD	229
	230	/* block */
	231	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
	232	return -EINTR;
	233	}
6684bfa4 CB	234
	235	shm_path = lttcomm_recv_string(sock);
	236
	237	if (!shm_path) {
	238	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
	239	return -1;
	240	}
	241
	242	wait_pipe_path = lttcomm_recv_string(sock);
	243
	244	if (!wait_pipe_path) {
3bd1e081	245	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
6684bfa4 CB	246	free(shm_path);
	247	return -1;
	248	}
	249
	250	DBG("consumer_add_stream received path %s", shm_path);
	251	DBG("consumer_add_stream received path %s", wait_pipe_path);
	252
	253	fds[0] = open(shm_path, O_RDWR);
	254
	255	if (fds[0] < 0) {
	256	DBG("consumer_add_stream open error on path %s", shm_path);
	257	free(shm_path);
	258	free(wait_pipe_path);
	259	return -1;
	260	}
	261
	262	fds[1] = open(wait_pipe_path, O_RDONLY);
	263
	264	if (fds[1] < 0) {
	265	DBG("consumer_add_stream open error on path %s", wait_pipe_path);
	266	PERROR("open");
	267	free(shm_path);
	268	free(wait_pipe_path);
	269	return -1;
	270	}
	271
	272	free(shm_path);
	273	free(wait_pipe_path);
	274
	275	for (i = 0; i < 2; ++i) {
	276	if (fcntl(fds[i], F_SETFD, FD_CLOEXEC) < 0) {
	277	DBG("consumer_add_stream fcntl error");
	278	return -1;
	279	}
3bd1e081 MD	280	}
	281
	282	DBG("consumer_add_stream %s (%d,%d)", msg.u.stream.path_name,
	283	fds[0], fds[1]);
d41f73b7	284	assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
7ad0a0cb	285	new_stream = consumer_allocate_stream(msg.u.channel.channel_key,
3bd1e081 MD	286	msg.u.stream.stream_key,
	287	fds[0], fds[1],
	288	msg.u.stream.state,
	289	msg.u.stream.mmap_len,
	290	msg.u.stream.output,
6df2e2c9 MD	291	msg.u.stream.path_name,
	292	msg.u.stream.uid,
	293	msg.u.stream.gid);
3bd1e081 MD	294	if (new_stream == NULL) {
	295	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	296	goto end;
	297	}
	298	if (ctx->on_recv_stream != NULL) {
	299	ret = ctx->on_recv_stream(new_stream);
	300	if (ret == 0) {
	301	consumer_add_stream(new_stream);
	302	} else if (ret < 0) {
	303	goto end;
	304	}
	305	} else {
	306	consumer_add_stream(new_stream);
	307	}
	308	break;
	309	}
	310	case LTTNG_CONSUMER_UPDATE_STREAM:
	311	{
7ad0a0cb MD	312	return -ENOSYS;
7ad0a0cb MD	313	#if 0
3bd1e081 MD	314	if (ctx->on_update_stream != NULL) {
	315	ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
	316	if (ret == 0) {
	317	consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
	318	} else if (ret < 0) {
	319	goto end;
	320	}
	321	} else {
	322	consumer_change_stream_state(msg.u.stream.stream_key,
	323	msg.u.stream.state);
	324	}
7ad0a0cb	325	#endif
3bd1e081 MD	326	break;
	327	}
	328	default:
	329	break;
	330	}
	331	end:
5a4b955e MD	332	/*
	333	* Wake-up the other end by writing a null byte in the pipe
	334	* (non-blocking). Important note: Because writing into the
	335	* pipe is non-blocking (and therefore we allow dropping wakeup
	336	* data, as long as there is wakeup data present in the pipe
	337	* buffer to wake up the other end), the other end should
	338	* perform the following sequence for waiting:
	339	* 1) empty the pipe (reads).
	340	* 2) perform update operation.
	341	* 3) wait on the pipe (poll).
	342	*/
	343	do {
	344	ret = write(ctx->consumer_poll_pipe[1], "", 1);
	345	} while (ret == -1UL && errno == EINTR);
3bd1e081	346	end_nosignal:
3fa5d621 DG	347
	348	/*
	349	* Return 1 to indicate success since the 0 value can be a socket shutdown
	350	* during the recv() or send() call.
	351	*/
	352	return 1;
3bd1e081 MD	353	}
	354
	355	int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
	356	{
13161846	357	struct lttng_ust_object_data obj;
3bd1e081 MD	358
	359	obj.handle = -1;
	360	obj.shm_fd = chan->shm_fd;
	361	obj.wait_fd = chan->wait_fd;
	362	obj.memory_map_size = chan->mmap_len;
	363	chan->handle = ustctl_map_channel(&obj);
	364	if (!chan->handle) {
	365	return -ENOMEM;
	366	}
b5c5fc29	367	chan->wait_fd_is_copy = 1;
2c1dd183	368	chan->shm_fd = -1;
b5c5fc29	369
3bd1e081 MD	370	return 0;
	371	}
	372
d056b477 MD	373	void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
	374	{
	375	ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
effcf122	376	stream->hangup_flush_done = 1;
d056b477 MD	377	}
d056b477 MD	378
3bd1e081 MD	379	void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
	380	{
	381	ustctl_unmap_channel(chan->handle);
	382	}
	383
	384	int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
	385	{
13161846	386	struct lttng_ust_object_data obj;
3bd1e081 MD	387	int ret;
	388
	389	obj.handle = -1;
	390	obj.shm_fd = stream->shm_fd;
	391	obj.wait_fd = stream->wait_fd;
	392	obj.memory_map_size = stream->mmap_len;
	393	ret = ustctl_add_stream(stream->chan->handle, &obj);
	394	if (ret)
	395	return ret;
	396	stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
	397	if (!stream->buf)
	398	return -EBUSY;
2c1dd183 MD	399	/* ustctl_open_stream_read has closed the shm fd. */
	400	stream->wait_fd_is_copy = 1;
	401	stream->shm_fd = -1;
	402
3bd1e081 MD	403	stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
	404	if (!stream->mmap_base) {
	405	return -EINVAL;
	406	}
ee77a7b0	407
3bd1e081 MD	408	return 0;
	409	}
	410
	411	void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
	412	{
	413	ustctl_close_stream_read(stream->chan->handle, stream->buf);
	414	}
d41f73b7 MD	415
	416
	417	int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
	418	struct lttng_consumer_local_data *ctx)
	419	{
	420	unsigned long len;
	421	int err;
	422	long ret = 0;
	423	struct lttng_ust_shm_handle *handle;
	424	struct lttng_ust_lib_ring_buffer *buf;
	425	char dummy;
	426	ssize_t readlen;
	427
	428	DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
	429	stream->wait_fd, stream->key);
	430
	431	/* We can consume the 1 byte written into the wait_fd by UST */
effcf122 MD	432	if (!stream->hangup_flush_done) {
	433	do {
	434	readlen = read(stream->wait_fd, &dummy, 1);
87dc6a9c	435	} while (readlen == -1 && errno == EINTR);
effcf122 MD	436	if (readlen == -1) {
	437	ret = readlen;
	438	goto end;
	439	}
d41f73b7 MD	440	}
	441
	442	buf = stream->buf;
	443	handle = stream->chan->handle;
	444	/* Get the next subbuffer */
	445	err = ustctl_get_next_subbuf(handle, buf);
	446	if (err != 0) {
effcf122	447	ret = -ret; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
d41f73b7 MD	448	/*
	449	* This is a debug message even for single-threaded consumer,
	450	* because poll() have more relaxed criterions than get subbuf,
	451	* so get_subbuf may fail for short race windows where poll()
	452	* would issue wakeups.
	453	*/
	454	DBG("Reserving sub buffer failed (everything is normal, "
	455	"it is due to concurrency)");
	456	goto end;
	457	}
	458	assert(stream->output == LTTNG_EVENT_MMAP);
	459	/* read the used subbuffer size */
	460	err = ustctl_get_padded_subbuf_size(handle, buf, &len);
effcf122	461	assert(err == 0);
d41f73b7 MD	462	/* write the subbuffer to the tracefile */
d41f73b7 MD	463	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
325933ef	464	if (ret != len) {
d41f73b7 MD	465	/*
	466	* display the error but continue processing to try
	467	* to release the subbuffer
	468	*/
	469	ERR("Error writing to tracefile");
	470	}
	471	err = ustctl_put_next_subbuf(handle, buf);
effcf122	472	assert(err == 0);
d41f73b7 MD	473	end:
	474	return ret;
	475	}
	476
	477	int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
	478	{
	479	int ret;
	480
	481	/* Opening the tracefile in write mode */
	482	if (stream->path_name != NULL) {
e11d277b	483	ret = run_as_open(stream->path_name,
60b6c79c MD	484	O_WRONLY\|O_CREAT\|O_TRUNC,
	485	S_IRWXU\|S_IRWXG\|S_IRWXO,
	486	stream->uid, stream->gid);
d41f73b7 MD	487	if (ret < 0) {
d41f73b7 MD	488	ERR("Opening %s", stream->path_name);
4c462e79	489	PERROR("open");
d41f73b7 MD	490	goto error;
	491	}
	492	stream->out_fd = ret;
	493	}
	494
	495	/* we return 0 to let the library handle the FD internally */
	496	return 0;
	497
	498	error:
	499	return ret;
	500	}