pq_ext_merger.h

/***************************************************************************
 * include/stxxl/bits/containers/pq_ext_merger.h
 *
 * Part of the STXXL. See http://stxxl.sourceforge.net
 *
 * Copyright (C) 1999 Peter Sanders <[email protected]>
 * Copyright (C) 2003, 2004, 2007 Roman Dementiev <[email protected]>
 * Copyright (C) 2007-2009 Johannes Singler <[email protected]>
 * Copyright (C) 2007-2010 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_PQ_EXT_MERGER_HEADER
#define STXXL_PQ_EXT_MERGER_HEADER

#include <stxxl/bits/mng/mng.h>

__STXXL_BEGIN_NAMESPACE


namespace priority_queue_local
{
 template <typename Iterator>
 class short_sequence : public std::pair<Iterator, Iterator>
 {
 typedef std::pair<Iterator, Iterator> pair;

 public:
 typedef Iterator iterator;
 typedef const iterator const_iterator;
 typedef typename std::iterator_traits<iterator>::value_type value_type;
 typedef typename std::iterator_traits<iterator>::difference_type size_type;
 typedef value_type & reference;
 typedef const value_type & const_reference;
 typedef unsigned_type origin_type;

 private:
 origin_type m_origin;

 public:
 short_sequence(Iterator first, Iterator last, origin_type origin) :
 pair(first, last), m_origin(origin)
 { }

 iterator begin()
 {
 return this->first;
 }

 const_iterator begin() const
 {
 return this->first;
 }

 const_iterator cbegin() const
 {
 return begin();
 }

 iterator end()
 {
 return this->second;
 }

 const_iterator end() const
 {
 return this->second;
 }

 const_iterator cend() const
 {
 return end();
 }

 reference front()
 {
 return *begin();
 }

 const_reference front() const
 {
 return *begin();
 }

 reference back()
 {
 return *(end() - 1);
 }

 const_reference back() const
 {
 return *(end() - 1);
 }

 size_type size() const
 {
 return end() - begin();
 }

 bool empty() const
 {
 return size() == 0;
 }

 origin_type origin() const
 {
 return m_origin;
 }
 };

 template <class BlockType_,
 class Cmp_,
 unsigned Arity_,
 class AllocStr_ = STXXL_DEFAULT_ALLOC_STRATEGY>
 class ext_merger : private noncopyable
 {
 public:
 typedef stxxl::uint64 size_type;
 typedef BlockType_ block_type;
 typedef typename block_type::bid_type bid_type;
 typedef typename block_type::value_type value_type;
 typedef Cmp_ comparator_type;
 typedef AllocStr_ alloc_strategy;
 typedef read_write_pool<block_type> pool_type;

 // arity_bound / 2 < arity <= arity_bound
 enum { arity = Arity_, arity_bound = 1UL << (LOG2<Arity_>::ceil) };

 protected:
 comparator_type cmp;

 bool is_sentinel(const value_type & a) const
 {
 return !(cmp(cmp.min_value(), a)); // a <= cmp.min_value()
 }

 bool not_sentinel(const value_type & a) const
 {
 return cmp(cmp.min_value(), a); // a > cmp.min_value()
 }

 struct sequence_state : private noncopyable
 {
 block_type * block; //current block
 unsigned_type current; //current index in current block
 std::list<bid_type> * bids; //list of blocks forming this sequence
 comparator_type cmp;
 ext_merger * merger;
 bool allocated;

 const value_type & operator * () const
 {
 return (*block)[current];
 }

 sequence_state() : bids(NULL), allocated(false)
 { }

 ~sequence_state()
 {
 STXXL_VERBOSE2("ext_merger sequence_state::~sequence_state()");
 if (bids != NULL)
 {
 block_manager * bm = block_manager::get_instance();
 bm->delete_blocks(bids->begin(), bids->end());
 delete bids;
 }
 }

 void make_inf()
 {
 current = 0;
 (*block)[0] = cmp.min_value();
 }

 bool is_sentinel(const value_type & a) const
 {
 return !(cmp(cmp.min_value(), a));
 }

 bool not_sentinel(const value_type & a) const
 {
 return cmp(cmp.min_value(), a);
 }

 void swap(sequence_state & obj)
 {
 if (&obj != this)
 {
 std::swap(current, obj.current);
 std::swap(block, obj.block);
 std::swap(bids, obj.bids);
 assert(merger == obj.merger);
 std::swap(allocated, obj.allocated);
 }
 }

 sequence_state & operator ++ ()
 {
 assert(not_sentinel((*block)[current]));
 assert(current < block->size);

 ++current;

 if (current == block->size)
 {
 STXXL_VERBOSE2("ext_merger sequence_state operator++ crossing block border ");
 // go to the next block
 assert(bids != NULL);
 if (bids->empty()) // if there is no next block
 {
 STXXL_VERBOSE2("ext_merger sequence_state operator++ it was the last block in the sequence ");
 delete bids;
 bids = NULL;
 make_inf();
 }
 else
 {
 STXXL_VERBOSE2("ext_merger sequence_state operator++ there is another block ");
 bid_type bid = bids->front();
 bids->pop_front();
 merger->pool->hint(bid);
 if (!(bids->empty()))
 {
 STXXL_VERBOSE2("ext_merger sequence_state operator++ more blocks exist in a sequence, hinting the next");
 merger->pool->hint(bids->front());
 }
 merger->pool->read(block, bid)->wait();
 STXXL_VERBOSE2("first element of read block " << bid << " " << *(block->begin()) << " cached in " << block);
 if (!(bids->empty()))
 merger->pool->hint(bids->front()); // re-hint, reading might have made a block free
 block_manager::get_instance()->delete_block(bid);
 current = 0;
 }
 }
 return *this;
 }
 };


#if STXXL_PQ_EXTERNAL_LOSER_TREE
 struct Entry
 {
 value_type key; // key of loser element (winner for 0)
 unsigned_type index; // the number of the losing segment
 };
#endif //STXXL_PQ_EXTERNAL_LOSER_TREE

 size_type size_; // total number of elements stored
 unsigned_type log_k; // log of current tree size
 unsigned_type k; // invariant (k == 1 << log_k), always a power of 2
 // only entries 0 .. arity-1 may hold actual sequences, the other
 // entries arity .. arity_bound-1 are sentinels to make the size of the tree
 // a power of 2 always

 // stack of empty segment indices
 internal_bounded_stack<unsigned_type, arity> free_segments;

#if STXXL_PQ_EXTERNAL_LOSER_TREE
 // upper levels of loser trees
 // entry[0] contains the winner info
 Entry entry[arity_bound];
#endif //STXXL_PQ_EXTERNAL_LOSER_TREE

 // leaf information
 // note that Knuth uses indices k..k-1
 // while we use 0..k-1
 sequence_state states[arity_bound]; // sequence including current position, dereference gives current element

 pool_type * pool;

 block_type * sentinel_block;

 public:
 ext_merger() :
 size_(0), log_k(0), k(1), pool(0)
 {
 init();
 }

 ext_merger(pool_type * pool_) :
 size_(0), log_k(0), k(1),
 pool(pool_)
 {
 init();
 }

 virtual ~ext_merger()
 {
 STXXL_VERBOSE1("ext_merger::~ext_merger()");
 for (unsigned_type i = 0; i < arity; ++i)
 {
 delete states[i].block;
 }
 delete sentinel_block;
 }

 void set_pool(pool_type * pool_)
 {
 pool = pool_;
 }

 private:
 void init()
 {
 STXXL_VERBOSE2("ext_merger::init()");
 assert(!cmp(cmp.min_value(), cmp.min_value())); // verify strict weak ordering

 sentinel_block = NULL;
 if (arity < arity_bound)
 {
 sentinel_block = new block_type;
 for (unsigned_type i = 0; i < block_type::size; ++i)
 (*sentinel_block)[i] = cmp.min_value();
 if (arity + 1 == arity_bound) {
 // same memory consumption, but smaller merge width, better use arity = arity_bound
 STXXL_ERRMSG("inefficient PQ parameters for ext_merger: arity + 1 == arity_bound");
 }
 }

 for (unsigned_type i = 0; i < arity_bound; ++i)
 {
 states[i].merger = this;
 if (i < arity)
 states[i].block = new block_type;
 else
 states[i].block = sentinel_block;

 states[i].make_inf();
 }

 assert(k == 1);
 free_segments.push(0); //total state: one free sequence

 rebuild_loser_tree();
#if STXXL_PQ_EXTERNAL_LOSER_TREE
 assert(is_sentinel(*states[entry[0].index]));
#endif //STXXL_PQ_EXTERNAL_LOSER_TREE
 }

 // rebuild loser tree information from the values in current
 void rebuild_loser_tree()
 {
#if STXXL_PQ_EXTERNAL_LOSER_TREE
 unsigned_type winner = init_winner(1);
 entry[0].index = winner;
 entry[0].key = *(states[winner]);
#endif //STXXL_PQ_EXTERNAL_LOSER_TREE
 }


#if STXXL_PQ_EXTERNAL_LOSER_TREE
 // given any values in the leaves this
 // routing recomputes upper levels of the tree
 // from scratch in linear time
 // initialize entry[root].index and the subtree rooted there
 // return winner index
 unsigned_type init_winner(unsigned_type root)
 {
 if (root >= k)
 { // leaf reached
 return root - k;
 }
 else
 {
 unsigned_type left = init_winner(2 * root);
 unsigned_type right = init_winner(2 * root + 1);
 value_type lk = *(states[left]);
 value_type rk = *(states[right]);
 if (!(cmp(lk, rk)))
 { // right subtree looses
 entry[root].index = right;
 entry[root].key = rk;
 return left;
 }
 else
 {
 entry[root].index = left;
 entry[root].key = lk;
 return right;
 }
 }
 }

 // first go up the tree all the way to the root
 // hand down old winner for the respective subtree
 // based on new value, and old winner and loser
 // update each node on the path to the root top down.
 // This is implemented recursively
 void update_on_insert(
 unsigned_type node,
 const value_type & new_key,
 unsigned_type new_index,
 value_type * winner_key,
 unsigned_type * winner_index, // old winner
 unsigned_type * mask) // 1 << (ceil(log KNK) - dist-from-root)
 {
 if (node == 0)
 { // winner part of root
 *mask = 1 << (log_k - 1);
 *winner_key = entry[0].key;
 *winner_index = entry[0].index;
 if (cmp(entry[node].key, new_key))
 {
 entry[node].key = new_key;
 entry[node].index = new_index;
 }
 }
 else
 {
 update_on_insert(node >> 1, new_key, new_index, winner_key, winner_index, mask);
 value_type loserKey = entry[node].key;
 unsigned_type loserIndex = entry[node].index;
 if ((*winner_index & *mask) != (new_index & *mask))
 { // different subtrees
 if (cmp(loserKey, new_key))
 { // new_key will have influence here
 if (cmp(*winner_key, new_key))
 { // old winner loses here
 entry[node].key = *winner_key;
 entry[node].index = *winner_index;
 }
 else
 { // new entry looses here
 entry[node].key = new_key;
 entry[node].index = new_index;
 }
 }
 *winner_key = loserKey;
 *winner_index = loserIndex;
 }
 // note that nothing needs to be done if
 // the winner came from the same subtree
 // a) new_key <= winner_key => even more reason for the other tree to lose
 // b) new_key > winner_key => the old winner will beat the new
 // entry further down the tree
 // also the same old winner is handed down the tree

 *mask >>= 1; // next level
 }
 }
#endif //STXXL_PQ_EXTERNAL_LOSER_TREE

 // make the tree two times as wide
 void double_k()
 {
 STXXL_VERBOSE1("ext_merger::double_k (before) k=" << k << " log_k=" << log_k << " arity_bound=" << arity_bound << " arity=" << arity << " #free=" << free_segments.size());
 assert(k > 0);
 assert(k < arity);
 assert(free_segments.empty()); // stack was free (probably not needed)

 // make all new entries free
 // and push them on the free stack
 for (unsigned_type i = 2 * k - 1; i >= k; i--) //backwards
 {
 states[i].make_inf();
 if (i < arity)
 free_segments.push(i);
 }

 // double the size
 k *= 2;
 log_k++;

 STXXL_VERBOSE1("ext_merger::double_k (after) k=" << k << " log_k=" << log_k << " arity_bound=" << arity_bound << " arity=" << arity << " #free=" << free_segments.size());
 assert(!free_segments.empty());

 // recompute loser tree information
 rebuild_loser_tree();
 }


 // compact nonempty segments in the left half of the tree
 void compact_tree()
 {
 STXXL_VERBOSE1("ext_merger::compact_tree (before) k=" << k << " log_k=" << log_k << " #free=" << free_segments.size());
 assert(log_k > 0);

 // compact all nonempty segments to the left

 unsigned_type target = 0;
 for (unsigned_type from = 0; from < k; from++)
 {
 if (!is_segment_empty(from))
 {
 assert(is_segment_allocated(from));
 if (from != target)
 {
 assert(!is_segment_allocated(target));
 states[target].swap(states[from]);
 }
 ++target;
 }
 }

 // half degree as often as possible
 while (k > 1 && target <= (k / 2))
 {
 k /= 2;
 log_k--;
 }

 // overwrite garbage and compact the stack of free segment indices
 free_segments.clear(); // none free
 for ( ; target < k; target++)
 {
 assert(!is_segment_allocated(target));
 states[target].make_inf();
 if (target < arity)
 free_segments.push(target);
 }

 STXXL_VERBOSE1("ext_merger::compact_tree (after) k=" << k << " log_k=" << log_k << " #free=" << free_segments.size());
 assert(k > 0);

 // recompute loser tree information
 rebuild_loser_tree();
 }


#if 0
 void swap(ext_merger & obj)
 {
 std::swap(cmp, obj.cmp);
 std::swap(free_segments, obj.free_segments);
 std::swap(size_, obj.size_);
 std::swap(log_k, obj.log_k);
 std::swap(k, obj.k);
 swap_1D_arrays(entry, obj.entry, arity_bound);
 swap_1D_arrays(states, obj.states, arity_bound);

 // std::swap(pool,obj.pool);
 }
#endif

 public:
 unsigned_type mem_cons() const // only rough estimation
 {
 return (STXXL_MIN<unsigned_type>(arity + 1, arity_bound) * block_type::raw_size);
 }

 // delete the (length = end-begin) smallest elements and write them to [begin..end)
 // empty segments are deallocated
 // requires:
 // - there are at least length elements
 // - segments are ended by sentinels
 template <class OutputIterator>
 void multi_merge(OutputIterator begin, OutputIterator end)
 {
 size_type length = end - begin;

 STXXL_VERBOSE1("ext_merger::multi_merge from " << k << " sequence(s), length = " << length);

 if (length == 0)
 return;

 assert(k > 0);
 assert(length <= size_);

 //This is the place to make statistics about external multi_merge calls.

#if STXXL_PARALLEL && STXXL_PARALLEL_PQ_MULTIWAY_MERGE_EXTERNAL
 typedef stxxl::int64 diff_type;
 typedef std::pair<typename block_type::iterator, typename block_type::iterator> sequence;

 std::vector<sequence> seqs;
 std::vector<unsigned_type> orig_seq_index;

 Cmp_ cmp;
 priority_queue_local::invert_order<Cmp_, value_type, value_type> inv_cmp(cmp);

 for (unsigned_type i = 0; i < k; ++i) //initialize sequences
 {
 if (states[i].current == states[i].block->size || is_sentinel(*states[i]))
 continue;

 seqs.push_back(std::make_pair(states[i].block->begin() + states[i].current, states[i].block->end()));
 orig_seq_index.push_back(i);

#if STXXL_CHECK_ORDER_IN_SORTS
 if (!is_sentinel(*seqs.back().first) && !stxxl::is_sorted(seqs.back().first, seqs.back().second, inv_cmp))
 {
 STXXL_VERBOSE1("length " << i << " " << (seqs.back().second - seqs.back().first));
 for (value_type * v = seqs.back().first + 1; v < seqs.back().second; ++v)
 {
 if (inv_cmp(*v, *(v - 1)))
 {
 STXXL_VERBOSE1("Error at position " << i << "/" << (v - seqs.back().first - 1) << "/" << (v - seqs.back().first) << " " << *(v - 1) << " " << *v);
 }
 if (is_sentinel(*v))
 {
 STXXL_VERBOSE1("Wrong sentinel at position " << (v - seqs.back().first));
 }
 }
 assert(false);
 }
#endif

 //Hint first non-internal (actually second) block of this sequence.
 if (states[i].bids != NULL && !states[i].bids->empty())
 pool->hint(states[i].bids->front());
 }

 assert(seqs.size() > 0);

#if STXXL_CHECK_ORDER_IN_SORTS
 value_type last_elem;
#endif

 diff_type rest = length; //elements still to merge for this output block

 while (rest > 0)
 {
 value_type min_last = cmp.min_value(); // minimum of the sequences' last elements
 diff_type total_size = 0;

 for (unsigned_type i = 0; i < seqs.size(); ++i)
 {
 diff_type seq_i_size = seqs[i].second - seqs[i].first;
 if (seq_i_size > 0)
 {
 total_size += seq_i_size;
 if (inv_cmp(*(seqs[i].second - 1), min_last))
 min_last = *(seqs[i].second - 1);

 STXXL_VERBOSE1("front block of seq " << i << ": front=" << *(seqs[i].first) << " back=" << *(seqs[i].second - 1) << " len=" << seq_i_size);
 } else {
 STXXL_VERBOSE1("front block of seq " << i << ": empty");
 }
 }

 assert(total_size > 0);
 assert(!is_sentinel(min_last));

 STXXL_VERBOSE1("min_last " << min_last << " total size " << total_size << " num_seq " << seqs.size());

 diff_type less_equal_than_min_last = 0;
 //locate this element in all sequences
 for (unsigned_type i = 0; i < seqs.size(); ++i)
 {
 //assert(seqs[i].first < seqs[i].second);

 typename block_type::iterator position =
 std::upper_bound(seqs[i].first, seqs[i].second, min_last, inv_cmp);

 //no element larger than min_last is merged

 STXXL_VERBOSE1("seq " << i << ": " << (position - seqs[i].first) << " greater equal than " << min_last);

 less_equal_than_min_last += (position - seqs[i].first);
 }

 diff_type output_size = STXXL_MIN(less_equal_than_min_last, rest); // at most rest elements

 STXXL_VERBOSE1("output_size=" << output_size << " = min(leq_t_ml=" << less_equal_than_min_last << ", rest=" << rest << ")");

 assert(output_size > 0);

 //main call

 begin = parallel::multiway_merge(seqs.begin(), seqs.end(), begin, inv_cmp, output_size); //sequence iterators are progressed appropriately

 rest -= output_size;
 size_ -= output_size;

 for (unsigned_type i = 0; i < seqs.size(); ++i)
 {
 sequence_state & state = states[orig_seq_index[i]];

 state.current = seqs[i].first - state.block->begin();

 assert(seqs[i].first <= seqs[i].second);

 if (seqs[i].first == seqs[i].second) //has run empty
 {
 assert(state.current == state.block->size);
 if (state.bids == NULL || state.bids->empty()) // if there is no next block
 {
 STXXL_VERBOSE1("seq " << i << ": ext_merger::multi_merge(...) it was the last block in the sequence ");
 state.make_inf();
 }
 else
 {
#if STXXL_CHECK_ORDER_IN_SORTS
 last_elem = *(seqs[i].second - 1);
#endif
 STXXL_VERBOSE1("seq " << i << ": ext_merger::multi_merge(...) there is another block ");
 bid_type bid = state.bids->front();
 state.bids->pop_front();
 pool->hint(bid);
 if (!(state.bids->empty()))
 {
 STXXL_VERBOSE2("seq " << i << ": ext_merger::multi_merge(...) more blocks exist, hinting the next");
 pool->hint(state.bids->front());
 }
 pool->read(state.block, bid)->wait();
 STXXL_VERBOSE1("seq " << i << ": first element of read block " << bid << " " << *(state.block->begin()) << " cached in " << state.block);
 if (!(state.bids->empty()))
 pool->hint(state.bids->front()); // re-hint, reading might have made a block free
 state.current = 0;
 seqs[i] = std::make_pair(state.block->begin() + state.current, state.block->end());
 block_manager::get_instance()->delete_block(bid);

 #if STXXL_CHECK_ORDER_IN_SORTS
 STXXL_VERBOSE1("before " << last_elem << " after " << *seqs[i].first << " newly loaded block " << bid);
 if (!stxxl::is_sorted(seqs[i].first, seqs[i].second, inv_cmp))
 {
 STXXL_VERBOSE1("length " << i << " " << (seqs[i].second - seqs[i].first));
 for (value_type * v = seqs[i].first + 1; v < seqs[i].second; ++v)
 {
 if (inv_cmp(*v, *(v - 1)))
 {
 STXXL_VERBOSE1("Error at position " << i << "/" << (v - seqs[i].first - 1) << "/" << (v - seqs[i].first) << " " << *(v - 1) << " " << *v);
 }
 if (is_sentinel(*v))
 {
 STXXL_VERBOSE1("Wrong sentinel at position " << (v - seqs[i].first));
 }
 }
 assert(false);
 }
 #endif
 }
 }
 }
 } //while (rest > 1)

 for (unsigned_type i = 0; i < seqs.size(); ++i)
 {
 unsigned_type seg = orig_seq_index[i];
 if (is_segment_empty(seg))
 {
 STXXL_VERBOSE1("deallocated " << seg);
 deallocate_segment(seg);
 }
 }

#else // STXXL_PARALLEL && STXXL_PARALLEL_PQ_MULTIWAY_MERGE_EXTERNAL

 //Hint first non-internal (actually second) block of each sequence.
 for (unsigned_type i = 0; i < k; ++i)
 {
 if (states[i].bids != NULL && !states[i].bids->empty())
 pool->hint(states[i].bids->front());
 }

 switch (log_k) {
 case 0:
 assert(k == 1);
 assert(entry[0].index == 0);
 assert(free_segments.empty());
 //memcpy(target, states[0], length * sizeof(value_type));
 //std::copy(states[0],states[0]+length,target);
 for (size_type i = 0; i < length; ++i, ++(states[0]), ++begin)
 *begin = *(states[0]);

 entry[0].key = **states;
 if (is_segment_empty(0))
 deallocate_segment(0);

 break;
 case 1:
 assert(k == 2);
 merge_iterator(states[0], states[1], begin, length, cmp);
 rebuild_loser_tree();
 if (is_segment_empty(0) && is_segment_allocated(0))
 deallocate_segment(0);

 if (is_segment_empty(1) && is_segment_allocated(1))
 deallocate_segment(1);

 break;
 case 2:
 assert(k == 4);
 if (is_segment_empty(3))
 merge3_iterator(states[0], states[1], states[2], begin, length, cmp);
 else
 merge4_iterator(states[0], states[1], states[2], states[3], begin, length, cmp);
 rebuild_loser_tree();
 if (is_segment_empty(0) && is_segment_allocated(0))
 deallocate_segment(0);

 if (is_segment_empty(1) && is_segment_allocated(1))
 deallocate_segment(1);

 if (is_segment_empty(2) && is_segment_allocated(2))
 deallocate_segment(2);

 if (is_segment_empty(3) && is_segment_allocated(3))
 deallocate_segment(3);

 break;
 case 3: multi_merge_f<OutputIterator, 3>(begin, end);
 break;
 case 4: multi_merge_f<OutputIterator, 4>(begin, end);
 break;
 case 5: multi_merge_f<OutputIterator, 5>(begin, end);
 break;
 case 6: multi_merge_f<OutputIterator, 6>(begin, end);
 break;
 case 7: multi_merge_f<OutputIterator, 7>(begin, end);
 break;
 case 8: multi_merge_f<OutputIterator, 8>(begin, end);
 break;
 case 9: multi_merge_f<OutputIterator, 9>(begin, end);
 break;
 case 10: multi_merge_f<OutputIterator, 10>(begin, end);
 break;
 default: multi_merge_k(begin, end);
 break;
 }

 size_ -= length;
#endif

 // compact tree if it got considerably smaller
 {
 const unsigned_type num_segments_used = std::min<unsigned_type>(arity, k) - free_segments.size();
 const unsigned_type num_segments_trigger = k - (3 * k / 5);
 // using k/2 would be worst case inefficient (for large k)
 // for k \in {2, 4, 8} the trigger is k/2 which is good
 // because we have special mergers for k \in {1, 2, 4}
 // there is also a special 3-way-merger, that will be
 // triggered if k == 4 && is_segment_empty(3)
 STXXL_VERBOSE3("ext_merger compact? k=" << k << " #used=" << num_segments_used
 << " <= #trigger=" << num_segments_trigger << " ==> "
 << ((k > 1 && num_segments_used <= num_segments_trigger) ? "yes" : "no ")
 << " || "
 << ((k == 4 && !free_segments.empty() && !is_segment_empty(3)) ? "yes" : "no ")
 << " #free=" << free_segments.size());
 if (k > 1 && ((num_segments_used <= num_segments_trigger) ||
 (k == 4 && !free_segments.empty() && !is_segment_empty(3))))
 {
 compact_tree();
 }
 }
 }

 private:
#if STXXL_PQ_EXTERNAL_LOSER_TREE
 // multi-merge for arbitrary K
 template <class OutputIterator>
 void multi_merge_k(OutputIterator begin, OutputIterator end)
 {
 Entry * current_pos;
 value_type current_key;
 unsigned_type current_index; // leaf pointed to by current entry
 unsigned_type kReg = k;
 OutputIterator done = end;
 OutputIterator target = begin;
 unsigned_type winner_index = entry[0].index;
 value_type winner_key = entry[0].key;

 while (target != done)
 {
 // write result
 *target = *(states[winner_index]);

 // advance winner segment
 ++(states[winner_index]);

 winner_key = *(states[winner_index]);

 // remove winner segment if empty now
 if (is_sentinel(winner_key)) //
 deallocate_segment(winner_index);


 // go up the entry-tree
 for (unsigned_type i = (winner_index + kReg) >> 1; i > 0; i >>= 1)
 {
 current_pos = entry + i;
 current_key = current_pos->key;
 if (cmp(winner_key, current_key))
 {
 current_index = current_pos->index;
 current_pos->key = winner_key;
 current_pos->index = winner_index;
 winner_key = current_key;
 winner_index = current_index;
 }
 }

 ++target;
 }
 entry[0].index = winner_index;
 entry[0].key = winner_key;
 }

 template <class OutputIterator, unsigned LogK>
 void multi_merge_f(OutputIterator begin, OutputIterator end)
 {
 OutputIterator done = end;
 OutputIterator target = begin;
 unsigned_type winner_index = entry[0].index;
 Entry * reg_entry = entry;
 sequence_state * reg_states = states;
 value_type winner_key = entry[0].key;

 assert(log_k >= LogK);
 while (target != done)
 {
 // write result
 *target = *(reg_states[winner_index]);

 // advance winner segment
 ++(reg_states[winner_index]);

 winner_key = *(reg_states[winner_index]);


 // remove winner segment if empty now
 if (is_sentinel(winner_key))
 deallocate_segment(winner_index);


 ++target;

 // update loser tree
#define TreeStep(L) \
 if (1 << LogK >= 1 << L) { \
 Entry * pos ## L = reg_entry + ((winner_index + (1 << LogK)) >> (((int(LogK - L) + 1) >= 0) ? ((LogK - L) + 1) : 0)); \
 value_type key ## L = pos ## L->key; \
 if (cmp(winner_key, key ## L)) { \
 unsigned_type index ## L = pos ## L->index; \
 pos ## L->key = winner_key; \
 pos ## L->index = winner_index; \
 winner_key = key ## L; \
 winner_index = index ## L; \
 } \
 }
 TreeStep(10);
 TreeStep(9);
 TreeStep(8);
 TreeStep(7);
 TreeStep(6);
 TreeStep(5);
 TreeStep(4);
 TreeStep(3);
 TreeStep(2);
 TreeStep(1);
#undef TreeStep
 }
 reg_entry[0].index = winner_index;
 reg_entry[0].key = winner_key;
 }
#endif //STXXL_PQ_EXTERNAL_LOSER_TREE

 public:
 bool is_space_available() const // for new segment
 {
 return k < arity || !free_segments.empty();
 }


 // insert segment beginning at target
 // require: is_space_available() == 1
 template <class Merger>
 void insert_segment(Merger & another_merger, size_type segment_size)
 {
 STXXL_VERBOSE1("ext_merger::insert_segment(merger,...)" << this);

 if (segment_size > 0)
 {
 // get a free slot
 if (free_segments.empty())
 { // tree is too small
 double_k();
 }
 assert(!free_segments.empty());
 unsigned_type free_slot = free_segments.top();
 free_segments.pop();


 // link new segment
 assert(segment_size);
 unsigned_type nblocks = segment_size / block_type::size;
 //assert(nblocks); // at least one block
 STXXL_VERBOSE1("ext_merger::insert_segment nblocks=" << nblocks);
 if (nblocks == 0)
 {
 STXXL_VERBOSE1("ext_merger::insert_segment(merger,...) WARNING: inserting a segment with " <<
 nblocks << " blocks");
 STXXL_VERBOSE1("THIS IS INEFFICIENT: TRY TO CHANGE PRIORITY QUEUE PARAMETERS");
 }
 unsigned_type first_size = segment_size % block_type::size;
 if (first_size == 0)
 {
 first_size = block_type::size;
 --nblocks;
 }
 block_manager * bm = block_manager::get_instance();
 std::list<bid_type> * bids = new std::list<bid_type>(nblocks);
 bm->new_blocks(alloc_strategy(), bids->begin(), bids->end());
 block_type * first_block = new block_type;

 another_merger.multi_merge(
 first_block->begin() + (block_type::size - first_size),
 first_block->end());

 STXXL_VERBOSE1("last element of first block " << *(first_block->end() - 1));
 assert(!cmp(*(first_block->begin() + (block_type::size - first_size)), *(first_block->end() - 1)));

 assert(pool->size_write() > 0);

 for (typename std::list<bid_type>::iterator curbid = bids->begin(); curbid != bids->end(); ++curbid)
 {
 block_type * b = pool->steal();
 another_merger.multi_merge(b->begin(), b->end());
 STXXL_VERBOSE1("first element of following block " << *curbid << " " << *(b->begin()));
 STXXL_VERBOSE1("last element of following block " << *curbid << " " << *(b->end() - 1));
 assert(!cmp(*(b->begin()), *(b->end() - 1)));
 pool->write(b, *curbid);
 STXXL_VERBOSE1("written to block " << *curbid << " cached in " << b);
 }

 insert_segment(bids, first_block, first_size, free_slot);

 size_ += segment_size;

#if STXXL_PQ_EXTERNAL_LOSER_TREE
 // propagate new information up the tree
 value_type dummyKey;
 unsigned_type dummyIndex;
 unsigned_type dummyMask;
 update_on_insert((free_slot + k) >> 1, *(states[free_slot]), free_slot,
 &dummyKey, &dummyIndex, &dummyMask);
#endif //STXXL_PQ_EXTERNAL_LOSER_TREE
 }
 else
 {
 // deallocate memory ?
 STXXL_VERBOSE1("Merged segment with zero size.");
 }
 }

 size_type size() const { return size_; }

 protected:
 void insert_segment(std::list<bid_type> * bidlist, block_type * first_block,
 unsigned_type first_size, unsigned_type slot)
 {
 STXXL_VERBOSE1("ext_merger::insert_segment(bidlist,...) " << this << " " << bidlist->size() << " " << slot);
 assert(!is_segment_allocated(slot));
 assert(first_size > 0);

 sequence_state & new_sequence = states[slot];
 new_sequence.current = block_type::size - first_size;
 std::swap(new_sequence.block, first_block);
 delete first_block;
 std::swap(new_sequence.bids, bidlist);
 if (bidlist) // the old list
 {
 assert(bidlist->empty());
 delete bidlist;
 }
 new_sequence.allocated = true;
 assert(is_segment_allocated(slot));
 }

 // free an empty segment .
 void deallocate_segment(unsigned_type slot)
 {
 STXXL_VERBOSE1("ext_merger::deallocate_segment() deleting segment " << slot << " allocated=" << int(is_segment_allocated(slot)));
 assert(is_segment_allocated(slot));
 states[slot].allocated = false;
 states[slot].make_inf();

 // push on the stack of free segment indices
 free_segments.push(slot);
 }

 // is this segment empty ?
 bool is_segment_empty(unsigned_type slot) const
 {
 return is_sentinel(*(states[slot]));
 }

 // Is this segment allocated? Otherwise it's empty,
 // already on the stack of free segment indices and can be reused.
 bool is_segment_allocated(unsigned_type slot) const
 {
 return states[slot].allocated;
 }
 }; // ext_merger
} // priority_queue_local


__STXXL_END_NAMESPACE

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