[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/types.h>
#include <unistd.h>
#include <lttng/ust-ctl.h>

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