Morris无饥饿互斥算法实现中的竞态条件排查求助

我几乎完全照搬《Little Book of Semaphores》第85页的Morris无饥饿互斥算法实现，但程序约半数时间能正常终止，另一半时间会中途冻结，说明存在竞态条件，但无法定位原因。查阅网上的Morris算法伪代码，和书中内容完全一致。猜测可能是C++的semaphore与研究领域或其他语言实现有差异，但不确定具体原因，请问是什么导致了这个竞态条件？

实现代码 #

#include <mutex>
#include <semaphore>
#include <thread>
#include <vector>
#include <iostream>
#include <chrono>
#include <syncstream>

std::binary_semaphore t1(1);
std::binary_semaphore t2(0);
std::mutex mutex;
int room1 = 0;
int room2 = 0;

void Morris(std::stop_token stop_token){
    while (!stop_token.stop_requested()){
        mutex.lock();
            room1 += 1;
        mutex.unlock();
        
        t1.acquire();
            room2 += 1;
            mutex.lock();
            room1 -= 1;
            if (room1 == 0){
                mutex.unlock();
                t2.release();
            } else {
                mutex.unlock();
                t1.release();
            }

        t2.acquire();
            room2 -= 1;

            // critical region
            {
                std::osyncstream synced_out(std::cout);
                synced_out << "critical region of thread " << std::hex << std::this_thread::get_id() << "\n";
            }
            // end critical region

            if (room2 == 0){
                t1.release();
            } else {
                t2.release();
            }
    }
}

int main(){
    std::vector<std::jthread> threads;
    for (int i = 0; i < 50; ++i){
        threads.emplace_back(Morris);
    }
    std::this_thread::sleep_for(std::chrono::seconds(5));
    for (std::jthread& thread: threads){
        thread.request_stop();
    }
}

设备与编译信息 #

compiler: g++12.3.0 -std=c++20 -lpthread -o
pc: Ubuntu 22.04.3 LTS, Intel Core i7-6700K, 4.00GHz x 8 (4 cores, 8 threads). processor is 64-bit x86

GDB回溯信息 #

(gdb) thread 11
[Switching to thread 11 (Thread 0x7fffde7fc640 (LWP 115600))]
#0  syscall () at ../sysdeps/unix/sysv/linux/x86_64/syscall.S:38
38  ../sysdeps/unix/sysv/linux/x86_64/syscall.S: No such file or directory.
(gdb) bt
#0  syscall () at ../sysdeps/unix/sysv/linux/x86_64/syscall.S:38
#1  0x0000555555558fd7 in std::__detail::__platform_wait<int> (__addr=0x5555555611a0 <t1>, __val=0) at /usr/include/c++/12/bits/atomic_wait.h:108
#2  0x000055555555906c in std::__atomic_wait_address_bare<std::__atomic_semaphore::_M_acquire()::{lambda()#1}>(int const*, std::__atomic_semaphore::_M_acquire()::{lambda()#1}) (
    __addr=0x5555555611a0 <t1>, __pred=...) at /usr/include/c++/12/bits/atomic_wait.h:444
#3  0x000055555555921c in std::__atomic_semaphore::_M_acquire (this=0x5555555611a0 <t1>) at /usr/include/c++/12/bits/semaphore_base.h:215
#4  std::counting_semaphore<1l>::acquire (this=0x5555555611a0 <t1>) at /usr/include/c++/12/semaphore:72
#5  0x0000555555557806 in Morris (stop_token=...) at /home/me/Desktop/nostarve/nostarve2.cpp:21
#6  0x000055555555be50 in std::__invoke_impl<void, void (*)(std::stop_token), std::stop_token> (__f=@0x555555574df0: 0x55555555779f <Morris(std::stop_token)>) at /usr/include/c++/12/bits/invoke.h:61
#7  0x000055555555bda1 in std::__invoke<void (*)(std::stop_token), std::stop_token> (__fn=@0x555555574df0: 0x55555555779f <Morris(std::stop_token)>) at /usr/include/c++/12/bits/invoke.h:96
#8  0x000055555555bd01 in std::thread::_Invoker<std::tuple<void (*)(std::stop_token), std::stop_token> >::_M_invoke<0ul, 1ul> (this=0x555555574de8) at /usr/include/c++/12/bits/std_thread.h:279
#9  0x000055555555bcb6 in std::thread::_Invoker<std::tuple<void (*)(std::stop_token), std::stop_token> >::operator() (this=0x555555574de8) at /usr/include/c++/12/bits/std_thread.h:286
#10 0x000055555555bb8a in std::thread::_State_impl<std::thread::_Invoker<std::tuple<void (*)(std::stop_token), std::stop_token> > >::_M_run (this=0x555555574de0)
    at /usr/include/c++/12/bits/std_thread.h:231
#11 0x00007ffff7cdc253 in ?? () from /lib/x86_64-linux-gnu/libstdc++.so.6
#12 0x00007ffff7894ac3 in start_thread (arg=<optimized out>) at ./nptl/pthread_create.c:442
#13 0x00007ffff7926660 in clone3 () at ../sysdeps/unix/sysv/linux/x86_64/clone3.S:81

已尝试的解决方法（均无效） #

• 将binary_semaphore替换为counting_semaphore
• 将room1和room2改为std::atomic<int>

问题原因与修复方案 #

核心问题：信号量类型不匹配+临界区后操作无保护 #

1. 信号量类型差异：《Little Book of Semaphores》中的信号量是通用计数信号量，支持累加计数；而C++的std::binary_semaphore是严格二元信号量，计数只能是0或1，多次release不会累加，会直接覆盖为1，导致本该传递的通知丢失，引发线程无法被唤醒的死锁。
2. 临界区后操作无互斥保护：退出临界区时修改room2和释放信号量的操作没有被互斥锁保护，导致room2的状态判断与信号量释放之间出现竞态，可能出现信号量释放时机错误，最终导致线程阻塞在t1.acquire()。

修复步骤 #

1. 替换为计数信号量：将t1和t2改为std::counting_semaphore<100>（上限设为大于线程数的数值，确保计数可以正常累加）。
2. 保护临界区后的状态修改：在退出临界区时，用互斥锁包裹room2的修改和信号量释放操作，保证这两个动作的原子性。

完整修复代码 #

#include <mutex>
#include <semaphore>
#include <thread>
#include <vector>
#include <iostream>
#include <chrono>
#include <syncstream>

std::counting_semaphore<100> t1(1);
std::counting_semaphore<100> t2(0);
std::mutex mutex;
int room1 = 0;
int room2 = 0;

void Morris(std::stop_token stop_token){
    while (!stop_token.stop_requested()){
        mutex.lock();
            room1 += 1;
        mutex.unlock();
        
        t1.acquire();
            mutex.lock();
            room2 += 1;
            room1 -= 1;
            if (room1 == 0){
                mutex.unlock();
                t2.release();
            } else {
                mutex.unlock();
                t1.release();
            }

        t2.acquire();
            // critical region
            {
                std::osyncstream synced_out(std::cout);
                synced_out << "critical region of thread " << std::hex << std::this_thread::get_id() << "\n";
            }
            // end critical region

            mutex.lock();
            room2 -= 1;
            if (room2 == 0){
                t1.release();
            } else {
                t2.release();
            }
            mutex.unlock();
    }
}

int main(){
    std::vector<std::jthread> threads;
    for (int i = 0; i < 50; ++i){
        threads.emplace_back(Morris);
    }
    std::this_thread::sleep_for(std::chrono::seconds(5));
    for (std::jthread& thread: threads){
        thread.request_stop();
    }
}

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