基于锁的并发程序设计

2020-12-27

基于锁的并发数据结构设计，需要确保访问线程持有锁的时间最短。对于只有一个互斥量的数据结构来说，需要保证数据不被锁之外的操作所访问到，并且还要保证不会在固有结构上产生条件竞争。但是使用多个互斥量可能会导致死锁。

下面是一个使用锁的的线程安全的数据队列：

#include <mutex>
#include <deque>
#include <thread>
#include <queue>
#include <exception>
class empty_queue : public std::exception{
public:
    const char* what() const throw() {
        throw "empty queue";
    }
};
template <typename T>
class ThreadSafe_que{
public:
    ThreadSafe_que(){}
    ThreadSafe_que(const ThreadSafe_que& rhs){
        std::unique_lock<std::mutex> ul(rhs._m);
        _queu = rhs._queu;
    }
    ThreadSafe_que& operator=(const ThreadSafe_que&) = delete;

    void push(T value){
        std::lock_guard<std::mutex> lg(_m);
        _queu.push(std::move(value));
    }

    std::shared_ptr<T> pop(){
        std::lock_guard<std::mutex> lg(_m);
        if (empty()) throw empty_queue();
        std::shared_ptr<T> const res(std::make_shared<T>(std::move(_queu.front())));
        _queu.pop();
        return res;
    }

    void pop(T& value){
        std::lock_guard<std::mutex> lg(_m);
        value = std::move(_queu.front());
        _queu.pop();
    }
    bool empty() noexcept {
        std::lock_guard<std::mutex> lg(_m);
        return _queu.empty();
    }
private:
    std::queue<T> _queu;
    mutable std::mutex _m;
};

互斥量_m能保证基本的线程安全，那就是对每个成员函数进行加锁保护。这就保证在同
一时间内，只有一个线程可以访问到数据，所以能够保证，数据结构的“不变量”被破坏时，不会被其他线程看到。在empty()和pop()成员函数之间会存在潜在的竞争，不过代码会在pop()函数上锁时，显式的查询栈是否为空，所以这里的竞争是非恶性的所有成员函数都使用 st::lock_guard<> 来保护数据，所以队列的成员函数能有“线程安全”的表现。

下面使用条件变量和锁是实现一个线程安全的队列：

#include <mutex>
#include <deque>
#include <thread>
#include <queue>
#include <condition_variable>
template <typename T>
class ThreadSafe_que{
public:
    ThreadSafe_que(){}
    ThreadSafe_que(const ThreadSafe_que& rhs){
        std::unique_lock<std::mutex> ul(rhs._m);
        _queu = rhs._queu;
    }
    ThreadSafe_que& operator=(const ThreadSafe_que&) = delete;

    void push(T value){
        std::lock_guard<std::mutex> lg(_m);
        _queu.push(std::move(value));
        _cv.notify_one();
    }

    std::shared_ptr<T> pop(){
        std::unique_lock<std::mutex> lk(_m);
        _cv.wait(lk,[this]{return !_queu.empty(); });
        std::shared_ptr<T> const res(std::make_shared<T>(std::move(_queu.front())));
        _queu.pop();
        return res;
    }

    void pop(T& value){
        std::unique_lock<std::mutex> lk(_m);
        _cv.wait(lk,[this]{return !_queu.empty(); });
        value = std::move(_queu.front());
        _queu.pop();
    }
    std::shared_ptr<T> try_pop(){
        std::lock_guard<std::mutex> lg(_m);
        if (_queu.empty())
            return false;
        std::shared_ptr<T> const res(std::make_shared<T>(std::move(_queu.front())));
        _queu.pop();
        return res;
    }
    bool try_pop(T& value){
        std::lock_guard<std::mutex> lg(_m);
        if (_queu.empty())
            return false;
        value = std::move(_queu.front());
        _queu.pop();
        return true;
    }
    bool empty() noexcept {
        std::lock_guard<std::mutex> lg(_m);
        return _queu.empty();
    }
private:
    std::queue<T> _queu;
    std::condition_variable _cv;
    mutable std::mutex _m;
};