src/plugins/utils/muxer/muxer.c

   1 /*
   2  * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
   3  *
   4  * Permission is hereby granted, free of charge, to any person obtaining a copy
   5  * of this software and associated documentation files (the "Software"), to deal
   6  * in the Software without restriction, including without limitation the rights
   7  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   8  * copies of the Software, and to permit persons to whom the Software is
   9  * furnished to do so, subject to the following conditions:
  10  *
  11  * The above copyright notice and this permission notice shall be included in
  12  * all copies or substantial portions of the Software.
  13  *
  14  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  19  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  20  * SOFTWARE.
  21  */
  22
  23 #define BT_LOG_TAG "PLUGIN-UTILS-MUXER-FLT"
  24 #include "logging.h"
  25
  26 #include "common/babeltrace.h"
  27 #include "compat/uuid.h"
  28 #include <babeltrace2/babeltrace.h>
  29 #include "lib/value.h"
  30 #include "lib/graph/component.h"
  31 #include "lib/graph/message/iterator.h"
  32 #include "lib/graph/connection.h"
  33 #include "plugins/plugins-common.h"
  34 #include <glib.h>
  35 #include <stdbool.h>
  36 #include <inttypes.h>
  37 #include "common/assert.h"
  38 #include "common/common.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         const bt_stream_class *stream_class = NULL;
 510         bt_message_type msg_type;
 511
 512         BT_ASSERT(msg);
 513         BT_ASSERT(ts_ns);
 514         BT_LOGV("Getting message's timestamp: "
 515                 "muxer-msg-iter-addr=%p, msg-addr=%p, "
 516                 "last-returned-ts=%" PRId64,
 517                 muxer_msg_iter, msg, last_returned_ts_ns);
 518
 519         if (unlikely(muxer_msg_iter->clock_class_expectation ==
 520                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE)) {
 521                 *ts_ns = last_returned_ts_ns;
 522                 goto end;
 523         }
 524
 525         msg_type = bt_message_get_type(msg);
 526
 527         if (unlikely(msg_type == BT_MESSAGE_TYPE_PACKET_BEGINNING)) {
 528                 stream_class = bt_stream_borrow_class_const(
 529                         bt_packet_borrow_stream_const(
 530                                 bt_message_packet_beginning_borrow_packet_const(
 531                                         msg)));
 532         } else if (unlikely(msg_type == BT_MESSAGE_TYPE_PACKET_END)) {
 533                 stream_class = bt_stream_borrow_class_const(
 534                         bt_packet_borrow_stream_const(
 535                                 bt_message_packet_end_borrow_packet_const(
 536                                         msg)));
 537         } else if (unlikely(msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS)) {
 538                 stream_class = bt_stream_borrow_class_const(
 539                         bt_message_discarded_events_borrow_stream_const(msg));
 540         } else if (unlikely(msg_type == BT_MESSAGE_TYPE_DISCARDED_PACKETS)) {
 541                 stream_class = bt_stream_borrow_class_const(
 542                         bt_message_discarded_packets_borrow_stream_const(msg));
 543         }
 544
 545         switch (msg_type) {
 546         case BT_MESSAGE_TYPE_EVENT:
 547                 BT_ASSERT(bt_message_event_borrow_stream_class_default_clock_class_const(
 548                                 msg));
 549                 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
 550                         msg);
 551                 break;
 552         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 553                 if (bt_stream_class_packets_have_beginning_default_clock_snapshot(
 554                                 stream_class)) {
 555                         clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
 556                                 msg);
 557                 } else {
 558                         goto no_clock_snapshot;
 559                 }
 560
 561                 break;
 562         case BT_MESSAGE_TYPE_PACKET_END:
 563                 if (bt_stream_class_packets_have_end_default_clock_snapshot(
 564                                 stream_class)) {
 565                         clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
 566                                 msg);
 567                 } else {
 568                         goto no_clock_snapshot;
 569                 }
 570
 571                 break;
 572         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 573                 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
 574                                 stream_class)) {
 575                         clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
 576                                 msg);
 577                 } else {
 578                         goto no_clock_snapshot;
 579                 }
 580
 581                 break;
 582         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 583                 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
 584                                 stream_class)) {
 585                         clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
 586                                 msg);
 587                 } else {
 588                         goto no_clock_snapshot;
 589                 }
 590
 591                 break;
 592         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
 593                 BT_ASSERT(bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
 594                                 msg));
 595                 sa_cs_state = bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
 596                         msg, &clock_snapshot);
 597                 if (sa_cs_state != BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN) {
 598                         goto no_clock_snapshot;
 599                 }
 600
 601                 break;
 602         case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
 603                 BT_ASSERT(bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
 604                                 msg));
 605                 sa_cs_state = bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
 606                         msg, &clock_snapshot);
 607                 if (sa_cs_state != BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN) {
 608                         goto no_clock_snapshot;
 609                 }
 610
 611                 break;
 612         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 613                 clock_snapshot = bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
 614                         msg);
 615                 break;
 616         default:
 617                 /* All the other messages have a higher priority */
 618                 BT_LOGV_STR("Message has no timestamp: using the last returned timestamp.");
 619                 *ts_ns = last_returned_ts_ns;
 620                 goto end;
 621         }
 622
 623         ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
 624         if (ret) {
 625                 BT_LOGE("Cannot get nanoseconds from Epoch of clock snapshot: "
 626                         "clock-snapshot-addr=%p", clock_snapshot);
 627                 goto error;
 628         }
 629
 630         goto end;
 631
 632 no_clock_snapshot:
 633         BT_LOGV_STR("Message's default clock snapshot is missing: "
 634                 "using the last returned timestamp.");
 635         *ts_ns = last_returned_ts_ns;
 636         goto end;
 637
 638 error:
 639         ret = -1;
 640
 641 end:
 642         if (ret == 0) {
 643                 BT_LOGV("Found message's timestamp: "
 644                         "muxer-msg-iter-addr=%p, msg-addr=%p, "
 645                         "last-returned-ts=%" PRId64 ", ts=%" PRId64,
 646                         muxer_msg_iter, msg, last_returned_ts_ns,
 647                         *ts_ns);
 648         }
 649
 650         return ret;
 651 }
 652
 653 static inline
 654 int validate_clock_class(struct muxer_msg_iter *muxer_msg_iter,
 655                 struct muxer_comp *muxer_comp,
 656                 const bt_clock_class *clock_class)
 657 {
 658         int ret = 0;
 659         const unsigned char *cc_uuid;
 660         const char *cc_name;
 661
 662         BT_ASSERT(clock_class);
 663         cc_uuid = bt_clock_class_get_uuid(clock_class);
 664         cc_name = bt_clock_class_get_name(clock_class);
 665
 666         if (muxer_msg_iter->clock_class_expectation ==
 667                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 668                 /*
 669                  * This is the first clock class that this muxer
 670                  * message iterator encounters. Its properties
 671                  * determine what to expect for the whole lifetime of
 672                  * the iterator without a true
 673                  * `assume-absolute-clock-classes` parameter.
 674                  */
 675                 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 676                         /* Expect absolute clock classes */
 677                         muxer_msg_iter->clock_class_expectation =
 678                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE;
 679                 } else {
 680                         if (cc_uuid) {
 681                                 /*
 682                                  * Expect non-absolute clock classes
 683                                  * with a specific UUID.
 684                                  */
 685                                 muxer_msg_iter->clock_class_expectation =
 686                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID;
 687                                 memcpy(muxer_msg_iter->expected_clock_class_uuid,
 688                                         cc_uuid, BABELTRACE_UUID_LEN);
 689                         } else {
 690                                 /*
 691                                  * Expect non-absolute clock classes
 692                                  * with no UUID.
 693                                  */
 694                                 muxer_msg_iter->clock_class_expectation =
 695                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID;
 696                         }
 697                 }
 698         }
 699
 700         if (!muxer_comp->assume_absolute_clock_classes) {
 701                 switch (muxer_msg_iter->clock_class_expectation) {
 702                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE:
 703                         if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
 704                                 BT_LOGE("Expecting an absolute clock class, "
 705                                         "but got a non-absolute one: "
 706                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 707                                         clock_class, cc_name);
 708                                 goto error;
 709                         }
 710                         break;
 711                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID:
 712                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 713                                 BT_LOGE("Expecting a non-absolute clock class with no UUID, "
 714                                         "but got an absolute one: "
 715                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 716                                         clock_class, cc_name);
 717                                 goto error;
 718                         }
 719
 720                         if (cc_uuid) {
 721                                 BT_LOGE("Expecting a non-absolute clock class with no UUID, "
 722                                         "but got one with a UUID: "
 723                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 724                                         "uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\"",
 725                                         clock_class, cc_name,
 726                                         (unsigned int) cc_uuid[0],
 727                                         (unsigned int) cc_uuid[1],
 728                                         (unsigned int) cc_uuid[2],
 729                                         (unsigned int) cc_uuid[3],
 730                                         (unsigned int) cc_uuid[4],
 731                                         (unsigned int) cc_uuid[5],
 732                                         (unsigned int) cc_uuid[6],
 733                                         (unsigned int) cc_uuid[7],
 734                                         (unsigned int) cc_uuid[8],
 735                                         (unsigned int) cc_uuid[9],
 736                                         (unsigned int) cc_uuid[10],
 737                                         (unsigned int) cc_uuid[11],
 738                                         (unsigned int) cc_uuid[12],
 739                                         (unsigned int) cc_uuid[13],
 740                                         (unsigned int) cc_uuid[14],
 741                                         (unsigned int) cc_uuid[15]);
 742                                 goto error;
 743                         }
 744                         break;
 745                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID:
 746                         if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 747                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 748                                         "but got an absolute one: "
 749                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 750                                         clock_class, cc_name);
 751                                 goto error;
 752                         }
 753
 754                         if (!cc_uuid) {
 755                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 756                                         "but got one with no UUID: "
 757                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 758                                         clock_class, cc_name);
 759                                 goto error;
 760                         }
 761
 762                         if (memcmp(muxer_msg_iter->expected_clock_class_uuid,
 763                                         cc_uuid, BABELTRACE_UUID_LEN) != 0) {
 764                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 765                                         "but got one with different UUID: "
 766                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 767                                         "expected-uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\", "
 768                                         "uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\"",
 769                                         clock_class, cc_name,
 770                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[0],
 771                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[1],
 772                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[2],
 773                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[3],
 774                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[4],
 775                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[5],
 776                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[6],
 777                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[7],
 778                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[8],
 779                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[9],
 780                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[10],
 781                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[11],
 782                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[12],
 783                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[13],
 784                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[14],
 785                                         (unsigned int) muxer_msg_iter->expected_clock_class_uuid[15],
 786                                         (unsigned int) cc_uuid[0],
 787                                         (unsigned int) cc_uuid[1],
 788                                         (unsigned int) cc_uuid[2],
 789                                         (unsigned int) cc_uuid[3],
 790                                         (unsigned int) cc_uuid[4],
 791                                         (unsigned int) cc_uuid[5],
 792                                         (unsigned int) cc_uuid[6],
 793                                         (unsigned int) cc_uuid[7],
 794                                         (unsigned int) cc_uuid[8],
 795                                         (unsigned int) cc_uuid[9],
 796                                         (unsigned int) cc_uuid[10],
 797                                         (unsigned int) cc_uuid[11],
 798                                         (unsigned int) cc_uuid[12],
 799                                         (unsigned int) cc_uuid[13],
 800                                         (unsigned int) cc_uuid[14],
 801                                         (unsigned int) cc_uuid[15]);
 802                                 goto error;
 803                         }
 804                         break;
 805                 case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE:
 806                         BT_LOGE("Expecting no clock class, but got one: "
 807                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 808                                 clock_class, cc_name);
 809                         goto error;
 810                 default:
 811                         /* Unexpected */
 812                         BT_LOGF("Unexpected clock class expectation: "
 813                                 "expectation-code=%d",
 814                                 muxer_msg_iter->clock_class_expectation);
 815                         abort();
 816                 }
 817         }
 818
 819         goto end;
 820
 821 error:
 822         ret = -1;
 823
 824 end:
 825         return ret;
 826 }
 827
 828 static inline
 829 int validate_new_stream_clock_class(struct muxer_msg_iter *muxer_msg_iter,
 830                 struct muxer_comp *muxer_comp, const bt_stream *stream)
 831 {
 832         int ret = 0;
 833         const bt_stream_class *stream_class =
 834                 bt_stream_borrow_class_const(stream);
 835         const bt_clock_class *clock_class =
 836                 bt_stream_class_borrow_default_clock_class_const(stream_class);
 837
 838         if (!clock_class) {
 839                 if (muxer_msg_iter->clock_class_expectation ==
 840                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 841                         /* Expect no clock class */
 842                         muxer_msg_iter->clock_class_expectation =
 843                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE;
 844                 } else {
 845                         BT_LOGE("Expecting stream class with a default clock class: "
 846                                 "stream-class-addr=%p, stream-class-name=\"%s\", "
 847                                 "stream-class-id=%" PRIu64,
 848                                 stream_class, bt_stream_class_get_name(stream_class),
 849                                 bt_stream_class_get_id(stream_class));
 850                         ret = -1;
 851                 }
 852
 853                 goto end;
 854         }
 855
 856         ret = validate_clock_class(muxer_msg_iter, muxer_comp, clock_class);
 857
 858 end:
 859         return ret;
 860 }
 861
 862 /*
 863  * This function finds the youngest available message amongst the
 864  * non-ended upstream message iterators and returns the upstream
 865  * message iterator which has it, or
 866  * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
 867  * message.
 868  *
 869  * This function does NOT:
 870  *
 871  * * Update any upstream message iterator.
 872  * * Check the upstream message iterators to retry.
 873  *
 874  * On sucess, this function sets *muxer_upstream_msg_iter to the
 875  * upstream message iterator of which the current message is
 876  * the youngest, and sets *ts_ns to its time.
 877  */
 878 static
 879 bt_self_message_iterator_status
 880 muxer_msg_iter_youngest_upstream_msg_iter(
 881                 struct muxer_comp *muxer_comp,
 882                 struct muxer_msg_iter *muxer_msg_iter,
 883                 struct muxer_upstream_msg_iter **muxer_upstream_msg_iter,
 884                 int64_t *ts_ns)
 885 {
 886         size_t i;
 887         int ret;
 888         int64_t youngest_ts_ns = INT64_MAX;
 889         bt_self_message_iterator_status status =
 890                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 891
 892         BT_ASSERT(muxer_comp);
 893         BT_ASSERT(muxer_msg_iter);
 894         BT_ASSERT(muxer_upstream_msg_iter);
 895         *muxer_upstream_msg_iter = NULL;
 896
 897         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
 898                         i++) {
 899                 const bt_message *msg;
 900                 struct muxer_upstream_msg_iter *cur_muxer_upstream_msg_iter =
 901                         g_ptr_array_index(
 902                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
 903                                 i);
 904                 int64_t msg_ts_ns;
 905
 906                 if (!cur_muxer_upstream_msg_iter->msg_iter) {
 907                         /* This upstream message iterator is ended */
 908                         BT_LOGV("Skipping ended upstream message iterator: "
 909                                 "muxer-upstream-msg-iter-wrap-addr=%p",
 910                                 cur_muxer_upstream_msg_iter);
 911                         continue;
 912                 }
 913
 914                 BT_ASSERT(cur_muxer_upstream_msg_iter->msgs->length > 0);
 915                 msg = g_queue_peek_head(cur_muxer_upstream_msg_iter->msgs);
 916                 BT_ASSERT(msg);
 917
 918                 if (unlikely(bt_message_get_type(msg) ==
 919                                 BT_MESSAGE_TYPE_STREAM_BEGINNING)) {
 920                         ret = validate_new_stream_clock_class(
 921                                 muxer_msg_iter, muxer_comp,
 922                                 bt_message_stream_beginning_borrow_stream_const(
 923                                         msg));
 924                         if (ret) {
 925                                 /*
 926                                  * validate_new_stream_clock_class() logs
 927                                  * errors.
 928                                  */
 929                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 930                                 goto end;
 931                         }
 932                 } else if (unlikely(bt_message_get_type(msg) ==
 933                                 BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY)) {
 934                         const bt_clock_snapshot *cs;
 935
 936                         cs = bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
 937                                 msg);
 938                         ret = validate_clock_class(muxer_msg_iter, muxer_comp,
 939                                 bt_clock_snapshot_borrow_clock_class_const(cs));
 940                         if (ret) {
 941                                 /* validate_clock_class() logs errors */
 942                                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 943                                 goto end;
 944                         }
 945                 }
 946
 947                 ret = get_msg_ts_ns(muxer_comp, muxer_msg_iter, msg,
 948                         muxer_msg_iter->last_returned_ts_ns, &msg_ts_ns);
 949                 if (ret) {
 950                         /* get_msg_ts_ns() logs errors */
 951                         *muxer_upstream_msg_iter = NULL;
 952                         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
 953                         goto end;
 954                 }
 955
 956                 if (msg_ts_ns <= youngest_ts_ns) {
 957                         *muxer_upstream_msg_iter =
 958                                 cur_muxer_upstream_msg_iter;
 959                         youngest_ts_ns = msg_ts_ns;
 960                         *ts_ns = youngest_ts_ns;
 961                 }
 962         }
 963
 964         if (!*muxer_upstream_msg_iter) {
 965                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
 966                 *ts_ns = INT64_MIN;
 967         }
 968
 969 end:
 970         return status;
 971 }
 972
 973 static
 974 bt_self_message_iterator_status validate_muxer_upstream_msg_iter(
 975         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
 976         bool *is_ended)
 977 {
 978         bt_self_message_iterator_status status =
 979                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
 980
 981         BT_LOGV("Validating muxer's upstream message iterator wrapper: "
 982                 "muxer-upstream-msg-iter-wrap-addr=%p",
 983                 muxer_upstream_msg_iter);
 984
 985         if (muxer_upstream_msg_iter->msgs->length > 0 ||
 986                         !muxer_upstream_msg_iter->msg_iter) {
 987                 BT_LOGV("Already valid or not considered: "
 988                         "queue-len=%u, upstream-msg-iter-addr=%p",
 989                         muxer_upstream_msg_iter->msgs->length,
 990                         muxer_upstream_msg_iter->msg_iter);
 991                 goto end;
 992         }
 993
 994         /* muxer_upstream_msg_iter_next() logs details/errors */
 995         status = muxer_upstream_msg_iter_next(muxer_upstream_msg_iter,
 996                 is_ended);
 997
 998 end:
 999         return status;
1000 }
1001
1002 static
1003 bt_self_message_iterator_status validate_muxer_upstream_msg_iters(
1004                 struct muxer_msg_iter *muxer_msg_iter)
1005 {
1006         bt_self_message_iterator_status status =
1007                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1008         size_t i;
1009
1010         BT_LOGV("Validating muxer's upstream message iterator wrappers: "
1011                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
1012
1013         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
1014                         i++) {
1015                 bool is_ended = false;
1016                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
1017                         g_ptr_array_index(
1018                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
1019                                 i);
1020
1021                 status = validate_muxer_upstream_msg_iter(
1022                         muxer_upstream_msg_iter, &is_ended);
1023                 if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1024                         if (status < 0) {
1025                                 BT_LOGE("Cannot validate muxer's upstream message iterator wrapper: "
1026                                         "muxer-msg-iter-addr=%p, "
1027                                         "muxer-upstream-msg-iter-wrap-addr=%p",
1028                                         muxer_msg_iter,
1029                                         muxer_upstream_msg_iter);
1030                         } else {
1031                                 BT_LOGV("Cannot validate muxer's upstream message iterator wrapper: "
1032                                         "muxer-msg-iter-addr=%p, "
1033                                         "muxer-upstream-msg-iter-wrap-addr=%p",
1034                                         muxer_msg_iter,
1035                                         muxer_upstream_msg_iter);
1036                         }
1037
1038                         goto end;
1039                 }
1040
1041                 /*
1042                  * Move this muxer upstream message iterator to the
1043                  * array of ended iterators if it's ended.
1044                  */
1045                 if (unlikely(is_ended)) {
1046                         BT_LOGV("Muxer's upstream message iterator wrapper: ended or canceled: "
1047                                 "muxer-msg-iter-addr=%p, "
1048                                 "muxer-upstream-msg-iter-wrap-addr=%p",
1049                                 muxer_msg_iter, muxer_upstream_msg_iter);
1050                         g_ptr_array_add(
1051                                 muxer_msg_iter->ended_muxer_upstream_msg_iters,
1052                                 muxer_upstream_msg_iter);
1053                         muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i] = NULL;
1054
1055                         /*
1056                          * Use g_ptr_array_remove_fast() because the
1057                          * order of those elements is not important.
1058                          */
1059                         g_ptr_array_remove_index_fast(
1060                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
1061                                 i);
1062                         i--;
1063                 }
1064         }
1065
1066 end:
1067         return status;
1068 }
1069
1070 static inline
1071 bt_self_message_iterator_status muxer_msg_iter_do_next_one(
1072                 struct muxer_comp *muxer_comp,
1073                 struct muxer_msg_iter *muxer_msg_iter,
1074                 const bt_message **msg)
1075 {
1076         bt_self_message_iterator_status status;
1077         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter = NULL;
1078         int64_t next_return_ts;
1079
1080         status = validate_muxer_upstream_msg_iters(muxer_msg_iter);
1081         if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1082                 /* validate_muxer_upstream_msg_iters() logs details */
1083                 goto end;
1084         }
1085
1086         /*
1087          * At this point we know that all the existing upstream
1088          * message iterators are valid. We can find the one,
1089          * amongst those, of which the current message is the
1090          * youngest.
1091          */
1092         status = muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp,
1093                         muxer_msg_iter, &muxer_upstream_msg_iter,
1094                         &next_return_ts);
1095         if (status < 0 || status == BT_SELF_MESSAGE_ITERATOR_STATUS_END) {
1096                 if (status < 0) {
1097                         BT_LOGE("Cannot find the youngest upstream message iterator wrapper: "
1098                                 "status=%s",
1099                                 bt_common_self_message_iterator_status_string(status));
1100                 } else {
1101                         BT_LOGV("Cannot find the youngest upstream message iterator wrapper: "
1102                                 "status=%s",
1103                                 bt_common_self_message_iterator_status_string(status));
1104                 }
1105
1106                 goto end;
1107         }
1108
1109         if (next_return_ts < muxer_msg_iter->last_returned_ts_ns) {
1110                 BT_LOGE("Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
1111                         "muxer-msg-iter-addr=%p, ts=%" PRId64 ", "
1112                         "last-returned-ts=%" PRId64,
1113                         muxer_msg_iter, next_return_ts,
1114                         muxer_msg_iter->last_returned_ts_ns);
1115                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1116                 goto end;
1117         }
1118
1119         BT_LOGV("Found youngest upstream message iterator wrapper: "
1120                 "muxer-msg-iter-addr=%p, "
1121                 "muxer-upstream-msg-iter-wrap-addr=%p, "
1122                 "ts=%" PRId64,
1123                 muxer_msg_iter, muxer_upstream_msg_iter, next_return_ts);
1124         BT_ASSERT(status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK);
1125         BT_ASSERT(muxer_upstream_msg_iter);
1126
1127         /*
1128          * Consume from the queue's head: other side
1129          * (muxer_upstream_msg_iter_next()) writes to the tail.
1130          */
1131         *msg = g_queue_pop_head(muxer_upstream_msg_iter->msgs);
1132         BT_ASSERT(*msg);
1133         muxer_msg_iter->last_returned_ts_ns = next_return_ts;
1134
1135 end:
1136         return status;
1137 }
1138
1139 static
1140 bt_self_message_iterator_status muxer_msg_iter_do_next(
1141                 struct muxer_comp *muxer_comp,
1142                 struct muxer_msg_iter *muxer_msg_iter,
1143                 bt_message_array_const msgs, uint64_t capacity,
1144                 uint64_t *count)
1145 {
1146         bt_self_message_iterator_status status =
1147                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1148         uint64_t i = 0;
1149
1150         while (i < capacity && status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1151                 status = muxer_msg_iter_do_next_one(muxer_comp,
1152                         muxer_msg_iter, &msgs[i]);
1153                 if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
1154                         i++;
1155                 }
1156         }
1157
1158         if (i > 0) {
1159                 /*
1160                  * Even if muxer_msg_iter_do_next_one() returned
1161                  * something else than
1162                  * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1163                  * message objects in the output message
1164                  * array, so we need to return
1165                  * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1166                  * transfered to downstream. This other status occurs
1167                  * again the next time muxer_msg_iter_do_next() is
1168                  * called, possibly without any accumulated
1169                  * message, in which case we'll return it.
1170                  */
1171                 *count = i;
1172                 status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1173         }
1174
1175         return status;
1176 }
1177
1178 static
1179 void destroy_muxer_msg_iter(struct muxer_msg_iter *muxer_msg_iter)
1180 {
1181         if (!muxer_msg_iter) {
1182                 return;
1183         }
1184
1185         BT_LOGD("Destroying muxer component's message iterator: "
1186                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
1187
1188         if (muxer_msg_iter->active_muxer_upstream_msg_iters) {
1189                 BT_LOGD_STR("Destroying muxer's active upstream message iterator wrappers.");
1190                 g_ptr_array_free(
1191                         muxer_msg_iter->active_muxer_upstream_msg_iters, TRUE);
1192         }
1193
1194         if (muxer_msg_iter->ended_muxer_upstream_msg_iters) {
1195                 BT_LOGD_STR("Destroying muxer's ended upstream message iterator wrappers.");
1196                 g_ptr_array_free(
1197                         muxer_msg_iter->ended_muxer_upstream_msg_iters, TRUE);
1198         }
1199
1200         g_free(muxer_msg_iter);
1201 }
1202
1203 static
1204 int muxer_msg_iter_init_upstream_iterators(struct muxer_comp *muxer_comp,
1205                 struct muxer_msg_iter *muxer_msg_iter)
1206 {
1207         int64_t count;
1208         int64_t i;
1209         int ret = 0;
1210
1211         count = bt_component_filter_get_input_port_count(
1212                 bt_self_component_filter_as_component_filter(
1213                         muxer_comp->self_comp));
1214         if (count < 0) {
1215                 BT_LOGD("No input port to initialize for muxer component's message iterator: "
1216                         "muxer-comp-addr=%p, muxer-msg-iter-addr=%p",
1217                         muxer_comp, muxer_msg_iter);
1218                 goto end;
1219         }
1220
1221         for (i = 0; i < count; i++) {
1222                 bt_self_component_port_input_message_iterator *upstream_msg_iter;
1223                 bt_self_component_port_input *self_port =
1224                         bt_self_component_filter_borrow_input_port_by_index(
1225                                 muxer_comp->self_comp, i);
1226                 const bt_port *port;
1227
1228                 BT_ASSERT(self_port);
1229                 port = bt_self_component_port_as_port(
1230                         bt_self_component_port_input_as_self_component_port(
1231                                 self_port));
1232                 BT_ASSERT(port);
1233
1234                 if (!bt_port_is_connected(port)) {
1235                         /* Skip non-connected port */
1236                         continue;
1237                 }
1238
1239                 upstream_msg_iter = create_msg_iter_on_input_port(self_port);
1240                 if (!upstream_msg_iter) {
1241                         /* create_msg_iter_on_input_port() logs errors */
1242                         BT_ASSERT(!upstream_msg_iter);
1243                         ret = -1;
1244                         goto end;
1245                 }
1246
1247                 ret = muxer_msg_iter_add_upstream_msg_iter(muxer_msg_iter,
1248                         upstream_msg_iter);
1249                 bt_self_component_port_input_message_iterator_put_ref(
1250                         upstream_msg_iter);
1251                 if (ret) {
1252                         /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
1253                         goto end;
1254                 }
1255         }
1256
1257 end:
1258         return ret;
1259 }
1260
1261 BT_HIDDEN
1262 bt_self_message_iterator_status muxer_msg_iter_init(
1263                 bt_self_message_iterator *self_msg_iter,
1264                 bt_self_component_filter *self_comp,
1265                 bt_self_component_port_output *port)
1266 {
1267         struct muxer_comp *muxer_comp = NULL;
1268         struct muxer_msg_iter *muxer_msg_iter = NULL;
1269         bt_self_message_iterator_status status =
1270                 BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
1271         int ret;
1272
1273         muxer_comp = bt_self_component_get_data(
1274                 bt_self_component_filter_as_self_component(self_comp));
1275         BT_ASSERT(muxer_comp);
1276         BT_LOGD("Initializing muxer component's message iterator: "
1277                 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1278                 self_comp, muxer_comp, self_msg_iter);
1279
1280         if (muxer_comp->initializing_muxer_msg_iter) {
1281                 /*
1282                  * Weird, unhandled situation detected: downstream
1283                  * creates a muxer message iterator while creating
1284                  * another muxer message iterator (same component).
1285                  */
1286                 BT_LOGE("Recursive initialization of muxer component's message iterator: "
1287                         "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1288                         self_comp, muxer_comp, self_msg_iter);
1289                 goto error;
1290         }
1291
1292         muxer_comp->initializing_muxer_msg_iter = true;
1293         muxer_msg_iter = g_new0(struct muxer_msg_iter, 1);
1294         if (!muxer_msg_iter) {
1295                 BT_LOGE_STR("Failed to allocate one muxer component's message iterator.");
1296                 goto error;
1297         }
1298
1299         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1300         muxer_msg_iter->active_muxer_upstream_msg_iters =
1301                 g_ptr_array_new_with_free_func(
1302                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1303         if (!muxer_msg_iter->active_muxer_upstream_msg_iters) {
1304                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
1305                 goto error;
1306         }
1307
1308         muxer_msg_iter->ended_muxer_upstream_msg_iters =
1309                 g_ptr_array_new_with_free_func(
1310                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1311         if (!muxer_msg_iter->ended_muxer_upstream_msg_iters) {
1312                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
1313                 goto error;
1314         }
1315
1316         ret = muxer_msg_iter_init_upstream_iterators(muxer_comp,
1317                 muxer_msg_iter);
1318         if (ret) {
1319                 BT_LOGE("Cannot initialize connected input ports for muxer component's message iterator: "
1320                         "comp-addr=%p, muxer-comp-addr=%p, "
1321                         "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1322                         self_comp, muxer_comp, muxer_msg_iter,
1323                         self_msg_iter, ret);
1324                 goto error;
1325         }
1326
1327         bt_self_message_iterator_set_data(self_msg_iter, muxer_msg_iter);
1328         BT_LOGD("Initialized muxer component's message iterator: "
1329                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1330                 "msg-iter-addr=%p",
1331                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1332         goto end;
1333
1334 error:
1335         destroy_muxer_msg_iter(muxer_msg_iter);
1336         bt_self_message_iterator_set_data(self_msg_iter, NULL);
1337         status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
1338
1339 end:
1340         muxer_comp->initializing_muxer_msg_iter = false;
1341         return status;
1342 }
1343
1344 BT_HIDDEN
1345 void muxer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
1346 {
1347         struct muxer_msg_iter *muxer_msg_iter =
1348                 bt_self_message_iterator_get_data(self_msg_iter);
1349         bt_self_component *self_comp = NULL;
1350         struct muxer_comp *muxer_comp = NULL;
1351
1352         self_comp = bt_self_message_iterator_borrow_component(
1353                 self_msg_iter);
1354         BT_ASSERT(self_comp);
1355         muxer_comp = bt_self_component_get_data(self_comp);
1356         BT_LOGD("Finalizing muxer component's message iterator: "
1357                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1358                 "msg-iter-addr=%p",
1359                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1360
1361         if (muxer_msg_iter) {
1362                 destroy_muxer_msg_iter(muxer_msg_iter);
1363         }
1364 }
1365
1366 BT_HIDDEN
1367 bt_self_message_iterator_status muxer_msg_iter_next(
1368                 bt_self_message_iterator *self_msg_iter,
1369                 bt_message_array_const msgs, uint64_t capacity,
1370                 uint64_t *count)
1371 {
1372         bt_self_message_iterator_status status;
1373         struct muxer_msg_iter *muxer_msg_iter =
1374                 bt_self_message_iterator_get_data(self_msg_iter);
1375         bt_self_component *self_comp = NULL;
1376         struct muxer_comp *muxer_comp = NULL;
1377
1378         BT_ASSERT(muxer_msg_iter);
1379         self_comp = bt_self_message_iterator_borrow_component(
1380                 self_msg_iter);
1381         BT_ASSERT(self_comp);
1382         muxer_comp = bt_self_component_get_data(self_comp);
1383         BT_ASSERT(muxer_comp);
1384         BT_LOGV("Muxer component's message iterator's \"next\" method called: "
1385                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1386                 "msg-iter-addr=%p",
1387                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1388
1389         status = muxer_msg_iter_do_next(muxer_comp, muxer_msg_iter,
1390                 msgs, capacity, count);
1391         if (status < 0) {
1392                 BT_LOGE("Cannot get next message: "
1393                         "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1394                         "msg-iter-addr=%p, status=%s",
1395                         self_comp, muxer_comp, muxer_msg_iter, self_msg_iter,
1396                         bt_common_self_message_iterator_status_string(status));
1397         } else {
1398                 BT_LOGV("Returning from muxer component's message iterator's \"next\" method: "
1399                         "status=%s",
1400                         bt_common_self_message_iterator_status_string(status));
1401         }
1402
1403         return status;
1404 }
1405
1406 BT_HIDDEN
1407 bt_self_component_status muxer_input_port_connected(
1408                 bt_self_component_filter *self_comp,
1409                 bt_self_component_port_input *self_port,
1410                 const bt_port_output *other_port)
1411 {
1412         bt_self_component_status status;
1413
1414         status = add_available_input_port(self_comp);
1415         if (status) {
1416                 /*
1417                  * Only way to report an error later since this
1418                  * method does not return anything.
1419                  */
1420                 BT_LOGE("Cannot add one muxer component's input port: "
1421                         "status=%s",
1422                         bt_self_component_status_string(status));
1423                 goto end;
1424         }
1425
1426 end:
1427         return status;
1428 }
1429
1430 static inline
1431 bt_bool muxer_upstream_msg_iters_can_all_seek_beginning(
1432                 GPtrArray *muxer_upstream_msg_iters)
1433 {
1434         uint64_t i;
1435         bt_bool ret = BT_TRUE;
1436
1437         for (i = 0; i < muxer_upstream_msg_iters->len; i++) {
1438                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1439                         muxer_upstream_msg_iters->pdata[i];
1440
1441                 if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
1442                                 upstream_msg_iter->msg_iter)) {
1443                         ret = BT_FALSE;
1444                         goto end;
1445                 }
1446         }
1447
1448 end:
1449         return ret;
1450 }
1451
1452 BT_HIDDEN
1453 bt_bool muxer_msg_iter_can_seek_beginning(
1454                 bt_self_message_iterator *self_msg_iter)
1455 {
1456         struct muxer_msg_iter *muxer_msg_iter =
1457                 bt_self_message_iterator_get_data(self_msg_iter);
1458         bt_bool ret = BT_TRUE;
1459
1460         if (!muxer_upstream_msg_iters_can_all_seek_beginning(
1461                         muxer_msg_iter->active_muxer_upstream_msg_iters)) {
1462                 ret = BT_FALSE;
1463                 goto end;
1464         }
1465
1466         if (!muxer_upstream_msg_iters_can_all_seek_beginning(
1467                         muxer_msg_iter->ended_muxer_upstream_msg_iters)) {
1468                 ret = BT_FALSE;
1469                 goto end;
1470         }
1471
1472 end:
1473         return ret;
1474 }
1475
1476 BT_HIDDEN
1477 bt_self_message_iterator_status muxer_msg_iter_seek_beginning(
1478                 bt_self_message_iterator *self_msg_iter)
1479 {
1480         struct muxer_msg_iter *muxer_msg_iter =
1481                 bt_self_message_iterator_get_data(self_msg_iter);
1482         bt_message_iterator_status status = BT_MESSAGE_ITERATOR_STATUS_OK;
1483         uint64_t i;
1484
1485         /* Seek all ended upstream iterators first */
1486         for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
1487                         i++) {
1488                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1489                         muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
1490
1491                 status = bt_self_component_port_input_message_iterator_seek_beginning(
1492                         upstream_msg_iter->msg_iter);
1493                 if (status != BT_MESSAGE_ITERATOR_STATUS_OK) {
1494                         goto end;
1495                 }
1496
1497                 empty_message_queue(upstream_msg_iter);
1498         }
1499
1500         /* Seek all previously active upstream iterators */
1501         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
1502                         i++) {
1503                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1504                         muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i];
1505
1506                 status = bt_self_component_port_input_message_iterator_seek_beginning(
1507                         upstream_msg_iter->msg_iter);
1508                 if (status != BT_MESSAGE_ITERATOR_STATUS_OK) {
1509                         goto end;
1510                 }
1511
1512                 empty_message_queue(upstream_msg_iter);
1513         }
1514
1515         /* Make them all active */
1516         for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
1517                         i++) {
1518                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1519                         muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
1520
1521                 g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
1522                         upstream_msg_iter);
1523                 muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i] = NULL;
1524         }
1525
1526         g_ptr_array_remove_range(muxer_msg_iter->ended_muxer_upstream_msg_iters,
1527                 0, muxer_msg_iter->ended_muxer_upstream_msg_iters->len);
1528         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1529         muxer_msg_iter->clock_class_expectation =
1530                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY;
1531
1532 end:
1533         return (bt_self_message_iterator_status) status;
1534 }