TBB Flow Graph：如何等待输出中的特定令牌？答案

【问题标题】：TBB Flow Graph: how wait for a specific token from the output?TBB Flow Graph：如何等待输出中的特定令牌？
【发布时间】：2013-07-26 11:28:28
【问题描述】：

我正在尝试为 Flow Graph 编写一个适配器，以模仿类似管道的同步函数调用。但我不明白如何阻止并等待特定令牌的输出。在图表上调用 wait_for_all 并没有帮助，因为我不需要等待所有值。有人可以提出解决方案吗？

template <typename TOutput, typename TInput>
class FlowPathAdapter {
public:
    TOutput operator()(const TInput& val) {
        m_input->try_put(val);
        TOutput result;
        // What should be done here to ensure that
        // m_output returns the result corresponding to this specific token?
        m_output->try_get(result);
        return result;
    }

private:
    // input and output are connected in some graph constructed outside the adapter
    std::shared_ptr<tbb::flow::receiver<TInput>> m_input;
    std::shared_ptr<tbb::flow::sender<TOutput>> m_output;
};

【问题讨论】：

标签： c++ multithreading concurrency tbb tbb-flow-graph

【解决方案1】：

接收者和发送者类是“纯虚拟”类（try_* 方法有默认行为以简化节点的定义。）对于每个 flow::graph 节点，它们都被特定的行为覆盖类。

如果您希望创建特定类型的节点（或适配器），您应该覆盖虚拟方法。但是，您可以使用 function_node 来执行您尝试使用此适配器执行的操作。您可以删除 try_get 和 try_put 调用（这是 function_node 行为的一部分。）

// notice TInput and TOutput are exchanged
template < typename TInput, typename TOutput >
class FlowPathAdapterBody {
public:
    TOutput operator()(const TInput& val) {
        // val has the value passed to us
        TOutput result;
        // computation turning val into result
        return result;
    }
};

tbb::flow::graph g;
// node is fully-parallel in this case
tbb::flow::function_node<int, int> fnode(g, tbb::flow::unlimited, FlowPathAdapterBody<int,int>());

// ...

您可以将多个前任附加到 function_node 的输入，并将多个后继附加到其输出。

默认情况下，function_node 在其输入上有一个缓冲区，因此即使节点是串行的，它也将始终接受输入。如果您希望它拒绝输入而不是缓冲它们，您可以将节点设为 function_node<T,U,tbb::flow::rejecting>.

如果您想要更高级的行为（例如每个输入发出多条消息的能力，或者可选择不为某些输入发出消息，或者发出特殊条件的信号，请查看 multifunction_node。如果是，multifunction_node 只能有一个输出所有这些都需要。）

问候，克里斯

【讨论】：

你只是建议写一个函数节点。我想要一个现有图的适配器，它将一个令牌发送到一个节点并等待它来自另一个节点。

【解决方案2】：

在 TBB 中通常避免等待；等待是徒劳的。

TBB 中的任务执行时间也不能保证，因为我们依赖于操作系统进行调度。也就是说，在转发来自节点的消息之后，不会发生诸如原子操作之类的带外信令，因此结果不会准备好。您还必须使用queue_node 或其他缓冲节点来保存结果。

鉴于此，如果您希望进行显式等待，您可以创建一个function_node 来接收最终答案，将其分配给结果位置，并发出信号以告知结果已准备好。（信号应该是一个原子以强制执行围栏。）您的其他任务可以在原子上旋转等待。或者你可以使用条件变量。

问候，克里斯

【讨论】：