C++并发编程 - 同步并发操作

发表于 2年以前 | 总阅读数：252 次

C++并发编程 - 同步并发操作

前言

一个好的程序员是那种过单行线马路都要往两边看的人。

线程同步的本质是防止临界区(公共资源)并发操作，即多个线程禁止同时操作临界区。为此，在程序中以某种手段，将多个线程按照先后顺序访问临界区。

临界区的操作一直要保持谨慎。多线程访问临界区，同时读还好；假如一个写的同时，另一个读，那这个读的值可能不是确定的，有可能是写之前的也有可能是写之后的，这可能会引发重大bug，且难以排查。

本文主要介绍如何使用c++11中条件变量以及期望来设计并发操作。

条件变量 std::condition_variable

在多线程任务中，线程通常使用条件变量阻塞自身，直至条件发生。即A线程阻塞等待某个条件变量，B线程通知条件变量变化解除A线程阻塞。为了防止竞争，条件变量的使用总是和互斥锁结合在一起。

C++标准库对条件变量有两套实现: std::condition_variable 和 std::condition_variable_any。其中前者仅限于与std:mutex配合使用，后者可以和任何满足最低标准的互斥量一起使用。

从体积、性能以及系统资源的使用方面，推荐使用 std::condition_variable。仅当对灵活性有硬性要求的情况下，才会选择 std::condition_variable_any。

std::condition_variable 实例：

static std::mutex mut;
static bool cond_value = false;
static std::condition_variable cond;

// 阻塞线程
void wait_thread()
{
    std::unique_lock<std::mutex> lk(mut); //a
    cond.wait(
        lk, []{ return cond_value; }); 
    //lk.unlock();    // b
    //...  //c
}

// 通知线程
void notify()
{
    std::lock_guard<std::mutex> lk(mut); 
    cond_value = true;
    cond.notify_one();
}

流程: step 1. wait_thread先获取锁，然后查询cond_value为false时，先解锁，然后阻塞线程，等待其他线程通知。 step2. notify()获取锁，更改cond_value为true，并通过notify_one()通知阻塞的线程，并解锁。 step3. wait_thread接收到通知，重新获取锁，检查cond_value为true，从wait()返回（解除阻塞）。返回时仍处于持有锁状态，直至互斥锁被析构或者手动解锁。

在多线程中持有锁时间过长是一件糟糕的事情，当处理完与互斥锁相关的共享数据时，就应该立刻解锁。故上述c处还有其他业务时，b处有必要解锁。

另外，还存在多个线程等待同一事件。此种场景，可通过notify_all()通知所有阻塞的线程检查条件。

「注」：wait()会去检查这些条件(通过调用所提供的lambda函数)，当条件满足(lambda函数返回true)时返回。如果条件不满足(lambda函数返回false)， wait()函数将解锁互斥量，并且将这个线程(上段提到的处理数据的线程)置于阻塞或等待状态。当准备数据的线程调用notify_one()通知条件变量时，处理数据的线程从睡眠状态中苏醒，重新获取互斥锁，并且对条件再次检查，在条件满足的情况下，从wait()返回并继续持有锁。当条件不满足时，线程将对互斥量解锁，并且重新开始等待。

期望 std::future

是指某个线程只等待一个特定的一次性事件。C++标准库将这种一次性事件称为“期望” (future)。

当一个线程需要等待一个特定的一次性事件时，在某种程度上来说它就需要知道这个事件在未来的表现形式。之后，这个线程会周期性(较短的周期)的等待或检查，事件是否触发(检查信息板)；在检查期间也会执行其他任务。另外，在等待任务期间它可以先执行另外一些任务，直到对应的任务触发，而后等待期望的状态会变为“就绪”(ready)。

在C++标准库中，有两种“期望”，使用两种类型模板实现，声明在头文件中: 唯一期望(uniquefutures)( std::future<> )和共享期望(shared futures)( std::shared_future<> )。这是仿照 std::unique_ptr 和 std::shared_ptr 。std::future 的实例只能与一个指定事件相关联，而 std::shared_future 的实例就能关联多个事件。后者的实现中，所有实例会在同时变为就绪状态，并且他们可以访问与事件相关的任何数据。

「注」: 以上两段描述，摘抄于《C++并发编程实战》

std::future并非单独使用，在C++标准库std::async、std::packaged_task和std::promise关联了std::future。即std::async、std::packaged_task和std::promise会返回std::future类型，线程通过std::future获取任务执行的结果。

了解这些，我们可以使用std::future程序上实现业务与任务的分离。即业务线程只负责处理逻辑，任务线程负责任务执行，业务线程又能获取到任务执行的结果或其他的设计。

std::async 带返回值的后台任务

当需要执行一个耗时的任务，在不阻塞主线程的条件下，还需要关心这个任务执行的结果时(例如是获取计算结果)。在不考虑使用全局变量情况下（全局变量，代码大忌），std::thread就不能满足这个需求了，此时std::async可以满足需求。

std::async使用案例

std::async可启动一个异步任务。与std::thread对象等待运行方式的不同，std::async会返回一个std::future对象，这个对象持有最终计算出来的结果。当需要这个结果时，只需要调用这个对象的get()成员函数；并且直到“期望”状态为就绪的情况下，线程才会阻塞；之后，返回计算结果。

「std::async 实例」:

void async_test()
{
    struct SThreadParam
    {
        int func(int val1, int val2)
        {
            LOGD("async func start...\n");
            sleep(2);           // Simulated 2S time consumption
            return val1 + val2;
        }
    };
    SThreadParam paramFunc;

    LOGD("create async...\n");
    std::future<int> f4 = std::async(std::launch::async, &SThreadParam::func, &paramFunc, 10, 12);
    LOGD("async debug...\n");
    //sleep(2);
    LOGD("async return %d.\n", f4.get());
}

「执行结果」：

2022-08-27 00:46:44.951  179 D: create async...
2022-08-27 00:46:44.951  181 D: async debug...
2022-08-27 00:46:44.951  173 D: async func start...
2022-08-27 00:46:46.953  183 D: async return 22.

一般情况下，std::async()任务函数默认情况下在创建时启动，在调用f4.get()时，若线程未执行完，则阻塞至执行完；否则，立刻返回结果（主线程sleep(2)是否注释，后两条打印都相差2s）。当然也有不一般的情况，见下文「注」。

指定async任务执行时机

std::async()任务执行时机可在创建时传入额外参数指定，这个参数的类型是std::launch，有以下含义：

std::launch::deferred: 函数调用被延迟到wait()或get()函数调用时才执行
std::launch::async: 函数必须在其所在的独立线程上执行
std::launch::deferred | std::launch::async: 表明实现可以选择这两种方式的一种

auto f6=std::async(std::launch::async,Y(),1.2); // 在新线程上执行
auto f7=std::async(std::launch::deferred, func, std::ref(x)); // 在wait()或get()调用时执行
auto f8=std::async(
        std::launch::deferred | std::launch::async,
        func, std::ref(x)); // 实现选择执行方式
auto f9=std::async(func, std::ref(x));
f7.wait(); // 调用延迟函数

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

std::async避坑

接下说的很重要!!! 在std::async时，一定要处理其返回的std::future。否则，async会以同步的方式执行任务，即当前线程会阻塞到任务执行完。

「错误的示例」：

void async_case()
{
    struct SThreadParam
    {
        int func(int val1, int val2)
        {
            LOGD("async func start...\n"); // a
            sleep(2);           // Simulated 2S time consumption
            return val1 + val2;
        }
    };
    SThreadParam paramFunc;
    LOGD("create async...\n"); // b
    //std::future<int> f4 = std::async(&SThreadParam::func, &paramFunc, 10, 12);
    std::async(std::launch::async, &SThreadParam::func, &paramFunc, 10, 12);
    LOGD("async debug...\n"); // c
    //sleep(2);
    //LOGD("async return %d.\n", f4.get());
}

「执行结果」：

2022-08-27 00:55:15.886  179 D: create async...
2022-08-27 00:55:15.886  173 D: async func start...
2022-08-27 00:55:17.887  182 D: async debug...

设计上我们希望async不会阻塞，即SThreadParam::func应该在另一个线程执行，b和c的打印时间应该接近相同。但实际上却相差2s，刚好是SThreadParam::func的执行时间。

这是很隐蔽且很重大的bug，假设任务时间更长，会导致主线长时间阻塞。这个bug网上大佬们已经给出很多解释了，这里不再赘诉，可参考https://blog.csdn.net/weixin_44537992/article/details/108438379。

这就警告我们在使用新的接口时，还是要搞清楚其原型及原理，规范地使用未知的接口。说到这里我也慢慢懂了，为什么有些接口明明部分参数可以缺省，但是大佬们在使用时还是会明确指定这些参数值，虽然使用起来更复杂，但是也更安全。

std::packaged_task 任务与期望

std::packaged_task<>对一个函数或可调用对象，绑定一个期望。当std::packaged_task<>对象被调用，它就会调用相关函数或可调用对象，将期望状态置为就绪，返回值也会被存储为相关数据。

与std::async不同的是，std::packaged_task<>可以将任务与期望打包，移动到指定线程显示调用packaged_task，future才会就绪；而std::async()不能指定线程运行任务。这是两者最显著的差异。

「packaged_task实例」：

void task_execute_thread()
{
    std::thread th5([](const char *name) {
        static std::packaged_task<int(int)> task;

        while(!packaged_shutdown) {
            std::unique_lock<std::mutex> lk(package_mut);
            LOG("%s: wait task...\n", name);
            pacakge_cond.wait(lk, [](){
                LOG("check queue %ld.\n", task_queue.size());
                return !task_queue.empty();
            });
            lk.unlock();
            LOG("%s: task execute.\n", name);
            task = std::move(task_queue.front()); // 5
            task_queue.pop_front();
            task(2); // a. 执行任务，参数未使用
        }
    }, "thread5");

    th5.detach();
}

void task_post()
{
    static int i = 0;
    auto f = [](int num){
        return num;
    };
    std::packaged_task<int(int)> task(std::bind(f, ++i));
    std::future<int> fut = task.get_future();
    std::unique_lock<std::mutex> lk(package_mut);

    task_queue.push_back(std::move(task));
    lk.unlock();
    pacakge_cond.notify_one();
    LOG("task return %d.\n", fut.get()); // b. 阻塞至a执行完成
}

「执行结果」：

input: f
thread5: wait task...
check queue 0.
Input: g
check queue 1.
thread5: task execute 0.
thread5: wait task...
check queue 0.
task return 1.

「注」std::future只适用于一次期望。即获取的future只允许使用一次get。std::shared_future似乎可以解决这个问题，未确认。

std::promises 承诺与期望

std::promise对象可以保存某一类型 T 的值，该值可被 future 对象读取（可能在另外一个线程中），因此 promise 也提供了一种线程同步的手段。在 promise 对象构造时可以和一个共享状态（通常是std::future）相关联，并可以在相关联的共享状态(std::future)上保存一个类型为 T 的值。

与std::packaged_task差异在于，std::promise可在线程运行时，通过set_value()向阻塞线程传递值。即A线程通过std::promise::set_value()传递某类型T的值，B线程std::future::get()会解除阻塞，同时获取到A线程传的T类型的值。

「std::promise实例」：

void pomise_case()
{
    std::promise<int> promise;  // save type is int
    std::future<int> future = promise.get_future(); // from promise

    LOGD("-> create thread6.\n");
    std::thread th6([](const char *name, std::promise<int> &prm) {
        static int value = 1024;
        std::this_thread::sleep_for(std::chrono::seconds(2)); // To highlight the delay of the effect

        prm.set_value(++value);  // a. refer step b
        LOGD("%s: set value: %d.\n", name, value);
    }, "thread6", std::ref(promise));

    LOGD("-> create thread7.\n"); 
    std::thread th7([](const char *name, std::future<int> &fut) {
        auto value = fut.get(); // b. wait step a set_value

        LOGD("%s: get value: %d.\n", name, value);
    }, "thread7", std::ref(future));

    th6.join();
    th7.join();
}

「执行效果」：

2022-08-27 02:24:08.826  233 D: -> create thread6.
2022-08-27 02:24:08.826  242 D: -> create thread7.
2022-08-27 02:24:10.827  239 D: thread6: set value: 1025.
2022-08-27 02:24:10.827  246 D: thread7: get value: 1025.

std::promise还是挺有意思的，可以实现线程间的值传递且无并发问题。原来我们通过全局变量实现线程间通信的方法，还要考虑上锁，以后就可以通过promise来实现了。

总结

本文主要总结了同步并发常用的一些技巧，其中条件变量使用被广泛使用，而c++11新引进的std::future相对来说很少被用到。
大致概括一下std::async、std::packaged_task、std::promise使用的场景： std::async: 另起线程异步处理耗时的任务，并返回处理结果。std::packaged_task: 将任务函数打包丢到指定线程处理，并返回处理结果。std::promise: 实现两个线程间的数据传递。
刚开始看std::async、std::packaged_task、std::promise，多少有一点抵触。一个同步搞这么多接口，仔细慢看发现每一种都使用都是针对不同的场景。
std::async在使用时一定主要处理其返回的std::future，否则会引发隐蔽且严重的bug。详见「std::async避坑章节」。
针对std::async可能引发的问题，告诫程序员们，一定要规范的使用每一个接口，缺省的参数及返回值不处理的话可能让代码裂成“东非大裂谷”。对于使用到的接口的参数和返回值，不管有没有必要，都要处理。
同步并发问题一直是程序中隐蔽而又严重的bug，不到万不得已尽量不要使用。同步并发bug就像是程序里的"蒙娜丽莎"，它在对你笑，你却一脸懵逼。