cmpxchg16b指令的实现分析

我们知道C++11以后提供了原子操作类型：atomic<T>，该原子模板类可以实现原子操作，比如exchange、compare_exchange_weak、fecth_add等，T类型一般都是非常简单的类型，比如整型、指针等类型，这些类型都可以实现指令级的原子操作，所操作的数据类型的长度一般也不能大于内存数据总线的宽度，比如在32/64bit的处理器中，T类型的数据长度不能超过32/64bit。如果数据类型的长度大于内存数据总线宽度，当读写内存访问这些数据时，就无法使用一条指令对它们进行操作，需要使用多条指令来访问内存，因此在对它们进行操作时，为了保证原子性，只能使用外部的锁加以保护，比如mutex或者自旋锁，显然这种实现机制不是lock-free，既然涉及到锁，进行原子操作时开销比较大。

不过在x86处理器中，因为它的指令集是CISC，可以用一条指令做复杂的操作，它有一条特殊的指令，可以实现对2倍内存数据总线宽度的数据的原子操作，即32/64位的处理器，可以用一条指令来实现64/128位数据的原子操作，下面就以64位的x86处理器来讨论一下。

在X86-64的处理器中，提供了一个可对128bit数据（也就是16字节）进行CAS算法操作的指令：cmpxchg16b（在x86-32处理器中对应的指令是cmpxchg8b）。该指令可以把内存中16字节连续地址的数据进行CAS算法，它的形式是：cmpxchg16b m128，其中m128是一段16字节的连续内存存储位置（即指针）。该指令把rdx:rax寄存器中存放的128位值与内存地址m128中存放的128位值进行比较：如果值相等，则设置零标志（ZF），并将寄存器rcx:rbx存放的值复制到内存位置m128处，否则，ZF标志将被清除，并将内存位置m128处的数据复制到寄存器rdx:rax，可以通过ZF标志来判断操作是否成功。

既然X86处理器有指令可以进行128bit数据的CAS算法，x86环境的C++编译器在编译128bit数据的原子操作函数时，是不是都使用cmpxchg16b指令来实现呢？经过实际分析，发现不同编译器有不同的方案，所产生的汇编代码并不一定使用了cmpxchg16b指令。下面简单地通过atomic<__int128_t>的store()的原子操作来分析一下实现机制。

atomic<__int128_t> am16;
void bar(void)
{am16.store(42);
}

上面代码片段定义了一个类型为__int128_t，即16字节的atomic对象变量am16，bar()函数调用了am16对象的store()成员函数，分析一下不同编译器生成的汇编代码：
1、编译器GCC，优化选项-O3，所产生的汇编代码如下，没有使用cmpxchg16b指令，而是调用了一个函数：__atomic_store_16：

bar():mov     ecx, 5mov     esi, 42xor     edx, edxmov     edi, OFFSET FLAT:am16jmp     __atomic_store_16

如果运行am16.is_lock_free()，结果返回false，可见GCC所实现的atomic<__int128_t>的原子操作不是lock-free算法。

2、同GCC，编译器CLANG，在-O3优化选项下，生成的汇编代码也没有使用cmpxchg16b指令，也是使用了函数：__atomic_store_16：

bar():push    raxlea     rdi, [rip + am16]mov     esi, 42xor     edx, edxmov     ecx, 5call    __atomic_store_16@PLTpop     raxret

如果运行am16.is_lock_free()，结果返回false，可见CLANG实现的atomic<__int128_t>的原子操作也不是lock-free算法。

3、ICC编译器，即Intel C++Compiler编译器，毕竟是Intel自己家的编译器，肯定要为自己家的处理器生成最优的汇编代码了，即使在使用较为低级的-O1优化选项时，它也会产生cmpxchg16b指令：

bar():
..B1.1:                         # Preds ..B1.0push      rbx                                           #549.1mov       esi, offset flat: am16                        #271.2mov       r8d, 42                                       #271.2xor       r9d, r9d                                      #271.2mov       rax, QWORD PTR [rsi]                          #271.2mov       rdx, QWORD PTR [8+rsi]                        #271.2
..L7:                                                           #mov       rbx, r8                                       #271.2mov       rcx, r9                                       #271.2lock      cmpxchg16b QWORD PTR [rsi]                              #271.2jnz       ..L7          # Prob 5%                       #271.2
..B1.2:                         # Preds ..B1.1pop       rbx                                           #551.1ret

如果运行am16.is_lock_free()，结果返回true，可见ICC为atomic<__int128_t>实现的原子操作是lock-free算法。

那么，如果要操作atomic<__int128_t>的原子操作，在gcc和clang编译环境下，是不是只能使用non-lock-free算法吗？

查看GCC的手册，发现它有一个与此相关的编译选项：-mcx16
This option enables GCC to generate CMPXCHG16B instructions in 64-bit code to implement compare-and-exchange operations on 16-byte aligned 128-bit objects. This is useful for atomic updates of data structures exceeding one machine word in size. （此选项使GCC能够为64位代码生成CMPXCHG16B指令，以在16字节对齐的128位对象上实现比较和交换操作。这对于大小超过一个机器字的数据结构的原子更新非常有用。）

好，那就在编译时加上-mcx16，再分析GCC和CLANG编译器所生成的汇编代码。
1、很遗憾，GCC编译时加上-mcx16，仍然没有生成cmpxchg16b指令代码：

bar():mov     ecx, 5mov     esi, 42xor     edx, edxmov     edi, OFFSET FLAT:am16jmp     __atomic_store_16

2、不过，clang编译时加上-mcx16，生成了cmpxchg16b指令代码：

bar():push    rbxmov     rdx, qword ptr [rip + am16+8]mov     rax, qword ptr [rip + am16]mov     ebx, 42
.LBB0_1:xor     ecx, ecxlock    cmpxchg16b xmmword ptr [rip + am16]jne     .LBB0_1pop     rbxret

如果运行am16.is_lock_free()，返回true，可见在-mcx16编译选项的加持下，CLANG编译器为128bit的原子变量使用了lock-free的原子操作算法。

同样，对于atomic<__int128_t>原子类型的其它原子操作，比如load、exchange、compare_exchange_weak等，各个编译器所生成汇编指令的方案与前面的store操作的方案相同。

由此可见，对于128bit数据的原子类型的原子操作，在ICC编译环境下支持lock-free算法，CLANG编译器只有在使用-mcx16编译选项时，才支持lock-free算法，而GCC编译器无论是否使用-mcx16选项，都不支持lock-free算法。因此，如果我们使用atomic<T>类为128bit的数据实现原子操作时，可能有的编译器并不支持lock-free算法，这样在执行相关原子操作时可能会带来锁的开销。

那么，难道在GCC环境下，要想为128bit的数据实现lock-free算法的原子操作，只能使用汇编语言来手动实现吗？不是的，在GCC编译环境下也能生成cmpxchg16b的指令，只不过不是使用atomic模板类，而是使用它的内置函数：bool __sync_bool_compare_and_swap(type *ptr, type oldval, type newval);
该函数是GCC提供的内置函数，用于实现CAS操作的原子性，它可以和编译选项-mcx16结合来实现128bit数据的CAS原子操作。它会比较指针ptr指向的值与给定的旧值oldval，如果相同，则将新值newval写入，并返回真（true）；如果不同，则不写入，并返回假（false）。
下面看一下使用它如何来实现__int128_t类型的store原子写操作：

__int128_t atomic_val;
void store() {__int128_t val = 42;__int128_t cmp;do {cmp = atomic_val;} while (!__sync_bool_compare_and_swap(&atomic_val, cmp, val));
}

下面是GCC使用-mcx16编译选项产生的汇编代码，此时编译器使用了cmpxchg16b指令，使用lock-free算法实现了原子赋值操作。

store():push    rbxxor     ecx, ecxmov     ebx, 42
.L20:mov     rax, QWORD PTR global[rip]mov     rdx, QWORD PTR global[rip+8]lock 	cmpxchg16b XMMWORD PTR global[rip]jne     .L20pop     rbxret

同样，编译器ICC以及CLANG在使用-mcx16编译选项时，也生成了类似的lock-free算法的原子赋值操作。

不过，需要说明的是，因为bool __sync_bool_compare_and_swap(__int128_t *ptr, __int128_t cmp, __int128_t val)是C函数，它的第二个参数cmp是值类型，当CAS运算失败时无法同时把指针ptr指向的原值atomic_val通过cmp参数返回。因此，在CAS失败后循环重试时，需要再次通过cmp = atomic_val;把原值赋值给cmp值。而指令cmpxchg16b在失败时，会把原子变量的原值通过rdx:rax返回，因此，失败后循环重试时第6、7行代码，其实寄存器rdx:rax中所存放的值和寄存器rdi所指向位置的128bit的值完全相同，没必要再为rdx:rax做一次赋值操作，而atomic的store操作就没有这样的冗余操作。可见__sync_bool_compare_and_swap()生成的store原子操作的汇编代码没有前面的am.store()生成的汇编代码高效。

此外，指令cmpxchg16b都使用了前缀lock，以保证CAS操作是原子的，在X86处理器环境中，lock指令前缀还能保证sequence-consistency的内存序，也就是说store()等同于atomic.store(42, memory_order_seq_cst)的操作。

最后，就用cmpxchg16b的CAS算法来实现一个128bit的加法原子操作，128bit的数据是由两个64bit的数据组成，假设它们是两个计数器，我们使用多线程同时分别为这两个进行累加操作。因为atomic<__int128_t>类型的原子类，不支持fetch_add等算术运算，而compared_exchange_weak等CAS函数在GCC编译器中又不是lock-free算法实现，基于前面的分析，不建议直接使用原子类来实现，可以使用__sync_bool_compare_and_swap()内置函数来实现功能。

void atomic_add(__int128_t &data, int64_t x, int64_t y) {__int128_t val;__int128_t cmp;do {cmp = data;val = cmp;int64_t *p = (int64_t *) &val;p[0] += x;p[1] += y;} while (!__sync_bool_compare_and_swap(&data, cmp, val));
}#define NUM 10000000
int main(void)
{__int128_t m16 = 0;int64_t *p = (int64_t *) &m16;printf("init: %d-%d\n", *p++, *p);auto lamb = [&m16]() {for (volatile int i=0; i<NUM; i++) {atomic_add(m16, 1, 1);}};thread t1(lamb);thread t2(lamb);thread t3(lamb);thread t4(lamb);t1.join();t2.join();t3.join();t4.join();p = (int64_t *) &m16;printf("finish: %d-%d\n", *p++, *p);
}