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