mtQueue.h

/* Copyright (c) 2005-2012, Stefan Eilemann <eile@equalizergraphics.com> 
 *
 * This library is free software; you can redistribute it and/or modify it under
 * the terms of the GNU Lesser General Public License version 2.1 as published
 * by the Free Software Foundation.
 *  
 * This library 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 this library; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#ifndef LUNCHBOX_MTQUEUE_H
#define LUNCHBOX_MTQUEUE_H

#include <lunchbox/condition.h>
#include <lunchbox/debug.h>

#include <algorithm>
#include <limits.h>
#include <queue>
#include <string.h>

namespace lunchbox
{
    template< typename T, size_t S = ULONG_MAX > class MTQueue
    // S = std::numeric_limits< size_t >::max() does not work:
    //   http://gcc.gnu.org/bugzilla/show_bug.cgi?id=6424
    {
    public:
        MTQueue( size_t maxSize = S ) : _maxSize( maxSize ) {}

        MTQueue( const MTQueue< T, S >& from )  { *this = from; }

        ~MTQueue() {}

        MTQueue< T, S >& operator = ( const MTQueue< T, S >& from )
            {
                if( this != &from )
                {
                    from._cond.lock();
                    std::deque< T > copy = from._queue;
                    const size_t maxSize = from._maxSize;
                    from._cond.unlock();

                    _cond.lock();
                    _maxSize = maxSize;
                    _queue.swap( copy );
                    _cond.signal();
                    _cond.unlock();
                }
                return *this;
            }

        const T& operator[]( const size_t index ) const
            {
                _cond.lock();
                while( _queue.size() <= index )
                    _cond.wait();

                LBASSERT( _queue.size() > index );
                const T& element = _queue[index];
                _cond.unlock();
                return element;
            }

        bool isEmpty() const { return _queue.empty(); }

        size_t getSize() const { return _queue.size(); }

        void setMaxSize( const size_t maxSize )
            {
                _cond.lock();
                while( _queue.size() > maxSize )
                    _cond.wait();
                _maxSize = maxSize;
                _cond.signal();
                _cond.unlock();
            }

        size_t getMaxSize() const { return _maxSize; }

        size_t waitSize( const size_t minSize ) const
            {
                LBASSERT( minSize <= _maxSize );
                _cond.lock();
                while( _queue.size() < minSize )
                    _cond.wait();
                const size_t size = _queue.size();
                _cond.unlock();
                return size;
            }

        void clear()
            {
                _cond.lock();
                _queue.clear();
                _cond.signal();
                _cond.unlock();
            }

        T pop()
            {
                _cond.lock();
                while( _queue.empty( ))
                    _cond.wait();
                
                LBASSERT( !_queue.empty( ));
                T element = _queue.front();
                _queue.pop_front();
                _cond.signal();
                _cond.unlock();
                return element;
            }

        bool timedPop( const unsigned timeout, T& element )
            {
                _cond.lock();
                while( _queue.empty( ))
                {
                    if( !_cond.timedWait( timeout ))
                    {
                        _cond.unlock();
                        return false;
                    }
                }
                LBASSERT( !_queue.empty( ));
                element = _queue.front();
                _queue.pop_front();
                _cond.signal();
                _cond.unlock();
                return true;
            }

        bool tryPop( T& result )
            {
                _cond.lock();
                if( _queue.empty( ))
                {
                    _cond.unlock();
                    return false;
                }

                result = _queue.front();
                _queue.pop_front();
                _cond.signal();
                _cond.unlock();
                return true;
            }   

        void tryPop( const size_t num, std::vector< T >& result )
            {
                _cond.lock();
                const size_t size = LB_MIN( num, _queue.size( ));
                if( size > 0 )
                {
                    result.reserve( result.size() + size );
                    for( size_t i = 0; i < size; ++i )
                    {
                        result.push_back( _queue.front( ));
                        _queue.pop_front();
                    }
                    _cond.signal();
                }
                _cond.unlock();
            }   

        bool getFront( T& result ) const
            {
                _cond.lock();
                if( _queue.empty( ))
                {
                    _cond.unlock();
                    return false;
                }
                // else
                result = _queue.front();
                _cond.unlock();
                return true;
            }

        bool getBack( T& result ) const
            {
                _cond.lock();
                if( _queue.empty( ))
                {
                    _cond.unlock();
                    return false;
                }
                // else
                result = _queue.back();
                _cond.unlock();
                return true;
            }

        void push( const T& element )
            {
                _cond.lock();
                while( _queue.size() >= _maxSize )
                    _cond.wait();
                _queue.push_back( element );
                _cond.signal();
                _cond.unlock();
            }

        void push( const std::vector< T >& elements )
            {
                _cond.lock();
                LBASSERT( elements.size() <= _maxSize );
                while( (_maxSize - _queue.size( )) < elements.size( ))
                    _cond.wait();
                _queue.insert( _queue.end(), elements.begin(), elements.end( ));
                _cond.signal();
                _cond.unlock();
            }

        void pushFront( const T& element )
            {
                _cond.lock();
                while( _queue.size() >= _maxSize )
                    _cond.wait();
                _queue.push_front( element );
                _cond.signal();
                _cond.unlock();
            }

        void pushFront( const std::vector< T >& elements )
            {
                _cond.lock();
                LBASSERT( elements.size() <= _maxSize );
                while( (_maxSize - _queue.size( )) < elements.size( ))
                    _cond.wait();
                _queue.insert(_queue.begin(), elements.begin(), elements.end());
                _cond.signal();
                _cond.unlock();
            }


    private:
        std::deque< T > _queue;
        mutable Condition _cond;
        size_t _maxSize;
    };
}
#endif //LUNCHBOX_MTQUEUE_H