plugins/utils/muxer/muxer.c

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