提升进程间消息队列和分叉

Question

我正在尝试使用来自 boost 进程间库的消息队列与分叉的子进程进行通信。当子进程调用 receive 它会导致异常消息

boost::interprocess_exception::library_error

我在 Debian 9 x64 上使用 GCC 6.3。

#include <iostream>
#include <unistd.h>
#include <boost/interprocess/ipc/message_queue.hpp>
#include <memory>

int main(int argc, char* argv[])
{
    using namespace boost::interprocess;

    const char* name = "foo-552b8ae9-6037-4b77-aa0d-d4dc9dad790b";
    const int max_num_msg = 100;
    const int max_msg_size = 32;
    bool is_child = false;

    message_queue::remove(name);
    auto mq = std::make_unique<message_queue>(create_only, name, max_num_msg, max_msg_size);

    auto child_pid = fork();
    if (child_pid == -1)
    {
        std::cout << "fork failed" << std::endl;
        return -1;
    }
    else if (child_pid == 0)
    {
        is_child = true;
    }

    if (is_child)
    {
        // does child needs to reopen it?
        mq.reset( new message_queue(open_only, name) );
    }

    int send_num = 0;
    while(true)
    {
        unsigned int priority = 0;
        if (is_child)
        {
            message_queue::size_type bytes = 0;
            try
            {
                int num;
                // Always throws. What is wrong ???????
                mq->receive(&num, sizeof(num), bytes, priority);
                std::cout <<  num << std::endl;
            }
            catch(const std::exception& e)
            {
                std::cout << "Receive caused execption " << e.what() << std::endl;
            }
            sleep(1);
        }
        else
        {
            mq->send(&send_num, sizeof(send_num), priority);
            send_num++;
            sleep(5);
        }
    }


    return 0;
}

另外，在子进程中是否需要重新打开父进程创建的消息队列？我尝试了两种方式，但都没有奏效。我在 receive 上遇到了同样的异常。

Answer 1

问题是您的接收缓冲区小于max_msg_size。假设 4 字节整数，这应该工作：

int num[8];
mq.receive(num, sizeof(num), bytes, priority);
std::cout << *num << std::endl;

另外，我认为没有理由对实际队列实例进行快速而松散的操作。只需为每个进程创建它：

#include <boost/interprocess/ipc/message_queue.hpp>
#include <boost/exception/diagnostic_information.hpp>
#include <iostream>
#include <memory>
#include <unistd.h>

int main() {
    namespace bip = boost::interprocess;

    const char *name = "foo-552b8ae9-6037-4b77-aa0d-d4dc9dad790b";
    {
        const int max_num_msg = 100;
        const int max_msg_size = 32;
        bip::message_queue::remove(name);
        bip::message_queue mq(bip::create_only, name, max_num_msg, max_msg_size);
    }

    auto child_pid = fork();
    if (child_pid == -1) {
        std::cout << "fork failed" << std::endl;
        return -1;
    }
    bip::message_queue mq(bip::open_only, name);

    if (bool const is_child = (child_pid == 0)) {
        while (true) {
            unsigned int priority = 0;
            bip::message_queue::size_type bytes = 0;

            try {
                int num[8];
                mq.receive(num, sizeof(num), bytes, priority);
                std::cout << *num << std::endl;
            } catch (const bip::interprocess_exception &e) {
                std::cout << "Receive caused execption " << boost::diagnostic_information(e, true) << std::endl;
            }
            sleep(1);
        }
    } else {
        // parent
        int send_num = 0;
        while (true) {
            unsigned int priority = 0;

            mq.send(&send_num, sizeof(send_num), priority);
            send_num++;
            sleep(5);
        }
    }
}