plugins/utils/muxer/muxer.c

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