src/plugins/utils/muxer/muxer.c

   1 /*
   2  * SPDX-License-Identifier: MIT
   3  *
   4  * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
   5  */
   6
   7 #define BT_COMP_LOG_SELF_COMP (muxer_comp->self_comp)
   8 #define BT_LOG_OUTPUT_LEVEL (muxer_comp->log_level)
   9 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.MUXER"
  10 #include "logging/comp-logging.h"
  11
  12 #include "common/macros.h"
  13 #include "common/uuid.h"
  14 #include <babeltrace2/babeltrace.h>
  15 #include <glib.h>
  16 #include <stdbool.h>
  17 #include <inttypes.h>
  18 #include "common/assert.h"
  19 #include "common/common.h"
  20 #include <stdlib.h>
  21 #include <string.h>
  22
  23 #include "plugins/common/muxing/muxing.h"
  24 #include "plugins/common/param-validation/param-validation.h"
  25
  26 #include "muxer.h"
  27
  28 struct muxer_comp {
  29         /* Weak refs */
  30         bt_self_component_filter *self_comp_flt;
  31         bt_self_component *self_comp;
  32
  33         unsigned int next_port_num;
  34         size_t available_input_ports;
  35         bool initializing_muxer_msg_iter;
  36         bt_logging_level log_level;
  37 };
  38
  39 struct muxer_upstream_msg_iter {
  40         struct muxer_comp *muxer_comp;
  41
  42         /* Owned by this, NULL if ended */
  43         bt_message_iterator *msg_iter;
  44
  45         /* Contains `const bt_message *`, owned by this */
  46         GPtrArray *msgs;
  47
  48         /* Index of the next message in `msgs` to return */
  49         guint next_msg;
  50 };
  51
  52 enum muxer_msg_iter_clock_class_expectation {
  53         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY = 0,
  54         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE,
  55         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE,
  56         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID,
  57         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID,
  58 };
  59
  60 struct muxer_msg_iter {
  61         struct muxer_comp *muxer_comp;
  62
  63         /* Weak */
  64         bt_self_message_iterator *self_msg_iter;
  65
  66         /*
  67          * Array of struct muxer_upstream_msg_iter * (owned by this).
  68          *
  69          * NOTE: This array is searched in linearly to find the youngest
  70          * current message. Keep this until benchmarks confirm that
  71          * another data structure is faster than this for our typical
  72          * use cases.
  73          */
  74         GPtrArray *active_muxer_upstream_msg_iters;
  75
  76         /*
  77          * Array of struct muxer_upstream_msg_iter * (owned by this).
  78          *
  79          * We move ended message iterators from
  80          * `active_muxer_upstream_msg_iters` to this array so as to be
  81          * able to restore them when seeking.
  82          */
  83         GPtrArray *ended_muxer_upstream_msg_iters;
  84
  85         /* Last time returned in a message */
  86         int64_t last_returned_ts_ns;
  87
  88         /* Clock class expectation state */
  89         enum muxer_msg_iter_clock_class_expectation clock_class_expectation;
  90
  91         /*
  92          * Expected clock class UUID, only valid when
  93          * clock_class_expectation is
  94          * MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID.
  95          */
  96         bt_uuid_t expected_clock_class_uuid;
  97
  98         /*
  99          * Saved error.  If we hit an error in the _next method, but have some
 100          * messages ready to return, we save the error here and return it on
 101          * the next _next call.
 102          */
 103         bt_message_iterator_class_next_method_status next_saved_status;
 104         const struct bt_error *next_saved_error;
 105 };
 106
 107 static
 108 void empty_message_queue(struct muxer_upstream_msg_iter *upstream_msg_iter)
 109 {
 110         g_ptr_array_set_size(upstream_msg_iter->msgs, 0);
 111 }
 112
 113 static
 114 void destroy_muxer_upstream_msg_iter(
 115                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
 116 {
 117         struct muxer_comp *muxer_comp;
 118
 119         if (!muxer_upstream_msg_iter) {
 120                 return;
 121         }
 122
 123         muxer_comp = muxer_upstream_msg_iter->muxer_comp;
 124         BT_COMP_LOGD("Destroying muxer's upstream message iterator wrapper: "
 125                 "addr=%p, msg-iter-addr=%p, queue-len=%u, next-msg=%u",
 126                 muxer_upstream_msg_iter,
 127                 muxer_upstream_msg_iter->msg_iter,
 128                 muxer_upstream_msg_iter->msgs->len,
 129                 muxer_upstream_msg_iter->next_msg);
 130
 131         bt_message_iterator_put_ref(
 132                 muxer_upstream_msg_iter->msg_iter);
 133
 134         if (muxer_upstream_msg_iter->msgs) {
 135                 g_ptr_array_free(muxer_upstream_msg_iter->msgs, TRUE);
 136         }
 137
 138         g_free(muxer_upstream_msg_iter);
 139 }
 140
 141 static
 142 int muxer_msg_iter_add_upstream_msg_iter(struct muxer_msg_iter *muxer_msg_iter,
 143                 bt_message_iterator *self_msg_iter)
 144 {
 145         int ret = 0;
 146         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
 147                 g_new0(struct muxer_upstream_msg_iter, 1);
 148         struct muxer_comp *muxer_comp = muxer_msg_iter->muxer_comp;
 149
 150         if (!muxer_upstream_msg_iter) {
 151                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 152                         "Failed to allocate one muxer's upstream message iterator wrapper.");
 153                 goto error;
 154         }
 155
 156         muxer_upstream_msg_iter->muxer_comp = muxer_comp;
 157         muxer_upstream_msg_iter->msg_iter = self_msg_iter;
 158         bt_message_iterator_get_ref(muxer_upstream_msg_iter->msg_iter);
 159         muxer_upstream_msg_iter->msgs =
 160                 g_ptr_array_new_with_free_func((GDestroyNotify) bt_message_put_ref);
 161         if (!muxer_upstream_msg_iter->msgs) {
 162                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 163                         "Failed to allocate a GPtrArray.");
 164                 goto error;
 165         }
 166
 167         g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
 168                 muxer_upstream_msg_iter);
 169         BT_COMP_LOGD("Added muxer's upstream message iterator wrapper: "
 170                 "addr=%p, muxer-msg-iter-addr=%p, msg-iter-addr=%p",
 171                 muxer_upstream_msg_iter, muxer_msg_iter,
 172                 self_msg_iter);
 173
 174         goto end;
 175
 176 error:
 177         destroy_muxer_upstream_msg_iter(muxer_upstream_msg_iter);
 178         ret = -1;
 179
 180 end:
 181         return ret;
 182 }
 183
 184 static
 185 bt_self_component_add_port_status add_available_input_port(
 186                 bt_self_component_filter *self_comp)
 187 {
 188         struct muxer_comp *muxer_comp = bt_self_component_get_data(
 189                 bt_self_component_filter_as_self_component(self_comp));
 190         bt_self_component_add_port_status status =
 191                 BT_SELF_COMPONENT_ADD_PORT_STATUS_OK;
 192         GString *port_name = NULL;
 193
 194         BT_ASSERT(muxer_comp);
 195         port_name = g_string_new("in");
 196         if (!port_name) {
 197                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp, "Failed to allocate a GString.");
 198                 status = BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR;
 199                 goto end;
 200         }
 201
 202         g_string_append_printf(port_name, "%u", muxer_comp->next_port_num);
 203         status = bt_self_component_filter_add_input_port(
 204                 self_comp, port_name->str, NULL, NULL);
 205         if (status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 206                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 207                         "Cannot add input port to muxer component: "
 208                         "port-name=\"%s\", comp-addr=%p, status=%s",
 209                         port_name->str, self_comp,
 210                         bt_common_func_status_string(status));
 211                 goto end;
 212         }
 213
 214         muxer_comp->available_input_ports++;
 215         muxer_comp->next_port_num++;
 216         BT_COMP_LOGI("Added one input port to muxer component: "
 217                 "port-name=\"%s\", comp-addr=%p",
 218                 port_name->str, self_comp);
 219
 220 end:
 221         if (port_name) {
 222                 g_string_free(port_name, TRUE);
 223         }
 224
 225         return status;
 226 }
 227
 228 static
 229 bt_self_component_add_port_status create_output_port(
 230                 bt_self_component_filter *self_comp)
 231 {
 232         return bt_self_component_filter_add_output_port(
 233                 self_comp, "out", NULL, NULL);
 234 }
 235
 236 static
 237 void destroy_muxer_comp(struct muxer_comp *muxer_comp)
 238 {
 239         if (!muxer_comp) {
 240                 return;
 241         }
 242
 243         g_free(muxer_comp);
 244 }
 245
 246 static
 247 struct bt_param_validation_map_value_entry_descr muxer_params[] = {
 248         BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
 249 };
 250
 251 bt_component_class_initialize_method_status muxer_init(
 252                 bt_self_component_filter *self_comp_flt,
 253                 bt_self_component_filter_configuration *config __attribute__((unused)),
 254                 const bt_value *params,
 255                 void *init_data __attribute__((unused)))
 256 {
 257         bt_component_class_initialize_method_status status;
 258         bt_self_component_add_port_status add_port_status;
 259         bt_self_component *self_comp =
 260                 bt_self_component_filter_as_self_component(self_comp_flt);
 261         struct muxer_comp *muxer_comp = g_new0(struct muxer_comp, 1);
 262         bt_logging_level log_level = bt_component_get_logging_level(
 263                 bt_self_component_as_component(self_comp));
 264         enum bt_param_validation_status validation_status;
 265         gchar *validate_error = NULL;
 266
 267         BT_COMP_LOG_CUR_LVL(BT_LOG_INFO, log_level, self_comp,
 268                 "Initializing muxer component: "
 269                 "comp-addr=%p, params-addr=%p", self_comp, params);
 270
 271         if (!muxer_comp) {
 272                 /*
 273                  * Don't use BT_COMP_LOGE_APPEND_CAUSE, as `muxer_comp` is not
 274                  * initialized.
 275                  */
 276                 BT_COMP_LOG_CUR_LVL(BT_LOG_ERROR, log_level, self_comp,
 277                         "Failed to allocate one muxer component.");
 278                 BT_CURRENT_THREAD_ERROR_APPEND_CAUSE_FROM_COMPONENT(self_comp,
 279                         "Failed to allocate one muxer component.");
 280                         status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
 281                 goto error;
 282         }
 283
 284         muxer_comp->log_level = log_level;
 285         muxer_comp->self_comp = self_comp;
 286         muxer_comp->self_comp_flt = self_comp_flt;
 287
 288         validation_status = bt_param_validation_validate(params,
 289                 muxer_params, &validate_error);
 290         if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
 291                 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
 292                 goto error;
 293         } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
 294                 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
 295                 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "%s", validate_error);
 296                 goto error;
 297         }
 298
 299         bt_self_component_set_data(self_comp, muxer_comp);
 300         add_port_status = add_available_input_port(self_comp_flt);
 301         if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 302                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 303                         "Cannot ensure that at least one muxer component's input port is available: "
 304                         "muxer-comp-addr=%p, status=%s",
 305                         muxer_comp, bt_common_func_status_string(add_port_status));
 306                 status = (int) add_port_status;
 307                 goto error;
 308         }
 309
 310         add_port_status = create_output_port(self_comp_flt);
 311         if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 312                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 313                         "Cannot create muxer component's output port: "
 314                         "muxer-comp-addr=%p, status=%s",
 315                         muxer_comp, bt_common_func_status_string(add_port_status));
 316                 status = (int) add_port_status;
 317                 goto error;
 318         }
 319
 320         BT_COMP_LOGI("Initialized muxer component: "
 321                 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
 322                 self_comp, params, muxer_comp);
 323
 324         status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
 325         goto end;
 326
 327 error:
 328         destroy_muxer_comp(muxer_comp);
 329         bt_self_component_set_data(self_comp, NULL);
 330
 331 end:
 332         g_free(validate_error);
 333         return status;
 334 }
 335
 336 void muxer_finalize(bt_self_component_filter *self_comp)
 337 {
 338         struct muxer_comp *muxer_comp = bt_self_component_get_data(
 339                 bt_self_component_filter_as_self_component(self_comp));
 340
 341         BT_COMP_LOGI("Finalizing muxer component: comp-addr=%p",
 342                 self_comp);
 343         destroy_muxer_comp(muxer_comp);
 344 }
 345
 346 static
 347 bt_message_iterator_create_from_message_iterator_status
 348 create_msg_iter_on_input_port(struct muxer_comp *muxer_comp,
 349                 struct muxer_msg_iter *muxer_msg_iter,
 350                 bt_self_component_port_input *self_port,
 351                 bt_message_iterator **msg_iter)
 352 {
 353         const bt_port *port = bt_self_component_port_as_port(
 354                 bt_self_component_port_input_as_self_component_port(
 355                         self_port));
 356         bt_message_iterator_create_from_message_iterator_status
 357                 status;
 358
 359         BT_ASSERT(port);
 360         BT_ASSERT(bt_port_is_connected(port));
 361
 362         // TODO: Advance the iterator to >= the time of the latest
 363         //       returned message by the muxer message
 364         //       iterator which creates it.
 365         status = bt_message_iterator_create_from_message_iterator(
 366                 muxer_msg_iter->self_msg_iter, self_port, msg_iter);
 367         if (status != BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK) {
 368                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 369                         "Cannot create upstream message iterator on input port: "
 370                         "port-addr=%p, port-name=\"%s\"",
 371                         port, bt_port_get_name(port));
 372                 goto end;
 373         }
 374
 375         BT_COMP_LOGI("Created upstream message iterator on input port: "
 376                 "port-addr=%p, port-name=\"%s\", msg-iter-addr=%p",
 377                 port, bt_port_get_name(port), msg_iter);
 378
 379 end:
 380         return status;
 381 }
 382
 383 static
 384 bt_message_iterator_class_next_method_status muxer_upstream_msg_iter_next(
 385                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
 386                 bool *is_ended)
 387 {
 388         struct muxer_comp *muxer_comp = muxer_upstream_msg_iter->muxer_comp;
 389         bt_message_iterator_class_next_method_status status;
 390         bt_message_iterator_next_status input_port_iter_status;
 391         bt_message_array_const msgs;
 392         uint64_t i;
 393         uint64_t count;
 394
 395         BT_COMP_LOGD("Calling upstream message iterator's \"next\" method: "
 396                 "muxer-upstream-msg-iter-wrap-addr=%p, msg-iter-addr=%p",
 397                 muxer_upstream_msg_iter,
 398                 muxer_upstream_msg_iter->msg_iter);
 399         input_port_iter_status = bt_message_iterator_next(
 400                 muxer_upstream_msg_iter->msg_iter, &msgs, &count);
 401         BT_COMP_LOGD("Upstream message iterator's \"next\" method returned: "
 402                 "status=%s",
 403                 bt_common_func_status_string(input_port_iter_status));
 404
 405         switch (input_port_iter_status) {
 406         case BT_MESSAGE_ITERATOR_NEXT_STATUS_OK:
 407                 /*
 408                  * Message iterator's current message is
 409                  * valid: it must be considered for muxing operations.
 410                  */
 411                 BT_COMP_LOGD_STR("Validated upstream message iterator wrapper.");
 412                 BT_ASSERT_DBG(count > 0);
 413
 414                 g_ptr_array_set_size(muxer_upstream_msg_iter->msgs, count);
 415                 muxer_upstream_msg_iter->next_msg = 0;
 416
 417                 /* Move messages to our queue */
 418                 for (i = 0; i < count; i++) {
 419                         /*
 420                          * Push to tail in order; other side
 421                          * (muxer_msg_iter_do_next_one()) consumes
 422                          * from the head first.
 423                          */
 424                         g_ptr_array_index(muxer_upstream_msg_iter->msgs, i)
 425                                 = (gpointer *) msgs[i];
 426                 }
 427                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
 428                 break;
 429         case BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN:
 430                 /*
 431                  * Message iterator's current message is not
 432                  * valid anymore. Return
 433                  * BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN immediately.
 434                  */
 435                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
 436                 break;
 437         case BT_MESSAGE_ITERATOR_NEXT_STATUS_END:       /* Fall-through. */
 438                 /*
 439                  * Message iterator reached the end: release it. It
 440                  * won't be considered again to find the youngest
 441                  * message.
 442                  */
 443                 *is_ended = true;
 444                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
 445                 break;
 446         case BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR:
 447         case BT_MESSAGE_ITERATOR_NEXT_STATUS_MEMORY_ERROR:
 448                 /* Error status code */
 449                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 450                         "Upstream iterator's next method returned an error: status=%s",
 451                         bt_common_func_status_string(input_port_iter_status));
 452                 status = (int) input_port_iter_status;
 453                 break;
 454         default:
 455                 /* Unsupported status code */
 456                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 457                         "Unsupported status code: status=%s",
 458                         bt_common_func_status_string(input_port_iter_status));
 459                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
 460                 break;
 461         }
 462
 463         return status;
 464 }
 465
 466 static
 467 int get_msg_ts_ns(struct muxer_comp *muxer_comp,
 468                 struct muxer_msg_iter *muxer_msg_iter,
 469                 const bt_message *msg, int64_t last_returned_ts_ns,
 470                 int64_t *ts_ns)
 471 {
 472         const bt_clock_snapshot *clock_snapshot = NULL;
 473         int ret = 0;
 474         const bt_stream_class *stream_class = NULL;
 475         bt_message_type msg_type;
 476
 477         BT_ASSERT_DBG(msg);
 478         BT_ASSERT_DBG(ts_ns);
 479         BT_COMP_LOGD("Getting message's timestamp: "
 480                 "muxer-msg-iter-addr=%p, msg-addr=%p, "
 481                 "last-returned-ts=%" PRId64,
 482                 muxer_msg_iter, msg, last_returned_ts_ns);
 483
 484         if (G_UNLIKELY(muxer_msg_iter->clock_class_expectation ==
 485                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE)) {
 486                 *ts_ns = last_returned_ts_ns;
 487                 goto end;
 488         }
 489
 490         msg_type = bt_message_get_type(msg);
 491
 492         if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_PACKET_BEGINNING)) {
 493                 stream_class = bt_stream_borrow_class_const(
 494                         bt_packet_borrow_stream_const(
 495                                 bt_message_packet_beginning_borrow_packet_const(
 496                                         msg)));
 497         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_PACKET_END)) {
 498                 stream_class = bt_stream_borrow_class_const(
 499                         bt_packet_borrow_stream_const(
 500                                 bt_message_packet_end_borrow_packet_const(
 501                                         msg)));
 502         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS)) {
 503                 stream_class = bt_stream_borrow_class_const(
 504                         bt_message_discarded_events_borrow_stream_const(msg));
 505         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_DISCARDED_PACKETS)) {
 506                 stream_class = bt_stream_borrow_class_const(
 507                         bt_message_discarded_packets_borrow_stream_const(msg));
 508         }
 509
 510         switch (msg_type) {
 511         case BT_MESSAGE_TYPE_EVENT:
 512                 BT_ASSERT_DBG(bt_message_event_borrow_stream_class_default_clock_class_const(
 513                                 msg));
 514                 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
 515                         msg);
 516                 break;
 517         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 518                 if (bt_stream_class_packets_have_beginning_default_clock_snapshot(
 519                                 stream_class)) {
 520                         clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
 521                                 msg);
 522                 } else {
 523                         goto no_clock_snapshot;
 524                 }
 525
 526                 break;
 527         case BT_MESSAGE_TYPE_PACKET_END:
 528                 if (bt_stream_class_packets_have_end_default_clock_snapshot(
 529                                 stream_class)) {
 530                         clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
 531                                 msg);
 532                 } else {
 533                         goto no_clock_snapshot;
 534                 }
 535
 536                 break;
 537         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 538         {
 539                 enum bt_message_stream_clock_snapshot_state snapshot_state =
 540                         bt_message_stream_beginning_borrow_default_clock_snapshot_const(
 541                                 msg, &clock_snapshot);
 542                 if (snapshot_state == BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN) {
 543                         goto no_clock_snapshot;
 544                 }
 545
 546                 break;
 547         }
 548         case BT_MESSAGE_TYPE_STREAM_END:
 549         {
 550                 enum bt_message_stream_clock_snapshot_state snapshot_state =
 551                         bt_message_stream_end_borrow_default_clock_snapshot_const(
 552                                 msg, &clock_snapshot);
 553                 if (snapshot_state == BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN) {
 554                         goto no_clock_snapshot;
 555                 }
 556
 557                 break;
 558         }
 559         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 560                 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
 561                                 stream_class)) {
 562                         clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
 563                                 msg);
 564                 } else {
 565                         goto no_clock_snapshot;
 566                 }
 567
 568                 break;
 569         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 570                 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
 571                                 stream_class)) {
 572                         clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
 573                                 msg);
 574                 } else {
 575                         goto no_clock_snapshot;
 576                 }
 577
 578                 break;
 579         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 580                 clock_snapshot = bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
 581                         msg);
 582                 break;
 583         default:
 584                 /* All the other messages have a higher priority */
 585                 BT_COMP_LOGD_STR("Message has no timestamp: using the last returned timestamp.");
 586                 *ts_ns = last_returned_ts_ns;
 587                 goto end;
 588         }
 589
 590         ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
 591         if (ret) {
 592                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 593                         "Cannot get nanoseconds from Epoch of clock snapshot: "
 594                         "clock-snapshot-addr=%p", clock_snapshot);
 595                 goto error;
 596         }
 597
 598         goto end;
 599
 600 no_clock_snapshot:
 601         BT_COMP_LOGD_STR("Message's default clock snapshot is missing: "
 602                 "using the last returned timestamp.");
 603         *ts_ns = last_returned_ts_ns;
 604         goto end;
 605
 606 error:
 607         ret = -1;
 608
 609 end:
 610         if (ret == 0) {
 611                 BT_COMP_LOGD("Found message's timestamp: "
 612                         "muxer-msg-iter-addr=%p, msg-addr=%p, "
 613                         "last-returned-ts=%" PRId64 ", ts=%" PRId64,
 614                         muxer_msg_iter, msg, last_returned_ts_ns,
 615                         *ts_ns);
 616         }
 617
 618         return ret;
 619 }
 620
 621 static inline
 622 int validate_clock_class(struct muxer_msg_iter *muxer_msg_iter,
 623                 struct muxer_comp *muxer_comp,
 624                 const bt_clock_class *clock_class)
 625 {
 626         int ret = 0;
 627         const uint8_t *cc_uuid;
 628         const char *cc_name;
 629
 630         BT_ASSERT_DBG(clock_class);
 631         cc_uuid = bt_clock_class_get_uuid(clock_class);
 632         cc_name = bt_clock_class_get_name(clock_class);
 633
 634         if (muxer_msg_iter->clock_class_expectation ==
 635                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 636                 /*
 637                  * This is the first clock class that this muxer message
 638                  * iterator encounters. Its properties determine what to expect
 639                  * for the whole lifetime of the iterator.
 640                  */
 641                 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 642                         /* Expect absolute clock classes */
 643                         muxer_msg_iter->clock_class_expectation =
 644                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE;
 645                 } else {
 646                         if (cc_uuid) {
 647                                 /*
 648                                  * Expect non-absolute clock classes
 649                                  * with a specific UUID.
 650                                  */
 651                                 muxer_msg_iter->clock_class_expectation =
 652                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID;
 653                                 bt_uuid_copy(muxer_msg_iter->expected_clock_class_uuid, cc_uuid);
 654                         } else {
 655                                 /*
 656                                  * Expect non-absolute clock classes
 657                                  * with no UUID.
 658                                  */
 659                                 muxer_msg_iter->clock_class_expectation =
 660                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID;
 661                         }
 662                 }
 663         }
 664
 665         switch (muxer_msg_iter->clock_class_expectation) {
 666         case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE:
 667                 if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
 668                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 669                                 "Expecting an absolute clock class, "
 670                                 "but got a non-absolute one: "
 671                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 672                                 clock_class, cc_name);
 673                         goto error;
 674                 }
 675                 break;
 676         case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID:
 677                 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 678                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 679                                 "Expecting a non-absolute clock class with no UUID, "
 680                                 "but got an absolute one: "
 681                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 682                                 clock_class, cc_name);
 683                         goto error;
 684                 }
 685
 686                 if (cc_uuid) {
 687                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 688                                 "Expecting a non-absolute clock class with no UUID, "
 689                                 "but got one with a UUID: "
 690                                 "clock-class-addr=%p, clock-class-name=\"%s\", "
 691                                 "uuid=\"" BT_UUID_FMT "\"",
 692                                 clock_class, cc_name, BT_UUID_FMT_VALUES(cc_uuid));
 693                         goto error;
 694                 }
 695                 break;
 696         case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID:
 697                 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 698                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 699                                 "Expecting a non-absolute clock class with a specific UUID, "
 700                                 "but got an absolute one: "
 701                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 702                                 clock_class, cc_name);
 703                         goto error;
 704                 }
 705
 706                 if (!cc_uuid) {
 707                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 708                                 "Expecting a non-absolute clock class with a specific UUID, "
 709                                 "but got one with no UUID: "
 710                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 711                                 clock_class, cc_name);
 712                         goto error;
 713                 }
 714
 715                 if (bt_uuid_compare(muxer_msg_iter->expected_clock_class_uuid, cc_uuid) != 0) {
 716                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 717                                 "Expecting a non-absolute clock class with a specific UUID, "
 718                                 "but got one with different UUID: "
 719                                 "clock-class-addr=%p, clock-class-name=\"%s\", "
 720                                 "expected-uuid=\"" BT_UUID_FMT "\", "
 721                                 "uuid=\"" BT_UUID_FMT "\"",
 722                                 clock_class, cc_name,
 723                                 BT_UUID_FMT_VALUES(muxer_msg_iter->expected_clock_class_uuid),
 724                                 BT_UUID_FMT_VALUES(cc_uuid));
 725                         goto error;
 726                 }
 727                 break;
 728         case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE:
 729                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 730                         "Expecting no clock class, but got one: "
 731                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 732                         clock_class, cc_name);
 733                 goto error;
 734         default:
 735                 /* Unexpected */
 736                 BT_COMP_LOGF("Unexpected clock class expectation: "
 737                         "expectation-code=%d",
 738                         muxer_msg_iter->clock_class_expectation);
 739                 bt_common_abort();
 740         }
 741
 742         goto end;
 743
 744 error:
 745         ret = -1;
 746
 747 end:
 748         return ret;
 749 }
 750
 751 static inline
 752 int validate_new_stream_clock_class(struct muxer_msg_iter *muxer_msg_iter,
 753                 struct muxer_comp *muxer_comp, const bt_stream *stream)
 754 {
 755         int ret = 0;
 756         const bt_stream_class *stream_class =
 757                 bt_stream_borrow_class_const(stream);
 758         const bt_clock_class *clock_class =
 759                 bt_stream_class_borrow_default_clock_class_const(stream_class);
 760
 761         if (!clock_class) {
 762                 if (muxer_msg_iter->clock_class_expectation ==
 763                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 764                         /* Expect no clock class */
 765                         muxer_msg_iter->clock_class_expectation =
 766                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE;
 767                 } else if (muxer_msg_iter->clock_class_expectation !=
 768                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE) {
 769                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 770                                 "Expecting stream class without a default clock class: "
 771                                 "stream-class-addr=%p, stream-class-name=\"%s\", "
 772                                 "stream-class-id=%" PRIu64,
 773                                 stream_class, bt_stream_class_get_name(stream_class),
 774                                 bt_stream_class_get_id(stream_class));
 775                         ret = -1;
 776                 }
 777
 778                 goto end;
 779         }
 780
 781         ret = validate_clock_class(muxer_msg_iter, muxer_comp, clock_class);
 782
 783 end:
 784         return ret;
 785 }
 786
 787 /*
 788  * This function finds the youngest available message amongst the
 789  * non-ended upstream message iterators and returns the upstream
 790  * message iterator which has it, or
 791  * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
 792  * message.
 793  *
 794  * This function does NOT:
 795  *
 796  * * Update any upstream message iterator.
 797  * * Check the upstream message iterators to retry.
 798  *
 799  * On sucess, this function sets *muxer_upstream_msg_iter to the
 800  * upstream message iterator of which the current message is
 801  * the youngest, and sets *ts_ns to its time.
 802  */
 803 static
 804 bt_message_iterator_class_next_method_status
 805 muxer_msg_iter_youngest_upstream_msg_iter(
 806                 struct muxer_comp *muxer_comp,
 807                 struct muxer_msg_iter *muxer_msg_iter,
 808                 struct muxer_upstream_msg_iter **muxer_upstream_msg_iter,
 809                 int64_t *ts_ns)
 810 {
 811         size_t i;
 812         int ret;
 813         int64_t youngest_ts_ns = INT64_MAX;
 814         bt_message_iterator_class_next_method_status status =
 815                 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
 816
 817         BT_ASSERT_DBG(muxer_comp);
 818         BT_ASSERT_DBG(muxer_msg_iter);
 819         BT_ASSERT_DBG(muxer_upstream_msg_iter);
 820         *muxer_upstream_msg_iter = NULL;
 821
 822         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
 823                         i++) {
 824                 const bt_message *msg;
 825                 struct muxer_upstream_msg_iter *cur_muxer_upstream_msg_iter =
 826                         g_ptr_array_index(
 827                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
 828                                 i);
 829                 int64_t msg_ts_ns;
 830
 831                 if (!cur_muxer_upstream_msg_iter->msg_iter) {
 832                         /* This upstream message iterator is ended */
 833                         BT_COMP_LOGT("Skipping ended upstream message iterator: "
 834                                 "muxer-upstream-msg-iter-wrap-addr=%p",
 835                                 cur_muxer_upstream_msg_iter);
 836                         continue;
 837                 }
 838
 839                 BT_ASSERT_DBG(cur_muxer_upstream_msg_iter->next_msg <
 840                         cur_muxer_upstream_msg_iter->msgs->len);
 841                 msg = g_ptr_array_index(cur_muxer_upstream_msg_iter->msgs,
 842                         cur_muxer_upstream_msg_iter->next_msg);
 843                 BT_ASSERT_DBG(msg);
 844
 845                 if (G_UNLIKELY(bt_message_get_type(msg) ==
 846                                 BT_MESSAGE_TYPE_STREAM_BEGINNING)) {
 847                         ret = validate_new_stream_clock_class(
 848                                 muxer_msg_iter, muxer_comp,
 849                                 bt_message_stream_beginning_borrow_stream_const(
 850                                         msg));
 851                         if (ret) {
 852                                 /*
 853                                  * validate_new_stream_clock_class() logs
 854                                  * errors.
 855                                  */
 856                                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
 857                                 goto end;
 858                         }
 859                 } else if (G_UNLIKELY(bt_message_get_type(msg) ==
 860                                 BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY)) {
 861                         const bt_clock_snapshot *cs;
 862
 863                         cs = bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
 864                                 msg);
 865                         ret = validate_clock_class(muxer_msg_iter, muxer_comp,
 866                                 bt_clock_snapshot_borrow_clock_class_const(cs));
 867                         if (ret) {
 868                                 /* validate_clock_class() logs errors */
 869                                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
 870                                 goto end;
 871                         }
 872                 }
 873
 874                 ret = get_msg_ts_ns(muxer_comp, muxer_msg_iter, msg,
 875                         muxer_msg_iter->last_returned_ts_ns, &msg_ts_ns);
 876                 if (ret) {
 877                         /* get_msg_ts_ns() logs errors */
 878                         *muxer_upstream_msg_iter = NULL;
 879                         status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
 880                         goto end;
 881                 }
 882
 883                 /*
 884                  * Update the current message iterator if it has not been set
 885                  * yet, or if its current message has a timestamp smaller than
 886                  * the previously selected youngest message.
 887                  */
 888                 if (G_UNLIKELY(*muxer_upstream_msg_iter == NULL) ||
 889                                 msg_ts_ns < youngest_ts_ns) {
 890                         *muxer_upstream_msg_iter =
 891                                 cur_muxer_upstream_msg_iter;
 892                         youngest_ts_ns = msg_ts_ns;
 893                         *ts_ns = youngest_ts_ns;
 894                 } else if (msg_ts_ns == youngest_ts_ns) {
 895                         /*
 896                          * The currently selected message to be sent downstream
 897                          * next has the exact same timestamp that of the
 898                          * current candidate message. We must break the tie
 899                          * in a predictable manner.
 900                          */
 901                         BT_ASSERT_DBG((*muxer_upstream_msg_iter)->next_msg <
 902                                 (*muxer_upstream_msg_iter)->msgs->len);
 903                         const bt_message *selected_msg =
 904                                 g_ptr_array_index((*muxer_upstream_msg_iter)->msgs,
 905                                         (*muxer_upstream_msg_iter)->next_msg);
 906                         BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
 907
 908                         /*
 909                          * Order the messages in an arbitrary but determinitic
 910                          * way.
 911                          */
 912                         ret = common_muxing_compare_messages(msg, selected_msg);
 913                         if (ret < 0) {
 914                                 /*
 915                                  * The `msg` should go first. Update the next
 916                                  * iterator and the current timestamp.
 917                                  */
 918                                 *muxer_upstream_msg_iter =
 919                                         cur_muxer_upstream_msg_iter;
 920                                 youngest_ts_ns = msg_ts_ns;
 921                                 *ts_ns = youngest_ts_ns;
 922                         } else if (ret == 0) {
 923                                 /* Unable to pick which one should go first. */
 924                                 BT_COMP_LOGW("Cannot deterministically pick next upstream message iterator because they have identical next messages: "
 925                                         "muxer-upstream-msg-iter-wrap-addr=%p"
 926                                         "cur-muxer-upstream-msg-iter-wrap-addr=%p",
 927                                         *muxer_upstream_msg_iter,
 928                                         cur_muxer_upstream_msg_iter);
 929                         }
 930                 }
 931         }
 932
 933         if (!*muxer_upstream_msg_iter) {
 934                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
 935                 *ts_ns = INT64_MIN;
 936         }
 937
 938 end:
 939         return status;
 940 }
 941
 942 static
 943 bt_message_iterator_class_next_method_status
 944 validate_muxer_upstream_msg_iter(
 945         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
 946         bool *is_ended)
 947 {
 948         struct muxer_comp *muxer_comp = muxer_upstream_msg_iter->muxer_comp;
 949         bt_message_iterator_class_next_method_status status;
 950
 951         BT_COMP_LOGD("Validating muxer's upstream message iterator wrapper: "
 952                 "muxer-upstream-msg-iter-wrap-addr=%p",
 953                 muxer_upstream_msg_iter);
 954
 955         if (muxer_upstream_msg_iter->next_msg < muxer_upstream_msg_iter->msgs->len ||
 956                         !muxer_upstream_msg_iter->msg_iter) {
 957                 BT_COMP_LOGD("Already valid or not considered: "
 958                         "queue-len=%u, next-msg=%u, upstream-msg-iter-addr=%p",
 959                         muxer_upstream_msg_iter->msgs->len,
 960                         muxer_upstream_msg_iter->next_msg,
 961                         muxer_upstream_msg_iter->msg_iter);
 962                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
 963                 goto end;
 964         }
 965
 966         /* muxer_upstream_msg_iter_next() logs details/errors */
 967         status = muxer_upstream_msg_iter_next(muxer_upstream_msg_iter,
 968                 is_ended);
 969
 970 end:
 971         return status;
 972 }
 973
 974 static
 975 bt_message_iterator_class_next_method_status
 976 validate_muxer_upstream_msg_iters(
 977                 struct muxer_msg_iter *muxer_msg_iter)
 978 {
 979         struct muxer_comp *muxer_comp = muxer_msg_iter->muxer_comp;
 980         bt_message_iterator_class_next_method_status status;
 981         size_t i;
 982
 983         BT_COMP_LOGD("Validating muxer's upstream message iterator wrappers: "
 984                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
 985
 986         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
 987                         i++) {
 988                 bool is_ended = false;
 989                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
 990                         g_ptr_array_index(
 991                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
 992                                 i);
 993
 994                 status = validate_muxer_upstream_msg_iter(
 995                         muxer_upstream_msg_iter, &is_ended);
 996                 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
 997                         if (status < 0) {
 998                                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 999                                         "Cannot validate muxer's upstream message iterator wrapper: "
1000                                         "muxer-msg-iter-addr=%p, "
1001                                         "muxer-upstream-msg-iter-wrap-addr=%p",
1002                                         muxer_msg_iter,
1003                                         muxer_upstream_msg_iter);
1004                         } else {
1005                                 BT_COMP_LOGD("Cannot validate muxer's upstream message iterator wrapper: "
1006                                         "muxer-msg-iter-addr=%p, "
1007                                         "muxer-upstream-msg-iter-wrap-addr=%p",
1008                                         muxer_msg_iter,
1009                                         muxer_upstream_msg_iter);
1010                         }
1011
1012                         goto end;
1013                 }
1014
1015                 /*
1016                  * Move this muxer upstream message iterator to the
1017                  * array of ended iterators if it's ended.
1018                  */
1019                 if (G_UNLIKELY(is_ended)) {
1020                         BT_COMP_LOGD("Muxer's upstream message iterator wrapper: ended or canceled: "
1021                                 "muxer-msg-iter-addr=%p, "
1022                                 "muxer-upstream-msg-iter-wrap-addr=%p",
1023                                 muxer_msg_iter, muxer_upstream_msg_iter);
1024                         g_ptr_array_add(
1025                                 muxer_msg_iter->ended_muxer_upstream_msg_iters,
1026                                 muxer_upstream_msg_iter);
1027                         muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i] = NULL;
1028
1029                         /*
1030                          * Use g_ptr_array_remove_fast() because the
1031                          * order of those elements is not important.
1032                          */
1033                         g_ptr_array_remove_index_fast(
1034                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
1035                                 i);
1036                         i--;
1037                 }
1038         }
1039
1040         status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1041
1042 end:
1043         return status;
1044 }
1045
1046 static inline
1047 bt_message_iterator_class_next_method_status muxer_msg_iter_do_next_one(
1048                 struct muxer_comp *muxer_comp,
1049                 struct muxer_msg_iter *muxer_msg_iter,
1050                 const bt_message **msg)
1051 {
1052         bt_message_iterator_class_next_method_status status;
1053         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter = NULL;
1054         /* Initialize to avoid -Wmaybe-uninitialized warning with gcc 4.8. */
1055         int64_t next_return_ts = 0;
1056
1057         status = validate_muxer_upstream_msg_iters(muxer_msg_iter);
1058         if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1059                 /* validate_muxer_upstream_msg_iters() logs details */
1060                 goto end;
1061         }
1062
1063         /*
1064          * At this point we know that all the existing upstream
1065          * message iterators are valid. We can find the one,
1066          * amongst those, of which the current message is the
1067          * youngest.
1068          */
1069         status = muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp,
1070                         muxer_msg_iter, &muxer_upstream_msg_iter,
1071                         &next_return_ts);
1072         if (status < 0 || status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END) {
1073                 if (status < 0) {
1074                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1075                                 "Cannot find the youngest upstream message iterator wrapper: "
1076                                 "status=%s",
1077                                 bt_common_func_status_string(status));
1078                 } else {
1079                         BT_COMP_LOGD("Cannot find the youngest upstream message iterator wrapper: "
1080                                 "status=%s",
1081                                 bt_common_func_status_string(status));
1082                 }
1083
1084                 goto end;
1085         }
1086
1087         if (next_return_ts < muxer_msg_iter->last_returned_ts_ns) {
1088                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1089                         "Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
1090                         "muxer-msg-iter-addr=%p, ts=%" PRId64 ", "
1091                         "last-returned-ts=%" PRId64,
1092                         muxer_msg_iter, next_return_ts,
1093                         muxer_msg_iter->last_returned_ts_ns);
1094                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1095                 goto end;
1096         }
1097
1098         BT_COMP_LOGD("Found youngest upstream message iterator wrapper: "
1099                 "muxer-msg-iter-addr=%p, "
1100                 "muxer-upstream-msg-iter-wrap-addr=%p, "
1101                 "ts=%" PRId64,
1102                 muxer_msg_iter, muxer_upstream_msg_iter, next_return_ts);
1103         BT_ASSERT_DBG(status ==
1104                 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK);
1105         BT_ASSERT_DBG(muxer_upstream_msg_iter);
1106
1107         /*
1108          * Consume from the queue's head: other side
1109          * (muxer_upstream_msg_iter_next()) writes to the tail.
1110          */
1111         *msg = g_ptr_array_index(muxer_upstream_msg_iter->msgs,
1112                 muxer_upstream_msg_iter->next_msg);
1113         g_ptr_array_index(muxer_upstream_msg_iter->msgs,
1114                 muxer_upstream_msg_iter->next_msg) = NULL;
1115         ++muxer_upstream_msg_iter->next_msg;
1116         BT_ASSERT_DBG(*msg);
1117         muxer_msg_iter->last_returned_ts_ns = next_return_ts;
1118
1119 end:
1120         return status;
1121 }
1122
1123 static
1124 bt_message_iterator_class_next_method_status muxer_msg_iter_do_next(
1125                 struct muxer_comp *muxer_comp,
1126                 struct muxer_msg_iter *muxer_msg_iter,
1127                 bt_message_array_const msgs, uint64_t capacity,
1128                 uint64_t *count)
1129 {
1130         bt_message_iterator_class_next_method_status status;
1131         uint64_t i = 0;
1132
1133         if (G_UNLIKELY(muxer_msg_iter->next_saved_error)) {
1134                 /*
1135                  * Last time we were called, we hit an error but had some
1136                  * messages to deliver, so we stashed the error here.  Return
1137                  * it now.
1138                  */
1139                 BT_CURRENT_THREAD_MOVE_ERROR_AND_RESET(muxer_msg_iter->next_saved_error);
1140                 status = muxer_msg_iter->next_saved_status;
1141                 goto end;
1142         }
1143
1144         do {
1145                 status = muxer_msg_iter_do_next_one(muxer_comp,
1146                         muxer_msg_iter, &msgs[i]);
1147                 if (status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1148                         i++;
1149                 }
1150         } while (i < capacity && status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK);
1151
1152         if (i > 0) {
1153                 /*
1154                  * Even if muxer_msg_iter_do_next_one() returned
1155                  * something else than
1156                  * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1157                  * message objects in the output message
1158                  * array, so we need to return
1159                  * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1160                  * transfered to downstream. This other status occurs
1161                  * again the next time muxer_msg_iter_do_next() is
1162                  * called, possibly without any accumulated
1163                  * message, in which case we'll return it.
1164                  */
1165                 if (status < 0) {
1166                         /*
1167                          * Save this error for the next _next call.  Assume that
1168                          * this component always appends error causes when
1169                          * returning an error status code, which will cause the
1170                          * current thread error to be non-NULL.
1171                          */
1172                         muxer_msg_iter->next_saved_error = bt_current_thread_take_error();
1173                         BT_ASSERT(muxer_msg_iter->next_saved_error);
1174                         muxer_msg_iter->next_saved_status = status;
1175                 }
1176
1177                 *count = i;
1178                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1179         }
1180
1181 end:
1182         return status;
1183 }
1184
1185 static
1186 void destroy_muxer_msg_iter(struct muxer_msg_iter *muxer_msg_iter)
1187 {
1188         struct muxer_comp *muxer_comp;
1189
1190         if (!muxer_msg_iter) {
1191                 return;
1192         }
1193
1194         muxer_comp = muxer_msg_iter->muxer_comp;
1195         BT_COMP_LOGD("Destroying muxer component's message iterator: "
1196                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
1197
1198         if (muxer_msg_iter->active_muxer_upstream_msg_iters) {
1199                 BT_COMP_LOGD_STR("Destroying muxer's active upstream message iterator wrappers.");
1200                 g_ptr_array_free(
1201                         muxer_msg_iter->active_muxer_upstream_msg_iters, TRUE);
1202         }
1203
1204         if (muxer_msg_iter->ended_muxer_upstream_msg_iters) {
1205                 BT_COMP_LOGD_STR("Destroying muxer's ended upstream message iterator wrappers.");
1206                 g_ptr_array_free(
1207                         muxer_msg_iter->ended_muxer_upstream_msg_iters, TRUE);
1208         }
1209
1210         g_free(muxer_msg_iter);
1211 }
1212
1213 static
1214 bt_message_iterator_class_initialize_method_status
1215 muxer_msg_iter_init_upstream_iterators(struct muxer_comp *muxer_comp,
1216                 struct muxer_msg_iter *muxer_msg_iter,
1217                 struct bt_self_message_iterator_configuration *config)
1218 {
1219         int64_t count;
1220         int64_t i;
1221         bt_message_iterator_class_initialize_method_status status;
1222         bool can_seek_forward = true;
1223
1224         count = bt_component_filter_get_input_port_count(
1225                 bt_self_component_filter_as_component_filter(
1226                         muxer_comp->self_comp_flt));
1227         if (count < 0) {
1228                 BT_COMP_LOGD("No input port to initialize for muxer component's message iterator: "
1229                         "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
1230                         muxer_comp, muxer_msg_iter);
1231                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
1232                 goto end;
1233         }
1234
1235         for (i = 0; i < count; i++) {
1236                 bt_message_iterator *upstream_msg_iter;
1237                 bt_self_component_port_input *self_port =
1238                         bt_self_component_filter_borrow_input_port_by_index(
1239                                 muxer_comp->self_comp_flt, i);
1240                 const bt_port *port;
1241                 bt_message_iterator_create_from_message_iterator_status
1242                         msg_iter_status;
1243                 int int_status;
1244
1245                 BT_ASSERT(self_port);
1246                 port = bt_self_component_port_as_port(
1247                         bt_self_component_port_input_as_self_component_port(
1248                                 self_port));
1249                 BT_ASSERT(port);
1250
1251                 if (!bt_port_is_connected(port)) {
1252                         /* Skip non-connected port */
1253                         continue;
1254                 }
1255
1256                 msg_iter_status = create_msg_iter_on_input_port(muxer_comp,
1257                         muxer_msg_iter, self_port, &upstream_msg_iter);
1258                 if (msg_iter_status != BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK) {
1259                         /* create_msg_iter_on_input_port() logs errors */
1260                         status = (int) msg_iter_status;
1261                         goto end;
1262                 }
1263
1264                 int_status = muxer_msg_iter_add_upstream_msg_iter(muxer_msg_iter,
1265                         upstream_msg_iter);
1266                 bt_message_iterator_put_ref(
1267                         upstream_msg_iter);
1268                 if (int_status) {
1269                         status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1270                         /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
1271                         goto end;
1272                 }
1273
1274                 can_seek_forward = can_seek_forward &&
1275                         bt_message_iterator_can_seek_forward(
1276                                 upstream_msg_iter);
1277         }
1278
1279         /*
1280          * This iterator can seek forward if all of its iterators can seek
1281          * forward.
1282          */
1283         bt_self_message_iterator_configuration_set_can_seek_forward(
1284                 config, can_seek_forward);
1285
1286         status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
1287
1288 end:
1289         return status;
1290 }
1291
1292 bt_message_iterator_class_initialize_method_status muxer_msg_iter_init(
1293                 bt_self_message_iterator *self_msg_iter,
1294                 bt_self_message_iterator_configuration *config,
1295                 bt_self_component_port_output *port __attribute__((unused)))
1296 {
1297         struct muxer_comp *muxer_comp = NULL;
1298         struct muxer_msg_iter *muxer_msg_iter = NULL;
1299         bt_message_iterator_class_initialize_method_status status;
1300         bt_self_component *self_comp =
1301                 bt_self_message_iterator_borrow_component(self_msg_iter);
1302
1303         muxer_comp = bt_self_component_get_data(self_comp);
1304         BT_ASSERT(muxer_comp);
1305         BT_COMP_LOGD("Initializing muxer component's message iterator: "
1306                 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1307                 self_comp, muxer_comp, self_msg_iter);
1308
1309         if (muxer_comp->initializing_muxer_msg_iter) {
1310                 /*
1311                  * Weird, unhandled situation detected: downstream
1312                  * creates a muxer message iterator while creating
1313                  * another muxer message iterator (same component).
1314                  */
1315                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1316                         "Recursive initialization of muxer component's message iterator: "
1317                         "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1318                         self_comp, muxer_comp, self_msg_iter);
1319                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1320                 goto error;
1321         }
1322
1323         muxer_comp->initializing_muxer_msg_iter = true;
1324         muxer_msg_iter = g_new0(struct muxer_msg_iter, 1);
1325         if (!muxer_msg_iter) {
1326                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1327                         "Failed to allocate one muxer component's message iterator.");
1328                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1329                 goto error;
1330         }
1331
1332         muxer_msg_iter->muxer_comp = muxer_comp;
1333         muxer_msg_iter->self_msg_iter = self_msg_iter;
1334         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1335         muxer_msg_iter->active_muxer_upstream_msg_iters =
1336                 g_ptr_array_new_with_free_func(
1337                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1338         if (!muxer_msg_iter->active_muxer_upstream_msg_iters) {
1339                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp, "Failed to allocate a GPtrArray.");
1340                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1341                 goto error;
1342         }
1343
1344         muxer_msg_iter->ended_muxer_upstream_msg_iters =
1345                 g_ptr_array_new_with_free_func(
1346                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1347         if (!muxer_msg_iter->ended_muxer_upstream_msg_iters) {
1348                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp, "Failed to allocate a GPtrArray.");
1349                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1350                 goto error;
1351         }
1352
1353         status = muxer_msg_iter_init_upstream_iterators(muxer_comp,
1354                 muxer_msg_iter, config);
1355         if (status) {
1356                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1357                         "Cannot initialize connected input ports for muxer component's message iterator: "
1358                         "comp-addr=%p, muxer-comp-addr=%p, "
1359                         "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1360                         self_comp, muxer_comp, muxer_msg_iter,
1361                         self_msg_iter, status);
1362                 goto error;
1363         }
1364
1365         bt_self_message_iterator_set_data(self_msg_iter, muxer_msg_iter);
1366         BT_COMP_LOGD("Initialized muxer component's message iterator: "
1367                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1368                 "msg-iter-addr=%p",
1369                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1370         goto end;
1371
1372 error:
1373         destroy_muxer_msg_iter(muxer_msg_iter);
1374         bt_self_message_iterator_set_data(self_msg_iter, NULL);
1375
1376 end:
1377         muxer_comp->initializing_muxer_msg_iter = false;
1378         return status;
1379 }
1380
1381 void muxer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
1382 {
1383         struct muxer_msg_iter *muxer_msg_iter =
1384                 bt_self_message_iterator_get_data(self_msg_iter);
1385         bt_self_component *self_comp = NULL;
1386         struct muxer_comp *muxer_comp = NULL;
1387
1388         self_comp = bt_self_message_iterator_borrow_component(
1389                 self_msg_iter);
1390         BT_ASSERT(self_comp);
1391         muxer_comp = bt_self_component_get_data(self_comp);
1392         BT_COMP_LOGD("Finalizing muxer component's message iterator: "
1393                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1394                 "msg-iter-addr=%p",
1395                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1396
1397         if (muxer_msg_iter) {
1398                 destroy_muxer_msg_iter(muxer_msg_iter);
1399         }
1400 }
1401
1402 bt_message_iterator_class_next_method_status muxer_msg_iter_next(
1403                 bt_self_message_iterator *self_msg_iter,
1404                 bt_message_array_const msgs, uint64_t capacity,
1405                 uint64_t *count)
1406 {
1407         bt_message_iterator_class_next_method_status status;
1408         struct muxer_msg_iter *muxer_msg_iter =
1409                 bt_self_message_iterator_get_data(self_msg_iter);
1410         bt_self_component *self_comp = NULL;
1411         struct muxer_comp *muxer_comp = NULL;
1412
1413         BT_ASSERT_DBG(muxer_msg_iter);
1414         self_comp = bt_self_message_iterator_borrow_component(
1415                 self_msg_iter);
1416         BT_ASSERT_DBG(self_comp);
1417         muxer_comp = bt_self_component_get_data(self_comp);
1418         BT_ASSERT_DBG(muxer_comp);
1419         BT_COMP_LOGT("Muxer component's message iterator's \"next\" method called: "
1420                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1421                 "msg-iter-addr=%p",
1422                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1423
1424         status = muxer_msg_iter_do_next(muxer_comp, muxer_msg_iter,
1425                 msgs, capacity, count);
1426         if (status < 0) {
1427                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1428                         "Cannot get next message: "
1429                         "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1430                         "msg-iter-addr=%p, status=%s",
1431                         self_comp, muxer_comp, muxer_msg_iter, self_msg_iter,
1432                         bt_common_func_status_string(status));
1433         } else {
1434                 BT_COMP_LOGT("Returning from muxer component's message iterator's \"next\" method: "
1435                         "status=%s",
1436                         bt_common_func_status_string(status));
1437         }
1438
1439         return status;
1440 }
1441
1442 bt_component_class_port_connected_method_status muxer_input_port_connected(
1443                 bt_self_component_filter *self_comp,
1444                 bt_self_component_port_input *self_port __attribute__((unused)),
1445                 const bt_port_output *other_port __attribute__((unused)))
1446 {
1447         bt_component_class_port_connected_method_status status =
1448                 BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_OK;
1449         bt_self_component_add_port_status add_port_status;
1450         struct muxer_comp *muxer_comp = bt_self_component_get_data(
1451                 bt_self_component_filter_as_self_component(self_comp));
1452
1453         add_port_status = add_available_input_port(self_comp);
1454         if (add_port_status) {
1455                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1456                         "Cannot add one muxer component's input port: status=%s",
1457                         bt_common_func_status_string(add_port_status));
1458
1459                 if (add_port_status ==
1460                                 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR) {
1461                         status = BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_MEMORY_ERROR;
1462                 } else {
1463                         status = BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_ERROR;
1464                 }
1465
1466                 goto end;
1467         }
1468
1469 end:
1470         return status;
1471 }
1472
1473 static inline
1474 bt_message_iterator_class_can_seek_beginning_method_status
1475 muxer_upstream_msg_iters_can_all_seek_beginning(
1476                 struct muxer_comp *muxer_comp,
1477                 GPtrArray *muxer_upstream_msg_iters, bt_bool *can_seek)
1478 {
1479         bt_message_iterator_class_can_seek_beginning_method_status status =
1480                 BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK;
1481         uint64_t i;
1482
1483         for (i = 0; i < muxer_upstream_msg_iters->len; i++) {
1484                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1485                         muxer_upstream_msg_iters->pdata[i];
1486                 status = (int) bt_message_iterator_can_seek_beginning(
1487                         upstream_msg_iter->msg_iter, can_seek);
1488                 if (status != BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK) {
1489                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1490                                 "Failed to determine whether upstream message iterator can seek beginning: "
1491                                 "msg-iter-addr=%p", upstream_msg_iter->msg_iter);
1492                         goto end;
1493                 }
1494
1495                 if (!*can_seek) {
1496                         goto end;
1497                 }
1498         }
1499
1500         *can_seek = BT_TRUE;
1501
1502 end:
1503         return status;
1504 }
1505
1506 bt_message_iterator_class_can_seek_beginning_method_status
1507 muxer_msg_iter_can_seek_beginning(
1508                 bt_self_message_iterator *self_msg_iter, bt_bool *can_seek)
1509 {
1510         struct muxer_msg_iter *muxer_msg_iter =
1511                 bt_self_message_iterator_get_data(self_msg_iter);
1512         bt_message_iterator_class_can_seek_beginning_method_status status;
1513
1514         status = muxer_upstream_msg_iters_can_all_seek_beginning(
1515                 muxer_msg_iter->muxer_comp,
1516                 muxer_msg_iter->active_muxer_upstream_msg_iters, can_seek);
1517         if (status != BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK) {
1518                 goto end;
1519         }
1520
1521         if (!*can_seek) {
1522                 goto end;
1523         }
1524
1525         status = muxer_upstream_msg_iters_can_all_seek_beginning(
1526                 muxer_msg_iter->muxer_comp,
1527                 muxer_msg_iter->ended_muxer_upstream_msg_iters, can_seek);
1528
1529 end:
1530         return status;
1531 }
1532
1533 bt_message_iterator_class_seek_beginning_method_status muxer_msg_iter_seek_beginning(
1534                 bt_self_message_iterator *self_msg_iter)
1535 {
1536         struct muxer_msg_iter *muxer_msg_iter =
1537                 bt_self_message_iterator_get_data(self_msg_iter);
1538         bt_message_iterator_class_seek_beginning_method_status status =
1539                 BT_MESSAGE_ITERATOR_CLASS_SEEK_BEGINNING_METHOD_STATUS_OK;
1540         bt_message_iterator_seek_beginning_status seek_beg_status;
1541         uint64_t i;
1542
1543         /* Seek all ended upstream iterators first */
1544         for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
1545                         i++) {
1546                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1547                         muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
1548
1549                 seek_beg_status = bt_message_iterator_seek_beginning(
1550                         upstream_msg_iter->msg_iter);
1551                 if (seek_beg_status != BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK) {
1552                         status = (int) seek_beg_status;
1553                         goto end;
1554                 }
1555
1556                 empty_message_queue(upstream_msg_iter);
1557         }
1558
1559         /* Seek all previously active upstream iterators */
1560         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
1561                         i++) {
1562                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1563                         muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i];
1564
1565                 seek_beg_status = bt_message_iterator_seek_beginning(
1566                         upstream_msg_iter->msg_iter);
1567                 if (seek_beg_status != BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK) {
1568                         status = (int) seek_beg_status;
1569                         goto end;
1570                 }
1571
1572                 empty_message_queue(upstream_msg_iter);
1573         }
1574
1575         /* Make them all active */
1576         for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
1577                         i++) {
1578                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1579                         muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
1580
1581                 g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
1582                         upstream_msg_iter);
1583                 muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i] = NULL;
1584         }
1585
1586         /*
1587          * GLib < 2.48.0 asserts when g_ptr_array_remove_range() is
1588          * called on an empty array.
1589          */
1590         if (muxer_msg_iter->ended_muxer_upstream_msg_iters->len > 0) {
1591                 g_ptr_array_remove_range(muxer_msg_iter->ended_muxer_upstream_msg_iters,
1592                         0, muxer_msg_iter->ended_muxer_upstream_msg_iters->len);
1593         }
1594         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1595         muxer_msg_iter->clock_class_expectation =
1596                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY;
1597
1598 end:
1599         return status;
1600 }