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  * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
   8  *
   9  * Permission is hereby granted, free of charge, to any person obtaining a copy
  10  * of this software and associated documentation files (the "Software"), to deal
  11  * in the Software without restriction, including without limitation the rights
  12  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  13  * copies of the Software, and to permit persons to whom the Software is
  14  * furnished to do so, subject to the following conditions:
  15  *
  16  * The above copyright notice and this permission notice shall be included in
  17  * all copies or substantial portions of the Software.
  18  *
  19  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  20  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  21  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  22  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  23  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  24  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  25  * SOFTWARE.
  26  */
  27
  28 #define BT_LOG_TAG "GRAPH"
  29 #include <babeltrace/lib-logging-internal.h>
  30
  31 #include <babeltrace/graph/component-internal.h>
  32 #include <babeltrace/graph/graph-internal.h>
  33 #include <babeltrace/graph/connection-internal.h>
  34 #include <babeltrace/graph/component-sink-internal.h>
  35 #include <babeltrace/graph/component-source.h>
  36 #include <babeltrace/graph/component-filter.h>
  37 #include <babeltrace/graph/port.h>
  38 #include <babeltrace/compiler-internal.h>
  39 #include <babeltrace/types.h>
  40 #include <babeltrace/values.h>
  41 #include <babeltrace/values-internal.h>
  42 #include <unistd.h>
  43 #include <glib.h>
  44
  45 struct bt_graph_listener {
  46         void *func;
  47         bt_graph_listener_removed removed;
  48         void *data;
  49 };
  50
  51 static
  52 int init_listeners_array(GArray **listeners)
  53 {
  54         int ret = 0;
  55
  56         assert(listeners);
  57         *listeners = g_array_new(FALSE, TRUE, sizeof(struct bt_graph_listener));
  58         if (!*listeners) {
  59                 BT_LOGE_STR("Failed to allocate one GArray.");
  60                 ret = -1;
  61                 goto end;
  62         }
  63
  64 end:
  65         return ret;
  66 }
  67
  68 static
  69 void call_remove_listeners(GArray *listeners)
  70 {
  71         size_t i;
  72
  73         for (i = 0; i < listeners->len; i++) {
  74                 struct bt_graph_listener listener =
  75                         g_array_index(listeners, struct bt_graph_listener, i);
  76
  77                 if (listener.removed) {
  78                         listener.removed(listener.data);
  79                 }
  80         }
  81 }
  82
  83 static
  84 void bt_graph_destroy(struct bt_object *obj)
  85 {
  86         struct bt_graph *graph = container_of(obj,
  87                         struct bt_graph, base);
  88
  89         /*
  90          * The graph's reference count is 0 if we're here. Increment
  91          * it to avoid a double-destroy (possibly infinitely recursive)
  92          * in this situation:
  93          *
  94          * 1. We put and destroy a connection.
  95          * 2. This connection's destructor finalizes its active
  96          *    notification iterators.
  97          * 3. A notification iterator's finalization function gets a
  98          *    new reference on its component (reference count goes from
  99          *    0 to 1).
 100          * 4. Since this component's reference count goes to 1, it takes
 101          *    a reference on its parent (this graph). This graph's
 102          *    reference count goes from 0 to 1.
 103          * 5. The notification iterator's finalization function puts its
 104          *    component reference (reference count goes from 1 to 0).
 105          * 6. Since this component's reference count goes from 1 to 0,
 106          *    it puts its parent (this graph). This graph's reference
 107          *    count goes from 1 to 0.
 108          * 7. Since this graph's reference count goes from 1 to 0,
 109          *    its destructor is called (this function).
 110          *
 111          * With the incrementation below, the graph's reference count at
 112          * step 4 goes from 1 to 2, and from 2 to 1 at step 6. This
 113          * ensures that this function is not called two times.
 114          */
 115         BT_LOGD("Destroying graph: addr=%p", graph);
 116         obj->ref_count.count++;
 117
 118         /*
 119          * Cancel the graph to disallow some operations, like creating
 120          * notification iterators and adding ports to components.
 121          */
 122         (void) bt_graph_cancel(graph);
 123
 124         /* Call all remove listeners */
 125         call_remove_listeners(graph->listeners.port_added);
 126         call_remove_listeners(graph->listeners.port_removed);
 127         call_remove_listeners(graph->listeners.ports_connected);
 128         call_remove_listeners(graph->listeners.ports_disconnected);
 129
 130         if (graph->connections) {
 131                 BT_LOGD_STR("Destroying connections.");
 132                 g_ptr_array_free(graph->connections, TRUE);
 133         }
 134         if (graph->components) {
 135                 BT_LOGD_STR("Destroying components.");
 136                 g_ptr_array_free(graph->components, TRUE);
 137         }
 138         if (graph->sinks_to_consume) {
 139                 g_queue_free(graph->sinks_to_consume);
 140         }
 141
 142         if (graph->listeners.port_added) {
 143                 g_array_free(graph->listeners.port_added, TRUE);
 144         }
 145
 146         if (graph->listeners.port_removed) {
 147                 g_array_free(graph->listeners.port_removed, TRUE);
 148         }
 149
 150         if (graph->listeners.ports_connected) {
 151                 g_array_free(graph->listeners.ports_connected, TRUE);
 152         }
 153
 154         if (graph->listeners.ports_disconnected) {
 155                 g_array_free(graph->listeners.ports_disconnected, TRUE);
 156         }
 157
 158         g_free(graph);
 159 }
 160
 161 struct bt_graph *bt_graph_create(void)
 162 {
 163         struct bt_graph *graph;
 164         int ret;
 165
 166         BT_LOGD_STR("Creating graph object.");
 167         graph = g_new0(struct bt_graph, 1);
 168         if (!graph) {
 169                 BT_LOGE_STR("Failed to allocate one graph.");
 170                 goto end;
 171         }
 172
 173         bt_object_init(graph, bt_graph_destroy);
 174
 175         graph->connections = g_ptr_array_new_with_free_func(bt_object_release);
 176         if (!graph->connections) {
 177                 BT_LOGE_STR("Failed to allocate one GPtrArray.");
 178                 goto error;
 179         }
 180         graph->components = g_ptr_array_new_with_free_func(bt_object_release);
 181         if (!graph->components) {
 182                 BT_LOGE_STR("Failed to allocate one GPtrArray.");
 183                 goto error;
 184         }
 185         graph->sinks_to_consume = g_queue_new();
 186         if (!graph->sinks_to_consume) {
 187                 BT_LOGE_STR("Failed to allocate one GQueue.");
 188                 goto error;
 189         }
 190
 191         graph->can_consume = BT_TRUE;
 192         ret = init_listeners_array(&graph->listeners.port_added);
 193         if (ret) {
 194                 BT_LOGE_STR("Cannot create the \"port added\" listener array.");
 195                 goto error;
 196         }
 197
 198         ret = init_listeners_array(&graph->listeners.port_removed);
 199         if (ret) {
 200                 BT_LOGE_STR("Cannot create the \"port removed\" listener array.");
 201                 goto error;
 202         }
 203
 204         ret = init_listeners_array(&graph->listeners.ports_connected);
 205         if (ret) {
 206                 BT_LOGE_STR("Cannot create the \"port connected\" listener array.");
 207                 goto error;
 208         }
 209
 210         ret = init_listeners_array(&graph->listeners.ports_disconnected);
 211         if (ret) {
 212                 BT_LOGE_STR("Cannot create the \"port disconneted\" listener array.");
 213                 goto error;
 214         }
 215
 216         BT_LOGD("Created graph object: addr=%p", graph);
 217
 218 end:
 219         return graph;
 220 error:
 221         BT_PUT(graph);
 222         goto end;
 223 }
 224
 225 enum bt_graph_status bt_graph_connect_ports(struct bt_graph *graph,
 226                 struct bt_port *upstream_port, struct bt_port *downstream_port,
 227                 struct bt_connection **user_connection)
 228 {
 229         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 230         struct bt_connection *connection = NULL;
 231         struct bt_graph *upstream_graph = NULL;
 232         struct bt_graph *downstream_graph = NULL;
 233         struct bt_component *upstream_component = NULL;
 234         struct bt_component *downstream_component = NULL;
 235         enum bt_component_status component_status;
 236         bt_bool init_can_consume;
 237
 238         if (!graph) {
 239                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 240                 status = BT_GRAPH_STATUS_INVALID;
 241                 goto end;
 242         }
 243         init_can_consume = graph->can_consume;
 244
 245         if (!upstream_port) {
 246                 BT_LOGW_STR("Invalid parameter: upstream port is NULL.");
 247                 status = BT_GRAPH_STATUS_INVALID;
 248                 goto end;
 249         }
 250
 251         if (!downstream_port) {
 252                 BT_LOGW_STR("Invalid parameter: downstream port is NULL.");
 253                 status = BT_GRAPH_STATUS_INVALID;
 254                 goto end;
 255         }
 256
 257         BT_LOGD("Connecting component ports within graph: "
 258                 "graph-addr=%p, "
 259                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 260                 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
 261                 graph, upstream_port, bt_port_get_name(upstream_port),
 262                 downstream_port, bt_port_get_name(downstream_port));
 263
 264         if (graph->canceled) {
 265                 BT_LOGW_STR("Invalid parameter: graph is canceled.");
 266                 status = BT_GRAPH_STATUS_CANCELED;
 267                 goto end;
 268         }
 269
 270         graph->can_consume = BT_FALSE;
 271
 272         /* Ensure appropriate types for upstream and downstream ports. */
 273         if (bt_port_get_type(upstream_port) != BT_PORT_TYPE_OUTPUT) {
 274                 BT_LOGW_STR("Invalid parameter: upstream port is not an output port.");
 275                 status = BT_GRAPH_STATUS_INVALID;
 276                 goto end;
 277         }
 278         if (bt_port_get_type(downstream_port) != BT_PORT_TYPE_INPUT) {
 279                 BT_LOGW_STR("Invalid parameter: downstream port is not an input port.");
 280                 status = BT_GRAPH_STATUS_INVALID;
 281                 goto end;
 282         }
 283
 284         /* Ensure that both ports are currently unconnected. */
 285         if (bt_port_is_connected(upstream_port)) {
 286                 BT_LOGW_STR("Invalid parameter: upstream port is already connected.");
 287                 status = BT_GRAPH_STATUS_INVALID;
 288                 goto end;
 289         }
 290
 291         if (bt_port_is_connected(downstream_port)) {
 292                 BT_LOGW_STR("Invalid parameter: downstream port is already connected.");
 293                 status = BT_GRAPH_STATUS_INVALID;
 294                 goto end;
 295         }
 296
 297         /*
 298          * Ensure that both ports are still attached to their creating
 299          * component.
 300          */
 301         upstream_component = bt_port_get_component(upstream_port);
 302         if (!upstream_component) {
 303                 BT_LOGW_STR("Invalid parameter: upstream port is loose (does not belong to a component)");
 304                 status = BT_GRAPH_STATUS_INVALID;
 305                 goto end;
 306         }
 307
 308         downstream_component = bt_port_get_component(downstream_port);
 309         if (!downstream_component) {
 310                 BT_LOGW_STR("Invalid parameter: downstream port is loose (does not belong to a component)");
 311                 status = BT_GRAPH_STATUS_INVALID;
 312                 goto end;
 313         }
 314
 315         BT_LOGD("Connecting component ports: "
 316                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 317                 "downstream-comp-addr=%p, downstream-comp-name=\"%s\"",
 318                 upstream_component, bt_component_get_name(upstream_component),
 319                 downstream_component, bt_component_get_name(downstream_component));
 320
 321         /*
 322          * At this point the ports are not connected yet. Both
 323          * components need to accept an eventual connection to their
 324          * port by the other port before we continue.
 325          */
 326         BT_LOGD_STR("Asking upstream component to accept the connection.");
 327         component_status = bt_component_accept_port_connection(
 328                 upstream_component, upstream_port, downstream_port);
 329         if (component_status != BT_COMPONENT_STATUS_OK) {
 330                 if (component_status == BT_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
 331                         BT_LOGD_STR("Upstream component refused the connection.");
 332                 } else {
 333                         BT_LOGW("Cannot ask upstream component to accept the connection: "
 334                                 "status=%s", bt_component_status_string(component_status));
 335                 }
 336
 337                 status = bt_graph_status_from_component_status(
 338                         component_status);
 339                 goto end;
 340         }
 341
 342         BT_LOGD_STR("Asking downstream component to accept the connection.");
 343         component_status = bt_component_accept_port_connection(
 344                 downstream_component, downstream_port, upstream_port);
 345         if (component_status != BT_COMPONENT_STATUS_OK) {
 346                 if (component_status == BT_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
 347                         BT_LOGD_STR("Downstream component refused the connection.");
 348                 } else {
 349                         BT_LOGW("Cannot ask downstream component to accept the connection: "
 350                                 "status=%s", bt_component_status_string(component_status));
 351                 }
 352
 353                 status = bt_graph_status_from_component_status(
 354                         component_status);
 355                 goto end;
 356         }
 357
 358         BT_LOGD_STR("Creating connection.");
 359         connection = bt_connection_create(graph, upstream_port,
 360                         downstream_port);
 361         if (!connection) {
 362                 BT_LOGW("Cannot create connection object.");
 363                 status = BT_GRAPH_STATUS_NOMEM;
 364                 goto end;
 365         }
 366
 367         BT_LOGD("Connection object created: conn-addr=%p", connection);
 368
 369         /*
 370          * Ownership of upstream_component/downstream_component and of
 371          * the connection object is transferred to the graph.
 372          */
 373         g_ptr_array_add(graph->connections, connection);
 374
 375         /*
 376          * Notify both components that their port is connected.
 377          */
 378         BT_LOGD_STR("Notifying upstream component that its port is connected.");
 379         bt_component_port_connected(upstream_component, upstream_port,
 380                 downstream_port);
 381         BT_LOGD_STR("Notifying downstream component that its port is connected.");
 382         bt_component_port_connected(downstream_component, downstream_port,
 383                 upstream_port);
 384
 385         /*
 386          * Notify the graph's creator that both ports are connected.
 387          */
 388         BT_LOGD_STR("Notifying graph's user that new component ports are connected.");
 389         bt_graph_notify_ports_connected(graph, upstream_port, downstream_port);
 390         BT_LOGD("Connected component ports within graph: "
 391                 "graph-addr=%p, "
 392                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 393                 "downstream-comp-addr=%p, downstream-comp-name=\"%s\", "
 394                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 395                 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
 396                 graph,
 397                 upstream_component, bt_component_get_name(upstream_component),
 398                 downstream_component, bt_component_get_name(downstream_component),
 399                 upstream_port, bt_port_get_name(upstream_port),
 400                 downstream_port, bt_port_get_name(downstream_port));
 401
 402         if (user_connection) {
 403                 /* Move reference to user */
 404                 *user_connection = connection;
 405                 connection = NULL;
 406         }
 407
 408 end:
 409         bt_put(upstream_graph);
 410         bt_put(downstream_graph);
 411         bt_put(upstream_component);
 412         bt_put(downstream_component);
 413         bt_put(connection);
 414         if (graph) {
 415                 graph->can_consume = init_can_consume;
 416         }
 417         return status;
 418 }
 419
 420 static
 421 enum bt_graph_status consume_graph_sink(struct bt_component *sink)
 422 {
 423         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 424         enum bt_component_status comp_status;
 425
 426         assert(sink);
 427         comp_status = bt_component_sink_consume(sink);
 428         BT_LOGV("Consumed from sink: addr=%p, name=\"%s\", status=%s",
 429                 sink, bt_component_get_name(sink),
 430                 bt_component_status_string(comp_status));
 431
 432         switch (comp_status) {
 433         case BT_COMPONENT_STATUS_OK:
 434                 break;
 435         case BT_COMPONENT_STATUS_END:
 436                 status = BT_GRAPH_STATUS_END;
 437                 break;
 438         case BT_COMPONENT_STATUS_AGAIN:
 439                 status = BT_GRAPH_STATUS_AGAIN;
 440                 break;
 441         case BT_COMPONENT_STATUS_INVALID:
 442                 status = BT_GRAPH_STATUS_INVALID;
 443                 break;
 444         default:
 445                 status = BT_GRAPH_STATUS_ERROR;
 446                 break;
 447         }
 448
 449         return status;
 450 }
 451
 452 /*
 453  * `node` is removed from the queue of sinks to consume when passed to
 454  * this function. This function adds it back to the queue if there's
 455  * still something to consume afterwards.
 456  */
 457 static
 458 enum bt_graph_status consume_sink_node(struct bt_graph *graph,
 459                 GList *node)
 460 {
 461         enum bt_graph_status status;
 462         struct bt_component *sink;
 463
 464         sink = node->data;
 465         status = consume_graph_sink(sink);
 466         if (status != BT_GRAPH_STATUS_END) {
 467                 g_queue_push_tail_link(graph->sinks_to_consume, node);
 468                 goto end;
 469         }
 470
 471         /* End reached, the node is not added back to the queue and free'd. */
 472         g_queue_delete_link(graph->sinks_to_consume, node);
 473
 474         /* Don't forward an END status if there are sinks left to consume. */
 475         if (!g_queue_is_empty(graph->sinks_to_consume)) {
 476                 status = BT_GRAPH_STATUS_OK;
 477                 goto end;
 478         }
 479
 480 end:
 481         BT_LOGV("Consumed sink node: status=%s", bt_graph_status_string(status));
 482         return status;
 483 }
 484
 485 BT_HIDDEN
 486 enum bt_graph_status bt_graph_consume_sink_no_check(struct bt_graph *graph,
 487                 struct bt_component *sink)
 488 {
 489         enum bt_graph_status status;
 490         GList *sink_node;
 491         int index;
 492
 493         BT_LOGV("Making specific sink consume: addr=%p, "
 494                 "comp-addr=%p, comp-name=\"%s\"",
 495                 graph, sink, bt_component_get_name(sink));
 496
 497         assert(bt_component_borrow_graph(sink) == graph);
 498
 499         if (g_queue_is_empty(graph->sinks_to_consume)) {
 500                 BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
 501                 status = BT_GRAPH_STATUS_END;
 502                 goto end;
 503         }
 504
 505         index = g_queue_index(graph->sinks_to_consume, sink);
 506         if (index < 0) {
 507                 BT_LOGV_STR("Sink is not marked as consumable: sink is ended.");
 508                 status = BT_GRAPH_STATUS_END;
 509                 goto end;
 510         }
 511
 512         sink_node = g_queue_pop_nth_link(graph->sinks_to_consume, index);
 513         assert(sink_node);
 514         status = consume_sink_node(graph, sink_node);
 515
 516 end:
 517         return status;
 518 }
 519
 520 BT_HIDDEN
 521 enum bt_graph_status bt_graph_consume_no_check(struct bt_graph *graph)
 522 {
 523         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 524         struct bt_component *sink;
 525         GList *current_node;
 526
 527         BT_LOGV("Making next sink consume: addr=%p", graph);
 528
 529         if (!graph->has_sink) {
 530                 BT_LOGW_STR("Graph has no sink component.");
 531                 status = BT_GRAPH_STATUS_NO_SINK;
 532                 goto end;
 533         }
 534
 535         if (g_queue_is_empty(graph->sinks_to_consume)) {
 536                 BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
 537                 status = BT_GRAPH_STATUS_END;
 538                 goto end;
 539         }
 540
 541         current_node = g_queue_pop_head_link(graph->sinks_to_consume);
 542         sink = current_node->data;
 543         BT_LOGV("Chose next sink to consume: comp-addr=%p, comp-name=\"%s\"",
 544                 sink, bt_component_get_name(sink));
 545         status = consume_sink_node(graph, current_node);
 546
 547 end:
 548         return status;
 549 }
 550
 551 enum bt_graph_status bt_graph_consume(struct bt_graph *graph)
 552 {
 553         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 554
 555         if (!graph) {
 556                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 557                 status = BT_GRAPH_STATUS_INVALID;
 558                 goto end;
 559         }
 560
 561         if (graph->canceled) {
 562                 BT_LOGW("Invalid parameter: graph is canceled: "
 563                         "graph-addr=%p", graph);
 564                 status = BT_GRAPH_STATUS_CANCELED;
 565                 goto end;
 566         }
 567
 568         if (!graph->can_consume) {
 569                 BT_LOGW_STR("Cannot consume graph in its current state.");
 570                 status = BT_GRAPH_STATUS_CANNOT_CONSUME;
 571                 goto end;
 572         }
 573
 574         graph->can_consume = BT_FALSE;
 575         status = bt_graph_consume_no_check(graph);
 576         graph->can_consume = BT_TRUE;
 577
 578 end:
 579         return status;
 580 }
 581
 582 enum bt_graph_status bt_graph_run(struct bt_graph *graph)
 583 {
 584         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 585
 586         if (!graph) {
 587                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 588                 status = BT_GRAPH_STATUS_INVALID;
 589                 goto end;
 590         }
 591
 592         if (graph->canceled) {
 593                 BT_LOGW("Invalid parameter: graph is canceled: "
 594                         "graph-addr=%p", graph);
 595                 status = BT_GRAPH_STATUS_CANCELED;
 596                 goto end;
 597         }
 598
 599         if (!graph->can_consume) {
 600                 BT_LOGW_STR("Cannot run graph in its current state.");
 601                 status = BT_GRAPH_STATUS_CANNOT_CONSUME;
 602                 goto end;
 603         }
 604
 605         graph->can_consume = BT_FALSE;
 606         BT_LOGV("Running graph: addr=%p", graph);
 607
 608         do {
 609                 /*
 610                  * Check if the graph is canceled at each iteration. If
 611                  * the graph was canceled by another thread or by a
 612                  * signal, this is not a warning nor an error, it was
 613                  * intentional: log with a DEBUG level only.
 614                  */
 615                 if (graph->canceled) {
 616                         BT_LOGD("Stopping the graph: graph is canceled: "
 617                                 "graph-addr=%p", graph);
 618                         status = BT_GRAPH_STATUS_CANCELED;
 619                         goto end;
 620                 }
 621
 622                 status = bt_graph_consume_no_check(graph);
 623                 if (status == BT_GRAPH_STATUS_AGAIN) {
 624                         /*
 625                          * If AGAIN is received and there are multiple
 626                          * sinks, go ahead and consume from the next
 627                          * sink.
 628                          *
 629                          * However, in the case where a single sink is
 630                          * left, the caller can decide to busy-wait and
 631                          * call bt_graph_run() continuously until the
 632                          * source is ready or it can decide to sleep for
 633                          * an arbitrary amount of time.
 634                          */
 635                         if (graph->sinks_to_consume->length > 1) {
 636                                 status = BT_GRAPH_STATUS_OK;
 637                         }
 638                 } else if (status == BT_GRAPH_STATUS_NO_SINK) {
 639                         goto end;
 640                 }
 641         } while (status == BT_GRAPH_STATUS_OK);
 642
 643         if (g_queue_is_empty(graph->sinks_to_consume)) {
 644                 status = BT_GRAPH_STATUS_END;
 645         }
 646
 647 end:
 648         BT_LOGV("Graph ran: status=%s", bt_graph_status_string(status));
 649         if (graph) {
 650                 graph->can_consume = BT_TRUE;
 651         }
 652         return status;
 653 }
 654
 655 static
 656 int add_listener(GArray *listeners, void *func, void *removed, void *data)
 657 {
 658         struct bt_graph_listener listener = {
 659                 .func = func,
 660                 .removed = removed,
 661                 .data = data,
 662         };
 663
 664         g_array_append_val(listeners, listener);
 665         return listeners->len - 1;
 666 }
 667
 668 int bt_graph_add_port_added_listener(
 669                 struct bt_graph *graph,
 670                 bt_graph_port_added_listener listener,
 671                 bt_graph_listener_removed listener_removed, void *data)
 672 {
 673         int ret;
 674
 675         if (!graph) {
 676                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 677                 ret = -1;
 678                 goto end;
 679         }
 680
 681         if (graph->in_remove_listener) {
 682                 BT_LOGW("Cannot call this function during the execution of a remove listener: "
 683                         "addr=%p", graph);
 684                 ret = -1;
 685                 goto end;
 686         }
 687
 688         if (!listener) {
 689                 BT_LOGW_STR("Invalid parameter: listener is NULL.");
 690                 ret = -1;
 691                 goto end;
 692         }
 693
 694         ret = add_listener(graph->listeners.port_added, listener,
 695                 listener_removed, data);
 696         BT_LOGV("Added \"port added\" listener to graph: "
 697                 "graph-addr=%p, listener-addr=%p, pos=%d",
 698                 graph, listener, ret);
 699
 700 end:
 701         return ret;
 702 }
 703
 704 int bt_graph_add_port_removed_listener(
 705                 struct bt_graph *graph,
 706                 bt_graph_port_removed_listener listener,
 707                 bt_graph_listener_removed listener_removed, void *data)
 708 {
 709         int ret;
 710
 711         if (!graph) {
 712                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 713                 ret = -1;
 714                 goto end;
 715         }
 716
 717         if (graph->in_remove_listener) {
 718                 BT_LOGW("Cannot call this function during the execution of a remove listener: "
 719                         "addr=%p", graph);
 720                 ret = -1;
 721                 goto end;
 722         }
 723
 724         if (!listener) {
 725                 BT_LOGW_STR("Invalid parameter: listener is NULL.");
 726                 ret = -1;
 727                 goto end;
 728         }
 729
 730         ret = add_listener(graph->listeners.port_removed, listener,
 731                 listener_removed, data);
 732         BT_LOGV("Added \"port removed\" listener to graph: "
 733                 "graph-addr=%p, listener-addr=%p, pos=%d",
 734                 graph, listener, ret);
 735
 736 end:
 737         return ret;
 738 }
 739
 740 int bt_graph_add_ports_connected_listener(
 741                 struct bt_graph *graph,
 742                 bt_graph_ports_connected_listener listener,
 743                 bt_graph_listener_removed listener_removed, void *data)
 744 {
 745         int ret;
 746
 747         if (!graph) {
 748                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 749                 ret = -1;
 750                 goto end;
 751         }
 752
 753         if (graph->in_remove_listener) {
 754                 BT_LOGW("Cannot call this function during the execution of a remove listener: "
 755                         "addr=%p", graph);
 756                 ret = -1;
 757                 goto end;
 758         }
 759
 760         if (!listener) {
 761                 BT_LOGW_STR("Invalid parameter: listener is NULL.");
 762                 ret = -1;
 763                 goto end;
 764         }
 765
 766         ret = add_listener(graph->listeners.ports_connected, listener,
 767                 listener_removed, data);
 768         BT_LOGV("Added \"port connected\" listener to graph: "
 769                 "graph-addr=%p, listener-addr=%p, pos=%d",
 770                 graph, listener, ret);
 771
 772 end:
 773         return ret;
 774 }
 775
 776 int bt_graph_add_ports_disconnected_listener(
 777                 struct bt_graph *graph,
 778                 bt_graph_ports_disconnected_listener listener,
 779                 bt_graph_listener_removed listener_removed, void *data)
 780 {
 781         int ret;
 782
 783         if (!graph) {
 784                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 785                 ret = -1;
 786                 goto end;
 787         }
 788
 789         if (graph->in_remove_listener) {
 790                 BT_LOGW("Cannot call this function during the execution of a remove listener: "
 791                         "addr=%p", graph);
 792                 ret = -1;
 793                 goto end;
 794         }
 795
 796         if (!listener) {
 797                 BT_LOGW_STR("Invalid parameter: listener is NULL.");
 798                 ret = -1;
 799                 goto end;
 800         }
 801
 802         ret = add_listener(graph->listeners.ports_disconnected, listener,
 803                 listener_removed, data);
 804         BT_LOGV("Added \"port disconnected\" listener to graph: "
 805                 "graph-addr=%p, listener-addr=%p, pos=%d",
 806                 graph, listener, ret);
 807
 808 end:
 809         return ret;
 810 }
 811
 812 BT_HIDDEN
 813 void bt_graph_notify_port_added(struct bt_graph *graph, struct bt_port *port)
 814 {
 815         size_t i;
 816
 817         BT_LOGV("Notifying graph listeners that a port was added: "
 818                 "graph-addr=%p, port-addr=%p, port-name=\"%s\"",
 819                 graph, port, bt_port_get_name(port));
 820
 821         for (i = 0; i < graph->listeners.port_added->len; i++) {
 822                 struct bt_graph_listener listener =
 823                         g_array_index(graph->listeners.port_added,
 824                                 struct bt_graph_listener, i);
 825                 bt_graph_port_added_listener func = listener.func;
 826
 827                 assert(func);
 828                 func(port, listener.data);
 829         }
 830 }
 831
 832 BT_HIDDEN
 833 void bt_graph_notify_port_removed(struct bt_graph *graph,
 834                 struct bt_component *comp, struct bt_port *port)
 835 {
 836         size_t i;
 837
 838         BT_LOGV("Notifying graph listeners that a port was removed: "
 839                 "graph-addr=%p, port-addr=%p, port-name=\"%s\"",
 840                 graph, port, bt_port_get_name(port));
 841
 842         for (i = 0; i < graph->listeners.port_removed->len; i++) {
 843                 struct bt_graph_listener listener =
 844                         g_array_index(graph->listeners.port_removed,
 845                                 struct bt_graph_listener, i);
 846                 bt_graph_port_removed_listener func = listener.func;
 847
 848                 assert(func);
 849                 func(comp, port, listener.data);
 850         }
 851 }
 852
 853 BT_HIDDEN
 854 void bt_graph_notify_ports_connected(struct bt_graph *graph,
 855                 struct bt_port *upstream_port, struct bt_port *downstream_port)
 856 {
 857         size_t i;
 858
 859         BT_LOGV("Notifying graph listeners that two ports were connected: "
 860                 "graph-addr=%p, "
 861                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 862                 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
 863                 graph, upstream_port, bt_port_get_name(upstream_port),
 864                 downstream_port, bt_port_get_name(downstream_port));
 865
 866         for (i = 0; i < graph->listeners.ports_connected->len; i++) {
 867                 struct bt_graph_listener listener =
 868                         g_array_index(graph->listeners.ports_connected,
 869                                 struct bt_graph_listener, i);
 870                 bt_graph_ports_connected_listener func = listener.func;
 871
 872                 assert(func);
 873                 func(upstream_port, downstream_port, listener.data);
 874         }
 875 }
 876
 877 BT_HIDDEN
 878 void bt_graph_notify_ports_disconnected(struct bt_graph *graph,
 879                 struct bt_component *upstream_comp,
 880                 struct bt_component *downstream_comp,
 881                 struct bt_port *upstream_port, struct bt_port *downstream_port)
 882 {
 883         size_t i;
 884
 885         BT_LOGV("Notifying graph listeners that two ports were disconnected: "
 886                 "graph-addr=%p, "
 887                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 888                 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
 889                 graph, upstream_port, bt_port_get_name(upstream_port),
 890                 downstream_port, bt_port_get_name(downstream_port));
 891
 892         for (i = 0; i < graph->listeners.ports_disconnected->len; i++) {
 893                 struct bt_graph_listener listener =
 894                         g_array_index(graph->listeners.ports_disconnected,
 895                                 struct bt_graph_listener, i);
 896                 bt_graph_ports_disconnected_listener func = listener.func;
 897
 898                 assert(func);
 899                 func(upstream_comp, downstream_comp, upstream_port,
 900                         downstream_port, listener.data);
 901         }
 902 }
 903
 904 enum bt_graph_status bt_graph_cancel(struct bt_graph *graph)
 905 {
 906         enum bt_graph_status ret = BT_GRAPH_STATUS_OK;
 907
 908         if (!graph) {
 909                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 910                 ret = BT_GRAPH_STATUS_INVALID;
 911                 goto end;
 912         }
 913
 914         graph->canceled = BT_TRUE;
 915         BT_LOGV("Canceled graph: addr=%p", graph);
 916
 917 end:
 918         return ret;
 919 }
 920
 921 bt_bool bt_graph_is_canceled(struct bt_graph *graph)
 922 {
 923         bt_bool canceled = BT_FALSE;
 924
 925         if (!graph) {
 926                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 927                 goto end;
 928         }
 929
 930         canceled = graph->canceled;
 931
 932 end:
 933         return canceled;
 934 }
 935
 936 BT_HIDDEN
 937 void bt_graph_remove_connection(struct bt_graph *graph,
 938                 struct bt_connection *connection)
 939 {
 940         assert(graph);
 941         assert(connection);
 942         BT_LOGV("Removing graph's connection: graph-addr=%p, conn-addr=%p",
 943                 graph, connection);
 944         g_ptr_array_remove(graph->connections, connection);
 945 }
 946
 947 enum bt_graph_status bt_graph_add_component_with_init_method_data(
 948                 struct bt_graph *graph,
 949                 struct bt_component_class *component_class,
 950                 const char *name, struct bt_value *params,
 951                 void *init_method_data,
 952                 struct bt_component **user_component)
 953 {
 954         enum bt_graph_status graph_status = BT_GRAPH_STATUS_OK;
 955         enum bt_component_status comp_status;
 956         struct bt_component *component = NULL;
 957         enum bt_component_class_type type;
 958         size_t i;
 959         bt_bool init_can_consume;
 960
 961         bt_get(params);
 962
 963         if (!graph) {
 964                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 965                 graph_status = BT_GRAPH_STATUS_INVALID;
 966                 goto end;
 967         }
 968         init_can_consume = graph->can_consume;
 969
 970         if (!component_class) {
 971                 BT_LOGW_STR("Invalid parameter: component class is NULL.");
 972                 graph_status = BT_GRAPH_STATUS_INVALID;
 973                 goto end;
 974         }
 975
 976         graph->can_consume = BT_FALSE;
 977         type = bt_component_class_get_type(component_class);
 978         BT_LOGD("Adding component to graph: "
 979                 "graph-addr=%p, comp-cls-addr=%p, "
 980                 "comp-cls-type=%s, name=\"%s\", params-addr=%p, "
 981                 "init-method-data-addr=%p",
 982                 graph, component_class, bt_component_class_type_string(type),
 983                 name, params, init_method_data);
 984
 985         if (!name) {
 986                 BT_LOGW_STR("Invalid parameter: name is NULL.");
 987                 graph_status = BT_GRAPH_STATUS_INVALID;
 988                 goto end;
 989         }
 990
 991         if (graph->canceled) {
 992                 BT_LOGW_STR("Invalid parameter: graph is canceled.");
 993                 graph_status = BT_GRAPH_STATUS_CANCELED;
 994                 goto end;
 995         }
 996
 997         if (type != BT_COMPONENT_CLASS_TYPE_SOURCE &&
 998                         type != BT_COMPONENT_CLASS_TYPE_FILTER &&
 999                         type != BT_COMPONENT_CLASS_TYPE_SINK) {
1000                 BT_LOGW("Invalid parameter: unknown component class type: "
1001                         "type=%d", type);
1002                 graph_status = BT_GRAPH_STATUS_INVALID;
1003                 goto end;
1004         }
1005
1006         for (i = 0; i < graph->components->len; i++) {
1007                 void *other_comp = graph->components->pdata[i];
1008
1009                 if (strcmp(name, bt_component_get_name(other_comp)) == 0) {
1010                         BT_LOGW("Invalid parameter: another component with the same name already exists in the graph: "
1011                                 "other-comp-addr=%p, name=\"%s\"",
1012                                 other_comp, name);
1013                         graph_status = BT_GRAPH_STATUS_INVALID;
1014                         goto end;
1015                 }
1016         }
1017
1018         /*
1019          * Parameters must be a map value, but we create a convenient
1020          * empty one if it's NULL.
1021          */
1022         if (params) {
1023                 if (!bt_value_is_map(params)) {
1024                         BT_LOGW("Invalid parameter: initialization parameters must be a map value: "
1025                                 "type=%s",
1026                                 bt_value_type_string(bt_value_get_type(params)));
1027                         graph_status = BT_GRAPH_STATUS_INVALID;
1028                         goto end;
1029                 }
1030         } else {
1031                 params = bt_value_map_create();
1032                 if (!params) {
1033                         BT_LOGE_STR("Cannot create map value object.");
1034                         graph_status = BT_GRAPH_STATUS_NOMEM;
1035                         goto end;
1036                 }
1037         }
1038
1039         comp_status = bt_component_create(component_class, name, &component);
1040         if (comp_status != BT_COMPONENT_STATUS_OK) {
1041                 BT_LOGE("Cannot create empty component object: status=%s",
1042                         bt_component_status_string(comp_status));
1043                 graph_status = bt_graph_status_from_component_status(
1044                         comp_status);
1045                 goto end;
1046         }
1047
1048         /*
1049          * The user's initialization method needs to see that this
1050          * component is part of the graph. If the user method fails, we
1051          * immediately remove the component from the graph's components.
1052          */
1053         g_ptr_array_add(graph->components, component);
1054         bt_component_set_graph(component, graph);
1055
1056         if (component_class->methods.init) {
1057                 BT_LOGD_STR("Calling user's initialization method.");
1058                 comp_status = component_class->methods.init(
1059                         bt_private_component_from_component(component), params,
1060                         init_method_data);
1061                 BT_LOGD("User method returned: status=%s",
1062                         bt_component_status_string(comp_status));
1063                 if (comp_status != BT_COMPONENT_STATUS_OK) {
1064                         BT_LOGW_STR("Initialization method failed.");
1065                         graph_status = bt_graph_status_from_component_status(
1066                                 comp_status);
1067                         bt_component_set_graph(component, NULL);
1068                         g_ptr_array_remove_fast(graph->components, component);
1069                         goto end;
1070                 }
1071         }
1072
1073         /*
1074          * Mark the component as initialized so that its finalization
1075          * method is called when it is destroyed.
1076          */
1077         component->initialized = true;
1078
1079         /*
1080          * If it's a sink component, it needs to be part of the graph's
1081          * sink queue to be consumed by bt_graph_consume().
1082          */
1083         if (bt_component_is_sink(component)) {
1084                 graph->has_sink = BT_TRUE;
1085                 g_queue_push_tail(graph->sinks_to_consume, component);
1086         }
1087
1088         /*
1089          * Freeze the component class now that it's instantiated at
1090          * least once.
1091          */
1092         BT_LOGD_STR("Freezing component class.");
1093         bt_component_class_freeze(component->class);
1094         BT_LOGD("Added component to graph: "
1095                 "graph-addr=%p, comp-cls-addr=%p, "
1096                 "comp-cls-type=%s, name=\"%s\", params-addr=%p, "
1097                 "init-method-data-addr=%p, comp-addr=%p",
1098                 graph, component_class, bt_component_class_type_string(type),
1099                 name, params, init_method_data, component);
1100
1101         if (user_component) {
1102                 /* Move reference to user */
1103                 *user_component = component;
1104                 component = NULL;
1105         }
1106
1107 end:
1108         bt_put(component);
1109         bt_put(params);
1110         if (graph) {
1111                 graph->can_consume = init_can_consume;
1112         }
1113         return graph_status;
1114 }
1115
1116 enum bt_graph_status bt_graph_add_component(
1117                 struct bt_graph *graph,
1118                 struct bt_component_class *component_class,
1119                 const char *name, struct bt_value *params,
1120                 struct bt_component **component)
1121 {
1122         return bt_graph_add_component_with_init_method_data(graph,
1123                 component_class, name, params, NULL, component);
1124 }
1125
1126 BT_HIDDEN
1127 int bt_graph_remove_unconnected_component(struct bt_graph *graph,
1128                 struct bt_component *component)
1129 {
1130         bt_bool init_can_consume;
1131         int64_t count;
1132         uint64_t i;
1133         int ret = 0;
1134
1135         assert(graph);
1136         assert(component);
1137         assert(component->base.ref_count.count == 0);
1138         assert(bt_component_borrow_graph(component) == graph);
1139
1140         init_can_consume = graph->can_consume;
1141         count = bt_component_get_input_port_count(component);
1142
1143         for (i = 0; i < count; i++) {
1144                 struct bt_port *port =
1145                         bt_component_get_input_port_by_index(component, i);
1146
1147                 assert(port);
1148                 bt_put(port);
1149
1150                 if (bt_port_is_connected(port)) {
1151                         BT_LOGW("Cannot remove component from graph: "
1152                                 "an input port is connected: "
1153                                 "graph-addr=%p, comp-addr=%p, "
1154                                 "comp-name=\"%s\", connected-port-addr=%p, "
1155                                 "connected-port-name=\"%s\"",
1156                                 graph, component,
1157                                 bt_component_get_name(component),
1158                                 port, bt_port_get_name(port));
1159                         goto error;
1160                 }
1161         }
1162
1163         count = bt_component_get_output_port_count(component);
1164
1165         for (i = 0; i < count; i++) {
1166                 struct bt_port *port =
1167                         bt_component_get_output_port_by_index(component, i);
1168
1169                 assert(port);
1170                 bt_put(port);
1171
1172                 if (bt_port_is_connected(port)) {
1173                         BT_LOGW("Cannot remove component from graph: "
1174                                 "an output port is connected: "
1175                                 "graph-addr=%p, comp-addr=%p, "
1176                                 "comp-name=\"%s\", connected-port-addr=%p, "
1177                                 "connected-port-name=\"%s\"",
1178                                 graph, component,
1179                                 bt_component_get_name(component),
1180                                 port, bt_port_get_name(port));
1181                         goto error;
1182                 }
1183         }
1184
1185         graph->can_consume = BT_FALSE;
1186
1187         /* Possibly remove from sinks to consume */
1188         (void) g_queue_remove(graph->sinks_to_consume, component);
1189
1190         if (graph->sinks_to_consume->length == 0) {
1191                 graph->has_sink = BT_FALSE;
1192         }
1193
1194         /*
1195          * This calls bt_object_release() on the component, and since
1196          * its reference count is 0, its destructor is called. Its
1197          * destructor calls the user's finalization method (if set).
1198          */
1199         g_ptr_array_remove(graph->components, component);
1200         goto end;
1201
1202 error:
1203         ret = -1;
1204
1205 end:
1206         graph->can_consume = init_can_consume;
1207         return ret;
1208 }