/*
- * lttng-live.c
- *
- * Babeltrace CTF LTTng-live Client Component
+ * SPDX-License-Identifier: MIT
*
* Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
* Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
* Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@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.
+ * Babeltrace CTF LTTng-live Client Component
*/
#define BT_COMP_LOG_SELF_COMP self_comp
}
static
-struct lttng_live_trace *lttng_live_find_trace(struct lttng_live_session *session,
+struct lttng_live_trace *lttng_live_session_borrow_trace_by_id(struct lttng_live_session *session,
uint64_t trace_id)
{
uint64_t trace_idx;
trace->stream_iterators = g_ptr_array_new_with_free_func(
(GDestroyNotify) lttng_live_stream_iterator_destroy);
BT_ASSERT(trace->stream_iterators);
- trace->new_metadata_needed = true;
+ trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
g_ptr_array_add(session->traces, trace);
goto end;
}
BT_HIDDEN
-struct lttng_live_trace *lttng_live_borrow_trace(
+struct lttng_live_trace *lttng_live_session_borrow_or_create_trace_by_id(
struct lttng_live_session *session, uint64_t trace_id)
{
struct lttng_live_trace *trace;
- trace = lttng_live_find_trace(session, trace_id);
+ trace = lttng_live_session_borrow_trace_by_id(session, trace_id);
if (trace) {
goto end;
}
"session-id=%"PRIu64", session-name=\"%s\"",
session->id, session->session_name->str);
if (session->id != -1ULL) {
- if (lttng_live_detach_session(session)) {
+ if (lttng_live_session_detach(session)) {
if (!lttng_live_graph_is_canceled(
session->lttng_live_msg_iter)) {
/* Old relayd cannot detach sessions. */
enum lttng_live_stream_state orig_state = lttng_live_stream->state;
struct packet_index index;
- if (lttng_live_stream->trace->new_metadata_needed) {
+ if (lttng_live_stream->trace->metadata_stream_state ==
+ LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
goto end;
}
if (!session->attached) {
enum lttng_live_viewer_status attach_status =
- lttng_live_attach_session(session,
+ lttng_live_session_attach(session,
lttng_live_msg_iter->self_msg_iter);
if (attach_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
}
}
- status = lttng_live_get_new_streams(session,
+ status = lttng_live_session_get_new_streams(session,
lttng_live_msg_iter->self_msg_iter);
if (status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
status != LTTNG_LIVE_ITERATOR_STATUS_END) {
return status;
}
-BT_HIDDEN
-void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter)
-{
- uint64_t session_idx;
-
- for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
- session_idx++) {
- struct lttng_live_session *session =
- g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
- session->new_streams_needed = true;
- }
-}
-
static
void lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
{
trace_idx++) {
struct lttng_live_trace *trace =
g_ptr_array_index(session->traces, trace_idx);
- trace->new_metadata_needed = true;
+
+ BT_ASSERT(trace->metadata_stream_state !=
+ LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED);
+
+ trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
}
}
}
enum lttng_live_iterator_status emit_inactivity_message(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_stream_iterator *stream_iter,
- bt_message **message, uint64_t timestamp)
+ const bt_message **message, uint64_t timestamp)
{
enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
bt_logging_level log_level = lttng_live_msg_iter->log_level;
enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_quiescent_stream(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_stream_iterator *lttng_live_stream,
- bt_message **message)
+ const bt_message **message)
{
enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
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:
enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_stream(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_stream_iterator *lttng_live_stream,
- bt_message **message)
+ const bt_message **message)
{
enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
bt_logging_level log_level = lttng_live_msg_iter->log_level;
trace_idx++) {
struct lttng_live_trace *trace =
g_ptr_array_index(session->traces, trace_idx);
- if (trace->new_metadata_needed) {
+ if (trace->metadata_stream_state == LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
goto end;
}
*/
ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
break;
- case CTF_MSG_ITER_STATUS_INVAL:
- /* No argument provided by the user, so don't return INVAL. */
case CTF_MSG_ITER_STATUS_ERROR:
default:
ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
enum lttng_live_iterator_status lttng_live_iterator_close_stream(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_stream_iterator *stream_iter,
- bt_message **curr_msg)
+ const bt_message **curr_msg)
{
enum lttng_live_iterator_status live_status =
LTTNG_LIVE_ITERATOR_STATUS_OK;
"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);
enum lttng_live_iterator_status lttng_live_iterator_next_msg_on_stream(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_stream_iterator *stream_iter,
- bt_message **curr_msg)
+ const bt_message **curr_msg)
{
bt_logging_level log_level = lttng_live_msg_iter->log_level;
bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
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,
stream_iter_idx);
/*
- * Find if there is are now current message for this stream
- * iterator get it.
+ * If there is no current message for this stream, go fetch
+ * one.
*/
while (!stream_iter->current_msg) {
- bt_message *msg = NULL;
+ const bt_message *msg = NULL;
int64_t curr_msg_ts_ns = INT64_MAX;
stream_iter_status = lttng_live_iterator_next_msg_on_stream(
lttng_live_msg_iter, stream_iter, &msg);
/*
* 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.
*/
if (curr_msg_ts_ns >= lttng_live_msg_iter->last_msg_ts_ns) {
stream_iter->current_msg = msg;
stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
+ } else if (stream_iter->last_inactivity_ts > 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 =
+ stream_iter->last_inactivity_ts;
} else {
/*
* We received a message in the past. To ensure
}
BT_HIDDEN
-bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
+bt_message_iterator_class_next_method_status lttng_live_msg_iter_next(
bt_self_message_iterator *self_msg_it,
bt_message_array_const msgs, uint64_t capacity,
uint64_t *count)
{
- bt_component_class_message_iterator_next_method_status status;
+ bt_message_iterator_class_next_method_status status;
enum lttng_live_viewer_status viewer_status;
struct lttng_live_msg_iter *lttng_live_msg_iter =
bt_self_message_iterator_get_data(self_msg_it);
* is to prevent other graph users from using this live
* iterator in an messed up internal state.
*/
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
BT_COMP_LOGE_APPEND_CAUSE(self_comp,
"Message iterator was interrupted during a previous call to the `next()` and currently does not support continuing after such event.");
goto end;
if (lttng_live_msg_iter->sessions->len == 0) {
if (lttng_live->params.sess_not_found_act !=
SESSION_NOT_FOUND_ACTION_CONTINUE) {
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
goto end;
} else {
/*
viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
BT_COMP_LOGE_APPEND_CAUSE(self_comp,
"Error creating LTTng live viewer session");
} else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
} else {
bt_common_abort();
}
* doesn't support restarting after an interruption.
*/
if (*count > 0) {
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
} else {
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
}
break;
case LTTNG_LIVE_ITERATOR_STATUS_END:
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
break;
case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
BT_COMP_LOGE_APPEND_CAUSE(self_comp,
"Memory error preparing the next batch of messages: "
"live-iter-status=%s",
lttng_live_iterator_status_string(stream_iter_status));
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
break;
case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
"live-iter-status=%s",
lttng_live_iterator_status_string(stream_iter_status));
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
+ status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
/* Put all existing messages on error. */
put_messages(msgs, *count);
break;
}
BT_HIDDEN
-bt_component_class_message_iterator_initialize_method_status lttng_live_msg_iter_init(
+bt_message_iterator_class_initialize_method_status lttng_live_msg_iter_init(
bt_self_message_iterator *self_msg_it,
bt_self_message_iterator_configuration *config,
- bt_self_component_source *self_comp_src,
bt_self_component_port_output *self_port)
{
- bt_component_class_message_iterator_initialize_method_status status;
- bt_self_component *self_comp =
- bt_self_component_source_as_self_component(self_comp_src);
+ bt_message_iterator_class_initialize_method_status status;
struct lttng_live_component *lttng_live;
struct lttng_live_msg_iter *lttng_live_msg_iter;
enum lttng_live_viewer_status viewer_status;
bt_logging_level log_level;
-
- BT_ASSERT(self_msg_it);
+ bt_self_component *self_comp =
+ bt_self_message_iterator_borrow_component(self_msg_it);
lttng_live = bt_self_component_get_data(self_comp);
log_level = lttng_live->log_level;
lttng_live_msg_iter = lttng_live_msg_iter_create(lttng_live,
self_msg_it);
if (!lttng_live_msg_iter) {
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
+ status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
BT_COMP_LOGE_APPEND_CAUSE(self_comp,
"Failed to create lttng_live_msg_iter");
goto error;
}
bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_OK;
+ status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
goto end;
error:
- status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_ERROR;
+ status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
lttng_live_msg_iter_destroy(lttng_live_msg_iter);
end:
return status;