/*
- * graph.c
- *
- * Babeltrace Plugin Component Graph
- *
* Copyright 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
* Copyright 2017 Philippe Proulx <pproulx@efficios.com>
*
#include <babeltrace/lib-logging-internal.h>
#include <babeltrace/graph/component-internal.h>
+#include <babeltrace/graph/private-graph.h>
#include <babeltrace/graph/graph-internal.h>
#include <babeltrace/graph/connection-internal.h>
#include <babeltrace/graph/component-sink-internal.h>
#include <babeltrace/graph/notification-event-internal.h>
#include <babeltrace/graph/notification-packet-internal.h>
#include <babeltrace/compiler-internal.h>
+#include <babeltrace/common-internal.h>
#include <babeltrace/types.h>
#include <babeltrace/values.h>
+#include <babeltrace/private-values.h>
#include <babeltrace/values-internal.h>
#include <babeltrace/assert-internal.h>
#include <babeltrace/assert-pre-internal.h>
struct bt_graph_listener {
void *func;
- bt_graph_listener_removed removed;
+ bt_private_graph_listener_removed removed;
void *data;
};
* Cancel the graph to disallow some operations, like creating
* notification iterators and adding ports to components.
*/
- (void) bt_graph_cancel(graph);
+ (void) bt_private_graph_cancel((void *) graph);
/* Call all remove listeners */
call_remove_listeners(graph->listeners.port_added);
bt_notification_unlink_graph(notif);
}
-struct bt_graph *bt_graph_create(void)
+struct bt_private_graph *bt_private_graph_create(void)
{
struct bt_graph *graph;
int ret;
goto error;
}
- graph->can_consume = BT_TRUE;
+ bt_graph_set_can_consume(graph, BT_TRUE);
ret = init_listeners_array(&graph->listeners.port_added);
if (ret) {
BT_LOGE_STR("Cannot create the \"port added\" listener array.");
BT_LOGD("Created graph object: addr=%p", graph);
end:
- return graph;
+ return (void *) graph;
error:
- BT_PUT(graph);
+ BT_OBJECT_PUT_REF_AND_RESET(graph);
goto end;
}
-enum bt_graph_status bt_graph_connect_ports(struct bt_graph *graph,
+enum bt_graph_status bt_private_graph_connect_ports(
+ struct bt_private_graph *priv_graph,
struct bt_port *upstream_port, struct bt_port *downstream_port,
struct bt_connection **user_connection)
{
+ struct bt_graph *graph = (void *) priv_graph;
enum bt_graph_status status = BT_GRAPH_STATUS_OK;
struct bt_connection *connection = NULL;
struct bt_graph *upstream_graph = NULL;
goto end;
}
- graph->can_consume = BT_FALSE;
+ bt_graph_set_can_consume(graph, BT_FALSE);
/* Ensure appropriate types for upstream and downstream ports. */
if (bt_port_get_type(upstream_port) != BT_PORT_TYPE_OUTPUT) {
* Notify both components that their port is connected.
*/
BT_LOGD_STR("Notifying upstream component that its port is connected.");
- bt_component_port_connected(upstream_component, upstream_port,
- downstream_port);
+ component_status = bt_component_port_connected(upstream_component,
+ upstream_port, downstream_port);
+ if (component_status != BT_COMPONENT_STATUS_OK) {
+ BT_LOGW("Error while notifying upstream component that its port is connected: "
+ "status=%s, graph-addr=%p, "
+ "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
+ "downstream-comp-addr=%p, downstream-comp-name=\"%s\", "
+ "upstream-port-addr=%p, upstream-port-name=\"%s\", "
+ "downstream-port-addr=%p, downstream-port-name=\"%s\"",
+ bt_component_status_string(component_status), graph,
+ upstream_component, bt_component_get_name(upstream_component),
+ downstream_component, bt_component_get_name(downstream_component),
+ upstream_port, bt_port_get_name(upstream_port),
+ downstream_port, bt_port_get_name(downstream_port));
+ bt_connection_end(connection, true);
+ status = bt_graph_status_from_component_status(
+ component_status);
+ goto end;
+ }
+
+ connection->notified_upstream_port_connected = true;
BT_LOGD_STR("Notifying downstream component that its port is connected.");
- bt_component_port_connected(downstream_component, downstream_port,
- upstream_port);
+ component_status = bt_component_port_connected(downstream_component,
+ downstream_port, upstream_port);
+ if (component_status != BT_COMPONENT_STATUS_OK) {
+ BT_LOGW("Error while notifying downstream component that its port is connected: "
+ "status=%s, graph-addr=%p, "
+ "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
+ "downstream-comp-addr=%p, downstream-comp-name=\"%s\", "
+ "upstream-port-addr=%p, upstream-port-name=\"%s\", "
+ "downstream-port-addr=%p, downstream-port-name=\"%s\"",
+ bt_component_status_string(component_status), graph,
+ upstream_component, bt_component_get_name(upstream_component),
+ downstream_component, bt_component_get_name(downstream_component),
+ upstream_port, bt_port_get_name(upstream_port),
+ downstream_port, bt_port_get_name(downstream_port));
+ bt_connection_end(connection, true);
+ status = bt_graph_status_from_component_status(
+ component_status);
+ goto end;
+ }
+
+ connection->notified_downstream_port_connected = true;
/*
* Notify the graph's creator that both ports are connected.
*/
BT_LOGD_STR("Notifying graph's user that new component ports are connected.");
bt_graph_notify_ports_connected(graph, upstream_port, downstream_port);
+ connection->notified_graph_ports_connected = true;
BT_LOGD("Connected component ports within graph: "
"graph-addr=%p, "
"upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
}
end:
- bt_put(upstream_graph);
- bt_put(downstream_graph);
- bt_put(upstream_component);
- bt_put(downstream_component);
- bt_put(connection);
+ bt_object_put_ref(upstream_graph);
+ bt_object_put_ref(downstream_graph);
+ bt_object_put_ref(upstream_component);
+ bt_object_put_ref(downstream_component);
+ bt_object_put_ref(connection);
if (graph) {
- graph->can_consume = init_can_consume;
+ (void) init_can_consume;
+ bt_graph_set_can_consume(graph, init_can_consume);
}
return status;
}
-static
+static inline
enum bt_graph_status consume_graph_sink(struct bt_component *sink)
{
- enum bt_graph_status status = BT_GRAPH_STATUS_OK;
enum bt_component_status comp_status;
BT_ASSERT(sink);
BT_LOGV("Consumed from sink: addr=%p, name=\"%s\", status=%s",
sink, bt_component_get_name(sink),
bt_component_status_string(comp_status));
-
- switch (comp_status) {
- case BT_COMPONENT_STATUS_OK:
- break;
- case BT_COMPONENT_STATUS_END:
- status = BT_GRAPH_STATUS_END;
- break;
- case BT_COMPONENT_STATUS_AGAIN:
- status = BT_GRAPH_STATUS_AGAIN;
- break;
- case BT_COMPONENT_STATUS_INVALID:
- status = BT_GRAPH_STATUS_INVALID;
- break;
- default:
- status = BT_GRAPH_STATUS_ERROR;
- break;
- }
-
- return status;
+ BT_ASSERT_PRE(comp_status == BT_COMPONENT_STATUS_OK ||
+ comp_status == BT_COMPONENT_STATUS_END ||
+ comp_status == BT_COMPONENT_STATUS_AGAIN ||
+ comp_status == BT_COMPONENT_STATUS_ERROR ||
+ comp_status == BT_COMPONENT_STATUS_NOMEM,
+ "Invalid component status returned by consuming function: "
+ "status=%s", bt_component_status_string(comp_status));
+ return (enum bt_graph_status) comp_status;
}
/*
* this function. This function adds it back to the queue if there's
* still something to consume afterwards.
*/
-static
+static inline
enum bt_graph_status consume_sink_node(struct bt_graph *graph,
GList *node)
{
sink = node->data;
status = consume_graph_sink(sink);
- if (status != BT_GRAPH_STATUS_END) {
+ if (unlikely(status != BT_GRAPH_STATUS_END)) {
g_queue_push_tail_link(graph->sinks_to_consume, node);
goto end;
}
return status;
}
-BT_HIDDEN
+static inline
enum bt_graph_status bt_graph_consume_no_check(struct bt_graph *graph)
{
enum bt_graph_status status = BT_GRAPH_STATUS_OK;
BT_ASSERT_PRE(graph->has_sink,
"Graph has no sink component: %!+g", graph);
- if (g_queue_is_empty(graph->sinks_to_consume)) {
+ if (unlikely(g_queue_is_empty(graph->sinks_to_consume))) {
BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
status = BT_GRAPH_STATUS_END;
goto end;
return status;
}
-enum bt_graph_status bt_graph_consume(struct bt_graph *graph)
+enum bt_graph_status bt_private_graph_consume(
+ struct bt_private_graph *priv_graph)
{
+ struct bt_graph *graph = (void *) priv_graph;
enum bt_graph_status status;
BT_ASSERT_PRE_NON_NULL(graph, "Graph");
BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
BT_ASSERT_PRE(graph->can_consume,
"Cannot consume graph in its current state: %!+g", graph);
- graph->can_consume = BT_FALSE;
+ bt_graph_set_can_consume(graph, BT_FALSE);
status = bt_graph_consume_no_check(graph);
- graph->can_consume = BT_TRUE;
+ bt_graph_set_can_consume(graph, BT_TRUE);
return status;
}
-enum bt_graph_status bt_graph_run(struct bt_graph *graph)
+enum bt_graph_status bt_private_graph_run(
+ struct bt_private_graph *priv_graph)
{
+ struct bt_graph *graph = (void *) priv_graph;
enum bt_graph_status status = BT_GRAPH_STATUS_OK;
if (!graph) {
goto end;
}
- if (!graph->can_consume) {
- BT_LOGW_STR("Cannot run graph in its current state.");
- status = BT_GRAPH_STATUS_CANNOT_CONSUME;
- goto end;
- }
-
- graph->can_consume = BT_FALSE;
+ BT_ASSERT_PRE(graph->can_consume,
+ "Cannot consume graph in its current state: %!+g", graph);
+ bt_graph_set_can_consume(graph, BT_FALSE);
BT_LOGV("Running graph: addr=%p", graph);
do {
* signal, this is not a warning nor an error, it was
* intentional: log with a DEBUG level only.
*/
- if (graph->canceled) {
+ if (unlikely(graph->canceled)) {
BT_LOGD("Stopping the graph: graph is canceled: "
"graph-addr=%p", graph);
status = BT_GRAPH_STATUS_CANCELED;
}
status = bt_graph_consume_no_check(graph);
- if (status == BT_GRAPH_STATUS_AGAIN) {
+ if (unlikely(status == BT_GRAPH_STATUS_AGAIN)) {
/*
* If AGAIN is received and there are multiple
* sinks, go ahead and consume from the next
end:
BT_LOGV("Graph ran: status=%s", bt_graph_status_string(status));
if (graph) {
- graph->can_consume = BT_TRUE;
+ bt_graph_set_can_consume(graph, BT_TRUE);
}
return status;
}
return listeners->len - 1;
}
-int bt_graph_add_port_added_listener(
- struct bt_graph *graph,
- bt_graph_port_added_listener listener,
- bt_graph_listener_removed listener_removed, void *data)
+int bt_private_graph_add_port_added_listener(
+ struct bt_private_graph *priv_graph,
+ bt_private_graph_port_added_listener listener,
+ bt_private_graph_listener_removed listener_removed, void *data)
{
+ struct bt_graph *graph = (void *) priv_graph;
int ret;
if (!graph) {
return ret;
}
-int bt_graph_add_port_removed_listener(
- struct bt_graph *graph,
- bt_graph_port_removed_listener listener,
- bt_graph_listener_removed listener_removed, void *data)
+int bt_private_graph_add_port_removed_listener(
+ struct bt_private_graph *priv_graph,
+ bt_private_graph_port_removed_listener listener,
+ bt_private_graph_listener_removed listener_removed, void *data)
{
+ struct bt_graph *graph = (void *) priv_graph;
int ret;
if (!graph) {
return ret;
}
-int bt_graph_add_ports_connected_listener(
- struct bt_graph *graph,
- bt_graph_ports_connected_listener listener,
- bt_graph_listener_removed listener_removed, void *data)
+int bt_private_graph_add_ports_connected_listener(
+ struct bt_private_graph *priv_graph,
+ bt_private_graph_ports_connected_listener listener,
+ bt_private_graph_listener_removed listener_removed, void *data)
{
+ struct bt_graph *graph = (void *) priv_graph;
int ret;
if (!graph) {
return ret;
}
-int bt_graph_add_ports_disconnected_listener(
- struct bt_graph *graph,
- bt_graph_ports_disconnected_listener listener,
- bt_graph_listener_removed listener_removed, void *data)
+int bt_private_graph_add_ports_disconnected_listener(
+ struct bt_private_graph *priv_graph,
+ bt_private_graph_ports_disconnected_listener listener,
+ bt_private_graph_listener_removed listener_removed, void *data)
{
+ struct bt_graph *graph = (void *) priv_graph;
int ret;
if (!graph) {
struct bt_graph_listener listener =
g_array_index(graph->listeners.port_added,
struct bt_graph_listener, i);
- bt_graph_port_added_listener func = listener.func;
+ bt_private_graph_port_added_listener func = listener.func;
BT_ASSERT(func);
func(port, listener.data);
struct bt_graph_listener listener =
g_array_index(graph->listeners.port_removed,
struct bt_graph_listener, i);
- bt_graph_port_removed_listener func = listener.func;
+ bt_private_graph_port_removed_listener func = listener.func;
BT_ASSERT(func);
func(comp, port, listener.data);
struct bt_graph_listener listener =
g_array_index(graph->listeners.ports_connected,
struct bt_graph_listener, i);
- bt_graph_ports_connected_listener func = listener.func;
+ bt_private_graph_ports_connected_listener func = listener.func;
BT_ASSERT(func);
func(upstream_port, downstream_port, listener.data);
struct bt_graph_listener listener =
g_array_index(graph->listeners.ports_disconnected,
struct bt_graph_listener, i);
- bt_graph_ports_disconnected_listener func = listener.func;
+ bt_private_graph_ports_disconnected_listener func = listener.func;
BT_ASSERT(func);
func(upstream_comp, downstream_comp, upstream_port,
}
}
-enum bt_graph_status bt_graph_cancel(struct bt_graph *graph)
+enum bt_graph_status bt_private_graph_cancel(
+ struct bt_private_graph *priv_graph)
{
+ struct bt_graph *graph = (void *) priv_graph;
enum bt_graph_status ret = BT_GRAPH_STATUS_OK;
if (!graph) {
g_ptr_array_remove(graph->connections, connection);
}
-enum bt_graph_status bt_graph_add_component_with_init_method_data(
- struct bt_graph *graph,
+enum bt_graph_status bt_private_graph_add_component_with_init_method_data(
+ struct bt_private_graph *priv_graph,
struct bt_component_class *component_class,
const char *name, struct bt_value *params,
void *init_method_data,
struct bt_component **user_component)
{
+ struct bt_graph *graph = (void *) priv_graph;
enum bt_graph_status graph_status = BT_GRAPH_STATUS_OK;
enum bt_component_status comp_status;
struct bt_component *component = NULL;
size_t i;
bt_bool init_can_consume;
- bt_get(params);
+ bt_object_get_ref(params);
if (!graph) {
BT_LOGW_STR("Invalid parameter: graph is NULL.");
if (!bt_value_is_map(params)) {
BT_LOGW("Invalid parameter: initialization parameters must be a map value: "
"type=%s",
- bt_value_type_string(bt_value_get_type(params)));
+ bt_common_value_type_string(
+ bt_value_get_type(params)));
graph_status = BT_GRAPH_STATUS_INVALID;
goto end;
}
} else {
- params = bt_value_map_create();
+ params = bt_value_borrow_from_private(
+ bt_private_value_map_create());
if (!params) {
BT_LOGE_STR("Cannot create map value object.");
graph_status = BT_GRAPH_STATUS_NOMEM;
}
end:
- bt_put(component);
- bt_put(params);
+ bt_object_put_ref(component);
+ bt_object_put_ref(params);
if (graph) {
graph->can_consume = init_can_consume;
}
return graph_status;
}
-enum bt_graph_status bt_graph_add_component(
- struct bt_graph *graph,
+enum bt_graph_status bt_private_graph_add_component(
+ struct bt_private_graph *graph,
struct bt_component_class *component_class,
const char *name, struct bt_value *params,
struct bt_component **component)
{
- return bt_graph_add_component_with_init_method_data(graph,
+ return bt_private_graph_add_component_with_init_method_data(graph,
component_class, name, params, NULL, component);
}
bt_component_get_input_port_by_index(component, i);
BT_ASSERT(port);
- bt_put(port);
+ bt_object_put_ref(port);
if (bt_port_is_connected(port)) {
BT_LOGW("Cannot remove component from graph: "
bt_component_get_output_port_by_index(component, i);
BT_ASSERT(port);
- bt_put(port);
+ bt_object_put_ref(port);
if (bt_port_is_connected(port)) {
BT_LOGW("Cannot remove component from graph: "
*/
g_ptr_array_add(graph->notifications, notif);
}
+
+struct bt_graph *bt_graph_borrow_from_private(
+ struct bt_private_graph *priv_graph)
+{
+ return (void *) priv_graph;
+}
projects / babeltrace.git / blobdiff