如何用c ++ 11 CAS实现ABA计数器？

Question

我正在实现一个基于此算法的无锁队列,该算法使用计数器来解决ABA问题.但我不知道如何用c ++ 11 CAS实现这个计数器.例如,从算法:

E9:    if CAS(&tail.ptr->next, next, <node, next.count+1>)

它是一个原子操作,意思是如果tail.ptr->next等于next,则tail.ptr->next指向node并同时(原子地)产生next.count+1.但是,使用C++ 11 CAS,我只能实现:

std::atomic_compare_exchange_weak(&tail.ptr->next, next, node);

这不可能next.count+1同时发生.

Answer 1

要使用单个原子操作一次原子地修改两个东西,您需要将它们放在相邻的内存中,例如在两个成员的结构中.然后你可以std::atomic<my_struct>用来让gcc lock cmpxchg16b在x86-64 上发光.

你不需要内联asm,为了避免它,值得一点C++语法痛苦. https://gcc.gnu.org/wiki/DontUseInlineAsm.

不幸的是,对于当前的编译器,你需要使用a union来获得有效的代码来读取其中一对.进行原子加载结构然后只使用一个成员的"显而易见"的方法仍然导致lock cmpxchg16b读取整个结构,即使我们只需要一个成员.我相信指针的正常64b加载仍然可以正确地实现x86上的获取内存排序语义(以及原子性),但是当前的编译器甚至没有进行优化std::memory_order_relaxed,所以我们用它来欺骗它们工会.

(已提交GCC错误80835有关此问题.TODO:如果这是一个有用的想法,则同样适用于clang.)

清单:

• 确保您的编译器生成有效的代码,只在一个只读的情况下加载一个成员,而不是lock cmpxchg16b该对中的一个成员.例如使用联盟.
• 确保您的编译器保证在写入不同的union成员后访问union的一个成员在该实现中具有明确定义的行为. 联合类型 - 惩罚在C99中是合法的(因此这应该适用于C11 stdatomic),但它在ISO C++ 11中是UB.但是,它在C++的GNU方言中是合法的(由gcc,clang和ICC等支持).
• 确保对象是16B对齐的,或者对于32位指针进行8B对齐.更一般地说,alignas(2*sizeof(void*))应该工作.在x86上,未对齐的locked指令可能会非常慢,尤其是当它们跨越缓存行边界时.如果对象未对齐,clang3.8甚至会将其编译为库调用.

• 编译-mcx16为x86-64版本. cmpxchg16b最早的x86-64 CPU(AMD K8)不支持,但在此之后应该是所有内容.没有-mcx16,你得到一个库函数调用(可能使用全局锁).32位等价物,cmpxchg8b足以让现代编译器得到它的支持.(并且可以使用SSE,MMX甚至x87来执行64b原子加载/存储,因此在读取一个成员时,使用union对于获得良好性能并不那么重要).

• 确保指针+ uintptr_t原子对象是无锁的.对于x32和32位ABI(8B对象),这几乎是有保证的,但对于16B对象则没有.例如,MSVC使用锁定x86-64.

  gcc7和更高版本将调用libatomic而不是内联lock cmpxchg16b,并将返回false atomic_is_lock_free(原因包括它的速度太慢,而不是用户期望is_lock_free的意思),但至少目前libatomic实现仍然使用lock cmpxchg16b该指令可用的目标.(它甚至可以对只读原子对象进行分段,因此它实际上并不理想.)

这是一个带有CAS重试循环的代码示例,它编译为看起来正确的asm,而且我认为对于允许联合类型双关语的实现,它没有UB或其他不安全的C++.它是用C风格编写的(非成员函数等),但如果编写成员函数则会相同.

请参阅Godbolt编译器资源管理器中gcc6.3的 asm输出代码.有-m32,它使用cmpxchg8b64位代码使用相同的方式cmpxchg16b.使用-mx32(长模式下的32位指针),它可以简单地使用64位cmpxchg和普通的64位整数加载来捕获一个原子负载中的两个成员.

这是可移植的C++ 11(联合类型 - 惩罚除外),没有任何x86特定的.但它只能在CAS上使用两个指针大小的对象才有效. 例如,它编译为调用__atomic_compare_exchange_16ARM/ARM64和MIPS64 的库函数,正如您在Godbolt上看到的那样.

它不能在MSVC上编译,其中atomic<counted_ptr>大于counted_ptr_separate,所以static_assert捕获它.据推测,MSVC在原子对象中包含一个锁定成员.

#include <atomic>
#include <stdint.h>

using namespace std;

struct node {
  // This alignas is essential for clang to use cmpxchg16b instead of a function call
  // Apparently just having it on the union member isn't enough.
  struct alignas(2*sizeof(node*)) counted_ptr {
    node *    ptr;
    uintptr_t count;  // use pointer-sized integers to avoid padding
  };

  // hack to allow reading just the pointer without lock-cmpxchg16b,
  // but still without any C++ data race
  struct counted_ptr_separate {
    atomic<node *>    ptr;
    atomic<uintptr_t> count_separate;  // var name emphasizes that accessing this way isn't atomic with ptr
  };

  static_assert(sizeof(atomic<counted_ptr>) == sizeof(counted_ptr_separate), "atomic<counted_ptr> isn't the same size as the separate version; union type-punning will be bogus");
  //static_assert(std::atomic<counted_ptr>{}.is_lock_free());

  union {  // anonymous union: the members are directly part of struct node
    alignas(2*sizeof(node*)) atomic<counted_ptr> next_and_count;
    counted_ptr_separate  next;
  };
  // TODO: write member functions to read next.ptr or read/write next_and_count

  int data[4];
};


// make sure read-only access is efficient.
node *follow(node *p) {   // good asm, just a mov load
  return p->next.ptr.load(memory_order_acquire);
}
node *follow_nounion(node *p) {  // really bad asm, using cmpxchg16b to load the whole thing
  return p->next_and_count.load(memory_order_acquire).ptr;
}


void update_next(node &target, node *desired)
{
  // read the old value efficiently to avoid overhead for the no-contention case
  // tearing (or stale data from a relaxed load) will just lead to a retry
  node::counted_ptr expected = {
      target.next.ptr.load(memory_order_relaxed),
      target.next.count_separate.load(memory_order_relaxed) };

  bool success;
  do {
    node::counted_ptr newval = { desired, expected.count + 1 };
    // x86-64: compiles to cmpxchg16b
    success = target.next_and_count.compare_exchange_weak(
                           expected, newval, memory_order_acq_rel);
    // updates exected on failure
  } while( !success );
}

clang 4.0的asm输出-O3 -mcx16是:

update_next(node&, node*):
    push    rbx             # cmpxchg16b uses rbx implicitly so it has to be saved/restored
    mov     rbx, rsi
    mov     rax, qword ptr [rdi]     # load the pointer
    mov     rdx, qword ptr [rdi + 8] # load the counter
.LBB2_1:                        # =>This Inner Loop Header: Depth=1
    lea     rcx, [rdx + 1]
    lock
    cmpxchg16b      xmmword ptr [rdi]
    jne     .LBB2_1
    pop     rbx
    ret

gcc做了一些笨重的存储/重新加载,但基本上是相同的逻辑.

follow(node*)编译为mov rax, [rdi]/ ret,因此对指针的只读访问就像它应该的那样便宜,这要归功于联合黑客攻击.

它确实依赖于通过一个成员编写一个联合并通过另一个成员读取它,以便在不使用指针的情况下有效地读取指针lock cmpxchg16b.这保证可以在GNU C++(和ISO C99/C11)中使用,但不能在ISO C++中使用.许多其他C++编译器确实可以保证联合类型 - 惩罚工作,但即使没有它,它可能仍然有效:我们总是使用std::atomic必须假设值被异步修改的负载.因此,在通过另一个指针(或联合成员)写入值之后,寄存器中的值仍被认为是实时的,我们应该免受类似于混叠的问题的影响.编译时认为独立的编译时重新排序可能是一个问题.

在指针+计数器的原子cmpxchg之后原子读取指针仍应在x86上给出获取/释放语义,但我不认为ISO C++对此有任何说明. 我猜想一个广泛的发布商店(作为一部分compare_exchange_weak将与大多数架构上相同地址的较窄负载同步(就像在x86上一样),但AFAIK C++ std::atomic并不保证任何关于类型惩罚的东西.

与指针+ ABA计数器无关,但可能会出现其他使用联合的应用程序以允许访问较大原子对象的子集: 不要使用联合允许原子存储仅指向指针或只是计数器.至少不是如果你关心与对的获取负载的同步.即使是强烈排序的x86也可以对具有更宽容量的窄存储进行重新排序,该存储完全包含它.一切都仍然是原子的,但就内存排序而言,你会进入奇怪的领域.

在x86-64上,原子16B负载需要一个lock cmpxchg16b(这是一个完整的内存屏障,防止前面的窄存储在其后变为全局可见).但是,如果您使用32位指针(或32位数组索引),则很容易出现问题,因为两个半部分都可以加载常规的64b负载.如果您需要与其他线程同步而不仅仅是原子性,我不知道您可以在其他架构上看到哪些问题.

要了解有关std :: memory_order获取和发布的更多信息,请参阅Jeff Preshing的优秀文章.