Tests: flt.utils.muxer: run `black` on bt_plugin_muxer_test.py

[babeltrace.git] / src / plugins / utils / muxer / muxer.c

diff --git a/src/plugins/utils/muxer/muxer.c b/src/plugins/utils/muxer/muxer.c
index 31d374fc0fcf4b242e1001c75a727a0408323b64..6cf8ae6c65751a437214b8b6dc781f9fa14a59e8 100644 (file)
--- a/src/plugins/utils/muxer/muxer.c
+++ b/src/plugins/utils/muxer/muxer.c
@@ -23,7 +23,7 @@
 #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"
@@ -36,6 +36,8 @@
 #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"
@@ -943,11 +945,44 @@ muxer_msg_iter_youngest_upstream_msg_iter(
                        goto end;
                }
 
-               if (msg_ts_ns <= youngest_ts_ns) {
+               if (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);
+                       }
                }
        }