c++11单实例(singleton)初始化的几种方法(memory fence,atomic,call_once)

10km

发布于 2022-05-07 10:02:34

7220

发布于 2022-05-07 10:02:34

单实例模式(singleton)下要求一个类只能有一个实例,如何保证只创建一个实例？类的静态成员延迟初始化要求静态成员只能被初始化一次，也有类似的问题。在单线程环境下，这事儿很好办。

Singleton* Singleton::getInstance() {
    if (m_instance == nullptr) {
        m_instance = new Singleton;
    }

但是在多线程环境下，上面的方式显然是不安全的，有可能造成多个线程同时创建多个不同实例，并且除了最后一个实例被引用，其他实例都被丢弃而引起内存泄漏。

scope-based lock

所以如果在多线程编程中安全使用单实例对象（Singleton），最简单的做法是在访问时对函数加锁，使用这种方式，假定多个线程同时调用Singleton::getInstance方法，第一个获得锁的线程负责创建实例,其他线程则直接返回已经创建的实例：

Singleton* Singleton::getInstance() {
// lock是基于作用域的锁(scope-based lock),作用域结束自动释放，相当于java中的synchronized关键字起到的作用，
//本例中，函数返回时作用域结束，相当于对函数加锁，boost中scope_lock类实现此功能
    Lock lock;  
    if (m_instance == nullptr) {
        m_instance = new Singleton;
    }
    return m_instance;
}

这个方法无疑是安全的，但是当实例被创建之后，实际上已经不需要再对其进行加锁，加锁虽然不一定导致性能低下，但是在重负载情况下，这也可能导致响应缓慢。所以对于追求完美的人来说，这个办法确实有些让人不爽啊。

双重检查锁定模式(DCLP)

为了解决上面单实例初始化多次加锁的问题，程序员们想出了双重检查锁定模式(DCLP),估计你也想到了这个办法。代码原型如下：

Singleton* Singleton::getInstance() {
    if (pInstance == nullptr) { 
        // 第一次检查
        Lock lock;
        if (pInstance == nullptr) { 
            //  第二次检查
            pInstance = new Singleton;
        }
    }
    return pInstance;
}

在上面的代码中，第一次检查并没有加锁，就避免了每次调用getInstance时都要加锁的问题。貌似这个方法很完美了吧，逻辑上无懈可击，但是深入研究就会发现DCLP也并不是可靠的。具体的原因参见此下文，说得很详细了

C++和双重检查锁定模式(DCLP)的风险

读过上面这篇文章，我们可以得出一个结论:因为c++编译器在编译过程中会对代码进行优化，所以实际的代码执行顺序可能被打乱，另外因为CPU有一级二级缓存(cache),CPU的计算结果并不是及时更新到内存的,所以在多线程环境，不同线程间共享内存数据存在可见性问题，从而导致使用DCLP也存在风险。关于多线程间的数据可见性，就要涉及到c++的内存模型(memory model)的话题，这个事吧还真不太容易说明白，推荐一篇比较浅显易懂的文章

漫谈C++11多线程内存模型

memory fence/barrier

在上节，我们知道的双重检查锁定模式存在风险，那么有没有办法改进呢？办法是有，这就是内存栅栏技术(memory fence),也称内存栅障(memory barrier) 内存栅栏的作用在于保证内存操作的相对顺序，但并不保证内存操作的严格时序，确保第一个线程更新的数据对其他线程可见。一个 memory fence之前的内存访问操作必定先于其之后的完成关于内存栅栏的详细概念参见：

理解 Memory barrier（内存屏障）

以下是使用内存栅栏技术来实现DCLP的伪代码

Singleton* Singleton::getInstance() {
    Singleton* tmp = m_instance;
    ...                     
// 插入内存栅栏指令
    if (tmp == nullptr) {
        Lock lock;
        tmp = m_instance;
        if (tmp == nullptr) {
            tmp = new Singleton; // 语句1
            ...             
// 插入内存栅栏指令,确保语句2执行时，tmp指向的对象已经完成初始化构造函数
            m_instance = tmp;//语句2            
        }
    }
    return tmp;
}

这里，我们可以看到：在m_instance指针为NULL时，我们做了一次锁定，这个锁定确保创建该对象的线程对m_instance 的操作对其他线程可见。在创建线程内部构造块中，m_instance被再一次检查，以确保该线程仅创建了一份对象副本。

atomic_thread_fence

关于memory fence，不同的CPU，不同的编译器有不同的实现方式，要是直接使用还真是麻烦，不过，c++11中对这一概念进行了抽象，提供了方便的使用方式在c++11中，可以获取(acquire/consume)和释放(release)内存栅栏来实现上述功能。使用c++11中的atomic类型来包装m_instance指针，这使得对m_instance的操作是一个原子操作。下面的代码演示了如何使用内存栅栏：

std::atomic<Singleton*> Singleton::m_instance;
std::mutex Singleton::m_mutex;

Singleton* Singleton::getInstance() {
    Singleton* tmp = m_instance.load(std::memory_order_relaxed);
    std::atomic_thread_fence(std::memory_order_acquire);  
    if (tmp == nullptr) {
        std::lock_guard<std::mutex> lock(m_mutex);
        tmp = m_instance.load(std::memory_order_relaxed);
        if (tmp == nullptr) {
            tmp = new Singleton;
            std::atomic_thread_fence(std::memory_order_release); 
            m_instance.store(tmp, std::memory_order_relaxed);
        }
    }
    return tmp;
}

上面的代码中atomic_thread_fence在创建对象线程和使用对象线程之间建立了一种“同步-与”的关系（synchronizes-with）。

以下是摘自cplusplus关于atomic_thread_fence函数的说明:

Establishes a multi-thread fence: The point of call to this function becomes either an acquire or a release synchronization point (or both). All visible side effects from the releasing thread that happen before the call to this function are synchronized to also happen before the call this function in the acquiring thread. Calling this function has the same effects as a load or store atomic operation, but without involving an atomic value

中文大意是创建一个多线程栅栏,调用该函数的位置成为一个(acquire或release或两者)的同步点, 在release线程中此同步点之前的数据更新都将同步于acquire 线程的同步点之前，这就实现线程可见性一致

atomic

上节的代码使用内存栅栏锁定技术可以很方便地实现双重检查锁定。但是看着实在有点麻烦，在C++11中更好的实现方式是直接使用原子操作。

std::atomic<Singleton*> Singleton::m_instance;
std::mutex Singleton::m_mutex;

Singleton* Singleton::getInstance() {
    Singleton* tmp = m_instance.load(std::memory_order_acquire);
    if (tmp == nullptr) {
        std::lock_guard<std::mutex> lock(m_mutex);
        tmp = m_instance.load(std::memory_order_relaxed);
        if (tmp == nullptr) {
            tmp = new Singleton;
            m_instance.store(tmp, std::memory_order_release);
        }
    }
    return tmp;
}

如果你对memory_order的概念还是不太清楚，那么就使用C++顺序一致的原子操作，所有std::atomic的操作如果不带参数默认都是std::memory_order_seq_cst,即顺序的原子操作（sequentially consistent）,简称SC,使用（SC）原子操作库，整个函数执行指令都将保证顺序执行,这是一种最保守的内存模型策略。下面的代码就是使用SC原子操作实现双重检查锁定

std::atomic<Singleton*> Singleton::m_instance;
std::mutex Singleton::m_mutex;

Singleton* Singleton::getInstance() {
    Singleton* tmp = m_instance.load();
    if (tmp == nullptr) {
        std::lock_guard<std::mutex> lock(m_mutex);
        tmp = m_instance.load();
        if (tmp == nullptr) {
            tmp = new Singleton;
            m_instance.store(tmp);
        }
    }
    return tmp;
}

call_once(最简单的实现)

前面讲了辣么多就为了一个单实例初始化，太复杂啦，说实话我看明白上面的这内容也花了几天时间补充各种知识，觉得总算搞明白了，可以消停一下了，但是我脑子里突然了闪过一个名字，”call_once”。。。这是前阵子翻c++11标准头文件《mutex》时看到的一个函数,于是赶紧去查资料, 以下是对std::call_once的原文说明：

from:std::call_once@cplusplus.com Calls fn passing args as arguments, unless another thread has already executed (or is currently executing) a call to call_once with the same flag. If another thread is already actively executing a call to call_once with the same flag, it causes a passive execution: Passive executions do not call fn but do not return until the active execution itself has returned, and all visible side effects are synchronized at that point among all concurrent calls to this function with the same flag. If an active call to call_once ends by throwing an exception (which is propagated to its calling thread) and passive executions exist, one is selected among these passive executions, and called to be the new active call instead. Note that once an active execution has returned, all current passive executions and future calls to call_once (with the same flag) also return without becoming active executions. The active execution uses decay copies of the lvalue or rvalue references of fn and args, ignoring the value returned by fn. also see: call_once 函数 @microsoft std:call_once@cppreference.com

大意就是

call_one保证函数fn只被执行一次，如果有多个线程同时执行函数fn调用，则只有一个活动线程(active call)会执行函数，其他的线程在这个线程执行返回之前会处于”passive execution”(被动执行状态)—不会直接返回，直到活动线程对fn调用结束才返回。对于所有调用函数fn的并发线程的数据可见性都是同步的(一致的)。如果活动线程在执行fn时抛出异常，则会从处于”passive execution”状态的线程中挑一个线程成为活动线程继续执行fn,依此类推。一旦活动线程返回,所有”passive execution”状态的线程也返回,不会成为活动线程。

由上面的说明，我们可以确信call_once完全满足对多线程状态下对数据可见性的要求。所以利用call_once再结合lambda表达式,前面几节那么多复杂代码，在这里千言万语凝聚为一句话：

Singleton* Singleton::m_instance;
Singleton* Singleton::getInstance() {
    static std::once_flag oc;//用于call_once的局部静态变量
    std::call_once(oc, [&] {  m_instance = new Singleton();});
    return m_instance;
}