[babeltrace.git] / plugins / ctf / lttng-live / data-stream.c

/*
 * Copyright 2016 - Philippe Proulx <pproulx@efficios.com>
 * Copyright 2016 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
 * Copyright 2010-2011 - EfficiOS Inc. and Linux Foundation
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

#define BT_LOG_TAG "PLUGIN-CTF-LTTNG-LIVE-SRC-DS"
#include "logging.h"

#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <stdbool.h>
#include <glib.h>
#include <inttypes.h>
#include <babeltrace/compat/mman-internal.h>
#include <babeltrace/babeltrace.h>
#include "../common/notif-iter/notif-iter.h"
#include <babeltrace/assert-internal.h>

#include "data-stream.h"

static
enum bt_notif_iter_medium_status medop_request_bytes(
		size_t request_sz, uint8_t **buffer_addr,
		size_t *buffer_sz, void *data)
{
	enum bt_notif_iter_medium_status status =
		BT_NOTIF_ITER_MEDIUM_STATUS_OK;
	struct lttng_live_stream_iterator *stream = data;
	struct lttng_live_trace *trace = stream->trace;
	struct lttng_live_session *session = trace->session;
	struct lttng_live_component *lttng_live = session->lttng_live;
	uint64_t recv_len = 0;
	uint64_t len_left;
	uint64_t read_len;
	//int i;

	len_left = stream->base_offset + stream->len - stream->offset;
	if (!len_left) {
		stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
		status = BT_NOTIF_ITER_MEDIUM_STATUS_AGAIN;
		return status;
	}
	read_len = MIN(request_sz, stream->buflen);
	read_len = MIN(read_len, len_left);
	status = lttng_live_get_stream_bytes(lttng_live,
			stream, stream->buf, stream->offset,
			read_len, &recv_len);
	*buffer_addr = stream->buf;
	*buffer_sz = recv_len;
	stream->offset += recv_len;
	return status;
}

static
const struct bt_stream *medop_get_stream(
		const struct bt_stream_class *stream_class,
		uint64_t stream_id, void *data)
{
	struct lttng_live_stream_iterator *lttng_live_stream = data;

	if (!lttng_live_stream->stream) {
		int64_t stream_class_id =
			bt_stream_class_get_id(stream_class);

		BT_LOGD("Creating stream %s (ID: %" PRIu64 ") out of stream class %" PRId64,
			lttng_live_stream->name, stream_id, stream_class_id);

		if (stream_id == -1ULL) {
			/* No stream ID */
			lttng_live_stream->stream = bt_stream_create(
				stream_class, lttng_live_stream->name);
		} else {
			lttng_live_stream->stream =
				bt_stream_create_with_id(stream_class,
					lttng_live_stream->name, stream_id);
		}

		if (!lttng_live_stream->stream) {
			BT_LOGE("Cannot create stream %s (stream class %" PRId64 ", stream ID %" PRIu64 ")",
					lttng_live_stream->name,
					stream_class_id, stream_id);
		}
	}

	return lttng_live_stream->stream;
}

static struct bt_notif_iter_medium_ops medops = {
	.request_bytes = medop_request_bytes,
	.get_stream = medop_get_stream,
};

BT_HIDDEN
enum bt_lttng_live_iterator_status lttng_live_lazy_notif_init(
		struct lttng_live_session *session)
{
	struct lttng_live_component *lttng_live = session->lttng_live;
	struct lttng_live_trace *trace;

	if (!session->lazy_stream_notif_init) {
		return BT_LTTNG_LIVE_ITERATOR_STATUS_OK;
	}

	bt_list_for_each_entry(trace, &session->traces, node) {
		struct lttng_live_stream_iterator *stream;

		bt_list_for_each_entry(stream, &trace->streams, node) {
			if (stream->notif_iter) {
				continue;
			}
			stream->notif_iter = bt_notif_iter_create(trace->trace,
					lttng_live->max_query_size, medops,
					stream);
			if (!stream->notif_iter) {
				goto error;
			}
		}
	}

	session->lazy_stream_notif_init = false;

	return BT_LTTNG_LIVE_ITERATOR_STATUS_OK;

error:
	return BT_LTTNG_LIVE_ITERATOR_STATUS_ERROR;
}

BT_HIDDEN
struct lttng_live_stream_iterator *lttng_live_stream_iterator_create(
		struct lttng_live_session *session,
		uint64_t ctf_trace_id,
		uint64_t stream_id)
{
	struct lttng_live_component *lttng_live = session->lttng_live;
	struct lttng_live_stream_iterator *stream =
			g_new0(struct lttng_live_stream_iterator, 1);
	struct lttng_live_trace *trace;
	int ret;

	trace = lttng_live_ref_trace(session, ctf_trace_id);
	if (!trace) {
		goto error;
	}

	stream->p.type = LIVE_STREAM_TYPE_STREAM;
	stream->trace = trace;
	stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
	stream->viewer_stream_id = stream_id;
	stream->ctf_stream_class_id = -1ULL;
	stream->last_returned_inactivity_timestamp = INT64_MIN;

	if (trace->trace) {
		stream->notif_iter = bt_notif_iter_create(trace->trace,
				lttng_live->max_query_size, medops,
				stream);
		if (!stream->notif_iter) {
			goto error;
		}
	}
	stream->buf = g_new0(uint8_t, session->lttng_live->max_query_size);
	stream->buflen = session->lttng_live->max_query_size;

	ret = lttng_live_add_port(lttng_live, stream);
	BT_ASSERT(!ret);

	bt_list_add(&stream->node, &trace->streams);

	goto end;
error:
	/* Do not touch "borrowed" file. */
	lttng_live_stream_iterator_destroy(stream);
	stream = NULL;
end:
	return stream;
}

BT_HIDDEN
void lttng_live_stream_iterator_destroy(struct lttng_live_stream_iterator *stream)
{
	struct lttng_live_component *lttng_live;
	int ret;

	if (!stream) {
		return;
	}

	lttng_live = stream->trace->session->lttng_live;
	ret = lttng_live_remove_port(lttng_live, stream->port);
	BT_ASSERT(!ret);

	if (stream->stream) {
		BT_OBJECT_PUT_REF_AND_RESET(stream->stream);
	}

	if (stream->notif_iter) {
		bt_notif_iter_destroy(stream->notif_iter);
	}
	g_free(stream->buf);
	BT_OBJECT_PUT_REF_AND_RESET(stream->packet_end_notif_queue);
	bt_list_del(&stream->node);
	/*
	 * Ensure we poke the trace metadata in the future, which is
	 * required to release the metadata reference on the trace.
	 */
	stream->trace->new_metadata_needed = true;
	lttng_live_unref_trace(stream->trace);
	g_free(stream);
}
Commit	Line	Data
7cdc2bab MD	1	/*
	2	* Copyright 2016 - Philippe Proulx <pproulx@efficios.com>
	3	* Copyright 2016 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
	4	* Copyright 2010-2011 - EfficiOS Inc. and Linux Foundation
	5	*
	6	* Permission is hereby granted, free of charge, to any person obtaining a copy
	7	* of this software and associated documentation files (the "Software"), to deal
	8	* in the Software without restriction, including without limitation the rights
	9	* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
	10	* copies of the Software, and to permit persons to whom the Software is
	11	* furnished to do so, subject to the following conditions:
	12	*
	13	* The above copyright notice and this permission notice shall be included in
	14	* all copies or substantial portions of the Software.
	15	*
	16	* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
	17	* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
	18	* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
	19	* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
	20	* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
	21	* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
	22	* SOFTWARE.
	23	*/
	24
020bc26f PP	25	#define BT_LOG_TAG "PLUGIN-CTF-LTTNG-LIVE-SRC-DS"
	26	#include "logging.h"
	27
7cdc2bab MD	28	#include <stdio.h>
	29	#include <stdint.h>
	30	#include <stdlib.h>
	31	#include <stdbool.h>
	32	#include <glib.h>
	33	#include <inttypes.h>
8f76831a	34	#include <babeltrace/compat/mman-internal.h>
9d408fca	35	#include <babeltrace/babeltrace.h>
7cdc2bab	36	#include "../common/notif-iter/notif-iter.h"
f6ccaed9	37	#include <babeltrace/assert-internal.h>
7cdc2bab	38
087bc060	39	#include "data-stream.h"
7cdc2bab MD	40
7cdc2bab MD	41	static
50842bdc	42	enum bt_notif_iter_medium_status medop_request_bytes(
7cdc2bab MD	43	size_t request_sz, uint8_t **buffer_addr,
	44	size_t buffer_sz, void data)
	45	{
50842bdc PP	46	enum bt_notif_iter_medium_status status =
50842bdc PP	47	BT_NOTIF_ITER_MEDIUM_STATUS_OK;
7cdc2bab MD	48	struct lttng_live_stream_iterator *stream = data;
	49	struct lttng_live_trace *trace = stream->trace;
	50	struct lttng_live_session *session = trace->session;
	51	struct lttng_live_component *lttng_live = session->lttng_live;
	52	uint64_t recv_len = 0;
	53	uint64_t len_left;
	54	uint64_t read_len;
	55	//int i;
	56
	57	len_left = stream->base_offset + stream->len - stream->offset;
	58	if (!len_left) {
	59	stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
50842bdc	60	status = BT_NOTIF_ITER_MEDIUM_STATUS_AGAIN;
7cdc2bab MD	61	return status;
	62	}
	63	read_len = MIN(request_sz, stream->buflen);
	64	read_len = MIN(read_len, len_left);
	65	status = lttng_live_get_stream_bytes(lttng_live,
	66	stream, stream->buf, stream->offset,
	67	read_len, &recv_len);
7cdc2bab MD	68	*buffer_addr = stream->buf;
	69	*buffer_sz = recv_len;
	70	stream->offset += recv_len;
	71	return status;
	72	}
	73
	74	static
40f4ba76 PP	75	const struct bt_stream *medop_get_stream(
40f4ba76 PP	76	const struct bt_stream_class *stream_class,
b92735af	77	uint64_t stream_id, void *data)
7cdc2bab MD	78	{
7cdc2bab MD	79	struct lttng_live_stream_iterator *lttng_live_stream = data;
7cdc2bab MD	80
7cdc2bab MD	81	if (!lttng_live_stream->stream) {
b92735af	82	int64_t stream_class_id =
50842bdc	83	bt_stream_class_get_id(stream_class);
b92735af PP	84
	85	BT_LOGD("Creating stream %s (ID: %" PRIu64 ") out of stream class %" PRId64,
	86	lttng_live_stream->name, stream_id, stream_class_id);
	87
	88	if (stream_id == -1ULL) {
	89	/* No stream ID */
50842bdc	90	lttng_live_stream->stream = bt_stream_create(
b92735af PP	91	stream_class, lttng_live_stream->name);
	92	} else {
	93	lttng_live_stream->stream =
50842bdc	94	bt_stream_create_with_id(stream_class,
b92735af PP	95	lttng_live_stream->name, stream_id);
b92735af PP	96	}
7cdc2bab	97
7cdc2bab	98	if (!lttng_live_stream->stream) {
b92735af PP	99	BT_LOGE("Cannot create stream %s (stream class %" PRId64 ", stream ID %" PRIu64 ")",
	100	lttng_live_stream->name,
	101	stream_class_id, stream_id);
7cdc2bab MD	102	}
	103	}
	104
	105	return lttng_live_stream->stream;
	106	}
	107
50842bdc	108	static struct bt_notif_iter_medium_ops medops = {
7cdc2bab MD	109	.request_bytes = medop_request_bytes,
	110	.get_stream = medop_get_stream,
	111	};
	112
	113	BT_HIDDEN
50842bdc	114	enum bt_lttng_live_iterator_status lttng_live_lazy_notif_init(
7cdc2bab MD	115	struct lttng_live_session *session)
	116	{
	117	struct lttng_live_component *lttng_live = session->lttng_live;
	118	struct lttng_live_trace *trace;
	119
	120	if (!session->lazy_stream_notif_init) {
50842bdc	121	return BT_LTTNG_LIVE_ITERATOR_STATUS_OK;
7cdc2bab MD	122	}
	123
	124	bt_list_for_each_entry(trace, &session->traces, node) {
	125	struct lttng_live_stream_iterator *stream;
	126
	127	bt_list_for_each_entry(stream, &trace->streams, node) {
	128	if (stream->notif_iter) {
	129	continue;
	130	}
50842bdc	131	stream->notif_iter = bt_notif_iter_create(trace->trace,
7cdc2bab	132	lttng_live->max_query_size, medops,
55314f2a	133	stream);
7cdc2bab MD	134	if (!stream->notif_iter) {
	135	goto error;
	136	}
	137	}
	138	}
	139
	140	session->lazy_stream_notif_init = false;
	141
50842bdc	142	return BT_LTTNG_LIVE_ITERATOR_STATUS_OK;
7cdc2bab MD	143
7cdc2bab MD	144	error:
50842bdc	145	return BT_LTTNG_LIVE_ITERATOR_STATUS_ERROR;
7cdc2bab MD	146	}
	147
	148	BT_HIDDEN
	149	struct lttng_live_stream_iterator *lttng_live_stream_iterator_create(
	150	struct lttng_live_session *session,
	151	uint64_t ctf_trace_id,
	152	uint64_t stream_id)
	153	{
	154	struct lttng_live_component *lttng_live = session->lttng_live;
	155	struct lttng_live_stream_iterator *stream =
	156	g_new0(struct lttng_live_stream_iterator, 1);
	157	struct lttng_live_trace *trace;
	158	int ret;
	159
	160	trace = lttng_live_ref_trace(session, ctf_trace_id);
	161	if (!trace) {
	162	goto error;
	163	}
	164
	165	stream->p.type = LIVE_STREAM_TYPE_STREAM;
	166	stream->trace = trace;
	167	stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
	168	stream->viewer_stream_id = stream_id;
	169	stream->ctf_stream_class_id = -1ULL;
	170	stream->last_returned_inactivity_timestamp = INT64_MIN;
	171
	172	if (trace->trace) {
50842bdc	173	stream->notif_iter = bt_notif_iter_create(trace->trace,
7cdc2bab	174	lttng_live->max_query_size, medops,
55314f2a	175	stream);
7cdc2bab MD	176	if (!stream->notif_iter) {
	177	goto error;
	178	}
	179	}
	180	stream->buf = g_new0(uint8_t, session->lttng_live->max_query_size);
	181	stream->buflen = session->lttng_live->max_query_size;
	182
	183	ret = lttng_live_add_port(lttng_live, stream);
f6ccaed9	184	BT_ASSERT(!ret);
7cdc2bab MD	185
	186	bt_list_add(&stream->node, &trace->streams);
	187
	188	goto end;
	189	error:
	190	/* Do not touch "borrowed" file. */
	191	lttng_live_stream_iterator_destroy(stream);
	192	stream = NULL;
	193	end:
	194	return stream;
	195	}
	196
	197	BT_HIDDEN
	198	void lttng_live_stream_iterator_destroy(struct lttng_live_stream_iterator *stream)
	199	{
	200	struct lttng_live_component *lttng_live;
	201	int ret;
	202
	203	if (!stream) {
	204	return;
	205	}
	206
	207	lttng_live = stream->trace->session->lttng_live;
	208	ret = lttng_live_remove_port(lttng_live, stream->port);
f6ccaed9	209	BT_ASSERT(!ret);
7cdc2bab MD	210
7cdc2bab MD	211	if (stream->stream) {
65300d60	212	BT_OBJECT_PUT_REF_AND_RESET(stream->stream);
7cdc2bab MD	213	}
	214
	215	if (stream->notif_iter) {
50842bdc	216	bt_notif_iter_destroy(stream->notif_iter);
7cdc2bab MD	217	}
7cdc2bab MD	218	g_free(stream->buf);
65300d60	219	BT_OBJECT_PUT_REF_AND_RESET(stream->packet_end_notif_queue);
7cdc2bab MD	220	bt_list_del(&stream->node);
	221	/*
	222	* Ensure we poke the trace metadata in the future, which is
	223	* required to release the metadata reference on the trace.
	224	*/
	225	stream->trace->new_metadata_needed = true;
	226	lttng_live_unref_trace(stream->trace);
	227	g_free(stream);
	228	}