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