第26课 std::async异步任务

一. std::async函数模板

（一）std::async和std::thread的区别

　　1. 两者最明显的区别在于async采用默认启动策略时并不一定创建新的线程。如果系统资源紧张，那么std::thread创建线程可能失败，系统报告异常，整个程序可能崩溃。而std::async一般则不会，它在无法创建新线程时，会将任务分配给后续调用future.get()函数的线程，并以同步的方式执行（即不创建新线程）。

　　2. std::async表现为更高阶的抽象，它把用户从线程管理的细节解放出来，将这些责任转交给C++标准库的实现者。而std::thread要求自行处理线程耗尽、超订、负载均衡以及新平台适配问题。

　　3. std::thread未提供直接获取线程函数返回值的方法。但std::async可以通过future对象来获取。

（二）std::async函数模板及分析 

　　1. “共享状态”对象，用于保存线程函数(一般是可调用对象)及其参数、返回值以及新线程状态等信息。该对象保存在堆中，由std::async、std::promise或std::package_task提供，并交由future或shared_future管理其生命期。被调方(通常指调用promise.set_value()的一方)将计算所得的结果写入“共享状态”，而调用方通过std::future的get()读取该结果。

 

　　2. 调用std::async是会创建一个“_Deferred_async_state”或_“Task_async_state”类的“共享状态”对象，该对象是_Packaged_state的子类。注意，直接创建std::promise时，生成的是“_associated_state”类的共享状态对象，而std::package_task创建的是“_Packaged_state”类的共享状态对象。

　　3. _Get_associated_state是个工厂函数，通过不同的策略创建不同的“共享状态”对象，并将其交由future管理，负责其生命周期。future类似于std::unique_ptr，对“共享状态”对象“独占”所有权。

　　4. 与std::thread一样，传入std::async中的可调用对象及其参数会被按值以副本形成保存成一个tuple对象，然后再以右值的方式传入线程函数中对应的参数。

【编程实验】创建异步任务

#include <iostream>
#include <thread>
#include <future>
#include <mutex>
#include <vector>
#include <numeric> //for std::accumulate

using namespace std;

std::mutex mtx;

class Widget
{
public:
    void foo(int x, const std::string& s)
    {
        std::lock_guard<std::mutex> lk(mtx);
        cout << "thread id = "<<std::this_thread::get_id()<<
            " void Foo::foo(int, const std::string&): x = " <<  x << ", s = " << s<< endl;
    }

    void bar(const std::string& s)
    {
        std::lock_guard<std::mutex> lk(mtx);
        cout << "thread id = " << std::this_thread::get_id()  
             <<" void Widget::bar(const std::string&): s = " << s << endl;
    }

    void operator()(double val)
    {
        std::lock_guard<std::mutex> lk(mtx);
        cout << "thread id = " << std::this_thread::get_id()
             << " void Widget::operator(): val = " << val << endl;
    }
};

class NonCopyable //只移动对象
{
public:
    NonCopyable() {};

    NonCopyable(const NonCopyable&) = delete;
    NonCopyable& operator=(const NonCopyable&) = delete;

    NonCopyable(NonCopyable&&) = default;
    NonCopyable& operator=(NonCopyable&&) = default;

    double operator()(double d)
    {
        std::lock_guard<std::mutex> lk(mtx);
        cout << "thread id = " << std::this_thread::get_id()
             << " void NonCopyable::operator(): d = " << d << endl;
        return d;
    }
};

//并行计算
template<typename RandomIt>
int parallel_sum(RandomIt beg, RandomIt end)
{
    auto len = end - beg;
    if (len < 1000)
    {
        std::lock_guard<std::mutex> lk(mtx);
        cout << "thread id = " << std::this_thread::get_id()
            << " invoke parallel_sum()" << endl;
        return std::accumulate(beg, end, 0); //遍历[beg,end)区别的每个元素并累加。初始值为0
    }
        

    RandomIt mid = beg + len / 2;
    auto handle = std::async(std::launch::async,  //子线程将[mid,end)元素进行累加
                            parallel_sum<RandomIt>, mid, end);

    int sum = parallel_sum(beg, mid);//本线程将[begin,mid)区间元素进行累加

    return sum + handle.get(); //返回两个区间结果的累加和
}

int main()
{
    Widget w;

    cout << "main thread id = " << std::this_thread::get_id() << endl;
    //1. 参数传递
    auto fut1 = std::async(&Widget::foo, &w, 42, "hello"); //传入this指针：&w
    auto fut2 = std::async(&Widget::bar, w, "goodbye"); //传入x的副本如tmp。 tmp.bar(...)

    auto fut3 = std::async(Widget(), 3.14159); //传入Widget临时对象，调用operator()
    auto fut4 = std::async(std::ref(w), 2.718);  //传入w的引用，调用operator();

    NonCopyable mo;    //只移动对象
    auto fut5 = std::async(std::move(mo),3.14159); //mo是只移动对象，必须被转为右值

    //2. 同步、异步
    auto fut6 = std::async(std::launch::async, Widget(), 1.2); //在新线程上运行,operator()
    auto fut7 = std::async(std::launch::deferred, &Widget::bar, &w, "deferred"); //线程延迟到调用get或wait才执行

    auto fut8 = std::async(std::launch::async | std::launch::deferred, //等价于默认启动策略
                         &Widget::bar, &w, "async | deferred");

    fut7.get(); //主线程阻塞，等待fut7子线程。（子线程延迟到这时才执行）。

    //3. 并行计算
    std::vector<int> vec(10000, 1); //10000个1
    int res = parallel_sum(vec.begin(), vec.end());
    
    {
        std::lock_guard<std::mutex> lk(mtx);
        cout << "The sum is: " << res << endl;

        cout << "main thread  end." << endl;
    }

    return 0;
}
/*输出结果
main thread id = 16756
thread id = 1928 void Foo::foo(int, const std::string&): x = 42, s = hello
thread id = 16756 void Widget::bar(const std::string&): s = deferred  //注意，由主线程执行
thread id = 13216 void Widget::bar(const std::string&): s = goodbye
thread id = 7940 void Widget::operator(): val = 3.14159
thread id = 16080 void Widget::operator(): val = 2.718
thread id = 11492 void NonCopyable::operator(): d = 3.14159
thread id = 1928 void Widget::operator(): val = 1.2
thread id = 13216 void Widget::bar(const std::string&): s = async | deferred
thread id = 16756 invoke parallel_sum()
thread id = 7940 invoke parallel_sum()
thread id = 16080 invoke parallel_sum()
thread id = 11492 invoke parallel_sum()
thread id = 1928 invoke parallel_sum()
thread id = 13216 invoke parallel_sum()
thread id = 1928 invoke parallel_sum()
thread id = 7636 invoke parallel_sum()
thread id = 5816 invoke parallel_sum()
thread id = 15856 invoke parallel_sum()
thread id = 15832 invoke parallel_sum()
thread id = 7636 invoke parallel_sum()
thread id = 15400 invoke parallel_sum()
thread id = 16968 invoke parallel_sum()
thread id = 15856 invoke parallel_sum()
thread id = 15476 invoke parallel_sum()
The sum is: 10000
main thread  end.
*/

二. std::async的启动策略

（一）std::async的启动策略

　　1. 三种启动策略（std::async通过指定不同的启动策略来决定创建是“共享状态”对象）

　　（1）异步方式（std::launch::async）：会创建一个“_Task_async_state”类的共享状态对象。使用该策略时异味着线程函数必须以异步的方式运行，即在另一个线程之上执行。

　　（2）同步方式（std::launch::deferred）：会创建一个“_Deferred_async_state”类的共享状态对象。使用该策略意味着线程函数延迟到调用future的get/wait时才得以运行，而且两者是在同一线程上以同步的方式运行。即调用future的一方会阻塞至线程函数运行结束为止。如果get/wait没有得到调用，则线程函数不会被执行。

　　（3）默认启动策略（std::launch::async|std::launch::deferred）：即两者或运算的结果，这意味着任务可能以异步或同步的方式被运行。也就是说是否创建新线程来运行任务，取决于系统资源是否紧张，由标准库的线程管理组件承担线程创建和销毁、避免超订以及负载均衡的责任。

（二）默认启动策略

　　1. 带来的问题

　　（1）用户无法预知是异步还是同步运行，因为线程函数可能被调度为延迟执行。

　　（2）无法预知线程函数是否与调用future的get/wait函数线程是否在同一线程运行。如果此时线程函数会读取线程局部存储（thread_local storage, TLS），那么也就无法预知会取到哪个线程的局部存储。

　　（3）有时甚至连线程函数是否会运行，这件起码的事情都是无法预知的。这是因此无法保证在程序的每条路径上future的get或wait都会得以调用。

　　2. 注意事项：

　　（1）默认启动策略能正常工作需要满足以下所有条件。

　　　　①任务不需要与调用get/wait的线程并发执行。

　　　　②读/写哪个线程的thread_local变量无关紧要。

　　　　③可以保证在std::async返回的future上调用get/wait，或者可以接受任务可能永不执行。

　　　　④用户已清楚使用wait_for或wait_unitil的代码任务可能被推迟执行，这种可能性己被纳入考量。

　　（2）只要其中一个条件不满足，就必须手动指定启动策略以保证任务以异步或同步的方式运行。

【编程实验】默认启动策略问题的解决

#include <iostream>
#include <future>

using namespace std;
using namespace literals; //for duration suffixes（时长后缀，如1s)

void func()
{
    std::this_thread::sleep_for(1s);
}

//reallyAsync函数模板：用于保证任务被异步执行
template<typename Func, typename ...Args>
inline auto reallyAsync(Func&& f, Args... args)
{
    return std::async(std::launch::async,
                      std::forward<Func>(func),
                      std::forward<Args>(args)...);
}

int main()
{
    //wait_for函数必须可虑任务是同步或异步运行

    auto fut1 = std::async(func); //默认启动策略，无法预估任务是被同步还是异步运行

    //解决方案1：wait_for(0s)
    if (fut1.wait_for(0s) == std::future_status::deferred){ //同步运行，wait_for(0s)
        fut1.get(); //等待结果
    }else { //异步运行
        while (fut1.wait_for(100ms) != std::future_status::ready) { //轮询子线程是否结束
            //...   //并发做其他任务
        }

        //... //fut is ready
    }

    //解决方案2：确实以异步运行任务
    auto fut2 = reallyAsync(func);
    while (fut2.wait_for(100ms) != std::future_status::ready) //异步方式，确保wait_for返回ready的结果
    {                                                         //从而消除future_status::deferred的可能

    }

    return 0;
}