I didn't get a response to my email, but I ended up figuring out how
to use POSIX message queues with the boost::asio::io_service. Under
Linux you use select/epoll with message queue descriptors. I ended up
using boost::asio::posix::stream_descriptor to interact with message
queues. You can specify boost::asio::null_buffers() for the
async_read_some() and async_write_some() calls. These usually do
some reading or writing, but with null_buffers() they do not do any
reading or writing. You get a callbacks when the select/epoll
indicates that a message queue can be written to or read from. In the
callbacks, you can do the actual mq_send() and mq_receive() calls.
I've written some test code and included it below. There is one
message queue, a reader thread, and a writer thread. The writer
thread uses a timer to send a message through the message queue every
100 ms. The reader thread wakes up immediately to process the
messages.
-- Sergio
#include <iostream>
#include <string>
#include
#include
#include
#include
#include
using namespace std;
const std::string MSG_QUEUE_NAME = "/test_msg_queue";
struct message
{
unsigned char buffer[128];
};
class Writer
{
public:
Writer(boost::asio::io_service& ioService)
: ioService(ioService),
streamDescriptor(ioService),
timer(ioService)
{
struct mq_attr mattr;
mattr.mq_maxmsg = 10;
mattr.mq_msgsize = sizeof(struct message);
mqid = mq_open(MSG_QUEUE_NAME.c_str(), O_CREAT | O_WRONLY |
O_NONBLOCK, S_IREAD | S_IWRITE, &mattr);
cout << "writer mqid = " << mqid << endl;
streamDescriptor.assign(mqid);
streamDescriptor.async_write_some(
boost::asio::null_buffers(),
boost::bind(&Writer::handleWrite,
this,
boost::asio::placeholders::error));
}
void handleWrite(boost::system::error_code ec)
{
timer.expires_from_now(boost::posix_time::microseconds(100));
timer.async_wait(boost::bind(&Writer::handleTimer, this,
boost::asio::placeholders::error));
}
void handleTimer(boost::system::error_code& ec)
{
message msg;
int sendRet = mq_send(mqid, (const char*)&msg, sizeof(message), 0);
cout << "sendRet = " << sendRet << endl;
streamDescriptor.async_write_some(
boost::asio::null_buffers(),
boost::bind(&Writer::handleWrite,
this,
boost::asio::placeholders::error));
}
private:
boost::asio::io_service& ioService;
boost::asio::posix::stream_descriptor streamDescriptor;
boost::asio::deadline_timer timer;
mqd_t mqid;
};
void writerThread()
{
try
{
boost::asio::io_service ioService;
Writer writer(ioService);
ioService.run();
}
catch (std::exception& e)
{
std::cerr << "Exception: " << e.what() << "\n";
}
}
class Reader
{
public:
Reader(boost::asio::io_service& ioService)
: ioService(ioService),
streamDescriptor(ioService)
{
struct mq_attr mattr;
mattr.mq_maxmsg = 10;
mattr.mq_msgsize = sizeof(struct message);
mqid = mq_open(MSG_QUEUE_NAME.c_str(), O_CREAT | O_RDONLY |
O_NONBLOCK, S_IREAD | S_IWRITE, &mattr);
cout << "reader mqid = " << mqid << endl;
streamDescriptor.assign(mqid);
streamDescriptor.async_read_some(
boost::asio::null_buffers(),
boost::bind(&Reader::handleRead,
this,
boost::asio::placeholders::error));
}
void handleRead(boost::system::error_code ec)
{
u_int pri;
message msg;
ssize_t receiveRet = mq_receive(mqid, (char *) &msg, sizeof(msg), &pri);
cout << "receiveRet = " << receiveRet << endl;
streamDescriptor.async_read_some(
boost::asio::null_buffers(),
boost::bind(&Reader::handleRead,
this,
boost::asio::placeholders::error));
}
private:
boost::asio::io_service& ioService;
boost::asio::posix::stream_descriptor streamDescriptor;
mqd_t mqid;
};
void readerThread()
{
try
{
boost::asio::io_service ioService;
Reader reader(ioService);
ioService.run();
}
catch (std::exception& e)
{
std::cerr << "Exception: " << e.what() << "\n";
}
}
int main(int argc, char* argv[])
{
mq_unlink(MSG_QUEUE_NAME.c_str());
boost::thread writer(writerThread);
boost::thread reader(readerThread);
writer.join();
reader.join();
return 0;
}
On Tue, Aug 25, 2009 at 8:52 AM, Sergio Martinez wrote:
I'm working on a project where we want a single thread to
asynchronously interact with posix message queues, TCP sockets and
drivers. From what I've read, boost::asio::io_service provides a
common framework dealing with asynchronous IO. It looks like there's
a asio library for TCP and posix::stream_descriptor could be used for
interacting with a driver file descriptor. What I don't see is
support for posix message queues.
The io_service documentation states the following: "The io_service
class also includes facilities intended for developers of custom
asynchronous services.".
I was thinking of adding a custom service for posix message queues.
Has anyone already done this? Is there any fundamental reason why it
could not be done? (i.e. message queues are not byte streams, etc.)
Thanks,
Sergio
