X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fplugins%2Fctf%2Fcommon%2Fmsg-iter%2Fmsg-iter.c;h=fc617f3d366ccee0f8a9b8815a23b3222c432602;hb=7b4311c13e6c335ae193fafed8060ef5bcec635e;hp=c094acc4c002fcf0b03ec4a348543abc1709d416;hpb=cad707e2548d7c9ea8501f12aec7ed28eaaf7975;p=babeltrace.git

diff --git a/src/plugins/ctf/common/msg-iter/msg-iter.c b/src/plugins/ctf/common/msg-iter/msg-iter.c
index c094acc4..fc617f3d 100644
--- a/src/plugins/ctf/common/msg-iter/msg-iter.c
+++ b/src/plugins/ctf/common/msg-iter/msg-iter.c
@@ -87,7 +87,6 @@ enum state {
 	STATE_DSCOPE_STREAM_PACKET_CONTEXT_BEGIN,
 	STATE_DSCOPE_STREAM_PACKET_CONTEXT_CONTINUE,
 	STATE_AFTER_STREAM_PACKET_CONTEXT,
-	STATE_CHECK_EMIT_MSG_STREAM_BEGINNING,
 	STATE_EMIT_MSG_STREAM_BEGINNING,
 	STATE_CHECK_EMIT_MSG_DISCARDED_EVENTS,
 	STATE_CHECK_EMIT_MSG_DISCARDED_PACKETS,
@@ -108,7 +107,6 @@ enum state {
 	STATE_SKIP_PACKET_PADDING,
 	STATE_EMIT_MSG_PACKET_END_MULTI,
 	STATE_EMIT_MSG_PACKET_END_SINGLE,
-	STATE_CHECK_EMIT_MSG_STREAM_END,
 	STATE_EMIT_QUEUED_MSG_PACKET_END,
 	STATE_EMIT_MSG_STREAM_END,
 	STATE_DONE,
@@ -129,21 +127,12 @@ struct ctf_msg_iter {
 	/* Current message iterator to create messages (weak) */
 	bt_self_message_iterator *self_msg_iter;
 
-	/* True to emit a stream beginning message. */
-	bool emit_stream_begin_msg;
-
-	/* True to emit a stream end message. */
-	bool emit_stream_end_msg;
-
 	/*
 	 * True if library objects are unavailable during the decoding and
 	 * should not be created/used.
 	 */
 	bool dry_run;
 
-	/* True to set the stream */
-	bool set_stream;
-
 	/*
 	 * Current dynamic scope field pointer.
 	 *
@@ -168,9 +157,6 @@ struct ctf_msg_iter {
 		struct ctf_event_class *ec;
 	} meta;
 
-	/* Current packet context field wrapper (NULL if not created yet) */
-	bt_packet_context_field *packet_context_field;
-
 	/* Current packet (NULL if not created yet) */
 	bt_packet *packet;
 
@@ -189,6 +175,12 @@ struct ctf_msg_iter {
 	 */
 	bool emit_delayed_packet_beginning_msg;
 
+	/*
+	 * True if this is the first packet we are reading, and therefore if we
+	 * should emit a stream beginning message.
+	 */
+	bool emit_stream_beginning_message;
+
 	/* Database of current dynamic scopes */
 	struct {
 		bt_field *stream_packet_context;
@@ -697,18 +689,13 @@ void release_all_dscopes(struct ctf_msg_iter *msg_it)
 {
 	msg_it->dscopes.stream_packet_context = NULL;
 
-	if (msg_it->packet_context_field) {
-		bt_packet_context_field_release(msg_it->packet_context_field);
-		msg_it->packet_context_field = NULL;
-	}
-
 	release_event_dscopes(msg_it);
 }
 
 static
 enum ctf_msg_iter_status switch_packet_state(struct ctf_msg_iter *msg_it)
 {
-	enum ctf_msg_iter_status status = CTF_MSG_ITER_STATUS_OK;
+	enum ctf_msg_iter_status status;
 	bt_self_component *self_comp = msg_it->self_comp;
 
 	/*
@@ -732,6 +719,30 @@ enum ctf_msg_iter_status switch_packet_state(struct ctf_msg_iter *msg_it)
 	release_all_dscopes(msg_it);
 	msg_it->cur_dscope_field = NULL;
 
+	if (msg_it->medium.medops.switch_packet) {
+		enum ctf_msg_iter_medium_status medium_status;
+
+		medium_status = msg_it->medium.medops.switch_packet(msg_it->medium.data);
+		if (medium_status == CTF_MSG_ITER_MEDIUM_STATUS_EOF) {
+			/* No more packets. */
+			msg_it->state = STATE_EMIT_MSG_STREAM_END;
+			status = CTF_MSG_ITER_STATUS_OK;
+			goto end;
+		} else if (medium_status != CTF_MSG_ITER_MEDIUM_STATUS_OK) {
+			status = (int) medium_status;
+			goto end;
+		}
+
+		/*
+		 * After the packet switch, the medium might want to give us a
+		 * different buffer for the new packet.
+		 */
+		status = request_medium_bytes(msg_it);
+		if (status != CTF_MSG_ITER_STATUS_OK) {
+			goto end;
+		}
+	}
+
 	/*
 	 * Adjust current buffer so that addr points to the beginning of the new
 	 * packet.
@@ -768,6 +779,7 @@ enum ctf_msg_iter_status switch_packet_state(struct ctf_msg_iter *msg_it)
 	msg_it->snapshots.end_clock = UINT64_C(-1);
 	msg_it->state = STATE_DSCOPE_TRACE_PACKET_HEADER_BEGIN;
 
+	status = CTF_MSG_ITER_STATUS_OK;
 end:
 	return status;
 }
@@ -791,7 +803,7 @@ enum ctf_msg_iter_status read_packet_header_begin_state(
 		break;
 	case CTF_MSG_ITER_STATUS_EOF:
 		status = CTF_MSG_ITER_STATUS_OK;
-		msg_it->state = STATE_CHECK_EMIT_MSG_STREAM_END;
+		msg_it->state = STATE_EMIT_MSG_STREAM_END;
 		goto end;
 	default:
 		goto end;
@@ -987,8 +999,22 @@ enum ctf_msg_iter_status after_packet_header_state(
 		goto end;
 	}
 
+	if (!msg_it->dry_run) {
+		status = set_current_stream(msg_it);
+		if (status != CTF_MSG_ITER_STATUS_OK) {
+			goto end;
+		}
+
+		status = set_current_packet(msg_it);
+		if (status != CTF_MSG_ITER_STATUS_OK) {
+			goto end;
+		}
+	}
+
 	msg_it->state = STATE_DSCOPE_STREAM_PACKET_CONTEXT_BEGIN;
 
+	status = CTF_MSG_ITER_STATUS_OK;
+
 end:
 	return status;
 }
@@ -1013,31 +1039,11 @@ enum ctf_msg_iter_status read_packet_context_begin_state(
 		goto end;
 	}
 
-	BT_ASSERT(!msg_it->packet_context_field);
-
 	if (packet_context_fc->in_ir && !msg_it->dry_run) {
-		/*
-		 * Create free packet context field from stream class.
-		 * This field is going to be moved to the packet once we
-		 * create it. We cannot create the packet now because a
-		 * packet is created from a stream, and this API must be
-		 * able to return the packet context properties without
-		 * creating a stream
-		 * (ctf_msg_iter_get_packet_properties()).
-		 */
-		msg_it->packet_context_field =
-			bt_packet_context_field_create(
-				msg_it->meta.sc->ir_sc);
-		if (!msg_it->packet_context_field) {
-			BT_COMP_LOGE_APPEND_CAUSE(self_comp,
-				"Cannot create packet context field wrapper from stream class.");
-			status = CTF_MSG_ITER_STATUS_ERROR;
-			goto end;
-		}
-
+		BT_ASSERT(!msg_it->dscopes.stream_packet_context);
+		BT_ASSERT(msg_it->packet);
 		msg_it->dscopes.stream_packet_context =
-			bt_packet_context_field_borrow_field(
-				msg_it->packet_context_field);
+			bt_packet_borrow_context_field(msg_it->packet);
 		BT_ASSERT(msg_it->dscopes.stream_packet_context);
 	}
 
@@ -1129,15 +1135,10 @@ enum ctf_msg_iter_status after_packet_context_state(struct ctf_msg_iter *msg_it)
 		goto end;
 	}
 
-	if (msg_it->stream) {
-		/*
-		 * Stream exists, which means we already emitted at
-		 * least one packet beginning message, so the initial
-		 * stream beginning message was also emitted.
-		 */
-		msg_it->state = STATE_CHECK_EMIT_MSG_DISCARDED_EVENTS;
+	if (msg_it->emit_stream_beginning_message) {
+		msg_it->state = STATE_EMIT_MSG_STREAM_BEGINNING;
 	} else {
-		msg_it->state = STATE_CHECK_EMIT_MSG_STREAM_BEGINNING;
+		msg_it->state = STATE_CHECK_EMIT_MSG_DISCARDED_EVENTS;
 	}
 
 end:
@@ -1572,30 +1573,6 @@ end:
 	return status;
 }
 
-static
-enum ctf_msg_iter_status check_emit_msg_stream_beginning_state(
-		struct ctf_msg_iter *msg_it)
-{
-	enum ctf_msg_iter_status status = CTF_MSG_ITER_STATUS_OK;
-
-	if (msg_it->set_stream) {
-		status = set_current_stream(msg_it);
-		if (status != CTF_MSG_ITER_STATUS_OK) {
-			goto end;
-		}
-	}
-
-	if (msg_it->emit_stream_begin_msg) {
-		msg_it->state = STATE_EMIT_MSG_STREAM_BEGINNING;
-	} else {
-		/* Stream's first packet */
-		msg_it->state = STATE_CHECK_EMIT_MSG_DISCARDED_EVENTS;
-	}
-
-end:
-	return status;
-}
-
 static
 enum ctf_msg_iter_status check_emit_msg_discarded_events(
 		struct ctf_msg_iter *msg_it)
@@ -1680,19 +1657,6 @@ end:
 	return CTF_MSG_ITER_STATUS_OK;
 }
 
-static
-enum ctf_msg_iter_status check_emit_msg_stream_end(
-		struct ctf_msg_iter *msg_it)
-{
-	if (msg_it->emit_stream_end_msg) {
-		msg_it->state = STATE_EMIT_MSG_STREAM_END;
-	} else {
-		msg_it->state = STATE_DONE;
-	}
-
-	return CTF_MSG_ITER_STATUS_OK;
-}
-
 static inline
 enum ctf_msg_iter_status handle_state(struct ctf_msg_iter *msg_it)
 {
@@ -1728,9 +1692,6 @@ enum ctf_msg_iter_status handle_state(struct ctf_msg_iter *msg_it)
 	case STATE_AFTER_STREAM_PACKET_CONTEXT:
 		status = after_packet_context_state(msg_it);
 		break;
-	case STATE_CHECK_EMIT_MSG_STREAM_BEGINNING:
-		status = check_emit_msg_stream_beginning_state(msg_it);
-		break;
 	case STATE_EMIT_MSG_STREAM_BEGINNING:
 		msg_it->state = STATE_CHECK_EMIT_MSG_DISCARDED_EVENTS;
 		break;
@@ -1789,10 +1750,7 @@ enum ctf_msg_iter_status handle_state(struct ctf_msg_iter *msg_it)
 		msg_it->state = STATE_SKIP_PACKET_PADDING;
 		break;
 	case STATE_EMIT_MSG_PACKET_END_SINGLE:
-		msg_it->state = STATE_CHECK_EMIT_MSG_STREAM_END;
-		break;
-	case STATE_CHECK_EMIT_MSG_STREAM_END:
-		status = check_emit_msg_stream_end(msg_it);
+		msg_it->state = STATE_EMIT_MSG_STREAM_END;
 		break;
 	case STATE_EMIT_QUEUED_MSG_PACKET_END:
 		msg_it->state = STATE_EMIT_MSG_PACKET_END_SINGLE;
@@ -1829,11 +1787,6 @@ void ctf_msg_iter_reset_for_next_stream_file(struct ctf_msg_iter *msg_it)
 	release_all_dscopes(msg_it);
 	msg_it->cur_dscope_field = NULL;
 
-	if (msg_it->packet_context_field) {
-		bt_packet_context_field_release(msg_it->packet_context_field);
-		msg_it->packet_context_field = NULL;
-	}
-
 	msg_it->buf.addr = NULL;
 	msg_it->buf.sz = 0;
 	msg_it->buf.at = 0;
@@ -1863,6 +1816,7 @@ void ctf_msg_iter_reset(struct ctf_msg_iter *msg_it)
 	msg_it->prev_packet_snapshots.packets = UINT64_C(-1);
 	msg_it->prev_packet_snapshots.beginning_clock = UINT64_C(-1);
 	msg_it->prev_packet_snapshots.end_clock = UINT64_C(-1);
+	msg_it->emit_stream_beginning_message = true;
 }
 
 static
@@ -2469,7 +2423,7 @@ struct ctf_field_class *bfcr_borrow_variant_selected_field_class_cb(
 	if (selected_option->fc->in_ir && !msg_it->dry_run) {
 		bt_field *var_field = stack_top(msg_it->stack)->base;
 
-		ret = bt_field_variant_select_option_field_by_index(
+		ret = bt_field_variant_select_option_by_index(
 			var_field, option_index);
 		if (ret) {
 			BT_COMP_LOGE_APPEND_CAUSE(self_comp,
@@ -2541,33 +2495,11 @@ bt_message *create_msg_packet_beginning(struct ctf_msg_iter *msg_it,
 		bool use_default_cs)
 {
 	bt_self_component *self_comp = msg_it->self_comp;
-	int ret;
 	bt_message *msg;
 	const bt_stream_class *sc = msg_it->meta.sc->ir_sc;
 
 	BT_ASSERT(msg_it->packet);
 	BT_ASSERT(sc);
-
-	if (msg_it->packet_context_field) {
-		ret = bt_packet_move_context_field(
-			msg_it->packet, msg_it->packet_context_field);
-		if (ret) {
-			msg = NULL;
-			goto end;
-		}
-
-		msg_it->packet_context_field = NULL;
-
-		/*
-		 * At this point msg_it->dscopes.stream_packet_context
-		 * has the same value as the packet context field within
-		 * msg_it->packet.
-		 */
-		BT_ASSERT(bt_packet_borrow_context_field(
-			msg_it->packet) ==
-			msg_it->dscopes.stream_packet_context);
-	}
-
 	BT_ASSERT(msg_it->self_msg_iter);
 
 	if (msg_it->meta.sc->packets_have_ts_begin) {
@@ -2837,6 +2769,8 @@ struct ctf_msg_iter *ctf_msg_iter_create(
 	BT_ASSERT(tc);
 	BT_ASSERT(medops.request_bytes);
 	BT_ASSERT(medops.borrow_stream);
+	BT_ASSERT(max_request_sz > 0);
+
 	BT_COMP_LOG_CUR_LVL(BT_LOG_DEBUG, log_level, self_comp,
 		"Creating CTF plugin message iterator: "
 		"trace-addr=%p, max-request-size=%zu, "
@@ -2923,7 +2857,6 @@ enum ctf_msg_iter_status ctf_msg_iter_get_next_message(
 
 	BT_ASSERT_DBG(msg_it);
 	BT_ASSERT_DBG(message);
-	msg_it->set_stream = true;
 	BT_COMP_LOGD("Getting next message: msg-it-addr=%p", msg_it);
 
 	while (true) {
@@ -2982,11 +2915,6 @@ enum ctf_msg_iter_status ctf_msg_iter_get_next_message(
 
 			goto end;
 		case STATE_EMIT_MSG_PACKET_BEGINNING:
-			status = set_current_packet(msg_it);
-			if (status != CTF_MSG_ITER_STATUS_OK) {
-				goto end;
-			}
-
 			if (G_UNLIKELY(msg_it->meta.tc->quirks.barectf_event_before_packet)) {
 				msg_it->emit_delayed_packet_beginning_msg = true;
 				/*
@@ -3018,6 +2946,7 @@ enum ctf_msg_iter_status ctf_msg_iter_get_next_message(
 		case STATE_EMIT_MSG_STREAM_BEGINNING:
 			/* create_msg_stream_beginning() logs errors */
 			*message = create_msg_stream_beginning(msg_it);
+			msg_it->emit_stream_beginning_message = false;
 
 			if (!*message) {
 				status = CTF_MSG_ITER_STATUS_ERROR;
@@ -3054,7 +2983,6 @@ enum ctf_msg_iter_status decode_until_state( struct ctf_msg_iter *msg_it,
 	bt_self_component *self_comp = msg_it->self_comp;
 
 	BT_ASSERT_DBG(msg_it);
-	msg_it->set_stream = false;
 
 	do {
 		/*
@@ -3086,7 +3014,6 @@ enum ctf_msg_iter_status decode_until_state( struct ctf_msg_iter *msg_it,
 		case STATE_DSCOPE_STREAM_PACKET_CONTEXT_BEGIN:
 		case STATE_DSCOPE_STREAM_PACKET_CONTEXT_CONTINUE:
 		case STATE_AFTER_STREAM_PACKET_CONTEXT:
-		case STATE_CHECK_EMIT_MSG_STREAM_BEGINNING:
 		case STATE_EMIT_MSG_STREAM_BEGINNING:
 		case STATE_CHECK_EMIT_MSG_DISCARDED_EVENTS:
 		case STATE_EMIT_MSG_DISCARDED_EVENTS:
@@ -3108,7 +3035,6 @@ enum ctf_msg_iter_status decode_until_state( struct ctf_msg_iter *msg_it,
 		case STATE_EMIT_MSG_PACKET_END_MULTI:
 		case STATE_EMIT_MSG_PACKET_END_SINGLE:
 		case STATE_EMIT_QUEUED_MSG_PACKET_END:
-		case STATE_CHECK_EMIT_MSG_STREAM_END:
 		case STATE_EMIT_MSG_STREAM_END:
 			break;
 		case STATE_DONE:
@@ -3147,39 +3073,18 @@ end:
 	return status;
 }
 
-BT_HIDDEN
-void ctf_msg_iter_set_medops_data(struct ctf_msg_iter *msg_it,
-		void *medops_data)
-{
-	BT_ASSERT(msg_it);
-	msg_it->medium.data = medops_data;
-}
-
 BT_HIDDEN
 enum ctf_msg_iter_status ctf_msg_iter_seek(struct ctf_msg_iter *msg_it,
 		off_t offset)
 {
 	enum ctf_msg_iter_status status = CTF_MSG_ITER_STATUS_OK;
-	bt_self_component *self_comp = msg_it->self_comp;
 	enum ctf_msg_iter_medium_status medium_status;
 
 	BT_ASSERT(msg_it);
-	if (offset < 0) {
-		BT_COMP_LOGE_APPEND_CAUSE(self_comp,
-			"Cannot seek to negative offset: offset=%jd",
-			(intmax_t) offset);
-		status = CTF_MSG_ITER_STATUS_INVAL;
-		goto end;
-	}
-
-	if (!msg_it->medium.medops.seek) {
-		status = CTF_MSG_ITER_STATUS_UNSUPPORTED;
-		BT_COMP_LOGD("Aborting seek as the iterator's underlying media does not implement seek support.");
-		goto end;
-	}
+	BT_ASSERT(offset >= 0);
+	BT_ASSERT(msg_it->medium.medops.seek);
 
-	medium_status = msg_it->medium.medops.seek(
-		CTF_MSG_ITER_SEEK_WHENCE_SET, offset, msg_it->medium.data);
+	medium_status = msg_it->medium.medops.seek(offset, msg_it->medium.data);
 	if (medium_status != CTF_MSG_ITER_MEDIUM_STATUS_OK) {
 		if (medium_status == CTF_MSG_ITER_MEDIUM_STATUS_EOF) {
 			status = CTF_MSG_ITER_STATUS_EOF;
@@ -3259,20 +3164,6 @@ end:
 	return status;
 }
 
-BT_HIDDEN
-void ctf_msg_iter_set_emit_stream_beginning_message(struct ctf_msg_iter *msg_it,
-		bool val)
-{
-	msg_it->emit_stream_begin_msg = val;
-}
-
-BT_HIDDEN
-void ctf_msg_iter_set_emit_stream_end_message(struct ctf_msg_iter *msg_it,
-		bool val)
-{
-	msg_it->emit_stream_end_msg = val;
-}
-
 BT_HIDDEN
 void ctf_msg_iter_set_dry_run(struct ctf_msg_iter *msg_it,
 		bool val)