c++ - 如何在 tbb 流程图中中止节点及其子节点的执行

Question

我目前正在测试 tbb 的流程图功能。为了使用它，我必须能够中止图中某些节点的执行，包括所有依赖它的子节点，但让其他不依赖它的子节点继续执行。从主体抛出异常或调用 task::cancel_group_execution() 中止所有节点的执行。

#include <cstdio>
#include "tbb/flow_graph.h"

using namespace tbb::flow;

struct body
{   std::string my_name;
    body( const char *name ) : my_name(name)
    {
    }
    void operator()( continue_msg ) const
    {   if (my_name == "B")
            tbb::task::self().group()->cancel_group_execution();
        else
        {   sleep(1);
            printf("%s\n", my_name.c_str());
        }
    }
};

int main()
{
    graph g;

    broadcast_node< continue_msg > start(g);
    continue_node<continue_msg> a( g, body("A"));
    continue_node<continue_msg> b( g, body("B"));
    continue_node<continue_msg> c( g, body("C"));
    continue_node<continue_msg> d( g, body("D"));
    continue_node<continue_msg> e( g, body("E"));

    make_edge( start, a );
    make_edge( start, b );
    make_edge( a, c );
    make_edge( b, c );
    make_edge( c, d );
    make_edge( a, e );

    for (int i = 0; i < 3; ++i )
        try
        {   start.try_put( continue_msg() );
            g.wait_for_all();
        } catch (...)
        {   printf("Caught exception\n");
        }
    return 0;
}

Answer 1

如果你希望能够取消部分图的执行，你需要使用task_group_contexts。添加以下内容:

#include "tbb/task.h"

并将您的主程序更改为以下内容:

int main()
{
    tbb::task_group_context tgc1;
    tbb::task_group_context tgc2;
    graph g1(tgc1);
    graph g2(tgc2);
    printf("Constructing graph\n");
    broadcast_node< continue_msg > start(g1);
    continue_node<continue_msg> a( g1, body("A"));
    continue_node<continue_msg> b( g2, body("B"));
    continue_node<continue_msg> c( g2, body("C"));
    continue_node<continue_msg> d( g2, body("D"));
    continue_node<continue_msg> e( g1, body("E"));

    make_edge( start, a );
    make_edge( start, b );
    make_edge( a, c );
    make_edge( b, c );
    make_edge( c, d );
    make_edge( a, e );

    for (int i = 0; i < 3; ++i ) {
        try
        {   
            printf("broadcasting graph %d\n", i);
            start.try_put( continue_msg() );
            g1.wait_for_all();
            g2.wait_for_all();
        } catch (...)
        {   printf("Caught exception\n");
        }
        g1.wait_for_all();
        g1.reset();
        g2.reset();
    }
    return 0;
}

每个 task_group_context 都是(默认)父上下文的子上下文。取消 g2 不会影响 g1。如果 B 抛出而不是取消，您的 catch 将确保异常不会传递给父级。如果不捕获异常，父上下文也将被取消，A 和 E 的上下文也是如此。

请注意，您必须等待两个图表完成。此外，您必须 reset() 图表以重置 continue_nodes 计数器。实际上，在抛出并捕获异常的情况下，并不能保证catch(...)完成后g1就完成了，所以需要做一个g1。 wait_for_all() 在 try/catch 之外。我编辑了代码以显示这一点。

不是使用取消来停止部分计算，您可以使 B 成为一个 multifunction_node，输入为 continue_msg，输出为 continue_msg:

typedef multifunction_node<continue_msg, tuple<continue_msg> > mf_type;

struct mf_body {
    std::string my_name;
    mf_body(const char *name) : my_name(name) {}
    void operator()(continue_msg, mf_type::output_ports_type &op) {
        if(my_name == "B") {
            printf("%s returning without sending\n", my_name.c_str());
            return;
        }
        sleep(1);
        get<0>(op).try_put(continue_msg());
        return;
    }
};

然后你创建节点B:

mf_type b( g, unlimited, mf_body("B"));

从 B 到 C 的边会像这样设置:

make_edge( output_port<0>(b), c ); 

在这种情况下，您不需要将图拆分为两个子图。如果节点 B 会取消，它会返回而不会将 continue_msg 转发给它的后继者。如果节点 B 不转发消息，节点 C 将不会执行，因为它需要两个 continue_msgs 才能启动。之后您仍然需要重置图表，以重置 C 的计数。

multifunction_node 的优点是您可以选择是否转发消息。这里需要注意的是，带有 continue_msg 输入的 multifunction_node 与 continue_node 不同。 continue_node 需要与其前任一样多的 continue_msgs(加上构造时的初始化值)。multifunction_node 主体在收到 时执行code>continue_msg，不管它有多少前驱。因此，对于您的图表，您不能只将所有节点设为 multifunction_nodes。