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 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

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++:

7

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Elena Rostova

Elena Rostova

Lead Health, Wellness & Medical Journalist

Elena Rostova holds a Master's degree in Public Health Journalism. She covers groundbreaking medical research, holistic wellness trends, mental health awareness, and nutritional science.

Share this article
Twitter Facebook Pinterest