src/lib/graph/graph.c

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