代码之家 › 专栏 › 技术社区 › ridiculous_fish

用32位原子实现64位原子计数器

lockless stdatomic c++11 c++

ridiculous_fish · 技术社区 · 7 年前

我想从原子uint32中拼凑出一个uint64原子计数器。计数器只有一个编写器和多个阅读器。编写器是一个信号处理程序,因此它不能阻塞。

我的想法是使用低位的代计数作为读取锁。读卡器重试,直到生成计数在整个读卡过程中稳定,并且低位未设置。

以下代码在设计和使用内存排序时是否正确?有更好的方法吗?

using namespace std;
class counter {
    atomic<uint32_t> lo_{};
    atomic<uint32_t> hi_{};
    atomic<uint32_t> gen_{};

    uint64_t read() const {
        auto acquire = memory_order_acquire;
        uint32_t lo, hi, gen1, gen2;
        do {
            gen1 = gen_.load(acquire);
            lo = lo_.load(acquire);
            hi = hi_.load(acquire);
            gen2 = gen_.load(acquire);
        } while (gen1 != gen2 || (gen1 & 1));
        return (uint64_t(hi) << 32) | lo;
    }

    void increment() {
        auto release = memory_order_release;
        gen_.fetch_add(1, release);
        uint32_t newlo = 1 + lo_.fetch_add(1, release);
        if (newlo == 0) {
            hi_.fetch_add(1, release);
        }
        gen_.fetch_add(1, release);
    }
};

编辑哎呀,固定 auto acquire = memory_order_release;

1 回复 | 直到 7 年前

Peter Cordes 7 年前

这是一个已知的模式,称为seqlock。 https://en.wikipedia.org/wiki/Seqlock .(简化为只有一个作者,因此不需要额外的支持来排除同时的作者。)

您不需要或不希望计数器变量本身的增量使用原子RMW操作 .您可以使用原子32位加载来加载这两个部分,增加它,并原子地存储结果。(便宜的 relaxed 或 release 内存顺序,并使用 释放 存储以进行第二次计数器更新)。

同样,计数器也不需要是原子RMW。

作者只需要纯负载和纯存储,只需要发布顺序,这比原子RMW便宜很多,或者使用seq-cst顺序的存储。 :

按任意顺序加载计数器和值
存储新计数器(旧的+1)
存储新值(如果要无进位分支,则只更新下半部分)
存储最后一个计数器。

这三个要点中的商店订购是唯一重要的。在第一家店之后设置一个写围栏可能会很好,因为我们不想知道制作成本 二者都 价值两半的存储 释放 在CPU上,这比放松更昂贵。

不幸的是,要满足C++规则, value 必须是 atomic<T> 这使得编译器无法生成加载这两部分的最有效代码。例如手臂 ldp / stp 负载对可能不是原子的,但这并不重要。(编译器通常不会将两个单独的原子32位加载优化为一个更宽的加载。)

当序列计数器为奇数时其他线程读取的值是不相关的,但我们希望避免未定义的行为。也许我们可以用一个 volatile uint64_t 和一个 atomic<uint64_t>

我写的这个C++ SeqLock<class T> 模板 another question 我没有为(弄清楚哪些版本的ARM有64位的原子加载和存储)写出答案。

这将尝试检查目标是否已支持上的无锁原子操作 原子& T; 阻止你在毫无意义的时候使用它。(通过定义 IGNORE_SIZECHECK 。)TODO:很明显,可以使用模板专门化,而不是使用 static_assert .

我提供了一个 inc() 功能 T 支持一个 ++ 操作员。托多会是一个 apply() 接受lambda对 T ,并在序列计数器更新之间存储结果。

// **UNTESTED**

#include <atomic>

#ifdef UNIPROCESSOR
// all readers and writers run on the same core
// ordering instructions at compile time is all that's necessary
#define ATOMIC_FENCE std::atomic_signal_fence
#else
// A reader can be running on another core while writing
// memory barriers or ARMv8 acquire / release loads / store are needed
#define ATOMIC_FENCE std::atomic_thread_fence
#endif
// using fences instead of .store(std::memory_order_release) will stop the compiler
// from taking advantage of a release-store instruction, like on AArch64 or x86


// SINGLE WRITER only.
// uses volatile + barriers for the data itself, like pre-C++11
template <class T>
class SeqLocked
{
#ifndef IGNORE_SIZECHECK
    // sizeof(T) > sizeof(unsigned)
    static_assert(!std::atomic<T>::is_always_lock_free, "A Seq Lock with a type small enough to be atomic on its own is totally pointless, and we don't have a specialization that replaces it with a straight wrapper for atomic<T>");
#endif

       // C++17 doesn't have a good way to express a load that doesn't care about tearing
       //  without explicitly writing it as multiple small parts and thus gimping the compiler if it can use larger loads
    volatile T data;          // volatile should be fine on any implementation where pre-C++11 lockless code was possible with volatile,
                              //  even though Data Race UB does apply to volatile variables in ISO C++11 and later.

    std::atomic<unsigned> seqcount{0};  // Even means valid, odd means modification in progress.
                                        //  unsigned wraps around at a power of 2 on overflow

public:
    T get() const {
        unsigned c0, c1;
        T tmp;

        do {
            c0 = seqcount.load(std::memory_order_relaxed);  // or this can be a std::memory_order_acquire for multicore so AArch64 can use LDAR
            ATOMIC_FENCE(std::memory_order_acquire);

            tmp = (T)data;       // load

            ATOMIC_FENCE(std::memory_order_acquire);  // LoadLoad barrier
            c1 = seqcount.load(std::memory_order_relaxed);
        } while(c0&1 || c0 != c1);     // retry if the counter changed or is odd

        return tmp;
    }

    // TODO: a version of this that takes a lambda for the operation on tmp
    T inc() {
        unsigned orig_count = seqcount.load(std::memory_order_relaxed);

        seqcount.store(orig_count+1, std::memory_order_relaxed);
        ATOMIC_FENCE(std::memory_order_release);
        // make sure the data stores appear after the first counter update.

        T tmp = data;  // load
        ++tmp;
        data = tmp;    // store

        ATOMIC_FENCE(std::memory_order_release);
        seqcount.store(orig_count+2, std::memory_order_relaxed);  // Or use mo_release here, better on AArch64

        return tmp;
    }

    void set(T newval) {
        unsigned orig_count = seqcount.load(std::memory_order_relaxed);

        seqcount.store(orig_count+1, std::memory_order_relaxed);
        ATOMIC_FENCE(std::memory_order_release);
        // make sure the data stores appear after the first counter update.

        data = newval;    // store

        ATOMIC_FENCE(std::memory_order_release);
        seqcount.store(orig_count+2, std::memory_order_relaxed);  // Or use mo_release here, better on AArch64
    }

};


/***** test callers *******/
#include <stdint.h>

struct sixteenbyte {
    //unsigned arr[4];
    unsigned long  a,b,c,d;
    sixteenbyte() = default;
    sixteenbyte(const volatile sixteenbyte &old)
         : a(old.a), b(old.b), c(old.c), d(old.d) {}
    //arr(old.arr) {}
};

void test_inc(SeqLocked<uint64_t> &obj) {  obj.inc(); }
sixteenbyte test_get(SeqLocked<sixteenbyte> &obj) { return obj.get(); }
//void test_set(SeqLocked<sixteenbyte> &obj, sixteenbyte val) { obj.set(val); }

uint64_t test_get(SeqLocked<uint64_t> &obj) {
    return obj.get();
}

// void atomic_inc_u64_seq_cst(std::atomic<uint64_t> &a) { ++a; }
uint64_t u64_inc_relaxed(std::atomic<uint64_t> &a) {
    // same but without dmb barriers
    return 1 + a.fetch_add(1, std::memory_order_relaxed);
}

uint64_t u64_load_relaxed(std::atomic<uint64_t> &a) {
    // gcc uses LDREXD, not just LDRD?
    return a.load(std::memory_order_relaxed);
}

void u64_store_relaxed(std::atomic<uint64_t> &a, uint64_t val) {
    // gcc uses a LL/SC retry loop even for a pure store?
    a.store(val, std::memory_order_relaxed);
}

它编译成我们想要的ASM on the Godbolt compiler explorer 对于ARM和其他ISA。至少对于Int64而言,由于繁琐,更大的结构类型的复制效率可能更低。 volatile 规则。

它使用非原子 volatile T data 共享数据。从技术上讲,这是一种未定义的数据竞争行为,但是我们在实践中使用的所有编译器都可以通过pre-C++11多线程访问 不稳定的 物体。而在C++11之前,人们甚至在某些尺寸上依赖原子性。我们做不 ,我们检查计数器,仅在没有并发写入时使用读取的值。(这就是Seqlock的全部意义。)

一个问题是 易失性T数据 是在ISO C++中, T foo = data 除非提供来自 不稳定的 对象,像

sixteenbyte(const volatile sixteenbyte &old)
         : a(old.a), b(old.b), c(old.c), d(old.d) {}

这对我们来说真的很烦人,因为我们不关心如何读取内存的细节,只是多个读取没有优化为一个。

不稳定的 这里的工具真的不对吗 和平原 T data 有足够的围栏,以确保在原子计数器的读取之间实际发生的读取会更好。例如,我们可以在GNU C中使用 asm("":::"memory"); 编译器阻止在访问之前/之后重新排序。这样编译器就可以用SIMD向量或其他任何方法复制更大的对象,而这是不需要单独处理的。 不稳定的 访问。

我想 std::atomic_thread_fence(mo_acquire) 这也是一个足够的障碍,但我不能百分之百确定。

在ISO C中,您可以复制 不稳定的 聚合(struct),编译器将发出通常复制那么多字节所需的任何asm。但是在C++中,我们显然不能有好的东西。