tpie::merge_sorter< T, UseProgress, pred_t, store_t > Class Template Reference

Merge sorting consists of three phases. More...

#include <tpie/pipelining/merge_sorter.h>

Inherits tpie::merge_sorter_base.

Public Types
typedef std::shared_ptr< merge_sorter >	ptr
typedef progress_types< UseProgress >	Progress

Public Member Functions
	merge_sorter (pred_t pred=pred_t(), store_t store=store_t())
void	begin ()
	Initiate phase 1: Formation of input runs. More...
void	push (item_type &&item)
	Push item to merge sorter during phase 1. More...
void	push (const item_type &item)
void	end ()
	End phase 1. More...
void	calc (typename Progress::base &pi)
	Perform phase 2: Performing all merges in the merge tree except the last one. More...
void	evacuate ()
void	evacuate_before_merging ()
void	evacuate_before_reporting ()
void	reinitialize_final_merger ()
bool	can_pull ()
	In phase 3, return true if there are more items in the final merge phase. More...
item_type	pull ()
	In phase 3, fetch next item in the final merge phase. More...
memory_size_type	actual_memory_phase_3 ()
void	set_parameters (memory_size_type runLength, memory_size_type fanout)
	Enable setting run length and fanout manually (for testing purposes). More...
void	set_available_files (memory_size_type f)
	Calculate parameters from given amount of files. More...
void	set_available_files (memory_size_type f1, memory_size_type f2, memory_size_type f3)
	Calculate parameters from given amount of files. More...
void	set_available_memory (memory_size_type m)
	Calculate parameters from given memory amount. More...
void	set_available_memory (memory_size_type m1, memory_size_type m2, memory_size_type m3)
	Calculate parameters from given memory amount. More...
stream_size_type	item_count ()
memory_size_type	evacuated_memory_usage () const
void	set_items (stream_size_type n)
	Set upper bound on number of items pushed. More...
void	set_owner (tpie::pipelining::node *n)
void	set_phase_1_files (memory_size_type f1)
void	set_phase_2_files (memory_size_type f2)
void	set_phase_3_files (memory_size_type f3)
void	set_phase_1_memory (memory_size_type m1)
void	set_phase_2_memory (memory_size_type m2)
void	set_phase_3_memory (memory_size_type m3)
bool	is_calc_free () const
memory_size_type	minimum_memory_phase_1 () noexcept
memory_size_type	minimum_memory_phase_2 () noexcept
memory_size_type	minimum_memory_phase_3 () noexcept
memory_size_type	maximum_memory_phase_3 () noexcept
memory_size_type	phase_1_memory (const sort_parameters &params) noexcept
memory_size_type	phase_2_memory (const sort_parameters &params) noexcept
memory_size_type	phase_3_memory (const sort_parameters &params) noexcept
memory_size_type	calculate_fanout (memory_size_type availableMemory, memory_size_type availableFiles) noexcept
	calculate_parameters helper More...

Static Public Attributes
static const memory_size_type	defaultFiles = 253
static const memory_size_type	minimumFilesPhase1 = 1
static const memory_size_type	maximumFilesPhase1 = 1
static const memory_size_type	minimumFilesPhase2 = 5
static const memory_size_type	maximumFilesPhase2 = std::numeric_limits<memory_size_type>::max()
static const memory_size_type	minimumFilesPhase3 = 5
static const memory_size_type	maximumFilesPhase3 = std::numeric_limits<memory_size_type>::max()

Protected Types
enum	state_type { stNotStarted, stRunFormation, stMerge, stReport }

Protected Member Functions
void	calculate_parameters ()
	Calculate parameters from given memory amount. More...
void	check_not_started ()

Static Protected Member Functions
static stream_size_type	calculate_run_length (stream_size_type initialRunLength, memory_size_type fanout, memory_size_type mergeLevel)
	initialize_merger helper. More...

Protected Attributes
const linear_memory_usage	m_fanout_memory_usage
const memory_size_type	m_item_size
const memory_size_type	m_element_file_stream_memory_usage
std::unique_ptr< memory_bucket >	m_bucketPtr
memory_bucket_ref	m_bucket
array< temp_file >	m_runFiles
state_type	m_state
sort_parameters	p
bool	m_parametersSet
bits::run_positions	m_runPositions
stream_size_type	m_finishedRuns
memory_size_type	m_currentRunItemCount
bool	m_reportInternal
memory_size_type	m_itemsPulled
stream_size_type	m_itemCount
stream_size_type	m_maxItems
bool	m_evacuated
bool	m_finalMergeInitialized
memory_size_type	m_finalMergeLevel
memory_size_type	m_finalRunCount
memory_size_type	m_finalMergeSpecialRunNumber
tpie::pipelining::node *	m_owning_node

Detailed Description

template<typename T, bool UseProgress, typename pred_t = std::less<T>, typename store_t = default_store>
class tpie::merge_sorter< T, UseProgress, pred_t, store_t >

Merge sorting consists of three phases.

1. Sorting and forming runs
2. Merging runs
3. Final merge and report

If the number of elements received during phase 1 is less than the length of a single run, we are in "report internal" mode, meaning we do not write anything to disk. This causes phase 2 to be a no-op and phase 3 to be a simple array traversal.

Definition at line 396 of file merge_sorter.h.

Member Function Documentation

begin()

template<typename T , bool UseProgress, typename pred_t = std::less<T>, typename store_t = default_store>

void tpie::merge_sorter< T, UseProgress, pred_t, store_t >::begin ( )

inline

Initiate phase 1: Formation of input runs.

Definition at line 423 of file merge_sorter.h.

                 {
        tp_assert(m_state == stNotStarted, "Merge sorting already begun");
        if (!m_parametersSet) calculate_parameters();
        log_pipe_debug() << "Start forming input runs" << std::endl;
        m_currentRunItems = array<store_type>(0, allocator<store_type>(m_bucket));
        m_currentRunItems.resize((size_t)p.runLength);
        m_runFiles.resize(p.fanout*2);
        m_currentRunItemCount = 0;
        m_finishedRuns = 0;
        m_state = stRunFormation;
        m_itemCount = 0;
    }

References tpie::merge_sorter_base::calculate_parameters(), tpie::sort_parameters::fanout, tpie::log_pipe_debug(), tpie::array< T, Allocator >::resize(), tpie::sort_parameters::runLength, and tp_assert.

calc()

template<typename T , bool UseProgress, typename pred_t = std::less<T>, typename store_t = default_store>

void tpie::merge_sorter< T, UseProgress, pred_t, store_t >::calc ( typename Progress::base &  pi )

inline

Perform phase 2: Performing all merges in the merge tree except the last one.

Definition at line 511 of file merge_sorter.h.

                                          {
        tp_assert(m_state == stMerge, "Wrong phase");
        if (!m_reportInternal) {
            prepare_pull(pi);
        } else {
            pi.init(1);
            pi.step();
            pi.done();
        }
        m_state = stReport;
    }

References tpie::progress_indicator_base::done(), tpie::progress_indicator_base::init(), tpie::progress_indicator_base::step(), and tp_assert.

calculate_fanout()

memory_size_type tpie::merge_sorter_base::calculate_fanout ( memory_size_type  availableMemory, memory_size_type  availableFiles  )

noexceptinherited

calculate_parameters helper

calculate_parameters()

void tpie::merge_sorter_base::calculate_parameters ( )

protectedinherited

Calculate parameters from given memory amount.

Referenced by tpie::merge_sorter< T, UseProgress, pred_t, store_t >::begin().

calculate_run_length()

static stream_size_type tpie::merge_sorter_base::calculate_run_length ( stream_size_type  initialRunLength, memory_size_type  fanout, memory_size_type  mergeLevel  )

inlinestaticprotectedinherited

initialize_merger helper.

Definition at line 324 of file merge_sorter.h.

                                                                                                                                          {
        stream_size_type runLength = initialRunLength;
        for (memory_size_type i = 0; i < mergeLevel; ++i) {
            runLength *= fanout;
        }
        return runLength;
    }

can_pull()

template<typename T , bool UseProgress, typename pred_t = std::less<T>, typename store_t = default_store>

bool tpie::merge_sorter< T, UseProgress, pred_t, store_t >::can_pull ( )

inline

In phase 3, return true if there are more items in the final merge phase.

Definition at line 705 of file merge_sorter.h.

                    {
        tp_assert(m_state == stReport, "Wrong phase");
        if (m_reportInternal) return m_itemsPulled < m_currentRunItemCount;
        else {
            if (m_evacuated) reinitialize_final_merger();
            return m_merger.can_pull();
        }
    }

References tp_assert.

Referenced by tpie::merge_sorter< T, UseProgress, pred_t, store_t >::pull().

end()

template<typename T , bool UseProgress, typename pred_t = std::less<T>, typename store_t = default_store>

void tpie::merge_sorter< T, UseProgress, pred_t, store_t >::end ( )

inline

End phase 1.

Definition at line 464 of file merge_sorter.h.

               {
        tp_assert(m_state == stRunFormation, "Wrong phase");
        sort_current_run();
 
        if (m_itemCount == 0) {
            tp_assert(m_currentRunItemCount == 0, "m_itemCount == 0, but m_currentRunItemCount != 0");
            m_reportInternal = true;
            m_itemsPulled = 0;
            m_currentRunItems.resize(0);
            log_debug() << "Got no items. Internal reporting mode." << std::endl;
        } else if (m_finishedRuns == 0 && m_currentRunItems.size() <= p.internalReportThreshold) {
            // Our current buffer fits within the memory requirements of phase 2.
            m_reportInternal = true;
            m_itemsPulled = 0;
            log_debug() << "Got " << m_currentRunItemCount << " items. Internal reporting mode." << std::endl;
 
        } else if (m_finishedRuns == 0
                   && m_currentRunItemCount <= p.internalReportThreshold
                   && array<store_type>::memory_usage(m_currentRunItemCount) <= get_memory_manager().available()) {
            // Our current buffer does not fit within the memory requirements
            // of phase 2, but we have enough temporary memory to copy and
            // resize the buffer.
 
            array<store_type> currentRun(m_currentRunItemCount);
            for (size_t i=0; i < m_currentRunItemCount; ++i)
                currentRun[i] = std::move(m_currentRunItems[i]);
            m_currentRunItems.swap(currentRun);
 
            m_reportInternal = true;
            m_itemsPulled = 0;
            log_debug() << "Got " << m_currentRunItemCount << " items. Internal reporting mode "
                << "after resizing item buffer." << std::endl;
 
        } else {
            m_reportInternal = false;
            empty_current_run();
            m_currentRunItems.resize(0);
            log_debug() << "Got " << m_finishedRuns << " runs. External reporting mode." << std::endl;
        }
        m_state = stMerge;
    }

References tpie::get_memory_manager(), tpie::sort_parameters::internalReportThreshold, tpie::log_debug(), tpie::array< T, Allocator >::resize(), tpie::array< T, Allocator >::size(), tpie::array< T, Allocator >::swap(), and tp_assert.

pull()

template<typename T , bool UseProgress, typename pred_t = std::less<T>, typename store_t = default_store>

item_type tpie::merge_sorter< T, UseProgress, pred_t, store_t >::pull ( )

inline

In phase 3, fetch next item in the final merge phase.

Definition at line 717 of file merge_sorter.h.

                     {
        tp_assert(m_state == stReport, "Wrong phase");
        if (m_reportInternal && m_itemsPulled < m_currentRunItemCount) {
            store_type el = std::move(m_currentRunItems[m_itemsPulled++]);
            if (!can_pull()) m_currentRunItems.resize(0);
            return m_store.store_to_outer(std::move(el));
        } else {
            if (m_evacuated) reinitialize_final_merger();
            m_runPositions.close();
            return m_store.store_to_outer(m_merger.pull());
        }
    }

References tpie::merge_sorter< T, UseProgress, pred_t, store_t >::can_pull(), tpie::bits::run_positions::close(), tpie::array< T, Allocator >::resize(), and tp_assert.

push()

template<typename T , bool UseProgress, typename pred_t = std::less<T>, typename store_t = default_store>

void tpie::merge_sorter< T, UseProgress, pred_t, store_t >::push ( item_type &&  item )

inline

Push item to merge sorter during phase 1.

Definition at line 439 of file merge_sorter.h.

                                 {
        tp_assert(m_state == stRunFormation, "Wrong phase");
        if (m_currentRunItemCount >= p.runLength) {
            sort_current_run();
            empty_current_run();
        }
        m_currentRunItems[m_currentRunItemCount] = m_store.outer_to_store(std::move(item));
        ++m_currentRunItemCount;
        ++m_itemCount;
    }

References tpie::sort_parameters::runLength, and tp_assert.

set_available_files() [1/2]

void tpie::merge_sorter_base::set_available_files ( memory_size_type  f )

inlineinherited

Calculate parameters from given amount of files.

Parameters

f	Files available for phase 1, 2 and 3

Definition at line 163 of file merge_sorter.h.

                                                 {
        p.filesPhase1 = p.filesPhase2 = p.filesPhase3 = f;
        check_not_started();
    }

set_available_files() [2/2]

void tpie::merge_sorter_base::set_available_files ( memory_size_type  f1, memory_size_type  f2, memory_size_type  f3  )

inlineinherited

Calculate parameters from given amount of files.

Parameters

f1	Files available for phase 1
f2	Files available for phase 2
f3	Files available for phase 3

Definition at line 174 of file merge_sorter.h.

                                                                                            {
        p.filesPhase1 = f1;
        p.filesPhase2 = f2;
        p.filesPhase3 = f3;
        check_not_started();
    }

set_available_memory() [1/2]

void tpie::merge_sorter_base::set_available_memory ( memory_size_type  m )

inlineinherited

Calculate parameters from given memory amount.

Parameters

m	Memory available for phase 1, 2 and 3

Definition at line 185 of file merge_sorter.h.

                                                  {
        p.memoryPhase1 = p.memoryPhase2 = p.memoryPhase3 = m;
        check_not_started();
    }

set_available_memory() [2/2]

void tpie::merge_sorter_base::set_available_memory ( memory_size_type  m1, memory_size_type  m2, memory_size_type  m3  )

inlineinherited

Calculate parameters from given memory amount.

Parameters

m1	Memory available for phase 1
m2	Memory available for phase 2
m3	Memory available for phase 3

Definition at line 196 of file merge_sorter.h.

                                                                                             {
        p.memoryPhase1 = m1;
        p.memoryPhase2 = m2;
        p.memoryPhase3 = m3;
        check_not_started();
    }

set_items()

void tpie::merge_sorter_base::set_items ( stream_size_type  n )

inherited

Set upper bound on number of items pushed.

If the number of items to push is less than the size of a single run, this method will decrease the run size to that. This may make it easier for the sorter to go into internal reporting mode.

set_parameters()

void tpie::merge_sorter_base::set_parameters ( memory_size_type  runLength, memory_size_type  fanout  )

inherited

Enable setting run length and fanout manually (for testing purposes).

---

The documentation for this class was generated from the following file:

• pipelining/merge_sorter.h