ddcThreadQueue.h

Go to the documentation of this file.

//-*- Mode: C++ -*-
//
//  DDC originally by Alexey Sokirko
//  This file by Bryan Jurish, adapted from https://vichargrave.github.io/articles/2013-01/multithreaded-work-queue-in-cpp
//
//  This file is part of DDC.
// 
//  DDC is free software: you can redistribute it and/or modify
//  it under the terms of the GNU Lesser General Public License as published by
//  the Free Software Foundation, either version 3 of the License, or
//  (at your option) any later version.
// 
//  DDC is distributed in the hope that it will be useful,
//  but WITHOUT ANY WARRANTY; without even the implied warranty of
//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//  GNU Lesser General Public License for more details.
// 
//  You should have received a copy of the GNU Lesser General Public License
//  along with DDC.  If not, see <http://www.gnu.org/licenses/>.
//
#ifndef DDC_THREAD_QUEUE_H
#define DDC_THREAD_QUEUE_H

#include "ddcConfig.h"
#include <pthread.h>
#include <list>

#define DDC_DEFAULT_THREAD_QUEUE_SIZE 256

//======================================================================
template<typename T> class ddcThreadQueue
{
public:
    //--------------------------------------------------------------

    typedef T ItemT;

public:
    //--------------------------------------------------------------
    // (all public, because private data is annoying)

    std::list<T>   m_queue;

    size_t m_max_size;

    pthread_mutex_t m_mutex;
    
    pthread_cond_t  m_condv_added;
    
    pthread_cond_t  m_condv_removed;

public:
    //--------------------------------------------------------------


    ddcThreadQueue(size_t max_size=DDC_DEFAULT_THREAD_QUEUE_SIZE)
        : m_max_size(max_size)
    {
        pthread_mutex_init(&m_mutex, NULL);
        pthread_cond_init(&m_condv_added, NULL);
        pthread_cond_init(&m_condv_removed, NULL);
    };

    ~ddcThreadQueue()
    {
        pthread_mutex_destroy(&m_mutex);
        pthread_cond_destroy(&m_condv_added);
        pthread_cond_destroy(&m_condv_removed);
    };

    //--------------------------------------------------------------


    void add(T item)
    {
        pthread_mutex_lock(&m_mutex);
        while (m_queue.size() == m_max_size) {
            //fprintf(stderr, "ddcThreadQueue::add(): max queue size (%zu) reached; waiting...\n", m_queue.size());
            pthread_cond_wait(&m_condv_removed, &m_mutex);
        }
        //fprintf(stderr, "ddcThreadQueue::add(): appending queue item to queue of size %zu\n", m_queue.size());
        m_queue.push_back(item);
        pthread_cond_signal(&m_condv_added);
        pthread_mutex_unlock(&m_mutex);
    };

    T remove()
    {
        pthread_mutex_lock(&m_mutex);
        while (m_queue.size() == 0) {
            pthread_cond_wait(&m_condv_added, &m_mutex);
        }

        T item = m_queue.front();
        m_queue.pop_front();

        pthread_cond_signal(&m_condv_removed);
        pthread_mutex_unlock(&m_mutex);
        return item;
    };


    size_t size() const
    {
        pthread_mutex_lock(&m_mutex);
        size_t sz = m_queue.size();
        pthread_mutex_unlock(&m_mutex);
        return sz;
    };
};

#endif /* DDC_THREAD_QUEUE_H */

/*--- emacs style variables ---
 * Local Variables:
 *  mode: C++
 *  c-file-style: "ellemtel"
 *  c-basic-offset: 4
 *  tab-width: 8
 *  indent-tabs-mode: nil
 * End:
 */