现代多处理器系统下无锁环形队列的线程饥饿问题

我当前正在实现生产者-消费者模型。常规实现方案中，需要结合std::condition_variable与std::mutex两类同步原语实现线程唤醒逻辑：生产者可通过该机制通知消费者唤醒，消费者也可反向通知生产者唤醒，这类唤醒机制通常被用于规避线程饥饿问题。
此处存在核心疑问：线程饥饿问题在多处理器系统中是否仍然真实存在？

实现背景 #

我正基于无锁环形缓冲区实现上述生产者-消费者模型，目标是在生产者、消费者两侧完全不使用std::mutex与std::condition_variable；所用队列的enqueue()、dequeue()接口不存在数据竞争问题，完整实现代码如下：

template<typename MessageType>
class MessageProcessor
{
public:
    ~MessageProcessor()
    {
        stop();

        if (workerThread_.joinable())
            workerThread_.join();
    }

    bool postMessage(MessageType const &msg)
    {
        return queue_.enqueue(msg);
    }

    void registerHandler(std::function<void(MessageType)> handler, int32_t coreId=-1, std::string_view const &name="")
    {
        std::call_once(init_, [&](){
            handler_ = std::move(handler);
            workerThread_ = std::thread{&MessageProcessor::process, this};

            if (!setAffinity(coreId, workerThread_))
                LOG("Msg Processing thread couldn't be pinned to core: " << coreId);
            else
                LOG("Msg Processing thread pinned to core: " << coreId);

            if (! name.empty())
                pthread_setname_np(workerThread_.native_handle(), name.data());
        });
    }

    void stop()
    {
        stop_ = true;
    }

private:
    void process() // 消费者逻辑，运行在独立线程
    {
        while(!stop_.load(std::memory_order_acquire))
        {
            MessageType msg;

            if (! queue_.dequeue(msg))
                continue;

            try
            {
                handler_(msg);
            }
            catch(std::exception const &ex)
            {
                LOG("Error while processing data: " << msg << ", Exception: " << ex.what());
            }
            catch(...)
            {
                LOG("UNKOWN Error while processing data: " << msg);
            }
        }
    }

    bool setAffinity(int32_t const coreId, std::thread &thread)
    {
        int cpuCoreCount = __sysconf(_GLIBCXX_USE_SC_NPROCESSORS_ONLN);

        if (coreId < 0 || coreId >= cpuCoreCount)
            return false;

        cpu_set_t cpuset;

        CPU_ZERO(&cpuset);
        CPU_SET(coreId, &cpuset);

        pthread_t currentThread = thread.native_handle();

        return pthread_setaffinity_np(currentThread, sizeof(cpu_set_t), &cpuset) == 0;
    }

    std::thread workerThread_;
    std::atomic<bool> stop_{false};
    MPMC_Circular_Queue<MessageType, 1024> queue_;
    std::function<void(MessageType)> handler_{};
    std::once_flag init_;
};

int main()
{
    pthread_setname_np(pthread_self(), "MAIN");

    MessageProcessor<int> processor;

    processor.registerHandler([](int i){
        LOG("Received value: " << i);
    }, 2, "PROCESSOR");

    std::thread t1([&]() { // 生产者线程1
        for (int i = 1; i <= 100000; i += 2)
        {
            LOG("Submitting value: " << i);
            processor.postMessage(i);
        }
    });

    pthread_setname_np(t1.native_handle(), "ODD ");

    std::thread t2([&]() { // 生产者线程2
        for (int i = 2; i <= 100000; i += 2)
        {
            LOG("Submitting value: " << i);
            processor.postMessage(i);
        }
    });

    pthread_setname_np(t2.native_handle(), "EVEN");

    for (int i = 1; i <= 100000; ++i)
    {
        LOG("Runing main thread: " << i);
    }

    t1.join();
    t2.join();

    return 0;
}

核心问题 #

上述基于有界无锁MPMC_Circular_Queue实现的消息处理逻辑，在现代多处理器系统中是否会引发线程饥饿问题？

直接给结论：哪怕是现代多处理器系统，线程饥饿也绝对是真实存在的问题，你现在写的这套纯自旋无锁实现不仅有饥饿风险，还有好几个其他的生产环境隐患，和CPU核心数量多少没有本质关系。

先理清几个最容易搞错的认知 #

• 别觉得核心多就所有线程都能同时跑：任何正常运行的系统，总就绪线程数远大于CPU核心数是常态，操作系统调度器是按时间片、线程优先级来分配CPU资源的。就算你给消费者绑了核，要是这个核上跑了个更高优先级的实时线程把资源占满，或者整台机器负载打满了，你的消费者线程照样可能几十毫秒甚至更久拿不到执行时间片。
• 你现在的消费者是死循环空转：队列空的时候直接continue进下一轮，一点不让出CPU，会硬生生占满一个核的全部算力，哪怕半条消息都没有也白烧电。要是你后续给生产者也加上入队失败重试的空转逻辑，一堆空转线程抢CPU，只会把调度压力拉得更高，饥饿问题只会更严重。
• 无锁只解决死锁和数据竞争，管不了调度层面的饥饿：无锁结构的进度保证是“整个系统整体能往前走”，根本不保证每个单独的线程能在有限时间内拿到执行权。举个真实生产环境踩过的例子：生产者一直不停发消息，消费者每次刚要执行出队操作就被调度切走，等再切回来队列又被新消息占满；或者反过来消费者一直占着核空转，生产者根本抢不到时间片入队，这些都不是理论上的极端情况。

你这段代码里现成的问题 #

1. 空转的资源浪费完全不可接受：队列空的时候没有任何停顿、让出CPU的操作，哪怕加个_mm_pause()或者短时间自旋后yield都比现在强，现在的写法空载就能占满一个核，嵌入式、云原生这种按算力/功耗算成本的场景根本用不了。
2. 入队失败直接丢消息：你的环形队列深度只有1024，只要生产者速度比消费者处理速度快一点，队列瞬间就满，postMessage直接返回false，你测试代码里根本没处理这个返回值，跑起来大概率丢一半以上的消息，这和饥不饥饿没关系，是逻辑本身就有缺失。
3. 硬绑核反而放大饥饿风险：你把消费者钉死在固定核心上，万一这个核被其他高优先级任务占了，消费者连被调度到其他空闲核心的机会都没有，延迟直接炸。

实际开发的建议 #

• 真要坚持无锁方案，别纯自旋：空队列/满队列的时候先自旋个几十上百轮（x86平台加_mm_pause()能降功耗、提性能），还拿不到数据/空位就主动调用std::this_thread::yield()把CPU让出去，别死占着核。
• 不是极致性能要求别硬上无锁自旋：std::mutex+std::condition_variable的组合没你想的那么慢，没消息的时候线程完全休眠不占资源，新消息来了才唤醒，不管是功耗还是平均延迟都比纯自旋好，只有吞吐量到百万级、延迟要求亚微秒的场景才值得为自旋付出代价。
• 别迷信多核心调度：调度器不是神仙，高负载下优先级反转、时间片分配不均导致的饥饿太常见了，和是不是多CPU系统半毛钱关系都没有。

内容的提问来源于stack exchange，提问作者Daya Shanker Prasad