[lttng-tools.git] / src / bin / lttng-relayd / viewer-stream.c

/*
 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
 *                      David Goulet <dgoulet@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
#define _LGPL_SOURCE
#include <common/common.h>
#include <common/index/index.h>

#include "lttng-relayd.h"
#include "viewer-stream.h"

static void free_stream(struct relay_viewer_stream *stream)
{
	assert(stream);

	free(stream->path_name);
	free(stream->channel_name);
	free(stream);
}

static void deferred_free_viewer_stream(struct rcu_head *head)
{
	struct relay_viewer_stream *stream =
		caa_container_of(head, struct relay_viewer_stream, rcu_node);

	free_stream(stream);
}

struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
		enum lttng_viewer_seek seek_t, struct ctf_trace *ctf_trace)
{
	struct relay_viewer_stream *vstream;

	assert(stream);
	assert(ctf_trace);

	vstream = zmalloc(sizeof(*vstream));
	if (!vstream) {
		PERROR("relay viewer stream zmalloc");
		goto error;
	}

	vstream->session_id = stream->session_id;
	vstream->stream_handle = stream->stream_handle;
	vstream->path_name = strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
	vstream->channel_name = strndup(stream->channel_name,
			LTTNG_VIEWER_NAME_MAX);
	vstream->tracefile_count = stream->tracefile_count;
	vstream->metadata_flag = stream->metadata_flag;
	vstream->tracefile_count_last = -1ULL;

	switch (seek_t) {
	case LTTNG_VIEWER_SEEK_BEGINNING:
		vstream->tracefile_count_current = stream->oldest_tracefile_id;
		break;
	case LTTNG_VIEWER_SEEK_LAST:
		vstream->tracefile_count_current = stream->tracefile_count_current;
		break;
	default:
		assert(0);
		goto error;
	}

	if (vstream->metadata_flag) {
		ctf_trace->viewer_metadata_stream = vstream;
	}

	/* Globally visible after the add unique. */
	lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
	lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);

	vstream->index_read_fd = -1;
	vstream->read_fd = -1;

	/*
	 * This is to avoid a race between the initialization of this object and
	 * the close of the given stream. If the stream is unable to find this
	 * viewer stream when closing, this copy will at least take the latest
	 * value. We also need that for the seek_last.
	 */
	vstream->total_index_received = stream->total_index_received;

	/*
	 * If we never received an index for the current stream, delay the opening
	 * of the index, otherwise open it right now.
	 */
	if (vstream->tracefile_count_current == stream->tracefile_count_current
			&& vstream->total_index_received == 0) {
		vstream->index_read_fd = -1;
	} else {
		int read_fd;

		read_fd = index_open(vstream->path_name, vstream->channel_name,
				vstream->tracefile_count, vstream->tracefile_count_current);
		if (read_fd < 0) {
			goto error;
		}
		vstream->index_read_fd = read_fd;
	}

	if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_read_fd >= 0) {
		off_t lseek_ret;

		lseek_ret = lseek(vstream->index_read_fd,
				vstream->total_index_received * sizeof(struct ctf_packet_index),
				SEEK_CUR);
		if (lseek_ret < 0) {
			goto error;
		}
		vstream->last_sent_index = vstream->total_index_received;
	}

	return vstream;

error:
	if (vstream) {
		free_stream(vstream);
	}
	return NULL;
}

void viewer_stream_delete(struct relay_viewer_stream *stream)
{
	int ret;
	struct lttng_ht_iter iter;

	iter.iter.node = &stream->stream_n.node;
	ret = lttng_ht_del(viewer_streams_ht, &iter);
	assert(!ret);
}

void viewer_stream_destroy(struct ctf_trace *ctf_trace,
		struct relay_viewer_stream *stream)
{
	int ret;

	assert(stream);

	if (ctf_trace) {
		ctf_trace_put_ref(ctf_trace);
	}

	if (stream->read_fd >= 0) {
		ret = close(stream->read_fd);
		if (ret < 0) {
			PERROR("close read_fd");
		}
	}
	if (stream->index_read_fd >= 0) {
		ret = close(stream->index_read_fd);
		if (ret < 0) {
			PERROR("close index_read_fd");
		}
	}

	call_rcu(&stream->rcu_node, deferred_free_viewer_stream);
}

/*
 * Find viewer stream by id. RCU read side lock MUST be acquired.
 *
 * Return stream if found else NULL.
 */
struct relay_viewer_stream *viewer_stream_find_by_id(uint64_t id)
{
	struct lttng_ht_node_u64 *node;
	struct lttng_ht_iter iter;
	struct relay_viewer_stream *stream = NULL;

	lttng_ht_lookup(viewer_streams_ht, &id, &iter);
	node = lttng_ht_iter_get_node_u64(&iter);
	if (!node) {
		DBG("Relay viewer stream %" PRIu64 " not found", id);
		goto end;
	}
	stream = caa_container_of(node, struct relay_viewer_stream, stream_n);

end:
	return stream;
}

/*
 * Rotate a stream to the next tracefile.
 *
 * Must be called with viewer_stream_rotation_lock held.
 * Returns 0 on success, 1 on EOF, a negative value on error.
 */
int viewer_stream_rotate(struct relay_viewer_stream *vstream,
		struct relay_stream *stream)
{
	int ret;
	uint64_t tracefile_id;

	assert(vstream);
	assert(stream);

	if (vstream->tracefile_count == 0) {
		/* Ignore rotation, there is none to do. */
		ret = 0;
		goto end;
	}

	tracefile_id = (vstream->tracefile_count_current + 1) %
		vstream->tracefile_count;

	/* Detect the last tracefile to open. */
	if (vstream->tracefile_count_last != -1ULL &&
			vstream->tracefile_count_last ==
			vstream->tracefile_count_current) {
		ret = 1;
		goto end;
	}

	/*
	 * The writer and the reader are not working in the same tracefile, we can
	 * read up to EOF, we don't care about the total_index_received.
	 */
	if (stream->close_flag || (stream->tracefile_count_current != tracefile_id)) {
		vstream->close_write_flag = 1;
	} else {
		/*
		 * We are opening a file that is still open in write, make sure we
		 * limit our reading to the number of indexes received.
		 */
		vstream->close_write_flag = 0;
		if (stream->close_flag) {
			vstream->total_index_received = stream->total_index_received;
		}
	}
	vstream->tracefile_count_current = tracefile_id;

	ret = close(vstream->index_read_fd);
	if (ret < 0) {
		PERROR("close index file %d", vstream->index_read_fd);
	}
	vstream->index_read_fd = -1;

	ret = close(vstream->read_fd);
	if (ret < 0) {
		PERROR("close tracefile %d", vstream->read_fd);
	}
	vstream->read_fd = -1;

	pthread_mutex_lock(&vstream->overwrite_lock);
	vstream->abort_flag = 0;
	pthread_mutex_unlock(&vstream->overwrite_lock);

	ret = index_open(vstream->path_name, vstream->channel_name,
			vstream->tracefile_count, vstream->tracefile_count_current);
	if (ret < 0) {
		goto error;
	}
	vstream->index_read_fd = ret;

	ret = 0;

end:
error:
	return ret;
}
Commit	Line	Data
	1	/*
	2	* Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
	3	* David Goulet <dgoulet@efficios.com>
	4	*
	5	* This program is free software; you can redistribute it and/or modify it
	6	* under the terms of the GNU General Public License, version 2 only, as
	7	* published by the Free Software Foundation.
	8	*
	9	* This program is distributed in the hope that it will be useful, but WITHOUT
	10	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	11	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	12	* more details.
	13	*
	14	* You should have received a copy of the GNU General Public License along with
	15	* this program; if not, write to the Free Software Foundation, Inc., 51
	16	* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	17	*/
	18
	19	#define _GNU_SOURCE
	20	#define _LGPL_SOURCE
	21	#include <common/common.h>
	22	#include <common/index/index.h>
	23
	24	#include "lttng-relayd.h"
	25	#include "viewer-stream.h"
	26
	27	static void free_stream(struct relay_viewer_stream *stream)
	28	{
	29	assert(stream);
	30
	31	free(stream->path_name);
	32	free(stream->channel_name);
	33	free(stream);
	34	}
	35
	36	static void deferred_free_viewer_stream(struct rcu_head *head)
	37	{
	38	struct relay_viewer_stream *stream =
	39	caa_container_of(head, struct relay_viewer_stream, rcu_node);
	40
	41	free_stream(stream);
	42	}
	43
	44	struct relay_viewer_stream viewer_stream_create(struct relay_stream stream,
	45	enum lttng_viewer_seek seek_t, struct ctf_trace *ctf_trace)
	46	{
	47	struct relay_viewer_stream *vstream;
	48
	49	assert(stream);
	50	assert(ctf_trace);
	51
	52	vstream = zmalloc(sizeof(*vstream));
	53	if (!vstream) {
	54	PERROR("relay viewer stream zmalloc");
	55	goto error;
	56	}
	57
	58	vstream->session_id = stream->session_id;
	59	vstream->stream_handle = stream->stream_handle;
	60	vstream->path_name = strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
	61	vstream->channel_name = strndup(stream->channel_name,
	62	LTTNG_VIEWER_NAME_MAX);
	63	vstream->tracefile_count = stream->tracefile_count;
	64	vstream->metadata_flag = stream->metadata_flag;
	65	vstream->tracefile_count_last = -1ULL;
	66
	67	switch (seek_t) {
	68	case LTTNG_VIEWER_SEEK_BEGINNING:
	69	vstream->tracefile_count_current = stream->oldest_tracefile_id;
	70	break;
	71	case LTTNG_VIEWER_SEEK_LAST:
	72	vstream->tracefile_count_current = stream->tracefile_count_current;
	73	break;
	74	default:
	75	assert(0);
	76	goto error;
	77	}
	78
	79	if (vstream->metadata_flag) {
	80	ctf_trace->viewer_metadata_stream = vstream;
	81	}
	82
	83	/* Globally visible after the add unique. */
	84	lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
	85	lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
	86
	87	vstream->index_read_fd = -1;
	88	vstream->read_fd = -1;
	89
	90	/*
	91	* This is to avoid a race between the initialization of this object and
	92	* the close of the given stream. If the stream is unable to find this
	93	* viewer stream when closing, this copy will at least take the latest
	94	* value. We also need that for the seek_last.
	95	*/
	96	vstream->total_index_received = stream->total_index_received;
	97
	98	/*
	99	* If we never received an index for the current stream, delay the opening
	100	* of the index, otherwise open it right now.
	101	*/
	102	if (vstream->tracefile_count_current == stream->tracefile_count_current
	103	&& vstream->total_index_received == 0) {
	104	vstream->index_read_fd = -1;
	105	} else {
	106	int read_fd;
	107
	108	read_fd = index_open(vstream->path_name, vstream->channel_name,
	109	vstream->tracefile_count, vstream->tracefile_count_current);
	110	if (read_fd < 0) {
	111	goto error;
	112	}
	113	vstream->index_read_fd = read_fd;
	114	}
	115
	116	if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_read_fd >= 0) {
	117	off_t lseek_ret;
	118
	119	lseek_ret = lseek(vstream->index_read_fd,
	120	vstream->total_index_received * sizeof(struct ctf_packet_index),
	121	SEEK_CUR);
	122	if (lseek_ret < 0) {
	123	goto error;
	124	}
	125	vstream->last_sent_index = vstream->total_index_received;
	126	}
	127
	128	return vstream;
	129
	130	error:
	131	if (vstream) {
	132	free_stream(vstream);
	133	}
	134	return NULL;
	135	}
	136
	137	void viewer_stream_delete(struct relay_viewer_stream *stream)
	138	{
	139	int ret;
	140	struct lttng_ht_iter iter;
	141
	142	iter.iter.node = &stream->stream_n.node;
	143	ret = lttng_ht_del(viewer_streams_ht, &iter);
	144	assert(!ret);
	145	}
	146
	147	void viewer_stream_destroy(struct ctf_trace *ctf_trace,
	148	struct relay_viewer_stream *stream)
	149	{
	150	int ret;
	151
	152	assert(stream);
	153
	154	if (ctf_trace) {
	155	ctf_trace_put_ref(ctf_trace);
	156	}
	157
	158	if (stream->read_fd >= 0) {
	159	ret = close(stream->read_fd);
	160	if (ret < 0) {
	161	PERROR("close read_fd");
	162	}
	163	}
	164	if (stream->index_read_fd >= 0) {
	165	ret = close(stream->index_read_fd);
	166	if (ret < 0) {
	167	PERROR("close index_read_fd");
	168	}
	169	}
	170
	171	call_rcu(&stream->rcu_node, deferred_free_viewer_stream);
	172	}
	173
	174	/*
	175	* Find viewer stream by id. RCU read side lock MUST be acquired.
	176	*
	177	* Return stream if found else NULL.
	178	*/
	179	struct relay_viewer_stream *viewer_stream_find_by_id(uint64_t id)
	180	{
	181	struct lttng_ht_node_u64 *node;
	182	struct lttng_ht_iter iter;
	183	struct relay_viewer_stream *stream = NULL;
	184
	185	lttng_ht_lookup(viewer_streams_ht, &id, &iter);
	186	node = lttng_ht_iter_get_node_u64(&iter);
	187	if (!node) {
	188	DBG("Relay viewer stream %" PRIu64 " not found", id);
	189	goto end;
	190	}
	191	stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
	192
	193	end:
	194	return stream;
	195	}
	196
	197	/*
	198	* Rotate a stream to the next tracefile.
	199	*
	200	* Must be called with viewer_stream_rotation_lock held.
	201	* Returns 0 on success, 1 on EOF, a negative value on error.
	202	*/
	203	int viewer_stream_rotate(struct relay_viewer_stream *vstream,
	204	struct relay_stream *stream)
	205	{
	206	int ret;
	207	uint64_t tracefile_id;
	208
	209	assert(vstream);
	210	assert(stream);
	211
	212	if (vstream->tracefile_count == 0) {
	213	/* Ignore rotation, there is none to do. */
	214	ret = 0;
	215	goto end;
	216	}
	217
	218	tracefile_id = (vstream->tracefile_count_current + 1) %
	219	vstream->tracefile_count;
	220
	221	/* Detect the last tracefile to open. */
	222	if (vstream->tracefile_count_last != -1ULL &&
	223	vstream->tracefile_count_last ==
	224	vstream->tracefile_count_current) {
	225	ret = 1;
	226	goto end;
	227	}
	228
	229	/*
	230	* The writer and the reader are not working in the same tracefile, we can
	231	* read up to EOF, we don't care about the total_index_received.
	232	*/
	233	if (stream->close_flag \|\| (stream->tracefile_count_current != tracefile_id)) {
	234	vstream->close_write_flag = 1;
	235	} else {
	236	/*
	237	* We are opening a file that is still open in write, make sure we
	238	* limit our reading to the number of indexes received.
	239	*/
	240	vstream->close_write_flag = 0;
	241	if (stream->close_flag) {
	242	vstream->total_index_received = stream->total_index_received;
	243	}
	244	}
	245	vstream->tracefile_count_current = tracefile_id;
	246
	247	ret = close(vstream->index_read_fd);
	248	if (ret < 0) {
	249	PERROR("close index file %d", vstream->index_read_fd);
	250	}
	251	vstream->index_read_fd = -1;
	252
	253	ret = close(vstream->read_fd);
	254	if (ret < 0) {
	255	PERROR("close tracefile %d", vstream->read_fd);
	256	}
	257	vstream->read_fd = -1;
	258
	259	pthread_mutex_lock(&vstream->overwrite_lock);
	260	vstream->abort_flag = 0;
	261	pthread_mutex_unlock(&vstream->overwrite_lock);
	262
	263	ret = index_open(vstream->path_name, vstream->channel_name,
	264	vstream->tracefile_count, vstream->tracefile_count_current);
	265	if (ret < 0) {
	266	goto error;
	267	}
	268	vstream->index_read_fd = ret;
	269
	270	ret = 0;
	271
	272	end:
	273	error:
	274	return ret;
	275	}