基于boost::fibers的简单生产者-消费者程序异常原因及修复

问题描述 #

我们有一套Producer-Consumer架构的简化程序：Producer每秒生成一个int类型数据，通过boost::signals2::signal通知Consumer；Consumer收到通知后把值推入自己的boost::fibers::unbuffered_channel，同时Consumer的纤程负责从通道中取出值。但实际运行时，Consumer始终阻塞在pop()操作上，程序最终因长时间无响应被终止。

原因分析 #

• 无缓冲通道的阻塞特性：boost::fibers::unbuffered_channel是无缓冲通道，push()操作会一直阻塞，直到有对应的pop()操作接收数据。
• 信号槽的执行上下文冲突：Producer的纤程调用sig(i)时，信号的槽函数（Consumer的operator()）是在Producer的纤程上下文中执行的。也就是说，push()操作是在Producer纤程里运行的，而此时Consumer的纤程正阻塞在pop()上——两边互相等待，形成死锁。
• 主线程过早退出：main函数执行完p.start()后直接返回，主线程退出会导致整个进程终止，后台纤程还没来得及完成正常逻辑。
• Consumer只执行一次pop：Producer会发送10个数据，但Consumer的start()函数只调用了一次pop()，即使第一个数据处理完成，后续的数据也会导致Producer的push()阻塞。

修复方案（保留原架构） #

• 将unbuffered_channel替换为buffered_channel，设置最小缓冲（比如1），避免push()因等待pop()而阻塞；
• 修改Consumer的start()函数为循环pop()，处理Producer发送的所有10个数据；
• 在main函数中等待所有纤程执行完成，避免进程过早终止；

修复后的代码 #

#include <boost/fiber/all.hpp>
#include <boost/signals2/signal.hpp>
#include <chrono>
#include <iostream>

using channel_t = boost::fibers::buffered_channel<int>;
namespace fibers = boost::fibers;

struct Producer {
    boost::signals2::signal<void(int)> sig;
    void notify() {
        std::cout << "enter Producer::notify()\n";
        for (int i = 0; i < 10; i++ ) {
            sig(i);
            std::this_thread::sleep_for(std::chrono::seconds{1});
        }
        std::cout << "complete Producer::notify()\n";
    }
    void start() {
        std::cout << "enter Producer::start()\n";
        fibers::fiber(fibers::launch::dispatch, &Producer::notify, this).detach();
        std::cout << "complete Producer::start()\n";
    }        
};

struct Consumer {
    channel_t chan{1}; // 设置缓冲大小为1
    fibers::fiber consumer_fiber;

    Consumer() {
        std::cout << "enter Consumer()\n";
        consumer_fiber = fibers::fiber(fibers::launch::dispatch, &Consumer::start, this);
        std::cout << "complete Consumer()\n";
    }

    ~Consumer() {
        if (consumer_fiber.joinable()) {
            consumer_fiber.join();
        }
    }

    void start() {
        std::cout << "enter Consumer::start()\n";
        int i;
        // 循环pop直到通道关闭
        while (chan.pop(i) == fibers::channel_op_status::success) {
            std::cout << "Consumer received: " << i << "\n";
        }
        std::cout << "complete Consumer::start()\n";
    }

    void operator()(int i) {
       std::cout << "enter Consumer::operator()\n";
       // 推送数据，缓冲通道不会阻塞（只要缓冲未满）
       auto status = chan.push(i);
       if (status != fibers::channel_op_status::success) {
           std::cout << "Push failed: " << static_cast<int>(status) << "\n";
       }
       std::cout << "complete Consumer::operator()\n";
    }    
};

int main() {
    Consumer c;
    Producer p;

    p.sig.connect(std::ref(c));
    p.start();

    // 等待Producer发送完所有数据后关闭通道
    std::this_thread::sleep_for(std::chrono::seconds{11});
    c.chan.close();

    std::cout << "done." << std::endl;
    return EXIT_SUCCESS;
}

修复说明 #

1. 缓冲通道：buffered_channel<int>{1}设置了大小为1的缓冲，Producer的push()操作可以立即返回，无需等待Consumer的pop()，直接避免死锁。
2. 循环pop：Consumer的start()函数循环调用pop()，直到通道关闭，确保处理Producer发送的全部10个数据。
3. 等待与通道关闭：主线程等待Producer发送完所有数据后关闭通道，让Consumer纤程能正常退出，避免进程过早终止。
4. 资源清理：Consumer的析构函数等待纤程完成，避免资源泄漏。

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