src/plugins/utils/muxer/muxer.c

   1 /*
   2  * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
   3  *
   4  * Permission is hereby granted, free of charge, to any person obtaining a copy
   5  * of this software and associated documentation files (the "Software"), to deal
   6  * in the Software without restriction, including without limitation the rights
   7  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   8  * copies of the Software, and to permit persons to whom the Software is
   9  * furnished to do so, subject to the following conditions:
  10  *
  11  * The above copyright notice and this permission notice shall be included in
  12  * all copies or substantial portions of the Software.
  13  *
  14  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  19  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  20  * SOFTWARE.
  21  */
  22
  23 #define BT_COMP_LOG_SELF_COMP (muxer_comp->self_comp)
  24 #define BT_LOG_OUTPUT_LEVEL (muxer_comp->log_level)
  25 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.MUXER"
  26 #include "logging/comp-logging.h"
  27
  28 #include "common/macros.h"
  29 #include "common/uuid.h"
  30 #include <babeltrace2/babeltrace.h>
  31 #include <glib.h>
  32 #include <stdbool.h>
  33 #include <inttypes.h>
  34 #include "common/assert.h"
  35 #include "common/common.h"
  36 #include <stdlib.h>
  37 #include <string.h>
  38
  39 #include "muxer.h"
  40
  41 #define ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME        "assume-absolute-clock-classes"
  42
  43 struct muxer_comp {
  44         /* Weak refs */
  45         bt_self_component_filter *self_comp_flt;
  46         bt_self_component *self_comp;
  47
  48         unsigned int next_port_num;
  49         size_t available_input_ports;
  50         bool initializing_muxer_msg_iter;
  51         bool assume_absolute_clock_classes;
  52         bt_logging_level log_level;
  53 };
  54
  55 struct muxer_upstream_msg_iter {
  56         struct muxer_comp *muxer_comp;
  57
  58         /* Owned by this, NULL if ended */
  59         bt_self_component_port_input_message_iterator *msg_iter;
  60
  61         /* Contains `const bt_message *`, owned by this */
  62         GQueue *msgs;
  63 };
  64
  65 enum muxer_msg_iter_clock_class_expectation {
  66         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY = 0,
  67         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE,
  68         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE,
  69         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID,
  70         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID,
  71 };
  72
  73 struct muxer_msg_iter {
  74         struct muxer_comp *muxer_comp;
  75
  76         /* Weak */
  77         bt_self_message_iterator *self_msg_iter;
  78
  79         /*
  80          * Array of struct muxer_upstream_msg_iter * (owned by this).
  81          *
  82          * NOTE: This array is searched in linearly to find the youngest
  83          * current message. Keep this until benchmarks confirm that
  84          * another data structure is faster than this for our typical
  85          * use cases.
  86          */
  87         GPtrArray *active_muxer_upstream_msg_iters;
  88
  89         /*
  90          * Array of struct muxer_upstream_msg_iter * (owned by this).
  91          *
  92          * We move ended message iterators from
  93          * `active_muxer_upstream_msg_iters` to this array so as to be
  94          * able to restore them when seeking.
  95          */
  96         GPtrArray *ended_muxer_upstream_msg_iters;
  97
  98         /* Last time returned in a message */
  99         int64_t last_returned_ts_ns;
 100
 101         /* Clock class expectation state */
 102         enum muxer_msg_iter_clock_class_expectation clock_class_expectation;
 103
 104         /*
 105          * Expected clock class UUID, only valid when
 106          * clock_class_expectation is
 107          * MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID.
 108          */
 109         bt_uuid_t expected_clock_class_uuid;
 110 };
 111
 112 static
 113 void empty_message_queue(struct muxer_upstream_msg_iter *upstream_msg_iter)
 114 {
 115         const bt_message *msg;
 116
 117         while ((msg = g_queue_pop_head(upstream_msg_iter->msgs))) {
 118                 bt_message_put_ref(msg);
 119         }
 120 }
 121
 122 static
 123 void destroy_muxer_upstream_msg_iter(
 124                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
 125 {
 126         struct muxer_comp *muxer_comp;
 127
 128         if (!muxer_upstream_msg_iter) {
 129                 return;
 130         }
 131
 132         muxer_comp = muxer_upstream_msg_iter->muxer_comp;
 133         BT_COMP_LOGD("Destroying muxer's upstream message iterator wrapper: "
 134                 "addr=%p, msg-iter-addr=%p, queue-len=%u",
 135                 muxer_upstream_msg_iter,
 136                 muxer_upstream_msg_iter->msg_iter,
 137                 muxer_upstream_msg_iter->msgs->length);
 138         bt_self_component_port_input_message_iterator_put_ref(
 139                 muxer_upstream_msg_iter->msg_iter);
 140
 141         if (muxer_upstream_msg_iter->msgs) {
 142                 empty_message_queue(muxer_upstream_msg_iter);
 143                 g_queue_free(muxer_upstream_msg_iter->msgs);
 144         }
 145
 146         g_free(muxer_upstream_msg_iter);
 147 }
 148
 149 static
 150 int muxer_msg_iter_add_upstream_msg_iter(struct muxer_msg_iter *muxer_msg_iter,
 151                 bt_self_component_port_input_message_iterator *self_msg_iter)
 152 {
 153         int ret = 0;
 154         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
 155                 g_new0(struct muxer_upstream_msg_iter, 1);
 156         struct muxer_comp *muxer_comp = muxer_msg_iter->muxer_comp;
 157
 158         if (!muxer_upstream_msg_iter) {
 159                 BT_COMP_LOGE_STR("Failed to allocate one muxer's upstream message iterator wrapper.");
 160                 goto error;
 161         }
 162
 163         muxer_upstream_msg_iter->muxer_comp = muxer_comp;
 164         muxer_upstream_msg_iter->msg_iter = self_msg_iter;
 165         bt_self_component_port_input_message_iterator_get_ref(muxer_upstream_msg_iter->msg_iter);
 166         muxer_upstream_msg_iter->msgs = g_queue_new();
 167         if (!muxer_upstream_msg_iter->msgs) {
 168                 BT_COMP_LOGE_STR("Failed to allocate a GQueue.");
 169                 goto error;
 170         }
 171
 172         g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
 173                 muxer_upstream_msg_iter);
 174         BT_COMP_LOGD("Added muxer's upstream message iterator wrapper: "
 175                 "addr=%p, muxer-msg-iter-addr=%p, msg-iter-addr=%p",
 176                 muxer_upstream_msg_iter, muxer_msg_iter,
 177                 self_msg_iter);
 178
 179         goto end;
 180
 181 error:
 182         g_free(muxer_upstream_msg_iter);
 183         ret = -1;
 184
 185 end:
 186         return ret;
 187 }
 188
 189 static
 190 bt_self_component_add_port_status add_available_input_port(
 191                 bt_self_component_filter *self_comp)
 192 {
 193         struct muxer_comp *muxer_comp = bt_self_component_get_data(
 194                 bt_self_component_filter_as_self_component(self_comp));
 195         bt_self_component_add_port_status status =
 196                 BT_SELF_COMPONENT_ADD_PORT_STATUS_OK;
 197         GString *port_name = NULL;
 198
 199         BT_ASSERT(muxer_comp);
 200         port_name = g_string_new("in");
 201         if (!port_name) {
 202                 BT_COMP_LOGE_STR("Failed to allocate a GString.");
 203                 status = BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR;
 204                 goto end;
 205         }
 206
 207         g_string_append_printf(port_name, "%u", muxer_comp->next_port_num);
 208         status = bt_self_component_filter_add_input_port(
 209                 self_comp, port_name->str, NULL, NULL);
 210         if (status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 211                 BT_COMP_LOGE("Cannot add input port to muxer component: "
 212                         "port-name=\"%s\", comp-addr=%p, status=%s",
 213                         port_name->str, self_comp,
 214                         bt_common_func_status_string(status));
 215                 goto end;
 216         }
 217
 218         muxer_comp->available_input_ports++;
 219         muxer_comp->next_port_num++;
 220         BT_COMP_LOGI("Added one input port to muxer component: "
 221                 "port-name=\"%s\", comp-addr=%p",
 222                 port_name->str, self_comp);
 223
 224 end:
 225         if (port_name) {
 226                 g_string_free(port_name, TRUE);
 227         }
 228
 229         return status;
 230 }
 231
 232 static
 233 bt_self_component_add_port_status create_output_port(
 234                 bt_self_component_filter *self_comp)
 235 {
 236         return bt_self_component_filter_add_output_port(
 237                 self_comp, "out", NULL, NULL);
 238 }
 239
 240 static
 241 void destroy_muxer_comp(struct muxer_comp *muxer_comp)
 242 {
 243         if (!muxer_comp) {
 244                 return;
 245         }
 246
 247         g_free(muxer_comp);
 248 }
 249
 250 static
 251 bt_value *get_default_params(struct muxer_comp *muxer_comp)
 252 {
 253         bt_value *params;
 254         int ret;
 255
 256         params = bt_value_map_create();
 257         if (!params) {
 258                 BT_COMP_LOGE_STR("Cannot create a map value object.");
 259                 goto error;
 260         }
 261
 262         ret = bt_value_map_insert_bool_entry(params,
 263                 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, false);
 264         if (ret) {
 265                 BT_COMP_LOGE_STR("Cannot add boolean value to map value object.");
 266                 goto error;
 267         }
 268
 269         goto end;
 270
 271 error:
 272         BT_VALUE_PUT_REF_AND_RESET(params);
 273
 274 end:
 275         return params;
 276 }
 277
 278 static
 279 int configure_muxer_comp(struct muxer_comp *muxer_comp,
 280                 const bt_value *params)
 281 {
 282         bt_value *default_params = NULL;
 283         bt_value *real_params = NULL;
 284         const bt_value *assume_absolute_clock_classes = NULL;
 285         int ret = 0;
 286         bt_bool bool_val;
 287
 288         default_params = get_default_params(muxer_comp);
 289         if (!default_params) {
 290                 BT_COMP_LOGE("Cannot get default parameters: "
 291                         "muxer-comp-addr=%p", muxer_comp);
 292                 goto error;
 293         }
 294
 295         ret = bt_value_map_extend(default_params, params, &real_params);
 296         if (ret) {
 297                 BT_COMP_LOGE("Cannot extend default parameters map value: "
 298                         "muxer-comp-addr=%p, def-params-addr=%p, "
 299                         "params-addr=%p", muxer_comp, default_params,
 300                         params);
 301                 goto error;
 302         }
 303
 304         assume_absolute_clock_classes = bt_value_map_borrow_entry_value(real_params,
 305                                                                         ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME);
 306         if (assume_absolute_clock_classes &&
 307                         !bt_value_is_bool(assume_absolute_clock_classes)) {
 308                 BT_COMP_LOGE("Expecting a boolean value for the `%s` parameter: "
 309                         "muxer-comp-addr=%p, value-type=%s",
 310                         ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, muxer_comp,
 311                         bt_common_value_type_string(
 312                                 bt_value_get_type(assume_absolute_clock_classes)));
 313                 goto error;
 314         }
 315
 316         bool_val = bt_value_bool_get(assume_absolute_clock_classes);
 317         muxer_comp->assume_absolute_clock_classes = (bool) bool_val;
 318         BT_COMP_LOGI("Configured muxer component: muxer-comp-addr=%p, "
 319                 "assume-absolute-clock-classes=%d",
 320                 muxer_comp, muxer_comp->assume_absolute_clock_classes);
 321         goto end;
 322
 323 error:
 324         ret = -1;
 325
 326 end:
 327         bt_value_put_ref(default_params);
 328         bt_value_put_ref(real_params);
 329         return ret;
 330 }
 331
 332 BT_HIDDEN
 333 bt_component_class_init_method_status muxer_init(
 334                 bt_self_component_filter *self_comp_flt,
 335                 const bt_value *params, void *init_data)
 336 {
 337         int ret;
 338         bt_component_class_init_method_status status =
 339                 BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK;
 340         bt_self_component_add_port_status add_port_status;
 341         bt_self_component *self_comp =
 342                 bt_self_component_filter_as_self_component(self_comp_flt);
 343         struct muxer_comp *muxer_comp = g_new0(struct muxer_comp, 1);
 344         bt_logging_level log_level = bt_component_get_logging_level(
 345                 bt_self_component_as_component(self_comp));
 346
 347         BT_COMP_LOG_CUR_LVL(BT_LOG_INFO, log_level, self_comp,
 348                 "Initializing muxer component: "
 349                 "comp-addr=%p, params-addr=%p", self_comp, params);
 350
 351         if (!muxer_comp) {
 352                 BT_COMP_LOG_CUR_LVL(BT_LOG_ERROR, log_level, self_comp,
 353                         "Failed to allocate one muxer component.");
 354                 goto error;
 355         }
 356
 357         muxer_comp->log_level = log_level;
 358         muxer_comp->self_comp = self_comp;
 359         muxer_comp->self_comp_flt = self_comp_flt;
 360         ret = configure_muxer_comp(muxer_comp, params);
 361         if (ret) {
 362                 BT_COMP_LOGE("Cannot configure muxer component: "
 363                         "muxer-comp-addr=%p, params-addr=%p",
 364                         muxer_comp, params);
 365                 goto error;
 366         }
 367
 368         bt_self_component_set_data(self_comp, muxer_comp);
 369         add_port_status = add_available_input_port(self_comp_flt);
 370         if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 371                 BT_COMP_LOGE("Cannot ensure that at least one muxer component's input port is available: "
 372                         "muxer-comp-addr=%p, status=%s",
 373                         muxer_comp,
 374                         bt_common_func_status_string(add_port_status));
 375                 if (add_port_status ==
 376                                 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR) {
 377                         status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
 378                 } else {
 379                         status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
 380                 }
 381
 382                 goto error;
 383         }
 384
 385         add_port_status = create_output_port(self_comp_flt);
 386         if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 387                 BT_COMP_LOGE("Cannot create muxer component's output port: "
 388                         "muxer-comp-addr=%p, status=%s",
 389                         muxer_comp,
 390                         bt_common_func_status_string(add_port_status));
 391                 if (add_port_status ==
 392                                 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR) {
 393                         status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_MEMORY_ERROR;
 394                 } else {
 395                         status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
 396                 }
 397
 398                 goto error;
 399         }
 400
 401         BT_COMP_LOGI("Initialized muxer component: "
 402                 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
 403                 self_comp, params, muxer_comp);
 404
 405         goto end;
 406
 407 error:
 408         destroy_muxer_comp(muxer_comp);
 409         bt_self_component_set_data(self_comp, NULL);
 410
 411         if (status == BT_COMPONENT_CLASS_INIT_METHOD_STATUS_OK) {
 412                 status = BT_COMPONENT_CLASS_INIT_METHOD_STATUS_ERROR;
 413         }
 414
 415 end:
 416         return status;
 417 }
 418
 419 BT_HIDDEN
 420 void muxer_finalize(bt_self_component_filter *self_comp)
 421 {
 422         struct muxer_comp *muxer_comp = bt_self_component_get_data(
 423                 bt_self_component_filter_as_self_component(self_comp));
 424
 425         BT_COMP_LOGI("Finalizing muxer component: comp-addr=%p",
 426                 self_comp);
 427         destroy_muxer_comp(muxer_comp);
 428 }
 429
 430 static
 431 bt_self_component_port_input_message_iterator *
 432 create_msg_iter_on_input_port(struct muxer_comp *muxer_comp,
 433                 struct muxer_msg_iter *muxer_msg_iter,
 434                 bt_self_component_port_input *self_port)
 435 {
 436         const bt_port *port = bt_self_component_port_as_port(
 437                 bt_self_component_port_input_as_self_component_port(
 438                         self_port));
 439         bt_self_component_port_input_message_iterator *msg_iter =
 440                 NULL;
 441
 442         BT_ASSERT(port);
 443         BT_ASSERT(bt_port_is_connected(port));
 444
 445         // TODO: Advance the iterator to >= the time of the latest
 446         //       returned message by the muxer message
 447         //       iterator which creates it.
 448         msg_iter = bt_self_component_port_input_message_iterator_create_from_message_iterator(
 449                 muxer_msg_iter->self_msg_iter, self_port);
 450         if (!msg_iter) {
 451                 BT_COMP_LOGE("Cannot create upstream message iterator on input port: "
 452                         "port-addr=%p, port-name=\"%s\"",
 453                         port, bt_port_get_name(port));
 454                 goto end;
 455         }
 456
 457         BT_COMP_LOGI("Created upstream message iterator on input port: "
 458                 "port-addr=%p, port-name=\"%s\", msg-iter-addr=%p",
 459                 port, bt_port_get_name(port), msg_iter);
 460
 461 end:
 462         return msg_iter;
 463 }
 464
 465 static
 466 bt_component_class_message_iterator_next_method_status muxer_upstream_msg_iter_next(
 467                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
 468                 bool *is_ended)
 469 {
 470         struct muxer_comp *muxer_comp =
 471                 muxer_upstream_msg_iter->muxer_comp;
 472         bt_component_class_message_iterator_next_method_status status;
 473         bt_message_iterator_next_status input_port_iter_status;
 474         bt_message_array_const msgs;
 475         uint64_t i;
 476         uint64_t count;
 477
 478         BT_COMP_LOGD("Calling upstream message iterator's \"next\" method: "
 479                 "muxer-upstream-msg-iter-wrap-addr=%p, msg-iter-addr=%p",
 480                 muxer_upstream_msg_iter,
 481                 muxer_upstream_msg_iter->msg_iter);
 482         input_port_iter_status = bt_self_component_port_input_message_iterator_next(
 483                 muxer_upstream_msg_iter->msg_iter, &msgs, &count);
 484         BT_COMP_LOGD("Upstream message iterator's \"next\" method returned: "
 485                 "status=%s",
 486                 bt_common_func_status_string(input_port_iter_status));
 487
 488         switch (input_port_iter_status) {
 489         case BT_MESSAGE_ITERATOR_NEXT_STATUS_OK:
 490                 /*
 491                  * Message iterator's current message is
 492                  * valid: it must be considered for muxing operations.
 493                  */
 494                 BT_COMP_LOGD_STR("Validated upstream message iterator wrapper.");
 495                 BT_ASSERT(count > 0);
 496
 497                 /* Move messages to our queue */
 498                 for (i = 0; i < count; i++) {
 499                         /*
 500                          * Push to tail in order; other side
 501                          * (muxer_msg_iter_do_next_one()) consumes
 502                          * from the head first.
 503                          */
 504                         g_queue_push_tail(muxer_upstream_msg_iter->msgs,
 505                                 (void *) msgs[i]);
 506                 }
 507                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
 508                 break;
 509         case BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN:
 510                 /*
 511                  * Message iterator's current message is not
 512                  * valid anymore. Return
 513                  * BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN immediately.
 514                  */
 515                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
 516                 break;
 517         case BT_MESSAGE_ITERATOR_NEXT_STATUS_END:       /* Fall-through. */
 518                 /*
 519                  * Message iterator reached the end: release it. It
 520                  * won't be considered again to find the youngest
 521                  * message.
 522                  */
 523                 *is_ended = true;
 524                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
 525                 break;
 526         default:
 527                 /* Error or unsupported status code */
 528                 BT_COMP_LOGE("Error or unsupported status code: "
 529                         "status-code=%d", input_port_iter_status);
 530                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
 531                 break;
 532         }
 533
 534         return status;
 535 }
 536
 537 static
 538 int get_msg_ts_ns(struct muxer_comp *muxer_comp,
 539                 struct muxer_msg_iter *muxer_msg_iter,
 540                 const bt_message *msg, int64_t last_returned_ts_ns,
 541                 int64_t *ts_ns)
 542 {
 543         const bt_clock_snapshot *clock_snapshot = NULL;
 544         int ret = 0;
 545         const bt_stream_class *stream_class = NULL;
 546         bt_message_type msg_type;
 547
 548         BT_ASSERT(msg);
 549         BT_ASSERT(ts_ns);
 550         BT_COMP_LOGD("Getting message's timestamp: "
 551                 "muxer-msg-iter-addr=%p, msg-addr=%p, "
 552                 "last-returned-ts=%" PRId64,
 553                 muxer_msg_iter, msg, last_returned_ts_ns);
 554
 555         if (G_UNLIKELY(muxer_msg_iter->clock_class_expectation ==
 556                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE)) {
 557                 *ts_ns = last_returned_ts_ns;
 558                 goto end;
 559         }
 560
 561         msg_type = bt_message_get_type(msg);
 562
 563         if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_PACKET_BEGINNING)) {
 564                 stream_class = bt_stream_borrow_class_const(
 565                         bt_packet_borrow_stream_const(
 566                                 bt_message_packet_beginning_borrow_packet_const(
 567                                         msg)));
 568         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_PACKET_END)) {
 569                 stream_class = bt_stream_borrow_class_const(
 570                         bt_packet_borrow_stream_const(
 571                                 bt_message_packet_end_borrow_packet_const(
 572                                         msg)));
 573         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS)) {
 574                 stream_class = bt_stream_borrow_class_const(
 575                         bt_message_discarded_events_borrow_stream_const(msg));
 576         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_DISCARDED_PACKETS)) {
 577                 stream_class = bt_stream_borrow_class_const(
 578                         bt_message_discarded_packets_borrow_stream_const(msg));
 579         }
 580
 581         switch (msg_type) {
 582         case BT_MESSAGE_TYPE_EVENT:
 583                 BT_ASSERT(bt_message_event_borrow_stream_class_default_clock_class_const(
 584                                 msg));
 585                 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
 586                         msg);
 587                 break;
 588         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 589                 if (bt_stream_class_packets_have_beginning_default_clock_snapshot(
 590                                 stream_class)) {
 591                         clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
 592                                 msg);
 593                 } else {
 594                         goto no_clock_snapshot;
 595                 }
 596
 597                 break;
 598         case BT_MESSAGE_TYPE_PACKET_END:
 599                 if (bt_stream_class_packets_have_end_default_clock_snapshot(
 600                                 stream_class)) {
 601                         clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
 602                                 msg);
 603                 } else {
 604                         goto no_clock_snapshot;
 605                 }
 606
 607                 break;
 608         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 609         {
 610                 enum bt_message_stream_clock_snapshot_state snapshot_state =
 611                         bt_message_stream_beginning_borrow_default_clock_snapshot_const(
 612                                 msg, &clock_snapshot);
 613                 if (snapshot_state == BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN) {
 614                         goto no_clock_snapshot;
 615                 }
 616
 617                 break;
 618         }
 619         case BT_MESSAGE_TYPE_STREAM_END:
 620         {
 621                 enum bt_message_stream_clock_snapshot_state snapshot_state =
 622                         bt_message_stream_end_borrow_default_clock_snapshot_const(
 623                                 msg, &clock_snapshot);
 624                 if (snapshot_state == BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN) {
 625                         goto no_clock_snapshot;
 626                 }
 627
 628                 break;
 629         }
 630         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 631                 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
 632                                 stream_class)) {
 633                         clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
 634                                 msg);
 635                 } else {
 636                         goto no_clock_snapshot;
 637                 }
 638
 639                 break;
 640         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 641                 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
 642                                 stream_class)) {
 643                         clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
 644                                 msg);
 645                 } else {
 646                         goto no_clock_snapshot;
 647                 }
 648
 649                 break;
 650         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 651                 clock_snapshot = bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
 652                         msg);
 653                 break;
 654         default:
 655                 /* All the other messages have a higher priority */
 656                 BT_COMP_LOGD_STR("Message has no timestamp: using the last returned timestamp.");
 657                 *ts_ns = last_returned_ts_ns;
 658                 goto end;
 659         }
 660
 661         ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
 662         if (ret) {
 663                 BT_COMP_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
 664                         "clock-snapshot-addr=%p", clock_snapshot);
 665                 goto error;
 666         }
 667
 668         goto end;
 669
 670 no_clock_snapshot:
 671         BT_COMP_LOGD_STR("Message's default clock snapshot is missing: "
 672                 "using the last returned timestamp.");
 673         *ts_ns = last_returned_ts_ns;
 674         goto end;
 675
 676 error:
 677         ret = -1;
 678
 679 end:
 680         if (ret == 0) {
 681                 BT_COMP_LOGD("Found message's timestamp: "
 682                         "muxer-msg-iter-addr=%p, msg-addr=%p, "
 683                         "last-returned-ts=%" PRId64 ", ts=%" PRId64,
 684                         muxer_msg_iter, msg, last_returned_ts_ns,
 685                         *ts_ns);
 686         }
 687
 688         return ret;
 689 }
 690
 691 static inline
 692 int validate_clock_class(struct muxer_msg_iter *muxer_msg_iter,
 693                 struct muxer_comp *muxer_comp,
 694                 const bt_clock_class *clock_class)
 695 {
 696         int ret = 0;
 697         const uint8_t *cc_uuid;
 698         const char *cc_name;
 699
 700         BT_ASSERT(clock_class);
 701         cc_uuid = bt_clock_class_get_uuid(clock_class);
 702         cc_name = bt_clock_class_get_name(clock_class);
 703
 704         if (muxer_msg_iter->clock_class_expectation ==
 705                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 706                 /*
 707                  * This is the first clock class that this muxer
 708                  * message iterator encounters. Its properties
 709                  * determine what to expect for the whole lifetime of
 710                  * the iterator without a true
 711                  * `assume-absolute-clock-classes` parameter.
 712                  */
 713                 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 714                         /* Expect absolute clock classes */
 715                         muxer_msg_iter->clock_class_expectation =
 716                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE;
 717                 } else {
 718                         if (cc_uuid) {
 719                                 /*
 720                                  * Expect non-absolute clock classes
 721                                  * with a specific UUID.
 722                                  */
 723                                 muxer_msg_iter->clock_class_expectation =
 724                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID;
 725                                 bt_uuid_copy(muxer_msg_iter->expected_clock_class_uuid, cc_uuid);
 726                         } else {
 727                                 /*
 728                                  * Expect non-absolute clock classes
 729                                  * with no UUID.
 730                                  */
 731                                 muxer_msg_iter->clock_class_expectation =
 732                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID;
 733                         }
 734                 }
 735         }
 736
 737         if (!muxer_comp->assume_absolute_clock_classes) {
 738                 switch (muxer_msg_iter->clock_class_expectation) {
 739                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE:
 740                         if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
 741                                 BT_COMP_LOGE("Expecting an absolute clock class, "
 742                                         "but got a non-absolute one: "
 743                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 744                                         clock_class, cc_name);
 745                                 goto error;
 746                         }
 747                         break;
 748                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID:
 749                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 750                                 BT_COMP_LOGE("Expecting a non-absolute clock class with no UUID, "
 751                                         "but got an absolute one: "
 752                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 753                                         clock_class, cc_name);
 754                                 goto error;
 755                         }
 756
 757                         if (cc_uuid) {
 758                                 BT_COMP_LOGE("Expecting a non-absolute clock class with no UUID, "
 759                                         "but got one with a UUID: "
 760                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 761                                         "uuid=\"" BT_UUID_FMT "\"",
 762                                         clock_class, cc_name, BT_UUID_FMT_VALUES(cc_uuid));
 763                                 goto error;
 764                         }
 765                         break;
 766                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID:
 767                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 768                                 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 769                                         "but got an absolute one: "
 770                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 771                                         clock_class, cc_name);
 772                                 goto error;
 773                         }
 774
 775                         if (!cc_uuid) {
 776                                 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 777                                         "but got one with no UUID: "
 778                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 779                                         clock_class, cc_name);
 780                                 goto error;
 781                         }
 782
 783                         if (bt_uuid_compare(muxer_msg_iter->expected_clock_class_uuid, cc_uuid) != 0) {
 784                                 BT_COMP_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 785                                         "but got one with different UUID: "
 786                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 787                                         "expected-uuid=\"" BT_UUID_FMT "\", "
 788                                         "uuid=\"" BT_UUID_FMT "\"",
 789                                         clock_class, cc_name,
 790                                         BT_UUID_FMT_VALUES(muxer_msg_iter->expected_clock_class_uuid),
 791                                         BT_UUID_FMT_VALUES(cc_uuid));
 792                                 goto error;
 793                         }
 794                         break;
 795                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE:
 796                         BT_COMP_LOGE("Expecting no clock class, but got one: "
 797                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 798                                 clock_class, cc_name);
 799                         goto error;
 800                 default:
 801                         /* Unexpected */
 802                         BT_COMP_LOGF("Unexpected clock class expectation: "
 803                                 "expectation-code=%d",
 804                                 muxer_msg_iter->clock_class_expectation);
 805                         abort();
 806                 }
 807         }
 808
 809         goto end;
 810
 811 error:
 812         ret = -1;
 813
 814 end:
 815         return ret;
 816 }
 817
 818 static inline
 819 int validate_new_stream_clock_class(struct muxer_msg_iter *muxer_msg_iter,
 820                 struct muxer_comp *muxer_comp, const bt_stream *stream)
 821 {
 822         int ret = 0;
 823         const bt_stream_class *stream_class =
 824                 bt_stream_borrow_class_const(stream);
 825         const bt_clock_class *clock_class =
 826                 bt_stream_class_borrow_default_clock_class_const(stream_class);
 827
 828         if (!clock_class) {
 829                 if (muxer_msg_iter->clock_class_expectation ==
 830                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 831                         /* Expect no clock class */
 832                         muxer_msg_iter->clock_class_expectation =
 833                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE;
 834                 } else {
 835                         BT_COMP_LOGE("Expecting stream class with a default clock class: "
 836                                 "stream-class-addr=%p, stream-class-name=\"%s\", "
 837                                 "stream-class-id=%" PRIu64,
 838                                 stream_class, bt_stream_class_get_name(stream_class),
 839                                 bt_stream_class_get_id(stream_class));
 840                         ret = -1;
 841                 }
 842
 843                 goto end;
 844         }
 845
 846         ret = validate_clock_class(muxer_msg_iter, muxer_comp, clock_class);
 847
 848 end:
 849         return ret;
 850 }
 851
 852 struct message_to_compare {
 853         const bt_message *msg;
 854         const bt_trace *trace;
 855         const bt_stream *stream;
 856 };
 857
 858 struct messages_to_compare {
 859         struct message_to_compare left;
 860         struct message_to_compare right;
 861 };
 862
 863 static
 864 int message_type_weight(const bt_message_type msg_type)
 865 {
 866         int weight;
 867
 868         switch (msg_type) {
 869         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 870                 weight = 7;
 871                 break;
 872         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 873                 weight = 6;
 874                 break;
 875         case BT_MESSAGE_TYPE_EVENT:
 876                 weight = 5;
 877                 break;
 878         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 879                 weight = 4;
 880                 break;
 881         case BT_MESSAGE_TYPE_PACKET_END:
 882                 weight = 3;
 883                 break;
 884         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 885                 weight = 2;
 886                 break;
 887         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 888                 weight = 1;
 889                 break;
 890         case BT_MESSAGE_TYPE_STREAM_END:
 891                 weight = 0;
 892                 break;
 893         default:
 894                 abort();
 895         }
 896
 897         return weight;
 898 }
 899
 900 /*
 901  * Compare 2 messages to order them in a determinitic way based on their
 902  * types.
 903  * Returns -1 is left mesage must go first
 904  * Returns 1 is right mesage must go first
 905  */
 906 static
 907 int compare_messages_by_type(struct messages_to_compare *msgs)
 908 {
 909         bt_message_type left_msg_type = bt_message_get_type(msgs->left.msg);
 910         bt_message_type right_msg_type = bt_message_get_type(msgs->right.msg);
 911
 912         return message_type_weight(right_msg_type) -
 913                 message_type_weight(left_msg_type);
 914 }
 915
 916 static
 917 int compare_events(const bt_event *left_event, const bt_event *right_event)
 918 {
 919         int ret;
 920         const bt_event_class *left_event_class, *right_event_class;
 921         uint64_t left_event_class_id, right_event_class_id;
 922         const char *left_event_class_name, *right_event_class_name,
 923                         *left_event_class_emf_uri, *right_event_class_emf_uri;
 924         bt_event_class_log_level left_event_class_log_level, right_event_class_log_level;
 925         bt_property_availability left_log_level_avail, right_log_level_avail;
 926
 927         left_event_class = bt_event_borrow_class_const(left_event);
 928         right_event_class = bt_event_borrow_class_const(right_event);
 929
 930         left_event_class_id = bt_event_class_get_id(left_event_class);
 931         right_event_class_id = bt_event_class_get_id(right_event_class);
 932
 933         if (left_event_class_id > right_event_class_id) {
 934                 ret = 1;
 935                 goto end;
 936         } else if (left_event_class_id < right_event_class_id) {
 937                 ret = -1;
 938                 goto end;
 939         }
 940
 941         left_event_class_name = bt_event_class_get_name(left_event_class);
 942         right_event_class_name = bt_event_class_get_name(right_event_class);
 943         if (left_event_class_name && right_event_class_name) {
 944                 ret = strcmp(left_event_class_name, right_event_class_name);
 945                 if (ret != 0) {
 946                         goto end;
 947                 }
 948         } else if (!left_event_class_name && right_event_class_name) {
 949                 ret = -1;
 950                 goto end;
 951         } else if (left_event_class_name && !right_event_class_name) {
 952                 ret = 1;
 953                 goto end;
 954         }
 955
 956         left_log_level_avail = bt_event_class_get_log_level(left_event_class,
 957                 &left_event_class_log_level);
 958         right_log_level_avail = bt_event_class_get_log_level(right_event_class,
 959                 &right_event_class_log_level);
 960
 961         if (left_log_level_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE &&
 962                         right_log_level_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE) {
 963                 ret = left_event_class_log_level - right_event_class_log_level;
 964                 if (ret) {
 965                         goto end;
 966                 }
 967         } else if (left_log_level_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE &&
 968                         right_log_level_avail == BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE) {
 969                 ret = -1;
 970                 goto end;
 971         } else if (left_log_level_avail == BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE &&
 972                         right_log_level_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE) {
 973                 ret = 1;
 974                 goto end;
 975         }
 976
 977         left_event_class_emf_uri = bt_event_class_get_emf_uri(left_event_class);
 978         right_event_class_emf_uri = bt_event_class_get_emf_uri(right_event_class);
 979         if (left_event_class_emf_uri && right_event_class_emf_uri) {
 980                 ret = strcmp(left_event_class_emf_uri, right_event_class_emf_uri);
 981                 if (ret != 0) {
 982                         goto end;
 983                 }
 984         } else if (!left_event_class_emf_uri && right_event_class_emf_uri) {
 985                 ret = -1;
 986                 goto end;
 987         } else if (left_event_class_emf_uri && !right_event_class_emf_uri) {
 988                 ret = 1;
 989                 goto end;
 990         }
 991
 992 end:
 993         return ret;
 994 }
 995
 996 static
 997 int compare_streams(const bt_stream *left_stream, const bt_stream *right_stream)
 998 {
 999         int ret = 0;
1000         const char *left_stream_name, *right_stream_name,
1001               *left_stream_class_name, *right_stream_class_name;
1002         const bt_stream_class *left_stream_class, *right_stream_class;
1003
1004         /*
1005          * No need to compare stream id as it was checked earlier and if we are
1006          * here it means they are identical or both absent.
1007          */
1008         BT_ASSERT(bt_stream_get_id(left_stream) ==
1009                 bt_stream_get_id(right_stream));
1010
1011         /* Compare stream name. */
1012         left_stream_name = bt_stream_get_name(left_stream);
1013         right_stream_name = bt_stream_get_name(right_stream);
1014
1015         if (left_stream_name && right_stream_name) {
1016                 ret = strcmp(left_stream_name, right_stream_name);
1017                 if (ret != 0) {
1018                         goto end;
1019                 }
1020         } else if (!left_stream_name && right_stream_name) {
1021                 ret = -1;
1022                 goto end;
1023         } else if (left_stream_name && !right_stream_name) {
1024                 ret = 1;
1025                 goto end;
1026         }
1027
1028         left_stream_class = bt_stream_borrow_class_const(left_stream);
1029         right_stream_class = bt_stream_borrow_class_const(right_stream);
1030
1031         /*
1032          * No need to compare stream class id as it was checked earlier and if
1033          * we are here it means they are identical.
1034          */
1035         BT_ASSERT(bt_stream_class_get_id(left_stream_class) ==
1036                 bt_stream_class_get_id(right_stream_class));
1037
1038         /* Compare stream class name. */
1039         left_stream_class_name = bt_stream_class_get_name(left_stream_class);
1040         right_stream_class_name = bt_stream_class_get_name(right_stream_class);
1041
1042         if (left_stream_class_name && right_stream_class_name) {
1043                 ret = strcmp(left_stream_class_name, right_stream_class_name);
1044                 if (ret != 0) {
1045                         goto end;
1046                 }
1047         } else if (!left_stream_class_name && right_stream_class_name) {
1048                 ret = -1;
1049                 goto end;
1050         } else if (left_stream_class_name && !right_stream_class_name) {
1051                 ret = 1;
1052                 goto end;
1053         }
1054
1055         /* Compare stream class automatic event class id assignment. */
1056         if (bt_stream_class_assigns_automatic_event_class_id(left_stream_class) &&
1057                         !bt_stream_class_assigns_automatic_event_class_id(right_stream_class)) {
1058                 ret = 1;
1059                 goto end;
1060         } else if (!bt_stream_class_assigns_automatic_event_class_id(left_stream_class) &&
1061                         bt_stream_class_assigns_automatic_event_class_id(right_stream_class)) {
1062                 ret = -1;
1063                 goto end;
1064         }
1065
1066         /* Compare stream class automatic stream id assignment. */
1067         if (bt_stream_class_assigns_automatic_stream_id(left_stream_class) &&
1068                         !bt_stream_class_assigns_automatic_stream_id(right_stream_class)) {
1069                 ret = 1;
1070                 goto end;
1071         } else if (!bt_stream_class_assigns_automatic_stream_id(left_stream_class) &&
1072                         bt_stream_class_assigns_automatic_stream_id(right_stream_class)) {
1073                 ret = -1;
1074                 goto end;
1075         }
1076
1077         /* Compare stream class support of discarded events. */
1078         if (bt_stream_class_supports_discarded_events(left_stream_class) &&
1079                         !bt_stream_class_supports_discarded_events(right_stream_class)) {
1080                 ret = 1;
1081                 goto end;
1082         } else if (!bt_stream_class_supports_discarded_events(left_stream_class) &&
1083                         bt_stream_class_supports_discarded_events(right_stream_class)) {
1084                 ret = -1;
1085                 goto end;
1086         }
1087
1088         /* Compare stream class discarded events default clock snapshot. */
1089         if (bt_stream_class_discarded_events_have_default_clock_snapshots(left_stream_class) &&
1090                         !bt_stream_class_discarded_events_have_default_clock_snapshots(right_stream_class)) {
1091                 ret = 1;
1092                 goto end;
1093         } else if (!bt_stream_class_discarded_events_have_default_clock_snapshots(left_stream_class) &&
1094                         bt_stream_class_discarded_events_have_default_clock_snapshots(right_stream_class)) {
1095                 ret = -1;
1096                 goto end;
1097         }
1098
1099         /* Compare stream class support of packets. */
1100         if (bt_stream_class_supports_packets(left_stream_class) &&
1101                         !bt_stream_class_supports_packets(right_stream_class)) {
1102                 ret = 1;
1103                 goto end;
1104         } else if (!bt_stream_class_supports_packets(left_stream_class) &&
1105                         bt_stream_class_supports_packets(right_stream_class)) {
1106                 ret = -1;
1107                 goto end;
1108         }
1109
1110         if (!bt_stream_class_supports_packets(left_stream_class)) {
1111                 /* Skip all packet related checks. */
1112                 goto end;
1113         }
1114
1115         /*
1116          * Compare stream class presence of discarded packets beginning default
1117          * clock snapshot.
1118          */
1119         if (bt_stream_class_packets_have_beginning_default_clock_snapshot(left_stream_class) &&
1120                         !bt_stream_class_packets_have_beginning_default_clock_snapshot(right_stream_class)) {
1121                 ret = 1;
1122                 goto end;
1123         } else if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(left_stream_class) &&
1124                         bt_stream_class_packets_have_beginning_default_clock_snapshot(right_stream_class)) {
1125                 ret = -1;
1126                 goto end;
1127         }
1128
1129         /*
1130          * Compare stream class presence of discarded packets end default clock
1131          * snapshot.
1132          */
1133         if (bt_stream_class_packets_have_end_default_clock_snapshot(left_stream_class) &&
1134                         !bt_stream_class_packets_have_end_default_clock_snapshot(right_stream_class)) {
1135                 ret = 1;
1136                 goto end;
1137         } else if (!bt_stream_class_packets_have_end_default_clock_snapshot(left_stream_class) &&
1138                         bt_stream_class_packets_have_end_default_clock_snapshot(right_stream_class)) {
1139                 ret = -1;
1140                 goto end;
1141         }
1142
1143         /* Compare stream class support of discarded packets. */
1144         if (bt_stream_class_supports_discarded_packets(left_stream_class) &&
1145                         !bt_stream_class_supports_discarded_packets(right_stream_class)) {
1146                 ret = 1;
1147                 goto end;
1148         } else if (!bt_stream_class_supports_discarded_packets(left_stream_class) &&
1149                         bt_stream_class_supports_discarded_packets(right_stream_class)) {
1150                 ret = -1;
1151                 goto end;
1152         }
1153
1154         /* Compare stream class discarded packets default clock snapshot. */
1155         if (bt_stream_class_discarded_packets_have_default_clock_snapshots(left_stream_class) &&
1156                         !bt_stream_class_discarded_packets_have_default_clock_snapshots(right_stream_class)) {
1157                 ret = 1;
1158                 goto end;
1159         } else if (!bt_stream_class_discarded_packets_have_default_clock_snapshots(left_stream_class) &&
1160                         bt_stream_class_discarded_packets_have_default_clock_snapshots(right_stream_class)) {
1161                 ret = -1;
1162                 goto end;
1163         }
1164
1165 end:
1166         return ret;
1167 }
1168
1169 static
1170 int compare_clock_snapshots_and_clock_classes(const bt_clock_snapshot *left_cs,
1171                 const bt_clock_snapshot *right_cs)
1172 {
1173         int ret;
1174         uint64_t left_freq, right_freq, left_prec, right_prec;
1175         uint64_t left_cs_value, right_cs_value;
1176         const bt_clock_class *left_clock_class, *right_clock_class;
1177         const char *left_clock_class_name, *right_clock_class_name;
1178         left_cs_value = bt_clock_snapshot_get_value(left_cs);
1179         right_cs_value = bt_clock_snapshot_get_value(right_cs);
1180         bt_uuid left_clock_class_uuid, right_clock_class_uuid;
1181
1182         ret = left_cs_value - right_cs_value;
1183         if (ret != 0) {
1184                 goto end;
1185         }
1186
1187         left_clock_class = bt_clock_snapshot_borrow_clock_class_const(left_cs);
1188         right_clock_class = bt_clock_snapshot_borrow_clock_class_const(right_cs);
1189
1190         left_clock_class_uuid = bt_clock_class_get_uuid(left_clock_class);
1191         right_clock_class_uuid = bt_clock_class_get_uuid(right_clock_class);
1192
1193         if (left_clock_class_uuid && !right_clock_class_uuid) {
1194                 ret = -1;
1195                 goto end;
1196         } else if (!left_clock_class_uuid && right_clock_class_uuid) {
1197                 ret = 1;
1198                 goto end;
1199         } else if (left_clock_class_uuid && right_clock_class_uuid) {
1200                 ret = bt_uuid_compare(left_clock_class_uuid,
1201                         right_clock_class_uuid);
1202                 if (ret != 0) {
1203                         goto end;
1204                 }
1205         }
1206
1207
1208         left_clock_class_name = bt_clock_class_get_name(left_clock_class);
1209         right_clock_class_name = bt_clock_class_get_name(right_clock_class);
1210
1211         if (left_clock_class_name && !right_clock_class_name) {
1212                 ret = -1;
1213                 goto end;
1214         } else if (!left_clock_class_name && right_clock_class_name) {
1215                 ret = 1;
1216                 goto end;
1217         } else if (left_clock_class_name && right_clock_class_name) {
1218                 ret = strcmp(left_clock_class_name, right_clock_class_name);
1219                 if (ret != 0) {
1220                         goto end;
1221                 }
1222         }
1223
1224         left_freq = bt_clock_class_get_frequency(left_clock_class);
1225         right_freq = bt_clock_class_get_frequency(right_clock_class);
1226
1227         ret = right_freq - left_freq;
1228         if (ret != 0) {
1229                 goto end;
1230         }
1231
1232         left_prec = bt_clock_class_get_precision(left_clock_class);
1233         right_prec = bt_clock_class_get_precision(right_clock_class);
1234
1235         ret = right_prec - left_prec;
1236         if (ret != 0) {
1237                 goto end;
1238         }
1239
1240 end:
1241         return ret;
1242 }
1243
1244 static
1245 const bt_stream *borrow_stream(const bt_message *msg)
1246 {
1247         bt_message_type msg_type = bt_message_get_type(msg);
1248         const bt_stream *stream = NULL;
1249         const bt_packet *packet = NULL;
1250         const bt_event *event = NULL;
1251
1252         switch (msg_type) {
1253         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1254                 stream = bt_message_stream_beginning_borrow_stream_const(msg);
1255                 break;
1256         case BT_MESSAGE_TYPE_STREAM_END:
1257                 stream = bt_message_stream_end_borrow_stream_const(msg);
1258                 break;
1259         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1260                 packet = bt_message_packet_beginning_borrow_packet_const(msg);
1261                 stream = bt_packet_borrow_stream_const(packet);
1262                 break;
1263         case BT_MESSAGE_TYPE_PACKET_END:
1264                 packet = bt_message_packet_end_borrow_packet_const(msg);
1265                 stream = bt_packet_borrow_stream_const(packet);
1266                 break;
1267         case BT_MESSAGE_TYPE_EVENT:
1268                 event = bt_message_event_borrow_event_const(msg);
1269                 stream = bt_event_borrow_stream_const(event);
1270                 break;
1271         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1272                 stream = bt_message_discarded_events_borrow_stream_const(msg);
1273                 break;
1274         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1275                 stream = bt_message_discarded_packets_borrow_stream_const(msg);
1276                 break;
1277         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
1278                 goto end;
1279         default:
1280                 abort();
1281         }
1282
1283 end:
1284         return stream;
1285 }
1286
1287 static
1288 const bt_trace *borrow_trace(const bt_message *msg)
1289 {
1290         const bt_trace *trace = NULL;
1291         const bt_stream *stream = NULL;
1292
1293         stream = borrow_stream(msg);
1294         if (stream) {
1295                 trace = bt_stream_borrow_trace_const(stream);
1296         }
1297
1298         return trace;
1299 }
1300
1301 static
1302 int compare_messages_by_trace_name(struct messages_to_compare *msgs)
1303 {
1304         int ret = 0;
1305         const char *left_trace_name = NULL, *right_trace_name = NULL;
1306
1307         if (msgs->left.trace && !msgs->right.trace) {
1308                 ret = -1;
1309                 goto end;
1310         }
1311
1312         if (!msgs->left.trace && msgs->right.trace) {
1313                 ret = 1;
1314                 goto end;
1315         }
1316
1317         if (!msgs->left.trace && !msgs->right.trace) {
1318                 ret = 0;
1319                 goto end;
1320         }
1321
1322         left_trace_name = bt_trace_get_name(msgs->left.trace);
1323         right_trace_name = bt_trace_get_name(msgs->right.trace);
1324
1325         if (left_trace_name && !right_trace_name) {
1326                 ret = -1;
1327                 goto end;
1328         }
1329
1330         if (!left_trace_name && right_trace_name) {
1331                 ret = 1;
1332                 goto end;
1333         }
1334
1335         if (!left_trace_name && !right_trace_name) {
1336                 ret = 0;
1337                 goto end;
1338         }
1339
1340         ret = strcmp(left_trace_name, right_trace_name);
1341 end:
1342         return ret;
1343 }
1344
1345 static
1346 int compare_messages_by_trace_uuid(struct messages_to_compare *msgs)
1347 {
1348         int ret = 0;
1349         bt_uuid left_trace_uuid = NULL, right_trace_uuid = NULL;
1350
1351         if (msgs->left.trace && !msgs->right.trace) {
1352                 ret = -1;
1353                 goto end;
1354         }
1355
1356         if (!msgs->left.trace && msgs->right.trace) {
1357                 ret = 1;
1358                 goto end;
1359         }
1360
1361         if (!msgs->left.trace && !msgs->right.trace) {
1362                 ret = 0;
1363                 goto end;
1364         }
1365
1366         left_trace_uuid = bt_trace_get_uuid(msgs->left.trace);
1367         right_trace_uuid = bt_trace_get_uuid(msgs->right.trace);
1368
1369         if (left_trace_uuid && !right_trace_uuid) {
1370                 ret = -1;
1371                 goto end;
1372         }
1373
1374         if (!left_trace_uuid && right_trace_uuid) {
1375                 ret = 1;
1376                 goto end;
1377         }
1378
1379         if (!left_trace_uuid && !right_trace_uuid) {
1380                 ret = 0;
1381                 goto end;
1382         }
1383
1384         ret = bt_uuid_compare(left_trace_uuid, right_trace_uuid);
1385 end:
1386         return ret;
1387 }
1388
1389 static
1390 int compare_messages_by_stream_class_id(struct messages_to_compare *msgs)
1391 {
1392         int ret = 0;
1393         uint64_t left_stream_class_id = 0, right_stream_class_id = 0;
1394
1395         if (msgs->left.stream && !msgs->right.stream) {
1396                 ret = -1;
1397                 goto end;
1398         }
1399
1400         if (!msgs->left.stream && msgs->right.stream) {
1401                 ret = 1;
1402                 goto end;
1403         }
1404
1405         if (!msgs->left.stream && !msgs->right.stream) {
1406                 ret = 0;
1407                 goto end;
1408         }
1409
1410         left_stream_class_id = bt_stream_class_get_id(
1411                 bt_stream_borrow_class_const(msgs->left.stream));
1412
1413         right_stream_class_id = bt_stream_class_get_id(
1414                 bt_stream_borrow_class_const(msgs->right.stream));
1415
1416         if (left_stream_class_id == right_stream_class_id) {
1417                 ret = 0;
1418                 goto end;
1419         }
1420
1421         ret = (left_stream_class_id < right_stream_class_id) ? -1 : 1;
1422
1423 end:
1424         return ret;
1425 }
1426
1427 static
1428 int compare_messages_by_stream_id(struct messages_to_compare *msgs)
1429 {
1430         int ret = 0;
1431         uint64_t left_stream_id = 0, right_stream_id = 0;
1432
1433         if (msgs->left.stream && !msgs->right.stream) {
1434                 ret = -1;
1435                 goto end;
1436         }
1437
1438         if (!msgs->left.stream && msgs->right.stream) {
1439                 ret = 1;
1440                 goto end;
1441         }
1442
1443         if (!msgs->left.stream && !msgs->right.stream) {
1444                 ret = 0;
1445                 goto end;
1446         }
1447
1448         left_stream_id = bt_stream_get_id(msgs->left.stream);
1449         right_stream_id = bt_stream_get_id(msgs->right.stream);
1450
1451         if (left_stream_id == right_stream_id) {
1452                 ret = 0;
1453                 goto end;
1454         }
1455
1456         ret = (left_stream_id < right_stream_id) ? -1 : 1;
1457
1458 end:
1459         return ret;
1460 }
1461
1462 static
1463 int compare_messages_same_type(struct messages_to_compare *msgs)
1464 {
1465         int ret = 0;
1466
1467         /*
1468          * Both messages are of the same type, we must compare characterics of
1469          * the messages such as the attributes of the event in a event message.
1470          */
1471         BT_ASSERT(bt_message_get_type(msgs->left.msg) ==
1472                 bt_message_get_type(msgs->right.msg));
1473
1474         switch (bt_message_get_type(msgs->left.msg)) {
1475         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
1476                 ret = compare_streams(msgs->left.stream, msgs->right.stream);
1477                 if (ret) {
1478                         goto end;
1479                 }
1480         case BT_MESSAGE_TYPE_STREAM_END:
1481                 ret = compare_streams(msgs->left.stream, msgs->right.stream);
1482                 if (ret) {
1483                         goto end;
1484                 }
1485
1486                 break;
1487         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
1488                 ret = compare_streams(msgs->left.stream, msgs->right.stream);
1489                 if (ret) {
1490                         goto end;
1491                 }
1492
1493                 break;
1494         case BT_MESSAGE_TYPE_PACKET_END:
1495                 ret = compare_streams(msgs->left.stream, msgs->right.stream);
1496                 if (ret) {
1497                         goto end;
1498                 }
1499
1500                 break;
1501         case BT_MESSAGE_TYPE_EVENT:
1502         {
1503                 const bt_event *left_event, *right_event;
1504                 left_event = bt_message_event_borrow_event_const(msgs->left.msg);
1505                 right_event = bt_message_event_borrow_event_const(msgs->right.msg);
1506
1507                 ret = compare_events(left_event, right_event);
1508                 if (ret) {
1509                         goto end;
1510                 }
1511
1512                 ret = compare_streams(msgs->left.stream, msgs->right.stream);
1513                 if (ret) {
1514                         goto end;
1515                 }
1516                 break;
1517         }
1518         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1519         {
1520                 const bt_stream_class *left_stream_class;
1521                 bt_property_availability left_event_count_avail,
1522                                          right_event_count_avail;
1523                 uint64_t left_event_count, right_event_count;
1524
1525                 /*
1526                  * Compare streams first to check if there is a
1527                  * mismatch about discarded event related configuration
1528                  * in the stream class.
1529                  */
1530                 ret = compare_streams(msgs->left.stream, msgs->right.stream);
1531                 if (ret) {
1532                         goto end;
1533                 }
1534
1535                 left_stream_class = bt_stream_borrow_class_const(msgs->left.stream);
1536                 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
1537                                 left_stream_class)) {
1538                         const bt_clock_snapshot *left_beg_cs =
1539                                 bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(msgs->left.msg);
1540                         const bt_clock_snapshot *right_beg_cs =
1541                                 bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(msgs->right.msg);
1542                         const bt_clock_snapshot *left_end_cs =
1543                                 bt_message_discarded_events_borrow_end_default_clock_snapshot_const(msgs->left.msg);
1544                         const bt_clock_snapshot *right_end_cs =
1545                                 bt_message_discarded_events_borrow_end_default_clock_snapshot_const(msgs->right.msg);
1546
1547                         ret = compare_clock_snapshots_and_clock_classes(
1548                                 left_beg_cs, right_beg_cs);
1549                         if (ret) {
1550                                 goto end;
1551                         }
1552
1553                         ret = compare_clock_snapshots_and_clock_classes(
1554                                 left_end_cs, right_end_cs);
1555                         if (ret) {
1556                                 goto end;
1557                         }
1558                 }
1559
1560                 left_event_count_avail =
1561                         bt_message_discarded_events_get_count(
1562                                 msgs->left.msg, &left_event_count);
1563                 right_event_count_avail =
1564                         bt_message_discarded_events_get_count(
1565                                 msgs->right.msg, &right_event_count);
1566                 if (left_event_count_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE &&
1567                         right_event_count_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE) {
1568                         ret = left_event_count - right_event_count;
1569                         if (ret != 0) {
1570                                 goto end;
1571                         }
1572                 } else if (left_event_count_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE &&
1573                         right_event_count_avail == BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE) {
1574                         ret = -1;
1575                         goto end;
1576                 } else if (left_event_count_avail == BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE &&
1577                         right_event_count_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE) {
1578                         ret = 1;
1579                         goto end;
1580                 }
1581
1582                 break;
1583         }
1584         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1585         {
1586                 const bt_stream_class *left_stream_class;
1587                 bt_property_availability left_packet_count_avail,
1588                                          right_packet_count_avail;
1589                 uint64_t left_packet_count, right_packet_count;
1590
1591                 /*
1592                  * Compare streams first to check if there is a
1593                  * mismatch about discarded packets related
1594                  * configuration in the stream class.
1595                  */
1596                 ret = compare_streams(msgs->left.stream, msgs->right.stream);
1597                 if (ret) {
1598                         goto end;
1599                 }
1600
1601                 left_stream_class = bt_stream_borrow_class_const(msgs->left.stream);
1602
1603                 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
1604                                 left_stream_class)) {
1605                         const bt_clock_snapshot *left_beg_cs =
1606                                 bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(msgs->left.msg);
1607                         const bt_clock_snapshot *right_beg_cs =
1608                                 bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(msgs->right.msg);
1609                         const bt_clock_snapshot *left_end_cs =
1610                                 bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(msgs->left.msg);
1611                         const bt_clock_snapshot *right_end_cs =
1612                                 bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(msgs->right.msg);
1613
1614                         ret = compare_clock_snapshots_and_clock_classes(
1615                                 left_beg_cs, right_beg_cs);
1616                         if (ret) {
1617                                 goto end;
1618                         }
1619
1620                         ret = compare_clock_snapshots_and_clock_classes(
1621                                 left_end_cs, right_end_cs);
1622                         if (ret) {
1623                                 goto end;
1624                         }
1625                 }
1626
1627                 left_packet_count_avail = bt_message_discarded_packets_get_count(
1628                         msgs->left.msg, &left_packet_count);
1629                 right_packet_count_avail = bt_message_discarded_packets_get_count(
1630                         msgs->right.msg, &right_packet_count);
1631                 if (left_packet_count_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE &&
1632                         right_packet_count_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE) {
1633                         ret = left_packet_count - right_packet_count;
1634                         if (ret != 0) {
1635                                 goto end;
1636                         }
1637                 } else if (left_packet_count_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE &&
1638                         right_packet_count_avail == BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE) {
1639                         ret = -1;
1640                         goto end;
1641                 } else if (left_packet_count_avail == BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE &&
1642                         right_packet_count_avail == BT_PROPERTY_AVAILABILITY_AVAILABLE) {
1643                         ret = 1;
1644                         goto end;
1645                 }
1646
1647                 break;
1648         }
1649         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
1650         {
1651                 const bt_clock_snapshot *left_cs =
1652                         bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(msgs->left.msg);
1653                 const bt_clock_snapshot *right_cs =
1654                         bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(msgs->right.msg);
1655
1656                 ret = compare_clock_snapshots_and_clock_classes(
1657                         left_cs, right_cs);
1658                 if (ret != 0) {
1659                         goto end;
1660                 }
1661
1662                 break;
1663         }
1664         default:
1665                 abort();
1666         }
1667
1668 end:
1669         return ret;
1670 }
1671
1672 static
1673 int compare_messages(const bt_message *left_msg, const bt_message *right_msg)
1674 {
1675         int ret = 0;
1676         struct messages_to_compare msgs;
1677
1678         BT_ASSERT(left_msg != right_msg);
1679
1680         msgs.left.msg = left_msg;
1681         msgs.left.trace = borrow_trace(left_msg);
1682         msgs.left.stream = borrow_stream(left_msg);
1683
1684         msgs.right.msg = right_msg;
1685         msgs.right.trace = borrow_trace(right_msg);
1686         msgs.right.stream = borrow_stream(right_msg);
1687
1688         /* Same timestamp: compare trace names. */
1689         ret = compare_messages_by_trace_name(&msgs);
1690         if (ret) {
1691                 goto end;
1692         }
1693
1694         /* Same timestamp and trace name: compare trace UUIDs. */
1695         ret = compare_messages_by_trace_uuid(&msgs);
1696         if (ret) {
1697                 goto end;
1698         }
1699
1700         /*
1701          * Same timestamp, trace name, and trace UUID: compare stream class
1702          * IDs.
1703          */
1704         ret = compare_messages_by_stream_class_id(&msgs);
1705         if (ret) {
1706                 goto end;
1707         }
1708
1709         /*
1710          * Same timestamp, trace name, trace UUID, and stream class ID: compare
1711          * stream IDs.
1712          */
1713         ret = compare_messages_by_stream_id(&msgs);
1714         if (ret) {
1715                 goto end;
1716         }
1717
1718         if (bt_message_get_type(msgs.left.msg) !=
1719                         bt_message_get_type(msgs.right.msg)) {
1720                 /*
1721                  * The messages are of different type, we order (arbitrarily)
1722                  * in the following way:
1723                  * SB < PB < EV < DE < MI < PE < DP < SE
1724                  */
1725                 ret = compare_messages_by_type(&msgs);
1726                 if (ret) {
1727                         goto end;
1728                 }
1729         } else {
1730                 /* The messages are of the same type. */
1731                 ret = compare_messages_same_type(&msgs);
1732                 if (ret) {
1733                         goto end;
1734                 }
1735         }
1736
1737 end:
1738         return ret;
1739 }
1740
1741 /*
1742  * This function finds the youngest available message amongst the
1743  * non-ended upstream message iterators and returns the upstream
1744  * message iterator which has it, or
1745  * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
1746  * message.
1747  *
1748  * This function does NOT:
1749  *
1750  * * Update any upstream message iterator.
1751  * * Check the upstream message iterators to retry.
1752  *
1753  * On sucess, this function sets *muxer_upstream_msg_iter to the
1754  * upstream message iterator of which the current message is
1755  * the youngest, and sets *ts_ns to its time.
1756  */
1757 static
1758 bt_component_class_message_iterator_next_method_status
1759 muxer_msg_iter_youngest_upstream_msg_iter(
1760                 struct muxer_comp *muxer_comp,
1761                 struct muxer_msg_iter *muxer_msg_iter,
1762                 struct muxer_upstream_msg_iter **muxer_upstream_msg_iter,
1763                 int64_t *ts_ns)
1764 {
1765         size_t i;
1766         int ret;
1767         int64_t youngest_ts_ns = INT64_MAX;
1768         bt_component_class_message_iterator_next_method_status status =
1769                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1770
1771         BT_ASSERT(muxer_comp);
1772         BT_ASSERT(muxer_msg_iter);
1773         BT_ASSERT(muxer_upstream_msg_iter);
1774         *muxer_upstream_msg_iter = NULL;
1775
1776         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
1777                         i++) {
1778                 const bt_message *msg;
1779                 struct muxer_upstream_msg_iter *cur_muxer_upstream_msg_iter =
1780                         g_ptr_array_index(
1781                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
1782                                 i);
1783                 int64_t msg_ts_ns;
1784
1785                 if (!cur_muxer_upstream_msg_iter->msg_iter) {
1786                         /* This upstream message iterator is ended */
1787                         BT_COMP_LOGT("Skipping ended upstream message iterator: "
1788                                 "muxer-upstream-msg-iter-wrap-addr=%p",
1789                                 cur_muxer_upstream_msg_iter);
1790                         continue;
1791                 }
1792
1793                 BT_ASSERT(cur_muxer_upstream_msg_iter->msgs->length > 0);
1794                 msg = g_queue_peek_head(cur_muxer_upstream_msg_iter->msgs);
1795                 BT_ASSERT(msg);
1796
1797                 if (G_UNLIKELY(bt_message_get_type(msg) ==
1798                                 BT_MESSAGE_TYPE_STREAM_BEGINNING)) {
1799                         ret = validate_new_stream_clock_class(
1800                                 muxer_msg_iter, muxer_comp,
1801                                 bt_message_stream_beginning_borrow_stream_const(
1802                                         msg));
1803                         if (ret) {
1804                                 /*
1805                                  * validate_new_stream_clock_class() logs
1806                                  * errors.
1807                                  */
1808                                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1809                                 goto end;
1810                         }
1811                 } else if (G_UNLIKELY(bt_message_get_type(msg) ==
1812                                 BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY)) {
1813                         const bt_clock_snapshot *cs;
1814
1815                         cs = bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
1816                                 msg);
1817                         ret = validate_clock_class(muxer_msg_iter, muxer_comp,
1818                                 bt_clock_snapshot_borrow_clock_class_const(cs));
1819                         if (ret) {
1820                                 /* validate_clock_class() logs errors */
1821                                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1822                                 goto end;
1823                         }
1824                 }
1825
1826                 ret = get_msg_ts_ns(muxer_comp, muxer_msg_iter, msg,
1827                         muxer_msg_iter->last_returned_ts_ns, &msg_ts_ns);
1828                 if (ret) {
1829                         /* get_msg_ts_ns() logs errors */
1830                         *muxer_upstream_msg_iter = NULL;
1831                         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
1832                         goto end;
1833                 }
1834
1835                 if (msg_ts_ns < youngest_ts_ns) {
1836                         *muxer_upstream_msg_iter =
1837                                 cur_muxer_upstream_msg_iter;
1838                         youngest_ts_ns = msg_ts_ns;
1839                         *ts_ns = youngest_ts_ns;
1840                 } else if (msg_ts_ns == youngest_ts_ns) {
1841                         /*
1842                          * The currently selected message to be sent downstream
1843                          * next has the exact same timestamp that of the
1844                          * current candidate message. We must break the tie
1845                          * in a predictable manner.
1846                          */
1847                         const bt_message *selected_msg = g_queue_peek_head(
1848                                 (*muxer_upstream_msg_iter)->msgs);
1849                         BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
1850
1851                         /*
1852                          * Order the messages in an arbitrary but determinitic
1853                          * way.
1854                          */
1855                         ret = compare_messages(msg, selected_msg);
1856                         if (ret < 0) {
1857                                 /*
1858                                  * The `msg` should go first. Update the next
1859                                  * iterator and the current timestamp.
1860                                  */
1861                                 *muxer_upstream_msg_iter =
1862                                         cur_muxer_upstream_msg_iter;
1863                                 youngest_ts_ns = msg_ts_ns;
1864                                 *ts_ns = youngest_ts_ns;
1865                         } else if (ret == 0) {
1866                                 /* Unable to pick which one should go first. */
1867                                 BT_COMP_LOGW("Cannot deterministically pick next upstream message iterator because they have identical next messages: "
1868                                         "muxer-upstream-msg-iter-wrap-addr=%p"
1869                                         "cur-muxer-upstream-msg-iter-wrap-addr=%p",
1870                                         *muxer_upstream_msg_iter,
1871                                         cur_muxer_upstream_msg_iter);
1872                         }
1873                 }
1874         }
1875
1876         if (!*muxer_upstream_msg_iter) {
1877                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
1878                 *ts_ns = INT64_MIN;
1879         }
1880
1881 end:
1882         return status;
1883 }
1884
1885 static
1886 bt_component_class_message_iterator_next_method_status
1887 validate_muxer_upstream_msg_iter(
1888         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
1889         bool *is_ended)
1890 {
1891         struct muxer_comp *muxer_comp =
1892                 muxer_upstream_msg_iter->muxer_comp;
1893         bt_component_class_message_iterator_next_method_status status =
1894                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1895
1896         BT_COMP_LOGD("Validating muxer's upstream message iterator wrapper: "
1897                 "muxer-upstream-msg-iter-wrap-addr=%p",
1898                 muxer_upstream_msg_iter);
1899
1900         if (muxer_upstream_msg_iter->msgs->length > 0 ||
1901                         !muxer_upstream_msg_iter->msg_iter) {
1902                 BT_COMP_LOGD("Already valid or not considered: "
1903                         "queue-len=%u, upstream-msg-iter-addr=%p",
1904                         muxer_upstream_msg_iter->msgs->length,
1905                         muxer_upstream_msg_iter->msg_iter);
1906                 goto end;
1907         }
1908
1909         /* muxer_upstream_msg_iter_next() logs details/errors */
1910         status = muxer_upstream_msg_iter_next(muxer_upstream_msg_iter,
1911                 is_ended);
1912
1913 end:
1914         return status;
1915 }
1916
1917 static
1918 bt_component_class_message_iterator_next_method_status
1919 validate_muxer_upstream_msg_iters(
1920                 struct muxer_msg_iter *muxer_msg_iter)
1921 {
1922         struct muxer_comp *muxer_comp = muxer_msg_iter->muxer_comp;
1923         bt_component_class_message_iterator_next_method_status status =
1924                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
1925         size_t i;
1926
1927         BT_COMP_LOGD("Validating muxer's upstream message iterator wrappers: "
1928                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
1929
1930         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
1931                         i++) {
1932                 bool is_ended = false;
1933                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
1934                         g_ptr_array_index(
1935                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
1936                                 i);
1937
1938                 status = validate_muxer_upstream_msg_iter(
1939                         muxer_upstream_msg_iter, &is_ended);
1940                 if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1941                         if (status < 0) {
1942                                 BT_COMP_LOGE("Cannot validate muxer's upstream message iterator wrapper: "
1943                                         "muxer-msg-iter-addr=%p, "
1944                                         "muxer-upstream-msg-iter-wrap-addr=%p",
1945                                         muxer_msg_iter,
1946                                         muxer_upstream_msg_iter);
1947                         } else {
1948                                 BT_COMP_LOGD("Cannot validate muxer's upstream message iterator wrapper: "
1949                                         "muxer-msg-iter-addr=%p, "
1950                                         "muxer-upstream-msg-iter-wrap-addr=%p",
1951                                         muxer_msg_iter,
1952                                         muxer_upstream_msg_iter);
1953                         }
1954
1955                         goto end;
1956                 }
1957
1958                 /*
1959                  * Move this muxer upstream message iterator to the
1960                  * array of ended iterators if it's ended.
1961                  */
1962                 if (G_UNLIKELY(is_ended)) {
1963                         BT_COMP_LOGD("Muxer's upstream message iterator wrapper: ended or canceled: "
1964                                 "muxer-msg-iter-addr=%p, "
1965                                 "muxer-upstream-msg-iter-wrap-addr=%p",
1966                                 muxer_msg_iter, muxer_upstream_msg_iter);
1967                         g_ptr_array_add(
1968                                 muxer_msg_iter->ended_muxer_upstream_msg_iters,
1969                                 muxer_upstream_msg_iter);
1970                         muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i] = NULL;
1971
1972                         /*
1973                          * Use g_ptr_array_remove_fast() because the
1974                          * order of those elements is not important.
1975                          */
1976                         g_ptr_array_remove_index_fast(
1977                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
1978                                 i);
1979                         i--;
1980                 }
1981         }
1982
1983 end:
1984         return status;
1985 }
1986
1987 static inline
1988 bt_component_class_message_iterator_next_method_status muxer_msg_iter_do_next_one(
1989                 struct muxer_comp *muxer_comp,
1990                 struct muxer_msg_iter *muxer_msg_iter,
1991                 const bt_message **msg)
1992 {
1993         bt_component_class_message_iterator_next_method_status status;
1994         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter = NULL;
1995         int64_t next_return_ts;
1996
1997         status = validate_muxer_upstream_msg_iters(muxer_msg_iter);
1998         if (status != BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
1999                 /* validate_muxer_upstream_msg_iters() logs details */
2000                 goto end;
2001         }
2002
2003         /*
2004          * At this point we know that all the existing upstream
2005          * message iterators are valid. We can find the one,
2006          * amongst those, of which the current message is the
2007          * youngest.
2008          */
2009         status = muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp,
2010                         muxer_msg_iter, &muxer_upstream_msg_iter,
2011                         &next_return_ts);
2012         if (status < 0 || status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END) {
2013                 if (status < 0) {
2014                         BT_COMP_LOGE("Cannot find the youngest upstream message iterator wrapper: "
2015                                 "status=%s",
2016                                 bt_common_func_status_string(status));
2017                 } else {
2018                         BT_COMP_LOGD("Cannot find the youngest upstream message iterator wrapper: "
2019                                 "status=%s",
2020                                 bt_common_func_status_string(status));
2021                 }
2022
2023                 goto end;
2024         }
2025
2026         if (next_return_ts < muxer_msg_iter->last_returned_ts_ns) {
2027                 BT_COMP_LOGE("Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
2028                         "muxer-msg-iter-addr=%p, ts=%" PRId64 ", "
2029                         "last-returned-ts=%" PRId64,
2030                         muxer_msg_iter, next_return_ts,
2031                         muxer_msg_iter->last_returned_ts_ns);
2032                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
2033                 goto end;
2034         }
2035
2036         BT_COMP_LOGD("Found youngest upstream message iterator wrapper: "
2037                 "muxer-msg-iter-addr=%p, "
2038                 "muxer-upstream-msg-iter-wrap-addr=%p, "
2039                 "ts=%" PRId64,
2040                 muxer_msg_iter, muxer_upstream_msg_iter, next_return_ts);
2041         BT_ASSERT(status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK);
2042         BT_ASSERT(muxer_upstream_msg_iter);
2043
2044         /*
2045          * Consume from the queue's head: other side
2046          * (muxer_upstream_msg_iter_next()) writes to the tail.
2047          */
2048         *msg = g_queue_pop_head(muxer_upstream_msg_iter->msgs);
2049         BT_ASSERT(*msg);
2050         muxer_msg_iter->last_returned_ts_ns = next_return_ts;
2051
2052 end:
2053         return status;
2054 }
2055
2056 static
2057 bt_component_class_message_iterator_next_method_status muxer_msg_iter_do_next(
2058                 struct muxer_comp *muxer_comp,
2059                 struct muxer_msg_iter *muxer_msg_iter,
2060                 bt_message_array_const msgs, uint64_t capacity,
2061                 uint64_t *count)
2062 {
2063         bt_component_class_message_iterator_next_method_status status =
2064                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
2065         uint64_t i = 0;
2066
2067         while (i < capacity && status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
2068                 status = muxer_msg_iter_do_next_one(muxer_comp,
2069                         muxer_msg_iter, &msgs[i]);
2070                 if (status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
2071                         i++;
2072                 }
2073         }
2074
2075         if (i > 0) {
2076                 /*
2077                  * Even if muxer_msg_iter_do_next_one() returned
2078                  * something else than
2079                  * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
2080                  * message objects in the output message
2081                  * array, so we need to return
2082                  * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
2083                  * transfered to downstream. This other status occurs
2084                  * again the next time muxer_msg_iter_do_next() is
2085                  * called, possibly without any accumulated
2086                  * message, in which case we'll return it.
2087                  */
2088                 *count = i;
2089                 status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
2090         }
2091
2092         return status;
2093 }
2094
2095 static
2096 void destroy_muxer_msg_iter(struct muxer_msg_iter *muxer_msg_iter)
2097 {
2098         struct muxer_comp *muxer_comp;
2099
2100         if (!muxer_msg_iter) {
2101                 return;
2102         }
2103
2104         muxer_comp = muxer_msg_iter->muxer_comp;
2105         BT_COMP_LOGD("Destroying muxer component's message iterator: "
2106                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
2107
2108         if (muxer_msg_iter->active_muxer_upstream_msg_iters) {
2109                 BT_COMP_LOGD_STR("Destroying muxer's active upstream message iterator wrappers.");
2110                 g_ptr_array_free(
2111                         muxer_msg_iter->active_muxer_upstream_msg_iters, TRUE);
2112         }
2113
2114         if (muxer_msg_iter->ended_muxer_upstream_msg_iters) {
2115                 BT_COMP_LOGD_STR("Destroying muxer's ended upstream message iterator wrappers.");
2116                 g_ptr_array_free(
2117                         muxer_msg_iter->ended_muxer_upstream_msg_iters, TRUE);
2118         }
2119
2120         g_free(muxer_msg_iter);
2121 }
2122
2123 static
2124 int muxer_msg_iter_init_upstream_iterators(struct muxer_comp *muxer_comp,
2125                 struct muxer_msg_iter *muxer_msg_iter)
2126 {
2127         int64_t count;
2128         int64_t i;
2129         int ret = 0;
2130
2131         count = bt_component_filter_get_input_port_count(
2132                 bt_self_component_filter_as_component_filter(
2133                         muxer_comp->self_comp_flt));
2134         if (count < 0) {
2135                 BT_COMP_LOGD("No input port to initialize for muxer component's message iterator: "
2136                         "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
2137                         muxer_comp, muxer_msg_iter);
2138                 goto end;
2139         }
2140
2141         for (i = 0; i < count; i++) {
2142                 bt_self_component_port_input_message_iterator *upstream_msg_iter;
2143                 bt_self_component_port_input *self_port =
2144                         bt_self_component_filter_borrow_input_port_by_index(
2145                                 muxer_comp->self_comp_flt, i);
2146                 const bt_port *port;
2147
2148                 BT_ASSERT(self_port);
2149                 port = bt_self_component_port_as_port(
2150                         bt_self_component_port_input_as_self_component_port(
2151                                 self_port));
2152                 BT_ASSERT(port);
2153
2154                 if (!bt_port_is_connected(port)) {
2155                         /* Skip non-connected port */
2156                         continue;
2157                 }
2158
2159                 upstream_msg_iter = create_msg_iter_on_input_port(muxer_comp,
2160                         muxer_msg_iter, self_port);
2161                 if (!upstream_msg_iter) {
2162                         /* create_msg_iter_on_input_port() logs errors */
2163                         BT_ASSERT(!upstream_msg_iter);
2164                         ret = -1;
2165                         goto end;
2166                 }
2167
2168                 ret = muxer_msg_iter_add_upstream_msg_iter(muxer_msg_iter,
2169                         upstream_msg_iter);
2170                 bt_self_component_port_input_message_iterator_put_ref(
2171                         upstream_msg_iter);
2172                 if (ret) {
2173                         /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
2174                         goto end;
2175                 }
2176         }
2177
2178 end:
2179         return ret;
2180 }
2181
2182 BT_HIDDEN
2183 bt_component_class_message_iterator_init_method_status muxer_msg_iter_init(
2184                 bt_self_message_iterator *self_msg_iter,
2185                 bt_self_component_filter *self_comp,
2186                 bt_self_component_port_output *port)
2187 {
2188         struct muxer_comp *muxer_comp = NULL;
2189         struct muxer_msg_iter *muxer_msg_iter = NULL;
2190         bt_component_class_message_iterator_init_method_status status =
2191                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_OK;
2192         int ret;
2193
2194         muxer_comp = bt_self_component_get_data(
2195                 bt_self_component_filter_as_self_component(self_comp));
2196         BT_ASSERT(muxer_comp);
2197         BT_COMP_LOGD("Initializing muxer component's message iterator: "
2198                 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
2199                 self_comp, muxer_comp, self_msg_iter);
2200
2201         if (muxer_comp->initializing_muxer_msg_iter) {
2202                 /*
2203                  * Weird, unhandled situation detected: downstream
2204                  * creates a muxer message iterator while creating
2205                  * another muxer message iterator (same component).
2206                  */
2207                 BT_COMP_LOGE("Recursive initialization of muxer component's message iterator: "
2208                         "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
2209                         self_comp, muxer_comp, self_msg_iter);
2210                 goto error;
2211         }
2212
2213         muxer_comp->initializing_muxer_msg_iter = true;
2214         muxer_msg_iter = g_new0(struct muxer_msg_iter, 1);
2215         if (!muxer_msg_iter) {
2216                 BT_COMP_LOGE_STR("Failed to allocate one muxer component's message iterator.");
2217                 goto error;
2218         }
2219
2220         muxer_msg_iter->muxer_comp = muxer_comp;
2221         muxer_msg_iter->self_msg_iter = self_msg_iter;
2222         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
2223         muxer_msg_iter->active_muxer_upstream_msg_iters =
2224                 g_ptr_array_new_with_free_func(
2225                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
2226         if (!muxer_msg_iter->active_muxer_upstream_msg_iters) {
2227                 BT_COMP_LOGE_STR("Failed to allocate a GPtrArray.");
2228                 goto error;
2229         }
2230
2231         muxer_msg_iter->ended_muxer_upstream_msg_iters =
2232                 g_ptr_array_new_with_free_func(
2233                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
2234         if (!muxer_msg_iter->ended_muxer_upstream_msg_iters) {
2235                 BT_COMP_LOGE_STR("Failed to allocate a GPtrArray.");
2236                 goto error;
2237         }
2238
2239         ret = muxer_msg_iter_init_upstream_iterators(muxer_comp,
2240                 muxer_msg_iter);
2241         if (ret) {
2242                 BT_COMP_LOGE("Cannot initialize connected input ports for muxer component's message iterator: "
2243                         "comp-addr=%p, muxer-comp-addr=%p, "
2244                         "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
2245                         self_comp, muxer_comp, muxer_msg_iter,
2246                         self_msg_iter, ret);
2247                 goto error;
2248         }
2249
2250         bt_self_message_iterator_set_data(self_msg_iter, muxer_msg_iter);
2251         BT_COMP_LOGD("Initialized muxer component's message iterator: "
2252                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
2253                 "msg-iter-addr=%p",
2254                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
2255         goto end;
2256
2257 error:
2258         destroy_muxer_msg_iter(muxer_msg_iter);
2259         bt_self_message_iterator_set_data(self_msg_iter, NULL);
2260         status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_ERROR;
2261
2262 end:
2263         muxer_comp->initializing_muxer_msg_iter = false;
2264         return status;
2265 }
2266
2267 BT_HIDDEN
2268 void muxer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
2269 {
2270         struct muxer_msg_iter *muxer_msg_iter =
2271                 bt_self_message_iterator_get_data(self_msg_iter);
2272         bt_self_component *self_comp = NULL;
2273         struct muxer_comp *muxer_comp = NULL;
2274
2275         self_comp = bt_self_message_iterator_borrow_component(
2276                 self_msg_iter);
2277         BT_ASSERT(self_comp);
2278         muxer_comp = bt_self_component_get_data(self_comp);
2279         BT_COMP_LOGD("Finalizing muxer component's message iterator: "
2280                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
2281                 "msg-iter-addr=%p",
2282                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
2283
2284         if (muxer_msg_iter) {
2285                 destroy_muxer_msg_iter(muxer_msg_iter);
2286         }
2287 }
2288
2289 BT_HIDDEN
2290 bt_component_class_message_iterator_next_method_status muxer_msg_iter_next(
2291                 bt_self_message_iterator *self_msg_iter,
2292                 bt_message_array_const msgs, uint64_t capacity,
2293                 uint64_t *count)
2294 {
2295         bt_component_class_message_iterator_next_method_status status;
2296         struct muxer_msg_iter *muxer_msg_iter =
2297                 bt_self_message_iterator_get_data(self_msg_iter);
2298         bt_self_component *self_comp = NULL;
2299         struct muxer_comp *muxer_comp = NULL;
2300
2301         BT_ASSERT(muxer_msg_iter);
2302         self_comp = bt_self_message_iterator_borrow_component(
2303                 self_msg_iter);
2304         BT_ASSERT(self_comp);
2305         muxer_comp = bt_self_component_get_data(self_comp);
2306         BT_ASSERT(muxer_comp);
2307         BT_COMP_LOGT("Muxer component's message iterator's \"next\" method called: "
2308                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
2309                 "msg-iter-addr=%p",
2310                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
2311
2312         status = muxer_msg_iter_do_next(muxer_comp, muxer_msg_iter,
2313                 msgs, capacity, count);
2314         if (status < 0) {
2315                 BT_COMP_LOGE("Cannot get next message: "
2316                         "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
2317                         "msg-iter-addr=%p, status=%s",
2318                         self_comp, muxer_comp, muxer_msg_iter, self_msg_iter,
2319                         bt_common_func_status_string(status));
2320         } else {
2321                 BT_COMP_LOGT("Returning from muxer component's message iterator's \"next\" method: "
2322                         "status=%s",
2323                         bt_common_func_status_string(status));
2324         }
2325
2326         return status;
2327 }
2328
2329 BT_HIDDEN
2330 bt_component_class_port_connected_method_status muxer_input_port_connected(
2331                 bt_self_component_filter *self_comp,
2332                 bt_self_component_port_input *self_port,
2333                 const bt_port_output *other_port)
2334 {
2335         bt_component_class_port_connected_method_status status =
2336                 BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_OK;
2337         bt_self_component_add_port_status add_port_status;
2338         struct muxer_comp *muxer_comp = bt_self_component_get_data(
2339                 bt_self_component_filter_as_self_component(self_comp));
2340
2341         add_port_status = add_available_input_port(self_comp);
2342         if (add_port_status) {
2343                 BT_COMP_LOGE("Cannot add one muxer component's input port: "
2344                         "status=%s",
2345                         bt_common_func_status_string(status));
2346
2347                 if (add_port_status ==
2348                                 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR) {
2349                         status = BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_MEMORY_ERROR;
2350                 } else {
2351                         status = BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_ERROR;
2352                 }
2353
2354                 goto end;
2355         }
2356
2357 end:
2358         return status;
2359 }
2360
2361 static inline
2362 bt_bool muxer_upstream_msg_iters_can_all_seek_beginning(
2363                 GPtrArray *muxer_upstream_msg_iters)
2364 {
2365         uint64_t i;
2366         bt_bool ret = BT_TRUE;
2367
2368         for (i = 0; i < muxer_upstream_msg_iters->len; i++) {
2369                 struct muxer_upstream_msg_iter *upstream_msg_iter =
2370                         muxer_upstream_msg_iters->pdata[i];
2371
2372                 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
2373                                 upstream_msg_iter->msg_iter)) {
2374                         ret = BT_FALSE;
2375                         goto end;
2376                 }
2377         }
2378
2379 end:
2380         return ret;
2381 }
2382
2383 BT_HIDDEN
2384 bt_bool muxer_msg_iter_can_seek_beginning(
2385                 bt_self_message_iterator *self_msg_iter)
2386 {
2387         struct muxer_msg_iter *muxer_msg_iter =
2388                 bt_self_message_iterator_get_data(self_msg_iter);
2389         bt_bool ret = BT_TRUE;
2390
2391         if (!muxer_upstream_msg_iters_can_all_seek_beginning(
2392                         muxer_msg_iter->active_muxer_upstream_msg_iters)) {
2393                 ret = BT_FALSE;
2394                 goto end;
2395         }
2396
2397         if (!muxer_upstream_msg_iters_can_all_seek_beginning(
2398                         muxer_msg_iter->ended_muxer_upstream_msg_iters)) {
2399                 ret = BT_FALSE;
2400                 goto end;
2401         }
2402
2403 end:
2404         return ret;
2405 }
2406
2407 BT_HIDDEN
2408 bt_component_class_message_iterator_seek_beginning_method_status muxer_msg_iter_seek_beginning(
2409                 bt_self_message_iterator *self_msg_iter)
2410 {
2411         struct muxer_msg_iter *muxer_msg_iter =
2412                 bt_self_message_iterator_get_data(self_msg_iter);
2413         bt_component_class_message_iterator_seek_beginning_method_status status =
2414                 BT_COMPONENT_CLASS_MESSAGE_ITERATOR_SEEK_BEGINNING_METHOD_STATUS_OK;
2415         bt_message_iterator_seek_beginning_status seek_beg_status;
2416         uint64_t i;
2417
2418         /* Seek all ended upstream iterators first */
2419         for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
2420                         i++) {
2421                 struct muxer_upstream_msg_iter *upstream_msg_iter =
2422                         muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
2423
2424                 seek_beg_status = bt_self_component_port_input_message_iterator_seek_beginning(
2425                         upstream_msg_iter->msg_iter);
2426                 if (seek_beg_status != BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK) {
2427                         status = (int) seek_beg_status;
2428                         goto end;
2429                 }
2430
2431                 empty_message_queue(upstream_msg_iter);
2432         }
2433
2434         /* Seek all previously active upstream iterators */
2435         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
2436                         i++) {
2437                 struct muxer_upstream_msg_iter *upstream_msg_iter =
2438                         muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i];
2439
2440                 seek_beg_status = bt_self_component_port_input_message_iterator_seek_beginning(
2441                         upstream_msg_iter->msg_iter);
2442                 if (seek_beg_status != BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK) {
2443                         status = (int) seek_beg_status;
2444                         goto end;
2445                 }
2446
2447                 empty_message_queue(upstream_msg_iter);
2448         }
2449
2450         /* Make them all active */
2451         for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
2452                         i++) {
2453                 struct muxer_upstream_msg_iter *upstream_msg_iter =
2454                         muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
2455
2456                 g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
2457                         upstream_msg_iter);
2458                 muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i] = NULL;
2459         }
2460
2461         /*
2462          * GLib < 2.48.0 asserts when g_ptr_array_remove_range() is
2463          * called on an empty array.
2464          */
2465         if (muxer_msg_iter->ended_muxer_upstream_msg_iters->len > 0) {
2466                 g_ptr_array_remove_range(muxer_msg_iter->ended_muxer_upstream_msg_iters,
2467                         0, muxer_msg_iter->ended_muxer_upstream_msg_iters->len);
2468         }
2469         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
2470         muxer_msg_iter->clock_class_expectation =
2471                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY;
2472
2473 end:
2474         return status;
2475 }