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