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