单实时写/多非实时读无锁同步的C语言实现方案问询

论文中的原始伪代码 #

论文《Non-blocking synchronization between real-time and non-real-time applications》清单2中的无锁同步伪代码如下：

volatile counter : pointer to shared memory area of an integer
data_size : unsigned
pt_base : pointer to shared memory area of data_size bytes

initialization (d_size: unsigned)
    counter := 0
    data_size := d_size
    pt_base := allocate data_size bytes in the shared memory

//Real-time operation
write (new_value: pointer)
    local_counter := *counter
    *counter := local_counter + 1
    copy data_size bytes from *new_value
    *pt_base*data_size
    *counter := local_counter + 2

//Non-real-time operation
read (pt_data: pointer)
    loop
        counter_begin := *counter
        copy data_size bytes from *pt_base to *pt_data *data_size
        counter_end := *counter
        if (counter_end == counter_begin and counter_begin is even)
            break;
        endif
    endloop

需求与当前实现 #

需求是基于该伪代码，用C11 _Atomic、gcc/clang __atomic 或 volatile+内存屏障实现正确的无锁同步，并评估以下用户实现的正确性：

用户当前实现代码 #

struct latency
{
    uint64_t counter;
    uint64_t active;
    uint64_t minimal;
    uint64_t maximal;
    uint64_t sum;
};

struct timer
{
     //timer related stuff
     struct latency latency;
     _Atomic uint32_t latency_seqcount;
};

static struct timer *timer;

//writer invoked by real-time thread
static void writer(void)
{
    uint32_t seqcount;
    struct latency *latency;

    //setup timer / next period
    //timer wait
    //calculate latency

    seqcount = atomic_load_explicit(&timer->latency_seqcount, memory_order_relaxed);
    atomic_store_explicit(&timer->latency_seqcount, seqcount + 1, memory_order_relaxed);
    atomic_thread_fence(memory_order_release);

    latency = &timer->latency;
    latency->active = latency_domain;

    if (latency->counter == 0)
    {
        latency->maximal = latency->active;
        latency->minimal = latency->active;
    }
    else if (latency->active > latency->maximal)
    {
        latency->maximal = latency->active;
    }
    else if (latency->active < latency->minimal)
    {
        latency->minimal = latency->active;
    }

    latency->counter++;
    latency->sum += latency->active;

    atomic_store_explicit(&timer->latency_seqcount, seqcount + 2, memory_order_release);
}

//reader invoked by non-real-time thread
static void reader(void)
{
    struct latency latency;
    uint32_t seqcount_begin;
    uint32_t seqcount_end;

    do
    {
        seqcount_begin = atomic_load_explicit(&timer->latency_seqcount, memory_order_acquire);

        latency.counter = timer->latency.counter;
        latency.active = timer->latency.active;
        latency.minimal = timer->latency.minimal;
        latency.maximal = timer->latency.maximal;
        latency.sum = timer->latency.sum;

        atomic_thread_fence(memory_order_acquire);
        seqcount_end = atomic_load_explicit(&timer->latency_seqcount, memory_order_relaxed);
    }
    while (seqcount_begin != seqcount_end || seqcount_end & 0x01);
    
    //print
}

当前实现的问题分析 #

1. 内存屏障位置错误：
   writer中的atomic_thread_fence(memory_order_release)放在seqcount+1写入之后、数据修改之前，无法保证共享数据的修改操作在seqcount+2写入前完成，违背了伪代码中"先标记写中、再写数据、最后标记写完成"的顺序约束。
2. 冗余的内存屏障：
   reader中的atomic_thread_fence(memory_order_acquire)是多余的——seqcount_begin的读取已经使用memory_order_acquire，已确保后续的数据读取不会被重排到该操作之前。
3. 共享数据未做约束：
   struct latency的成员既非volatile也非原子类型，编译器可能对其读写做优化（如缓存寄存器值），导致读取到过期数据，seqcount的检查机制可能失效。

优化后的实现方案 #

方案1：C11 _Atomic 实现 #

该方案利用C11原子类型的内存顺序约束，确保同步逻辑的正确性：

#include <stdatomic.h>
#include <stdint.h>

struct latency
{
    uint64_t counter;
    uint64_t active;
    uint64_t minimal;
    uint64_t maximal;
    uint64_t sum;
};

struct timer
{
    // 定时器相关内容
    struct latency latency;
    _Atomic uint32_t latency_seqcount;
};

static struct timer *timer;

// 实时线程调用的写函数
static void writer(void)
{
    uint32_t seqcount = atomic_load_explicit(&timer->latency_seqcount, memory_order_relaxed);
    
    // 标记写开始（奇数）：release确保后续数据修改不会重排到该操作之前
    atomic_store_explicit(&timer->latency_seqcount, seqcount + 1, memory_order_release);

    struct latency *latency = &timer->latency;
    latency->active = latency_domain; // 假设latency_domain为已定义变量

    if (latency->counter == 0)
    {
        latency->maximal = latency->active;
        latency->minimal = latency->active;
    }
    else if (latency->active > latency->maximal)
    {
        latency->maximal = latency->active;
    }
    else if (latency->active < latency->minimal)
    {
        latency->minimal = latency->active;
    }

    latency->counter++;
    latency->sum += latency->active;

    // 标记写完成（偶数）：release确保所有数据修改完成后，该状态才会被其他线程看到
    atomic_store_explicit(&timer->latency_seqcount, seqcount + 2, memory_order_release);
}

// 非实时线程调用的读函数
static void reader(void)
{
    struct latency latency;
    uint32_t seqcount_begin;
    uint32_t seqcount_end;

    do
    {
        // acquire确保后续的数据读取不会重排到该操作之前
        seqcount_begin = atomic_load_explicit(&timer->latency_seqcount, memory_order_acquire);

        // 复制共享数据
        latency.counter = timer->latency.counter;
        latency.active = timer->latency.active;
        latency.minimal = timer->latency.minimal;
        latency.maximal = timer->latency.maximal;
        latency.sum = timer->latency.sum;

        // acquire确保数据读取完成后，才读取最新的seqcount
        seqcount_end = atomic_load_explicit(&timer->latency_seqcount, memory_order_acquire);
    }
    while (seqcount_begin != seqcount_end || (seqcount_end & 0x01));
    
    // 处理读取到的latency数据（如打印）
}

方案2：volatile+内存屏障实现（gcc/clang） #

针对不支持C11原子类型的场景，用volatile防止编译器优化，配合全内存屏障保证内存操作顺序：

#include <stdint.h>

struct latency
{
    volatile uint64_t counter;
    volatile uint64_t active;
    volatile uint64_t minimal;
    volatile uint64_t maximal;
    volatile uint64_t sum;
};

struct timer
{
    // 定时器相关内容
    struct latency latency;
    volatile uint32_t latency_seqcount;
};

static struct timer *timer;

// 实时线程调用的写函数
static void writer(void)
{
    uint32_t seqcount = timer->latency_seqcount;
    
    // 标记写开始（奇数）
    timer->latency_seqcount = seqcount + 1;
    // 全内存屏障：确保seqcount的写在数据修改前完成
    __sync_synchronize();

    struct latency *latency = &timer->latency;
    latency->active = latency_domain;

    if (latency->counter == 0)
    {
        latency->maximal = latency->active;
        latency->minimal = latency->active;
    }
    else if (latency->active > latency->maximal)
    {
        latency->maximal = latency->active;
    }
    else if (latency->active < latency->minimal)
    {
        latency->minimal = latency->active;
    }

    latency->counter++;
    latency->sum += latency->active;

    // 全内存屏障：确保所有数据修改完成后，才写入seqcount
    __sync_synchronize();
    // 标记写完成（偶数）
    timer->latency_seqcount = seqcount + 2;
}

// 非实时线程调用的读函数
static void reader(void)
{
    struct latency latency;
    uint32_t seqcount_begin;
    uint32_t seqcount_end;

    do
    {
        seqcount_begin = timer->latency_seqcount;
        // 全内存屏障：确保seqcount的读在数据读取前完成
        __sync_synchronize();

        latency.counter = timer->latency.counter;
        latency.active = timer->latency.active;
        latency.minimal = timer->latency.minimal;
        latency.maximal = timer->latency.maximal;
        latency.sum = timer->latency.sum;

        // 全内存屏障：确保数据读取完成后，才读取最新的seqcount
        __sync_synchronize();
        seqcount_end = timer->latency_seqcount;
    }
    while (seqcount_begin != seqcount_end || (seqcount_end & 0x01));
    
    // 处理读取到的latency数据
}

内容的提问来源于stack exchange，提问作者Markus Fuchs