C++并发与多线程 9_asyncfuturepackaged_taskpromise

Posted 2021-10-05 TianSong

std::future

类模板

template <class T>  future;
template <class R&> future<R&>;     // specialization : T is a reference type (R&)
template <>         future<void>;   // specialization : T is void

• future 是一个对象，可以从某个提供对象或函数中检索值，如果在不同线程中，则可以正确同步此访问。

• “有效” future 对象，通过调用一下函数之一来构造：

  • async
  • promise::get_future
  • packaged_task::get_future
• future 对象仅在他们有效时才有用。默认构造的 future 对象五项（除非移动分配一个有效的 future）。
• 在有效的 future 上调用 future::get 会阻塞线程，直到提供程序准备好共享状态（通过设置值或异常）。这样，两个线程可以通过一个线程同步，等待另一个线程设置值。
• 共享状态的生存期至少要持续到与之关联的最后一个对象释放它或销毁它为止。因此，如果与 future 相关联，共享状态可以在最初获得它的对象（如果有的话）之后继续存在。

future::valid()

bool valid() const noexcept;

检查有效的共享状态

• 返回 future 对象当前是否与共享状态关联。
• 对于默认构造的 future 对象，此函数返回 false (除非将有效的 future 分配给移动对象)。
• future 只能由某些提供函数（如， async, promise::get_future 或 packaged_task::get_future）使用有效的共享状态进行初始化。
• 一旦使用 future::get 检索了共享状态的值，则调用此函数返回 false (除非移动分配了一个新的 future).

返回值

• 如果对象与共享状态关联，则为 ture。
• 否则为假。

// future::valid
#include <iostream>       // std::cout
#include <future>         // std::async, std::future
#include <utility>        // std::move

int get_value() { return 10; }

int main ()
{
  std::future<int> foo,bar;
  foo = std::async (get_value);
  bar = std::move(foo);

  if (foo.valid())
    std::cout << "foo\'s value: " << foo.get() << \'\\n\';
  else
    std::cout << "foo is not valid\\n";

  if (bar.valid())
    std::cout << "bar\'s value: " << bar.get() << \'\\n\';
  else
    std::cout << "bar is not valid\\n";

  return 0;
}

输出：

foo is not valid
bar\'s value: 10

future::get()、wait()、wait_for(xxx)、wait_until(xxx)

future::get	当共享状态就绪时，返回存储在共享状态中的值（或引发其异常）
future::wait	等待共享状态就绪
future::wait_for	等待共享状态在rel_time指定的时间内准备就绪。
future::wait_until	等待共享状态准备就绪，最多等到abs_time时间点

future::get

获取值

generic template (1)    
T get();

reference specialization (2)    
R& future<R&>::get();       // when T is a reference type (R&)

void specialization (3)    
void future<void>::get();   // when T is void

• 当共享状态就绪时，返回存储在共享状态中的值（或引发其异常）。
• 如果共享状态尚未准备好（即提供程序尚未设置其值或异常），则该函数将阻塞调用线程直到准备就绪。
• 共享状态就绪后，该函数将取消阻塞并返回（或引发异常）以释放其共享状态。这时 future 对象不再有效：对于每个 future 的共享状态，此成员函数最多应被调用一次。
• 提供者准备好共享状态和返回此函数之间是同步的。
• 版本 (3) 不返回任何值，但仍等待共享状态准备就绪并释放它。

参数

• 无

返回值

• （1） std :: move（x），其中x是共享状态下存储的值。
• （2） 是对在共享状态下存储的值的引用。
• （3） 无。

// future::get
#include <iostream>       // std::cout, std::ios
#include <future>         // std::async, std::future
#include <exception>      // std::exception

int get_int() {
  std::cin.exceptions (std::ios::failbit);   // throw on failbit set
  int x;
  std::cin >> x;                             // sets failbit if invalid
  return x;
}

int main ()
{
  std::future<int> fut = std::async (get_int);

  std::cout << "Please, enter an integer value: ";

  try {
    int x = fut.get();
    std::cout << "You entered: " << x << \'\\n\';
  }
  catch (std::exception&) {
    std::cout << "[exception caught]";
  }

  return 0;
}

输出：

Please, enter an integer value: 101
You entered: 101

future::wait

void wait() const;

等待共享状态就绪

// future::wait
#include <iostream>       // std::cout
#include <future>         // std::async, std::future
#include <chrono>         // std::chrono::milliseconds

// a non-optimized way of checking for prime numbers:
bool is_prime (int x) {
  for (int i=2; i<x; ++i) if (x%i==0) return false;
  return true;
}

int main ()
{
  // call function asynchronously:
  std::future<bool> fut = std::async (is_prime,194232491);

  std::cout << "checking...\\n";
  fut.wait();

  std::cout << "\\n194232491 ";
  if (fut.get())      // guaranteed to be ready (and not block) after wait returns
    std::cout << "is prime.\\n";
  else
    std::cout << "is not prime.\\n";

  return 0;
}

输出：

checking...

194232491 is prime.

future::wait_for

template <class Rep, class Period>
  future_status wait_for (const chrono::duration<Rep,Period>& rel_time) const;

等待共享状态在rel_time指定的时间内准备就绪

值	描述
future_status::ready	共享状态已就绪：生产者已设置值或异常
future_status::timeout	该函数等待rel_time，但共享状态未准备就绪
future_status::deferred	共享状态包含延迟功能

// future::wait_for
#include <iostream>       // std::cout
#include <future>         // std::async, std::future
#include <chrono>         // std::chrono::milliseconds

// a non-optimized way of checking for prime numbers:
bool is_prime (int x) {
  for (int i=2; i<x; ++i) 
      if (x%i==0) 
          return false;
  return true;
}

int main ()
{
  // call function asynchronously:
  std::future<bool> fut = std::async (is_prime,700020007);

  // do something while waiting for function to set future:
  std::cout << "checking, please wait";
  std::chrono::milliseconds span (100);
  while (fut.wait_for(span)==std::future_status::timeout)
    std::cout << \'.\';

  bool x = fut.get();

  std::cout << "\\n700020007 " << (x?"is":"is not") << " prime.\\n";

  return 0;
}

输出：

checking, please wait.........................
700020007 is prime.

future::wait_until

template <class Clock, class Duration>
  future_status wait_until (const chrono::time_point<Clock,Duration>& abs_time) const;

等待共享状态准备就绪，最多等到 abs_time 时间点

返回值同 future::wait_for

std::async

函数模板

unspecified policy (1)    
template <class Fn, class... Args>
  future<typename result_of<Fn(Args...)>::type>
    async (Fn&& fn, Args&&... args);

specific policy (2)    
template <class Fn, class... Args>
  future<typename result_of<Fn(Args...)>::type>
    async (launch policy, Fn&& fn, Args&&... args);

异步调用函数

• 异步调用函数在某个时刻调用 fn (以 args 作为参数)，返回时无需等待 fn 执行完成。
• 可以通过返回的 future 对象（通过调用其成员 future::get）来访问 fn 返回的值。
• 第二个版本 （2） 允许调用者选择特定的启动策略，而第一个版本 （1） 使用自动选择，就好像调用 (2) 并将 launch::aysnc | launch::deferred 作为策略。
• 该函数在共享状态下临时存储使用的线程处理程序。一旦完成 fn 的执行，共享状态将包含 fn 返回的值并准备就绪。

参数

policy

枚举，启动类型的位掩码值，指示启动策略：

enum class launch : /* unspecified */ {
    async =    /* unspecified */,
    deferred = /* unspecified */,
    /* implementation-defined */
};

策略	描述
launch::async	启动一个新的线程以调用 fn （就像使用 fn 和 args 作为参数构造线程对象，并访问返回的 future 的共享状态将其联接）
launch::deferred	对 fn 的调用将推迟到访问返回的共享状态 （使用 wait 或 get）之前。此时，将在调用线程调用 fn, 并且不再考虑函数被推迟。当此调用返回时，返回的 future 的共享状态已准备就绪
launch::async | launch::deferred	该功能自动（在某个时候）选择策略。这取决于系统和库的实现，他们通常会针对系统中当前的并发可用性进行优化

fn

• 指向函数的指针，指向成员的指针或任何类型的可移动构造的函数对象（即，其类定义了 operator() 的对象，包括闭包和函数对象）。该函数使用此参数的std::decay副本。 fn 的返回值存储为共享状态，以供异步返回的 future 对象检索。如果 fn 抛出异常，则将异常设置为共享状态，以供 future 对象检索。

args...

• 传递给 fn 调用的参数（如果有）。他们的类型应是可移动构造的。如果 fn 是成员指针，则第一个参数应是为其定义了该成员的对象（或其引用或指向它的指针）。该函数使用这些参数的std::decay副本

返回值

• 当 fn 的执行结束时，一个共享状态的 future 对象准备就绪。其成员 future::get 检索的值是 fn 返回的值（如果有）。
• 当选择 launch::async，返回的 future 链接到创建的线程的末尾，即使它的共享状态从未访问：在这种情况下，它的析构函数与 fn 的返回同步。因此，即使 fn 返回 void, 异步行为也不应该忽略该返回值。

Test1.cpp

// async example
#include <iostream>       // std::cout
#include <future>         // std::async, std::future

// a non-optimized way of checking for prime numbers:
bool is_prime (int x) {
  std::cout << "Calculating. Please, wait...\\n";
  for (int i=2; i<x; ++i)
      if (x%i==0)
          return false;
  return true;
}

int main ()
{
  // call is_prime(313222313) asynchronously:
  std::future<bool> fut = std::async (is_prime,313222313);

  std::cout << "Checking whether 313222313 is prime.\\n";
  // ...

  bool ret = fut.get();      // waits for is_prime to return !!!!

  if (ret)
      std::cout << "It is prime!\\n";
  else
      std::cout << "It is not prime.\\n";

  return 0;
}

输出：

Checking whether 313222313 is prime.
Calculating. Please, wait...
It is prime!

Test2.cpp

#include <iostream>
#include <future>

using namespace std;

bool async_func () {

    cout << "async_func begin" << endl;

    cout << "async_func end" << endl;

    return true;
}

int main ()
{
  cout << "main begin" << endl;

  std::future<bool> fut = std::async (async_func);

  cout << "main end" << endl;

  return 0;
}

输出：

main begin
main end
async_func begin
async_func end

说明：依据打印信息，主线程（进程）等待 async 创建的子线程执行结束；
因为fut 的析构函数与 fn 的返回同步，阻塞主线程直到子线程 fn 执行结束。

Test3.cpp

#include <iostream>
#include <future>
#include <thread>

using namespace std;

bool async_func () {

    cout << "async_func begin " << std::this_thread::get_id() << endl;

    cout << "async_func end" << endl;

    return true;
}

int main ()
{
  cout << "main begin " << std::this_thread::get_id() << endl;

  std::future<bool> fut = std::async (launch::deferred, async_func);

  if (fut.wait_for(std::chrono::microseconds(10)) == std::future_status::deferred)
  {
    cout << "----" << endl;
    fut.wait();
  }

  cout << "main end" << endl;

  return 0;
}

输出：[注意，async_func 在主线程中被调用]

main begin 1
----
async_func begin 1
async_func end
main end

std::packaged_task

类模板

template <class T> packaged_task;     // undefined
template <class Ret, class... Args> class packaged_task<Ret(Args...)>;

• 包装可调用元素，并允许异步检索其结果。
• 类似于 std::function, 但是会自动将其结果传输到 future 对象。

• 对象内部包含两个元素：

  • 存储的任务是一些可调用对象（例如，函数指针，成员或函数对象的指针），其调用签名应采用 Args... 中类型的参数，并返回 Ret 类型的值。
  • 共享状态，该状态能够存储调用存储的任务（类型为 Ret）的结果，并且可以通过 future 来异步访问。

• 通过调用成员 get_future 将共享状态与 future 对象关联。调用之后，两个对象共享相同的共享状态：

  • packaged_task 对象是异步提供程序，通过调用存储的任务，可以在某个时刻将共享状态设置为就绪。
  • future 对象是一个异步返回对象，可以检索共享状态的值，并在必要时等待其准备就绪。

// packaged_task example
#include <iostream>     // std::cout
#include <future>       // std::packaged_task, std::future
#include <chrono>       // std::chrono::seconds
#include <thread>       // std::thread, std::this_thread::sleep_for

// count down taking a second for each value:
int countdown (int from, int to) {
  for (int i=from; i!=to; --i) {
    std::cout << i << \'\\n\';
    std::this_thread::sleep_for(std::chrono::seconds(1));
  }
  std::cout << "Lift off!\\n";
  return from-to;
}

int main ()
{
  std::packaged_task<int(int,int)> tsk (countdown);   // set up packaged_task
  std::future<int> ret = tsk.get_future();            // get future

  std::thread th (std::move(tsk),10,0);   // spawn thread to count down from 10 to 0

  // ...

  int value = ret.get();                  // wait for the task to finish and get result

  std::cout << "The countdown lasted for " << value << " seconds.\\n";

  th.join();

  return 0;
}

输出：

10
9
8
7
6
5
4
3
2
1
Lift off!
The countdown lasted for 10 seconds

std::promise

类模板

template <class T>  promise;
template <class R&> promise<R&>;     // specialization : T is a reference type (R&)
template <>         promise<void>;   // specialization : T is void

• promise 是一个对象，它可以存储 T 类型的值，由 future 对象（可能在另一个线程中）检索，并提供一个同步点。

• 通过调用成员get_future，可以将该共享状态与 future 对象关联。 调用之后，两个对象共享相同的共享状态：

  • promise对象是异步提供程序，应在某个时候为共享状态设置一个值。
  • future 对象是一个异步返回对象，可以检索共享状态的值，并在必要时等待其准备就绪。

// promise example
#include <iostream>       // std::cout
#include <functional>     // std::ref
#include <thread>         // std::thread
#include <future>         // std::promise, std::future

void print_int (std::future<int>& fut) {
  std::cout << "print_int begin " << std::endl;

  int x = fut.get();
  std::cout << "value: " << x << std::endl;

  std::cout << "print_int end " << std::endl;
}

int main ()
{
  std::promise<int> prom;                      // create promise

  std::future<int> fut = prom.get_future();    // 0 with future

  std::thread th1 (print_int, std::ref(fut));  // send future to new thread

  std::cout << "set_value begin " << std::endl;

  std::this_thread::sleep_for(std::chrono::milliseconds(1000));

  prom.set_value (10);                         // fulfill promise
                                               // (synchronizes with getting the future)

  std::cout << "set_value  end" << std::endl;

  th1.join();

  return 0;
}

输出：

set_value begin 
print_int begin 
set_value  end
value: 10
print_int end 

promise::set_value

generic template (1)    
void set_value (const T& val);
void set_value (T&& val);

specializations (2)    
void promise<R&>::set_value (R& val);   // when T is a reference type (R&)
void promise<void>::set_value (void);   // when T is void

• 将val存储为共享状态下的值，该状态变为就绪状态。
• 如果与同一共享状态关联的Future对象当前正在等待对future :: get的调用，则它将取消阻塞并返回val。
• void 特化的成员只是简单地使共享状态就绪，而无需设置任何值。