c++11 多线程间共享数据 <c++ concurrency in action>

本章主要描述多线程之间共享数据的方法、存在问题、解决方案。

第一部分：mutex在保护共享数据中的使用

 

1、最简单使用：

#include<mutex>
std::mutex some_mutex;
void func(){
    some_mutex.lock();
    //访问共享数据
    ....
    some_mutex.unlock();
}

2、向lock_guard推进：

但是不推荐直接使用lock、unlock，因为unlock一定要调用，如果由于你的疏忽或前面的异常将会导致问题，再次利用RAII思想，用对象管理资源就有了标准库的std::lock_guard,在构造函数中lock，析构函数中unlock。

std::mutex some_mutex;
void func(){
    lock_guard<std::mutex> some_guard(some_mutex);
    //访问共享数据
    ....
}

3、向封装前进：

　　每次数据访问都要记得加解锁，如果能让用户从加解锁中解脱就好了。将共享数据、mutex、对共享数据的访问函数（接口）封装到一起，这样用户就可以在多线程下安全使用该共享数据了。示例如下：

 

class EncapShareData{
    std::mutex m_mutex;
    Data m_data;
public:
    void Func(){
        std::lock_guard<std::mutex> mutexGuard;
        //对数据访问
        .....
    }
    //下述方法曝光了m_data，是危险的
    //返回共享数据的引用
    Data& DangerFunc1();
    //返回共享数据的指针
    Data* DangerFunc2();
    //参数中返回了共享数据的引用
    void DangerFunc3(Data&);
    //参数中返回了共享数据的指针
    void DangerFunc4(Data*);
    //函数f做了什么？将m_data的指针或引用保存到其他地方了（糟糕）？
    template<typename Function>
    void DangerFunc5(Function f){
        std::lock_guard<std::mutex> mutexGuard(m_mutex);
        f(m_data);
    }
    //friend!class、func...
    friend void DangerFunc6();
}

 

注意：示例中的DangerFunc*任意一种都会让对共享数据的多线程安全访问毁于一旦。

 

4、到此为止？

 

当你走完上述封装之路，并且确保封装类的每个接口都是多线程安全的，是不是真的就多线程安全了呢？看下面例子所示：

template<typename T>
class ThreadSafeStack{
private:
    std::stack<T> m_data;
    std::mutex m_mutex;
public:
    ThreadSafeStack(){}
    ThreadSafeStack(const ThreadSafeStack& other){
        std::lock_guard<std::mutex> lock(other.m_mutex);
        data = other.data;
    }
    ThreadSafeStack& operator=(const ThreadSafeStack&) = delete;
    //多线程安全的push
    void push(T v){
        std::lock_guard<std::mutex> lock(m_mutex);
        data.push(v);
    }
    //多线程安全的pop
    void pop(){
        std::lock_guard<std::mutex> lock(m_mutex);
        data.pop();
    }
    //stack::top返回内部元素引用，这里为了安全返回拷贝
    T top(){ return data.top(); }
    //empty/size是只读，是多线程安全的，只需要转发
    bool empty(){ return data.empty(); }
    size_t size(){ return data.size(); }
};

　　ThreadSafeStack类的每一个接口单独拿出来都是线程安全的，我们知道stl中stack的入栈只需要一个函数：stack::push()就可以了，但是如果要出栈就需要一连串函数调用，先要判断栈是否为空(top并不进行是否为空检查，所以你要检查)stack::empty(),然后取得栈顶元素stack::top()，最后才能stack::pop()，如下代码所示：

 

if(!mStack.empty()){
    //①
    T const v=mStack.top();
    //②
    mStack.pop();
}

 

　　单线程下这是安全的胡庸置疑。多线程线考虑以下情况：

　　a、栈中只有一个元素，两个独立要pop的线程可能都走到①处，然后在②处出现在空栈上pop，悲剧！

　　b、栈中有两个元素，两个独立要pop的线程可能都走到①处，但是取得的是同一个元素，然后pop两次，导致数据丢失！

　　函数中每个元素访问是多线程安全的，不代表这个函数是多线程安全的，一个类中每个函数是多线程安全的，不代表将这些函的数组合是多线程安全的，如果这些组合是多线程安全的，并不代表更高层的组合是多线程安全的...。你要根据你的需要权衡提供哪一个层次的多线程安全（即确定mutex的作用范围）。比如上述的问题：类中每个单独函数都是多线程安全的，组合到一起不再多线程安全，要求的多线程范围是“函数组合”不再是单独的函数了，解决方法是将这个“函数组合”封装起来，将mutex作用范围扩展到这个组合，最简单的封装就是封装成一个单独的函数，将上面所有步骤封装成一个新的pop如下：

 

bool pop(T& v){
    std::lock_guard<std::mutex> lock(m_mutex);
    if(m_data.empty()) return false;
    value=data.top();
    data.pop();
    return true;
}

 

　　mutex的作用范围要设置合理，如果太小，只能保证小范围的多线程安全，但是小范围能确保足够少的函数需要mutex同步，效率高；如果作用范围太大，可以保证大范围的线程安全，但是大范围内很多操作都是没必要同步的本身是线程安全的，这样会降低性能。

 

第二部分：多个mutex导致的死锁：

 

1、最基本的最核心的解决方法（按顺序加锁）：

　　写字需要纸和笔，只有一张纸、一支笔（资源有限），两个人同时决定去写字，A拿到了笔，B拿到了纸（推进顺序不当），如果他们谁都不妥协就会导致死锁。教科书中死锁产生原因：资源有限、推进顺序不当。如果规定先拿到笔才能去拿纸，在这种情况下就不会产生死锁。

 

　　由此可知在多个mutex情况下规定加锁顺序可以避免死锁，PV操作如下：

 

P(pen);
P(paper);
    写字；
V(paper);
V(pen);

 

2、当按顺序加锁行不通时：

　　按规定顺序加锁可以避免死锁，但是有的情况下这种顺序并“无法”确定，例如Swap(C,D)需要访问需要同时访问C、D的内部数据，理所当然对C、D都加锁再访问，按常规思路首先对第一个参数加锁，然后对第二个参数加锁，如下Swap(C,D)所示，那另一个线程同时执行Swap(D,C)将参数调用来了，结果如何呢？死锁！

Swap(C,D):P(C);P(D);执行内部数据交换;V(D);V(C);
Swap(D,C):P(D);P(C);执行内部数据交换;V(C);V(D);

　　该如何解决呢？stl提供std::lock(mutex1,mutex2)将mutex1和mutex2的两个加锁操作当成原子一步操作（但你要记得对mutex1和mutex2进行unlock），这就不会导致死锁,std::lock要么将mutex1和mutex2都加锁也么一个都不锁。如下代码所示：

 

void Swap(T&C, T&D){
    if (&T == &D) return;    //对std::mutex两次lock是未定义的，从效率、安全考虑这都很必要
    std::lock(C.m_mutex, D.m_mutex);
    std::lock_guard<std::mutex> lock1(C.m_mutex, std::adopt_lock);
    std::lock_guard<std::mutex> lock2(D.m_mutex, std::adopt_lock);
    //数据交换的操作
    .....
}

 

　　std::adopt_lock告诉lock_guard在构造函数中不用在调用std::mutex::lock了，在前面已经调用过了。

 

3、hierarchical_mutex实现在运行时检查死锁的出现。

　　hierarchical_mutex规则思想是：将mutex分层，规定加锁顺序是由高层到底层才能进行，底层到高层报出运行时错误，这样就可以利用编程的方法检测死锁。书中实现了hierarchical_mutex类作为可分层的mutex，先列出使用方法如下：

 

 

 

 

 

 

hierarchical_mutex high_level_mutex(10000);
hierarchical_mutex low_level_mutex(500);
void ThreadA(){
    std::lock_guard<hierarchical_mutex> lock1(high_level_mutex);
    ...    //做一些使用high_level_mutex就可以干的事
        std::lock_guard<hierarchical_mutex> lock1(low_level_mutex);
    ...    //需要两个mutex同时加锁才可以干的事
}
void ThreadB(){
    std::lock_guard<hierarchical_mutex> lock1(low_level_mutex);
    ...    //做一些使用low_level_mutex就可以干的事
        //对高低层mutex加锁的情况下，对高层mutex加锁，不符合规定的顺序，抛出异常！
        std::lock_guard<hierarchical_mutex> lock1(high_level_mutex);
}

 

　　ThreadA符合hierarchical_mutex使用规定不会出现死锁是多线程安全的，ThreadB没按hierarchical_mutex 规定，有出现死锁的危险，在运行时就抛出异常。

　　hierarchical_mutex实现方法，要能使用lock_guard对hierarchical_mutex进行管理必须实现lock/unlock/try_lock方法；为了实现层次间的比较进而决定能不能加锁，需要记录准备加锁mutex的层号、该线程中当前已经加锁的mutex的层号、在解锁时恢复原先现场就要记录先前mutex的层号，综上hierarchical_mutex定义如下：

 

class hierarchical_mutex{
    std::mutex internal_mutex;
    //准备加锁mutex的层号
    const unsigned hierarchical_value;
    //在解锁时恢复原先现场就要记录先前mutex的层号
    unsigned previous_hierarchical_value;
    //该线程中当前已经加锁的mutex的层号
    static thread_local unsigned this_thread_hierarchical_value;
    //检查是否满足hierarchical_mutex规则
    void check_for_hierarchical_violation(){
        if (this_thread_hierarchical_value <= hierarchical_value){
            throw std::logic_error(“mutex hierarchical violated!”);
        }
    }
    void update_hierarchical_value(){
        previous_hierarchical_value = this_thread_hierarchical_value;
        this_thread_hierarchical_value = hierarchical_value;
    }
public:
    explicit hierarchical_mutex(unsigned value) :hierarchical_value(value), previous_hierarchical_value(0){}
    //满足hierarchical_mutex规则时加锁并更新内部记录变量
    void lock(){
        check_for_hierarchical_violation();
        internal_mutex.lock();
        update_hierarchical_value;
    }
    //恢复到调用lock之前的现场，解锁
    void unlock(){
        this_thread_hierarchical_value = previous_hierarchical_value;
        internal_mutex.unlock();
    }
    bool try_lock(){
        check_for_hierarchical_violation();
        if (!internal_mutex.try_lock())
            return false;
        update_hierarchical_value();
        return true;
    }
};

//使用UNSIGND_MAX初始化表明刚开始任何层的mutex都可以加锁成功
thread_local unsigned hierarchical_mutex::this_thread_hierarchical_value(UNSIGND_MAX);

 

　　标注：mutex::try_lock()检查mutex是否可以成功lock，如果可以就lock，如果不行立刻返回false.

 

第三部分：扩展

 

一、unique_lock:

 

1、相对lock_guard加锁、解锁更灵活的控制。

　　内部保存的是相关联mutex的状态，使用std::lock调用或std::unique_lock::lock()或使用std::unique_lock(mutex)构造时打开状态，使用unique_lock::unlock时关闭状态，在析构时候检查状态以决定是否调用std::mutex::unlock()，灵活就在随时调用lock、unlock。

std::unique_lock lock(m_mutex,std::defer_lock);    //std::defer_lock说明定义时对m_mutex不加锁
....
std::lock(lock);        //需要加锁时再加锁
....
lock.unlock();        //使用结束手动解锁

2、对mutex的控制权转移：

//*1作为参数
void f2(std::unique_lock&& p){ ..此范围内p关联的m_mutex已经加锁.. }
void f1(){
    std::unique_lock lock(m_mutex);        //对m_mutex加锁
    f2(lock);        //m_mutex管理权限转移到f2中
}
//*2作为返回值
unique_lock f2(){
    std::unique_lock lock(m_mutex);
    prepareData();
    return lock;
}
void f1(){
    std::unique_lock lock(f2());
    processData();
}

　　unique_lock内部保存的是关联mutex的状态标记，可以利用unique_lock::owe_lock()检查关联的mutex是否已经加锁。管理权转移为什么不用lock_guard呢？从源代码分析可知lock_guard没有拷贝构造、赋值、移动构造、移动赋值，意味着不能作为参数、返回值传递，但是unique_lock跟unique_ptr一样虽然没有拷贝构造、赋值，但有移动构造、移动赋值，“天生”是用做权限转移的。

 

二、Lazy-initialization:

 

单线程中：

std::shared_ptr<some_resource> resource_ptr;
void foo(){
    if (!resource_ptr){
        resource_ptr.reset(new some_resource);
    }
    resource_ptr->do_something();
}

多线程中：

std::shared_ptr<some_resource> resource_ptr;
std::mutex resource_mutex;
void foo(){
    std::unique_lock<std::mutex> lock(resource_mutex);
    if (!resource_ptr){
        resource_ptr.reset(new some_resource);
    }
    lock.unlock();
    resource_ptr->do_something();
}

　　我们要的是只在resource_ptr还没初始化时才加锁、解锁，这个版本任何时候进入foo都加锁、解锁，明显效率下降。然后就有了“infamous”双重锁：

 

std::shared_ptr<some_resource> resource_ptr;
std::mutex resource_mutex;
void foo(){
    if (!resource_ptr){//①
        std::lock_guard<std::mutex> lock(resource_mutex);
        if (!resource_ptr){
            resource_ptr.reset(new some_resource);//②
        }
    }
    resource_ptr->do_something();//③
}

 

　　看似很合理，但这却变成了一个让很多学者头疼的问题。问题出在②处对resource_ptr指针的赋值和some_resource构造函数的调用谁先谁后不确定（编译器要重排代码顺序，究竟怎么重排没规定），如果首先对resource_ptr指针赋值但是some_resource构造函数还没有调用，另一个线程在①处检查到resource_ptr有值了(true)，立刻执行③，其实是错误的。

　　标准库提供了一种解决方案，具体怎么解决的很复杂参考http://blog.jobbole.com/52164/。这里只谈使用方法，你只需知道这是多线程安全的并且比在正确情况下的双重检查锁效率还高，代码示例如下：

 

std::shared_ptr<some_resource> resource_ptr;
std::once_flag resource_flag;
void init_resource(){
    resource_ptr.reset(new some_resource);
}
void foo(){
    std::call_once(resource_flag, init_resource);
    resource_ptr->do_something();
}

 

　　你可以将此方法很容易扩展到多线程安全的单例模式，实现真正的多线程安全。

三、保护很少更新，但是经常是只读的数据：

　　可以想到使用读者优先，你可以自己利用mutex去设计自己的读者优先的函数，标准库并没有提供任何读写锁。但是boost库提供了boost::shared_mutex和boost::shared_lock函数可以很简单的解决这个问题：当使用lock_guard或unique_lock对boost::shared_mutex加锁后，对该boost::shared_mutex的获取方式（lock_guard/unique_lock/shared_lock）都会阻塞；当使用boost::shared_lock对该boost::shared_mutex加锁后，使用lock_guard和unique_lock对该boost::shared_lock的获取就会阻塞，但是使用shared_lock对该boost::shared_mutex的获取就不会阻塞。使用实例如下：

#include <map>
#include <string>
#include <mutex>
#include <boost/thread/shared_mutex.hpp>
class dns_entry;
class dns_cache
{
    std::map<std::string, dns_entry> entries;
    mutable boost::shared_mutex entry_mutex;
public:
    //读者
    dns_entry find_entry(std::string const& domain) const
    {
        boost::shared_lock<boost::shared_mutex> lk(entry_mutex);
        //阻塞写者但不阻塞读者
        std::map<std::string, dns_entry>::const_iterator const it =
            entries.find(domain);
        return (it == entries.end()) ? dns_entry() : it->second;
    }
    //写者
    void update_or_add_entry(std::string const& domain, dns_entry const& dns_details)
    {
        std::lock_guard<boost::shared_mutex> lk(entry_mutex);
        //将其他的写者和所有的读者都阻塞
        entries[domain] = dns_details;
    }
};

　　对recursive_mutex的使用，作者不推荐“Most of the time, if you think you want a recursive mutex, you probably need to change your design instead”，这里就不再详述了。

　　本帖全自己在阅读《c++ concurrency in action》中的总结，如果对你有帮助，请点个赞^-^