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