c++multithreadingunixparallel-processingpthreads

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: http://www.asic-world.com/systemverilog/sema_mail_events2.html

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

Solution

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