src/plugins/utils/muxer/muxer.c

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