lib/graph/graph.c

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