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         const bt_bool init_can_consume = graph->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
 244         if (!upstream_port) {
 245                 BT_LOGW_STR("Invalid parameter: upstream port is NULL.");
 246                 status = BT_GRAPH_STATUS_INVALID;
 247                 goto end;
 248         }
 249
 250         if (!downstream_port) {
 251                 BT_LOGW_STR("Invalid parameter: downstream port is NULL.");
 252                 status = BT_GRAPH_STATUS_INVALID;
 253                 goto end;
 254         }
 255
 256         BT_LOGD("Connecting component ports within graph: "
 257                 "graph-addr=%p, "
 258                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 259                 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
 260                 graph, upstream_port, bt_port_get_name(upstream_port),
 261                 downstream_port, bt_port_get_name(downstream_port));
 262
 263         if (graph->canceled) {
 264                 BT_LOGW_STR("Invalid parameter: graph is canceled.");
 265                 status = BT_GRAPH_STATUS_CANCELED;
 266                 goto end;
 267         }
 268
 269         graph->can_consume = BT_FALSE;
 270
 271         /* Ensure appropriate types for upstream and downstream ports. */
 272         if (bt_port_get_type(upstream_port) != BT_PORT_TYPE_OUTPUT) {
 273                 BT_LOGW_STR("Invalid parameter: upstream port is not an output port.");
 274                 status = BT_GRAPH_STATUS_INVALID;
 275                 goto end;
 276         }
 277         if (bt_port_get_type(downstream_port) != BT_PORT_TYPE_INPUT) {
 278                 BT_LOGW_STR("Invalid parameter: downstream port is not an input port.");
 279                 status = BT_GRAPH_STATUS_INVALID;
 280                 goto end;
 281         }
 282
 283         /* Ensure that both ports are currently unconnected. */
 284         if (bt_port_is_connected(upstream_port)) {
 285                 BT_LOGW_STR("Invalid parameter: upstream port is already connected.");
 286                 status = BT_GRAPH_STATUS_INVALID;
 287                 goto end;
 288         }
 289
 290         if (bt_port_is_connected(downstream_port)) {
 291                 BT_LOGW_STR("Invalid parameter: downstream port is already connected.");
 292                 status = BT_GRAPH_STATUS_INVALID;
 293                 goto end;
 294         }
 295
 296         /*
 297          * Ensure that both ports are still attached to their creating
 298          * component.
 299          */
 300         upstream_component = bt_port_get_component(upstream_port);
 301         if (!upstream_component) {
 302                 BT_LOGW_STR("Invalid parameter: upstream port is loose (does not belong to a component)");
 303                 status = BT_GRAPH_STATUS_INVALID;
 304                 goto end;
 305         }
 306
 307         downstream_component = bt_port_get_component(downstream_port);
 308         if (!downstream_component) {
 309                 BT_LOGW_STR("Invalid parameter: downstream port is loose (does not belong to a component)");
 310                 status = BT_GRAPH_STATUS_INVALID;
 311                 goto end;
 312         }
 313
 314         BT_LOGD("Connecting component ports: "
 315                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 316                 "downstream-comp-addr=%p, downstream-comp-name=\"%s\"",
 317                 upstream_component, bt_component_get_name(upstream_component),
 318                 downstream_component, bt_component_get_name(downstream_component));
 319
 320         /*
 321          * At this point the ports are not connected yet. Both
 322          * components need to accept an eventual connection to their
 323          * port by the other port before we continue.
 324          */
 325         BT_LOGD_STR("Asking upstream component to accept the connection.");
 326         component_status = bt_component_accept_port_connection(
 327                 upstream_component, upstream_port, downstream_port);
 328         if (component_status != BT_COMPONENT_STATUS_OK) {
 329                 if (component_status == BT_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
 330                         BT_LOGD_STR("Upstream component refused the connection.");
 331                 } else {
 332                         BT_LOGW("Cannot ask upstream component to accept the connection: "
 333                                 "status=%s", bt_component_status_string(component_status));
 334                 }
 335
 336                 status = bt_graph_status_from_component_status(
 337                         component_status);
 338                 goto end;
 339         }
 340
 341         BT_LOGD_STR("Asking downstream component to accept the connection.");
 342         component_status = bt_component_accept_port_connection(
 343                 downstream_component, downstream_port, upstream_port);
 344         if (component_status != BT_COMPONENT_STATUS_OK) {
 345                 if (component_status == BT_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
 346                         BT_LOGD_STR("Downstream component refused the connection.");
 347                 } else {
 348                         BT_LOGW("Cannot ask downstream component to accept the connection: "
 349                                 "status=%s", bt_component_status_string(component_status));
 350                 }
 351
 352                 status = bt_graph_status_from_component_status(
 353                         component_status);
 354                 goto end;
 355         }
 356
 357         BT_LOGD_STR("Creating connection.");
 358         connection = bt_connection_create(graph, upstream_port,
 359                         downstream_port);
 360         if (!connection) {
 361                 BT_LOGW("Cannot create connection object.");
 362                 status = BT_GRAPH_STATUS_NOMEM;
 363                 goto end;
 364         }
 365
 366         BT_LOGD("Connection object created: conn-addr=%p", connection);
 367
 368         /*
 369          * Ownership of upstream_component/downstream_component and of
 370          * the connection object is transferred to the graph.
 371          */
 372         g_ptr_array_add(graph->connections, connection);
 373
 374         /*
 375          * Notify both components that their port is connected.
 376          */
 377         BT_LOGD_STR("Notifying upstream component that its port is connected.");
 378         bt_component_port_connected(upstream_component, upstream_port,
 379                 downstream_port);
 380         BT_LOGD_STR("Notifying downstream component that its port is connected.");
 381         bt_component_port_connected(downstream_component, downstream_port,
 382                 upstream_port);
 383
 384         /*
 385          * Notify the graph's creator that both ports are connected.
 386          */
 387         BT_LOGD_STR("Notifying graph's user that new component ports are connected.");
 388         bt_graph_notify_ports_connected(graph, upstream_port, downstream_port);
 389         BT_LOGD("Connected component ports within graph: "
 390                 "graph-addr=%p, "
 391                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 392                 "downstream-comp-addr=%p, downstream-comp-name=\"%s\", "
 393                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 394                 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
 395                 graph,
 396                 upstream_component, bt_component_get_name(upstream_component),
 397                 downstream_component, bt_component_get_name(downstream_component),
 398                 upstream_port, bt_port_get_name(upstream_port),
 399                 downstream_port, bt_port_get_name(downstream_port));
 400
 401         if (user_connection) {
 402                 /* Move reference to user */
 403                 *user_connection = connection;
 404                 connection = NULL;
 405         }
 406
 407 end:
 408         bt_put(upstream_graph);
 409         bt_put(downstream_graph);
 410         bt_put(upstream_component);
 411         bt_put(downstream_component);
 412         bt_put(connection);
 413         graph->can_consume = init_can_consume;
 414         return status;
 415 }
 416
 417 static
 418 enum bt_graph_status consume_graph_sink(struct bt_component *sink)
 419 {
 420         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 421         enum bt_component_status comp_status;
 422
 423         assert(sink);
 424         comp_status = bt_component_sink_consume(sink);
 425         BT_LOGV("Consumed from sink: addr=%p, name=\"%s\", status=%s",
 426                 sink, bt_component_get_name(sink),
 427                 bt_component_status_string(comp_status));
 428
 429         switch (comp_status) {
 430         case BT_COMPONENT_STATUS_OK:
 431                 break;
 432         case BT_COMPONENT_STATUS_END:
 433                 status = BT_GRAPH_STATUS_END;
 434                 break;
 435         case BT_COMPONENT_STATUS_AGAIN:
 436                 status = BT_GRAPH_STATUS_AGAIN;
 437                 break;
 438         case BT_COMPONENT_STATUS_INVALID:
 439                 status = BT_GRAPH_STATUS_INVALID;
 440                 break;
 441         default:
 442                 status = BT_GRAPH_STATUS_ERROR;
 443                 break;
 444         }
 445
 446         return status;
 447 }
 448
 449 /*
 450  * `node` is removed from the queue of sinks to consume when passed to
 451  * this function. This function adds it back to the queue if there's
 452  * still something to consume afterwards.
 453  */
 454 static
 455 enum bt_graph_status consume_sink_node(struct bt_graph *graph,
 456                 GList *node)
 457 {
 458         enum bt_graph_status status;
 459         struct bt_component *sink;
 460
 461         sink = node->data;
 462         status = consume_graph_sink(sink);
 463         if (status != BT_GRAPH_STATUS_END) {
 464                 g_queue_push_tail_link(graph->sinks_to_consume, node);
 465                 goto end;
 466         }
 467
 468         /* End reached, the node is not added back to the queue and free'd. */
 469         g_queue_delete_link(graph->sinks_to_consume, node);
 470
 471         /* Don't forward an END status if there are sinks left to consume. */
 472         if (!g_queue_is_empty(graph->sinks_to_consume)) {
 473                 status = BT_GRAPH_STATUS_OK;
 474                 goto end;
 475         }
 476
 477 end:
 478         BT_LOGV("Consumed sink node: status=%s", bt_graph_status_string(status));
 479         return status;
 480 }
 481
 482 BT_HIDDEN
 483 enum bt_graph_status bt_graph_consume_sink_no_check(struct bt_graph *graph,
 484                 struct bt_component *sink)
 485 {
 486         enum bt_graph_status status;
 487         GList *sink_node;
 488         int index;
 489
 490         BT_LOGV("Making specific sink consume: addr=%p, "
 491                 "comp-addr=%p, comp-name=\"%s\"",
 492                 graph, sink, bt_component_get_name(sink));
 493
 494         assert(bt_component_borrow_graph(sink) == graph);
 495
 496         if (g_queue_is_empty(graph->sinks_to_consume)) {
 497                 BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
 498                 status = BT_GRAPH_STATUS_END;
 499                 goto end;
 500         }
 501
 502         index = g_queue_index(graph->sinks_to_consume, sink);
 503         if (index < 0) {
 504                 BT_LOGV_STR("Sink is not marked as consumable: sink is ended.");
 505                 status = BT_GRAPH_STATUS_END;
 506                 goto end;
 507         }
 508
 509         sink_node = g_queue_pop_nth_link(graph->sinks_to_consume, index);
 510         assert(sink_node);
 511         status = consume_sink_node(graph, sink_node);
 512
 513 end:
 514         return status;
 515 }
 516
 517 BT_HIDDEN
 518 enum bt_graph_status bt_graph_consume_no_check(struct bt_graph *graph)
 519 {
 520         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 521         struct bt_component *sink;
 522         GList *current_node;
 523
 524         BT_LOGV("Making next sink consume: addr=%p", graph);
 525
 526         if (!graph->has_sink) {
 527                 BT_LOGW_STR("Graph has no sink component.");
 528                 status = BT_GRAPH_STATUS_NO_SINK;
 529                 goto end;
 530         }
 531
 532         if (g_queue_is_empty(graph->sinks_to_consume)) {
 533                 BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
 534                 status = BT_GRAPH_STATUS_END;
 535                 goto end;
 536         }
 537
 538         current_node = g_queue_pop_head_link(graph->sinks_to_consume);
 539         sink = current_node->data;
 540         BT_LOGV("Chose next sink to consume: comp-addr=%p, comp-name=\"%s\"",
 541                 sink, bt_component_get_name(sink));
 542         status = consume_sink_node(graph, current_node);
 543
 544 end:
 545         return status;
 546 }
 547
 548 enum bt_graph_status bt_graph_consume(struct bt_graph *graph)
 549 {
 550         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 551
 552         if (!graph) {
 553                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 554                 status = BT_GRAPH_STATUS_INVALID;
 555                 goto end;
 556         }
 557
 558         if (graph->canceled) {
 559                 BT_LOGW("Invalid parameter: graph is canceled: "
 560                         "graph-addr=%p", graph);
 561                 status = BT_GRAPH_STATUS_CANCELED;
 562                 goto end;
 563         }
 564
 565         if (!graph->can_consume) {
 566                 BT_LOGW_STR("Cannot consume graph in its current state.");
 567                 status = BT_GRAPH_STATUS_CANNOT_CONSUME;
 568                 goto end;
 569         }
 570
 571         graph->can_consume = BT_FALSE;
 572         status = bt_graph_consume_no_check(graph);
 573         graph->can_consume = BT_TRUE;
 574
 575 end:
 576         return status;
 577 }
 578
 579 enum bt_graph_status bt_graph_run(struct bt_graph *graph)
 580 {
 581         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 582
 583         if (!graph) {
 584                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 585                 status = BT_GRAPH_STATUS_INVALID;
 586                 goto end;
 587         }
 588
 589         if (graph->canceled) {
 590                 BT_LOGW("Invalid parameter: graph is canceled: "
 591                         "graph-addr=%p", graph);
 592                 status = BT_GRAPH_STATUS_CANCELED;
 593                 goto end;
 594         }
 595
 596         if (!graph->can_consume) {
 597                 BT_LOGW_STR("Cannot run graph in its current state.");
 598                 status = BT_GRAPH_STATUS_CANNOT_CONSUME;
 599                 goto end;
 600         }
 601
 602         graph->can_consume = BT_FALSE;
 603         BT_LOGV("Running graph: addr=%p", graph);
 604
 605         do {
 606                 /*
 607                  * Check if the graph is canceled at each iteration. If
 608                  * the graph was canceled by another thread or by a
 609                  * signal, this is not a warning nor an error, it was
 610                  * intentional: log with a DEBUG level only.
 611                  */
 612                 if (graph->canceled) {
 613                         BT_LOGD("Stopping the graph: graph is canceled: "
 614                                 "graph-addr=%p", graph);
 615                         status = BT_GRAPH_STATUS_CANCELED;
 616                         goto end;
 617                 }
 618
 619                 status = bt_graph_consume_no_check(graph);
 620                 if (status == BT_GRAPH_STATUS_AGAIN) {
 621                         /*
 622                          * If AGAIN is received and there are multiple
 623                          * sinks, go ahead and consume from the next
 624                          * sink.
 625                          *
 626                          * However, in the case where a single sink is
 627                          * left, the caller can decide to busy-wait and
 628                          * call bt_graph_run() continuously until the
 629                          * source is ready or it can decide to sleep for
 630                          * an arbitrary amount of time.
 631                          */
 632                         if (graph->sinks_to_consume->length > 1) {
 633                                 status = BT_GRAPH_STATUS_OK;
 634                         }
 635                 } else if (status == BT_GRAPH_STATUS_NO_SINK) {
 636                         goto end;
 637                 }
 638         } while (status == BT_GRAPH_STATUS_OK);
 639
 640         if (g_queue_is_empty(graph->sinks_to_consume)) {
 641                 status = BT_GRAPH_STATUS_END;
 642         }
 643
 644 end:
 645         BT_LOGV("Graph ran: status=%s", bt_graph_status_string(status));
 646         graph->can_consume = BT_TRUE;
 647         return status;
 648 }
 649
 650 static
 651 int add_listener(GArray *listeners, void *func, void *removed, void *data)
 652 {
 653         struct bt_graph_listener listener = {
 654                 .func = func,
 655                 .removed = removed,
 656                 .data = data,
 657         };
 658
 659         g_array_append_val(listeners, listener);
 660         return listeners->len - 1;
 661 }
 662
 663 int bt_graph_add_port_added_listener(
 664                 struct bt_graph *graph,
 665                 bt_graph_port_added_listener listener,
 666                 bt_graph_listener_removed listener_removed, void *data)
 667 {
 668         int ret;
 669
 670         if (!graph) {
 671                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 672                 ret = -1;
 673                 goto end;
 674         }
 675
 676         if (graph->in_remove_listener) {
 677                 BT_LOGW("Cannot call this function during the execution of a remove listener: "
 678                         "addr=%p", graph);
 679                 ret = -1;
 680                 goto end;
 681         }
 682
 683         if (!listener) {
 684                 BT_LOGW_STR("Invalid parameter: listener is NULL.");
 685                 ret = -1;
 686                 goto end;
 687         }
 688
 689         ret = add_listener(graph->listeners.port_added, listener,
 690                 listener_removed, data);
 691         BT_LOGV("Added \"port added\" listener to graph: "
 692                 "graph-addr=%p, listener-addr=%p, pos=%d",
 693                 graph, listener, ret);
 694
 695 end:
 696         return ret;
 697 }
 698
 699 int bt_graph_add_port_removed_listener(
 700                 struct bt_graph *graph,
 701                 bt_graph_port_removed_listener listener,
 702                 bt_graph_listener_removed listener_removed, void *data)
 703 {
 704         int ret;
 705
 706         if (!graph) {
 707                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 708                 ret = -1;
 709                 goto end;
 710         }
 711
 712         if (graph->in_remove_listener) {
 713                 BT_LOGW("Cannot call this function during the execution of a remove listener: "
 714                         "addr=%p", graph);
 715                 ret = -1;
 716                 goto end;
 717         }
 718
 719         if (!listener) {
 720                 BT_LOGW_STR("Invalid parameter: listener is NULL.");
 721                 ret = -1;
 722                 goto end;
 723         }
 724
 725         ret = add_listener(graph->listeners.port_removed, listener,
 726                 listener_removed, data);
 727         BT_LOGV("Added \"port removed\" listener to graph: "
 728                 "graph-addr=%p, listener-addr=%p, pos=%d",
 729                 graph, listener, ret);
 730
 731 end:
 732         return ret;
 733 }
 734
 735 int bt_graph_add_ports_connected_listener(
 736                 struct bt_graph *graph,
 737                 bt_graph_ports_connected_listener listener,
 738                 bt_graph_listener_removed listener_removed, void *data)
 739 {
 740         int ret;
 741
 742         if (!graph) {
 743                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 744                 ret = -1;
 745                 goto end;
 746         }
 747
 748         if (graph->in_remove_listener) {
 749                 BT_LOGW("Cannot call this function during the execution of a remove listener: "
 750                         "addr=%p", graph);
 751                 ret = -1;
 752                 goto end;
 753         }
 754
 755         if (!listener) {
 756                 BT_LOGW_STR("Invalid parameter: listener is NULL.");
 757                 ret = -1;
 758                 goto end;
 759         }
 760
 761         ret = add_listener(graph->listeners.ports_connected, listener,
 762                 listener_removed, data);
 763         BT_LOGV("Added \"port connected\" listener to graph: "
 764                 "graph-addr=%p, listener-addr=%p, pos=%d",
 765                 graph, listener, ret);
 766
 767 end:
 768         return ret;
 769 }
 770
 771 int bt_graph_add_ports_disconnected_listener(
 772                 struct bt_graph *graph,
 773                 bt_graph_ports_disconnected_listener listener,
 774                 bt_graph_listener_removed listener_removed, void *data)
 775 {
 776         int ret;
 777
 778         if (!graph) {
 779                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 780                 ret = -1;
 781                 goto end;
 782         }
 783
 784         if (graph->in_remove_listener) {
 785                 BT_LOGW("Cannot call this function during the execution of a remove listener: "
 786                         "addr=%p", graph);
 787                 ret = -1;
 788                 goto end;
 789         }
 790
 791         if (!listener) {
 792                 BT_LOGW_STR("Invalid parameter: listener is NULL.");
 793                 ret = -1;
 794                 goto end;
 795         }
 796
 797         ret = add_listener(graph->listeners.ports_disconnected, listener,
 798                 listener_removed, data);
 799         BT_LOGV("Added \"port disconnected\" listener to graph: "
 800                 "graph-addr=%p, listener-addr=%p, pos=%d",
 801                 graph, listener, ret);
 802
 803 end:
 804         return ret;
 805 }
 806
 807 BT_HIDDEN
 808 void bt_graph_notify_port_added(struct bt_graph *graph, struct bt_port *port)
 809 {
 810         size_t i;
 811
 812         BT_LOGV("Notifying graph listeners that a port was added: "
 813                 "graph-addr=%p, port-addr=%p, port-name=\"%s\"",
 814                 graph, port, bt_port_get_name(port));
 815
 816         for (i = 0; i < graph->listeners.port_added->len; i++) {
 817                 struct bt_graph_listener listener =
 818                         g_array_index(graph->listeners.port_added,
 819                                 struct bt_graph_listener, i);
 820                 bt_graph_port_added_listener func = listener.func;
 821
 822                 assert(func);
 823                 func(port, listener.data);
 824         }
 825 }
 826
 827 BT_HIDDEN
 828 void bt_graph_notify_port_removed(struct bt_graph *graph,
 829                 struct bt_component *comp, struct bt_port *port)
 830 {
 831         size_t i;
 832
 833         BT_LOGV("Notifying graph listeners that a port was removed: "
 834                 "graph-addr=%p, port-addr=%p, port-name=\"%s\"",
 835                 graph, port, bt_port_get_name(port));
 836
 837         for (i = 0; i < graph->listeners.port_removed->len; i++) {
 838                 struct bt_graph_listener listener =
 839                         g_array_index(graph->listeners.port_removed,
 840                                 struct bt_graph_listener, i);
 841                 bt_graph_port_removed_listener func = listener.func;
 842
 843                 assert(func);
 844                 func(comp, port, listener.data);
 845         }
 846 }
 847
 848 BT_HIDDEN
 849 void bt_graph_notify_ports_connected(struct bt_graph *graph,
 850                 struct bt_port *upstream_port, struct bt_port *downstream_port)
 851 {
 852         size_t i;
 853
 854         BT_LOGV("Notifying graph listeners that two ports were connected: "
 855                 "graph-addr=%p, "
 856                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 857                 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
 858                 graph, upstream_port, bt_port_get_name(upstream_port),
 859                 downstream_port, bt_port_get_name(downstream_port));
 860
 861         for (i = 0; i < graph->listeners.ports_connected->len; i++) {
 862                 struct bt_graph_listener listener =
 863                         g_array_index(graph->listeners.ports_connected,
 864                                 struct bt_graph_listener, i);
 865                 bt_graph_ports_connected_listener func = listener.func;
 866
 867                 assert(func);
 868                 func(upstream_port, downstream_port, listener.data);
 869         }
 870 }
 871
 872 BT_HIDDEN
 873 void bt_graph_notify_ports_disconnected(struct bt_graph *graph,
 874                 struct bt_component *upstream_comp,
 875                 struct bt_component *downstream_comp,
 876                 struct bt_port *upstream_port, struct bt_port *downstream_port)
 877 {
 878         size_t i;
 879
 880         BT_LOGV("Notifying graph listeners that two ports were disconnected: "
 881                 "graph-addr=%p, "
 882                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 883                 "downstream-port-addr=%p, downstream-port-name=\"%s\"",
 884                 graph, upstream_port, bt_port_get_name(upstream_port),
 885                 downstream_port, bt_port_get_name(downstream_port));
 886
 887         for (i = 0; i < graph->listeners.ports_disconnected->len; i++) {
 888                 struct bt_graph_listener listener =
 889                         g_array_index(graph->listeners.ports_disconnected,
 890                                 struct bt_graph_listener, i);
 891                 bt_graph_ports_disconnected_listener func = listener.func;
 892
 893                 assert(func);
 894                 func(upstream_comp, downstream_comp, upstream_port,
 895                         downstream_port, listener.data);
 896         }
 897 }
 898
 899 enum bt_graph_status bt_graph_cancel(struct bt_graph *graph)
 900 {
 901         enum bt_graph_status ret = BT_GRAPH_STATUS_OK;
 902
 903         if (!graph) {
 904                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 905                 ret = BT_GRAPH_STATUS_INVALID;
 906                 goto end;
 907         }
 908
 909         graph->canceled = BT_TRUE;
 910         BT_LOGV("Canceled graph: addr=%p", graph);
 911
 912 end:
 913         return ret;
 914 }
 915
 916 bt_bool bt_graph_is_canceled(struct bt_graph *graph)
 917 {
 918         bt_bool canceled = BT_FALSE;
 919
 920         if (!graph) {
 921                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 922                 goto end;
 923         }
 924
 925         canceled = graph->canceled;
 926
 927 end:
 928         return canceled;
 929 }
 930
 931 BT_HIDDEN
 932 void bt_graph_remove_connection(struct bt_graph *graph,
 933                 struct bt_connection *connection)
 934 {
 935         assert(graph);
 936         assert(connection);
 937         BT_LOGV("Removing graph's connection: graph-addr=%p, conn-addr=%p",
 938                 graph, connection);
 939         g_ptr_array_remove(graph->connections, connection);
 940 }
 941
 942 enum bt_graph_status bt_graph_add_component_with_init_method_data(
 943                 struct bt_graph *graph,
 944                 struct bt_component_class *component_class,
 945                 const char *name, struct bt_value *params,
 946                 void *init_method_data,
 947                 struct bt_component **user_component)
 948 {
 949         enum bt_graph_status graph_status = BT_GRAPH_STATUS_OK;
 950         enum bt_component_status comp_status;
 951         struct bt_component *component = NULL;
 952         enum bt_component_class_type type;
 953         size_t i;
 954         const bt_bool init_can_consume = graph->can_consume;
 955
 956         bt_get(params);
 957
 958         if (!graph) {
 959                 BT_LOGW_STR("Invalid parameter: graph is NULL.");
 960                 graph_status = BT_GRAPH_STATUS_INVALID;
 961                 goto end;
 962         }
 963
 964         if (!component_class) {
 965                 BT_LOGW_STR("Invalid parameter: component class is NULL.");
 966                 graph_status = BT_GRAPH_STATUS_INVALID;
 967                 goto end;
 968         }
 969
 970         graph->can_consume = BT_FALSE;
 971         type = bt_component_class_get_type(component_class);
 972         BT_LOGD("Adding component to graph: "
 973                 "graph-addr=%p, comp-cls-addr=%p, "
 974                 "comp-cls-type=%s, name=\"%s\", params-addr=%p, "
 975                 "init-method-data-addr=%p",
 976                 graph, component_class, bt_component_class_type_string(type),
 977                 name, params, init_method_data);
 978
 979         if (!name) {
 980                 BT_LOGW_STR("Invalid parameter: name is NULL.");
 981                 graph_status = BT_GRAPH_STATUS_INVALID;
 982                 goto end;
 983         }
 984
 985         if (graph->canceled) {
 986                 BT_LOGW_STR("Invalid parameter: graph is canceled.");
 987                 graph_status = BT_GRAPH_STATUS_CANCELED;
 988                 goto end;
 989         }
 990
 991         if (type != BT_COMPONENT_CLASS_TYPE_SOURCE &&
 992                         type != BT_COMPONENT_CLASS_TYPE_FILTER &&
 993                         type != BT_COMPONENT_CLASS_TYPE_SINK) {
 994                 BT_LOGW("Invalid parameter: unknown component class type: "
 995                         "type=%d", type);
 996                 graph_status = BT_GRAPH_STATUS_INVALID;
 997                 goto end;
 998         }
 999
1000         for (i = 0; i < graph->components->len; i++) {
1001                 void *other_comp = graph->components->pdata[i];
1002
1003                 if (strcmp(name, bt_component_get_name(other_comp)) == 0) {
1004                         BT_LOGW("Invalid parameter: another component with the same name already exists in the graph: "
1005                                 "other-comp-addr=%p, name=\"%s\"",
1006                                 other_comp, name);
1007                         graph_status = BT_GRAPH_STATUS_INVALID;
1008                         goto end;
1009                 }
1010         }
1011
1012         /*
1013          * Parameters must be a map value, but we create a convenient
1014          * empty one if it's NULL.
1015          */
1016         if (params) {
1017                 if (!bt_value_is_map(params)) {
1018                         BT_LOGW("Invalid parameter: initialization parameters must be a map value: "
1019                                 "type=%s",
1020                                 bt_value_type_string(bt_value_get_type(params)));
1021                         graph_status = BT_GRAPH_STATUS_INVALID;
1022                         goto end;
1023                 }
1024         } else {
1025                 params = bt_value_map_create();
1026                 if (!params) {
1027                         BT_LOGE_STR("Cannot create map value object.");
1028                         graph_status = BT_GRAPH_STATUS_NOMEM;
1029                         goto end;
1030                 }
1031         }
1032
1033         comp_status = bt_component_create(component_class, name, &component);
1034         if (comp_status != BT_COMPONENT_STATUS_OK) {
1035                 BT_LOGE("Cannot create empty component object: status=%s",
1036                         bt_component_status_string(comp_status));
1037                 graph_status = bt_graph_status_from_component_status(
1038                         comp_status);
1039                 goto end;
1040         }
1041
1042         /*
1043          * The user's initialization method needs to see that this
1044          * component is part of the graph. If the user method fails, we
1045          * immediately remove the component from the graph's components.
1046          */
1047         g_ptr_array_add(graph->components, component);
1048         bt_component_set_graph(component, graph);
1049
1050         if (component_class->methods.init) {
1051                 BT_LOGD_STR("Calling user's initialization method.");
1052                 comp_status = component_class->methods.init(
1053                         bt_private_component_from_component(component), params,
1054                         init_method_data);
1055                 BT_LOGD("User method returned: status=%s",
1056                         bt_component_status_string(comp_status));
1057                 if (comp_status != BT_COMPONENT_STATUS_OK) {
1058                         BT_LOGW_STR("Initialization method failed.");
1059                         graph_status = bt_graph_status_from_component_status(
1060                                 comp_status);
1061                         bt_component_set_graph(component, NULL);
1062                         g_ptr_array_remove_fast(graph->components, component);
1063                         goto end;
1064                 }
1065         }
1066
1067         /*
1068          * Mark the component as initialized so that its finalization
1069          * method is called when it is destroyed.
1070          */
1071         component->initialized = true;
1072
1073         /*
1074          * If it's a sink component, it needs to be part of the graph's
1075          * sink queue to be consumed by bt_graph_consume().
1076          */
1077         if (bt_component_is_sink(component)) {
1078                 graph->has_sink = BT_TRUE;
1079                 g_queue_push_tail(graph->sinks_to_consume, component);
1080         }
1081
1082         /*
1083          * Freeze the component class now that it's instantiated at
1084          * least once.
1085          */
1086         BT_LOGD_STR("Freezing component class.");
1087         bt_component_class_freeze(component->class);
1088         BT_LOGD("Added component to graph: "
1089                 "graph-addr=%p, comp-cls-addr=%p, "
1090                 "comp-cls-type=%s, name=\"%s\", params-addr=%p, "
1091                 "init-method-data-addr=%p, comp-addr=%p",
1092                 graph, component_class, bt_component_class_type_string(type),
1093                 name, params, init_method_data, component);
1094
1095         if (user_component) {
1096                 /* Move reference to user */
1097                 *user_component = component;
1098                 component = NULL;
1099         }
1100
1101 end:
1102         bt_put(component);
1103         bt_put(params);
1104         graph->can_consume = init_can_consume;
1105         return graph_status;
1106 }
1107
1108 enum bt_graph_status bt_graph_add_component(
1109                 struct bt_graph *graph,
1110                 struct bt_component_class *component_class,
1111                 const char *name, struct bt_value *params,
1112                 struct bt_component **component)
1113 {
1114         return bt_graph_add_component_with_init_method_data(graph,
1115                 component_class, name, params, NULL, component);
1116 }
1117
1118 BT_HIDDEN
1119 int bt_graph_remove_unconnected_component(struct bt_graph *graph,
1120                 struct bt_component *component)
1121 {
1122         const bt_bool init_can_consume = graph->can_consume;
1123         int64_t count;
1124         uint64_t i;
1125         int ret = 0;
1126
1127         assert(graph);
1128         assert(component);
1129         assert(component->base.ref_count.count == 0);
1130         assert(bt_component_borrow_graph(component) == graph);
1131
1132         count = bt_component_get_input_port_count(component);
1133
1134         for (i = 0; i < count; i++) {
1135                 struct bt_port *port =
1136                         bt_component_get_input_port_by_index(component, i);
1137
1138                 assert(port);
1139                 bt_put(port);
1140
1141                 if (bt_port_is_connected(port)) {
1142                         BT_LOGW("Cannot remove component from graph: "
1143                                 "an input port is connected: "
1144                                 "graph-addr=%p, comp-addr=%p, "
1145                                 "comp-name=\"%s\", connected-port-addr=%p, "
1146                                 "connected-port-name=\"%s\"",
1147                                 graph, component,
1148                                 bt_component_get_name(component),
1149                                 port, bt_port_get_name(port));
1150                         goto error;
1151                 }
1152         }
1153
1154         count = bt_component_get_output_port_count(component);
1155
1156         for (i = 0; i < count; i++) {
1157                 struct bt_port *port =
1158                         bt_component_get_output_port_by_index(component, i);
1159
1160                 assert(port);
1161                 bt_put(port);
1162
1163                 if (bt_port_is_connected(port)) {
1164                         BT_LOGW("Cannot remove component from graph: "
1165                                 "an output port is connected: "
1166                                 "graph-addr=%p, comp-addr=%p, "
1167                                 "comp-name=\"%s\", connected-port-addr=%p, "
1168                                 "connected-port-name=\"%s\"",
1169                                 graph, component,
1170                                 bt_component_get_name(component),
1171                                 port, bt_port_get_name(port));
1172                         goto error;
1173                 }
1174         }
1175
1176         graph->can_consume = BT_FALSE;
1177
1178         /* Possibly remove from sinks to consume */
1179         (void) g_queue_remove(graph->sinks_to_consume, component);
1180
1181         if (graph->sinks_to_consume->length == 0) {
1182                 graph->has_sink = BT_FALSE;
1183         }
1184
1185         /*
1186          * This calls bt_object_release() on the component, and since
1187          * its reference count is 0, its destructor is called. Its
1188          * destructor calls the user's finalization method (if set).
1189          */
1190         g_ptr_array_remove(graph->components, component);
1191         goto end;
1192
1193 error:
1194         ret = -1;
1195
1196 end:
1197         graph->can_consume = init_can_consume;
1198         return ret;
1199 }