You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

MSVC下atomic<pair<uintptr_t,uintptr_t>>无内核依赖DWCAS替代实现问询

问题:MSVC对大尺寸atomic类型的compare_exchange实现逻辑识别

我想测试编译器能否识别如下类型:

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针对超过硬件默认支持的原子操作尺寸的类型的用户态自旋锁实现,并非软件事务内存,具体逻辑拆解如下:

  1. 内存布局解释:你看到的24字节atomic<pair<uintptr_t, uintptr_t>>,前4字节(对齐后占8字节)就是嵌入式自旋锁的状态位,后面16字节存储你定义的uip_pair数据,所以总大小是8+16=24字节,和你观察到的规律一致:不管内部存储的结构多大,std::atomic都会额外多占8字节存自旋锁状态。
  2. 加锁逻辑:开头的xchg DWORD PTR [rcx], eax是原子抢锁操作,xchg自带lock语义,抢到锁(锁原值为0)就跳转到label8执行实际的比较交换逻辑,抢不到就进入自旋流程:自旋过程会用pause指令降低功耗,并且自旋等待次数会指数退避(上限64次),避免频繁抢锁导致的缓存一致性流量爆炸。
  3. 比较交换逻辑:抢到锁后会先比对atomic中存储的两个uintptr_t值和传入的cmp值是否完全一致:
    • 一致就把xchg的值写入atomic的数据区,返回true
    • 不一致就把当前atomic的数据区值写回cmp参数,返回false
  4. 解锁逻辑:无论比较交换是否成功,最后都会用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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.03 04:48:02