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