Dynamic Flow Control

Dynamic Flow Control is a feature introduced in Holoscan SDK v3.0 that allows operators to modify their connections with other operators at runtime. This enables creating complex workflows with conditional branching, loops, and dynamic routing patterns.

Overview

Traditional static workflows in Holoscan define fixed connections between operators at application composition time. Dynamic Flow Control extends this by allowing operators to:

• Modify their connections during execution
• Route data conditionally to different operators
• Create loops and iterative patterns
• Implement complex branching logic

Common use cases include:

• Conditional processing pipelines
• Adaptive workflow routing
• Iterative processing with dynamic termination
• Error handling and recovery flows

Quick Start

Here’s a simple example to get started with Dynamic Flow Control in Holoscan:

1. Basic Sequential Flow

The simplest use of dynamic flow control is creating a sequential chain of operators that doesn’t have any input or output ports:

1
  #include <holoscan/holoscan.hpp>
2
   
3
  // Define a simple operator
4
  class SimpleOp : public holoscan::Operator {
5
   public:
6
    HOLOSCAN_OPERATOR_FORWARD_ARGS(SimpleOp)
7
    SimpleOp() = default;
8
   
9
    void compute(holoscan::InputContext&, holoscan::OutputContext&,
10
                 holoscan::ExecutionContext&) override {
11
      // Simple computation
12
      HOLOSCAN_LOG_INFO("Executing {}", name());
13
    }
14
  };
15
   
16
  // Define the application
17
  class SimpleSequentialApp : public holoscan::Application {
18
   public:
19
    void compose() override {
20
      // Create operators
21
      auto op1 = make_operator<SimpleOp>("op1");
22
      auto op2 = make_operator<SimpleOp>("op2");
23
   
24
      // Connect operators sequentially
25
      add_flow(start_op(), op1);  // Start with op1
26
      add_flow(op1, op2);         // Then op2
27
    }
28
  };
29
   
30
  int main(int argc, char** argv) {
31
    auto app = holoscan::make_application<SimpleSequentialApp>();
32
    app->run();
33
    return 0;
34
  }

2. Basic Conditional Flow

Here’s a simple example of conditional routing between operators:

1
  #include <holoscan/holoscan.hpp>
2
   
3
  // Define a simple operator with a value
4
  class SimpleOp : public holoscan::Operator {
5
   public:
6
    HOLOSCAN_OPERATOR_FORWARD_ARGS(SimpleOp)
7
   
8
    SimpleOp() = default;
9
   
10
    void setup(holoscan::OperatorSpec& spec) override {}
11
   
12
    void compute(holoscan::InputContext&, holoscan::OutputContext&,
13
                 holoscan::ExecutionContext&) override {
14
      value_++;  // Increment value each time
15
      HOLOSCAN_LOG_INFO("Executing {} with value {}", name(), value_);
16
    }
17
   
18
    int get_value() const { return value_; }
19
   
20
   private:
21
    int value_ = 0;
22
  };
23
   
24
  // Define the application
25
  class SimpleConditionalApp : public holoscan::Application {
26
   public:
27
    void compose() override {
28
      // Create operators
29
      auto op1 = make_operator<SimpleOp>("op1", make_condition&lt;holoscan::CountCondition&gt;(3));
30
      auto path_a = make_operator<SimpleOp>("path_a");
31
      auto path_b = make_operator<SimpleOp>("path_b");
32
   
33
      // Define possible flows
34
      add_flow(op1, path_a);
35
      add_flow(op1, path_b);
36
   
37
      // Set dynamic flow based on condition
38
      set_dynamic_flows(op1, [path_a, path_b](const std::shared_ptr&lt;holoscan::Operator&gt;& op) {
39
        auto simple_op = std::static_pointer_cast<SimpleOp>(op);
40
        if (simple_op->get_value() % 2 == 0) {
41
          simple_op->add_dynamic_flow(path_a);  // Even values go to path_a
42
        } else {
43
          simple_op->add_dynamic_flow(path_b);  // Odd values go to path_b
44
        }
45
      });
46
    }
47
  };
48
   
49
  int main(int argc, char** argv) {
50
    auto app = holoscan::make_application<SimpleConditionalApp>();
51
    app->run();
52
    return 0;
53
  }

Key Points to Remember

1. Optionally use start_op() (C++ (holoscan::Fragment::start_op)/Python (holoscan.core.Fragment.start_op)) to get the initial operator in your flow
2. Connect operators with add_flow()
3. Use set_dynamic_flows() (C++ (holoscan::Fragment::set_dynamic_flows)/Python (holoscan.core.Application.set_dynamic_flows)) to define runtime routing logic. An identically named method is also available on Subgraph (C++ (holoscan::Subgraph::set_dynamic_flows)/Python (holoscan.core.Subgraph.set_dynamic_flows)) for applications making use of subgraphs containing dynamic flows.
4. Implement flow control logic in the callback function passed to set_dynamic_flows() (C++ (holoscan::Fragment::set_dynamic_flows)/Python (holoscan.core.Application.set_dynamic_flows))

• The callback function takes an operator as input and returns void
• The callback function can add dynamic flows using the operator’s add_dynamic_flow() (C++ (holoscan::Operator::add_dynamic_flow)/Python (holoscan.core.Operator.add_dynamic_flow)) methods

For more complex patterns and detailed explanations, see the sections below.

Key Concepts

Input and Output Execution Ports

Before Holoscan SDK v3.0, operators needed input and output ports to be connected via add_flow() and there is no way to specify the execution dependency if the operator does not have any input or output ports.

However, in some cases, the requirements were different:

• An ‘execution order dependency’ was needed instead of a ‘data flow dependency’.
• Execution control was required rather than keeping a node running continuously.
• The pipeline should run only once unless explicitly specified to loop.

To address these needs, Holoscan SDK v3.0 introduced implicit input/output ‘execution ports’ (__input_exec__ / __output_exec__), inspired by Unreal Engine’s Blueprints (particularly execution pins).

The output execution port (__output_exec__. holoscan::Operator::kOutputExecPortName in C++ and holoscan.core.Operator.OUTPUT_EXEC_PORT_NAME in Python) of a source operator and the input execution port (__input_exec__, holoscan::Operator::kInputExecPortName in C++ and holoscan.core.Operator.INPUT_EXEC_PORT_NAME in Python) of a target operator are implicitly added when both of the following are true:

• Two operators are connected using add_flow() without specifying a port map.
• The target operator does not have an explicit input port.

During execution, after the source operator’s compute() method is called, the Holoscan executor emits an empty message (Entity) to the implicit output execution port as a signal. This can then trigger the target operator’s execution, as the Holoscan executor attaches a MessageAvailableCondition to the target operator’s implicit input execution port. Before the target operator’s compute() method runs, the executor collects (pops) all messages from the implicit input execution port, enabling execution dependencies without requiring explicit input and output execution ports.

Starting with Holoscan SDK v3.0, operators can be connected via add_flow() without the need for explicit input and output execution ports, allowing for more flexible and dynamic operator connections.

Start Operator

In Holoscan, when the workflow graph is executed, root operators who do not have any input ports are first executed, and unless any condition is specified to the root operator (such as CountCondition or PeriodicCondition), it will execute continuously.

Inspired by LangGraph’s start node (langgraph.graph.START), Holoscan SDK v3.0 introduces a new concept of the start operator.

The start operator is the first operator in an application fragment, serving as the entry point to the workflow. It is simply the first operator added to the fragment.

This operator is named <|start|> and has a condition of CountCondition(1), ensuring it executes only once. Other entry operators that initiate fragment execution should connect to this operator.

In Holoscan, you can retrieve the start operator by calling start_op() (C++ (holoscan::Fragment::start_op)/Python (holoscan.core.Fragment.start_op)) within the compose() method. If this method is called multiple times, it will return the same start operator

This API is available in both C++ and Python (see flow_control/sequential for an example):

1
  class SequentialExecutionApp : public holoscan::Application {
2
   public:
3
    void compose() override {
4
      using namespace holoscan;
5
   
6
      // Define the operators
7
      auto node1 = make_operator<SimpleOp>("node1");
8
      auto node2 = make_operator<SimpleOp>("node2");
9
      auto node3 = make_operator<SimpleOp>("node3");
10
   
11
      // Define the three-operator workflow
12
      add_flow(start_op(), node1);
13
      add_flow(node1, node2);
14
      add_flow(node2, node3);
15
    }
16
  };

In this example, the start operator is connected to node1, making node1 the first operator to execute. Since the start operator has a condition of CountCondition(1), it will only trigger once, ensuring node1 runs a single time.

In this example, each node (operator) is executed sequentially and executed only once. The start_op() method retrieves the start operator, which is connected to node1. This makes node1 the first operator to execute. After node1 completes its execution, node2 is triggered, followed by node3. The CountCondition(1) in the start_op() method ensures that each operator in the sequence runs a single time, maintaining a clear and predictable flow of execution.

Setting Dynamic Flows

The set_dynamic_flows() (C++ (holoscan::Fragment::set_dynamic_flows)/Python (holoscan.core.Application.set_dynamic_flows)) method allows for dynamic flow control in a Holoscan application. This method sets a callback function that determines the flow of execution based on the state of the operator at runtime.

In the example from flow_control/conditional, the set_dynamic_flows() method is used to dynamically control the flow between node1, node2, and node4 based on the value of node1. The callback function checks the value of node1 and adds a dynamic flow to either node2 or node4:

 Node Graph:
 
       node1 (launch twice)
       /   \
   node2   node4
     |       |
   node3   node5

1
  class ConditionalExecutionApp : public holoscan::Application {
2
   public:
3
    void compose() override {
4
      using namespace holoscan;
5
   
6
      // Define the operators
7
      auto node1 = make_operator<SimpleOp>("node1", make_condition<CountCondition>(2));
8
      auto node2 = make_operator<SimpleOp>("node2");
9
      auto node3 = make_operator<SimpleOp>("node3");
10
      auto node4 = make_operator<SimpleOp>("node4");
11
      auto node5 = make_operator<SimpleOp>("node5");
12
   
13
      add_flow(node1, node2);
14
      add_flow(node2, node3);
15
      add_flow(node1, node4);
16
      add_flow(node4, node5);
17
   
18
      set_dynamic_flows(node1, [node2, node4](const std::shared_ptr<Operator>& op) {
19
        auto simple_op = std::static_pointer_cast<SimpleOp>(op);
20
        if (simple_op->get_value() % 2 == 1) {
21
          simple_op->add_dynamic_flow(node2);
22
        } else {
23
          simple_op->add_dynamic_flow(node4);
24
        }
25
      });
26
    }
27
  };

In the above example, the set_dynamic_flows() (C++ (holoscan::Fragment::set_dynamic_flows)/Python (holoscan.core.Application.set_dynamic_flows)) methods are used to define and manage dynamic workflows in the application.

The set_dynamic_flows() (C++ (holoscan::Fragment::set_dynamic_flows)/Python (holoscan.core.Application.set_dynamic_flows)) method takes an operator and a callback function as arguments. The callback function is called with the operator as an argument. Inside the callback function, the add_dynamic_flow() (C++ (holoscan::Operator::add_dynamic_flow)/Python (holoscan.core.Operator.add_dynamic_flow)) method is used to add a dynamic flow to the operator.

In the example, the callback function checks the value of node1 and adds a dynamic flow to either node2 or node4.

The add_dynamic_flow() (C++ (holoscan::Operator::add_dynamic_flow)/Python (holoscan.core.Operator.add_dynamic_flow)) method has several overloads to support different ways of adding dynamic flows:

1
  /// Basic connection using default output port. This is the simplest form for connecting
2
  /// two operators when you only need to specify the destination.
3
  void add_dynamic_flow(const std::shared_ptr<Operator>& next_op,
4
                       const std::string& next_input_port = "");
5
   
6
  /// Connection with explicit output port specification. Use this when the source operator
7
  /// has multiple output ports and you need to specify which one to use.
8
  void add_dynamic_flow(const std::string& curr_output_port,
9
                       const std::shared_ptr<Operator>& next_op,
10
                       const std::string& next_input_port = "");
11
   
12
  /// Connection using a FlowInfo object, which encapsulates all connection details including:
13
  /// - Source operator and its output port specification
14
  /// - Destination operator and its input port specification
15
  /// - Port names and associated IOSpecs
16
  void add_dynamic_flow(const std::shared_ptr<FlowInfo>& flow);
17
   
18
  /// Batch connection using multiple FlowInfo objects. Use this to set up multiple
19
  /// connections in a single call, which is more efficient than making multiple
20
  /// individual connections.
21
  void add_dynamic_flow(const std::vector<std::shared_ptr<FlowInfo>>& flows);

The simple form of add_dynamic_flow() (C++ (holoscan::Operator::add_dynamic_flow)/Python (holoscan.core.Operator.add_dynamic_flow)) is passing just the next operator (and optionally the next input port name).

If the next operator does not have any explicit input, you can omit the next input port name. In this case, current operator’s implicit output execution port will be connected to the next operator’s implicit input execution port.

Flow Information

The FlowInfo (C++ (holoscan::Operator::FlowInfo)/Python (holoscan.core.FlowInfo)) class represents information about a connection between operators and takes the following arguments in the constructor:

• curr_operator: The source operator of the flow connection
• curr_output_port: The name of the output port on the source operator
• next_operator: The destination operator of the flow connection
• next_input_port: The name of the input port on the destination operator

Inside the callback function, you can use the find_flow_info() (C++ (holoscan::Operator::find_flow_info)/Python (holoscan.core.Operator.find_flow_info)) method and find_all_flow_info() (C++ (holoscan::Operator::find_all_flow_info)/Python (holoscan.core.Operator.find_all_flow_info)) method to find the FlowInfo object(s) that matches the predicate.

The following example shows how to find the FlowInfo object(s) that matches the predicate:

1
  class ConditionalExecutionApp : public holoscan::Application {
2
   public:
3
    void compose() override {
4
      using namespace holoscan;
5
   
6
      // Define the operators
7
      auto node1 = make_operator<SimpleOp>("node1", make_condition<CountCondition>(2));
8
      auto node2 = make_operator<SimpleOp>("node2");
9
      auto node3 = make_operator<SimpleOp>("node3");
10
      auto node4 = make_operator<SimpleOp>("node4");
11
      auto node5 = make_operator<SimpleOp>("node5");
12
   
13
      add_flow(node1, node2);
14
      add_flow(node2, node3);
15
      add_flow(node1, node4);
16
      add_flow(node4, node5);
17
   
18
      // // If you want to add all the next flows, you can use the following code:
19
      // set_dynamic_flows(
20
      //     node1, [](const std::shared_ptr<Operator>& op) { op->add_dynamic_flow(op->next_flows());
21
      //     });
22
   
23
      set_dynamic_flows(node1, [](const std::shared_ptr<Operator>& op) {
24
        auto simple_op = std::static_pointer_cast<SimpleOp>(op);
25
        static const auto& node2_flow = op->find_flow_info(
26
            [](const auto& flow) { return flow->next_operator->name() == "node2"; });
27
        static const auto& node4_flow = op->find_flow_info(
28
            [](const auto& flow) { return flow->next_operator->name() == "node4"; });
29
        //static const auto& all_next_flows = op->find_all_flow_info(
30
        //    [](const auto& flow) { return true; });
31
   
32
        //std::cout << "All next flows: ";
33
        //for (const auto& flow : all_next_flows) {
34
        //  std::cout << flow->next_operator->name() << " ";
35
        //}
36
        //std::cout << std::endl;
37
   
38
        if (simple_op->get_value() % 2 == 1) {
39
          simple_op->add_dynamic_flow(node2_flow);
40
        } else {
41
          simple_op->add_dynamic_flow(node4_flow);
42
        }
43
      });
44
    }
45
  };

In the above example, instead of using op.add_dynamic_flow(node2) or op.add_dynamic_flow(node4), we use op.add_dynamic_flow(node2_flow) or op.add_dynamic_flow(node4_flow). And the node2_flow and node4_flow are FlowInfo objects that are found using the find_flow_info() method.

The find_flow_info() (C++ (holoscan::Operator::find_flow_info)/Python (holoscan.core.Operator.find_flow_info)) method takes a predicate as an argument and returns a FlowInfo object that matches the predicate.

The find_all_flow_info() (C++ (holoscan::Operator::find_all_flow_info)/Python (holoscan.core.Operator.find_all_flow_info)) method takes a predicate as an argument and returns a vector (list) of FlowInfo objects that match the predicate.

If you want to get a vector of all the next flows, you can use op->next_flows() in C++ or op.next_flows in Python.

Using Dynamic Flow Control with Subgraphs

Subgraphs can expose execution interface ports that work seamlessly with dynamic flow control features. This allows you to encapsulate sequences of operators and treat them as single units in your control flow.

Subgraphs support both data and execution interface ports:

• Data interface ports: Expose operator data input/output ports using add_input_interface_port() and add_output_interface_port()
• Execution interface ports: Expose operator execution ports using add_input_exec_interface_port() and add_output_exec_interface_port()

When connecting Subgraphs in control flow:

• If a Subgraph has only one execution interface port (input or output), add_flow() will automatically resolve to that port
• Subgraphs can be used with start_op(), just like regular operators
• Dynamic flow routing works with Subgraphs through their interface ports
• Nested Subgraphs are supported, allowing hierarchical control flow composition

For a complete example, see sequential_with_subgraph, which demonstrates a workflow where node2 and node3 are encapsulated in a Subgraph with execution interface ports.

Flow Control Pattern Selection Guide

Here’s when to choose different flow control patterns:

start_op() + Cyclic Flow

• Best for: Dynamic routing, feedback loops, runtime-adaptive flows
• Use when: Flow patterns depend on data content or need to change during execution
• Advantages: Flexible, handles complex routing
• Trade-offs: More complex to debug, slightly higher runtime overhead

Generator (root operator) with condition (CountCondition, PeriodicCondition, etc.)

• Best for: Fixed iteration counts, simple linear flows
• Use when: Number of iterations is known in advance (or infinite), static flow patterns
• Advantages: Simple to implement, better performance, easier to debug
• Trade-offs: Less flexible, cannot adapt to runtime conditions

Examples

Please see full examples under the examples/flow_control folder in the Holoscan SDK repository for more detailed implementations and use cases.

Note that the execution control examples are also related to the dynamic behavior of operators, and are available in the execution_control directory.

Best Practices

1. Clear Flow Logic: Keep dynamic flow logic clear and well-documented. Use meaningful names for operators and document the conditions that trigger different flow paths.

2. Error Handling: Handle edge cases in flow callbacks. Consider what happens if expected operators are not available or if flow conditions are invalid.

3. State Management: Be careful with shared state in dynamic flows. Ensure thread safety when multiple operators access shared resources.

4. Performance: Consider the overhead of frequent flow changes. Cache flow information when possible and avoid unnecessary flow modifications.

5. Testing: Test all possible flow paths thoroughly. Create unit tests that verify both normal operation and edge cases for each flow pattern.

Limitations

The dynamic flow control feature has the following limitations:

• It can only be used to connect operators within the same fragment. For inter-fragment flows (connecting operators across different fragments), explicit non-execution ports must be used instead of dynamic flows.
• Both the source and target operators must be native Holoscan operators (not GXF Operators).