#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);
}

x86上的原子清​​除最低位没有直接的硬件支持. BMI1 blsr仅以内存源形式提供,而不是以内存 – 目标形式提供; lock blsr [shared_var]不存在. (当使用-march = haswell进行编译时,编译器知道如何优化(var-1)&(var)为局部变量的blsr,否则启用假定BMI1支持的代码.)所以即使你可以假设BMI1支持1,它不会让你做任何根本不同的事情.

您在x86上可以做的最好的事情是您在问题中提出的锁定cmpxchg重试循环.这是比在旧版本的变量中找到正确的位然后使用lock btr更好的选择,特别是如果在位扫描和锁之间设置较低的位来清除错误的位将是正确性问题BTR.如果另一个线程已经清除了你想要的位,你仍然需要一个重试循环.

CAS重试循环在实践中并不坏：重试非常罕见

(除非你的程序对共享变量有很高的争用,即使在没有尝试的情况下使用锁定添加也会对性能造成问题,只需要硬件仲裁来访问缓存行.如果是这种情况,你应该重新设计你的无锁算法,/或考虑某种操作系统辅助的睡眠/唤醒,而不是让很多内核花费大量的cpu时间来攻击同一个缓存线.在低争用的情况下,无锁是很好的.)

cpu失去高速缓存之间的缓存线以获得预期和运行锁定cmpxchg以及该值的几个指令结果的窗口很小.通常它会在第一次成功时通过,因为当cmpxchg运行时,缓存行仍将存在于L1d缓存中.当高速缓存行到达时,它(希望)已经处于MESI Exclusive状态,如果cpu看到足够远,可以为它做RFO.

您可以检测cmpxchg重试循环以查看实际程序中实际获得的争用程度. (例如,通过递增循环内的局部并在成功后使用原子relax =进入共享计数器,或者使用线程局部计数器.)

请记住,您的真实代码(希望)在此位掩码上的原子操作之间做了大量工作,因此它与微基准测试非常不同,在微基准测试中,所有线程都花费所有时间来锤击该缓存线.

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,它只进行比较,因此加载相同预期的另一个线程将成功.在LL / SC机器上,希望一个核心上的SC故障不会导致其他核心上的SC故障,否则它可能会活锁.这是你可以假设cpu架构师想到的东西.

它不是等待的,因为单个线程理论上不得不重试无限次.

你的代码用gcc8.1编译-O3 -march = haswell到这个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各自变成两条指令.考虑到锁定指令的成本,这几乎与性能无关.

不幸的是,clang和MSVC稍微笨拙,在无争用的快速路径上采取了分支. (并且clang通过剥离第一次迭代来填充函数.IDK如果这有助于分支预测或其他什么,或者如果它纯粹是错过的优化.我们可以通过使用不太可能的()宏来生成快速路径而没有采用分支. How do the likely() and unlikely() macros in the Linux kernel work and what is their benefit?).

脚注1：

除非您为一组已知的机器构建二进制文件,否则无论如何都不能假设BMI1可用.所以编译器需要做一些像lea rdx,[rax-1] /和rdx,rax而不是blsr rdx,rax.这对于这个功能来说是微不足道的.即使在非竞争情况下,大部分成本也是原子操作,即使对于热插入高速缓存也没有争用情况,查看无序执行吞吐量对周围代码的影响. (例如,对于Skylake(http://agner.org/optimize/)锁定cmpxchg为10 uops而不是使用blsr而不是其他2个指令保存1 uop.假设前端是瓶颈,而不是内存或其他东西.作为完整的内存屏障会对在周围的代码中加载/存储,但幸运的是,不是在无序执行独立的ALU指令.)

非x86：LL/SC machines

大多数非x86机器使用加载链接/存储条件重试循环执行所有原子操作.有点不幸的是,C 11不允许您创建自定义LL / SC操作(例如,在LL / SC中使用(x-1)& x而不是仅使用fetch_add获得的添加),但CAS机器(如x86)无法为您提供LL / SC提供的ABA检测.因此,目前尚不清楚如何设计一个可以在x86上高效编译的C类,还可以直接在ARM和其他LL / SC ISA上进行LL / SC重试循环. (见this discussion)

因此,如果C不提供您想要的操作(例如fetch_or或fetch_and),则必须使用compare_exchange_weak编写代码.

从当前编译器实际获得的是使用LL / SC实现的compare_exchange_weak,并且您的算术运算与之分离. asm实际上在独占 – 加载(ldaxr)之前执行宽松的加载,而不是仅仅基于ldaxr结果的计算.并且还有额外的分支来检查上次尝试的预期与尝试存储之前的新加载结果相匹配.

LL / SC窗口可能比加载和存储之间的2个相关ALU指令短,以防万一. cpu提前准备好所需的值,而不依赖于负载结果. (这取决于之前的加载结果.)Clang的code-gen将sub /和循环内部,但依赖于前一次迭代的加载,因此在乱序执行时,它们仍然可以提前准备好.

但如果这确实是最有效的方法,那么编译器也应该将它用于fetch_add.他们没有,因为它可能不是. LL / SC重试很少见,就像x86上的CAS重试一样.我不知道它是否可以在非常高争用的情况下有所不同.也许,但编译器在编译fetch_add时不会减慢快速路径以进行优化.

为了便于阅读,我重新命名了你的函数并重新格式化了while(),因为对于一行而言,它已经太长了而没有用distinct()包装它.

这个版本编译成可能比你原来的更好的asm,与clang.我还修改了你的函数名,所以它实际上是编译的;你的问题不匹配.我选择了与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_relaxed)))
        {}

    return isolate_lowest_set(expected);
}

Clar -O3 for AArch64(-target aarch64-linux-android -stdlib = libc -mcpu = cortex-a57 on Godbolt)制作了一些奇怪的代码.这是来自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,[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,.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,x8
    ret

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

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

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

## 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,#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,x10
    ret

用于AArch64的gcc6.3也会产生奇怪的代码,存储在堆栈中. (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,x9                   @ compare part of the CAS
    b.ne    .Lcas_fail
    sub     x10,x9             @ clear_lowest (relaxed load result)
    stlxr   w11,.Lllsc_fail

    # LL/SC success: isolate_lowest_set and return
.Lepilogue:
    neg     x8,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为fetch_add()做了很好的代码：

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

而不是添加#1,我们希望有子x9,#1 /和x9,所以我们只得到一个LL / SC重试循环.这样可以节省代码大小,但速度不会快得多.在大多数情况下可能甚至没有明显更快,特别是作为整个计划的一部分,它没有使用疯狂的数量.

替代方案：您是否甚至需要这个位图操作？你可以分手以减少争用吗？

你可以使用原子计数器而不是位图,并在需要时将其映射到位图吗？想要位图的操作可以使用uint64_t(~0ULL)<<<<<< (64计数器)(仅适用于非零计数器).或者也许tmp = 1ULL<<计数器; tmp ^ = tmp-1; (即x86 xor eax,eax / bts rax,rdi(rax =位置0..63处的1位)/ blsmsk rax,rax(rax =设置到该位置的所有位).嗯,还需要一个特殊情况下的掩码= 0,因为连续位掩码有65种可能的状态：64位之一的最高位(或最低位),或根本没有位.

或者,如果位图有一些模式,x86 BMI2 pdep可以将连续位分散到该模式中.获得N个连续的设置位(1ULL<< counter)-1,再次用于计数器< 64. 或者它可能必须是一个位掩码,但不需要是一个单一的位掩码？ 不知道你的用例是什么,但这种想法可能有用： 您是否需要一个所有线程必须争用的单个原子位图？也许你可以把它分成多个块,每个块都在一个单独的缓存行中. (但这使得无法以原子方式对整个地图进行快照.)但是,如果每个线程都有一个首选块,并且在继续寻找另一个块中的一个槽之前总是尝试,那么您可以减少平均情况下的争用. 在asm中(或者在C中使用可怕的联合攻击),你可以尝试通过找到要更新的64位整数的正确字节来减少cmpxchg失败而不减少争用,然后只锁定cmpxchg.这对于这种情况实际上没有用,因为看到相同整数的两个线程都会尝试清除相同的位.但是它可以减少这与在qword中设置其他位的东西之间的重试.当然,只有当qword的其他字节发生变化时锁定cmpxchg成功的问题不正确时,这才有效.