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_LOG_TAG "PLUGIN-UTILS-MUXER-FLT"
  24 #include "logging.h"
  25
  26 #include <babeltrace/babeltrace-internal.h>
  27 #include <babeltrace/compat/uuid-internal.h>
  28 #include <babeltrace/babeltrace.h>
  29 #include <babeltrace/value-internal.h>
  30 #include <babeltrace/graph/component-internal.h>
  31 #include <babeltrace/graph/message-iterator-internal.h>
  32 #include <babeltrace/graph/connection-internal.h>
  33 #include <plugins-common.h>
  34 #include <glib.h>
  35 #include <stdbool.h>
  36 #include <inttypes.h>
  37 #include <babeltrace/assert-internal.h>
  38 #include <babeltrace/common-internal.h>
  39 #include <stdlib.h>
  40 #include <string.h>
  41
  42 #include "muxer.h"
  43
  44 #define ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME        "assume-absolute-clock-classes"
  45
  46 struct muxer_comp {
  47         /* Weak ref */
  48         bt_self_component_filter *self_comp;
  49
  50         unsigned int next_port_num;
  51         size_t available_input_ports;
  52         bool initializing_muxer_msg_iter;
  53         bool assume_absolute_clock_classes;
  54 };
  55
  56 struct muxer_upstream_msg_iter {
  57         /* Owned by this, NULL if ended */
  58         bt_self_component_port_input_message_iterator *msg_iter;
  59
  60         /* Contains `const bt_message *`, owned by this */
  61         GQueue *msgs;
  62 };
  63
  64 enum muxer_msg_iter_clock_class_expectation {
  65         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY = 0,
  66         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE,
  67         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID,
  68         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID,
  69 };
  70
  71 struct muxer_msg_iter {
  72         /*
  73          * Array of struct muxer_upstream_msg_iter * (owned by this).
  74          *
  75          * NOTE: This array is searched in linearly to find the youngest
  76          * current message. Keep this until benchmarks confirm that
  77          * another data structure is faster than this for our typical
  78          * use cases.
  79          */
  80         GPtrArray *muxer_upstream_msg_iters;
  81
  82         /* Last time returned in a message */
  83         int64_t last_returned_ts_ns;
  84
  85         /* Clock class expectation state */
  86         enum muxer_msg_iter_clock_class_expectation clock_class_expectation;
  87
  88         /*
  89          * Expected clock class UUID, only valid when
  90          * clock_class_expectation is
  91          * MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID.
  92          */
  93         unsigned char expected_clock_class_uuid[BABELTRACE_UUID_LEN];
  94 };
  95
  96 static
  97 void destroy_muxer_upstream_msg_iter(
  98                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
  99 {
 100         if (!muxer_upstream_msg_iter) {
 101                 return;
 102         }
 103
 104         BT_LOGD("Destroying muxer's upstream message iterator wrapper: "
 105                 "addr=%p, msg-iter-addr=%p, queue-len=%u",
 106                 muxer_upstream_msg_iter,
 107                 muxer_upstream_msg_iter->msg_iter,
 108                 muxer_upstream_msg_iter->msgs->length);
 109         bt_self_component_port_input_message_iterator_put_ref(muxer_upstream_msg_iter->msg_iter);
 110
 111         if (muxer_upstream_msg_iter->msgs) {
 112                 const bt_message *msg;
 113
 114                 while ((msg = g_queue_pop_head(
 115                                 muxer_upstream_msg_iter->msgs))) {
 116                         bt_message_put_ref(msg);
 117                 }
 118
 119                 g_queue_free(muxer_upstream_msg_iter->msgs);
 120         }
 121
 122         g_free(muxer_upstream_msg_iter);
 123 }
 124
 125 static
 126 struct muxer_upstream_msg_iter *muxer_msg_iter_add_upstream_msg_iter(
 127                 struct muxer_msg_iter *muxer_msg_iter,
 128                 bt_self_component_port_input_message_iterator *self_msg_iter)
 129 {
 130         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
 131                 g_new0(struct muxer_upstream_msg_iter, 1);
 132
 133         if (!muxer_upstream_msg_iter) {
 134                 BT_LOGE_STR("Failed to allocate one muxer's upstream message iterator wrapper.");
 135                 goto end;
 136         }
 137
 138         muxer_upstream_msg_iter->msg_iter = self_msg_iter;
 139         bt_self_component_port_input_message_iterator_get_ref(muxer_upstream_msg_iter->msg_iter);
 140         muxer_upstream_msg_iter->msgs = g_queue_new();
 141         if (!muxer_upstream_msg_iter->msgs) {
 142                 BT_LOGE_STR("Failed to allocate a GQueue.");
 143                 goto end;
 144         }
 145
 146         g_ptr_array_add(muxer_msg_iter->muxer_upstream_msg_iters,
 147                 muxer_upstream_msg_iter);
 148         BT_LOGD("Added muxer's upstream message iterator wrapper: "
 149                 "addr=%p, muxer-msg-iter-addr=%p, msg-iter-addr=%p",
 150                 muxer_upstream_msg_iter, muxer_msg_iter,
 151                 self_msg_iter);
 152
 153 end:
 154         return muxer_upstream_msg_iter;
 155 }
 156
 157 static
 158 bt_self_component_status add_available_input_port(
 159                 bt_self_component_filter *self_comp)
 160 {
 161         struct muxer_comp *muxer_comp = bt_self_component_get_data(
 162                 bt_self_component_filter_as_self_component(self_comp));
 163         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 164         GString *port_name = NULL;
 165
 166         BT_ASSERT(muxer_comp);
 167         port_name = g_string_new("in");
 168         if (!port_name) {
 169                 BT_LOGE_STR("Failed to allocate a GString.");
 170                 status = BT_SELF_COMPONENT_STATUS_NOMEM;
 171                 goto end;
 172         }
 173
 174         g_string_append_printf(port_name, "%u", muxer_comp->next_port_num);
 175         status = bt_self_component_filter_add_input_port(
 176                 self_comp, port_name->str, NULL, NULL);
 177         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 178                 BT_LOGE("Cannot add input port to muxer component: "
 179                         "port-name=\"%s\", comp-addr=%p, status=%s",
 180                         port_name->str, self_comp,
 181                         bt_self_component_status_string(status));
 182                 goto end;
 183         }
 184
 185         muxer_comp->available_input_ports++;
 186         muxer_comp->next_port_num++;
 187         BT_LOGD("Added one input port to muxer component: "
 188                 "port-name=\"%s\", comp-addr=%p",
 189                 port_name->str, self_comp);
 190
 191 end:
 192         if (port_name) {
 193                 g_string_free(port_name, TRUE);
 194         }
 195
 196         return status;
 197 }
 198
 199 static
 200 bt_self_component_status create_output_port(
 201                 bt_self_component_filter *self_comp)
 202 {
 203         return bt_self_component_filter_add_output_port(
 204                 self_comp, "out", NULL, NULL);
 205 }
 206
 207 static
 208 void destroy_muxer_comp(struct muxer_comp *muxer_comp)
 209 {
 210         if (!muxer_comp) {
 211                 return;
 212         }
 213
 214         g_free(muxer_comp);
 215 }
 216
 217 static
 218 bt_value *get_default_params(void)
 219 {
 220         bt_value *params;
 221         int ret;
 222
 223         params = bt_value_map_create();
 224         if (!params) {
 225                 BT_LOGE_STR("Cannot create a map value object.");
 226                 goto error;
 227         }
 228
 229         ret = bt_value_map_insert_bool_entry(params,
 230                 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, false);
 231         if (ret) {
 232                 BT_LOGE_STR("Cannot add boolean value to map value object.");
 233                 goto error;
 234         }
 235
 236         goto end;
 237
 238 error:
 239         BT_VALUE_PUT_REF_AND_RESET(params);
 240
 241 end:
 242         return params;
 243 }
 244
 245 static
 246 int configure_muxer_comp(struct muxer_comp *muxer_comp,
 247                 const bt_value *params)
 248 {
 249         bt_value *default_params = NULL;
 250         bt_value *real_params = NULL;
 251         const bt_value *assume_absolute_clock_classes = NULL;
 252         int ret = 0;
 253         bt_bool bool_val;
 254
 255         default_params = get_default_params();
 256         if (!default_params) {
 257                 BT_LOGE("Cannot get default parameters: "
 258                         "muxer-comp-addr=%p", muxer_comp);
 259                 goto error;
 260         }
 261
 262         ret = bt_value_map_extend(default_params, params, &real_params);
 263         if (ret) {
 264                 BT_LOGE("Cannot extend default parameters map value: "
 265                         "muxer-comp-addr=%p, def-params-addr=%p, "
 266                         "params-addr=%p", muxer_comp, default_params,
 267                         params);
 268                 goto error;
 269         }
 270
 271         assume_absolute_clock_classes = bt_value_map_borrow_entry_value(real_params,
 272                                                                         ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME);
 273         if (assume_absolute_clock_classes &&
 274                         !bt_value_is_bool(assume_absolute_clock_classes)) {
 275                 BT_LOGE("Expecting a boolean value for the `%s` parameter: "
 276                         "muxer-comp-addr=%p, value-type=%s",
 277                         ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, muxer_comp,
 278                         bt_common_value_type_string(
 279                                 bt_value_get_type(assume_absolute_clock_classes)));
 280                 goto error;
 281         }
 282
 283         bool_val = bt_value_bool_get(assume_absolute_clock_classes);
 284         muxer_comp->assume_absolute_clock_classes = (bool) bool_val;
 285         BT_LOGD("Configured muxer component: muxer-comp-addr=%p, "
 286                 "assume-absolute-clock-classes=%d",
 287                 muxer_comp, muxer_comp->assume_absolute_clock_classes);
 288         goto end;
 289
 290 error:
 291         ret = -1;
 292
 293 end:
 294         bt_value_put_ref(default_params);
 295         bt_value_put_ref(real_params);
 296         return ret;
 297 }
 298
 299 BT_HIDDEN
 300 bt_self_component_status muxer_init(
 301                 bt_self_component_filter *self_comp,
 302                 const bt_value *params, void *init_data)
 303 {
 304         int ret;
 305         bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
 306         struct muxer_comp *muxer_comp = g_new0(struct muxer_comp, 1);
 307
 308         BT_LOGD("Initializing muxer component: "
 309                 "comp-addr=%p, params-addr=%p", self_comp, params);
 310
 311         if (!muxer_comp) {
 312                 BT_LOGE_STR("Failed to allocate one muxer component.");
 313                 goto error;
 314         }
 315
 316         ret = configure_muxer_comp(muxer_comp, params);
 317         if (ret) {
 318                 BT_LOGE("Cannot configure muxer component: "
 319                         "muxer-comp-addr=%p, params-addr=%p",
 320                         muxer_comp, params);
 321                 goto error;
 322         }
 323
 324         muxer_comp->self_comp = self_comp;
 325         bt_self_component_set_data(
 326                 bt_self_component_filter_as_self_component(self_comp),
 327                 muxer_comp);
 328         status = add_available_input_port(self_comp);
 329         if (status != BT_SELF_COMPONENT_STATUS_OK) {
 330                 BT_LOGE("Cannot ensure that at least one muxer component's input port is available: "
 331                         "muxer-comp-addr=%p, status=%s",
 332                         muxer_comp,
 333                         bt_self_component_status_string(status));
 334                 goto error;
 335         }
 336
 337         status = create_output_port(self_comp);
 338         if (status) {
 339                 BT_LOGE("Cannot create muxer component's output port: "
 340                         "muxer-comp-addr=%p, status=%s",
 341                         muxer_comp,
 342                         bt_self_component_status_string(status));
 343                 goto error;
 344         }
 345
 346         BT_LOGD("Initialized muxer component: "
 347                 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
 348                 self_comp, params, muxer_comp);
 349
 350         goto end;
 351
 352 error:
 353         destroy_muxer_comp(muxer_comp);
 354         bt_self_component_set_data(
 355                 bt_self_component_filter_as_self_component(self_comp),
 356                 NULL);
 357
 358         if (status == BT_SELF_COMPONENT_STATUS_OK) {
 359                 status = BT_SELF_COMPONENT_STATUS_ERROR;
 360         }
 361
 362 end:
 363         return status;
 364 }
 365
 366 BT_HIDDEN
 367 void muxer_finalize(bt_self_component_filter *self_comp)
 368 {
 369         struct muxer_comp *muxer_comp = bt_self_component_get_data(
 370                 bt_self_component_filter_as_self_component(self_comp));
 371
 372         BT_LOGD("Finalizing muxer component: comp-addr=%p",
 373                 self_comp);
 374         destroy_muxer_comp(muxer_comp);
 375 }
 376
 377 static
 378 bt_self_component_port_input_message_iterator *
 379 create_msg_iter_on_input_port(bt_self_component_port_input *self_port)
 380 {
 381         const bt_port *port = bt_self_component_port_as_port(
 382                 bt_self_component_port_input_as_self_component_port(
 383                         self_port));
 384         bt_self_component_port_input_message_iterator *msg_iter =
 385                 NULL;
 386
 387         BT_ASSERT(port);
 388         BT_ASSERT(bt_port_is_connected(port));
 389
 390         // TODO: Advance the iterator to >= the time of the latest
 391         //       returned message by the muxer message
 392         //       iterator which creates it.
 393         msg_iter = bt_self_component_port_input_message_iterator_create(
 394                 self_port);
 395         if (!msg_iter) {
 396                 BT_LOGE("Cannot create upstream message iterator on input port: "
 397                         "port-addr=%p, port-name=\"%s\"",
 398                         port, bt_port_get_name(port));
 399                 goto end;
 400         }
 401
 402         BT_LOGD("Created upstream message iterator on input port: "
 403                 "port-addr=%p, port-name=\"%s\", msg-iter-addr=%p",
 404                 port, bt_port_get_name(port), msg_iter);
 405
 406 end:
 407         return msg_iter;
 408 }
 409
 410 static
 411 bt_self_message_iterator_status muxer_upstream_msg_iter_next(
 412                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
 413 {
 414         bt_self_message_iterator_status status;
 415         bt_message_iterator_status input_port_iter_status;
 416         bt_message_array_const msgs;
 417         uint64_t i;
 418         uint64_t count;
 419
 420         BT_LOGV("Calling upstream message iterator's \"next\" method: "
 421                 "muxer-upstream-msg-iter-wrap-addr=%p, msg-iter-addr=%p",
 422                 muxer_upstream_msg_iter,
 423                 muxer_upstream_msg_iter->msg_iter);
 424         input_port_iter_status = bt_self_component_port_input_message_iterator_next(
 425                 muxer_upstream_msg_iter->msg_iter, &msgs, &count);
 426         BT_LOGV("Upstream message iterator's \"next\" method returned: "
 427                 "status=%s", bt_message_iterator_status_string(input_port_iter_status));
 428
 429         switch (input_port_iter_status) {
 430         case BT_MESSAGE_ITERATOR_STATUS_OK:
 431                 /*
 432                  * Message iterator's current message is
 433                  * valid: it must be considered for muxing operations.
 434                  */
 435                 BT_LOGV_STR("Validated upstream message iterator wrapper.");
 436                 BT_ASSERT(count > 0);
 437
 438                 /* Move messages to our queue */
 439                 for (i = 0; i < count; i++) {
 440                         /*
 441                          * Push to tail in order; other side
 442                          * (muxer_msg_iter_do_next_one()) consumes
 443                          * from the head first.
 444                          */
 445                         g_queue_push_tail(muxer_upstream_msg_iter->msgs,
 446                                 (void *) msgs[i]);
 447                 }
 448                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 449                 break;
 450         case BT_MESSAGE_ITERATOR_STATUS_AGAIN:
 451                 /*
 452                  * Message iterator's current message is not
 453                  * valid anymore. Return
 454                  * BT_MESSAGE_ITERATOR_STATUS_AGAIN immediately.
 455                  */
 456                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_AGAIN;
 457                 break;
 458         case BT_MESSAGE_ITERATOR_STATUS_END:    /* Fall-through. */
 459                 /*
 460                  * Message iterator reached the end: release it. It
 461                  * won't be considered again to find the youngest
 462                  * message.
 463                  */
 464                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_PUT_REF_AND_RESET(muxer_upstream_msg_iter->msg_iter);
 465                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 466                 break;
 467         default:
 468                 /* Error or unsupported status code */
 469                 BT_LOGE("Error or unsupported status code: "
 470                         "status-code=%d", input_port_iter_status);
 471                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 472                 break;
 473         }
 474
 475         return status;
 476 }
 477
 478 static
 479 int get_msg_ts_ns(struct muxer_comp *muxer_comp,
 480                 struct muxer_msg_iter *muxer_msg_iter,
 481                 const bt_message *msg, int64_t last_returned_ts_ns,
 482                 int64_t *ts_ns)
 483 {
 484         const bt_clock_class *clock_class = NULL;
 485         const bt_clock_snapshot *clock_snapshot = NULL;
 486         int ret = 0;
 487         const unsigned char *cc_uuid;
 488         const char *cc_name;
 489         bt_clock_snapshot_state cs_state = BT_CLOCK_SNAPSHOT_STATE_KNOWN;
 490         bt_message_stream_activity_clock_snapshot_state sa_cs_state;
 491
 492         BT_ASSERT(msg);
 493         BT_ASSERT(ts_ns);
 494
 495         BT_LOGV("Getting message's timestamp: "
 496                 "muxer-msg-iter-addr=%p, msg-addr=%p, "
 497                 "last-returned-ts=%" PRId64,
 498                 muxer_msg_iter, msg, last_returned_ts_ns);
 499
 500         switch (bt_message_get_type(msg)) {
 501         case BT_MESSAGE_TYPE_EVENT:
 502                 clock_class =
 503                         bt_message_event_borrow_stream_class_default_clock_class_const(
 504                                 msg);
 505                 if (!clock_class) {
 506                         goto no_clock_snapshot;
 507                 }
 508
 509                 cs_state = bt_message_event_borrow_default_clock_snapshot_const(
 510                         msg, &clock_snapshot);
 511                 break;
 512         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 513                 bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
 514                         msg);
 515                 if (!clock_class) {
 516                         goto no_clock_snapshot;
 517                 }
 518
 519                 cs_state = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
 520                         msg, &clock_snapshot);
 521                 break;
 522         case BT_MESSAGE_TYPE_PACKET_END:
 523                 bt_message_packet_end_borrow_stream_class_default_clock_class_const(
 524                         msg);
 525                 if (!clock_class) {
 526                         goto no_clock_snapshot;
 527                 }
 528
 529                 cs_state = bt_message_packet_end_borrow_default_clock_snapshot_const(
 530                         msg, &clock_snapshot);
 531                 break;
 532         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 533                 bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
 534                         msg);
 535                 if (!clock_class) {
 536                         goto no_clock_snapshot;
 537                 }
 538
 539                 cs_state = bt_message_discarded_events_borrow_default_beginning_clock_snapshot_const(
 540                         msg, &clock_snapshot);
 541                 break;
 542         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 543                 bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
 544                         msg);
 545                 if (!clock_class) {
 546                         goto no_clock_snapshot;
 547                 }
 548
 549                 cs_state = bt_message_discarded_packets_borrow_default_beginning_clock_snapshot_const(
 550                         msg, &clock_snapshot);
 551                 break;
 552         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
 553                 bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
 554                         msg);
 555                 if (!clock_class) {
 556                         goto no_clock_snapshot;
 557                 }
 558
 559                 sa_cs_state = bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
 560                         msg, &clock_snapshot);
 561                 if (sa_cs_state != BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN) {
 562                         goto no_clock_snapshot;
 563                 }
 564
 565                 break;
 566         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
 567                 bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
 568                         msg);
 569                 if (!clock_class) {
 570                         goto no_clock_snapshot;
 571                 }
 572
 573                 sa_cs_state = bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
 574                         msg, &clock_snapshot);
 575                 if (sa_cs_state != BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN) {
 576                         goto no_clock_snapshot;
 577                 }
 578
 579                 break;
 580         case BT_MESSAGE_TYPE_INACTIVITY:
 581                 cs_state =
 582                         bt_message_inactivity_borrow_default_clock_snapshot_const(
 583                                 msg, &clock_snapshot);
 584                 break;
 585         default:
 586                 /* All the other messages have a higher priority */
 587                 BT_LOGV_STR("Message has no timestamp: using the last returned timestamp.");
 588                 *ts_ns = last_returned_ts_ns;
 589                 goto end;
 590         }
 591
 592         if (cs_state != BT_CLOCK_SNAPSHOT_STATE_KNOWN) {
 593                 BT_LOGE_STR("Unsupported unknown clock snapshot.");
 594                 ret = -1;
 595                 goto end;
 596         }
 597
 598         clock_class = bt_clock_snapshot_borrow_clock_class_const(clock_snapshot);
 599         BT_ASSERT(clock_class);
 600         cc_uuid = bt_clock_class_get_uuid(clock_class);
 601         cc_name = bt_clock_class_get_name(clock_class);
 602
 603         if (muxer_msg_iter->clock_class_expectation ==
 604                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 605                 /*
 606                  * This is the first clock class that this muxer
 607                  * message iterator encounters. Its properties
 608                  * determine what to expect for the whole lifetime of
 609                  * the iterator without a true
 610                  * `assume-absolute-clock-classes` parameter.
 611                  */
 612                 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 613                         /* Expect absolute clock classes */
 614                         muxer_msg_iter->clock_class_expectation =
 615                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE;
 616                 } else {
 617                         if (cc_uuid) {
 618                                 /*
 619                                  * Expect non-absolute clock classes
 620                                  * with a specific UUID.
 621                                  */
 622                                 muxer_msg_iter->clock_class_expectation =
 623                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID;
 624                                 memcpy(muxer_msg_iter->expected_clock_class_uuid,
 625                                         cc_uuid, BABELTRACE_UUID_LEN);
 626                         } else {
 627                                 /*
 628                                  * Expect non-absolute clock classes
 629                                  * with no UUID.
 630                                  */
 631                                 muxer_msg_iter->clock_class_expectation =
 632                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID;
 633                         }
 634                 }
 635         }
 636
 637         if (!muxer_comp->assume_absolute_clock_classes) {
 638                 switch (muxer_msg_iter->clock_class_expectation) {
 639                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE:
 640                         if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
 641                                 BT_LOGE("Expecting an absolute clock class, "
 642                                         "but got a non-absolute one: "
 643                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 644                                         clock_class, cc_name);
 645                                 goto error;
 646                         }
 647                         break;
 648                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID:
 649                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 650                                 BT_LOGE("Expecting a non-absolute clock class with no UUID, "
 651                                         "but got an absolute one: "
 652                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 653                                         clock_class, cc_name);
 654                                 goto error;
 655                         }
 656
 657                         if (cc_uuid) {
 658                                 BT_LOGE("Expecting a non-absolute clock class with no UUID, "
 659                                         "but got one with a UUID: "
 660                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 661                                         "uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\"",
 662                                         clock_class, cc_name,
 663                                         (unsigned int) cc_uuid[0],
 664                                         (unsigned int) cc_uuid[1],
 665                                         (unsigned int) cc_uuid[2],
 666                                         (unsigned int) cc_uuid[3],
 667                                         (unsigned int) cc_uuid[4],
 668                                         (unsigned int) cc_uuid[5],
 669                                         (unsigned int) cc_uuid[6],
 670                                         (unsigned int) cc_uuid[7],
 671                                         (unsigned int) cc_uuid[8],
 672                                         (unsigned int) cc_uuid[9],
 673                                         (unsigned int) cc_uuid[10],
 674                                         (unsigned int) cc_uuid[11],
 675                                         (unsigned int) cc_uuid[12],
 676                                         (unsigned int) cc_uuid[13],
 677                                         (unsigned int) cc_uuid[14],
 678                                         (unsigned int) cc_uuid[15]);
 679                                 goto error;
 680                         }
 681                         break;
 682                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID:
 683                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 684                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 685                                         "but got an absolute one: "
 686                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 687                                         clock_class, cc_name);
 688                                 goto error;
 689                         }
 690
 691                         if (!cc_uuid) {
 692                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 693                                         "but got one with no UUID: "
 694                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 695                                         clock_class, cc_name);
 696                                 goto error;
 697                         }
 698
 699                         if (memcmp(muxer_msg_iter->expected_clock_class_uuid,
 700                                         cc_uuid, BABELTRACE_UUID_LEN) != 0) {
 701                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 702                                         "but got one with different UUID: "
 703                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 704                                         "expected-uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\", "
 705                                         "uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\"",
 706                                         clock_class, cc_name,
 707                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[0],
 708                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[1],
 709                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[2],
 710                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[3],
 711                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[4],
 712                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[5],
 713                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[6],
 714                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[7],
 715                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[8],
 716                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[9],
 717                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[10],
 718                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[11],
 719                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[12],
 720                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[13],
 721                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[14],
 722                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[15],
 723                                         (unsigned int) cc_uuid[0],
 724                                         (unsigned int) cc_uuid[1],
 725                                         (unsigned int) cc_uuid[2],
 726                                         (unsigned int) cc_uuid[3],
 727                                         (unsigned int) cc_uuid[4],
 728                                         (unsigned int) cc_uuid[5],
 729                                         (unsigned int) cc_uuid[6],
 730                                         (unsigned int) cc_uuid[7],
 731                                         (unsigned int) cc_uuid[8],
 732                                         (unsigned int) cc_uuid[9],
 733                                         (unsigned int) cc_uuid[10],
 734                                         (unsigned int) cc_uuid[11],
 735                                         (unsigned int) cc_uuid[12],
 736                                         (unsigned int) cc_uuid[13],
 737                                         (unsigned int) cc_uuid[14],
 738                                         (unsigned int) cc_uuid[15]);
 739                                 goto error;
 740                         }
 741                         break;
 742                 default:
 743                         /* Unexpected */
 744                         BT_LOGF("Unexpected clock class expectation: "
 745                                 "expectation-code=%d",
 746                                 muxer_msg_iter->clock_class_expectation);
 747                         abort();
 748                 }
 749         }
 750
 751         ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
 752         if (ret) {
 753                 BT_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
 754                         "clock-snapshot-addr=%p", clock_snapshot);
 755                 goto error;
 756         }
 757
 758         goto end;
 759
 760 no_clock_snapshot:
 761         BT_LOGV_STR("Message's default clock snapshot is missing: "
 762                 "using the last returned timestamp.");
 763         *ts_ns = last_returned_ts_ns;
 764         goto end;
 765
 766 error:
 767         ret = -1;
 768
 769 end:
 770         if (ret == 0) {
 771                 BT_LOGV("Found message's timestamp: "
 772                         "muxer-msg-iter-addr=%p, msg-addr=%p, "
 773                         "last-returned-ts=%" PRId64 ", ts=%" PRId64,
 774                         muxer_msg_iter, msg, last_returned_ts_ns,
 775                         *ts_ns);
 776         }
 777
 778         return ret;
 779 }
 780
 781 /*
 782  * This function finds the youngest available message amongst the
 783  * non-ended upstream message iterators and returns the upstream
 784  * message iterator which has it, or
 785  * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
 786  * message.
 787  *
 788  * This function does NOT:
 789  *
 790  * * Update any upstream message iterator.
 791  * * Check the upstream message iterators to retry.
 792  *
 793  * On sucess, this function sets *muxer_upstream_msg_iter to the
 794  * upstream message iterator of which the current message is
 795  * the youngest, and sets *ts_ns to its time.
 796  */
 797 static
 798 bt_self_message_iterator_status
 799 muxer_msg_iter_youngest_upstream_msg_iter(
 800                 struct muxer_comp *muxer_comp,
 801                 struct muxer_msg_iter *muxer_msg_iter,
 802                 struct muxer_upstream_msg_iter **muxer_upstream_msg_iter,
 803                 int64_t *ts_ns)
 804 {
 805         size_t i;
 806         int ret;
 807         int64_t youngest_ts_ns = INT64_MAX;
 808         bt_self_message_iterator_status status =
 809                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 810
 811         BT_ASSERT(muxer_comp);
 812         BT_ASSERT(muxer_msg_iter);
 813         BT_ASSERT(muxer_upstream_msg_iter);
 814         *muxer_upstream_msg_iter = NULL;
 815
 816         for (i = 0; i < muxer_msg_iter->muxer_upstream_msg_iters->len; i++) {
 817                 const bt_message *msg;
 818                 struct muxer_upstream_msg_iter *cur_muxer_upstream_msg_iter =
 819                         g_ptr_array_index(muxer_msg_iter->muxer_upstream_msg_iters, i);
 820                 int64_t msg_ts_ns;
 821
 822                 if (!cur_muxer_upstream_msg_iter->msg_iter) {
 823                         /* This upstream message iterator is ended */
 824                         BT_LOGV("Skipping ended upstream message iterator: "
 825                                 "muxer-upstream-msg-iter-wrap-addr=%p",
 826                                 cur_muxer_upstream_msg_iter);
 827                         continue;
 828                 }
 829
 830                 BT_ASSERT(cur_muxer_upstream_msg_iter->msgs->length > 0);
 831                 msg = g_queue_peek_head(cur_muxer_upstream_msg_iter->msgs);
 832                 BT_ASSERT(msg);
 833                 ret = get_msg_ts_ns(muxer_comp, muxer_msg_iter, msg,
 834                         muxer_msg_iter->last_returned_ts_ns, &msg_ts_ns);
 835                 if (ret) {
 836                         /* get_msg_ts_ns() logs errors */
 837                         *muxer_upstream_msg_iter = NULL;
 838                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 839                         goto end;
 840                 }
 841
 842                 if (msg_ts_ns <= youngest_ts_ns) {
 843                         *muxer_upstream_msg_iter =
 844                                 cur_muxer_upstream_msg_iter;
 845                         youngest_ts_ns = msg_ts_ns;
 846                         *ts_ns = youngest_ts_ns;
 847                 }
 848         }
 849
 850         if (!*muxer_upstream_msg_iter) {
 851                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
 852                 *ts_ns = INT64_MIN;
 853         }
 854
 855 end:
 856         return status;
 857 }
 858
 859 static
 860 bt_self_message_iterator_status validate_muxer_upstream_msg_iter(
 861         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
 862 {
 863         bt_self_message_iterator_status status =
 864                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 865
 866         BT_LOGV("Validating muxer's upstream message iterator wrapper: "
 867                 "muxer-upstream-msg-iter-wrap-addr=%p",
 868                 muxer_upstream_msg_iter);
 869
 870         if (muxer_upstream_msg_iter->msgs->length > 0 ||
 871                         !muxer_upstream_msg_iter->msg_iter) {
 872                 BT_LOGV("Already valid or not considered: "
 873                         "queue-len=%u, upstream-msg-iter-addr=%p",
 874                         muxer_upstream_msg_iter->msgs->length,
 875                         muxer_upstream_msg_iter->msg_iter);
 876                 goto end;
 877         }
 878
 879         /* muxer_upstream_msg_iter_next() logs details/errors */
 880         status = muxer_upstream_msg_iter_next(muxer_upstream_msg_iter);
 881
 882 end:
 883         return status;
 884 }
 885
 886 static
 887 bt_self_message_iterator_status validate_muxer_upstream_msg_iters(
 888                 struct muxer_msg_iter *muxer_msg_iter)
 889 {
 890         bt_self_message_iterator_status status =
 891                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 892         size_t i;
 893
 894         BT_LOGV("Validating muxer's upstream message iterator wrappers: "
 895                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
 896
 897         for (i = 0; i < muxer_msg_iter->muxer_upstream_msg_iters->len; i++) {
 898                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
 899                         g_ptr_array_index(
 900                                 muxer_msg_iter->muxer_upstream_msg_iters,
 901                                 i);
 902
 903                 status = validate_muxer_upstream_msg_iter(
 904                         muxer_upstream_msg_iter);
 905                 if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
 906                         if (status < 0) {
 907                                 BT_LOGE("Cannot validate muxer's upstream message iterator wrapper: "
 908                                         "muxer-msg-iter-addr=%p, "
 909                                         "muxer-upstream-msg-iter-wrap-addr=%p",
 910                                         muxer_msg_iter,
 911                                         muxer_upstream_msg_iter);
 912                         } else {
 913                                 BT_LOGV("Cannot validate muxer's upstream message iterator wrapper: "
 914                                         "muxer-msg-iter-addr=%p, "
 915                                         "muxer-upstream-msg-iter-wrap-addr=%p",
 916                                         muxer_msg_iter,
 917                                         muxer_upstream_msg_iter);
 918                         }
 919
 920                         goto end;
 921                 }
 922
 923                 /*
 924                  * Remove this muxer upstream message iterator
 925                  * if it's ended or canceled.
 926                  */
 927                 if (!muxer_upstream_msg_iter->msg_iter) {
 928                         /*
 929                          * Use g_ptr_array_remove_fast() because the
 930                          * order of those elements is not important.
 931                          */
 932                         BT_LOGV("Removing muxer's upstream message iterator wrapper: ended or canceled: "
 933                                 "muxer-msg-iter-addr=%p, "
 934                                 "muxer-upstream-msg-iter-wrap-addr=%p",
 935                                 muxer_msg_iter, muxer_upstream_msg_iter);
 936                         g_ptr_array_remove_index_fast(
 937                                 muxer_msg_iter->muxer_upstream_msg_iters,
 938                                 i);
 939                         i--;
 940                 }
 941         }
 942
 943 end:
 944         return status;
 945 }
 946
 947 static inline
 948 bt_self_message_iterator_status muxer_msg_iter_do_next_one(
 949                 struct muxer_comp *muxer_comp,
 950                 struct muxer_msg_iter *muxer_msg_iter,
 951                 const bt_message **msg)
 952 {
 953         bt_self_message_iterator_status status;
 954         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter = NULL;
 955         int64_t next_return_ts;
 956
 957         status = validate_muxer_upstream_msg_iters(muxer_msg_iter);
 958         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
 959                 /* validate_muxer_upstream_msg_iters() logs details */
 960                 goto end;
 961         }
 962
 963         /*
 964          * At this point we know that all the existing upstream
 965          * message iterators are valid. We can find the one,
 966          * amongst those, of which the current message is the
 967          * youngest.
 968          */
 969         status = muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp,
 970                         muxer_msg_iter, &muxer_upstream_msg_iter,
 971                         &next_return_ts);
 972         if (status < 0 || status == BT_SELF_MESSAGE_ITERATOR_STATUS_END) {
 973                 if (status < 0) {
 974                         BT_LOGE("Cannot find the youngest upstream message iterator wrapper: "
 975                                 "status=%s",
 976                                 bt_self_message_iterator_status_string(status));
 977                 } else {
 978                         BT_LOGV("Cannot find the youngest upstream message iterator wrapper: "
 979                                 "status=%s",
 980                                 bt_self_message_iterator_status_string(status));
 981                 }
 982
 983                 goto end;
 984         }
 985
 986         if (next_return_ts < muxer_msg_iter->last_returned_ts_ns) {
 987                 BT_LOGE("Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
 988                         "muxer-msg-iter-addr=%p, ts=%" PRId64 ", "
 989                         "last-returned-ts=%" PRId64,
 990                         muxer_msg_iter, next_return_ts,
 991                         muxer_msg_iter->last_returned_ts_ns);
 992                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 993                 goto end;
 994         }
 995
 996         BT_LOGV("Found youngest upstream message iterator wrapper: "
 997                 "muxer-msg-iter-addr=%p, "
 998                 "muxer-upstream-msg-iter-wrap-addr=%p, "
 999                 "ts=%" PRId64,
1000                 muxer_msg_iter, muxer_upstream_msg_iter, next_return_ts);
1001         BT_ASSERT(status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK);
1002         BT_ASSERT(muxer_upstream_msg_iter);
1003
1004         /*
1005          * Consume from the queue's head: other side
1006          * (muxer_upstream_msg_iter_next()) writes to the tail.
1007          */
1008         *msg = g_queue_pop_head(muxer_upstream_msg_iter->msgs);
1009         BT_ASSERT(*msg);
1010         muxer_msg_iter->last_returned_ts_ns = next_return_ts;
1011
1012 end:
1013         return status;
1014 }
1015
1016 static
1017 bt_self_message_iterator_status muxer_msg_iter_do_next(
1018                 struct muxer_comp *muxer_comp,
1019                 struct muxer_msg_iter *muxer_msg_iter,
1020                 bt_message_array_const msgs, uint64_t capacity,
1021                 uint64_t *count)
1022 {
1023         bt_self_message_iterator_status status =
1024                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1025         uint64_t i = 0;
1026
1027         while (i < capacity && status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1028                 status = muxer_msg_iter_do_next_one(muxer_comp,
1029                         muxer_msg_iter, &msgs[i]);
1030                 if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1031                         i++;
1032                 }
1033         }
1034
1035         if (i > 0) {
1036                 /*
1037                  * Even if muxer_msg_iter_do_next_one() returned
1038                  * something else than
1039                  * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1040                  * message objects in the output message
1041                  * array, so we need to return
1042                  * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1043                  * transfered to downstream. This other status occurs
1044                  * again the next time muxer_msg_iter_do_next() is
1045                  * called, possibly without any accumulated
1046                  * message, in which case we'll return it.
1047                  */
1048                 *count = i;
1049                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1050         }
1051
1052         return status;
1053 }
1054
1055 static
1056 void destroy_muxer_msg_iter(struct muxer_msg_iter *muxer_msg_iter)
1057 {
1058         if (!muxer_msg_iter) {
1059                 return;
1060         }
1061
1062         BT_LOGD("Destroying muxer component's message iterator: "
1063                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
1064
1065         if (muxer_msg_iter->muxer_upstream_msg_iters) {
1066                 BT_LOGD_STR("Destroying muxer's upstream message iterator wrappers.");
1067                 g_ptr_array_free(
1068                         muxer_msg_iter->muxer_upstream_msg_iters, TRUE);
1069         }
1070
1071         g_free(muxer_msg_iter);
1072 }
1073
1074 static
1075 int muxer_msg_iter_init_upstream_iterators(struct muxer_comp *muxer_comp,
1076                 struct muxer_msg_iter *muxer_msg_iter)
1077 {
1078         int64_t count;
1079         int64_t i;
1080         int ret = 0;
1081
1082         count = bt_component_filter_get_input_port_count(
1083                 bt_self_component_filter_as_component_filter(
1084                         muxer_comp->self_comp));
1085         if (count < 0) {
1086                 BT_LOGD("No input port to initialize for muxer component's message iterator: "
1087                         "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
1088                         muxer_comp, muxer_msg_iter);
1089                 goto end;
1090         }
1091
1092         for (i = 0; i < count; i++) {
1093                 bt_self_component_port_input_message_iterator *upstream_msg_iter;
1094                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter;
1095                 bt_self_component_port_input *self_port =
1096                         bt_self_component_filter_borrow_input_port_by_index(
1097                                 muxer_comp->self_comp, i);
1098                 const bt_port *port;
1099
1100                 BT_ASSERT(self_port);
1101                 port = bt_self_component_port_as_port(
1102                         bt_self_component_port_input_as_self_component_port(
1103                                 self_port));
1104                 BT_ASSERT(port);
1105
1106                 if (!bt_port_is_connected(port)) {
1107                         /* Skip non-connected port */
1108                         continue;
1109                 }
1110
1111                 upstream_msg_iter = create_msg_iter_on_input_port(self_port);
1112                 if (!upstream_msg_iter) {
1113                         /* create_msg_iter_on_input_port() logs errors */
1114                         BT_ASSERT(!upstream_msg_iter);
1115                         ret = -1;
1116                         goto end;
1117                 }
1118
1119                 muxer_upstream_msg_iter =
1120                         muxer_msg_iter_add_upstream_msg_iter(
1121                                 muxer_msg_iter, upstream_msg_iter);
1122                 bt_self_component_port_input_message_iterator_put_ref(
1123                         upstream_msg_iter);
1124                 if (!muxer_upstream_msg_iter) {
1125                         /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
1126                         ret = -1;
1127                         goto end;
1128                 }
1129         }
1130
1131 end:
1132         return ret;
1133 }
1134
1135 BT_HIDDEN
1136 bt_self_message_iterator_status muxer_msg_iter_init(
1137                 bt_self_message_iterator *self_msg_iter,
1138                 bt_self_component_filter *self_comp,
1139                 bt_self_component_port_output *port)
1140 {
1141         struct muxer_comp *muxer_comp = NULL;
1142         struct muxer_msg_iter *muxer_msg_iter = NULL;
1143         bt_self_message_iterator_status status =
1144                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1145         int ret;
1146
1147         muxer_comp = bt_self_component_get_data(
1148                 bt_self_component_filter_as_self_component(self_comp));
1149         BT_ASSERT(muxer_comp);
1150         BT_LOGD("Initializing muxer component's message iterator: "
1151                 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1152                 self_comp, muxer_comp, self_msg_iter);
1153
1154         if (muxer_comp->initializing_muxer_msg_iter) {
1155                 /*
1156                  * Weird, unhandled situation detected: downstream
1157                  * creates a muxer message iterator while creating
1158                  * another muxer message iterator (same component).
1159                  */
1160                 BT_LOGE("Recursive initialization of muxer component's message iterator: "
1161                         "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1162                         self_comp, muxer_comp, self_msg_iter);
1163                 goto error;
1164         }
1165
1166         muxer_comp->initializing_muxer_msg_iter = true;
1167         muxer_msg_iter = g_new0(struct muxer_msg_iter, 1);
1168         if (!muxer_msg_iter) {
1169                 BT_LOGE_STR("Failed to allocate one muxer component's message iterator.");
1170                 goto error;
1171         }
1172
1173         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1174         muxer_msg_iter->muxer_upstream_msg_iters =
1175                 g_ptr_array_new_with_free_func(
1176                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1177         if (!muxer_msg_iter->muxer_upstream_msg_iters) {
1178                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
1179                 goto error;
1180         }
1181
1182         ret = muxer_msg_iter_init_upstream_iterators(muxer_comp,
1183                 muxer_msg_iter);
1184         if (ret) {
1185                 BT_LOGE("Cannot initialize connected input ports for muxer component's message iterator: "
1186                         "comp-addr=%p, muxer-comp-addr=%p, "
1187                         "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1188                         self_comp, muxer_comp, muxer_msg_iter,
1189                         self_msg_iter, ret);
1190                 goto error;
1191         }
1192
1193         bt_self_message_iterator_set_data(self_msg_iter, muxer_msg_iter);
1194         BT_LOGD("Initialized muxer component's message iterator: "
1195                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1196                 "msg-iter-addr=%p",
1197                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1198         goto end;
1199
1200 error:
1201         destroy_muxer_msg_iter(muxer_msg_iter);
1202         bt_self_message_iterator_set_data(self_msg_iter, NULL);
1203         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1204
1205 end:
1206         muxer_comp->initializing_muxer_msg_iter = false;
1207         return status;
1208 }
1209
1210 BT_HIDDEN
1211 void muxer_msg_iter_finalize(
1212                 bt_self_message_iterator *self_msg_iter)
1213 {
1214         struct muxer_msg_iter *muxer_msg_iter =
1215                 bt_self_message_iterator_get_data(self_msg_iter);
1216         bt_self_component *self_comp = NULL;
1217         struct muxer_comp *muxer_comp = NULL;
1218
1219         self_comp = bt_self_message_iterator_borrow_component(
1220                 self_msg_iter);
1221         BT_ASSERT(self_comp);
1222         muxer_comp = bt_self_component_get_data(self_comp);
1223         BT_LOGD("Finalizing muxer component's message iterator: "
1224                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1225                 "msg-iter-addr=%p",
1226                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1227
1228         if (muxer_msg_iter) {
1229                 destroy_muxer_msg_iter(muxer_msg_iter);
1230         }
1231 }
1232
1233 BT_HIDDEN
1234 bt_self_message_iterator_status muxer_msg_iter_next(
1235                 bt_self_message_iterator *self_msg_iter,
1236                 bt_message_array_const msgs, uint64_t capacity,
1237                 uint64_t *count)
1238 {
1239         bt_self_message_iterator_status status;
1240         struct muxer_msg_iter *muxer_msg_iter =
1241                 bt_self_message_iterator_get_data(self_msg_iter);
1242         bt_self_component *self_comp = NULL;
1243         struct muxer_comp *muxer_comp = NULL;
1244
1245         BT_ASSERT(muxer_msg_iter);
1246         self_comp = bt_self_message_iterator_borrow_component(
1247                 self_msg_iter);
1248         BT_ASSERT(self_comp);
1249         muxer_comp = bt_self_component_get_data(self_comp);
1250         BT_ASSERT(muxer_comp);
1251         BT_LOGV("Muxer component's message iterator's \"next\" method called: "
1252                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1253                 "msg-iter-addr=%p",
1254                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1255
1256         status = muxer_msg_iter_do_next(muxer_comp, muxer_msg_iter,
1257                 msgs, capacity, count);
1258         if (status < 0) {
1259                 BT_LOGE("Cannot get next message: "
1260                         "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1261                         "msg-iter-addr=%p, status=%s",
1262                         self_comp, muxer_comp, muxer_msg_iter, self_msg_iter,
1263                         bt_self_message_iterator_status_string(status));
1264         } else {
1265                 BT_LOGV("Returning from muxer component's message iterator's \"next\" method: "
1266                         "status=%s",
1267                         bt_self_message_iterator_status_string(status));
1268         }
1269
1270         return status;
1271 }
1272
1273 BT_HIDDEN
1274 bt_self_component_status muxer_input_port_connected(
1275                 bt_self_component_filter *self_comp,
1276                 bt_self_component_port_input *self_port,
1277                 const bt_port_output *other_port)
1278 {
1279         bt_self_component_status status;
1280
1281         status = add_available_input_port(self_comp);
1282         if (status) {
1283                 /*
1284                  * Only way to report an error later since this
1285                  * method does not return anything.
1286                  */
1287                 BT_LOGE("Cannot add one muxer component's input port: "
1288                         "status=%s",
1289                         bt_self_component_status_string(status));
1290                 goto end;
1291         }
1292
1293 end:
1294         return status;
1295 }
1296
1297 BT_HIDDEN
1298 bt_bool muxer_msg_iter_can_seek_beginning(
1299                 bt_self_message_iterator *self_msg_iter)
1300 {
1301         struct muxer_msg_iter *muxer_msg_iter =
1302                 bt_self_message_iterator_get_data(self_msg_iter);
1303         uint64_t i;
1304         bt_bool ret = BT_TRUE;
1305
1306         for (i = 0; i < muxer_msg_iter->muxer_upstream_msg_iters->len; i++) {
1307                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1308                         muxer_msg_iter->muxer_upstream_msg_iters->pdata[i];
1309
1310                 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
1311                                 upstream_msg_iter->msg_iter)) {
1312                         ret = BT_FALSE;
1313                         goto end;
1314                 }
1315         }
1316
1317 end:
1318         return ret;
1319 }
1320
1321 BT_HIDDEN
1322 bt_self_message_iterator_status muxer_msg_iter_seek_beginning(
1323                 bt_self_message_iterator *self_msg_iter)
1324 {
1325         struct muxer_msg_iter *muxer_msg_iter =
1326                 bt_self_message_iterator_get_data(self_msg_iter);
1327         int status;
1328         uint64_t i;
1329
1330         for (i = 0; i < muxer_msg_iter->muxer_upstream_msg_iters->len; i++) {
1331                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1332                         muxer_msg_iter->muxer_upstream_msg_iters->pdata[i];
1333
1334                 status = bt_self_component_port_input_message_iterator_seek_beginning(
1335                         upstream_msg_iter->msg_iter);
1336                 if (status != BT_MESSAGE_ITERATOR_STATUS_OK) {
1337                         goto end;
1338                 }
1339         }
1340
1341         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1342
1343 end:
1344         return status;
1345 }