mng.h

/***************************************************************************
 *  include/stxxl/bits/mng/mng.h
 *
 *  Part of the STXXL. See http://stxxl.sourceforge.net
 *
 *  Copyright (C) 2002-2004 Roman Dementiev <[email protected]>
 *  Copyright (C) 2007 Johannes Singler <[email protected]>
 *  Copyright (C) 2008 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_MNG_HEADER
#define STXXL_MNG_HEADER

#include <memory>
#include <iostream>
#include <fstream>
#include <vector>
#include <list>
#include <map>
#include <algorithm>
#include <string>
#include <cstdlib>

#ifdef STXXL_BOOST_CONFIG
 #include <boost/config.hpp>
#endif

#ifdef BOOST_MSVC
#include <memory.h>
#endif

#include <stxxl/io>
#include <stxxl/bits/noncopyable.h>
#include <stxxl/bits/common/rand.h>
#include <stxxl/bits/common/aligned_alloc.h>
#include <stxxl/bits/common/debug.h>


__STXXL_BEGIN_NAMESPACE



template <unsigned SIZE>
struct BID
{
    enum
    {
        size = SIZE,         
        t_size = SIZE        
    };
    file * storage;          
    stxxl::int64 offset;     
    BID() : storage(NULL), offset(0) { }
    bool valid() const
    {
        return storage;
    }
    BID(file * s, stxxl::int64 o) : storage(s), offset(o) { }
    BID(const BID & obj) : storage(obj.storage), offset(obj.offset) { }
    template <unsigned BlockSize>
    explicit BID(const BID<BlockSize> & obj) : storage(obj.storage), offset(obj.offset) { }
};



template <>
struct BID<0>
{
    file * storage;          
    stxxl::int64 offset;     
    unsigned size;           
    enum
    {
        t_size = 0           
    };
    BID() : storage(NULL), offset(0), size(0) { }
    BID(file * f, stxxl::int64 o, unsigned s) : storage(f), offset(o), size(s) { }
    bool valid() const
    {
        return storage;
    }
};

template <unsigned blk_sz>
bool operator == (const BID<blk_sz> & a, const BID<blk_sz> & b)
{
    return (a.storage == b.storage) && (a.offset == b.offset) && (a.size == b.size);
}

template <unsigned blk_sz>
bool operator != (const BID<blk_sz> & a, const BID<blk_sz> & b)
{
    return (a.storage != b.storage) || (a.offset != b.offset) || (a.size != b.size);
}


template <unsigned blk_sz>
std::ostream & operator << (std::ostream & s, const BID<blk_sz> & bid)
{
    s << " storage file addr: " << bid.storage;
    s << " offset: " << bid.offset;
    s << " size: " << bid.size;
    return s;
}


template <unsigned bytes>
class filler_struct__
{
    typedef unsigned char byte_type;
    byte_type filler_array_[bytes];

public:
    filler_struct__() { STXXL_VERBOSE2("filler_struct__ is allocated"); }
};

template <>
class filler_struct__<0>
{
    typedef unsigned char byte_type;

public:
    filler_struct__() { STXXL_VERBOSE2("filler_struct__ is allocated"); }
};

template <class T, unsigned Size_>
class element_block
{
public:
    typedef T type;
    typedef T value_type;
    typedef T & reference;
    typedef const T & const_reference;
    typedef type * pointer;
    typedef pointer iterator;
    typedef const type * const_iterator;

    enum
    {
        size = Size_ 
    };

    T elem[size];

    element_block() { STXXL_VERBOSE2("element_block is allocated"); }

    reference operator [] (int i)
    {
        return elem[i];
    }

    iterator begin()
    {
        return elem;
    }
    const_iterator begin() const
    {
        return elem;
    }
    iterator end()
    {
        return elem + size;
    }
    const_iterator end() const
    {
        return elem + size;
    }
};

template <class T, unsigned Size_, unsigned RawSize_, unsigned NBids_ = 0>
class block_w_bids : public element_block<T, Size_>
{
public:
    enum
    {
        raw_size = RawSize_,
        nbids = NBids_
    };
    typedef BID<raw_size> bid_type;

    bid_type ref[nbids];

    bid_type & operator () (int i)
    {
        return ref[i];
    }

    block_w_bids() { STXXL_VERBOSE2("block_w_bids is allocated"); }
};

template <class T, unsigned Size_, unsigned RawSize_>
class block_w_bids<T, Size_, RawSize_, 0>: public element_block<T, Size_>
{
public:
    enum
    {
        raw_size = RawSize_,
        nbids = 0
    };
    typedef BID<raw_size> bid_type;

    block_w_bids() { STXXL_VERBOSE2("block_w_bids is allocated"); }
};

template <class T_, unsigned RawSize_, unsigned NBids_, class InfoType_ = void>
class block_w_info :
    public block_w_bids<T_, ((RawSize_ - sizeof(BID<RawSize_>) * NBids_ - sizeof(InfoType_)) / sizeof(T_)), RawSize_, NBids_>
{
public:
    typedef InfoType_ info_type;

    info_type info;

    enum { size = ((RawSize_ - sizeof(BID<RawSize_>) * NBids_ - sizeof(InfoType_)) / sizeof(T_)) };

public:
    block_w_info() { STXXL_VERBOSE2("block_w_info is allocated"); }
};

template <class T_, unsigned RawSize_, unsigned NBids_>
class block_w_info<T_, RawSize_, NBids_, void>:
    public block_w_bids<T_, ((RawSize_ - sizeof(BID<RawSize_>) * NBids_) / sizeof(T_)), RawSize_, NBids_>
{
public:
    typedef void info_type;
    enum { size = ((RawSize_ - sizeof(BID<RawSize_>) * NBids_) / sizeof(T_)) };

public:
    block_w_info() { STXXL_VERBOSE2("block_w_info is allocated"); }
};


template <unsigned RawSize_, class T_, unsigned NRef_ = 0, class InfoType_ = void>
class typed_block :
    public block_w_info<T_, RawSize_, NRef_, InfoType_>,
    public filler_struct__<(RawSize_ - sizeof(block_w_info<T_, RawSize_, NRef_, InfoType_>))>
{
public:
    typedef T_ type;
    typedef T_ value_type;
    typedef T_ & reference;
    typedef const T_ & const_reference;
    typedef type * pointer;
    typedef pointer iterator;
    typedef type const * const_iterator;

    enum { has_filler = (RawSize_ != sizeof(block_w_info<T_, RawSize_, NRef_, InfoType_>)) };

    typedef BID<RawSize_> bid_type;

    typed_block()
    {
        STXXL_VERBOSE2("typed_block is allocated");
    }

    enum
    {
        raw_size = RawSize_,                                      
        size = block_w_info<T_, RawSize_, NRef_, InfoType_>::size 
    };

    request_ptr write(const BID<raw_size> & bid,
                      completion_handler on_cmpl = default_completion_handler())
    {
        return bid.storage->awrite(
                   this,
                   bid.offset,
                   raw_size,
                   on_cmpl);
    }

    request_ptr read(const BID<raw_size> & bid,
                     completion_handler on_cmpl = default_completion_handler())
    {
        return bid.storage->aread(this, bid.offset, raw_size, on_cmpl);
    }

    static void * operator new[] (size_t bytes)
    {
        unsigned_type meta_info_size = bytes % raw_size;
        STXXL_VERBOSE1("typed::block operator new[]: Meta info size: " << meta_info_size);

        void * result = aligned_alloc<BLOCK_ALIGN>(bytes, meta_info_size);
        #ifdef STXXL_VALGRIND_TYPED_BLOCK_INITIALIZE_ZERO
        memset(result, 0, bytes);
        #endif
        char * tmp = (char *)result;
        debugmon::get_instance()->block_allocated(tmp, tmp + bytes, RawSize_);
        tmp += RawSize_;
        while (tmp < ((char *)result) + bytes)
        {
            debugmon::get_instance()->block_allocated(tmp, ((char *)result) + bytes, RawSize_);
            tmp += RawSize_;
        }
        return result;
    }

    static void * operator new (size_t bytes)
    {
        unsigned_type meta_info_size = bytes % raw_size;
        STXXL_VERBOSE1("typed::block operator new: Meta info size: " << meta_info_size);

        void * result = aligned_alloc<BLOCK_ALIGN>(bytes, meta_info_size);
        #ifdef STXXL_VALGRIND_TYPED_BLOCK_INITIALIZE_ZERO
        memset(result, 0, bytes);
        #endif
        char * tmp = (char *)result;
        debugmon::get_instance()->block_allocated(tmp, tmp + bytes, RawSize_);
        tmp += RawSize_;
        while (tmp < ((char *)result) + bytes)
        {
            debugmon::get_instance()->block_allocated(tmp, ((char *)result) + bytes, RawSize_);
            tmp += RawSize_;
        }
        return result;
    }

    static void * operator new (size_t /*bytes*/, void * ptr)     // construct object in existing memory
    {
        return ptr;
    }

    static void operator delete[] (void * ptr)
    {
        debugmon::get_instance()->block_deallocated((char *)ptr);
        aligned_dealloc<BLOCK_ALIGN>(ptr);
    }

    static void operator delete (void * ptr)
    {
        debugmon::get_instance()->block_deallocated((char *)ptr);
        aligned_dealloc<BLOCK_ALIGN>(ptr);
    }

    static void operator delete (void *, void *)
    { }

    // STRANGE: implementing destructor makes g++ allocate
    // additional 4 bytes in the beginning of every array
    // of this type !? makes aligning to 4K boundaries difficult
    //
    // http://www.cc.gatech.edu/grads/j/Seung.Won.Jun/tips/pl/node4.html :
    // "One interesting thing is the array allocator requires more memory
    //  than the array size multiplied by the size of an element, by a
    //  difference of delta for metadata a compiler needs. It happens to
    //  be 8 bytes long in g++."
    // ~typed_block() { }
};


/*
   template <unsigned BLK_SIZE>
   class BIDArray: public std::vector< BID <BLK_SIZE> >
   {
   public:
   BIDArray(std::vector< BID <BLK_SIZE> >::size_type size = 0) : std::vector< BID <BLK_SIZE> >(size) {};
   };
 */

template <unsigned BLK_SIZE>
class BIDArray : private noncopyable
{
protected:
    unsigned_type _size;
    BID<BLK_SIZE> * array;

public:
    typedef BID<BLK_SIZE> & reference;
    typedef BID<BLK_SIZE> * iterator;
    typedef const BID<BLK_SIZE> * const_iterator;
    BIDArray() : _size(0), array(NULL)
    { }
    iterator begin()
    {
        return array;
    }
    iterator end()
    {
        return array + _size;
    }

    BIDArray(unsigned_type size) : _size(size)
    {
        array = new BID<BLK_SIZE>[size];
    }
    unsigned_type size() const
    {
        return _size;
    }
    reference operator [] (int_type i)
    {
        return array[i];
    }
    void resize(unsigned_type newsize)
    {
        if (array)
        {
            STXXL_MSG("Warning: resizing nonempty BIDArray");
            BID<BLK_SIZE> * tmp = array;
            array = new BID<BLK_SIZE>[newsize];
            memcpy((void *)array, (void *)tmp,
                   sizeof(BID<BLK_SIZE>) * (STXXL_MIN(_size, newsize)));
            delete[] tmp;
            _size = newsize;
        }
        else
        {
            array = new BID<BLK_SIZE>[newsize];
            _size = newsize;
        }
    }
    ~BIDArray()
    {
        if (array)
            delete[] array;
    }
};


class DiskAllocator : private noncopyable
{
#ifdef STXXL_BOOST_THREADS
    boost::mutex mutex;
#else
    stxxl::mutex mutex;
#endif

    typedef std::pair<stxxl::int64, stxxl::int64> place;
    struct FirstFit : public std::binary_function<place, stxxl::int64, bool>
    {
        bool operator () (
            const place & entry,
            const stxxl::int64 size) const
        {
            return (entry.second >= size);
        }
    };
    struct OffCmp
    {
        bool operator () (const stxxl::int64 & off1, const stxxl::int64 & off2)
        {
            return off1 < off2;
        }
    };

    DiskAllocator()
    { }

protected:
    typedef std::map<stxxl::int64, stxxl::int64> sortseq;
    sortseq free_space;
    //  sortseq used_space;
    stxxl::int64 free_bytes;
    stxxl::int64 disk_bytes;

    void dump();

    void check_corruption(stxxl::int64 region_pos, stxxl::int64 region_size,
                          sortseq::iterator pred, sortseq::iterator succ)
    {
        if (pred != free_space.end())
        {
            if (pred->first <= region_pos && pred->first + pred->second > region_pos)
            {
                STXXL_THROW(bad_ext_alloc, "DiskAllocator::check_corruption", "Error: double deallocation of external memory " <<
                            "System info: P " << pred->first << " " << pred->second << " " << region_pos);
            }
        }
        if (succ != free_space.end())
        {
            if (region_pos <= succ->first && region_pos + region_size > succ->first)
            {
                STXXL_THROW(bad_ext_alloc, "DiskAllocator::check_corruption", "Error: double deallocation of external memory "
                     << "System info: S " << region_pos << " " << region_size << " " << succ->first);
            }
        }
    }

public:
    inline DiskAllocator(stxxl::int64 disk_size);

    inline stxxl::int64 get_free_bytes() const
    {
        return free_bytes;
    }
    inline stxxl::int64 get_used_bytes() const
    {
        return disk_bytes - free_bytes;
    }
    inline stxxl::int64 get_total_bytes() const
    {
        return disk_bytes;
    }

    template <unsigned BLK_SIZE>
    stxxl::int64 new_blocks(BIDArray<BLK_SIZE> & bids);

    template <unsigned BLK_SIZE>
    stxxl::int64 new_blocks(BID<BLK_SIZE> * begin,
                            BID<BLK_SIZE> * end);
/*
        template < unsigned BLK_SIZE >
        void delete_blocks (const BIDArray < BLK_SIZE > &bids); */
    template <unsigned BLK_SIZE>
    void delete_block(const BID<BLK_SIZE> & bid);
};

DiskAllocator::DiskAllocator(stxxl::int64 disk_size) :
    free_bytes(disk_size),
    disk_bytes(disk_size)
{
    free_space[0] = disk_size;
}


template <unsigned BLK_SIZE>
stxxl::int64 DiskAllocator::new_blocks(BIDArray<BLK_SIZE> & bids)
{
    return new_blocks(bids.begin(), bids.end());
}

template <unsigned BLK_SIZE>
stxxl::int64 DiskAllocator::new_blocks(BID<BLK_SIZE> * begin,
                                       BID<BLK_SIZE> * end)
{
#ifdef STXXL_BOOST_THREADS
    boost::mutex::scoped_lock lock(mutex);
#else
    mutex.lock();
#endif

    STXXL_VERBOSE2("DiskAllocator::new_blocks<BLK_SIZE>,  BLK_SIZE = " << BLK_SIZE
                                                                       << ", free:" << free_bytes << " total:" << disk_bytes <<
                   " begin: " << ((void *)(begin)) << " end: " << ((void *)(end)));

    stxxl::int64 requested_size = 0;

    typename BIDArray<BLK_SIZE>::iterator cur = begin;
    for ( ; cur != end; ++cur)
    {
        STXXL_VERBOSE2("Asking for a block with size: " << (cur->size));
        requested_size += cur->size;
    }

    if (free_bytes < requested_size)
    {
        STXXL_ERRMSG("External memory block allocation error: " << requested_size <<
                     " bytes requested, " << free_bytes <<
                     " bytes free. Trying to extend the external memory space...");


        begin->offset = disk_bytes; // allocate at the end
        for (++begin; begin != end; ++begin)
        {
            begin->offset = (begin - 1)->offset + (begin - 1)->size;
        }
        disk_bytes += requested_size;

#ifndef STXXL_BOOST_THREADS
        mutex.unlock();
#endif

        return disk_bytes;
    }

    // dump();

    sortseq::iterator space =
        std::find_if(free_space.begin(), free_space.end(),
                     bind2nd(FirstFit(), requested_size));

    if (space != free_space.end())
    {
        stxxl::int64 region_pos = (*space).first;
        stxxl::int64 region_size = (*space).second;
        free_space.erase(space);
        if (region_size > requested_size)
            free_space[region_pos + requested_size] = region_size - requested_size;

        begin->offset = region_pos;
        for (++begin; begin != end; ++begin)
        {
            begin->offset = (begin - 1)->offset + (begin - 1)->size;
        }
        free_bytes -= requested_size;
        //dump();

#ifndef STXXL_BOOST_THREADS
        mutex.unlock();
#endif

        return disk_bytes;
    }

    // no contiguous region found
    STXXL_VERBOSE1("Warning, when allocation an external memory space, no contiguous region found");
    STXXL_VERBOSE1("It might harm the performance");
    if (requested_size == BLK_SIZE)
    {
        assert(end - begin == 1);

        STXXL_ERRMSG("Warning: Severe external memory space fragmentation!");
        dump();

        STXXL_ERRMSG("External memory block allocation error: " << requested_size <<
                     " bytes requested, " << free_bytes <<
                     " bytes free. Trying to extend the external memory space...");

        begin->offset = disk_bytes; // allocate at the end
        disk_bytes += BLK_SIZE;

#ifndef STXXL_BOOST_THREADS
        mutex.unlock();
#endif

        return disk_bytes;
    }

    assert(requested_size > BLK_SIZE);
    assert(end - begin > 1);

    typename  BIDArray<BLK_SIZE>::iterator middle = begin + ((end - begin) / 2);
    new_blocks(begin, middle);
    new_blocks(middle, end);

#ifndef STXXL_BOOST_THREADS
    mutex.unlock();
#endif

    return disk_bytes;
}


template <unsigned BLK_SIZE>
void DiskAllocator::delete_block(const BID<BLK_SIZE> & bid)
{
#ifdef STXXL_BOOST_THREADS
    boost::mutex::scoped_lock lock(mutex);
#else
    mutex.lock();
#endif

    STXXL_VERBOSE2("DiskAllocator::delete_block<BLK_SIZE>,  BLK_SIZE = " << BLK_SIZE
                                                                         << ", free:" << free_bytes << " total:" << disk_bytes);
    STXXL_VERBOSE2("Deallocating a block with size: " << bid.size);
    //assert(bid.size);

    //dump();
    stxxl::int64 region_pos = bid.offset;
    stxxl::int64 region_size = bid.size;
    STXXL_VERBOSE2("Deallocating a block with size: " << region_size << " position: " << region_pos);
    if (!free_space.empty())
    {
        sortseq::iterator succ = free_space.upper_bound(region_pos);
        sortseq::iterator pred = succ;
        pred--;
        check_corruption(region_pos, region_size, pred, succ);
        if (succ == free_space.end())
        {
            if (pred == free_space.end())
            {
                STXXL_ERRMSG("Error deallocating block at " << bid.offset << " size " << bid.size);
                STXXL_ERRMSG(((pred == succ) ? "pred==succ" : "pred!=succ"));
                STXXL_ERRMSG(((pred == free_space.begin()) ? "pred==free_space.begin()" : "pred!=free_space.begin()"));
                STXXL_ERRMSG(((pred == free_space.end()) ? "pred==free_space.end()" : "pred!=free_space.end()"));
                STXXL_ERRMSG(((succ == free_space.begin()) ? "succ==free_space.begin()" : "succ!=free_space.begin()"));
                STXXL_ERRMSG(((succ == free_space.end()) ? "succ==free_space.end()" : "succ!=free_space.end()"));
                dump();
                assert(pred != free_space.end());
            }
            if ((*pred).first + (*pred).second == region_pos)
            {
                // coalesce with predecessor
                region_size += (*pred).second;
                region_pos = (*pred).first;
                free_space.erase(pred);
            }
        }
        else
        {
            if (free_space.size() > 1)
            {
                /*
                   if(pred == succ)
                   {
                      STXXL_ERRMSG("Error deallocating block at "<<bid.offset<<" size "<<bid.size);
                      STXXL_ERRMSG(((pred==succ)?"pred==succ":"pred!=succ"));
                      STXXL_ERRMSG(((pred==free_space.begin ())?"pred==free_space.begin()":"pred!=free_space.begin()"));
                      STXXL_ERRMSG(((pred==free_space.end ())?"pred==free_space.end()":"pred!=free_space.end()"));
                      STXXL_ERRMSG(((succ==free_space.begin ())?"succ==free_space.begin()":"succ!=free_space.begin()"));
                      STXXL_ERRMSG(((succ==free_space.end ())?"succ==free_space.end()":"succ!=free_space.end()"));
                      dump();
                      assert(pred != succ);
                   } */
                bool succ_is_not_the_first = (succ != free_space.begin());
                if ((*succ).first == region_pos + region_size)
                {
                    // coalesce with successor
                    region_size += (*succ).second;
                    free_space.erase(succ);
                }
                if (succ_is_not_the_first)
                {
                    if (pred == free_space.end())
                    {
                        STXXL_ERRMSG("Error deallocating block at " << bid.offset << " size " << bid.size);
                        STXXL_ERRMSG(((pred == succ) ? "pred==succ" : "pred!=succ"));
                        STXXL_ERRMSG(((pred == free_space.begin()) ? "pred==free_space.begin()" : "pred!=free_space.begin()"));
                        STXXL_ERRMSG(((pred == free_space.end()) ? "pred==free_space.end()" : "pred!=free_space.end()"));
                        STXXL_ERRMSG(((succ == free_space.begin()) ? "succ==free_space.begin()" : "succ!=free_space.begin()"));
                        STXXL_ERRMSG(((succ == free_space.end()) ? "succ==free_space.end()" : "succ!=free_space.end()"));
                        dump();
                        assert(pred != free_space.end());
                    }
                    if ((*pred).first + (*pred).second == region_pos)
                    {
                        // coalesce with predecessor
                        region_size += (*pred).second;
                        region_pos = (*pred).first;
                        free_space.erase(pred);
                    }
                }
            }
            else
            {
                if ((*succ).first == region_pos + region_size)
                {
                    // coalesce with successor
                    region_size += (*succ).second;
                    free_space.erase(succ);
                }
            }
        }
    }

    free_space[region_pos] = region_size;
    free_bytes += stxxl::int64(bid.size);

    //dump();

#ifndef STXXL_BOOST_THREADS
    mutex.unlock();
#endif
}
/*
    template < unsigned BLK_SIZE >
        void DiskAllocator::delete_blocks (const BIDArray < BLK_SIZE > &bids)
    {
        STXXL_VERBOSE2("DiskAllocator::delete_blocks<BLK_SIZE> BLK_SIZE="<< BLK_SIZE <<
      ", free:" << free_bytes << " total:"<< disk_bytes );

        unsigned i=0;
        for (; i < bids.size (); ++i)
        {
            stxxl::int64 region_pos = bids[i].offset;
            stxxl::int64 region_size = bids[i].size;
              STXXL_VERBOSE2("Deallocating a block with size: "<<region_size);
              assert(bids[i].size);

            if(!free_space.empty())
            {
                sortseq::iterator succ =
                    free_space.upper_bound (region_pos);
                sortseq::iterator pred = succ;
                pred--;

                if (succ != free_space.end ()
                    && (*succ).first == region_pos + region_size)
                {
                    // coalesce with successor

                    region_size += (*succ).second;
                    free_space.erase (succ);
                }
                if (pred != free_space.end ()
                    && (*pred).first + (*pred).second == region_pos)
                {
                    // coalesce with predecessor

                    region_size += (*pred).second;
                    region_pos = (*pred).first;
                    free_space.erase (pred);
                }
            }
            free_space[region_pos] = region_size;

        }
        for(i=0;i<bids.size();++i)
                  free_bytes += stxxl::int64(bids[i].size);
    } */

class config : private noncopyable
{
    struct DiskEntry
    {
        std::string path;
        std::string io_impl;
        stxxl::int64 size;
    };
    std::vector<DiskEntry> disks_props;

    // in disks_props, flash devices come after all regular disks
    unsigned first_flash;

    config(const char * config_path = "./.stxxl");

public:
    inline unsigned disks_number() const
    {
        return disks_props.size();
    }

    inline std::pair<unsigned, unsigned> regular_disk_range() const
    {
        return std::pair<unsigned, unsigned>(0, first_flash);
    }

    inline std::pair<unsigned, unsigned> flash_range() const
    {
        return std::pair<unsigned, unsigned>(first_flash, disks_props.size());
    }

    inline const std::string & disk_path(int disk) const
    {
        return disks_props[disk].path;
    }
    inline stxxl::int64 disk_size(int disk) const
    {
        return disks_props[disk].size;
    }
    inline const std::string & disk_io_impl(int disk) const
    {
        return disks_props[disk].io_impl;
    }

    static config * get_instance();

private:
    static config * instance;
};


class FileCreator
{
public:
    virtual stxxl::file * create(const std::string & io_impl,
                                 const std::string & filename,
                                 int options, int disk)
    {
        if (io_impl == "syscall")
        {
            stxxl::ufs_file_base * result = new stxxl::syscall_file(filename,
                                                                    options,
                                                                    disk);
            result->lock();
            return result;
        }
            #ifndef BOOST_MSVC
        else if (io_impl == "mmap")
        {
            stxxl::ufs_file_base * result = new stxxl::mmap_file(filename,
                                                                 options, disk);
            result->lock();
            return result;
        }
        else if (io_impl == "simdisk")
        {
            stxxl::ufs_file_base * result = new stxxl::sim_disk_file(filename,
                                                                     options,
                                                                     disk);
            result->lock();
            return result;
        }
            #else
        else if (io_impl == "wincall")
        {
            stxxl::wfs_file_base * result = new stxxl::wincall_file(filename,
                                                                    options, disk);
            result->lock();
            return result;
        }
            #endif
      #ifdef STXXL_BOOST_CONFIG
        else if (io_impl == "boostfd")
            return new stxxl::boostfd_file(filename,
                                           options, disk);
      #endif

        STXXL_THROW(std::runtime_error, "FileCreator::create", "Unsupported disk I/O implementation " <<
                    io_impl << " .");

        return NULL;
    }

    virtual ~FileCreator() { }
};


struct basic_allocation_strategy
{
    basic_allocation_strategy(int disks_begin, int disks_end);
    basic_allocation_strategy();
    int operator () (int i) const;
    static const char * name();
};

struct striping
{
    int begin, diff;
    striping(int b, int e) : begin(b), diff(e - b)
    { }
    striping() : begin(0)
    {
        diff = config::get_instance()->disks_number();
    }
    int operator () (int i) const
    {
        return begin + i % diff;
    }
    static const char * name()
    {
        return "striping";
    }
    // FIXME WHY?
    virtual ~striping()
    { }
};

struct FR : public striping
{
    random_number<random_uniform_fast> rnd;
    FR(int b, int e) : striping(b, e)
    { }
    FR() : striping()
    { }
    int operator () (int /*i*/) const
    {
        return begin + rnd(diff);
    }
    static const char * name()
    {
        return "fully randomized striping";
    }
};

struct SR : public striping
{
    random_number<random_uniform_fast> rnd;
    int offset;
    SR(int b, int e) : striping(b, e)
    {
        offset = rnd(diff);
    }
    SR() : striping()
    {
        offset = rnd(diff);
    }
    int operator () (int i) const
    {
        return begin + (i + offset) % diff;
    }
    static const char * name()
    {
        return "simple randomized striping";
    }
};

struct RC : public striping
{
    std::vector<int> perm;

    RC(int b, int e) : striping(b, e), perm(diff)
    {
        for (int i = 0; i < diff; i++)
            perm[i] = i;

        stxxl::random_number<random_uniform_fast> rnd;
        std::random_shuffle(perm.begin(), perm.end(), rnd __STXXL_FORCE_SEQUENTIAL);
    }
    RC() : striping(), perm(diff)
    {
        for (int i = 0; i < diff; i++)
            perm[i] = i;

        random_number<random_uniform_fast> rnd;
        std::random_shuffle(perm.begin(), perm.end(), rnd __STXXL_FORCE_SEQUENTIAL);
    }
    int operator () (int i) const
    {
        return begin + perm[i % diff];
    }
    static const char * name()
    {
        return "randomized cycling striping";
    }
};

struct RC_disk : public RC
{
    RC_disk(int b, int e) : RC(b, e)
    { }
    RC_disk() : RC(config::get_instance()->regular_disk_range().first, config::get_instance()->regular_disk_range().second)
    { }
    static const char * name()
    {
        return "Randomized cycling striping on regular disks";
    }
};

struct RC_flash : public RC
{
    RC_flash(int b, int e) : RC(b, e)
    { }
    RC_flash() : RC(config::get_instance()->flash_range().first, config::get_instance()->flash_range().second)
    { }
    static const char * name()
    {
        return "Randomized cycling striping on flash devices";
    }
};

struct single_disk
{
    const int disk;
    single_disk(int d, int = 0) : disk(d)
    { }

    single_disk() : disk(0)
    { }
    int operator () (int /*i*/) const
    {
        return disk;
    }
    static const char * name()
    {
        return "single disk";
    }
};


template <class BaseAllocator_>
struct offset_allocator
{
    BaseAllocator_ base;
    int_type offset;
    offset_allocator(int_type offset_) : base(), offset(offset_) { }
    offset_allocator(int_type offset_, BaseAllocator_ & base_) : base(base_), offset(offset_) { }
    int operator () (int_type i) const
    {
        return base(offset + i);
    }
};


/* deprecated

   template < class bid_it >
   struct bid_iterator_traits
   {
      bid_it *a;
      enum
      {
          block_size = bid_it::block_size
      };
   };

   template < unsigned blk_sz >
   struct bid_iterator_traits <BID <blk_sz > *>
   {
      enum
      {
          block_size = blk_sz
      };
   };


   template < unsigned blk_sz,class X >
   struct bid_iterator_traits< __gnu_cxx::__normal_iterator< BID<blk_sz> *,  X> >
   {
      enum
      {
          block_size = blk_sz
      };
   };

   template < unsigned blk_sz,class X , class Y>
   struct bid_iterator_traits< std::_List_iterator<BID<blk_sz>,X,Y > >
   {
      enum
      {
          block_size = blk_sz
      };
   };

   template < unsigned blk_sz, class X >
   struct bid_iterator_traits< typename std::vector< BID<blk_sz> , X >::iterator >
   {
      enum
      {
          block_size = blk_sz
      };
   };
 */

class block_manager : private noncopyable
{
    DiskAllocator ** disk_allocators;
    file ** disk_files;

    unsigned ndisks;
    block_manager();

protected:
    template <class BIDType, class DiskAssgnFunctor, class BIDIteratorClass>
    void new_blocks_int(
        const unsigned_type nblocks,
        DiskAssgnFunctor functor,
        BIDIteratorClass out);

public:
    static block_manager * get_instance();


    template <class DiskAssgnFunctor, class BIDIteratorClass>
    void new_blocks(
        DiskAssgnFunctor functor,
        BIDIteratorClass bidbegin,
        BIDIteratorClass bidend);

    template <class BlockType, class DiskAssgnFunctor, class BIDIteratorClass>
    void new_blocks(
        const unsigned_type nblocks,
        DiskAssgnFunctor functor,
        BIDIteratorClass out);



    template <class BIDIteratorClass>
    void delete_blocks(const BIDIteratorClass & bidbegin, const BIDIteratorClass & bidend);

    template <unsigned BLK_SIZE>
    void delete_block(const BID<BLK_SIZE> & bid);

    ~block_manager();

private:
    static block_manager * instance;
};


template <class BIDType, class DiskAssgnFunctor, class OutputIterator>
void block_manager::new_blocks_int(
    const unsigned_type nblocks,
    DiskAssgnFunctor functor,
    OutputIterator out)
{
    typedef BIDType bid_type;
    typedef  BIDArray<bid_type::t_size> bid_array_type;

    // bid_type tmpbid;
    int_type * bl = new int_type[ndisks];
    bid_array_type * disk_bids = new bid_array_type[ndisks];
    file ** disk_ptrs = new file *[nblocks];

    memset(bl, 0, ndisks * sizeof(int_type));

    unsigned_type i = 0;
    //OutputIterator  it = out;
    for ( ; i < nblocks; ++i /* , ++it*/)
    {
        const int disk = functor(i);
        disk_ptrs[i] = disk_files[disk];
        //(*it).storage = disk_files[disk];
        bl[disk]++;
    }

    for (i = 0; i < ndisks; ++i)
    {
        if (bl[i])
        {
            disk_bids[i].resize(bl[i]);
            const stxxl::int64 old_capacity =
                disk_allocators[i]->get_total_bytes();
            const stxxl::int64 new_capacity =
                disk_allocators[i]->new_blocks(disk_bids[i]);
            if (old_capacity != new_capacity)
            {
                // resize the file
                disk_files[i]->set_size(new_capacity);
            }
        }
    }

    memset(bl, 0, ndisks * sizeof(int_type));

    OutputIterator it = out;
    for (i = 0 /*,it = out */; i != nblocks; ++it, ++i)
    {
        //int disk = (*it).storage->get_disk_number ();
        //(*it).offset = disk_bids[disk][bl[disk]++].offset;
        const int disk = disk_ptrs[i]->get_disk_number();
        *it = bid_type(disk_ptrs[i], disk_bids[disk][bl[disk]++].offset);
    }

    delete[] bl;
    delete[] disk_bids;
    delete[] disk_ptrs;
}

template <class BlockType, class DiskAssgnFunctor, class OutputIterator>
void block_manager::new_blocks(
    const unsigned_type nblocks,
    DiskAssgnFunctor functor,
    OutputIterator out)
{
    typedef typename BlockType::bid_type bid_type;
    new_blocks_int<bid_type>(nblocks, functor, out);
}

template <class DiskAssgnFunctor, class BIDIteratorClass>
void block_manager::new_blocks(
    DiskAssgnFunctor functor,
    BIDIteratorClass bidbegin,
    BIDIteratorClass bidend)
{
    typedef typename std::iterator_traits<BIDIteratorClass>::value_type bid_type;

    unsigned_type nblocks = 0;

    BIDIteratorClass bidbegin_copy(bidbegin);
    while (bidbegin_copy != bidend)
    {
        ++bidbegin_copy;
        ++nblocks;
    }

    new_blocks_int<bid_type>(nblocks, functor, bidbegin);
}


template <unsigned BLK_SIZE>
void block_manager::delete_block(const BID<BLK_SIZE> & bid)
{
    // do not uncomment it
    //assert(bid.storage->get_disk_number () < config::get_instance ()->disks_number ());
    if (bid.storage->get_disk_number() == -1)
        return; // self managed disk
    assert(bid.storage->get_disk_number() >= 0);
    disk_allocators[bid.storage->get_disk_number()]->delete_block(bid);
}


template <class BIDIteratorClass>
void block_manager::delete_blocks(
    const BIDIteratorClass & bidbegin,
    const BIDIteratorClass & bidend)
{
    for (BIDIteratorClass it = bidbegin; it != bidend; it++)
    {
        delete_block(*it);
    }
}

#ifndef STXXL_DEFAULT_ALLOC_STRATEGY
    #define STXXL_DEFAULT_ALLOC_STRATEGY stxxl::RC
#endif

// in bytes
#ifndef STXXL_DEFAULT_BLOCK_SIZE
    #define STXXL_DEFAULT_BLOCK_SIZE(type) (2 * 1024 * 1024) // use traits
#endif


__STXXL_END_NAMESPACE


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

---