lib/graph/graph.c

   1 /*
   2  * graph.c
   3  *
   4  * Babeltrace Plugin Component Graph
   5  *
   6  * Copyright 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   7  *
   8  * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
   9  *
  10  * Permission is hereby granted, free of charge, to any person obtaining a copy
  11  * of this software and associated documentation files (the "Software"), to deal
  12  * in the Software without restriction, including without limitation the rights
  13  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  14  * copies of the Software, and to permit persons to whom the Software is
  15  * furnished to do so, subject to the following conditions:
  16  *
  17  * The above copyright notice and this permission notice shall be included in
  18  * all copies or substantial portions of the Software.
  19  *
  20  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  21  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  22  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  23  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  24  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  25  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  26  * SOFTWARE.
  27  */
  28
  29 #include <babeltrace/graph/component-internal.h>
  30 #include <babeltrace/graph/graph-internal.h>
  31 #include <babeltrace/graph/connection-internal.h>
  32 #include <babeltrace/graph/component-sink-internal.h>
  33 #include <babeltrace/graph/component-source.h>
  34 #include <babeltrace/graph/component-filter.h>
  35 #include <babeltrace/graph/port.h>
  36 #include <babeltrace/compiler-internal.h>
  37 #include <babeltrace/types.h>
  38 #include <unistd.h>
  39 #include <glib.h>
  40
  41 struct bt_graph_listener {
  42         void *func;
  43         void *data;
  44 };
  45
  46 static
  47 void bt_graph_destroy(struct bt_object *obj)
  48 {
  49         struct bt_graph *graph = container_of(obj,
  50                         struct bt_graph, base);
  51
  52         /*
  53          * The graph's reference count is 0 if we're here. Increment
  54          * it to avoid a double-destroy (possibly infinitely recursive)
  55          * in this situation:
  56          *
  57          * 1. We put and destroy a connection.
  58          * 2. This connection's destructor finalizes its active
  59          *    notification iterators.
  60          * 3. A notification iterator's finalization function gets a
  61          *    new reference on its component (reference count goes from
  62          *    0 to 1).
  63          * 4. Since this component's reference count goes to 1, it takes
  64          *    a reference on its parent (this graph). This graph's
  65          *    reference count goes from 0 to 1.
  66          * 5. The notification iterator's finalization function puts its
  67          *    component reference (reference count goes from 1 to 0).
  68          * 6. Since this component's reference count goes from 1 to 0,
  69          *    it puts its parent (this graph). This graph's reference
  70          *    count goes from 1 to 0.
  71          * 7. Since this graph's reference count goes from 1 to 0,
  72          *    its destructor is called (this function).
  73          *
  74          * With the incrementation below, the graph's reference count at
  75          * step 4 goes from 1 to 2, and from 2 to 1 at step 6. This
  76          * ensures that this function is not called two times.
  77          */
  78         obj->ref_count.count++;
  79
  80         if (graph->connections) {
  81                 g_ptr_array_free(graph->connections, TRUE);
  82         }
  83         if (graph->components) {
  84                 g_ptr_array_free(graph->components, TRUE);
  85         }
  86         if (graph->sinks_to_consume) {
  87                 g_queue_free(graph->sinks_to_consume);
  88         }
  89
  90         if (graph->listeners.port_added) {
  91                 g_array_free(graph->listeners.port_added, TRUE);
  92         }
  93
  94         if (graph->listeners.port_removed) {
  95                 g_array_free(graph->listeners.port_removed, TRUE);
  96         }
  97
  98         if (graph->listeners.ports_connected) {
  99                 g_array_free(graph->listeners.ports_connected, TRUE);
 100         }
 101
 102         if (graph->listeners.ports_disconnected) {
 103                 g_array_free(graph->listeners.ports_disconnected, TRUE);
 104         }
 105
 106         g_free(graph);
 107 }
 108
 109 static
 110 int init_listeners_array(GArray **listeners)
 111 {
 112         int ret = 0;
 113
 114         assert(listeners);
 115         *listeners = g_array_new(FALSE, TRUE, sizeof(struct bt_graph_listener));
 116         if (!*listeners) {
 117                 ret = -1;
 118                 goto end;
 119         }
 120
 121 end:
 122         return ret;
 123 }
 124
 125 struct bt_graph *bt_graph_create(void)
 126 {
 127         struct bt_graph *graph;
 128         int ret;
 129
 130         graph = g_new0(struct bt_graph, 1);
 131         if (!graph) {
 132                 goto end;
 133         }
 134
 135         bt_object_init(graph, bt_graph_destroy);
 136
 137         graph->connections = g_ptr_array_new_with_free_func(bt_object_release);
 138         if (!graph->connections) {
 139                 goto error;
 140         }
 141         graph->components = g_ptr_array_new_with_free_func(bt_object_release);
 142         if (!graph->components) {
 143                 goto error;
 144         }
 145         graph->sinks_to_consume = g_queue_new();
 146         if (!graph->sinks_to_consume) {
 147                 goto error;
 148         }
 149
 150         ret = init_listeners_array(&graph->listeners.port_added);
 151         if (ret) {
 152                 goto error;
 153         }
 154
 155         ret = init_listeners_array(&graph->listeners.port_removed);
 156         if (ret) {
 157                 goto error;
 158         }
 159
 160         ret = init_listeners_array(&graph->listeners.ports_connected);
 161         if (ret) {
 162                 goto error;
 163         }
 164
 165         ret = init_listeners_array(&graph->listeners.ports_disconnected);
 166         if (ret) {
 167                 goto error;
 168         }
 169
 170 end:
 171         return graph;
 172 error:
 173         BT_PUT(graph);
 174         goto end;
 175 }
 176
 177 struct bt_connection *bt_graph_connect_ports(struct bt_graph *graph,
 178                 struct bt_port *upstream_port,
 179                 struct bt_port *downstream_port)
 180 {
 181         struct bt_connection *connection = NULL;
 182         struct bt_graph *upstream_graph = NULL;
 183         struct bt_graph *downstream_graph = NULL;
 184         struct bt_component *upstream_component = NULL;
 185         struct bt_component *downstream_component = NULL;
 186         enum bt_component_status component_status;
 187         bt_bool upstream_was_already_in_graph;
 188         bt_bool downstream_was_already_in_graph;
 189
 190         if (!graph || !upstream_port || !downstream_port) {
 191                 goto end;
 192         }
 193
 194         if (graph->canceled) {
 195                 goto end;
 196         }
 197
 198         /* Ensure appropriate types for upstream and downstream ports. */
 199         if (bt_port_get_type(upstream_port) != BT_PORT_TYPE_OUTPUT) {
 200                 goto end;
 201         }
 202         if (bt_port_get_type(downstream_port) != BT_PORT_TYPE_INPUT) {
 203                 goto end;
 204         }
 205
 206         /* Ensure that both ports are currently unconnected. */
 207         if (bt_port_is_connected(upstream_port)) {
 208                 fprintf(stderr, "Upstream port is already connected\n");
 209                 goto end;
 210         }
 211
 212         if (bt_port_is_connected(downstream_port)) {
 213                 fprintf(stderr, "Downstream port is already connected\n");
 214                 goto end;
 215         }
 216
 217         /*
 218          * Ensure that both ports are still attached to their creating
 219          * component.
 220          */
 221         upstream_component = bt_port_get_component(upstream_port);
 222         if (!upstream_component) {
 223                 fprintf(stderr, "Upstream port does not belong to a component\n");
 224                 goto end;
 225         }
 226
 227         downstream_component = bt_port_get_component(downstream_port);
 228         if (!downstream_component) {
 229                 fprintf(stderr, "Downstream port does not belong to a component\n");
 230                 goto end;
 231         }
 232
 233         /* Ensure the components are not already part of another graph. */
 234         upstream_graph = bt_component_get_graph(upstream_component);
 235         if (upstream_graph && (graph != upstream_graph)) {
 236                 fprintf(stderr, "Upstream component is already part of another graph\n");
 237                 goto error;
 238         }
 239         upstream_was_already_in_graph = (graph == upstream_graph);
 240         downstream_graph = bt_component_get_graph(downstream_component);
 241         if (downstream_graph && (graph != downstream_graph)) {
 242                 fprintf(stderr, "Downstream component is already part of another graph\n");
 243                 goto error;
 244         }
 245         downstream_was_already_in_graph = (graph == downstream_graph);
 246
 247         /*
 248          * At this point the ports are not connected yet. Both
 249          * components need to accept an eventual connection to their
 250          * port by the other port before we continue.
 251          */
 252         component_status = bt_component_accept_port_connection(
 253                 upstream_component, upstream_port, downstream_port);
 254         if (component_status != BT_COMPONENT_STATUS_OK) {
 255                 goto error;
 256         }
 257         component_status = bt_component_accept_port_connection(
 258                 downstream_component, downstream_port, upstream_port);
 259         if (component_status != BT_COMPONENT_STATUS_OK) {
 260                 goto error;
 261         }
 262
 263         connection = bt_connection_create(graph, upstream_port,
 264                         downstream_port);
 265         if (!connection) {
 266                 goto error;
 267         }
 268
 269         /*
 270          * Ownership of upstream_component/downstream_component and of
 271          * the connection object is transferred to the graph.
 272          */
 273         g_ptr_array_add(graph->connections, connection);
 274
 275         if (!upstream_was_already_in_graph) {
 276                 g_ptr_array_add(graph->components, upstream_component);
 277                 bt_component_set_graph(upstream_component, graph);
 278         }
 279         if (!downstream_was_already_in_graph) {
 280                 g_ptr_array_add(graph->components, downstream_component);
 281                 bt_component_set_graph(downstream_component, graph);
 282                 if (bt_component_get_class_type(downstream_component) ==
 283                                 BT_COMPONENT_CLASS_TYPE_SINK) {
 284                         g_queue_push_tail(graph->sinks_to_consume,
 285                                         downstream_component);
 286                 }
 287         }
 288
 289         /*
 290          * The graph is now the parent of these components which
 291          * garantees their existence for the duration of the graph's
 292          * lifetime.
 293          */
 294
 295         /*
 296          * Notify both components that their port is connected.
 297          */
 298         bt_component_port_connected(upstream_component, upstream_port,
 299                 downstream_port);
 300         bt_component_port_connected(downstream_component, downstream_port,
 301                 upstream_port);
 302
 303         /*
 304          * Notify the graph's creator that both ports are connected.
 305          */
 306         bt_graph_notify_ports_connected(graph, upstream_port, downstream_port);
 307
 308 end:
 309         bt_put(upstream_graph);
 310         bt_put(downstream_graph);
 311         bt_put(upstream_component);
 312         bt_put(downstream_component);
 313         return connection;
 314
 315 error:
 316         BT_PUT(upstream_component);
 317         BT_PUT(downstream_component);
 318         goto end;
 319 }
 320
 321 static
 322 enum bt_component_status get_component_port_counts(
 323                 struct bt_component *component, int64_t *input_count,
 324                 int64_t *output_count)
 325 {
 326         enum bt_component_status ret;
 327
 328         switch (bt_component_get_class_type(component)) {
 329         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 330                 *output_count =
 331                         bt_component_source_get_output_port_count(component);
 332                 if (*output_count < 0) {
 333                         ret = BT_COMPONENT_STATUS_ERROR;
 334                         goto end;
 335                 }
 336                 break;
 337         case BT_COMPONENT_CLASS_TYPE_FILTER:
 338                 *output_count =
 339                         bt_component_filter_get_output_port_count(component);
 340                 if (*output_count < 0) {
 341                         ret = BT_COMPONENT_STATUS_ERROR;
 342                         goto end;
 343                 }
 344                 *input_count =
 345                         bt_component_filter_get_input_port_count(component);
 346                 if (*input_count < 0) {
 347                         ret = BT_COMPONENT_STATUS_ERROR;
 348                         goto end;
 349                 }
 350                 break;
 351         case BT_COMPONENT_CLASS_TYPE_SINK:
 352                 *input_count =
 353                         bt_component_sink_get_input_port_count(component);
 354                 if (*input_count < 0) {
 355                         ret = BT_COMPONENT_STATUS_ERROR;
 356                         goto end;
 357                 }
 358                 break;
 359         default:
 360                 assert(BT_FALSE);
 361                 break;
 362         }
 363         ret = BT_COMPONENT_STATUS_OK;
 364 end:
 365         return ret;
 366 }
 367
 368 enum bt_graph_status bt_graph_add_component_as_sibling(struct bt_graph *graph,
 369                 struct bt_component *origin,
 370                 struct bt_component *new_component)
 371 {
 372         int64_t origin_input_port_count = 0;
 373         int64_t origin_output_port_count = 0;
 374         int64_t new_input_port_count = 0;
 375         int64_t new_output_port_count = 0;
 376         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 377         struct bt_graph *origin_graph = NULL;
 378         struct bt_graph *new_graph = NULL;
 379         struct bt_port *origin_port = NULL;
 380         struct bt_port *new_port = NULL;
 381         struct bt_port *upstream_port = NULL;
 382         struct bt_port *downstream_port = NULL;
 383         struct bt_connection *origin_connection = NULL;
 384         struct bt_connection *new_connection = NULL;
 385         int64_t port_index;
 386
 387         if (!graph || !origin || !new_component) {
 388                 status = BT_GRAPH_STATUS_INVALID;
 389                 goto end;
 390         }
 391
 392         if (graph->canceled) {
 393                 status = BT_GRAPH_STATUS_CANCELED;
 394                 goto end;
 395         }
 396
 397         if (bt_component_get_class_type(origin) !=
 398                         bt_component_get_class_type(new_component)) {
 399                 status = BT_GRAPH_STATUS_INVALID;
 400                 goto end;
 401         }
 402
 403         origin_graph = bt_component_get_graph(origin);
 404         if (!origin_graph || (origin_graph != graph)) {
 405                 status = BT_GRAPH_STATUS_INVALID;
 406                 goto end;
 407         }
 408
 409         new_graph = bt_component_get_graph(new_component);
 410         if (new_graph) {
 411                 status = BT_GRAPH_STATUS_ALREADY_IN_A_GRAPH;
 412                 goto end;
 413         }
 414
 415         if (get_component_port_counts(origin, &origin_input_port_count,
 416                         &origin_output_port_count) != BT_COMPONENT_STATUS_OK) {
 417                 status = BT_GRAPH_STATUS_INVALID;
 418                 goto end;
 419         }
 420         if (get_component_port_counts(new_component, &new_input_port_count,
 421                         &new_output_port_count) != BT_COMPONENT_STATUS_OK) {
 422                 status = BT_GRAPH_STATUS_INVALID;
 423                 goto end;
 424         }
 425
 426         if (origin_input_port_count != new_input_port_count ||
 427                         origin_output_port_count != new_output_port_count) {
 428                 status = BT_GRAPH_STATUS_INVALID;
 429                 goto end;
 430         }
 431
 432         /* Replicate input connections. */
 433         for (port_index = 0; port_index< origin_input_port_count; port_index++) {
 434                 origin_port = bt_component_get_input_port_by_index(origin,
 435                         port_index);
 436                 if (!origin_port) {
 437                         status = BT_GRAPH_STATUS_ERROR;
 438                         goto error_disconnect;
 439                 }
 440
 441                 new_port = bt_component_get_input_port_by_index(new_component,
 442                         port_index);
 443                 if (!new_port) {
 444                         status = BT_GRAPH_STATUS_ERROR;
 445                         goto error_disconnect;
 446                 }
 447
 448                 origin_connection = bt_port_get_connection(origin_port);
 449                 if (origin_connection) {
 450                         upstream_port = bt_connection_get_upstream_port(
 451                                 origin_connection);
 452                         if (!upstream_port) {
 453                                 goto error_disconnect;
 454                         }
 455
 456                         new_connection = bt_graph_connect_ports(graph,
 457                                 upstream_port, new_port);
 458                         if (!new_connection) {
 459                                 goto error_disconnect;
 460                         }
 461                 }
 462
 463                 BT_PUT(upstream_port);
 464                 BT_PUT(origin_connection);
 465                 BT_PUT(new_connection);
 466                 BT_PUT(origin_port);
 467                 BT_PUT(new_port);
 468         }
 469
 470         /* Replicate output connections. */
 471         for (port_index = 0; port_index < origin_output_port_count; port_index++) {
 472                 origin_port = bt_component_get_output_port_by_index(origin,
 473                         port_index);
 474                 if (!origin_port) {
 475                         status = BT_GRAPH_STATUS_ERROR;
 476                         goto error_disconnect;
 477                 }
 478                 new_port = bt_component_get_output_port_by_index(new_component,
 479                         port_index);
 480                 if (!new_port) {
 481                         status = BT_GRAPH_STATUS_ERROR;
 482                         goto error_disconnect;
 483                 }
 484
 485                 origin_connection = bt_port_get_connection(origin_port);
 486                 if (origin_connection) {
 487                         downstream_port = bt_connection_get_downstream_port(
 488                                         origin_connection);
 489                         if (!downstream_port) {
 490                                 goto error_disconnect;
 491                         }
 492
 493                         new_connection = bt_graph_connect_ports(graph,
 494                                 new_port, downstream_port);
 495                         if (!new_connection) {
 496                                 goto error_disconnect;
 497                         }
 498                 }
 499
 500                 BT_PUT(downstream_port);
 501                 BT_PUT(origin_connection);
 502                 BT_PUT(new_connection);
 503                 BT_PUT(origin_port);
 504                 BT_PUT(new_port);
 505         }
 506 end:
 507         bt_put(origin_graph);
 508         bt_put(new_graph);
 509         bt_put(origin_port);
 510         bt_put(new_port);
 511         bt_put(upstream_port);
 512         bt_put(downstream_port);
 513         bt_put(origin_connection);
 514         bt_put(new_connection);
 515         return status;
 516 error_disconnect:
 517         /* Destroy all connections of the new component. */
 518         /* FIXME. */
 519         goto end;
 520 }
 521
 522 enum bt_graph_status bt_graph_consume(struct bt_graph *graph)
 523 {
 524         struct bt_component *sink;
 525         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 526         enum bt_component_status comp_status;
 527         GList *current_node;
 528
 529         if (!graph) {
 530                 status = BT_GRAPH_STATUS_INVALID;
 531                 goto end;
 532         }
 533
 534         if (graph->canceled) {
 535                 status = BT_GRAPH_STATUS_CANCELED;
 536                 goto end;
 537         }
 538
 539         if (g_queue_is_empty(graph->sinks_to_consume)) {
 540                 status = BT_GRAPH_STATUS_END;
 541                 goto end;
 542         }
 543
 544         current_node = g_queue_pop_head_link(graph->sinks_to_consume);
 545         sink = current_node->data;
 546         comp_status = bt_component_sink_consume(sink);
 547         switch (comp_status) {
 548         case BT_COMPONENT_STATUS_OK:
 549                 break;
 550         case BT_COMPONENT_STATUS_END:
 551                 status = BT_GRAPH_STATUS_END;
 552                 break;
 553         case BT_COMPONENT_STATUS_AGAIN:
 554                 status = BT_GRAPH_STATUS_AGAIN;
 555                 break;
 556         case BT_COMPONENT_STATUS_INVALID:
 557                 status = BT_GRAPH_STATUS_INVALID;
 558                 break;
 559         default:
 560                 status = BT_GRAPH_STATUS_ERROR;
 561                 break;
 562         }
 563
 564         if (status != BT_GRAPH_STATUS_END) {
 565                 g_queue_push_tail_link(graph->sinks_to_consume, current_node);
 566                 goto end;
 567         }
 568
 569         /* End reached, the node is not added back to the queue and free'd. */
 570         g_queue_delete_link(graph->sinks_to_consume, current_node);
 571
 572         /* Don't forward an END status if there are sinks left to consume. */
 573         if (!g_queue_is_empty(graph->sinks_to_consume)) {
 574                 status = BT_GRAPH_STATUS_OK;
 575                 goto end;
 576         }
 577 end:
 578         return status;
 579 }
 580
 581 enum bt_graph_status bt_graph_run(struct bt_graph *graph)
 582 {
 583         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 584
 585         if (!graph) {
 586                 status = BT_GRAPH_STATUS_INVALID;
 587                 goto end;
 588         }
 589
 590         do {
 591                 if (graph->canceled) {
 592                         status = BT_GRAPH_STATUS_CANCELED;
 593                         goto end;
 594                 }
 595
 596                 status = bt_graph_consume(graph);
 597                 if (status == BT_GRAPH_STATUS_AGAIN) {
 598                         /*
 599                          * If AGAIN is received and there are multiple
 600                          * sinks, go ahead and consume from the next
 601                          * sink.
 602                          *
 603                          * However, in the case where a single sink is
 604                          * left, the caller can decide to busy-wait and
 605                          * call bt_graph_run() continuously until the
 606                          * source is ready or it can decide to sleep for
 607                          * an arbitrary amount of time.
 608                          */
 609                         if (graph->sinks_to_consume->length > 1) {
 610                                 status = BT_GRAPH_STATUS_OK;
 611                         }
 612                 }
 613         } while (status == BT_GRAPH_STATUS_OK);
 614
 615         if (g_queue_is_empty(graph->sinks_to_consume)) {
 616                 status = BT_GRAPH_STATUS_END;
 617         }
 618 end:
 619         return status;
 620 }
 621
 622 static
 623 void add_listener(GArray *listeners, void *func, void *data)
 624 {
 625         struct bt_graph_listener listener = {
 626                 .func = func,
 627                 .data = data,
 628         };
 629
 630         g_array_append_val(listeners, listener);
 631 }
 632
 633 enum bt_graph_status bt_graph_add_port_added_listener(
 634                 struct bt_graph *graph,
 635                 bt_graph_port_added_listener listener, void *data)
 636 {
 637         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 638
 639         if (!graph || !listener) {
 640                 status = BT_GRAPH_STATUS_INVALID;
 641                 goto end;
 642         }
 643
 644         add_listener(graph->listeners.port_added, listener, data);
 645
 646 end:
 647         return status;
 648 }
 649
 650 enum bt_graph_status bt_graph_add_port_removed_listener(
 651                 struct bt_graph *graph,
 652                 bt_graph_port_removed_listener listener, void *data)
 653 {
 654         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 655
 656         if (!graph || !listener) {
 657                 status = BT_GRAPH_STATUS_INVALID;
 658                 goto end;
 659         }
 660
 661         add_listener(graph->listeners.port_removed, listener, data);
 662
 663 end:
 664         return status;
 665 }
 666
 667 enum bt_graph_status bt_graph_add_ports_connected_listener(
 668                 struct bt_graph *graph,
 669                 bt_graph_ports_connected_listener listener, void *data)
 670 {
 671         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 672
 673         if (!graph || !listener) {
 674                 status = BT_GRAPH_STATUS_INVALID;
 675                 goto end;
 676         }
 677
 678         add_listener(graph->listeners.ports_connected, listener, data);
 679
 680 end:
 681         return status;
 682 }
 683
 684 enum bt_graph_status bt_graph_add_ports_disconnected_listener(
 685                 struct bt_graph *graph,
 686                 bt_graph_ports_disconnected_listener listener, void *data)
 687 {
 688         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 689
 690         if (!graph || !listener) {
 691                 status = BT_GRAPH_STATUS_INVALID;
 692                 goto end;
 693         }
 694
 695         add_listener(graph->listeners.ports_disconnected, listener, data);
 696
 697 end:
 698         return status;
 699 }
 700
 701 BT_HIDDEN
 702 void bt_graph_notify_port_added(struct bt_graph *graph, struct bt_port *port)
 703 {
 704         size_t i;
 705
 706         for (i = 0; i < graph->listeners.port_added->len; i++) {
 707                 struct bt_graph_listener listener =
 708                         g_array_index(graph->listeners.port_added,
 709                                 struct bt_graph_listener, i);
 710                 bt_graph_port_added_listener func = listener.func;
 711
 712                 assert(func);
 713                 func(port, listener.data);
 714         }
 715 }
 716
 717 BT_HIDDEN
 718 void bt_graph_notify_port_removed(struct bt_graph *graph,
 719                 struct bt_component *comp, struct bt_port *port)
 720 {
 721         size_t i;
 722
 723         for (i = 0; i < graph->listeners.port_removed->len; i++) {
 724                 struct bt_graph_listener listener =
 725                         g_array_index(graph->listeners.port_removed,
 726                                 struct bt_graph_listener, i);
 727                 bt_graph_port_removed_listener func = listener.func;
 728
 729                 assert(func);
 730                 func(comp, port, listener.data);
 731         }
 732 }
 733
 734 BT_HIDDEN
 735 void bt_graph_notify_ports_connected(struct bt_graph *graph,
 736                 struct bt_port *upstream_port, struct bt_port *downstream_port)
 737 {
 738         size_t i;
 739
 740         for (i = 0; i < graph->listeners.ports_connected->len; i++) {
 741                 struct bt_graph_listener listener =
 742                         g_array_index(graph->listeners.ports_connected,
 743                                 struct bt_graph_listener, i);
 744                 bt_graph_ports_connected_listener func = listener.func;
 745
 746                 assert(func);
 747                 func(upstream_port, downstream_port, listener.data);
 748         }
 749 }
 750
 751 BT_HIDDEN
 752 void bt_graph_notify_ports_disconnected(struct bt_graph *graph,
 753                 struct bt_component *upstream_comp,
 754                 struct bt_component *downstream_comp,
 755                 struct bt_port *upstream_port, struct bt_port *downstream_port)
 756 {
 757         size_t i;
 758
 759         for (i = 0; i < graph->listeners.ports_disconnected->len; i++) {
 760                 struct bt_graph_listener listener =
 761                         g_array_index(graph->listeners.ports_disconnected,
 762                                 struct bt_graph_listener, i);
 763                 bt_graph_ports_disconnected_listener func = listener.func;
 764
 765                 assert(func);
 766                 func(upstream_comp, downstream_comp, upstream_port,
 767                         downstream_port, listener.data);
 768         }
 769 }
 770
 771 extern enum bt_graph_status bt_graph_cancel(struct bt_graph *graph)
 772 {
 773         enum bt_graph_status ret = BT_GRAPH_STATUS_OK;
 774
 775         if (!graph) {
 776                 ret = BT_GRAPH_STATUS_INVALID;
 777                 goto end;
 778         }
 779
 780         graph->canceled = BT_TRUE;
 781
 782 end:
 783         return ret;
 784 }
 785
 786 extern bt_bool bt_graph_is_canceled(struct bt_graph *graph)
 787 {
 788         return graph ? graph->canceled : BT_FALSE;
 789 }