block_scheduler.h

Go to the documentation of this file.

/***************************************************************************
 *  include/stxxl/bits/mng/block_scheduler.h
 *
 *  Part of the STXXL. See http://stxxl.sourceforge.net
 *
 *  Copyright (C) 2010-2011 Raoul Steffen <[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_MNG_BLOCK_SCHEDULER_HEADER
#define STXXL_MNG_BLOCK_SCHEDULER_HEADER

#include <stack>
#include <queue>
#include <limits>

#include <stxxl/bits/mng/block_manager.h>
#include <stxxl/bits/mng/typed_block.h>
#include <stxxl/bits/common/addressable_queues.h>

STXXL_BEGIN_NAMESPACE

//! Virtualization of a block of data.
//! Holds information for allocating and swapping. To use in cooperation with block_scheduler.
//!
//! A swappable_block can be uninitialized, i.e. it holds no data.
//! When access is required, is has to be acquired first, and released afterwards, so it can be swapped in and out as required.
//! If the stored data is no longer needed, it can get uninitialized, freeing both internal and external memory.
//! \tparam ValueType type of contained objects (POD with no references to internal memory).
//! \tparam BlockSize Number of objects in one block.
//!         BlockSize*sizeof(ValueType) must be divisible by 4096.
template <typename ValueType, unsigned BlockSize>
class swappable_block
{
protected:
    static const unsigned_type raw_block_size = BlockSize * sizeof(ValueType);

public:
    typedef typed_block<raw_block_size, ValueType> internal_block_type;
    typedef typename internal_block_type::bid_type external_block_type;

protected:
    external_block_type external_data;      //!external_data.valid if no associated space on disk
    internal_block_type* internal_data;     //NULL if there is no internal memory reserved
    bool dirty;
    int_type reference_count;

    static unsigned_type disk_allocation_offset;

    void get_external_block()
    { block_manager::get_instance()->new_block(striping(), external_data, ++disk_allocation_offset); }

    void free_external_block()
    {
        block_manager::get_instance()->delete_block(external_data);
        external_data = external_block_type(); // make invalid
    }

public:
    //! Create in uninitialized state.
    swappable_block()
        : external_data() /*!valid*/, internal_data(0), dirty(false), reference_count(0) { }

    ~swappable_block() { }

    //! If it has an internal_block. The internal_block implicitly holds valid data.
    bool is_internal() const
    { return (internal_data != NULL); }

    //! If the external_block does not hold valid data.
    bool is_dirty() const
    { return dirty; }

    //! If it has an external_block.
    bool has_external_block() const
    { return external_data.valid(); }

    //! If it has an external_block that holds valid data.
    bool is_external() const
    { return has_external_block() && ! is_dirty(); }

    //! If it is acquired.
    bool is_acquired() const
    { return reference_count > 0; }

    //! If it holds an internal_block but does not need it.
    bool is_evictable() const
    { return ! is_acquired() && is_internal(); }

    //! If it has some valid data (in- or external).
    bool is_initialized() const
    { return is_internal() || is_external(); }

    //! Invalidate external data if true.
    //! \return is_dirty()
    bool make_dirty_if(const bool make_dirty)
    {
        assert(is_acquired());
        return dirty = make_dirty || dirty;
    }

    //! Acquire the block, i.e. add a reference. Has to be internal.
    //! \return A reference to the data-block.
    internal_block_type & acquire()
    {
        assert(is_internal());
        ++reference_count;
        return *internal_data;
    }

    //! Release the block, i.e. subduct a reference. Has to be acquired.
    void release()
    {
        assert(is_acquired());
        --reference_count;
    }

    //! Get a reference to the data-block. Has to be acquired.
    const internal_block_type & get_internal_block() const
    {
        assert(is_acquired());
        return *internal_data;
    }

    //! Get a reference to the data-block. Has to be acquired.
    internal_block_type & get_internal_block()
    {
        assert(is_acquired());
        return *internal_data;
    }

    //! Fill block with default data, is supposed to be overwritten by subclass. Has to be internal.
    void fill_default() { }

    //! Read asyncronusly from external_block to internal_block. Has to be internal and have an external_block.
    //! \return A request pointer to the I/O.
    request_ptr read_async(completion_handler on_cmpl = default_completion_handler())
    {
        assert(is_internal());
        assert(has_external_block());
        #ifdef RW_VERBOSE
        STXXL_MSG("reading block");
        #endif
        dirty = false;
        return internal_data->read(external_data, on_cmpl);
    }

    //! Read synchronously from external_block to internal_block. Has to be internal and have an external_block.
    void read_sync()
    { read_async()->wait(); }

    //! Write asyncronusly from internal_block to external_block if necessary.
    //! \return A request pointer to the I/O, an invalid request pointer if not necessary.
    request_ptr clean_async(completion_handler on_cmpl = default_completion_handler())
    {
        if (! is_dirty())
            return request_ptr();
        if (! has_external_block())
            get_external_block();
        #ifdef RW_VERBOSE
        STXXL_MSG("writing block");
        #endif
        dirty = false;
        return internal_data->write(external_data, on_cmpl);
    }

    //! Write synchronously from internal_block to external_block if necessary.
    void clean_sync()
    {
        request_ptr rp = clean_async();
        if (rp.valid())
            rp->wait();
    }

    //! Attach an internal_block, making the block internal. Has to be not internal.
    void attach_internal_block(internal_block_type* iblock)
    {
        assert(! is_internal());
        internal_data = iblock;
    }

    //! Detach the internal_block. Writes to external_block if necessary. Has to be evictable.
    //! \return A pointer to the internal_block.
    internal_block_type * detach_internal_block()
    {
        assert(is_evictable());
        clean_sync();
        internal_block_type* iblock = internal_data;
        internal_data = 0;
        return iblock;
    }

    //! Bring the block in uninitialized state, freeing external and internal memory.
    //! Returns a pointer to the internal_block, NULL if it had none.
    //! \return A pointer to the freed internal_block, NULL if it had none.
    internal_block_type * deinitialize()
    {
        assert(! is_acquired());
        dirty = false;
        // free external_block (so that it becomes invalid and the disk-space can be used again)
        if (has_external_block())
            free_external_block();
        // free internal_block
        internal_block_type* iblock = internal_data;
        internal_data = 0;
        return iblock;
    }

    //! Set the external_block that holds the swappable_block's data. The block gets initialized with it.
    //! \param eblock The external_block holding initial data.
    void initialize(external_block_type eblock)
    {
        assert(! is_initialized());
        external_data = eblock;
    }

    //! Extract the swappable_blocks data in an external_block. The block gets uninitialized.
    //! \return The external_block that holds the swappable_block's data.
    external_block_type extract_external_block()
    {
        assert(! is_internal());
        external_block_type eblock = external_data;
        external_data = external_block_type();
        return eblock;
    }
};

template <typename ValueType, unsigned BlockSize>
unsigned_type swappable_block<ValueType, BlockSize>::disk_allocation_offset = 0;

template <class SwappableBlockType>
class block_scheduler_algorithm;

template <class SwappableBlockType>
class block_scheduler_algorithm_online_lru;

//! Schedules swapping of blocks and provides blocks for temporary storage.
//!
//! In simple mode, it tries to save I/Os through caching only.
//! In simulation mode, it records access patterns into a prediction sequence.
//! The prediction sequence can then be used for prefetching in the (offline) execute mode.
//! This will only work for algorithms with deterministic, data oblivious access patterns.
//! In simulation mode, no I/O is performed; the data provided is accessible but undefined.
//! In execute mode, it does caching, prefetching, and possibly other optimizations.
//! \tparam SwappableBlockType Type of swappable_blocks to manage. Can be some specialized subclass.
template <class SwappableBlockType>
class block_scheduler : private noncopyable
{
protected:
    // tuning-parameter: To acquire blocks, internal memory has to be allocated.
    // This constant limits the number of internal_blocks allocated at once.
    static const int_type max_internal_blocks_alloc_at_once;

    typedef int_type time_type;

public:
    typedef typename SwappableBlockType::internal_block_type internal_block_type;
    typedef typename SwappableBlockType::external_block_type external_block_type;
    typedef typename std::vector<SwappableBlockType>::size_type swappable_block_identifier_type;

    /*/! Mode the block scheduler currently works in
    enum mode
    {
        online,         //serve requests immediately, without any prediction, LRU caching
        simulation,     //record prediction sequence only, do not serve requests, (returned blocks must not be accessed)
        offline_lfd,    //serve requests based on prediction sequence, using longest-forward-distance caching
        offline_lfd_prefetch     //serve requests based on prediction sequence, using longest-forward-distance caching, and prefetching
    };*/

public:
    // -------- prediction_sequence -------
    enum block_scheduler_operation
    {
        op_acquire,
        op_acquire_uninitialized,
        op_release,
        op_release_dirty,
        op_deinitialize,
        op_initialize,
        op_extract_external_block
    };

    struct prediction_sequence_element
    {
        block_scheduler_operation op;
        swappable_block_identifier_type id;
        time_type time;

        prediction_sequence_element(block_scheduler_operation op,
                                    swappable_block_identifier_type id, int_type time)
            : op(op), id(id), time(time) { }
    };

    typedef std::list<prediction_sequence_element> prediction_sequence_type;
    // ---- end prediction_sequence -------

protected:
    template <class SBT>
    friend class block_scheduler_algorithm;

    const int_type max_internal_blocks;
    int_type remaining_internal_blocks;
    //! Stores pointers to arrays of internal_blocks. Used to deallocate them only.
    std::stack<internal_block_type*> internal_blocks_blocks;
    //! holds free internal_blocks with attributes reset.
    std::stack<internal_block_type*> free_internal_blocks;
    //! temporary blocks that will not be needed after algorithm termination.
    mutable std::vector<SwappableBlockType> swappable_blocks;
    //! holds indices of free swappable_blocks with attributes reset.
    std::priority_queue<swappable_block_identifier_type, std::vector<swappable_block_identifier_type>,
                        std::greater<swappable_block_identifier_type> > free_swappable_blocks;
    block_manager* bm;
    block_scheduler_algorithm<SwappableBlockType>* algo;

    //! Get an internal_block from the freelist or a newly allocated one if available.
    //! \return Pointer to the internal_block. NULL if none available.
    internal_block_type * get_free_internal_block()
    {
        if (! free_internal_blocks.empty())
        {
            // => there are internal_blocks in the free-list
            internal_block_type* iblock = free_internal_blocks.top();
            free_internal_blocks.pop();
            return iblock;
        }
        else if (remaining_internal_blocks > 0)
        {
            // => more internal_blocks can be allocated
            int_type num_blocks = std::min(max_internal_blocks_alloc_at_once, remaining_internal_blocks);
            remaining_internal_blocks -= num_blocks;
            internal_block_type* iblocks = new internal_block_type[num_blocks];
            internal_blocks_blocks.push(iblocks);
            for (int_type i = num_blocks - 1; i > 0; --i)
                free_internal_blocks.push(iblocks + i);
            return iblocks;
        }
        else
        {
            // => no internal_block available
            return 0;
        }
    }

    //! Return an internal_block to the freelist.
    void return_free_internal_block(internal_block_type* iblock)
    { free_internal_blocks.push(iblock); }

public:
    //! Create a block_scheduler with empty prediction sequence in simple mode.
    //! \param max_internal_memory Amount of internal memory (in bytes) the scheduler is allowed to use for acquiring, prefetching and caching.
    explicit block_scheduler(const int_type max_internal_memory)
        : max_internal_blocks(div_ceil(max_internal_memory, sizeof(internal_block_type))),
          remaining_internal_blocks(max_internal_blocks),
          bm(block_manager::get_instance()),
          algo(0)
    {
        algo = new block_scheduler_algorithm_online_lru<SwappableBlockType>(*this);
    }

    ~block_scheduler()
    {
        delete algo;
        int_type num_freed_internal_blocks = 0;
        if (free_swappable_blocks.size() != swappable_blocks.size())
        {
            // => not all swappable_blocks are free, at least deinitialize them
            STXXL_ERRMSG("not all swappable_blocks are free, those not acquired will be deinitialized");
            // evictable_blocks would suffice
            for (typename std::vector<SwappableBlockType>::iterator it = swappable_blocks.begin();
                 it != swappable_blocks.end(); ++it)
            {
                if (! it->is_acquired() && it->deinitialize())
                    // count internal_blocks that get freed
                    num_freed_internal_blocks++;
            }
        }
        if (int_type nlost = (max_internal_blocks - remaining_internal_blocks)
                             - (free_internal_blocks.size() + num_freed_internal_blocks)) {
            STXXL_ERRMSG(nlost << " internal_blocks are lost. They will get deallocated.");
        }
        while (! internal_blocks_blocks.empty())
        {
            delete[] internal_blocks_blocks.top();
            internal_blocks_blocks.pop();
        }
    }

    //! Acquire the given block.
    //! Has to be in pairs with release. Pairs may be nested and interleaved.
    //! \return Reference to the block's data.
    //! param sbid Swappable block to acquire.
    internal_block_type & acquire(const swappable_block_identifier_type sbid, const bool uninitialized = false)
    { return algo->acquire(sbid, uninitialized); }

    //! Release the given block.
    //! Has to be in pairs with acquire. Pairs may be nested and interleaved.
    //! \param sbid Swappable block to release.
    //! \param dirty If the data has been changed, invalidating possible data in external storage.
    void release(const swappable_block_identifier_type sbid, const bool dirty)
    { algo->release(sbid, dirty); }

    //! Drop all data in the given block, freeing in- and external memory.
    void deinitialize(const swappable_block_identifier_type sbid)
    { algo->deinitialize(sbid); }

    //! Initialize the swappable_block with the given external_block.
    //!
    //! It will use the the external_block for swapping and take care about
    //! it's deallocation. Has to be uninitialized.
    //! \param sbid identifier to the swappable_block
    //! \param eblock external_block a.k.a. bid
    void initialize(const swappable_block_identifier_type sbid, external_block_type eblock)
    { algo->initialize(sbid, eblock); }

    //! Deinitialize the swappable_block and return it's contents in an external_block.
    //!
    //! \param sbid identifier to the swappable_block
    //! \return external_block a.k.a. bid
    external_block_type extract_external_block(const swappable_block_identifier_type sbid)
    { return algo->extract_external_block(sbid); }

    //! check if the swappable_block is initialized.
    //! \param sbid identifier to the swappable_block
    //! \return if the swappable_block is initialized
    bool is_initialized(const swappable_block_identifier_type sbid) const
    { return algo->is_initialized(sbid); }

    //! Record a timestep in the prediction sequence to seperate consecutive
    //! acquire rsp. release-operations. Has an effect only in simulation mode.
    void explicit_timestep()
    { algo->explicit_timestep(); }

    //! Get a const reference to given block's data. Block has to be already acquired.
    //! \param sbid Swappable block to access.
    internal_block_type & get_internal_block(const swappable_block_identifier_type sbid) const
    { return swappable_blocks[sbid].get_internal_block(); }

    //! Allocate an uninitialized swappable_block.
    //! \return An identifier of the block.
    swappable_block_identifier_type allocate_swappable_block()
    {
        swappable_block_identifier_type sbid;
        if (free_swappable_blocks.empty())
        {
            // create new swappable_block
            sbid = swappable_blocks.size();
            swappable_blocks.resize(sbid + 1);
            algo->swappable_blocks_resize(sbid + 1);
        }
        else
        {
            // take swappable_block from freelist
            sbid = free_swappable_blocks.top();
            free_swappable_blocks.pop();
        }
        return sbid;
    }

    //! Free given no longer used temporary swappable_block.
    //! \param sbid Temporary swappable_block to free.
    void free_swappable_block(const swappable_block_identifier_type sbid)
    {
        deinitialize(sbid);
        free_swappable_blocks.push(sbid);
    }

    //! Returns if simulation mode is on, i.e. if a prediction sequence is being recorded.
    //! \return If simulation mode is on.
    bool is_simulating() const
    { return algo->is_simulating(); }

    //! Switch the used algorithm, e.g. to simulation etc..
    //! \param new_algo Pointer to the new algorithm object. Has to be instantiated to the block scheduler (or the old algorithm object).
    //! \return Pointer to the old algorithm object.
    block_scheduler_algorithm<SwappableBlockType> * switch_algorithm_to(block_scheduler_algorithm<SwappableBlockType>* new_algo)
    {
        assert(&new_algo->bs == this);
        block_scheduler_algorithm<SwappableBlockType>* old_algo = algo;
        algo = new_algo;
        return old_algo;
    }

    //! Return the current algorithm.
    block_scheduler_algorithm<SwappableBlockType> * get_current_algorithm() const
    {
        return algo;
    }

    //! Get the prediction_sequence.
    //! \return reference to the prediction_sequence
    const prediction_sequence_type & get_prediction_sequence() const
    { return algo->get_prediction_sequence(); }

    void flush()
    {
        std::vector<request_ptr> requests;
        while (! algo->evictable_blocks_empty())
        {
            swappable_block_identifier_type sbid = algo->evictable_blocks_pop();
            request_ptr rq = swappable_blocks[sbid].clean_async();
            if (rq.valid())
                requests.push_back(rq);
            return_free_internal_block(swappable_blocks[sbid].detach_internal_block());
        }
        for (typename std::vector<request_ptr>::reverse_iterator it = requests.rbegin();
             it != requests.rend(); ++it)
        {
            (*it)->wait();
        }
    }
};

template <class SwappableBlockType>
const int_type block_scheduler<SwappableBlockType>::max_internal_blocks_alloc_at_once = 128;

//! Interface of a block scheduling algorithm.
template <class SwappableBlockType>
class block_scheduler_algorithm : private noncopyable
{
protected:
    typedef block_scheduler<SwappableBlockType> block_scheduler_type;
    typedef typename block_scheduler_type::internal_block_type internal_block_type;
    typedef typename block_scheduler_type::external_block_type external_block_type;
    typedef typename block_scheduler_type::swappable_block_identifier_type swappable_block_identifier_type;
    typedef typename block_scheduler_type::prediction_sequence_type prediction_sequence_type;
    typedef typename block_scheduler_type::time_type time_type;

public:
    block_scheduler_type& bs;

protected:
    std::vector<SwappableBlockType>& swappable_blocks;
    prediction_sequence_type prediction_sequence;

    block_scheduler_algorithm * get_algorithm_from_block_scheduler()
    { return bs.algo; }

    //! Get an internal_block from the block_scheduler.
    //! \return Pointer to the internal_block. NULL if none available.
    internal_block_type * get_free_internal_block_from_block_scheduler()
    { return bs.get_free_internal_block(); }

    //! Return an internal_block to the block_scheduler.
    void return_free_internal_block_to_block_scheduler(internal_block_type* iblock)
    { bs.return_free_internal_block(iblock); }

public:
    block_scheduler_algorithm(block_scheduler_type& bs)
        : bs(bs),
          swappable_blocks(bs.swappable_blocks)
    { }

    block_scheduler_algorithm(block_scheduler_algorithm* old)
        : bs(old->bs),
          swappable_blocks(bs.swappable_blocks)
    { }

    virtual ~block_scheduler_algorithm() { }

    virtual bool evictable_blocks_empty() = 0;
    virtual swappable_block_identifier_type evictable_blocks_pop() = 0;
    virtual void swappable_blocks_resize(swappable_block_identifier_type /*size*/) { }

    virtual internal_block_type & acquire(const swappable_block_identifier_type sbid, const bool uninitialized = false) = 0;
    virtual void release(swappable_block_identifier_type sbid, const bool dirty) = 0;
    virtual void deinitialize(swappable_block_identifier_type sbid) = 0;
    virtual void initialize(swappable_block_identifier_type sbid, external_block_type eblock) = 0;
    virtual external_block_type extract_external_block(swappable_block_identifier_type sbid) = 0;

    virtual bool is_initialized(const swappable_block_identifier_type sbid) const
    { return swappable_blocks[sbid].is_initialized(); }

    virtual void explicit_timestep() { }
    virtual bool is_simulating() const
    { return false; }
    virtual const prediction_sequence_type & get_prediction_sequence() const
    { return prediction_sequence; }
};

//! Block scheduling algorithm caching via the least recently used policy (online).
template <class SwappableBlockType>
class block_scheduler_algorithm_online_lru : public block_scheduler_algorithm<SwappableBlockType>
{
protected:
    typedef block_scheduler<SwappableBlockType> block_scheduler_type;
    typedef block_scheduler_algorithm<SwappableBlockType> block_scheduler_algorithm_type;
    typedef typename block_scheduler_type::internal_block_type internal_block_type;
    typedef typename block_scheduler_type::external_block_type external_block_type;
    typedef typename block_scheduler_type::swappable_block_identifier_type swappable_block_identifier_type;

    using block_scheduler_algorithm_type::bs;
    using block_scheduler_algorithm_type::swappable_blocks;
    using block_scheduler_algorithm_type::get_algorithm_from_block_scheduler;
    using block_scheduler_algorithm_type::get_free_internal_block_from_block_scheduler;
    using block_scheduler_algorithm_type::return_free_internal_block_to_block_scheduler;

    //! Holds swappable blocks, whose internal block can be freed, i.e. that are internal but unacquired.
    addressable_fifo_queue<swappable_block_identifier_type> evictable_blocks;

    internal_block_type * get_free_internal_block()
    {
        // try to get a free internal_block
        if (internal_block_type * iblock = get_free_internal_block_from_block_scheduler())
            return iblock;
        // evict block
        assert(! evictable_blocks.empty()); // fails it there is not enough memory available
        return swappable_blocks[evictable_blocks.pop()].detach_internal_block();
    }

    void return_free_internal_block(internal_block_type* iblock)
    { return_free_internal_block_to_block_scheduler(iblock); }

    void init()
    {
        if (get_algorithm_from_block_scheduler())
            while (! get_algorithm_from_block_scheduler()->evictable_blocks_empty())
                evictable_blocks.insert(get_algorithm_from_block_scheduler()->evictable_blocks_pop());
    }

public:
    block_scheduler_algorithm_online_lru(block_scheduler_type& bs)
        : block_scheduler_algorithm_type(bs)
    { init(); }

    block_scheduler_algorithm_online_lru(block_scheduler_algorithm_type* old)
        : block_scheduler_algorithm_type(old)
    { init(); }

    virtual ~block_scheduler_algorithm_online_lru()
    {
        if (! evictable_blocks.empty())
            STXXL_ERRMSG("Destructing block_scheduler_algorithm_online that still holds evictable blocks. They get deinitialized.");
        while (! evictable_blocks.empty())
        {
            SwappableBlockType& sblock = swappable_blocks[evictable_blocks.pop()];
            if (internal_block_type * iblock = sblock.deinitialize())
                return_free_internal_block(iblock);
        }
    }

    virtual bool evictable_blocks_empty()
    { return evictable_blocks.empty(); }

    virtual swappable_block_identifier_type evictable_blocks_pop()
    { return evictable_blocks.pop(); }

    virtual internal_block_type & acquire(const swappable_block_identifier_type sbid, const bool uninitialized = false)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        /* acquired => internal -> increase reference count
           internal but not acquired -> remove from evictable_blocks, increase reference count
           not internal => uninitialized or external -> get internal_block, increase reference count
           uninitialized -> fill with default value
           external -> read */
        if (sblock.is_internal())
        {
            if (! sblock.is_acquired())
                // not acquired yet -> remove from evictable_blocks
                evictable_blocks.erase(sbid);
            sblock.acquire();
        }
        else if (sblock.is_initialized())
        {
            // => external but not internal
            //get internal_block
            sblock.attach_internal_block(get_free_internal_block());
            if (! uninitialized)
                //load block synchronously
                sblock.read_sync();
            sblock.acquire();
        }
        else
        {
            // => ! sblock.is_initialized()
            //get internal_block
            sblock.attach_internal_block(get_free_internal_block());
            sblock.acquire();
            //initialize new block
            if (! uninitialized)
                sblock.fill_default();
        }
        return sblock.get_internal_block();
    }

    virtual void release(swappable_block_identifier_type sbid, const bool dirty)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        sblock.make_dirty_if(dirty);
        sblock.release();
        if (! sblock.is_acquired())
        {
            if (sblock.is_dirty() || sblock.is_external())
                // => evictable, put in pq
                evictable_blocks.insert(sbid);
            else
                // => uninitialized, release internal block and put it in freelist
                return_free_internal_block(sblock.detach_internal_block());
        }
    }

    virtual void deinitialize(swappable_block_identifier_type sbid)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        if (sblock.is_evictable())
            evictable_blocks.erase(sbid);
        if (internal_block_type * iblock = sblock.deinitialize())
            return_free_internal_block(iblock);
    }

    virtual void initialize(swappable_block_identifier_type sbid, external_block_type eblock)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        sblock.initialize(eblock);
    }

    virtual external_block_type extract_external_block(swappable_block_identifier_type sbid)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        if (sblock.is_evictable())
            evictable_blocks.erase(sbid);
        if (sblock.is_internal())
            return_free_internal_block(sblock.detach_internal_block());
        return sblock.extract_external_block();
    }
};

//! Pseudo block scheduling algorithm only recording the request sequence.
template <class SwappableBlockType>
class block_scheduler_algorithm_simulation : public block_scheduler_algorithm<SwappableBlockType>
{
protected:
    typedef block_scheduler<SwappableBlockType> block_scheduler_type;
    typedef block_scheduler_algorithm<SwappableBlockType> block_scheduler_algorithm_type;
    typedef typename block_scheduler_type::internal_block_type internal_block_type;
    typedef typename block_scheduler_type::external_block_type external_block_type;
    typedef typename block_scheduler_type::swappable_block_identifier_type swappable_block_identifier_type;
    typedef typename block_scheduler_type::prediction_sequence_element prediction_sequence_element_type;
    typedef typename block_scheduler_algorithm_type::time_type time_type;

    using block_scheduler_algorithm_type::bs;
    using block_scheduler_algorithm_type::prediction_sequence;
    using block_scheduler_algorithm_type::swappable_blocks;
    using block_scheduler_algorithm_type::get_algorithm_from_block_scheduler;
    using block_scheduler_algorithm_type::get_free_internal_block_from_block_scheduler;
    using block_scheduler_algorithm_type::return_free_internal_block_to_block_scheduler;

    //! Holds swappable blocks, whose internal block can be freed, i.e. that are internal but unacquired.
    std::stack<swappable_block_identifier_type> evictable_blocks;
    time_type time_count;
    bool last_op_release;
    std::vector<int_type> reference_counts;
    internal_block_type dummy_block;

    void return_free_internal_block(internal_block_type* iblock)
    { return_free_internal_block_to_block_scheduler(iblock); }

    void init()
    {
        if (get_algorithm_from_block_scheduler())
            while (! get_algorithm_from_block_scheduler()->evictable_blocks_empty())
                evictable_blocks.push(get_algorithm_from_block_scheduler()->evictable_blocks_pop());
        for (swappable_block_identifier_type i = 0; i < reference_counts.size(); ++i)
            reference_counts[i] = swappable_blocks[i].is_initialized();
    }

public:
    block_scheduler_algorithm_simulation(block_scheduler_type& bs)
        : block_scheduler_algorithm_type(bs),
          time_count(0),
          last_op_release(false),
          reference_counts(swappable_blocks.size())
    { init(); }

    block_scheduler_algorithm_simulation(block_scheduler_algorithm_type* old)
        : block_scheduler_algorithm_type(old),
          time_count(0),
          last_op_release(false),
          reference_counts(swappable_blocks.size())
    { init(); }

    virtual ~block_scheduler_algorithm_simulation()
    {
        if (! evictable_blocks.empty())
            STXXL_ERRMSG("Destructing block_scheduler_algorithm_record_prediction_sequence that still holds evictable blocks. They get deinitialized.");
        while (! evictable_blocks.empty())
        {
            SwappableBlockType& sblock = swappable_blocks[evictable_blocks.top()];
            if (internal_block_type * iblock = sblock.deinitialize())
                return_free_internal_block(iblock);
            evictable_blocks.pop();
        }
    }

    virtual bool evictable_blocks_empty()
    { return evictable_blocks.empty(); }

    virtual swappable_block_identifier_type evictable_blocks_pop()
    {
        swappable_block_identifier_type sbid = evictable_blocks.top();
        evictable_blocks.pop();
        return sbid;
    }

    virtual internal_block_type & acquire(const swappable_block_identifier_type sbid, const bool uninitialized = false)
    {
        ++reference_counts[sbid];
        last_op_release = false;
        if (uninitialized)
            prediction_sequence.push_back(
                prediction_sequence_element_type(block_scheduler_type::op_acquire_uninitialized, sbid, time_count)
                );
        else
            prediction_sequence.push_back(
                prediction_sequence_element_type(block_scheduler_type::op_acquire, sbid, time_count)
                );
        return dummy_block;
    }

    virtual void release(swappable_block_identifier_type sbid, const bool dirty)
    {
        --reference_counts[sbid] += dirty;
        time_count += ! last_op_release;
        last_op_release = true;
        if (dirty)
            prediction_sequence.push_back(
                prediction_sequence_element_type(block_scheduler_type::op_release_dirty, sbid, time_count)
                );
        else
            prediction_sequence.push_back(
                prediction_sequence_element_type(block_scheduler_type::op_release, sbid, time_count)
                );
    }

    virtual void deinitialize(swappable_block_identifier_type sbid)
    {
        reference_counts[sbid] = false;
        prediction_sequence.push_back(
            prediction_sequence_element_type(block_scheduler_type::op_deinitialize, sbid, time_count)
            );
    }

    virtual void initialize(swappable_block_identifier_type sbid, external_block_type)
    {
        reference_counts[sbid] = true;
        prediction_sequence.push_back(
            prediction_sequence_element_type(block_scheduler_type::op_initialize, sbid, time_count)
            );
    }

    virtual external_block_type extract_external_block(swappable_block_identifier_type sbid)
    {
        reference_counts[sbid] = false;
        prediction_sequence.push_back(
            prediction_sequence_element_type(block_scheduler_type::op_extract_external_block, sbid, time_count)
            );
        return external_block_type();
    }

    virtual void swappable_blocks_resize(swappable_block_identifier_type size)
    {
        reference_counts.resize(size, 0);
    }

    virtual bool is_initialized(const swappable_block_identifier_type sbid) const
    {
        return reference_counts[sbid] > 0;
    }

    virtual void explicit_timestep()
    { ++time_count; }

    virtual bool is_simulating() const
    { return true; }
};

//! Block scheduling algorithm caching via the longest forward distance policy (offline).
template <class SwappableBlockType>
class block_scheduler_algorithm_offline_lfd : public block_scheduler_algorithm<SwappableBlockType>
{
protected:
    typedef block_scheduler<SwappableBlockType> block_scheduler_type;
    typedef block_scheduler_algorithm<SwappableBlockType> block_scheduler_algorithm_type;
    typedef typename block_scheduler_type::internal_block_type internal_block_type;
    typedef typename block_scheduler_type::external_block_type external_block_type;
    typedef typename block_scheduler_type::swappable_block_identifier_type swappable_block_identifier_type;
    typedef typename block_scheduler_algorithm_type::time_type time_type;
    typedef typename block_scheduler_type::prediction_sequence_type prediction_sequence_type;

    using block_scheduler_algorithm_type::bs;
    using block_scheduler_algorithm_type::swappable_blocks;
    using block_scheduler_algorithm_type::get_algorithm_from_block_scheduler;
    using block_scheduler_algorithm_type::get_free_internal_block_from_block_scheduler;
    using block_scheduler_algorithm_type::return_free_internal_block_to_block_scheduler;

    class priority
    {
        unsigned_type p;

    public:
        priority(const SwappableBlockType& sblock, const std::pair<bool, time_type>& t)
        {
            // p larger => evict earlier
            if (t.first)
            {
                // most significant: next use
                p = unsigned_type(t.second) << 2;
                // less significant: not dirty
                p |= unsigned_type(! sblock.is_dirty()) << 1;
                // less significant: has external_block
                p |= unsigned_type(sblock.has_external_block()) << 0;
            }
            else
            {
                // most significant: next use
                p = std::numeric_limits<unsigned_type>::max() << 2;
                // less significant: next operation: extract > accessed no more > deinitialize
                p |= unsigned_type(t.second) << 0;
            }
        }

        // less => evict earlier
        bool operator < (const priority& right) const
        { return p > right.p; }
    };

    //! Holds swappable blocks, whose internal block can be freed, i.e. that are internal but unacquired.
    addressable_priority_queue<swappable_block_identifier_type, priority> evictable_blocks;
    /*!
     * Stores for the sequence of releases extracted from the prediction_sequence:
     * (true, timestamp of the blocks next acquire) if it is acquired next
     * (false, 0) if it is deinitialized next
     * (false, 1) if it is not accessed any more
     * (false, 2) if it is extracted next
     * (false, 3) if it is initialized next
     */
    std::deque<std::pair<bool, time_type> > next_use;

    internal_block_type * get_free_internal_block()
    {
        // try to get a free internal_block
        if (internal_block_type * iblock = get_free_internal_block_from_block_scheduler())
            return iblock;
        // evict block
        assert(! evictable_blocks.empty()); // fails it there is not enough memory available
        return swappable_blocks[evictable_blocks.pop()].detach_internal_block();
    }

    void return_free_internal_block(internal_block_type* iblock)
    { return_free_internal_block_to_block_scheduler(iblock); }

    void init(block_scheduler_algorithm_type* old_algo)
    {
        std::vector<std::pair<bool, time_type> >
        blocks_next_acquire(swappable_blocks.size(), std::make_pair(false, 1));
        if (old_algo)
        {
            // precomputations for priorities: init next_acquires
            const prediction_sequence_type& ps = old_algo->get_prediction_sequence();
            for (typename prediction_sequence_type::const_reverse_iterator it = ps.rbegin(); it != ps.rend(); ++it)
            {
                switch (it->op)
                {
                case (block_scheduler_type::op_acquire):
                case (block_scheduler_type::op_acquire_uninitialized):
                    blocks_next_acquire[it->id] = std::make_pair(true, it->time);
                    break;
                case (block_scheduler_type::op_release):
                case (block_scheduler_type::op_release_dirty):
                    next_use.push_front(blocks_next_acquire[it->id]);
                    break;
                case (block_scheduler_type::op_deinitialize):
                    blocks_next_acquire[it->id] = std::make_pair(false, 0);
                    break;
                case (block_scheduler_type::op_initialize):
                    blocks_next_acquire[it->id] = std::make_pair(false, 3);
                    break;
                case (block_scheduler_type::op_extract_external_block):
                    blocks_next_acquire[it->id] = std::make_pair(false, 2);
                    break;
                }
            }
        }
        if (get_algorithm_from_block_scheduler())
        {
            while (! get_algorithm_from_block_scheduler()->evictable_blocks_empty())
            {
                // insert already evictable blocks with the right priority
                const swappable_block_identifier_type sbid = get_algorithm_from_block_scheduler()->evictable_blocks_pop();
                evictable_blocks.insert(sbid, priority(swappable_blocks[sbid], blocks_next_acquire[sbid]));
            }
        }
    }

public:
    block_scheduler_algorithm_offline_lfd(block_scheduler_type& bs)
        : block_scheduler_algorithm_type(bs)
    { init(get_algorithm_from_block_scheduler()); }

    // It is possible to keep an old simulation-algorithm object and reuse it's prediction sequence
    block_scheduler_algorithm_offline_lfd(block_scheduler_algorithm_type* old)
        : block_scheduler_algorithm_type(old)
    { init(old); }

    virtual ~block_scheduler_algorithm_offline_lfd()
    {
        if (! evictable_blocks.empty())
            STXXL_ERRMSG("Destructing block_scheduler_algorithm_offline_lfd that still holds evictable blocks. They get deinitialized.");
        while (! evictable_blocks.empty())
        {
            SwappableBlockType& sblock = swappable_blocks[evictable_blocks.pop()];
            if (internal_block_type * iblock = sblock.deinitialize())
                return_free_internal_block(iblock);
        }
    }

    virtual bool evictable_blocks_empty()
    { return evictable_blocks.empty(); }

    virtual swappable_block_identifier_type evictable_blocks_pop()
    { return evictable_blocks.pop(); }

    virtual internal_block_type & acquire(const swappable_block_identifier_type sbid, const bool uninitialized = false)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        /* acquired => internal -> increase reference count
           internal but not acquired -> remove from evictable_blocks, increase reference count
           not intern => uninitialized or external -> get internal_block, increase reference count
           uninitialized -> fill with default value
           external -> read */
        if (sblock.is_internal())
        {
            if (! sblock.is_acquired())
                // not acquired yet -> remove from evictable_blocks
                evictable_blocks.erase(sbid);
            sblock.acquire();
        }
        else if (sblock.is_initialized())
        {
            // => external but not internal
            //get internal_block
            sblock.attach_internal_block(get_free_internal_block());
            if (! uninitialized)
                //load block synchronously
                sblock.read_sync();
            sblock.acquire();
        }
        else
        {
            // => ! sblock.is_initialized()
            //get internal_block
            sblock.attach_internal_block(get_free_internal_block());
            sblock.acquire();
            //initialize new block
            if (! uninitialized)
                sblock.fill_default();
        }
        return sblock.get_internal_block();
    }

    virtual void release(swappable_block_identifier_type sbid, const bool dirty)
    {
        if (next_use.empty())
        {
            STXXL_ERRMSG("block_scheduler_algorithm_offline_lfd got release-request but prediction sequence ended. Switching to block_scheduler_algorithm_online.");
            // switch algorithm
            block_scheduler_algorithm_type* new_algo, * old_algo;
            new_algo = new block_scheduler_algorithm_online_lru<SwappableBlockType>(bs);
            old_algo = bs.switch_algorithm_to(new_algo);
            // redirect call
            new_algo->release(sbid, dirty);
            // delete self
            delete old_algo;
            return;
        }
        SwappableBlockType& sblock = swappable_blocks[sbid];
        sblock.make_dirty_if(dirty);
        sblock.release();
        if (! sblock.is_acquired())
        {
            if (sblock.is_dirty() || sblock.is_external())
                // => evictable, put in pq
                evictable_blocks.insert(sbid, priority(swappable_blocks[sbid], next_use.front()));
            else
                // => uninitialized, release internal block and put it in freelist
                return_free_internal_block(sblock.detach_internal_block());
        }
        next_use.pop_front();
    }

    virtual void deinitialize(swappable_block_identifier_type sbid)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        if (sblock.is_evictable())
            evictable_blocks.erase(sbid);
        if (internal_block_type * iblock = sblock.deinitialize())
            return_free_internal_block(iblock);
    }

    virtual void initialize(swappable_block_identifier_type sbid, external_block_type eblock)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        sblock.initialize(eblock);
    }

    virtual external_block_type extract_external_block(swappable_block_identifier_type sbid)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        if (sblock.is_evictable())
            evictable_blocks.erase(sbid);
        if (sblock.is_internal())
            return_free_internal_block(sblock.detach_internal_block());
        return sblock.extract_external_block();
    }
};

//! Block scheduling algorithm caching via the least recently used policy
//! (offline), and prefetching in addition.
template <class SwappableBlockType>
class block_scheduler_algorithm_offline_lru_prefetching : public block_scheduler_algorithm<SwappableBlockType>
{
protected:
    struct scheduled_block_meta;
    struct write_read_request;

    typedef block_scheduler<SwappableBlockType> block_scheduler_type;
    typedef block_scheduler_algorithm<SwappableBlockType> block_scheduler_algorithm_type;
    typedef typename block_scheduler_type::internal_block_type internal_block_type;
    typedef typename block_scheduler_type::external_block_type external_block_type;
    typedef typename block_scheduler_type::swappable_block_identifier_type swappable_block_identifier_type;
    typedef typename block_scheduler_algorithm_type::time_type time_type;
    typedef typename block_scheduler_type::prediction_sequence_type prediction_sequence_type;
    typedef typename block_scheduler_type::block_scheduler_operation block_scheduler_operation;
    typedef typename std::vector<SwappableBlockType>::iterator swappable_blocks_iterator;

    typedef std::map<swappable_block_identifier_type, scheduled_block_meta> scheduled_blocks_type;
    typedef typename scheduled_blocks_type::iterator scheduled_blocks_iterator;
    typedef typename scheduled_blocks_type::reference scheduled_blocks_reference;
    typedef std::map<swappable_block_identifier_type, write_read_request*> write_scheduled_blocks_type;
    typedef typename write_scheduled_blocks_type::iterator write_scheduled_blocks_iterator;
    typedef typename write_scheduled_blocks_type::reference write_scheduled_blocks_reference;

    using block_scheduler_algorithm_type::bs;
    using block_scheduler_algorithm_type::swappable_blocks;
    using block_scheduler_algorithm_type::get_algorithm_from_block_scheduler;
    using block_scheduler_algorithm_type::get_free_internal_block_from_block_scheduler;
    using block_scheduler_algorithm_type::return_free_internal_block_to_block_scheduler;
    using block_scheduler_algorithm_type::prediction_sequence;

    struct write_read_request
    {
        bool write_done_soon;                          // set by read_after_write, checked by schedule_read()
        bool shall_read;                               // checked by read_after_write, set by schedule_read()
        swappable_blocks_iterator block_to_start_read; // used by read_after_write, set by schedule_read()
        scheduled_blocks_iterator taker;               // read_req set by read_after_write
        request_ptr write_req;                         // completes with read_after_write

        write_read_request()
            : write_done_soon(false), shall_read(false), block_to_start_read(), taker(), write_req(0) { }
    };

    struct read_after_write
    {
        write_read_request* wrr;

        read_after_write(write_read_request* write_read_req)
            : wrr(write_read_req) { }

        void operator () (request*)
        {
            wrr->write_done_soon = true;
            if (wrr->shall_read)
                wrr->taker->second.read_req = wrr->block_to_start_read->read_async();
        }
    };

    struct scheduled_block_meta
    {
        internal_block_type* reserved_iblock;
        std::pair<bool, swappable_block_identifier_type> giver;
        request_ptr read_req;
        std::deque<block_scheduler_operation> operations; // invariant: not empty; front: last scheduled operation, back: upcoming operation

        scheduled_block_meta(block_scheduler_operation op)
            : reserved_iblock(0),
              giver(false, 0),
              read_req(0),
              operations()
        { operations.push_front(op); }
    };

    //! Holds swappable blocks, whose internal block can be freed, i.e. that are internal but unacquired.
    std::set<swappable_block_identifier_type> free_evictable_blocks;
    std::set<swappable_block_identifier_type> scheduled_evictable_blocks;

    //! Holds not internal swappable_blocks, whose next access has already been scheduled.
    scheduled_blocks_type scheduled_blocks;

    //! Holds swappable_blocks, whose internal block has been taken away but the clean did not finish yet.
    write_scheduled_blocks_type write_scheduled_blocks;
    typename prediction_sequence_type::iterator next_op_to_schedule;

    //! Schedule an internal, possibly dirty swappable_block to write.
    //!
    //! The block becomes not dirty. if it was dirty, an entry in write_scheduled_blocks is made referencing the write_read_request.
    //! \param sbid block to write
    //! \return pointer to the write_read_request
    write_read_request * schedule_write(const swappable_block_identifier_type sbid)
    {
        SwappableBlockType& sblock = swappable_blocks[sbid];
        write_read_request* wrr = new write_read_request;
        wrr->write_req = sblock.clean_async(read_after_write(wrr));
        if (wrr->write_req.valid())
        {
            bool t = write_scheduled_blocks.insert(std::make_pair(sbid, wrr)).second;
            STXXL_ASSERT(t);
            return wrr;
        }
        else
        {
            delete wrr;
            return 0;
        }
    }

    //! Try to interrupt a read scheduled in a write_read_request.
    //!
    //! side-effect: possibly erases entry from write_scheduled_blocks, so the iterator writing_block may become invalid
    //! \return if successful
    bool try_interrupt_read(const write_scheduled_blocks_iterator& writing_block)
    {
        // stop read
        writing_block->second->shall_read = false;
        // check if stopped
        if (! writing_block->second->write_done_soon)
            return true;
        // => possibly to late
        scheduled_blocks_reference taker = *writing_block->second->taker;
        // wait
        wait_on_write(writing_block);
        // check if read started
        if (taker.second.read_req.valid())
            // => read started, to late
            return false;
        else
            // => just in time
            return true;
    }

    //! Schedule an internal and external block to read.
    //!
    //! If the giver is still writing, schedule read via its write_read_request.
    void schedule_read(scheduled_blocks_iterator block_to_read)
    {
        // first check if block_to_read is still writing. do not read before write finished
        // wait_on_write(block_to_read->first);

        write_scheduled_blocks_iterator it = write_scheduled_blocks.find(block_to_read->first);
        if (it != write_scheduled_blocks.end())
        {
            scheduled_blocks_iterator other_block_to_read = it->second->taker;
            // check if scheduled to read
            if (it->second->shall_read)
            {
                if (try_interrupt_read(it))
                {
                    // => interrupted, swap internal_blocks
                    std::swap(other_block_to_read->second.giver, block_to_read->second.giver);
                    if (other_block_to_read->second.giver.first)
                    {
                        write_scheduled_blocks_iterator it = write_scheduled_blocks.find(other_block_to_read->second.giver.second);
                        if (it != write_scheduled_blocks.end())
                            it->second->taker = other_block_to_read;
                        else
                            other_block_to_read->second.giver.first = false;
                    }
                    if (block_to_read->second.giver.first)
                    {
                        write_scheduled_blocks_iterator it = write_scheduled_blocks.find(block_to_read->second.giver.second);
                        if (it != write_scheduled_blocks.end())
                            it->second->taker = block_to_read;
                        else
                            block_to_read->second.giver.first = false;
                    }

                    internal_block_type* tmp_iblock = swappable_blocks[block_to_read->first].detach_internal_block();
                    swappable_blocks[block_to_read->first].attach_internal_block(
                        swappable_blocks[other_block_to_read->first].detach_internal_block());
                    swappable_blocks[other_block_to_read->first].attach_internal_block(tmp_iblock);
                    // => this block has its internal_block back, no need to read
                    // reschedule other
                    schedule_read(other_block_to_read);
                    return;
                }
                // else => read already started, but write done -> read this
            }
            else
            {
                // => no read scheduled, swap internal_blocks
                std::swap(other_block_to_read->second.giver, block_to_read->second.giver);
                if (other_block_to_read->second.giver.first)
                {
                    write_scheduled_blocks_iterator it = write_scheduled_blocks.find(other_block_to_read->second.giver.second);
                    if (it != write_scheduled_blocks.end())
                        it->second->taker = other_block_to_read;
                    else
                        other_block_to_read->second.giver.first = false;
                }
                if (block_to_read->second.giver.first)
                {
                    write_scheduled_blocks_iterator it = write_scheduled_blocks.find(block_to_read->second.giver.second);
                    if (it != write_scheduled_blocks.end())
                        it->second->taker = block_to_read;
                    else
                        block_to_read->second.giver.first = false;
                }

                internal_block_type* tmp_iblock = swappable_blocks[block_to_read->first].detach_internal_block();
                swappable_blocks[block_to_read->first].attach_internal_block(
                    other_block_to_read->second.reserved_iblock);
                other_block_to_read->second.reserved_iblock = tmp_iblock;
                // => this block has its internal_block back, no need to read
                return;
            }
        }

        // schedule block_to_read to read
        if (block_to_read->second.giver.first)
        {
            write_scheduled_blocks_iterator writing_block = write_scheduled_blocks.find(block_to_read->second.giver.second);
            if (writing_block != write_scheduled_blocks.end())
            {
                // => there is a write scheduled
                // tell the completion handler that we want a read
                writing_block->second->block_to_start_read = swappable_blocks.begin() + block_to_read->first;
                writing_block->second->taker = block_to_read;
                writing_block->second->shall_read = true;
                // and check if it is not to late
                if (writing_block->second->write_done_soon)
                {
                    // => the completion handler may have missed our wish to read
                    // so wait for it to finish and check
                    wait_on_write(writing_block);
                    block_to_read->second.giver.first = false;
                    if (block_to_read->second.read_req.valid())
                        // read scheduled
                        return;
                }
                else
                    // read scheduled
                    return;
            }
            else
                block_to_read->second.giver.first = false;
        }
        // => read could not be scheduled through the completion handler
        block_to_read->second.read_req = swappable_blocks[block_to_read->first].read_async();
    }

    //! Wait for the write to finish.
    //!
    //! side-effect: erases entry from write_scheduled_blocks
    void wait_on_write(const write_scheduled_blocks_iterator& writing_block)
    {
        writing_block->second->write_req->wait();
        delete writing_block->second;
        write_scheduled_blocks.erase(writing_block);
    }

    //! Wait for the write to finish.
    //!
    //! side-effect: erases entry from write_scheduled_blocks
    void wait_on_write(const swappable_block_identifier_type& writing_block)
    {
        write_scheduled_blocks_iterator it = write_scheduled_blocks.find(writing_block);
        if (it != write_scheduled_blocks.end())
            wait_on_write(it);
    }

    //! Wait for the write of the giver to finish.
    //!
    //! side-effect: erases entry from write_scheduled_blocks
    void wait_on_write(const scheduled_blocks_iterator& schedule_meta)
    {
        if (schedule_meta->second.giver.first)
        {
            wait_on_write(schedule_meta->second.giver.second);
            schedule_meta->second.giver.first = false;
        }
    }

    //! Wait for the read to finish.
    //!
    //! side-effect: erases entry for the write of the giver from write_scheduled_blocks
    void wait_on_read(const scheduled_blocks_iterator& schedule_meta)
    {
        wait_on_write(schedule_meta);
        if (schedule_meta->second.read_req.valid())
        {
            schedule_meta->second.read_req->wait();
            schedule_meta->second.read_req = 0;
        }
    }

    //! Wait for the write of the giver to finish and return reserved internal_block.
    //!
    //! side-effect: erases entry for the write of the giver from write_scheduled_blocks
    internal_block_type * get_ready_block(const scheduled_blocks_iterator& schedule_meta)
    {
        wait_on_write(schedule_meta);
        internal_block_type* r = schedule_meta->second.reserved_iblock;
        schedule_meta->second.reserved_iblock = 0;
        return r;
    }

    bool shall_keep_internal_block(const scheduled_blocks_iterator& schedule_meta, const bool ignore_first = true) const
    {
        // returns true iif there is an acquire or acquire_uninitialized scheduled or there is a deinitialize scheduled and the block is dirty
        for (typename std::deque<block_scheduler_operation>::reverse_iterator
             rit = schedule_meta->second.operations.rbegin() + ignore_first;
             rit != schedule_meta->second.operations.rend(); ++rit)
        {
            switch (*rit)
            {
            case block_scheduler_type::op_acquire:
            case block_scheduler_type::op_acquire_uninitialized:
                return true;
                break;
            case block_scheduler_type::op_release:
            case block_scheduler_type::op_release_dirty:
                break;
            case block_scheduler_type::op_deinitialize:
                if (swappable_blocks[schedule_meta->first].is_dirty()) return true; break;
            case block_scheduler_type::op_initialize:
            case block_scheduler_type::op_extract_external_block:
                break;
            }
        }
        return false;
    }

    // assumes the current operation to be still in operations
    bool shall_be_cleaned(const scheduled_blocks_iterator& schedule_meta) const
    {
        // returns true iif there is an extract_external_block scheduled and no release_dirty, deinitialize or initialize before
        for (typename std::deque<block_scheduler_operation>::reverse_iterator
             rit = schedule_meta->second.operations.rbegin() + 1;
             rit != schedule_meta->second.operations.rend(); ++rit)
        {
            switch (*rit)
            {
            case block_scheduler_type::op_acquire:
            case block_scheduler_type::op_acquire_uninitialized:
            case block_scheduler_type::op_release:
                break;
            case block_scheduler_type::op_release_dirty:
            case block_scheduler_type::op_deinitialize:
            case block_scheduler_type::op_initialize:
                return false;
                break;
            case block_scheduler_type::op_extract_external_block:
                return true;
                break;
            }
        }
        return false;
    }

    bool shall_be_read(const scheduled_blocks_iterator& schedule_meta, const bool ignore_first = true) const
    {
        // returns true iif there is an acquire scheduled next and the block is initialized
        return swappable_blocks[schedule_meta->first].is_initialized()
               && schedule_meta->second.operations.rbegin() + ignore_first != schedule_meta->second.operations.rend()
               && *(schedule_meta->second.operations.rbegin() + ignore_first) == block_scheduler_type::op_acquire;
    }

    void operation_done(scheduled_blocks_iterator& schedule_meta)
    {
        schedule_meta->second.operations.pop_back();
        if (schedule_meta->second.operations.empty())
        {
            assert(! schedule_meta->second.giver.first);
            scheduled_blocks.erase(schedule_meta);
        }
    }

    block_scheduler_algorithm_type * give_up(std::string err_msg = "detected some error in the prediction sequence")
    {
        STXXL_ERRMSG("block_scheduler_algorithm_offline_lru_prefetching: " << err_msg << ". Switching to block_scheduler_algorithm_online.");
        // switch algorithm
        block_scheduler_algorithm_type* new_algo
            = new block_scheduler_algorithm_online_lru<SwappableBlockType>(bs);
        // and delete self
        delete bs.switch_algorithm_to(new_algo);
        return new_algo;
    }

    void return_free_internal_block(internal_block_type* iblock)
    { return_free_internal_block_to_block_scheduler(iblock); }

    void schedule_next_operations()
    {
        while (next_op_to_schedule != prediction_sequence.end())
        {
            // list operation in scheduled_blocks
            std::pair<scheduled_blocks_iterator, bool> ins_res = scheduled_blocks.insert(
                std::make_pair(next_op_to_schedule->id, next_op_to_schedule->op));
            scheduled_blocks_iterator schedule_meta = ins_res.first;
            if (! ins_res.second)
                schedule_meta->second.operations.push_front(next_op_to_schedule->op);
            SwappableBlockType& sblock = swappable_blocks[next_op_to_schedule->id];

            // do appropriate preparations
            if (next_op_to_schedule->op == block_scheduler_type::op_acquire
                || next_op_to_schedule->op == block_scheduler_type::op_acquire_uninitialized)
            {
                if (sblock.is_internal())
                {
                    if (free_evictable_blocks.erase(next_op_to_schedule->id))
                        scheduled_evictable_blocks.insert(next_op_to_schedule->id);
                }
                else
                {
                    if (! schedule_meta->second.reserved_iblock)
                    {
                        // => needs internal_block -> try to get one
                        // -> try to get one from block_scheduler
                        schedule_meta->second.reserved_iblock = get_free_internal_block_from_block_scheduler();
                        if (! schedule_meta->second.reserved_iblock)
                        {
                            // -> try to get one by evicting
                            if (free_evictable_blocks.empty())
                            {
                                // => can not schedule acquire
                                // remove operation from scheduled_blocks
                                if (ins_res.second)
                                    scheduled_blocks.erase(ins_res.first);
                                else
                                    schedule_meta->second.operations.pop_front();
                                // stop scheduling
                                return;
                            }
                            swappable_block_identifier_type giver = pop_begin(free_evictable_blocks);
                            {
                                assert(scheduled_blocks.find(giver) == scheduled_blocks.end() ||
                                       !shall_keep_internal_block(scheduled_blocks.find(giver), false));
                            }
                            write_read_request* wrr = schedule_write(giver);
                            schedule_meta->second.giver.first = (wrr != NULL);
                            schedule_meta->second.giver.second = giver;
                            schedule_meta->second.reserved_iblock = swappable_blocks[giver].detach_internal_block();
                            if (wrr)
                                wrr->taker = schedule_meta;
                        }
                        // read if desired
                        if (shall_be_read(schedule_meta, false))
                        {
                            // => there is no operation scheduled for this block before this acquire and it is initialized
                            // -> start prefetching now
                            sblock.attach_internal_block(schedule_meta->second.reserved_iblock);
                            schedule_meta->second.reserved_iblock = 0;
                            scheduled_evictable_blocks.insert(next_op_to_schedule->id);
                            schedule_read(schedule_meta);
                        }
                    }
                }
            }
            else if (next_op_to_schedule->op == block_scheduler_type::op_deinitialize)
            {
                if (sblock.is_dirty())
                    if (free_evictable_blocks.erase(next_op_to_schedule->id))
                        scheduled_evictable_blocks.insert(next_op_to_schedule->id);
            }

            ++next_op_to_schedule;
        }
        for (typename std::set<swappable_block_identifier_type>::iterator it = free_evictable_blocks.begin();
             it != free_evictable_blocks.end(); ++it)
        {
            if (! write_scheduled_blocks.count(*it))
                schedule_write(*it);
        }
    }

    void init(block_scheduler_algorithm_type* old_algo)
    {
        if (old_algo)
            // copy prediction sequence
            prediction_sequence = old_algo->get_prediction_sequence();
        next_op_to_schedule = prediction_sequence.begin();
        if (get_algorithm_from_block_scheduler())
            while (! get_algorithm_from_block_scheduler()->evictable_blocks_empty())
                free_evictable_blocks.insert(get_algorithm_from_block_scheduler()->evictable_blocks_pop());
        schedule_next_operations();
    }

    void deinit()
    {
        // todo remove
        if (! scheduled_blocks.empty())
            STXXL_MSG("deinit while scheduled_blocks not empty");
        if (! scheduled_evictable_blocks.empty())
            STXXL_MSG("deinit while scheduled_evictable_blocks not empty");

        // empty scheduled_blocks
        free_evictable_blocks.insert(scheduled_evictable_blocks.begin(), scheduled_evictable_blocks.end());
        //for (typename std::set<swappable_block_identifier_type>::iterator it = scheduled_evictable_blocks.begin();
        //        it != scheduled_evictable_blocks.end(); ++it)
        //    free_evictable_blocks.insert(*it);
        scheduled_evictable_blocks.clear();
        while (! scheduled_blocks.empty())
        {
            scheduled_blocks_iterator it = scheduled_blocks.begin();
            wait_on_read(it);
            if (it->second.reserved_iblock)
                return_free_internal_block(it->second.reserved_iblock);
            scheduled_blocks.erase(it);
        }
        while (! write_scheduled_blocks.empty())
        {
            write_scheduled_blocks_iterator it = write_scheduled_blocks.begin();
            wait_on_write(it);
        }
    }

public:
    block_scheduler_algorithm_offline_lru_prefetching(block_scheduler_type& bs)
        : block_scheduler_algorithm_type(bs)
    { init(get_algorithm_from_block_scheduler()); }

    // It is possible to keep an old simulation-algorithm object and reuse it's prediction sequence
    block_scheduler_algorithm_offline_lru_prefetching(block_scheduler_algorithm_type* old)
        : block_scheduler_algorithm_type(old)
    { init(old); }

    virtual ~block_scheduler_algorithm_offline_lru_prefetching()
    {
        deinit();
        if (! free_evictable_blocks.empty())
            STXXL_ERRMSG("Destructing block_scheduler_algorithm_offline_lru_prefetching that still holds evictable blocks. They get deinitialized.");
        while (! free_evictable_blocks.empty())
        {
            SwappableBlockType& sblock = swappable_blocks[pop_begin(free_evictable_blocks)];
            if (internal_block_type * iblock = sblock.deinitialize())
                return_free_internal_block(iblock);
        }
    }

    virtual bool evictable_blocks_empty()
    {
        deinit();
        return free_evictable_blocks.empty();
    }

    virtual swappable_block_identifier_type evictable_blocks_pop()
    { return pop_begin(free_evictable_blocks); }

    virtual internal_block_type & acquire(const swappable_block_identifier_type sbid, const bool uninitialized = false)
    {
        assert(! prediction_sequence.empty());
        assert(prediction_sequence.front().op ==
               ((uninitialized) ? block_scheduler_type::op_acquire_uninitialized : block_scheduler_type::op_acquire));
        assert(prediction_sequence.front().id == sbid);
        prediction_sequence.pop_front();
        scheduled_blocks_iterator schedule_meta = scheduled_blocks.find(sbid);
        assert(schedule_meta != scheduled_blocks.end());                                                               // acquire not scheduled or out of internal_blocks (i.e. not enough internal memory)
        assert(schedule_meta->second.operations.back() ==
               ((uninitialized) ? block_scheduler_type::op_acquire_uninitialized : block_scheduler_type::op_acquire)); // acquire not scheduled or out of internal_blocks (i.e. not enough internal memory)

        SwappableBlockType& sblock = swappable_blocks[sbid];
        /* acquired => internal -> increase reference count
           internal but not acquired -> remove from scheduled_evictable_blocks, increase reference count
           not internal => uninitialized or external -> get internal_block, increase reference count
           uninitialized -> fill with default value
           external -> read */
        if (sblock.is_internal())
        {
            if (! sblock.is_acquired())
            {
                // not acquired yet -> remove from scheduled_evictable_blocks
                size_t t = scheduled_evictable_blocks.erase(sbid);
                STXXL_ASSERT(t != 0);
                wait_on_read(schedule_meta);
            }
            sblock.acquire();
        }
        else
        {
            assert(uninitialized || ! sblock.is_initialized()); // initialized blocks should be scheduled to read and thus internal
            //get internal_block
            sblock.attach_internal_block(get_ready_block(schedule_meta));
            sblock.acquire();
            //initialize new block
            if (! uninitialized)
                sblock.fill_default();
        }

        operation_done(schedule_meta);
        return sblock.get_internal_block();
    }

    virtual void release(swappable_block_identifier_type sbid, const bool dirty)
    {
        assert(! prediction_sequence.empty());
        assert(prediction_sequence.front().op ==
               ((dirty) ? block_scheduler_type::op_release_dirty : block_scheduler_type::op_release));
        assert(prediction_sequence.front().id == sbid);
        prediction_sequence.pop_front();
        scheduled_blocks_iterator schedule_meta = scheduled_blocks.find(sbid);
        assert(schedule_meta != scheduled_blocks.end());
        assert(schedule_meta->second.operations.back() ==
               ((dirty) ? block_scheduler_type::op_release_dirty : block_scheduler_type::op_release));

        SwappableBlockType& sblock = swappable_blocks[sbid];
        sblock.make_dirty_if(dirty);
        sblock.release();
        if (! sblock.is_acquired())
        {
            if (sblock.is_dirty() || sblock.is_external())
            {
                // => evictable
                if (shall_keep_internal_block(schedule_meta))
                {
                    // => swappable_block shall keep its internal_block
                    scheduled_evictable_blocks.insert(sbid);
                    if (shall_be_cleaned(schedule_meta))
                        schedule_write(sbid);
                }
                else
                {
                    // give block to scheduler
                    free_evictable_blocks.insert(sbid);
                    if (next_op_to_schedule != prediction_sequence.end())
                        schedule_next_operations();
                    else {
                        if (! write_scheduled_blocks.count(sbid))
                            schedule_write(sbid);
                    }
                }
            }
            else
            {
                // => uninitialized
                if (shall_keep_internal_block(schedule_meta))
                    // => swappable_block shall keep its internal_block
                    schedule_meta->second.reserved_iblock = sblock.detach_internal_block();
                else
                {
                    // release internal block and give it to prefetcher
                    return_free_internal_block(sblock.detach_internal_block());
                    if (next_op_to_schedule != prediction_sequence.end())
                        schedule_next_operations();
                }
            }
        }
        operation_done(schedule_meta);
    }

    virtual void deinitialize(swappable_block_identifier_type sbid)
    {
        assert(! prediction_sequence.empty());
        assert(prediction_sequence.front().op == block_scheduler_type::op_deinitialize);
        assert(prediction_sequence.front().id == sbid);
        prediction_sequence.pop_front();
        scheduled_blocks_iterator schedule_meta = scheduled_blocks.find(sbid);
        assert(schedule_meta != scheduled_blocks.end());
        assert(schedule_meta->second.operations.back() == block_scheduler_type::op_deinitialize);

        SwappableBlockType& sblock = swappable_blocks[sbid];
        if (sblock.is_evictable())
        {
            size_t t;
            if (shall_keep_internal_block(schedule_meta, false))
            {
                t = scheduled_evictable_blocks.erase(sbid);
                if (t == 0) {
                    STXXL_ERRMSG("dirty block not scheduled on deinitialize");
                    t = free_evictable_blocks.erase(sbid);
                }
            }
            else
                t = free_evictable_blocks.erase(sbid);
            assert(t != 0);
        }
        if (internal_block_type * iblock = sblock.deinitialize())
        {
            if (shall_keep_internal_block(schedule_meta))
                // => swappable_block shall keep its internal_block
                schedule_meta->second.reserved_iblock = iblock;
            else
            {
                // release internal block and give it to prefetcher
                return_free_internal_block(iblock);
                if (next_op_to_schedule != prediction_sequence.end())
                    schedule_next_operations();
            }
        }
        operation_done(schedule_meta);
    }

    virtual void initialize(swappable_block_identifier_type sbid, external_block_type eblock)
    {
        assert(! prediction_sequence.empty());
        assert(prediction_sequence.front().op == block_scheduler_type::op_initialize);
        assert(prediction_sequence.front().id == sbid);
        prediction_sequence.pop_front();
        scheduled_blocks_iterator schedule_meta = scheduled_blocks.find(sbid);
        assert(schedule_meta != scheduled_blocks.end());
        assert(schedule_meta->second.operations.back() == block_scheduler_type::op_initialize);

        SwappableBlockType& sblock = swappable_blocks[sbid];
        sblock.initialize(eblock);
        if (shall_be_read(schedule_meta))
        {
            sblock.attach_internal_block(schedule_meta->second.reserved_iblock);
            schedule_meta->second.reserved_iblock = 0;
            scheduled_evictable_blocks.insert(sbid);
            schedule_read(schedule_meta);
        }
        operation_done(schedule_meta);
    }

    virtual external_block_type extract_external_block(swappable_block_identifier_type sbid)
    {
        assert(! prediction_sequence.empty());
        assert(prediction_sequence.front().op == block_scheduler_type::op_extract_external_block);
        assert(prediction_sequence.front().id == sbid);
        prediction_sequence.pop_front();
        scheduled_blocks_iterator schedule_meta = scheduled_blocks.find(sbid);
        assert(schedule_meta != scheduled_blocks.end());
        assert(schedule_meta->second.operations.back() == block_scheduler_type::op_extract_external_block);

        SwappableBlockType& sblock = swappable_blocks[sbid];
        wait_on_write(sbid);
        operation_done(schedule_meta);
        return sblock.extract_external_block();
    }
};

STXXL_END_NAMESPACE

#endif // !STXXL_MNG_BLOCK_SCHEDULER_HEADER