X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Flib%2Fgraph%2Fiterator.c;h=760affc7ddcae8dfe501883f87598478f2955d5e;hb=5a3fec55322bf221441f86d5ffedeb049d08d200;hp=9bcbab55ee6b59627c710c6146566fb6ca593d3d;hpb=ecd7492f21a492b70569d5ecc1d3a808241b63f0;p=babeltrace.git

diff --git a/src/lib/graph/iterator.c b/src/lib/graph/iterator.c
index 9bcbab55..760affc7 100644
--- a/src/lib/graph/iterator.c
+++ b/src/lib/graph/iterator.c
@@ -18,6 +18,7 @@
 #include <babeltrace2/trace-ir/packet.h>
 #include "lib/trace-ir/packet.h"
 #include "lib/trace-ir/stream.h"
+#include "lib/trace-ir/stream-class.h"
 #include <babeltrace2/trace-ir/clock-class.h>
 #include <babeltrace2/trace-ir/stream-class.h>
 #include <babeltrace2/trace-ir/stream.h>
@@ -38,14 +39,12 @@
 
 #include "component-class.h"
 #include "component.h"
-#include "component-sink.h"
-#include "component-source.h"
 #include "connection.h"
 #include "graph.h"
+#include "iterator.h"
 #include "message-iterator-class.h"
 #include "message/discarded-items.h"
 #include "message/event.h"
-#include "message/iterator.h"
 #include "message/message.h"
 #include "message/message-iterator-inactivity.h"
 #include "message/stream.h"
@@ -66,6 +65,26 @@
 		(_iter)->state == BT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR, \
 		"Message iterator is in the wrong state: %!+i", (_iter))
 
+#ifdef BT_DEV_MODE
+struct per_stream_state
+{
+	bt_packet *cur_packet;
+
+	/* Bit mask of expected message types. */
+	guint expected_msg_types;
+};
+#endif
+
+static void
+clear_per_stream_state (struct bt_message_iterator *iterator)
+{
+#ifdef BT_DEV_MODE
+	g_hash_table_remove_all(iterator->per_stream_state);
+#else
+	BT_USE_EXPR(iterator);
+#endif
+}
+
 static inline
 void set_msg_iterator_state(struct bt_message_iterator *iterator,
 		enum bt_message_iterator_state state)
@@ -100,6 +119,12 @@ void bt_message_iterator_destroy(struct bt_object *obj)
 		"%!+i", iterator);
 	bt_message_iterator_try_finalize(iterator);
 
+	if (iterator->clock_expectation.type ==
+			CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID) {
+		BT_CLOCK_CLASS_PUT_REF_AND_RESET(
+			iterator->clock_expectation.clock_class);
+	}
+
 	if (iterator->connection) {
 		/*
 		 * Remove ourself from the originating connection so
@@ -135,6 +160,10 @@ void bt_message_iterator_destroy(struct bt_object *obj)
 		iterator->msgs = NULL;
 	}
 
+#ifdef BT_DEV_MODE
+	g_hash_table_destroy(iterator->per_stream_state);
+#endif
+
 	g_free(iterator);
 }
 
@@ -341,6 +370,16 @@ int create_self_component_input_port_message_iterator(
 
 	g_ptr_array_set_size(iterator->msgs, MSG_BATCH_SIZE);
 	iterator->last_ns_from_origin = INT64_MIN;
+
+#ifdef BT_DEV_MODE
+	/* The per-stream state is only used for dev assertions right now. */
+	iterator->per_stream_state = g_hash_table_new_full(
+		g_direct_hash,
+		g_direct_equal,
+		NULL,
+		g_free);
+#endif
+
 	iterator->auto_seek.msgs = g_queue_new();
 	if (!iterator->auto_seek.msgs) {
 		BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GQueue.");
@@ -668,6 +707,8 @@ bool clock_classes_are_compatible_one(struct bt_message_iterator *iterator,
 				bt_uuid_copy(iterator->clock_expectation.uuid, clock_class_uuid);
 			} else {
 				iterator->clock_expectation.type = CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID;
+				iterator->clock_expectation.clock_class = clock_class;
+				bt_clock_class_get_ref(iterator->clock_expectation.clock_class);
 			}
 			break;
 
@@ -736,26 +777,21 @@ bool clock_classes_are_compatible_one(struct bt_message_iterator *iterator,
 		case CLOCK_EXPECTATION_ORIGIN_OTHER_NO_UUID:
 			if (!clock_class) {
 				BT_ASSERT_COND_DEV_MSG(
-					"Expecting a clock class, got none.");
+					"Expecting clock class %![cc-]+K, got none.",
+					iterator->clock_expectation.clock_class);
 				result = false;
 				goto end;
 			}
 
-			if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
+			if (clock_class != iterator->clock_expectation.clock_class) {
 				BT_ASSERT_COND_DEV_MSG(
-					"Expecting a clock class without Unix epoch origin: %![cc-]+K",
+					"Expecting clock class %![cc-]+K, got %![cc-]+K.",
+					iterator->clock_expectation.clock_class,
 					clock_class);
 				result = false;
 				goto end;
 			}
 
-			if (clock_class_uuid) {
-				BT_ASSERT_COND_DEV_MSG(
-					"Expecting a clock class without UUID: %![cc-]+K",
-					clock_class);
-				result = false;
-				goto end;
-			}
 			break;
 		}
 	}
@@ -788,13 +824,350 @@ end:
 	return result;
 }
 
+#ifdef BT_DEV_MODE
+static
+const bt_stream *get_stream_from_msg(const struct bt_message *msg)
+{
+	struct bt_stream *stream;
+
+	switch (msg->type) {
+        case BT_MESSAGE_TYPE_STREAM_BEGINNING:
+        case BT_MESSAGE_TYPE_STREAM_END:
+	{
+		struct bt_message_stream *msg_stream =
+			(struct bt_message_stream *) msg;
+		stream = msg_stream->stream;
+		break;
+	}
+        case BT_MESSAGE_TYPE_EVENT:
+	{
+		struct bt_message_event *msg_event =
+			(struct bt_message_event *) msg;
+		stream = msg_event->event->stream;
+		break;
+	}
+        case BT_MESSAGE_TYPE_PACKET_BEGINNING:
+        case BT_MESSAGE_TYPE_PACKET_END:
+	{
+		struct bt_message_packet *msg_packet =
+			(struct bt_message_packet *) msg;
+		stream = msg_packet->packet->stream;
+		break;
+	}
+        case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
+        case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
+	{
+		struct bt_message_discarded_items *msg_discarded =
+			(struct bt_message_discarded_items *) msg;
+		stream = msg_discarded->stream;
+		break;
+	}
+        case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
+		stream = NULL;
+		break;
+	default:
+		bt_common_abort();
+        }
+
+	return stream;
+}
+
+static
+GString *message_types_to_string(guint msg_types)
+{
+	GString *str = g_string_new("");
+
+	for (int msg_type = 1; msg_type <= BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY;
+			msg_type <<= 1) {
+		if (msg_type & msg_types) {
+			if (str->len > 0) {
+				g_string_append_c(str, '|');
+			}
+
+			g_string_append(str,
+				bt_common_message_type_string(msg_type));
+		}
+	}
+
+	return str;
+}
+
+static
+void update_expected_msg_type(const struct bt_stream *stream,
+		struct per_stream_state *state,
+		const struct bt_message *msg)
+{
+	switch (msg->type) {
+        case BT_MESSAGE_TYPE_STREAM_BEGINNING:
+		state->expected_msg_types = BT_MESSAGE_TYPE_STREAM_END;
+
+		if (stream->class->supports_packets) {
+			state->expected_msg_types |=
+				BT_MESSAGE_TYPE_PACKET_BEGINNING;
+
+			if (stream->class->supports_discarded_packets) {
+				state->expected_msg_types |=
+					BT_MESSAGE_TYPE_DISCARDED_PACKETS;
+			}
+		} else {
+			state->expected_msg_types |= BT_MESSAGE_TYPE_EVENT;
+		}
+
+		if (stream->class->supports_discarded_events) {
+			state->expected_msg_types |=
+				BT_MESSAGE_TYPE_DISCARDED_EVENTS;
+		}
+
+		break;
+        case BT_MESSAGE_TYPE_STREAM_END:
+		state->expected_msg_types = 0;
+		break;
+        case BT_MESSAGE_TYPE_EVENT:
+	{
+		state->expected_msg_types = BT_MESSAGE_TYPE_EVENT;
+
+		if (stream->class->supports_packets) {
+			state->expected_msg_types |= BT_MESSAGE_TYPE_PACKET_END;
+		} else {
+			state->expected_msg_types |= BT_MESSAGE_TYPE_STREAM_END;
+		}
+
+		if (stream->class->supports_discarded_events) {
+			state->expected_msg_types |=
+				BT_MESSAGE_TYPE_DISCARDED_EVENTS;
+		}
+
+		break;
+	}
+        case BT_MESSAGE_TYPE_PACKET_BEGINNING:
+	{
+		state->expected_msg_types = BT_MESSAGE_TYPE_EVENT |
+			BT_MESSAGE_TYPE_PACKET_END;
+
+		if (stream->class->supports_discarded_events) {
+			state->expected_msg_types |=
+				BT_MESSAGE_TYPE_DISCARDED_EVENTS;
+		}
+
+		break;
+	}
+        case BT_MESSAGE_TYPE_PACKET_END:
+	{
+		state->expected_msg_types = BT_MESSAGE_TYPE_PACKET_BEGINNING |
+			BT_MESSAGE_TYPE_STREAM_END;
+
+		if (stream->class->supports_discarded_events) {
+			state->expected_msg_types |=
+				BT_MESSAGE_TYPE_DISCARDED_EVENTS;
+		}
+
+		if (stream->class->supports_discarded_packets) {
+			state->expected_msg_types |=
+				BT_MESSAGE_TYPE_DISCARDED_PACKETS;
+		}
+
+		break;
+	}
+        case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
+		state->expected_msg_types = BT_MESSAGE_TYPE_DISCARDED_EVENTS;
+
+		if (state->cur_packet) {
+			state->expected_msg_types |= BT_MESSAGE_TYPE_EVENT |
+				BT_MESSAGE_TYPE_PACKET_END;
+		} else {
+			state->expected_msg_types |= BT_MESSAGE_TYPE_STREAM_END;
+
+			if (stream->class->supports_packets) {
+				state->expected_msg_types |=
+					BT_MESSAGE_TYPE_PACKET_BEGINNING;
+
+				if (stream->class->supports_discarded_packets) {
+					state->expected_msg_types |=
+						BT_MESSAGE_TYPE_DISCARDED_PACKETS;
+				}
+			} else {
+				state->expected_msg_types |=
+					BT_MESSAGE_TYPE_EVENT;
+			}
+		}
+
+		break;
+        case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
+		state->expected_msg_types = BT_MESSAGE_TYPE_DISCARDED_PACKETS |
+			BT_MESSAGE_TYPE_PACKET_BEGINNING |
+			BT_MESSAGE_TYPE_STREAM_END;
+
+		if (stream->class->supports_discarded_events) {
+			state->expected_msg_types |=
+				BT_MESSAGE_TYPE_DISCARDED_EVENTS;
+		}
+		break;
+	default:
+		/*
+		 * Other message types are not associated to a stream, so we
+		 * should not get them here.
+		 */
+                bt_common_abort();
+        }
+}
+
+static
+struct per_stream_state *get_per_stream_state(
+		struct bt_message_iterator *iterator,
+		const struct bt_stream *stream)
+{
+	struct per_stream_state *state = g_hash_table_lookup(
+		iterator->per_stream_state, stream);
+
+	if (!state) {
+		state = g_new0(struct per_stream_state, 1);
+		state->expected_msg_types = BT_MESSAGE_TYPE_STREAM_BEGINNING;
+		g_hash_table_insert(iterator->per_stream_state,
+			(gpointer) stream, state);
+	}
+
+	return state;
+}
+#endif
+
+#define NEXT_METHOD_NAME	"bt_message_iterator_class_next_method"
+
+#ifdef BT_DEV_MODE
+static
+void assert_post_dev_expected_sequence(struct bt_message_iterator *iterator,
+		const struct bt_message *msg)
+{
+	const bt_stream *stream = get_stream_from_msg(msg);
+	struct per_stream_state *state;
+
+	if (!stream) {
+		goto end;
+	}
+
+	state = get_per_stream_state(iterator, stream);
+
+	/*
+	 * We don't free the return value of message_types_to_string(), but
+	 * that's because we know the program is going to abort anyway, and
+	 * we don't want to call it if the assertion holds.
+	 */
+	BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
+		"message-type-is-expected",
+		msg->type & state->expected_msg_types,
+		"Unexpected message type: %![stream-]s, %![iterator-]i, "
+		"%![message-]n, expected-msg-types=%s",
+		stream, iterator, msg,
+		message_types_to_string(state->expected_msg_types)->str);
+
+	update_expected_msg_type(stream, state, msg);
+
+end:
+	return;
+}
+
+static
+void assert_post_dev_expected_packet(struct bt_message_iterator *iterator,
+		const struct bt_message *msg)
+{
+	const bt_stream *stream = get_stream_from_msg(msg);
+	struct per_stream_state *state;
+	const bt_packet *actual_packet = NULL;
+	const bt_packet *expected_packet = NULL;
+
+	if (!stream) {
+		goto end;
+	}
+
+	state = get_per_stream_state(iterator, stream);
+
+	switch (msg->type) {
+	case BT_MESSAGE_TYPE_EVENT:
+	{
+		const struct bt_message_event *msg_event =
+			(const struct bt_message_event *) msg;
+
+		actual_packet = msg_event->event->packet;
+		expected_packet = state->cur_packet;
+		break;
+	}
+	case BT_MESSAGE_TYPE_PACKET_BEGINNING:
+	{
+		const struct bt_message_packet *msg_packet =
+			(const struct bt_message_packet *) msg;
+
+		BT_ASSERT(!state->cur_packet);
+		state->cur_packet = msg_packet->packet;
+		break;
+	}
+	case BT_MESSAGE_TYPE_PACKET_END:
+	{
+		const struct bt_message_packet *msg_packet =
+			(const struct bt_message_packet *) msg;
+
+		actual_packet = msg_packet->packet;
+		expected_packet = state->cur_packet;
+		BT_ASSERT(state->cur_packet);
+		state->cur_packet = NULL;
+		break;
+	}
+	default:
+		break;
+	}
+
+	BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
+		"message-packet-is-expected",
+		actual_packet == expected_packet,
+		"Message's packet is not expected: %![stream-]s, %![iterator-]i, "
+		"%![message-]n, %![received-packet-]a, %![expected-packet-]a",
+		stream, iterator, msg, actual_packet, expected_packet);
+
+end:
+	return;
+}
+
+static
+void assert_post_dev_next(
+		struct bt_message_iterator *iterator,
+		bt_message_iterator_class_next_method_status status,
+		bt_message_array_const msgs, uint64_t msg_count)
+{
+	if (status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
+		uint64_t i;
+
+		for (i = 0; i < msg_count; i++) {
+			assert_post_dev_expected_sequence(iterator, msgs[i]);
+			assert_post_dev_expected_packet(iterator, msgs[i]);
+		}
+	} else if (status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END) {
+		GHashTableIter iter;
+
+		gpointer stream_v, stream_state_v;
+
+		g_hash_table_iter_init(&iter, iterator->per_stream_state);
+		while (g_hash_table_iter_next(&iter, &stream_v,
+				&stream_state_v)) {
+			struct bt_stream *stream = stream_v;
+			struct per_stream_state *stream_state = stream_state_v;
+
+			BT_ASSERT_POST_DEV(NEXT_METHOD_NAME,
+				"stream-is-ended",
+				stream_state->expected_msg_types == 0,
+				"Stream is not ended: %![stream-]s, "
+				"%![iterator-]i, expected-msg-types=%s",
+				stream, iterator,
+				message_types_to_string(
+					stream_state->expected_msg_types)->str);
+		}
+
+	}
+}
+#endif
+
 /*
  * Call the `next` method of the iterator.  Do some validation on the returned
  * messages.
  */
 
-#define NEXT_METHOD_NAME	"bt_message_iterator_class_next_method"
-
 static
 enum bt_message_iterator_class_next_method_status
 call_iterator_next_method(
@@ -822,6 +1195,10 @@ call_iterator_next_method(
 			"Clock snapshots are not monotonic");
 	}
 
+#ifdef BT_DEV_MODE
+	assert_post_dev_next(iterator, status, msgs, *user_count);
+#endif
+
 	BT_ASSERT_POST_DEV_NO_ERROR_IF_NO_ERROR_STATUS(NEXT_METHOD_NAME,
 		status);
 
@@ -882,7 +1259,7 @@ bt_message_iterator_next(
 	 * by its downstream owner.
 	 *
 	 * For the same reason, there is no way that this iterator could
-	 * have seeked (cannot seek a self message iterator).
+	 * have sought (cannot seek a self message iterator).
 	 */
 	BT_ASSERT_DBG(iterator->state ==
 		BT_MESSAGE_ITERATOR_STATE_ACTIVE);
@@ -1169,6 +1546,8 @@ bt_message_iterator_seek_beginning(struct bt_message_iterator *iterator)
 			iterator, bt_common_func_status_string(status));
 	}
 
+	clear_per_stream_state(iterator);
+
 	set_iterator_state_after_seeking(iterator, status);
 	return status;
 }
@@ -1651,7 +2030,7 @@ end:
 
 /*
  * This function is installed as the iterator's next callback after we have
- * auto-seeked (seeked to the beginning and fast-forwarded) to send the
+ * auto-sought (sought to the beginning and fast-forwarded) to send the
  * messages saved in iterator->auto_seek.msgs.  Once this is done, the original
  * next callback is put back.
  */
@@ -1826,6 +2205,8 @@ bt_message_iterator_seek_ns_from_origin(
 				iterator, bt_common_func_status_string(status));
 		}
 
+		clear_per_stream_state(iterator);
+
 		switch (status) {
 		case BT_FUNC_STATUS_OK:
 			break;
@@ -1982,6 +2363,8 @@ bt_message_iterator_seek_ns_from_origin(
 		}
 	}
 
+	clear_per_stream_state(iterator);
+
 	/*
 	 * The following messages returned by the next method (including
 	 * post_auto_seek_next) must be after (or at) `ns_from_origin`.