[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/relayd/relayd.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;
	uint64_t metadata_id;
	struct consumer_relayd_sock_pair *relayd = NULL;

	/* RCU lock for the relayd pointer */
	rcu_read_lock();

	/* Flag that the current stream if set for network streaming. */
	if (stream->net_seq_idx != -1) {
		relayd = consumer_find_relayd(stream->net_seq_idx);
		if (relayd == NULL) {
			ERR("UST consumer mmap(), unable to find relay for index %d",
					stream->net_seq_idx);
			goto end;
		}
	}

	/* 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;
	}

	/* Handle stream on the relayd if the output is on the network */
	if (relayd) {
		unsigned long netlen = len;

		if (stream->metadata_flag) {
			/* Only lock if metadata since we use the control socket. */
			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
			netlen += sizeof(stream->relayd_stream_id);
		}

		ret = consumer_handle_stream_before_relayd(stream, netlen);
		if (ret >= 0) {
			outfd = ret;

			/* Write metadata stream id before payload */
			if (stream->metadata_flag) {
				metadata_id = htobe64(stream->relayd_stream_id);
				do {
					ret = write(outfd, (void *) &metadata_id,
							sizeof(stream->relayd_stream_id));
				} while (ret < 0 && errno == EINTR);
				if (ret < 0) {
					PERROR("write metadata stream id");
					written = ret;
					goto end;
				}
				DBG("Metadata stream id %zu written before data",
						stream->relayd_stream_id);
			}
		}
		/* Else, use the default set before which is the filesystem. */
	}

	while (len > 0) {
		do {
			ret = write(outfd, stream->mmap_base + mmap_offset, len);
		} while (ret < 0 && errno == EINTR);
		if (ret < 0) {
			PERROR("Error in file write");
			if (written == 0) {
				written = ret;
			}
			goto end;
		} else if (ret > len) {
			PERROR("ret %ld > len %lu", ret, len);
			written += ret;
			goto end;
		} else {
			len -= ret;
			mmap_offset += ret;
		}
		DBG("UST mmap write() ret %ld (len %lu)", ret, len);

		/* This call is useless on a socket so better save a syscall. */
		if (!relayd) {
			/* 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:
	if (relayd && stream->metadata_flag) {
		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
	}
	rcu_read_unlock();
	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;
	}

	/* relayd need RCU read-side lock */
	rcu_read_lock();

	switch (msg.cmd_type) {
	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
	{
		int fd;
		struct consumer_relayd_sock_pair *relayd;

		DBG("UST Consumer adding relayd socket");

		/* Get relayd reference if exists. */
		relayd = consumer_find_relayd(msg.u.relayd_sock.net_index);
		if (relayd == NULL) {
			/* Not found. Allocate one. */
			relayd = consumer_allocate_relayd_sock_pair(
					msg.u.relayd_sock.net_index);
			if (relayd == NULL) {
				lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
				goto end_nosignal;
			}
		}

		/* Poll on consumer socket. */
		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
			return -EINTR;
		}

		/* Get relayd socket from session daemon */
		ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
		if (ret != sizeof(fd)) {
			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
			goto end_nosignal;
		}

		/* Copy socket information and received FD */
		switch (msg.u.relayd_sock.type) {
		case LTTNG_STREAM_CONTROL:
			/* Copy received lttcomm socket */
			lttcomm_copy_sock(&relayd->control_sock, &msg.u.relayd_sock.sock);
			ret = lttcomm_create_sock(&relayd->control_sock);
			if (ret < 0) {
				goto end_nosignal;
			}

			/* Close the created socket fd which is useless */
			close(relayd->control_sock.fd);

			/* Assign new file descriptor */
			relayd->control_sock.fd = fd;
			break;
		case LTTNG_STREAM_DATA:
			/* Copy received lttcomm socket */
			lttcomm_copy_sock(&relayd->data_sock, &msg.u.relayd_sock.sock);
			ret = lttcomm_create_sock(&relayd->data_sock);
			if (ret < 0) {
				goto end_nosignal;
			}

			/* Close the created socket fd which is useless */
			close(relayd->data_sock.fd);

			/* Assign new file descriptor */
			relayd->data_sock.fd = fd;
			break;
		default:
			ERR("Unknown relayd socket type");
			goto end_nosignal;
		}

		DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
				msg.u.relayd_sock.type == LTTNG_STREAM_CONTROL ? "control" : "data",
				relayd->net_seq_idx, fd);

		/*
		 * Add relayd socket pair to consumer data hashtable. If object already
		 * exists or on error, the function gracefully returns.
		 */
		consumer_add_relayd(relayd);

		goto end_nosignal;
	}
	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;
		struct consumer_relayd_sock_pair *relayd = NULL;

		/* 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;
		}

		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,
				msg.u.stream.net_index,
				msg.u.stream.metadata_flag);
		if (new_stream == NULL) {
			lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
			goto end;
		}

		/* The stream is not metadata. Get relayd reference if exists. */
		relayd = consumer_find_relayd(msg.u.stream.net_index);
		if (relayd != NULL) {
			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
			/* Add stream on the relayd */
			ret = relayd_add_stream(&relayd->control_sock,
					msg.u.stream.name, msg.u.stream.path_name,
					&new_stream->relayd_stream_id);
			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
			if (ret < 0) {
				goto end;
			}
		} else if (msg.u.stream.net_index != -1) {
			ERR("Network sequence index %d unknown. Not adding stream.",
					msg.u.stream.net_index);
			free(new_stream);
			goto end;
		}

		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);
		}

		DBG("UST consumer_add_stream %s (%d,%d) with relayd id %lu",
				msg.u.stream.path_name, fds[0], fds[1],
				new_stream->relayd_stream_id);
		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 < 0 && errno == EINTR);
end_nosignal:
	rcu_read_unlock();
	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 != 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 && stream->net_seq_idx == -1) {
		ret = run_as_open(stream->path_name,
				O_WRONLY|O_CREAT|O_TRUNC,
				S_IRWXU|S_IRWXG|S_IRWXO,
				stream->uid, stream->gid);
		if (ret < 0) {
			ERR("Opening %s", stream->path_name);
			PERROR("open");
			goto error;
		}
		stream->out_fd = ret;
	}

	/* 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>
00e2e675	34	#include <common/relayd/relayd.h>
dbb5dfe6	35	#include <common/compat/fcntl.h>
10a8a223 DG	36
10a8a223 DG	37	#include "ust-consumer.h"
3bd1e081 MD	38
	39	extern struct lttng_consumer_global_data consumer_data;
	40	extern int consumer_poll_timeout;
	41	extern volatile int consumer_quit;
	42
	43	/*
	44	* Mmap the ring buffer, read it and write the data to the tracefile.
	45	*
4078b776	46	* Returns the number of bytes written, else negative value on error.
3bd1e081	47	*/
4078b776	48	ssize_t lttng_ustconsumer_on_read_subbuffer_mmap(
3bd1e081 MD	49	struct lttng_consumer_local_data *ctx,
	50	struct lttng_consumer_stream *stream, unsigned long len)
	51	{
	52	unsigned long mmap_offset;
47e81c02	53	long ret = 0, written = 0;
3bd1e081 MD	54	off_t orig_offset = stream->out_fd_offset;
3bd1e081 MD	55	int outfd = stream->out_fd;
00e2e675 DG	56	uint64_t metadata_id;
	57	struct consumer_relayd_sock_pair *relayd = NULL;
	58
b0b335c8 MD	59	/* RCU lock for the relayd pointer */
	60	rcu_read_lock();
	61
00e2e675 DG	62	/* Flag that the current stream if set for network streaming. */
	63	if (stream->net_seq_idx != -1) {
	64	relayd = consumer_find_relayd(stream->net_seq_idx);
	65	if (relayd == NULL) {
87c1611d DG	66	ERR("UST consumer mmap(), unable to find relay for index %d",
87c1611d DG	67	stream->net_seq_idx);
00e2e675 DG	68	goto end;
	69	}
	70	}
3bd1e081 MD	71
	72	/* get the offset inside the fd to mmap */
	73	ret = ustctl_get_mmap_read_offset(stream->chan->handle,
	74	stream->buf, &mmap_offset);
	75	if (ret != 0) {
87dc6a9c	76	errno = -ret;
4c462e79	77	PERROR("ustctl_get_mmap_read_offset");
47e81c02	78	written = ret;
3bd1e081 MD	79	goto end;
3bd1e081 MD	80	}
00e2e675 DG	81
	82	/* Handle stream on the relayd if the output is on the network */
	83	if (relayd) {
f6416125 MD	84	unsigned long netlen = len;
f6416125 MD	85
00e2e675 DG	86	if (stream->metadata_flag) {
	87	/* Only lock if metadata since we use the control socket. */
	88	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
f6416125	89	netlen += sizeof(stream->relayd_stream_id);
00e2e675 DG	90	}
00e2e675 DG	91
f6416125	92	ret = consumer_handle_stream_before_relayd(stream, netlen);
00e2e675 DG	93	if (ret >= 0) {
	94	outfd = ret;
	95
	96	/* Write metadata stream id before payload */
	97	if (stream->metadata_flag) {
	98	metadata_id = htobe64(stream->relayd_stream_id);
	99	do {
	100	ret = write(outfd, (void *) &metadata_id,
	101	sizeof(stream->relayd_stream_id));
6f94560a MD	102	} while (ret < 0 && errno == EINTR);
	103	if (ret < 0) {
	104	PERROR("write metadata stream id");
	105	written = ret;
	106	goto end;
	107	}
00e2e675 DG	108	DBG("Metadata stream id %zu written before data",
	109	stream->relayd_stream_id);
	110	}
	111	}
	112	/* Else, use the default set before which is the filesystem. */
	113	}
	114
3bd1e081	115	while (len > 0) {
6f94560a MD	116	do {
	117	ret = write(outfd, stream->mmap_base + mmap_offset, len);
	118	} while (ret < 0 && errno == EINTR);
47e81c02	119	if (ret < 0) {
6f94560a MD	120	PERROR("Error in file write");
	121	if (written == 0) {
	122	written = ret;
47e81c02	123	}
6f94560a	124	goto end;
47e81c02	125	} else if (ret > len) {
00e2e675	126	PERROR("ret %ld > len %lu", ret, len);
47e81c02	127	written += ret;
3bd1e081	128	goto end;
47e81c02 MD	129	} else {
	130	len -= ret;
	131	mmap_offset += ret;
3bd1e081	132	}
00e2e675 DG	133	DBG("UST mmap write() ret %ld (len %lu)", ret, len);
	134
	135	/* This call is useless on a socket so better save a syscall. */
	136	if (!relayd) {
	137	/* This won't block, but will start writeout asynchronously */
	138	lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
	139	SYNC_FILE_RANGE_WRITE);
	140	stream->out_fd_offset += ret;
	141	}
47e81c02	142	written += ret;
3bd1e081	143	}
3bd1e081	144	lttng_consumer_sync_trace_file(stream, orig_offset);
00e2e675	145
3bd1e081	146	end:
00e2e675 DG	147	if (relayd && stream->metadata_flag) {
	148	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
	149	}
b0b335c8	150	rcu_read_unlock();
47e81c02	151	return written;
3bd1e081 MD	152	}
	153
	154	/*
	155	* Splice the data from the ring buffer to the tracefile.
	156	*
	157	* Returns the number of bytes spliced.
	158	*/
4078b776	159	ssize_t lttng_ustconsumer_on_read_subbuffer_splice(
3bd1e081 MD	160	struct lttng_consumer_local_data *ctx,
	161	struct lttng_consumer_stream *stream, unsigned long len)
	162	{
	163	return -ENOSYS;
	164	}
	165
	166	/*
	167	* Take a snapshot for a specific fd
	168	*
	169	* Returns 0 on success, < 0 on error
	170	*/
	171	int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
	172	struct lttng_consumer_stream *stream)
	173	{
	174	int ret = 0;
	175
	176	ret = ustctl_snapshot(stream->chan->handle, stream->buf);
	177	if (ret != 0) {
87dc6a9c	178	errno = -ret;
4c462e79	179	PERROR("Getting sub-buffer snapshot.");
3bd1e081 MD	180	}
	181
	182	return ret;
	183	}
	184
	185	/*
	186	* Get the produced position
	187	*
	188	* Returns 0 on success, < 0 on error
	189	*/
	190	int lttng_ustconsumer_get_produced_snapshot(
	191	struct lttng_consumer_local_data *ctx,
	192	struct lttng_consumer_stream *stream,
	193	unsigned long *pos)
	194	{
	195	int ret;
	196
	197	ret = ustctl_snapshot_get_produced(stream->chan->handle,
	198	stream->buf, pos);
	199	if (ret != 0) {
87dc6a9c	200	errno = -ret;
4c462e79	201	PERROR("kernctl_snapshot_get_produced");
3bd1e081 MD	202	}
	203
	204	return ret;
	205	}
	206
	207	int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
	208	int sock, struct pollfd *consumer_sockpoll)
	209	{
	210	ssize_t ret;
	211	struct lttcomm_consumer_msg msg;
	212
	213	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
	214	if (ret != sizeof(msg)) {
	215	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
	216	return ret;
	217	}
	218	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
	219	return -ENOENT;
	220	}
	221
b0b335c8 MD	222	/* relayd need RCU read-side lock */
	223	rcu_read_lock();
	224
3bd1e081	225	switch (msg.cmd_type) {
00e2e675 DG	226	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
	227	{
	228	int fd;
	229	struct consumer_relayd_sock_pair *relayd;
	230
	231	DBG("UST Consumer adding relayd socket");
	232
	233	/* Get relayd reference if exists. */
	234	relayd = consumer_find_relayd(msg.u.relayd_sock.net_index);
	235	if (relayd == NULL) {
	236	/* Not found. Allocate one. */
	237	relayd = consumer_allocate_relayd_sock_pair(
	238	msg.u.relayd_sock.net_index);
	239	if (relayd == NULL) {
	240	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	241	goto end_nosignal;
	242	}
	243	}
	244
	245	/* Poll on consumer socket. */
	246	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
	247	return -EINTR;
	248	}
	249
	250	/* Get relayd socket from session daemon */
	251	ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
	252	if (ret != sizeof(fd)) {
	253	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
	254	goto end_nosignal;
	255	}
	256
	257	/* Copy socket information and received FD */
	258	switch (msg.u.relayd_sock.type) {
	259	case LTTNG_STREAM_CONTROL:
	260	/* Copy received lttcomm socket */
	261	lttcomm_copy_sock(&relayd->control_sock, &msg.u.relayd_sock.sock);
	262	ret = lttcomm_create_sock(&relayd->control_sock);
	263	if (ret < 0) {
	264	goto end_nosignal;
	265	}
	266
	267	/* Close the created socket fd which is useless */
	268	close(relayd->control_sock.fd);
	269
	270	/* Assign new file descriptor */
	271	relayd->control_sock.fd = fd;
	272	break;
	273	case LTTNG_STREAM_DATA:
	274	/* Copy received lttcomm socket */
	275	lttcomm_copy_sock(&relayd->data_sock, &msg.u.relayd_sock.sock);
	276	ret = lttcomm_create_sock(&relayd->data_sock);
	277	if (ret < 0) {
	278	goto end_nosignal;
	279	}
	280
	281	/* Close the created socket fd which is useless */
	282	close(relayd->data_sock.fd);
	283
	284	/* Assign new file descriptor */
	285	relayd->data_sock.fd = fd;
	286	break;
	287	default:
	288	ERR("Unknown relayd socket type");
	289	goto end_nosignal;
290	}
291
292	DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
293	msg.u.relayd_sock.type == LTTNG_STREAM_CONTROL ? "control" : "data",
294	relayd->net_seq_idx, fd);
295
296	/*
297	* Add relayd socket pair to consumer data hashtable. If object already
298	* exists or on error, the function gracefully returns.
299	*/
300	consumer_add_relayd(relayd);
301
302	goto end_nosignal;
303	}
3bd1e081 MD	304	case LTTNG_CONSUMER_ADD_CHANNEL:
	305	{
	306	struct lttng_consumer_channel *new_channel;
	307	int fds[1];
	308	size_t nb_fd = 1;
	309
	310	/* block */
	311	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
	312	return -EINTR;
	313	}
	314	ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
	315	if (ret != sizeof(fds)) {
	316	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
	317	return ret;
	318	}
	319
	320	DBG("consumer_add_channel %d", msg.u.channel.channel_key);
	321
	322	new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
	323	fds[0], -1,
	324	msg.u.channel.mmap_len,
	325	msg.u.channel.max_sb_size);
	326	if (new_channel == NULL) {
	327	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	328	goto end_nosignal;
	329	}
	330	if (ctx->on_recv_channel != NULL) {
	331	ret = ctx->on_recv_channel(new_channel);
	332	if (ret == 0) {
	333	consumer_add_channel(new_channel);
	334	} else if (ret < 0) {
	335	goto end_nosignal;
	336	}
	337	} else {
	338	consumer_add_channel(new_channel);
	339	}
	340	goto end_nosignal;
	341	}
	342	case LTTNG_CONSUMER_ADD_STREAM:
	343	{
	344	struct lttng_consumer_stream *new_stream;
	345	int fds[2];
	346	size_t nb_fd = 2;
00e2e675	347	struct consumer_relayd_sock_pair *relayd = NULL;
3bd1e081 MD	348
	349	/* block */
	350	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
	351	return -EINTR;
	352	}
	353	ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
	354	if (ret != sizeof(fds)) {
	355	lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
	356	return ret;
	357	}
	358
d41f73b7	359	assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
7ad0a0cb	360	new_stream = consumer_allocate_stream(msg.u.channel.channel_key,
3bd1e081 MD	361	msg.u.stream.stream_key,
	362	fds[0], fds[1],
	363	msg.u.stream.state,
	364	msg.u.stream.mmap_len,
	365	msg.u.stream.output,
6df2e2c9 MD	366	msg.u.stream.path_name,
6df2e2c9 MD	367	msg.u.stream.uid,
00e2e675 DG	368	msg.u.stream.gid,
	369	msg.u.stream.net_index,
	370	msg.u.stream.metadata_flag);
3bd1e081 MD	371	if (new_stream == NULL) {
	372	lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
	373	goto end;
	374	}
00e2e675 DG	375
	376	/* The stream is not metadata. Get relayd reference if exists. */
	377	relayd = consumer_find_relayd(msg.u.stream.net_index);
	378	if (relayd != NULL) {
	379	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
	380	/* Add stream on the relayd */
	381	ret = relayd_add_stream(&relayd->control_sock,
	382	msg.u.stream.name, msg.u.stream.path_name,
	383	&new_stream->relayd_stream_id);
	384	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
	385	if (ret < 0) {
	386	goto end;
	387	}
	388	} else if (msg.u.stream.net_index != -1) {
	389	ERR("Network sequence index %d unknown. Not adding stream.",
	390	msg.u.stream.net_index);
	391	free(new_stream);
	392	goto end;
	393	}
	394
3bd1e081 MD	395	if (ctx->on_recv_stream != NULL) {
	396	ret = ctx->on_recv_stream(new_stream);
	397	if (ret == 0) {
	398	consumer_add_stream(new_stream);
	399	} else if (ret < 0) {
	400	goto end;
	401	}
	402	} else {
	403	consumer_add_stream(new_stream);
	404	}
00e2e675 DG	405
	406	DBG("UST consumer_add_stream %s (%d,%d) with relayd id %lu",
	407	msg.u.stream.path_name, fds[0], fds[1],
	408	new_stream->relayd_stream_id);
3bd1e081 MD	409	break;
	410	}
	411	case LTTNG_CONSUMER_UPDATE_STREAM:
	412	{
7ad0a0cb MD	413	return -ENOSYS;
7ad0a0cb MD	414	#if 0
3bd1e081 MD	415	if (ctx->on_update_stream != NULL) {
	416	ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
	417	if (ret == 0) {
	418	consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
	419	} else if (ret < 0) {
	420	goto end;
	421	}
	422	} else {
	423	consumer_change_stream_state(msg.u.stream.stream_key,
	424	msg.u.stream.state);
	425	}
7ad0a0cb	426	#endif
3bd1e081 MD	427	break;
	428	}
	429	default:
	430	break;
	431	}
	432	end:
04fdd819 MD	433	/*
	434	* Wake-up the other end by writing a null byte in the pipe
	435	* (non-blocking). Important note: Because writing into the
	436	* pipe is non-blocking (and therefore we allow dropping wakeup
	437	* data, as long as there is wakeup data present in the pipe
	438	* buffer to wake up the other end), the other end should
	439	* perform the following sequence for waiting:
	440	* 1) empty the pipe (reads).
	441	* 2) perform update operation.
	442	* 3) wait on the pipe (poll).
	443	*/
	444	do {
	445	ret = write(ctx->consumer_poll_pipe[1], "", 1);
6f94560a	446	} while (ret < 0 && errno == EINTR);
3bd1e081	447	end_nosignal:
b0b335c8	448	rcu_read_unlock();
3bd1e081 MD	449	return 0;
	450	}
	451
	452	int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
	453	{
13161846	454	struct lttng_ust_object_data obj;
3bd1e081 MD	455
	456	obj.handle = -1;
	457	obj.shm_fd = chan->shm_fd;
	458	obj.wait_fd = chan->wait_fd;
	459	obj.memory_map_size = chan->mmap_len;
	460	chan->handle = ustctl_map_channel(&obj);
	461	if (!chan->handle) {
	462	return -ENOMEM;
	463	}
b5c5fc29	464	chan->wait_fd_is_copy = 1;
2c1dd183	465	chan->shm_fd = -1;
b5c5fc29	466
3bd1e081 MD	467	return 0;
	468	}
	469
d056b477 MD	470	void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
	471	{
	472	ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
effcf122	473	stream->hangup_flush_done = 1;
d056b477 MD	474	}
d056b477 MD	475
3bd1e081 MD	476	void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
	477	{
	478	ustctl_unmap_channel(chan->handle);
	479	}
	480
	481	int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
	482	{
13161846	483	struct lttng_ust_object_data obj;
3bd1e081 MD	484	int ret;
	485
	486	obj.handle = -1;
	487	obj.shm_fd = stream->shm_fd;
	488	obj.wait_fd = stream->wait_fd;
	489	obj.memory_map_size = stream->mmap_len;
	490	ret = ustctl_add_stream(stream->chan->handle, &obj);
	491	if (ret)
	492	return ret;
	493	stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
	494	if (!stream->buf)
	495	return -EBUSY;
2c1dd183 MD	496	/* ustctl_open_stream_read has closed the shm fd. */
	497	stream->wait_fd_is_copy = 1;
	498	stream->shm_fd = -1;
	499
3bd1e081 MD	500	stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
	501	if (!stream->mmap_base) {
	502	return -EINVAL;
	503	}
ee77a7b0	504
3bd1e081 MD	505	return 0;
	506	}
	507
	508	void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
	509	{
	510	ustctl_close_stream_read(stream->chan->handle, stream->buf);
	511	}
d41f73b7 MD	512
	513
	514	int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
	515	struct lttng_consumer_local_data *ctx)
	516	{
	517	unsigned long len;
	518	int err;
	519	long ret = 0;
	520	struct lttng_ust_shm_handle *handle;
	521	struct lttng_ust_lib_ring_buffer *buf;
	522	char dummy;
	523	ssize_t readlen;
	524
	525	DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
	526	stream->wait_fd, stream->key);
	527
	528	/* We can consume the 1 byte written into the wait_fd by UST */
effcf122 MD	529	if (!stream->hangup_flush_done) {
	530	do {
	531	readlen = read(stream->wait_fd, &dummy, 1);
87dc6a9c	532	} while (readlen == -1 && errno == EINTR);
effcf122 MD	533	if (readlen == -1) {
	534	ret = readlen;
	535	goto end;
	536	}
d41f73b7 MD	537	}
	538
	539	buf = stream->buf;
	540	handle = stream->chan->handle;
	541	/* Get the next subbuffer */
	542	err = ustctl_get_next_subbuf(handle, buf);
	543	if (err != 0) {
effcf122	544	ret = -ret; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
d41f73b7 MD	545	/*
	546	* This is a debug message even for single-threaded consumer,
	547	* because poll() have more relaxed criterions than get subbuf,
	548	* so get_subbuf may fail for short race windows where poll()
	549	* would issue wakeups.
	550	*/
	551	DBG("Reserving sub buffer failed (everything is normal, "
	552	"it is due to concurrency)");
	553	goto end;
	554	}
	555	assert(stream->output == LTTNG_EVENT_MMAP);
	556	/* read the used subbuffer size */
	557	err = ustctl_get_padded_subbuf_size(handle, buf, &len);
effcf122	558	assert(err == 0);
d41f73b7 MD	559	/* write the subbuffer to the tracefile */
d41f73b7 MD	560	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
47e81c02	561	if (ret != len) {
d41f73b7 MD	562	/*
	563	* display the error but continue processing to try
	564	* to release the subbuffer
	565	*/
	566	ERR("Error writing to tracefile");
	567	}
	568	err = ustctl_put_next_subbuf(handle, buf);
effcf122	569	assert(err == 0);
d41f73b7 MD	570	end:
	571	return ret;
	572	}
	573
	574	int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
	575	{
	576	int ret;
	577
	578	/* Opening the tracefile in write mode */
00e2e675	579	if (stream->path_name != NULL && stream->net_seq_idx == -1) {
e11d277b	580	ret = run_as_open(stream->path_name,
60b6c79c MD	581	O_WRONLY\|O_CREAT\|O_TRUNC,
	582	S_IRWXU\|S_IRWXG\|S_IRWXO,
	583	stream->uid, stream->gid);
d41f73b7 MD	584	if (ret < 0) {
d41f73b7 MD	585	ERR("Opening %s", stream->path_name);
4c462e79	586	PERROR("open");
d41f73b7 MD	587	goto error;
	588	}
	589	stream->out_fd = ret;
	590	}
	591
	592	/* we return 0 to let the library handle the FD internally */
	593	return 0;
	594
	595	error:
	596	return ret;
	597	}