sort_stream.h

Go to the documentation of this file.

/***************************************************************************
 * include/stxxl/bits/stream/sort_stream.h
 *
 * Part of the STXXL. See http://stxxl.sourceforge.net
 *
 * Copyright (C) 2002-2005 Roman Dementiev <[email protected]>
 * Copyright (C) 2006-2008 Johannes Singler <[email protected]>
 * Copyright (C) 2008-2010 Andreas Beckmann <[email protected]>
 * Copyright (C) 2013 Timo Bingmann <[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_STREAM_SORT_STREAM_HEADER
#define STXXL_STREAM_SORT_STREAM_HEADER

#include <stxxl/bits/config.h>
#include <stxxl/bits/stream/stream.h>
#include <stxxl/bits/mng/block_manager.h>
#include <stxxl/bits/algo/sort_base.h>
#include <stxxl/bits/algo/sort_helper.h>
#include <stxxl/bits/algo/adaptor.h>
#include <stxxl/bits/algo/run_cursor.h>
#include <stxxl/bits/algo/losertree.h>
#include <stxxl/bits/stream/sorted_runs.h>

STXXL_BEGIN_NAMESPACE

namespace stream {

//! \addtogroup streampack Stream Package
//! \{

////////////////////////////////////////////////////////////////////////
// CREATE RUNS //
////////////////////////////////////////////////////////////////////////

//! Forms sorted runs of data from a stream.
//!
//! \tparam Input_ type of the input stream
//! \tparam CompareType_ type of comparison object used for sorting the runs
//! \tparam BlockSize_ size of blocks used to store the runs (in bytes)
//! \tparam AllocStr_ functor that defines allocation strategy for the runs
template <
 class Input_,
 class CompareType_,
 unsigned BlockSize_ = STXXL_DEFAULT_BLOCK_SIZE(typename Input_::value_type),
 class AllocStr_ = STXXL_DEFAULT_ALLOC_STRATEGY>
class basic_runs_creator : private noncopyable
{
public:
 typedef Input_ input_type;
 typedef CompareType_ cmp_type;
 static const unsigned block_size = BlockSize_;
 typedef AllocStr_ allocation_strategy_type;

public:
 typedef typename Input_::value_type value_type;
 typedef typed_block<BlockSize_, value_type> block_type;
 typedef sort_helper::trigger_entry<block_type> trigger_entry_type;
 typedef sorted_runs<trigger_entry_type, cmp_type> sorted_runs_data_type;
 typedef typename sorted_runs_data_type::run_type run_type;
 typedef counting_ptr<sorted_runs_data_type> sorted_runs_type;

protected:
 Input_& m_input; //! reference to the input stream
 CompareType_ m_cmp; //! comparator used to sort block groups

private:
 sorted_runs_type m_result; //! stores the result (sorted runs) as smart pointer
 unsigned_type m_memsize; //! memory for internal use in blocks
 bool m_result_computed; //! true iff result is already computed (used in 'result()' method)

 //! Fetch data from input into blocks[first_idx,last_idx).
 unsigned_type fetch(block_type* blocks, unsigned_type first_idx, unsigned_type last_idx)
 {
 typename element_iterator_traits<block_type>::element_iterator output =
 make_element_iterator(blocks, first_idx);
 unsigned_type curr_idx = first_idx;
 while (!m_input.empty() && curr_idx != last_idx) {
 *output = *m_input;
 ++m_input;
 ++output;
 ++curr_idx;
 }
 return curr_idx;
 }

 //! fill the rest of the block with max values
 void fill_with_max_value(block_type* blocks, unsigned_type num_blocks, unsigned_type first_idx)
 {
 unsigned_type last_idx = num_blocks * block_type::size;
 if (first_idx < last_idx) {
 typename element_iterator_traits<block_type>::element_iterator curr =
 make_element_iterator(blocks, first_idx);
 while (first_idx != last_idx) {
 *curr = m_cmp.max_value();
 ++curr;
 ++first_idx;
 }
 }
 }

 //! Sort a specific run, contained in a sequences of blocks.
 void sort_run(block_type* run, unsigned_type elements)
 {
 check_sort_settings();
 potentially_parallel::sort(make_element_iterator(run, 0),
 make_element_iterator(run, elements),
 m_cmp);
 }

 void compute_result();

public:
 //! Create the object.
 //! \param input input stream
 //! \param cmp comparator object
 //! \param memory_to_use memory amount that is allowed to used by the sorter in bytes
 basic_runs_creator(Input_& input, CompareType_ cmp, unsigned_type memory_to_use)
 : m_input(input),
 m_cmp(cmp),
 m_result(new sorted_runs_data_type),
 m_memsize(memory_to_use / BlockSize_ / sort_memory_usage_factor()),
 m_result_computed(false)
 {
 sort_helper::verify_sentinel_strict_weak_ordering(cmp);
 if (!(2 * BlockSize_ * sort_memory_usage_factor() <= memory_to_use)) {
 throw bad_parameter("stxxl::runs_creator<>:runs_creator(): INSUFFICIENT MEMORY provided, please increase parameter 'memory_to_use'");
 }
 assert(m_memsize > 0);
 }

 //! Returns the sorted runs object.
 //! \return Sorted runs object. The result is computed lazily, i.e. on the first call
 //! \remark Returned object is intended to be used by \c runs_merger object as input
 sorted_runs_type & result()
 {
 if (!m_result_computed)
 {
 compute_result();
 m_result_computed = true;
#ifdef STXXL_PRINT_STAT_AFTER_RF
 STXXL_MSG(*stats::get_instance());
#endif //STXXL_PRINT_STAT_AFTER_RF
 }
 return m_result;
 }
};

//! Finish the results, i. e. create all runs.
//!
//! This is the main routine of this class.
template <class Input_, class CompareType_, unsigned BlockSize_, class AllocStr_>
void basic_runs_creator<Input_, CompareType_, BlockSize_, AllocStr_>::compute_result()
{
 unsigned_type i = 0;
 unsigned_type m2 = m_memsize / 2;
 const unsigned_type el_in_run = m2 * block_type::size; // # el in a run
 STXXL_VERBOSE1("basic_runs_creator::compute_result m2=" << m2);
 unsigned_type blocks1_length = 0, blocks2_length = 0;
 block_type* Blocks1 = NULL;

#ifndef STXXL_SMALL_INPUT_PSORT_OPT
 Blocks1 = new block_type[m2 * 2];
#else
 // push input element into small_run vector in result until it is full
 while (!input.empty() && blocks1_length != block_type::size)
 {
 m_result->small_run.push_back(*input);
 ++input;
 ++blocks1_length;
 }

 if (blocks1_length == block_type::size && !input.empty())
 {
 Blocks1 = new block_type[m2 * 2];
 std::copy(m_result->small_run.begin(), m_result->small_run.end(), Blocks1[0].begin());
 m_result->small_run.clear();
 }
 else
 {
 STXXL_VERBOSE1("basic_runs_creator: Small input optimization, input length: " << blocks1_length);
 m_result->elements = blocks1_length;
 check_sort_settings();
 potentially_parallel::sort(m_result->small_run.begin(), m_result->small_run.end(), cmp);
 return;
 }
#endif //STXXL_SMALL_INPUT_PSORT_OPT

 // the first block may be there already, now fetch until memsize is filled.
 blocks1_length = fetch(Blocks1, blocks1_length, el_in_run);

 // sort first run
 sort_run(Blocks1, blocks1_length);

 if (blocks1_length <= block_type::size && m_input.empty())
 {
 // small input, do not flush it on the disk(s)
 STXXL_VERBOSE1("basic_runs_creator: Small input optimization, input length: " << blocks1_length);
 assert(m_result->small_run.empty());
 m_result->small_run.assign(Blocks1[0].begin(), Blocks1[0].begin() + blocks1_length);
 m_result->elements = blocks1_length;
 delete[] Blocks1;
 return;
 }

 block_type* Blocks2 = Blocks1 + m2;
 block_manager* bm = block_manager::get_instance();
 request_ptr* write_reqs = new request_ptr[m2];
 run_type run;

 unsigned_type cur_run_size = div_ceil(blocks1_length, block_type::size); // in blocks
 run.resize(cur_run_size);
 bm->new_blocks(AllocStr_(), make_bid_iterator(run.begin()), make_bid_iterator(run.end()));

 disk_queues::get_instance()->set_priority_op(request_queue::WRITE);

 // fill the rest of the last block with max values
 fill_with_max_value(Blocks1, cur_run_size, blocks1_length);

 for (i = 0; i < cur_run_size; ++i)
 {
 run[i].value = Blocks1[i][0];
 write_reqs[i] = Blocks1[i].write(run[i].bid);
 }
 m_result->runs.push_back(run);
 m_result->runs_sizes.push_back(blocks1_length);
 m_result->elements += blocks1_length;

 if (m_input.empty())
 {
 // return
 wait_all(write_reqs, write_reqs + cur_run_size);
 delete[] write_reqs;
 delete[] Blocks1;
 return;
 }

 STXXL_VERBOSE1("Filling the second part of the allocated blocks");
 blocks2_length = fetch(Blocks2, 0, el_in_run);

 if (m_input.empty())
 {
 // optimization if the whole set fits into both halves
 // (re)sort internally and return
 blocks2_length += el_in_run;
 sort_run(Blocks1, blocks2_length); // sort first an second run together
 wait_all(write_reqs, write_reqs + cur_run_size);
 bm->delete_blocks(make_bid_iterator(run.begin()), make_bid_iterator(run.end()));

 cur_run_size = div_ceil(blocks2_length, block_type::size);
 run.resize(cur_run_size);
 bm->new_blocks(AllocStr_(), make_bid_iterator(run.begin()), make_bid_iterator(run.end()));

 // fill the rest of the last block with max values
 fill_with_max_value(Blocks1, cur_run_size, blocks2_length);

 assert(cur_run_size > m2);

 for (i = 0; i < m2; ++i)
 {
 run[i].value = Blocks1[i][0];
 write_reqs[i]->wait();
 write_reqs[i] = Blocks1[i].write(run[i].bid);
 }

 request_ptr* write_reqs1 = new request_ptr[cur_run_size - m2];

 for ( ; i < cur_run_size; ++i)
 {
 run[i].value = Blocks1[i][0];
 write_reqs1[i - m2] = Blocks1[i].write(run[i].bid);
 }

 m_result->runs[0] = run;
 m_result->runs_sizes[0] = blocks2_length;
 m_result->elements = blocks2_length;

 wait_all(write_reqs, write_reqs + m2);
 delete[] write_reqs;
 wait_all(write_reqs1, write_reqs1 + cur_run_size - m2);
 delete[] write_reqs1;

 delete[] Blocks1;

 return;
 }

 // more than 2 runs can be filled, i. e. the general case

 sort_run(Blocks2, blocks2_length);

 cur_run_size = div_ceil(blocks2_length, block_type::size); // in blocks
 run.resize(cur_run_size);
 bm->new_blocks(AllocStr_(), make_bid_iterator(run.begin()), make_bid_iterator(run.end()));

 for (i = 0; i < cur_run_size; ++i)
 {
 run[i].value = Blocks2[i][0];
 write_reqs[i]->wait();
 write_reqs[i] = Blocks2[i].write(run[i].bid);
 }
 assert((blocks2_length % el_in_run) == 0);

 m_result->add_run(run, blocks2_length);

 while (!m_input.empty())
 {
 blocks1_length = fetch(Blocks1, 0, el_in_run);
 sort_run(Blocks1, blocks1_length);
 cur_run_size = div_ceil(blocks1_length, block_type::size); // in blocks
 run.resize(cur_run_size);
 bm->new_blocks(AllocStr_(), make_bid_iterator(run.begin()), make_bid_iterator(run.end()));

 // fill the rest of the last block with max values (occurs only on the last run)
 fill_with_max_value(Blocks1, cur_run_size, blocks1_length);

 for (i = 0; i < cur_run_size; ++i)
 {
 run[i].value = Blocks1[i][0];
 write_reqs[i]->wait();
 write_reqs[i] = Blocks1[i].write(run[i].bid);
 }
 m_result->add_run(run, blocks1_length);

 std::swap(Blocks1, Blocks2);
 std::swap(blocks1_length, blocks2_length);
 }

 wait_all(write_reqs, write_reqs + m2);
 delete[] write_reqs;
 delete[] ((Blocks1 < Blocks2) ? Blocks1 : Blocks2);
}

//! Forms sorted runs of data from a stream.
//!
//! \tparam Input_ type of the input stream
//! \tparam CompareType_ type of omparison object used for sorting the runs
//! \tparam BlockSize_ size of blocks used to store the runs
//! \tparam AllocStr_ functor that defines allocation strategy for the runs
template <
 class Input_,
 class CompareType_,
 unsigned BlockSize_ = STXXL_DEFAULT_BLOCK_SIZE(typename Input_::value_type),
 class AllocStr_ = STXXL_DEFAULT_ALLOC_STRATEGY>
class runs_creator : public basic_runs_creator<Input_, CompareType_, BlockSize_, AllocStr_>
{
private:
 typedef basic_runs_creator<Input_, CompareType_, BlockSize_, AllocStr_> base;

public:
 typedef typename base::cmp_type cmp_type;
 typedef typename base::value_type value_type;
 typedef typename base::block_type block_type;
 typedef typename base::sorted_runs_data_type sorted_runs_data_type;
 typedef typename base::sorted_runs_type sorted_runs_type;

public:
 //! Creates the object.
 //! \param input input stream
 //! \param cmp comparator object
 //! \param memory_to_use memory amount that is allowed to used by the sorter in bytes
 runs_creator(Input_& input, CompareType_ cmp, unsigned_type memory_to_use)
 : base(input, cmp, memory_to_use)
 { }
};


//! Input strategy for \c runs_creator class.
//!
//! This strategy together with \c runs_creator class
//! allows to create sorted runs
//! data structure usable for \c runs_merger
//! pushing elements into the sorter
//! (using runs_creator::push())
template <class ValueType_>
struct use_push
{
 typedef ValueType_ value_type;
};

//! Forms sorted runs of elements passed in push() method.
//!
//! A specialization of \c runs_creator that
//! allows to create sorted runs
//! data structure usable for \c runs_merger from
//! elements passed in sorted push() method. <BR>
//! \tparam ValueType_ type of values (parameter for \c use_push strategy)
//! \tparam CompareType_ type of comparison object used for sorting the runs
//! \tparam BlockSize_ size of blocks used to store the runs
//! \tparam AllocStr_ functor that defines allocation strategy for the runs
template <
 class ValueType_,
 class CompareType_,
 unsigned BlockSize_,
 class AllocStr_>
class runs_creator<
 use_push<ValueType_>,
 CompareType_,
 BlockSize_,
 AllocStr_>
 : private noncopyable
{
public:
 typedef CompareType_ cmp_type;
 typedef ValueType_ value_type;
 typedef typed_block<BlockSize_, value_type> block_type;
 typedef sort_helper::trigger_entry<block_type> trigger_entry_type;
 typedef sorted_runs<trigger_entry_type, cmp_type> sorted_runs_data_type;
 typedef counting_ptr<sorted_runs_data_type> sorted_runs_type;
 typedef sorted_runs_type result_type;

private:
 //! comparator object to sort runs
 CompareType_ m_cmp;

 typedef typename sorted_runs_data_type::run_type run_type;

 //! stores the result (sorted runs) in a reference counted object
 sorted_runs_type m_result;

 //! memory size in bytes to use
 const unsigned_type m_memory_to_use;

 //! memory size in numberr of blocks for internal use
 const unsigned_type m_memsize;

 //! m_memsize / 2
 const unsigned_type m_m2;

 //! true after the result() method was called for the first time
 bool m_result_computed;

 //! total number of elements in a run
 const unsigned_type m_el_in_run;

 //! current number of elements in the run m_blocks1
 unsigned_type m_cur_el;

 //! accumulation buffer of size m_m2 blocks, half the available memory size
 block_type* m_blocks1;

 //! accumulation buffer that is currently being written to disk
 block_type* m_blocks2;

 //! reference to write requests transporting the last accumulation buffer to disk
 request_ptr* m_write_reqs;

 //! run object containing block ids of the run being written to disk
 run_type run;

protected:
 //! fill the rest of the block with max values
 void fill_with_max_value(block_type* blocks, unsigned_type num_blocks, unsigned_type first_idx)
 {
 unsigned_type last_idx = num_blocks * block_type::size;
 if (first_idx < last_idx) {
 typename element_iterator_traits<block_type>::element_iterator curr =
 make_element_iterator(blocks, first_idx);
 while (first_idx != last_idx) {
 *curr = m_cmp.max_value();
 ++curr;
 ++first_idx;
 }
 }
 }

 //! Sort a specific run, contained in a sequences of blocks.
 void sort_run(block_type* run, unsigned_type elements)
 {
 check_sort_settings();
 potentially_parallel::sort(make_element_iterator(run, 0),
 make_element_iterator(run, elements),
 m_cmp);
 }

 void compute_result()
 {
 if (m_cur_el == 0)
 return;

 sort_run(m_blocks1, m_cur_el);

 if (m_cur_el <= block_type::size && m_result->elements == 0)
 {
 // small input, do not flush it on the disk(s)
 STXXL_VERBOSE1("runs_creator(use_push): Small input optimization, input length: " << m_cur_el);
 m_result->small_run.assign(m_blocks1[0].begin(), m_blocks1[0].begin() + m_cur_el);
 m_result->elements = m_cur_el;
 return;
 }

 const unsigned_type cur_run_size = div_ceil(m_cur_el, block_type::size); // in blocks
 run.resize(cur_run_size);
 block_manager* bm = block_manager::get_instance();
 bm->new_blocks(AllocStr_(), make_bid_iterator(run.begin()), make_bid_iterator(run.end()));

 disk_queues::get_instance()->set_priority_op(request_queue::WRITE);

 // fill the rest of the last block with max values
 fill_with_max_value(m_blocks1, cur_run_size, m_cur_el);

 unsigned_type i = 0;
 for ( ; i < cur_run_size; ++i)
 {
 run[i].value = m_blocks1[i][0];
 if (m_write_reqs[i].get())
 m_write_reqs[i]->wait();

 m_write_reqs[i] = m_blocks1[i].write(run[i].bid);
 }
 m_result->add_run(run, m_cur_el);

 for (i = 0; i < m_m2; ++i)
 {
 if (m_write_reqs[i].get())
 m_write_reqs[i]->wait();
 }
 }

public:
 //! Creates the object.
 //! \param cmp comparator object
 //! \param memory_to_use memory amount that is allowed to used by the sorter in bytes
 runs_creator(CompareType_ cmp, unsigned_type memory_to_use) :
 m_cmp(cmp),
 m_memory_to_use(memory_to_use),
 m_memsize(memory_to_use / BlockSize_ / sort_memory_usage_factor()),
 m_m2(m_memsize / 2),
 m_el_in_run(m_m2 * block_type::size),
 m_blocks1(NULL), m_blocks2(NULL),
 m_write_reqs(NULL)
 {
 sort_helper::verify_sentinel_strict_weak_ordering(m_cmp);
 if (!(2 * BlockSize_ * sort_memory_usage_factor() <= m_memory_to_use)) {
 throw bad_parameter("stxxl::runs_creator<>:runs_creator(): INSUFFICIENT MEMORY provided, please increase parameter 'memory_to_use'");
 }
 assert(m_m2 > 0);

 allocate();
 }

 ~runs_creator()
 {
 m_result_computed = 1;
 deallocate();
 }

 //! Clear current state and remove all items.
 void clear()
 {
 if (!m_result)
 m_result = new sorted_runs_data_type;
 else
 m_result->clear();

 m_result_computed = false;
 m_cur_el = 0;

 for (unsigned_type i = 0; i < m_m2; ++i)
 {
 if (m_write_reqs[i].get())
 m_write_reqs[i]->cancel();
 }
 }

 //! Allocates input buffers and clears result.
 void allocate()
 {
 if (!m_blocks1)
 {
 m_blocks1 = new block_type[m_m2 * 2];
 m_blocks2 = m_blocks1 + m_m2;

 m_write_reqs = new request_ptr[m_m2];
 }

 clear();
 }

 //! Deallocates input buffers but not the current result.
 void deallocate()
 {
 result(); // finishes result

 if (m_blocks1)
 {
 delete[] ((m_blocks1 < m_blocks2) ? m_blocks1 : m_blocks2);
 m_blocks1 = m_blocks2 = NULL;

 delete[] m_write_reqs;
 m_write_reqs = NULL;
 }
 }

 //! Adds new element to the sorter.
 //! \param val value to be added
 void push(const value_type& val)
 {
 assert(m_result_computed == false);
 if (LIKELY(m_cur_el < m_el_in_run))
 {
 m_blocks1[m_cur_el / block_type::size][m_cur_el % block_type::size] = val;
 ++m_cur_el;
 return;
 }

 assert(m_el_in_run == m_cur_el);
 m_cur_el = 0;

 // sort and store m_blocks1
 sort_run(m_blocks1, m_el_in_run);

 const unsigned_type cur_run_blocks = div_ceil(m_el_in_run, block_type::size); // in blocks
 run.resize(cur_run_blocks);
 block_manager* bm = block_manager::get_instance();
 bm->new_blocks(AllocStr_(), make_bid_iterator(run.begin()), make_bid_iterator(run.end()));

 disk_queues::get_instance()->set_priority_op(request_queue::WRITE);

 for (unsigned_type i = 0; i < cur_run_blocks; ++i)
 {
 run[i].value = m_blocks1[i][0];
 if (m_write_reqs[i].get())
 m_write_reqs[i]->wait();

 m_write_reqs[i] = m_blocks1[i].write(run[i].bid);
 }

 m_result->add_run(run, m_el_in_run);

 std::swap(m_blocks1, m_blocks2);

 push(val);
 }

 //! Returns the sorted runs object.
 //! \return Sorted runs object.
 //! \remark Returned object is intended to be used by \c runs_merger object as input
 sorted_runs_type & result()
 {
 if (!m_result_computed)
 {
 compute_result();
 m_result_computed = true;
#ifdef STXXL_PRINT_STAT_AFTER_RF
 STXXL_MSG(*stats::get_instance());
#endif //STXXL_PRINT_STAT_AFTER_RF
 }
 return m_result;
 }

 //! number of items currently inserted.
 unsigned_type size() const
 {
 return m_result->elements + m_cur_el;
 }

 //! return comparator object.
 const cmp_type & cmp() const
 {
 return m_cmp;
 }

 //! return memory size used (in bytes).
 unsigned_type memory_used() const
 {
 return m_memory_to_use;
 }
};


//! Input strategy for \c runs_creator class.
//!
//! This strategy together with \c runs_creator class
//! allows to create sorted runs
//! data structure usable for \c runs_merger from
//! sequences of elements in sorted order
template <class ValueType_>
struct from_sorted_sequences
{
 typedef ValueType_ value_type;
};

//! Forms sorted runs of data taking elements in sorted order (element by element).
//!
//! A specialization of \c runs_creator that
//! allows to create sorted runs
//! data structure usable for \c runs_merger from
//! sequences of elements in sorted order. <BR>
//! \tparam ValueType_ type of values (parameter for \c from_sorted_sequences strategy)
//! \tparam CompareType_ type of comparison object used for sorting the runs
//! \tparam BlockSize_ size of blocks used to store the runs
//! \tparam AllocStr_ functor that defines allocation strategy for the runs
template <
 class ValueType_,
 class CompareType_,
 unsigned BlockSize_,
 class AllocStr_>
class runs_creator<
 from_sorted_sequences<ValueType_>,
 CompareType_,
 BlockSize_,
 AllocStr_>
 : private noncopyable
{
public:
 typedef ValueType_ value_type;
 typedef typed_block<BlockSize_, value_type> block_type;
 typedef sort_helper::trigger_entry<block_type> trigger_entry_type;
 typedef AllocStr_ alloc_strategy_type;

public:
 typedef CompareType_ cmp_type;
 typedef sorted_runs<trigger_entry_type, cmp_type> sorted_runs_data_type;
 typedef counting_ptr<sorted_runs_data_type> sorted_runs_type;
 typedef sorted_runs_type result_type;

private:
 typedef typename sorted_runs_data_type::run_type run_type;

 CompareType_ cmp;

 sorted_runs_type result_; // stores the result (sorted runs)
 unsigned_type m_; // memory for internal use in blocks
 buffered_writer<block_type> writer;
 block_type* cur_block;
 unsigned_type offset;
 unsigned_type iblock;
 unsigned_type irun;
 alloc_strategy_type alloc_strategy; // needs to be reset after each run

public:
 //! Creates the object.
 //! \param c comparator object
 //! \param memory_to_use memory amount that is allowed to used by the sorter in bytes.
 //! Recommended value: 2 * block_size * D
 runs_creator(CompareType_ c, unsigned_type memory_to_use) :
 cmp(c),
 result_(new sorted_runs_data_type),
 m_(memory_to_use / BlockSize_ / sort_memory_usage_factor()),
 writer(m_, m_ / 2),
 cur_block(writer.get_free_block()),
 offset(0),
 iblock(0),
 irun(0)
 {
 sort_helper::verify_sentinel_strict_weak_ordering(cmp);
 assert(m_ > 0);
 if (!(2 * BlockSize_ * sort_memory_usage_factor() <= memory_to_use)) {
 throw bad_parameter("stxxl::runs_creator<>:runs_creator(): INSUFFICIENT MEMORY provided, please increase parameter 'memory_to_use'");
 }
 }

 //! Adds new element to the current run.
 //! \param val value to be added to the current run
 void push(const value_type& val)
 {
 assert(offset < block_type::size);

 (*cur_block)[offset] = val;
 ++offset;

 if (offset == block_type::size)
 {
 // write current block

 block_manager* bm = block_manager::get_instance();
 // allocate space for the block
 result_->runs.resize(irun + 1);
 result_->runs[irun].resize(iblock + 1);
 bm->new_blocks(
 alloc_strategy,
 make_bid_iterator(result_->runs[irun].begin() + iblock),
 make_bid_iterator(result_->runs[irun].end()),
 iblock
 );

 result_->runs[irun][iblock].value = (*cur_block)[0]; // init trigger
 cur_block = writer.write(cur_block, result_->runs[irun][iblock].bid);
 ++iblock;

 offset = 0;
 }

 ++result_->elements;
 }

 //! Finishes current run and begins new one.
 void finish()
 {
 if (offset == 0 && iblock == 0) // current run is empty
 return;


 result_->runs_sizes.resize(irun + 1);
 result_->runs_sizes.back() = iblock * block_type::size + offset;

 if (offset) // if current block is partially filled
 {
 while (offset != block_type::size)
 {
 (*cur_block)[offset] = cmp.max_value();
 ++offset;
 }
 offset = 0;

 block_manager* bm = block_manager::get_instance();
 // allocate space for the block
 result_->runs.resize(irun + 1);
 result_->runs[irun].resize(iblock + 1);
 bm->new_blocks(
 alloc_strategy,
 make_bid_iterator(result_->runs[irun].begin() + iblock),
 make_bid_iterator(result_->runs[irun].end()),
 iblock
 );

 result_->runs[irun][iblock].value = (*cur_block)[0]; // init trigger
 cur_block = writer.write(cur_block, result_->runs[irun][iblock].bid);
 }
 else
 { }

 alloc_strategy = alloc_strategy_type(); // reinitialize block allocator for the next run
 iblock = 0;
 ++irun;
 }

 //! Returns the sorted runs object.
 //! \return Sorted runs object
 //! \remark Returned object is intended to be used by \c runs_merger object as input
 sorted_runs_type & result()
 {
 finish();
 writer.flush();

 return result_;
 }
};


//! Checker for the sorted runs object created by the \c runs_creator .
//! \param sruns sorted runs object
//! \param cmp comparison object used for checking the order of elements in runs
//! \return \c true if runs are sorted, \c false otherwise
template <class RunsType_, class CompareType_>
bool check_sorted_runs(const RunsType_& sruns, CompareType_ cmp)
{
 sort_helper::verify_sentinel_strict_weak_ordering(cmp);
 typedef typename RunsType_::element_type::block_type block_type;
 STXXL_VERBOSE2("Elements: " << sruns->elements);
 unsigned_type nruns = sruns->runs.size();
 STXXL_VERBOSE2("Runs: " << nruns);
 unsigned_type irun = 0;
 for (irun = 0; irun < nruns; ++irun)
 {
 const unsigned_type nblocks = sruns->runs[irun].size();
 block_type* blocks = new block_type[nblocks];
 request_ptr* reqs = new request_ptr[nblocks];
 for (unsigned_type j = 0; j < nblocks; ++j)
 {
 reqs[j] = blocks[j].read(sruns->runs[irun][j].bid);
 }
 wait_all(reqs, reqs + nblocks);
 delete[] reqs;

 for (unsigned_type j = 0; j < nblocks; ++j)
 {
 if (cmp(blocks[j][0], sruns->runs[irun][j].value) ||
 cmp(sruns->runs[irun][j].value, blocks[j][0])) //!=
 {
 STXXL_ERRMSG("check_sorted_runs wrong trigger in the run");
 delete[] blocks;
 return false;
 }
 }
 if (!stxxl::is_sorted(make_element_iterator(blocks, 0),
 make_element_iterator(blocks, sruns->runs_sizes[irun]),
 cmp))
 {
 STXXL_ERRMSG("check_sorted_runs wrong order in the run");
 delete[] blocks;
 return false;
 }

 delete[] blocks;
 }

 STXXL_MSG("Checking runs finished successfully");

 return true;
}


////////////////////////////////////////////////////////////////////////
// MERGE RUNS //
////////////////////////////////////////////////////////////////////////

//! Merges sorted runs.
//!
//! \tparam RunsType_ type of the sorted runs, available as \c runs_creator::sorted_runs_type ,
//! \tparam CompareType_ type of comparison object used for merging
//! \tparam AllocStr_ allocation strategy used to allocate the blocks for
//! storing intermediate results if several merge passes are required
template <class RunsType_,
 class CompareType_,
 class AllocStr_ = STXXL_DEFAULT_ALLOC_STRATEGY>
class basic_runs_merger : private noncopyable
{
public:
 typedef RunsType_ sorted_runs_type;
 typedef CompareType_ value_cmp;
 typedef AllocStr_ alloc_strategy;

 typedef typename sorted_runs_type::element_type sorted_runs_data_type;
 typedef typename sorted_runs_data_type::size_type size_type;
 typedef typename sorted_runs_data_type::run_type run_type;
 typedef typename sorted_runs_data_type::block_type block_type;
 typedef block_type out_block_type;
 typedef typename run_type::value_type trigger_entry_type;
 typedef block_prefetcher<block_type, typename run_type::iterator> prefetcher_type;
 typedef run_cursor2<block_type, prefetcher_type> run_cursor_type;
 typedef sort_helper::run_cursor2_cmp<block_type, prefetcher_type, value_cmp> run_cursor2_cmp_type;
 typedef loser_tree<run_cursor_type, run_cursor2_cmp_type> loser_tree_type;
 typedef stxxl::int64 diff_type;
 typedef std::pair<typename block_type::iterator, typename block_type::iterator> sequence;
 typedef typename std::vector<sequence>::size_type seqs_size_type;

public:
 //! Standard stream typedef.
 typedef typename sorted_runs_data_type::value_type value_type;

private:
 //! comparator object to sort runs
 value_cmp m_cmp;

 //! memory size in bytes to use
 unsigned_type m_memory_to_use;

 //! smart pointer to sorted_runs object
 sorted_runs_type m_sruns;

 //! items remaining in input
 size_type m_elements_remaining;

 //! memory buffer for merging from external streams
 out_block_type* m_buffer_block;

 //! pointer into current memory buffer: this is either m_buffer_block or the small_runs vector
 const value_type* m_current_ptr;

 //! pointer into current memory buffer: end after range of current values
 const value_type* m_current_end;

 //! sequence of block needed for merging
 run_type m_consume_seq;

 //! precalculated order of blocks in which they are prefetched
 int_type* m_prefetch_seq;

 //! prefetcher object
 prefetcher_type* m_prefetcher;

 //! loser tree used for native merging
 loser_tree_type* m_losers;

#if STXXL_PARALLEL_MULTIWAY_MERGE
 std::vector<sequence>* seqs;
 std::vector<block_type*>* buffers;
 diff_type num_currently_mergeable;
#endif

#if STXXL_CHECK_ORDER_IN_SORTS
 //! previous element to ensure the current output ordering
 value_type m_last_element;
#endif //STXXL_CHECK_ORDER_IN_SORTS

 ////////////////////////////////////////////////////////////////////

 void merge_recursively();

 void deallocate_prefetcher()
 {
 if (m_prefetcher)
 {
 delete m_losers;
#if STXXL_PARALLEL_MULTIWAY_MERGE
 delete seqs;
 delete buffers;
#endif
 delete m_prefetcher;
 delete[] m_prefetch_seq;
 m_prefetcher = NULL;
 }
 }

 void fill_buffer_block()
 {
 STXXL_VERBOSE1("fill_buffer_block");
 if (do_parallel_merge())
 {
#if STXXL_PARALLEL_MULTIWAY_MERGE
// begin of STL-style merging
 diff_type rest = out_block_type::size; // elements still to merge for this output block

 do // while rest > 0 and still elements available
 {
 if (num_currently_mergeable < rest)
 {
 if (!m_prefetcher || m_prefetcher->empty())
 {
 // anything remaining is already in memory
 num_currently_mergeable = m_elements_remaining;
 }
 else
 {
 num_currently_mergeable = sort_helper::count_elements_less_equal(
 *seqs, m_consume_seq[m_prefetcher->pos()].value, m_cmp);
 }
 }

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

 STXXL_VERBOSE1("before merge " << output_size);

 stxxl::parallel::multiway_merge((*seqs).begin(), (*seqs).end(), m_buffer_block->end() - rest, m_cmp, output_size);
 // sequence iterators are progressed appropriately

 rest -= output_size;
 num_currently_mergeable -= output_size;

 STXXL_VERBOSE1("after merge");

 sort_helper::refill_or_remove_empty_sequences(*seqs, *buffers, *m_prefetcher);
 } while (rest > 0 && (*seqs).size() > 0);

#if STXXL_CHECK_ORDER_IN_SORTS
 if (!stxxl::is_sorted(m_buffer_block->begin(), m_buffer_block->end(), cmp))
 {
 for (value_type* i = m_buffer_block->begin() + 1; i != m_buffer_block->end(); ++i)
 if (cmp(*i, *(i - 1)))
 {
 STXXL_VERBOSE1("Error at position " << (i - m_buffer_block->begin()));
 }
 assert(false);
 }
#endif //STXXL_CHECK_ORDER_IN_SORTS

// end of STL-style merging
#else
 STXXL_THROW_UNREACHABLE();
#endif //STXXL_PARALLEL_MULTIWAY_MERGE
 }
 else
 {
// begin of native merging procedure
 m_losers->multi_merge(m_buffer_block->elem, m_buffer_block->elem + STXXL_MIN<size_type>(out_block_type::size, m_elements_remaining));
// end of native merging procedure
 }
 STXXL_VERBOSE1("current block filled");

 m_current_ptr = m_buffer_block->elem;
 m_current_end = m_buffer_block->elem + STXXL_MIN<size_type>(out_block_type::size, m_elements_remaining);

 if (m_elements_remaining <= out_block_type::size)
 deallocate_prefetcher();
 }

public:
 //! Creates a runs merger object.
 //! \param c comparison object
 //! \param memory_to_use amount of memory available for the merger in bytes
 basic_runs_merger(value_cmp c, unsigned_type memory_to_use)
 : m_cmp(c),
 m_memory_to_use(memory_to_use),
 m_buffer_block(new out_block_type),
 m_prefetch_seq(NULL),
 m_prefetcher(NULL),
 m_losers(NULL)
#if STXXL_PARALLEL_MULTIWAY_MERGE
 , seqs(NULL),
 buffers(NULL),
 num_currently_mergeable(0)
#endif
#if STXXL_CHECK_ORDER_IN_SORTS
 , m_last_element(m_cmp.min_value())
#endif //STXXL_CHECK_ORDER_IN_SORTS
 {
 sort_helper::verify_sentinel_strict_weak_ordering(m_cmp);
 }

 //! Set memory amount to use for the merger in bytes.
 void set_memory_to_use(unsigned_type memory_to_use)
 {
 m_memory_to_use = memory_to_use;
 }

 //! Initialize the runs merger object with a new round of sorted_runs.
 void initialize(const sorted_runs_type& sruns)
 {
 m_sruns = sruns;
 m_elements_remaining = m_sruns->elements;

 if (empty())
 return;

 if (!m_sruns->small_run.empty())
 {
 // we have a small input <= B, that is kept in the main memory
 STXXL_VERBOSE1("basic_runs_merger: small input optimization, input length: " << m_elements_remaining);
 assert(m_elements_remaining == size_type(m_sruns->small_run.size()));

 m_current_ptr = &m_sruns->small_run[0];
 m_current_end = m_current_ptr + m_sruns->small_run.size();

 return;
 }

#if STXXL_CHECK_ORDER_IN_SORTS
 assert(check_sorted_runs(m_sruns, m_cmp));
#endif //STXXL_CHECK_ORDER_IN_SORTS

 // *** test whether recursive merging is necessary

 disk_queues::get_instance()->set_priority_op(request_queue::WRITE);

 int_type disks_number = config::get_instance()->disks_number();
 unsigned_type min_prefetch_buffers = 2 * disks_number;
 unsigned_type input_buffers = (m_memory_to_use > sizeof(out_block_type) ? m_memory_to_use - sizeof(out_block_type) : 0) / block_type::raw_size;
 unsigned_type nruns = m_sruns->runs.size();

 if (input_buffers < nruns + min_prefetch_buffers)
 {
 // can not merge runs in one pass. merge recursively:
 STXXL_WARNMSG_RECURSIVE_SORT("The implementation of sort requires more than one merge pass, therefore for a better");
 STXXL_WARNMSG_RECURSIVE_SORT("efficiency decrease block size of run storage (a parameter of the run_creator)");
 STXXL_WARNMSG_RECURSIVE_SORT("or increase the amount memory dedicated to the merger.");
 STXXL_WARNMSG_RECURSIVE_SORT("m=" << input_buffers << " nruns=" << nruns << " prefetch_blocks=" << min_prefetch_buffers);
 STXXL_WARNMSG_RECURSIVE_SORT("memory_to_use=" << m_memory_to_use << " bytes block_type::raw_size=" << block_type::raw_size << " bytes");

 // check whether we have enough memory to merge recursively
 unsigned_type recursive_merge_buffers = m_memory_to_use / block_type::raw_size;
 if (recursive_merge_buffers < 2 * min_prefetch_buffers + 1 + 2) {
 // recursive merge uses min_prefetch_buffers for input buffering and min_prefetch_buffers output buffering
 // as well as 1 current output block and at least 2 input blocks
 STXXL_ERRMSG("There are only m=" << recursive_merge_buffers << " blocks available for recursive merging, but "
 << min_prefetch_buffers << "+" << min_prefetch_buffers << "+1 are needed read-ahead/write-back/output, and");
 STXXL_ERRMSG("the merger requires memory to store at least two input blocks internally. Aborting.");
 throw bad_parameter("basic_runs_merger::sort(): INSUFFICIENT MEMORY provided, please increase parameter 'memory_to_use'");
 }

 merge_recursively();

 nruns = m_sruns->runs.size();
 }

 assert(nruns + min_prefetch_buffers <= input_buffers);

 // *** Allocate prefetcher and merge data structure

 deallocate_prefetcher();

 unsigned_type prefetch_seq_size = 0;
 for (unsigned_type i = 0; i < nruns; ++i)
 {
 prefetch_seq_size += m_sruns->runs[i].size();
 }

 m_consume_seq.resize(prefetch_seq_size);
 m_prefetch_seq = new int_type[prefetch_seq_size];

 typename run_type::iterator copy_start = m_consume_seq.begin();
 for (unsigned_type i = 0; i < nruns; ++i)
 {
 copy_start = std::copy(m_sruns->runs[i].begin(),
 m_sruns->runs[i].end(),
 copy_start);
 }

 std::stable_sort(m_consume_seq.begin(), m_consume_seq.end(),
 sort_helper::trigger_entry_cmp<trigger_entry_type, value_cmp>(m_cmp) _STXXL_SORT_TRIGGER_FORCE_SEQUENTIAL);

 const unsigned_type n_prefetch_buffers = STXXL_MAX(min_prefetch_buffers, input_buffers - nruns);

#if STXXL_SORT_OPTIMAL_PREFETCHING
 // heuristic
 const int_type n_opt_prefetch_buffers = min_prefetch_buffers + (3 * (n_prefetch_buffers - min_prefetch_buffers)) / 10;

 compute_prefetch_schedule(
 m_consume_seq,
 m_prefetch_seq,
 n_opt_prefetch_buffers,
 disks_number);
#else
 for (unsigned_type i = 0; i < prefetch_seq_size; ++i)
 m_prefetch_seq[i] = i;
#endif //STXXL_SORT_OPTIMAL_PREFETCHING

 m_prefetcher = new prefetcher_type(
 m_consume_seq.begin(),
 m_consume_seq.end(),
 m_prefetch_seq,
 STXXL_MIN(nruns + n_prefetch_buffers, prefetch_seq_size));

 if (do_parallel_merge())
 {
#if STXXL_PARALLEL_MULTIWAY_MERGE
// begin of STL-style merging
 seqs = new std::vector<sequence>(nruns);
 buffers = new std::vector<block_type*>(nruns);

 for (unsigned_type i = 0; i < nruns; ++i) //initialize sequences
 {
 (*buffers)[i] = m_prefetcher->pull_block(); //get first block of each run
 (*seqs)[i] = std::make_pair((*buffers)[i]->begin(), (*buffers)[i]->end()); //this memory location stays the same, only the data is exchanged
 }
// end of STL-style merging
#else
 STXXL_THROW_UNREACHABLE();
#endif //STXXL_PARALLEL_MULTIWAY_MERGE
 }
 else
 {
// begin of native merging procedure
 m_losers = new loser_tree_type(m_prefetcher, nruns, run_cursor2_cmp_type(m_cmp));
// end of native merging procedure
 }

 fill_buffer_block();
 }

 //! Deallocate temporary structures freeing memory prior to next initialize().
 void deallocate()
 {
 deallocate_prefetcher();
 m_sruns = NULL; // release reference on result object
 }

public:
 //! Standard stream method.
 bool empty() const
 {
 return (m_elements_remaining == 0);
 }

 //! Standard size method.
 size_type size() const
 {
 return m_elements_remaining;
 }

 //! Standard stream method.
 const value_type& operator * () const
 {
 assert(!empty());
 return *m_current_ptr;
 }

 //! Standard stream method.
 const value_type* operator -> () const
 {
 return &(operator * ());
 }

 //! Standard stream method.
 basic_runs_merger& operator ++ () // preincrement operator
 {
 assert(!empty());
 assert(m_current_ptr != m_current_end);

 --m_elements_remaining;
 ++m_current_ptr;

 if (LIKELY(m_current_ptr == m_current_end && !empty()))
 {
 fill_buffer_block();

#if STXXL_CHECK_ORDER_IN_SORTS
 assert(stxxl::is_sorted(m_buffer_block->elem, m_buffer_block->elem + STXXL_MIN<size_type>(m_elements_remaining, m_buffer_block->size), m_cmp));
#endif //STXXL_CHECK_ORDER_IN_SORTS
 }

#if STXXL_CHECK_ORDER_IN_SORTS
 if (!empty())
 {
 assert(!m_cmp(operator * (), m_last_element));
 m_last_element = operator * ();
 }
#endif //STXXL_CHECK_ORDER_IN_SORTS

 return *this;
 }

 //! Destructor.
 //! \remark Deallocates blocks of the input sorted runs object
 virtual ~basic_runs_merger()
 {
 deallocate_prefetcher();

 delete m_buffer_block;
 }
};


template <class RunsType_, class CompareType_, class AllocStr_>
void basic_runs_merger<RunsType_, CompareType_, AllocStr_>::merge_recursively()
{
 block_manager* bm = block_manager::get_instance();
 unsigned_type ndisks = config::get_instance()->disks_number();
 unsigned_type nwrite_buffers = 2 * ndisks;
 unsigned_type memory_for_write_buffers = nwrite_buffers * sizeof(block_type);

 // memory consumption of the recursive merger (uses block_type as out_block_type)
 unsigned_type recursive_merger_memory_prefetch_buffers = 2 * ndisks * sizeof(block_type);
 unsigned_type recursive_merger_memory_out_block = sizeof(block_type);
 unsigned_type memory_for_buffers = memory_for_write_buffers
 + recursive_merger_memory_prefetch_buffers
 + recursive_merger_memory_out_block;
 // maximum arity in the recursive merger
 unsigned_type max_arity = (m_memory_to_use > memory_for_buffers ? m_memory_to_use - memory_for_buffers : 0) / block_type::raw_size;

 unsigned_type nruns = m_sruns->runs.size();
 const unsigned_type merge_factor = optimal_merge_factor(nruns, max_arity);
 assert(merge_factor > 1);
 assert(merge_factor <= max_arity);

 while (nruns > max_arity)
 {
 unsigned_type new_nruns = div_ceil(nruns, merge_factor);
 STXXL_MSG("Starting new merge phase: nruns: " << nruns <<
 " opt_merge_factor: " << merge_factor <<
 " max_arity: " << max_arity << " new_nruns: " << new_nruns);

 // construct new sorted_runs data object which will be swapped into m_sruns

 sorted_runs_data_type new_runs;
 new_runs.runs.resize(new_nruns);
 new_runs.runs_sizes.resize(new_nruns);
 new_runs.elements = m_sruns->elements;

 // merge all runs from m_runs into news_runs

 unsigned_type runs_left = nruns;
 unsigned_type cur_out_run = 0;
 size_type elements_left = m_sruns->elements;

 while (runs_left > 0)
 {
 unsigned_type runs2merge = STXXL_MIN(runs_left, merge_factor);
 STXXL_MSG("Merging " << runs2merge << " runs");

 if (runs2merge > 1) // non-trivial merge
 {
 // count the number of elements in the run
 size_type elements_in_new_run = 0;
 for (unsigned_type i = nruns - runs_left; i < (nruns - runs_left + runs2merge); ++i)
 {
 elements_in_new_run += m_sruns->runs_sizes[i];
 }
 new_runs.runs_sizes[cur_out_run] = elements_in_new_run;

 // calculate blocks in run
 const unsigned_type blocks_in_new_run = div_ceil(elements_in_new_run, block_type::size);

 // allocate blocks for the new runs
 new_runs.runs[cur_out_run].resize(blocks_in_new_run);
 bm->new_blocks(alloc_strategy(), make_bid_iterator(new_runs.runs[cur_out_run].begin()), make_bid_iterator(new_runs.runs[cur_out_run].end()));

 // Construct temporary sorted_runs object as input into recursive merger.
 // This sorted_runs is copied a subset of the over-large set of runs, which
 // will be deallocated from external memory once the runs are merged.
 sorted_runs_type cur_runs = new sorted_runs_data_type;
 cur_runs->runs.resize(runs2merge);
 cur_runs->runs_sizes.resize(runs2merge);

 std::copy(m_sruns->runs.begin() + nruns - runs_left,
 m_sruns->runs.begin() + nruns - runs_left + runs2merge,
 cur_runs->runs.begin());
 std::copy(m_sruns->runs_sizes.begin() + nruns - runs_left,
 m_sruns->runs_sizes.begin() + nruns - runs_left + runs2merge,
 cur_runs->runs_sizes.begin());

 cur_runs->elements = elements_in_new_run;
 elements_left -= elements_in_new_run;

 // construct recursive merger

 basic_runs_merger<RunsType_, CompareType_, AllocStr_> merger(m_cmp, m_memory_to_use - memory_for_write_buffers);
 merger.initialize(cur_runs);

 { // make sure everything is being destroyed in right time
 buf_ostream<block_type, typename run_type::iterator> out(
 new_runs.runs[cur_out_run].begin(),
 nwrite_buffers);

 size_type cnt = 0;
 const size_type cnt_max = cur_runs->elements;

 while (cnt != cnt_max)
 {
 *out = *merger;
 if ((cnt % block_type::size) == 0) // have to write the trigger value
 new_runs.runs[cur_out_run][cnt / size_type(block_type::size)].value = *merger;

 ++cnt, ++out, ++merger;
 }
 assert(merger.empty());

 while (cnt % block_type::size)
 {
 *out = m_cmp.max_value();
 ++out, ++cnt;
 }
 }

 // deallocate merged runs by destroying cur_runs
 }
 else // runs2merge = 1 -> no merging needed
 {
 assert(cur_out_run + 1 == new_runs.runs.size());

 elements_left -= m_sruns->runs_sizes.back();

 // copy block identifiers into new sorted_runs object
 new_runs.runs.back() = m_sruns->runs.back();
 new_runs.runs_sizes.back() = m_sruns->runs_sizes.back();
 }

 runs_left -= runs2merge;
 ++cur_out_run;
 }

 assert(elements_left == 0);

 m_sruns->runs.clear(); // clear bid vector of m_sruns to skip deallocation of blocks in destructor

 std::swap(nruns, new_nruns);
 m_sruns->swap(new_runs); // replaces data in referenced counted object m_sruns
 } // end while (nruns > max_arity)
}


//! Merges sorted runs.
//!
//! \tparam RunsType_ type of the sorted runs, available as \c runs_creator::sorted_runs_type ,
//! \tparam CompareType_ type of comparison object used for merging
//! \tparam AllocStr_ allocation strategy used to allocate the blocks for
//! storing intermediate results if several merge passes are required
template <class RunsType_,
 class CompareType_ = typename RunsType_::element_type::cmp_type,
 class AllocStr_ = STXXL_DEFAULT_ALLOC_STRATEGY>
class runs_merger : public basic_runs_merger<RunsType_, CompareType_, AllocStr_>
{
protected:
 typedef basic_runs_merger<RunsType_, CompareType_, AllocStr_> base;

public:
 typedef RunsType_ sorted_runs_type;
 typedef typename base::value_cmp value_cmp;
 typedef typename base::value_cmp cmp_type;
 typedef typename base::block_type block_type;

public:
 //! Creates a runs merger object.
 //! \param sruns input sorted runs object
 //! \param cmp comparison object
 //! \param memory_to_use amount of memory available for the merger in bytes
 runs_merger(sorted_runs_type& sruns, value_cmp cmp, unsigned_type memory_to_use)
 : base(cmp, memory_to_use)
 {
 this->initialize(sruns);
 }

 //! Creates a runs merger object without initializing a round of sorted_runs.
 //! \param cmp comparison object
 //! \param memory_to_use amount of memory available for the merger in bytes
 runs_merger(value_cmp cmp, unsigned_type memory_to_use)
 : base(cmp, memory_to_use)
 { }
};


////////////////////////////////////////////////////////////////////////
// SORT //
////////////////////////////////////////////////////////////////////////

//! Produces sorted stream from input stream.
//!
//! \tparam Input_ type of the input stream
//! \tparam CompareType_ type of comparison object used for sorting the runs
//! \tparam BlockSize_ size of blocks used to store the runs
//! \tparam AllocStr_ functor that defines allocation strategy for the runs
//! \remark Implemented as the composition of \c runs_creator and \c runs_merger .
template <class Input_,
 class CompareType_,
 unsigned BlockSize_ = STXXL_DEFAULT_BLOCK_SIZE(typename Input_::value_type),
 class AllocStr_ = STXXL_DEFAULT_ALLOC_STRATEGY,
 class runs_creator_type = runs_creator<Input_, CompareType_, BlockSize_, AllocStr_> >
class sort : public noncopyable
{
 typedef typename runs_creator_type::sorted_runs_type sorted_runs_type;
 typedef runs_merger<sorted_runs_type, CompareType_, AllocStr_> runs_merger_type;

 runs_creator_type creator;
 runs_merger_type merger;

public:
 //! Standard stream typedef.
 typedef typename Input_::value_type value_type;

 //! Creates the object.
 //! \param in input stream
 //! \param c comparator object
 //! \param memory_to_use memory amount that is allowed to used by the sorter in bytes
 sort(Input_& in, CompareType_ c, unsigned_type memory_to_use) :
 creator(in, c, memory_to_use),
 merger(creator.result(), c, memory_to_use)
 {
 sort_helper::verify_sentinel_strict_weak_ordering(c);
 }

 //! Creates the object.
 //! \param in input stream
 //! \param c comparator object
 //! \param m_memory_to_userc memory amount that is allowed to used by the runs creator in bytes
 //! \param m_memory_to_use memory amount that is allowed to used by the merger in bytes
 sort(Input_& in, CompareType_ c, unsigned_type m_memory_to_userc, unsigned_type m_memory_to_use) :
 creator(in, c, m_memory_to_userc),
 merger(creator.result(), c, m_memory_to_use)
 {
 sort_helper::verify_sentinel_strict_weak_ordering(c);
 }


 //! Standard stream method.
 bool empty() const
 {
 return merger.empty();
 }

 //! Standard stream method.
 const value_type& operator * () const
 {
 assert(!empty());
 return *merger;
 }

 const value_type* operator -> () const
 {
 assert(!empty());
 return merger.operator -> ();
 }

 //! Standard stream method.
 sort& operator ++ ()
 {
 ++merger;
 return *this;
 }
};

//! Computes sorted runs type from value type and block size.
//!
//! \tparam ValueType_ type of values ins sorted runs
//! \tparam BlockSize_ size of blocks where sorted runs stored
template <
 class ValueType_,
 unsigned BlockSize_>
class compute_sorted_runs_type
{
 typedef ValueType_ value_type;
 typedef BID<BlockSize_> bid_type;
 typedef sort_helper::trigger_entry<bid_type, value_type> trigger_entry_type;

public:
 typedef sorted_runs<trigger_entry_type, std::less<value_type> > result;
};

//! \}

} // namespace stream

//! \addtogroup stlalgo
//! \{

//! Sorts range of any random access iterators externally.
//!
//! \param begin iterator pointing to the first element of the range
//! \param end iterator pointing to the last+1 element of the range
//! \param cmp comparison object
//! \param MemSize memory to use for sorting (in bytes)
//! \param AS allocation strategy
//!
//! The \c BlockSize template parameter defines the block size to use (in bytes)
//! \warning Slower than External Iterator Sort
template <unsigned BlockSize,
 class RandomAccessIterator,
 class CmpType,
 class AllocStr>
void sort(RandomAccessIterator begin,
 RandomAccessIterator end,
 CmpType cmp,
 unsigned_type MemSize,
 AllocStr AS)
{
 STXXL_UNUSED(AS);
#if STXXL_MSVC
 typedef typename streamify_traits<RandomAccessIterator>::stream_type InputType;
#else
 typedef __typeof__ (stream::streamify(begin, end)) InputType;
#endif // STXXL_MSVC
 InputType Input(begin, end);
 typedef stream::sort<InputType, CmpType, BlockSize, AllocStr> sorter_type;
 sorter_type Sort(Input, cmp, MemSize);
 stream::materialize(Sort, begin);
}

//! \}

STXXL_END_NAMESPACE

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