6

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

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

这是一个原子操作，意思是 iftail.ptr->next等于next， lettail.ptr->next指向node并同时（原子地） make next.count+1。但是，使用 C++11 CAS，我只能实现：

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

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

4

1 回答 1

18

要使用单个原子操作一次原子地修改两件事，您需要将它们放在相邻的内存中，例如放在两个成员的结构中。然后，您可以使用std::atomic<my_struct>让 gcclock cmpxchg16b在 x86-64 上发出，例如。

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

不幸的是，使用当前的编译器，您需要使用 aunion来获得有效的代码来读取其中一个。lock cmpxchg16b即使我们只需要一个成员，对结构进行原子加载然后只使用一个成员的“明显”方式仍然会导致读取整个结构。（速度慢得多，并且会弄脏缓存行，因此读者与其他读者竞争）。我相信指针的正常 64b 负载仍然可以正确实现 x86 上的获取内存排序语义（以及原子性），但是当前的编译器甚至不会对 进行这种优化std::memory_order_relaxed，所以我们用一个工会。

（为此提交了GCC 错误 80835。TODO：如果这是一个有用的想法，则与 clang 相同。）

清单：

  • 确保您的编译器生成有效的代码，以便在只读情况下仅加载一个成员，而不是一lock cmpxchg16b对中的一个。例如使用联合。

  • 确保您的编译器保证在编写不同的联合成员后访问联合的一个成员在该实现中具有明确定义的行为。 联合类型双关语在 C99 中是合法的（因此这应该适用于 C11 stdatomic），但它是 ISO C++11 中的 UB。但是，它在 C++ 的 GNU 方言中是合法的（由 gcc、clang 和 ICC 等支持）。

  • 确保您的对象是 16B 对齐的，或 8B 对齐的 32 位指针。更一般地说，alignas(2*sizeof(void*))应该工作。未对齐的locked 指令在 x86 上可能非常慢，尤其是当它们跨越缓存线边界时。如果对象未对齐，clang3.8 甚至会将其编译为库调用。

  • 编译-mcx16为 x86-64 版本。 cmpxchg16b最早的 x86-64 CPU (AMD K8) 不支持，但之后的一切都应该支持。没有-mcx16，你会得到一个库函数调用（它可能使用全局锁）。等效的 32 位cmpxchg8b, 已经足够老了，现代编译器假定它支持它。（并且可以使用 SSE、MMX 甚至 x87 来执行 64b 原子加载/存储，因此在读取一个成员时，使用联合对于获得良好性能而言不太重要）。

  • 确保指针+uintptr_t 原子对象是无锁的。对于 x32 和 32 位 ABI（8B 对象），这几乎可以保证，但对于 16B 对象则不然。例如，MSVC 使用 x86-64 的锁。

    gcc7 及更高版本将调用 libatomic 而不是 inlining lock cmpxchg16b，并将返回 false atomic_is_lock_free（原因包括它太慢了，这不是用户期望is_lock_free的意思），但至少目前 libatomic 实现仍然lock cmpxchg16b在该指令可用的目标上使用。（它甚至可以对只读原子对象进行段错误，所以它真的不理想。更重要的是，读取器与其他读取器争夺对缓存行的独占访问。这就是为什么我们要lock cmpxchg16b为读取端避免这么多麻烦当我们只想要一个 8 字节的一半时。）

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

在 Godbolt 编译器资源管理器上查看带有 gcc6.3 的 asm 输出的代码。使用-m32, 它使用cmpxchg8b与 64 位代码相同的方式使用cmpxchg16b. 使用-mx32（长模式下的 32 位指针）它可以简单地使用 64 位cmpxchg和正常的 64 位整数加载来在一个原子加载中抓取两个成员。

这是可移植的 C++11（除了联合类型双关语），没有特定于 x86 的内容。但是，它仅对可以 CAS 两个指针大小的对象的目标有效。 例如，它编译为调用__atomic_compare_exchange_16ARM / ARM64 和 MIPS64 的库函数，正如您在 Godbolt 上看到的那样。

它不能在atomic<counted_ptr>大于的 MSVC 上编译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，因此对指针的只读访问尽可能便宜，这要归功于联合黑客。

它确实依赖于通过一个成员编写联合并通过另一个成员读取它，以便在不使用 a 的情况下仅对指针进行有效读取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-counter 无关，但可能会出现在其他使用联合以允许访问更大原子对象的子集的应用程序中： 不要使用联合来允许原子存储仅指向指针或仅存储计数器。如果您关心与该对的获取负载同步，至少不会。即使是强排序的 x86 也可以用完全包含它的更宽负载重新排序狭窄的存储。一切仍然是原子的，但就内存排序而言，你会进入奇怪的领域。

在 x86-64 上，原子 16B 加载需要一个lock cmpxchg16b（这是一个完整的内存屏障，防止前面的窄存储在它之后变得全局可见）。但是，如果您将其与 32 位指针（或 32 位数组索引）一起使用，您很容易遇到问题，因为这两个部分都可以通过常规 64b 加载来加载。如果您需要与其他线程同步而不仅仅是原子性，我不知道您会在其他架构上看到什么问题。

要了解有关 std::memory_order 获取和释放的更多信息，请参阅Jeff Preshing 的优秀文章。

于 2016-08-17T08:38:29.247 回答