CODEWITHSUNDEEP
c++multithreadingasynchronoustbbtbb-flow-graph
lizarisk
lizarisk

Reputation: 7820

TBB flow graph conditional execution

It possible to control execution path in the TBB Flow Graph dynamically, using output of a node as a condition variable to determine whether another node should be launched?

Upvotes: 1

Views: 1694

Answers (1)

cahuson
cahuson

Reputation: 846

There are a couple ways to dynamically control where messages go in a flow::graph:

You can explicitly put messages to other nodes in the body of a node. Notice that func_body puts a message to f1 or f2, depending on the value of its input. The nodes are not attached by make_edge(), because the flow of messages is not controlled by the topology of the graph:

template<typename T>
struct func_body {
    typedef tbb::flow::function_node<T,T> target_node_type;
    target_node_type &my_n1;
    target_node_type &my_n2;
    func_body(target_node_type &node1, target_node_type &node2) : my_n1(node1), my_n2(node2) {}
    tbb::flow::continue_msg operator()(const T& in) {
        // do some computation
        bool send_to_one = in > 0;
        if(send_to_one) my_n1.try_put(in);
        else            my_n2.try_put(in);
        return tbb::flow::continue_msg();  // message is discarded if no successor exists
    }
};

struct otherbody {
    int operator()(const int& in) {
        return in;
    }
};

int
main() {
    tbb::flow::graph g;
    tbb::flow::function_node<int,int> f1(g, tbb::flow::unlimited, otherbody());
    tbb::flow::function_node<int,int> f2(g, tbb::flow::unlimited, otherbody());
    tbb::flow::function_node<int> cn(g, tbb::flow::unlimited, func_body<int>(f1,f2));
}

Or you can use a multifunction_node (note the types of the tuple and get templates are in the tbb::flow namespace, not std:: as in the old documentation.) Notice in this case we attach f1 and f2 to the output ports of the multifunction_node.

typedef tbb::flow::multifunction_node<int,tbb::flow::tuple<int,int> > mfnode;

struct mfunc_body {
    void operator()(const int& in, mfnode::output_ports_type &op) {
        // do some computation
        bool send_to_one = in > 0;
        if(send_to_one) tbb::flow::get<0>(op).try_put(in);
        else            tbb::flow::get<1>(op).try_put(in);
    }
};

struct otherbody {
    int operator()(const int& in) {
        return in;
    }
};

int
main() {
    tbb::flow::graph g;
    tbb::flow::function_node<int,int> f1(g, tbb::flow::unlimited, otherbody());
    tbb::flow::function_node<int,int> f2(g, tbb::flow::unlimited, otherbody());
    mfnode cn(g, tbb::flow::unlimited, mfunc_body());
    tbb::flow::make_edge(tbb::flow::output_port<0>(cn), f1);
    tbb::flow::make_edge(tbb::flow::output_port<1>(cn), f2);

    // ...
}

Currently these two methods are functionally-identical; each will spawn a task to execute the body of the function_nodes. In the future the multifunction_node case may be optimized to not spawn if only one output port is try_put() to.

Upvotes: 5

Related Questions