即使使用互斥元对数据进行保护,仍有可能会有竞争条件。

接口中固有的竞争条件

下面是一个简单的std::stack接口实现。

template<typename T, typename Container=std::deque<T>>
class stack
{
public:
    explicit stack(const Container&);
    explicit stack(Container&& = Container());
    template<class Alloc> explicit stack(const Alloc&);
    template<class Alloc> stack(const Container&, const Alloc&);
    template<class Alloc> stack(Container&&, const Alloc&);
    template<class Alloc> stack(stack&&, const Alloc&);

    bool empty() const;
    size_t size() const;
    T& top();
    T const& top() const;
    void push(T const&);
    void push (T&&);
    void pop();
    void swap(stack&&);
}

empty()size()都不可靠,原因是,在一个线程调用了这两个方法之后,使用获得的信息之前,其他线程可以访问并修改栈。类似的竞争条件也发生在top()pop()之间。考虑有两个线程共享同一个stack<int> s,并同时执行下面的代码段:

if(!s.empty())
{
    int const value = s.top();
    s.pop();
    do_something(value);
}

啊,先检查栈是否为空,如果不为空,则将栈顶的元素弹出,并对该元素做后续处理。如果栈的每个成员函数被互斥元保护,即某时刻只有一个线程可以运行成员函数——看起来这些调用被很好的交错开了,好像没什么问题?不是哦,两个线程从top()中获得了一样的值,但第二个执行pop()的线程中的栈已经执行过pop()一次了,因此传给do_something的参数并不是实际上的栈顶元素,这并不符合我们的期望!为此需要修改这个类的接口。《并发编程》给出了一个线程安全栈的详细定义:

#include <exception>
#include <memory>
#include <mutex>
#include <stack>

struct empty_stack: std::exception
{
    const char* what() const throw();
}

templace<typename T>
class threadsafe_stack
{
private:
    std::stack<T> data;
    mutable std::mutex m;
public:
    threadsafe_stackk() {}
    threadsafe_stack(const threadsafe_stack& other)
    {
        std::lock_guard<std::mutex> lock(other.m);
        data = other.data;
    }

    threadsafe_stack& operator=(const threadsafe_stack&) = delete;

    void push(T new_value)
    {
        std::lock_guard<std::mutex> lock(m);
        data.push(new_value);
    }

    std::shared_ptr<T> pop()
    {
        std::lock_guard<std::mutex> lock(m);
        if(data.empty()) throw empty_stack();
        std::shared_ptr<T> const res(std::make_shared<T>(data.top()));
        data.pop();
        return res;
    }

    void pop(T& value)
    {
        std::lock_guard<std::mutex> lock(m);
        if(data.empty()) throw empty_stack();
        value = data.top();
        data.pop();
    }

    bool empty() const
    {
        std::lock_guard<std::mutex> lock(m);
        return data.empty();
    }
};

该实现削减了接口数量,只保留了empty()pop()top()pop()有两个重载,一个接收存放元素的引用作为参数,另一个则返回std::shared_ptr<T>。·threadsafe_stack在接口层面避免了数据的竞争,消除了在调用接口时殚精竭虑仍无法避免的潜在问题。

死锁

标准库提供的std::lock可以同时锁定两个互斥元,从而避免了死锁。参考下面的代码:

class some_big_object;
void swap(some_big_object& lhs, some_big_object& rhs);

class X
{
private:
    some_big_object some_detail;
    std::mutex m;
public:
    X(some_big_object const& sd): some_detail(sd) {}

    friend void swap(X& lhs, X& rhs)
    {
        if(&lhs = &rhs)
            return;
        std::lock(lhs.m, rhs.m);
        std::lock_guard<std::mutex> lock_a(lhs.m, std::adopt_lock);
        std::lock_guard<std::mutex> lock_b(rhs.m, std::adopt_lock);
        swap(lhs.some_detail, rhs.some_detail);
    }
}

首先检查两个X是不同实例,因为试图获取已经锁定的std::mutex是未定义行为。再调用std::lock()锁定两个互斥元,并用这两个互斥元构造两个std::lock_guard,构造是额外指定参数std::adopt_lock,告知std::lock_guard互斥元已被锁定,只接受互斥元的所有权,无需再上锁。