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

Merge sorting consists of three phases. More...

#include <tpie/pipelining/merge_sorter.h>

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

Public Member Functions
	merge_sorter (pred_t pred=pred_t())
void	set_parameters (stream_size_type runLength, memory_size_type fanout)
	Enable setting run length and fanout manually (for testing purposes). 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_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 (const T &item)
	Push item to merge sorter during phase 1. More...
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...
T	pull ()
	In phase 3, fetch next item in the final merge phase. 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...

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	maximumFanout = 250

Detailed Description

template<typename T, bool UseProgress, typename pred_t = std::less<T>>
class tpie::merge_sorter< T, UseProgress, pred_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 44 of file merge_sorter.h.

Member Function Documentation

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

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

inline

Initiate phase 1: Formation of input runs.

Definition at line 124 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 == stParameters, "Merge sorting already begun");
        if (!m_parametersSet) throw merge_sort_not_ready();
        log_debug() << "Start forming input runs" << std::endl;
        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>>

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

inline

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

Definition at line 197 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>>

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

inline

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

Definition at line 394 of file merge_sorter.h.

References tp_assert.

Referenced by tpie::merge_sorter< T, UseProgress, pred_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>>

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

inline

End phase 1.

Definition at line 153 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<T>::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<T> currentRun(array_view<T>(m_currentRunItems, 0, m_currentRunItemCount));
            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>>

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

inline

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

Definition at line 406 of file merge_sorter.h.

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

                    {
        tp_assert(m_state == stReport, "Wrong phase");
        if (m_reportInternal && m_itemsPulled < m_currentRunItemCount) {
            T el = m_currentRunItems[m_itemsPulled++];
            if (!can_pull()) m_currentRunItems.resize(0);
            return el;
        } else {
            if (m_evacuated) reinitialize_final_merger();
            return m_merger.pull();
        }
    }

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

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

inline

Push item to merge sorter during phase 1.

Definition at line 139 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] = item;
        ++m_currentRunItemCount;
        ++m_itemCount;
    }

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

void tpie::merge_sorter< T, UseProgress, pred_t >::set_available_memory ( memory_size_type  m )

inline

Calculate parameters from given memory amount.

Parameters

m	Memory available for phase 2, 3 and 4

Definition at line 80 of file merge_sorter.h.

                                                         {
        calculate_parameters(m, m, m);
    }

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

void tpie::merge_sorter< T, UseProgress, pred_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 90 of file merge_sorter.h.

                                                                                                    {
        calculate_parameters(m1, m2, m3);
    }

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

void tpie::merge_sorter< T, UseProgress, pred_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 591 of file merge_sorter.h.

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

                                       {
        if (!m_parametersSet)
            throw exception("Wrong state in set_items: parameters not set");
        if (m_state != stParameters)
            throw exception("Wrong state in set_items: state is not stParameters");

        if (n < p.runLength) {
            log_debug() << "Decreasing run length from " << p.runLength
                << " to " << n << std::endl;

            p.runLength = n;

            // Mirror the restriction from calculate_parameters.
            if (p.internalReportThreshold > p.runLength)
                p.internalReportThreshold = p.runLength;

            log_debug() << "New merge sort parameters\n";
            p.dump(log_debug());
            log_debug() << std::endl;
        }
    }

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

void tpie::merge_sorter< T, UseProgress, pred_t >::set_parameters ( stream_size_type  runLength, memory_size_type  fanout  )

inline

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

Definition at line 66 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 == stParameters, "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*sizeof(T) + file_stream<T>::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