[babeltrace.git] / formats / ctf / ir / stream.c

/*
 * stream.c
 *
 * Babeltrace CTF IR - Stream
 *
 * Copyright 2013, 2014 Jérémie Galarneau <jeremie.galarneau@efficios.com>
 *
 * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
 *
 * 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.
 */

#include <babeltrace/ctf-ir/clock.h>
#include <babeltrace/ctf-ir/clock-internal.h>
#include <babeltrace/ctf-writer/event.h>
#include <babeltrace/ctf-ir/event-internal.h>
#include <babeltrace/ctf-ir/event-types-internal.h>
#include <babeltrace/ctf-ir/event-fields-internal.h>
#include <babeltrace/ctf-ir/stream.h>
#include <babeltrace/ctf-ir/stream-internal.h>
#include <babeltrace/ctf-ir/stream-class-internal.h>
#include <babeltrace/ctf-writer/functor-internal.h>
#include <babeltrace/compiler.h>
#include <babeltrace/align.h>

static
void bt_ctf_stream_destroy(struct bt_ctf_ref *ref);
static
int set_structure_field_integer(struct bt_ctf_field *, char *, uint64_t);

BT_HIDDEN
struct bt_ctf_stream *bt_ctf_stream_create(
		struct bt_ctf_stream_class *stream_class)
{
	int ret;
	struct bt_ctf_stream *stream = NULL;

	if (!stream_class) {
		goto end;
	}

	stream = g_new0(struct bt_ctf_stream, 1);
	if (!stream) {
		goto end;
	}

	bt_ctf_ref_init(&stream->ref_count);
	stream->packet_context = bt_ctf_field_create(
		stream_class->packet_context_type);
	if (!stream->packet_context) {
		goto error_destroy;
	}

	/*
	 * A stream class may not have a stream event context defined
	 * in which case this stream will never have a stream_event_context
	 * member since, after a stream's creation, the parent stream class
	 * is "frozen" (immutable).
	 */
	if (stream_class->event_context_type) {
		stream->event_context = bt_ctf_field_create(
			stream_class->event_context_type);
		if (!stream->packet_context) {
			goto error_destroy;
		}
	}

	/* Initialize events_discarded */
	ret = set_structure_field_integer(stream->packet_context,
		"events_discarded", 0);
	if (ret) {
		goto error_destroy;
	}

	stream->pos.fd = -1;
	stream->id = stream_class->next_stream_id++;
	stream->stream_class = stream_class;
	bt_ctf_stream_class_get(stream_class);
	bt_ctf_stream_class_freeze(stream_class);
	stream->events = g_ptr_array_new_with_free_func(
		(GDestroyNotify)bt_ctf_event_put);
end:
	return stream;
error_destroy:
	bt_ctf_stream_destroy(&stream->ref_count);
	return NULL;
}

BT_HIDDEN
int bt_ctf_stream_set_flush_callback(struct bt_ctf_stream *stream,
	flush_func callback, void *data)
{
	int ret = stream ? 0 : -1;

	if (!stream) {
		goto end;
	}

	stream->flush.func = callback;
	stream->flush.data = data;
end:
	return ret;
}

BT_HIDDEN
int bt_ctf_stream_set_fd(struct bt_ctf_stream *stream, int fd)
{
	int ret = 0;

	if (stream->pos.fd != -1) {
		ret = -1;
		goto end;
	}

	ctf_init_pos(&stream->pos, NULL, fd, O_RDWR);
	stream->pos.fd = fd;
end:
	return ret;
}

int bt_ctf_stream_get_discarded_events_count(
		struct bt_ctf_stream *stream, uint64_t *count)
{
	int64_t ret = 0;
	int field_signed;
	struct bt_ctf_field *events_discarded_field = NULL;
	struct bt_ctf_field_type *events_discarded_field_type = NULL;

	if (!stream || !count || !stream->packet_context) {
		ret = -1;
		goto end;
	}

	events_discarded_field = bt_ctf_field_structure_get_field(
		stream->packet_context, "events_discarded");
	if (!events_discarded_field) {
		ret = -1;
		goto end;
	}

	events_discarded_field_type = bt_ctf_field_get_type(
		events_discarded_field);
	if (!events_discarded_field_type) {
		ret = -1;
		goto end;
	}

	field_signed = bt_ctf_field_type_integer_get_signed(
		events_discarded_field_type);
	if (field_signed < 0) {
		ret = field_signed;
		goto end;
	}

	if (field_signed) {
		int64_t signed_count;

		ret = bt_ctf_field_signed_integer_get_value(
			events_discarded_field, &signed_count);
		if (ret) {
			goto end;
		}
		if (signed_count < 0) {
			/* Invalid value */
			ret = -1;
			goto end;
		}
		*count = (uint64_t) signed_count;
	} else {
		ret = bt_ctf_field_unsigned_integer_get_value(
			events_discarded_field, count);
		if (ret) {
			goto end;
		}
	}
end:
	if (events_discarded_field) {
		bt_ctf_field_put(events_discarded_field);
	}
	if (events_discarded_field_type) {
		bt_ctf_field_type_put(events_discarded_field_type);
	}
	return ret;
}

void bt_ctf_stream_append_discarded_events(struct bt_ctf_stream *stream,
		uint64_t event_count)
{
	int ret;
	int field_signed;
	uint64_t previous_count;
	uint64_t new_count;
	struct bt_ctf_field *events_discarded_field = NULL;
	struct bt_ctf_field_type *events_discarded_field_type = NULL;

	if (!stream || !stream->packet_context) {
		goto end;
	}

	ret = bt_ctf_stream_get_discarded_events_count(stream,
		&previous_count);
	if (ret) {
		goto end;
	}

	events_discarded_field = bt_ctf_field_structure_get_field(
		stream->packet_context, "events_discarded");
	if (!events_discarded_field) {
		goto end;
	}

	events_discarded_field_type = bt_ctf_field_get_type(
		events_discarded_field);
	if (!events_discarded_field_type) {
		goto end;
	}

	field_signed = bt_ctf_field_type_integer_get_signed(
		events_discarded_field_type);
	if (field_signed < 0) {
		goto end;
	}

	new_count = previous_count + event_count;
	if (field_signed) {
		ret = bt_ctf_field_signed_integer_set_value(
			events_discarded_field, (int64_t) new_count);
		if (ret) {
			goto end;
		}
	} else {
		ret = bt_ctf_field_unsigned_integer_set_value(
			events_discarded_field, new_count);
		if (ret) {
			goto end;
		}
	}

end:
	if (events_discarded_field) {
		bt_ctf_field_put(events_discarded_field);
	}
	if (events_discarded_field_type) {
		bt_ctf_field_type_put(events_discarded_field_type);
	}
}

int bt_ctf_stream_append_event(struct bt_ctf_stream *stream,
		struct bt_ctf_event *event)
{
	int ret = 0;
	uint64_t timestamp;
	struct bt_ctf_field *event_context_copy = NULL;

	if (!stream || !event) {
		ret = -1;
		goto end;
	}

	ret = bt_ctf_event_validate(event);
	if (ret) {
		goto end;
	}

	timestamp = bt_ctf_clock_get_time(stream->stream_class->clock);
	ret = bt_ctf_event_set_timestamp(event, timestamp);
	if (ret) {
		goto end;
	}

	bt_ctf_event_get(event);
	g_ptr_array_add(stream->events, event);
end:
	return ret;
}

struct bt_ctf_field *bt_ctf_stream_get_packet_context(
		struct bt_ctf_stream *stream)
{
	struct bt_ctf_field *packet_context = NULL;

	if (!stream) {
		goto end;
	}

	packet_context = stream->packet_context;
end:
	if (packet_context) {
		bt_ctf_field_get(packet_context);
	}
	return packet_context;
}

int bt_ctf_stream_set_packet_context(struct bt_ctf_stream *stream,
		struct bt_ctf_field *field)
{
	int ret = 0;
	struct bt_ctf_field_type *field_type;

	if (!stream || !field) {
		ret = -1;
		goto end;
	}

	field_type = bt_ctf_field_get_type(field);
	if (field_type != stream->stream_class->packet_context_type) {
		ret = -1;
		goto end;
	}

	bt_ctf_field_type_put(field_type);
	bt_ctf_field_get(field);
	bt_ctf_field_put(stream->packet_context);
	stream->packet_context = field;
end:
	return ret;
}

int bt_ctf_stream_flush(struct bt_ctf_stream *stream)
{
	int ret = 0;
	size_t i;
	uint64_t timestamp_begin, timestamp_end, events_discarded;
	struct bt_ctf_stream_class *stream_class;
	struct bt_ctf_field *integer = NULL;
	struct ctf_stream_pos packet_context_pos;

	if (!stream || stream->pos.fd < 0) {
		/*
		 * Stream does not have an associated fd. It is,
		 * therefore, not a stream being used to write events.
		 */
		ret = -1;
		goto end;
	}

	if (!stream->events->len) {
		goto end;
	}

	if (stream->flush.func) {
		stream->flush.func(stream, stream->flush.data);
	}

	stream_class = stream->stream_class;
	timestamp_begin = ((struct bt_ctf_event *) g_ptr_array_index(
		stream->events, 0))->timestamp;
	timestamp_end = ((struct bt_ctf_event *) g_ptr_array_index(
		stream->events, stream->events->len - 1))->timestamp;

	/* Set the default context attributes if present and unset. */
	ret = set_structure_field_integer(stream->packet_context,
		"timestamp_begin", timestamp_begin);
	if (ret) {
		goto end;
	}

	ret = set_structure_field_integer(stream->packet_context,
		"timestamp_end", timestamp_end);
	if (ret) {
		goto end;
	}

	ret = set_structure_field_integer(stream->packet_context,
		"content_size", UINT64_MAX);
	if (ret) {
		goto end;
	}

	ret = set_structure_field_integer(stream->packet_context,
		"packet_size", UINT64_MAX);
	if (ret) {
		goto end;
	}

	/* Write packet context */
	memcpy(&packet_context_pos, &stream->pos,
	       sizeof(struct ctf_stream_pos));
	ret = bt_ctf_field_serialize(stream->packet_context,
		&stream->pos);
	if (ret) {
		goto end;
	}

	ret = bt_ctf_stream_get_discarded_events_count(stream,
		&events_discarded);
	if (ret) {
		goto end;
	}

	/* Unset the packet context's fields. */
	ret = bt_ctf_field_reset(stream->packet_context);
	if (ret) {
		goto end;
	}

	/* Set the previous number of discarded events. */
	ret = set_structure_field_integer(stream->packet_context,
		"events_discarded", events_discarded);
	if (ret) {
		goto end;
	}

	for (i = 0; i < stream->events->len; i++) {
		struct bt_ctf_event *event = g_ptr_array_index(
			stream->events, i);
		uint32_t event_id = bt_ctf_event_class_get_id(
			event->event_class);
		uint64_t timestamp = bt_ctf_event_get_timestamp(event);

		ret = bt_ctf_field_reset(stream_class->event_header);
		if (ret) {
			goto end;
		}

		ret = set_structure_field_integer(stream_class->event_header,
			"id", event_id);
		if (ret) {
			goto end;
		}
		ret = set_structure_field_integer(stream_class->event_header,
			"timestamp", timestamp);
		if (ret) {
			goto end;
		}

		/* Write event header */
		ret = bt_ctf_field_serialize(stream_class->event_header,
			&stream->pos);
		if (ret) {
			goto end;
		}

		/* Write event content */
		ret = bt_ctf_event_serialize(event, &stream->pos);
		if (ret) {
			goto end;
		}
	}

	/*
	 * Update the packet total size and content size and overwrite the
	 * packet context.
	 * Copy base_mma as the packet may have been remapped (e.g. when a
	 * packet is resized).
	 */
	packet_context_pos.base_mma = stream->pos.base_mma;
	ret = set_structure_field_integer(stream->packet_context,
		"content_size", stream->pos.offset);
	if (ret) {
		goto end;
	}

	ret = set_structure_field_integer(stream->packet_context,
		"packet_size", stream->pos.packet_size);
	if (ret) {
		goto end;
	}

	ret = bt_ctf_field_serialize(stream->packet_context,
		&packet_context_pos);
	if (ret) {
		goto end;
	}

	g_ptr_array_set_size(stream->events, 0);
	stream->flushed_packet_count++;
end:
	bt_ctf_field_put(integer);
	return ret;
}

void bt_ctf_stream_get(struct bt_ctf_stream *stream)
{
	if (!stream) {
		return;
	}

	bt_ctf_ref_get(&stream->ref_count);
}

void bt_ctf_stream_put(struct bt_ctf_stream *stream)
{
	if (!stream) {
		return;
	}

	bt_ctf_ref_put(&stream->ref_count, bt_ctf_stream_destroy);
}

static
void bt_ctf_stream_destroy(struct bt_ctf_ref *ref)
{
	struct bt_ctf_stream *stream;

	if (!ref) {
		return;
	}

	stream = container_of(ref, struct bt_ctf_stream, ref_count);
	ctf_fini_pos(&stream->pos);
	if (close(stream->pos.fd)) {
		perror("close");
	}

	if (stream->stream_class) {
		bt_ctf_stream_class_put(stream->stream_class);
	}
	if (stream->events) {
		g_ptr_array_free(stream->events, TRUE);
	}
	if (stream->packet_context) {
		bt_ctf_field_put(stream->packet_context);
	}
	g_free(stream);
}

static
int set_structure_field_integer(struct bt_ctf_field *structure, char *name,
		uint64_t value)
{
	int ret = 0;
	struct bt_ctf_field *integer =
		bt_ctf_field_structure_get_field(structure, name);

	if (!structure || !name) {
		ret = -1;
		goto end;
	}

	if (!integer) {
		/* Field not found, not an error. */
		goto end;
	}

	/* Make sure the payload has not already been set. */
	if (!bt_ctf_field_validate(integer)) {
		/* Payload already set, not an error */
		goto end;
	}

	ret = bt_ctf_field_unsigned_integer_set_value(integer, value);
end:
	bt_ctf_field_put(integer);
	return ret;
}
Commit	Line	Data
273b65be JG	1	/*
	2	* stream.c
	3	*
d2dc44b6	4	* Babeltrace CTF IR - Stream
273b65be	5	*
de9dd397	6	* Copyright 2013, 2014 Jérémie Galarneau <jeremie.galarneau@efficios.com>
273b65be JG	7	*
	8	* Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
	9	*
	10	* Permission is hereby granted, free of charge, to any person obtaining a copy
	11	* of this software and associated documentation files (the "Software"), to deal
	12	* in the Software without restriction, including without limitation the rights
	13	* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
	14	* copies of the Software, and to permit persons to whom the Software is
	15	* furnished to do so, subject to the following conditions:
	16	*
	17	* The above copyright notice and this permission notice shall be included in
	18	* all copies or substantial portions of the Software.
	19	*
	20	* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
	21	* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
	22	* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
	23	* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
	24	* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
	25	* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
	26	* SOFTWARE.
	27	*/
	28
3f043b05	29	#include <babeltrace/ctf-ir/clock.h>
adc315b8	30	#include <babeltrace/ctf-ir/clock-internal.h>
273b65be	31	#include <babeltrace/ctf-writer/event.h>
adc315b8 JG	32	#include <babeltrace/ctf-ir/event-internal.h>
	33	#include <babeltrace/ctf-ir/event-types-internal.h>
	34	#include <babeltrace/ctf-ir/event-fields-internal.h>
3f043b05 JG	35	#include <babeltrace/ctf-ir/stream.h>
3f043b05 JG	36	#include <babeltrace/ctf-ir/stream-internal.h>
adc315b8	37	#include <babeltrace/ctf-ir/stream-class-internal.h>
273b65be JG	38	#include <babeltrace/ctf-writer/functor-internal.h>
	39	#include <babeltrace/compiler.h>
	40	#include <babeltrace/align.h>
	41
	42	static
	43	void bt_ctf_stream_destroy(struct bt_ctf_ref *ref);
	44	static
273b65be JG	45	int set_structure_field_integer(struct bt_ctf_field , char , uint64_t);
273b65be JG	46
273b65be JG	47	BT_HIDDEN
	48	struct bt_ctf_stream *bt_ctf_stream_create(
	49	struct bt_ctf_stream_class *stream_class)
	50	{
12c8a1a3	51	int ret;
273b65be JG	52	struct bt_ctf_stream *stream = NULL;
	53
	54	if (!stream_class) {
	55	goto end;
	56	}
	57
	58	stream = g_new0(struct bt_ctf_stream, 1);
	59	if (!stream) {
	60	goto end;
	61	}
	62
	63	bt_ctf_ref_init(&stream->ref_count);
12c8a1a3 JG	64	stream->packet_context = bt_ctf_field_create(
	65	stream_class->packet_context_type);
	66	if (!stream->packet_context) {
	67	goto error_destroy;
	68	}
	69
af181248 JG	70	/*
	71	* A stream class may not have a stream event context defined
	72	* in which case this stream will never have a stream_event_context
	73	* member since, after a stream's creation, the parent stream class
	74	* is "frozen" (immutable).
	75	*/
	76	if (stream_class->event_context_type) {
	77	stream->event_context = bt_ctf_field_create(
	78	stream_class->event_context_type);
	79	if (!stream->packet_context) {
	80	goto error_destroy;
	81	}
	82	}
	83
e19cc57d	84	/* Initialize events_discarded */
12c8a1a3 JG	85	ret = set_structure_field_integer(stream->packet_context,
	86	"events_discarded", 0);
	87	if (ret) {
	88	goto error_destroy;
	89	}
	90
273b65be JG	91	stream->pos.fd = -1;
	92	stream->id = stream_class->next_stream_id++;
	93	stream->stream_class = stream_class;
	94	bt_ctf_stream_class_get(stream_class);
	95	bt_ctf_stream_class_freeze(stream_class);
	96	stream->events = g_ptr_array_new_with_free_func(
	97	(GDestroyNotify)bt_ctf_event_put);
	98	end:
	99	return stream;
12c8a1a3 JG	100	error_destroy:
	101	bt_ctf_stream_destroy(&stream->ref_count);
	102	return NULL;
273b65be JG	103	}
	104
	105	BT_HIDDEN
	106	int bt_ctf_stream_set_flush_callback(struct bt_ctf_stream *stream,
	107	flush_func callback, void *data)
	108	{
	109	int ret = stream ? 0 : -1;
	110
	111	if (!stream) {
	112	goto end;
	113	}
	114
	115	stream->flush.func = callback;
	116	stream->flush.data = data;
	117	end:
	118	return ret;
	119	}
	120
	121	BT_HIDDEN
	122	int bt_ctf_stream_set_fd(struct bt_ctf_stream *stream, int fd)
	123	{
	124	int ret = 0;
	125
	126	if (stream->pos.fd != -1) {
	127	ret = -1;
	128	goto end;
	129	}
	130
	131	ctf_init_pos(&stream->pos, NULL, fd, O_RDWR);
	132	stream->pos.fd = fd;
	133	end:
	134	return ret;
	135	}
	136
a78a2e25 JG	137	int bt_ctf_stream_get_discarded_events_count(
	138	struct bt_ctf_stream stream, uint64_t count)
	139	{
	140	int64_t ret = 0;
12c8a1a3 JG	141	int field_signed;
	142	struct bt_ctf_field *events_discarded_field = NULL;
	143	struct bt_ctf_field_type *events_discarded_field_type = NULL;
a78a2e25	144
12c8a1a3	145	if (!stream \|\| !count \|\| !stream->packet_context) {
a78a2e25 JG	146	ret = -1;
	147	goto end;
	148	}
	149
12c8a1a3 JG	150	events_discarded_field = bt_ctf_field_structure_get_field(
	151	stream->packet_context, "events_discarded");
	152	if (!events_discarded_field) {
	153	ret = -1;
	154	goto end;
	155	}
	156
	157	events_discarded_field_type = bt_ctf_field_get_type(
	158	events_discarded_field);
	159	if (!events_discarded_field_type) {
	160	ret = -1;
	161	goto end;
	162	}
	163
	164	field_signed = bt_ctf_field_type_integer_get_signed(
	165	events_discarded_field_type);
	166	if (field_signed < 0) {
	167	ret = field_signed;
	168	goto end;
	169	}
	170
	171	if (field_signed) {
	172	int64_t signed_count;
	173
	174	ret = bt_ctf_field_signed_integer_get_value(
	175	events_discarded_field, &signed_count);
	176	if (ret) {
	177	goto end;
	178	}
	179	if (signed_count < 0) {
	180	/* Invalid value */
	181	ret = -1;
	182	goto end;
	183	}
	184	*count = (uint64_t) signed_count;
	185	} else {
	186	ret = bt_ctf_field_unsigned_integer_get_value(
	187	events_discarded_field, count);
	188	if (ret) {
	189	goto end;
	190	}
	191	}
a78a2e25	192	end:
12c8a1a3 JG	193	if (events_discarded_field) {
	194	bt_ctf_field_put(events_discarded_field);
	195	}
	196	if (events_discarded_field_type) {
	197	bt_ctf_field_type_put(events_discarded_field_type);
	198	}
a78a2e25 JG	199	return ret;
	200	}
	201
273b65be JG	202	void bt_ctf_stream_append_discarded_events(struct bt_ctf_stream *stream,
	203	uint64_t event_count)
	204	{
12c8a1a3 JG	205	int ret;
	206	int field_signed;
	207	uint64_t previous_count;
	208	uint64_t new_count;
	209	struct bt_ctf_field *events_discarded_field = NULL;
	210	struct bt_ctf_field_type *events_discarded_field_type = NULL;
	211
	212	if (!stream \|\| !stream->packet_context) {
	213	goto end;
	214	}
	215
	216	ret = bt_ctf_stream_get_discarded_events_count(stream,
	217	&previous_count);
	218	if (ret) {
	219	goto end;
	220	}
	221
	222	events_discarded_field = bt_ctf_field_structure_get_field(
	223	stream->packet_context, "events_discarded");
	224	if (!events_discarded_field) {
	225	goto end;
	226	}
	227
	228	events_discarded_field_type = bt_ctf_field_get_type(
	229	events_discarded_field);
	230	if (!events_discarded_field_type) {
	231	goto end;
	232	}
	233
	234	field_signed = bt_ctf_field_type_integer_get_signed(
	235	events_discarded_field_type);
	236	if (field_signed < 0) {
	237	goto end;
273b65be JG	238	}
273b65be JG	239
12c8a1a3 JG	240	new_count = previous_count + event_count;
	241	if (field_signed) {
	242	ret = bt_ctf_field_signed_integer_set_value(
	243	events_discarded_field, (int64_t) new_count);
	244	if (ret) {
	245	goto end;
	246	}
	247	} else {
	248	ret = bt_ctf_field_unsigned_integer_set_value(
	249	events_discarded_field, new_count);
	250	if (ret) {
	251	goto end;
	252	}
	253	}
	254
	255	end:
	256	if (events_discarded_field) {
	257	bt_ctf_field_put(events_discarded_field);
	258	}
	259	if (events_discarded_field_type) {
	260	bt_ctf_field_type_put(events_discarded_field_type);
	261	}
273b65be JG	262	}
	263
	264	int bt_ctf_stream_append_event(struct bt_ctf_stream *stream,
	265	struct bt_ctf_event *event)
	266	{
	267	int ret = 0;
	268	uint64_t timestamp;
af181248	269	struct bt_ctf_field *event_context_copy = NULL;
273b65be JG	270
	271	if (!stream \|\| !event) {
	272	ret = -1;
	273	goto end;
	274	}
	275
	276	ret = bt_ctf_event_validate(event);
	277	if (ret) {
	278	goto end;
	279	}
	280
	281	timestamp = bt_ctf_clock_get_time(stream->stream_class->clock);
	282	ret = bt_ctf_event_set_timestamp(event, timestamp);
	283	if (ret) {
	284	goto end;
	285	}
	286
	287	bt_ctf_event_get(event);
	288	g_ptr_array_add(stream->events, event);
	289	end:
	290	return ret;
	291	}
	292
12c8a1a3 JG	293	struct bt_ctf_field *bt_ctf_stream_get_packet_context(
	294	struct bt_ctf_stream *stream)
	295	{
	296	struct bt_ctf_field *packet_context = NULL;
	297
	298	if (!stream) {
	299	goto end;
	300	}
	301
	302	packet_context = stream->packet_context;
	303	end:
	304	if (packet_context) {
	305	bt_ctf_field_get(packet_context);
	306	}
	307	return packet_context;
	308	}
	309
	310	int bt_ctf_stream_set_packet_context(struct bt_ctf_stream *stream,
	311	struct bt_ctf_field *field)
	312	{
	313	int ret = 0;
	314	struct bt_ctf_field_type *field_type;
	315
	316	if (!stream \|\| !field) {
	317	ret = -1;
	318	goto end;
	319	}
	320
	321	field_type = bt_ctf_field_get_type(field);
	322	if (field_type != stream->stream_class->packet_context_type) {
	323	ret = -1;
	324	goto end;
	325	}
	326
	327	bt_ctf_field_type_put(field_type);
	328	bt_ctf_field_get(field);
	329	bt_ctf_field_put(stream->packet_context);
	330	stream->packet_context = field;
	331	end:
	332	return ret;
	333	}
	334
273b65be JG	335	int bt_ctf_stream_flush(struct bt_ctf_stream *stream)
	336	{
	337	int ret = 0;
	338	size_t i;
12c8a1a3	339	uint64_t timestamp_begin, timestamp_end, events_discarded;
9f56e450	340	struct bt_ctf_stream_class *stream_class;
273b65be JG	341	struct bt_ctf_field *integer = NULL;
	342	struct ctf_stream_pos packet_context_pos;
	343
bc37ae52 JG	344	if (!stream \|\| stream->pos.fd < 0) {
	345	/*
	346	* Stream does not have an associated fd. It is,
	347	* therefore, not a stream being used to write events.
	348	*/
273b65be JG	349	ret = -1;
	350	goto end;
	351	}
	352
	353	if (!stream->events->len) {
	354	goto end;
	355	}
	356
	357	if (stream->flush.func) {
	358	stream->flush.func(stream, stream->flush.data);
	359	}
	360
9f56e450	361	stream_class = stream->stream_class;
273b65be JG	362	timestamp_begin = ((struct bt_ctf_event *) g_ptr_array_index(
	363	stream->events, 0))->timestamp;
	364	timestamp_end = ((struct bt_ctf_event *) g_ptr_array_index(
	365	stream->events, stream->events->len - 1))->timestamp;
12c8a1a3 JG	366
	367	/* Set the default context attributes if present and unset. */
	368	ret = set_structure_field_integer(stream->packet_context,
273b65be JG	369	"timestamp_begin", timestamp_begin);
	370	if (ret) {
	371	goto end;
	372	}
	373
12c8a1a3	374	ret = set_structure_field_integer(stream->packet_context,
273b65be JG	375	"timestamp_end", timestamp_end);
	376	if (ret) {
	377	goto end;
	378	}
	379
12c8a1a3	380	ret = set_structure_field_integer(stream->packet_context,
273b65be JG	381	"content_size", UINT64_MAX);
	382	if (ret) {
	383	goto end;
	384	}
	385
12c8a1a3	386	ret = set_structure_field_integer(stream->packet_context,
273b65be JG	387	"packet_size", UINT64_MAX);
	388	if (ret) {
	389	goto end;
	390	}
	391
	392	/* Write packet context */
	393	memcpy(&packet_context_pos, &stream->pos,
	394	sizeof(struct ctf_stream_pos));
12c8a1a3	395	ret = bt_ctf_field_serialize(stream->packet_context,
273b65be JG	396	&stream->pos);
	397	if (ret) {
	398	goto end;
	399	}
	400
12c8a1a3 JG	401	ret = bt_ctf_stream_get_discarded_events_count(stream,
	402	&events_discarded);
	403	if (ret) {
	404	goto end;
	405	}
	406
	407	/* Unset the packet context's fields. */
	408	ret = bt_ctf_field_reset(stream->packet_context);
	409	if (ret) {
	410	goto end;
	411	}
	412
	413	/* Set the previous number of discarded events. */
	414	ret = set_structure_field_integer(stream->packet_context,
	415	"events_discarded", events_discarded);
	416	if (ret) {
	417	goto end;
	418	}
	419
273b65be JG	420	for (i = 0; i < stream->events->len; i++) {
	421	struct bt_ctf_event *event = g_ptr_array_index(
	422	stream->events, i);
	423	uint32_t event_id = bt_ctf_event_class_get_id(
	424	event->event_class);
	425	uint64_t timestamp = bt_ctf_event_get_timestamp(event);
	426
12c8a1a3 JG	427	ret = bt_ctf_field_reset(stream_class->event_header);
	428	if (ret) {
	429	goto end;
	430	}
	431
273b65be JG	432	ret = set_structure_field_integer(stream_class->event_header,
	433	"id", event_id);
	434	if (ret) {
	435	goto end;
	436	}
	437	ret = set_structure_field_integer(stream_class->event_header,
	438	"timestamp", timestamp);
	439	if (ret) {
	440	goto end;
	441	}
	442
	443	/* Write event header */
	444	ret = bt_ctf_field_serialize(stream_class->event_header,
	445	&stream->pos);
	446	if (ret) {
	447	goto end;
	448	}
	449
	450	/* Write event content */
	451	ret = bt_ctf_event_serialize(event, &stream->pos);
	452	if (ret) {
	453	goto end;
	454	}
	455	}
	456
	457	/*
	458	* Update the packet total size and content size and overwrite the
	459	* packet context.
3a092c05 JG	460	* Copy base_mma as the packet may have been remapped (e.g. when a
3a092c05 JG	461	* packet is resized).
273b65be	462	*/
3a092c05	463	packet_context_pos.base_mma = stream->pos.base_mma;
12c8a1a3	464	ret = set_structure_field_integer(stream->packet_context,
273b65be JG	465	"content_size", stream->pos.offset);
	466	if (ret) {
	467	goto end;
	468	}
	469
12c8a1a3	470	ret = set_structure_field_integer(stream->packet_context,
273b65be JG	471	"packet_size", stream->pos.packet_size);
	472	if (ret) {
	473	goto end;
	474	}
	475
12c8a1a3	476	ret = bt_ctf_field_serialize(stream->packet_context,
273b65be JG	477	&packet_context_pos);
	478	if (ret) {
	479	goto end;
	480	}
	481
	482	g_ptr_array_set_size(stream->events, 0);
	483	stream->flushed_packet_count++;
	484	end:
	485	bt_ctf_field_put(integer);
	486	return ret;
	487	}
	488
	489	void bt_ctf_stream_get(struct bt_ctf_stream *stream)
	490	{
	491	if (!stream) {
	492	return;
	493	}
	494
	495	bt_ctf_ref_get(&stream->ref_count);
	496	}
	497
	498	void bt_ctf_stream_put(struct bt_ctf_stream *stream)
	499	{
	500	if (!stream) {
	501	return;
	502	}
	503
	504	bt_ctf_ref_put(&stream->ref_count, bt_ctf_stream_destroy);
	505	}
	506
	507	static
	508	void bt_ctf_stream_destroy(struct bt_ctf_ref *ref)
	509	{
	510	struct bt_ctf_stream *stream;
	511
	512	if (!ref) {
	513	return;
	514	}
	515
	516	stream = container_of(ref, struct bt_ctf_stream, ref_count);
	517	ctf_fini_pos(&stream->pos);
9f56e450 JG	518	if (close(stream->pos.fd)) {
	519	perror("close");
	520	}
12c8a1a3 JG	521
	522	if (stream->stream_class) {
	523	bt_ctf_stream_class_put(stream->stream_class);
	524	}
	525	if (stream->events) {
	526	g_ptr_array_free(stream->events, TRUE);
	527	}
	528	if (stream->packet_context) {
	529	bt_ctf_field_put(stream->packet_context);
	530	}
273b65be JG	531	g_free(stream);
	532	}
	533
273b65be JG	534	static
	535	int set_structure_field_integer(struct bt_ctf_field structure, char name,
	536	uint64_t value)
	537	{
	538	int ret = 0;
273b65be JG	539	struct bt_ctf_field *integer =
273b65be JG	540	bt_ctf_field_structure_get_field(structure, name);
12c8a1a3 JG	541
12c8a1a3 JG	542	if (!structure \|\| !name) {
273b65be JG	543	ret = -1;
	544	goto end;
	545	}
	546
12c8a1a3 JG	547	if (!integer) {
	548	/* Field not found, not an error. */
	549	goto end;
	550	}
	551
	552	/* Make sure the payload has not already been set. */
	553	if (!bt_ctf_field_validate(integer)) {
	554	/* Payload already set, not an error */
	555	goto end;
	556	}
	557
273b65be JG	558	ret = bt_ctf_field_unsigned_integer_set_value(integer, value);
	559	end:
	560	bt_ctf_field_put(integer);
	561	return ret;
	562	}