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