控制流指的是程序中操作的执行顺序。 控制流是通过使用条件语句、循环等控制结构调节的。 此外，数据流指的是一种编程模型，在这种模型中，只有当所有需要的数据都可用时，才会进行计算。 数据流编程模型与消息传递这一概念相关，其中程序的独立组件通过发送消息相互通信。
异步代理同时支持控制流和数据流编程模型。 尽管控制流模型适用于许多情况，但数据流模型也适用于其他一些情况，例如，当代理接收数据并执行基于该数据的有效负载的操作时。下面演示演示如何创建基于数据流，而不是基于控制流的基于代理的应用程序。
创建基本控制流代理
请看下面的示例，该示例定义了 control_flow_agent 类。 control_flow_agent 类对三个消息缓冲区(一个输入缓冲区和两个输出缓冲区)执行操作。 run 方法在一个循环中从源消息缓冲区读取，并使用条件语句来引导程序执行流。 代理针对非零负值递增一个计数器值，并为非零正值递增另一个计数器值。 代理收到为零的 Sentinel 值后，它会将计数器的值发送到输出消息缓冲区。 通过 negatives 和 positives 方法，应用程序可以从代理读取负值和正值的计数。
// A basic agent that uses control-flow to regulate the order of program
// execution. This agent reads numbers from a message buffer and counts the
// number of positive and negative values.
class control_flow_agent : public agent
{
public:explicit control_flow_agent(ISource<int>& source): _source(source){}// Retrieves the count of negative numbers that the agent received.size_t negatives() {return receive(_negatives);}// Retrieves the count of positive numbers that the agent received.size_t positives(){return receive(_positives);}protected:void run(){// Counts the number of negative and positive values that// the agent receives.size_t negative_count = 0;size_t positive_count = 0;// Read from the source buffer until we receive// the sentinel value of 0.int value = 0; while ((value = receive(_source)) != 0){// Send negative values to the first target and// non-negative values to the second target.if (value < 0)++negative_count;else++positive_count;}// Write the counts to the message buffers.send(_negatives, negative_count);send(_positives, positive_count);// Set the agent to the completed state.done();}
private:// Source message buffer to read from.ISource<int>& _source;// Holds the number of negative and positive numbers that the agent receives.single_assignment<size_t> _negatives;single_assignment<size_t> _positives;
};虽然在此示例中看到的是代理中的控制流的基本用法，但它展示了基于控制流的编程的连续性。 必须按顺序处理每个消息，即使在输入消息缓冲区中可能有多个消息。 数据流模型使条件语句的两个分支能同时求值。 数据流模型还使你能够创建更复杂的消息传送网络，在数据可用时对其执行操作。
创建基本数据流代理
本部分介绍如何转换 control_flow_agent 类以使用数据流模型执行相同的任务。
数据流代理的工作方式是创建一个消息缓冲区网络，每个缓冲区都有特定用途。 某些消息块使用筛选功能，根据消息的有效负载接受或拒绝消息。 筛选器函数可确保消息块仅接收特定值。
将控制流代理转换为数据流代理
1. 将 control_flow_agent 类的主体复制到另一个类，例如 dataflow_agent。 或者，可以重命名 control_flow_agent 类。
2. 从 run 方法移除用来调用 receive 的循环主体。
void run()
{// Counts the number of negative and positive values that// the agent receives.size_t negative_count = 0;size_t positive_count = 0;// Write the counts to the message buffers.send(_negatives, negative_count);send(_positives, positive_count);// Set the agent to the completed state.done();
}3. 在 run 方法中，在变量 negative_count 和 positive_count 初始化后，添加 countdown_event 对象，用于跟踪活动操作的计数。
// Tracks the count of active operations.
countdown_event active;
// An event that is set by the sentinel.
event received_sentinel;4. 创建将参与数据流网络的消息缓冲区对象。
//// Create the members of the dataflow network.//// Increments the active counter.transformer<int, int> increment_active([&active](int value) -> int {active.add_count();return value;});// Increments the count of negative values.call<int> negatives([&](int value) {++negative_count;// Decrement the active counter.active.signal();},[](int value) -> bool {return value < 0;});// Increments the count of positive values.call<int> positives([&](int value) {++positive_count;// Decrement the active counter.active.signal();},[](int value) -> bool {return value > 0;});// Receives only the sentinel value of 0.call<int> sentinel([&](int value) { // Decrement the active counter.active.signal();// Set the sentinel event.received_sentinel.set();},[](int value) -> bool { return value == 0; });// Connects the _source message buffer to the rest of the network.unbounded_buffer<int> connector;5. 连接消息缓冲区以构成网络。
//
// Connect the network.
//// Connect the internal nodes of the network.
connector.link_target(&negatives);
connector.link_target(&positives);
connector.link_target(&sentinel);
increment_active.link_target(&connector);// Connect the _source buffer to the internal network to
// begin data flow.
_source.link_target(&increment_active);6.等待 event 和 countdown event 对象设置完成。 这些事件表明代理已收到 Sentinel 值，并且所有操作都已完成。
// Wait for the sentinel event and for all operations to finish.
received_sentinel.wait();
active.wait();
