queue.h

Go to the documentation of this file.

/***************************************************************************
 *  include/stxxl/bits/containers/queue.h
 *
 *  Part of the STXXL. See http://stxxl.sourceforge.net
 *
 *  Copyright (C) 2005 Roman Dementiev <[email protected]>
 *  Copyright (C) 2009, 2010 Andreas Beckmann <[email protected]>
 *
 *  Distributed under the Boost Software License, Version 1.0.
 *  (See accompanying file LICENSE_1_0.txt or copy at
 *  http://www.boost.org/LICENSE_1_0.txt)
 **************************************************************************/

#ifndef STXXL_CONTAINERS_QUEUE_HEADER
#define STXXL_CONTAINERS_QUEUE_HEADER

#include <vector>
#include <queue>
#include <deque>

#include <stxxl/bits/deprecated.h>
#include <stxxl/bits/mng/block_manager.h>
#include <stxxl/bits/mng/typed_block.h>
#include <stxxl/bits/common/simple_vector.h>
#include <stxxl/bits/common/tmeta.h>
#include <stxxl/bits/mng/read_write_pool.h>
#include <stxxl/bits/mng/write_pool.h>
#include <stxxl/bits/mng/prefetch_pool.h>

STXXL_BEGIN_NAMESPACE

#ifndef STXXL_VERBOSE_QUEUE
#define STXXL_VERBOSE_QUEUE STXXL_VERBOSE2
#endif

//! \addtogroup stlcont
//! \{

//! External FIFO queue container. \n
//! <b> Introduction </b> to queue container: see \ref tutorial_queue tutorial\n
//! <b> Design and Internals </b> of queue container: see \ref design_queue.
//!
//! \tparam ValueType type of the contained objects (POD with no references to internal memory)
//! \tparam BlockSize size of the external memory block in bytes, default is \c STXXL_DEFAULT_BLOCK_SIZE(ValueType)
//! \tparam AllocStr parallel disk allocation strategy, default is \c STXXL_DEFAULT_ALLOC_STRATEGY
//! \tparam SizeType size data type, default is \c stxxl::uint64
template <class ValueType,
          unsigned BlockSize = STXXL_DEFAULT_BLOCK_SIZE(ValueType),
          class AllocStr = STXXL_DEFAULT_ALLOC_STRATEGY,
          class SizeType = stxxl::uint64>
class queue : private noncopyable
{
public:
    typedef ValueType value_type;
    typedef AllocStr alloc_strategy_type;
    typedef SizeType size_type;
    enum {
        block_size = BlockSize
    };

    typedef typed_block<block_size, value_type> block_type;
    typedef BID<block_size> bid_type;

private:
    typedef read_write_pool<block_type> pool_type;

    size_type m_size;
    bool delete_pool;
    pool_type* pool;
    block_type* front_block;
    block_type* back_block;
    value_type* front_element;
    value_type* back_element;
    alloc_strategy_type alloc_strategy;
    unsigned_type alloc_count;
    std::deque<bid_type> bids;
    block_manager* bm;
    unsigned_type blocks2prefetch;

public:
    //! \name Constructors/Destructors
    //! \{

    //! Constructs empty queue with own write and prefetch block pool.
    //!
    //! \param D  number of parallel disks, defaulting to the configured number of scratch disks,
    //!           memory consumption will be 2 * D + 2 blocks
    //!           (first and last block, D blocks as write cache, D block for prefetching)
    explicit queue(int_type D = -1)
        : m_size(0),
          delete_pool(true),
          alloc_count(0),
          bm(block_manager::get_instance())
    {
        if (D < 1)
            D = config::get_instance()->disks_number();
        STXXL_VERBOSE_QUEUE("queue[" << this << "]::queue(D)");
        pool = new pool_type(D, D + 2);
        init();
    }

    //! Constructs empty queue with own write and prefetch block pool.
    //!
    //! \param w_pool_size  number of blocks in the write pool, must be at least 2, recommended at least 3
    //! \param p_pool_size  number of blocks in the prefetch pool, recommended at least 1
    //! \param blocks2prefetch_  defines the number of blocks to prefetch (\c front side),
    //!                          default is number of block in the prefetch pool
    explicit queue(unsigned_type w_pool_size, unsigned_type p_pool_size, int blocks2prefetch_ = -1)
        : m_size(0),
          delete_pool(true),
          alloc_count(0),
          bm(block_manager::get_instance())
    {
        STXXL_VERBOSE_QUEUE("queue[" << this << "]::queue(sizes)");
        pool = new pool_type(p_pool_size, w_pool_size);
        init(blocks2prefetch_);
    }

    //! Constructs empty queue.
    //!
    //! \param w_pool write pool
    //! \param p_pool prefetch pool
    //! \param blocks2prefetch_  defines the number of blocks to prefetch (\c front side),
    //!                          default is number of blocks in the prefetch pool
    //!  \warning Number of blocks in the write pool must be at least 2, recommended at least 3
    //!  \warning Number of blocks in the prefetch pool recommended at least 1
    STXXL_DEPRECATED(
        queue(write_pool<block_type>& w_pool, prefetch_pool<block_type>& p_pool, int blocks2prefetch_ = -1))
        : m_size(0),
          delete_pool(true),
          alloc_count(0),
          bm(block_manager::get_instance())
    {
        STXXL_VERBOSE_QUEUE("queue[" << this << "]::queue(pools)");
        pool = new pool_type(p_pool, w_pool);
        init(blocks2prefetch_);
    }

    //! Constructs empty queue.
    //!
    //! \param pool_ block write/prefetch pool
    //! \param blocks2prefetch_  defines the number of blocks to prefetch (\c front side),
    //!                          default is number of blocks in the prefetch pool
    //!  \warning Number of blocks in the write pool must be at least 2, recommended at least 3
    //!  \warning Number of blocks in the prefetch pool recommended at least 1
    queue(pool_type& pool_, int blocks2prefetch_ = -1)
        : m_size(0),
          delete_pool(false),
          pool(&pool_),
          alloc_count(0),
          bm(block_manager::get_instance())
    {
        STXXL_VERBOSE_QUEUE("queue[" << this << "]::queue(pool)");
        init(blocks2prefetch_);
    }

    //! \}

    //! \name Modifiers
    //! \{

    void swap(queue& obj)
    {
        std::swap(m_size, obj.m_size);
        std::swap(delete_pool, obj.delete_pool);
        std::swap(pool, obj.pool);
        std::swap(front_block, obj.front_block);
        std::swap(back_block, obj.back_block);
        std::swap(front_element, obj.front_element);
        std::swap(back_element, obj.back_element);
        std::swap(alloc_strategy, obj.alloc_strategy);
        std::swap(alloc_count, obj.alloc_count);
        std::swap(bids, obj.bids);
        std::swap(bm, obj.bm);
        std::swap(blocks2prefetch, obj.blocks2prefetch);
    }

    //! \}

private:
    void init(int blocks2prefetch_ = -1)
    {
        if (pool->size_write() < 2) {
            STXXL_ERRMSG("queue: invalid configuration, not enough blocks (" << pool->size_write() <<
                         ") in write pool, at least 2 are needed, resizing to 3");
            pool->resize_write(3);
        }

        if (pool->size_write() < 3) {
            STXXL_MSG("queue: inefficient configuration, no blocks for buffered writing available");
        }

        if (pool->size_prefetch() < 1) {
            STXXL_MSG("queue: inefficient configuration, no blocks for prefetching available");
        }

        front_block = back_block = pool->steal();
        back_element = back_block->begin() - 1;
        front_element = back_block->begin();
        set_prefetch_aggr(blocks2prefetch_);
    }

public:
    //! \name Miscellaneous
    //! \{

    //! Defines the number of blocks to prefetch (\c front side).
    //! This method should be called whenever the prefetch pool is resized
    //! \param blocks2prefetch_  defines the number of blocks to prefetch (\c front side),
    //!                          a negative value means to use the number of blocks in the prefetch pool
    void set_prefetch_aggr(int_type blocks2prefetch_)
    {
        if (blocks2prefetch_ < 0)
            blocks2prefetch = pool->size_prefetch();
        else
            blocks2prefetch = blocks2prefetch_;
    }

    //! Returns the number of blocks prefetched from the \c front side.
    unsigned_type get_prefetch_aggr() const
    {
        return blocks2prefetch;
    }
    //! \}

    //! \name Modifiers
    //! \{

    //! Adds an element in the queue.
    void push(const value_type& val)
    {
        if (UNLIKELY(back_element == back_block->begin() + (block_type::size - 1)))
        {
            // back block is filled
            if (front_block == back_block)
            {             // can not write the back block because it
                // is the same as the front block, must keep it memory
                STXXL_VERBOSE1("queue::push Case 1");
            }
            else if (size() < 2 * block_type::size)
            {
                STXXL_VERBOSE1("queue::push Case 1.5");
                // only two blocks with a gap in the beginning, move elements within memory
                assert(bids.empty());
                size_t gap = front_element - front_block->begin();
                assert(gap > 0);
                std::copy(front_element, front_block->end(), front_block->begin());
                std::copy(back_block->begin(), back_block->begin() + gap, front_block->begin() + (block_type::size - gap));
                std::copy(back_block->begin() + gap, back_block->end(), back_block->begin());
                front_element -= gap;
                back_element -= gap;

                ++back_element;
                *back_element = val;
                ++m_size;
                return;
            }
            else
            {
                STXXL_VERBOSE1("queue::push Case 2");
                // write the back block
                // need to allocate new block
                bid_type newbid;

                bm->new_block(alloc_strategy, newbid, alloc_count++);

                STXXL_VERBOSE_QUEUE("queue[" << this << "]: push block " << back_block << " @ " << FMT_BID(newbid));
                bids.push_back(newbid);
                pool->write(back_block, newbid);
                if (bids.size() <= blocks2prefetch) {
                    STXXL_VERBOSE1("queue::push Case Hints");
                    pool->hint(newbid);
                }
            }
            back_block = pool->steal();

            back_element = back_block->begin();
            *back_element = val;
            ++m_size;
            return;
        }
        ++back_element;
        *back_element = val;
        ++m_size;
    }

    //! Removes element from the queue.
    void pop()
    {
        assert(!empty());

        if (UNLIKELY(front_element == front_block->begin() + (block_type::size - 1)))
        {
            // if there is only one block, it implies ...
            if (back_block == front_block)
            {
                STXXL_VERBOSE1("queue::pop Case 3");
                assert(size() == 1);
                assert(back_element == front_element);
                assert(bids.empty());
                // reset everything
                back_element = back_block->begin() - 1;
                front_element = back_block->begin();
                m_size = 0;
                return;
            }

            --m_size;
            if (m_size <= block_type::size)
            {
                STXXL_VERBOSE1("queue::pop Case 4");
                assert(bids.empty());
                // the back_block is the next block
                pool->add(front_block);
                front_block = back_block;
                front_element = back_block->begin();
                return;
            }
            STXXL_VERBOSE1("queue::pop Case 5");

            assert(!bids.empty());
            request_ptr req = pool->read(front_block, bids.front());
            STXXL_VERBOSE_QUEUE("queue[" << this << "]: pop block  " << front_block << " @ " << FMT_BID(bids.front()));

            // give prefetching hints
            for (unsigned_type i = 0; i < blocks2prefetch && i < bids.size() - 1; ++i)
            {
                STXXL_VERBOSE1("queue::pop Case Hints");
                pool->hint(bids[i + 1]);
            }

            front_element = front_block->begin();
            req->wait();

            bm->delete_block(bids.front());
            bids.pop_front();
            return;
        }

        ++front_element;
        --m_size;
    }
    //! \}

    //! \name Operators
    //! \{

    //! Returns a mutable reference at the back of the queue.
    value_type & back()
    {
        assert(!empty());
        return *back_element;
    }

    //! Returns a const reference at the back of the queue.
    const value_type & back() const
    {
        assert(!empty());
        return *back_element;
    }

    //! Returns a mutable reference at the front of the queue.
    value_type & front()
    {
        assert(!empty());
        return *front_element;
    }

    //! Returns a const reference at the front of the queue.
    const value_type & front() const
    {
        assert(!empty());
        return *front_element;
    }

    //! \}

    //! \name Constructors/Destructors
    //! \{

    ~queue()
    {
        if (front_block != back_block)
            pool->add(back_block);
        pool->add(front_block);

        if (delete_pool)
        {
            delete pool;
        }

        if (!bids.empty())
            bm->delete_blocks(bids.begin(), bids.end());
    }

    //! \}

    //! \name Capacity
    //! \{

    //! Returns the size of the queue.
    size_type size() const
    {
        return m_size;
    }

    //! Returns \c true if queue is empty.
    bool empty() const
    {
        return (m_size == 0);
    }

    //! \}
};

//! \}

STXXL_END_NAMESPACE

#endif // !STXXL_CONTAINERS_QUEUE_HEADER
// vim: et:ts=4:sw=4