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 #include <babeltrace/babeltrace-internal.h>
  24 #include <babeltrace/ctf-ir/clock-class.h>
  25 #include <babeltrace/ctf-ir/event.h>
  26 #include <babeltrace/graph/clock-class-priority-map.h>
  27 #include <babeltrace/graph/component-filter.h>
  28 #include <babeltrace/graph/component.h>
  29 #include <babeltrace/graph/notification-event.h>
  30 #include <babeltrace/graph/notification-inactivity.h>
  31 #include <babeltrace/graph/notification-iterator.h>
  32 #include <babeltrace/graph/notification.h>
  33 #include <babeltrace/graph/port.h>
  34 #include <babeltrace/graph/private-component-filter.h>
  35 #include <babeltrace/graph/private-component.h>
  36 #include <babeltrace/graph/private-component.h>
  37 #include <babeltrace/graph/private-connection.h>
  38 #include <babeltrace/graph/private-notification-iterator.h>
  39 #include <babeltrace/graph/private-port.h>
  40 #include <plugins-common.h>
  41 #include <glib.h>
  42 #include <stdbool.h>
  43 #include <assert.h>
  44 #include <stdlib.h>
  45
  46 #define ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME        "assume-absolute-clock-classes"
  47
  48 struct muxer_comp {
  49         /* Array of struct bt_private_notification_iterator * (weak refs) */
  50         GPtrArray *muxer_notif_iters;
  51
  52         /* Weak ref */
  53         struct bt_private_component *priv_comp;
  54         unsigned int next_port_num;
  55         size_t available_input_ports;
  56         bool error;
  57         bool initializing_muxer_notif_iter;
  58         bool ignore_absolute;
  59 };
  60
  61 struct muxer_upstream_notif_iter {
  62         /* Owned by this, NULL if ended */
  63         struct bt_notification_iterator *notif_iter;
  64
  65         /*
  66          * This flag is true if the upstream notification iterator's
  67          * current notification must be considered for the multiplexing
  68          * operations. If the upstream iterator returns
  69          * BT_NOTIFICATION_ITERATOR_STATUS_AGAIN, then this object
  70          * is considered invalid, because its current notification is
  71          * still the previous one, but we already took it into account.
  72          *
  73          * The value of this flag is not important if notif_iter above
  74          * is NULL (which means the upstream iterator is finished).
  75          */
  76         bool is_valid;
  77 };
  78
  79 struct muxer_notif_iter {
  80         /*
  81          * Array of struct muxer_upstream_notif_iter * (owned by this).
  82          *
  83          * NOTE: This array is searched in linearly to find the youngest
  84          * current notification. Keep this until benchmarks confirm that
  85          * another data structure is faster than this for our typical
  86          * use cases.
  87          */
  88         GPtrArray *muxer_upstream_notif_iters;
  89
  90         /*
  91          * List of "recently" connected input ports (weak) to
  92          * handle by this muxer notification iterator.
  93          * muxer_port_connected() adds entries to this list, and the
  94          * entries are removed when a notification iterator is created
  95          * on the port's connection and put into
  96          * muxer_upstream_notif_iters above by
  97          * muxer_notif_iter_handle_newly_connected_ports().
  98          */
  99         GList *newly_connected_priv_ports;
 100
 101         /* Next thing to return by the "next" method */
 102         struct bt_notification_iterator_next_return next_next_return;
 103
 104         /* Last time returned in a notification */
 105         int64_t last_returned_ts_ns;
 106 };
 107
 108 static
 109 void destroy_muxer_upstream_notif_iter(
 110                 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
 111 {
 112         if (!muxer_upstream_notif_iter) {
 113                 return;
 114         }
 115
 116         bt_put(muxer_upstream_notif_iter->notif_iter);
 117         g_free(muxer_upstream_notif_iter);
 118 }
 119
 120 static
 121 struct muxer_upstream_notif_iter *muxer_notif_iter_add_upstream_notif_iter(
 122                 struct muxer_notif_iter *muxer_notif_iter,
 123                 struct bt_notification_iterator *notif_iter,
 124                 struct bt_private_port *priv_port)
 125 {
 126         struct muxer_upstream_notif_iter *muxer_upstream_notif_iter =
 127                 g_new0(struct muxer_upstream_notif_iter, 1);
 128
 129         if (!muxer_upstream_notif_iter) {
 130                 goto end;
 131         }
 132
 133         muxer_upstream_notif_iter->notif_iter = bt_get(notif_iter);
 134         muxer_upstream_notif_iter->is_valid = false;
 135         g_ptr_array_add(muxer_notif_iter->muxer_upstream_notif_iters,
 136                 muxer_upstream_notif_iter);
 137
 138 end:
 139         return muxer_upstream_notif_iter;
 140 }
 141
 142 static
 143 enum bt_component_status ensure_available_input_port(
 144                 struct bt_private_component *priv_comp)
 145 {
 146         struct muxer_comp *muxer_comp =
 147                 bt_private_component_get_user_data(priv_comp);
 148         enum bt_component_status status = BT_COMPONENT_STATUS_OK;
 149         GString *port_name = NULL;
 150
 151         assert(muxer_comp);
 152
 153         if (muxer_comp->available_input_ports >= 1) {
 154                 goto end;
 155         }
 156
 157         port_name = g_string_new("in");
 158         if (!port_name) {
 159                 status = BT_COMPONENT_STATUS_NOMEM;
 160                 goto end;
 161         }
 162
 163         g_string_append_printf(port_name, "%u", muxer_comp->next_port_num);
 164         status = bt_private_component_filter_add_input_private_port(
 165                 priv_comp, port_name->str, NULL, NULL);
 166         if (status != BT_COMPONENT_STATUS_OK) {
 167                 goto end;
 168         }
 169
 170         muxer_comp->available_input_ports++;
 171         muxer_comp->next_port_num++;
 172
 173 end:
 174         if (port_name) {
 175                 g_string_free(port_name, TRUE);
 176         }
 177
 178         return status;
 179 }
 180
 181 static
 182 enum bt_component_status create_output_port(
 183                 struct bt_private_component *priv_comp)
 184 {
 185         return bt_private_component_filter_add_output_private_port(
 186                 priv_comp, "out", NULL, NULL);
 187 }
 188
 189 static
 190 void destroy_muxer_comp(struct muxer_comp *muxer_comp)
 191 {
 192         if (!muxer_comp) {
 193                 return;
 194         }
 195
 196         if (muxer_comp->muxer_notif_iters) {
 197                 g_ptr_array_free(muxer_comp->muxer_notif_iters, TRUE);
 198         }
 199
 200         g_free(muxer_comp);
 201 }
 202
 203 static
 204 struct bt_value *get_default_params(void)
 205 {
 206         struct bt_value *params;
 207         int ret;
 208
 209         params = bt_value_map_create();
 210         if (!params) {
 211                 goto error;
 212         }
 213
 214         ret = bt_value_map_insert_bool(params,
 215                 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME, false);
 216         if (ret) {
 217                 goto error;
 218         }
 219
 220         goto end;
 221
 222 error:
 223         BT_PUT(params);
 224
 225 end:
 226         return params;
 227 }
 228
 229 static
 230 int configure_muxer_comp(struct muxer_comp *muxer_comp, struct bt_value *params)
 231 {
 232         struct bt_value *default_params = NULL;
 233         struct bt_value *real_params = NULL;
 234         struct bt_value *ignore_absolute = NULL;
 235         int ret = 0;
 236         bt_bool bool_val;
 237
 238         default_params = get_default_params();
 239         if (!default_params) {
 240                 goto error;
 241         }
 242
 243         real_params = bt_value_map_extend(default_params, params);
 244         if (!real_params) {
 245                 goto error;
 246         }
 247
 248         ignore_absolute = bt_value_map_get(real_params,
 249                 ASSUME_ABSOLUTE_CLOCK_CLASSES_PARAM_NAME);
 250         if (!bt_value_is_bool(ignore_absolute)) {
 251                 goto error;
 252         }
 253
 254         if (bt_value_bool_get(ignore_absolute, &bool_val)) {
 255                 goto error;
 256         }
 257
 258         muxer_comp->ignore_absolute = (bool) bool_val;
 259
 260         goto end;
 261
 262 error:
 263         ret = -1;
 264
 265 end:
 266         bt_put(default_params);
 267         bt_put(real_params);
 268         bt_put(ignore_absolute);
 269         return ret;
 270 }
 271
 272 BT_HIDDEN
 273 enum bt_component_status muxer_init(
 274                 struct bt_private_component *priv_comp,
 275                 struct bt_value *params, void *init_data)
 276 {
 277         int ret;
 278         enum bt_component_status status = BT_COMPONENT_STATUS_OK;
 279         struct muxer_comp *muxer_comp = g_new0(struct muxer_comp, 1);
 280
 281         if (!muxer_comp) {
 282                 goto error;
 283         }
 284
 285         ret = configure_muxer_comp(muxer_comp, params);
 286         if (ret) {
 287                 goto error;
 288         }
 289
 290         muxer_comp->muxer_notif_iters = g_ptr_array_new();
 291         if (!muxer_comp->muxer_notif_iters) {
 292                 goto error;
 293         }
 294
 295         muxer_comp->priv_comp = priv_comp;
 296         ret = bt_private_component_set_user_data(priv_comp, muxer_comp);
 297         assert(ret == 0);
 298         status = ensure_available_input_port(priv_comp);
 299         if (status != BT_COMPONENT_STATUS_OK) {
 300                 goto error;
 301         }
 302
 303         ret = create_output_port(priv_comp);
 304         if (ret) {
 305                 goto error;
 306         }
 307
 308         goto end;
 309
 310 error:
 311         destroy_muxer_comp(muxer_comp);
 312         ret = bt_private_component_set_user_data(priv_comp, NULL);
 313         assert(ret == 0);
 314
 315         if (status == BT_COMPONENT_STATUS_OK) {
 316                 status = BT_COMPONENT_STATUS_ERROR;
 317         }
 318
 319 end:
 320         return status;
 321 }
 322
 323 BT_HIDDEN
 324 void muxer_finalize(struct bt_private_component *priv_comp)
 325 {
 326         struct muxer_comp *muxer_comp =
 327                 bt_private_component_get_user_data(priv_comp);
 328
 329         destroy_muxer_comp(muxer_comp);
 330 }
 331
 332 static
 333 struct bt_notification_iterator *create_notif_iter_on_input_port(
 334                 struct bt_private_port *priv_port, int *ret)
 335 {
 336         struct bt_port *port = bt_port_from_private_port(priv_port);
 337         struct bt_notification_iterator *notif_iter = NULL;
 338         struct bt_private_connection *priv_conn = NULL;
 339
 340         assert(ret);
 341         *ret = 0;
 342         assert(port);
 343
 344         assert(bt_port_is_connected(port));
 345         priv_conn = bt_private_port_get_private_connection(priv_port);
 346         if (!priv_conn) {
 347                 *ret = -1;
 348                 goto end;
 349         }
 350
 351         // TODO: Advance the iterator to >= the time of the latest
 352         //       returned notification by the muxer notification
 353         //       iterator which creates it.
 354         notif_iter = bt_private_connection_create_notification_iterator(
 355                 priv_conn, NULL);
 356         if (!notif_iter) {
 357                 *ret = -1;
 358                 goto end;
 359         }
 360
 361 end:
 362         bt_put(port);
 363         bt_put(priv_conn);
 364         return notif_iter;
 365 }
 366
 367 static
 368 enum bt_notification_iterator_status muxer_upstream_notif_iter_next(
 369                 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
 370 {
 371         enum bt_notification_iterator_status status;
 372
 373         status = bt_notification_iterator_next(
 374                 muxer_upstream_notif_iter->notif_iter);
 375
 376         switch (status) {
 377         case BT_NOTIFICATION_ITERATOR_STATUS_OK:
 378                 /*
 379                  * Notification iterator's current notification is valid:
 380                  * it must be considered for muxing operations.
 381                  */
 382                 muxer_upstream_notif_iter->is_valid = true;
 383                 break;
 384         case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
 385                 /*
 386                  * Notification iterator's current notification is not
 387                  * valid anymore. Return
 388                  * BT_NOTIFICATION_ITERATOR_STATUS_AGAIN
 389                  * immediately.
 390                  */
 391                 muxer_upstream_notif_iter->is_valid = false;
 392                 break;
 393         case BT_NOTIFICATION_ITERATOR_STATUS_END:       /* Fall-through. */
 394         case BT_NOTIFICATION_ITERATOR_STATUS_CANCELED:
 395                 /*
 396                  * Notification iterator reached the end: release it. It
 397                  * won't be considered again to find the youngest
 398                  * notification.
 399                  */
 400                 BT_PUT(muxer_upstream_notif_iter->notif_iter);
 401                 muxer_upstream_notif_iter->is_valid = false;
 402                 status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
 403                 break;
 404         default:
 405                 /* Error or unsupported status code */
 406                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 407                 break;
 408         }
 409
 410         return status;
 411 }
 412
 413 static
 414 int muxer_notif_iter_handle_newly_connected_ports(
 415                 struct muxer_notif_iter *muxer_notif_iter)
 416 {
 417         int ret = 0;
 418
 419         /*
 420          * Here we create one upstream notification iterator for each
 421          * newly connected port. We do not perform an initial "next" on
 422          * those new upstream notification iterators: they are
 423          * invalidated, to be validated later. The list of newly
 424          * connected ports to handle here is updated by
 425          * muxer_port_connected().
 426          */
 427         while (true) {
 428                 GList *node = muxer_notif_iter->newly_connected_priv_ports;
 429                 struct bt_private_port *priv_port;
 430                 struct bt_port *port;
 431                 struct bt_notification_iterator *upstream_notif_iter = NULL;
 432                 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter;
 433
 434                 if (!node) {
 435                         break;
 436                 }
 437
 438                 priv_port = node->data;
 439                 port = bt_port_from_private_port(priv_port);
 440                 assert(port);
 441
 442                 if (!bt_port_is_connected(port)) {
 443                         /*
 444                          * Looks like this port is not connected
 445                          * anymore: we can't create an upstream
 446                          * notification iterator on its (non-existing)
 447                          * connection in this case.
 448                          */
 449                         goto remove_node;
 450                 }
 451
 452                 BT_PUT(port);
 453                 upstream_notif_iter = create_notif_iter_on_input_port(priv_port,
 454                         &ret);
 455                 if (ret) {
 456                         assert(!upstream_notif_iter);
 457                         goto error;
 458                 }
 459
 460                 muxer_upstream_notif_iter =
 461                         muxer_notif_iter_add_upstream_notif_iter(
 462                                 muxer_notif_iter, upstream_notif_iter,
 463                                 priv_port);
 464                 BT_PUT(upstream_notif_iter);
 465                 if (!muxer_upstream_notif_iter) {
 466                         goto error;
 467                 }
 468
 469 remove_node:
 470                 bt_put(upstream_notif_iter);
 471                 bt_put(port);
 472                 muxer_notif_iter->newly_connected_priv_ports =
 473                         g_list_delete_link(
 474                                 muxer_notif_iter->newly_connected_priv_ports,
 475                                 node);
 476         }
 477
 478         goto end;
 479
 480 error:
 481         if (ret >= 0) {
 482                 ret = -1;
 483         }
 484
 485 end:
 486         return ret;
 487 }
 488
 489 static
 490 int get_notif_ts_ns(struct muxer_comp *muxer_comp,
 491                 struct bt_notification *notif, int64_t last_returned_ts_ns,
 492                 int64_t *ts_ns)
 493 {
 494         struct bt_clock_class_priority_map *cc_prio_map = NULL;
 495         struct bt_ctf_clock_class *clock_class = NULL;
 496         struct bt_ctf_clock_value *clock_value = NULL;
 497         struct bt_ctf_event *event = NULL;
 498         int ret = 0;
 499
 500         assert(notif);
 501         assert(ts_ns);
 502
 503         switch (bt_notification_get_type(notif)) {
 504         case BT_NOTIFICATION_TYPE_EVENT:
 505                 cc_prio_map =
 506                         bt_notification_event_get_clock_class_priority_map(
 507                                 notif);
 508                 break;
 509
 510         case BT_NOTIFICATION_TYPE_INACTIVITY:
 511                 cc_prio_map =
 512                         bt_notification_inactivity_get_clock_class_priority_map(
 513                                 notif);
 514                 break;
 515         default:
 516                 /* All the other notifications have a higher priority */
 517                 *ts_ns = last_returned_ts_ns;
 518                 goto end;
 519         }
 520
 521         if (!cc_prio_map) {
 522                 goto error;
 523         }
 524
 525         /*
 526          * If the clock class priority map is empty, then we consider
 527          * that this notification has no time. In this case it's always
 528          * the youngest.
 529          */
 530         if (bt_clock_class_priority_map_get_clock_class_count(cc_prio_map) == 0) {
 531                 *ts_ns = last_returned_ts_ns;
 532                 goto end;
 533         }
 534
 535         clock_class =
 536                 bt_clock_class_priority_map_get_highest_priority_clock_class(
 537                         cc_prio_map);
 538         if (!clock_class) {
 539                 goto error;
 540         }
 541
 542         if (!muxer_comp->ignore_absolute &&
 543                         !bt_ctf_clock_class_is_absolute(clock_class)) {
 544                 goto error;
 545         }
 546
 547         switch (bt_notification_get_type(notif)) {
 548         case BT_NOTIFICATION_TYPE_EVENT:
 549                 event = bt_notification_event_get_event(notif);
 550                 assert(event);
 551                 clock_value = bt_ctf_event_get_clock_value(event,
 552                         clock_class);
 553                 break;
 554         case BT_NOTIFICATION_TYPE_INACTIVITY:
 555                 clock_value = bt_notification_inactivity_get_clock_value(
 556                         notif, clock_class);
 557                 break;
 558         default:
 559                 abort();
 560         }
 561
 562         if (!clock_value) {
 563                 goto error;
 564         }
 565
 566         ret = bt_ctf_clock_value_get_value_ns_from_epoch(clock_value, ts_ns);
 567         if (ret) {
 568                 goto error;
 569         }
 570
 571         goto end;
 572
 573 error:
 574         ret = -1;
 575
 576 end:
 577         bt_put(cc_prio_map);
 578         bt_put(event);
 579         bt_put(clock_class);
 580         bt_put(clock_value);
 581         return ret;
 582 }
 583
 584 /*
 585  * This function finds the youngest available notification amongst the
 586  * non-ended upstream notification iterators and returns the upstream
 587  * notification iterator which has it, or
 588  * BT_NOTIFICATION_ITERATOR_STATUS_END if there's no available
 589  * notification.
 590  *
 591  * This function does NOT:
 592  *
 593  * * Update any upstream notification iterator.
 594  * * Check for newly connected ports.
 595  * * Check the upstream notification iterators to retry.
 596  *
 597  * On sucess, this function sets *muxer_upstream_notif_iter to the
 598  * upstream notification iterator of which the current notification is
 599  * the youngest, and sets *ts_ns to its time.
 600  */
 601 static
 602 enum bt_notification_iterator_status
 603 muxer_notif_iter_youngest_upstream_notif_iter(
 604                 struct muxer_comp *muxer_comp,
 605                 struct muxer_notif_iter *muxer_notif_iter,
 606                 struct muxer_upstream_notif_iter **muxer_upstream_notif_iter,
 607                 int64_t *ts_ns)
 608 {
 609         size_t i;
 610         int ret;
 611         int64_t youngest_ts_ns = INT64_MAX;
 612         enum bt_notification_iterator_status status =
 613                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
 614
 615         assert(muxer_comp);
 616         assert(muxer_notif_iter);
 617         assert(muxer_upstream_notif_iter);
 618         *muxer_upstream_notif_iter = NULL;
 619
 620         for (i = 0; i < muxer_notif_iter->muxer_upstream_notif_iters->len; i++) {
 621                 struct bt_notification *notif;
 622                 struct muxer_upstream_notif_iter *cur_muxer_upstream_notif_iter =
 623                         g_ptr_array_index(muxer_notif_iter->muxer_upstream_notif_iters, i);
 624                 int64_t notif_ts_ns;
 625
 626                 if (!cur_muxer_upstream_notif_iter->notif_iter) {
 627                         /* This upstream notification iterator is ended */
 628                         continue;
 629                 }
 630
 631                 assert(cur_muxer_upstream_notif_iter->is_valid);
 632                 notif = bt_notification_iterator_get_notification(
 633                         cur_muxer_upstream_notif_iter->notif_iter);
 634                 assert(notif);
 635                 ret = get_notif_ts_ns(muxer_comp, notif,
 636                         muxer_notif_iter->last_returned_ts_ns, &notif_ts_ns);
 637                 bt_put(notif);
 638                 if (ret) {
 639                         *muxer_upstream_notif_iter = NULL;
 640                         status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 641                         goto end;
 642                 }
 643
 644                 if (notif_ts_ns <= youngest_ts_ns) {
 645                         *muxer_upstream_notif_iter =
 646                                 cur_muxer_upstream_notif_iter;
 647                         youngest_ts_ns = notif_ts_ns;
 648                         *ts_ns = youngest_ts_ns;
 649                 }
 650         }
 651
 652         if (!*muxer_upstream_notif_iter) {
 653                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
 654                 *ts_ns = INT64_MIN;
 655         }
 656
 657 end:
 658         return status;
 659 }
 660
 661 static
 662 enum bt_notification_iterator_status validate_muxer_upstream_notif_iter(
 663         struct muxer_upstream_notif_iter *muxer_upstream_notif_iter)
 664 {
 665         enum bt_notification_iterator_status status =
 666                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
 667
 668         if (muxer_upstream_notif_iter->is_valid ||
 669                         !muxer_upstream_notif_iter->notif_iter) {
 670                 goto end;
 671         }
 672
 673         status = muxer_upstream_notif_iter_next(muxer_upstream_notif_iter);
 674
 675 end:
 676         return status;
 677 }
 678
 679 static
 680 enum bt_notification_iterator_status validate_muxer_upstream_notif_iters(
 681         struct muxer_notif_iter *muxer_notif_iter)
 682 {
 683         enum bt_notification_iterator_status status =
 684                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
 685         size_t i;
 686
 687         for (i = 0; i < muxer_notif_iter->muxer_upstream_notif_iters->len; i++) {
 688                 struct muxer_upstream_notif_iter *muxer_upstream_notif_iter =
 689                         g_ptr_array_index(
 690                                 muxer_notif_iter->muxer_upstream_notif_iters,
 691                                 i);
 692
 693                 status = validate_muxer_upstream_notif_iter(
 694                         muxer_upstream_notif_iter);
 695                 if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
 696                         goto end;
 697                 }
 698
 699                 /*
 700                  * Remove this muxer upstream notification iterator
 701                  * if it's ended or canceled.
 702                  */
 703                 if (!muxer_upstream_notif_iter->notif_iter) {
 704                         /*
 705                          * Use g_ptr_array_remove_fast() because the
 706                          * order of those elements is not important.
 707                          */
 708                         g_ptr_array_remove_index_fast(
 709                                 muxer_notif_iter->muxer_upstream_notif_iters,
 710                                 i);
 711                         i--;
 712                 }
 713         }
 714
 715 end:
 716         return status;
 717 }
 718
 719 static
 720 struct bt_notification_iterator_next_return muxer_notif_iter_do_next(
 721                 struct muxer_comp *muxer_comp,
 722                 struct muxer_notif_iter *muxer_notif_iter)
 723 {
 724         struct muxer_upstream_notif_iter *muxer_upstream_notif_iter = NULL;
 725         struct bt_notification_iterator_next_return next_return = {
 726                 .notification = NULL,
 727                 .status = BT_NOTIFICATION_ITERATOR_STATUS_OK,
 728         };
 729         int64_t next_return_ts;
 730
 731         while (true) {
 732                 int ret = muxer_notif_iter_handle_newly_connected_ports(
 733                         muxer_notif_iter);
 734
 735                 if (ret) {
 736                         next_return.status =
 737                                 BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 738                         goto end;
 739                 }
 740
 741                 next_return.status =
 742                         validate_muxer_upstream_notif_iters(muxer_notif_iter);
 743                 if (next_return.status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
 744                         goto end;
 745                 }
 746
 747                 /*
 748                  * At this point, we know that all the existing upstream
 749                  * notification iterators are valid. However the
 750                  * operations to validate them (during
 751                  * validate_muxer_upstream_notif_iters()) may have
 752                  * connected new ports. If no ports were connected
 753                  * during this operation, exit the loop.
 754                  */
 755                 if (!muxer_notif_iter->newly_connected_priv_ports) {
 756                         break;
 757                 }
 758         }
 759
 760         assert(!muxer_notif_iter->newly_connected_priv_ports);
 761
 762         /*
 763          * At this point we know that all the existing upstream
 764          * notification iterators are valid. We can find the one,
 765          * amongst those, of which the current notification is the
 766          * youngest.
 767          */
 768         next_return.status =
 769                 muxer_notif_iter_youngest_upstream_notif_iter(muxer_comp,
 770                         muxer_notif_iter, &muxer_upstream_notif_iter,
 771                         &next_return_ts);
 772         if (next_return.status < 0 ||
 773                         next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_END ||
 774                         next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_CANCELED) {
 775                 goto end;
 776         }
 777
 778         if (next_return_ts < muxer_notif_iter->last_returned_ts_ns) {
 779                 next_return.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 780                 goto end;
 781         }
 782
 783         assert(next_return.status == BT_NOTIFICATION_ITERATOR_STATUS_OK);
 784         assert(muxer_upstream_notif_iter);
 785         next_return.notification = bt_notification_iterator_get_notification(
 786                 muxer_upstream_notif_iter->notif_iter);
 787         assert(next_return.notification);
 788
 789         /*
 790          * We invalidate the upstream notification iterator so that, the
 791          * next time this function is called,
 792          * validate_muxer_upstream_notif_iters() will make it valid.
 793          */
 794         muxer_upstream_notif_iter->is_valid = false;
 795         muxer_notif_iter->last_returned_ts_ns = next_return_ts;
 796
 797 end:
 798         return next_return;
 799 }
 800
 801 static
 802 void destroy_muxer_notif_iter(struct muxer_notif_iter *muxer_notif_iter)
 803 {
 804         if (!muxer_notif_iter) {
 805                 return;
 806         }
 807
 808         if (muxer_notif_iter->muxer_upstream_notif_iters) {
 809                 g_ptr_array_free(
 810                         muxer_notif_iter->muxer_upstream_notif_iters, TRUE);
 811         }
 812
 813         g_list_free(muxer_notif_iter->newly_connected_priv_ports);
 814         g_free(muxer_notif_iter);
 815 }
 816
 817 static
 818 int muxer_notif_iter_init_newly_connected_ports(struct muxer_comp *muxer_comp,
 819                 struct muxer_notif_iter *muxer_notif_iter)
 820 {
 821         struct bt_component *comp;
 822         int64_t count;
 823         int64_t i;
 824         int ret = 0;
 825
 826         /*
 827          * Add the connected input ports to this muxer notification
 828          * iterator's list of newly connected ports. They will be
 829          * handled by muxer_notif_iter_handle_newly_connected_ports().
 830          */
 831         comp = bt_component_from_private_component(muxer_comp->priv_comp);
 832         assert(comp);
 833         count = bt_component_filter_get_input_port_count(comp);
 834         if (count < 0) {
 835                 goto end;
 836         }
 837
 838         for (i = 0; i < count; i++) {
 839                 struct bt_private_port *priv_port =
 840                         bt_private_component_filter_get_input_private_port_by_index(
 841                                 muxer_comp->priv_comp, i);
 842                 struct bt_port *port;
 843
 844                 assert(priv_port);
 845                 port = bt_port_from_private_port(priv_port);
 846                 assert(port);
 847
 848                 if (!bt_port_is_connected(port)) {
 849                         bt_put(priv_port);
 850                         bt_put(port);
 851                         continue;
 852                 }
 853
 854                 bt_put(port);
 855                 bt_put(priv_port);
 856                 muxer_notif_iter->newly_connected_priv_ports =
 857                         g_list_append(
 858                                 muxer_notif_iter->newly_connected_priv_ports,
 859                                 priv_port);
 860                 if (!muxer_notif_iter->newly_connected_priv_ports) {
 861                         ret = -1;
 862                         goto end;
 863                 }
 864         }
 865
 866 end:
 867         bt_put(comp);
 868         return ret;
 869 }
 870
 871 BT_HIDDEN
 872 enum bt_notification_iterator_status muxer_notif_iter_init(
 873                 struct bt_private_notification_iterator *priv_notif_iter,
 874                 struct bt_private_port *output_priv_port)
 875 {
 876         struct muxer_comp *muxer_comp = NULL;
 877         struct muxer_notif_iter *muxer_notif_iter = NULL;
 878         struct bt_private_component *priv_comp = NULL;
 879         enum bt_notification_iterator_status status =
 880                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
 881         int ret;
 882
 883         priv_comp = bt_private_notification_iterator_get_private_component(
 884                 priv_notif_iter);
 885         assert(priv_comp);
 886         muxer_comp = bt_private_component_get_user_data(priv_comp);
 887         assert(muxer_comp);
 888
 889         if (muxer_comp->initializing_muxer_notif_iter) {
 890                 /*
 891                  * Weird, unhandled situation detected: downstream
 892                  * creates a muxer notification iterator while creating
 893                  * another muxer notification iterator (same component).
 894                  */
 895                 goto error;
 896         }
 897
 898         muxer_comp->initializing_muxer_notif_iter = true;
 899         muxer_notif_iter = g_new0(struct muxer_notif_iter, 1);
 900         if (!muxer_notif_iter) {
 901                 goto error;
 902         }
 903
 904         muxer_notif_iter->last_returned_ts_ns = INT64_MIN;
 905         muxer_notif_iter->muxer_upstream_notif_iters =
 906                 g_ptr_array_new_with_free_func(
 907                         (GDestroyNotify) destroy_muxer_upstream_notif_iter);
 908         if (!muxer_notif_iter->muxer_upstream_notif_iters) {
 909                 goto error;
 910         }
 911
 912         /*
 913          * Add the muxer notification iterator to the component's array
 914          * of muxer notification iterators here because
 915          * muxer_notif_iter_init_newly_connected_ports() can cause
 916          * muxer_port_connected() to be called, which adds the newly
 917          * connected port to each muxer notification iterator's list of
 918          * newly connected ports.
 919          */
 920         g_ptr_array_add(muxer_comp->muxer_notif_iters, muxer_notif_iter);
 921         ret = muxer_notif_iter_init_newly_connected_ports(muxer_comp,
 922                 muxer_notif_iter);
 923         if (ret) {
 924                 goto error;
 925         }
 926
 927         ret = bt_private_notification_iterator_set_user_data(priv_notif_iter,
 928                 muxer_notif_iter);
 929         assert(ret == 0);
 930         goto end;
 931
 932 error:
 933         if (g_ptr_array_index(muxer_comp->muxer_notif_iters,
 934                         muxer_comp->muxer_notif_iters->len - 1) == muxer_notif_iter) {
 935                 g_ptr_array_remove_index(muxer_comp->muxer_notif_iters,
 936                         muxer_comp->muxer_notif_iters->len - 1);
 937         }
 938
 939         destroy_muxer_notif_iter(muxer_notif_iter);
 940         ret = bt_private_notification_iterator_set_user_data(priv_notif_iter,
 941                 NULL);
 942         assert(ret == 0);
 943         status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 944
 945 end:
 946         muxer_comp->initializing_muxer_notif_iter = false;
 947         bt_put(priv_comp);
 948         return status;
 949 }
 950
 951 BT_HIDDEN
 952 void muxer_notif_iter_finalize(
 953                 struct bt_private_notification_iterator *priv_notif_iter)
 954 {
 955         struct muxer_notif_iter *muxer_notif_iter =
 956                 bt_private_notification_iterator_get_user_data(priv_notif_iter);
 957         struct bt_private_component *priv_comp = NULL;
 958         struct muxer_comp *muxer_comp = NULL;
 959
 960         priv_comp = bt_private_notification_iterator_get_private_component(
 961                 priv_notif_iter);
 962         assert(priv_comp);
 963         muxer_comp = bt_private_component_get_user_data(priv_comp);
 964
 965         if (muxer_comp) {
 966                 (void) g_ptr_array_remove_fast(muxer_comp->muxer_notif_iters,
 967                         muxer_notif_iter);
 968                 destroy_muxer_notif_iter(muxer_notif_iter);
 969         }
 970
 971         bt_put(priv_comp);
 972 }
 973
 974 BT_HIDDEN
 975 struct bt_notification_iterator_next_return muxer_notif_iter_next(
 976                 struct bt_private_notification_iterator *priv_notif_iter)
 977 {
 978         struct bt_notification_iterator_next_return next_ret;
 979         struct muxer_notif_iter *muxer_notif_iter =
 980                 bt_private_notification_iterator_get_user_data(priv_notif_iter);
 981         struct bt_private_component *priv_comp = NULL;
 982         struct muxer_comp *muxer_comp = NULL;
 983
 984         assert(muxer_notif_iter);
 985         priv_comp = bt_private_notification_iterator_get_private_component(
 986                 priv_notif_iter);
 987         assert(priv_comp);
 988         muxer_comp = bt_private_component_get_user_data(priv_comp);
 989         assert(muxer_comp);
 990
 991         /* Are we in an error state set elsewhere? */
 992         if (unlikely(muxer_comp->error)) {
 993                 next_ret.notification = NULL;
 994                 next_ret.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 995                 goto end;
 996         }
 997
 998         next_ret = muxer_notif_iter_do_next(muxer_comp, muxer_notif_iter);
 999
1000 end:
1001         bt_put(priv_comp);
1002         return next_ret;
1003 }
1004
1005 BT_HIDDEN
1006 void muxer_port_connected(
1007                 struct bt_private_component *priv_comp,
1008                 struct bt_private_port *self_private_port,
1009                 struct bt_port *other_port)
1010 {
1011         struct bt_port *self_port =
1012                 bt_port_from_private_port(self_private_port);
1013         struct muxer_comp *muxer_comp =
1014                 bt_private_component_get_user_data(priv_comp);
1015         size_t i;
1016         int ret;
1017
1018         assert(self_port);
1019         assert(muxer_comp);
1020
1021         if (bt_port_get_type(self_port) == BT_PORT_TYPE_OUTPUT) {
1022                 goto end;
1023         }
1024
1025         for (i = 0; i < muxer_comp->muxer_notif_iters->len; i++) {
1026                 struct muxer_notif_iter *muxer_notif_iter =
1027                         g_ptr_array_index(muxer_comp->muxer_notif_iters, i);
1028
1029                 /*
1030                  * Add this port to the list of newly connected ports
1031                  * for this muxer notification iterator. We append at
1032                  * the end of this list while
1033                  * muxer_notif_iter_handle_newly_connected_ports()
1034                  * removes the nodes from the beginning.
1035                  *
1036                  * The list node owns the private port.
1037                  */
1038                 muxer_notif_iter->newly_connected_priv_ports =
1039                         g_list_append(
1040                                 muxer_notif_iter->newly_connected_priv_ports,
1041                                 self_private_port);
1042                 if (!muxer_notif_iter->newly_connected_priv_ports) {
1043                         /* Put reference taken by bt_get() above */
1044                         muxer_comp->error = true;
1045                         goto end;
1046                 }
1047         }
1048
1049         /* One less available input port */
1050         muxer_comp->available_input_ports--;
1051         ret = ensure_available_input_port(priv_comp);
1052         if (ret) {
1053                 /*
1054                  * Only way to report an error later since this
1055                  * method does not return anything.
1056                  */
1057                 muxer_comp->error = true;
1058                 goto end;
1059         }
1060
1061 end:
1062         bt_put(self_port);
1063 }
1064
1065 BT_HIDDEN
1066 void muxer_port_disconnected(struct bt_private_component *priv_comp,
1067                 struct bt_private_port *priv_port)
1068 {
1069         struct bt_port *port = bt_port_from_private_port(priv_port);
1070         struct muxer_comp *muxer_comp =
1071                 bt_private_component_get_user_data(priv_comp);
1072
1073         assert(port);
1074         assert(muxer_comp);
1075
1076         /*
1077          * There's nothing special to do when a port is disconnected
1078          * because this component deals with upstream notification
1079          * iterators which were already created thanks to connected
1080          * ports. The fact that the port is disconnected does not cancel
1081          * the upstream notification iterators created using its
1082          * connection: they still exist, even if the connection is dead.
1083          * The only way to remove an upstream notification iterator is
1084          * for its "next" operation to return
1085          * BT_NOTIFICATION_ITERATOR_STATUS_END.
1086          */
1087         if (bt_port_get_type(port) == BT_PORT_TYPE_INPUT) {
1088                 /* One more available input port */
1089                 muxer_comp->available_input_ports++;
1090         }
1091
1092         bt_put(port);
1093 }