这个信封实现是否正确使用 C++11 原子? [英] Does this envelope implementation correctly use C++11 atomics?

问题描述

我编写了一个简单的信封"类，以确保我正确理解 C++11 原子语义.我有一个标头和一个有效载荷，作者清除标头，填充有效载荷，然后用递增的整数填充标头.这个想法是，读取器然后可以读取头，memcpy 输出有效载荷，再次读取头，如果头相同，则读取器可以假设他们成功复制了有效载荷.读者可能会错过一些更新是可以的，但他们无法获得撕裂的更新(其中有来自不同更新的混合字节).只有一个读者和一个作者.

I have written a simple 'envelope' class to make sure I understand the C++11 atomic semantics correctly. I have a header and a payload, where the writer clears the header, fills in the payload, then fills the header with an increasing integer. The idea is that a reader then can read the header, memcpy out the payload, read the header again, and if the header is the same the reader can then assume they successfully copied the payload. It's OK that the reader may miss some updates, but it's not OK for them to get a torn update (where there is a mix of bytes from different updates). There is only ever a single reader and a single writer.

作者使用释放内存顺序，读者使用获取内存顺序.

The writer uses release memory order and the reader uses acquire memory order.

使用原子存储/加载调用重新排序 memcpy 是否有任何风险?或者负载可以相互重新排序吗?这对我来说永远不会中止，但也许我很幸运.

Is there any risk of the memcpy being reordered with the atomic store/load calls? Or can the loads be reordered with each other? This never aborts for me but maybe I'm lucky.

#include <iostream>
#include <atomic>
#include <thread>
#include <cstring>

struct envelope {
    alignas(64) uint64_t writer_sequence_number = 1;
    std::atomic<uint64_t> sequence_number;
    char payload[5000];

    void start_writing()
    {
        sequence_number.store(0, std::memory_order::memory_order_release);
    }

    void publish()
    {
        sequence_number.store(++writer_sequence_number, std::memory_order::memory_order_release);
    }

    bool try_copy(char* copy)
    {
        auto before = sequence_number.load(std::memory_order::memory_order_acquire);
        if(!before) {
            return false;
        }
        ::memcpy(copy, payload, 5000);
        auto after = sequence_number.load(std::memory_order::memory_order_acquire);
        return before == after;
    }
};

envelope g_envelope;

void reader_thread()
{
    char local_copy[5000];
    unsigned messages_received = 0;
    while(true) {
        if(g_envelope.try_copy(local_copy)) {
            for(int i = 0; i < 5000; ++i) {
                // if there is no tearing we should only see the same letter over and over
                if(local_copy[i] != local_copy[0]) {
                    abort();
                }
            }
            if(messages_received++ % 64 == 0) {
                std::cout << "successfully received=" << messages_received << std::endl;
            }
        }
    }
}

void writer_thread()
{
    const char alphabet[] = {"ABCDEFGHIJKLMNOPQRSTUVWXYZ"};
    unsigned i = 0;
    while(true) {
        char to_write = alphabet[i % (sizeof(alphabet)-1)];
        g_envelope.start_writing();
        ::memset(g_envelope.payload, to_write, 5000);
        g_envelope.publish();
        ++i;
    }
}

int main(int argc, char** argv)
{
    std::thread writer(&writer_thread);
    std::thread reader(&reader_thread);

    writer.join();
    reader.join();

    return 0;
}

推荐答案

这称为 seqlock；仅仅因为对memset 和memcpy 的调用发生冲突，它就会发生数据竞争.已经有人提议提供类似 memcpy 的工具来使这种代码正确；最近的不太可能出现在 C++26 之前(即使被批准).

This is called a seqlock; it has a data race simply because of the conflicting calls to memset and memcpy. There have been proposals to provide a memcpy-like facility to make this sort of code correct; the most recent is not likely to appear before C++26 (even if approved).

这篇关于这个信封实现是否正确使用 C++11 原子?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！