第28课 “共享状态”提供者(std::promise/std::package_task)

一. std::promise和std::package_task

(一)共享状态、提供者和管理者

 

  1.“共享状态”作为异步结果的传输通道,由std::async、std::promise和std::package_task等提供(Provider),并交由future/shared_future管理。Provider将计算结果写入“共享状态”对象,而future/shared_future通过get()函数来读取该结果。

  2. std::promise用于包装一个值,将数据和future绑定起来,方便线程赋值。而std::package_task用来包装一个可调用对象,将函数与future绑定以便异步调用。

  3. std::async、std::promise和std::package_task都是“共享状态”对象的创建者,它们创建“共享状态”类型有所不同。

  (1)std::async:创建_Deferred_async_state和_Task_async_state类型的共享状态。

  (2)std::promise:创建_Associated_state类型的“共享状态”。这种类型比较简单,内部只能保存某个类型的值(如返回值)。

  (3)std::package_task:创建_Package_state类型的“共享状态”,这种类型内部是通过std::function来可以包装可调用对象的

  4. std::promise和std::package_task都只有移动语义而没有拷贝语义

(二)std::promise类

  1. 用来保存某一类型的值,也可以用来保存线程函数的返回值,该值可被future读取。它为线程同步提供了一种手段。

  2. 可以通过 get_future 来获取future 对象,该对象与promise通过“共享状态”这个通道进行异步结果传输。std::promise通常在某个时间点通过设置一个值或异常对象, future通过调用get()来获取这个结果。

  3. set_value_at_thread_exit函数:设置共享状态的值,但不会将共享状态的标志设置为ready。当有当线程退出时,该标志位才设置为true,同时唤醒所有阻塞在future的get()函数的线程。

  4. 如果销毁std::promise时未设置值,则会存入一个异常。

(三)std::package_task类

  1. std::package_task 用来包装可调用对象,其本身也是一个可调用对象(因为重载了operator()(Args…args)函数。它可以作为线程函数传递给std::thread,或传给需要可调用对象的另一个函数,或者干脆直接调用。这与std::function类似,但std::package_task会将其包装的可调用对象执行结果(返回值)保存起来,并传递给了future对象

  2. 通过get_future()返回一个与“共享状态”相关联的future对象。其他线程可以通过std::package_task对象在“共享状态”上设置某个值或者异常。

  3. make_ready_at_thread_exit(Arg...args):该函数会调用被包装的任务,并向任务传递参数,类似于std::package_task的operator()成员函数,但不同的是make_read_at_thread_exit并不会立即设置“共享状态”的ready标志,而是在线程退出时才设置它

  4. reset()函数会重置“共享状态”,但是保留了之前被包装的任务。它使得package_task可以被重复使用,这点与std::promise一次性使用不同。

  5. std::package_task对象一般与std::thread配合使用,而不是std::async。如果要使用std::async运行任务,就没有理由去创建std::package_task对象。因为std::async调用时,内部会创建一个基类为_Packaged_state类 “共享状态”的子类对象,而std::package_task也会创建_Packaged_state类的对象。可见std::async能够在调用任务执行之前就做到std::package_task能做到的任何事情,也可以避免重复创建“共享状态”对象。

【编程实验】初探std::promise和std::pack_task

#include <iostream>
#include <future>
#include <thread>
#include <chrono>
#include <queue>

using namespace std;

void func(std::promise<int>& pr, int param)
{
    int res = param * 10;
    pr.set_value_at_thread_exit(res); //线程退出时,设置需要输出的值
}

//计算阶乘
int factorial(int n)
{
    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    if (n == 1)
        return 1;

    return n * factorial(n - 1);
}

void get_result(std::future<int>& fut) //获取结果
{
    while (fut.wait_for(std::chrono::milliseconds(20)) == std::future_status::timeout) {
        std::cout << ".";
    }
    std::cout << std::endl;
    std::cout << "the factorial result is " << fut.get() << std::endl;
}

int main()
{
    //1. std::promise/std::future配合使用

    //1.1 主线程等待子线程的结果
    std::promise<int> pr1;
    std::future<int>  fut1 = pr1.get_future();

    std::thread t1(func, std::ref(pr1), 5); 
    t1.join();  //等待t1线程退出

    std::cout << "The func output: " << fut1.get() << std::endl;
    //1.2 子线程等待主线程的计算结果
    std::promise<int> pr2;
    std::future<int> fut2 = pr2.get_future(); //创建通道

    std::thread t2([](std::future<int>& fut) {
        int res = fut.get();
        cout <<"thread id: "<<std::this_thread::get_id() << " get result " << res << endl;
    }, std::ref(fut2));

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

    pr2.set_value(100);
    t2.join();

    //2. std::package_task与std::future配合使用
    //2.1 直接将package_task作为函数对象使用
    std::packaged_task<double(int, int)> task1([](int a, int b) {
        return std::pow(a, b);
        });

    std::future<double> res = task1.get_future();

    task1(2, 9);

    std::cout << "task_lambda: " << res.get() << endl;

    //2.2 将std::package_task作为任务传递给std::thread线程
    std::packaged_task<int(int)> task2(factorial);
    std::future<int> fut3 = task2.get_future();

    std::thread t3(std::ref(task2), 7); //t3线程计算7的阶乘
    std::thread t4(get_result, std::ref(fut3));
    t3.join();
    t4.join();

    task2.reset();  //重置task,使得task2可以被重复使用
    std::future<int> fut5 = task2.get_future();
    std::thread t5(std::ref(task2), 8); //计算8的阶乘
    cout << fut5.get() << endl;
    t5.join();

    //3.std::async与std::future配合使用
    std::future<int> fut6 = std::async(std::launch::async, factorial, 7);
    fut6.wait();

    cout <<"aync calc result is: " << fut6.get() << endl;

    return 0;
}
/*输出结果
The func output: 50
thread id: 6532 get result 100
task_lambda: 512
.................................
the factorial result is 5040
40320
aync calc result is: 5040
*/

二. std::promise/std::package_task的应用

(一)一次性事件及建模

  1. 一个线程在完成其任务之前,可能需要等待特定的一次性事件的发生。在等待期间,线程可以去轮询事件是否发生,也可以去做其他任务。C++标准库使用std::future为这类一次性事件建模,

  2.一旦事件发生,future变为就绪,而std::future的get/wait()只能被调用一次,无法重复使用。如果多线程等待同一个事件,就需要使用std::shared_future,当事件发生时所有相关的shared_future对象均会变为就绪,并且可以访问其关联的任务结果。

  3.期值对象本身并不提供同步访问,当多个线程需要访问一个独立的期值对象时,必须使用互斥量或类似同步机制对访问进行保护。而如果仅为了实现一次性的事件通信,基于条件变量的设计会要求多余的互斥量和标志位,这显然不够干净利落,而使用期值可以很好的处理这个问题。

(二)线程间传递任务(以GUI消息处理为例)

  1. 在GUI编程中,当一个线程计算完结果,它要发出一条信息给GUI线程,以通知更新界面。

  2. std::package_task提供了实现这种功能的方法,且不需要发送一条自定义信息给GUI线程,而是将函数包装成任务,并传递到GUI线程,使任务在GUI线程中运行。

【编程实验】std::promise和std::package_task的应用

#include <iostream>
#include <future>
#include <thread>
#include <mutex>
#include <queue>

using namespace std;

//1.一次性事件及建模(以实现暂停状态启动的线程为例)
//创建暂停状态的线程:std::thread类创建的线程,一启动线程就运行起来。但是如果希望在线程运行前设置优先级和内核亲和性,
//就需要一个可以创建一个暂停的线程,然后通过其提供的native_handle成员,利用平台底层API配置这些线程特征。为达到这一
//目的,可以利用std::promise / std::future提供的一次性机制来实现暂停状态的线程。
class MyThread
{
private:
    std::thread mThread;
    std::promise<void> mPromise;
    std::future<void> mFuture;
    bool bStart;
public:
    template<typename Fn, typename ...ArgTypes>
    MyThread(Fn&& fn, ArgTypes&&... args):bStart(false)
    {
        mFuture = mPromise.get_future();
        
        mThread = std::move(std::thread([this, &fn, &args...] {
            mFuture.wait();
            std::forward<Fn>(fn)(std::forward<ArgTypes>(args)...);
            }));
    }

    void start()
    {
        if (!bStart) {
            mPromise.set_value();
            bStart = true;
        }
    }

    void join()
    {
        mThread.join();
    }

    void detach()
    {
        mThread.detach();
    }
 
    bool joinable() {
        return mThread.joinable();
    }
};

//2. 利用shared_future处理多个反应任务
//反应任务
std::mutex g_mtx;
void reach()
{
    std::lock_guard<std::mutex> lck(g_mtx);
    cout << "thread(id= " <<std::this_thread::get_id() <<") react"<< endl;
}

//检测任务(可处理多个反应任务)
void detech() 
{
    std::promise<void> pr;
    std::shared_future<void> sf = pr.get_future().share();
    std::vector<std::thread> vec; //反应任务的容器

    for (int i = 0; i < 10; ++i)
    {
        vec.emplace_back([sf]   //在sf按值传递,在其副本上wait
        {
                sf.wait();
                reach();
        });
    }

    //...      //注意,如果此处抛出异常,则detech会失去响应

    pr.set_value();  //让所有线程取消暂停

    for (auto& t : vec) {
        t.join(); 
    } 
}

//3. gui消息处理(在线程间传递任务,而不是消息)
class MessageManager
{
    std::queue<std::shared_ptr<std::function<void()>>> mQueue;
    std::mutex mtx;
    bool bShutdown = false;
public:
    void shutDown() { bShutdown = true; }

    //将任务包装成package_task
    template<typename Fn, typename... Args>
    std::future<std::result_of_t<Fn && (Args&& ...)>> //postTask函数的返回值类型,future配合package_task使用
    postTask(Fn&& fn, Args&& ...args)
    {
        using Ret = std::result_of_t <Fn && (Args && ...)>; //Fn函数的返回值类型

        std::lock_guard<std::mutex> lck(mtx);

        ////将任务包装成package_task类型(注意,由于package_task为只移动类型,不能复制。这里在堆上创建)
        auto ptrPA = std::make_shared<std::packaged_task<Ret()>>(std::bind(std::forward<Fn>(fn), std::forward<Args>(args)...));
        auto fut = ptrPA->get_future();

        //利用lambda将“std::packaged_task<Ret()>”重新包装成queue队列所需的元素类型:std::function<void()>共享指针类型
        auto pTask = std::make_shared<std::function<void()>>([ptrPA]()->void {(*ptrPA)(); });  

        mQueue.push(pTask);

        return fut;
    }

    void guiThread()
    {
        while (!bShutdown)
        {
            //...     //处理其它gui界面消息

            //获取并执行用户任务
            std::shared_ptr<std::function<void()>> task;
            {
                std::lock_guard<std::mutex> lk(mtx);
                if (mQueue.empty())
                    continue;

                task = mQueue.front(); //取出队列中的用户任务
                mQueue.pop();
            }
            (*task)();  //执行任务
        }
    }
};

int main()
{
    cout << "main thread running..." << endl;

    //1. 以暂停状态启动的线程
    MyThread th([](int x, int y) {
        int res = x + y;

        cout << x << " + " << y << " = " << res << endl;
        return res;
    }, 10, 20);

    th.start();

    th.join();

    //2.处理多反应任务(在shared_future上等待)
    detech();

    //3. 在线程间传递任务
    MessageManager mm;
    std::thread guiThread(&MessageManager::guiThread,&mm);

    auto fut1 = mm.postTask([](int x, int y)->int { return x + y; }, 1, 2);
    auto fut2 = mm.postTask([](int x, int y, int z) {return x * y * z; }, 10, 20, 30);
    auto fut3 = mm.postTask([](const std::string& str) {return str; }, "SantaClaus");
    cout << fut1.get() << endl;  //3
    cout << fut2.get() << endl;  //6000
    cout << fut3.get() << endl;  //SantaClaus

    mm.shutDown();

    guiThread.join();

    return 0;
}
/*输出结果
main thread running...
10 + 20 = 30
thread(id= 8792) react
thread(id= 2600) react
thread(id= 2604) react
thread(id= 13384) react
thread(id= 14864) react
thread(id= 14884) react
thread(id= 13588) react
thread(id= 13516) react
thread(id= 11956) react
thread(id= 13540) react
3
6000
SantaClaus
*/

猜你喜欢

转载自www.cnblogs.com/5iedu/p/11815582.html