Top: Multithreading: msgqueue
#include <pasync.h>
class msgqueue {
msgqueue(int limit = DEF_QUEUE_LIMIT);
// functions calling from the owner thread:
void processone();
void processmsgs();
void run();
// functions calling from any thread:
void post(message* msg);
void post(int id, pintptr param = 0);
void posturgent(message* msg);
void posturgent(int id, pintptr param = 0);
int send(message* msg);
int send(int id, pintptr param = 0);
int get_count();
int get_limit();
// message handlers
virtual void msghandler(message& msg) = 0;
void defhandler(message& msg);
}
The msgqueue class implements a queue of message objects and is typically used to synchronously exchange data in a multithreaded environment.
The thread which created the msgqueue object can retrieve messages from the queue using one of run(), processmsgs(), processone(). Applications always define a class derived from msgqueue and override the pure virtual method msghandler(). The overridden method receives message structures and performs appropriate actions depending on the message ID. Any other thread or even multiple threads in your application can send or post messages to the given queue using post(), posturgent() or send().
Msgqueue can serve as a synchronization object between threads. Unlike semaphores, where both sending and receiving threads can "hang" when waiting for readiness of the peer, msgqueue allows the sender to send data and immediately continue the execution. The receiver in its turn processes messages one by one in the same order as they were posted.
Threads can not only exchange data through a message queue, but also send simple notifications about various events. Message queues can as well be used in single-threaded applications with event-driven logic.
A simple example of using msgqueue could be a server application with multiple threads, each serving one client; the server maintains a log file or a table in a database where it records various events. To record events synchronously the client threads are sending appropriate messages to the main thread. The client threads never waste time, they just post their messages and immediately continue their work.
IMPORTANT NOTES: (1) a message object should always be constructed dynamically, i.e. using operator new; (2) a message object is always destroyed by the queue manager after it has been processed; (3) a message object can be sent and processed only once.
A slower but more universal alternative to the message queue is local pipe (see infile::pipe()).
msgqueue::msgqueue(int limit = DEF_QUEUE_LIMIT) constructs a message queue object. It doesn't matter which thread is creating this object, but later only one thread can process the queue and handle messages. Limit specifies the maximum number of unhandled messages this queue can hold. If the limit is reached, the next thread that posts a message will wait until the queue becomes available again. In this version the default for limit is 5000.
void msgqueue::processone() processes one message from the queue. This method may "hang" if no messages are available. processone() calls the overridden msghandler() and then destroys the message object.
void msgqueue::processmsgs() processes all messages in the queue and returns to the caller. If there are no messages in the queue, processmsgs() returns immediately. Each message is processed as described for processone().
void msgqueue::run() enters an infinite loop of message processing which can only be terminated by sending or posting a special message MSG_QUIT (e.g. post(MSG_QUIT)). Each message is processed as described for processone().
void msgqueue::post(message* msg) adds a message to the queue. Msg can be an object of class message or any derivative class. The message object should always be created dynamically using operator new. The messages in the queue are processed in order they were posted, i.e. on first-in-first-out basis. post() can be called from any thread, including the thread owning the queue.
void msgqueue::post(int id, pintptr param = 0) creates a message object using id and param and calls post(message*).
void msgqueue::posturgent(message* msg) posts a message object "out of turn", i.e. this message will be processed first. The messages posted through this method are processed on first-in-last-out basis. post() and posturgent() can be used alternately on the same queue. Like post(), this method can be called from any thread.
void msgqueue::posturgent(int id, pintptr param = 0) creates a message object using id and param and calls posturgent(message*).
pintptr msgqueue::send(message* msg) calls the message handler directly, by-passing the queue. If the sender is the same as the thread owning the queue, send() simply calls msghandler(). Otherwise, if the sender is a concurrent thread, send() enters an effective wait state until the message is processed by the owner thread. The return value is the value of result in the message object. In both cases the message is destroyed upon return from send().
pintptr msgqueue::send(int id, pintptr param = 0) creates a message object using id and param and calls send(message*). The return value is the value of result in the message object.
int msgqueue::get_count() returns the number of messages currently in the queue.
int msgqueue::get_limit() returns the queue limit set by the constructor.
virtual void msgqueue::msghandler(message& msg) this pure virtual method should be overridden to provide application-specific message handling functionality. msghandler() usually checks the message ID through a switch statement. If the message ID is unknown to the application, defhandler() should be called. The message object msg CAN NOT be reused with post(), posturgent() or send(), neither it can be destroyed within the message handler. The message handler can assign some value to msg.result to return a simple answer to the caller of send().
void msgqueue::defhandler(message& msg) is called from within user-defined msghandler() when the message ID is unknown to the application. defhandler() processes some messages internally used by the library, e.g. MSG_QUIT.