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_LOG_TAG "PLUGIN-UTILS-MUXER-FLT"
  24 #include "logging.h"
  25
  26 #include <babeltrace/babeltrace-internal.h>
  27 #include <babeltrace/compat/uuid-internal.h>
  28 #include <babeltrace/babeltrace.h>
  29 #include <babeltrace/values-internal.h>
  30 #include <babeltrace/graph/component-internal.h>
  31 #include <babeltrace/graph/notification-iterator-internal.h>
  32 #include <babeltrace/graph/connection-internal.h>
  33 #include <plugins-common.h>
  34 #include <glib.h>
  35 #include <stdbool.h>
  36 #include <inttypes.h>
  37 #include <babeltrace/assert-internal.h>
  38 #include <stdlib.h>
  39 #include <string.h>
  40
  41 #define ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME        "assume-absolute-clock-classes"
  42
  43 struct muxer_comp {
  44         /*
  45          * Array of struct
  46          * bt_private_connection_private_notification_iterator *
  47          * (weak refs)
  48          */
  49         GPtrArray *muxer_notif_iters;
  50
  51         /* Weak ref */
  52         struct bt_private_component *priv_comp;
  53         unsigned int next_port_num;
  54         size_t available_input_ports;
  55         bool initializing_muxer_notif_iter;
  56         bool assume_absolute_clock_classes;
  57 };
  58
  59 struct muxer_upstream_notif_iter {
  60         /* Owned by this, NULL if ended */
  61         struct bt_notification_iterator *notif_iter;
  62
  63         /* Contains `struct bt_notification *`, owned by this */
  64         GQueue *notifs;
  65 };
  66
  67 enum muxer_notif_iter_clock_class_expectation {
  68         MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_ANY = 0,
  69         MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE,
  70         MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID,
  71         MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID,
  72 };
  73
  74 struct muxer_notif_iter {
  75         /*
  76          * Array of struct muxer_upstream_notif_iter * (owned by this).
  77          *
  78          * NOTE: This array is searched in linearly to find the youngest
  79          * current notification. Keep this until benchmarks confirm that
  80          * another data structure is faster than this for our typical
  81          * use cases.
  82          */
  83         GPtrArray *muxer_upstream_notif_iters;
  84
  85         /*
  86          * List of "recently" connected input ports (weak) to
  87          * handle by this muxer notification iterator.
  88          * muxer_port_connected() adds entries to this list, and the
  89          * entries are removed when a notification iterator is created
  90          * on the port's connection and put into
  91          * muxer_upstream_notif_iters above by
  92          * muxer_notif_iter_handle_newly_connected_ports().
  93          */
  94         GList *newly_connected_priv_ports;
  95
  96         /* Last time returned in a notification */
  97         int64_t last_returned_ts_ns;
  98
  99         /* Clock class expectation state */
 100         enum muxer_notif_iter_clock_class_expectation clock_class_expectation;
 101
 102         /*
 103          * Expected clock class UUID, only valid when
 104          * clock_class_expectation is
 105          * MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID.
 106          */
 107         unsigned char expected_clock_class_uuid[BABELTRACE_UUID_LEN];
 108 };
 109
 110 static
 111 void destroy_muxer_upstream_notif_iter(
 112                 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
 113 {
 114         if (!muxer_upstream_notif_iter) {
 115                 return;
 116         }
 117
 118         BT_LOGD("Destroying muxer's upstream notification iterator wrapper: "
 119                 "addr=%p, notif-iter-addr=%p, queue-len=%u",
 120                 muxer_upstream_notif_iter,
 121                 muxer_upstream_notif_iter->notif_iter,
 122                 muxer_upstream_notif_iter->notifs->length);
 123         bt_put(muxer_upstream_notif_iter->notif_iter);
 124
 125         if (muxer_upstream_notif_iter->notifs) {
 126                 struct bt_notification *notif;
 127
 128                 while ((notif = g_queue_pop_head(
 129                                 muxer_upstream_notif_iter->notifs))) {
 130                         bt_put(notif);
 131                 }
 132
 133                 g_queue_free(muxer_upstream_notif_iter->notifs);
 134         }
 135
 136         g_free(muxer_upstream_notif_iter);
 137 }
 138
 139 static
 140 struct muxer_upstream_notif_iter *muxer_notif_iter_add_upstream_notif_iter(
 141                 struct muxer_notif_iter *muxer_notif_iter,
 142                 struct bt_notification_iterator *notif_iter,
 143                 struct bt_private_port *priv_port)
 144 {
 145         struct muxer_upstream_notif_iter *muxer_upstream_notif_iter =
 146                 g_new0(struct muxer_upstream_notif_iter, 1);
 147
 148         if (!muxer_upstream_notif_iter) {
 149                 BT_LOGE_STR("Failed to allocate one muxer's upstream notification iterator wrapper.");
 150                 goto end;
 151         }
 152
 153         muxer_upstream_notif_iter->notif_iter = bt_get(notif_iter);
 154         muxer_upstream_notif_iter->notifs = g_queue_new();
 155         if (!muxer_upstream_notif_iter->notifs) {
 156                 BT_LOGE_STR("Failed to allocate a GQueue.");
 157
 158                 goto end;
 159         }
 160
 161         g_ptr_array_add(muxer_notif_iter->muxer_upstream_notif_iters,
 162                 muxer_upstream_notif_iter);
 163         BT_LOGD("Added muxer's upstream notification iterator wrapper: "
 164                 "addr=%p, muxer-notif-iter-addr=%p, notif-iter-addr=%p",
 165                 muxer_upstream_notif_iter, muxer_notif_iter,
 166                 notif_iter);
 167
 168 end:
 169         return muxer_upstream_notif_iter;
 170 }
 171
 172 static
 173 enum bt_component_status ensure_available_input_port(
 174                 struct bt_private_component *priv_comp)
 175 {
 176         struct muxer_comp *muxer_comp =
 177                 bt_private_component_get_user_data(priv_comp);
 178         enum bt_component_status status = BT_COMPONENT_STATUS_OK;
 179         GString *port_name = NULL;
 180
 181         BT_ASSERT(muxer_comp);
 182
 183         if (muxer_comp->available_input_ports >= 1) {
 184                 goto end;
 185         }
 186
 187         port_name = g_string_new("in");
 188         if (!port_name) {
 189                 BT_LOGE_STR("Failed to allocate a GString.");
 190                 status = BT_COMPONENT_STATUS_NOMEM;
 191                 goto end;
 192         }
 193
 194         g_string_append_printf(port_name, "%u", muxer_comp->next_port_num);
 195         status = bt_private_component_filter_add_input_private_port(
 196                 priv_comp, port_name->str, NULL, NULL);
 197         if (status != BT_COMPONENT_STATUS_OK) {
 198                 BT_LOGE("Cannot add input port to muxer component: "
 199                         "port-name=\"%s\", comp-addr=%p, status=%s",
 200                         port_name->str, priv_comp,
 201                         bt_component_status_string(status));
 202                 goto end;
 203         }
 204
 205         muxer_comp->available_input_ports++;
 206         muxer_comp->next_port_num++;
 207         BT_LOGD("Added one input port to muxer component: "
 208                 "port-name=\"%s\", comp-addr=%p",
 209                 port_name->str, priv_comp);
 210 end:
 211         if (port_name) {
 212                 g_string_free(port_name, TRUE);
 213         }
 214
 215         return status;
 216 }
 217
 218 static
 219 enum bt_component_status create_output_port(
 220                 struct bt_private_component *priv_comp)
 221 {
 222         return bt_private_component_filter_add_output_private_port(
 223                 priv_comp, "out", NULL, NULL);
 224 }
 225
 226 static
 227 void destroy_muxer_comp(struct muxer_comp *muxer_comp)
 228 {
 229         if (!muxer_comp) {
 230                 return;
 231         }
 232
 233         BT_LOGD("Destroying muxer component: muxer-comp-addr=%p, "
 234                 "muxer-notif-iter-count=%u", muxer_comp,
 235                 muxer_comp->muxer_notif_iters ?
 236                         muxer_comp->muxer_notif_iters->len : 0);
 237
 238         if (muxer_comp->muxer_notif_iters) {
 239                 g_ptr_array_free(muxer_comp->muxer_notif_iters, TRUE);
 240         }
 241
 242         g_free(muxer_comp);
 243 }
 244
 245 static
 246 struct bt_value *get_default_params(void)
 247 {
 248         struct bt_value *params;
 249         int ret;
 250
 251         params = bt_value_map_create();
 252         if (!params) {
 253                 BT_LOGE_STR("Cannot create a map value object.");
 254                 goto error;
 255         }
 256
 257         ret = bt_value_map_insert_bool(params,
 258                 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, false);
 259         if (ret) {
 260                 BT_LOGE_STR("Cannot add boolean value to map value object.");
 261                 goto error;
 262         }
 263
 264         goto end;
 265
 266 error:
 267         BT_PUT(params);
 268
 269 end:
 270         return params;
 271 }
 272
 273 static
 274 int configure_muxer_comp(struct muxer_comp *muxer_comp, struct bt_value *params)
 275 {
 276         struct bt_value *default_params = NULL;
 277         struct bt_value *real_params = NULL;
 278         struct bt_value *assume_absolute_clock_classes = NULL;
 279         int ret = 0;
 280         bt_bool bool_val;
 281
 282         default_params = get_default_params();
 283         if (!default_params) {
 284                 BT_LOGE("Cannot get default parameters: "
 285                         "muxer-comp-addr=%p", muxer_comp);
 286                 goto error;
 287         }
 288
 289         real_params = bt_value_map_extend(default_params, params);
 290         if (!real_params) {
 291                 BT_LOGE("Cannot extend default parameters map value: "
 292                         "muxer-comp-addr=%p, def-params-addr=%p, "
 293                         "params-addr=%p", muxer_comp, default_params,
 294                         params);
 295                 goto error;
 296         }
 297
 298         assume_absolute_clock_classes = bt_value_map_borrow(real_params,
 299                 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME);
 300         if (assume_absolute_clock_classes &&
 301                         !bt_value_is_bool(assume_absolute_clock_classes)) {
 302                 BT_LOGE("Expecting a boolean value for the `%s` parameter: "
 303                         "muxer-comp-addr=%p, value-type=%s",
 304                         ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, muxer_comp,
 305                         bt_value_type_string(
 306                                 bt_value_get_type(assume_absolute_clock_classes)));
 307                 goto error;
 308         }
 309
 310         ret = bt_value_bool_get(assume_absolute_clock_classes, &bool_val);
 311         BT_ASSERT(ret == 0);
 312         muxer_comp->assume_absolute_clock_classes = (bool) bool_val;
 313         BT_LOGD("Configured muxer component: muxer-comp-addr=%p, "
 314                 "assume-absolute-clock-classes=%d",
 315                 muxer_comp, muxer_comp->assume_absolute_clock_classes);
 316         goto end;
 317
 318 error:
 319         ret = -1;
 320
 321 end:
 322         bt_put(default_params);
 323         bt_put(real_params);
 324         return ret;
 325 }
 326
 327 BT_HIDDEN
 328 enum bt_component_status muxer_init(
 329                 struct bt_private_component *priv_comp,
 330                 struct bt_value *params, void *init_data)
 331 {
 332         int ret;
 333         enum bt_component_status status = BT_COMPONENT_STATUS_OK;
 334         struct muxer_comp *muxer_comp = g_new0(struct muxer_comp, 1);
 335
 336         BT_LOGD("Initializing muxer component: "
 337                 "comp-addr=%p, params-addr=%p", priv_comp, params);
 338
 339         if (!muxer_comp) {
 340                 BT_LOGE_STR("Failed to allocate one muxer component.");
 341                 goto error;
 342         }
 343
 344         ret = configure_muxer_comp(muxer_comp, params);
 345         if (ret) {
 346                 BT_LOGE("Cannot configure muxer component: "
 347                         "muxer-comp-addr=%p, params-addr=%p",
 348                         muxer_comp, params);
 349                 goto error;
 350         }
 351
 352         muxer_comp->muxer_notif_iters = g_ptr_array_new();
 353         if (!muxer_comp->muxer_notif_iters) {
 354                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
 355                 goto error;
 356         }
 357
 358         muxer_comp->priv_comp = priv_comp;
 359         ret = bt_private_component_set_user_data(priv_comp, muxer_comp);
 360         BT_ASSERT(ret == 0);
 361         status = ensure_available_input_port(priv_comp);
 362         if (status != BT_COMPONENT_STATUS_OK) {
 363                 BT_LOGE("Cannot ensure that at least one muxer component's input port is available: "
 364                         "muxer-comp-addr=%p, status=%s",
 365                         muxer_comp,
 366                         bt_component_status_string(status));
 367                 goto error;
 368         }
 369
 370         status = create_output_port(priv_comp);
 371         if (status) {
 372                 BT_LOGE("Cannot create muxer component's output port: "
 373                         "muxer-comp-addr=%p, status=%s",
 374                         muxer_comp,
 375                         bt_component_status_string(status));
 376                 goto error;
 377         }
 378
 379         BT_LOGD("Initialized muxer component: "
 380                 "comp-addr=%p, params-addr=%p, muxer-comp-addr=%p",
 381                 priv_comp, params, muxer_comp);
 382
 383         goto end;
 384
 385 error:
 386         destroy_muxer_comp(muxer_comp);
 387         ret = bt_private_component_set_user_data(priv_comp, NULL);
 388         BT_ASSERT(ret == 0);
 389
 390         if (status == BT_COMPONENT_STATUS_OK) {
 391                 status = BT_COMPONENT_STATUS_ERROR;
 392         }
 393
 394 end:
 395         return status;
 396 }
 397
 398 BT_HIDDEN
 399 void muxer_finalize(struct bt_private_component *priv_comp)
 400 {
 401         struct muxer_comp *muxer_comp =
 402                 bt_private_component_get_user_data(priv_comp);
 403
 404         BT_LOGD("Finalizing muxer component: comp-addr=%p",
 405                 priv_comp);
 406         destroy_muxer_comp(muxer_comp);
 407 }
 408
 409 static
 410 struct bt_notification_iterator *create_notif_iter_on_input_port(
 411                 struct bt_private_port *priv_port, int *ret)
 412 {
 413         struct bt_port *port = bt_port_borrow_from_private(priv_port);
 414         struct bt_notification_iterator *notif_iter = NULL;
 415         struct bt_private_connection *priv_conn = NULL;
 416         enum bt_connection_status conn_status;
 417
 418         BT_ASSERT(ret);
 419         *ret = 0;
 420         BT_ASSERT(port);
 421         BT_ASSERT(bt_port_is_connected(port));
 422         priv_conn = bt_private_port_get_private_connection(priv_port);
 423         BT_ASSERT(priv_conn);
 424
 425         // TODO: Advance the iterator to >= the time of the latest
 426         //       returned notification by the muxer notification
 427         //       iterator which creates it.
 428         conn_status = bt_private_connection_create_notification_iterator(
 429                 priv_conn, &notif_iter);
 430         if (conn_status != BT_CONNECTION_STATUS_OK) {
 431                 BT_LOGE("Cannot create upstream notification iterator on input port's connection: "
 432                         "port-addr=%p, port-name=\"%s\", conn-addr=%p, "
 433                         "status=%s",
 434                         port, bt_port_get_name(port), priv_conn,
 435                         bt_connection_status_string(conn_status));
 436                 *ret = -1;
 437                 goto end;
 438         }
 439
 440         BT_LOGD("Created upstream notification iterator on input port's connection: "
 441                 "port-addr=%p, port-name=\"%s\", conn-addr=%p, "
 442                 "notif-iter-addr=%p",
 443                 port, bt_port_get_name(port), priv_conn, notif_iter);
 444
 445 end:
 446         bt_put(priv_conn);
 447         return notif_iter;
 448 }
 449
 450 static
 451 enum bt_notification_iterator_status muxer_upstream_notif_iter_next(
 452                 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
 453 {
 454         enum bt_notification_iterator_status status;
 455         bt_notification_array notifs;
 456         uint64_t i;
 457         uint64_t count;
 458
 459         BT_LOGV("Calling upstream notification iterator's \"next\" method: "
 460                 "muxer-upstream-notif-iter-wrap-addr=%p, notif-iter-addr=%p",
 461                 muxer_upstream_notif_iter,
 462                 muxer_upstream_notif_iter->notif_iter);
 463         status = bt_private_connection_notification_iterator_next(
 464                 muxer_upstream_notif_iter->notif_iter, &notifs, &count);
 465         BT_LOGV("Upstream notification iterator's \"next\" method returned: "
 466                 "status=%s", bt_notification_iterator_status_string(status));
 467
 468         switch (status) {
 469         case BT_NOTIFICATION_ITERATOR_STATUS_OK:
 470                 /*
 471                  * Notification iterator's current notification is
 472                  * valid: it must be considered for muxing operations.
 473                  */
 474                 BT_LOGV_STR("Validated upstream notification iterator wrapper.");
 475                 BT_ASSERT(count > 0);
 476
 477                 /* Move notifications to our queue */
 478                 for (i = 0; i < count; i++) {
 479                         /*
 480                          * Push to tail in order; other side
 481                          * (muxer_notif_iter_do_next_one()) consumes
 482                          * from the head first.
 483                          */
 484                         g_queue_push_tail(muxer_upstream_notif_iter->notifs,
 485                                 notifs[i]);
 486                 }
 487                 break;
 488         case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
 489                 /*
 490                  * Notification iterator's current notification is not
 491                  * valid anymore. Return
 492                  * BT_NOTIFICATION_ITERATOR_STATUS_AGAIN immediately.
 493                  */
 494                 break;
 495         case BT_NOTIFICATION_ITERATOR_STATUS_END:       /* Fall-through. */
 496         case BT_NOTIFICATION_ITERATOR_STATUS_CANCELED:
 497                 /*
 498                  * Notification iterator reached the end: release it. It
 499                  * won't be considered again to find the youngest
 500                  * notification.
 501                  */
 502                 BT_PUT(muxer_upstream_notif_iter->notif_iter);
 503                 status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
 504                 break;
 505         default:
 506                 /* Error or unsupported status code */
 507                 BT_LOGE("Error or unsupported status code: "
 508                         "status-code=%d", status);
 509                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 510                 break;
 511         }
 512
 513         return status;
 514 }
 515
 516 static
 517 int muxer_notif_iter_handle_newly_connected_ports(
 518                 struct muxer_notif_iter *muxer_notif_iter)
 519 {
 520         int ret = 0;
 521
 522         BT_LOGV("Handling newly connected ports: "
 523                 "muxer-notif-iter-addr=%p", muxer_notif_iter);
 524
 525         /*
 526          * Here we create one upstream notification iterator for each
 527          * newly connected port. We do NOT perform an initial "next" on
 528          * those new upstream notification iterators: they are
 529          * invalidated, to be validated later. The list of newly
 530          * connected ports to handle here is updated by
 531          * muxer_port_connected().
 532          */
 533         while (true) {
 534                 GList *node = muxer_notif_iter->newly_connected_priv_ports;
 535                 struct bt_private_port *priv_port;
 536                 struct bt_port *port;
 537                 struct bt_notification_iterator *upstream_notif_iter = NULL;
 538                 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter;
 539
 540                 if (!node) {
 541                         break;
 542                 }
 543
 544                 priv_port = node->data;
 545                 port = bt_port_borrow_from_private(priv_port);
 546                 BT_ASSERT(port);
 547
 548                 if (!bt_port_is_connected(port)) {
 549                         /*
 550                          * Looks like this port is not connected
 551                          * anymore: we can't create an upstream
 552                          * notification iterator on its (non-existing)
 553                          * connection in this case.
 554                          */
 555                         goto remove_node;
 556                 }
 557
 558                 upstream_notif_iter = create_notif_iter_on_input_port(priv_port,
 559                         &ret);
 560                 if (ret) {
 561                         /* create_notif_iter_on_input_port() logs errors */
 562                         BT_ASSERT(!upstream_notif_iter);
 563                         goto error;
 564                 }
 565
 566                 muxer_upstream_notif_iter =
 567                         muxer_notif_iter_add_upstream_notif_iter(
 568                                 muxer_notif_iter, upstream_notif_iter,
 569                                 priv_port);
 570                 BT_PUT(upstream_notif_iter);
 571                 if (!muxer_upstream_notif_iter) {
 572                         /*
 573                          * muxer_notif_iter_add_upstream_notif_iter()
 574                          * logs errors.
 575                          */
 576                         goto error;
 577                 }
 578
 579 remove_node:
 580                 bt_put(upstream_notif_iter);
 581                 muxer_notif_iter->newly_connected_priv_ports =
 582                         g_list_delete_link(
 583                                 muxer_notif_iter->newly_connected_priv_ports,
 584                                 node);
 585         }
 586
 587         goto end;
 588
 589 error:
 590         if (ret >= 0) {
 591                 ret = -1;
 592         }
 593
 594 end:
 595         return ret;
 596 }
 597
 598 static
 599 int get_notif_ts_ns(struct muxer_comp *muxer_comp,
 600                 struct muxer_notif_iter *muxer_notif_iter,
 601                 struct bt_notification *notif, int64_t last_returned_ts_ns,
 602                 int64_t *ts_ns)
 603 {
 604         struct bt_clock_class_priority_map *cc_prio_map = NULL;
 605         struct bt_clock_class *clock_class = NULL;
 606         struct bt_clock_value *clock_value = NULL;
 607         struct bt_event *event = NULL;
 608         int ret = 0;
 609         const unsigned char *cc_uuid;
 610         const char *cc_name;
 611
 612         BT_ASSERT(notif);
 613         BT_ASSERT(ts_ns);
 614
 615         BT_LOGV("Getting notification's timestamp: "
 616                 "muxer-notif-iter-addr=%p, notif-addr=%p, "
 617                 "last-returned-ts=%" PRId64,
 618                 muxer_notif_iter, notif, last_returned_ts_ns);
 619
 620         switch (bt_notification_get_type(notif)) {
 621         case BT_NOTIFICATION_TYPE_EVENT:
 622                 cc_prio_map =
 623                         bt_notification_event_borrow_clock_class_priority_map(
 624                                 notif);
 625                 break;
 626
 627         case BT_NOTIFICATION_TYPE_INACTIVITY:
 628                 cc_prio_map =
 629                         bt_notification_inactivity_borrow_clock_class_priority_map(
 630                                 notif);
 631                 break;
 632         default:
 633                 /* All the other notifications have a higher priority */
 634                 BT_LOGV_STR("Notification has no timestamp: using the last returned timestamp.");
 635                 *ts_ns = last_returned_ts_ns;
 636                 goto end;
 637         }
 638
 639         if (!cc_prio_map) {
 640                 BT_LOGE("Cannot get notification's clock class priority map: "
 641                         "notif-addr=%p", notif);
 642                 goto error;
 643         }
 644
 645         /*
 646          * If the clock class priority map is empty, then we consider
 647          * that this notification has no time. In this case it's always
 648          * the youngest.
 649          */
 650         if (bt_clock_class_priority_map_get_clock_class_count(cc_prio_map) == 0) {
 651                 BT_LOGV_STR("Notification's clock class priority map contains 0 clock classes: "
 652                         "using the last returned timestamp.");
 653                 *ts_ns = last_returned_ts_ns;
 654                 goto end;
 655         }
 656
 657         clock_class =
 658                 bt_clock_class_priority_map_borrow_highest_priority_clock_class(
 659                         cc_prio_map);
 660         if (!clock_class) {
 661                 BT_LOGE("Cannot get the clock class with the highest priority from clock class priority map: "
 662                         "cc-prio-map-addr=%p", cc_prio_map);
 663                 goto error;
 664         }
 665
 666         cc_uuid = bt_clock_class_get_uuid(clock_class);
 667         cc_name = bt_clock_class_get_name(clock_class);
 668
 669         if (muxer_notif_iter->clock_class_expectation ==
 670                         MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_ANY) {
 671                 /*
 672                  * This is the first clock class that this muxer
 673                  * notification iterator encounters. Its properties
 674                  * determine what to expect for the whole lifetime of
 675                  * the iterator without a true
 676                  * `assume-absolute-clock-classes` parameter.
 677                  */
 678                 if (bt_clock_class_is_absolute(clock_class)) {
 679                         /* Expect absolute clock classes */
 680                         muxer_notif_iter->clock_class_expectation =
 681                                 MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE;
 682                 } else {
 683                         if (cc_uuid) {
 684                                 /*
 685                                  * Expect non-absolute clock classes
 686                                  * with a specific UUID.
 687                                  */
 688                                 muxer_notif_iter->clock_class_expectation =
 689                                         MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID;
 690                                 memcpy(muxer_notif_iter->expected_clock_class_uuid,
 691                                         cc_uuid, BABELTRACE_UUID_LEN);
 692                         } else {
 693                                 /*
 694                                  * Expect non-absolute clock classes
 695                                  * with no UUID.
 696                                  */
 697                                 muxer_notif_iter->clock_class_expectation =
 698                                         MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID;
 699                         }
 700                 }
 701         }
 702
 703         if (!muxer_comp->assume_absolute_clock_classes) {
 704                 switch (muxer_notif_iter->clock_class_expectation) {
 705                 case MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_ABSOLUTE:
 706                         if (!bt_clock_class_is_absolute(clock_class)) {
 707                                 BT_LOGE("Expecting an absolute clock class, "
 708                                         "but got a non-absolute one: "
 709                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 710                                         clock_class, cc_name);
 711                                 goto error;
 712                         }
 713                         break;
 714                 case MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_NO_UUID:
 715                         if (bt_clock_class_is_absolute(clock_class)) {
 716                                 BT_LOGE("Expecting a non-absolute clock class with no UUID, "
 717                                         "but got an absolute one: "
 718                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 719                                         clock_class, cc_name);
 720                                 goto error;
 721                         }
 722
 723                         if (cc_uuid) {
 724                                 BT_LOGE("Expecting a non-absolute clock class with no UUID, "
 725                                         "but got one with a UUID: "
 726                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 727                                         "uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\"",
 728                                         clock_class, cc_name,
 729                                         (unsigned int) cc_uuid[0],
 730                                         (unsigned int) cc_uuid[1],
 731                                         (unsigned int) cc_uuid[2],
 732                                         (unsigned int) cc_uuid[3],
 733                                         (unsigned int) cc_uuid[4],
 734                                         (unsigned int) cc_uuid[5],
 735                                         (unsigned int) cc_uuid[6],
 736                                         (unsigned int) cc_uuid[7],
 737                                         (unsigned int) cc_uuid[8],
 738                                         (unsigned int) cc_uuid[9],
 739                                         (unsigned int) cc_uuid[10],
 740                                         (unsigned int) cc_uuid[11],
 741                                         (unsigned int) cc_uuid[12],
 742                                         (unsigned int) cc_uuid[13],
 743                                         (unsigned int) cc_uuid[14],
 744                                         (unsigned int) cc_uuid[15]);
 745                                 goto error;
 746                         }
 747                         break;
 748                 case MUXER_NOTIF_ITER_CLOCK_CLASS_EXPECTATION_NOT_ABS_SPEC_UUID:
 749                         if (bt_clock_class_is_absolute(clock_class)) {
 750                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 751                                         "but got an absolute one: "
 752                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 753                                         clock_class, cc_name);
 754                                 goto error;
 755                         }
 756
 757                         if (!cc_uuid) {
 758                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 759                                         "but got one with no UUID: "
 760                                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 761                                         clock_class, cc_name);
 762                                 goto error;
 763                         }
 764
 765                         if (memcmp(muxer_notif_iter->expected_clock_class_uuid,
 766                                         cc_uuid, BABELTRACE_UUID_LEN) != 0) {
 767                                 BT_LOGE("Expecting a non-absolute clock class with a specific UUID, "
 768                                         "but got one with different UUID: "
 769                                         "clock-class-addr=%p, clock-class-name=\"%s\", "
 770                                         "expected-uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\", "
 771                                         "uuid=\"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x\"",
 772                                         clock_class, cc_name,
 773                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[0],
 774                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[1],
 775                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[2],
 776                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[3],
 777                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[4],
 778                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[5],
 779                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[6],
 780                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[7],
 781                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[8],
 782                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[9],
 783                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[10],
 784                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[11],
 785                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[12],
 786                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[13],
 787                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[14],
 788                                         (unsigned int) muxer_notif_iter->expected_clock_class_uuid[15],
 789                                         (unsigned int) cc_uuid[0],
 790                                         (unsigned int) cc_uuid[1],
 791                                         (unsigned int) cc_uuid[2],
 792                                         (unsigned int) cc_uuid[3],
 793                                         (unsigned int) cc_uuid[4],
 794                                         (unsigned int) cc_uuid[5],
 795                                         (unsigned int) cc_uuid[6],
 796                                         (unsigned int) cc_uuid[7],
 797                                         (unsigned int) cc_uuid[8],
 798                                         (unsigned int) cc_uuid[9],
 799                                         (unsigned int) cc_uuid[10],
 800                                         (unsigned int) cc_uuid[11],
 801                                         (unsigned int) cc_uuid[12],
 802                                         (unsigned int) cc_uuid[13],
 803                                         (unsigned int) cc_uuid[14],
 804                                         (unsigned int) cc_uuid[15]);
 805                                 goto error;
 806                         }
 807                         break;
 808                 default:
 809                         /* Unexpected */
 810                         BT_LOGF("Unexpected clock class expectation: "
 811                                 "expectation-code=%d",
 812                                 muxer_notif_iter->clock_class_expectation);
 813                         abort();
 814                 }
 815         }
 816
 817         switch (bt_notification_get_type(notif)) {
 818         case BT_NOTIFICATION_TYPE_EVENT:
 819                 event = bt_notification_event_borrow_event(notif);
 820                 BT_ASSERT(event);
 821                 clock_value = bt_event_borrow_clock_value(event,
 822                         clock_class);
 823                 break;
 824         case BT_NOTIFICATION_TYPE_INACTIVITY:
 825                 clock_value = bt_notification_inactivity_borrow_clock_value(
 826                         notif, clock_class);
 827                 break;
 828         default:
 829                 BT_LOGF("Unexpected notification type at this point: "
 830                         "type=%d", bt_notification_get_type(notif));
 831                 abort();
 832         }
 833
 834         if (!clock_value) {
 835                 BT_LOGE("Cannot get notification's clock value for clock class: "
 836                         "clock-class-addr=%p, clock-class-name=\"%s\"",
 837                         clock_class, cc_name);
 838                 goto error;
 839         }
 840
 841         ret = bt_clock_value_get_value_ns_from_epoch(clock_value, ts_ns);
 842         if (ret) {
 843                 BT_LOGE("Cannot get nanoseconds from Epoch of clock value: "
 844                         "clock-value-addr=%p", clock_value);
 845                 goto error;
 846         }
 847
 848         goto end;
 849
 850 error:
 851         ret = -1;
 852
 853 end:
 854         if (ret == 0) {
 855                 BT_LOGV("Found notification's timestamp: "
 856                         "muxer-notif-iter-addr=%p, notif-addr=%p, "
 857                         "last-returned-ts=%" PRId64 ", ts=%" PRId64,
 858                         muxer_notif_iter, notif, last_returned_ts_ns,
 859                         *ts_ns);
 860         }
 861
 862         return ret;
 863 }
 864
 865 /*
 866  * This function finds the youngest available notification amongst the
 867  * non-ended upstream notification iterators and returns the upstream
 868  * notification iterator which has it, or
 869  * BT_NOTIFICATION_ITERATOR_STATUS_END if there's no available
 870  * notification.
 871  *
 872  * This function does NOT:
 873  *
 874  * * Update any upstream notification iterator.
 875  * * Check for newly connected ports.
 876  * * Check the upstream notification iterators to retry.
 877  *
 878  * On sucess, this function sets *muxer_upstream_notif_iter to the
 879  * upstream notification iterator of which the current notification is
 880  * the youngest, and sets *ts_ns to its time.
 881  */
 882 static
 883 enum bt_notification_iterator_status
 884 muxer_notif_iter_youngest_upstream_notif_iter(
 885                 struct muxer_comp *muxer_comp,
 886                 struct muxer_notif_iter *muxer_notif_iter,
 887                 struct muxer_upstream_notif_iter **muxer_upstream_notif_iter,
 888                 int64_t *ts_ns)
 889 {
 890         size_t i;
 891         int ret;
 892         int64_t youngest_ts_ns = INT64_MAX;
 893         enum bt_notification_iterator_status status =
 894                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
 895
 896         BT_ASSERT(muxer_comp);
 897         BT_ASSERT(muxer_notif_iter);
 898         BT_ASSERT(muxer_upstream_notif_iter);
 899         *muxer_upstream_notif_iter = NULL;
 900
 901         for (i = 0; i < muxer_notif_iter->muxer_upstream_notif_iters->len; i++) {
 902                 struct bt_notification *notif;
 903                 struct muxer_upstream_notif_iter *cur_muxer_upstream_notif_iter =
 904                         g_ptr_array_index(muxer_notif_iter->muxer_upstream_notif_iters, i);
 905                 int64_t notif_ts_ns;
 906
 907                 if (!cur_muxer_upstream_notif_iter->notif_iter) {
 908                         /* This upstream notification iterator is ended */
 909                         BT_LOGV("Skipping ended upstream notification iterator: "
 910                                 "muxer-upstream-notif-iter-wrap-addr=%p",
 911                                 cur_muxer_upstream_notif_iter);
 912                         continue;
 913                 }
 914
 915                 BT_ASSERT(cur_muxer_upstream_notif_iter->notifs->length > 0);
 916                 notif = g_queue_peek_head(cur_muxer_upstream_notif_iter->notifs);
 917                 BT_ASSERT(notif);
 918                 ret = get_notif_ts_ns(muxer_comp, muxer_notif_iter, notif,
 919                         muxer_notif_iter->last_returned_ts_ns, &notif_ts_ns);
 920                 if (ret) {
 921                         /* get_notif_ts_ns() logs errors */
 922                         *muxer_upstream_notif_iter = NULL;
 923                         status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 924                         goto end;
 925                 }
 926
 927                 if (notif_ts_ns <= youngest_ts_ns) {
 928                         *muxer_upstream_notif_iter =
 929                                 cur_muxer_upstream_notif_iter;
 930                         youngest_ts_ns = notif_ts_ns;
 931                         *ts_ns = youngest_ts_ns;
 932                 }
 933         }
 934
 935         if (!*muxer_upstream_notif_iter) {
 936                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
 937                 *ts_ns = INT64_MIN;
 938         }
 939
 940 end:
 941         return status;
 942 }
 943
 944 static
 945 enum bt_notification_iterator_status validate_muxer_upstream_notif_iter(
 946         struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
 947 {
 948         enum bt_notification_iterator_status status =
 949                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
 950
 951         BT_LOGV("Validating muxer's upstream notification iterator wrapper: "
 952                 "muxer-upstream-notif-iter-wrap-addr=%p",
 953                 muxer_upstream_notif_iter);
 954
 955         if (muxer_upstream_notif_iter->notifs->length > 0 ||
 956                         !muxer_upstream_notif_iter->notif_iter) {
 957                 BT_LOGV("Already valid or not considered: "
 958                         "queue-len=%u, upstream-notif-iter-addr=%p",
 959                         muxer_upstream_notif_iter->notifs->length,
 960                         muxer_upstream_notif_iter->notif_iter);
 961                 goto end;
 962         }
 963
 964         /* muxer_upstream_notif_iter_next() logs details/errors */
 965         status = muxer_upstream_notif_iter_next(muxer_upstream_notif_iter);
 966
 967 end:
 968         return status;
 969 }
 970
 971 static
 972 enum bt_notification_iterator_status validate_muxer_upstream_notif_iters(
 973                 struct muxer_notif_iter *muxer_notif_iter)
 974 {
 975         enum bt_notification_iterator_status status =
 976                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
 977         size_t i;
 978
 979         BT_LOGV("Validating muxer's upstream notification iterator wrappers: "
 980                 "muxer-notif-iter-addr=%p", muxer_notif_iter);
 981
 982         for (i = 0; i < muxer_notif_iter->muxer_upstream_notif_iters->len; i++) {
 983                 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter =
 984                         g_ptr_array_index(
 985                                 muxer_notif_iter->muxer_upstream_notif_iters,
 986                                 i);
 987
 988                 status = validate_muxer_upstream_notif_iter(
 989                         muxer_upstream_notif_iter);
 990                 if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
 991                         if (status < 0) {
 992                                 BT_LOGE("Cannot validate muxer's upstream notification iterator wrapper: "
 993                                         "muxer-notif-iter-addr=%p, "
 994                                         "muxer-upstream-notif-iter-wrap-addr=%p",
 995                                         muxer_notif_iter,
 996                                         muxer_upstream_notif_iter);
 997                         } else {
 998                                 BT_LOGV("Cannot validate muxer's upstream notification iterator wrapper: "
 999                                         "muxer-notif-iter-addr=%p, "
1000                                         "muxer-upstream-notif-iter-wrap-addr=%p",
1001                                         muxer_notif_iter,
1002                                         muxer_upstream_notif_iter);
1003                         }
1004
1005                         goto end;
1006                 }
1007
1008                 /*
1009                  * Remove this muxer upstream notification iterator
1010                  * if it's ended or canceled.
1011                  */
1012                 if (!muxer_upstream_notif_iter->notif_iter) {
1013                         /*
1014                          * Use g_ptr_array_remove_fast() because the
1015                          * order of those elements is not important.
1016                          */
1017                         BT_LOGV("Removing muxer's upstream notification iterator wrapper: ended or canceled: "
1018                                 "muxer-notif-iter-addr=%p, "
1019                                 "muxer-upstream-notif-iter-wrap-addr=%p",
1020                                 muxer_notif_iter, muxer_upstream_notif_iter);
1021                         g_ptr_array_remove_index_fast(
1022                                 muxer_notif_iter->muxer_upstream_notif_iters,
1023                                 i);
1024                         i--;
1025                 }
1026         }
1027
1028 end:
1029         return status;
1030 }
1031
1032 static inline
1033 enum bt_notification_iterator_status muxer_notif_iter_do_next_one(
1034                 struct muxer_comp *muxer_comp,
1035                 struct muxer_notif_iter *muxer_notif_iter,
1036                 struct bt_notification **notif)
1037 {
1038         enum bt_notification_iterator_status status =
1039                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
1040         struct muxer_upstream_notif_iter *muxer_upstream_notif_iter = NULL;
1041         int64_t next_return_ts;
1042
1043         while (true) {
1044                 int ret = muxer_notif_iter_handle_newly_connected_ports(
1045                         muxer_notif_iter);
1046
1047                 if (ret) {
1048                         BT_LOGE("Cannot handle newly connected input ports for muxer's notification iterator: "
1049                                 "muxer-comp-addr=%p, muxer-notif-iter-addr=%p, "
1050                                 "ret=%d",
1051                                 muxer_comp, muxer_notif_iter, ret);
1052                         status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1053                         goto end;
1054                 }
1055
1056                 status = validate_muxer_upstream_notif_iters(muxer_notif_iter);
1057                 if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
1058                         /* validate_muxer_upstream_notif_iters() logs details */
1059                         goto end;
1060                 }
1061
1062                 /*
1063                  * At this point, we know that all the existing upstream
1064                  * notification iterators are valid. However the
1065                  * operations to validate them (during
1066                  * validate_muxer_upstream_notif_iters()) may have
1067                  * connected new ports. If no ports were connected
1068                  * during this operation, exit the loop.
1069                  */
1070                 if (!muxer_notif_iter->newly_connected_priv_ports) {
1071                         BT_LOGV("Not breaking this loop: muxer's notification iterator still has newly connected input ports to handle: "
1072                                 "muxer-comp-addr=%p", muxer_comp);
1073                         break;
1074                 }
1075         }
1076
1077         BT_ASSERT(!muxer_notif_iter->newly_connected_priv_ports);
1078
1079         /*
1080          * At this point we know that all the existing upstream
1081          * notification iterators are valid. We can find the one,
1082          * amongst those, of which the current notification is the
1083          * youngest.
1084          */
1085         status = muxer_notif_iter_youngest_upstream_notif_iter(muxer_comp,
1086                         muxer_notif_iter, &muxer_upstream_notif_iter,
1087                         &next_return_ts);
1088         if (status < 0 || status == BT_NOTIFICATION_ITERATOR_STATUS_END ||
1089                         status == BT_NOTIFICATION_ITERATOR_STATUS_CANCELED) {
1090                 if (status < 0) {
1091                         BT_LOGE("Cannot find the youngest upstream notification iterator wrapper: "
1092                                 "status=%s",
1093                                 bt_notification_iterator_status_string(status));
1094                 } else {
1095                         BT_LOGV("Cannot find the youngest upstream notification iterator wrapper: "
1096                                 "status=%s",
1097                                 bt_notification_iterator_status_string(status));
1098                 }
1099
1100                 goto end;
1101         }
1102
1103         if (next_return_ts < muxer_notif_iter->last_returned_ts_ns) {
1104                 BT_LOGE("Youngest upstream notification iterator wrapper's timestamp is less than muxer's notification iterator's last returned timestamp: "
1105                         "muxer-notif-iter-addr=%p, ts=%" PRId64 ", "
1106                         "last-returned-ts=%" PRId64,
1107                         muxer_notif_iter, next_return_ts,
1108                         muxer_notif_iter->last_returned_ts_ns);
1109                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1110                 goto end;
1111         }
1112
1113         BT_LOGV("Found youngest upstream notification iterator wrapper: "
1114                 "muxer-notif-iter-addr=%p, "
1115                 "muxer-upstream-notif-iter-wrap-addr=%p, "
1116                 "ts=%" PRId64,
1117                 muxer_notif_iter, muxer_upstream_notif_iter, next_return_ts);
1118         BT_ASSERT(status == BT_NOTIFICATION_ITERATOR_STATUS_OK);
1119         BT_ASSERT(muxer_upstream_notif_iter);
1120
1121         /*
1122          * Consume from the queue's head: other side
1123          * (muxer_upstream_notif_iter_next()) writes to the tail.
1124          */
1125         *notif = g_queue_pop_head(muxer_upstream_notif_iter->notifs);
1126         BT_ASSERT(*notif);
1127         muxer_notif_iter->last_returned_ts_ns = next_return_ts;
1128
1129 end:
1130         return status;
1131 }
1132
1133 static
1134 enum bt_notification_iterator_status muxer_notif_iter_do_next(
1135                 struct muxer_comp *muxer_comp,
1136                 struct muxer_notif_iter *muxer_notif_iter,
1137                 bt_notification_array notifs, uint64_t capacity,
1138                 uint64_t *count)
1139 {
1140         enum bt_notification_iterator_status status =
1141                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
1142         uint64_t i = 0;
1143
1144         while (i < capacity && status == BT_NOTIFICATION_ITERATOR_STATUS_OK) {
1145                 status = muxer_notif_iter_do_next_one(muxer_comp,
1146                         muxer_notif_iter, &notifs[i]);
1147                 if (status == BT_NOTIFICATION_ITERATOR_STATUS_OK) {
1148                         i++;
1149                 }
1150         }
1151
1152         if (i > 0) {
1153                 /*
1154                  * Even if muxer_notif_iter_do_next_one() returned
1155                  * something else than
1156                  * BT_NOTIFICATION_ITERATOR_STATUS_OK, we accumulated
1157                  * notification objects in the output notification
1158                  * array, so we need to return
1159                  * BT_NOTIFICATION_ITERATOR_STATUS_OK so that they are
1160                  * transfered to downstream. This other status occurs
1161                  * again the next time muxer_notif_iter_do_next() is
1162                  * called, possibly without any accumulated
1163                  * notification, in which case we'll return it.
1164                  */
1165                 *count = i;
1166                 status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
1167         }
1168
1169         return status;
1170 }
1171
1172 static
1173 void destroy_muxer_notif_iter(struct muxer_notif_iter *muxer_notif_iter)
1174 {
1175         if (!muxer_notif_iter) {
1176                 return;
1177         }
1178
1179         BT_LOGD("Destroying muxer component's notification iterator: "
1180                 "muxer-notif-iter-addr=%p", muxer_notif_iter);
1181
1182         if (muxer_notif_iter->muxer_upstream_notif_iters) {
1183                 BT_LOGD_STR("Destroying muxer's upstream notification iterator wrappers.");
1184                 g_ptr_array_free(
1185                         muxer_notif_iter->muxer_upstream_notif_iters, TRUE);
1186         }
1187
1188         g_list_free(muxer_notif_iter->newly_connected_priv_ports);
1189         g_free(muxer_notif_iter);
1190 }
1191
1192 static
1193 int muxer_notif_iter_init_newly_connected_ports(struct muxer_comp *muxer_comp,
1194                 struct muxer_notif_iter *muxer_notif_iter)
1195 {
1196         struct bt_component *comp;
1197         int64_t count;
1198         int64_t i;
1199         int ret = 0;
1200
1201         /*
1202          * Add the connected input ports to this muxer notification
1203          * iterator's list of newly connected ports. They will be
1204          * handled by muxer_notif_iter_handle_newly_connected_ports().
1205          */
1206         comp = bt_component_borrow_from_private(muxer_comp->priv_comp);
1207         BT_ASSERT(comp);
1208         count = bt_component_filter_get_input_port_count(comp);
1209         if (count < 0) {
1210                 BT_LOGD("No input port to initialize for muxer component's notification iterator: "
1211                         "muxer-comp-addr=%p, muxer-notif-iter-addr=%p",
1212                         muxer_comp, muxer_notif_iter);
1213                 goto end;
1214         }
1215
1216         for (i = 0; i < count; i++) {
1217                 struct bt_private_port *priv_port =
1218                         bt_private_component_filter_get_input_private_port_by_index(
1219                                 muxer_comp->priv_comp, i);
1220                 struct bt_port *port;
1221
1222                 BT_ASSERT(priv_port);
1223                 port = bt_port_borrow_from_private(priv_port);
1224                 BT_ASSERT(port);
1225
1226                 if (!bt_port_is_connected(port)) {
1227                         BT_LOGD("Skipping input port: not connected: "
1228                                 "muxer-comp-addr=%p, port-addr=%p, port-name\"%s\"",
1229                                 muxer_comp, port, bt_port_get_name(port));
1230                         bt_put(priv_port);
1231                         continue;
1232                 }
1233
1234                 bt_put(priv_port);
1235                 muxer_notif_iter->newly_connected_priv_ports =
1236                         g_list_append(
1237                                 muxer_notif_iter->newly_connected_priv_ports,
1238                                 priv_port);
1239                 if (!muxer_notif_iter->newly_connected_priv_ports) {
1240                         BT_LOGE("Cannot append port to muxer's notification iterator list of newly connected input ports: "
1241                                 "port-addr=%p, port-name=\"%s\", "
1242                                 "muxer-notif-iter-addr=%p", port,
1243                                 bt_port_get_name(port), muxer_notif_iter);
1244                         ret = -1;
1245                         goto end;
1246                 }
1247
1248                 BT_LOGD("Appended port to muxer's notification iterator list of newly connected input ports: "
1249                         "port-addr=%p, port-name=\"%s\", "
1250                         "muxer-notif-iter-addr=%p", port,
1251                         bt_port_get_name(port), muxer_notif_iter);
1252         }
1253
1254 end:
1255         return ret;
1256 }
1257
1258 BT_HIDDEN
1259 enum bt_notification_iterator_status muxer_notif_iter_init(
1260                 struct bt_private_connection_private_notification_iterator *priv_notif_iter,
1261                 struct bt_private_port *output_priv_port)
1262 {
1263         struct muxer_comp *muxer_comp = NULL;
1264         struct muxer_notif_iter *muxer_notif_iter = NULL;
1265         struct bt_private_component *priv_comp = NULL;
1266         enum bt_notification_iterator_status status =
1267                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
1268         int ret;
1269
1270         priv_comp = bt_private_connection_private_notification_iterator_get_private_component(
1271                 priv_notif_iter);
1272         BT_ASSERT(priv_comp);
1273         muxer_comp = bt_private_component_get_user_data(priv_comp);
1274         BT_ASSERT(muxer_comp);
1275         BT_LOGD("Initializing muxer component's notification iterator: "
1276                 "comp-addr=%p, muxer-comp-addr=%p, notif-iter-addr=%p",
1277                 priv_comp, muxer_comp, priv_notif_iter);
1278
1279         if (muxer_comp->initializing_muxer_notif_iter) {
1280                 /*
1281                  * Weird, unhandled situation detected: downstream
1282                  * creates a muxer notification iterator while creating
1283                  * another muxer notification iterator (same component).
1284                  */
1285                 BT_LOGE("Recursive initialization of muxer component's notification iterator: "
1286                         "comp-addr=%p, muxer-comp-addr=%p, notif-iter-addr=%p",
1287                         priv_comp, muxer_comp, priv_notif_iter);
1288                 goto error;
1289         }
1290
1291         muxer_comp->initializing_muxer_notif_iter = true;
1292         muxer_notif_iter = g_new0(struct muxer_notif_iter, 1);
1293         if (!muxer_notif_iter) {
1294                 BT_LOGE_STR("Failed to allocate one muxer component's notification iterator.");
1295                 goto error;
1296         }
1297
1298         muxer_notif_iter->last_returned_ts_ns = INT64_MIN;
1299         muxer_notif_iter->muxer_upstream_notif_iters =
1300                 g_ptr_array_new_with_free_func(
1301                         (GDestroyNotify) destroy_muxer_upstream_notif_iter);
1302         if (!muxer_notif_iter->muxer_upstream_notif_iters) {
1303                 BT_LOGE_STR("Failed to allocate a GPtrArray.");
1304                 goto error;
1305         }
1306
1307         /*
1308          * Add the muxer notification iterator to the component's array
1309          * of muxer notification iterators here because
1310          * muxer_notif_iter_init_newly_connected_ports() can cause
1311          * muxer_port_connected() to be called, which adds the newly
1312          * connected port to each muxer notification iterator's list of
1313          * newly connected ports.
1314          */
1315         g_ptr_array_add(muxer_comp->muxer_notif_iters, muxer_notif_iter);
1316         ret = muxer_notif_iter_init_newly_connected_ports(muxer_comp,
1317                 muxer_notif_iter);
1318         if (ret) {
1319                 BT_LOGE("Cannot initialize newly connected input ports for muxer component's notification iterator: "
1320                         "comp-addr=%p, muxer-comp-addr=%p, "
1321                         "muxer-notif-iter-addr=%p, notif-iter-addr=%p, ret=%d",
1322                         priv_comp, muxer_comp, muxer_notif_iter,
1323                         priv_notif_iter, ret);
1324                 goto error;
1325         }
1326
1327         ret = bt_private_connection_private_notification_iterator_set_user_data(priv_notif_iter,
1328                 muxer_notif_iter);
1329         BT_ASSERT(ret == 0);
1330         BT_LOGD("Initialized muxer component's notification iterator: "
1331                 "comp-addr=%p, muxer-comp-addr=%p, muxer-notif-iter-addr=%p, "
1332                 "notif-iter-addr=%p",
1333                 priv_comp, muxer_comp, muxer_notif_iter, priv_notif_iter);
1334         goto end;
1335
1336 error:
1337         if (g_ptr_array_index(muxer_comp->muxer_notif_iters,
1338                         muxer_comp->muxer_notif_iters->len - 1) == muxer_notif_iter) {
1339                 g_ptr_array_remove_index(muxer_comp->muxer_notif_iters,
1340                         muxer_comp->muxer_notif_iters->len - 1);
1341         }
1342
1343         destroy_muxer_notif_iter(muxer_notif_iter);
1344         ret = bt_private_connection_private_notification_iterator_set_user_data(priv_notif_iter,
1345                 NULL);
1346         BT_ASSERT(ret == 0);
1347         status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1348
1349 end:
1350         muxer_comp->initializing_muxer_notif_iter = false;
1351         bt_put(priv_comp);
1352         return status;
1353 }
1354
1355 BT_HIDDEN
1356 void muxer_notif_iter_finalize(
1357                 struct bt_private_connection_private_notification_iterator *priv_notif_iter)
1358 {
1359         struct muxer_notif_iter *muxer_notif_iter =
1360                 bt_private_connection_private_notification_iterator_get_user_data(priv_notif_iter);
1361         struct bt_private_component *priv_comp = NULL;
1362         struct muxer_comp *muxer_comp = NULL;
1363
1364         priv_comp = bt_private_connection_private_notification_iterator_get_private_component(
1365                 priv_notif_iter);
1366         BT_ASSERT(priv_comp);
1367         muxer_comp = bt_private_component_get_user_data(priv_comp);
1368         BT_LOGD("Finalizing muxer component's notification iterator: "
1369                 "comp-addr=%p, muxer-comp-addr=%p, muxer-notif-iter-addr=%p, "
1370                 "notif-iter-addr=%p",
1371                 priv_comp, muxer_comp, muxer_notif_iter, priv_notif_iter);
1372
1373         if (muxer_comp) {
1374                 (void) g_ptr_array_remove_fast(muxer_comp->muxer_notif_iters,
1375                         muxer_notif_iter);
1376                 destroy_muxer_notif_iter(muxer_notif_iter);
1377         }
1378
1379         bt_put(priv_comp);
1380 }
1381
1382 BT_HIDDEN
1383 enum bt_notification_iterator_status muxer_notif_iter_next(
1384                 struct bt_private_connection_private_notification_iterator *priv_notif_iter,
1385                 bt_notification_array notifs, uint64_t capacity,
1386                 uint64_t *count)
1387 {
1388         enum bt_notification_iterator_status status;
1389         struct muxer_notif_iter *muxer_notif_iter =
1390                 bt_private_connection_private_notification_iterator_get_user_data(priv_notif_iter);
1391         struct bt_private_component *priv_comp = NULL;
1392         struct muxer_comp *muxer_comp = NULL;
1393
1394         BT_ASSERT(muxer_notif_iter);
1395         priv_comp = bt_private_connection_private_notification_iterator_get_private_component(
1396                 priv_notif_iter);
1397         BT_ASSERT(priv_comp);
1398         muxer_comp = bt_private_component_get_user_data(priv_comp);
1399         BT_ASSERT(muxer_comp);
1400         BT_LOGV("Muxer component's notification iterator's \"next\" method called: "
1401                 "comp-addr=%p, muxer-comp-addr=%p, muxer-notif-iter-addr=%p, "
1402                 "notif-iter-addr=%p",
1403                 priv_comp, muxer_comp, muxer_notif_iter, priv_notif_iter);
1404
1405         status = muxer_notif_iter_do_next(muxer_comp, muxer_notif_iter,
1406                 notifs, capacity, count);
1407         if (status < 0) {
1408                 BT_LOGE("Cannot get next notification: "
1409                         "comp-addr=%p, muxer-comp-addr=%p, muxer-notif-iter-addr=%p, "
1410                         "notif-iter-addr=%p, status=%s",
1411                         priv_comp, muxer_comp, muxer_notif_iter, priv_notif_iter,
1412                         bt_notification_iterator_status_string(status));
1413         } else {
1414                 BT_LOGV("Returning from muxer component's notification iterator's \"next\" method: "
1415                         "status=%s",
1416                         bt_notification_iterator_status_string(status));
1417         }
1418
1419         bt_put(priv_comp);
1420         return status;
1421 }
1422
1423 BT_HIDDEN
1424 enum bt_component_status muxer_port_connected(
1425                 struct bt_private_component *priv_comp,
1426                 struct bt_private_port *self_private_port,
1427                 struct bt_port *other_port)
1428 {
1429         enum bt_component_status status = BT_COMPONENT_STATUS_OK;
1430         struct bt_port *self_port =
1431                 bt_port_borrow_from_private(self_private_port);
1432         struct muxer_comp *muxer_comp =
1433                 bt_private_component_get_user_data(priv_comp);
1434         size_t i;
1435         int ret;
1436
1437         BT_ASSERT(self_port);
1438         BT_ASSERT(muxer_comp);
1439         BT_LOGD("Port connected: "
1440                 "comp-addr=%p, muxer-comp-addr=%p, "
1441                 "port-addr=%p, port-name=\"%s\", "
1442                 "other-port-addr=%p, other-port-name=\"%s\"",
1443                 priv_comp, muxer_comp, self_port, bt_port_get_name(self_port),
1444                 other_port, bt_port_get_name(other_port));
1445
1446         if (bt_port_get_type(self_port) == BT_PORT_TYPE_OUTPUT) {
1447                 goto end;
1448         }
1449
1450         for (i = 0; i < muxer_comp->muxer_notif_iters->len; i++) {
1451                 struct muxer_notif_iter *muxer_notif_iter =
1452                         g_ptr_array_index(muxer_comp->muxer_notif_iters, i);
1453
1454                 /*
1455                  * Add this port to the list of newly connected ports
1456                  * for this muxer notification iterator. We append at
1457                  * the end of this list while
1458                  * muxer_notif_iter_handle_newly_connected_ports()
1459                  * removes the nodes from the beginning.
1460                  */
1461                 muxer_notif_iter->newly_connected_priv_ports =
1462                         g_list_append(
1463                                 muxer_notif_iter->newly_connected_priv_ports,
1464                                 self_private_port);
1465                 if (!muxer_notif_iter->newly_connected_priv_ports) {
1466                         BT_LOGE("Cannot append port to muxer's notification iterator list of newly connected input ports: "
1467                                 "port-addr=%p, port-name=\"%s\", "
1468                                 "muxer-notif-iter-addr=%p", self_port,
1469                                 bt_port_get_name(self_port), muxer_notif_iter);
1470                         status = BT_COMPONENT_STATUS_ERROR;
1471                         goto end;
1472                 }
1473
1474                 BT_LOGD("Appended port to muxer's notification iterator list of newly connected input ports: "
1475                         "port-addr=%p, port-name=\"%s\", "
1476                         "muxer-notif-iter-addr=%p", self_port,
1477                         bt_port_get_name(self_port), muxer_notif_iter);
1478         }
1479
1480         /* One less available input port */
1481         muxer_comp->available_input_ports--;
1482         ret = ensure_available_input_port(priv_comp);
1483         if (ret) {
1484                 /*
1485                  * Only way to report an error later since this
1486                  * method does not return anything.
1487                  */
1488                 BT_LOGE("Cannot ensure that at least one muxer component's input port is available: "
1489                         "muxer-comp-addr=%p, status=%s",
1490                         muxer_comp, bt_component_status_string(ret));
1491                 status = BT_COMPONENT_STATUS_ERROR;
1492                 goto end;
1493         }
1494
1495 end:
1496         return status;
1497 }
1498
1499 BT_HIDDEN
1500 void muxer_port_disconnected(struct bt_private_component *priv_comp,
1501                 struct bt_private_port *priv_port)
1502 {
1503         struct bt_port *port = bt_port_borrow_from_private(priv_port);
1504         struct muxer_comp *muxer_comp =
1505                 bt_private_component_get_user_data(priv_comp);
1506
1507         BT_ASSERT(port);
1508         BT_ASSERT(muxer_comp);
1509         BT_LOGD("Port disconnected: "
1510                 "comp-addr=%p, muxer-comp-addr=%p, port-addr=%p, "
1511                 "port-name=\"%s\"", priv_comp, muxer_comp,
1512                 port, bt_port_get_name(port));
1513
1514         /*
1515          * There's nothing special to do when a port is disconnected
1516          * because this component deals with upstream notification
1517          * iterators which were already created thanks to connected
1518          * ports. The fact that the port is disconnected does not cancel
1519          * the upstream notification iterators created using its
1520          * connection: they still exist, even if the connection is dead.
1521          * The only way to remove an upstream notification iterator is
1522          * for its "next" operation to return
1523          * BT_NOTIFICATION_ITERATOR_STATUS_END.
1524          */
1525         if (bt_port_get_type(port) == BT_PORT_TYPE_INPUT) {
1526                 /* One more available input port */
1527                 muxer_comp->available_input_ports++;
1528                 BT_LOGD("Leaving disconnected input port available for future connections: "
1529                         "comp-addr=%p, muxer-comp-addr=%p, port-addr=%p, "
1530                         "port-name=\"%s\", avail-input-port-count=%zu",
1531                         priv_comp, muxer_comp, port, bt_port_get_name(port),
1532                         muxer_comp->available_input_ports);
1533         }
1534 }