基于《多处理器编程艺术》实现MCS锁时程序卡死求助

我参考《多处理器编程艺术》实现了MCS锁，在Ubuntu系统运行时出现卡死问题。代码逻辑和书中一致，但只有用-O0无优化编译时程序能正常运行，开启优化编译（比如-O1/-O2）就会卡死。以下是我的C语言实现代码：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdatomic.h>

#define THREADS 2

pthread_t thread_t[THREADS];

struct Lock
{
    struct Lock* next;
    bool locked;
};

_Atomic(struct Lock*) tail;

_Thread_local struct Lock* my_lock;

void lock_()
{
    struct Lock* lock = (struct Lock*)malloc(sizeof(struct Lock));

    struct Lock* pred = atomic_exchange(&tail, lock);

    my_lock = lock;

    if (pred)
    {
        lock->locked = true;
        pred->next = lock;
        __sync_synchronize();
        while (lock->locked) {
        }
    }
}

void unlock_()
{
    struct Lock* succ = my_lock->next;

    if (succ == NULL)
    {
        struct Lock* expected = my_lock;
        if (atomic_compare_exchange_strong(&tail, &expected, NULL))
        {
            return;
        }

        while (succ == NULL) {
            succ = my_lock->next;
        }
    }

    succ->locked = false;
}

void* lock_func()
{
    int j = 0;
    for (int i = 0; i < 10000; ++i)
    {
        lock_();
        j++;
        unlock_();
    }
}

int main(int argc, char *argv[])
{
    int i;

    for (i = 0; i < THREADS; ++i)
    {
        if (pthread_create(&thread_t[i], NULL, lock_func, NULL) < 0)
        {
            perror("thread create error:");
            exit(0);
        }
    }

    for (i = 0; i < THREADS; ++i)
        pthread_join(thread_t[i], NULL);
}

问题原因分析 #

优化编译时卡死的核心原因是普通字段的读写缺乏线程可见性保证，编译器会做如下优化导致逻辑错误：

1. lock->locked是普通bool类型，while (lock->locked)循环会被编译器优化成死循环——编译器认为没有其他线程会修改这个值，直接将其缓存到寄存器，不再从内存读取最新值。
2. pred->next = lock和lock->locked = true的操作顺序，在没有内存屏障约束时，编译器可能重排序，导致前驱线程还没设置好next，当前线程就进入等待，后续解锁时找不到后继节点。
3. unlock_中读取my_lock->next同样没有可见性保证，优化后可能一直读取到NULL，陷入无限等待。

修正后的代码 #

将struct Lock中的字段改为原子类型，同时使用原子操作保证读写的可见性和顺序性：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdatomic.h>

#define THREADS 2

pthread_t thread_t[THREADS];

struct Lock
{
    _Atomic(struct Lock*) next;
    _Atomic(bool) locked;
};

_Atomic(struct Lock*) tail;

_Thread_local struct Lock* my_lock;

void lock_()
{
    struct Lock* lock = (struct Lock*)malloc(sizeof(struct Lock));
    // 初始化节点状态
    atomic_init(&lock->next, NULL);
    atomic_init(&lock->locked, false);

    struct Lock* pred = atomic_exchange(&tail, lock);
    my_lock = lock;

    if (pred)
    {
        atomic_store(&lock->locked, true);
        // 确保locked设置完成后再设置pred的next，避免重排序
        atomic_thread_fence(memory_order_release);
        atomic_store(&pred->next, lock);
        // 循环等待时原子加载，保证读取最新值
        while (atomic_load(&lock->locked)) {
            // 加入pause指令减少空耗
            __builtin_ia32_pause();
        }
    }
    // 没有前驱时，当前节点直接获得锁，locked保持false
}

void unlock_()
{
    struct Lock* succ = atomic_load(&my_lock->next);

    if (succ == NULL)
    {
        struct Lock* expected = my_lock;
        // CAS尝试将tail置为NULL，成功说明没有等待线程
        if (atomic_compare_exchange_strong(&tail, &expected, NULL))
        {
            free(my_lock); // 释放当前节点
            return;
        }
        // 失败说明有线程已经加入等待，循环等待后继节点出现
        while ((succ = atomic_load(&my_lock->next)) == NULL) {
            __builtin_ia32_pause();
        }
    }
    // 解锁后继节点
    atomic_store(&succ->locked, false);
    free(my_lock); // 释放当前节点
}

void* lock_func()
{
    int j = 0;
    for (int i = 0; i < 10000; ++i)
    {
        lock_();
        j++;
        unlock_();
    }
    return NULL;
}

int main(int argc, char *argv[])
{
    int i;
    atomic_init(&tail, NULL);

    for (i = 0; i < THREADS; ++i)
    {
        if (pthread_create(&thread_t[i], NULL, lock_func, NULL) < 0)
        {
            perror("thread create error:");
            exit(EXIT_FAILURE);
        }
    }

    for (i = 0; i < THREADS; ++i)
        pthread_join(thread_t[i], NULL);

    return 0;
}

关键修正点 #

1. 原子字段声明：将struct Lock的next和locked改为_Atomic类型，确保线程间的读写可见性。
2. 内存屏障与顺序约束：使用atomic_thread_fence(memory_order_release)保证locked赋值完成后再设置前驱的next，避免编译器重排序。
3. 原子加载等待：循环等待时使用atomic_load读取locked，强制从内存获取最新值，避免编译器优化成死循环。
4. 节点内存管理：在解锁时释放当前节点，避免内存泄漏（原代码未释放内存）。
5. 初始化原子变量：显式初始化原子字段和全局tail，保证初始状态正确。

内容的提问来源于stack exchange，提问作者Kim Ki Hwan