src/lib/graph/graph.c

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