src/plugins/utils/muxer/muxer.c

   1 /*
   2  * SPDX-License-Identifier: MIT
   3  *
   4  * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
   5  */
   6
   7 #define BT_COMP_LOG_SELF_COMP (muxer_comp->self_comp)
   8 #define BT_LOG_OUTPUT_LEVEL (muxer_comp->log_level)
   9 #define BT_LOG_TAG "PLUGIN/FLT.UTILS.MUXER"
  10 #include "logging/comp-logging.h"
  11
  12 #include "common/macros.h"
  13 #include "common/uuid.h"
  14 #include <babeltrace2/babeltrace.h>
  15 #include <glib.h>
  16 #include <stdbool.h>
  17 #include <inttypes.h>
  18 #include "common/assert.h"
  19 #include "common/common.h"
  20 #include <stdlib.h>
  21 #include <string.h>
  22
  23 #include "plugins/common/muxing/muxing.h"
  24 #include "plugins/common/param-validation/param-validation.h"
  25
  26 #include "muxer.h"
  27
  28 struct muxer_comp {
  29         /* Weak refs */
  30         bt_self_component_filter *self_comp_flt;
  31         bt_self_component *self_comp;
  32
  33         unsigned int next_port_num;
  34         size_t available_input_ports;
  35         bool initializing_muxer_msg_iter;
  36         bt_logging_level log_level;
  37 };
  38
  39 struct muxer_upstream_msg_iter {
  40         struct muxer_comp *muxer_comp;
  41
  42         /* Owned by this, NULL if ended */
  43         bt_message_iterator *msg_iter;
  44
  45         /* Contains `const bt_message *`, owned by this */
  46         GQueue *msgs;
  47 };
  48
  49 enum muxer_msg_iter_clock_class_expectation {
  50         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY = 0,
  51         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE,
  52         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE,
  53         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID,
  54         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID,
  55 };
  56
  57 struct muxer_msg_iter {
  58         struct muxer_comp *muxer_comp;
  59
  60         /* Weak */
  61         bt_self_message_iterator *self_msg_iter;
  62
  63         /*
  64          * Array of struct muxer_upstream_msg_iter * (owned by this).
  65          *
  66          * NOTE: This array is searched in linearly to find the youngest
  67          * current message. Keep this until benchmarks confirm that
  68          * another data structure is faster than this for our typical
  69          * use cases.
  70          */
  71         GPtrArray *active_muxer_upstream_msg_iters;
  72
  73         /*
  74          * Array of struct muxer_upstream_msg_iter * (owned by this).
  75          *
  76          * We move ended message iterators from
  77          * `active_muxer_upstream_msg_iters` to this array so as to be
  78          * able to restore them when seeking.
  79          */
  80         GPtrArray *ended_muxer_upstream_msg_iters;
  81
  82         /* Last time returned in a message */
  83         int64_t last_returned_ts_ns;
  84
  85         /* Clock class expectation state */
  86         enum muxer_msg_iter_clock_class_expectation clock_class_expectation;
  87
  88         /*
  89          * Expected clock class UUID, only valid when
  90          * clock_class_expectation is
  91          * MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID.
  92          */
  93         bt_uuid_t expected_clock_class_uuid;
  94
  95         /*
  96          * Saved error.  If we hit an error in the _next method, but have some
  97          * messages ready to return, we save the error here and return it on
  98          * the next _next call.
  99          */
 100         bt_message_iterator_class_next_method_status next_saved_status;
 101         const struct bt_error *next_saved_error;
 102 };
 103
 104 static
 105 void empty_message_queue(struct muxer_upstream_msg_iter *upstream_msg_iter)
 106 {
 107         const bt_message *msg;
 108
 109         while ((msg = g_queue_pop_head(upstream_msg_iter->msgs))) {
 110                 bt_message_put_ref(msg);
 111         }
 112 }
 113
 114 static
 115 void destroy_muxer_upstream_msg_iter(
 116                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter)
 117 {
 118         struct muxer_comp *muxer_comp;
 119
 120         if (!muxer_upstream_msg_iter) {
 121                 return;
 122         }
 123
 124         muxer_comp = muxer_upstream_msg_iter->muxer_comp;
 125         BT_COMP_LOGD("Destroying muxer's upstream message iterator wrapper: "
 126                 "addr=%p, msg-iter-addr=%p, queue-len=%u",
 127                 muxer_upstream_msg_iter,
 128                 muxer_upstream_msg_iter->msg_iter,
 129                 muxer_upstream_msg_iter->msgs->length);
 130         bt_message_iterator_put_ref(
 131                 muxer_upstream_msg_iter->msg_iter);
 132
 133         if (muxer_upstream_msg_iter->msgs) {
 134                 empty_message_queue(muxer_upstream_msg_iter);
 135                 g_queue_free(muxer_upstream_msg_iter->msgs);
 136         }
 137
 138         g_free(muxer_upstream_msg_iter);
 139 }
 140
 141 static
 142 int muxer_msg_iter_add_upstream_msg_iter(struct muxer_msg_iter *muxer_msg_iter,
 143                 bt_message_iterator *self_msg_iter)
 144 {
 145         int ret = 0;
 146         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
 147                 g_new0(struct muxer_upstream_msg_iter, 1);
 148         struct muxer_comp *muxer_comp = muxer_msg_iter->muxer_comp;
 149
 150         if (!muxer_upstream_msg_iter) {
 151                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 152                         "Failed to allocate one muxer's upstream message iterator wrapper.");
 153                 goto error;
 154         }
 155
 156         muxer_upstream_msg_iter->muxer_comp = muxer_comp;
 157         muxer_upstream_msg_iter->msg_iter = self_msg_iter;
 158         bt_message_iterator_get_ref(muxer_upstream_msg_iter->msg_iter);
 159         muxer_upstream_msg_iter->msgs = g_queue_new();
 160         if (!muxer_upstream_msg_iter->msgs) {
 161                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 162                         "Failed to allocate a GQueue.");
 163                 goto error;
 164         }
 165
 166         g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
 167                 muxer_upstream_msg_iter);
 168         BT_COMP_LOGD("Added muxer's upstream message iterator wrapper: "
 169                 "addr=%p, muxer-msg-iter-addr=%p, msg-iter-addr=%p",
 170                 muxer_upstream_msg_iter, muxer_msg_iter,
 171                 self_msg_iter);
 172
 173         goto end;
 174
 175 error:
 176         destroy_muxer_upstream_msg_iter(muxer_upstream_msg_iter);
 177         ret = -1;
 178
 179 end:
 180         return ret;
 181 }
 182
 183 static
 184 bt_self_component_add_port_status add_available_input_port(
 185                 bt_self_component_filter *self_comp)
 186 {
 187         struct muxer_comp *muxer_comp = bt_self_component_get_data(
 188                 bt_self_component_filter_as_self_component(self_comp));
 189         bt_self_component_add_port_status status =
 190                 BT_SELF_COMPONENT_ADD_PORT_STATUS_OK;
 191         GString *port_name = NULL;
 192
 193         BT_ASSERT(muxer_comp);
 194         port_name = g_string_new("in");
 195         if (!port_name) {
 196                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp, "Failed to allocate a GString.");
 197                 status = BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR;
 198                 goto end;
 199         }
 200
 201         g_string_append_printf(port_name, "%u", muxer_comp->next_port_num);
 202         status = bt_self_component_filter_add_input_port(
 203                 self_comp, port_name->str, NULL, NULL);
 204         if (status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 205                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 206                         "Cannot add input port to muxer component: "
 207                         "port-name=\"%s\", comp-addr=%p, status=%s",
 208                         port_name->str, self_comp,
 209                         bt_common_func_status_string(status));
 210                 goto end;
 211         }
 212
 213         muxer_comp->available_input_ports++;
 214         muxer_comp->next_port_num++;
 215         BT_COMP_LOGI("Added one input port to muxer component: "
 216                 "port-name=\"%s\", comp-addr=%p",
 217                 port_name->str, self_comp);
 218
 219 end:
 220         if (port_name) {
 221                 g_string_free(port_name, TRUE);
 222         }
 223
 224         return status;
 225 }
 226
 227 static
 228 bt_self_component_add_port_status create_output_port(
 229                 bt_self_component_filter *self_comp)
 230 {
 231         return bt_self_component_filter_add_output_port(
 232                 self_comp, "out", NULL, NULL);
 233 }
 234
 235 static
 236 void destroy_muxer_comp(struct muxer_comp *muxer_comp)
 237 {
 238         if (!muxer_comp) {
 239                 return;
 240         }
 241
 242         g_free(muxer_comp);
 243 }
 244
 245 static
 246 struct bt_param_validation_map_value_entry_descr muxer_params[] = {
 247         BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
 248 };
 249
 250 BT_HIDDEN
 251 bt_component_class_initialize_method_status muxer_init(
 252                 bt_self_component_filter *self_comp_flt,
 253                 bt_self_component_filter_configuration *config,
 254                 const bt_value *params, void *init_data)
 255 {
 256         bt_component_class_initialize_method_status status;
 257         bt_self_component_add_port_status add_port_status;
 258         bt_self_component *self_comp =
 259                 bt_self_component_filter_as_self_component(self_comp_flt);
 260         struct muxer_comp *muxer_comp = g_new0(struct muxer_comp, 1);
 261         bt_logging_level log_level = bt_component_get_logging_level(
 262                 bt_self_component_as_component(self_comp));
 263         enum bt_param_validation_status validation_status;
 264         gchar *validate_error = NULL;
 265
 266         BT_COMP_LOG_CUR_LVL(BT_LOG_INFO, log_level, self_comp,
 267                 "Initializing muxer component: "
 268                 "comp-addr=%p, params-addr=%p", self_comp, params);
 269
 270         if (!muxer_comp) {
 271                 /*
 272                  * Don't use BT_COMP_LOGE_APPEND_CAUSE, as `muxer_comp` is not
 273                  * initialized.
 274                  */
 275                 BT_COMP_LOG_CUR_LVL(BT_LOG_ERROR, log_level, self_comp,
 276                         "Failed to allocate one muxer component.");
 277                 BT_CURRENT_THREAD_ERROR_APPEND_CAUSE_FROM_COMPONENT(self_comp,
 278                         "Failed to allocate one muxer component.");
 279                         status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
 280                 goto error;
 281         }
 282
 283         muxer_comp->log_level = log_level;
 284         muxer_comp->self_comp = self_comp;
 285         muxer_comp->self_comp_flt = self_comp_flt;
 286
 287         validation_status = bt_param_validation_validate(params,
 288                 muxer_params, &validate_error);
 289         if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
 290                 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
 291                 goto error;
 292         } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
 293                 status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
 294                 BT_COMP_LOGE_APPEND_CAUSE(self_comp, "%s", validate_error);
 295                 goto error;
 296         }
 297
 298         bt_self_component_set_data(self_comp, muxer_comp);
 299         add_port_status = add_available_input_port(self_comp_flt);
 300         if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 301                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 302                         "Cannot ensure that at least one muxer component's input port is available: "
 303                         "muxer-comp-addr=%p, status=%s",
 304                         muxer_comp, bt_common_func_status_string(add_port_status));
 305                 status = (int) add_port_status;
 306                 goto error;
 307         }
 308
 309         add_port_status = create_output_port(self_comp_flt);
 310         if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
 311                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
 312                         "Cannot create muxer component's output port: "
 313                         "muxer-comp-addr=%p, status=%s",
 314                         muxer_comp, bt_common_func_status_string(add_port_status));
 315                 status = (int) add_port_status;
 316                 goto error;
 317         }
 318
 319         BT_COMP_LOGI("Initialized muxer component: "
 320                 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
 321                 self_comp, params, muxer_comp);
 322
 323         status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
 324         goto end;
 325
 326 error:
 327         destroy_muxer_comp(muxer_comp);
 328         bt_self_component_set_data(self_comp, NULL);
 329
 330 end:
 331         g_free(validate_error);
 332         return status;
 333 }
 334
 335 BT_HIDDEN
 336 void muxer_finalize(bt_self_component_filter *self_comp)
 337 {
 338         struct muxer_comp *muxer_comp = bt_self_component_get_data(
 339                 bt_self_component_filter_as_self_component(self_comp));
 340
 341         BT_COMP_LOGI("Finalizing muxer component: comp-addr=%p",
 342                 self_comp);
 343         destroy_muxer_comp(muxer_comp);
 344 }
 345
 346 static
 347 bt_message_iterator_create_from_message_iterator_status
 348 create_msg_iter_on_input_port(struct muxer_comp *muxer_comp,
 349                 struct muxer_msg_iter *muxer_msg_iter,
 350                 bt_self_component_port_input *self_port,
 351                 bt_message_iterator **msg_iter)
 352 {
 353         const bt_port *port = bt_self_component_port_as_port(
 354                 bt_self_component_port_input_as_self_component_port(
 355                         self_port));
 356         bt_message_iterator_create_from_message_iterator_status
 357                 status;
 358
 359         BT_ASSERT(port);
 360         BT_ASSERT(bt_port_is_connected(port));
 361
 362         // TODO: Advance the iterator to >= the time of the latest
 363         //       returned message by the muxer message
 364         //       iterator which creates it.
 365         status = bt_message_iterator_create_from_message_iterator(
 366                 muxer_msg_iter->self_msg_iter, self_port, msg_iter);
 367         if (status != BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK) {
 368                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 369                         "Cannot create upstream message iterator on input port: "
 370                         "port-addr=%p, port-name=\"%s\"",
 371                         port, bt_port_get_name(port));
 372                 goto end;
 373         }
 374
 375         BT_COMP_LOGI("Created upstream message iterator on input port: "
 376                 "port-addr=%p, port-name=\"%s\", msg-iter-addr=%p",
 377                 port, bt_port_get_name(port), msg_iter);
 378
 379 end:
 380         return status;
 381 }
 382
 383 static
 384 bt_message_iterator_class_next_method_status muxer_upstream_msg_iter_next(
 385                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
 386                 bool *is_ended)
 387 {
 388         struct muxer_comp *muxer_comp = muxer_upstream_msg_iter->muxer_comp;
 389         bt_message_iterator_class_next_method_status status;
 390         bt_message_iterator_next_status input_port_iter_status;
 391         bt_message_array_const msgs;
 392         uint64_t i;
 393         uint64_t count;
 394
 395         BT_COMP_LOGD("Calling upstream message iterator's \"next\" method: "
 396                 "muxer-upstream-msg-iter-wrap-addr=%p, msg-iter-addr=%p",
 397                 muxer_upstream_msg_iter,
 398                 muxer_upstream_msg_iter->msg_iter);
 399         input_port_iter_status = bt_message_iterator_next(
 400                 muxer_upstream_msg_iter->msg_iter, &msgs, &count);
 401         BT_COMP_LOGD("Upstream message iterator's \"next\" method returned: "
 402                 "status=%s",
 403                 bt_common_func_status_string(input_port_iter_status));
 404
 405         switch (input_port_iter_status) {
 406         case BT_MESSAGE_ITERATOR_NEXT_STATUS_OK:
 407                 /*
 408                  * Message iterator's current message is
 409                  * valid: it must be considered for muxing operations.
 410                  */
 411                 BT_COMP_LOGD_STR("Validated upstream message iterator wrapper.");
 412                 BT_ASSERT_DBG(count > 0);
 413
 414                 /* Move messages to our queue */
 415                 for (i = 0; i < count; i++) {
 416                         /*
 417                          * Push to tail in order; other side
 418                          * (muxer_msg_iter_do_next_one()) consumes
 419                          * from the head first.
 420                          */
 421                         g_queue_push_tail(muxer_upstream_msg_iter->msgs,
 422                                 (void *) msgs[i]);
 423                 }
 424                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
 425                 break;
 426         case BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN:
 427                 /*
 428                  * Message iterator's current message is not
 429                  * valid anymore. Return
 430                  * BT_MESSAGE_ITERATOR_NEXT_STATUS_AGAIN immediately.
 431                  */
 432                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
 433                 break;
 434         case BT_MESSAGE_ITERATOR_NEXT_STATUS_END:       /* Fall-through. */
 435                 /*
 436                  * Message iterator reached the end: release it. It
 437                  * won't be considered again to find the youngest
 438                  * message.
 439                  */
 440                 *is_ended = true;
 441                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
 442                 break;
 443         case BT_MESSAGE_ITERATOR_NEXT_STATUS_ERROR:
 444         case BT_MESSAGE_ITERATOR_NEXT_STATUS_MEMORY_ERROR:
 445                 /* Error status code */
 446                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 447                         "Upstream iterator's next method returned an error: status=%s",
 448                         bt_common_func_status_string(input_port_iter_status));
 449                 status = (int) input_port_iter_status;
 450                 break;
 451         default:
 452                 /* Unsupported status code */
 453                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 454                         "Unsupported status code: status=%s",
 455                         bt_common_func_status_string(input_port_iter_status));
 456                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
 457                 break;
 458         }
 459
 460         return status;
 461 }
 462
 463 static
 464 int get_msg_ts_ns(struct muxer_comp *muxer_comp,
 465                 struct muxer_msg_iter *muxer_msg_iter,
 466                 const bt_message *msg, int64_t last_returned_ts_ns,
 467                 int64_t *ts_ns)
 468 {
 469         const bt_clock_snapshot *clock_snapshot = NULL;
 470         int ret = 0;
 471         const bt_stream_class *stream_class = NULL;
 472         bt_message_type msg_type;
 473
 474         BT_ASSERT_DBG(msg);
 475         BT_ASSERT_DBG(ts_ns);
 476         BT_COMP_LOGD("Getting message's timestamp: "
 477                 "muxer-msg-iter-addr=%p, msg-addr=%p, "
 478                 "last-returned-ts=%" PRId64,
 479                 muxer_msg_iter, msg, last_returned_ts_ns);
 480
 481         if (G_UNLIKELY(muxer_msg_iter->clock_class_expectation ==
 482                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE)) {
 483                 *ts_ns = last_returned_ts_ns;
 484                 goto end;
 485         }
 486
 487         msg_type = bt_message_get_type(msg);
 488
 489         if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_PACKET_BEGINNING)) {
 490                 stream_class = bt_stream_borrow_class_const(
 491                         bt_packet_borrow_stream_const(
 492                                 bt_message_packet_beginning_borrow_packet_const(
 493                                         msg)));
 494         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_PACKET_END)) {
 495                 stream_class = bt_stream_borrow_class_const(
 496                         bt_packet_borrow_stream_const(
 497                                 bt_message_packet_end_borrow_packet_const(
 498                                         msg)));
 499         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS)) {
 500                 stream_class = bt_stream_borrow_class_const(
 501                         bt_message_discarded_events_borrow_stream_const(msg));
 502         } else if (G_UNLIKELY(msg_type == BT_MESSAGE_TYPE_DISCARDED_PACKETS)) {
 503                 stream_class = bt_stream_borrow_class_const(
 504                         bt_message_discarded_packets_borrow_stream_const(msg));
 505         }
 506
 507         switch (msg_type) {
 508         case BT_MESSAGE_TYPE_EVENT:
 509                 BT_ASSERT_DBG(bt_message_event_borrow_stream_class_default_clock_class_const(
 510                                 msg));
 511                 clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
 512                         msg);
 513                 break;
 514         case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 515                 if (bt_stream_class_packets_have_beginning_default_clock_snapshot(
 516                                 stream_class)) {
 517                         clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
 518                                 msg);
 519                 } else {
 520                         goto no_clock_snapshot;
 521                 }
 522
 523                 break;
 524         case BT_MESSAGE_TYPE_PACKET_END:
 525                 if (bt_stream_class_packets_have_end_default_clock_snapshot(
 526                                 stream_class)) {
 527                         clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
 528                                 msg);
 529                 } else {
 530                         goto no_clock_snapshot;
 531                 }
 532
 533                 break;
 534         case BT_MESSAGE_TYPE_STREAM_BEGINNING:
 535         {
 536                 enum bt_message_stream_clock_snapshot_state snapshot_state =
 537                         bt_message_stream_beginning_borrow_default_clock_snapshot_const(
 538                                 msg, &clock_snapshot);
 539                 if (snapshot_state == BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN) {
 540                         goto no_clock_snapshot;
 541                 }
 542
 543                 break;
 544         }
 545         case BT_MESSAGE_TYPE_STREAM_END:
 546         {
 547                 enum bt_message_stream_clock_snapshot_state snapshot_state =
 548                         bt_message_stream_end_borrow_default_clock_snapshot_const(
 549                                 msg, &clock_snapshot);
 550                 if (snapshot_state == BT_MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN) {
 551                         goto no_clock_snapshot;
 552                 }
 553
 554                 break;
 555         }
 556         case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 557                 if (bt_stream_class_discarded_events_have_default_clock_snapshots(
 558                                 stream_class)) {
 559                         clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
 560                                 msg);
 561                 } else {
 562                         goto no_clock_snapshot;
 563                 }
 564
 565                 break;
 566         case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 567                 if (bt_stream_class_discarded_packets_have_default_clock_snapshots(
 568                                 stream_class)) {
 569                         clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
 570                                 msg);
 571                 } else {
 572                         goto no_clock_snapshot;
 573                 }
 574
 575                 break;
 576         case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 577                 clock_snapshot = bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
 578                         msg);
 579                 break;
 580         default:
 581                 /* All the other messages have a higher priority */
 582                 BT_COMP_LOGD_STR("Message has no timestamp: using the last returned timestamp.");
 583                 *ts_ns = last_returned_ts_ns;
 584                 goto end;
 585         }
 586
 587         ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
 588         if (ret) {
 589                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 590                         "Cannot get nanoseconds from Epoch of clock snapshot: "
 591                         "clock-snapshot-addr=%p", clock_snapshot);
 592                 goto error;
 593         }
 594
 595         goto end;
 596
 597 no_clock_snapshot:
 598         BT_COMP_LOGD_STR("Message's default clock snapshot is missing: "
 599                 "using the last returned timestamp.");
 600         *ts_ns = last_returned_ts_ns;
 601         goto end;
 602
 603 error:
 604         ret = -1;
 605
 606 end:
 607         if (ret == 0) {
 608                 BT_COMP_LOGD("Found message's timestamp: "
 609                         "muxer-msg-iter-addr=%p, msg-addr=%p, "
 610                         "last-returned-ts=%" PRId64 ", ts=%" PRId64,
 611                         muxer_msg_iter, msg, last_returned_ts_ns,
 612                         *ts_ns);
 613         }
 614
 615         return ret;
 616 }
 617
 618 static inline
 619 int validate_clock_class(struct muxer_msg_iter *muxer_msg_iter,
 620                 struct muxer_comp *muxer_comp,
 621                 const bt_clock_class *clock_class)
 622 {
 623         int ret = 0;
 624         const uint8_t *cc_uuid;
 625         const char *cc_name;
 626
 627         BT_ASSERT_DBG(clock_class);
 628         cc_uuid = bt_clock_class_get_uuid(clock_class);
 629         cc_name = bt_clock_class_get_name(clock_class);
 630
 631         if (muxer_msg_iter->clock_class_expectation ==
 632                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 633                 /*
 634                  * This is the first clock class that this muxer message
 635                  * iterator encounters. Its properties determine what to expect
 636                  * for the whole lifetime of the iterator.
 637                  */
 638                 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 639                         /* Expect absolute clock classes */
 640                         muxer_msg_iter->clock_class_expectation =
 641                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE;
 642                 } else {
 643                         if (cc_uuid) {
 644                                 /*
 645                                  * Expect non-absolute clock classes
 646                                  * with a specific UUID.
 647                                  */
 648                                 muxer_msg_iter->clock_class_expectation =
 649                                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID;
 650                                 bt_uuid_copy(muxer_msg_iter->expected_clock_class_uuid, cc_uuid);
 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         switch (muxer_msg_iter->clock_class_expectation) {
 663         case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE:
 664                 if (!bt_clock_class_origin_is_unix_epoch(clock_class)) {
 665                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 666                                 "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_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 676                                 "Expecting a non-absolute clock class with no UUID, "
 677                                 "but got an absolute one: "
 678                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 679                                 clock_class, cc_name);
 680                         goto error;
 681                 }
 682
 683                 if (cc_uuid) {
 684                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 685                                 "Expecting a non-absolute clock class with no UUID, "
 686                                 "but got one with a UUID: "
 687                                 "clock-class-addr=%p, clock-class-name=\"%s\", "
 688                                 "uuid=\"" BT_UUID_FMT "\"",
 689                                 clock_class, cc_name, BT_UUID_FMT_VALUES(cc_uuid));
 690                         goto error;
 691                 }
 692                 break;
 693         case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID:
 694                 if (bt_clock_class_origin_is_unix_epoch(clock_class)) {
 695                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 696                                 "Expecting a non-absolute clock class with a specific UUID, "
 697                                 "but got an absolute one: "
 698                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 699                                 clock_class, cc_name);
 700                         goto error;
 701                 }
 702
 703                 if (!cc_uuid) {
 704                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 705                                 "Expecting a non-absolute clock class with a specific UUID, "
 706                                 "but got one with no UUID: "
 707                                 "clock-class-addr=%p, clock-class-name=\"%s\"",
 708                                 clock_class, cc_name);
 709                         goto error;
 710                 }
 711
 712                 if (bt_uuid_compare(muxer_msg_iter->expected_clock_class_uuid, cc_uuid) != 0) {
 713                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 714                                 "Expecting a non-absolute clock class with a specific UUID, "
 715                                 "but got one with different UUID: "
 716                                 "clock-class-addr=%p, clock-class-name=\"%s\", "
 717                                 "expected-uuid=\"" BT_UUID_FMT "\", "
 718                                 "uuid=\"" BT_UUID_FMT "\"",
 719                                 clock_class, cc_name,
 720                                 BT_UUID_FMT_VALUES(muxer_msg_iter->expected_clock_class_uuid),
 721                                 BT_UUID_FMT_VALUES(cc_uuid));
 722                         goto error;
 723                 }
 724                 break;
 725         case MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE:
 726                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 727                         "Expecting no clock class, but got one: "
 728                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 729                         clock_class, cc_name);
 730                 goto error;
 731         default:
 732                 /* Unexpected */
 733                 BT_COMP_LOGF("Unexpected clock class expectation: "
 734                         "expectation-code=%d",
 735                         muxer_msg_iter->clock_class_expectation);
 736                 bt_common_abort();
 737         }
 738
 739         goto end;
 740
 741 error:
 742         ret = -1;
 743
 744 end:
 745         return ret;
 746 }
 747
 748 static inline
 749 int validate_new_stream_clock_class(struct muxer_msg_iter *muxer_msg_iter,
 750                 struct muxer_comp *muxer_comp, const bt_stream *stream)
 751 {
 752         int ret = 0;
 753         const bt_stream_class *stream_class =
 754                 bt_stream_borrow_class_const(stream);
 755         const bt_clock_class *clock_class =
 756                 bt_stream_class_borrow_default_clock_class_const(stream_class);
 757
 758         if (!clock_class) {
 759                 if (muxer_msg_iter->clock_class_expectation ==
 760                         MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 761                         /* Expect no clock class */
 762                         muxer_msg_iter->clock_class_expectation =
 763                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE;
 764                 } else if (muxer_msg_iter->clock_class_expectation !=
 765                                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_NONE) {
 766                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 767                                 "Expecting stream class without a default clock class: "
 768                                 "stream-class-addr=%p, stream-class-name=\"%s\", "
 769                                 "stream-class-id=%" PRIu64,
 770                                 stream_class, bt_stream_class_get_name(stream_class),
 771                                 bt_stream_class_get_id(stream_class));
 772                         ret = -1;
 773                 }
 774
 775                 goto end;
 776         }
 777
 778         ret = validate_clock_class(muxer_msg_iter, muxer_comp, clock_class);
 779
 780 end:
 781         return ret;
 782 }
 783
 784 /*
 785  * This function finds the youngest available message amongst the
 786  * non-ended upstream message iterators and returns the upstream
 787  * message iterator which has it, or
 788  * BT_MESSAGE_ITERATOR_STATUS_END if there's no available
 789  * message.
 790  *
 791  * This function does NOT:
 792  *
 793  * * Update any upstream message iterator.
 794  * * Check the upstream message iterators to retry.
 795  *
 796  * On sucess, this function sets *muxer_upstream_msg_iter to the
 797  * upstream message iterator of which the current message is
 798  * the youngest, and sets *ts_ns to its time.
 799  */
 800 static
 801 bt_message_iterator_class_next_method_status
 802 muxer_msg_iter_youngest_upstream_msg_iter(
 803                 struct muxer_comp *muxer_comp,
 804                 struct muxer_msg_iter *muxer_msg_iter,
 805                 struct muxer_upstream_msg_iter **muxer_upstream_msg_iter,
 806                 int64_t *ts_ns)
 807 {
 808         size_t i;
 809         int ret;
 810         int64_t youngest_ts_ns = INT64_MAX;
 811         bt_message_iterator_class_next_method_status status =
 812                 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
 813
 814         BT_ASSERT_DBG(muxer_comp);
 815         BT_ASSERT_DBG(muxer_msg_iter);
 816         BT_ASSERT_DBG(muxer_upstream_msg_iter);
 817         *muxer_upstream_msg_iter = NULL;
 818
 819         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
 820                         i++) {
 821                 const bt_message *msg;
 822                 struct muxer_upstream_msg_iter *cur_muxer_upstream_msg_iter =
 823                         g_ptr_array_index(
 824                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
 825                                 i);
 826                 int64_t msg_ts_ns;
 827
 828                 if (!cur_muxer_upstream_msg_iter->msg_iter) {
 829                         /* This upstream message iterator is ended */
 830                         BT_COMP_LOGT("Skipping ended upstream message iterator: "
 831                                 "muxer-upstream-msg-iter-wrap-addr=%p",
 832                                 cur_muxer_upstream_msg_iter);
 833                         continue;
 834                 }
 835
 836                 BT_ASSERT_DBG(cur_muxer_upstream_msg_iter->msgs->length > 0);
 837                 msg = g_queue_peek_head(cur_muxer_upstream_msg_iter->msgs);
 838                 BT_ASSERT_DBG(msg);
 839
 840                 if (G_UNLIKELY(bt_message_get_type(msg) ==
 841                                 BT_MESSAGE_TYPE_STREAM_BEGINNING)) {
 842                         ret = validate_new_stream_clock_class(
 843                                 muxer_msg_iter, muxer_comp,
 844                                 bt_message_stream_beginning_borrow_stream_const(
 845                                         msg));
 846                         if (ret) {
 847                                 /*
 848                                  * validate_new_stream_clock_class() logs
 849                                  * errors.
 850                                  */
 851                                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
 852                                 goto end;
 853                         }
 854                 } else if (G_UNLIKELY(bt_message_get_type(msg) ==
 855                                 BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY)) {
 856                         const bt_clock_snapshot *cs;
 857
 858                         cs = bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(
 859                                 msg);
 860                         ret = validate_clock_class(muxer_msg_iter, muxer_comp,
 861                                 bt_clock_snapshot_borrow_clock_class_const(cs));
 862                         if (ret) {
 863                                 /* validate_clock_class() logs errors */
 864                                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
 865                                 goto end;
 866                         }
 867                 }
 868
 869                 ret = get_msg_ts_ns(muxer_comp, muxer_msg_iter, msg,
 870                         muxer_msg_iter->last_returned_ts_ns, &msg_ts_ns);
 871                 if (ret) {
 872                         /* get_msg_ts_ns() logs errors */
 873                         *muxer_upstream_msg_iter = NULL;
 874                         status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
 875                         goto end;
 876                 }
 877
 878                 /*
 879                  * Update the current message iterator if it has not been set
 880                  * yet, or if its current message has a timestamp smaller than
 881                  * the previously selected youngest message.
 882                  */
 883                 if (G_UNLIKELY(*muxer_upstream_msg_iter == NULL) ||
 884                                 msg_ts_ns < youngest_ts_ns) {
 885                         *muxer_upstream_msg_iter =
 886                                 cur_muxer_upstream_msg_iter;
 887                         youngest_ts_ns = msg_ts_ns;
 888                         *ts_ns = youngest_ts_ns;
 889                 } else if (msg_ts_ns == youngest_ts_ns) {
 890                         /*
 891                          * The currently selected message to be sent downstream
 892                          * next has the exact same timestamp that of the
 893                          * current candidate message. We must break the tie
 894                          * in a predictable manner.
 895                          */
 896                         const bt_message *selected_msg = g_queue_peek_head(
 897                                 (*muxer_upstream_msg_iter)->msgs);
 898                         BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
 899
 900                         /*
 901                          * Order the messages in an arbitrary but determinitic
 902                          * way.
 903                          */
 904                         ret = common_muxing_compare_messages(msg, selected_msg);
 905                         if (ret < 0) {
 906                                 /*
 907                                  * The `msg` should go first. Update the next
 908                                  * iterator and the current timestamp.
 909                                  */
 910                                 *muxer_upstream_msg_iter =
 911                                         cur_muxer_upstream_msg_iter;
 912                                 youngest_ts_ns = msg_ts_ns;
 913                                 *ts_ns = youngest_ts_ns;
 914                         } else if (ret == 0) {
 915                                 /* Unable to pick which one should go first. */
 916                                 BT_COMP_LOGW("Cannot deterministically pick next upstream message iterator because they have identical next messages: "
 917                                         "muxer-upstream-msg-iter-wrap-addr=%p"
 918                                         "cur-muxer-upstream-msg-iter-wrap-addr=%p",
 919                                         *muxer_upstream_msg_iter,
 920                                         cur_muxer_upstream_msg_iter);
 921                         }
 922                 }
 923         }
 924
 925         if (!*muxer_upstream_msg_iter) {
 926                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
 927                 *ts_ns = INT64_MIN;
 928         }
 929
 930 end:
 931         return status;
 932 }
 933
 934 static
 935 bt_message_iterator_class_next_method_status
 936 validate_muxer_upstream_msg_iter(
 937         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter,
 938         bool *is_ended)
 939 {
 940         struct muxer_comp *muxer_comp = muxer_upstream_msg_iter->muxer_comp;
 941         bt_message_iterator_class_next_method_status status;
 942
 943         BT_COMP_LOGD("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_COMP_LOGD("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                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
 954                 goto end;
 955         }
 956
 957         /* muxer_upstream_msg_iter_next() logs details/errors */
 958         status = muxer_upstream_msg_iter_next(muxer_upstream_msg_iter,
 959                 is_ended);
 960
 961 end:
 962         return status;
 963 }
 964
 965 static
 966 bt_message_iterator_class_next_method_status
 967 validate_muxer_upstream_msg_iters(
 968                 struct muxer_msg_iter *muxer_msg_iter)
 969 {
 970         struct muxer_comp *muxer_comp = muxer_msg_iter->muxer_comp;
 971         bt_message_iterator_class_next_method_status status;
 972         size_t i;
 973
 974         BT_COMP_LOGD("Validating muxer's upstream message iterator wrappers: "
 975                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
 976
 977         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
 978                         i++) {
 979                 bool is_ended = false;
 980                 struct muxer_upstream_msg_iter *muxer_upstream_msg_iter =
 981                         g_ptr_array_index(
 982                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
 983                                 i);
 984
 985                 status = validate_muxer_upstream_msg_iter(
 986                         muxer_upstream_msg_iter, &is_ended);
 987                 if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
 988                         if (status < 0) {
 989                                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
 990                                         "Cannot validate muxer's upstream message iterator wrapper: "
 991                                         "muxer-msg-iter-addr=%p, "
 992                                         "muxer-upstream-msg-iter-wrap-addr=%p",
 993                                         muxer_msg_iter,
 994                                         muxer_upstream_msg_iter);
 995                         } else {
 996                                 BT_COMP_LOGD("Cannot validate muxer's upstream message iterator wrapper: "
 997                                         "muxer-msg-iter-addr=%p, "
 998                                         "muxer-upstream-msg-iter-wrap-addr=%p",
 999                                         muxer_msg_iter,
1000                                         muxer_upstream_msg_iter);
1001                         }
1002
1003                         goto end;
1004                 }
1005
1006                 /*
1007                  * Move this muxer upstream message iterator to the
1008                  * array of ended iterators if it's ended.
1009                  */
1010                 if (G_UNLIKELY(is_ended)) {
1011                         BT_COMP_LOGD("Muxer's upstream message iterator wrapper: ended or canceled: "
1012                                 "muxer-msg-iter-addr=%p, "
1013                                 "muxer-upstream-msg-iter-wrap-addr=%p",
1014                                 muxer_msg_iter, muxer_upstream_msg_iter);
1015                         g_ptr_array_add(
1016                                 muxer_msg_iter->ended_muxer_upstream_msg_iters,
1017                                 muxer_upstream_msg_iter);
1018                         muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i] = NULL;
1019
1020                         /*
1021                          * Use g_ptr_array_remove_fast() because the
1022                          * order of those elements is not important.
1023                          */
1024                         g_ptr_array_remove_index_fast(
1025                                 muxer_msg_iter->active_muxer_upstream_msg_iters,
1026                                 i);
1027                         i--;
1028                 }
1029         }
1030
1031         status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1032
1033 end:
1034         return status;
1035 }
1036
1037 static inline
1038 bt_message_iterator_class_next_method_status muxer_msg_iter_do_next_one(
1039                 struct muxer_comp *muxer_comp,
1040                 struct muxer_msg_iter *muxer_msg_iter,
1041                 const bt_message **msg)
1042 {
1043         bt_message_iterator_class_next_method_status status;
1044         struct muxer_upstream_msg_iter *muxer_upstream_msg_iter = NULL;
1045         /* Initialize to avoid -Wmaybe-uninitialized warning with gcc 4.8. */
1046         int64_t next_return_ts = 0;
1047
1048         status = validate_muxer_upstream_msg_iters(muxer_msg_iter);
1049         if (status != BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1050                 /* validate_muxer_upstream_msg_iters() logs details */
1051                 goto end;
1052         }
1053
1054         /*
1055          * At this point we know that all the existing upstream
1056          * message iterators are valid. We can find the one,
1057          * amongst those, of which the current message is the
1058          * youngest.
1059          */
1060         status = muxer_msg_iter_youngest_upstream_msg_iter(muxer_comp,
1061                         muxer_msg_iter, &muxer_upstream_msg_iter,
1062                         &next_return_ts);
1063         if (status < 0 || status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END) {
1064                 if (status < 0) {
1065                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1066                                 "Cannot find the youngest upstream message iterator wrapper: "
1067                                 "status=%s",
1068                                 bt_common_func_status_string(status));
1069                 } else {
1070                         BT_COMP_LOGD("Cannot find the youngest upstream message iterator wrapper: "
1071                                 "status=%s",
1072                                 bt_common_func_status_string(status));
1073                 }
1074
1075                 goto end;
1076         }
1077
1078         if (next_return_ts < muxer_msg_iter->last_returned_ts_ns) {
1079                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1080                         "Youngest upstream message iterator wrapper's timestamp is less than muxer's message iterator's last returned timestamp: "
1081                         "muxer-msg-iter-addr=%p, ts=%" PRId64 ", "
1082                         "last-returned-ts=%" PRId64,
1083                         muxer_msg_iter, next_return_ts,
1084                         muxer_msg_iter->last_returned_ts_ns);
1085                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1086                 goto end;
1087         }
1088
1089         BT_COMP_LOGD("Found youngest upstream message iterator wrapper: "
1090                 "muxer-msg-iter-addr=%p, "
1091                 "muxer-upstream-msg-iter-wrap-addr=%p, "
1092                 "ts=%" PRId64,
1093                 muxer_msg_iter, muxer_upstream_msg_iter, next_return_ts);
1094         BT_ASSERT_DBG(status ==
1095                 BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK);
1096         BT_ASSERT_DBG(muxer_upstream_msg_iter);
1097
1098         /*
1099          * Consume from the queue's head: other side
1100          * (muxer_upstream_msg_iter_next()) writes to the tail.
1101          */
1102         *msg = g_queue_pop_head(muxer_upstream_msg_iter->msgs);
1103         BT_ASSERT_DBG(*msg);
1104         muxer_msg_iter->last_returned_ts_ns = next_return_ts;
1105
1106 end:
1107         return status;
1108 }
1109
1110 static
1111 bt_message_iterator_class_next_method_status muxer_msg_iter_do_next(
1112                 struct muxer_comp *muxer_comp,
1113                 struct muxer_msg_iter *muxer_msg_iter,
1114                 bt_message_array_const msgs, uint64_t capacity,
1115                 uint64_t *count)
1116 {
1117         bt_message_iterator_class_next_method_status status;
1118         uint64_t i = 0;
1119
1120         if (G_UNLIKELY(muxer_msg_iter->next_saved_error)) {
1121                 /*
1122                  * Last time we were called, we hit an error but had some
1123                  * messages to deliver, so we stashed the error here.  Return
1124                  * it now.
1125                  */
1126                 BT_CURRENT_THREAD_MOVE_ERROR_AND_RESET(muxer_msg_iter->next_saved_error);
1127                 status = muxer_msg_iter->next_saved_status;
1128                 goto end;
1129         }
1130
1131         do {
1132                 status = muxer_msg_iter_do_next_one(muxer_comp,
1133                         muxer_msg_iter, &msgs[i]);
1134                 if (status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK) {
1135                         i++;
1136                 }
1137         } while (i < capacity && status == BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK);
1138
1139         if (i > 0) {
1140                 /*
1141                  * Even if muxer_msg_iter_do_next_one() returned
1142                  * something else than
1143                  * BT_MESSAGE_ITERATOR_STATUS_OK, we accumulated
1144                  * message objects in the output message
1145                  * array, so we need to return
1146                  * BT_MESSAGE_ITERATOR_STATUS_OK so that they are
1147                  * transfered to downstream. This other status occurs
1148                  * again the next time muxer_msg_iter_do_next() is
1149                  * called, possibly without any accumulated
1150                  * message, in which case we'll return it.
1151                  */
1152                 if (status < 0) {
1153                         /*
1154                          * Save this error for the next _next call.  Assume that
1155                          * this component always appends error causes when
1156                          * returning an error status code, which will cause the
1157                          * current thread error to be non-NULL.
1158                          */
1159                         muxer_msg_iter->next_saved_error = bt_current_thread_take_error();
1160                         BT_ASSERT(muxer_msg_iter->next_saved_error);
1161                         muxer_msg_iter->next_saved_status = status;
1162                 }
1163
1164                 *count = i;
1165                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1166         }
1167
1168 end:
1169         return status;
1170 }
1171
1172 static
1173 void destroy_muxer_msg_iter(struct muxer_msg_iter *muxer_msg_iter)
1174 {
1175         struct muxer_comp *muxer_comp;
1176
1177         if (!muxer_msg_iter) {
1178                 return;
1179         }
1180
1181         muxer_comp = muxer_msg_iter->muxer_comp;
1182         BT_COMP_LOGD("Destroying muxer component's message iterator: "
1183                 "muxer-msg-iter-addr=%p", muxer_msg_iter);
1184
1185         if (muxer_msg_iter->active_muxer_upstream_msg_iters) {
1186                 BT_COMP_LOGD_STR("Destroying muxer's active upstream message iterator wrappers.");
1187                 g_ptr_array_free(
1188                         muxer_msg_iter->active_muxer_upstream_msg_iters, TRUE);
1189         }
1190
1191         if (muxer_msg_iter->ended_muxer_upstream_msg_iters) {
1192                 BT_COMP_LOGD_STR("Destroying muxer's ended upstream message iterator wrappers.");
1193                 g_ptr_array_free(
1194                         muxer_msg_iter->ended_muxer_upstream_msg_iters, TRUE);
1195         }
1196
1197         g_free(muxer_msg_iter);
1198 }
1199
1200 static
1201 bt_message_iterator_class_initialize_method_status
1202 muxer_msg_iter_init_upstream_iterators(struct muxer_comp *muxer_comp,
1203                 struct muxer_msg_iter *muxer_msg_iter,
1204                 struct bt_self_message_iterator_configuration *config)
1205 {
1206         int64_t count;
1207         int64_t i;
1208         bt_message_iterator_class_initialize_method_status status;
1209         bool can_seek_forward = true;
1210
1211         count = bt_component_filter_get_input_port_count(
1212                 bt_self_component_filter_as_component_filter(
1213                         muxer_comp->self_comp_flt));
1214         if (count < 0) {
1215                 BT_COMP_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                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
1219                 goto end;
1220         }
1221
1222         for (i = 0; i < count; i++) {
1223                 bt_message_iterator *upstream_msg_iter;
1224                 bt_self_component_port_input *self_port =
1225                         bt_self_component_filter_borrow_input_port_by_index(
1226                                 muxer_comp->self_comp_flt, i);
1227                 const bt_port *port;
1228                 bt_message_iterator_create_from_message_iterator_status
1229                         msg_iter_status;
1230                 int int_status;
1231
1232                 BT_ASSERT(self_port);
1233                 port = bt_self_component_port_as_port(
1234                         bt_self_component_port_input_as_self_component_port(
1235                                 self_port));
1236                 BT_ASSERT(port);
1237
1238                 if (!bt_port_is_connected(port)) {
1239                         /* Skip non-connected port */
1240                         continue;
1241                 }
1242
1243                 msg_iter_status = create_msg_iter_on_input_port(muxer_comp,
1244                         muxer_msg_iter, self_port, &upstream_msg_iter);
1245                 if (msg_iter_status != BT_MESSAGE_ITERATOR_CREATE_FROM_MESSAGE_ITERATOR_STATUS_OK) {
1246                         /* create_msg_iter_on_input_port() logs errors */
1247                         status = (int) msg_iter_status;
1248                         goto end;
1249                 }
1250
1251                 int_status = muxer_msg_iter_add_upstream_msg_iter(muxer_msg_iter,
1252                         upstream_msg_iter);
1253                 bt_message_iterator_put_ref(
1254                         upstream_msg_iter);
1255                 if (int_status) {
1256                         status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1257                         /* muxer_msg_iter_add_upstream_msg_iter() logs errors */
1258                         goto end;
1259                 }
1260
1261                 can_seek_forward = can_seek_forward &&
1262                         bt_message_iterator_can_seek_forward(
1263                                 upstream_msg_iter);
1264         }
1265
1266         /*
1267          * This iterator can seek forward if all of its iterators can seek
1268          * forward.
1269          */
1270         bt_self_message_iterator_configuration_set_can_seek_forward(
1271                 config, can_seek_forward);
1272
1273         status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
1274
1275 end:
1276         return status;
1277 }
1278
1279 BT_HIDDEN
1280 bt_message_iterator_class_initialize_method_status muxer_msg_iter_init(
1281                 bt_self_message_iterator *self_msg_iter,
1282                 bt_self_message_iterator_configuration *config,
1283                 bt_self_component_port_output *port)
1284 {
1285         struct muxer_comp *muxer_comp = NULL;
1286         struct muxer_msg_iter *muxer_msg_iter = NULL;
1287         bt_message_iterator_class_initialize_method_status status;
1288         bt_self_component *self_comp =
1289                 bt_self_message_iterator_borrow_component(self_msg_iter);
1290
1291         muxer_comp = bt_self_component_get_data(self_comp);
1292         BT_ASSERT(muxer_comp);
1293         BT_COMP_LOGD("Initializing muxer component's message iterator: "
1294                 "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1295                 self_comp, muxer_comp, self_msg_iter);
1296
1297         if (muxer_comp->initializing_muxer_msg_iter) {
1298                 /*
1299                  * Weird, unhandled situation detected: downstream
1300                  * creates a muxer message iterator while creating
1301                  * another muxer message iterator (same component).
1302                  */
1303                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1304                         "Recursive initialization of muxer component's message iterator: "
1305                         "comp-addr=%p, muxer-comp-addr=%p, msg-iter-addr=%p",
1306                         self_comp, muxer_comp, self_msg_iter);
1307                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1308                 goto error;
1309         }
1310
1311         muxer_comp->initializing_muxer_msg_iter = true;
1312         muxer_msg_iter = g_new0(struct muxer_msg_iter, 1);
1313         if (!muxer_msg_iter) {
1314                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1315                         "Failed to allocate one muxer component's message iterator.");
1316                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1317                 goto error;
1318         }
1319
1320         muxer_msg_iter->muxer_comp = muxer_comp;
1321         muxer_msg_iter->self_msg_iter = self_msg_iter;
1322         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1323         muxer_msg_iter->active_muxer_upstream_msg_iters =
1324                 g_ptr_array_new_with_free_func(
1325                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1326         if (!muxer_msg_iter->active_muxer_upstream_msg_iters) {
1327                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp, "Failed to allocate a GPtrArray.");
1328                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1329                 goto error;
1330         }
1331
1332         muxer_msg_iter->ended_muxer_upstream_msg_iters =
1333                 g_ptr_array_new_with_free_func(
1334                         (GDestroyNotify) destroy_muxer_upstream_msg_iter);
1335         if (!muxer_msg_iter->ended_muxer_upstream_msg_iters) {
1336                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp, "Failed to allocate a GPtrArray.");
1337                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1338                 goto error;
1339         }
1340
1341         status = muxer_msg_iter_init_upstream_iterators(muxer_comp,
1342                 muxer_msg_iter, config);
1343         if (status) {
1344                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1345                         "Cannot initialize connected input ports for muxer component's message iterator: "
1346                         "comp-addr=%p, muxer-comp-addr=%p, "
1347                         "muxer-msg-iter-addr=%p, msg-iter-addr=%p, ret=%d",
1348                         self_comp, muxer_comp, muxer_msg_iter,
1349                         self_msg_iter, status);
1350                 goto error;
1351         }
1352
1353         bt_self_message_iterator_set_data(self_msg_iter, muxer_msg_iter);
1354         BT_COMP_LOGD("Initialized muxer component's message iterator: "
1355                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1356                 "msg-iter-addr=%p",
1357                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1358         goto end;
1359
1360 error:
1361         destroy_muxer_msg_iter(muxer_msg_iter);
1362         bt_self_message_iterator_set_data(self_msg_iter, NULL);
1363
1364 end:
1365         muxer_comp->initializing_muxer_msg_iter = false;
1366         return status;
1367 }
1368
1369 BT_HIDDEN
1370 void muxer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
1371 {
1372         struct muxer_msg_iter *muxer_msg_iter =
1373                 bt_self_message_iterator_get_data(self_msg_iter);
1374         bt_self_component *self_comp = NULL;
1375         struct muxer_comp *muxer_comp = NULL;
1376
1377         self_comp = bt_self_message_iterator_borrow_component(
1378                 self_msg_iter);
1379         BT_ASSERT(self_comp);
1380         muxer_comp = bt_self_component_get_data(self_comp);
1381         BT_COMP_LOGD("Finalizing muxer component's message iterator: "
1382                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1383                 "msg-iter-addr=%p",
1384                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1385
1386         if (muxer_msg_iter) {
1387                 destroy_muxer_msg_iter(muxer_msg_iter);
1388         }
1389 }
1390
1391 BT_HIDDEN
1392 bt_message_iterator_class_next_method_status muxer_msg_iter_next(
1393                 bt_self_message_iterator *self_msg_iter,
1394                 bt_message_array_const msgs, uint64_t capacity,
1395                 uint64_t *count)
1396 {
1397         bt_message_iterator_class_next_method_status status;
1398         struct muxer_msg_iter *muxer_msg_iter =
1399                 bt_self_message_iterator_get_data(self_msg_iter);
1400         bt_self_component *self_comp = NULL;
1401         struct muxer_comp *muxer_comp = NULL;
1402
1403         BT_ASSERT_DBG(muxer_msg_iter);
1404         self_comp = bt_self_message_iterator_borrow_component(
1405                 self_msg_iter);
1406         BT_ASSERT_DBG(self_comp);
1407         muxer_comp = bt_self_component_get_data(self_comp);
1408         BT_ASSERT_DBG(muxer_comp);
1409         BT_COMP_LOGT("Muxer component's message iterator's \"next\" method called: "
1410                 "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1411                 "msg-iter-addr=%p",
1412                 self_comp, muxer_comp, muxer_msg_iter, self_msg_iter);
1413
1414         status = muxer_msg_iter_do_next(muxer_comp, muxer_msg_iter,
1415                 msgs, capacity, count);
1416         if (status < 0) {
1417                 BT_COMP_LOGE_APPEND_CAUSE(self_comp,
1418                         "Cannot get next message: "
1419                         "comp-addr=%p, muxer-comp-addr=%p, muxer-msg-iter-addr=%p, "
1420                         "msg-iter-addr=%p, status=%s",
1421                         self_comp, muxer_comp, muxer_msg_iter, self_msg_iter,
1422                         bt_common_func_status_string(status));
1423         } else {
1424                 BT_COMP_LOGT("Returning from muxer component's message iterator's \"next\" method: "
1425                         "status=%s",
1426                         bt_common_func_status_string(status));
1427         }
1428
1429         return status;
1430 }
1431
1432 BT_HIDDEN
1433 bt_component_class_port_connected_method_status muxer_input_port_connected(
1434                 bt_self_component_filter *self_comp,
1435                 bt_self_component_port_input *self_port,
1436                 const bt_port_output *other_port)
1437 {
1438         bt_component_class_port_connected_method_status status =
1439                 BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_OK;
1440         bt_self_component_add_port_status add_port_status;
1441         struct muxer_comp *muxer_comp = bt_self_component_get_data(
1442                 bt_self_component_filter_as_self_component(self_comp));
1443
1444         add_port_status = add_available_input_port(self_comp);
1445         if (add_port_status) {
1446                 BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1447                         "Cannot add one muxer component's input port: status=%s",
1448                         bt_common_func_status_string(add_port_status));
1449
1450                 if (add_port_status ==
1451                                 BT_SELF_COMPONENT_ADD_PORT_STATUS_MEMORY_ERROR) {
1452                         status = BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_MEMORY_ERROR;
1453                 } else {
1454                         status = BT_COMPONENT_CLASS_PORT_CONNECTED_METHOD_STATUS_ERROR;
1455                 }
1456
1457                 goto end;
1458         }
1459
1460 end:
1461         return status;
1462 }
1463
1464 static inline
1465 bt_message_iterator_class_can_seek_beginning_method_status
1466 muxer_upstream_msg_iters_can_all_seek_beginning(
1467                 struct muxer_comp *muxer_comp,
1468                 GPtrArray *muxer_upstream_msg_iters, bt_bool *can_seek)
1469 {
1470         bt_message_iterator_class_can_seek_beginning_method_status status =
1471                 BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK;
1472         uint64_t i;
1473
1474         for (i = 0; i < muxer_upstream_msg_iters->len; i++) {
1475                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1476                         muxer_upstream_msg_iters->pdata[i];
1477                 status = (int) bt_message_iterator_can_seek_beginning(
1478                         upstream_msg_iter->msg_iter, can_seek);
1479                 if (status != BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK) {
1480                         BT_COMP_LOGE_APPEND_CAUSE(muxer_comp->self_comp,
1481                                 "Failed to determine whether upstream message iterator can seek beginning: "
1482                                 "msg-iter-addr=%p", upstream_msg_iter->msg_iter);
1483                         goto end;
1484                 }
1485
1486                 if (!*can_seek) {
1487                         goto end;
1488                 }
1489         }
1490
1491         *can_seek = BT_TRUE;
1492
1493 end:
1494         return status;
1495 }
1496
1497 BT_HIDDEN
1498 bt_message_iterator_class_can_seek_beginning_method_status
1499 muxer_msg_iter_can_seek_beginning(
1500                 bt_self_message_iterator *self_msg_iter, bt_bool *can_seek)
1501 {
1502         struct muxer_msg_iter *muxer_msg_iter =
1503                 bt_self_message_iterator_get_data(self_msg_iter);
1504         bt_message_iterator_class_can_seek_beginning_method_status status;
1505
1506         status = muxer_upstream_msg_iters_can_all_seek_beginning(
1507                 muxer_msg_iter->muxer_comp,
1508                 muxer_msg_iter->active_muxer_upstream_msg_iters, can_seek);
1509         if (status != BT_MESSAGE_ITERATOR_CLASS_CAN_SEEK_BEGINNING_METHOD_STATUS_OK) {
1510                 goto end;
1511         }
1512
1513         if (!*can_seek) {
1514                 goto end;
1515         }
1516
1517         status = muxer_upstream_msg_iters_can_all_seek_beginning(
1518                 muxer_msg_iter->muxer_comp,
1519                 muxer_msg_iter->ended_muxer_upstream_msg_iters, can_seek);
1520
1521 end:
1522         return status;
1523 }
1524
1525 BT_HIDDEN
1526 bt_message_iterator_class_seek_beginning_method_status muxer_msg_iter_seek_beginning(
1527                 bt_self_message_iterator *self_msg_iter)
1528 {
1529         struct muxer_msg_iter *muxer_msg_iter =
1530                 bt_self_message_iterator_get_data(self_msg_iter);
1531         bt_message_iterator_class_seek_beginning_method_status status =
1532                 BT_MESSAGE_ITERATOR_CLASS_SEEK_BEGINNING_METHOD_STATUS_OK;
1533         bt_message_iterator_seek_beginning_status seek_beg_status;
1534         uint64_t i;
1535
1536         /* Seek all ended upstream iterators first */
1537         for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
1538                         i++) {
1539                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1540                         muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
1541
1542                 seek_beg_status = bt_message_iterator_seek_beginning(
1543                         upstream_msg_iter->msg_iter);
1544                 if (seek_beg_status != BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK) {
1545                         status = (int) seek_beg_status;
1546                         goto end;
1547                 }
1548
1549                 empty_message_queue(upstream_msg_iter);
1550         }
1551
1552         /* Seek all previously active upstream iterators */
1553         for (i = 0; i < muxer_msg_iter->active_muxer_upstream_msg_iters->len;
1554                         i++) {
1555                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1556                         muxer_msg_iter->active_muxer_upstream_msg_iters->pdata[i];
1557
1558                 seek_beg_status = bt_message_iterator_seek_beginning(
1559                         upstream_msg_iter->msg_iter);
1560                 if (seek_beg_status != BT_MESSAGE_ITERATOR_SEEK_BEGINNING_STATUS_OK) {
1561                         status = (int) seek_beg_status;
1562                         goto end;
1563                 }
1564
1565                 empty_message_queue(upstream_msg_iter);
1566         }
1567
1568         /* Make them all active */
1569         for (i = 0; i < muxer_msg_iter->ended_muxer_upstream_msg_iters->len;
1570                         i++) {
1571                 struct muxer_upstream_msg_iter *upstream_msg_iter =
1572                         muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i];
1573
1574                 g_ptr_array_add(muxer_msg_iter->active_muxer_upstream_msg_iters,
1575                         upstream_msg_iter);
1576                 muxer_msg_iter->ended_muxer_upstream_msg_iters->pdata[i] = NULL;
1577         }
1578
1579         /*
1580          * GLib < 2.48.0 asserts when g_ptr_array_remove_range() is
1581          * called on an empty array.
1582          */
1583         if (muxer_msg_iter->ended_muxer_upstream_msg_iters->len > 0) {
1584                 g_ptr_array_remove_range(muxer_msg_iter->ended_muxer_upstream_msg_iters,
1585                         0, muxer_msg_iter->ended_muxer_upstream_msg_iters->len);
1586         }
1587         muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
1588         muxer_msg_iter->clock_class_expectation =
1589                 MUXER_MSG_ITER_CLOCK_CLASS_EXPECTATION_ANY;
1590
1591 end:
1592         return status;
1593 }