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>

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	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...
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)
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...
bool	is_calc_free () const
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...
stream_size_type	item_count ()
memory_size_type	actual_memory_phase_3 ()
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)

Static Public Member Functions
static memory_size_type	memory_usage_phase_1 (const sort_parameters &params)
static memory_size_type	minimum_memory_phase_1 ()
static memory_size_type	memory_usage_phase_2 (const sort_parameters &params)
static memory_size_type	minimum_memory_phase_2 ()
static memory_size_type	memory_usage_phase_3 (const sort_parameters &params)
static memory_size_type	minimum_memory_phase_3 ()
static memory_size_type	maximum_memory_phase_3 ()

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()

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 150 of file merge_sorter.h.

Member Function Documentation

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 288 of file merge_sorter.h.

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

                        {
        tp_assert(m_state == stNotStarted, "Merge sorting already begun");
        if (!m_parametersSet) calculate_parameters();
        log_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;
    }

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 380 of file merge_sorter.h.

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

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

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 585 of file merge_sorter.h.

References tp_assert.

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

                           {
        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();
        }
    }

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 329 of file merge_sorter.h.

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.

                      {
        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;
    }

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 597 of file merge_sorter.h.

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.

                            {
        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());
        }
    }

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 304 of file merge_sorter.h.

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

                                        {
        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;
    }

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 >::set_available_files ( memory_size_type  f )

inline

Calculate parameters from given amount of files.

Parameters

f	Files available for phase 1, 2 and 3

Definition at line 206 of file merge_sorter.h.

References tpie::sort_parameters::filesPhase1, tpie::sort_parameters::filesPhase2, and tpie::sort_parameters::filesPhase3.

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

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 >::set_available_files ( memory_size_type  f1, memory_size_type  f2, memory_size_type  f3  )

inline

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 217 of file merge_sorter.h.

References tpie::sort_parameters::filesPhase1, tpie::sort_parameters::filesPhase2, and tpie::sort_parameters::filesPhase3.

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

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 >::set_available_memory ( memory_size_type  m )

inline

Calculate parameters from given memory amount.

Parameters

m	Memory available for phase 1, 2 and 3

Definition at line 228 of file merge_sorter.h.

References tpie::sort_parameters::memoryPhase1, tpie::sort_parameters::memoryPhase2, and tpie::sort_parameters::memoryPhase3.

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

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 >::set_available_memory ( memory_size_type  m1, memory_size_type  m2, memory_size_type  m3  )

inline

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 239 of file merge_sorter.h.

References tpie::sort_parameters::memoryPhase1, tpie::sort_parameters::memoryPhase2, and tpie::sort_parameters::memoryPhase3.

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

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 >::set_items ( stream_size_type  n )

inline

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.

Definition at line 814 of file merge_sorter.h.

References tpie::sort_parameters::internalReportThreshold, tpie::log_debug(), and tpie::sort_parameters::runLength.

                                       {
        if (m_state != stNotStarted)
            throw exception("Wrong state in set_items: state is not stNotStarted");

        m_maxItems = n;

        if (!m_parametersSet) {
            // We will handle this later in calculate_parameters
            return;
        }

        // If the item upper bound is less than a run,
        // then it might pay off to decrease the length of a run
        // so that we can avoid I/O altogether.
        if (m_maxItems < p.runLength) {
            memory_size_type newRunLength =
                std::max(memory_size_type(m_maxItems), p.internalReportThreshold);
            log_debug() << "Decreasing run length from " << p.runLength
                << " to " << newRunLength
                << " since at most " << m_maxItems << " items will be pushed,"
                << " and the internal report threshold is "
                << p.internalReportThreshold
                << ". New merge sort parameters:\n";
            // In principle, we could decrease runLength to m_maxItems,
            // but setting runLength below internalReportThreshold does not
            // give additional benefits.
            // Furthermore, buggy code could call set_items with a very low
            // upper bound, leading to unacceptable performance in practice;
            // thus, internalReportThreshold is used as a stopgap/failsafe.
            p.runLength = newRunLength;
            p.dump(log_debug());
            log_debug() << std::endl;
        }
    }

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 >::set_parameters ( memory_size_type  runLength, memory_size_type  fanout  )

inline

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

Definition at line 192 of file merge_sorter.h.

References tpie::sort_parameters::fanout, tpie::sort_parameters::finalFanout, tpie::sort_parameters::internalReportThreshold, tpie::log_debug(), tpie::sort_parameters::runLength, and tp_assert.

                                                                                    {
        tp_assert(m_state == stNotStarted, "Merge sorting already begun");
        p.runLength = p.internalReportThreshold = runLength;
        p.fanout = p.finalFanout = fanout;
        m_parametersSet = true;
        log_debug() << "Manually set merge sort run length and fanout\n";
        log_debug() << "Run length =       " << p.runLength << " (uses memory " << (p.runLength*item_size + file_stream<element_type>::memory_usage()) << ")\n";
        log_debug() << "Fanout =           " << p.fanout << " (uses memory " << fanout_memory_usage(p.fanout) << ")" << std::endl;
    }

---

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

• pipelining/merge_sorter.h