src/plugins/utils/muxer/muxer.c

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