带有回调的异步未来。 C++11

Question

我有一份期货清单。问题是我有很多文件，在创建每个文件后我需要进行一些长时间的操作。这就是为什么我想在每次“文件保存”之后进行回调。

例如，

  (new thread; saveFile 1.txt -> new thread; do a long operation after the file has been created)
  (new thread; saveFile 2.pdf -> new thread; do a long operation after the file has been created).

我需要在单独的线程中完成所有操作。文件的保存至关重要，在创建文件之前无法运行第二个任务。 我该怎么做？ 我有以下代码：

 void save_file() {
     // preparing data...
     saving a file
   } 

   std::vector<std::future<void>> saveFileTasks;
   for (int n = 0; n < p.size(); ++n)
   {
      saveFileTasks.push_back(std::async(std::bind(&saveFile, filename)));
   }

   for (auto &e : saveFileTasks) {
      e.get();
   }

如何在 C++11 中使用 future/promise 进行 回调？我不允许在我的项目中使用 boost。

我真的很困惑，一个非常简单的任务有这么多复杂的例子。许多示例无法编译，例如，promise.set_wait_callback 在 C++11 中不存在，但许多函数已迁移到 C++11。 如果我使用 Python 或 Clojure，我可以很容易地做到这一点。我怎样才能用 C++ 做到这一点？

Answer 1

未来，future 将有一个 .then 运算符，可以让您链接任务。

没有它我们可以写它。

// complete named operator library in about a dozen lines of code:
namespace named_operator {
  template<class D>struct make_operator{ constexpr make_operator() {}; };

  template<class T, class O> struct half_apply { T&& lhs; };

  template<class Lhs, class Op>
  half_apply<Lhs, Op> operator*( Lhs&& lhs, make_operator<Op> ) {
    return {std::forward<Lhs>(lhs)};
  }
  template<class Lhs, class Op, class Rhs>
  decltype(auto) operator*( half_apply<Lhs, Op>&& lhs, Rhs&& rhs )
  {
    return named_invoke( std::forward<Lhs>(lhs.lhs), Op{}, std::forward<Rhs>(rhs) );
  }
}

// create a named operator then:
namespace then_ns {
  static const struct then_t:named_operator::make_operator<then_t> {} then{};

  namespace details {
    template<size_t...Is, class Tup, class F>
    auto invoke_helper( std::index_sequence<Is...>, Tup&& tup, F&& f )
    ->decltype(std::forward<F>(f)( std::get<Is>(std::forward<Tup>(tup))... ))
    {
      return std::forward<F>(f)( std::get<Is>(std::forward<Tup>(tup))... );
    }
  }

  // first overload of A *then* B handles tuple and tuple-like return values:
  template<class Tup, class F>
  auto named_invoke( Tup&& tup, then_t, F&& f )
  -> decltype( details::invoke_helper( std::make_index_sequence< std::tuple_size<std::decay_t<Tup>>{} >{}, std::forward<Tup>(tup), std::forward<F>(f) ) )
  {
    return details::invoke_helper( std::make_index_sequence< std::tuple_size<std::decay_t<Tup>>{} >{}, std::forward<Tup>(tup), std::forward<F>(f) );
  }

  // second overload of A *then* B
  // only applies if above does not:
  template<class T, class F>
  auto named_invoke( T&& t, then_t, F&& f, ... )
  -> std::result_of_t< F(T) >
  {
    return std::forward<F>(f)(std::forward<T>(t));
  }
  // *then* with a future; unpack the future
  // into a call to f within an async:
  template<class X, class F>
  auto named_invoke( std::future<X> x, then_t, F&& f )
  -> std::future< std::decay_t<decltype( std::move(x).get() *then* std::declval<F>() )> >
  {
    return std::async( std::launch::async,
      [x = std::move(x), f = std::forward<F>(f)]() mutable {
        return std::move(x).get() *then* std::move(f);
      }
    );
  }
  // void future, don't try to pass void to f:
  template<class F>
  auto named_invoke( std::future<void> x, then_t, F&& f )
  -> std::future< std::decay_t<decltype( std::declval<F>()() )> >
  {
    return std::async( std::launch::async,
      [x = std::move(x), f = std::forward<F>(f)]() mutable {
        std::move(x).get();
        return std::move(f)();
      }
    );
  }
}
using then_ns::then;

看，这并不难。

a *then* f，如果a 是一个元组（或对或数组），将使用a 的内容调用f。

如果a 不是类似元组的，或者f 不接受a 的内容，它会使用a 调用f。

如果a 是一个未来，它会创建一个新的异步未来，使用*then* 消耗a.get()。

Live example.

假设你想在文件保存时增加一个原子 int：

std::vector<std::future<void>> saveFileTasks;
for (int n = 0; n < p.size(); ++n)
{
  saveFileTasks.push_back(
    std::async(std::launch::async, [filename]{
      saveFile(filename);
    })
  );
}
std::atomic<int> count;
for (auto &e : saveFileTasks) {
  e = std::move(e) *then* [&count]{
    ++count;
  });
}

当然，这一切都可以在没有命名运算符*then* 样式语法的情况下完成，但这样做有什么乐趣呢？

如果第一个异步返回一个元组，第二个可以将它作为一个元组或解压缩的“平面”参数。

Answer 2

不幸的是，std::future 的当前版本中没有 .then 延续 - 它与类似的实用程序一起被提议用于未来的 C++ 标准。

如果你不能使用boost，你可以用函数组合构建你自己的延续：

string save_file(string data)      { /* ... */ return filename; } // step 1
void do_operation(string filename) { /* ... */ }                  // step 2

// ... 

std::vector<std::future<void>> fileTasks;
for(const auto& data : /* ... */)
{
    fileTasks.emplace_back(std::async(std::launch::async, 
        [data]{ do_operation(save_file(data)); });
}

注意std::async([data]{ do_operation(save_file(data)); } 将在同一个线程中执行这两个函数。如果您希望每个函数在单独的线程中执行，您可以多次调用async：

std::async(std::launch::async, [data]
{
    auto result = save_file(data);
    std::async(std::launch::async, [r = std::move(result)]
    {
        do_operation(std::move(r));
    });
});

---

对于boost::future 或标准的未来版本，您可以简单地说：

std::async(std::launch::async, [data]{ save_file(data); })
    .then([](string filename){ do_operation(filename); );

Answer 3

尽管我迟到了，但我还是要提到，回调可以很容易地用范围保护来实现。

长话短说：

• 添加一个作用域保护，利用 RAII 在作用域末尾调用回调，这发生在长时间运行的函数退出后，

• 创建一个包装函数，该函数调用长时间运行的函数，该函数还通过回调声明范围保护，并且

• 创建一个std::future 来启动封装的长时间运行函数。

这是一个最小的工作示例：

#include <iostream>
#include <future>
#include <functional>

class ScopedGuard
{
public:
    ScopedGuard(std::function<void()> callback)
        : m_callback(callback) {}

    ~ScopedGuard() {
        m_callback();
    }
private:
    std::function<void()> m_callback;
};

void my_callback() {
    std::cout << "Calling callback." << std::endl;
}

int my_long_running_task() {
    std::cout << "Calling long running task." << std::endl;
    return 8;
}

int main()
{
    std::cout << "Hello World!" << std::endl;

    std::future<int> f2 = std::async(std::launch::async, []{
        ScopedGuard sg(my_callback);
        return my_long_running_task();
    });
    return 0;
}

通过这种方法，您还可以定义一个特殊的 ScopeGuard 类，它包装了长时间运行的操作和回调，并将其作为函子传递给 std::async。