22
\$\begingroup\$

Is this the correct way to implement a thread-safe concurrent FIFO queue in C++? It requires passing unsigned char* arrays of binary data.

Thread Safe Concurrent Queue

#include <queue>
#include <pthread.h>

class concurrent_queue
{

private:

    std::queue<unsigned char*> _queue_;
    pthread_mutex_t push_mutex;
    pthread_mutex_t pop_mutex;
    pthread_cond_t cond;

public:

    concurrent_queue()
    {
        pthread_mutex_init(&push_mutex, NULL);
        pthread_mutex_init(&pop_mutex, NULL);
        pthread_cond_init(&cond, NULL);
    }

    void push(unsigned char* data)
    {
        pthread_mutex_lock(&push_mutex);

        _queue_.push(data);

        pthread_cond_signal(&cond);
        pthread_mutex_unlock(&push_mutex);
    }

    void pop(unsigned char** popped_data)
    {
        pthread_mutex_lock(&pop_mutex);

        while (_queue_.empty() == true)
        {
            pthread_cond_wait(&cond, &pop_mutex);
        }

        *popped_data = _queue_.front();
        _queue_.pop();

        pthread_mutex_unlock(&pop_mutex);
    }
};

CONSUMER TEST:

void *consumer_thread(void *arguments)
{
    concurrent_queue *cq = static_cast<concurrent_queue*>(arguments);

    while (true)
    {
        unsigned char* data = NULL;

        cq->pop(&data);

        if (data != NULL)
        {
            // Eureka! Received from the other thread!!!
            // Delete it so memory keeps free.
            // NOTE: In the real scenario for which I need
            // this class, the data received are bitmap pixels
            // and at this point it would be processed

            delete[] data;
        }
    }

    return 0;
}

PRODUCER TEST:

void main()
{
    concurrent_queue cq;

    // Create the consumer
    pthread_t consumer;
    pthread_create(&consumer, NULL, consumer_thread, &cq));

    // Start producing
    while(true)
    {
        // Push data. 
        // Expected behaviour: memory should never run out, as the
        // consumer should receive the data and delete it.
        // NOTE: test_data in the real purpose scenario for which I 
        // need this class would hold bitmap pixels, so it's meant to 
        // hold binary data and not a string

        unsigned char* test_data = new unsigned char [8192];
        cq.push(test_data);
    }

    return 0;
}
\$\endgroup\$
4
  • 1
    \$\begingroup\$ Please read Can I edit my own question to include suggested changes from answers? \$\endgroup\$
    – ChrisW
    Commented Feb 14, 2014 at 13:10
  • \$\begingroup\$ Zhenya - editing the question to incorporate suggested changes, 'invalidates' the answers that have been given. Please read the meta-post/link that ChrisW posted above, and I repeat here.... \$\endgroup\$
    – rolfl
    Commented Apr 20, 2014 at 15:09
  • \$\begingroup\$ I see you're deleting the char * in consumer Thread. Isn't that a bug ? \$\endgroup\$
    – navderm
    Commented Aug 9, 2015 at 19:57
  • \$\begingroup\$ The consumer is just a test. In the comment right before the delete, on the second line, is explained why is deleted. In a real scenario you will not delete it until you have used for your own purposes, but after using it you will need to delete it to avoid memory leaks. \$\endgroup\$
    – Perraco
    Commented Aug 9, 2015 at 20:00

1 Answer 1

12
\$\begingroup\$

You are guarding the state of one variable

std::queue<unsigned char*> _queue_;

So you only have one mutex.

pthread_mutex_t push_mutex;
pthread_mutex_t pop_mutex;

If you have two then push and pop can modify queue at the same time.

Make sure your mutex locking is exception safe by using RAII. Note all operations were you do start operation stop should be done by using RAII. This is the most important concept in C++ look it up.

Don't use pointers in C++. They do not convey ownership symantics (which means you don't know who should delete it (or when it should be deleted)). In this case use std::string it solves all the problems.

I see calls pthread_XXX_init but not the equivalent destroy.

Because you have not defined them the default copy constructor and assignment operator will be created. Since your object contains resources (mutext and conditional) you probably don't want the default behavior. Either define them or disable them.

C++11 has a better set of threading constructs use those rather than the C pthread library.

\$\endgroup\$
2
  • 1
    \$\begingroup\$ I find std::vector easier to deal with when handling binary data (assuming the unsigned char * pointer is for binary data.) \$\endgroup\$ Commented Feb 14, 2014 at 4:42
  • \$\begingroup\$ @AlexisWilke: Me too. I was assuming a string. But for binary data I would use a vector. \$\endgroup\$ Commented Feb 14, 2014 at 5:47

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.