C++ / a working model

76 / 80   ·   C++11   ·   约 11 分钟

任务结果:async、future、promise 与启动策略

先记住这句话

future 是一次性结果的接收端,不等于后台线程。async 的启动策略决定独立执行还是延迟求值;promise 负责手动提交值或异常。既要处理结果,也要明确等待、异常传播以及关联任务何时结束。

本篇内容
  1. 先选择执行策略,再讨论并发
  2. future 管理结果与异常
  3. promise 是提交端,不是启动器
  4. 运行示例
  5. 动手练习

先选择执行策略,再讨论并发

std::async 返回关联共享状态的 future。显式 launch::async 表示函数按独立执行线程的语义运行;launch::deferred 把函数保存起来,直到第一次非定时等待或 get,才由调用等待的线程执行。默认策略允许实现选择,不能把默认 async 当成保证后台执行的接口。

示例把计算四十二的任务设为 async,把修改 ran 的任务设为 deferred。读取 ran 不存在并发冲突:延迟任务尚未执行,get 又由主线程调用。若任务被丢弃且从未触发延迟求值,它可以根本不运行。因此必须按副作用是否必需选择策略,不能只看返回值是否方便。

future 管理结果与异常

wait() 只等待共享状态就绪,不取走结果;get() 等待并取值,或重抛任务保存的异常。普通 future 的 get 只能消费一次,之后不再关联该共享状态。需要多个接收者时可转成 shared_future,各接收者持有自己的副本;结果里的可变对象依旧需要同步。

async 会把工作函数的异常存入结果通道,而不是让异常逃出线程入口。调用方应在 get 周围捕获,不应以为 wait 没抛就代表计算成功。另一个容易忽略的行为是:async 异步任务尚未结束时,释放最后一个相关共享状态引用可能等待任务完成。丢掉临时 future 可能让本想并行的连续调用实际串行。

promise 是提交端,不是启动器

promise 和 future 共享一个结果槽,前者调用 set_value 或 set_exception,后者等待接收。promise 自己不会创建线程。正常提交结果会让等待方恢复,并建立所需同步;但仅仅结果就绪,不代表手工创建的生产线程已经结束,仍要管理线程生命周期。

示例在线程中捕获异常,再用 current_exception 交给 promise;主线程先 join,再 get 并捕获原异常类型。每个共享状态只能完成一次,重复提交会报错。若 promise 销毁时尚未完成,接收方会得到 broken_promise,而不是永远等待。实际接口应约定值、失败和取消如何表达,不能把未设置结果当成正常取消。

容易答错的地方

  • 不要假设所有 future 析构都会等待线程;普通 promise 创建的共享状态不提供这种线程回收保证,手工线程仍需 join。
  • 不要持有异步任务需要的锁时 get、wait 或销毁可能等待的 async future,否则会把结果等待变成死锁。

运行一个例子

最低标准 C++11 · 完整程序 · 下载 .cpp

#include <cassert>
#include <exception>
#include <future>
#include <iostream>
#include <stdexcept>
#include <thread>

int main() {
    auto answer = std::async(std::launch::async, [] { return 6 * 7; });
    bool ran = false;
    auto deferred = std::async(std::launch::deferred, [&] {
        ran = true;
        return 7;
    });
    assert(!ran);
    const int delayed = deferred.get();
    assert(ran && delayed == 7);
    assert(!deferred.valid());
    const int value = answer.get();
    assert(value == 42);

    std::promise<int> promise;
    auto failure = promise.get_future();
    std::thread worker([&] {
        try {
            throw std::invalid_argument("negative input");
        } catch (...) {
            promise.set_exception(std::current_exception());
        }
    });
    worker.join();
    bool caught = false;
    try {
        (void)failure.get();
    } catch (const std::invalid_argument&) {
        caught = true;
    }
    assert(caught);
    std::cout << value << ' ' << delayed << ' ' << caught << '\n';
}

在本地编译

g++ -std=c++11 -Wall -Wextra -Wpedantic -pthread concurrency-async.cpp -o example && ./example

预期结果

42 7 1

CHECK YOUR UNDERSTANDING

合上答案,试着解释。

把示例中的 promise 生产线程改成正常提交整数 9,接收端如何调整?能否连续调用两次 get 取得同一个值?

查看参考答案

工作函数改为 promise.set_value(9),主线程仍先 worker.join(),再用 const int result = failure.get() 并断言 result == 9,删除异常捕获分支。不能再次调用这个普通 future 的 get;第一次已消费共享状态。需要反复读取时,应在第一次消费前调用 share() 得到 shared_future,并通过它的 get 读取。

继续查证

标准草案链接会随工作草案更新;本文版本标记对应示例最低要求,不表示草案中的所有新规则都适用于旧标准。

回到目录