[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
 * 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 <fcntl.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
 */
int 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;
	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) {
		ret = -errno;
		perror("ustctl_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) {
			ret = -errno;
			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_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) {
		ret = errno;
		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) {
		ret = errno;
		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];
		size_t nb_fd = 1;

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

		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];
		size_t nb_fd = 2;

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

		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:
	/* signal the poll thread */
	ret = write(ctx->consumer_poll_pipe[1], "4", 1);
	if (ret < 0) {
		perror("write consumer poll");
	}
end_nosignal:
	return 0;
}

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 < 0) {
		/*
		 * 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	*
	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>
ca4537d3	22	#include <fcntl.h>
3bd1e081 MD	23	#include <poll.h>
	24	#include <pthread.h>
	25	#include <stdlib.h>
	26	#include <string.h>
	27	#include <sys/mman.h>
	28	#include <sys/socket.h>
dbb5dfe6	29	#include <sys/stat.h>
3bd1e081 MD	30	#include <sys/types.h>
3bd1e081 MD	31	#include <unistd.h>
9df8df5e	32	#include <lttng/ust-ctl.h>
0857097f	33
990570ed	34	#include <common/common.h>
10a8a223	35	#include <common/sessiond-comm/sessiond-comm.h>
dbb5dfe6	36	#include <common/compat/fcntl.h>
10a8a223 DG	37
10a8a223 DG	38	#include "ust-consumer.h"
3bd1e081 MD	39
	40	extern struct lttng_consumer_global_data consumer_data;
	41	extern int consumer_poll_timeout;
	42	extern volatile int consumer_quit;
	43
	44	/*
	45	* Mmap the ring buffer, read it and write the data to the tracefile.
	46	*
	47	* Returns the number of bytes written
	48	*/
	49	int lttng_ustconsumer_on_read_subbuffer_mmap(
	50	struct lttng_consumer_local_data *ctx,
	51	struct lttng_consumer_stream *stream, unsigned long len)
	52	{
	53	unsigned long mmap_offset;
	54	long ret = 0;
	55	off_t orig_offset = stream->out_fd_offset;
	56	int outfd = stream->out_fd;
	57
	58	/* get the offset inside the fd to mmap */
	59	ret = ustctl_get_mmap_read_offset(stream->chan->handle,
	60	stream->buf, &mmap_offset);
	61	if (ret != 0) {
	62	ret = -errno;
	63	perror("ustctl_get_mmap_read_offset");
	64	goto end;
	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) {
	71	ret = -errno;
	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_ustconsumer_on_read_subbuffer_splice(
	95	struct lttng_consumer_local_data *ctx,
	96	struct lttng_consumer_stream *stream, unsigned long len)
	97	{
	98	return -ENOSYS;
	99	}
	100
	101	/*
	102	* Take a snapshot for a specific fd
	103	*
	104	* Returns 0 on success, < 0 on error
	105	*/
	106	int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
	107	struct lttng_consumer_stream *stream)
	108	{
	109	int ret = 0;
	110
	111	ret = ustctl_snapshot(stream->chan->handle, stream->buf);
	112	if (ret != 0) {
	113	ret = errno;
	114	perror("Getting sub-buffer snapshot.");
	115	}
	116
	117	return ret;
	118	}
	119
	120	/*
	121	* Get the produced position
	122	*
	123	* Returns 0 on success, < 0 on error
	124	*/
	125	int lttng_ustconsumer_get_produced_snapshot(
	126	struct lttng_consumer_local_data *ctx,
	127	struct lttng_consumer_stream *stream,
	128	unsigned long *pos)
	129	{
	130	int ret;
	131
	132	ret = ustctl_snapshot_get_produced(stream->chan->handle,
	133	stream->buf, pos);
	134	if (ret != 0) {
	135	ret = errno;
	136	perror("kernctl_snapshot_get_produced");
	137	}
	138
	139	return ret;
	140	}
141
142	int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
143	int sock, struct pollfd *consumer_sockpoll)
144	{
145	ssize_t ret;
146	struct lttcomm_consumer_msg msg;
147
148	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
149	if (ret != sizeof(msg)) {
150	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
151	return ret;
152	}
153	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
154	return -ENOENT;
155	}
156
157	switch (msg.cmd_type) {
158	case LTTNG_CONSUMER_ADD_CHANNEL:
159	{
160	struct lttng_consumer_channel *new_channel;
161	int fds[1];
162	size_t nb_fd = 1;
163
164	/* block */
165	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
166	return -EINTR;
167	}
168	ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
169	if (ret != sizeof(fds)) {
170	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
171	return ret;
172	}
173
174	DBG("consumer_add_channel %d", msg.u.channel.channel_key);
175
176	new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
177	fds[0], -1,
178	msg.u.channel.mmap_len,
179	msg.u.channel.max_sb_size);
180	if (new_channel == NULL) {
181	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
182	goto end_nosignal;
183	}
184	if (ctx->on_recv_channel != NULL) {
185	ret = ctx->on_recv_channel(new_channel);
186	if (ret == 0) {
187	consumer_add_channel(new_channel);
188	} else if (ret < 0) {
189	goto end_nosignal;
190	}
191	} else {
192	consumer_add_channel(new_channel);
193	}
194	goto end_nosignal;
195	}
196	case LTTNG_CONSUMER_ADD_STREAM:
197	{
198	struct lttng_consumer_stream *new_stream;
199	int fds[2];
200	size_t nb_fd = 2;
201
202	/* block */
203	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
204	return -EINTR;
205	}
206	ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
207	if (ret != sizeof(fds)) {
208	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
209	return ret;
210	}
211
212	DBG("consumer_add_stream %s (%d,%d)", msg.u.stream.path_name,
213	fds[0], fds[1]);
d41f73b7	214	assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
7ad0a0cb	215	new_stream = consumer_allocate_stream(msg.u.channel.channel_key,
3bd1e081 MD	216	msg.u.stream.stream_key,
	217	fds[0], fds[1],
	218	msg.u.stream.state,
	219	msg.u.stream.mmap_len,
	220	msg.u.stream.output,
6df2e2c9 MD	221	msg.u.stream.path_name,
	222	msg.u.stream.uid,
	223	msg.u.stream.gid);
3bd1e081 MD	224	if (new_stream == NULL) {
	225	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	226	goto end;
	227	}
	228	if (ctx->on_recv_stream != NULL) {
	229	ret = ctx->on_recv_stream(new_stream);
	230	if (ret == 0) {
	231	consumer_add_stream(new_stream);
	232	} else if (ret < 0) {
	233	goto end;
	234	}
	235	} else {
	236	consumer_add_stream(new_stream);
	237	}
	238	break;
	239	}
	240	case LTTNG_CONSUMER_UPDATE_STREAM:
	241	{
7ad0a0cb MD	242	return -ENOSYS;
7ad0a0cb MD	243	#if 0
3bd1e081 MD	244	if (ctx->on_update_stream != NULL) {
	245	ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
	246	if (ret == 0) {
	247	consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
	248	} else if (ret < 0) {
	249	goto end;
	250	}
	251	} else {
	252	consumer_change_stream_state(msg.u.stream.stream_key,
	253	msg.u.stream.state);
	254	}
7ad0a0cb	255	#endif
3bd1e081 MD	256	break;
	257	}
	258	default:
	259	break;
	260	}
	261	end:
	262	/* signal the poll thread */
	263	ret = write(ctx->consumer_poll_pipe[1], "4", 1);
	264	if (ret < 0) {
	265	perror("write consumer poll");
	266	}
	267	end_nosignal:
	268	return 0;
	269	}
	270
	271	int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
	272	{
13161846	273	struct lttng_ust_object_data obj;
3bd1e081 MD	274
	275	obj.handle = -1;
	276	obj.shm_fd = chan->shm_fd;
	277	obj.wait_fd = chan->wait_fd;
	278	obj.memory_map_size = chan->mmap_len;
	279	chan->handle = ustctl_map_channel(&obj);
	280	if (!chan->handle) {
	281	return -ENOMEM;
	282	}
b5c5fc29	283	chan->wait_fd_is_copy = 1;
2c1dd183	284	chan->shm_fd = -1;
b5c5fc29	285
3bd1e081 MD	286	return 0;
	287	}
	288
d056b477 MD	289	void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
	290	{
	291	ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
effcf122	292	stream->hangup_flush_done = 1;
d056b477 MD	293	}
d056b477 MD	294
3bd1e081 MD	295	void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
	296	{
	297	ustctl_unmap_channel(chan->handle);
	298	}
	299
	300	int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
	301	{
13161846	302	struct lttng_ust_object_data obj;
3bd1e081 MD	303	int ret;
	304
	305	obj.handle = -1;
	306	obj.shm_fd = stream->shm_fd;
	307	obj.wait_fd = stream->wait_fd;
	308	obj.memory_map_size = stream->mmap_len;
	309	ret = ustctl_add_stream(stream->chan->handle, &obj);
	310	if (ret)
	311	return ret;
	312	stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
	313	if (!stream->buf)
	314	return -EBUSY;
2c1dd183 MD	315	/* ustctl_open_stream_read has closed the shm fd. */
	316	stream->wait_fd_is_copy = 1;
	317	stream->shm_fd = -1;
	318
3bd1e081 MD	319	stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
	320	if (!stream->mmap_base) {
	321	return -EINVAL;
	322	}
ee77a7b0	323
3bd1e081 MD	324	return 0;
	325	}
	326
	327	void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
	328	{
	329	ustctl_close_stream_read(stream->chan->handle, stream->buf);
	330	}
d41f73b7 MD	331
	332
	333	int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
	334	struct lttng_consumer_local_data *ctx)
	335	{
	336	unsigned long len;
	337	int err;
	338	long ret = 0;
	339	struct lttng_ust_shm_handle *handle;
	340	struct lttng_ust_lib_ring_buffer *buf;
	341	char dummy;
	342	ssize_t readlen;
	343
	344	DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
	345	stream->wait_fd, stream->key);
	346
	347	/* We can consume the 1 byte written into the wait_fd by UST */
effcf122 MD	348	if (!stream->hangup_flush_done) {
	349	do {
	350	readlen = read(stream->wait_fd, &dummy, 1);
	351	} while (readlen == -1 && errno == -EINTR);
	352	if (readlen == -1) {
	353	ret = readlen;
	354	goto end;
	355	}
d41f73b7 MD	356	}
	357
	358	buf = stream->buf;
	359	handle = stream->chan->handle;
	360	/* Get the next subbuffer */
	361	err = ustctl_get_next_subbuf(handle, buf);
	362	if (err != 0) {
effcf122	363	ret = -ret; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
d41f73b7 MD	364	/*
	365	* This is a debug message even for single-threaded consumer,
	366	* because poll() have more relaxed criterions than get subbuf,
	367	* so get_subbuf may fail for short race windows where poll()
	368	* would issue wakeups.
	369	*/
	370	DBG("Reserving sub buffer failed (everything is normal, "
	371	"it is due to concurrency)");
	372	goto end;
	373	}
	374	assert(stream->output == LTTNG_EVENT_MMAP);
	375	/* read the used subbuffer size */
	376	err = ustctl_get_padded_subbuf_size(handle, buf, &len);
effcf122	377	assert(err == 0);
d41f73b7 MD	378	/* write the subbuffer to the tracefile */
	379	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
	380	if (ret < 0) {
	381	/*
	382	* display the error but continue processing to try
	383	* to release the subbuffer
	384	*/
	385	ERR("Error writing to tracefile");
	386	}
	387	err = ustctl_put_next_subbuf(handle, buf);
effcf122	388	assert(err == 0);
d41f73b7 MD	389	end:
	390	return ret;
	391	}
	392
	393	int lttng_ustconsumer_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	/* we return 0 to let the library handle the FD internally */
	412	return 0;
	413
	414	error:
	415	return ret;
	416	}