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 enum 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         enum 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 enum 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 enum 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         enum 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 enum bt_message_iterator_status muxer_upstream_msg_iter_next(
 451                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
 452 {
 453         enum 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         case BT_MESSAGE_ITERATOR_STATUS_CANCELED:
 496                 /*
 497                  * Message iterator reached the end: release it. It
 498                  * won't be considered again to find the youngest
 499                  * message.
 500                  */
 501                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_PUT_REF_AND_RESET(muxer_upstream_msg_iter->msg_iter);
 502                 status = BT_MESSAGE_ITERATOR_STATUS_OK;
 503                 break;
 504         default:
 505                 /* Error or unsupported status code */
 506                 BT_LOGE("Error or unsupported status code: "
 507                         "status-code=%d", status);
 508                 status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
 509                 break;
 510         }
 511
 512         return status;
 513 }
 514
 515 static
 516 int muxer_msg_iter_handle_newly_connected_ports(
 517                 struct muxer_msg_iter *muxer_msg_iter)
 518 {
 519         int ret = 0;
 520
 521         BT_LOGV("Handling newly connected ports: "
 522                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
 523
 524         /*
 525          * Here we create one upstream message iterator for each
 526          * newly connected port. We do NOT perform an initial "next" on
 527          * those new upstream message iterators: they are
 528          * invalidated, to be validated later. The list of newly
 529          * connected ports to handle here is updated by
 530          * muxer_port_connected().
 531          */
 532         while (true) {
 533                 GList *node = muxer_msg_iter->newly_connected_self_ports;
 534                 bt_self_component_port_input *self_port;
 535                 const bt_port *port;
 536                 bt_self_component_port_input_message_iterator *
 537                         upstream_msg_iter = NULL;
 538                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter;
 539
 540                 if (!node) {
 541                         break;
 542                 }
 543
 544                 self_port = node->data;
 545                 port = bt_self_component_port_as_port(
 546                         bt_self_component_port_input_as_self_component_port(
 547                                 (self_port)));
 548                 BT_ASSERT(port);
 549
 550                 if (!bt_port_is_connected(port)) {
 551                         /*
 552                          * Looks like this port is not connected
 553                          * anymore: we can't create an upstream
 554                          * message iterator on its (non-existing)
 555                          * connection in this case.
 556                          */
 557                         goto remove_node;
 558                 }
 559
 560                 upstream_msg_iter = create_msg_iter_on_input_port(
 561                         self_port, &ret);
 562                 if (ret) {
 563                         /* create_msg_iter_on_input_port() logs errors */
 564                         BT_ASSERT(!upstream_msg_iter);
 565                         goto error;
 566                 }
 567
 568                 muxer_upstream_msg_iter =
 569                         muxer_msg_iter_add_upstream_msg_iter(
 570                                 muxer_msg_iter, upstream_msg_iter);
 571                 BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_PUT_REF_AND_RESET(upstream_msg_iter);
 572                 if (!muxer_upstream_msg_iter) {
 573                         /*
 574                          * muxer_msg_iter_add_upstream_msg_iter()
 575                          * logs errors.
 576                          */
 577                         goto error;
 578                 }
 579
 580 remove_node:
 581                 bt_self_component_port_input_message_iterator_put_ref(upstream_msg_iter);
 582                 muxer_msg_iter->newly_connected_self_ports =
 583                         g_list_delete_link(
 584                                 muxer_msg_iter->newly_connected_self_ports,
 585                                 node);
 586         }
 587
 588         goto end;
 589
 590 error:
 591         if (ret >= 0) {
 592                 ret = -1;
 593         }
 594
 595 end:
 596         return ret;
 597 }
 598
 599 static
 600 int get_msg_ts_ns(struct muxer_comp *muxer_comp,
 601                 struct muxer_msg_iter *muxer_msg_iter,
 602                 const bt_message *msg, int64_t last_returned_ts_ns,
 603                 int64_t *ts_ns)
 604 {
 605         const bt_clock_class *clock_class = NULL;
 606         const bt_clock_value *clock_value = NULL;
 607         const bt_event *event = NULL;
 608         int ret = 0;
 609         const unsigned char *cc_uuid;
 610         const char *cc_name;
 611         enum bt_clock_value_status cv_status = BT_CLOCK_VALUE_STATUS_KNOWN;
 612
 613         BT_ASSERT(msg);
 614         BT_ASSERT(ts_ns);
 615
 616         BT_LOGV("Getting message's timestamp: "
 617                 "muxer-msg-iter-addr=%p, msg-addr=%p, "
 618                 "last-returned-ts=%" PRId64,
 619                 muxer_msg_iter, msg, last_returned_ts_ns);
 620
 621         switch (bt_message_get_type(msg)) {
 622         case BT_MESSAGE_TYPE_EVENT:
 623                 event = bt_message_event_borrow_event_const(msg);
 624                 BT_ASSERT(event);
 625                 cv_status = bt_event_borrow_default_clock_value_const(event,
 626                         &clock_value);
 627                 break;
 628
 629         case BT_MESSAGE_TYPE_INACTIVITY:
 630                 clock_value =
 631                         bt_message_inactivity_borrow_default_clock_value_const(
 632                                 msg);
 633                 break;
 634         default:
 635                 /* All the other messages have a higher priority */
 636                 BT_LOGV_STR("Message has no timestamp: using the last returned timestamp.");
 637                 *ts_ns = last_returned_ts_ns;
 638                 goto end;
 639         }
 640
 641         if (cv_status != BT_CLOCK_VALUE_STATUS_KNOWN) {
 642                 BT_LOGE_STR("Unsupported unknown clock value.");
 643                 ret = -1;
 644                 goto end;
 645         }
 646
 647         /*
 648          * If the clock value is missing, then we consider that this
 649          * message has no time. In this case it's always the
 650          * youngest.
 651          */
 652         if (!clock_value) {
 653                 BT_LOGV_STR("Message's default clock value is missing: "
 654                         "using the last returned timestamp.");
 655                 *ts_ns = last_returned_ts_ns;
 656                 goto end;
 657         }
 658
 659         clock_class = bt_clock_value_borrow_clock_class_const(clock_value);
 660         BT_ASSERT(clock_class);
 661         cc_uuid = bt_clock_class_get_uuid(clock_class);
 662         cc_name = bt_clock_class_get_name(clock_class);
 663
 664         if (muxer_msg_iter->clock_class_expectation ==
 665                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 666                 /*
 667                  * This is the first clock class that this muxer
 668                  * message iterator encounters. Its properties
 669                  * determine what to expect for the whole lifetime of
 670                  * the iterator without a true
 671                  * `assume-absolute-clock-classes` parameter.
 672                  */
 673                 if (bt_clock_class_is_absolute(clock_class)) {
 674                         /* Expect absolute clock classes */
 675                         muxer_msg_iter->clock_class_expectation =
 676                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE;
 677                 } else {
 678                         if (cc_uuid) {
 679                                 /*
 680                                  * Expect non-absolute clock classes
 681                                  * with a specific UUID.
 682                                  */
 683                                 muxer_msg_iter->clock_class_expectation =
 684                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID;
 685                                 memcpy(muxer_msg_iter->expected_clock_class_uuid,
 686                                         cc_uuid, BABELTRACE_UUID_LEN);
 687                         } else {
 688                                 /*
 689                                  * Expect non-absolute clock classes
 690                                  * with no UUID.
 691                                  */
 692                                 muxer_msg_iter->clock_class_expectation =
 693                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID;
 694                         }
 695                 }
 696         }
 697
 698         if (!muxer_comp->assume_absolute_clock_classes) {
 699                 switch (muxer_msg_iter->clock_class_expectation) {
 700                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE:
 701                         if (!bt_clock_class_is_absolute(clock_class)) {
 702                                 BT_LOGE("Expecting an absolute clock class, "
 703                                         "but got a non-absolute one: "
 704                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 705                                         clock_class, cc_name);
 706                                 goto error;
 707                         }
 708                         break;
 709                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID:
 710                         if (bt_clock_class_is_absolute(clock_class)) {
 711                                 BT_LOGE("Expecting a non-absolute clock class with no UUID, "
 712                                         "but got an absolute one: "
 713                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 714                                         clock_class, cc_name);
 715                                 goto error;
 716                         }
 717
 718                         if (cc_uuid) {
 719                                 BT_LOGE("Expecting a non-absolute clock class with no UUID, "
 720                                         "but got one with a UUID: "
 721                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 722                                         "uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\"",
 723                                         clock_class, cc_name,
 724                                         (unsigned int) cc_uuid[0],
 725                                         (unsigned int) cc_uuid[1],
 726                                         (unsigned int) cc_uuid[2],
 727                                         (unsigned int) cc_uuid[3],
 728                                         (unsigned int) cc_uuid[4],
 729                                         (unsigned int) cc_uuid[5],
 730                                         (unsigned int) cc_uuid[6],
 731                                         (unsigned int) cc_uuid[7],
 732                                         (unsigned int) cc_uuid[8],
 733                                         (unsigned int) cc_uuid[9],
 734                                         (unsigned int) cc_uuid[10],
 735                                         (unsigned int) cc_uuid[11],
 736                                         (unsigned int) cc_uuid[12],
 737                                         (unsigned int) cc_uuid[13],
 738                                         (unsigned int) cc_uuid[14],
 739                                         (unsigned int) cc_uuid[15]);
 740                                 goto error;
 741                         }
 742                         break;
 743                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID:
 744                         if (bt_clock_class_is_absolute(clock_class)) {
 745                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 746                                         "but got an absolute one: "
 747                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 748                                         clock_class, cc_name);
 749                                 goto error;
 750                         }
 751
 752                         if (!cc_uuid) {
 753                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 754                                         "but got one with no UUID: "
 755                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 756                                         clock_class, cc_name);
 757                                 goto error;
 758                         }
 759
 760                         if (memcmp(muxer_msg_iter->expected_clock_class_uuid,
 761                                         cc_uuid, BABELTRACE_UUID_LEN) != 0) {
 762                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 763                                         "but got one with different UUID: "
 764                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 765                                         "expected-uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\", "
 766                                         "uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\"",
 767                                         clock_class, cc_name,
 768                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[0],
 769                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[1],
 770                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[2],
 771                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[3],
 772                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[4],
 773                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[5],
 774                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[6],
 775                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[7],
 776                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[8],
 777                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[9],
 778                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[10],
 779                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[11],
 780                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[12],
 781                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[13],
 782                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[14],
 783                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[15],
 784                                         (unsigned int) cc_uuid[0],
 785                                         (unsigned int) cc_uuid[1],
 786                                         (unsigned int) cc_uuid[2],
 787                                         (unsigned int) cc_uuid[3],
 788                                         (unsigned int) cc_uuid[4],
 789                                         (unsigned int) cc_uuid[5],
 790                                         (unsigned int) cc_uuid[6],
 791                                         (unsigned int) cc_uuid[7],
 792                                         (unsigned int) cc_uuid[8],
 793                                         (unsigned int) cc_uuid[9],
 794                                         (unsigned int) cc_uuid[10],
 795                                         (unsigned int) cc_uuid[11],
 796                                         (unsigned int) cc_uuid[12],
 797                                         (unsigned int) cc_uuid[13],
 798                                         (unsigned int) cc_uuid[14],
 799                                         (unsigned int) cc_uuid[15]);
 800                                 goto error;
 801                         }
 802                         break;
 803                 default:
 804                         /* Unexpected */
 805                         BT_LOGF("Unexpected clock class expectation: "
 806                                 "expectation-code=%d",
 807                                 muxer_msg_iter->clock_class_expectation);
 808                         abort();
 809                 }
 810         }
 811
 812         ret = bt_clock_value_get_ns_from_origin(clock_value, ts_ns);
 813         if (ret) {
 814                 BT_LOGE("Cannot get nanoseconds from Epoch of clock value: "
 815                         "clock-value-addr=%p", clock_value);
 816                 goto error;
 817         }
 818
 819         goto end;
 820
 821 error:
 822         ret = -1;
 823
 824 end:
 825         if (ret == 0) {
 826                 BT_LOGV("Found message's timestamp: "
 827                         "muxer-msg-iter-addr=%p, msg-addr=%p, "
 828                         "last-returned-ts=%" PRId64 ", ts=%" PRId64,
 829                         muxer_msg_iter, msg, last_returned_ts_ns,
 830                         *ts_ns);
 831         }
 832
 833         return ret;
 834 }
 835
 836 /*
 837  * This function finds the youngest available message amongst the
 838  * non-ended upstream message iterators and returns the upstream
 839  * message iterator which has it, or
 840  * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
 841  * message.
 842  *
 843  * This function does NOT:
 844  *
 845  * * Update any upstream message iterator.
 846  * * Check for newly connected ports.
 847  * * Check the upstream message iterators to retry.
 848  *
 849  * On sucess, this function sets *muxer_upstream_msg_iter to the
 850  * upstream message iterator of which the current message is
 851  * the youngest, and sets *ts_ns to its time.
 852  */
 853 static
 854 enum bt_message_iterator_status
 855 muxer_msg_iter_youngest_upstream_msg_iter(
 856                 struct muxer_comp *muxer_comp,
 857                 struct muxer_msg_iter *muxer_msg_iter,
 858                 struct muxer_upstream_msg_iter **muxer_upstream_msg_iter,
 859                 int64_t *ts_ns)
 860 {
 861         size_t i;
 862         int ret;
 863         int64_t youngest_ts_ns = INT64_MAX;
 864         enum bt_message_iterator_status status =
 865                 BT_MESSAGE_ITERATOR_STATUS_OK;
 866
 867         BT_ASSERT(muxer_comp);
 868         BT_ASSERT(muxer_msg_iter);
 869         BT_ASSERT(muxer_upstream_msg_iter);
 870         *muxer_upstream_msg_iter = NULL;
 871
 872         for (i = 0; i < muxer_msg_iter->muxer_upstream_msg_iters->len; i++) {
 873                 const bt_message *msg;
 874                 struct muxer_upstream_msg_iter *cur_muxer_upstream_msg_iter =
 875                         g_ptr_array_index(muxer_msg_iter->muxer_upstream_msg_iters, i);
 876                 int64_t msg_ts_ns;
 877
 878                 if (!cur_muxer_upstream_msg_iter->msg_iter) {
 879                         /* This upstream message iterator is ended */
 880                         BT_LOGV("Skipping ended upstream message iterator: "
 881                                 "muxer-upstream-msg-iter-wrap-addr=%p",
 882                                 cur_muxer_upstream_msg_iter);
 883                         continue;
 884                 }
 885
 886                 BT_ASSERT(cur_muxer_upstream_msg_iter->msgs->length > 0);
 887                 msg = g_queue_peek_head(cur_muxer_upstream_msg_iter->msgs);
 888                 BT_ASSERT(msg);
 889                 ret = get_msg_ts_ns(muxer_comp, muxer_msg_iter, msg,
 890                         muxer_msg_iter->last_returned_ts_ns, &msg_ts_ns);
 891                 if (ret) {
 892                         /* get_msg_ts_ns() logs errors */
 893                         *muxer_upstream_msg_iter = NULL;
 894                         status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
 895                         goto end;
 896                 }
 897
 898                 if (msg_ts_ns <= youngest_ts_ns) {
 899                         *muxer_upstream_msg_iter =
 900                                 cur_muxer_upstream_msg_iter;
 901                         youngest_ts_ns = msg_ts_ns;
 902                         *ts_ns = youngest_ts_ns;
 903                 }
 904         }
 905
 906         if (!*muxer_upstream_msg_iter) {
 907                 status = BT_MESSAGE_ITERATOR_STATUS_END;
 908                 *ts_ns = INT64_MIN;
 909         }
 910
 911 end:
 912         return status;
 913 }
 914
 915 static
 916 enum bt_message_iterator_status validate_muxer_upstream_msg_iter(
 917         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
 918 {
 919         enum bt_message_iterator_status status =
 920                 BT_MESSAGE_ITERATOR_STATUS_OK;
 921
 922         BT_LOGV("Validating muxer's upstream message iterator wrapper: "
 923                 "muxer-upstream-msg-iter-wrap-addr=%p",
 924                 muxer_upstream_msg_iter);
 925
 926         if (muxer_upstream_msg_iter->msgs->length > 0 ||
 927                         !muxer_upstream_msg_iter->msg_iter) {
 928                 BT_LOGV("Already valid or not considered: "
 929                         "queue-len=%u, upstream-msg-iter-addr=%p",
 930                         muxer_upstream_msg_iter->msgs->length,
 931                         muxer_upstream_msg_iter->msg_iter);
 932                 goto end;
 933         }
 934
 935         /* muxer_upstream_msg_iter_next() logs details/errors */
 936         status = muxer_upstream_msg_iter_next(muxer_upstream_msg_iter);
 937
 938 end:
 939         return status;
 940 }
 941
 942 static
 943 enum bt_message_iterator_status validate_muxer_upstream_msg_iters(
 944                 struct muxer_msg_iter *muxer_msg_iter)
 945 {
 946         enum bt_message_iterator_status status =
 947                 BT_MESSAGE_ITERATOR_STATUS_OK;
 948         size_t i;
 949
 950         BT_LOGV("Validating muxer's upstream message iterator wrappers: "
 951                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
 952
 953         for (i = 0; i < muxer_msg_iter->muxer_upstream_msg_iters->len; i++) {
 954                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
 955                         g_ptr_array_index(
 956                                 muxer_msg_iter->muxer_upstream_msg_iters,
 957                                 i);
 958
 959                 status = validate_muxer_upstream_msg_iter(
 960                         muxer_upstream_msg_iter);
 961                 if (status != BT_MESSAGE_ITERATOR_STATUS_OK) {
 962                         if (status < 0) {
 963                                 BT_LOGE("Cannot validate muxer's upstream message iterator wrapper: "
 964                                         "muxer-msg-iter-addr=%p, "
 965                                         "muxer-upstream-msg-iter-wrap-addr=%p",
 966                                         muxer_msg_iter,
 967                                         muxer_upstream_msg_iter);
 968                         } else {
 969                                 BT_LOGV("Cannot validate muxer's upstream message iterator wrapper: "
 970                                         "muxer-msg-iter-addr=%p, "
 971                                         "muxer-upstream-msg-iter-wrap-addr=%p",
 972                                         muxer_msg_iter,
 973                                         muxer_upstream_msg_iter);
 974                         }
 975
 976                         goto end;
 977                 }
 978
 979                 /*
 980                  * Remove this muxer upstream message iterator
 981                  * if it's ended or canceled.
 982                  */
 983                 if (!muxer_upstream_msg_iter->msg_iter) {
 984                         /*
 985                          * Use g_ptr_array_remove_fast() because the
 986                          * order of those elements is not important.
 987                          */
 988                         BT_LOGV("Removing muxer's upstream message iterator wrapper: ended or canceled: "
 989                                 "muxer-msg-iter-addr=%p, "
 990                                 "muxer-upstream-msg-iter-wrap-addr=%p",
 991                                 muxer_msg_iter, muxer_upstream_msg_iter);
 992                         g_ptr_array_remove_index_fast(
 993                                 muxer_msg_iter->muxer_upstream_msg_iters,
 994                                 i);
 995                         i--;
 996                 }
 997         }
 998
 999 end:
1000         return status;
1001 }
1002
1003 static inline
1004 enum bt_message_iterator_status muxer_msg_iter_do_next_one(
1005                 struct muxer_comp *muxer_comp,
1006                 struct muxer_msg_iter *muxer_msg_iter,
1007                 const bt_message **msg)
1008 {
1009         enum bt_message_iterator_status status =
1010                 BT_MESSAGE_ITERATOR_STATUS_OK;
1011         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter = NULL;
1012         int64_t next_return_ts;
1013
1014         while (true) {
1015                 int ret = muxer_msg_iter_handle_newly_connected_ports(
1016                         muxer_msg_iter);
1017
1018                 if (ret) {
1019                         BT_LOGE("Cannot handle newly connected input ports for muxer's message iterator: "
1020                                 "muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1021                                 "ret=%d",
1022                                 muxer_comp, muxer_msg_iter, ret);
1023                         status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
1024                         goto end;
1025                 }
1026
1027                 status = validate_muxer_upstream_msg_iters(muxer_msg_iter);
1028                 if (status != BT_MESSAGE_ITERATOR_STATUS_OK) {
1029                         /* validate_muxer_upstream_msg_iters() logs details */
1030                         goto end;
1031                 }
1032
1033                 /*
1034                  * At this point, we know that all the existing upstream
1035                  * message iterators are valid. However the
1036                  * operations to validate them (during
1037                  * validate_muxer_upstream_msg_iters()) may have
1038                  * connected new ports. If no ports were connected
1039                  * during this operation, exit the loop.
1040                  */
1041                 if (!muxer_msg_iter->newly_connected_self_ports) {
1042                         BT_LOGV("Not breaking this loop: muxer's message iterator still has newly connected input ports to handle: "
1043                                 "muxer-comp-addr=%p", muxer_comp);
1044                         break;
1045                 }
1046         }
1047
1048         BT_ASSERT(!muxer_msg_iter->newly_connected_self_ports);
1049
1050         /*
1051          * At this point we know that all the existing upstream
1052          * message iterators are valid. We can find the one,
1053          * amongst those, of which the current message is the
1054          * youngest.
1055          */
1056         status = muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp,
1057                         muxer_msg_iter, &muxer_upstream_msg_iter,
1058                         &next_return_ts);
1059         if (status < 0 || status == BT_MESSAGE_ITERATOR_STATUS_END ||
1060                         status == BT_MESSAGE_ITERATOR_STATUS_CANCELED) {
1061                 if (status < 0) {
1062                         BT_LOGE("Cannot find the youngest upstream message iterator wrapper: "
1063                                 "status=%s",
1064                                 bt_message_iterator_status_string(status));
1065                 } else {
1066                         BT_LOGV("Cannot find the youngest upstream message iterator wrapper: "
1067                                 "status=%s",
1068                                 bt_message_iterator_status_string(status));
1069                 }
1070
1071                 goto end;
1072         }
1073
1074         if (next_return_ts < muxer_msg_iter->last_returned_ts_ns) {
1075                 BT_LOGE("Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
1076                         "muxer-msg-iter-addr=%p, ts=%" PRId64 ", "
1077                         "last-returned-ts=%" PRId64,
1078                         muxer_msg_iter, next_return_ts,
1079                         muxer_msg_iter->last_returned_ts_ns);
1080                 status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
1081                 goto end;
1082         }
1083
1084         BT_LOGV("Found youngest upstream message iterator wrapper: "
1085                 "muxer-msg-iter-addr=%p, "
1086                 "muxer-upstream-msg-iter-wrap-addr=%p, "
1087                 "ts=%" PRId64,
1088                 muxer_msg_iter, muxer_upstream_msg_iter, next_return_ts);
1089         BT_ASSERT(status == BT_MESSAGE_ITERATOR_STATUS_OK);
1090         BT_ASSERT(muxer_upstream_msg_iter);
1091
1092         /*
1093          * Consume from the queue's head: other side
1094          * (muxer_upstream_msg_iter_next()) writes to the tail.
1095          */
1096         *msg = g_queue_pop_head(muxer_upstream_msg_iter->msgs);
1097         BT_ASSERT(*msg);
1098         muxer_msg_iter->last_returned_ts_ns = next_return_ts;
1099
1100 end:
1101         return status;
1102 }
1103
1104 static
1105 enum bt_message_iterator_status muxer_msg_iter_do_next(
1106                 struct muxer_comp *muxer_comp,
1107                 struct muxer_msg_iter *muxer_msg_iter,
1108                 bt_message_array_const msgs, uint64_t capacity,
1109                 uint64_t *count)
1110 {
1111         enum bt_message_iterator_status status =
1112                 BT_MESSAGE_ITERATOR_STATUS_OK;
1113         uint64_t i = 0;
1114
1115         while (i < capacity && status == BT_MESSAGE_ITERATOR_STATUS_OK) {
1116                 status = muxer_msg_iter_do_next_one(muxer_comp,
1117                         muxer_msg_iter, &msgs[i]);
1118                 if (status == BT_MESSAGE_ITERATOR_STATUS_OK) {
1119                         i++;
1120                 }
1121         }
1122
1123         if (i > 0) {
1124                 /*
1125                  * Even if muxer_msg_iter_do_next_one() returned
1126                  * something else than
1127                  * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1128                  * message objects in the output message
1129                  * array, so we need to return
1130                  * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1131                  * transfered to downstream. This other status occurs
1132                  * again the next time muxer_msg_iter_do_next() is
1133                  * called, possibly without any accumulated
1134                  * message, in which case we'll return it.
1135                  */
1136                 *count = i;
1137                 status = BT_MESSAGE_ITERATOR_STATUS_OK;
1138         }
1139
1140         return status;
1141 }
1142
1143 static
1144 void destroy_muxer_msg_iter(struct muxer_msg_iter *muxer_msg_iter)
1145 {
1146         if (!muxer_msg_iter) {
1147                 return;
1148         }
1149
1150         BT_LOGD("Destroying muxer component's message iterator: "
1151                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
1152
1153         if (muxer_msg_iter->muxer_upstream_msg_iters) {
1154                 BT_LOGD_STR("Destroying muxer's upstream message iterator wrappers.");
1155                 g_ptr_array_free(
1156                         muxer_msg_iter->muxer_upstream_msg_iters, TRUE);
1157         }
1158
1159         g_list_free(muxer_msg_iter->newly_connected_self_ports);
1160         g_free(muxer_msg_iter);
1161 }
1162
1163 static
1164 int muxer_msg_iter_init_newly_connected_ports(struct muxer_comp *muxer_comp,
1165                 struct muxer_msg_iter *muxer_msg_iter)
1166 {
1167         int64_t count;
1168         int64_t i;
1169         int ret = 0;
1170
1171         /*
1172          * Add the connected input ports to this muxer message
1173          * iterator's list of newly connected ports. They will be
1174          * handled by muxer_msg_iter_handle_newly_connected_ports().
1175          */
1176         count = bt_component_filter_get_input_port_count(
1177                 bt_self_component_filter_as_component_filter(
1178                         muxer_comp->self_comp));
1179         if (count < 0) {
1180                 BT_LOGD("No input port to initialize for muxer component's message iterator: "
1181                         "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
1182                         muxer_comp, muxer_msg_iter);
1183                 goto end;
1184         }
1185
1186         for (i = 0; i < count; i++) {
1187                 bt_self_component_port_input *self_port =
1188                         bt_self_component_filter_borrow_input_port_by_index(
1189                                 muxer_comp->self_comp, i);
1190                 const bt_port *port;
1191
1192                 BT_ASSERT(self_port);
1193                 port = bt_self_component_port_as_port(
1194                         bt_self_component_port_input_as_self_component_port(
1195                                 self_port));
1196                 BT_ASSERT(port);
1197
1198                 if (!bt_port_is_connected(port)) {
1199                         BT_LOGD("Skipping input port: not connected: "
1200                                 "muxer-comp-addr=%p, port-addr=%p, port-name\"%s\"",
1201                                 muxer_comp, port, bt_port_get_name(port));
1202                         continue;
1203                 }
1204
1205                 muxer_msg_iter->newly_connected_self_ports =
1206                         g_list_append(
1207                                 muxer_msg_iter->newly_connected_self_ports,
1208                                 self_port);
1209                 if (!muxer_msg_iter->newly_connected_self_ports) {
1210                         BT_LOGE("Cannot append port to muxer's message iterator list of newly connected input ports: "
1211                                 "port-addr=%p, port-name=\"%s\", "
1212                                 "muxer-msg-iter-addr=%p", port,
1213                                 bt_port_get_name(port), muxer_msg_iter);
1214                         ret = -1;
1215                         goto end;
1216                 }
1217
1218                 BT_LOGD("Appended port to muxer's message iterator list of newly connected input ports: "
1219                         "port-addr=%p, port-name=\"%s\", "
1220                         "muxer-msg-iter-addr=%p", port,
1221                         bt_port_get_name(port), muxer_msg_iter);
1222         }
1223
1224 end:
1225         return ret;
1226 }
1227
1228 BT_HIDDEN
1229 enum bt_self_message_iterator_status muxer_msg_iter_init(
1230                 bt_self_message_iterator *self_msg_iter,
1231                 bt_self_component_filter *self_comp,
1232                 bt_self_component_port_output *port)
1233 {
1234         struct muxer_comp *muxer_comp = NULL;
1235         struct muxer_msg_iter *muxer_msg_iter = NULL;
1236         enum bt_self_message_iterator_status status =
1237                 BT_MESSAGE_ITERATOR_STATUS_OK;
1238         int ret;
1239
1240         muxer_comp = bt_self_component_get_data(
1241                 bt_self_component_filter_as_self_component(self_comp));
1242         BT_ASSERT(muxer_comp);
1243         BT_LOGD("Initializing muxer component's message iterator: "
1244                 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1245                 self_comp, muxer_comp, self_msg_iter);
1246
1247         if (muxer_comp->initializing_muxer_msg_iter) {
1248                 /*
1249                  * Weird, unhandled situation detected: downstream
1250                  * creates a muxer message iterator while creating
1251                  * another muxer message iterator (same component).
1252                  */
1253                 BT_LOGE("Recursive initialization of muxer component's message iterator: "
1254                         "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1255                         self_comp, muxer_comp, self_msg_iter);
1256                 goto error;
1257         }
1258
1259         muxer_comp->initializing_muxer_msg_iter = true;
1260         muxer_msg_iter = g_new0(struct muxer_msg_iter, 1);
1261         if (!muxer_msg_iter) {
1262                 BT_LOGE_STR("Failed to allocate one muxer component's message iterator.");
1263                 goto error;
1264         }
1265
1266         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1267         muxer_msg_iter->muxer_upstream_msg_iters =
1268                 g_ptr_array_new_with_free_func(
1269                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1270         if (!muxer_msg_iter->muxer_upstream_msg_iters) {
1271                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
1272                 goto error;
1273         }
1274
1275         /*
1276          * Add the muxer message iterator to the component's array
1277          * of muxer message iterators here because
1278          * muxer_msg_iter_init_newly_connected_ports() can cause
1279          * muxer_port_connected() to be called, which adds the newly
1280          * connected port to each muxer message iterator's list of
1281          * newly connected ports.
1282          */
1283         g_ptr_array_add(muxer_comp->muxer_msg_iters, muxer_msg_iter);
1284         ret = muxer_msg_iter_init_newly_connected_ports(muxer_comp,
1285                 muxer_msg_iter);
1286         if (ret) {
1287                 BT_LOGE("Cannot initialize newly connected input ports for muxer component's message iterator: "
1288                         "comp-addr=%p, muxer-comp-addr=%p, "
1289                         "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1290                         self_comp, muxer_comp, muxer_msg_iter,
1291                         self_msg_iter, ret);
1292                 goto error;
1293         }
1294
1295         bt_self_message_iterator_set_data(self_msg_iter,
1296                 muxer_msg_iter);
1297         BT_LOGD("Initialized muxer component's message iterator: "
1298                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1299                 "msg-iter-addr=%p",
1300                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1301         goto end;
1302
1303 error:
1304         if (g_ptr_array_index(muxer_comp->muxer_msg_iters,
1305                         muxer_comp->muxer_msg_iters->len - 1) == muxer_msg_iter) {
1306                 g_ptr_array_remove_index(muxer_comp->muxer_msg_iters,
1307                         muxer_comp->muxer_msg_iters->len - 1);
1308         }
1309
1310         destroy_muxer_msg_iter(muxer_msg_iter);
1311         bt_self_message_iterator_set_data(self_msg_iter,
1312                 NULL);
1313         status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
1314
1315 end:
1316         muxer_comp->initializing_muxer_msg_iter = false;
1317         return status;
1318 }
1319
1320 BT_HIDDEN
1321 void muxer_msg_iter_finalize(
1322                 bt_self_message_iterator *self_msg_iter)
1323 {
1324         struct muxer_msg_iter *muxer_msg_iter =
1325                 bt_self_message_iterator_get_data(self_msg_iter);
1326         bt_self_component *self_comp = NULL;
1327         struct muxer_comp *muxer_comp = NULL;
1328
1329         self_comp = bt_self_message_iterator_borrow_component(
1330                 self_msg_iter);
1331         BT_ASSERT(self_comp);
1332         muxer_comp = bt_self_component_get_data(self_comp);
1333         BT_LOGD("Finalizing muxer component's message iterator: "
1334                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1335                 "msg-iter-addr=%p",
1336                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1337
1338         if (muxer_comp) {
1339                 (void) g_ptr_array_remove_fast(muxer_comp->muxer_msg_iters,
1340                         muxer_msg_iter);
1341                 destroy_muxer_msg_iter(muxer_msg_iter);
1342         }
1343 }
1344
1345 BT_HIDDEN
1346 enum bt_message_iterator_status muxer_msg_iter_next(
1347                 bt_self_message_iterator *self_msg_iter,
1348                 bt_message_array_const msgs, uint64_t capacity,
1349                 uint64_t *count)
1350 {
1351         enum bt_message_iterator_status status;
1352         struct muxer_msg_iter *muxer_msg_iter =
1353                 bt_self_message_iterator_get_data(self_msg_iter);
1354         bt_self_component *self_comp = NULL;
1355         struct muxer_comp *muxer_comp = NULL;
1356
1357         BT_ASSERT(muxer_msg_iter);
1358         self_comp = bt_self_message_iterator_borrow_component(
1359                 self_msg_iter);
1360         BT_ASSERT(self_comp);
1361         muxer_comp = bt_self_component_get_data(self_comp);
1362         BT_ASSERT(muxer_comp);
1363         BT_LOGV("Muxer component's message iterator's \"next\" method called: "
1364                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1365                 "msg-iter-addr=%p",
1366                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1367
1368         status = muxer_msg_iter_do_next(muxer_comp, muxer_msg_iter,
1369                 msgs, capacity, count);
1370         if (status < 0) {
1371                 BT_LOGE("Cannot get next message: "
1372                         "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1373                         "msg-iter-addr=%p, status=%s",
1374                         self_comp, muxer_comp, muxer_msg_iter, self_msg_iter,
1375                         bt_message_iterator_status_string(status));
1376         } else {
1377                 BT_LOGV("Returning from muxer component's message iterator's \"next\" method: "
1378                         "status=%s",
1379                         bt_message_iterator_status_string(status));
1380         }
1381
1382         return status;
1383 }
1384
1385 BT_HIDDEN
1386 enum bt_self_component_status muxer_input_port_connected(
1387                 bt_self_component_filter *self_comp,
1388                 bt_self_component_port_input *self_port,
1389                 const bt_port_output *other_port)
1390 {
1391         enum bt_self_component_status status = BT_SELF_COMPONENT_STATUS_OK;
1392         const bt_port *port = bt_self_component_port_as_port(
1393                 bt_self_component_port_input_as_self_component_port(
1394                         self_port));
1395         struct muxer_comp *muxer_comp =
1396                 bt_self_component_get_data(
1397                         bt_self_component_filter_as_self_component(
1398                                 self_comp));
1399         size_t i;
1400         int ret;
1401
1402         BT_ASSERT(port);
1403         BT_ASSERT(muxer_comp);
1404         BT_LOGD("Port connected: "
1405                 "comp-addr=%p, muxer-comp-addr=%p, "
1406                 "port-addr=%p, port-name=\"%s\", "
1407                 "other-port-addr=%p, other-port-name=\"%s\"",
1408                 self_comp, muxer_comp, self_port, bt_port_get_name(port),
1409                 other_port,
1410                 bt_port_get_name(bt_port_output_as_port_const(other_port)));
1411
1412         for (i = 0; i < muxer_comp->muxer_msg_iters->len; i++) {
1413                 struct muxer_msg_iter *muxer_msg_iter =
1414                         g_ptr_array_index(muxer_comp->muxer_msg_iters, i);
1415
1416                 /*
1417                  * Add this port to the list of newly connected ports
1418                  * for this muxer message iterator. We append at
1419                  * the end of this list while
1420                  * muxer_msg_iter_handle_newly_connected_ports()
1421                  * removes the nodes from the beginning.
1422                  */
1423                 muxer_msg_iter->newly_connected_self_ports =
1424                         g_list_append(
1425                                 muxer_msg_iter->newly_connected_self_ports,
1426                                 self_port);
1427                 if (!muxer_msg_iter->newly_connected_self_ports) {
1428                         BT_LOGE("Cannot append port to muxer's message iterator list of newly connected input ports: "
1429                                 "port-addr=%p, port-name=\"%s\", "
1430                                 "muxer-msg-iter-addr=%p", self_port,
1431                                 bt_port_get_name(port), muxer_msg_iter);
1432                         status = BT_SELF_COMPONENT_STATUS_ERROR;
1433                         goto end;
1434                 }
1435
1436                 BT_LOGD("Appended port to muxer's message iterator list of newly connected input ports: "
1437                         "port-addr=%p, port-name=\"%s\", "
1438                         "muxer-msg-iter-addr=%p", self_port,
1439                         bt_port_get_name(port), muxer_msg_iter);
1440         }
1441
1442         /* One less available input port */
1443         muxer_comp->available_input_ports--;
1444         ret = ensure_available_input_port(self_comp);
1445         if (ret) {
1446                 /*
1447                  * Only way to report an error later since this
1448                  * method does not return anything.
1449                  */
1450                 BT_LOGE("Cannot ensure that at least one muxer component's input port is available: "
1451                         "muxer-comp-addr=%p, status=%s",
1452                         muxer_comp, bt_self_component_status_string(ret));
1453                 status = BT_SELF_COMPONENT_STATUS_ERROR;
1454                 goto end;
1455         }
1456
1457 end:
1458         return status;
1459 }
1460
1461 BT_HIDDEN
1462 void muxer_input_port_disconnected(
1463                 bt_self_component_filter *self_component,
1464                 bt_self_component_port_input *self_port)
1465 {
1466         struct muxer_comp *muxer_comp =
1467                 bt_self_component_get_data(
1468                         bt_self_component_filter_as_self_component(
1469                                 self_component));
1470         const bt_port *port =
1471                 bt_self_component_port_as_port(
1472                         bt_self_component_port_input_as_self_component_port(
1473                                 self_port));
1474
1475         BT_ASSERT(port);
1476         BT_ASSERT(muxer_comp);
1477
1478         /* One more available input port */
1479         muxer_comp->available_input_ports++;
1480         BT_LOGD("Leaving disconnected input port available for future connections: "
1481                 "comp-addr=%p, muxer-comp-addr=%p, port-addr=%p, "
1482                 "port-name=\"%s\", avail-input-port-count=%zu",
1483                 self_component, muxer_comp, port, bt_port_get_name(port),
1484                 muxer_comp->available_input_ports);
1485 }