#include "compat/compiler.h"
#include <babeltrace2/types.h>
+#include "plugins/common/muxing/muxing.h"
+
#include "data-stream.h"
#include "metadata.h"
#include "lttng-live.h"
* When disconnected from relayd: try to re-connect endlessly.
*/
static
-enum lttng_live_iterator_status lttng_live_iterator_next_on_stream(
+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)
enum lttng_live_iterator_status next_stream_iterator_for_trace(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_trace *live_trace,
- struct lttng_live_stream_iterator **candidate_stream_iter)
+ struct lttng_live_stream_iterator **youngest_trace_stream_iter)
{
- struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
+ struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
enum lttng_live_iterator_status stream_iter_status;;
- int64_t curr_candidate_msg_ts = INT64_MAX;
+ int64_t youngest_candidate_msg_ts = INT64_MAX;
uint64_t stream_iter_idx;
bt_logging_level log_level = lttng_live_msg_iter->log_level;
bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
g_ptr_array_index(live_trace->stream_iterators,
stream_iter_idx);
+ /*
+ * Since we may remove elements from the GPtrArray as we
+ * iterate over it, it's possible to see the same element more
+ * than once.
+ */
+ if (stream_iter == youngest_candidate_stream_iter) {
+ stream_iter_idx++;
+ continue;
+ }
+
/*
* Find if there is are now current message for this stream
* iterator get it.
while (!stream_iter->current_msg) {
bt_message *msg = NULL;
int64_t curr_msg_ts_ns = INT64_MAX;
- stream_iter_status = lttng_live_iterator_next_on_stream(
+ stream_iter_status = lttng_live_iterator_next_msg_on_stream(
lttng_live_msg_iter, stream_iter, &msg);
BT_COMP_LOGD("live stream iterator returned status :%s",
}
}
- if (!stream_iter_is_ended &&
- stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
- /*
- * Update the current best candidate message for the
- * stream iterator of thise live trace to be forwarded
- * downstream.
- */
- curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
- curr_candidate_stream_iter = stream_iter;
- }
+ BT_ASSERT(stream_iter != youngest_candidate_stream_iter);
+
+ if (!stream_iter_is_ended) {
+ if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
+ stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
+ /*
+ * Update the current best candidate message
+ * for the stream iterator of this live trace
+ * to be forwarded downstream.
+ */
+ youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
+ youngest_candidate_stream_iter = stream_iter;
+ } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
+ /*
+ * Order the messages in an arbitrary but
+ * deterministic way.
+ */
+ BT_ASSERT(stream_iter != youngest_candidate_stream_iter);
+ int ret = common_muxing_compare_messages(
+ stream_iter->current_msg,
+ youngest_candidate_stream_iter->current_msg);
+ if (ret < 0) {
+ /*
+ * The `youngest_candidate_stream_iter->current_msg`
+ * should go first. Update the next
+ * iterator and the current timestamp.
+ */
+ youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
+ youngest_candidate_stream_iter = stream_iter;
+ } else if (ret == 0) {
+ /*
+ * Unable to pick which one should go
+ * first.
+ */
+ BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
+ "stream-iter-addr=%p"
+ "stream-iter-addr=%p",
+ stream_iter,
+ youngest_candidate_stream_iter);
+ }
+ }
- if (stream_iter_is_ended) {
+ stream_iter_idx++;
+ } else {
/*
* The live stream iterator is ENDed. We remove that
* iterator from the list and we restart the iteration
g_ptr_array_remove_index_fast(live_trace->stream_iterators,
stream_iter_idx);
stream_iter_idx = 0;
- } else {
- stream_iter_idx++;
}
}
- if (curr_candidate_stream_iter) {
- *candidate_stream_iter = curr_candidate_stream_iter;
+ if (youngest_candidate_stream_iter) {
+ *youngest_trace_stream_iter = youngest_candidate_stream_iter;
stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
} else {
/*
enum lttng_live_iterator_status next_stream_iterator_for_session(
struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_session *session,
- struct lttng_live_stream_iterator **candidate_session_stream_iter)
+ struct lttng_live_stream_iterator **youngest_session_stream_iter)
{
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
enum lttng_live_iterator_status stream_iter_status;
uint64_t trace_idx = 0;
- int64_t curr_candidate_msg_ts = INT64_MAX;
- struct lttng_live_stream_iterator *curr_candidate_stream_iter = NULL;
+ int64_t youngest_candidate_msg_ts = INT64_MAX;
+ struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
/*
* Make sure we are attached to the session and look for new streams
if (!trace_is_ended) {
BT_ASSERT(stream_iter);
- if (stream_iter->current_msg_ts_ns <= curr_candidate_msg_ts) {
- curr_candidate_msg_ts = stream_iter->current_msg_ts_ns;
- curr_candidate_stream_iter = stream_iter;
+ if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
+ stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
+ youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
+ youngest_candidate_stream_iter = stream_iter;
+ } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
+ /*
+ * Order the messages in an arbitrary but
+ * deterministic way.
+ */
+ int ret = common_muxing_compare_messages(
+ stream_iter->current_msg,
+ youngest_candidate_stream_iter->current_msg);
+ if (ret < 0) {
+ /*
+ * The `youngest_candidate_stream_iter->current_msg`
+ * should go first. Update the next iterator
+ * and the current timestamp.
+ */
+ youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
+ youngest_candidate_stream_iter = stream_iter;
+ } else if (ret == 0) {
+ /* Unable to pick which one should go first. */
+ BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
+ "stream-iter-addr=%p" "stream-iter-addr=%p",
+ stream_iter, youngest_candidate_stream_iter);
+ }
}
trace_idx++;
} else {
trace_idx = 0;
}
}
- if (curr_candidate_stream_iter) {
- *candidate_session_stream_iter = curr_candidate_stream_iter;
+ if (youngest_candidate_stream_iter) {
+ *youngest_session_stream_iter = youngest_candidate_stream_iter;
stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
} else {
/*
bt_self_message_iterator_get_data(self_msg_it);
struct lttng_live_component *lttng_live =
lttng_live_msg_iter->lttng_live_comp;
+ bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
+ bt_logging_level log_level = lttng_live_msg_iter->log_level;
enum lttng_live_iterator_status stream_iter_status;
uint64_t session_idx;
* return it.
*/
while (*count < capacity) {
- struct lttng_live_stream_iterator *next_stream_iter = NULL,
- *candidate_stream_iter = NULL;
- int64_t next_msg_ts_ns = INT64_MAX;
+ struct lttng_live_stream_iterator *youngest_stream_iter = NULL,
+ *candidate_stream_iter = NULL;
+ int64_t youngest_msg_ts_ns = INT64_MAX;
BT_ASSERT(lttng_live_msg_iter->sessions);
session_idx = 0;
goto end;
}
- if (candidate_stream_iter->current_msg_ts_ns <= next_msg_ts_ns) {
- next_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
- next_stream_iter = candidate_stream_iter;
+ if (G_UNLIKELY(youngest_stream_iter == NULL) ||
+ candidate_stream_iter->current_msg_ts_ns <= youngest_msg_ts_ns) {
+ youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
+ youngest_stream_iter = candidate_stream_iter;
+ } else if (candidate_stream_iter->current_msg_ts_ns == youngest_msg_ts_ns) {
+ /*
+ * The currently selected message to be sent
+ * downstream next has the exact same timestamp
+ * that of the current candidate message. We
+ * must break the tie in a predictable manner.
+ */
+ BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
+ /*
+ * Order the messages in an arbitrary but
+ * deterministic way.
+ */
+ int ret = common_muxing_compare_messages(
+ candidate_stream_iter->current_msg,
+ youngest_stream_iter->current_msg);
+ if (ret < 0) {
+ /*
+ * The `candidate_stream_iter->current_msg`
+ * should go first. Update the next
+ * iterator and the current timestamp.
+ */
+ youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
+ youngest_stream_iter = candidate_stream_iter;
+ } else if (ret == 0) {
+ /* Unable to pick which one should go first. */
+ BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
+ "next-stream-iter-addr=%p" "candidate-stream-iter-addr=%p",
+ youngest_stream_iter, candidate_stream_iter);
+ }
}
session_idx++;
}
- if (!next_stream_iter) {
+ if (!youngest_stream_iter) {
stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
goto end;
}
- BT_ASSERT(next_stream_iter->current_msg);
+ BT_ASSERT(youngest_stream_iter->current_msg);
/* Ensure monotonicity. */
BT_ASSERT(lttng_live_msg_iter->last_msg_ts_ns <=
- next_stream_iter->current_msg_ts_ns);
+ youngest_stream_iter->current_msg_ts_ns);
/*
* Insert the next message to the message batch. This will set
* stream iterator current messsage to NULL so that next time
* we fetch the next message of that stream iterator
*/
- BT_MESSAGE_MOVE_REF(msgs[*count], next_stream_iter->current_msg);
+ BT_MESSAGE_MOVE_REF(msgs[*count], youngest_stream_iter->current_msg);
(*count)++;
/* Update the last timestamp in nanoseconds sent downstream. */
- lttng_live_msg_iter->last_msg_ts_ns = next_msg_ts_ns;
- next_stream_iter->current_msg_ts_ns = INT64_MAX;
+ lttng_live_msg_iter->last_msg_ts_ns = youngest_msg_ts_ns;
+ youngest_stream_iter->current_msg_ts_ns = INT64_MAX;
stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
}
projects / babeltrace.git / blobdiff