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