#define BT_COMP_LOG_SELF_COMP (muxer_comp->self_comp)
#define BT_LOG_OUTPUT_LEVEL (muxer_comp->log_level)
#define BT_LOG_TAG "PLUGIN/FLT.UTILS.MUXER"
-#include "plugins/comp-logging.h"
+#include "logging/comp-logging.h"
#include "common/macros.h"
#include "common/uuid.h"
#include <stdlib.h>
#include <string.h>
+#include "plugins/common/muxing/muxing.h"
+
#include "muxer.h"
#define ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME "assume-absolute-clock-classes"
struct muxer_msg_iter {
struct muxer_comp *muxer_comp;
+ /* Weak */
+ bt_self_message_iterator *self_msg_iter;
+
/*
* Array of struct muxer_upstream_msg_iter * (owned by this).
*
static
bt_self_component_port_input_message_iterator *
create_msg_iter_on_input_port(struct muxer_comp *muxer_comp,
+ struct muxer_msg_iter *muxer_msg_iter,
bt_self_component_port_input *self_port)
{
const bt_port *port = bt_self_component_port_as_port(
// TODO: Advance the iterator to >= the time of the latest
// returned message by the muxer message
// iterator which creates it.
- msg_iter = bt_self_component_port_input_message_iterator_create(
- self_port);
+ msg_iter = bt_self_component_port_input_message_iterator_create_from_message_iterator(
+ muxer_msg_iter->self_msg_iter, self_port);
if (!msg_iter) {
BT_COMP_LOGE("Cannot create upstream message iterator on input port: "
"port-addr=%p, port-name=\"%s\"",
/* Expect no clock class */
muxer_msg_iter->clock_class_expectation =
MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE;
- } else {
- BT_COMP_LOGE("Expecting stream class with a default clock class: "
+ } else if (muxer_msg_iter->clock_class_expectation !=
+ MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE) {
+ BT_COMP_LOGE("Expecting stream class without a default clock class: "
"stream-class-addr=%p, stream-class-name=\"%s\", "
"stream-class-id=%" PRIu64,
stream_class, bt_stream_class_get_name(stream_class),
goto end;
}
- if (msg_ts_ns <= youngest_ts_ns) {
+ /*
+ * Update the current message iterator if it has not been set
+ * yet, or if its current message has a timestamp smaller than
+ * the previously selected youngest message.
+ */
+ if (G_UNLIKELY(*muxer_upstream_msg_iter == NULL) ||
+ msg_ts_ns < youngest_ts_ns) {
*muxer_upstream_msg_iter =
cur_muxer_upstream_msg_iter;
youngest_ts_ns = msg_ts_ns;
*ts_ns = youngest_ts_ns;
+ } else if (msg_ts_ns == youngest_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.
+ */
+ const bt_message *selected_msg = g_queue_peek_head(
+ (*muxer_upstream_msg_iter)->msgs);
+ BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
+
+ /*
+ * Order the messages in an arbitrary but determinitic
+ * way.
+ */
+ ret = common_muxing_compare_messages(msg, selected_msg);
+ if (ret < 0) {
+ /*
+ * The `msg` should go first. Update the next
+ * iterator and the current timestamp.
+ */
+ *muxer_upstream_msg_iter =
+ cur_muxer_upstream_msg_iter;
+ youngest_ts_ns = msg_ts_ns;
+ *ts_ns = youngest_ts_ns;
+ } else if (ret == 0) {
+ /* Unable to pick which one should go first. */
+ BT_COMP_LOGW("Cannot deterministically pick next upstream message iterator because they have identical next messages: "
+ "muxer-upstream-msg-iter-wrap-addr=%p"
+ "cur-muxer-upstream-msg-iter-wrap-addr=%p",
+ *muxer_upstream_msg_iter,
+ cur_muxer_upstream_msg_iter);
+ }
}
}
}
upstream_msg_iter = create_msg_iter_on_input_port(muxer_comp,
- self_port);
+ muxer_msg_iter, self_port);
if (!upstream_msg_iter) {
/* create_msg_iter_on_input_port() logs errors */
BT_ASSERT(!upstream_msg_iter);
}
muxer_msg_iter->muxer_comp = muxer_comp;
+ muxer_msg_iter->self_msg_iter = self_msg_iter;
muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
muxer_msg_iter->active_muxer_upstream_msg_iters =
g_ptr_array_new_with_free_func(