msg);
break;
case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
- clock_snapshot = bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
+ clock_snapshot = bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
msg);
break;
default:
"Error getting the next message from CTF message iterator");
live_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
goto end;
+ } else if (status == CTF_MSG_ITER_STATUS_EOF) {
+ BT_COMP_LOGI("Reached the end of the live stream iterator.");
+ live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
+ goto end;
}
BT_ASSERT(status == CTF_MSG_ITER_STATUS_OK);
return live_status;
}
+static
+bool is_discarded_packet_or_event_message(const bt_message *msg)
+{
+ const enum bt_message_type msg_type = bt_message_get_type(msg);
+
+ return msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS ||
+ msg_type == BT_MESSAGE_TYPE_DISCARDED_PACKETS;
+}
+
+static
+enum lttng_live_iterator_status adjust_discarded_packets_message(
+ bt_self_message_iterator *iter,
+ const bt_stream *stream,
+ const bt_message *msg_in, bt_message **msg_out,
+ uint64_t new_begin_ts)
+{
+ enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
+ enum bt_property_availability availability;
+ const bt_clock_snapshot *clock_snapshot;
+ uint64_t end_ts;
+ uint64_t count;
+
+ clock_snapshot = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(msg_in);
+ end_ts = bt_clock_snapshot_get_value(clock_snapshot);
+
+ availability = bt_message_discarded_packets_get_count(msg_in, &count);
+ BT_ASSERT_DBG(availability == BT_PROPERTY_AVAILABILITY_AVAILABLE);
+
+ *msg_out = bt_message_discarded_packets_create_with_default_clock_snapshots(
+ iter, stream, new_begin_ts, end_ts);
+ if (!*msg_out) {
+ status = LTTNG_LIVE_ITERATOR_STATUS_NOMEM;
+ goto end;
+ }
+
+ bt_message_discarded_packets_set_count(*msg_out, count);
+end:
+ return status;
+}
+
+static
+enum lttng_live_iterator_status adjust_discarded_events_message(
+ bt_self_message_iterator *iter,
+ const bt_stream *stream,
+ const bt_message *msg_in, bt_message **msg_out,
+ uint64_t new_begin_ts)
+{
+ enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
+ enum bt_property_availability availability;
+ const bt_clock_snapshot *clock_snapshot;
+ uint64_t end_ts;
+ uint64_t count;
+
+ clock_snapshot = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(msg_in);
+ end_ts = bt_clock_snapshot_get_value(clock_snapshot);
+
+ availability = bt_message_discarded_events_get_count(msg_in, &count);
+ BT_ASSERT_DBG(availability == BT_PROPERTY_AVAILABILITY_AVAILABLE);
+
+ *msg_out = bt_message_discarded_events_create_with_default_clock_snapshots(
+ iter, stream, new_begin_ts, end_ts);
+ if (!*msg_out) {
+ status = LTTNG_LIVE_ITERATOR_STATUS_NOMEM;
+ goto end;
+ }
+
+ bt_message_discarded_events_set_count(*msg_out, count);
+end:
+ return status;
+}
+
static
enum lttng_live_iterator_status next_stream_iterator_for_trace(
struct lttng_live_msg_iter *lttng_live_msg_iter,
/*
* Check if the message of the current live stream
- * iterator occured at the exact same time or after the
+ * iterator occurred at the exact same time or after the
* last message returned by this component's message
* iterator. If not, we return an error.
*/
stream_iter->current_msg = msg;
stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
} else {
- /*
- * We received a message in the past. To ensure
- * monotonicity, we can't send it forward.
- */
- BT_COMP_LOGE_APPEND_CAUSE(self_comp,
- "Message's timestamp is less than "
- "lttng-live's message iterator's last "
- "returned timestamp: "
- "lttng-live-msg-iter-addr=%p, ts=%" PRId64 ", "
- "last-msg-ts=%" PRId64,
- lttng_live_msg_iter, curr_msg_ts_ns,
- lttng_live_msg_iter->last_msg_ts_ns);
- stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
- goto end;
+ const bt_clock_class *clock_class;
+ const bt_stream_class *stream_class;
+ enum bt_clock_class_cycles_to_ns_from_origin_status ts_ns_status;
+ int64_t last_inactivity_ts_ns;
+
+ stream_class = bt_stream_borrow_class_const(stream_iter->stream);
+ clock_class = bt_stream_class_borrow_default_clock_class_const(stream_class);
+
+ ts_ns_status = bt_clock_class_cycles_to_ns_from_origin(
+ clock_class,
+ stream_iter->last_inactivity_ts,
+ &last_inactivity_ts_ns);
+ if (ts_ns_status != BT_CLOCK_CLASS_CYCLES_TO_NS_FROM_ORIGIN_STATUS_OK) {
+ stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
+ goto end;
+ }
+
+ if (last_inactivity_ts_ns > curr_msg_ts_ns &&
+ is_discarded_packet_or_event_message(msg)) {
+ /*
+ * The CTF message iterator emits Discarded
+ * Packets and Events with synthesized begin and
+ * end timestamps from the bounds of the last
+ * known packet and the newly decoded packet
+ * header.
+ *
+ * The CTF message iterator is not aware of
+ * stream inactivity beacons. Hence, we have
+ * to adjust the begin timestamp of those types
+ * of messages if a stream signaled its
+ * inactivity up until _after_ the last known
+ * packet's begin timestamp.
+ *
+ * Otherwise, the monotonicity guarantee would
+ * not be preserved.
+ */
+ const enum bt_message_type msg_type =
+ bt_message_get_type(msg);
+ enum lttng_live_iterator_status adjust_status =
+ LTTNG_LIVE_ITERATOR_STATUS_OK;
+ bt_message *adjusted_message;
+
+ switch (msg_type) {
+ case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
+ adjust_status = adjust_discarded_events_message(
+ lttng_live_msg_iter->self_msg_iter,
+ stream_iter->stream,
+ msg, &adjusted_message,
+ stream_iter->last_inactivity_ts);
+ break;
+ case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
+ adjust_status = adjust_discarded_packets_message(
+ lttng_live_msg_iter->self_msg_iter,
+ stream_iter->stream,
+ msg, &adjusted_message,
+ stream_iter->last_inactivity_ts);
+ break;
+ default:
+ bt_common_abort();
+ }
+
+ if (adjust_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
+ stream_iter_status = adjust_status;
+ goto end;
+ }
+
+ BT_ASSERT_DBG(adjusted_message);
+ stream_iter->current_msg = adjusted_message;
+ stream_iter->current_msg_ts_ns =
+ last_inactivity_ts_ns;
+ } else {
+ /*
+ * We received a message in the past. To ensure
+ * monotonicity, we can't send it forward.
+ */
+ BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+ "Message's timestamp is less than "
+ "lttng-live's message iterator's last "
+ "returned timestamp: "
+ "lttng-live-msg-iter-addr=%p, ts=%" PRId64 ", "
+ "last-msg-ts=%" PRId64,
+ lttng_live_msg_iter, curr_msg_ts_ns,
+ lttng_live_msg_iter->last_msg_ts_ns);
+ stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
+ goto end;
+ }
}
}
projects / babeltrace.git / blobdiff