关于std::thread以及std::condition_variable的一些细节备忘

Posted 2021-01-13 j1ac

　　也算是看过不少多线程相关的资料了，但是一直对于其中的一些细节没有太好的把握，比如std::thread线程真正开始运行的时机，比如join、detch等真正的作用。

跟着《Cplusplus Concurrency In Action_Practical Multithreading》又过了一遍相关的细节，下面记录一下一些个人所获得的收获。

std::thread真正开始运行的时机

下面是我尝试写的一个基于条件变量和互斥量的生产者消费者模型的Demo，就从这里开始说起

#include<iostream>
#include<thread>
#include<unistd.h>
#include<vector>
#include<mutex>
#include<condition_variable>
#include<cmath>

std::vector<int> resource;
std::mutex m;
std::condition_variable cv;


void thread_procedure(){
    while(true){
        int n;
        while(true){
            std::unique_lock<std::mutex> ul(m);
            cv.wait(ul,[&]{return !resource.empty();});
            if(!resource.empty()) {
                n = resource.back();
                resource.pop_back();
                break;
            }
        }

        int res = 0;
        for(int i = 0; i<n; ++i){
            //sleep(2);
            res+=i;
        }
        std::lock_guard<std::mutex> ll(m);
        std::cout<<"This is from Thread "<<std::this_thread::get_id()<<", "<<"The sum of 0+1+2+...+"<<n<<" is "<<res<<std::endl;
    }
}


int main(){
    std::thread tt[5];
    for(int i = 0; i<5; ++i){
        tt[i] = std::thread(thread_procedure);
    }
    
    for(int i = 0; i<5; ++i){
        tt[i].join();
    }
    while(true){
        std::lock_guard<std::mutex> lg(m);
        resource.push_back(rand()%100);
        cv.notify_one();
    }
}

这段代码使用了一个std::mutex和一个std::condition_variable控制相应的线程，尝试实现一个简单的打印功能。

但在运行时该代码会卡在生产者处，即join代码之后的while循环不会运行下去。。。

这里几乎就涉及了std::thread线程库里面对于线程启动的机制以及join的真正语义了。

下面是一段GNU对于std::thread的实现代码：

class thread
{
    ...
public:
    thread() noexcept = default;
    thread(thread&) = delete;
    thread(const thread&) = delete;
    thread(thread&& __t) noexcept
    { swap(__t); }
    template<typename _Callable, typename... _Args>
    explicit thread(_Callable&& __f, _Args&&... __args)
    {
        _M_start_thread(_M_make_routine(std::__bind_simple(
        std::forward<_Callable>(__f),
        std::forward<_Args>(__args)...)));
     }
    ...
};

可以看到thread的构造函数传入了一个_Callable可调用对象以及相关的参数，然后使用了std::__bind_simple进行了包装，相当于std::bind，然后使用_M_start_thread直接使用平台相关线程实现开启了这个线程！

从这里我们可以看出在每个std::thread构造完成的时候新线程就已经开启了！

而join函数的作用就是等待join的线程执行结束，在join返回之后继续运行后续代码。

这样上面的代码会卡住也就理所应当了，在开启新线程之后资源池就用完了，然后启动的线程都阻塞在条件变量上面，而后续的while循环里面的生产过程则是由于join函数在等待已经开启的线程结束而无法运行。

整个程序就会在所有线程都处于阻塞状态下停在那里。

而解决这个问题的方法倒也简单，另开一个生产者线程就行，如下代码：

#include<iostream>
#include<thread>
#include<unistd.h>
#include<vector>
#include<mutex>
#include<condition_variable>
#include<cmath>

std::vector<int> resource;
std::mutex m;
std::condition_variable cv;


void thread_procedure(){
    while(true){
        int n;
        while(true){
            std::unique_lock<std::mutex> ul(m);
            cv.wait(ul,[&]{return !resource.empty();});
            if(!resource.empty()) {
                n = resource.back();
                resource.pop_back();
                break;
            }
        }

        int res = 0;
        for(int i = 0; i<n; ++i){
            res+=i;
        }
        std::lock_guard<std::mutex> ll(m);
        std::cout<<"This is from Thread "<<std::this_thread::get_id()<<", "<<"The sum of 0+1+2+...+"<<n<<" is "<<res<<std::endl;
    }
}

void producer(){
    while(true){
        std::lock_guard<std::mutex> lg(m);
        resource.push_back(rand()%100);
        cv.notify_one();
    }
}

int main(){
    std::thread tt[6];
    for(int i = 0; i<5; ++i){
        tt[i] = std::thread(thread_procedure);
    }
    
    tt[5] = std::thread(producer);
    
    for(int i = 0; i<6; ++i){
        tt[i].join();
    }
}

这种情况下，所有工作线程都在join函数调用之前就开启了，也就不会存在上述问题了。