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