#include <babeltrace/graph/component-filter.h>
#include <babeltrace/graph/port.h>
#include <babeltrace/compiler-internal.h>
+#include <babeltrace/types.h>
#include <unistd.h>
#include <glib.h>
struct bt_graph *graph = container_of(obj,
struct bt_graph, base);
- if (graph->components) {
- g_ptr_array_free(graph->components, TRUE);
- }
+ /*
+ * The graph's reference count is 0 if we're here. Increment
+ * it to avoid a double-destroy (possibly infinitely recursive)
+ * in this situation:
+ *
+ * 1. We put and destroy a connection.
+ * 2. This connection's destructor finalizes its active
+ * notification iterators.
+ * 3. A notification iterator's finalization function gets a
+ * new reference on its component (reference count goes from
+ * 0 to 1).
+ * 4. Since this component's reference count goes to 1, it takes
+ * a reference on its parent (this graph). This graph's
+ * reference count goes from 0 to 1.
+ * 5. The notification iterator's finalization function puts its
+ * component reference (reference count goes from 1 to 0).
+ * 6. Since this component's reference count goes from 1 to 0,
+ * it puts its parent (this graph). This graph's reference
+ * count goes from 1 to 0.
+ * 7. Since this graph's reference count goes from 1 to 0,
+ * its destructor is called (this function).
+ *
+ * With the incrementation below, the graph's reference count at
+ * step 4 goes from 1 to 2, and from 2 to 1 at step 6. This
+ * ensures that this function is not called two times.
+ */
+ obj->ref_count.count++;
+
if (graph->connections) {
g_ptr_array_free(graph->connections, TRUE);
}
+ if (graph->components) {
+ g_ptr_array_free(graph->components, TRUE);
+ }
if (graph->sinks_to_consume) {
g_queue_free(graph->sinks_to_consume);
}
struct bt_component *upstream_component = NULL;
struct bt_component *downstream_component = NULL;
enum bt_component_status component_status;
- bool upstream_was_already_in_graph;
- bool downstream_was_already_in_graph;
+ bt_bool upstream_was_already_in_graph;
+ bt_bool downstream_was_already_in_graph;
if (!graph || !upstream_port || !downstream_port) {
goto end;
}
+ if (graph->canceled) {
+ goto end;
+ }
+
/* Ensure appropriate types for upstream and downstream ports. */
if (bt_port_get_type(upstream_port) != BT_PORT_TYPE_OUTPUT) {
goto end;
goto end;
}
-static
-enum bt_component_status get_component_port_counts(
- struct bt_component *component, uint64_t *input_count,
- uint64_t *output_count)
-{
- enum bt_component_status ret;
-
- switch (bt_component_get_class_type(component)) {
- case BT_COMPONENT_CLASS_TYPE_SOURCE:
- ret = bt_component_source_get_output_port_count(component,
- output_count);
- if (ret != BT_COMPONENT_STATUS_OK) {
- goto end;
- }
- break;
- case BT_COMPONENT_CLASS_TYPE_FILTER:
- ret = bt_component_filter_get_output_port_count(component,
- output_count);
- if (ret != BT_COMPONENT_STATUS_OK) {
- goto end;
- }
- ret = bt_component_filter_get_input_port_count(component,
- input_count);
- if (ret != BT_COMPONENT_STATUS_OK) {
- goto end;
- }
- break;
- case BT_COMPONENT_CLASS_TYPE_SINK:
- ret = bt_component_sink_get_input_port_count(component,
- input_count);
- if (ret != BT_COMPONENT_STATUS_OK) {
- goto end;
- }
- break;
- default:
- assert(false);
- break;
- }
- ret = BT_COMPONENT_STATUS_OK;
-end:
- return ret;
-}
-
-enum bt_graph_status bt_graph_add_component_as_sibling(struct bt_graph *graph,
- struct bt_component *origin,
- struct bt_component *new_component)
-{
- uint64_t origin_input_port_count = 0;
- uint64_t origin_output_port_count = 0;
- uint64_t new_input_port_count = 0;
- uint64_t new_output_port_count = 0;
- enum bt_graph_status status = BT_GRAPH_STATUS_OK;
- struct bt_graph *origin_graph = NULL;
- struct bt_graph *new_graph = NULL;
- struct bt_port *origin_port = NULL;
- struct bt_port *new_port = NULL;
- struct bt_port *upstream_port = NULL;
- struct bt_port *downstream_port = NULL;
- struct bt_connection *origin_connection = NULL;
- struct bt_connection *new_connection = NULL;
- int port_index;
-
- if (!graph || !origin || !new_component) {
- status = BT_GRAPH_STATUS_INVALID;
- goto end;
- }
-
- if (bt_component_get_class_type(origin) !=
- bt_component_get_class_type(new_component)) {
- status = BT_GRAPH_STATUS_INVALID;
- goto end;
- }
-
- origin_graph = bt_component_get_graph(origin);
- if (!origin_graph || (origin_graph != graph)) {
- status = BT_GRAPH_STATUS_INVALID;
- goto end;
- }
-
- new_graph = bt_component_get_graph(new_component);
- if (new_graph) {
- status = BT_GRAPH_STATUS_ALREADY_IN_A_GRAPH;
- goto end;
- }
-
- if (get_component_port_counts(origin, &origin_input_port_count,
- &origin_output_port_count) != BT_COMPONENT_STATUS_OK) {
- status = BT_GRAPH_STATUS_INVALID;
- goto end;
- }
- if (get_component_port_counts(new_component, &new_input_port_count,
- &new_output_port_count) != BT_COMPONENT_STATUS_OK) {
- status = BT_GRAPH_STATUS_INVALID;
- goto end;
- }
-
- if (origin_input_port_count != new_input_port_count ||
- origin_output_port_count != new_output_port_count) {
- status = BT_GRAPH_STATUS_INVALID;
- goto end;
- }
-
- /* Replicate input connections. */
- for (port_index = 0; port_index< origin_input_port_count; port_index++) {
- origin_port = bt_component_get_input_port_at_index(origin,
- port_index);
- if (!origin_port) {
- status = BT_GRAPH_STATUS_ERROR;
- goto error_disconnect;
- }
-
- new_port = bt_component_get_input_port_at_index(new_component,
- port_index);
- if (!new_port) {
- status = BT_GRAPH_STATUS_ERROR;
- goto error_disconnect;
- }
-
- origin_connection = bt_port_get_connection(origin_port);
- if (origin_connection) {
- upstream_port = bt_connection_get_upstream_port(
- origin_connection);
- if (!upstream_port) {
- goto error_disconnect;
- }
-
- new_connection = bt_graph_connect_ports(graph,
- upstream_port, new_port);
- if (!new_connection) {
- goto error_disconnect;
- }
- }
-
- BT_PUT(upstream_port);
- BT_PUT(origin_connection);
- BT_PUT(new_connection);
- BT_PUT(origin_port);
- BT_PUT(new_port);
- }
-
- /* Replicate output connections. */
- for (port_index = 0; port_index < origin_output_port_count; port_index++) {
- origin_port = bt_component_get_output_port_at_index(origin,
- port_index);
- if (!origin_port) {
- status = BT_GRAPH_STATUS_ERROR;
- goto error_disconnect;
- }
- new_port = bt_component_get_output_port_at_index(new_component,
- port_index);
- if (!new_port) {
- status = BT_GRAPH_STATUS_ERROR;
- goto error_disconnect;
- }
-
- origin_connection = bt_port_get_connection(origin_port);
- if (origin_connection) {
- downstream_port = bt_connection_get_downstream_port(
- origin_connection);
- if (!downstream_port) {
- goto error_disconnect;
- }
-
- new_connection = bt_graph_connect_ports(graph,
- new_port, downstream_port);
- if (!new_connection) {
- goto error_disconnect;
- }
- }
-
- BT_PUT(downstream_port);
- BT_PUT(origin_connection);
- BT_PUT(new_connection);
- BT_PUT(origin_port);
- BT_PUT(new_port);
- }
-end:
- bt_put(origin_graph);
- bt_put(new_graph);
- bt_put(origin_port);
- bt_put(new_port);
- bt_put(upstream_port);
- bt_put(downstream_port);
- bt_put(origin_connection);
- bt_put(new_connection);
- return status;
-error_disconnect:
- /* Destroy all connections of the new component. */
- /* FIXME. */
- goto end;
-}
-
enum bt_graph_status bt_graph_consume(struct bt_graph *graph)
{
struct bt_component *sink;
goto end;
}
+ if (graph->canceled) {
+ status = BT_GRAPH_STATUS_CANCELED;
+ goto end;
+ }
+
if (g_queue_is_empty(graph->sinks_to_consume)) {
status = BT_GRAPH_STATUS_END;
goto end;
if (!graph) {
status = BT_GRAPH_STATUS_INVALID;
- goto error;
+ goto end;
}
do {
status = bt_graph_consume(graph);
if (status == BT_GRAPH_STATUS_AGAIN) {
/*
- * If AGAIN is received and there are multiple sinks,
- * go ahead and consume from the next sink.
+ * If AGAIN is received and there are multiple
+ * sinks, go ahead and consume from the next
+ * sink.
*
- * However, in the case where a single sink is left,
- * the caller can decide to busy-wait and call
- * bt_graph_run continuously until the source is ready
- * or it can decide to sleep for an arbitrary amount of
- * time.
+ * However, in the case where a single sink is
+ * left, the caller can decide to busy-wait and
+ * call bt_graph_run() continuously until the
+ * source is ready or it can decide to sleep for
+ * an arbitrary amount of time.
*/
if (graph->sinks_to_consume->length > 1) {
status = BT_GRAPH_STATUS_OK;
if (g_queue_is_empty(graph->sinks_to_consume)) {
status = BT_GRAPH_STATUS_END;
}
-error:
+end:
return status;
}
downstream_port, listener.data);
}
}
+
+extern enum bt_graph_status bt_graph_cancel(struct bt_graph *graph)
+{
+ enum bt_graph_status ret = BT_GRAPH_STATUS_OK;
+
+ if (!graph) {
+ ret = BT_GRAPH_STATUS_INVALID;
+ goto end;
+ }
+
+ graph->canceled = BT_TRUE;
+
+end:
+ return ret;
+}
+
+extern bt_bool bt_graph_is_canceled(struct bt_graph *graph)
+{
+ return graph ? graph->canceled : BT_FALSE;
+}
+
+BT_HIDDEN
+void bt_graph_remove_connection(struct bt_graph *graph,
+ struct bt_connection *connection)
+{
+ assert(graph);
+ assert(connection);
+ g_ptr_array_remove(graph->connections, connection);
+}
projects / babeltrace.git / blobdiff