lib: graph: disallow recursive consuming

[babeltrace.git] / lib / graph / graph.c

diff --git a/lib/graph/graph.c b/lib/graph/graph.c
index 747a282601d5622543139a683b15e62572db9515..b6460f74e228bd684c6ad9f59b8fa139f440475a 100644 (file)
--- a/lib/graph/graph.c
+++ b/lib/graph/graph.c
@@ -188,6 +188,7 @@ struct bt_graph *bt_graph_create(void)
                goto error;
        }
 
+       graph->can_consume = BT_TRUE;
        ret = init_listeners_array(&graph->listeners.port_added);
        if (ret) {
                BT_LOGE_STR("Cannot create the \"port added\" listener array.");
@@ -232,6 +233,7 @@ enum bt_graph_status bt_graph_connect_ports(struct bt_graph *graph,
        struct bt_component *upstream_component = NULL;
        struct bt_component *downstream_component = NULL;
        enum bt_component_status component_status;
+       const bt_bool init_can_consume = graph->can_consume;
 
        if (!graph) {
                BT_LOGW_STR("Invalid parameter: graph is NULL.");
@@ -264,6 +266,8 @@ enum bt_graph_status bt_graph_connect_ports(struct bt_graph *graph,
                goto end;
        }
 
+       graph->can_consume = BT_FALSE;
+
        /* Ensure appropriate types for upstream and downstream ports. */
        if (bt_port_get_type(upstream_port) != BT_PORT_TYPE_OUTPUT) {
                BT_LOGW_STR("Invalid parameter: upstream port is not an output port.");
@@ -406,6 +410,7 @@ end:
        bt_put(upstream_component);
        bt_put(downstream_component);
        bt_put(connection);
+       graph->can_consume = init_can_consume;
        return status;
 }
 
@@ -419,6 +424,12 @@ enum bt_graph_status bt_graph_consume_no_check(struct bt_graph *graph)
 
        BT_LOGV("Making next sink consume: addr=%p", graph);
 
+       if (!graph->has_sink) {
+               BT_LOGW_STR("Graph has no sink component.");
+               status = BT_GRAPH_STATUS_NO_SINK;
+               goto end;
+       }
+
        if (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;
@@ -484,7 +495,15 @@ enum bt_graph_status bt_graph_consume(struct bt_graph *graph)
                goto end;
        }
 
+       if (!graph->can_consume) {
+               BT_LOGW_STR("Cannot consume graph in its current state.");
+               status = BT_GRAPH_STATUS_CANNOT_CONSUME;
+               goto end;
+       }
+
+       graph->can_consume = BT_FALSE;
        status = bt_graph_consume_no_check(graph);
+       graph->can_consume = BT_TRUE;
 
 end:
        return status;
@@ -507,6 +526,13 @@ enum bt_graph_status bt_graph_run(struct bt_graph *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_LOGV("Running graph: addr=%p", graph);
 
        do {
@@ -523,7 +549,7 @@ enum bt_graph_status bt_graph_run(struct bt_graph *graph)
                        goto end;
                }
 
-               status = bt_graph_consume(graph);
+               status = bt_graph_consume_no_check(graph);
                if (status == BT_GRAPH_STATUS_AGAIN) {
                        /*
                         * If AGAIN is received and there are multiple
@@ -539,6 +565,8 @@ enum bt_graph_status bt_graph_run(struct bt_graph *graph)
                        if (graph->sinks_to_consume->length > 1) {
                                status = BT_GRAPH_STATUS_OK;
                        }
+               } else if (status == BT_GRAPH_STATUS_NO_SINK) {
+                       goto end;
                }
        } while (status == BT_GRAPH_STATUS_OK);
 
@@ -548,6 +576,7 @@ enum bt_graph_status bt_graph_run(struct bt_graph *graph)
 
 end:
        BT_LOGV("Graph ran: status=%s", bt_graph_status_string(status));
+       graph->can_consume = BT_TRUE;
        return status;
 }
 
@@ -800,7 +829,7 @@ void bt_graph_notify_ports_disconnected(struct bt_graph *graph,
        }
 }
 
-extern enum bt_graph_status bt_graph_cancel(struct bt_graph *graph)
+enum bt_graph_status bt_graph_cancel(struct bt_graph *graph)
 {
        enum bt_graph_status ret = BT_GRAPH_STATUS_OK;
 
@@ -817,9 +846,19 @@ end:
        return ret;
 }
 
-extern bt_bool bt_graph_is_canceled(struct bt_graph *graph)
+bt_bool bt_graph_is_canceled(struct bt_graph *graph)
 {
-       return graph ? graph->canceled : BT_FALSE;
+       bt_bool canceled = BT_FALSE;
+
+       if (!graph) {
+               BT_LOGW_STR("Invalid parameter: graph is NULL.");
+               goto end;
+       }
+
+       canceled = graph->canceled;
+
+end:
+       return canceled;
 }
 
 BT_HIDDEN
@@ -845,6 +884,7 @@ enum bt_graph_status bt_graph_add_component_with_init_method_data(
        struct bt_component *component = NULL;
        enum bt_component_class_type type;
        size_t i;
+       const bt_bool init_can_consume = graph->can_consume;
 
        bt_get(params);
 
@@ -860,6 +900,7 @@ enum bt_graph_status bt_graph_add_component_with_init_method_data(
                goto end;
        }
 
+       graph->can_consume = BT_FALSE;
        type = bt_component_class_get_type(component_class);
        BT_LOGD("Adding component to graph: "
                "graph-addr=%p, comp-cls-addr=%p, "
@@ -967,6 +1008,7 @@ enum bt_graph_status bt_graph_add_component_with_init_method_data(
         * sink queue to be consumed by bt_graph_consume().
         */
        if (bt_component_is_sink(component)) {
+               graph->has_sink = BT_TRUE;
                g_queue_push_tail(graph->sinks_to_consume, component);
        }
 
@@ -992,6 +1034,7 @@ enum bt_graph_status bt_graph_add_component_with_init_method_data(
 end:
        bt_put(component);
        bt_put(params);
+       graph->can_consume = init_can_consume;
        return graph_status;
 }