深入探讨C++多线程性能优化

深入探讨C++多线程性能优化

在当代软件开辟中，多线程编程已成为提升应用程序性能和相应速度的关键技术之一。尤其在C++领域，多线程编程不仅能充分利用多核处置处罚器的优势，还能显著提高计算密集型使命的服从。然而，多线程编程也带来了诸多挑战，特别是在性能优化方面。本文将深入探讨影响C++多线程性能的一些关键因素，比较锁机制与原子操作的性能。通过这些内容，盼望能为开辟者提供有代价的见解和实用的优化策略，助力于更高效的多线程编程实践。
先在开头给一个例子，你以为下面这段benchmark代码效果会是怎样的。这里的逻辑很简朴，将0-20000按线程切成n片，每个线程在一个Set里查找这个数字存不存在，存在则计数+1。

1. #include <benchmark/benchmark.h>
2. #include <iostream>
3. #include <memory>
4. #include <mutex>
5. #include <unordered_set>
7. constexpr int kSetSize = 10000;
9. class MyBenchmark : public benchmark::Fixture {
10. public:
11.   void SetUp(const ::benchmark::State& state) override {
12.     std::call_once(flag, [this]() {
13.       s = std::make_shared<std::unordered_set<int>>();
14.       for (int i = 0; i < kSetSize; i++) {
15.         s->insert(i);
16.       }
17.     });
18.   }
20.   std::shared_ptr<std::unordered_set<int>> GetSet() { return s; }
22. private:
23.   std::shared_ptr<std::unordered_set<int>> s;
24.   std::once_flag flag;
25. };
27. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWork)(benchmark::State& state) {
28.   for (auto _ : state) {
29.     int size_sum = kSetSize * 2;
30.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
31.     int sum = 0;
32.     int start = state.thread_index() * size_per_thread;
33.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
34.     for (int i = start; i < end; i++) {
35.       auto inst = GetSet();
36.       if (inst->count(i) > 0) {
37.         sum++;
38.       }
39.     }
40.   }
41. }
43. // 注册基准测试，并指定线程数
44. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(1);
45. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(2);
46. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(4);
47. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(8);
49. BENCHMARK_MAIN();

复制代码

跑出来的效果如下：

将使命切分成多个片断并交由多线程执行后，团体性能不仅没有提升，反而降落了，且性能与线程数成反比。那么，题目来了：导致这种效果的原因是什么？怎样才华实现合理的并行执行，从而低落CPU的执行时间？在接下来的部门，笔者将为你揭示答案。
影响多线程性能的因素

笔者以为，影响多线程性能的重要因素有以下两个：
 1.Lock Contention。
 2.Cache Coherency。
Lock Contention对应使用锁来处置处罚多线程同步题目的场景，而Cache Coherency则对应使用原子操作来处置处罚多线程同步题目的场景。
Lock Contention

在多线程环境中，多个线程同时实验获取同一个锁（Lock）时，会发生竞争征象，这就是所谓的锁竞争（Lock Contention）。锁竞争会导致线程或进程被阻塞，等待锁被释放，从而影响系统的性能和相应时间。大多数情况下，开辟人员会选择使用锁来办理线程间的同步题目，因此锁竞争题目也变得广为人知且轻易明白。由于锁的存在，位于临界区的代码在同一时候只能由一个线程执行。因此，优化的思路就是尽量制止多个线程同时访问同一资源。常见的优化方向有两种：
 1.淘汰临界区大小：临界区越小，这段代码的执行时间就越短，从而在团体程序运行时间中所占的比例也越小，冲突也就越少。
 2.对共享资源进行分桶操作：每个线程只会在某个桶上访问资源，理想情况下，每个线程都会访问不同的桶，这样就不会有冲突。
淘汰临界区大小需要开辟者对自己的代码进行细致思考，将不必要的操作放在临界区外，比方一些初始化和内存分配操作。
对共享资源进行分桶操作在工程实践中也非经常见。比方，LevelDB的LRUCache中，每个Key只会固定在一个桶上。如果hash函数充足优秀且数据分布充足随机，这种方法可以大大提高LRUCache的性能。
Cache Coherency

缓存一致性（Cache Coherency）是指在多处置处罚器系统中，确保各个处置处罚器的缓存中的数据保持一致的机制。由于当代计算机系统通常包罗多个处置处罚器，每个处置处罚器都有自己的缓存（如L1、L2、L3缓存），因此在并发访问共享内存时，可能会出现缓存数据不一致的题目。缓存一致性协议旨在办理这些题目，确保所有处置处罚器在访问共享内存时看到的是一致的数据。
当我们对一个共享变量进行写入操作时，实际上需要通过缓存一致性协议将该变量的更新同步到其他线程的缓存中，否则可能会读到不一致的值。实际上，这个同步过程的单位是一个缓存行（Cache Line），而且同步过程相对较慢，由于涉及到跨核通讯。
由此引申出两个严重影响性能的征象：
 1.Cache Ping-Pong。
 2.False Sharing。
Cache Ping-Pong

缓存乒乓效应（Cache Ping-Pong）是指在多处置处罚器系统中，多个处置处罚器频繁地对同一个缓存行（Cache Line）进行读写操作，导致该缓存行在不同处置处罚器的缓存之间频繁地来回通报。这种征象会导致系统性能降落，由于缓存行的频繁通报会引起大量的缓存一致性流量和处置处罚器间通讯开销。
讲到这里，其实就可以解释为什么开头那段代码会随着线程数量的增长而性能反而降落。代码中的变量 s 是一个共享资源，但它使用了 shared_ptr。在复制 shared_ptr 时，会引起引用计数的增长（计数+1），多个线程频繁对同一个缓存行进行读写操作，从而引发缓存乒乓效应，导致性能降落。最简朴的修改方式就是去掉 shared_ptr，代码如下，同时还可以得到我们预期的效果，即CPU时间随着线程数的增长而低落：

1. #include <benchmark/benchmark.h>
2. #include <iostream>
3. #include <memory>
4. #include <mutex>
5. #include <unordered_set>
7. constexpr int kSetSize = 10000;
9. class MyBenchmark : public benchmark::Fixture {
10. public:
11.   void SetUp(const ::benchmark::State& state) override {
12.     std::call_once(flag, [this]() {
13.       for (int i = 0; i < kSetSize; i++) {
14.         s.insert(i);
15.       }
16.     });
17.   }
19.   const std::unordered_set<int>& GetSet() { return s; }
21. private:
22.   std::unordered_set<int> s;
23.   std::once_flag flag;
24. };
26. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWork)(benchmark::State& state) {
27.   for (auto _ : state) {
28.     int size_sum = kSetSize * 2;
29.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
30.     int sum = 0;
31.     int start = state.thread_index() * size_per_thread;
32.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
33.     for (int i = start; i < end; i++) {
34.       const auto& inst = GetSet();
35.       if (inst.count(i) > 0) {
36.         benchmark::DoNotOptimize(sum++);
37.       }
38.     }
39.   }
40. }
42. // 注册基准测试，并指定线程数
43. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(1);
44. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(2);
45. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(4);
46. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(8);
48. BENCHMARK_MAIN();

复制代码

False Sharing

伪共享（False Sharing）实际上是一种特殊的缓存乒乓效应（Cache Ping-Pong）。它指的是在多处置处罚器系统中，多个处置处罚器访问不同的数据，但这些数据恰好位于同一个缓存行中，导致该缓存行在不同处置处罚器的缓存之间频繁通报。尽管处置处罚器访问的是不同的数据，但由于它们共享同一个缓存行，仍旧会引发缓存一致性流量，导致性能降落。
为了更好地明白这一征象，我们可以对上面的代码进行一些修改。假设我们使用一个 vector<atomic> 来记录不同线程的 sum 值，这样固然不同线程修改的是不同的sum，但是照旧在一个缓存行上。使用 atomic 是为了强制触发缓存一致性协议，否则操作系统可能会进行优化，不会立即将修改反映到主存。

1. #include <benchmark/benchmark.h>
2. #include <atomic>
3. #include <iostream>
4. #include <memory>
5. #include <mutex>
6. #include <unordered_set>
8. constexpr int kSetSize = 10000;
10. class MyBenchmark : public benchmark::Fixture {
11. public:
12.   void SetUp(const ::benchmark::State& state) override {
13.     std::call_once(flag, [this, &state]() {
14.       for (int i = 0; i < kSetSize; i++) {
15.         s.insert(i);
16.       }
17.       sums = std::vector<std::atomic<int>>(state.threads());
18.     });
19.   }
21.   const std::unordered_set<int>& GetSet() { return s; }
22.   std::vector<std::atomic<int>>& GetSums() { return sums; }
24. private:
25.   std::unordered_set<int> s;
26.   std::once_flag flag;
27.   std::vector<std::atomic<int>> sums;
28. };
30. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWork)(benchmark::State& state) {
31.   for (auto _ : state) {
32.     int size_sum = kSetSize;
33.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
34.     int sum = 0;
35.     int start = state.thread_index() * size_per_thread;
36.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
37.     for (int i = start; i < end; i++) {
38.       const auto& inst = GetSet();
39.       if (inst.count(i) > 0) {
40.         benchmark::DoNotOptimize(GetSums()[state.thread_index()]++);
41.       }
42.     }
43.   }
44. }
46. // 注册基准测试，并指定线程数
47. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(1);
48. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(2);
49. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(4);
50. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(8);
52. BENCHMARK_MAIN();

复制代码

可以看到，尽管不同线程没有使用同一个变量，但由于 sums 里面的元素共享同一个缓存行（Cache Line），同样会导致性能急剧降落。
针对这种情况，只要我们将 sums 中的元素隔离，使它们不在同一个缓存行上，就不会引发这个题目。一样平常来说，缓存行的大小为64字节，我们可以使用一个类填充到64个字节来实现隔离。优化后的代码如下：

1. #include <benchmark/benchmark.h>
2. #include <atomic>
3. #include <iostream>
4. #include <memory>
5. #include <mutex>
6. #include <unordered_set>
8. constexpr int kSetSize = 10000;
10. struct alignas(64) PaddedCounter {
11.   std::atomic<int> value{0};
12.   char padding[64 - sizeof(std::atomic<int>)];  // 填充到缓存行大小
13. };
15. class MyBenchmark : public benchmark::Fixture {
16. public:
17.   void SetUp(const ::benchmark::State& state) override {
18.     std::call_once(flag, [this, &state]() {
19.       for (int i = 0; i < kSetSize; i++) {
20.         s.insert(i);
21.       }
22.       sums = std::vector<PaddedCounter>(state.threads());
23.     });
24.   }
26.   const std::unordered_set<int>& GetSet() { return s; }
27.   std::vector<PaddedCounter>& GetSums() { return sums; }
29. private:
30.   std::unordered_set<int> s;
31.   std::once_flag flag;
32.   std::vector<PaddedCounter> sums;
33. };
35. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWork)(benchmark::State& state) {
36.   for (auto _ : state) {
37.     int size_sum = kSetSize;
38.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
39.     int sum = 0;
40.     int start = state.thread_index() * size_per_thread;
41.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
42.     for (int i = start; i < end; i++) {
43.       const auto& inst = GetSet();
44.       if (inst.count(i) > 0) {
45.         benchmark::DoNotOptimize(GetSums()[state.thread_index()].value++);
46.       }
47.     }
48.   }
49. }
51. // 注册基准测试，并指定线程数
52. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(1);
53. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(2);
54. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(4);
55. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(8);
57. BENCHMARK_MAIN();

复制代码

Lock VS Atomic

Lock Atomic Benchmark

很多人都以为锁（lock）比原子操作（atomic）要更慢，那么实际上真的是这样吗？下面我们通过两个测试来进行对比。
公平起见，我们将使用一个基于 atomic 变量实现的自旋锁（SpinLock）与 std::mutex 进行性能对比。自旋锁的实现摘自 Folly 库。其原理是使用一个 atomic 变量来标志是否被占用，并使用 acquire-release 内存序来保证临界区的精确性。在冲突过大时，自旋锁会使用 sleep 让出 CPU。代码如下：

1. #pragma once
3. #include <atomic>
4. #include <cstdint>
6. class Sleeper {
7.   static const uint32_t kMaxActiveSpin = 4000;
8.   uint32_t spin_count_;
10. public:
11.   constexpr Sleeper() noexcept : spin_count_(0) {}
13.   inline __attribute__((always_inline)) static void sleep() noexcept {
14.     struct timespec ts = {0, 500000};
15.     nanosleep(&ts, nullptr);
16.   }
18.   inline __attribute__((always_inline)) void wait() noexcept {
19.     if (spin_count_ < kMaxActiveSpin) {
20.       ++spin_count_;
21. #ifdef __x86_64__
22.       asm volatile("pause" ::: "memory");
23. #elif defined(__aarch64__)
24.       asm volatile("yield" ::: "memory");
25. #else
26.       // Fallback for other architectures
27. #endif
29.     } else {
30.       sleep();
31.     }
32.   }
33. };
35. class SpinLock {
36.   enum { FREE = 0, LOCKED = 1 };
38. public:
39.   constexpr SpinLock() : lock_(FREE) {}
41.   inline __attribute__((always_inline)) bool try_lock() noexcept { return cas(FREE, LOCKED); }
43.   inline __attribute__((always_inline)) void lock() noexcept {
44.     Sleeper sleeper;
45.     while (!try_lock()) {
46.       do {
47.         sleeper.wait();
48.       } while (AtomicCast(&lock_)->load(std::memory_order_relaxed) == LOCKED);
49.     }
50.   }
52.   inline __attribute__((always_inline)) void unlock() noexcept {
53.     AtomicCast(&lock_)->store(FREE, std::memory_order_release);
54.   }
56. private:
57.   inline __attribute__((always_inline)) bool cas(uint8_t compare, uint8_t new_val) noexcept {
58.     return AtomicCast(&lock_)->compare_exchange_strong(compare, new_val, std::memory_order_acquire,
59.                                                        std::memory_order_relaxed);
60.   }
62.   inline __attribute__((always_inline)) static std::atomic<uint8_t>* AtomicCast(uint8_t* value) {
63.     return reinterpret_cast<std::atomic<uint8_t>*>(value);
64.   }
66. private:
67.   uint8_t lock_;
68. };

复制代码

在第一个benchmark中，我们测试了无竞争情况下的性能。也就是说，原子变量的CAS操作只会执行一次，不会进入 sleep 状态。在这种情况下，自旋锁（SpinLock）等价于一次原子 set 操作。
代码如下：

1. #include <benchmark/benchmark.h>
2. #include <atomic>
3. #include <iostream>
4. #include <memory>
5. #include <mutex>
6. #include <unordered_set>
7. #include "spin_lock.h"
9. constexpr int kSetSize = 10000;
11. // struct alignas(64) PaddedCounter {
12. //   std::atomic<int> value{0};
13. //   char padding[64 - sizeof(std::atomic<int>)];  // 填充到缓存行大小
14. // };
16. struct alignas(64) PaddedCounterLock {
17.   int value{0};
18.   char padding[64 - sizeof(std::atomic<int>)];  // 填充到缓存行大小
19. };
21. class MyBenchmark : public benchmark::Fixture {
22. public:
23.   void SetUp(const ::benchmark::State& state) override {
24.     std::call_once(flag, [this, &state]() {
25.       for (int i = 0; i < kSetSize; i++) {
26.         s.insert(i);
27.       }
28.       sums_atomic = std::vector<PaddedCounterLock>(state.threads());
29.       sum_lock = std::vector<PaddedCounterLock>(state.threads());
30.     });
31.   }
33.   const std::unordered_set<int>& GetSet() { return s; }
34.   std::vector<PaddedCounterLock>& GetSumsAtomic() { return sums_atomic; }
35.   std::vector<PaddedCounterLock>& GetSumLock() { return sum_lock; }
37. private:
38.   std::unordered_set<int> s;
39.   std::once_flag flag;
40.   std::vector<PaddedCounterLock> sums_atomic;
41.   std::vector<PaddedCounterLock> sum_lock;
42. };
44. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkAtomic)(benchmark::State& state) {
45.   SpinLock m;
46.   for (auto _ : state) {
47.     int size_sum = kSetSize;
48.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
49.     int sum = 0;
50.     int start = state.thread_index() * size_per_thread;
51.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
52.     for (int i = start; i < end; i++) {
53.       const auto& inst = GetSet();
54.       if (inst.count(i) > 0) {
55.         std::lock_guard<SpinLock> lg(m);
56.         benchmark::DoNotOptimize(GetSumsAtomic()[state.thread_index()].value++);
57.       }
58.     }
59.   }
60. }
62. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkLock)(benchmark::State& state) {
63.   std::mutex m;
64.   for (auto _ : state) {
65.     int size_sum = kSetSize;
66.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
67.     int sum = 0;
68.     int start = state.thread_index() * size_per_thread;
69.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
70.     for (int i = start; i < end; i++) {
71.       const auto& inst = GetSet();
72.       if (inst.count(i) > 0) {
73.         std::lock_guard<std::mutex> lg(m);
74.         benchmark::DoNotOptimize(GetSumLock()[state.thread_index()].value++);
75.       }
76.     }
77.   }
78. }
80. // 注册基准测试，并指定线程数
81. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(1);
82. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(2);
83. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(4);
84. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(8);
86. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(1);
87. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(2);
88. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(4);
89. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(8);
91. BENCHMARK_MAIN();

复制代码

benchmark效果：

第二个benchmark是对比竞争激烈时的性能，代码如下：

1. #include <benchmark/benchmark.h>
2. #include <atomic>
3. #include <iostream>
4. #include <memory>
5. #include <mutex>
6. #include <unordered_set>
7. #include "spin_lock.h"
9. constexpr int kSetSize = 10000;
11. class MyBenchmark : public benchmark::Fixture {
12. public:
13.   void SetUp(const ::benchmark::State& state) override {
14.     std::call_once(flag, [this, &state]() {
15.       for (int i = 0; i < kSetSize; i++) {
16.         s.insert(i);
17.       }
18.       count_ = 0;
19.     });
20.   }
22.   const std::unordered_set<int>& GetSet() { return s; }
23.   void SpinLockAndAdd() {
24.     std::lock_guard<SpinLock> lg(m1_);
25.     count_++;
26.   }
28.   void MutexLockAndAdd() {
29.     std::lock_guard<std::mutex> lg(m2_);
30.     count_++;
31.   }
33. private:
34.   std::unordered_set<int> s;
35.   std::once_flag flag;
36.   uint32_t count_;
37.   SpinLock m1_;
38.   std::mutex m2_;
39. };
41. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkAtomic)(benchmark::State& state) {
42.   for (auto _ : state) {
43.     int size_sum = kSetSize;
44.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
45.     int sum = 0;
46.     int start = state.thread_index() * size_per_thread;
47.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
48.     for (int i = start; i < end; i++) {
49.       const auto& inst = GetSet();
50.       if (inst.count(i) > 0) {
51.         SpinLockAndAdd();
52.       }
53.     }
54.   }
55. }
57. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkLock)(benchmark::State& state) {
58.   for (auto _ : state) {
59.     int size_sum = kSetSize;
60.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
61.     int sum = 0;
62.     int start = state.thread_index() * size_per_thread;
63.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
64.     for (int i = start; i < end; i++) {
65.       const auto& inst = GetSet();
66.       if (inst.count(i) > 0) {
67.         MutexLockAndAdd();
68.       }
69.     }
70.   }
71. }
73. // 注册基准测试，并指定线程数
74. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(1);
75. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(2);
76. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(4);
77. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(8);
79. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(1);
80. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(2);
81. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(4);
82. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(8);
84. BENCHMARK_MAIN();

复制代码

benchmark效果：

可以看到，无论是哪一种情况，std::mutex 的性能都更优。当然，这个测试效果可能会因不同的操作系统而有所不同，但至少可以得出一个结论：这两者的性能是一个量级的，并不存在 atomic 肯定比 std::mutex 更快的说法。这其实是由于当代 C++ 中的 std::mutex 实现已经高度优化，其实现与上面的自旋锁（SpinLock）非常相似，在低竞争的情况下并不会陷入内核态。
那么，按上面的说法，是不是我们根本不需要 atomic 变量呢？先来分析一下 atomic 的长处。
atomic 的长处有：
 1.可以实现内存占用极小的锁。
 2.当临界区操作可以等价于一个原子操作时，性能会更高。
对于第二个结论，我们可以做个测试。同样，拿前面的例子稍作修改。
case 1如下:

1. #include <benchmark/benchmark.h>
2. #include <atomic>
3. #include <iostream>
4. #include <memory>
5. #include <mutex>
6. #include <unordered_set>
7. #include "spin_lock.h"
9. constexpr int kSetSize = 10000;
11. class MyBenchmark : public benchmark::Fixture {
12. public:
13.   void SetUp(const ::benchmark::State& state) override {
14.     std::call_once(flag, [this, &state]() {
15.       for (int i = 0; i < kSetSize; i++) {
16.         s.insert(i);
17.       }
18.     });
19.   }
21.   const std::unordered_set<int>& GetSet() { return s; }
23. private:
24.   std::unordered_set<int> s;
25.   std::once_flag flag;
26. };
28. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkAtomic)(benchmark::State& state) {
29.   std::atomic<uint32_t> sum = 0;
30.   for (auto _ : state) {
31.     int size_sum = kSetSize;
32.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
33.     int sum = 0;
34.     int start = state.thread_index() * size_per_thread;
35.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
36.     for (int i = start; i < end; i++) {
37.       const auto& inst = GetSet();
38.       if (inst.count(i) > 0) {
39.         benchmark::DoNotOptimize(sum++);
40.       }
41.     }
42.   }
43. }
45. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkLock)(benchmark::State& state) {
46.   std::mutex m;
47.   uint32_t sum = 0;
48.   for (auto _ : state) {
49.     int size_sum = kSetSize;
50.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
51.     int sum = 0;
52.     int start = state.thread_index() * size_per_thread;
53.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
54.     for (int i = start; i < end; i++) {
55.       const auto& inst = GetSet();
56.       if (inst.count(i) > 0) {
57.         std::lock_guard<std::mutex> lg(m);
58.         sum++;
59.       }
60.     }
61.   }
62. }
64. // 注册基准测试，并指定线程数
65. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(1);
66. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(2);
67. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(4);
68. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(8);
70. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(1);
71. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(2);
72. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(4);
73. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(8);
75. BENCHMARK_MAIN();

复制代码

benchmark效果：

case 2如下:

1. #include <benchmark/benchmark.h>
2. #include <atomic>
3. #include <iostream>
4. #include <memory>
5. #include <mutex>
6. #include <unordered_set>
7. #include "spin_lock.h"
9. constexpr int kSetSize = 10000;
11. class MyBenchmark : public benchmark::Fixture {
12. public:
13.   void SetUp(const ::benchmark::State& state) override {
14.     std::call_once(flag, [this, &state]() {
15.       for (int i = 0; i < kSetSize; i++) {
16.         s.insert(i);
17.       }
18.       count_ = 0;
19.       atomic_count_ = 0;
20.     });
21.   }
23.   const std::unordered_set<int>& GetSet() { return s; }
24.   void AtomicAdd() { atomic_count_++; }
25.   void MutexLockAndAdd() {
26.     std::lock_guard<std::mutex> lg(m);
27.     count_++;
28.   }
30. private:
31.   std::unordered_set<int> s;
32.   std::once_flag flag;
33.   uint32_t count_;
34.   std::atomic<uint32_t> atomic_count_;
35.   std::mutex m;
36. };
38. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkAtomic)(benchmark::State& state) {
39.   for (auto _ : state) {
40.     int size_sum = kSetSize;
41.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
42.     int sum = 0;
43.     int start = state.thread_index() * size_per_thread;
44.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
45.     for (int i = start; i < end; i++) {
46.       const auto& inst = GetSet();
47.       if (inst.count(i) > 0) {
48.         AtomicAdd();
49.       }
50.     }
51.   }
52. }
54. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkLock)(benchmark::State& state) {
55.   for (auto _ : state) {
56.     int size_sum = kSetSize;
57.     int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
58.     int sum = 0;
59.     int start = state.thread_index() * size_per_thread;
60.     int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
61.     for (int i = start; i < end; i++) {
62.       const auto& inst = GetSet();
63.       if (inst.count(i) > 0) {
64.         MutexLockAndAdd();
65.       }
66.     }
67.   }
68. }
70. // 注册基准测试，并指定线程数
71. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(1);
72. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(2);
73. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(4);
74. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(8);
76. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(1);
77. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(2);
78. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(4);
79. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(8);
81. BENCHMARK_MAIN();

复制代码

benchmark效果：

接下来结合这两个长处来看，链式数据布局的场景非常适合使用 atomic 变量。
 1.内存占用少：即使每个节点都实现一个自旋锁（SpinLock），也不会浪费太多内存。

 2.链式数据布局的临界区通常可以优化成一个指针的 CAS 操作。

Epoch Based Reclamation

固然云云，但要写一个高性能的并发安全的链式数据布局是非常困难的，这重要是由于写操作包罗了删除操作。举个最简朴的例子：
假设有一个链表 A->B->C，一个线程正在读B节点，另一个线程正在删除B节点，怎样保证读线程在读B节点期间不会被另一个线程给删掉？
再举个更复杂的例子：
假设有一个链表 A->B->C。一个线程正在读取 B 节点，另一个线程正在修改 B 节点。显然，最简朴的实现是锁住 B，同时只允许一个操作，但显然这样从各方面来看性能都不是最佳的，这是第一个方法。
第二个方法是类似于 Copy On Write（COW）。写操作时先重新构造一个节点 B1，再修改对应的数据，末了通过 CAS 操作修改指针毗连 A->B1。
我们来分析一下为什么第二个方法远比第一个方法要好。
首先，上锁会触发原子写，意味着即便是你只是为了读数据，也会触发一次 Cache Line 一致性同步的题目。而且在找到 B 节点之前的每一个节点都要依次上锁来保证读取的精确性，这意味着极大概率会发生 Cache Ping-Pong 题目。
再来看写操作，写操作除了上锁以外还需要修改节点的数据。第二个方法需要先构造一个新的节点再修改，意味着这个节点在插入链表之前肯定不在其他线程的 Cache 里（清除刚好有某个变量和这个新节点的内存在同一个 Cache Line 的情况）。而第一个方法修改的节点已经在链表里，这表示在之前肯定有线程已经访问过这个节点，那么它很可能在 Cache 里面，从而触发一次 Cache Line 一致性同步的题目。
然而事情没有这么简朴。试想一下，在修改完指针 A->B1 后，B 节点需要被抛弃释放，这时候其他线程有可能正在访问 B 节点而导致瓦解。
可以看出这些题目都是由于删除操作引起的，这个题目有几个闻名的办理方案，比如 Epoch Based Reclamation 和 Hazard Pointer 等。这里只介绍其中的 Epoch Based Reclamation，感爱好的话请自行搜刮了解其他实现方式。
该算法的思路是删除操作会实验触发版本 +1，但只有当所有线程都是最新版本 e 时才华乐成，乐成后会接纳 e-1 版本的内存。因此，最多会累积 3 个版本未释放节点的内存。是个以空间换时间，轻读重写的方案。
首先，每个线程维护自己的线程变量：
 1.active：标志该线程是否正在读数据。

 2.epoch：标志该线程当前的版本。

全局维护变量：
 1.global_epoch：全局最新的版本。

 2.retire_list：等待释放的节点。

读操作：
 1.首先把线程 active 标志为 true，表示正在读数据。

 2.然后把 global_epoch 赋值给 epoch，记录当前正在读的版本。

 3.如果线程需要删除节点，则把节点放到全局的 retire_list 末尾。

 4.竣事读后，将 active 标志为 false。

写操作：
 1.如果要删除节点，则把节点放到全局的 retire_list 末尾，并且实验增长版本。

 2.增长版本时检查所有线程的状态，当所有线程满意 epoch 即是当前版本 e 大概 active 为 false 时，进行版本 e = e + 1 操作。

 3.清空 e-2 版本的 retire_list。

这里给出一个简朴的实现，代码如下：

1. #pragma once
3. #include <array>
4. #include <atomic>
5. #include <mutex>
6. #include <numeric>
7. #include <vector>
9. constexpr uint8_t kEpochSize = 3;
10. constexpr uint8_t kCacheLineSize = 64;
12. template <uint32_t kReadThreadNum>
13. class ThreadIDManager;
15. template <uint32_t kReadThreadNum>
16. struct ThreadID {
17.   ThreadID() { tid = ThreadIDManager<kReadThreadNum>::GetInstance().AcquireThreadID(); }
18.   ~ThreadID() { ThreadIDManager<kReadThreadNum>::GetInstance().ReleaseThreadID(tid); }
19.   uint32_t tid;
20. };
22. template <uint32_t kReadThreadNum>
23. class ThreadIDManager {
24. public:
25.   ThreadIDManager() : tid_list_(kReadThreadNum) { std::iota(tid_list_.begin(), tid_list_.end(), 1); }
26.   ThreadIDManager(const ThreadIDManager &) = delete;
27.   ThreadIDManager(ThreadIDManager &&) = delete;
28.   ThreadIDManager &operator=(const ThreadIDManager &) = delete;
29.   ~ThreadIDManager() = default;
31.   static ThreadIDManager &GetInstance() {
32.     static ThreadIDManager inst;
33.     return inst;
34.   }
36.   uint32_t AcquireThreadID() {
37.     std::lock_guard<std::mutex> lock(tid_list_mutex_);
38.     auto tid = tid_list_.back();
39.     tid_list_.pop_back();
40.     return tid;
41.   }
43.   void ReleaseThreadID(const uint32_t tid) {
44.     std::lock_guard<std::mutex> lock(tid_list_mutex_);
45.     tid_list_.emplace_back(tid);
46.   }
48. private:
49.   std::vector<uint32_t> tid_list_;
50.   std::mutex tid_list_mutex_;
51. };
53. struct TLS {
54.   TLS() : active(false), epoch(0) {}
55.   TLS(TLS &) = delete;
56.   TLS(TLS &&) = delete;
57.   void operator=(const TLS &) = delete;
58.   ~TLS() = default;
59.   std::atomic_flag active;
60.   std::atomic<uint8_t> epoch;
61. } __attribute__((aligned(kCacheLineSize)));
63. template <class RCObject, class DestroyClass, uint32_t kReadThreadNum>
64. class EbrManager {
65. public:
66.   EbrManager() : tls_list_(), global_epoch_(0), update_(false), write_cnt_(0) {
67.     for (int i = 0; i < kEpochSize; i++) {
68.       retire_list_[i].store(nullptr, std::memory_order_release);
69.     }
70.   }
71.   EbrManager(const EbrManager<RCObject, DestroyClass, kReadThreadNum> &) = delete;
72.   EbrManager(EbrManager<RCObject, DestroyClass, kReadThreadNum> &&) = delete;
73.   EbrManager &operator=(const EbrManager<RCObject, DestroyClass, kReadThreadNum> &) = delete;
74.   ~EbrManager() { ClearAllRetireList(); }
76.   void ClearAllRetireList() {
77.     for (int i = 0; i < kEpochSize; i++) {
78.       ClearRetireList(i);
79.     }
80.   }
82.   inline void StartRead() {
83.     auto &tls = GetTLS();
84.     tls.active.test_and_set(std::memory_order_release);
85.     tls.epoch.store(global_epoch_.load(std::memory_order_acquire), std::memory_order_release);
86.   }
88.   inline void EndRead() { GetTLS().active.clear(std::memory_order_release); }
90.   inline void FreeObject(RCObject *object) {
91.     auto epoch = global_epoch_.load(std::memory_order_acquire);
92.     auto *node = new RetireNode;
93.     node->obj = object;
94.     do {
95.       node->next = retire_list_[epoch].load(std::memory_order_acquire);
96.     } while (!retire_list_[epoch].compare_exchange_weak(node->next, node, std::memory_order_acq_rel));
98.     auto write_cnt = write_cnt_.fetch_add(1, std::memory_order_relaxed);
99.     if (write_cnt > kReadThreadNum) {
100.       if (!update_.test_and_set(std::memory_order_acq_rel)) {
101.         TryGC();
102.         update_.clear(std::memory_order_release);
103.       }
104.     }
105.   }
107. private:
108.   inline TLS &GetTLS() {
109.     thread_local ThreadID<kReadThreadNum> thread_id;
110.     return tls_list_[thread_id.tid];
111.   }
113.   inline void TryGC() {
114.     auto epoch = global_epoch_.load(std::memory_order_acquire);
115.     // TODO 优化记录上一次搜索到的位置
116.     for (int i = 0; i < tls_list_.size(); i++) {
117.       if (tls_list_[i].active.test(std::memory_order::memory_order_acquire) &&
118.           tls_list_[i].epoch.load(std::memory_order::memory_order_acquire) != epoch) {
119.         return;
120.       }
121.     }
122.     global_epoch_.store((epoch + 1) % kEpochSize, std::memory_order_release);
123.     ClearRetireList((epoch + 2) % kEpochSize);
124.     write_cnt_.store(0, std::memory_order_relaxed);
125.   }
127.   inline void ClearRetireList(int index) {
128.     auto *retire_node = retire_list_[index].load(std::memory_order_acquire);
129.     while (retire_node != nullptr) {
130.       DestroyClass destroy(retire_node->obj);
131.       auto *old_node = retire_node;
132.       retire_node = retire_node->next;
133.       delete old_node;
134.     }
135.     retire_list_[index].store(nullptr, std::memory_order_release);
136.   }
138.   struct RetireNode {
139.     RCObject *obj;
140.     RetireNode *next;
141.   };
142.   std::array<char, kCacheLineSize> start_padding_;
143.   std::array<TLS, kReadThreadNum> tls_list_;
144.   std::atomic<uint8_t> global_epoch_;
145.   std::array<char, kCacheLineSize> mid_padding_;
146.   std::atomic_flag update_;
147.   std::atomic<uint32_t> write_cnt_;
148.   std::atomic<RetireNode *> retire_list_[kEpochSize];
149.   std::array<char, kCacheLineSize> end_padding_;
150. };

复制代码

这里再给出一个benchmark，对比一下使用 Epoch Based Reclamation（EBR）和不使用 EBR 的区别。由于笔者时间有限，只能写一个非常简朴的版本，仅供参考。

1. #include <benchmark/benchmark.h>
2. #include <mutex>
3. #include "ebr.h"
4. #include "spin_lock.h"
6. struct Node {
7.   Node() : lock(), next(nullptr) {}
8.   int key;
9.   int value;
10.   Node *next;
11.   SpinLock lock;
12. };
14. class NodeFree {
15. public:
16.   NodeFree(Node *node) { delete node; }
17. };
19. /*
20. * 快速测试起见，简单写了个list版本的kv结构，里面只会有3个元素，然后只支持Get和Modify，Modify也必定会命中key。
21. * 不是直接把key，value，next设置成atomic变量而是使用SpinLock的原因是模拟复杂情况，真实情况下会存在Add和Remove操作，实现没有如此简单。
22. */
23. class MyList {
24. public:
25.   MyList() {
26.     Node *pre_node = nullptr;
27.     auto *&cur_node = root_;
28.     // 这里虽然插入了10个元素，但后面的实现会假设第一个key 9作为header是绝对不会被修改或者读到的。
29.     for (int i = 0; i < 10; i++) {
30.       cur_node = new Node;
31.       cur_node->key = i;
32.       cur_node->value = i;
33.       cur_node->next = pre_node;
34.       pre_node = cur_node;
35.     }
36.   }
38.   int Get(int key, int *value) {
39.     root_->lock.lock();
40.     auto *cur_node = root_->next;
41.     auto *pre_node = root_;
42.     while (cur_node != nullptr) {
43.       cur_node->lock.lock();
44.       pre_node->lock.unlock();
45.       if (key == cur_node->key) {
46.         *value = cur_node->value;
47.         cur_node->lock.unlock();
48.         return 0;
49.       }
50.       pre_node = cur_node;
51.       cur_node = cur_node->next;
52.     }
53.     pre_node->lock.unlock();
54.     return 1;
55.   }
56.   int Modify(int key, int value) {
57.     root_->lock.lock();
58.     auto *cur_node = root_->next;
59.     auto *pre_node = root_;
60.     while (cur_node != nullptr) {
61.       cur_node->lock.lock();
62.       pre_node->lock.unlock();
63.       if (key == cur_node->key) {
64.         cur_node->value = value;
65.         cur_node->lock.unlock();
66.         return 0;
67.       }
68.       pre_node = cur_node;
69.       cur_node = cur_node->next;
70.     }
71.     pre_node->lock.unlock();
72.     return 1;
73.   }
75.   int GetUseEbr(int key, int *value) {
76.     ebr_mgr_.StartRead();
77.     auto *cur_node = root_->next;
78.     while (cur_node != nullptr) {
79.       if (key == cur_node->key) {
80.         *value = cur_node->value;
81.         ebr_mgr_.EndRead();
82.         return 0;
83.       }
84.       cur_node = cur_node->next;
85.     }
86.     ebr_mgr_.EndRead();
87.     return 1;
88.   }
90.   int ModifyUseEbr(int key, int value) {
91.     root_->lock.lock();
92.     auto *cur_node = root_->next;
93.     auto *pre_node = root_;
94.     while (cur_node != nullptr) {
95.       cur_node->lock.lock();
96.       if (key == cur_node->key) {
97.         auto *new_node = new Node;
98.         new_node->key = cur_node->key;
99.         new_node->value = value;
100.         new_node->next = cur_node->next;
101.         pre_node->next = new_node;
102.         cur_node->lock.unlock();
103.         pre_node->lock.unlock();
104.         ebr_mgr_.FreeObject(cur_node);
105.         return 0;
106.       }
107.       auto *next_node = cur_node->next;
108.       pre_node->lock.unlock();
109.       pre_node = cur_node;
110.       cur_node = next_node;
111.     }
112.     pre_node->lock.unlock();
113.     return 1;
114.   }
116. private:
117.   Node *root_;
118.   EbrManager<Node, NodeFree, 15> ebr_mgr_;
119. };
121. class MyBenchmark : public benchmark::Fixture {
122. public:
123.   void SetUp(const ::benchmark::State &state) override {}
125.   MyList &GetMyList() { return l; }
127. private:
128.   MyList l;
129.   std::once_flag flag;
130. };
132. constexpr int kKeySize = 10000;
134. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkNoUseEbr)(benchmark::State &state) {
135.   for (auto _ : state) {
136.     auto &mylist = GetMyList();
137.     if (0 == state.thread_index()) {
138.       // modify
139.       for (int i = 0; i < kKeySize; i++) {
140.         mylist.Modify(i % 9, i);
141.       }
142.     } else {
143.       // get
144.       for (int i = 0; i < kKeySize; i++) {
145.         int value;
146.         mylist.Get(i % 9, &value);
147.       }
148.     }
149.   }
150. }
152. BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkUseEbr)(benchmark::State &state) {
153.   for (auto _ : state) {
154.     auto &mylist = GetMyList();
155.     if (0 == state.thread_index()) {
156.       // modify
157.       for (int i = 0; i < kKeySize; i++) {
158.         mylist.ModifyUseEbr(i % 9, i);
159.       }
160.     } else {
161.       // get
162.       for (int i = 0; i < kKeySize; i++) {
163.         int value;
164.         mylist.GetUseEbr(i % 9, &value);
165.       }
166.     }
167.   }
168. }
170. // 注册基准测试，并指定线程数
171. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkNoUseEbr)->Threads(4);
172. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkNoUseEbr)->Threads(8);
173. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkNoUseEbr)->Threads(12);
175. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkUseEbr)->Threads(4);
176. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkUseEbr)->Threads(8);
177. BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkUseEbr)->Threads(12);
179. BENCHMARK_MAIN();

复制代码

benchmark效果：

结语

以上是笔者的一些个人见解。由于水平和时间有限，测试用例尽可能地简化，可能不够全面。如果内容中有任何错误，欢迎指出。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。