src/plugins/utils/muxer/muxer.c

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