PERSONAL LAB / cpp/06-并发编程
关于项目博客知识库工具箱
原子变量与内存序.mdatomic 与 mutex 动手实验.mdC++：future、promise、packaged_task 与 async.md
C++ 并发编程

atomic 与 mutex 动手实验

难度：⭐⭐ | 高频指数：🔥🔥🔥

这篇是跑出来的笔记，不是背出来的。理论在 互斥锁与条件变量.md 和 原子变量与内存序.md，这里只做一件事：把「什么时候该用 atomic、什么时候 必须用 mutex」这条边界用五个能编译能运行的程序钉死。忘了就 g++ 一遍。

一句话结论

atomic 让「一次操作」不可分割。mutex 让「一段代码」不可分割。

判断方法只记一条：

看要保护的动作里，有没有「先把值读出来，再根据读到的值决定写什么」。

  • 没有 → 写一个值、读一个值、++ → atomic
  • 有 → 读完要判断，判断完要写 → mutex（或 atomic + CAS）

实验 1：count++ 不是一次操作

// d1.cpp
#include <iostream>
#include <thread>

int count = 0;

void work() {
    for (int i = 0; i < 100000; ++i) {
        count++;              // load -> add -> store，三步可被打断
    }
}

int main() {
    std::thread a(work), b(work);
    a.join(); b.join();
    std::cout << "期望 200000, 实际 " << count << "\n";
}
g++ -std=c++17 -O0 -o d1 d1.cpp && ./d1

实测输出（每次都不一样）：

期望 200000, 实际 101545
期望 200000, 实际 106222

一半的加法丢了。

⚠️ 必须用 -O0。开了 -O2 优化器会把整个循环折叠成一次加法，问题就看不见了 —— 这本身是个提醒：data race 是未定义行为，编译器有权假设它不存在。

丢的原因，count++ 在 CPU 眼里是三步：

1. 从内存读 count 到寄存器      (load)
2. 寄存器里 +1                  (add)
3. 写回内存                     (store)

两个线程一交错：

线程 A: 读到 count = 5
线程 B: 读到 count = 5      ← B 也读到 5
线程 A: 算出 6，写回        count = 6
线程 B: 算出 6，写回        count = 6   ← A 的那次加法没了

加了两次，只涨了 1。

「原子」的意思就是：这三步不能被打断。 要么没开始，要么全做完，别的线程看不到中间状态。

实验 2：atomic 和 mutex 都能修好「单个计数」

// d2.cpp
#include <iostream>
#include <thread>
#include <atomic>
#include <mutex>

std::atomic<int> atomicCount{0};
int mutexCount = 0;
std::mutex mtx;

void workAtomic() { for (int i = 0; i < 100000; ++i) atomicCount++; }
void workMutex()  { for (int i = 0; i < 100000; ++i) { std::lock_guard<std::mutex> lk(mtx); mutexCount++; } }

int main() {
    { std::thread a(workAtomic), b(workAtomic); a.join(); b.join(); }
    { std::thread a(workMutex),  b(workMutex);  a.join(); b.join(); }
    std::cout << "atomic: " << atomicCount << "\n";
    std::cout << "mutex : " << mutexCount  << "\n";
}
atomic: 200000
mutex : 200000

两个都对。这一步说明的是：只保护一个计数器时，atomic 和 mutex 等价，选 atomic 只是因为它更快。

这里容易得出错误结论——「那以后都用 atomic 好了」。下一个实验专门打这个结论。

实验 3：atomic 修不了 check-then-act（对应项目里的 double-free）

这段对应 CrossShellNext SshSession.cpp 的断开逻辑。假设把 status 直接换成 std::atomic<SessionStatus>：

// d3.cpp
#include <iostream>
#include <thread>
#include <atomic>
#include <chrono>

enum class SessionStatus { CONNECTED, DISCONNECTED };

std::atomic<SessionStatus> status{SessionStatus::CONNECTED};
std::atomic<int> releaseCount{0};

void releaseResource() { releaseCount++; }   // 真实代码里是 libssh2_session_free

void disconnect() {
    if (status.load() == SessionStatus::CONNECTED) {                // ① 读 + 判断
        std::this_thread::sleep_for(std::chrono::microseconds(1));  // ← 把缝放大
        status.store(SessionStatus::DISCONNECTED);                  // ② 写
        releaseResource();                                          // ③ 释放
    }
}

int main() {
    std::thread a(disconnect), b(disconnect);
    a.join(); b.join();
    std::cout << "资源释放次数（应为 1）: " << releaseCount << "\n";
}
资源释放次数（应为 1）: 2
资源释放次数（应为 1）: 2
资源释放次数（应为 1）: 2

status 明明是 atomic，资源还是释放了 2 次 —— 这就是 double-free。

线程 A: 读到 status == CONNECTED  ✓ 进入 if
线程 B: 读到 status == CONNECTED  ✓ 也进入 if     ← A 还没来得及改状态
线程 A: 释放资源                                    ← 第 1 次
线程 B: 释放资源                                    ← 第 2 次，崩溃

load() 是原子的，store() 也是原子的，但这两步之间是有缝的，别的线程能挤进来。

这就是 check-then-act（也叫 TOCTOU）。那句 sleep_for 只是把缝放大到必现， 去掉它问题依然存在，只是变成概率事件 —— 而概率事件在生产环境里就是「偶发崩溃」。

实验 4：两种正确修法

// d4.cpp
#include <iostream>
#include <thread>
#include <atomic>
#include <mutex>
#include <chrono>

enum class SessionStatus { CONNECTED, DISCONNECTED };

// 修法 A: mutex 罩住整段（项目里用的这个）
SessionStatus statusA = SessionStatus::CONNECTED;
std::mutex statusMutex;
std::atomic<int> releaseA{0};

void disconnectA() {
    std::lock_guard<std::mutex> lk(statusMutex);
    if (statusA == SessionStatus::CONNECTED) {
        std::this_thread::sleep_for(std::chrono::microseconds(1));
        statusA = SessionStatus::DISCONNECTED;
        releaseA++;
    }
}

// 修法 B: CAS 把「比较 + 写」合成一个原子动作
std::atomic<SessionStatus> statusB{SessionStatus::CONNECTED};
std::atomic<int> releaseB{0};

void disconnectB() {
    SessionStatus expected = SessionStatus::CONNECTED;
    // 「如果现在还是 CONNECTED，就改成 DISCONNECTED」——整体不可分割
    if (statusB.compare_exchange_strong(expected, SessionStatus::DISCONNECTED)) {
        std::this_thread::sleep_for(std::chrono::microseconds(1));
        releaseB++;   // 只有抢到的那个线程会进来
    }
}

int main() {
    { std::thread a(disconnectA), b(disconnectA); a.join(); b.join(); }
    { std::thread a(disconnectB), b(disconnectB); a.join(); b.join(); }
    std::cout << "mutex 修法，释放次数: " << releaseA << "\n";
    std::cout << "CAS   修法，释放次数: " << releaseB << "\n";
}
mutex 修法，释放次数: 1
CAS   修法，释放次数: 1

对比表：

做法能不能用为什么
光用 atomic<Status>❌load 和 store 之间有缝，double-free
mutex 罩住整段✅「一段代码」不可分割
atomic + CAS✅把「比较 + 写」压成一个动作

还有一层 mutex 独有的能力：atomic 只能保护它自己。两个 atomic 拼在一起不等于一个原子操作：

std::atomic<int> a, b;
a.store(1);
b.store(2);   // 别的线程可能看到 a=1 但 b 还是旧值

要让「改 a、改 b、再改 c」对外表现成一步，只能 mutex。

实验 5：内存序在汇编里长什么样

前四个实验都能稳定复现，内存序不行 —— 在 x86 和 Apple Silicon 上， relaxed 导致的重排要靠特定的硬件时序才会露头，写个循环跑几百万轮通常也是 0 次。 （实测在 arm64 上跑 100 万轮 relaxed 的 publish/subscribe，坏 0 次。）

所以换个能看见的角度：直接看编译器生成了什么指令。

// d6.cpp
#include <atomic>
std::atomic<bool> ready;
int data;
void publish_relaxed() { data = 42; ready.store(true, std::memory_order_relaxed); }
void publish_release() { data = 42; ready.store(true, std::memory_order_release); }
void publish_seqcst()  { data = 42; ready.store(true); }   // 默认 seq_cst
g++ -std=c++17 -O2 -S -o d6.s d6.cpp

arm64 上的关键差别：

publish_relaxed:
    str   w8, [x9, _data@PAGEOFF]     ; data = 42
    strb  w8, [x9, _ready@PAGEOFF]    ; 普通 store，前后可被重排

publish_release:
    str   w8, [x9, _data@PAGEOFF]     ; data = 42
    stlrb w9, [x8]                    ; store-release，挡住之前的写往后跑

publish_seqcst:                        ; 和 release 生成了同样的 stlrb
    str   w8, [x9, _data@PAGEOFF]
    stlrb w9, [x8]

strb → stlrb 一个字母的差别，就是「保证 data = 42 先于 ready = true 对别的线程可见」。 用 relaxed 的话，线程 B 可能看到 ready == true 但 data 还是旧值。

注：这台机器上 seq_cst 的 store 和 release 生成了相同指令，差别体现在 load 侧和 跨变量的全局顺序上，不同架构/编译器结果不同，别把这条当通则。

面试安全答法：「默认用 seq_cst，性能确实成瓶颈时才降到 acquire/release 或 relaxed， 而且要能说清为什么降了还正确。」

回到项目：CrossShellNext 里的划分

成员用什么为什么
SessionStatus statusstd::mutex statusMutex状态机转换，要「读 → 判断 → 写」三步连住
stopThread / shouldStopstd::atomic<bool>纯标志位，只有单次写和单次读
isCallbackValid / m_pfConnectedstd::atomic<bool>同上

这个划分是标准做法。 纯标志位配 atomic、状态机转换配 mutex，两边都选对了。

简历上曾写成「使用 atomic 管理会话状态、mutex 协调连接与断开操作」， 和代码里的实际分工正好写反了。面试被问到就按上表如实讲 —— 能说清两者边界在哪，比笼统说「用了 atomic」更值钱。

两个常见陷阱

1.「atomic 是不是就是自动加锁？」

不是。atomic 通常编译成单条 CPU 指令（x86 上带 lock 前缀，arm64 上是 ldxr/stxr 或 stlrb 这类），无锁。但类型太大时（比如 32 字节的结构体）标准库可能内部退化成加锁实现， is_lock_free() 可以查。

2.「atomic 更快，所以尽量用 atomic」

不对。mutex 在无竞争时也很快（现代实现走 futex，先在用户态自旋，不进内核）。 而且 atomic 用错是正确性问题，mutex 用错顶多是性能问题。正确性优先。

复习清单

跑一遍这五个程序，能答上这几问就算过：

  1. count++ 为什么不是原子的？拆成哪三步？
  2. 实验 1 为什么必须 -O0？
  3. status 换成 atomic<SessionStatus> 为什么还会 double-free？
  4. CAS 解决了什么？compare_exchange_strong 的语义是什么？
  5. 两个 atomic 变量连续写，能不能保证别的线程看到一致的中间状态？
  6. relaxed 和 release 在 arm64 上差哪条指令？
Related · 并发编程
互斥锁与条件变量死锁与锁顺序无锁编程基础线程池设计std::thread 生命周期、join() 与 detach()
⎇ main cpp/06-并发编程 9 节 253 notes UTF-8