btree.h

Go to the documentation of this file.

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

#ifndef STXXL_CONTAINERS_BTREE_BTREE_HEADER
#define STXXL_CONTAINERS_BTREE_BTREE_HEADER

#include <limits>
#include <stxxl/bits/namespace.h>
#include <stxxl/bits/containers/btree/iterator.h>
#include <stxxl/bits/containers/btree/iterator_map.h>
#include <stxxl/bits/containers/btree/leaf.h>
#include <stxxl/bits/containers/btree/node_cache.h>
#include <stxxl/bits/containers/btree/root_node.h>
#include <stxxl/bits/containers/btree/node.h>
#include <stxxl/vector>

STXXL_BEGIN_NAMESPACE

namespace btree {

template <class KeyType,
 class DataType,
 class CompareType,
 unsigned RawNodeSize,
 unsigned RawLeafSize,
 class PDAllocStrategy
 >
class btree : private noncopyable
{
public:
 typedef KeyType key_type;
 typedef DataType data_type;
 typedef CompareType key_compare;

 typedef btree<KeyType, DataType, CompareType,
 RawNodeSize, RawLeafSize, PDAllocStrategy> self_type;

 typedef PDAllocStrategy alloc_strategy_type;

 typedef stxxl::uint64 size_type;
 typedef stxxl::int64 difference_type;
 typedef std::pair<const key_type, data_type> value_type;
 typedef value_type& reference;
 typedef const value_type& const_reference;
 typedef value_type* pointer;
 typedef value_type const* const_pointer;

 // leaf type declarations
 typedef normal_leaf<key_type, data_type, key_compare, RawLeafSize, self_type> leaf_type;
 friend class normal_leaf<key_type, data_type, key_compare, RawLeafSize, self_type>;
 typedef typename leaf_type::block_type leaf_block_type;
 typedef typename leaf_type::bid_type leaf_bid_type;
 typedef node_cache<leaf_type, self_type> leaf_cache_type;
 friend class node_cache<leaf_type, self_type>;
 // iterator types
 typedef btree_iterator<self_type> iterator;
 typedef btree_const_iterator<self_type> const_iterator;
 friend class btree_iterator_base<self_type>;
 // iterator map type
 typedef iterator_map<self_type> iterator_map_type;
 // node type declarations
 typedef normal_node<key_type, key_compare, RawNodeSize, self_type> node_type;
 typedef typename node_type::block_type node_block_type;
 friend class normal_node<key_type, key_compare, RawNodeSize, self_type>;
 typedef typename node_type::bid_type node_bid_type;
 typedef node_cache<node_type, self_type> node_cache_type;
 friend class node_cache<node_type, self_type>;

 typedef typename leaf_type::value_compare value_compare;

 enum {
 min_node_size = node_type::min_size,
 max_node_size = node_type::max_size,
 min_leaf_size = leaf_type::min_size,
 max_leaf_size = leaf_type::max_size
 };

private:
 key_compare m_key_compare;
 mutable node_cache_type m_node_cache;
 mutable leaf_cache_type m_leaf_cache;
 iterator_map_type m_iterator_map;
 size_type m_size;
 unsigned int m_height;
 bool m_prefetching_enabled;
 block_manager* m_bm;
 alloc_strategy_type m_alloc_strategy;

 typedef std::map<key_type, node_bid_type, key_compare> root_node_type;
 typedef typename root_node_type::iterator root_node_iterator_type;
 typedef typename root_node_type::const_iterator root_node_const_iterator_type;
 typedef std::pair<key_type, node_bid_type> root_node_pair_type;

 root_node_type m_root_node;
 iterator m_end_iterator;

 void insert_into_root(const std::pair<key_type, node_bid_type>& splitter)
 {
 std::pair<root_node_iterator_type, bool> result =
 m_root_node.insert(splitter);
 STXXL_ASSERT(result.second == true);

 if (m_root_node.size() > max_node_size) // root overflow
 {
 STXXL_VERBOSE1("btree::insert_into_root, overflow happened, splitting");

 node_bid_type left_bid;
 node_type* left_node = m_node_cache.get_new_node(left_bid);
 assert(left_node);
 node_bid_type right_bid;
 node_type* right_node = m_node_cache.get_new_node(right_bid);
 assert(right_node);

 const unsigned_type old_size = m_root_node.size();
 const unsigned_type half = m_root_node.size() / 2;
 unsigned_type i = 0;
 root_node_iterator_type it = m_root_node.begin();
 typename node_block_type::iterator block_it = left_node->block().begin();
 while (i < half) // copy smaller part
 {
 *block_it = *it;
 ++i;
 ++block_it;
 ++it;
 }
 left_node->block().info.cur_size = (unsigned int)half;
 key_type left_key = (left_node->block()[half - 1]).first;

 block_it = right_node->block().begin();
 while (i < old_size) // copy larger part
 {
 *block_it = *it;
 ++i;
 ++block_it;
 ++it;
 }
 unsigned_type right_size = right_node->block().info.cur_size = (unsigned int)(old_size - half);
 key_type right_key = (right_node->block()[right_size - 1]).first;

 assert(old_size == right_node->size() + left_node->size());

 // create new root node
 m_root_node.clear();
 m_root_node.insert(root_node_pair_type(left_key, left_bid));
 m_root_node.insert(root_node_pair_type(right_key, right_bid));

 ++m_height;
 STXXL_VERBOSE1("btree Increasing height to " << m_height);
 if (m_node_cache.size() < (m_height - 1))
 {
 STXXL_THROW2(std::runtime_error, "btree::bulk_construction",
 "The height of the tree (" << m_height << ") has exceeded the required capacity (" << (m_node_cache.size() + 1) << ") of the node cache. Increase the node cache size.");
 }
 }
 }

 template <class CacheType>
 void fuse_or_balance(root_node_iterator_type uit, CacheType& cache)
 {
 typedef typename CacheType::node_type local_node_type;
 typedef typename local_node_type::bid_type local_bid_type;

 root_node_iterator_type left_it, right_it;
 if (uit->first == key_compare::max_value())
 {
 // uit is the last entry in the root
 assert(uit != m_root_node.begin());
 right_it = uit;
 left_it = --uit;
 }
 else
 {
 left_it = uit;
 right_it = ++uit;
 assert(right_it != m_root_node.end());
 }

 // now fuse or balance nodes pointed by leftIt and rightIt
 local_bid_type left_bid = (local_bid_type)left_it->second;
 local_bid_type right_bid = (local_bid_type)right_it->second;
 local_node_type* left_node = cache.get_node(left_bid, true);
 local_node_type* right_node = cache.get_node(right_bid, true);

 const unsigned_type total_size = left_node->size() + right_node->size();
 if (total_size <= right_node->max_nelements())
 {
 // --- fuse ---

 // add the content of left_node to right_node
 right_node->fuse(*left_node);

 cache.unfix_node(right_bid);
 // 'delete_node' unfixes left_bid also
 cache.delete_node(left_bid);

 // delete left BID from the root
 m_root_node.erase(left_it);
 }
 else
 {
 // --- balance ---

 key_type new_splitter = right_node->balance(*left_node);

 // delete left BID from the root
 m_root_node.erase(left_it);

 // reinsert with the new key
 m_root_node.insert(root_node_pair_type(new_splitter, (node_bid_type)left_bid));

 cache.unfix_node(left_bid);
 cache.unfix_node(right_bid);
 }
 }

 void create_empty_leaf()
 {
 leaf_bid_type new_bid;
 leaf_type* new_leaf = m_leaf_cache.get_new_node(new_bid);
 assert(new_leaf);
 m_end_iterator = new_leaf->end(); // initialize end() iterator
 m_root_node.insert(
 root_node_pair_type(key_compare::max_value(), (node_bid_type)new_bid)
 );
 }

 void deallocate_children()
 {
 if (m_height == 2)
 {
 // we have children leaves here
 for (root_node_const_iterator_type it = m_root_node.begin();
 it != m_root_node.end(); ++it)
 {
 // delete from leaf cache and deallocate bid
 m_leaf_cache.delete_node((leaf_bid_type)it->second);
 }
 }
 else
 {
 for (root_node_const_iterator_type it = m_root_node.begin();
 it != m_root_node.end(); ++it)
 {
 node_type* node = m_node_cache.get_node((node_bid_type)it->second);
 assert(node);
 node->deallocate_children(m_height - 1);
 // delete from node cache and deallocate bid
 m_node_cache.delete_node((node_bid_type)it->second);
 }
 }
 }

 template <class InputIterator>
 void bulk_construction(InputIterator begin, InputIterator end,
 double node_fill_factor, double leaf_fill_factor)
 {
 assert(node_fill_factor >= 0.5);
 assert(leaf_fill_factor >= 0.5);
 key_type last_key = key_compare::max_value();

 typedef std::pair<key_type, node_bid_type> key_bid_pair;
 typedef typename stxxl::VECTOR_GENERATOR<
 key_bid_pair, 1, 1, node_block_type::raw_size
 >::result key_bid_vector_type;

 key_bid_vector_type bids;

 leaf_bid_type new_bid;
 leaf_type* leaf = m_leaf_cache.get_new_node(new_bid);
 const unsigned_type max_leaf_elements = unsigned_type(
 (double)leaf->max_nelements() * leaf_fill_factor
 );

 while (begin != end)
 {
 // write data in leaves

 // if *b not equal to the last element
 if (m_key_compare(begin->first, last_key) || m_key_compare(last_key, begin->first))
 {
 ++m_size;
 if (leaf->size() == max_leaf_elements)
 {
 // overflow, need a new block
 bids.push_back(key_bid_pair(leaf->back().first, (node_bid_type)new_bid));

 leaf_type* new_leaf = m_leaf_cache.get_new_node(new_bid);
 assert(new_leaf);
 // Setting links
 leaf->succ() = new_leaf->my_bid();
 new_leaf->pred() = leaf->my_bid();

 leaf = new_leaf;
 }
 leaf->push_back(*begin);
 last_key = begin->first;
 }
 ++begin;
 }

 // rebalance the last leaf
 if (leaf->underflows() && !bids.empty())
 {
 leaf_type* left_leaf = m_leaf_cache.get_node((leaf_bid_type)(bids.back().second));
 assert(left_leaf);
 if (left_leaf->size() + leaf->size() <= leaf->max_nelements())
 {
 // can fuse
 leaf->fuse(*left_leaf);
 m_leaf_cache.delete_node((leaf_bid_type)(bids.back().second));
 bids.pop_back();
 assert(!leaf->overflows() && !leaf->underflows());
 }
 else
 {
 // need to rebalance
 const key_type new_splitter = leaf->balance(*left_leaf);
 bids.back().first = new_splitter;
 assert(!left_leaf->overflows() && !left_leaf->underflows());
 }
 }

 assert(!leaf->overflows() && (!leaf->underflows() || m_size <= max_leaf_size));

 m_end_iterator = leaf->end(); // initialize end() iterator

 bids.push_back(key_bid_pair(key_compare::max_value(), (node_bid_type)new_bid));

 const unsigned_type max_node_elements = unsigned_type(
 double(max_node_size) * node_fill_factor
 );

 //-tb fixes bug with only one child remaining in m_root_node
 while (bids.size() > node_type::max_nelements())
 {
 key_bid_vector_type parent_bids;

 stxxl::uint64 nparents = div_ceil(bids.size(), max_node_elements);
 assert(nparents >= 2);
 STXXL_VERBOSE1("btree bulk construct"
 << " bids.size=" << bids.size()
 << " nparents=" << nparents
 << " max_node_elements=" << max_node_elements
 << " node_type::max_nelements=" << node_type::max_nelements());

 for (typename key_bid_vector_type::const_iterator it = bids.begin();
 it != bids.end(); )
 {
 node_bid_type new_bid;
 node_type* node = m_node_cache.get_new_node(new_bid);
 assert(node);

 for (unsigned_type cnt = 0;
 cnt < max_node_elements && it != bids.end(); ++cnt, ++it)
 {
 node->push_back(*it);
 }

 STXXL_VERBOSE1("btree bulk construct node size : " << node->size() << " limits: " << node->min_nelements() << " " << node->max_nelements() << " max_node_elements: " << max_node_elements);

 if (node->underflows())
 {
 // this can happen only at the end
 assert(it == bids.end());
 assert(!parent_bids.empty());

 node_type* left_node = m_node_cache.get_node(parent_bids.back().second);
 assert(left_node);
 if (left_node->size() + node->size() <= node->max_nelements())
 {
 // can fuse
 STXXL_VERBOSE1("btree bulk construct fuse last nodes:"
 << " left_node.size=" << left_node->size()
 << " node.size=" << node->size());

 node->fuse(*left_node);
 m_node_cache.delete_node(parent_bids.back().second);
 parent_bids.pop_back();
 }
 else
 {
 // need to rebalance
 STXXL_VERBOSE1("btree bulk construct rebalance last nodes:"
 << " left_node.size=" << left_node->size()
 << " node.size=" << node->size());

 const key_type new_splitter = node->balance(*left_node, false);
 parent_bids.back().first = new_splitter;

 STXXL_VERBOSE1("btree bulk construct after rebalance:"
 << " left_node.size=" << left_node->size()
 << " node.size=" << node->size());

 assert(!left_node->overflows() && !left_node->underflows());
 }
 }
 assert(!node->overflows() && !node->underflows());

 parent_bids.push_back(key_bid_pair(node->back().first, new_bid));
 }

 STXXL_VERBOSE1("btree parent_bids.size()=" << parent_bids.size()
 << " bids.size()=" << bids.size());

 std::swap(parent_bids, bids);

 assert(nparents == bids.size() || (nparents - 1) == bids.size());

 ++m_height;
 STXXL_VERBOSE1("Increasing height to " << m_height);
 if (m_node_cache.size() < (m_height - 1))
 {
 STXXL_THROW2(std::runtime_error, "btree::bulk_construction",
 "The height of the tree (" << m_height << ") has exceeded the required capacity (" << (m_node_cache.size() + 1) << ") of the node cache. Increase the node cache size.");
 }
 }

 m_root_node.insert(bids.begin(), bids.end());

 STXXL_VERBOSE1("btree bulk root_node_.size()=" << m_root_node.size());
 }

public:
 btree(unsigned_type node_cache_size_in_bytes,
 unsigned_type leaf_cache_size_in_bytes)
 : m_node_cache(node_cache_size_in_bytes, this, m_key_compare),
 m_leaf_cache(leaf_cache_size_in_bytes, this, m_key_compare),
 m_iterator_map(this),
 m_size(0),
 m_height(2),
 m_prefetching_enabled(true),
 m_bm(block_manager::get_instance())
 {
 STXXL_VERBOSE1("Creating a btree, addr=" << this);
 STXXL_VERBOSE1(" bytes in a node: " << node_bid_type::size);
 STXXL_VERBOSE1(" bytes in a leaf: " << leaf_bid_type::size);
 STXXL_VERBOSE1(" elements in a node: " << node_block_type::size);
 STXXL_VERBOSE1(" elements in a leaf: " << leaf_block_type::size);
 STXXL_VERBOSE1(" size of a node element: " << sizeof(typename node_block_type::value_type));
 STXXL_VERBOSE1(" size of a leaf element: " << sizeof(typename leaf_block_type::value_type));

 create_empty_leaf();
 }

 btree(const key_compare& key_compare,
 unsigned_type node_cache_size_in_bytes,
 unsigned_type leaf_cache_size_in_bytes)
 : m_key_compare(key_compare),
 m_node_cache(node_cache_size_in_bytes, this, m_key_compare),
 m_leaf_cache(leaf_cache_size_in_bytes, this, m_key_compare),
 m_iterator_map(this),
 m_size(0),
 m_height(2),
 m_prefetching_enabled(true),
 m_bm(block_manager::get_instance())
 {
 STXXL_VERBOSE1("Creating a btree, addr=" << this);
 STXXL_VERBOSE1(" bytes in a node: " << node_bid_type::size);
 STXXL_VERBOSE1(" bytes in a leaf: " << leaf_bid_type::size);

 create_empty_leaf();
 }

 virtual ~btree()
 {
 try
 {
 deallocate_children();
 }
 catch (...)
 {
 // no exceptions in destructor
 }
 }

 size_type size() const
 {
 return m_size;
 }

 size_type max_size() const
 {
 return std::numeric_limits<size_type>::max();
 }

 bool empty() const
 {
 return !m_size;
 }

 std::pair<iterator, bool> insert(const value_type& x)
 {
 root_node_iterator_type it = m_root_node.lower_bound(x.first);
 assert(!m_root_node.empty());
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("Inserting new value into a leaf");
 leaf_type* leaf = m_leaf_cache.get_node((leaf_bid_type)it->second, true);
 assert(leaf);
 std::pair<key_type, leaf_bid_type> splitter;
 std::pair<iterator, bool> result = leaf->insert(x, splitter);
 if (result.second)
 ++m_size;

 m_leaf_cache.unfix_node((leaf_bid_type)it->second);
 //if(key_compare::max_value() == Splitter.first)
 if (!(m_key_compare(key_compare::max_value(), splitter.first) ||
 m_key_compare(splitter.first, key_compare::max_value())))
 return result;
 // no overflow/splitting happened

 STXXL_VERBOSE1("Inserting new value into root node");

 insert_into_root(std::make_pair(splitter.first, node_bid_type(splitter.second)));

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 // 'it' points to a node
 STXXL_VERBOSE1("Inserting new value into a node");
 node_type* node = m_node_cache.get_node((node_bid_type)it->second, true);
 assert(node);
 std::pair<key_type, node_bid_type> splitter;
 std::pair<iterator, bool> result = node->insert(x, m_height - 1, splitter);
 if (result.second)
 ++m_size;

 m_node_cache.unfix_node((node_bid_type)it->second);
 //if(key_compare::max_value() == Splitter.first)
 if (!(m_key_compare(key_compare::max_value(), splitter.first) ||
 m_key_compare(splitter.first, key_compare::max_value())))
 return result;
 // no overflow/splitting happened

 STXXL_VERBOSE1("Inserting new value into root node");

 insert_into_root(splitter);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);

 return result;
 }

 iterator begin()
 {
 root_node_iterator_type it = m_root_node.begin();
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("btree: retrieving begin() from the first leaf");
 leaf_type* leaf = m_leaf_cache.get_node((leaf_bid_type)it->second);
 assert(leaf);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return leaf->begin();
 }

 // 'it' points to a node
 STXXL_VERBOSE1("btree: retrieving begin() from the first node");
 node_type* node = m_node_cache.get_node((node_bid_type)it->second, true);
 assert(node);
 iterator result = node->begin(m_height - 1);
 m_node_cache.unfix_node((node_bid_type)it->second);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);

 return result;
 }

 const_iterator begin() const
 {
 root_node_const_iterator_type it = m_root_node.begin();
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("btree: retrieving begin() from the first leaf");
 const leaf_type* leaf = m_leaf_cache.get_const_node((leaf_bid_type)it->second);
 assert(leaf);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return leaf->begin();
 }

 // 'it' points to a node
 STXXL_VERBOSE1("btree: retrieving begin() from the first node");
 const node_type* node = m_node_cache.get_const_node((node_bid_type)it->second, true);
 assert(node);
 const_iterator result = node->begin(m_height - 1);
 m_node_cache.unfix_node((node_bid_type)it->second);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 iterator end()
 {
 return m_end_iterator;
 }

 const_iterator end() const
 {
 return m_end_iterator;
 }

 data_type& operator [] (const key_type& k)
 {
 return (*((insert(value_type(k, data_type()))).first)).second;
 }

 iterator find(const key_type& k)
 {
 root_node_iterator_type it = m_root_node.lower_bound(k);
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("Searching in a leaf");
 leaf_type* leaf = m_leaf_cache.get_node((leaf_bid_type)it->second, true);
 assert(leaf);
 iterator result = leaf->find(k);
 m_leaf_cache.unfix_node((leaf_bid_type)it->second);
 assert(result == end() || result->first == k);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 // 'it' points to a node
 STXXL_VERBOSE1("Searching in a node");
 node_type* node = m_node_cache.get_node((node_bid_type)it->second, true);
 assert(node);
 iterator result = node->find(k, m_height - 1);
 m_node_cache.unfix_node((node_bid_type)it->second);

 assert(result == end() || result->first == k);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 const_iterator find(const key_type& k) const
 {
 root_node_const_iterator_type it = m_root_node.lower_bound(k);
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("Searching in a leaf");
 const leaf_type* leaf = m_leaf_cache.get_const_node((leaf_bid_type)it->second, true);
 assert(leaf);
 const_iterator result = leaf->find(k);
 m_leaf_cache.unfix_node((leaf_bid_type)it->second);
 assert(result == end() || result->first == k);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 // 'it' points to a node
 STXXL_VERBOSE1("Searching in a node");
 const node_type* node = m_node_cache.get_const_node((node_bid_type)it->second, true);
 assert(node);
 const_iterator result = node->find(k, m_height - 1);
 m_node_cache.unfix_node((node_bid_type)it->second);

 assert(result == end() || result->first == k);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 iterator lower_bound(const key_type& k)
 {
 root_node_iterator_type it = m_root_node.lower_bound(k);
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("Searching lower bound in a leaf");
 leaf_type* leaf = m_leaf_cache.get_node((leaf_bid_type)it->second, true);
 assert(leaf);
 iterator result = leaf->lower_bound(k);
 m_leaf_cache.unfix_node((leaf_bid_type)it->second);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 // 'it' points to a node
 STXXL_VERBOSE1("Searching lower bound in a node");
 node_type* node = m_node_cache.get_node((node_bid_type)it->second, true);
 assert(node);
 iterator result = node->lower_bound(k, m_height - 1);
 m_node_cache.unfix_node((node_bid_type)it->second);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 const_iterator lower_bound(const key_type& k) const
 {
 root_node_const_iterator_type it = m_root_node.lower_bound(k);
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("Searching lower bound in a leaf");
 const leaf_type* leaf = m_leaf_cache.get_const_node((leaf_bid_type)it->second, true);
 assert(leaf);
 const_iterator result = leaf->lower_bound(k);
 m_leaf_cache.unfix_node((leaf_bid_type)it->second);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 // 'it' points to a node
 STXXL_VERBOSE1("Searching lower bound in a node");
 const node_type* node = m_node_cache.get_const_node((node_bid_type)it->second, true);
 assert(node);
 const_iterator result = node->lower_bound(k, m_height - 1);
 m_node_cache.unfix_node((node_bid_type)it->second);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 iterator upper_bound(const key_type& k)
 {
 root_node_iterator_type it = m_root_node.upper_bound(k);
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("Searching upper bound in a leaf");
 leaf_type* Leaf = m_leaf_cache.get_node((leaf_bid_type)it->second, true);
 assert(Leaf);
 iterator result = Leaf->upper_bound(k);
 m_leaf_cache.unfix_node((leaf_bid_type)it->second);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 // 'it' points to a node
 STXXL_VERBOSE1("Searching upper bound in a node");
 node_type* Node = m_node_cache.get_node((node_bid_type)it->second, true);
 assert(Node);
 iterator result = Node->upper_bound(k, m_height - 1);
 m_node_cache.unfix_node((node_bid_type)it->second);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 const_iterator upper_bound(const key_type& k) const
 {
 root_node_const_iterator_type it = m_root_node.upper_bound(k);
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("Searching upper bound in a leaf");
 const leaf_type* leaf = m_leaf_cache.get_const_node((leaf_bid_type)it->second, true);
 assert(leaf);
 const_iterator result = leaf->upper_bound(k);
 m_leaf_cache.unfix_node((leaf_bid_type)it->second);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 // 'it' points to a node
 STXXL_VERBOSE1("Searching upper bound in a node");
 const node_type* node = m_node_cache.get_const_node((node_bid_type)it->second, true);
 assert(node);
 const_iterator result = node->upper_bound(k, m_height - 1);
 m_node_cache.unfix_node((node_bid_type)it->second);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return result;
 }

 std::pair<iterator, iterator> equal_range(const key_type& k)
 {
 // l->first >= k
 iterator l = lower_bound(k);

 // if (k < l->first)
 if (l == end() || m_key_compare(k, l->first))
 // then upper_bound == lower_bound
 return std::pair<iterator, iterator>(l, l);

 iterator u = l;
 // only one element ==k can exist
 ++u;

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);

 // then upper_bound == (lower_bound+1)
 return std::pair<iterator, iterator>(l, u);
 }

 std::pair<const_iterator, const_iterator> equal_range(const key_type& k) const
 {
 // l->first >= k
 const_iterator l = lower_bound(k);

 // if (k < l->first)
 if (l == end() || m_key_compare(k, l->first))
 // then upper_bound == lower_bound
 return std::pair<const_iterator, const_iterator>(l, l);

 const_iterator u = l;
 // only one element ==k can exist
 ++u;

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 // then upper_bound == (lower_bound+1)
 return std::pair<const_iterator, const_iterator>(l, u);
 }

 size_type erase(const key_type& k)
 {
 root_node_iterator_type it = m_root_node.lower_bound(k);
 assert(it != m_root_node.end());

 if (m_height == 2) // 'it' points to a leaf
 {
 STXXL_VERBOSE1("Deleting key from a leaf");
 leaf_type* Leaf = m_leaf_cache.get_node((leaf_bid_type)it->second, true);
 assert(Leaf);
 size_type result = Leaf->erase(k);
 m_size -= result;
 m_leaf_cache.unfix_node((leaf_bid_type)it->second);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);

 if ((!Leaf->underflows()) || m_root_node.size() == 1)
 return result;
 // no underflow or root has a special degree 1 (too few elements)

 STXXL_VERBOSE1("btree: Fusing or rebalancing a leaf");
 fuse_or_balance(it, m_leaf_cache);

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);

 return result;
 }

 // 'it' points to a node
 STXXL_VERBOSE1("Deleting key from a node");
 assert(m_root_node.size() >= 2);
 node_type* node = m_node_cache.get_node((node_bid_type)it->second, true);
 assert(node);
 size_type result = node->erase(k, m_height - 1);
 m_size -= result;
 m_node_cache.unfix_node((node_bid_type)it->second);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 if (!node->underflows())
 return result;
 // no underflow happened

 STXXL_VERBOSE1("Fusing or rebalancing a node");
 fuse_or_balance(it, m_node_cache);

 if (m_root_node.size() == 1)
 {
 STXXL_VERBOSE1("btree Root has size 1 and height > 2");
 STXXL_VERBOSE1("btree Deallocate root and decrease height");
 it = m_root_node.begin();
 node_bid_type root_bid = it->second;
 assert(it->first == key_compare::max_value());
 node_type* root_node = m_node_cache.get_node(root_bid);
 assert(root_node);
 assert(root_node->back().first == key_compare::max_value());
 m_root_node.clear();
 m_root_node.insert(root_node->block().begin(),
 root_node->block().begin() + root_node->size());

 m_node_cache.delete_node(root_bid);
 --m_height;
 STXXL_VERBOSE1("btree Decreasing height to " << m_height);
 }

 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);

 return result;
 }

 size_type count(const key_type& k)
 {
 if (find(k) == end())
 return 0;

 return 1;
 }

 void erase(iterator pos)
 {
 assert(pos != end());
#ifndef NDEBUG
 size_type old_size = size();
#endif

 erase(pos->first);

 assert(size() == old_size - 1);
 }

 iterator insert(iterator /*pos*/, const value_type& x)
 {
 // pos ignored in the current version
 return insert(x).first;
 }

 void clear()
 {
 deallocate_children();

 m_root_node.clear();

 m_size = 0;
 m_height = 2,

 create_empty_leaf();
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 }

 template <class InputIterator>
 void insert(InputIterator b, InputIterator e)
 {
 while (b != e)
 {
 insert(*(b++));
 }
 }

 template <class InputIterator>
 btree(InputIterator begin,
 InputIterator end,
 const key_compare& c_,
 unsigned_type node_cache_size_in_bytes,
 unsigned_type leaf_cache_size_in_bytes,
 bool range_sorted = false,
 double node_fill_factor = 0.75,
 double leaf_fill_factor = 0.6)
 : m_key_compare(c_),
 m_node_cache(node_cache_size_in_bytes, this, m_key_compare),
 m_leaf_cache(leaf_cache_size_in_bytes, this, m_key_compare),
 m_iterator_map(this),
 m_size(0),
 m_height(2),
 m_prefetching_enabled(true),
 m_bm(block_manager::get_instance())
 {
 STXXL_VERBOSE1("Creating a btree, addr=" << this);
 STXXL_VERBOSE1(" bytes in a node: " << node_bid_type::size);
 STXXL_VERBOSE1(" bytes in a leaf: " << leaf_bid_type::size);

 if (range_sorted == false)
 {
 create_empty_leaf();
 insert(begin, end);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return;
 }

 bulk_construction(begin, end, node_fill_factor, leaf_fill_factor);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 }

 template <class InputIterator>
 btree(InputIterator begin,
 InputIterator end,
 unsigned_type node_cache_size_in_bytes,
 unsigned_type leaf_cache_size_in_bytes,
 bool range_sorted = false,
 double node_fill_factor = 0.75,
 double leaf_fill_factor = 0.6)
 : m_node_cache(node_cache_size_in_bytes, this, m_key_compare),
 m_leaf_cache(leaf_cache_size_in_bytes, this, m_key_compare),
 m_iterator_map(this),
 m_size(0),
 m_height(2),
 m_prefetching_enabled(true),
 m_bm(block_manager::get_instance())
 {
 STXXL_VERBOSE1("Creating a btree, addr=" << this);
 STXXL_VERBOSE1(" bytes in a node: " << node_bid_type::size);
 STXXL_VERBOSE1(" bytes in a leaf: " << leaf_bid_type::size);

 if (range_sorted == false)
 {
 create_empty_leaf();
 insert(begin, end);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 return;
 }

 bulk_construction(begin, end, node_fill_factor, leaf_fill_factor);
 assert(m_leaf_cache.nfixed() == 0);
 assert(m_node_cache.nfixed() == 0);
 }

 void erase(iterator first, iterator last)
 {
 if (first == begin() && last == end())
 clear();
 else
 while (first != last)
 erase(first++);
 }

 key_compare key_comp() const
 {
 return m_key_compare;
 }
 value_compare value_comp() const
 {
 return value_compare(m_key_compare);
 }

 void swap(btree& obj)
 {
 std::swap(m_key_compare, obj.m_key_compare); // OK

 std::swap(m_node_cache, obj.m_node_cache); // OK
 std::swap(m_leaf_cache, obj.m_leaf_cache); // OK

 std::swap(m_iterator_map, obj.m_iterator_map); // must update all iterators

 std::swap(m_end_iterator, obj.m_end_iterator);
 std::swap(m_size, obj.m_size);
 std::swap(m_height, obj.m_height);
 std::swap(m_alloc_strategy, obj.m_alloc_strategy);
 std::swap(m_root_node, obj.m_root_node);
 }

 void enable_prefetching()
 {
 m_prefetching_enabled = true;
 }
 void disable_prefetching()
 {
 m_prefetching_enabled = false;
 }
 bool prefetching_enabled()
 {
 return m_prefetching_enabled;
 }

 void print_statistics(std::ostream& o) const
 {
 o << "Node cache statistics:" << std::endl;
 m_node_cache.print_statistics(o);
 o << "Leaf cache statistics:" << std::endl;
 m_leaf_cache.print_statistics(o);
 }
 void reset_statistics()
 {
 m_node_cache.reset_statistics();
 m_leaf_cache.reset_statistics();
 }
};

template <class KeyType,
 class DataType,
 class CompareType,
 unsigned LogNodeSize,
 unsigned LogLeafSize,
 class PDAllocStrategy
 >
inline bool operator ==
 (const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& a,
 const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& b)
{
 return a.size() == b.size() && std::equal(a.begin(), a.end(), b.begin());
}

template <class KeyType,
 class DataType,
 class CompareType,
 unsigned LogNodeSize,
 unsigned LogLeafSize,
 class PDAllocStrategy
 >
inline bool operator !=
 (const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& a,
 const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& b)
{
 return !(a == b);
}

template <class KeyType,
 class DataType,
 class CompareType,
 unsigned LogNodeSize,
 unsigned LogLeafSize,
 class PDAllocStrategy
 >
inline bool operator <
 (const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& a,
 const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& b)
{
 return std::lexicographical_compare(a.begin(), a.end(), b.begin(), b.end());
}

template <class KeyType,
 class DataType,
 class CompareType,
 unsigned LogNodeSize,
 unsigned LogLeafSize,
 class PDAllocStrategy
 >
inline bool operator >
 (const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& a,
 const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& b)
{
 return b < a;
}

template <class KeyType,
 class DataType,
 class CompareType,
 unsigned LogNodeSize,
 unsigned LogLeafSize,
 class PDAllocStrategy
 >
inline bool operator <=
 (const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& a,
 const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& b)
{
 return !(b < a);
}

template <class KeyType,
 class DataType,
 class CompareType,
 unsigned LogNodeSize,
 unsigned LogLeafSize,
 class PDAllocStrategy
 >
inline bool operator >=
 (const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& a,
 const btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& b)
{
 return !(a < b);
}

} // namespace btree

STXXL_END_NAMESPACE

namespace std {

template <class KeyType,
 class DataType,
 class CompareType,
 unsigned LogNodeSize,
 unsigned LogLeafSize,
 class PDAllocStrategy
 >
void swap(stxxl::btree::btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& a,
 stxxl::btree::btree<KeyType, DataType, CompareType,
 LogNodeSize, LogLeafSize, PDAllocStrategy>& b)
{
 if (&a != &b)
 a.swap(b);
}

} // namespace std

#endif // !STXXL_CONTAINERS_BTREE_BTREE_HEADER