#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;
static
enum bt_component_status get_component_port_counts(
- struct bt_component *component, uint64_t *input_count,
- uint64_t *output_count)
+ struct bt_component *component, int64_t *input_count,
+ int64_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) {
+ *output_count =
+ bt_component_source_get_output_port_count(component);
+ if (*output_count < 0) {
+ ret = BT_COMPONENT_STATUS_ERROR;
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) {
+ *output_count =
+ bt_component_filter_get_output_port_count(component);
+ if (*output_count < 0) {
+ ret = BT_COMPONENT_STATUS_ERROR;
goto end;
}
- ret = bt_component_filter_get_input_port_count(component,
- input_count);
- if (ret != BT_COMPONENT_STATUS_OK) {
+ *input_count =
+ bt_component_filter_get_input_port_count(component);
+ if (*input_count < 0) {
+ ret = BT_COMPONENT_STATUS_ERROR;
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) {
+ *input_count =
+ bt_component_sink_get_input_port_count(component);
+ if (*input_count < 0) {
+ ret = BT_COMPONENT_STATUS_ERROR;
goto end;
}
break;
default:
- assert(false);
+ assert(BT_FALSE);
break;
}
ret = BT_COMPONENT_STATUS_OK;
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;
+ int64_t origin_input_port_count = 0;
+ int64_t origin_output_port_count = 0;
+ int64_t new_input_port_count = 0;
+ int64_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 *downstream_port = NULL;
struct bt_connection *origin_connection = NULL;
struct bt_connection *new_connection = NULL;
- int port_index;
+ int64_t port_index;
if (!graph || !origin || !new_component) {
status = BT_GRAPH_STATUS_INVALID;
goto end;
}
+ if (graph->canceled) {
+ status = BT_GRAPH_STATUS_CANCELED;
+ goto end;
+ }
+
if (bt_component_get_class_type(origin) !=
bt_component_get_class_type(new_component)) {
status = BT_GRAPH_STATUS_INVALID;
/* 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,
+ origin_port = bt_component_get_input_port_by_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,
+ new_port = bt_component_get_input_port_by_index(new_component,
port_index);
if (!new_port) {
status = BT_GRAPH_STATUS_ERROR;
/* 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,
+ origin_port = bt_component_get_output_port_by_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,
+ new_port = bt_component_get_output_port_by_index(new_component,
port_index);
if (!new_port) {
status = BT_GRAPH_STATUS_ERROR;
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 {
+ if (graph->canceled) {
+ status = BT_GRAPH_STATUS_CANCELED;
+ goto end;
+ }
+
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;
+}
projects / babeltrace.git / blobdiff