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