fork进程间Boost共享内存异常：重复取出元素问题排查

问题场景 #

我编写了一个基础程序，主进程通过fork创建子进程，子进程从共享列表中取出首个元素并打印。使用Boost库管理共享内存，但首次使用此类工具，可能存在错误。

原代码 #

#include <iostream>
#include <list>
#include <unistd.h>
#include <sys/wait.h>
#include <vector>
#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/mapped_region.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
using namespace std;
using namespace boost::interprocess;

struct shared_data{
    interprocess_mutex mtx;
    list<int> qqueue;
};

void modify_list(shared_data*& sdat);

int main(){
    shared_memory_object shm(open_or_create, "shm", read_write);
    shm.truncate(sizeof(shared_data));

    mapped_region reg(shm, read_write);
    void* address=reg.get_address();
    shared_data* shptr=new (address) shared_data;
    
    vector<pid_t> processors;
    size_t num_processors=4;
    
    for(int j=1; j<8; ++j){
        shptr->qqueue.push_back(j);
    }

    for(size_t i=0; i<num_processors; ++i){
        pid_t pid=fork();
        if(pid==0){
            modify_list(shptr);
        }
        else if(pid>0){
            processors.push_back(pid);
        }
    }

    for (pid_t cpu : processors) {
        int status;
        waitpid(cpu, &status, 0);
    }
}

void modify_list(shared_data*& sdat){
    while(true){
        sdat->mtx.lock();
        if(sdat->qqueue.empty()){
            sdat->mtx.unlock();
            break;
        }
        int ff = sdat->qqueue.front();
        sdat->qqueue.pop_front();
        sdat->mtx.unlock();
        cout<<"Process: "<<getpid()<<" dequeued "<<ff<<"\n";
        cout.flush();
    }
}

程序输出 #

Process: 320 dequeued 1
Process: 320 dequeued 2
Process: 320 dequeued 3
Process: 320 dequeued 4
Process: 320 dequeued 5
Process: 320 dequeued 6
Process: 320 dequeued 7
Process: 321 dequeued 2

疑问 #

元素2已被进程320取出，为何会被进程321再次取出，而列表中2仅存在一次？

问题根源 #

核心错误在于使用了标准库std::list，它的内部实现依赖进程私有堆内存，无法在多进程间共享：

• std::list的节点存储在进程自己的私有堆中，共享内存里的shared_data仅保存了指向这些私有节点的指针。
• fork创建子进程时，会复制父进程的地址空间，包括std::list的指针状态，但这些指针指向的是子进程自己私有堆里的链表副本，而非共享内存中的统一数据。
• 每个子进程操作的都是自己私有内存里的链表，导致各进程看到的链表状态不一致，出现重复读取同一元素的情况。

解决方案 #

必须使用Boost Interprocess提供的共享内存兼容容器，配合专用内存分配器，确保容器节点存储在共享内存区域：

修改后的代码示例 #

#include <iostream>
#include <unistd.h>
#include <sys/wait.h>
#include <vector>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/interprocess/containers/list.hpp>
#include <boost/interprocess/allocators/allocator.hpp>

using namespace std;
using namespace boost::interprocess;

// 定义共享内存分配器，用于分配链表节点
typedef allocator<int, managed_shared_memory::segment_manager> ShmemAllocator;
// 基于共享分配器的list
typedef list<int, ShmemAllocator> SharedList;

struct shared_data{
    interprocess_mutex mtx;
    SharedList qqueue;

    // 构造函数需要传入共享内存分配器
    shared_data(const ShmemAllocator& alloc) : qqueue(alloc) {}
};

void modify_list();

int main(){
    // 创建或打开共享内存段，指定足够的大小
    managed_shared_memory shm(open_or_create, "shm", 65536);

    // 初始化共享内存分配器
    ShmemAllocator alloc(shm.get_segment_manager());
    // 在共享内存中构造shared_data对象
    shared_data* shptr = shm.construct<shared_data>("shared_data")(alloc);
    
    // 填充共享列表
    for(int j=1; j<8; ++j){
        shptr->qqueue.push_back(j);
    }

    vector<pid_t> processors;
    size_t num_processors=4;

    for(size_t i=0; i<num_processors; ++i){
        pid_t pid=fork();
        if(pid==0){
            modify_list();
            exit(0); // 子进程完成后退出
        }
        else if(pid>0){
            processors.push_back(pid);
        }
    }

    // 等待所有子进程结束
    for (pid_t cpu : processors) {
        int status;
        waitpid(cpu, &status, 0);
    }

    // 清理共享内存（可选，根据需求决定是否保留）
    shm.destroy<shared_data>("shared_data");
    shared_memory_object::remove("shm");
}

void modify_list(){
    // 子进程重新打开共享内存段
    managed_shared_memory shm(open_only, "shm");
    // 获取共享数据的指针
    shared_data* sdat = shm.find<shared_data>("shared_data").first;

    while(true){
        sdat->mtx.lock();
        if(sdat->qqueue.empty()){
            sdat->mtx.unlock();
            break;
        }
        int ff = sdat->qqueue.front();
        sdat->qqueue.pop_front();
        sdat->mtx.unlock();
        cout<<"Process: "<<getpid()<<" dequeued "<<ff<<endl;
        cout.flush();
    }
}

关键修改点 #

1. 替换std::list为boost::interprocess::list，配合ShmemAllocator确保节点分配在共享内存。
2. 使用managed_shared_memory替代原始shared_memory_object，它提供内存管理能力，支持在共享内存中构造对象。
3. 子进程重新打开共享内存段并获取数据指针，避免依赖fork后的地址空间副本。
4. 子进程完成后调用exit(0)，确保正确退出。

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