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