[lttng-tools.git] / liblttng-ustconsumer / lttng-ustconsumer.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-sessiond-comm.h>
#include <lttng/lttng-ustconsumer.h>
#include <lttng/ust-ctl.h>
#include <lttngerr.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.stream.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);
		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:
	{
		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);
		}
		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;
	}
	/*
	 * The channel fds are passed to ustctl, we only keep a copy.
	 */
	chan->shm_fd_is_copy = 1;
	chan->wait_fd_is_copy = 1;

	return 0;
}

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

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;
	stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
	if (!stream->mmap_base) {
		return -EINVAL;
	}
	/*
	 * The stream fds are passed to ustctl, we only keep a copy.
	 */
	stream->shm_fd_is_copy = 1;
	stream->wait_fd_is_copy = 1;

	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 */
	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 = errno;
		/*
		 * 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);
	if (err != 0) {
		ret = errno;
		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 < 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);
	if (err != 0) {
		ret = errno;
		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_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
{
	int ret;

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