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