text: // A basic B^e-tree implementation templated on types Key and Value.
  // Keys and Values must be serializable (see swap_space.hpp).
  // Keys must be comparable (via operator< and operator==).
  // Values must be addable (via operator+).
  // See test.cpp for example usage.
  
  // This implementation represents in-memory nodes as objects with two
  // fields:
  // - a std::map mapping keys to child pointers
  // - a std::map mapping (key, timestamp) pairs to messages
  // Nodes are de/serialized to/from an on-disk representation.
  // I/O is managed transparently by a swap_space object.
  
  // This implementation deviates from a "textbook" implementation in
  // that there is not a fixed division of a node's space between pivots
  // and buffered messages.
  
  // In a textbook implementation, nodes have size B, B^e space is
  // devoted to pivots and child pointers, and B-B^e space is devoted to
  // buffering messages.  Whenever a leaf gets too many messages, it
  // splits.  Whenever an internal node gets too many messages, it
  // performs a flush.  Whenever an internal node gets too many
  // children, it splits.  This policy ensures that, whenever the tree
  // needs to flush messages from a node to one of its children, it can
  // always move a batch of size at least (B-B^e) / B^e = B^(1-e) - 1
  // messages.
  
  // In this implementation, nodes have a fixed maximum size.  Whenever
  // a leaf exceeds this max size, it splits.  Whenever an internal node
  // exceeds this maximum size, it checks to see if it can flush a large
  // batch of elements to one of its children.  If it can, it does so.
  // If it cannot, then it splits.
  
  // In-memory nodes may temporarily exceed the maximum size
  // restriction.  During a flush, we move all the incoming messages
  // into the destination node.  At that point the node may exceed the
  // max size.  The flushing procedure then performs further flushes or
  // splits to restore the max-size invariant.  Thus, whenever a flush
  // returns, all the nodes in the subtree of that node are guaranteed
  // to satisfy the max-size requirement.
  
  // This implementation also optimizes I/O based on which nodes are
  // on-disk, clean in memory, or dirty in memory.  For example,
  // inserted items are always immediately flushed as far down the tree
  // as they can go without dirtying any new nodes.  This is because
  // flushing an item to a node that is already dirty will not require
  // any additional I/O, since the node already has to be written back
  // anyway.  Furthermore, it will flush smaller batches to clean
  // in-memory nodes than to on-disk nodes.  This is because dirtying a
  // clean in-memory node only requires a write-back, whereas flushing
  // to an on-disk node requires reading it in and writing it out.
  
  #include <map>
  #include <vector>
  #include <cassert>
  #include "swap_space.hpp"
  #include "backing_store.hpp"
  
  ////////////////// Upserts
  
  // Internally, we store data indexed by both the user-specified key
  // and a timestamp, so that we can apply upserts in the correct order.
  template<class Key>
  class MessageKey {
  public:
    MessageKey(void) :
      key(),
      timestamp(0)
    {}
  
    MessageKey(const Key & k, uint64_t tstamp) :
      key(k),
      timestamp(tstamp)
    {}
  
    static MessageKey range_start(const Key &key) {
      return MessageKey(key, 0);
    }
    
    static MessageKey range_end(const Key &key) {
      return MessageKey(key, UINT64_MAX);
    }
    
    MessageKey range_start(void) const {
      return range_start(key);
    }
  
    MessageKey range_end(void) const {
      return range_end(key);
    }
  
    void _serialize(std::iostream &fs, serialization_context &context) const {
      fs << timestamp << " ";
      serialize(fs, context, key);
    } 
  
    void _deserialize(std::iostream &fs, serialization_context &context) {
      fs >> timestamp;
      deserialize(fs, context, key);
    }
  
    Key key;
    uint64_t timestamp;
  };
  
  template<class Key>
  bool operator<(const MessageKey<Key> & mkey1, const MessageKey<Key> & mkey2) {
    return mkey1.key < mkey2.key ||
  		     (mkey1.key == mkey2.key && mkey1.timestamp < mkey2.timestamp);
  }
  
  template<class Key>
  bool operator<(const Key & key, const MessageKey<Key> & mkey) {
    return key < mkey.key;
  }
  
  template<class Key>
  bool operator<(const MessageKey<Key> & mkey, const Key & key) {
    return mkey.key < key;
  }
  
  template<class Key>
  bool operator==(const MessageKey<Key> &a, const MessageKey<Key> &b) {
    return a.key == b.key && a.timestamp == b.timestamp;
  }
    
  
  // The three types of upsert.  An UPDATE specifies a value, v, that
  // will be added (using operator+) to the old value associated to some
  // key in the tree.  If there is no old value associated with the key,
  // then it will add v to the result of a Value obtained using the
  // default zero-argument constructor.
  #define INSERT (0)
  #define DELETE (1)
  #define UPDATE (2)
  
  template<class Value>
  class Message {
  public:
    Message(void) :
      opcode(INSERT),
      val()
    {}
  
    Message(int opc, const Value &v) :
      opcode(opc),
      val(v)
    {}
    
    void _serialize(std::iostream &fs, serialization_context &context) {
      fs << opcode << " ";
      serialize(fs, context, val);
    } 
  
    void _deserialize(std::iostream &fs, serialization_context &context) {
      fs >> opcode;
      deserialize(fs, context, val);
    }
  
    int opcode;
    Value val;
  };
  
  template <class Value>
  bool operator==(const Message<Value> &a, const Message<Value> &b) {
    return a.opcode == b.opcode && a.val == b.val;
  }
  
  // Measured in messages.
  #define DEFAULT_MAX_NODE_SIZE (1ULL<<18)
  
  // The minimum number of messages that we will flush to an out-of-cache node.
  // Note: we will flush even a single element to a child that is already dirty.
  // Note: we will flush MIN_FLUSH_SIZE/2 items to a clean in-memory child.
  #define DEFAULT_MIN_FLUSH_SIZE (DEFAULT_MAX_NODE_SIZE / 16ULL)
  
  
  template<class Key, class Value> class betree {
  private:
  
    class node;
    // We let a swap_space handle all the I/O.
    typedef typename swap_space::pointer<node> node_pointer;
    class child_info : public serializable {
    public:
      child_info(void)
        : child(),
  	child_size(0)
      {}
      
      child_info(node_pointer child, uint64_t child_size)
        : child(child),
  	child_size(child_size)
      {}
  
      void _serialize(std::iostream &fs, serialization_context &context) {
        serialize(fs, context, child);
        fs << " ";
        serialize(fs, context, child_size);
      }
  
      void _deserialize(std::iostream &fs, serialization_context &context) {
        deserialize(fs, context, child);
        deserialize(fs, context, child_size);
      }
      
      node_pointer child;
      uint64_t child_size;
    };
    typedef typename std::map<Key, child_info> pivot_map;
    typedef typename std::map<MessageKey<Key>, Message<Value> > message_map;
      
    class node : public serializable {
    public:
  
      // Child pointers
      pivot_map pivots;
      message_map elements;
  
      bool is_leaf(void) const {
        return pivots.empty();
      }
  
      // Holy frick-a-moly.  We want to write a const function that
      // returns a const_iterator when called from a const function and
      // a non-const function that returns a (non-const_)iterator when
      // called from a non-const function.  And we don't want to
      // duplicate the code.  The following solution is from
      //         http://stackoverflow.com/a/858893
      template<class OUT, class IN>
      static OUT get_pivot(IN & mp, const Key & k) {
        assert(mp.size() > 0);
        auto it = mp.lower_bound(k);
        if (it == mp.begin() && k < it->first)
  	throw std::out_of_range("Key does not exist "
  				"(it is smaller than any key in DB)");
        if (it == mp.end() || k < it->first)
  	--it;
        return it;      
      }
  
      // Instantiate the above template for const and non-const
      // calls. (template inference doesn't seem to work on this code)
      typename pivot_map::const_iterator get_pivot(const Key & k) const {
        return get_pivot<typename pivot_map::const_iterator,
  		       const pivot_map>(pivots, k);
      }
  
      typename pivot_map::iterator
      get_pivot(const Key & k) {
        return get_pivot<typename pivot_map::iterator, pivot_map>(pivots, k);
      }
  
      // Return iterator pointing to the first element with mk >= k.
      // (Same const/non-const templating trick as above)
      template<class OUT, class IN>
      static OUT get_element_begin(IN & elts, const Key &k) {
        return elts.lower_bound(MessageKey<Key>::range_start(k));
      }
  
      typename message_map::iterator get_element_begin(const Key &k) {
        return get_element_begin<typename message_map::iterator,
  			       message_map>(elements, k);
      }
  
      typename message_map::const_iterator get_element_begin(const Key &k) const {
        return get_element_begin<typename message_map::const_iterator,
  			       const message_map>(elements, k);
      }
  
      // Return iterator pointing to the first element that goes to
      // child indicated by it
      typename message_map::iterator
      get_element_begin(const typename pivot_map::iterator it) {
        return it == pivots.end() ? elements.end() : get_element_begin(it->first);
      }
  
      // Apply a message to ourself.
      void apply(const MessageKey<Key> &mkey, const Message<Value> &elt,
  	       Value &default_value) {
        switch (elt.opcode) {
        case INSERT:
  	elements.erase(elements.lower_bound(mkey.range_start()),
  		       elements.upper_bound(mkey.range_end()));
  	elements[mkey] = elt;
  	break;
  
        case DELETE:
  	elements.erase(elements.lower_bound(mkey.range_start()),
  		       elements.upper_bound(mkey.range_end()));
  	if (!is_leaf())
  	  elements[mkey] = elt;
  	break;
  
        case UPDATE:
  	{
  	  auto iter = elements.upper_bound(mkey.range_end());
  	  if (iter != elements.begin())
  	    iter--;
  	  if (iter == elements.end() || iter->first.key != mkey.key)
  	    if (is_leaf()) {
  	      Value dummy = default_value;
  	      apply(mkey, Message<Value>(INSERT, dummy + elt.val),
  		    default_value);
  	    } else {
  	      elements[mkey] = elt;
  	    }
  	  else {
  	    assert(iter != elements.end() && iter->first.key == mkey.key);
  	    if (iter->second.opcode == INSERT) {
  	      apply(mkey, Message<Value>(INSERT, iter->second.val + elt.val),
  		    default_value);	  
  	    } else {
  	      elements[mkey] = elt;	      
  	    }
  	  }
  	}
  	break;
  
        default:
  	assert(0);
        }
      }
      
      // Requires: there are less than MIN_FLUSH_SIZE things in elements
      //           destined for each child in pivots);
      pivot_map split(betree &bet) {
        assert(pivots.size() + elements.size() >= bet.max_node_size);
        // This size split does a good job of causing the resulting
        // nodes to have size between 0.4 * MAX_NODE_SIZE and 0.6 * MAX_NODE_SIZE.
        int num_new_leaves =
  	(pivots.size() + elements.size())  / (10 * bet.max_node_size / 24);
        int things_per_new_leaf =
  	(pivots.size() + elements.size() + num_new_leaves - 1) / num_new_leaves;
  
        pivot_map result;
        auto pivot_idx = pivots.begin();
        auto elt_idx = elements.begin();
        int things_moved = 0;
        for (int i = 0; i < num_new_leaves; i++) {
  	if (pivot_idx == pivots.end() && elt_idx == elements.end())
  	  break;
  	node_pointer new_node = bet.ss->allocate(new node);
  	result[pivot_idx != pivots.end() ?
  	       pivot_idx->first :
  	       elt_idx->first.key] = child_info(new_node,
  						new_node->elements.size() +
  						new_node->pivots.size());
  	while(things_moved < (i+1) * things_per_new_leaf &&
  	      (pivot_idx != pivots.end() || elt_idx != elements.end())) {
  	  if (pivot_idx != pivots.end()) {
  	    new_node->pivots[pivot_idx->first] = pivot_idx->second;
  	    ++pivot_idx;
  	    things_moved++;
  	    auto elt_end = get_element_begin(pivot_idx);
  	    while (elt_idx != elt_end) {
  	      new_node->elements[elt_idx->first] = elt_idx->second;
  	      ++elt_idx;
  	      things_moved++;
  	    }
  	  } else {
  	    // Must be a leaf
  	    assert(pivots.size() == 0);
  	    new_node->elements[elt_idx->first] = elt_idx->second;
  	    ++elt_idx;
  	    things_moved++;	    
  	  }
  	}
        }
        
        for (auto it = result.begin(); it != result.end(); ++it)
  	it->second.child_size = it->second.child->elements.size() +
  	  it->second.child->pivots.size();
        
        assert(pivot_idx == pivots.end());
        assert(elt_idx == elements.end());
        pivots.clear();
        elements.clear();
        return result;
      }
  
      node_pointer merge(betree &bet,
  		       typename pivot_map::iterator begin,
  		       typename pivot_map::iterator end) {
        node_pointer new_node = bet.ss->allocate(new node);
        for (auto it = begin; it != end; ++it) {
  	new_node->elements.insert(it->second.child->elements.begin(),
  				  it->second.child->elements.end());
  	new_node->pivots.insert(it->second.child->pivots.begin(),
  				  it->second.child->pivots.end());
        }
        return new_node;
      }
  
      void merge_small_children(betree &bet) {
        if (is_leaf())
  	return;
  
        for (auto beginit = pivots.begin(); beginit != pivots.end(); ++beginit) {
  	uint64_t total_size = 0;
  	auto endit = beginit;
  	while (endit != pivots.end()) {
  	  if (total_size + beginit->second.child_size > 6 * bet.max_node_size / 10)
  	    break;
  	  total_size += beginit->second.child_size;
  	  ++endit;
  	}
  	if (endit != beginit) {
  	  node_pointer merged_node = merge(bet, beginit, endit);
  	  for (auto tmp = beginit; tmp != endit; ++tmp) {
  	    tmp->second.child->elements.clear();
  	    tmp->second.child->pivots.clear();
  	  }
  	  Key key = beginit->first;
  	  pivots.erase(beginit, endit);
  	  pivots[key] = child_info(merged_node, merged_node->pivots.size() + merged_node->elements.size());
  	  beginit = pivots.lower_bound(key);
  	}
        }
      }
      
      // Receive a collection of new messages and perform recursive
      // flushes or splits as necessary.  If we split, return a
      // map with the new pivot keys pointing to the new nodes.
      // Otherwise return an empty map.
      pivot_map flush(betree &bet, message_map &elts)
      {
        debug(std::cout << "Flushing " << this << std::endl);
        pivot_map result;
  
        if (elts.size() == 0) {
  	debug(std::cout << "Done (empty input)" << std::endl);
  	return result;
        }
  
        if (is_leaf()) {
  	for (auto it = elts.begin(); it != elts.end(); ++it)
  	  apply(it->first, it->second, bet.default_value);
  	if (elements.size() + pivots.size() >= bet.max_node_size)
  	  result = split(bet);
  	return result;
        }	
  
        ////////////// Non-leaf
        
        // Update the key of the first child, if necessary
        Key oldmin = pivots.begin()->first;
        MessageKey<Key> newmin = elts.begin()->first;
        if (newmin < oldmin) {
  	pivots[newmin.key] = pivots[oldmin];
  	pivots.erase(oldmin);
        }
  
        // If everything is going to a single dirty child, go ahead
        // and put it there.
        auto first_pivot_idx = get_pivot(elts.begin()->first.key);
        auto last_pivot_idx = get_pivot((--elts.end())->first.key);
        if (first_pivot_idx == last_pivot_idx &&
  	  first_pivot_idx->second.child.is_dirty()) {
        	// There shouldn't be anything in our buffer for this child,
        	// but lets assert that just to be safe.
  	{
  	  auto next_pivot_idx = next(first_pivot_idx);
  	  auto elt_start = get_element_begin(first_pivot_idx);
  	  auto elt_end = get_element_begin(next_pivot_idx); 
  	  assert(elt_start == elt_end);
  	}
        	pivot_map new_children = first_pivot_idx->second.child->flush(bet, elts);
        	if (!new_children.empty()) {
        	  pivots.erase(first_pivot_idx);
        	  pivots.insert(new_children.begin(), new_children.end());
        	} else {
  	  first_pivot_idx->second.child_size =
  	    first_pivot_idx->second.child->pivots.size() +
  	    first_pivot_idx->second.child->elements.size();
  	}
  
        } else {
  	
  	for (auto it = elts.begin(); it != elts.end(); ++it)
  	  apply(it->first, it->second, bet.default_value);
  
  	// Now flush to out-of-core or clean children as necessary
  	while (elements.size() + pivots.size() >= bet.max_node_size) {
  	  // Find the child with the largest set of messages in our buffer
  	  unsigned int max_size = 0;
  	  auto child_pivot = pivots.begin();
  	  auto next_pivot = pivots.begin();
  	  for (auto it = pivots.begin(); it != pivots.end(); ++it) {
  	    auto it2 = next(it);
  	    auto elt_it = get_element_begin(it); 
  	    auto elt_it2 = get_element_begin(it2); 
  	    unsigned int dist = distance(elt_it, elt_it2);
  	    if (dist > max_size) {
  	      child_pivot = it;
  	      next_pivot = it2;
  	      max_size = dist;
  	    }
  	  }
  	  if (!(max_size > bet.min_flush_size ||
  		(max_size > bet.min_flush_size/2 &&
  		 child_pivot->second.child.is_in_memory())))
  	    break; // We need to split because we have too many pivots
  	  auto elt_child_it = get_element_begin(child_pivot);
  	  auto elt_next_it = get_element_begin(next_pivot);
  	  message_map child_elts(elt_child_it, elt_next_it);
  	  pivot_map new_children = child_pivot->second.child->flush(bet, child_elts);
  	  elements.erase(elt_child_it, elt_next_it);
  	  if (!new_children.empty()) {
  	    pivots.erase(child_pivot);
  	    pivots.insert(new_children.begin(), new_children.end());
  	  } else {
  	    first_pivot_idx->second.child_size =
  	      child_pivot->second.child->pivots.size() +
  	      child_pivot->second.child->elements.size();
  	  }
  	}
  
  	// We have too many pivots to efficiently flush stuff down, so split
  	if (elements.size() + pivots.size() > bet.max_node_size) {
  	  result = split(bet);
  	}
        }
  
        //merge_small_children(bet);
        
        debug(std::cout << "Done flushing " << this << std::endl);
        return result;
      }
  
      Value query(const betree & bet, const Key k) const
      {
        debug(std::cout << "Querying " << this << std::endl);
        if (is_leaf()) {
  	auto it = elements.lower_bound(MessageKey<Key>::range_start(k));
  	if (it != elements.end() && it->first.key == k) {
  	  assert(it->second.opcode == INSERT);
  	  return it->second.val;
  	} else {
  	  throw std::out_of_range("Key does not exist");
  	}
        }
  
        ///////////// Non-leaf
        
        auto message_iter = get_element_begin(k);
        Value v = bet.default_value;
  
        if (message_iter == elements.end() || k < message_iter->first)
  	// If we don't have any messages for this key, just search
  	// further down the tree.
  	v = get_pivot(k)->second.child->query(bet, k);
        else if (message_iter->second.opcode == UPDATE) {
  	// We have some updates for this key.  Search down the tree.
  	// If it has something, then apply our updates to that.  If it
  	// doesn't have anything, then apply our updates to the
  	// default initial value.
  	try {
  	  Value t = get_pivot(k)->second.child->query(bet, k);
  	  v = t;
  	} catch (std::out_of_range e) {}
        } else if (message_iter->second.opcode == DELETE) {
  	// We have a delete message, so we don't need to look further
  	// down the tree.  If we don't have any further update or
  	// insert messages, then we should return does-not-exist (in
  	// this subtree).
  	message_iter++;
  	if (message_iter == elements.end() || k < message_iter->first)
  	  throw std::out_of_range("Key does not exist");
        } else if (message_iter->second.opcode == INSERT) {
  	// We have an insert message, so we don't need to look further
  	// down the tree.  We'll apply any updates to this value.
  	v = message_iter->second.val;
  	message_iter++;
        }
  
        // Apply any updates to the value obtained above.
        while (message_iter != elements.end() && message_iter->first.key == k) {
  	assert(message_iter->second.opcode == UPDATE);
  	v = v + message_iter->second.val;
  	message_iter++;
        }
  
        return v;
      }
  
      std::pair<MessageKey<Key>, Message<Value> >
      get_next_message_from_children(const MessageKey<Key> *mkey) const {
        if (mkey && *mkey < pivots.begin()->first)
  	mkey = NULL;
        auto it = mkey ? get_pivot(mkey->key) : pivots.begin();
        while (it != pivots.end()) {
  	try {
  	  return it->second.child->get_next_message(mkey);
  	} catch (std::out_of_range e) {}
  	++it;
        }
        throw std::out_of_range("No more messages in any children");
      }
      
      std::pair<MessageKey<Key>, Message<Value> >
      get_next_message(const MessageKey<Key> *mkey) const {
        auto it = mkey ? elements.upper_bound(*mkey) : elements.begin();
  
        if (is_leaf()) {
  	if (it == elements.end())
  	  throw std::out_of_range("No more messages in sub-tree");
  	return std::make_pair(it->first, it->second);
        }
  
        if (it == elements.end())
  	return get_next_message_from_children(mkey);
        
        try {
  	auto kids = get_next_message_from_children(mkey);
  	if (kids.first < it->first)
  	  return kids;
  	else 
  	  return std::make_pair(it->first, it->second);
        } catch (std::out_of_range e) {
  	return std::make_pair(it->first, it->second);	
        }
      }
      
      void _serialize(std::iostream &fs, serialization_context &context) {
        fs << "pivots:" << std::endl;
        serialize(fs, context, pivots);
        fs << "elements:" << std::endl;
        serialize(fs, context, elements);
      }
      
      void _deserialize(std::iostream &fs, serialization_context &context) {
        std::string dummy;
        fs >> dummy;
        deserialize(fs, context, pivots);
        fs >> dummy;
        deserialize(fs, context, elements);
      }
  
      
    };
  
    swap_space *ss;
    uint64_t min_flush_size;
    uint64_t max_node_size;
    uint64_t min_node_size;
    node_pointer root;
    uint64_t next_timestamp = 1; // Nothing has a timestamp of 0
    Value default_value;
    
  public:
    betree(swap_space *sspace,
  	 uint64_t maxnodesize = DEFAULT_MAX_NODE_SIZE,
  	 uint64_t minnodesize = DEFAULT_MAX_NODE_SIZE / 4,
  	 uint64_t minflushsize = DEFAULT_MIN_FLUSH_SIZE) :
      ss(sspace),
      min_flush_size(minflushsize),
      max_node_size(maxnodesize),
      min_node_size(minnodesize)
    {
      root = ss->allocate(new node);
    }
  
    // Insert the specified message and handle a split of the root if it
    // occurs.
    void upsert(int opcode, Key k, Value v)
    {
      message_map tmp;
      tmp[MessageKey<Key>(k, next_timestamp++)] = Message<Value>(opcode, v);
      pivot_map new_nodes = root->flush(*this, tmp);
      if (new_nodes.size() > 0) {
        root = ss->allocate(new node);
        root->pivots = new_nodes;
      }
    }
  
    void insert(Key k, Value v)
    {
      upsert(INSERT, k, v);
    }
  
    void update(Key k, Value v)
    {
      upsert(UPDATE, k, v);
    }
  
    void erase(Key k)
    {
      upsert(DELETE, k, default_value);
    }
    
    Value query(Key k)
    {
      Value v = root->query(*this, k);
      return v;
    }
  
    void dump_messages(void) {
      std::pair<MessageKey<Key>, Message<Value> > current;
  
      std::cout << "############### BEGIN DUMP ##############" << std::endl;
      
      try {
        current = root->get_next_message(NULL);
        do { 
  	std::cout << current.first.key       << " "
  		  << current.first.timestamp << " "
  		  << current.second.opcode   << " "
  		  << current.second.val      << std::endl;
  	current = root->get_next_message(&current.first);
        } while (1);
      } catch (std::out_of_range e) {}
    }
  
    class iterator {
    public:
  
      iterator(const betree &bet)
        : bet(bet),
  	position(),
  	is_valid(false),
  	pos_is_valid(false),
  	first(),
  	second()
      {}
  
      iterator(const betree &bet, const MessageKey<Key> *mkey)
        : bet(bet),
  	position(),	
  	is_valid(false),
  	pos_is_valid(false),
  	first(),
  	second()
      {
        try {
  	position = bet.root->get_next_message(mkey);
  	pos_is_valid = true;
  	setup_next_element();
        } catch (std::out_of_range e) {}
      }
  
      void apply(const MessageKey<Key> &msgkey, const Message<Value> &msg) {
        switch (msg.opcode) {
        case INSERT:
    	first = msgkey.key;
    	second = msg.val;
    	is_valid = true;
    	break;
        case UPDATE:
    	first = msgkey.key;
    	if (is_valid == false)
    	  second = bet.default_value;
    	second = second + msg.val;
    	is_valid = true;
    	break;
        case DELETE:
    	is_valid = false;
    	break;
        default:
    	abort();
    	break;
        }
      }
  
      void setup_next_element(void) {
        is_valid = false;
        while (pos_is_valid && (!is_valid || position.first.key == first)) {
  	apply(position.first, position.second);
  	try {
  	  position = bet.root->get_next_message(&position.first);
  	} catch (std::exception e) {
  	  pos_is_valid = false;
  	}
        }
      }
  
      bool operator==(const iterator &other) {
        return &bet == &other.bet &&
  	is_valid == other.is_valid &&
  	pos_is_valid == other.pos_is_valid &&
  	(!pos_is_valid || position == other.position) &&
  	(!is_valid || (first == other.first && second == other.second));
      }
  
      bool operator!=(const iterator &other) {
        return !operator==(other);
      }
  
      iterator &operator++(void) {
        setup_next_element();
        return *this;
      }
      
      const betree &bet;
      std::pair<MessageKey<Key>, Message<Value> > position;
      bool is_valid;
      bool pos_is_valid;
      Key first;
      Value second;
    };
  
    iterator begin(void) const {
      return iterator(*this, NULL);
    }
  
    iterator lower_bound(Key key) const {
      MessageKey<Key> tmp = MessageKey<Key>::range_start(key);
      return iterator(*this, &tmp);
    }
    
    iterator upper_bound(Key key) const {
      MessageKey<Key> tmp = MessageKey<Key>::range_end(key);
      return iterator(*this, &tmp);
    }
    
    iterator end(void) const {
      return iterator(*this);
    }
  };
  