关于内存排序模型理解的续问及代码实现相关疑问

疑问1：设置“无更多bucket”标志时的CAS使用是否有误？ #

是的，这里确实用错了工具。你当前的CAS循环需要每次重置empty值，是因为atomic_compare_exchange_weak失败时会把empty更新为bucket_full的当前值，导致下一次循环的预期值错误。

更合理的做法是：生产者在发送完所有bucket后，先自旋等待bucket_full变为0（表示消费者处理完最后一个bucket），然后直接用atomic_store_explicit将其设置为-1即可。因为此时消费者只会把bucket_full从1改为0，不会有其他线程修改这个值，不需要CAS来做条件更新。修改后的代码片段如下：

// 替换原来的CAS循环
while(atomic_load_explicit(&bucket_full, memory_order_acquire))
    sched_yield();
atomic_store_explicit(&bucket_full, -1, memory_order_release);

疑问2：是否确实实现了并行工作？ #

是的，代码已经实现了生产者和消费者的并行工作：

• 生产者填充完一个bucket并设置bucket_full=1后，会进入等待状态，这段时间它可以提前准备下一个bucket的units（执行important_stuff(15)的循环）。
• 消费者在处理当前bucket的units时（执行important_stuff(20)的循环），生产者已经在并行准备下一批数据。
  你可以通过调整important_stuff的延迟参数，观察输出的时间间隔，或者用性能分析工具验证两个线程的CPU时间重叠情况，就能直观看到并行效果。

疑问3：代码是否不存在数据竞争？ #

代码不存在数据竞争，核心依据如下：

• 非原子变量bucket的访问完全由bucket_full的原子操作同步：生产者仅在bucket_full=0时写入bucket，消费者仅在bucket_full=1时读取/修改bucket，acquire/release语义保证了两者的访问区间完全互斥，不会出现同时读写的情况。
• 所有对bucket_full的操作都是标准C11原子操作，不存在原子性问题。
• 你使用了-fsanitize=thread编译选项，若运行时无报错，也能直接佐证代码无数据竞争——ThreadSanitizer会精准检测所有数据竞争场景。

完整代码 #

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <stdatomic.h>
#include <pthread.h>

#define tot_buckets 10
#define units_per_bucket 12

static int bucket = 0;
static _Atomic int bucket_full = 0;

// 模拟工作延迟的函数，h取值范围{0,99}
static inline void important_stuff(int h){
    struct timespec t = {0,h*10000000};
    nanosleep(&t,NULL);
}

void *producer(void *arg){
    int units_cumulator = 0;
    for(int i=0; i<tot_buckets; ++i){
        // 生成下一个bucket的单元（并行工作阶段）
        while(units_cumulator < units_per_bucket){
            important_stuff(15);
            units_cumulator += 1;
        }
        // 等待bucket空闲信号
        while(atomic_load_explicit(&bucket_full, memory_order_acquire))
            sched_yield();
        // 填充bucket
        bucket = units_cumulator;
        units_cumulator = 0;
        printf("\n[b%-2d:%2d]", i+1, bucket);
        fflush(stdout);
        // 发送bucket已填满信号
        atomic_store_explicit(&bucket_full, 1, memory_order_release);
    }
    // 发送无更多bucket的信号（优化后写法）
    while(atomic_load_explicit(&bucket_full, memory_order_acquire))
        sched_yield();
    atomic_store_explicit(&bucket_full, -1, memory_order_release);
    return arg;
}

void *consumer(void *arg){
    int tmp_bf;
    while(1){
        // 等待bucket已填满信号
        while(!(tmp_bf=atomic_load_explicit(&bucket_full, memory_order_acquire)))
            sched_yield();
         // 无更多bucket则退出
        if(tmp_bf == -1)
            break;
        // 处理当前bucket的单元（并行工作阶段）
        while(bucket--){
            important_stuff(20);
            printf(" %d", bucket);
            fflush(stdout);
        }
        // 发送bucket已清空信号
        atomic_store_explicit(&bucket_full, 0, memory_order_release);
    }
    return arg;
}

int main(){
    pthread_t t0, t1;
    pthread_create(&t0, NULL, consumer, NULL);
    pthread_create(&t1, NULL, producer, NULL);
    pthread_join(t0, NULL);
    pthread_join(t1, NULL);
    printf("\n\n");
    return 0;
}

编译命令 #

gcc -std=c11 -pthread -O1 -g -fsanitize=thread -fno-omit-frame-pointer -Wall -Wextra -Werror -pedantic -pedantic-errors main.c -o main

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