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