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