fairport/ndb/node.h

Go to the documentation of this file.

#ifndef FAIRPORT_NDB_NODE_H
#define FAIRPORT_NDB_NODE_H

#include <vector>
#include <algorithm>
#include <functional>
#include <memory>
#include <cassert>
#include <boost/iterator/transform_iterator.hpp>
#include <boost/iostreams/concepts.hpp>
#ifdef _MSC_VER
#pragma warning(push)
#pragma warning(disable:4244)
#endif
#include <boost/iostreams/stream.hpp>
#ifdef _MSC_VER
#pragma warning(pop)
#endif

#include "fairport/util/util.h"
#include "fairport/util/btree.h"

#include "fairport/ndb/database_iface.h"

#ifdef _MSC_VER
#pragma warning(push)
#pragma warning(disable:4250)
#endif

namespace fairport
{


class node_impl : public std::tr1::enable_shared_from_this<node_impl>
{
public:
    node_impl(const shared_db_ptr& db, const node_info& info)
        : m_id(info.id), m_original_data_id(info.data_bid), m_original_sub_id(info.sub_bid), m_original_parent_id(info.parent_id), m_parent_id(info.parent_id), m_db(db) { }

    node_impl(const std::tr1::shared_ptr<node_impl>& container_node, const subnode_info& info)
        : m_id(info.id), m_original_data_id(info.data_bid), m_original_sub_id(info.sub_bid), m_original_parent_id(0), m_parent_id(0), m_pcontainer_node(container_node), m_db(container_node->m_db) { }

    node_id get_id() const { return m_id; }
    block_id get_data_id() const;
    block_id get_sub_id() const;

    node_id get_parent_id() const { return m_parent_id; }

    bool is_subnode() const { return m_pcontainer_node; }

    std::tr1::shared_ptr<data_block> get_data_block() const
        { ensure_data_block(); return m_pdata; }
    std::tr1::shared_ptr<subnode_block> get_subnode_block() const 
        { ensure_sub_block(); return m_psub; }
  
    shared_db_ptr get_db() const { return m_db; }

    size_t read(std::vector<byte>& buffer, ulong offset) const;
    
    template<typename T> T read(ulong offset) const;

    size_t read(std::vector<byte>& buffer, uint page_num, ulong offset) const;
    
    template<typename T> T read(uint page_num, ulong offset) const;

    size_t read_raw(byte* pdest_buffer, size_t size, ulong offset) const;

    size_t size() const;

    size_t get_page_size(uint page_num) const;

    uint get_page_count() const;

    // iterate over subnodes
    const_subnodeinfo_iterator subnode_info_begin() const;
    const_subnodeinfo_iterator subnode_info_end() const;

    node lookup(node_id id) const;

private:
    node_impl& operator=(const node_impl& other); // = delete

    data_block* ensure_data_block() const;
    subnode_block* ensure_sub_block() const;

    const node_id m_id;             
    block_id m_original_data_id;    
    block_id m_original_sub_id;     
    node_id m_original_parent_id;   

    mutable std::tr1::shared_ptr<data_block> m_pdata;    
    mutable std::tr1::shared_ptr<subnode_block> m_psub;  
    node_id m_parent_id;                            

    std::tr1::shared_ptr<node_impl> m_pcontainer_node;   

    shared_db_ptr m_db; 
};

class subnode_transform_info : public std::unary_function<subnode_info, node>
{
public:
    subnode_transform_info(const std::tr1::shared_ptr<node_impl>& parent)
        : m_parent(parent) { }

    node operator()(const subnode_info& info) const;

private:
    std::tr1::shared_ptr<node_impl> m_parent; 
};

class node_stream_device : public boost::iostreams::device<boost::iostreams::input_seekable>
{
public:
    node_stream_device() : m_pos(0) { }

    std::streamsize read(char* pbuffer, std::streamsize n); 

    std::streampos seek(boost::iostreams::stream_offset off, std::ios_base::seekdir way);

private:
    friend class node;
    node_stream_device(std::tr1::shared_ptr<node_impl>& _node) : m_pos(0), m_pnode(_node) { }

    std::streamsize m_pos;              
    std::tr1::shared_ptr<node_impl> m_pnode; 
};

typedef boost::iostreams::stream<node_stream_device> node_stream;

class node
{
public:
    typedef boost::transform_iterator<subnode_transform_info, const_subnodeinfo_iterator> subnode_iterator;

    node(const shared_db_ptr& db, const node_info& info)
        : m_pimpl(new node_impl(db, info)) { }

    node(const node& container_node, const subnode_info& info)
        : m_pimpl(new node_impl(container_node.m_pimpl, info)) { }
    node(const std::tr1::shared_ptr<node_impl>& container_node, const subnode_info& info)
        : m_pimpl(new node_impl(container_node, info)) { }

    node(const node& other)
        : m_pimpl(other.m_pimpl) { }

    node_id get_id() const { return m_pimpl->get_id(); }
    block_id get_data_id() const { return m_pimpl->get_data_id(); }
    block_id get_sub_id() const { return m_pimpl->get_sub_id(); }

    node_id get_parent_id() const { return m_pimpl->get_parent_id(); } 
    bool is_subnode() const { return m_pimpl->is_subnode(); } 

    std::tr1::shared_ptr<data_block> get_data_block() const
        { return m_pimpl->get_data_block(); }
    std::tr1::shared_ptr<subnode_block> get_subnode_block() const 
        { return m_pimpl->get_subnode_block(); } 

    shared_db_ptr get_db() const { return m_pimpl->get_db(); }

    size_t read(std::vector<byte>& buffer, ulong offset) const
        { return m_pimpl->read(buffer, offset); }
    template<typename T> T read(ulong offset) const
        { return m_pimpl->read<T>(offset); }
    size_t read(std::vector<byte>& buffer, uint page_num, ulong offset) const
        { return m_pimpl->read(buffer, page_num, offset); }
    template<typename T> T read(uint page_num, ulong offset) const
        { return m_pimpl->read<T>(page_num, offset); }

    node_stream_device open_as_stream()
        { return node_stream_device(m_pimpl); }

    size_t size() const { return m_pimpl->size(); }
    size_t get_page_size(uint page_num) const 
        { return m_pimpl->get_page_size(page_num); }
    uint get_page_count() const { return m_pimpl->get_page_count(); }

    // iterate over subnodes
    const_subnodeinfo_iterator subnode_info_begin() const
        { return m_pimpl->subnode_info_begin(); }
    const_subnodeinfo_iterator subnode_info_end() const
        { return m_pimpl->subnode_info_end(); } 
    
    subnode_iterator subnode_begin() const
        { return boost::make_transform_iterator(subnode_info_begin(), subnode_transform_info(m_pimpl)); }
 
    subnode_iterator subnode_end() const
        { return boost::make_transform_iterator(subnode_info_end(), subnode_transform_info(m_pimpl)); }

    node lookup(node_id id) const
        { return m_pimpl->lookup(id); }

private:
    std::tr1::shared_ptr<node_impl> m_pimpl; 
};


class block
{
public:
    block(const shared_db_ptr& db, const block_info& info)
        : m_modified(false), m_size(info.size), m_id(info.id), m_address(info.address), m_db(db) { }

    virtual ~block() { }

    virtual bool is_internal() const = 0;

    size_t get_disk_size() const { return m_size; }

    block_id get_id() const { return m_id; }

    ulonglong get_address() const { return m_address; }

protected:
    shared_db_ptr get_db_ptr() const { return shared_db_ptr(m_db); }
    virtual void trim() { }

    bool m_modified;        
    size_t m_size;          
    block_id m_id;          
    ulonglong m_address;    

    weak_db_ptr m_db;
};

class data_block : public block
{
public:
    data_block(const shared_db_ptr& db, const block_info& info, size_t total_size)
        : block(db, info), m_total_size(total_size) { }
    virtual ~data_block() { }

    size_t read(std::vector<byte>& buffer, ulong offset) const;

    template<typename T> T read(ulong offset) const;

    virtual size_t read_raw(byte* pdest_buffer, size_t size, ulong offset) const = 0;

    virtual uint get_page_count() const = 0;

    virtual std::tr1::shared_ptr<external_block> get_page(uint page_num) const = 0;

    size_t get_total_size() const { return m_total_size; }

protected:
    size_t m_total_size;    
};

class extended_block : 
    public data_block, 
    public std::tr1::enable_shared_from_this<extended_block>
{
public:
#ifndef BOOST_NO_RVALUE_REFERENCES
    extended_block(const shared_db_ptr& db, const block_info& info, ushort level, size_t total_size, size_t child_max_total_size, ulong page_max_count, ulong child_page_max_count, std::vector<block_id> bi)
        : data_block(db, info, total_size), m_child_max_total_size(child_max_total_size), m_child_max_page_count(child_page_max_count), m_max_page_count(page_max_count), m_level(level), m_block_info(std::move(bi)), m_child_blocks(m_block_info.size()) { }
#else
    extended_block(const shared_db_ptr& db, const block_info& info, ushort level, size_t total_size, size_t child_max_total_size, ulong page_max_count, ulong child_page_max_count, const std::vector<block_id>& bi)
        : data_block(db, info, total_size), m_child_max_total_size(child_max_total_size), m_child_max_page_count(child_page_max_count), m_max_page_count(page_max_count), m_level(level), m_block_info(bi), m_child_blocks(m_block_info.size()) { }
#endif

    size_t read_raw(byte* pdest_buffer, size_t size, ulong offset) const;
    
    uint get_page_count() const;
    std::tr1::shared_ptr<external_block> get_page(uint page_num) const;
    
    ushort get_level() const { return m_level; }
    bool is_internal() const { return true; }

private:
    extended_block& operator=(const extended_block& other); // = delete
    data_block* get_child_block(uint index) const;

    const size_t m_child_max_total_size;    
    const ulong m_child_max_page_count;     
    const ulong m_max_page_count;           

    size_t get_max_size() const { return m_child_max_total_size * m_max_page_count; }

    const ushort m_level;                   
    std::vector<block_id> m_block_info;     
    mutable std::vector<std::tr1::shared_ptr<data_block> > m_child_blocks; 
};

class external_block : 
    public data_block, 
    public std::tr1::enable_shared_from_this<external_block>
{
public:
#ifndef BOOST_NO_RVALUE_REFERENCES
    external_block(const shared_db_ptr& db, const block_info& info, size_t max_size, std::vector<byte> buffer)
        : data_block(db, info, info.size), m_max_size(max_size), m_buffer(std::move(buffer)) { }
#else
    external_block(const shared_db_ptr& db, const block_info& info, size_t max_size, const std::vector<byte>& buffer)
        : data_block(db, info, info.size), m_max_size(max_size), m_buffer(buffer) { }
#endif

    size_t read_raw(byte* pdest_buffer, size_t size, ulong offset) const;

    uint get_page_count() const { return 1; }
    std::tr1::shared_ptr<external_block> get_page(uint page_num) const;

    bool is_internal() const { return false; }

private:
    external_block& operator=(const external_block& other); // = delete

    const size_t m_max_size;
    size_t get_max_size() const { return m_max_size; }

    std::vector<byte> m_buffer;
};


class subnode_block : 
    public block, 
    public virtual btree_node<node_id, subnode_info>
{
public:
    subnode_block(const shared_db_ptr& db, const block_info& info, ushort level)
        : block(db, info), m_level(level) { }

    virtual ~subnode_block() { }

    ushort get_level() const { return m_level; }

    bool is_internal() const { return true; }
    
protected:
    ushort m_level; 
};

class subnode_nonleaf_block : 
    public subnode_block, 
    public btree_node_nonleaf<node_id, subnode_info>, 
    public std::tr1::enable_shared_from_this<subnode_nonleaf_block>
{
public:
#ifndef BOOST_NO_RVALUE_REFERENCES
    subnode_nonleaf_block(const shared_db_ptr& db, const block_info& info, std::vector<std::pair<node_id, block_id> > subblocks)
        : subnode_block(db, info, 1), m_subnode_info(std::move(subblocks)), m_child_blocks(m_subnode_info.size()) { }
#else
    subnode_nonleaf_block(const shared_db_ptr& db, const block_info& info, const std::vector<std::pair<node_id, block_id> >& subblocks)
        : subnode_block(db, info, 1), m_subnode_info(subblocks), m_child_blocks(m_subnode_info.size()) { }
#endif

    // btree_node_nonleaf implementation
    const node_id& get_key(uint pos) const
        { return m_subnode_info[pos].first; }
    subnode_block* get_child(uint pos);
    const subnode_block* get_child(uint pos) const;
    uint num_values() const { return m_subnode_info.size(); }
    
private:
    std::vector<std::pair<node_id, block_id> > m_subnode_info;           
    mutable std::vector<std::tr1::shared_ptr<subnode_block> > m_child_blocks; 
};

class subnode_leaf_block : 
    public subnode_block, 
    public btree_node_leaf<node_id, subnode_info>, 
    public std::tr1::enable_shared_from_this<subnode_leaf_block>
{
public:
#ifndef BOOST_NO_RVALUE_REFERENCES
    subnode_leaf_block(const shared_db_ptr& db, const block_info& info, std::vector<std::pair<node_id, subnode_info> > subnodes)
        : subnode_block(db, info, 0), m_subnodes(std::move(subnodes)) { }
#else
    subnode_leaf_block(const shared_db_ptr& db, const block_info& info, const std::vector<std::pair<node_id, subnode_info> >& subnodes)
        : subnode_block(db, info, 0), m_subnodes(subnodes) { }
#endif

    // btree_node_leaf implementation
    const subnode_info& get_value(uint pos) const 
        { return m_subnodes[pos].second; }
    const node_id& get_key(uint pos) const
        { return m_subnodes[pos].first; }
    uint num_values() const
        { return m_subnodes.size(); }

private:
    std::vector<std::pair<node_id, subnode_info> > m_subnodes;   
};

} // end fairport namespace

inline fairport::node fairport::subnode_transform_info::operator()(const fairport::subnode_info& info) const
{ 
    return node(m_parent, info); 
}

inline fairport::block_id fairport::node_impl::get_data_id() const
{ 
    if(m_pdata)
        return m_pdata->get_id();
    
    return m_original_data_id;
}

inline fairport::block_id fairport::node_impl::get_sub_id() const
{ 
    if(m_psub)
        return m_psub->get_id();
    
    return m_original_sub_id;
}

inline size_t fairport::node_impl::size() const
{
    return ensure_data_block()->get_total_size();
}

inline size_t fairport::node_impl::get_page_size(uint page_num) const
{
    return ensure_data_block()->get_page(page_num)->get_total_size();
}
    
inline fairport::uint fairport::node_impl::get_page_count() const 
{ 
    return ensure_data_block()->get_page_count(); 
}

inline size_t fairport::node_impl::read(std::vector<byte>& buffer, ulong offset) const
{ 
    return ensure_data_block()->read(buffer, offset); 
}

inline size_t fairport::node_impl::read_raw(byte* pdest_buffer, size_t size, ulong offset) const
{ 
    return ensure_data_block()->read_raw(pdest_buffer, size, offset); 
}

template<typename T> 
inline T fairport::node_impl::read(ulong offset) const
{
    return ensure_data_block()->read<T>(offset); 
}
    
inline size_t fairport::node_impl::read(std::vector<byte>& buffer, uint page_num, ulong offset) const
{ 
    return ensure_data_block()->get_page(page_num)->read(buffer, offset); 
}

template<typename T> 
inline T fairport::node_impl::read(uint page_num, ulong offset) const
{
    return ensure_data_block()->get_page(page_num)->read<T>(offset); 
}

inline fairport::data_block* fairport::node_impl::ensure_data_block() const
{ 
    if(!m_pdata) 
        m_pdata = m_db->read_data_block(m_original_data_id); 

    return m_pdata.get();
}
    
inline fairport::subnode_block* fairport::node_impl::ensure_sub_block() const
{ 
    if(!m_psub) 
        m_psub = m_db->read_subnode_block(m_original_sub_id); 

    return m_psub.get();
}

inline std::streamsize fairport::node_stream_device::read(char* pbuffer, std::streamsize n)
{
    size_t read = m_pnode->read_raw(reinterpret_cast<byte*>(pbuffer), static_cast<size_t>(n), static_cast<size_t>(m_pos));
    m_pos += read;

    if(read)
        return read;
    else
        return -1;
}

inline std::streampos fairport::node_stream_device::seek(boost::iostreams::stream_offset off, std::ios_base::seekdir way)
{
#if defined(_MSC_VER) && (_MSC_VER < 1600)
#pragma warning(push)
#pragma warning(disable:4244)
#endif
    if(way == std::ios_base::beg)
            m_pos = off;
    else if(way == std::ios_base::end)
        m_pos = m_pnode->size() + off;
    else
        m_pos += off;
#if defined(_MSC_VER) && (_MSC_VER < 1600)
#pragma warning(pop)
#endif

    if(m_pos < 0)
        m_pos = 0;
    else if(static_cast<size_t>(m_pos) > m_pnode->size())
        m_pos = m_pnode->size();

    return m_pos;
}

inline fairport::subnode_block* fairport::subnode_nonleaf_block::get_child(uint pos)
{
    if(m_child_blocks[pos] == NULL)
    {
        m_child_blocks[pos] = get_db_ptr()->read_subnode_block(m_subnode_info[pos].second);
    }

    return m_child_blocks[pos].get();
}

inline const fairport::subnode_block* fairport::subnode_nonleaf_block::get_child(uint pos) const
{
    if(m_child_blocks[pos] == NULL)
    {
        m_child_blocks[pos] = get_db_ptr()->read_subnode_block(m_subnode_info[pos].second);
    }

    return m_child_blocks[pos].get();
}

inline size_t fairport::data_block::read(std::vector<byte>& buffer, ulong offset) const
{
    size_t read_size = buffer.size();
    
    if(read_size > 0)
    {
        if(offset >= get_total_size())
            throw std::out_of_range("offset >= size()");

        read_size = read_raw(&buffer[0], read_size, offset);
    }

    return read_size;
}

template<typename T> 
inline T fairport::data_block::read(ulong offset) const
{
    if(offset >= get_total_size())
        throw std::out_of_range("offset >= size()");
    if(sizeof(T) + offset > get_total_size())
        throw std::out_of_range("sizeof(T) + offset >= size()");

    T t;
    read_raw(reinterpret_cast<byte*>(&t), sizeof(T), offset);

    return t;
}

inline fairport::uint fairport::extended_block::get_page_count() const
{
    assert(m_child_max_total_size % m_child_max_page_count == 0);
    uint page_size = m_child_max_total_size / m_child_max_page_count;
    uint page_count = (get_total_size() / page_size) + ((get_total_size() % page_size) != 0 ? 1 : 0);
    assert(get_level() == 2 || page_count == m_block_info.size());

    return page_count;
}

inline fairport::data_block* fairport::extended_block::get_child_block(uint index) const
{
    if(index >= m_child_blocks.size())
        throw std::out_of_range("index >= m_child_blocks.size()");

    if(m_child_blocks[index] == NULL)
    {
        m_child_blocks[index] = get_db_ptr()->read_data_block(m_block_info[index]);
    }

    return m_child_blocks[index].get();
}

inline std::tr1::shared_ptr<fairport::external_block> fairport::extended_block::get_page(uint page_num) const
{
    uint page = page_num / m_child_max_page_count;
    return get_child_block(page)->get_page(page_num % m_child_max_page_count);
}

inline std::tr1::shared_ptr<fairport::external_block> fairport::external_block::get_page(uint index) const
{
    if(index != 0)
        throw std::out_of_range("index > 0");

    return std::tr1::const_pointer_cast<external_block>(this->shared_from_this());
}

inline size_t fairport::external_block::read_raw(byte* pdest_buffer, size_t size, ulong offset) const
{
    size_t read_size = size;

    assert(offset <= get_total_size());

    if(offset + size > get_total_size())
        read_size = get_total_size() - offset;

    memcpy(pdest_buffer, &m_buffer[offset], read_size);

    return read_size;
}

inline size_t fairport::extended_block::read_raw(byte* pdest_buffer, size_t size, ulong offset) const
{
    assert(offset <= get_total_size());

    if(offset + size > get_total_size())
        size = get_total_size() - offset;

    byte* pend = pdest_buffer + size;

    size_t total_bytes_read = 0;

    while(pdest_buffer != pend)
    {
        // the child this read starts on
        uint child_pos = offset / m_child_max_total_size;
        // offset into the child block this read starts on
        ulong child_offset = offset % m_child_max_total_size;

        // call into our child to read the data
        size_t bytes_read = get_child_block(child_pos)->read_raw(pdest_buffer, size, child_offset);
        assert(bytes_read <= size);
    
        // adjust pointers accordingly
        pdest_buffer += bytes_read;
        offset += bytes_read;
        size -= bytes_read;
        total_bytes_read += bytes_read;

        assert(pdest_buffer <= pend);
    }

    return total_bytes_read;
}

inline fairport::const_subnodeinfo_iterator fairport::node_impl::subnode_info_begin() const
{
    const subnode_block* pblock = ensure_sub_block();
    return pblock->begin();
}

inline fairport::const_subnodeinfo_iterator fairport::node_impl::subnode_info_end() const
{
    const subnode_block* pblock = ensure_sub_block();
    return pblock->end();
}

inline fairport::node fairport::node_impl::lookup(node_id id) const
{
    return node(std::tr1::const_pointer_cast<node_impl>(shared_from_this()), ensure_sub_block()->lookup(id));
}

#ifdef _MSC_VER
#pragma warning(pop)
#endif

#endif