C++进程间管道传输含vector的struct时vector值为0的问题

问题描述 #

通过write()函数经管道向另一个进程发送包含std::vector的结构体，接收进程读取vector时得到的都是零值。

场景为改进Peterson-Kearns协议：

• 发送进程发送消息前，会递增vector中自身索引对应的值
• 接收进程收到消息后，递增自身索引的值，并将其他索引的值更新为接收值与本地存储值的最大值

结构体定义 #

struct ptp_msg
{
    char msg_buf[50];
    int sending_process_nr;
};

struct ptp_err
{
    pid_t pid;
    int sending_process_nr;
    int fildes[2];
};

struct msg_t
{
    message_type msg_type;
    vector<int> time_v;
    vector<int> fail_v;
    union contents
    {
        struct ptp_msg ptp_msg;
        struct ptp_err ptp_err;
    } contents;
};

发送与接收代码 #

发送函数 #

int Pet_kea::send_msg(char *input, int fildes[2])
{
    // inc T^i
    time_v[id]++;

    struct msg_t msg;

    msg.msg_type = MSG;
    sprintf(msg.contents.ptp_msg.msg_buf, input);
    msg.time_v = time_v;
    msg.contents.ptp_msg.sending_process_nr = id;
    msg.fail_v = fail_v;

    // send the message
    cout << id << " vector sent =" << msg.time_v[0] << "-" << msg.time_v[1] << endl;

    if (write(fildes[1], &msg, sizeof(msg)) < 0)
    {
        // handle error
    }

    store_msg(&msg, false);

    return 0;
}

接收函数 #

int Pet_kea::recv_msg(int fildes[2], struct msg_t *ret_buf)
{
    // read message
    int ret;
    ret = read(fildes[0], ret_buf, sizeof(msg_t));
    // do err checking
    if (ret < 0)
    {
    }
    cout << id << " vector recieved =" << ret_buf->time_v[0] << "-" << ret_buf->time_v[1] << endl;

    // check if it is an errmsg

    // inc T^i and inc T^/j to max(T^j of send event, prev event T^j)
    time_v[id]++;
    for (int i = 0; i < (int)time_v.size(); i++)
    {
        if (i == id)
            continue;
        time_v.at(i) = max(ret_buf->time_v[i], time_v[i]);
    }

    store_msg(ret_buf, true);

    return 0;
}

测试输出 #

发送前与接收后的输出不一致：

0 vector sent =1-0
1 vector recieved =0-0

问题原因与解决方案 #

核心原因 #

std::vector是动态容器，内部包含指向堆内存的指针、大小、容量等成员变量。直接用write()发送结构体的内存镜像时，只会发送这些指针和元数据，不会传输堆上的实际元素数据。接收端读取后，指针指向的是自身进程的无效内存地址，访问时行为未定义，表现为读取到零值。

解决方法 #

1. 手动序列化传输

将vector的内容拆分为“大小+元素”的形式逐个发送，接收端按对应格式读取并重建vector：

修改后的发送函数：

int Pet_kea::send_msg(char *input, int fildes[2])
{
    time_v[id]++;

    // 定义消息头，传输固定长度的元数据
    struct msg_header {
        message_type msg_type;
        int sending_process_nr;
        size_t time_v_size;
        size_t fail_v_size;
        char msg_buf[50];
    } header;

    header.msg_type = MSG;
    header.sending_process_nr = id;
    header.time_v_size = time_v.size();
    header.fail_v_size = fail_v.size();
    sprintf(header.msg_buf, input);

    // 发送消息头
    if (write(fildes[1], &header, sizeof(header)) < 0) {
        // 处理错误
    }

    // 发送time_v的所有元素
    if (write(fildes[1], time_v.data(), time_v.size() * sizeof(int)) < 0) {
        // 处理错误
    }

    // 发送fail_v的所有元素
    if (write(fildes[1], fail_v.data(), fail_v.size() * sizeof(int)) < 0) {
        // 处理错误
    }

    cout << id << " vector sent =" << time_v[0] << "-" << time_v[1] << endl;

    // 构造msg_t用于本地存储
    struct msg_t msg;
    msg.msg_type = MSG;
    strcpy(msg.contents.ptp_msg.msg_buf, input);
    msg.time_v = time_v;
    msg.contents.ptp_msg.sending_process_nr = id;
    msg.fail_v = fail_v;
    store_msg(&msg, false);

    return 0;
}

修改后的接收函数：

int Pet_kea::recv_msg(int fildes[2], struct msg_t *ret_buf)
{
    struct msg_header {
        message_type msg_type;
        int sending_process_nr;
        size_t time_v_size;
        size_t fail_v_size;
        char msg_buf[50];
    } header;

    // 读取消息头
    int ret = read(fildes[0], &header, sizeof(header));
    if (ret < 0) {
        // 处理错误
    }

    // 读取并重建time_v
    ret_buf->time_v.resize(header.time_v_size);
    ret = read(fildes[0], ret_buf->time_v.data(), header.time_v_size * sizeof(int));
    if (ret < 0) {
        // 处理错误
    }

    // 读取并重建fail_v
    ret_buf->fail_v.resize(header.fail_v_size);
    ret = read(fildes[0], ret_buf->fail_v.data(), header.fail_v_size * sizeof(int));
    if (ret < 0) {
        // 处理错误
    }

    // 填充其他字段
    ret_buf->msg_type = header.msg_type;
    strcpy(ret_buf->contents.ptp_msg.msg_buf, header.msg_buf);
    ret_buf->contents.ptp_msg.sending_process_nr = header.sending_process_nr;

    cout << id << " vector recieved =" << ret_buf->time_v[0] << "-" << ret_buf->time_v[1] << endl;

    // 更新本地time_v
    time_v[id]++;
    for (int i = 0; i < (int)time_v.size(); i++)
    {
        if (i == id)
            continue;
        time_v.at(i) = max(ret_buf->time_v[i], time_v[i]);
    }

    store_msg(ret_buf, true);

    return 0;
}

2. 替换为静态数组

如果vector的大小是固定的（比如进程数量固定），可以将vector<int>替换为固定大小的数组（如int time_v[MAX_PROCESSES]），这样结构体可以直接通过内存拷贝传输，无需序列化。

3. 使用序列化库

对于复杂数据结构，可使用Protocol Buffers、FlatBuffers等成熟的序列化库，自动处理容器的序列化与反序列化，减少手动编码的工作量。

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