/usr/include/stxxl/bits/stream
Edit: /usr/include/stxxl/bits/stream/sort_stream.h (57845B)
/***************************************************************************
* include/stxxl/bits/stream/sort_stream.h
*
* Part of the STXXL. See http://stxxl.sourceforge.net
*
* Copyright (C) 2002-2005 Roman Dementiev
* Copyright (C) 2006-2008 Johannes Singler
* Copyright (C) 2008-2010 Andreas Beckmann
* Copyright (C) 2013 Timo Bingmann
*
* 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
#include
#include
#include
#include
#include
#include
#include
#include
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 block_type;
typedef sort_helper::trigger_entry trigger_entry_type;
typedef sorted_runs sorted_runs_data_type;
typedef typename sorted_runs_data_type::run_type run_type;
typedef counting_ptr sorted_runs_type;
typedef typename element_iterator_traits::element_iterator element_iterator;
protected:
//! reference to the input stream
Input& m_input;
//! comparator used to sort block groups
CompareType m_cmp;
private:
//! stores the result (sorted runs) as smart pointer
sorted_runs_type m_result;
//! memory for internal use in blocks
unsigned_type m_memsize;
//! true iff result is already computed (used in 'result()' method)
bool m_result_computed;
//! Fetch data from input into blocks[first_idx,last_idx).
unsigned_type fetch(block_type* blocks,
unsigned_type first_idx, unsigned_type last_idx)
{
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) {
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
void basic_runs_creator::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
{
private:
typedef basic_runs_creator 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
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.
//! \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,
CompareType,
BlockSize,
AllocStr
>: private noncopyable
{
public:
typedef CompareType cmp_type;
typedef ValueType value_type;
typedef typed_block block_type;
typedef sort_helper::trigger_entry trigger_entry_type;
typedef sorted_runs sorted_runs_data_type;
typedef counting_ptr sorted_runs_type;
typedef sorted_runs_type result_type;
typedef typename element_iterator_traits::element_iterator element_iterator;
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
internal_size_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) {
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.
external_size_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
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.
//! \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,
CompareType,
BlockSize,
AllocStr
>: private noncopyable
{
public:
typedef ValueType value_type;
typedef typed_block block_type;
typedef sort_helper::trigger_entry trigger_entry_type;
typedef AllocStr alloc_strategy_type;
public:
typedef CompareType cmp_type;
typedef sorted_runs sorted_runs_data_type;
typedef counting_ptr sorted_runs_type;
typedef sorted_runs_type result_type;
private:
typedef typename sorted_runs_data_type::run_type run_type;
CompareType cmp;
//! stores the result (sorted runs)
sorted_runs_type result_;
//! memory for internal use in blocks
unsigned_type m_;
buffered_writer writer;
block_type* cur_block;
unsigned_type offset;
unsigned_type iblock;
unsigned_type irun;
//! needs to be reset after each run
alloc_strategy_type alloc_strategy;
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
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 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 prefetcher_type;
typedef run_cursor2 run_cursor_type;
typedef sort_helper::run_cursor2_cmp run_cursor2_cmp_type;
typedef loser_tree loser_tree_type;
typedef stxxl::int64 diff_type;
typedef std::pair sequence;
typedef typename std::vector::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* seqs;
std::vector* 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(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(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(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,
config::get_instance()->get_max_device_id());
#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(nruns);
buffers = new std::vector(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(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
void basic_runs_merger::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 = (unsigned_type)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
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 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][(unsigned_type)(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);
// clear bid vector of m_sruns to skip deallocation of blocks in
// destructor
m_sruns->runs.clear();
// replaces data in referenced counted object m_sruns end while (nruns
// > max_arity)
std::swap(nruns, new_nruns);
m_sruns->swap(new_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 runs_merger : public basic_runs_merger
{
protected:
typedef basic_runs_merger 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 RunsCreatorType = runs_creator
>
class sort : public noncopyable
{
typedef RunsCreatorType runs_creator_type;
typedef typename runs_creator_type::sorted_runs_type sorted_runs_type;
typedef runs_merger 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 bid_type;
typedef sort_helper::trigger_entry trigger_entry_type;
public:
typedef sorted_runs > 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);
typedef typename stream::streamify_traits::stream_type InputType;
InputType Input(begin, end);
typedef stream::sort 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