通过命名管道无法正确接收nlohmann::json序列化的结构体

我用C++编写了两个进程，需要实现以下功能：

1. 服务端进程接收CAN BUS的信号信息，封装为struct后通过nlohmann::json序列化，再将序列化后的std::vector<uint8_t>字节流通过UNIX命名管道发送给客户端进程。
2. 客户端进程从管道读取序列化数据，反序列化后解析结构体内容。

相关代码 #

服务端代码 #

// 创建客户端要接收的信号信息，目前仅测试数值数据
sigHandlerLib::signalInfo sigToClient;
sigToClient.signalValue = value;

// 序列化内容
nlohmann::json json = sigToClient;
std::vector<uint8_t> serializedBuffer = nlohmann::json::to_bson(json);
std::size_t serializedBufferSize = sizeof(serializedBuffer);

std::cout << "Serialized buffer size: " << serializedBufferSize << '\n';
std::cout << "Buffer content size: " << serializedBuffer.size() << '\n';

...

// 通知客户端
const auto iter = std::find_if(clientSignalNotifications.begin(), clientSignalNotifications.end(), [&updatedSignal](const auto& iter) { return (iter.first.first.compare(updatedSignal) == 0); });
if(iter != clientSignalNotifications.end())
{
    for(const auto& client : iter->second)
    {
        writeToIpcNamedPipe(client, serializedBuffer, serializedBufferSize);
    }
}
else
{
    std::cout << "[Signal Handler Service]: No client subscribed for this signal, no notification is sent" << '\n';
}

写入管道函数 #

const std::string ipcLocationPath = "/tmp/" + processName + ".ipc";

int ipcFileDescriptor = open(ipcLocationPath.c_str(), O_WRONLY | O_NONBLOCK);
if(ipcFileDescriptor != -1 && ipcFileDescriptor != EEXIST)
{
    int nbytes = 0;
    nbytes = write(ipcFileDescriptor, &buffer, bufferSize);
    if(nbytes == -1)
    {
        std::cout << "[Signal Handler Service] Cannot write to IPC file data, error code = " << errno << '\n';
    }
    else
    {
        std::cout << "[Signal Handler Service] Data sent to client " << processName << '\n';
        std::cout << "[Signal Handler Service] Bytes written to IPC Pipe: " << nbytes << '\n';
        std::cout << "[Signal Handler Service] Number of the elements in the buffer " << buffer.size() << '\n';

        std::this_thread::sleep_for(std::chrono::milliseconds(150));
    }
}
else
{
    std::cout << "[Signal Handler Service] Cannot open IPC pipe for writing, error code = " << errno << '\n';
}

close(ipcFileDescriptor);

客户端代码（读取管道并反序列化逻辑） #

std::vector<uint8_t> deserializedBytes;

pollfd readPoll;

timeval selectTimeout;
selectTimeout.tv_sec = 2;
selectTimeout.tv_usec = 0;

while(1)
{
    memset(&readPoll, 0, sizeof(readPoll));
    readPoll.fd = ipcDescriptor;
    readPoll.events = POLLIN;

    std::thread::id this_id = std::this_thread::get_id();

    int pollReturn = 0;
    pollReturn = poll(&readPoll, 1, 2000);
    if(pollReturn > 0)
    {
        size_t nbytes = 0;
        nbytes = read(ipcDescriptor, deserializedBytes.data(), deserializedBytes.size());
        if(nbytes <= 0)
        {
             // 读取失败日志被注释
        }
        else                        
        {  
            std::cout << "Buffer received with size: " << nbytes << '\n';
            std::cout << "Buffer content size: " << deserializedBytes.size() << '\n';

            // 反序列化内容
            auto deserializedJson = nlohmann::json::from_bson(deserializedBytes);

            if(deserializedJson.size() != 0)
            {
                sigHandlerLib::signalInfo recvSignal = deserializedJson.get<sigHandlerLib::signalInfo>();

                std::cout << "[Signal Handler Library] Signal value: " << recvSignal.signalValue << '\n';    
            }
            else
            {
                std::cout << "[Signal Handler Library] There was an error at data deserialization!" << '\n';
            }       
        }        
    }
    else
    {
        // Poll错误日志被注释
    }
}

问题 #

客户端进程崩溃，日志显示调用read函数后deserializedBytes.size()变为极大值，管道读取的字节流内容异常，请问哪里出错了？

1. 服务端：错误计算序列化数据长度 #

你用sizeof(serializedBuffer)获取序列化字节流的大小，这完全错误。sizeof(std::vector)返回的是容器对象本身的内存占用（比如包含指针、容量、长度的结构体大小），不是容器内存储的数据长度。正确写法是用serializedBuffer.size()：

// 替换错误代码
std::size_t serializedBufferSize = sizeof(serializedBuffer);
// 改为
std::size_t serializedBufferSize = serializedBuffer.size();

2. 服务端：写入管道时传递错误的缓冲区地址 #

写入管道时用&buffer作为数据起始地址，但buffer是std::vector<uint8_t>类型，&buffer取的是容器对象的地址，而非内部存储数据的地址。必须用buffer.data()获取数据的起始指针：

// 替换错误代码
nbytes = write(ipcFileDescriptor, &buffer, bufferSize);
// 改为
nbytes = write(ipcFileDescriptor, buffer.data(), bufferSize);

另外，open函数的错误判断逻辑错误：ipcFileDescriptor != EEXIST是无效的，open返回的是文件描述符（整数），EEXIST是错误码，只有当open返回-1时，才需要检查errno是否为EEXIST。正确判断：

int ipcFileDescriptor = open(ipcLocationPath.c_str(), O_WRONLY | O_NONBLOCK);
if(ipcFileDescriptor != -1)
{
    // 写入逻辑
}
else
{
    if(errno == EEXIST)
    {
        std::cout << "[Signal Handler Service] IPC pipe already exists" << '\n';
    }
    else
    {
        std::cout << "[Signal Handler Service] Cannot open IPC pipe for writing, error code = " << errno << '\n';
    }
}

3. 客户端：读取前未初始化缓冲区大小 #

客户端的deserializedBytes是一个空的std::vector，此时deserializedBytes.data()返回空指针，deserializedBytes.size()为0，调用read会触发未定义行为（写入空指针指向的内存，或读取0字节）。

推荐采用「先传数据长度，再传实际数据」的方案，保证完整读取：

服务端修改：先发送数据长度，再发送数据 #

// 先发送数据长度（用uint32_t保证跨平台一致性）
uint32_t dataSize = static_cast<uint32_t>(serializedBuffer.size());
write(ipcFileDescriptor, &dataSize, sizeof(dataSize));
// 再发送实际数据
write(ipcFileDescriptor, serializedBuffer.data(), dataSize);

客户端修改：先读取长度，再读取数据 #

while(1)
{
    memset(&readPoll, 0, sizeof(readPoll));
    readPoll.fd = ipcDescriptor;
    readPoll.events = POLLIN;

    int pollReturn = poll(&readPoll, 1, 2000);
    if(pollReturn > 0)
    {
        // 第一步：读取数据长度
        uint32_t dataSize = 0;
        ssize_t lenRead = read(ipcDescriptor, &dataSize, sizeof(dataSize));
        if(lenRead != sizeof(dataSize))
        {
            // 处理长度读取失败，跳过本次循环
            continue;
        }

        // 第二步：初始化缓冲区大小
        deserializedBytes.resize(dataSize);
        // 第三步：读取实际数据
        lenRead = read(ipcDescriptor, deserializedBytes.data(), dataSize);
        if(lenRead != dataSize)
        {
            // 处理数据读取不完整，清空缓冲区
            deserializedBytes.clear();
            continue;
        }

        // 后续反序列化逻辑
        std::cout << "Buffer received with size: " << lenRead << '\n';
        std::cout << "Buffer content size: " << deserializedBytes.size() << '\n';

        auto deserializedJson = nlohmann::json::from_bson(deserializedBytes);
        if(!deserializedJson.empty())
        {
            sigHandlerLib::signalInfo recvSignal = deserializedJson.get<sigHandlerLib::signalInfo>();
            std::cout << "[Signal Handler Library] Signal value: " << recvSignal.signalValue << '\n';    
        }
        else
        {
            std::cout << "[Signal Handler Library] There was an error at data deserialization!" << '\n';
        }       
    }
}

另外，客户端中未使用的selectTimeout变量可以直接删除。

4. 额外注意点 #

• UNIX命名管道是字节流模式，read/write不能保证一次读写完整数据，需处理短读/短写情况（比如循环读取直到获取全部数据）。
• 服务端用O_NONBLOCK打开管道，需处理写入时的EAGAIN错误（管道满时会返回该错误）。
• 客户端的ipcDescriptor需正确初始化（比如用open打开管道，根据需求选择阻塞/非阻塞模式）。

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