src/lib/graph/graph.c

   1 /*
   2  * Copyright 2017-2018 Philippe Proulx <pproulx@efficios.com>
   3  * Copyright 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a copy
   6  * of this software and associated documentation files (the "Software"), to deal
   7  * in the Software without restriction, including without limitation the rights
   8  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9  * copies of the Software, and to permit persons to whom the Software is
  10  * furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  21  * SOFTWARE.
  22  */
  23
  24 #define BT_LOG_TAG "LIB/GRAPH"
  25 #include "lib/logging.h"
  26
  27 #include "common/assert.h"
  28 #include "lib/assert-pre.h"
  29 #include "lib/assert-post.h"
  30 #include <babeltrace2/graph/graph.h>
  31 #include <babeltrace2/graph/graph-const.h>
  32 #include <babeltrace2/graph/component-source-const.h>
  33 #include <babeltrace2/graph/component-filter-const.h>
  34 #include <babeltrace2/graph/port-const.h>
  35 #include "lib/graph/message/message.h"
  36 #include "compat/compiler.h"
  37 #include "common/common.h"
  38 #include <babeltrace2/types.h>
  39 #include <babeltrace2/value.h>
  40 #include <babeltrace2/value-const.h>
  41 #include "lib/value.h"
  42 #include <unistd.h>
  43 #include <stdbool.h>
  44 #include <glib.h>
  45
  46 #include "component-class-sink-simple.h"
  47 #include "component.h"
  48 #include "component-sink.h"
  49 #include "connection.h"
  50 #include "graph.h"
  51 #include "interrupter.h"
  52 #include "message/event.h"
  53 #include "message/packet.h"
  54
  55 typedef enum bt_graph_listener_func_status
  56 (*port_added_func_t)(const void *, const void *, void *);
  57
  58 typedef enum bt_component_class_initialize_method_status
  59 (*comp_init_method_t)(const void *, void *, const void *, void *);
  60
  61 struct bt_graph_listener {
  62         bt_graph_listener_removed_func removed;
  63         void *data;
  64 };
  65
  66 struct bt_graph_listener_port_added {
  67         struct bt_graph_listener base;
  68         port_added_func_t func;
  69 };
  70
  71 #define INIT_LISTENERS_ARRAY(_type, _listeners)                         \
  72         do {                                                            \
  73                 _listeners = g_array_new(FALSE, TRUE, sizeof(_type));   \
  74                 if (!(_listeners)) {                                    \
  75                         BT_LIB_LOGE_APPEND_CAUSE(                       \
  76                                 "Failed to allocate one GArray.");      \
  77                 }                                                       \
  78         } while (0)
  79
  80 #define CALL_REMOVE_LISTENERS(_type, _listeners)                        \
  81         do {                                                            \
  82                 size_t i;                                               \
  83                                                                         \
  84                 if (!_listeners) {                                      \
  85                         break;                                          \
  86                 }                                                       \
  87                 for (i = 0; i < (_listeners)->len; i++) {               \
  88                         _type *listener =                               \
  89                                 &g_array_index((_listeners), _type, i); \
  90                                                                         \
  91                         if (listener->base.removed) {                   \
  92                                 listener->base.removed(listener->base.data); \
  93                         }                                               \
  94                 }                                                       \
  95         } while (0)
  96
  97 static
  98 void destroy_graph(struct bt_object *obj)
  99 {
 100         struct bt_graph *graph = container_of(obj, struct bt_graph, base);
 101
 102         /*
 103          * The graph's reference count is 0 if we're here. Increment
 104          * it to avoid a double-destroy (possibly infinitely recursive)
 105          * in this situation:
 106          *
 107          * 1. We put and destroy a connection.
 108          * 2. This connection's destructor finalizes its active message
 109          *    iterators.
 110          * 3. A message iterator's finalization function gets a new
 111          *    reference on its component (reference count goes from 0 to
 112          *    1).
 113          * 4. Since this component's reference count goes to 1, it takes
 114          *    a reference on its parent (this graph). This graph's
 115          *    reference count goes from 0 to 1.
 116          * 5. The message iterator's finalization function puts its
 117          *    component reference (reference count goes from 1 to 0).
 118          * 6. Since this component's reference count goes from 1 to 0,
 119          *    it puts its parent (this graph). This graph's reference
 120          *    count goes from 1 to 0.
 121          * 7. Since this graph's reference count goes from 1 to 0, its
 122          *    destructor is called (this function).
 123          *
 124          * With the incrementation below, the graph's reference count at
 125          * step 4 goes from 1 to 2, and from 2 to 1 at step 6. This
 126          * ensures that this function is not called two times.
 127          */
 128         BT_LIB_LOGI("Destroying graph: %!+g", graph);
 129         obj->ref_count++;
 130         graph->config_state = BT_GRAPH_CONFIGURATION_STATE_DESTROYING;
 131
 132         /* Call all remove listeners */
 133         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
 134                 graph->listeners.source_output_port_added);
 135         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
 136                 graph->listeners.filter_output_port_added);
 137         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
 138                 graph->listeners.filter_input_port_added);
 139         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
 140                 graph->listeners.sink_input_port_added);
 141
 142         if (graph->messages) {
 143                 g_ptr_array_free(graph->messages, TRUE);
 144                 graph->messages = NULL;
 145         }
 146
 147         if (graph->connections) {
 148                 BT_LOGD_STR("Destroying connections.");
 149                 g_ptr_array_free(graph->connections, TRUE);
 150                 graph->connections = NULL;
 151         }
 152
 153         if (graph->components) {
 154                 BT_LOGD_STR("Destroying components.");
 155                 g_ptr_array_free(graph->components, TRUE);
 156                 graph->components = NULL;
 157         }
 158
 159         if (graph->interrupters) {
 160                 BT_LOGD_STR("Putting interrupters.");
 161                 g_ptr_array_free(graph->interrupters, TRUE);
 162                 graph->interrupters = NULL;
 163         }
 164
 165         BT_OBJECT_PUT_REF_AND_RESET(graph->default_interrupter);
 166
 167         if (graph->sinks_to_consume) {
 168                 g_queue_free(graph->sinks_to_consume);
 169                 graph->sinks_to_consume = NULL;
 170         }
 171
 172         if (graph->listeners.source_output_port_added) {
 173                 g_array_free(graph->listeners.source_output_port_added, TRUE);
 174                 graph->listeners.source_output_port_added = NULL;
 175         }
 176
 177         if (graph->listeners.filter_output_port_added) {
 178                 g_array_free(graph->listeners.filter_output_port_added, TRUE);
 179                 graph->listeners.filter_output_port_added = NULL;
 180         }
 181
 182         if (graph->listeners.filter_input_port_added) {
 183                 g_array_free(graph->listeners.filter_input_port_added, TRUE);
 184                 graph->listeners.filter_input_port_added = NULL;
 185         }
 186
 187         if (graph->listeners.sink_input_port_added) {
 188                 g_array_free(graph->listeners.sink_input_port_added, TRUE);
 189                 graph->listeners.sink_input_port_added = NULL;
 190         }
 191
 192         bt_object_pool_finalize(&graph->event_msg_pool);
 193         bt_object_pool_finalize(&graph->packet_begin_msg_pool);
 194         bt_object_pool_finalize(&graph->packet_end_msg_pool);
 195         g_free(graph);
 196 }
 197
 198 static
 199 void destroy_message_event(struct bt_message *msg,
 200                 struct bt_graph *graph)
 201 {
 202         bt_message_event_destroy(msg);
 203 }
 204
 205 static
 206 void destroy_message_packet_begin(struct bt_message *msg,
 207                 struct bt_graph *graph)
 208 {
 209         bt_message_packet_destroy(msg);
 210 }
 211
 212 static
 213 void destroy_message_packet_end(struct bt_message *msg,
 214                 struct bt_graph *graph)
 215 {
 216         bt_message_packet_destroy(msg);
 217 }
 218
 219 static
 220 void notify_message_graph_is_destroyed(struct bt_message *msg)
 221 {
 222         bt_message_unlink_graph(msg);
 223 }
 224
 225 struct bt_graph *bt_graph_create(uint64_t mip_version)
 226 {
 227         struct bt_graph *graph;
 228         int ret;
 229
 230         BT_ASSERT_PRE_NO_ERROR();
 231         BT_ASSERT_PRE(mip_version <= bt_get_maximal_mip_version(),
 232                 "Unknown MIP version: mip-version=%" PRIu64 ", "
 233                 "max-mip-version=%" PRIu64,
 234                 mip_version, bt_get_maximal_mip_version());
 235         BT_LOGI_STR("Creating graph object.");
 236         graph = g_new0(struct bt_graph, 1);
 237         if (!graph) {
 238                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate one graph.");
 239                 goto end;
 240         }
 241
 242         bt_object_init_shared(&graph->base, destroy_graph);
 243         graph->mip_version = mip_version;
 244         graph->connections = g_ptr_array_new_with_free_func(
 245                 (GDestroyNotify) bt_object_try_spec_release);
 246         if (!graph->connections) {
 247                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate one GPtrArray.");
 248                 goto error;
 249         }
 250         graph->components = g_ptr_array_new_with_free_func(
 251                 (GDestroyNotify) bt_object_try_spec_release);
 252         if (!graph->components) {
 253                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate one GPtrArray.");
 254                 goto error;
 255         }
 256         graph->sinks_to_consume = g_queue_new();
 257         if (!graph->sinks_to_consume) {
 258                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate one GQueue.");
 259                 goto error;
 260         }
 261
 262         bt_graph_set_can_consume(graph, true);
 263         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
 264                 graph->listeners.source_output_port_added);
 265
 266         if (!graph->listeners.source_output_port_added) {
 267                 goto error;
 268         }
 269
 270         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
 271                 graph->listeners.filter_output_port_added);
 272
 273         if (!graph->listeners.filter_output_port_added) {
 274                 goto error;
 275         }
 276
 277         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
 278                 graph->listeners.filter_input_port_added);
 279
 280         if (!graph->listeners.filter_input_port_added) {
 281                 goto error;
 282         }
 283
 284         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
 285                 graph->listeners.sink_input_port_added);
 286
 287         if (!graph->listeners.sink_input_port_added) {
 288                 goto error;
 289         }
 290
 291         graph->interrupters = g_ptr_array_new_with_free_func(
 292                 (GDestroyNotify) bt_object_put_ref_no_null_check);
 293         if (!graph->interrupters) {
 294                 BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate one GPtrArray.");
 295                 goto error;
 296         }
 297
 298         graph->default_interrupter = bt_interrupter_create();
 299         if (!graph->default_interrupter) {
 300                 BT_LIB_LOGE_APPEND_CAUSE(
 301                         "Failed to create one interrupter object.");
 302                 goto error;
 303         }
 304
 305         bt_graph_add_interrupter(graph, graph->default_interrupter);
 306         ret = bt_object_pool_initialize(&graph->event_msg_pool,
 307                 (bt_object_pool_new_object_func) bt_message_event_new,
 308                 (bt_object_pool_destroy_object_func) destroy_message_event,
 309                 graph);
 310         if (ret) {
 311                 BT_LIB_LOGE_APPEND_CAUSE(
 312                         "Failed to initialize event message pool: ret=%d",
 313                         ret);
 314                 goto error;
 315         }
 316
 317         ret = bt_object_pool_initialize(&graph->packet_begin_msg_pool,
 318                 (bt_object_pool_new_object_func) bt_message_packet_beginning_new,
 319                 (bt_object_pool_destroy_object_func) destroy_message_packet_begin,
 320                 graph);
 321         if (ret) {
 322                 BT_LIB_LOGE_APPEND_CAUSE(
 323                         "Failed to initialize packet beginning message pool: ret=%d",
 324                         ret);
 325                 goto error;
 326         }
 327
 328         ret = bt_object_pool_initialize(&graph->packet_end_msg_pool,
 329                 (bt_object_pool_new_object_func) bt_message_packet_end_new,
 330                 (bt_object_pool_destroy_object_func) destroy_message_packet_end,
 331                 graph);
 332         if (ret) {
 333                 BT_LIB_LOGE_APPEND_CAUSE(
 334                         "Failed to initialize packet end message pool: ret=%d",
 335                         ret);
 336                 goto error;
 337         }
 338
 339         graph->messages = g_ptr_array_new_with_free_func(
 340                 (GDestroyNotify) notify_message_graph_is_destroyed);
 341         BT_LIB_LOGI("Created graph object: %!+g", graph);
 342
 343 end:
 344         return (void *) graph;
 345
 346 error:
 347         BT_OBJECT_PUT_REF_AND_RESET(graph);
 348         goto end;
 349 }
 350
 351 enum bt_graph_connect_ports_status bt_graph_connect_ports(
 352                 struct bt_graph *graph,
 353                 const struct bt_port_output *upstream_port_out,
 354                 const struct bt_port_input *downstream_port_in,
 355                 const struct bt_connection **user_connection)
 356 {
 357         enum bt_graph_connect_ports_status status = BT_FUNC_STATUS_OK;
 358         struct bt_connection *connection = NULL;
 359         struct bt_port *upstream_port = (void *) upstream_port_out;
 360         struct bt_port *downstream_port = (void *) downstream_port_in;
 361         struct bt_component *upstream_component = NULL;
 362         struct bt_component *downstream_component = NULL;
 363         enum bt_component_class_port_connected_method_status port_connected_status;
 364         bool init_can_consume;
 365
 366         BT_ASSERT_PRE_NO_ERROR();
 367         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 368         BT_ASSERT_PRE_NON_NULL(upstream_port, "Upstream port");
 369         BT_ASSERT_PRE_NON_NULL(downstream_port, "Downstream port port");
 370         BT_ASSERT_PRE(
 371                 graph->config_state == BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
 372                 "Graph is not in the \"configuring\" state: %!+g", graph);
 373         BT_ASSERT_PRE(!bt_port_is_connected(upstream_port),
 374                 "Upstream port is already connected: %!+p", upstream_port);
 375         BT_ASSERT_PRE(!bt_port_is_connected(downstream_port),
 376                 "Downstream port is already connected: %!+p", downstream_port);
 377         BT_ASSERT_PRE(bt_port_borrow_component_inline((void *) upstream_port),
 378                 "Upstream port does not belong to a component: %!+p",
 379                 upstream_port);
 380         BT_ASSERT_PRE(bt_port_borrow_component_inline((void *) downstream_port),
 381                 "Downstream port does not belong to a component: %!+p",
 382                 downstream_port);
 383         init_can_consume = graph->can_consume;
 384         BT_LIB_LOGI("Connecting component ports within graph: "
 385                 "%![graph-]+g, %![up-port-]+p, %![down-port-]+p",
 386                 graph, upstream_port, downstream_port);
 387         bt_graph_set_can_consume(graph, false);
 388         upstream_component = bt_port_borrow_component_inline(
 389                 (void *) upstream_port);
 390         downstream_component = bt_port_borrow_component_inline(
 391                 (void *) downstream_port);
 392
 393         BT_LOGD_STR("Creating connection.");
 394         connection = bt_connection_create(graph, (void *) upstream_port,
 395                 (void *) downstream_port);
 396         if (!connection) {
 397                 BT_LIB_LOGE_APPEND_CAUSE("Cannot create connection object.");
 398                 status = BT_FUNC_STATUS_MEMORY_ERROR;
 399                 goto end;
 400         }
 401
 402         BT_LIB_LOGD("Connection object created: %!+x", connection);
 403
 404         /*
 405          * Ownership of upstream_component/downstream_component and of
 406          * the connection object is transferred to the graph.
 407          */
 408         g_ptr_array_add(graph->connections, connection);
 409
 410         /*
 411          * Notify both components that their port is connected.
 412          */
 413         BT_LIB_LOGD("Notifying upstream component that its port is connected: "
 414                 "%![comp-]+c, %![port-]+p", upstream_component, upstream_port);
 415         port_connected_status = bt_component_port_connected(upstream_component,
 416                 (void *) upstream_port, (void *) downstream_port);
 417         if (port_connected_status != BT_FUNC_STATUS_OK) {
 418                 if (port_connected_status < 0) {
 419                         BT_LIB_LOGW_APPEND_CAUSE(
 420                                 "Upstream component's \"port connected\" method failed: "
 421                                 "status=%s, %![graph-]+g, %![up-comp-]+c, "
 422                                 "%![down-comp-]+c, %![up-port-]+p, %![down-port-]+p",
 423                                 bt_common_func_status_string(
 424                                         port_connected_status),
 425                                 graph, upstream_component, downstream_component,
 426                                 upstream_port, downstream_port);
 427                 }
 428
 429                 bt_connection_end(connection, true);
 430                 status = (int) port_connected_status;
 431                 goto end;
 432         }
 433
 434         connection->notified_upstream_port_connected = true;
 435         BT_LIB_LOGD("Notifying downstream component that its port is connected: "
 436                 "%![comp-]+c, %![port-]+p", downstream_component,
 437                 downstream_port);
 438         port_connected_status = bt_component_port_connected(downstream_component,
 439                 (void *) downstream_port, (void *) upstream_port);
 440         if (port_connected_status != BT_FUNC_STATUS_OK) {
 441                 if (port_connected_status < 0) {
 442                         BT_LIB_LOGW_APPEND_CAUSE(
 443                                 "Downstream component's \"port connected\" method failed: "
 444                                 "status=%s, %![graph-]+g, %![up-comp-]+c, "
 445                                 "%![down-comp-]+c, %![up-port-]+p, %![down-port-]+p",
 446                                 bt_common_func_status_string(
 447                                         port_connected_status),
 448                                 graph, upstream_component, downstream_component,
 449                                 upstream_port, downstream_port);
 450                 }
 451
 452                 bt_connection_end(connection, true);
 453                 status = (int) port_connected_status;
 454                 goto end;
 455         }
 456
 457         connection->notified_downstream_port_connected = true;
 458
 459         BT_LIB_LOGI("Connected component ports within graph: "
 460                 "%![graph-]+g, %![up-comp-]+c, %![down-comp-]+c, "
 461                 "%![up-port-]+p, %![down-port-]+p",
 462                 graph, upstream_component, downstream_component,
 463                 upstream_port, downstream_port);
 464
 465         if (user_connection) {
 466                 /* Move reference to user */
 467                 *user_connection = connection;
 468                 connection = NULL;
 469         }
 470
 471 end:
 472         if (status != BT_FUNC_STATUS_OK) {
 473                 bt_graph_make_faulty(graph);
 474         }
 475
 476         bt_object_put_ref(connection);
 477         (void) init_can_consume;
 478         bt_graph_set_can_consume(graph, init_can_consume);
 479         return status;
 480 }
 481
 482 static inline
 483 int consume_graph_sink(struct bt_component_sink *comp)
 484 {
 485         enum bt_component_class_sink_consume_method_status consume_status;
 486         struct bt_component_class_sink *sink_class = NULL;
 487
 488         BT_ASSERT_DBG(comp);
 489         sink_class = (void *) comp->parent.class;
 490         BT_ASSERT_DBG(sink_class->methods.consume);
 491         BT_LIB_LOGD("Calling user's consume method: %!+c", comp);
 492         consume_status = sink_class->methods.consume((void *) comp);
 493         BT_LOGD("User method returned: status=%s",
 494                 bt_common_func_status_string(consume_status));
 495         BT_ASSERT_POST_DEV(consume_status == BT_FUNC_STATUS_OK ||
 496                 consume_status == BT_FUNC_STATUS_END ||
 497                 consume_status == BT_FUNC_STATUS_AGAIN ||
 498                 consume_status == BT_FUNC_STATUS_ERROR ||
 499                 consume_status == BT_FUNC_STATUS_MEMORY_ERROR,
 500                 "Invalid component status returned by consuming method: "
 501                 "status=%s", bt_common_func_status_string(consume_status));
 502         BT_ASSERT_POST_DEV_NO_ERROR_IF_NO_ERROR_STATUS(consume_status);
 503         if (consume_status) {
 504                 if (consume_status < 0) {
 505                         BT_LIB_LOGW_APPEND_CAUSE(
 506                                 "Component's \"consume\" method failed: "
 507                                 "status=%s, %![comp-]+c",
 508                                 bt_common_func_status_string(consume_status),
 509                                 comp);
 510                 }
 511
 512                 goto end;
 513         }
 514
 515         BT_LIB_LOGD("Consumed from sink: %![comp-]+c, status=%s",
 516                 comp, bt_common_func_status_string(consume_status));
 517
 518 end:
 519         return consume_status;
 520 }
 521
 522 /*
 523  * `node` is removed from the queue of sinks to consume when passed to
 524  * this function. This function adds it back to the queue if there's
 525  * still something to consume afterwards.
 526  */
 527 static inline
 528 int consume_sink_node(struct bt_graph *graph, GList *node)
 529 {
 530         int status;
 531         struct bt_component_sink *sink;
 532
 533         sink = node->data;
 534         status = consume_graph_sink(sink);
 535         if (G_UNLIKELY(status != BT_FUNC_STATUS_END)) {
 536                 g_queue_push_tail_link(graph->sinks_to_consume, node);
 537                 goto end;
 538         }
 539
 540         /* End reached, the node is not added back to the queue and free'd. */
 541         g_queue_delete_link(graph->sinks_to_consume, node);
 542
 543         /* Don't forward an END status if there are sinks left to consume. */
 544         if (!g_queue_is_empty(graph->sinks_to_consume)) {
 545                 status = BT_FUNC_STATUS_OK;
 546                 goto end;
 547         }
 548
 549 end:
 550         BT_LIB_LOGD("Consumed sink node: %![comp-]+c, status=%s",
 551                 sink, bt_common_func_status_string(status));
 552         return status;
 553 }
 554
 555 BT_HIDDEN
 556 int bt_graph_consume_sink_no_check(struct bt_graph *graph,
 557                 struct bt_component_sink *sink)
 558 {
 559         int status;
 560         GList *sink_node;
 561         int index;
 562
 563         BT_LIB_LOGD("Making specific sink consume: %![comp-]+c", sink);
 564         BT_ASSERT_DBG(bt_component_borrow_graph((void *) sink) == graph);
 565
 566         if (g_queue_is_empty(graph->sinks_to_consume)) {
 567                 BT_LOGD_STR("Graph's sink queue is empty: end of graph.");
 568                 status = BT_FUNC_STATUS_END;
 569                 goto end;
 570         }
 571
 572         index = g_queue_index(graph->sinks_to_consume, sink);
 573         if (index < 0) {
 574                 BT_LIB_LOGD("Sink component is not marked as consumable: "
 575                         "component sink is ended: %![comp-]+c", sink);
 576                 status = BT_FUNC_STATUS_END;
 577                 goto end;
 578         }
 579
 580         sink_node = g_queue_pop_nth_link(graph->sinks_to_consume, index);
 581         BT_ASSERT_DBG(sink_node);
 582         status = consume_sink_node(graph, sink_node);
 583
 584 end:
 585         return status;
 586 }
 587
 588 static inline
 589 int consume_no_check(struct bt_graph *graph)
 590 {
 591         int status = BT_FUNC_STATUS_OK;
 592         struct bt_component *sink;
 593         GList *current_node;
 594
 595         BT_ASSERT_PRE_DEV(graph->has_sink,
 596                 "Graph has no sink component: %!+g", graph);
 597         BT_LIB_LOGD("Making next sink component consume: %![graph-]+g", graph);
 598
 599         if (G_UNLIKELY(g_queue_is_empty(graph->sinks_to_consume))) {
 600                 BT_LOGD_STR("Graph's sink queue is empty: end of graph.");
 601                 status = BT_FUNC_STATUS_END;
 602                 goto end;
 603         }
 604
 605         current_node = g_queue_pop_head_link(graph->sinks_to_consume);
 606         sink = current_node->data;
 607         BT_LIB_LOGD("Chose next sink to consume: %!+c", sink);
 608         status = consume_sink_node(graph, current_node);
 609
 610 end:
 611         return status;
 612 }
 613
 614 enum bt_graph_run_once_status bt_graph_run_once(struct bt_graph *graph)
 615 {
 616         enum bt_graph_run_once_status status;
 617
 618         BT_ASSERT_PRE_NO_ERROR();
 619         BT_ASSERT_PRE_DEV_NON_NULL(graph, "Graph");
 620         BT_ASSERT_PRE_DEV(graph->can_consume,
 621                 "Cannot consume graph in its current state: %!+g", graph);
 622         BT_ASSERT_PRE_DEV(graph->config_state !=
 623                 BT_GRAPH_CONFIGURATION_STATE_FAULTY,
 624                 "Graph is in a faulty state: %!+g", graph);
 625         bt_graph_set_can_consume(graph, false);
 626         status = bt_graph_configure(graph);
 627         if (G_UNLIKELY(status)) {
 628                 /* bt_graph_configure() logs errors */
 629                 goto end;
 630         }
 631
 632         status = consume_no_check(graph);
 633         bt_graph_set_can_consume(graph, true);
 634
 635 end:
 636         return status;
 637 }
 638
 639 enum bt_graph_run_status bt_graph_run(struct bt_graph *graph)
 640 {
 641         enum bt_graph_run_status status;
 642
 643         BT_ASSERT_PRE_NO_ERROR();
 644         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 645         BT_ASSERT_PRE(graph->can_consume,
 646                 "Cannot consume graph in its current state: %!+g", graph);
 647         BT_ASSERT_PRE(graph->config_state != BT_GRAPH_CONFIGURATION_STATE_FAULTY,
 648                 "Graph is in a faulty state: %!+g", graph);
 649         bt_graph_set_can_consume(graph, false);
 650         status = bt_graph_configure(graph);
 651         if (G_UNLIKELY(status)) {
 652                 /* bt_graph_configure() logs errors */
 653                 goto end;
 654         }
 655
 656         BT_LIB_LOGI("Running graph: %!+g", graph);
 657
 658         do {
 659                 /*
 660                  * Check if the graph is interrupted at each iteration.
 661                  * If the graph was interrupted by another thread or by
 662                  * a signal handler, this is NOT a warning nor an error;
 663                  * it was intentional: log with an INFO level only.
 664                  */
 665                 if (G_UNLIKELY(bt_graph_is_interrupted(graph))) {
 666                         BT_LIB_LOGI("Stopping the graph: "
 667                                 "graph was interrupted: %!+g", graph);
 668                         status = BT_FUNC_STATUS_AGAIN;
 669                         goto end;
 670                 }
 671
 672                 status = consume_no_check(graph);
 673                 if (G_UNLIKELY(status == BT_FUNC_STATUS_AGAIN)) {
 674                         /*
 675                          * If AGAIN is received and there are multiple
 676                          * sinks, go ahead and consume from the next
 677                          * sink.
 678                          *
 679                          * However, in the case where a single sink is
 680                          * left, the caller can decide to busy-wait and
 681                          * call bt_graph_run() continuously
 682                          * until the source is ready or it can decide to
 683                          * sleep for an arbitrary amount of time.
 684                          */
 685                         if (graph->sinks_to_consume->length > 1) {
 686                                 status = BT_FUNC_STATUS_OK;
 687                         }
 688                 }
 689         } while (status == BT_FUNC_STATUS_OK);
 690
 691         if (status == BT_FUNC_STATUS_END) {
 692                 /*
 693                  * The last call to consume_no_check() returned
 694                  * `BT_FUNC_STATUS_END`, but bt_graph_run() has no
 695                  * `BT_GRAPH_RUN_STATUS_END` status: replace with
 696                  * `BT_GRAPH_RUN_STATUS_OK` (success: graph ran
 697                  * completely).
 698                  */
 699                 status = BT_FUNC_STATUS_OK;
 700         }
 701
 702 end:
 703         BT_LIB_LOGI("Graph ran: %![graph-]+g, status=%s", graph,
 704                 bt_common_func_status_string(status));
 705         bt_graph_set_can_consume(graph, true);
 706         return status;
 707 }
 708
 709 enum bt_graph_add_listener_status
 710 bt_graph_add_source_component_output_port_added_listener(
 711                 struct bt_graph *graph,
 712                 bt_graph_source_component_output_port_added_listener_func func,
 713                 bt_graph_listener_removed_func listener_removed, void *data,
 714                 bt_listener_id *out_listener_id)
 715 {
 716         struct bt_graph_listener_port_added listener = {
 717                 .base = {
 718                         .removed = listener_removed,
 719                         .data = data,
 720                 },
 721                 .func = (port_added_func_t) func,
 722         };
 723         bt_listener_id listener_id;
 724
 725         BT_ASSERT_PRE_NO_ERROR();
 726         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 727         BT_ASSERT_PRE_NON_NULL(func, "Listener");
 728         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
 729         BT_ASSERT_PRE(!graph->in_remove_listener,
 730                 "Graph currently executing a \"listener removed\" listener: "
 731                 "%!+g", graph);
 732         g_array_append_val(graph->listeners.source_output_port_added, listener);
 733         listener_id = graph->listeners.source_output_port_added->len - 1;
 734         BT_LIB_LOGD("Added \"source component output port added\" listener to graph: "
 735                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
 736                 listener_id);
 737
 738         if (listener_id) {
 739                 *out_listener_id = listener_id;
 740         }
 741
 742         return BT_FUNC_STATUS_OK;
 743 }
 744
 745 enum bt_graph_add_listener_status
 746 bt_graph_add_filter_component_output_port_added_listener(
 747                 struct bt_graph *graph,
 748                 bt_graph_filter_component_output_port_added_listener_func func,
 749                 bt_graph_listener_removed_func listener_removed, void *data,
 750                 bt_listener_id *out_listener_id)
 751 {
 752         struct bt_graph_listener_port_added listener = {
 753                 .base = {
 754                         .removed = listener_removed,
 755                         .data = data,
 756                 },
 757                 .func = (port_added_func_t) func,
 758         };
 759         bt_listener_id listener_id;
 760
 761         BT_ASSERT_PRE_NO_ERROR();
 762         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 763         BT_ASSERT_PRE_NON_NULL(func, "Listener");
 764         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
 765         BT_ASSERT_PRE(!graph->in_remove_listener,
 766                 "Graph currently executing a \"listener removed\" listener: "
 767                 "%!+g", graph);
 768         g_array_append_val(graph->listeners.filter_output_port_added, listener);
 769         listener_id = graph->listeners.filter_output_port_added->len - 1;
 770         BT_LIB_LOGD("Added \"filter component output port added\" listener to graph: "
 771                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
 772                 listener_id);
 773
 774         if (listener_id) {
 775                 *out_listener_id = listener_id;
 776         }
 777
 778         return BT_FUNC_STATUS_OK;
 779 }
 780
 781 enum bt_graph_add_listener_status
 782 bt_graph_add_filter_component_input_port_added_listener(
 783                 struct bt_graph *graph,
 784                 bt_graph_filter_component_input_port_added_listener_func func,
 785                 bt_graph_listener_removed_func listener_removed, void *data,
 786                 bt_listener_id *out_listener_id)
 787 {
 788         struct bt_graph_listener_port_added listener = {
 789                 .base = {
 790                         .removed = listener_removed,
 791                         .data = data,
 792                 },
 793                 .func = (port_added_func_t) func,
 794         };
 795         bt_listener_id listener_id;
 796
 797         BT_ASSERT_PRE_NO_ERROR();
 798         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 799         BT_ASSERT_PRE_NON_NULL(func, "Listener");
 800         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
 801         BT_ASSERT_PRE(!graph->in_remove_listener,
 802                 "Graph currently executing a \"listener removed\" listener: "
 803                 "%!+g", graph);
 804         g_array_append_val(graph->listeners.filter_input_port_added, listener);
 805         listener_id = graph->listeners.filter_input_port_added->len - 1;
 806         BT_LIB_LOGD("Added \"filter component input port added\" listener to graph: "
 807                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
 808                 listener_id);
 809
 810         if (listener_id) {
 811                 *out_listener_id = listener_id;
 812         }
 813
 814         return BT_FUNC_STATUS_OK;
 815 }
 816
 817 enum bt_graph_add_listener_status
 818 bt_graph_add_sink_component_input_port_added_listener(
 819                 struct bt_graph *graph,
 820                 bt_graph_sink_component_input_port_added_listener_func func,
 821                 bt_graph_listener_removed_func listener_removed, void *data,
 822                 bt_listener_id *out_listener_id)
 823 {
 824         struct bt_graph_listener_port_added listener = {
 825                 .base = {
 826                         .removed = listener_removed,
 827                         .data = data,
 828                 },
 829                 .func = (port_added_func_t) func,
 830         };
 831         bt_listener_id listener_id;
 832
 833         BT_ASSERT_PRE_NO_ERROR();
 834         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 835         BT_ASSERT_PRE_NON_NULL(func, "Listener");
 836         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
 837         BT_ASSERT_PRE(!graph->in_remove_listener,
 838                 "Graph currently executing a \"listener removed\" listener: "
 839                 "%!+g", graph);
 840         g_array_append_val(graph->listeners.sink_input_port_added, listener);
 841         listener_id = graph->listeners.sink_input_port_added->len - 1;
 842         BT_LIB_LOGD("Added \"sink component input port added\" listener to graph: "
 843                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
 844                 listener_id);
 845
 846         if (listener_id) {
 847                 *out_listener_id = listener_id;
 848         }
 849
 850         return BT_FUNC_STATUS_OK;
 851 }
 852
 853 BT_HIDDEN
 854 enum bt_graph_listener_func_status bt_graph_notify_port_added(
 855                 struct bt_graph *graph, struct bt_port *port)
 856 {
 857         uint64_t i;
 858         GArray *listeners;
 859         struct bt_component *comp;
 860         enum bt_graph_listener_func_status status = BT_FUNC_STATUS_OK;
 861
 862         BT_ASSERT(graph);
 863         BT_ASSERT(port);
 864         BT_LIB_LOGD("Notifying graph listeners that a port was added: "
 865                 "%![graph-]+g, %![port-]+p", graph, port);
 866         comp = bt_port_borrow_component_inline(port);
 867         BT_ASSERT(comp);
 868
 869         switch (comp->class->type) {
 870         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 871         {
 872                 switch (port->type) {
 873                 case BT_PORT_TYPE_OUTPUT:
 874                         listeners = graph->listeners.source_output_port_added;
 875                         break;
 876                 default:
 877                         bt_common_abort();
 878                 }
 879
 880                 break;
 881         }
 882         case BT_COMPONENT_CLASS_TYPE_FILTER:
 883         {
 884                 switch (port->type) {
 885                 case BT_PORT_TYPE_INPUT:
 886                         listeners = graph->listeners.filter_input_port_added;
 887                         break;
 888                 case BT_PORT_TYPE_OUTPUT:
 889                         listeners = graph->listeners.filter_output_port_added;
 890                         break;
 891                 default:
 892                         bt_common_abort();
 893                 }
 894
 895                 break;
 896         }
 897         case BT_COMPONENT_CLASS_TYPE_SINK:
 898         {
 899                 switch (port->type) {
 900                 case BT_PORT_TYPE_INPUT:
 901                         listeners = graph->listeners.sink_input_port_added;
 902                         break;
 903                 default:
 904                         bt_common_abort();
 905                 }
 906
 907                 break;
 908         }
 909         default:
 910                 bt_common_abort();
 911         }
 912
 913         for (i = 0; i < listeners->len; i++) {
 914                 struct bt_graph_listener_port_added *listener =
 915                         &g_array_index(listeners,
 916                                 struct bt_graph_listener_port_added, i);
 917
 918
 919                 BT_ASSERT(listener->func);
 920                 status = listener->func(comp, port, listener->base.data);
 921                 BT_ASSERT_POST_NO_ERROR_IF_NO_ERROR_STATUS(status);
 922                 if (status != BT_FUNC_STATUS_OK) {
 923                         goto end;
 924                 }
 925         }
 926
 927 end:
 928         return status;
 929 }
 930
 931 BT_HIDDEN
 932 void bt_graph_remove_connection(struct bt_graph *graph,
 933                 struct bt_connection *connection)
 934 {
 935         BT_ASSERT(graph);
 936         BT_ASSERT(connection);
 937         BT_LIB_LOGD("Removing graph's connection: %![graph-]+g, %![conn-]+x",
 938                 graph, connection);
 939         g_ptr_array_remove(graph->connections, connection);
 940 }
 941
 942 static inline
 943 bool component_name_exists(struct bt_graph *graph, const char *name)
 944 {
 945         bool exists = false;
 946         uint64_t i;
 947
 948         for (i = 0; i < graph->components->len; i++) {
 949                 struct bt_component *other_comp = graph->components->pdata[i];
 950
 951                 if (strcmp(name, bt_component_get_name(other_comp)) == 0) {
 952                         BT_ASSERT_PRE_MSG("Another component with the same name already exists in the graph: "
 953                                 "%![other-comp-]+c, name=\"%s\"",
 954                                 other_comp, name);
 955                         exists = true;
 956                         goto end;
 957                 }
 958         }
 959
 960 end:
 961         return exists;
 962 }
 963
 964 static
 965 int add_component_with_init_method_data(
 966                 struct bt_graph *graph,
 967                 struct bt_component_class *comp_cls,
 968                 comp_init_method_t init_method,
 969                 const char *name, const struct bt_value *params,
 970                 void *init_method_data, bt_logging_level log_level,
 971                 const struct bt_component **user_component)
 972 {
 973         int status = BT_FUNC_STATUS_OK;
 974         enum bt_component_class_initialize_method_status init_status;
 975         struct bt_component *component = NULL;
 976         int ret;
 977         bool init_can_consume;
 978         struct bt_value *new_params = NULL;
 979
 980         BT_ASSERT(comp_cls);
 981         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 982         BT_ASSERT_PRE_NON_NULL(name, "Name");
 983         BT_ASSERT_PRE(
 984                 graph->config_state == BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
 985                 "Graph is not in the \"configuring\" state: %!+g", graph);
 986         BT_ASSERT_PRE(!component_name_exists(graph, name),
 987                 "Duplicate component name: %!+g, name=\"%s\"", graph, name);
 988         BT_ASSERT_PRE(!params || bt_value_is_map(params),
 989                 "Parameter value is not a map value: %!+v", params);
 990         init_can_consume = graph->can_consume;
 991         bt_graph_set_can_consume(graph, false);
 992         BT_LIB_LOGI("Adding component to graph: "
 993                 "%![graph-]+g, %![cc-]+C, name=\"%s\", log-level=%s, "
 994                 "%![params-]+v, init-method-data-addr=%p",
 995                 graph, comp_cls, name,
 996                 bt_common_logging_level_string(log_level), params,
 997                 init_method_data);
 998
 999         if (!params) {
1000                 new_params = bt_value_map_create();
1001                 if (!new_params) {
1002                         BT_LIB_LOGE_APPEND_CAUSE(
1003                                 "Cannot create empty map value object.");
1004                         status = BT_FUNC_STATUS_MEMORY_ERROR;
1005                         goto end;
1006                 }
1007
1008                 params = new_params;
1009         }
1010
1011         ret = bt_component_create(comp_cls, name, log_level, &component);
1012         if (ret) {
1013                 BT_LIB_LOGE_APPEND_CAUSE(
1014                         "Cannot create empty component object: ret=%d",
1015                         ret);
1016                 status = BT_FUNC_STATUS_MEMORY_ERROR;
1017                 goto end;
1018         }
1019
1020         /*
1021          * The user's initialization method needs to see that this
1022          * component is part of the graph. If the user method fails, we
1023          * immediately remove the component from the graph's components.
1024          */
1025         g_ptr_array_add(graph->components, component);
1026         bt_component_set_graph(component, graph);
1027         bt_value_freeze(params);
1028
1029         if (init_method) {
1030                 /*
1031                  * There is no use for config objects right now, so just pass
1032                  * NULL.
1033                  */
1034                 BT_LOGD_STR("Calling user's initialization method.");
1035                 init_status = init_method(component, NULL, params, init_method_data);
1036                 BT_LOGD("User method returned: status=%s",
1037                         bt_common_func_status_string(init_status));
1038                 BT_ASSERT_POST_DEV_NO_ERROR_IF_NO_ERROR_STATUS(init_status);
1039                 if (init_status != BT_FUNC_STATUS_OK) {
1040                         if (init_status < 0) {
1041                                 BT_LIB_LOGW_APPEND_CAUSE(
1042                                         "Component initialization method failed: "
1043                                         "status=%s, %![comp-]+c",
1044                                         bt_common_func_status_string(init_status),
1045                                         component);
1046                         }
1047
1048                         status = init_status;
1049                         bt_component_set_graph(component, NULL);
1050                         g_ptr_array_remove_fast(graph->components, component);
1051                         goto end;
1052                 }
1053         }
1054
1055         /*
1056          * Mark the component as initialized so that its finalization
1057          * method is called when it is destroyed.
1058          */
1059         component->initialized = true;
1060
1061         /*
1062          * If it's a sink component, it needs to be part of the graph's
1063          * sink queue to be consumed by bt_graph_run() or
1064          * bt_graph_run_once().
1065          */
1066         if (bt_component_is_sink(component)) {
1067                 graph->has_sink = true;
1068                 g_queue_push_tail(graph->sinks_to_consume, component);
1069         }
1070
1071         /*
1072          * Freeze the component class now that it's instantiated at
1073          * least once.
1074          */
1075         BT_LOGD_STR("Freezing component class.");
1076         bt_component_class_freeze(comp_cls);
1077         BT_LIB_LOGI("Added component to graph: "
1078                 "%![graph-]+g, %![cc-]+C, name=\"%s\", log-level=%s, "
1079                 "%![params-]+v, init-method-data-addr=%p, %![comp-]+c",
1080                 graph, comp_cls, name,
1081                 bt_common_logging_level_string(log_level), params,
1082                 init_method_data, component);
1083
1084         if (user_component) {
1085                 /* Move reference to user */
1086                 *user_component = component;
1087                 component = NULL;
1088         }
1089
1090 end:
1091         if (status != BT_FUNC_STATUS_OK) {
1092                 bt_graph_make_faulty(graph);
1093         }
1094
1095         bt_object_put_ref(component);
1096         bt_object_put_ref(new_params);
1097         (void) init_can_consume;
1098         bt_graph_set_can_consume(graph, init_can_consume);
1099         return status;
1100 }
1101
1102 enum bt_graph_add_component_status
1103 bt_graph_add_source_component_with_initialize_method_data(
1104                 struct bt_graph *graph,
1105                 const struct bt_component_class_source *comp_cls,
1106                 const char *name, const struct bt_value *params,
1107                 void *init_method_data, bt_logging_level log_level,
1108                 const struct bt_component_source **component)
1109 {
1110         BT_ASSERT_PRE_NO_ERROR();
1111         BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1112         return add_component_with_init_method_data(graph,
1113                 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1114                 name, params, init_method_data, log_level, (void *) component);
1115 }
1116
1117 enum bt_graph_add_component_status bt_graph_add_source_component(
1118                 struct bt_graph *graph,
1119                 const struct bt_component_class_source *comp_cls,
1120                 const char *name, const struct bt_value *params,
1121                 enum bt_logging_level log_level,
1122                 const struct bt_component_source **component)
1123 {
1124         BT_ASSERT_PRE_NO_ERROR();
1125         return bt_graph_add_source_component_with_initialize_method_data(
1126                 graph, comp_cls, name, params, NULL, log_level, component);
1127 }
1128
1129 enum bt_graph_add_component_status
1130 bt_graph_add_filter_component_with_initialize_method_data(
1131                 struct bt_graph *graph,
1132                 const struct bt_component_class_filter *comp_cls,
1133                 const char *name, const struct bt_value *params,
1134                 void *init_method_data, enum bt_logging_level log_level,
1135                 const struct bt_component_filter **component)
1136 {
1137         BT_ASSERT_PRE_NO_ERROR();
1138         BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1139         return add_component_with_init_method_data(graph,
1140                 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1141                 name, params, init_method_data, log_level, (void *) component);
1142 }
1143
1144 enum bt_graph_add_component_status bt_graph_add_filter_component(
1145                 struct bt_graph *graph,
1146                 const struct bt_component_class_filter *comp_cls,
1147                 const char *name, const struct bt_value *params,
1148                 enum bt_logging_level log_level,
1149                 const struct bt_component_filter **component)
1150 {
1151         BT_ASSERT_PRE_NO_ERROR();
1152         return bt_graph_add_filter_component_with_initialize_method_data(
1153                 graph, comp_cls, name, params, NULL, log_level, component);
1154 }
1155
1156 enum bt_graph_add_component_status
1157 bt_graph_add_sink_component_with_initialize_method_data(
1158                 struct bt_graph *graph,
1159                 const struct bt_component_class_sink *comp_cls,
1160                 const char *name, const struct bt_value *params,
1161                 void *init_method_data, enum bt_logging_level log_level,
1162                 const struct bt_component_sink **component)
1163 {
1164         BT_ASSERT_PRE_NO_ERROR();
1165         BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1166         return add_component_with_init_method_data(graph,
1167                 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1168                 name, params, init_method_data, log_level, (void *) component);
1169 }
1170
1171 enum bt_graph_add_component_status bt_graph_add_sink_component(
1172                 struct bt_graph *graph,
1173                 const struct bt_component_class_sink *comp_cls,
1174                 const char *name, const struct bt_value *params,
1175                 enum bt_logging_level log_level,
1176                 const struct bt_component_sink **component)
1177 {
1178         BT_ASSERT_PRE_NO_ERROR();
1179         return bt_graph_add_sink_component_with_initialize_method_data(
1180                 graph, comp_cls, name, params, NULL, log_level, component);
1181 }
1182
1183 enum bt_graph_add_component_status
1184 bt_graph_add_simple_sink_component(struct bt_graph *graph, const char *name,
1185                 bt_graph_simple_sink_component_initialize_func init_func,
1186                 bt_graph_simple_sink_component_consume_func consume_func,
1187                 bt_graph_simple_sink_component_finalize_func finalize_func,
1188                 void *user_data, const bt_component_sink **component)
1189 {
1190         enum bt_graph_add_component_status status;
1191         struct bt_component_class_sink *comp_cls;
1192         struct simple_sink_init_method_data init_method_data = {
1193                 .init_func = init_func,
1194                 .consume_func = consume_func,
1195                 .finalize_func = finalize_func,
1196                 .user_data = user_data,
1197         };
1198
1199         BT_ASSERT_PRE_NO_ERROR();
1200
1201         /*
1202          * Other preconditions are checked by
1203          * bt_graph_add_sink_component_with_init_method_data().
1204          */
1205         BT_ASSERT_PRE_NON_NULL(consume_func, "Consume function");
1206
1207         comp_cls = bt_component_class_sink_simple_borrow();
1208         if (!comp_cls) {
1209                 BT_LIB_LOGE_APPEND_CAUSE(
1210                         "Cannot borrow simple sink component class.");
1211                 status = BT_FUNC_STATUS_MEMORY_ERROR;
1212                 goto end;
1213         }
1214
1215         status = bt_graph_add_sink_component_with_initialize_method_data(graph,
1216                 comp_cls, name, NULL, &init_method_data,
1217                 BT_LOGGING_LEVEL_NONE, component);
1218
1219 end:
1220         return status;
1221 }
1222
1223 BT_HIDDEN
1224 void bt_graph_add_message(struct bt_graph *graph,
1225                 struct bt_message *msg)
1226 {
1227         BT_ASSERT(graph);
1228         BT_ASSERT(msg);
1229
1230         /*
1231          * It's okay not to take a reference because, when a
1232          * message's reference count drops to 0, either:
1233          *
1234          * * It is recycled back to one of this graph's pool.
1235          * * It is destroyed because it doesn't have any link to any
1236          *   graph, which means the original graph is already destroyed.
1237          */
1238         g_ptr_array_add(graph->messages, msg);
1239 }
1240
1241 BT_HIDDEN
1242 bool bt_graph_is_interrupted(const struct bt_graph *graph)
1243 {
1244         BT_ASSERT_DBG(graph);
1245         return bt_interrupter_array_any_is_set(graph->interrupters);
1246 }
1247
1248 enum bt_graph_add_interrupter_status bt_graph_add_interrupter(
1249                 struct bt_graph *graph, const struct bt_interrupter *intr)
1250 {
1251         BT_ASSERT_PRE_NO_ERROR();
1252         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1253         BT_ASSERT_PRE_NON_NULL(intr, "Interrupter");
1254         g_ptr_array_add(graph->interrupters, (void *) intr);
1255         bt_object_get_ref_no_null_check(intr);
1256         BT_LIB_LOGD("Added interrupter to graph: %![graph-]+g, %![intr-]+z",
1257                 graph, intr);
1258         return BT_FUNC_STATUS_OK;
1259 }
1260
1261 void bt_graph_interrupt(struct bt_graph *graph)
1262 {
1263         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1264         bt_interrupter_set(graph->default_interrupter);
1265         BT_LIB_LOGI("Interrupted graph: %!+g", graph);
1266 }
1267
1268 void bt_graph_get_ref(const struct bt_graph *graph)
1269 {
1270         bt_object_get_ref(graph);
1271 }
1272
1273 void bt_graph_put_ref(const struct bt_graph *graph)
1274 {
1275         bt_object_put_ref(graph);
1276 }