MSVC下atomic<pair<uintptr_t,uintptr_t>>无内核依赖DWCAS替代实现问询
我想测试编译器能否识别如下类型:
atomic<pair<uintptr_t, uintptr_t>
并对其使用DWCAS指令(比如x86-64的lock cmpxchg16b),还是会给该pair附加常规锁实现原子操作。
首先我编写了一个最小程序,仅包含一个noinline函数来对原子pair执行比较交换操作。编译器生成了大量我无法理解的代码,且其中没有出现任何带LOCK前缀的指令。我好奇该实现是否在atomic内部嵌入了锁,于是打印了上述原子pair的大小:64位平台下为24字节,显然没有内置锁。
之后我编写了测试程序,让系统所有线程(锐龙Threadripper 64核,Win10,关闭SMT)对同一个原子pair的两个成员各递增预设次数,计算每次成功递增的耗时,单位为纳秒。测试得到的耗时很高,每次成功递增约20000ns,起初我以为是存在我没注意到的锁,但24字节的atomic大小否定了这个猜测。同时我在任务管理器中看到64核几乎全程跑满用户态CPU,说明不存在内核介入。
有没有相关从业者能从给出的汇编转储中识别出这个DWCAS替代方案的实现逻辑?
我的测试代码如下:
#include <iostream> #include <atomic> #include <utility> #include <thread> #include <mutex> #include <condition_variable> #include <chrono> #include <vector> using namespace std; using namespace chrono; struct uip_pair { uip_pair() = default; uip_pair( uintptr_t first, uintptr_t second ) : first( first ), second( second ) { } uintptr_t first, second; }; using atomic_pair = atomic<uip_pair>; int main() { cout << "sizeof(atomic<pair<uintptr_t, uintptr_t>>): " << sizeof(atomic_pair) << endl; atomic_pair ap( uip_pair( 0, 0 ) ); cout << "atomic<pair<uintptr_t, uintptr_t>>::is_lock_free: " << ap.is_lock_free() << endl; mutex mtx; unsigned nThreads = thread::hardware_concurrency(); unsigned ready = nThreads; condition_variable cvReady; bool run = false; condition_variable cvRun; atomic_int64_t sumDur = 0; auto theThread = [&]( size_t n ) { unique_lock<mutex> lock( mtx ); if( !--ready ) cvReady.notify_one(); cvRun.wait( lock, [&]() -> bool { return run; } ); lock.unlock(); auto start = high_resolution_clock::now(); uip_pair cmp = ap.load( memory_order_relaxed ); for( ; n--; ) while( !ap.compare_exchange_weak( cmp, uip_pair( cmp.first + 1, cmp.second + 1 ), memory_order_relaxed, memory_order_relaxed ) ); sumDur.fetch_add( duration_cast<nanoseconds>( high_resolution_clock::now() - start ).count(), memory_order_relaxed ); lock.lock(); }; vector<jthread> threads; threads.reserve( nThreads ); static size_t const ROUNDS = 100'000; for( unsigned t = nThreads; t--; ) threads.emplace_back( theThread, ROUNDS ); unique_lock<mutex> lock( mtx ); cvReady.wait( lock, [&]() -> bool { return !ready; } ); run = true; cvRun.notify_all(); lock.unlock(); for( jthread &thr : threads ) thr.join(); cout << (double)sumDur / ((double)nThreads * ROUNDS) << endl; uip_pair p = ap.load( memory_order_relaxed ); cout << "synch: " << (p.first == p.second ? "yes" : "no") << endl; }
[补充]:我将compare_exchange_weak函数提取为noinline函数并反汇编得到如下代码:
struct uip_pair { uip_pair() = default; uip_pair( uintptr_t first, uintptr_t second ) : first( first ), second( second ) { } uintptr_t first, second; }; using atomic_pair = atomic<uip_pair>; #if defined(_MSC_VER) #define NOINLINE __declspec(noinline) #elif defined(__GNUC__) || defined(__clang__) #define NOINLINE __attribute((noinline)) #endif NOINLINE bool cmpXchg( atomic_pair &ap, uip_pair &cmp, uip_pair xchg ) { return ap.compare_exchange_weak( cmp, xchg, memory_order_relaxed, memory_order_relaxed ); }
对应的汇编代码:
mov eax, 1 mov r10, rcx mov r9d, eax xchg DWORD PTR [rcx], eax test eax, eax je SHORT label8 label1: mov eax, DWORD PTR [rcx] test eax, eax je SHORT label7 label2: mov eax, r9d test r9d, r9d je SHORT label5 label4: pause sub eax, 1 jne SHORT label4 cmp r9d, 64 jl SHORT label5 lea r9d, QWORD PTR [rax+64] jmp SHORT label6 label5: add r9d, r9d label6: mov eax, DWORD PTR [rcx] test eax, eax jne SHORT label2 label7: mov eax, 1 xchg DWORD PTR [rcx], eax test eax, eax jne SHORT label1 label8: mov rax, QWORD PTR [rcx+8] sub rax, QWORD PTR [rdx] jne SHORT label9 mov rax, QWORD PTR [rcx+16] sub rax, QWORD PTR [rdx+8] label9: test rax, rax sete al test al, al je SHORT label10 movups xmm0, XMMWORD PTR [r8] movups XMMWORD PTR [rcx+8], xmm0 xor ecx, ecx xchg DWORD PTR [r10], ecx ret label10: movups xmm0, XMMWORD PTR [rcx+8] xor ecx, ecx movups XMMWORD PTR [rdx], xmm0 xchg DWORD PTR [r10], ecx ret
希望有人能理解这段反汇编逻辑。注意x86架构下XCHG指令隐式带有LOCK语义。在我看来MSVC在此处使用了某种软件事务内存实现:我可以任意扩展atomic中嵌入的共享结构,atomic大小始终比原结构大8字节,说明MSVC总是使用某种STM实现这类原子操作。
这段汇编对应的是MSVC标准库中std::atomic针对超过硬件默认支持的原子操作尺寸的类型的用户态自旋锁实现,并非软件事务内存,具体逻辑拆解如下:
- 内存布局解释:你看到的24字节
atomic<pair<uintptr_t, uintptr_t>>,前4字节(对齐后占8字节)就是嵌入式自旋锁的状态位,后面16字节存储你定义的uip_pair数据,所以总大小是8+16=24字节,和你观察到的规律一致:不管内部存储的结构多大,std::atomic都会额外多占8字节存自旋锁状态。 - 加锁逻辑:开头的
xchg DWORD PTR [rcx], eax是原子抢锁操作,xchg自带lock语义,抢到锁(锁原值为0)就跳转到label8执行实际的比较交换逻辑,抢不到就进入自旋流程:自旋过程会用pause指令降低功耗,并且自旋等待次数会指数退避(上限64次),避免频繁抢锁导致的缓存一致性流量爆炸。 - 比较交换逻辑:抢到锁后会先比对
atomic中存储的两个uintptr_t值和传入的cmp值是否完全一致:- 一致就把
xchg的值写入atomic的数据区,返回true - 不一致就把当前
atomic的数据区值写回cmp参数,返回false
- 一致就把
- 解锁逻辑:无论比较交换是否成功,最后都会用
xchg DWORD PTR [r10], ecx(ecx此时为0)原子释放锁。
为什么不用lock cmpxchg16b?
MSVC默认生成的x86-64代码的基线目标是不支持cmpxchg16b的初代x86-64处理器,你可以通过添加/arch:AVX或更高的编译选项,让MSVC启用硬件DWCAS指令,此时std::atomic<uip_pair>会变成16字节大小,is_lock_free()返回true,性能会提升一到两个数量级,远低于你现在测试的20000ns。
高耗时原因
你现在的实现本质上是所有线程抢同一个自旋锁,64核竞争下自旋等待的开销被急剧放大,所以单操作耗时很高,且所有核心都在用户态自旋等待,没有内核介入,符合你观察到的现象。
内容的提问来源于stack exchange,提问作者Bonita Montero

