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