使用C ++ 11异步功能的管道数据流 [英] Pipeline dataflow using C++11 async features

问题描述

我正在尝试实现具有以下功能的多线程管道数据流框架:

I'm trying to implement a multithreaded pipeline dataflow framework with the following features:

1. 管道可以描述为非循环有向图.每个节点执行一些处理，并具有任意数量的任意类型的输入和一个任意类型的输出.

1. The pipeline can be described as an acyclic directed graph. Each node performs some processing and has arbitrary number of inputs of arbitrary types and one output of arbitrary type.

对于每个给定的输入数据实例，每个节点执行的次数不应超过一次，之后应将结果缓存.尽管此高速缓存不应在内存中保留的时间更长，而应在其他任何节点不再需要时将其删除.

For each given input data instance each node should be executed not more that once, after that the result should be cached. Although this cache shouldn't persist in memory longer that required and should be deleted when no longer needed by any other nodes.

每个节点都应支持惰性求值，即仅应在某个其他节点需要其输出时才执行.

Each node should support lazy evaluation, i.e. should be executed only at the moment it's output is needed by some another node.

是否有可能通过使用C ++ 11多线程功能来实现，特别是 std :: future ， std :: promise 和 std :: async?有人可以提供线索吗?

Is it possible to implement this by using C++11 multithreading features, especially std::future, std::promise and std::async? Can anybody give a clue?

推荐答案

我相信使用 async 框架实际上确实很简单.

I believe it actually is fairly trivial using an async framework.

如果您查看 std :: launch ，您会发现其中存在一个延迟模式:

If you look at std::launch you will realize there is a deferred mode:

• std :: launch :: deferred :任务在第一次请求其结果时在调用线程上执行(延迟评估)

• std::launch::deferred: the task is executed on the calling thread the first time its result is requested (lazy evaluation)

因此，您可以启动任务并仅在需要结果时才执行任务.但是，由于提到了一个非循环图，您可能希望共享结果: std :: future (由对 std :: async 的调用返回)不能共享;为此，您需要一个 std :: shared_future .

Thus, you can launch a task and have it executed only when the result is needed. However, since you mention an acyclic graph, you probably want to share the result: a std::future (as returned by a call to std::async) cannot be shared; you need a std::shared_future for this.

因此，将其放在一起:

// Disclaimer:
// Compiles but does not run, but I have not figured it out.
// See: http://ideone.com/XZ49Dg

#include <future>
#include <iostream>

int main() {
    std::shared_future<std::string> greeting = std::async(std::launch::deferred, []() {
        std::string s;
        std::cout << "Enter how you would like to be greeted: ";
        std::getline(std::cin, s);
        return s;
    });

    std::shared_future<int> choice = std::async(std::launch::deferred, []() {
        int c = 0;
        std::cout << "Pick any integer: ";
        std::cin >> c;
        return c;
    });

    std::shared_future<void> complete = std::async(std::launch::deferred, [=]() mutable {
        std::string const g = greeting.get();
        int const c = choice.get();

        std::cout << "Hello " << g << ", you picked " << c << "!\n";
    });

    complete.wait();
}

这篇关于使用C ++ 11异步功能的管道数据流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！