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