c++ - 以原子方式清除无符号整数的最低非零位

Question

问题:
我正在寻找以线程安全的方式清除无符号原子(如std::atomic_uint64_t)的最低非零位的最佳方法，而无需使用额外的互斥锁等。另外，我还需要知道哪些地方已经清除。

示例:
可以说，如果当前存储的值为0b0110，我想知道最低的非零位是bit 1(0索引)，并将变量设置为0b0100。

我想出的最好的版本是this:

#include <atomic>
#include <cstdint>

inline uint64_t with_lowest_non_zero_cleared(std::uint64_t v){
    return v-1 & v;
}
inline uint64_t only_keep_lowest_non_zero(std::uint64_t v){
    return v & ~with_lowest_non_zero_cleared(v);
}

uint64_t pop_lowest_non_zero(std::atomic_uint64_t& foo)
{
    auto expected = foo.load(std::memory_order_relaxed);
    while(!foo.compare_exchange_weak(expected,with_lowest_non_zero_cleared(expected), std::memory_order_seq_cst, std::memory_order_relaxed))
    {;}
    return only_keep_lowest_non_zero(expected);
}

有更好的解决方案吗？

笔记:

不必一定是最低的非零位-我也很希望解决方案能够清除最高位或设置最高/最低“零位”(如果有区别的话)

我非常希望使用标准的c++(17)版本，但是会接受将内在函数用于clang和msvc的答案

它应该是可移植的(对于x64和AArch64，使用clang或msvc进行编译)，但是当使用clang进行编译时，我对最新的intel和AMD x64体系结构的性能最感兴趣。

编辑:更新必须是原子的，必须保证全局进度，但是就像上面的解决方案一样，它不必是免费的等待算法(当然，如果您能向我展示一个，我将非常高兴) 。

Answer 1

x86 上没有对原子清除最低位的直接硬件支持。 BMI1 blsr 仅以内存源形式提供，而不是内存目的地形式； lock blsr [shared_var]不存在。 (当您使用(var-1) & (var)进行编译或以其他方式启用假定BMI1支持的代码生成时，编译器知道如何针对本地var将blsr优化为-march=haswell。)因此，即使您可以假定BMI1 support1，也不会做任何根本不同的事情。

在x86上可以做的最好的事情就是在问题中建议的lock cmpxchg重试循环。与在变量的旧版本中找到正确的位然后使用lock btr相比，这是一个更好的选择，尤其是如果在bit-scan和lock btr之间设置了较低的位以清除错误的位会是一个正确性问题时，尤其如此。而且，如果另一个线程已经清除了您想要的位，您仍然需要重试循环。

CAS重试循环在实践中还不错:重试非常少见

(除非您的程序对共享变量的争用非常高，即使使用lock add进行尝试，只要尝试通过硬件仲裁来访问高速缓存行，即使使用expected，性能也会出现问题。如果是这种情况，您可能应该重新设计无锁算法和/或考虑采用某种操作系统辅助的 sleep /唤醒方式，而不是让大量内核在同一缓存行上花费大量的CPU时间。在低争用情况下，无锁非常有用。)

CPU丢失在获取lock cmpxchg的加载和运行cmpxchg之间的高速缓存行的窗口很小，并且在该值上有几个指令的结果。 通常，它将第一次通过成功，因为当+=运行时，高速缓存行仍将存在于L1d高速缓存中。当高速缓存行到达时，如果CPU能够看到足够远的距离为其执行RFO，则(希望)它已经处于MESI独占状态。

您可以检测cmpxchg重试循环，以查看实际程序中实际获得了多少竞争。 (例如，通过在循环内增加局部变量，并在成功后使用原子放松的expected到共享计数器中，或与线程局部计数器一起使用)。

请记住，您的真实代码(希望如此)在此位掩码上的原子操作之间完成了大量工作，因此，它与微基准测试有很大不同，在微基准测试中，所有线程将所有时间都花在了该高速缓存行上。

EDIT: The update has to be atomic and global progress must be guaranteed, but just as with solution above, it doesn't have to be a wait free algorithm (of course I'd be very happy if you can show me one).

从技术意义上讲，CAS重试循环(即使在LL/SC机器上编译，也请参见下文)在技术上是无锁的:您仅需重试其他线程就可以重试。

如果失败，则CAS会使高速缓存行保持不变。在x86上，它弄脏了缓存行(MESI M状态)，但是x86 cmpxchg不能检测到ABA，它仅进行比较，因此，另一个加载了相同 gcc8.1 -O3 -march=haswell的线程将成功。希望在LL/SC机器上，一个内核上的SC故障不会引起其他内核上的SSC故障，否则它可能会死锁。您可以假设CPU架构师想到了这一点。

它不是无休止的，因为理论上单个线程可能必须重试无数次。

您的代码使用 lock编译到此asm( from the Godbolt compiler explorer)

# gcc's code-gen for x86 with BMI1 looks optimal to me.  No wasted instructions!
# presumably you'll get something similar when inlining.
pop_lowest_non_zero(std::atomic<unsigned long>&):
    mov     rax, QWORD PTR [rdi]
.L2:
    blsr    rdx, rax                      # Bit Lowest Set Reset
    lock cmpxchg    QWORD PTR [rdi], rdx
    jne     .L2                           # fall through on success: cmpxchg sets ZF on equal like regular cmp
    blsi    rax, rax                      # Bit Lowest Set Isolate
    ret

如果没有BMI1，则blsr和blsi分别成为两条指令。鉴于 unlikely()指令的成本，这与性能几乎无关。

不幸的是，clang和MSVC有点笨拙，在无争用的快速路径上有一个采用的分支。 (然后clang通过剥离第一个迭代使函数膨胀。IDK如果这有助于分支预测或其他事情，或者纯粹是错过了优化。我们可以使用 lea rdx, [rax-1]宏让clang生成不带分支的快速路径。 How do the likely() and unlikely() macros in the Linux kernel work and what is their benefit?) 。

脚注1:

除非您要为一组已知的计算机构建二进制文件，否则您无法假设BMI1仍然可用。因此，编译器将需要执行 and rdx, rax/ blsr rdx, rax之类的操作，而不是 lock cmpxchg。这对于该功能而言是微不足道的。考虑到乱序执行吞吐量对周围代码的影响，即使在无竞争情况下，大部分成本也是原子操作，即使对于L1d高速缓存无争用情况也是如此。 (例如，在Skylake( http://agner.org/optimize/)上为 blsr设置10微秒，而不是使用 (x-1) & x而不是其他2条指令来保存1 uop。假定前端是瓶颈，而不是内存或其他东西。内存不足会影响负载/也存储在周围的代码中，但幸运的是不会在无序执行独立ALU指令的情况下。)

非x86: LL/SC machines

大多数非x86机器通过负载链接/存储条件重试循环执行其所有原子操作。不幸的是，C++ 11不允许您创建自定义的LL/SC操作(例如，在LL/SC中使用 fetch_add而不是仅使用 compare_exchange_weak获得的添加)，但是使用CAS机器(例如x86)无法为您提供LL/SC提供的ABA检测。因此，目前尚不清楚如何设计C++类，该类可以在x86上有效地编译，但也可以直接在ARM和其他LL/SC ISA上直接编译为LL/SC重试循环。 (请参阅 this discussion。)

因此，如果C++没有提供所需的操作(例如 fetch_or或 fetch_and)，则只需使用 compare_exchange_weak编写代码。

从当前的编译器实际得到的是使用LL/SC实现的 ldaxr，并且您的算术运算与此分开。实际上，asm会在排他性负载获取( ldaxr)之前执行宽松的负载，而不是仅仅基于 expected结果进行计算。并且还有一个额外的分支，可以在尝试存储之前检查上次尝试的 sub与新的加载结果是否匹配。

LL/SC窗口可能要短于在加载和存储之间使用2条从属ALU指令的情况，以防万一。 CPU可以提前准备好所需的值，与负载结果无关。 (这取决于先前的加载结果。)Clang的代码生成将 and/ fetch_add放入循环内，但取决于先前迭代的加载，因此在无序执行的情况下，它们仍可以提前准备就绪。

但是，如果这确实是最有效的处理方法，则编译器也应该将其用于 fetch_add。他们没有，因为可能不是。 LL/SC重试很少，就像x86上的CAS重试一样。我不知道在竞争非常激烈的情况下它是否会有所作为。也许可以，但是编译器在编译 while()时不会减慢为此优化的快速路径。

我重命名了您的函数并重新设置了 unlikely()的可读性，因为一行已经太长而没有用 -O3包裹它。

此版本使用clang 编译为可能比原始版本更好的asm。我还修复了您的函数名，以便它实际上可以编译；您的问题不匹配。我选择了完全不同的名称，它们与x86的BMI指令名称相似，因为它们简洁地描述了操作。

#include <atomic>
#include <cstdint>

#ifdef __GNUC__
#define unlikely(expr) __builtin_expect(!!(expr), 0)
#define likely(expr) __builtin_expect(!!(expr), 1)
#else
#define unlikely(expr) (expr)
#define likely(expr) (expr)
#endif

inline uint64_t clear_lowest_set(std::uint64_t v){
    return v-1 & v;
}
inline uint64_t isolate_lowest_set(std::uint64_t v){
    //return v & ~clear_lowest_set(v);
    return (-v) & v;
    // MSVC optimizes this better for ARM when used separately.
    // The other way (in terms of clear_lowest_set) does still CSE to 2 instructions
    //  when the clear_lowest_set result is already available
}

uint64_t pop_lowest_non_zero(std::atomic_uint64_t& foo)
{
    auto expected = foo.load(std::memory_order_relaxed);
    while(unlikely(!foo.compare_exchange_weak(
        expected, clear_lowest_set(expected), 
        std::memory_order_seq_cst, std::memory_order_relaxed)))
        {}

    return isolate_lowest_set(expected);
}

AArch64的Clang -target aarch64-linux-android -stdlib=libc++ -mcpu=cortex-a57(Godbolt上的 dmb ish)产生了一些奇怪的代码。这来自clang6.0，但是较旧的版本也很奇怪，在寄存器中创建一个0/1整数，然后对其进行测试，而不是仅仅跳到正确的位置。

pop_lowest_non_zero(std::__1::atomic<unsigned long long>&): // @pop_lowest_non_zero(std::__1::atomic<unsigned long long>&)

    ldr     x9, [x0]                   @ the relaxed load
    ldaxr   x8, [x0]                   @ the CAS load (a=acquire, x=exclusive: pairs with a later stxr)
    cmp     x8, x9                   @ compare part of the CAS
    b.ne    .LBB0_3
    sub     x10, x9, #1
    and     x10, x10, x9             @ clear_lowest( relaxed load result)
    stlxr   w11, x10, [x0]           @ the CAS store (sequential-release)
    cbnz    w11, .LBB0_4             @ if(SC failed) goto retry loop

          # else fall through and eventually reach the epilogue

    # this looks insane.  w10 = 0|1, then branch if bit[0] != 0.  Always taken, right?
    orr     w10, wzr, #0x1
    tbnz    w10, #0, .LBB0_5         @ test-bit not-zero will always jump to the epilogue
        b       .LBB0_6              # never reached


.LBB0_3:
    clrex                           @ clear the ldrx/stxr transaction
.LBB0_4:
    # This block is pointless; just should b to .LBB0_6
    mov     w10, wzr
    tbz     w10, #0, .LBB0_6    # go to the retry loop, below the ret (not shown here)

.LBB0_5:               # isolate_lowest_set and return
    neg     x8, x9
    and     x0, x9, x8
    ret

.LBB0_6:
     @ the retry loop.  Only reached if the compare or SC failed.
     ...

所有这些疯狂的分支都可能不是真正的性能问题，但是很显然，这可能会效率更高(即使没有教c如何直接使用LL/SC而不是模拟CAS)。 报告为clang/LLVM错误38173](https://bugs.llvm.org/show_bug.cgi?id=38173)

似乎MSVC并不知道AArch64的发布存储区是顺序发布的(不仅仅是像x86这样的常规版本)，因为它在 stlxr之后仍然使用 compare_exchange_weak指令(完整的内存屏障)。它具有更少的浪费指令，但是却显示出x86偏差: compare_exchange_strong像 dmb一样编译，带有可能没有用的重试循环，它将在LL/SC失败时以旧的预期/期望再次尝试CAS。这通常是因为另一个线程修改了该行，因此预期会不匹配。 (Godbolt在任何较旧的版本中都没有针对AArch64的MSVC，因此也许是全新的支持。这可能解释了 expected)

   ## MSVC Pre 2018 -Ox
|pop_lowest_non_zero| PROC
    ldr         x10,[x0]          # x10 = expected = foo.load(relaxed)

|$LL2@pop_lowest|           @ L2  # top of the while() loop
    sub         x8,x10,#1
    and         x11,x8,x10        # clear_lowest(relaxed load result)
|$LN76@pop_lowest|          @ LN76
    ldaxr       x8,[x0]
    cmp         x8,x10            # the compare part of CAS
    bne         |$LN77@pop_lowest|
    stlxr       w9,x11,[x0]
    cbnz        w9,|$LN76@pop_lowest|  # retry just the CAS on LL/SC fail; this looks like compare_exchange_strong
     # fall through on LL/SC success

|$LN77@pop_lowest|          @ LN77
    dmb         ish                # full memory barrier: slow
    cmp         x8,x10             # compare again, because apparently MSVC wants to share the `dmb` instruction between the CAS-fail and CAS-success paths.
    beq         |$LN75@pop_lowest| # if(expected matches) goto epilogue
    mov         x10,x8             # else update expected
    b           |$LL2@pop_lowest|  # and goto the top of the while loop

|$LN75@pop_lowest|          @ LN75   # function epilogue
    neg         x8,x10
    and         x0,x8,x10
    ret

用于AArch64的gcc6.3也会生成奇怪的代码，将 .fetch_add存储到堆栈中。 (Godbolt没有更新的AArch64 gcc)。

我对AArch64编译器不满意! IDK为什么它们不生成像这样的干净高效的东西，在快速路径上没有分支，只有一些脱机代码可以设置以跳回到CAS中重试。

pop_lowest_non_zero ## hand written based on clang
                    # clang could emit this even without turning CAS into LL/SC directly

    ldr     x9, [x0]                   @ x9 = expected = foo.load(relaxed)
 .Lcas_retry:
    ldaxr   x8, [x0]                   @ x8 = the CAS load (a=acquire, x=exclusive: pairs with a later stxr)
    cmp     x8, x9                   @ compare part of the CAS
    b.ne    .Lcas_fail
    sub     x10, x9, #1
    and     x10, x10, x9             @ clear_lowest (relaxed load result)
    stlxr   w11, x10, [x0]           @ the CAS store (sequential-release)
    cbnz    w11, .Lllsc_fail

    # LL/SC success: isolate_lowest_set and return
.Lepilogue:
    neg     x8, x9
    and     x0, x9, x8
    ret

.Lcas_fail:
    clrex                           @ clear the ldrx/stxr transaction
.Lllsc_fail:
    mov     x9, x8                  @ update expected
    b     .Lcas_retry           @ instead of duplicating the loop, jump back to the existing one with x9 = expected

与 fetch_add()比较:

Clang确实为 add #1编写了不错的代码:

    mov     x8, x0
.LBB1_1:
    ldxr    x0, [x8]                # relaxed exclusive load: I used mo_release
    add     x9, x0, #1
    stlxr   w10, x9, [x8]
    cbnz    w10, .LBB1_1            # retry if LL/SC failed
    ret

而不是 sub x9, x8, #1，我们想要 and x9, x9, x8/ uint64_t(~0ULL) << (64-counter)，所以我们只得到一个LL/SC重试循环。这样可以节省代码大小，但不会明显更快。在大多数情况下，甚至可能不是相当快的速度，特别是作为整个程序的一部分，而该程序没有用到疯狂的数量。

替代方案:您甚至需要此位图操作吗？您可以将其拆分以减少争用吗？

可以使用原子计数器代替位图，并在需要时将其映射到位图吗？需要位图的操作可以将计数器映射到带有 tmp=1ULL << counter; tmp ^= tmp-1;的位图(仅适用于非零计数器)。或 xor eax,eax(即x86 bts rax, rdi/ blsmsk rax, rax(rax = 1将位置0..63设置为位)/ pdep (rax =将所有位设置为该位置)。嗯，对于mask = 0仍需要特殊情况，因为连续的位掩码有65种可能的状态:最高(或最低)位在64个位置之一，或者根本没有设置任何位。

或者，如果位图有某种模式， x86 BMI2 (1ULL << counter) - 1 可以将连续的位分散到该模式中。再次使用 lock cmpxchg获取N个连续的设置位，用于计数器<64。

也许它必须是一个位掩码，但不必是一个位掩码？

不知道您的用例是什么，但是这种想法可能会有用:

您是否需要所有线程都必须竞争的单个原子位图？也许您可以将其分为多个块，每个块位于单独的缓存行中。 (但是，这样就不可能对整个 map 进行原子快照了。)不过，如果每个线程都有一个首选块，并且在继续尝试寻找另一个块中的插槽之前总是尝试这样做，则通常情况下，您可以减少争用。

在asm(或C++中可怕的 union 黑客)中，您可以尝试通过找到要更新的64位整数的正确字节，然后仅在其上添加 lock cmpxchg来减少cmpxchg故障，而不会减少争用。这实际上对这种情况没有帮助，因为看到相同整数的两个线程都将尝试清除同一位。但是，这可以减少在此操作与设置qword中其他位的操作之间的重试。当然，只有当qword的其他字节已更改时ojit_code成功不是正确性问题时，这才起作用。