【C++】C++11中局部static变量的线程安全问题
慕雪年华
  2023-02-20 21:13:46 2023-02-20 21:13   2023-04-03 09:23:39    2.2k 字  

C++11中局部static变量的线程安全问题

前言

大家都知道，在C++11标准中，要求局部静态变量初始化具有线程安全性，所以我们可以很容易实现一个线程安全的单例类：

class Foo
{
public:
    static Foo *getInstance()
    {
        static Foo s_instance;
        return &s_instance;
    }
private:
    Foo() {}
};

在C++标准中，是这样描述的（在标准草案的6.7节中）：

such a variable is initialized the first time control passes through its declaration; such a variable is considered initialized upon the completion of its initialization. If the initialization exits by throwing an exception, the initialization is not complete, so it will be tried again the next time control enters the declaration. If control enters the declaration concurrently while the variable is being initialized, the concurrent execution shall wait for completion of the initialization. If control re-enters the declaration recursively while the variable is being initialized, the behavior is undefined.

分析

标准关于局部静态变量初始化，有这么几点要求：

  1. 变量在代码第一次执行到变量声明的地方时初始化。
  2. 初始化过程中发生异常的话视为未完成初始化，未完成初始化的话，需要下次有代码执行到相同位置时再次初始化。
  3. 在当前线程执行到需要初始化变量的地方时，如果有其他线程正在初始化该变量，则阻塞当前线程，直到初始化完成为止。
  4. 如果初始化过程中发生了对初始化的递归调用，则视为未定义行为。

关于第4点，如果不明白，可以参考以下代码：

class Bar
{
public:
    static Bar *getInstance()
    {
        static Bar s_instance;
        return &s_instance;
    }
private:
    Bar()
    {
        getInstance();
    }
};

GCC的实现

以GCC 7.3.0版本为例，我们来分析GCC是如何实现标准的。

Foo::getInstance()

使用GCC编译后，我们使用gdb将文章开头的Foo::getInstance()反汇编：

Dump of assembler code for function Foo::getInstance():
   0x00005555555546ea <+0>:     push   %rbp
   0x00005555555546eb <+1>:     mov    %rsp,%rbp
=> 0x00005555555546ee <+4>:     movzbl 0x20092b(%rip),%eax        # 0x555555755020 <_ZGVZN3Foo11getInstanceEvE10s_instance>
   0x00005555555546f5 <+11>:    test   %al,%al
   0x00005555555546f7 <+13>:    sete   %al
   0x00005555555546fa <+16>:    test   %al,%al
   0x00005555555546fc <+18>:    je     0x55555555472b <Foo::getInstance()+65>
   0x00005555555546fe <+20>:    lea    0x20091b(%rip),%rdi        # 0x555555755020 <_ZGVZN3Foo11getInstanceEvE10s_instance>
   0x0000555555554705 <+27>:    callq  0x5555555545b0 <__cxa_guard_acquire@plt>
   0x000055555555470a <+32>:    test   %eax,%eax
   0x000055555555470c <+34>:    setne  %al
   0x000055555555470f <+37>:    test   %al,%al
   0x0000555555554711 <+39>:    je     0x55555555472b <Foo::getInstance()+65>
   0x0000555555554713 <+41>:    lea    0x2008fe(%rip),%rdi        # 0x555555755018 <_ZZN3Foo11getInstanceEvE10s_instance>
   0x000055555555471a <+48>:    callq  0x555555554734 <Foo::Foo()>
   0x000055555555471f <+53>:    lea    0x2008fa(%rip),%rdi        # 0x555555755020 <_ZGVZN3Foo11getInstanceEvE10s_instance>
   0x0000555555554726 <+60>:    callq  0x5555555545a0 <__cxa_guard_release@plt>
   0x000055555555472b <+65>:    lea    0x2008e6(%rip),%rax        # 0x555555755018 <_ZZN3Foo11getInstanceEvE10s_instance>
   0x0000555555554732 <+72>:    pop    %rbp
   0x0000555555554733 <+73>:    retq   
End of assembler dump.

在+4、+20、+53出现的_ZGVZN3Foo11getInstanceEvE10s_instance使用c++filt分析为guard variable for Foo::getInstance()::s_instance，而+41、+65位置出现的_ZZN3Foo11getInstanceEvE10s_instance则为Foo::getInstance()::s_instance。后者是s_instance这个局部静态变量，前者从名字看就知道是个guard标志变量，用来指示局部静态变量的初始化状态。

+4 ~ +18

测试guard变量的第一个字节，如果为0，代表s_instance未初始化，进入+27；否则代表s_instance已初始化，进入+65。

+20 ～ +27

以guard变量地址作为参数，执行__cxa_guard_acquire函数。

+32 ～ +39

测试返回值，如果为0，代表s_instance已初始化，进入+65；否则代表s_instance未初始化，进入+41。

+41 ～ +48

初始化s_instance

+53 ～ +60

以guard变量地址作为参数，执行__cxa_guard_release函数。

+65 ～ +73

返回s_instance地址

__cxa_guard_acquire

我们来看看__cxa_guard_acquire这个函数具体做了什么，该函数代码位于gcc-7-7.3.0/gcc-7.3.0/libstdc++-v3/libsupc++/guard.cc。由于这个函数针对不同平台做了不同的实现，有些我们不需要的代码，以我机器的设置，支持线程和futex系统调用，所以删除了一些不相关的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
int __cxa_guard_acquire (__guard *g)
{
    // If the target can reorder loads, we need to insert a read memory
    // barrier so that accesses to the guarded variable happen after the
    // guard test.

    // 1
    if (_GLIBCXX_GUARD_TEST_AND_ACQUIRE (g))
        return 0;

    // If __atomic_* and futex syscall are supported, don't use any global
    // mutex.

    // 2
    if (__gthread_active_p ())
    {
        int *gi = (int *) (void *) g;

        // 3
        const int guard_bit = _GLIBCXX_GUARD_BIT;
        const int pending_bit = _GLIBCXX_GUARD_PENDING_BIT;
        const int waiting_bit = _GLIBCXX_GUARD_WAITING_BIT;

        while (1)
        {
            // 4
            int expected(0);
            if (__atomic_compare_exchange_n(gi, &expected, pending_bit, false,
                                            __ATOMIC_ACQ_REL,
                                            __ATOMIC_ACQUIRE))
            {
                // This thread should do the initialization.
                return 1;
            }

            // 5
            if (expected == guard_bit)
            {
                // Already initialized.
                return 0;
            }

            // 6
            if (expected == pending_bit)
            {
                // Use acquire here.

                // 7
                int newv = expected | waiting_bit;

                // 8
                if (!__atomic_compare_exchange_n(gi, &expected, newv, false,
                                                 __ATOMIC_ACQ_REL,
                                                 __ATOMIC_ACQUIRE))
                {
                    // 9
                    if (expected == guard_bit)
                    {
                        // Make a thread that failed to set the
                        // waiting bit exit the function earlier,
                        // if it detects that another thread has
                        // successfully finished initialising.
                        return 0;
                    }

                    // 10
                    if (expected == 0)
                        continue;
                }

                // 11
                expected = newv;
            }

            // 12
            syscall (SYS_futex, gi, _GLIBCXX_FUTEX_WAIT, expected, 0);
        }
    }

    return acquire (g);
}
  1. 首先检测guard变量，guard变量等于1的话，直接返回0，代表s_instance已初始化，不需要再次初始化。
  2. 检测是否为多线程环境，如果没有多线程的话，也就没有必要去做额外工作来保证线程安全了。
  3. guard_bit表示s_instance已经初始化成功；pending_bit表示s_instance正在初始化；waiting_bit表示有其他线程正在等待s_instance的初始化。
  4. 使用一个原子操作来检测guard变量是否为0，如果为0，则由当前线程初始化s_instance，把pending_bit写入guard变量，返回1。如果不为0，则将guard当前值写入expected。
  5. 检测expected值是否为guard_bit，如果是，则s_instance已初始化完成，不再需要初始化，返回0。
  6. 检测expected值是否为pending_bit，如果是，说明s_instance正在初始化，且没有其他线程等待初始化。
  7. 将newv变量设置为pending_bit | waiting_bit，表示s_instance正在初始化且有线程正在等待初始化。
  8. 使用一个原子操作来检测guard变量是否为pending_bit，如果不是，说明有其他线程修改了guard变量，需要做进一步检测；如果是，说明没有其他线程修改guard变量，则将pending_bit | waiting_bit写入guard变量。
  9. 如果expected等于guard_bit，说明s_instance被初始化成功，不需要再初始化，返回0。
  10. 如果expected等于0，说明s_instance初始化失败，回到4重新开始检测。
  11. 如果在8中没有其他线程修改过guard变量，将expected设置为pending_bit | waiting_bit，表示s_instance正在初始化且有线程（也就是当前线程）正在等待初始化。
  12. 如果在6处没有进入if分支，说明expected等于pending_bit | waiting_bit，如果进入了if分支，由11可得，此时expected也被修改为了pending_bit | waiting_bit。总之，此时s_instance正在初始化且有线程正在等待初始化。利用futex系统调用，再次检测guard变量是否发生了变化，如果发生了变化，回到4重新开始检测；如果没有发生变化，仍然等于pending_bit | waiting_bit，则挂起当前线程。

总之，__cxa_guard_acquire要么返回0要么返回1，用来指示s_instance已初始化或未初始化。__cxa_guard_acquire可能会导致当前线程挂起，这发生在s_instance正在初始化的时候。

__cxa_guard_release

由于__cxa_guard_acquire可能导致当前线程挂起，因此需要在s_instance初始化完成后使用将__cxa_guard_release线程唤醒。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
void __cxa_guard_release (__guard *g) throw ()
{
    // If __atomic_* and futex syscall are supported, don't use any global
    // mutex.

    // 1
    if (__gthread_active_p ())
    {
        int *gi = (int *) (void *) g;
        const int guard_bit = _GLIBCXX_GUARD_BIT;
        const int waiting_bit = _GLIBCXX_GUARD_WAITING_BIT;

        // 2
        int old = __atomic_exchange_n (gi, guard_bit, __ATOMIC_ACQ_REL);

        // 3
        if ((old & waiting_bit) != 0)
            syscall (SYS_futex, gi, _GLIBCXX_FUTEX_WAKE, INT_MAX);
        return;
    }

    set_init_in_progress_flag(g, 0);
    _GLIBCXX_GUARD_SET_AND_RELEASE (g);
}
  1. 检测是否为多线程环境
  2. 使用原子操作将guard变量置为guard_bit，同时获取guard变量原始值。
  3. 如果guard变量原始值包含waiting_bit，说明有线程挂起（或将要调用futex欲使线程挂起），调用futex唤醒挂起的进程。

__cxa_guard_abort

由于s_instance可能初始化失败（本例中并未体现），因此还有一个__cxa_guard_abort函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
void __cxa_guard_abort (__guard *g) throw ()
{
    // If __atomic_* and futex syscall are supported, don't use any global
    // mutex.
    if (__gthread_active_p ())
    {
        int *gi = (int *) (void *) g;
        const int waiting_bit = _GLIBCXX_GUARD_WAITING_BIT;
        int old = __atomic_exchange_n (gi, 0, __ATOMIC_ACQ_REL);

        if ((old & waiting_bit) != 0)
            syscall (SYS_futex, gi, _GLIBCXX_FUTEX_WAKE, INT_MAX);
        return;
    }

    set_init_in_progress_flag(g, 0);
}

与__cxa_guard_release基本一致，不同的地方在于会将guard变量置0。

递归初始化调用

由于在C++11标准中，初始化如果发生了递归是未定义行为，所以GCC 7.3.0针对是否为多线程环境做了不同的处理。如果是多线程环境，不进行额外处理，会发生死锁；如果是单线程环境，则会抛异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// acquire() is a helper function used to acquire guard if thread support is
// not compiled in or is compiled in but not enabled at run-time.
static int
acquire(__guard *g)
{
    // Quit if the object is already initialized.
    if (_GLIBCXX_GUARD_TEST(g))
        return 0;

    if (init_in_progress_flag(g))
        throw_recursive_init_exception();

    set_init_in_progress_flag(g, 1);
    return 1;
}

总结

看到了GCC如此复杂的实现，我的个人感想是还是不要自己造轮子来保证单例类的线程安全了，想要做到和GCC一样的高效还是比较难的，利用C++11标准的带来的便利就挺好。

1
2
原文作者：imred
原文链接：https://blog.csdn.net/imred/article/details/89069750
  1. 前言
  2. 分析
  3. GCC的实现
    1. Foo::getInstance()
      1. +4 ~ +18
      2. +20 ～ +27
      3. +32 ～ +39
      4. +41 ～ +48
      5. +53 ～ +60
      6. +65 ～ +73
    2. __cxa_guard_acquire
    3. __cxa_guard_release
    4. __cxa_guard_abort
    5. 递归初始化调用
  4. 总结