使用互斥锁和条件变量的 C++ 线程同步

Question

我创建了一个程序来同步两个线程。 使用互斥量和条件变量。

#include <thread>
#include <string>
#include <iostream>
#include <mutex>
#include <condition_variable>
#include "signal.h"
#include <atomic>
class Class
{
public:
    Class(const std::string& s, const int& i) : m_data(s), threadName(i),counter(0) { }
    ~Class() { 
        if (m_thread.joinable())
        {
            m_thread.join();
        }
        if (m_thread2.joinable()){
            m_thread2.join();
        }

        std::cout << "destructor called" << std::endl;
        }
    void runThread() { 
        m_thread = std::thread(&Class::printThreadWrite, this); 
        m_thread2 = std::thread(&Class::printThreadRead,this);
    }

private:
    std::mutex m_printmutex;
    std::condition_variable m_conditionvar;
    std::string m_data;
    std::thread m_thread;
    std::thread m_thread2;
    std::atomic<bool> signalScheduleSwitch;
    int threadName;

    std::atomic<int> counter;
    void printThreadWrite()  { 
        
            while(1){
                {
                std::lock_guard<std::mutex> lg(m_printmutex);
                std::cout << "thread # " << std::this_thread::get_id() << " "  << m_data << '\n'; 
                counter ++;
                signalScheduleSwitch = true;
                }
                std::this_thread::sleep_for(std::chrono::milliseconds(2));
                m_conditionvar.notify_one();
            }
        
    }
    
    void printThreadRead()  { 
    
        while (1)
        {
            std::unique_lock<std::mutex> uLock(m_printmutex);
            m_conditionvar.wait(uLock,[this](){ return signalScheduleSwitch == true;});
            std::cout << "counter value is:" << counter << "\n";
            signalScheduleSwitch = false;

        }
    }
};

void signalHandler(int s){
    printf("caught signal %d Exiting\n", s);
    exit(1);

}

int main(int argc, char** argv)
{
    signal(SIGINT, signalHandler);
    try
    {
        Class c("Hello, world!",1);
        c.runThread();
    }
    catch(const std::exception& e)
    {
        std::cerr << e.what() << '\n';
    }

    return 0;
}

我无法获得线程同步。

使用条件变量等待标志被设置为真，代码应该在线程之间切换。我仍然没有得到干净的线程调度。我的理解有什么缺陷吗？

Answer 1

带有互斥量和两个条件变量的 Ping-Pong

这是使用互斥锁和条件变量的规范乒乓球。请注意，1) 您需要两个条件变量来使乒乓球工作，并且 2) 您必须小心地将输出语句放在仍然持有锁的块中。你的代码很接近。

#include <iostream>
#include <condition_variable>
#include <atomic>
#include <thread>

class PingPong {
public:
    PingPong() {
        t0_ = std::thread(&PingPong::ping, this);
        t1_ = std::thread(&PingPong::pong, this);
    }

    ~PingPong() {
        if (t0_.joinable())
            t0_.join();
        if (t1_.joinable())
            t1_.join();
    }

    void ping() {

        while(counter <= 20) {
            {
                std::unique_lock<std::mutex> lck(mutex_);
                cv0_.wait(lck, [this]{ return ready_ == false; });
                ready_ = true;
                std::cout << "ping counter: " << counter << std::endl;
            }
            ++counter;
            cv1_.notify_one();
        }
    }

    void pong() {

        while(counter < 20) {
            {
                std::unique_lock<std::mutex> lck(mutex_);
                cv1_.wait(lck, [this]{ return ready_ == true; });
                ready_ = false;
                std::cout << "pong counter: " << counter << std::endl;
            }
            cv0_.notify_one();
        }
    }

private:
    bool ready_{false};
    std::mutex mutex_;
    std::condition_variable cv0_, cv1_;
    std::atomic<int> counter{};
    std::thread t0_, t1_;
};

int main(){
    PingPong p{};
}

这将导致以下输出。

ping counter: 0
pong counter: 1
ping counter: 1
pong counter: 2
ping counter: 2
pong counter: 3
ping counter: 3
pong counter: 4
ping counter: 4
pong counter: 5
ping counter: 5
pong counter: 6
ping counter: 6
pong counter: 7
ping counter: 7
pong counter: 8
ping counter: 8
pong counter: 9
ping counter: 9
...

带有单个原子标志的乒乓球

根据您的平台，使用原子标志而不是条件变量可能性能更高（并且更容易理解）。这会产生与上面相同的输出。

class PingPongAtomicFlag {
public:
    PingPongAtomicFlag() {
        t0_ = std::thread([this]() { ping(); });
        t1_ = std::thread([this]() { pong(); });
    }

    ~PingPongAtomicFlag() {
        if (t0_.joinable())
            t0_.join();
        if (t1_.joinable())
            t1_.join();
    }

    void ping() {

        while(counter_ <= 20) {
            potato_.wait(true);
            std::cout << "ping counter: " << counter_ << std::endl;
            potato_.test_and_set();
            ++counter_;
            potato_.notify_one();
        }
    }

    void pong() {

        while(counter_ < 20) {
            potato_.wait(false);
            std::cout << "pong counter: " << counter_ << std::endl;
            potato_.clear();
            potato_.notify_one();
        }
    }

private:
    std::atomic_flag potato_;
    std::atomic<int> counter_{};
    std::thread t0_, t1_;
};