如何使用动态端口数制作Intel TBB multifunction_node？

Question

我是英特尔TBB库的新手.如您所见,我的问题与tbb :: flow :: graph有关.我需要实现如下逻辑:

用户使用一些逻辑块绘制图形.每个块(节点)可以具有无限的连接(边缘),因此每个块(节点)可以选择接下来放置数据的位置.然后我的程序将在TBB库的帮助下构建这样的图形并执行计算.

所以我不知道是否有可能构造具有动态输出端口数的节点(我想它必须是multifunction_node).你能告诉我这样做的方法吗？

Answer 1

遗憾的是,没有办法(没有动态编译)来改变multifunction_node中的输出端口数.您可以创建最大数量的端口(由宏开关控制并依赖于编译器),并且只是动态连接到端口.如果对端口执行try_put并且没有附加后继,则try_put将失败,您可以在运行时对此做出反应.

另一种方法(尽管有些挫折,我认为)是构建一个双端口多功能块的二叉树.如果使用具有输出目标的类作为字段,则构造每个节点以对目标的一位作出反应并输出到端口0或端口1,具体取决于掩码的结果.调度程序短路会在树中相对快速地控制输出,但是你需要为多个动态调用支付一点点罚款.

或者你可以使用除2之外的其他一些基础(例如,10)

附录:在与Mike(flow :: graph的设计者)交谈之后,我们意识到还有另一种方法来处理这个问题,这将允许动态数量的端口.你将不得不做一些低级别的东西,但它是这样的:

#include "tbb/tbb.h"
#include <iostream>

using namespace tbb::flow;

tbb::spin_mutex io_lock;
typedef broadcast_node<int> bnode_element_t;
typedef tbb::concurrent_vector<bnode_element_t *> output_port_vector_t;
struct multioutput_function_body {
    output_port_vector_t &my_ports;
    public:
    multioutput_function_body(output_port_vector_t &_ports) : my_ports(_ports) {}
    multioutput_function_body(const multioutput_function_body &other) : my_ports(other.my_ports) { }
    continue_msg operator()(const int in) {
        int current_size = my_ports.size();
        if(in >= current_size) {
            // error condition?  grow concurrent_vector?
            tbb::spin_mutex::scoped_lock gl(io_lock);
            std::cout << "Received input out of range(" << in << ")" << std::endl;
        }
        else {
            // do computation
            my_ports[in]->try_put(in*2);
        }
        return continue_msg();
    }
};

struct output_function_body {
    int my_prefix;
    output_function_body(int i) : my_prefix(i) { }
    int operator()(const int i) {
        tbb::spin_mutex::scoped_lock gl(io_lock);
        std::cout << " output node "<< my_prefix << " received " << i << std::endl;
        return i;
    }
};

int main() {
    graph g;
    output_port_vector_t output_ports;
    function_node<int> my_node(g, unlimited, multioutput_function_body(output_ports) );
    // create broadcast_nodes
    for( int i = 0; i < 20; ++i) {
        bnode_element_t *bp = new bnode_element_t(g);
        output_ports.push_back(bp);
    }

    // attach the output nodes to the broadcast_nodes
    for(int i = 0; i < 20; ++i) {
        function_node<int,int> *fp = new function_node<int,int>(g, unlimited, output_function_body(i));
        make_edge(*(output_ports[i]),*fp);
    }

    for( int i = 0; i < 21; ++i) {
        my_node.try_put(i);
    }
    g.wait_for_all();
    return 0;
}

以上注释:

• 我们正在创建一个concurrent_vector指针broadcast_nodes.这些的继承者function_node都附属于这些broadcast_nodes.function_node忽略输出.
• concurrent_vector被传递给的构造函数multioutput_function_body.在这种情况下,我们根本不需要多功能.在multioutput_function_body决定哪些broadcast_node要try_put在运行时. 请注意我们是明确try_puts的broadcast_nodes.这导致为每个任务生成一个任务try_put.生成的任务比排队的任务更快,但是比从节点返回值更多的调度开销.
• 我没有添加堆分配broadcast_nodes和输出的清理function_nodes.删除的"显而易见"的地方broadcast_nodes将在析构函数中multioutput_function_body.你不应该这样做,因为在function_node传入的函数体的复制构造中创建结果,并且function_body会有多个副本,它们会引用broadcast_node指针的concurrent_vector .删除后删除g.wait_for_all().

我concurrent_vector之所以使用,是因为它允许在修改时访问指针concurrent_vector.是否broadcast_node可以在执行图形期间添加其他指针的问题是开放的.我希望你只是创建节点并按原样使用它们,而不是即时修改它们. concurrent_vectors在增长结构时不要重新分配和移动已经初始化的元素; 这就是我使用它的原因,但如果您希望在图表运行时添加其他节点,请不要认为这是一个完整的答案.