C++ Implementation of a Mailbox for Inter-Thread Communication
I'm Wondering If Anyone Has Previously Implemented a Mailbox Class for Interthread Communication Using the Posix Library. for Reference, I'm Looking Similar to...
I'm wondering if anyone has previously implemented a mailbox class for interthread communication using the POSIX library. For reference, I'm looking similar to mailboxes used in SystemVerilog:
EDIT:
My attempt at a mailbox using STL queues, pthread conditions, and mutexes. It tries to copy the behavior of the SystemVerilog mailbox described in the link:
#include <cerrno>
#include <climits>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <queue>
#include <fcntl.h>
#include <pthread.h>
#include <sys/select.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
using namespace std;
class Mailbox{
private:
pthread_cond_t msg_available; // Message in the mailbox?
pthread_mutex_t queue_mutex; // Mutex for queue control
queue<messageT> messages; // Messages
public:
// Constructor
Mailbox(void){
msg_available = PTHREAD_COND_INITIALIZER;
queue_mutex = PTHREAD_MUTEX_INITIALIZER;
}
// Destructor
~Mailbox(void){
// Nothing to do here
}
// Put a single message into the mailbox
void put(messageT msg){
// Lock down queue
if(pthread_mutex_lock(queue_mutex)){
fprintf(stderr, "Queue Mutex Lock Error\n");
exit(EXIT_FAILURE);
}
// Push message into mailbox
messages.push(msg);
// Signal there is a message in the mailbox
if(pthread_cond_signal(&msg_available)){
fprintf(stderr, "cond error");
exit(EXIT_FAILURE);
}
// Unlock queue
if(pthread_mutex_unlock(queue_mutex)){
fprintf(stderr, "Queue Mutex Unlock Error\n");
exit(EXIT_FAILURE);
}
}
// Try to put a single message into the mailbox
int try_put(messageT msg){
// Try to lock down queue
if(pthread_mutex_trylock(queue_mutex) == 0){
// Push message into mailbox
messages.push(msg);
// Signal there is a message in the mailbox
if(pthread_cond_signal(&msg_available)){
fprintf(stderr, "cond error");
exit(EXIT_FAILURE);
}
// Unlock queue
if(pthread_mutex_unlock(queue_mutex)){
fprintf(stderr, "Queue Mutex Unlock Error\n");
exit(EXIT_FAILURE);
}
return 1;
}
// Otherwise, say mailbox is unavailable
else
return 0;
}
// Get single message from a mailbox
void get(mesageT *msg){
// Lock down queue
if(pthread_mutex_lock(queue_mutex)){
fprintf(stderr, "Queue Mutex Lock Error\n");
exit(EXIT_FAILURE);
}
// Wait for a message to come in
while(messages.empty()){
// Release hold of the lock until another thread
// signals that a message has been placed
if(pthread_cond_wait(&msg_available,&queue_mutex)){
fprintf(stderr, "cond_wait error");
exit(EXIT_FAILURE);
}
}
// Pop of least recent message
*msg = messages.front();
messages.pop();
// Unlock queue
if(pthread_mutex_unlock(queue_mutex)){
fprintf(stderr, "Queue Mutex Unlock Error\n");
exit(EXIT_FAILURE);
}
}
// Try to get single message from a mailbox
int try_get(mesageT *msg){
int mailbox_ready = 1; // Mailbox ready
// Lock down queue
if(pthread_mutex_lock(queue_mutex)){
fprintf(stderr, "Queue Mutex Lock Error\n");
exit(EXIT_FAILURE);
}
// Indicate if mailbox is empty
if(messages.empty())
mailbox_ready = 0
// Otherwise, grab the message
else {
// Pop of least recent message
*msg = messages.front();
messages.pop();
}
// Unlock queue
if(pthread_mutex_unlock(queue_mutex)){
fprintf(stderr, "Queue Mutex Unlock Error\n");
exit(EXIT_FAILURE);
}
return mailbox_ready;
}
// Peek at single message from a mailbox
void peek(mesageT *msg){
// Lock down queue
if(pthread_mutex_lock(queue_mutex)){
fprintf(stderr, "Queue Mutex Lock Error\n");
exit(EXIT_FAILURE);
}
// Wait for a message to come in
while(messages.empty()){
// Release hold of the lock until another thread
// signals that a message has been placed
if(pthread_cond_wait(&msg_available,&queue_mutex)){
fprintf(stderr, "cond_wait error");
exit(EXIT_FAILURE);
}
}
// Peek at most recent message
*msg = messages.front();
// Unlock queue
if(pthread_mutex_unlock(queue_mutex)){
fprintf(stderr, "Queue Mutex Unlock Error\n");
exit(EXIT_FAILURE);
}
}
// Try to peek at single message from a mailbox
int try_peek(mesageT *msg){
int mailbox_ready = 1; // Mailbox ready
// Lock down queue
if(pthread_mutex_lock(queue_mutex)){
fprintf(stderr, "Queue Mutex Lock Error\n");
exit(EXIT_FAILURE);
}
if(messages.empty()) // Indicate if mailbox is empty
mailbox_ready = 0
else // Otherwise, grab the message
*msg = messages.front();
// Unlock queue
if(pthread_mutex_unlock(queue_mutex)){
fprintf(stderr, "Queue Mutex Unlock Error\n");
exit(EXIT_FAILURE);
}
return mailbox_ready;
}
}
2 Answers
A simple, semaphore protected, queue should suffice.
If you want to be able to put different kind of data in the "mailbox", then use a common base structure, that can easily be extended and contains a single integer saying what kind of structure it is, and then typecast it to the correct structure (depending on the embedded type) when you need to use it.
If the threads are all in the same process you should use pthread_mutex and pthread_condition_variable, not semaphore. Unix semaphores allow inter-process synchronization, but they are less efficient within a process, and also have semantics that are much harder to reason about than mutexes and condition variables.
Here are a number of implementations with mutexes and condition variables in C or C++: