tpie::ami Namespace Reference

The namespace within TPIE for the Access Method Interface elements. More...

Classes
class	heap_element
	This is a heap element. More...
class	heap_ptr
	This is a record pointer element. More...
class	Internal_Sorter_Base
	The base class for internal sorters. More...
class	Internal_Sorter_Obj
	Comparision object based Internal_Sorter_base subclass implementation; uses quick_sort_obj(). More...
class	merge_base
	Superclass for merge management objects. More...
class	merge_heap_obj
class	merge_heap_op
	A merge heap object base class - also serves as the full implementation for objects with a < comparison operator. More...
class	merge_heap_ptr_obj
	A record pointer heap that uses a comparison object. More...
class	merge_heap_ptr_op
	A record pointer heap base class - also serves as the full implementation for objects with a < comparison operator. More...
class	qsort_item
	A simple class that facilitates doing key sorting followed by in-memory permuting to sort items in-memory. More...
class	stack
	An implementation of an external-memory stack compatible with the old AMI interface. More...
class	stream
	A Stream<T> object stores an ordered collection of objects of type T on external memory. More...

Typedefs
typedef TPIE_OS_SIZE_T	arity_t
	Intended to signal the number of input streams in a merge. More...

Enumerations
enum	err {   NO_ERROR = 0, IO_ERROR, END_OF_STREAM, READ_ONLY,   OS_ERROR, BASE_METHOD, BTE_ERROR, MM_ERROR,   OBJECT_INITIALIZATION, OBJECT_INVALID, PERMISSION_DENIED, INSUFFICIENT_MAIN_MEMORY,   INSUFFICIENT_AVAILABLE_STREAMS, ENV_UNDEFINED, NO_MAIN_MEMORY_OPERATION, BIT_MATRIX_BOUNDS,   NOT_POWER_OF_2, NULL_POINTER, GENERIC_ERROR = 0xfff, SCAN_DONE = 0x1000,   SCAN_CONTINUE, MERGE_DONE = 0x2000, MERGE_CONTINUE, MERGE_OUTPUT,   MERGE_READ_MULTIPLE, MATRIX_BOUNDS = 0x3000, SORT_ALREADY_SORTED = 0x4000 }
	Legacy TPIE error codes. More...
enum	merge_output_type { MERGE_OUTPUT_OVERWRITE = 1, MERGE_OUTPUT_APPEND }
enum	stream_type { READ_STREAM = 1, WRITE_STREAM, APPEND_STREAM, READ_WRITE_STREAM }
	AMI stream types passed to constructors. More...
enum	stream_status { STREAM_STATUS_VALID = 0, STREAM_STATUS_INVALID }
	AMI stream status. More...

Functions
template<class T , class M >
err	merge (stream< T > **instreams, arity_t arity, stream< T > *outstream, M *m_obj)
	Merges arity streams using a merge management object and writes result into outstream. More...
template<class T , class M >
err	partition_and_merge (stream< T > *instream, stream< T > *outstream, M *m_obj)
	Partitions a stream into substreams small enough to fit. More...
template<class T , class M >
err	single_merge (stream< T > **instreams, arity_t arity, stream< T > *outstream, M *m_obj)
	Merges arity streams in memory using a merge management object and write result into outstream. More...
template<class T , class M >
err	main_mem_merge (stream< T > *instream, stream< T > *outstream, M *m_obj)
	Reads instream in memory and merges it using m_obj->main_mem_operate(); if instream does not fit in main memory returns INSUFFICIENT_MAIN_MEMORY;. More...
template<class T , class M >
void	merge_sorted_runs (typename tpie::array< tpie::auto_ptr< file_stream< T > > >::iterator start, typename tpie::array< tpie::auto_ptr< file_stream< T > > >::iterator end, file_stream< T > *outStream, M *MergeHeap, TPIE_OS_OFFSET cutoff=-1, progress_indicator_base *indicator=NULL)
	This is a common merge routine for all of the AMI_merge_sorted, AMI_ptr_merge_sorted and AMI_key_merge_sorted entry points. More...
template<class T , class CMPR >
void	merge_sorted (typename tpie::array< std::auto_ptr< file_stream< T > > >::iterator start, typename tpie::array< std::auto_ptr< file_stream< T > > >::iterator end, file_stream< T > *outStream, CMPR *cmp)
	Merging with a heap that contains the records to be merged. More...
template<class T , class CMPR >
void	ptr_merge_sorted (typename tpie::array< tpie::auto_ptr< file_stream< T > > >::iterator start, typename tpie::array< tpie::auto_ptr< file_stream< T > > >::iterator end, file_stream< T > *outStream, CMPR *cmp)
	Merging with a heap that keeps a pointer to the records rather than the records themselves: CMPR is the class of the comparison object, and must contain the method compare() which is called from within the merge. More...
	TPIE_DEPRECATED_CLASS_A (template< typename T > class TPIE_DEPRECATED_CLASS_B queue:public tpie::queue< T >{public:queue(){}queue(const std::string &basename):tpie::queue< T >(basename){}})
err	exception_kind (const exception &e)
template<class T >
err	sort (stream< T > *instream_ami, stream< T > *outstream_ami, tpie::progress_indicator_base *indicator=NULL)
template<class T , class CMPR >
err	sort (stream< T > *instream_ami, stream< T > *outstream_ami, CMPR *cmp, progress_indicator_base *indicator=NULL)
template<class T >
err	ptr_sort (stream< T > *instream_ami, stream< T > *outstream_ami, progress_indicator_base *indicator=NULL)
template<class T , class CMPR >
err	ptr_sort (stream< T > *instream_ami, stream< T > *outstream_ami, CMPR *cmp, progress_indicator_base *indicator=NULL)
template<class T , class KEY , class CMPR >
err	key_sort (stream< T > *instream_ami, stream< T > *outstream_ami, KEY, CMPR *cmp, progress_indicator_base *indicator=NULL)
template<class T >
err	sort (stream< T > *instream_ami, progress_indicator_base *indicator=0)
template<class T , class CMPR >
err	sort (stream< T > *instream_ami, CMPR *cmp, progress_indicator_base *indicator=NULL)
template<class T >
err	ptr_sort (stream< T > *instream_ami, progress_indicator_base *indicator=NULL)
template<class T , class CMPR >
err	ptr_sort (stream< T > *instream_ami, CMPR *cmp, progress_indicator_base *indicator=NULL)
template<class T , class KEY , class CMPR >
err	key_sort (stream< T > *instream_ami, KEY, CMPR *cmp, progress_indicator_base *indicator=NULL)

Detailed Description

The namespace within TPIE for the Access Method Interface elements.

In-place sorting variant of key_sort(stream<T> instream_ami, stream<T> *outstream_ami, KEY dummykey, CMPR *cmp, progress_indicator_base indicator=NULL), see also In-place Variants for Sorting in TPIE.

In-place sorting variant of ptr_sort(stream<T> instream_ami, stream<T> *outstream_ami, CMPR *cmp, progress_indicator_base indicator=NULL), see also In-place Variants for Sorting in TPIE.

In-place sorting variant of ptr_sort(stream<T> instream_ami, stream<T> *outstream_ami, progress_indicator_base indicator=NULL), see also In-place Variants for Sorting in TPIE.

In-place sorting variant of sort(stream<T> instream_ami, stream<T> *outstream_ami, CMPR *cmp, progress_indicator_base indicator=NULL), see also In-place Variants for Sorting in TPIE.

In-place sorting variant of sort(stream<T> instream_ami, stream<T> *outstream_ami, tpie::progress_indicator_base indicator=NULL), see also In-place Variants for Sorting in TPIE.

A version of sort that takes an input stream of elements of type T, an output stream, and a user-specified comparison object.

A version of sort that takes an input stream of elements of type T, and an output stream, and and uses the < operator to sort, see also Sorting in TPIE.

Comparing within Sorting:

TPIE's sort() has two polymorphs, namely the comparison operator and comparison class polymorphs. The comparison operator version tends to be the fastest and most straightforward to use. The comparison class version is comparable in speed (maybe slightly slower), but somewhat more flexible, as it can support multiple, different sorts on the same keys.

Comparison operator version:

This version works on streams of objects for which the operator "<" is defined.

Comparison class version:

This version of sort() uses a method of a user-defined comparison object to determine the order of two input objects. The object must have a public member function named compare(), having the following prototype:

inline int compare (const KEY & k1, const KEY & k2);

The user-written compare() function computes the order of the two user-defined keys k1 and k2, and returns -1, 0, or +1 to indicate that k1<k2, k1==k2, or k1>k2 respectively.

The comparison object "cmp", of (user-defined) class represented by CMPR, must have a member function called "compare" which is used for sorting the input stream; see also Comparing within Sorting.

Typedef Documentation

typedef TPIE_OS_SIZE_T tpie::ami::arity_t

Intended to signal the number of input streams in a merge.

Definition at line 44 of file merge_sorted_runs.h.

Enumeration Type Documentation

enum tpie::ami::err

Legacy TPIE error codes.

Functions in the AMI interface of TPIE typically return error codes of the enumerated type err.

Enumerator
NO_ERROR	No error occurred. The call the the entry point returned normally.
IO_ERROR	A low level I/O error occurred.
END_OF_STREAM	An attempt was made to read past the end of a stream or write past the end of a substream.
READ_ONLY	An attempt was made to write to a read-only stream.
OS_ERROR	An unexpected operating system error occurred. Details should appear in the log file if logging is enabled. See also sec_logging
BASE_METHOD	An attempt was made to call a member function of the virtual base class of tpie::ami::stream. This indicates a bug in the implementation of TPIE streams.
BTE_ERROR	An error occurred at the BTE level.
MM_ERROR	An error occurred within the memory manager.
OBJECT_INITIALIZATION	An TPIE entry point was not able to properly initialize the operation management object that was passed to it. This generally indicates a bug in the operation management object's initialization code.
OBJECT_INVALID	A passed object is invalid.
PERMISSION_DENIED	A passed object is inaccessible due to insufficient permissions.
INSUFFICIENT_MAIN_MEMORY	The memory manager could not make adequate main memory available to complete the requested operation. Many operations adapt themselves to use whatever main memory is available, but in some cases, when memory is extremely tight, they may not be able to function.
INSUFFICIENT_AVAILABLE_STREAMS	TPIE could not allocate enough intermediate streams to perform the requested operation. Certain operating system restrictions limit the number of streams that can be created on certain platforms. Only in unusual circumstances, such as when the application itself has a very large number of open streams, will this error occur.
ENV_UNDEFINED	An environment variable necessary to initialize the TPIE accessing environment was not defined.
BIT_MATRIX_BOUNDS	A bit matrix larger than the number of bits in an offset into a stream was passed.
NOT_POWER_OF_2	The length of a stream on which a bit permutation was to be performed is not a power of two.
NULL_POINTER	An attempt was made to perform a matrix operation on matrices whose bounds did not match appropriately.
SCAN_DONE	Value returned by a scan_object: Indicates that the function should be called again with any "taken" inputs replaced by the next objects from their respective streams.
SCAN_CONTINUE	Value returned by a scan_object: Indicates that the scan is complete and no more input needs to be processed.
MERGE_DONE	Value returned by a merge_management_object, signaling that the merge() completed.
MERGE_CONTINUE	Value returned by a merge_management_object, telling merge() to continue to call the operate() member function of the management object with more data.
MERGE_OUTPUT	Value returned by a merge_management_object, signaling that the last merge() call generated output for the output stream.
MERGE_READ_MULTIPLE	Value returned by a merge_management_object, telling merge() that more than one input ob ject was consumed and thus the input flags should be consulted.
MATRIX_BOUNDS	Matrix related error.
SORT_ALREADY_SORTED	Values returned by sort routines if input does not require sorting.

Definition at line 45 of file err.h.

             {
      NO_ERROR = 0,
      IO_ERROR,
      END_OF_STREAM,
      READ_ONLY,
      OS_ERROR,
        BASE_METHOD,
        BTE_ERROR,
        MM_ERROR,
        OBJECT_INITIALIZATION,
        OBJECT_INVALID,
        PERMISSION_DENIED,
      INSUFFICIENT_MAIN_MEMORY,
        INSUFFICIENT_AVAILABLE_STREAMS,
        ENV_UNDEFINED,
        NO_MAIN_MEMORY_OPERATION,
        BIT_MATRIX_BOUNDS,
        NOT_POWER_OF_2,
        NULL_POINTER,

        GENERIC_ERROR = 0xfff,

        SCAN_DONE = 0x1000,
        SCAN_CONTINUE,

        MERGE_DONE = 0x2000,
       MERGE_CONTINUE,
        MERGE_OUTPUT,
        MERGE_READ_MULTIPLE,

        MATRIX_BOUNDS = 0x3000,

        SORT_ALREADY_SORTED = 0x4000
  
    };

enum tpie::ami::merge_output_type

Definition at line 45 of file merge.h.

                           {
        MERGE_OUTPUT_OVERWRITE = 1,
        MERGE_OUTPUT_APPEND
    };

enum tpie::ami::stream_status

AMI stream status.

Enumerator
STREAM_STATUS_VALID	Stream is valid.
STREAM_STATUS_INVALID	Stream is invalid.

Definition at line 60 of file stream.h.

                       {
        STREAM_STATUS_VALID = 0,
        STREAM_STATUS_INVALID
    };

enum tpie::ami::stream_type

AMI stream types passed to constructors.

Definition at line 52 of file stream.h.

                     {
        READ_STREAM = 1,    // Open existing stream for reading
        WRITE_STREAM,       // Open for writing.  Create if non-existent
        APPEND_STREAM,      // Open for writing at end.  Create if needed.
        READ_WRITE_STREAM   // Open to read and write.
    };

Function Documentation

template<class T , class M >

err tpie::ami::main_mem_merge	(	stream< T > *	instream,
		stream< T > *	outstream,
		M *	m_obj
	)

Reads instream in memory and merges it using m_obj->main_mem_operate(); if instream does not fit in main memory returns INSUFFICIENT_MAIN_MEMORY;.

Definition at line 442 of file merge.h.

References tpie::consecutive_memory_available(), INSUFFICIENT_MAIN_MEMORY, NO_ERROR, tpie::ami::stream< T >::read_array(), tpie::ami::stream< T >::seek(), tpie::ami::stream< T >::stream_len(), tp_assert, tpie::tpie_delete_array(), and tpie::ami::stream< T >::write_array().

Referenced by partition_and_merge().

                                                {

        err ae;
        TPIE_OS_OFFSET len;
        TPIE_OS_SIZE_T sz_avail;
  
        // How much memory is available?
        sz_avail = consecutive_memory_available ();

        len = instream->stream_len();
        if ((len * static_cast<TPIE_OS_OFFSET>(sizeof(T))) <= static_cast<TPIE_OS_OFFSET>(sz_avail)) {
    
        // If the whole input can fit in main memory just call
        // m_obj->main_mem_operate
    
        ae = instream->seek(0);
        assert(ae == NO_ERROR);
    
        // This code is sloppy and has to be rewritten correctly for
        // parallel buffer allocation.  It will not work with anything
        // other than a registration based memory manager.
        T *mm_stream;
        TPIE_OS_OFFSET len1;
        //allocate and read input stream in memory we know it fits, so we may cast.
        mm_stream = tpie_new_array<T>(static_cast<TPIE_OS_SIZE_T>(len));

        len1 = len;
        if ((ae = instream->read_array(mm_stream, &len1)) !=
            NO_ERROR) {
            return ae;
        }
        tp_assert(len1 == len, "Did not read the right amount; "
              "Allocated space for " << len << ", read " << len1 << '.');
    
        //just call m_obj->main_mem_operate. We know that len items fit into
        //main memory, so we may cast to TPIE_OS_SIZE_T
        if ((ae = m_obj->main_mem_operate(mm_stream, 
                          static_cast<TPIE_OS_SIZE_T>(len))) !=
            NO_ERROR) {
            TP_LOG_WARNING_ID("main_mem_operate failed");
            return ae;
        }

        //write array back to stream
        if ((ae = outstream->write_array(mm_stream, 
                         static_cast<TPIE_OS_SIZE_T>(len))) !=
            NO_ERROR) {
            TP_LOG_WARNING_ID("write array failed");
            return ae;
        }

        tpie_delete_array(mm_stream, static_cast<TPIE_OS_SIZE_T>(len));
        return NO_ERROR;
    
        } else {
    
        // Something went wrong.  We should not have called this
        // function, since we don't have enough main memory.
        TP_LOG_WARNING_ID("out of memory");
        return INSUFFICIENT_MAIN_MEMORY;
        }
    };

template<class T , class M >

err tpie::ami::merge	(	stream< T > **	instreams,
		arity_t	arity,
		stream< T > *	outstream,
		M *	m_obj
	)

Merges arity streams using a merge management object and writes result into outstream.

It is assumed that the available memory can fit the arity streams, the output stream and also the space required by the merge management object; merge() checks this and then calls single_merge();

Definition at line 253 of file merge.h.

References tpie::consecutive_memory_available(), INSUFFICIENT_MAIN_MEMORY, tpie::ami::stream< T >::main_memory_usage(), single_merge(), tpie::STREAM_USAGE_CURRENT, and tpie::STREAM_USAGE_MAXIMUM.

                                          {

        TPIE_OS_SIZE_T sz_avail;
        TPIE_OS_OFFSET sz_stream, sz_needed = 0;
  
        // How much main memory is available?
        sz_avail = consecutive_memory_available ();

        // Iterate through the streams, finding out how much additional
        // memory each stream will need in the worst case (the streams are
        // in memory, but their memory usage could be smaller then the
        // maximum one; one scenario is when the streams have been loaded
        // from disk with no subsequent read_item/write_item operation, in
        // which case their current memory usage is just the header block);
        // count also the output stream
        for (unsigned int ii = 0; ii < arity + 1; ii++) {
            instreams[ii]->main_memory_usage(&sz_stream, STREAM_USAGE_MAXIMUM);
            sz_needed += sz_stream;
            instreams[ii]->main_memory_usage(&sz_stream, STREAM_USAGE_CURRENT);
            sz_needed -= sz_stream;
        }                              
  
        //count the space used by the merge_management object (include
        //overhead added to a stream)
        sz_needed += m_obj->space_usage_overhead() + 
        arity * m_obj->space_usage_per_stream();
               
        //streams and m_obj must fit in memory!
        if (sz_needed >= static_cast<TPIE_OS_OFFSET>(sz_avail)) {
        TP_LOG_WARNING("Insufficient main memory to perform a merge.\n");
        return INSUFFICIENT_MAIN_MEMORY;
        }
        assert(sz_needed < sz_avail);
  
        //merge streams in memory
        return single_merge(instreams, arity, outstream, m_obj);
    };

template<class T , class CMPR >

void tpie::ami::merge_sorted	(	typename tpie::array< std::auto_ptr< file_stream< T > > >::iterator	start,
		typename tpie::array< std::auto_ptr< file_stream< T > > >::iterator	end,
		file_stream< T > *	outStream,
		CMPR *	cmp
	)

Merging with a heap that contains the records to be merged.

CMPR is the class of the comparison object, and must contain the method compare() which is called from within the merge.

This is one of the merge entry points for merging without the merge_management_object used by TPIE's merge. These routines perform the special case of merging when the the required output is the original records interleaved according to a comparison operator or function.

Definition at line 149 of file merge_sorted_runs.h.

References tpie::ami::merge_heap_op< REC, Compare >::allocate(), and merge_sorted_runs().

                                                               {
        
            // make a merge heap which uses the user's comparison object
            // and initialize it
            merge_heap_obj<T,CMPR> mrgheap (cmp);
            mrgheap.allocate(end-start);
        
            //Rewind all the input streams
            for (typename tpie::array<std::auto_ptr<file_stream<T> > >::iterator i=start; 
                 i != end; ++i)
                i->seek(0); 
        
            merge_sorted_runs(start, end, outStream, mrgheap);
        }

template<class T , class M >

void tpie::ami::merge_sorted_runs	(	typename tpie::array< tpie::auto_ptr< file_stream< T > > >::iterator	start,
		typename tpie::array< tpie::auto_ptr< file_stream< T > > >::iterator	end,
		file_stream< T > *	outStream,
		M *	MergeHeap,
		TPIE_OS_OFFSET	cutoff = -1,
		progress_indicator_base *	indicator = NULL
	)

This is a common merge routine for all of the AMI_merge_sorted, AMI_ptr_merge_sorted and AMI_key_merge_sorted entry points.

It is also used by the sort entry points AMI_sort, AMI_ptr_sort and AMI_key_sort and by the routine AMI_partition_and_merge. Differences are encapsulated within the merge heap object MergeHeap. It is assumed that MergeHeap::allocate() was called before entering ami_merge_sorted_runs. ami_merge_sorted_runs takes both the max number of elements to read from any stream and also a boolean flag for showing a progress indicator.

Sorted Stream Merging:

TPIE provides several merge entry points for merging sorted streams to produce a single, interleaved output stream. merge_sorted_runs has three polymorphs, namely the comparison operator, comparison class and the key-based polymorphs. The comparison operator version tends to be the fastest and most straightforward to use. The comparison class version is comparable in speed (maybe slightly slower), but somewhat more flexible, as it can support multiple, different merges on the same keys.

Definition at line 71 of file merge_sorted_runs.h.

References tpie::stream_crtp< child_t >::can_read(), tpie::file_stream< T >::read(), and tpie::file_stream< T >::write().

Referenced by merge_sorted(), and ptr_merge_sorted().

                                                                          {
        
            size_t i;
            size_t arity = end-start;

            //Pointers to current leading elements of streams
            tpie::array<const T *> in_objects(arity);
            tpie::array<TPIE_OS_OFFSET> nread(arity);
            
            // **************************************************************
            // * Read first element from stream. Do not rewind! We may read *
            // * more elements from the same stream later.                  *
            // **************************************************************

            for (i = 0; i < arity; i++) {
                file_stream<T> * stream = (start+i)->get();
                if (stream->can_read()) {
                    in_objects[i] = &stream->read();
                    MergeHeap->insert( in_objects[i], i );
                } else 
                    in_objects[i] = NULL;
                nread[i] = 1;
                if (indicator) indicator->step();
            }

            // *********************************************************
            // * Build a heap from the smallest items of each stream   *
            // *********************************************************
        
            MergeHeap->initialize ( );

            // *********************************************************
            // * Perform the merge until the inputs are exhausted.     *
            // *********************************************************
            while (MergeHeap->sizeofheap() > 0) {
                i = MergeHeap->get_min_run_id ();
                outStream->write(*in_objects[i]);

                bool eof = false;
        
                //Check if we read as many elements as we are allowed to
                if ( (cutoff != -1) && (nread[i]>=cutoff))
                    eof = true;
                else {
                    file_stream<T> * stream = (start+i)->get();
                    if (stream->can_read()) in_objects[i] = &stream->read();
                    else eof = true;
            
                    if (indicator) indicator->step();
                } 
        
                if (eof)
                    MergeHeap->delete_min_and_insert(NULL);
                else { 
                    nread[i]++;
                    MergeHeap->delete_min_and_insert (in_objects[i]);
                }
            }  //  while
        }

template<class T , class M >

err tpie::ami::partition_and_merge	(	stream< T > *	instream,
		stream< T > *	outstream,
		M *	m_obj
	)

Partitions a stream into substreams small enough to fit.

in main memory, operates on each in main memory, and then merges them together, possibly in several passes if low memory conditions dictate. This function takes three arguments: instream points to the input stream, outstream points to the output stream, and mo points to a merge management object that controls the merge. This function takes care of all the details of determining how much main memory is available, how big the initial substreams can be, how many streams can be merged at a time, and how many levels of merging must take place.

In order to complete the merge successfully, the function needs sufficient memory for a binary merge. If not enough memory is available, the function fails and it returns INSUFFICIENT_MAIN_MEMORY. Otherwise, it returns NO_ERROR.

Merge Management Objects for partition_and_merge:

The partition_and_merge() entry point requires a merge management object similar to the one described here. The following three additional member functions must also be provided.

main_mem_operate():

err main_mem_operate(T* mm_stream, size_t len); where mm_stream is a pointer to an array of objects that have been read into main memory, len is the number of objects in the array. This function is called by AMI_partition_and_merge() when a substream of the data is small enough to fit into main memory, and the (application-specific) processing of this subset of the data can therefore be completed in internal memory.

space_usage_per_stream():

size_t space_usage_per_stream(void); This function should return the amount of main memory that the merge management object will need per per input stream. Merge management objects are allowed to maintain data structures whose size is linear in the number of input streams being processed.

space_usage_overhead(): size_t space_usage_overhead(void); This function should return an upper bound on the number of bytes of

main memory the merge management object will allocate in addition to the portion that is linear in the number of streams.

Definition at line 508 of file merge.h.

References tpie::ami::stream< T >::available_streams(), tpie::ami::stream< T >::chunk_size(), tpie::consecutive_memory_available(), INSUFFICIENT_AVAILABLE_STREAMS, INSUFFICIENT_MAIN_MEMORY, main_mem_merge(), tpie::ami::stream< T >::main_memory_usage(), NO_ERROR, tpie::ami::stream< T >::persist(), tpie::ami::stream< T >::read_array(), tpie::ami::stream< T >::seek(), single_merge(), tpie::ami::stream< T >::stream_len(), tpie::STREAM_USAGE_MAXIMUM, tp_assert, tpie::tpie_delete(), tpie::tpie_delete_array(), and tpie::ami::stream< T >::write_array().

                                                {

        err ae;
        TPIE_OS_OFFSET len;
        TPIE_OS_SIZE_T sz_avail, sz_stream;
        unsigned int ii;
        int jj;
  
        //How much memory is available?
        sz_avail = consecutive_memory_available ();

        // If the whole input can fit in main memory then just call
        // main_mem_merge() to deal with it by loading it once and
        // processing it.
        len = instream->stream_len();
        if ((len * static_cast<TPIE_OS_OFFSET>(sizeof(T))) <= static_cast<TPIE_OS_OFFSET>(sz_avail)) {
        return main_mem_merge(instream, outstream, m_obj);
        } 
        //else {

  
 
        // The number of substreams that can be merged together at once; i
        // this many substreams (at most) we are dividing the input stream
        arity_t merge_arity;
  
        //nb of substreams the original input stream will be split into
        arity_t nb_orig_substr;
  
        // length (nb obj of type T) of the original substreams of the input
        // stream.  The last one may be shorter than this.
        TPIE_OS_OFFSET sz_orig_substr;
  
        // The initial temporary stream, to which substreams of the
        // original input stream are written.
        stream<T> *initial_tmp_stream;

        // A pointer to the buffer in main memory to read a memory load into.
        T *mm_stream;
  
  
        // Loop variables:
  
        // The stream being read at the current level.
        stream<T> *current_input;

        // The output stream for the current level if it is not outstream.
        stream<T> *intermediate_tmp_stream;
        
        // The size of substreams of *current_input that are being
        // merged.  The last one may be smaller.  This value should be
        // sz_orig_substr * (merge_arity ** k) where k is the
        // number of iterations the loop has gone through.
        TPIE_OS_OFFSET current_substream_len;

        // The exponenent used to verify that current_substream_len is
        // correct.
        unsigned int k;
  
        TPIE_OS_OFFSET sub_start, sub_end;
  
  
  
        // How many substreams will there be?  The main memory
        // available to us is the total amount available, minus what
        // is needed for the input stream and the temporary stream.
        if ((ae = instream->main_memory_usage(&sz_stream, STREAM_USAGE_MAXIMUM)) 
        != NO_ERROR) {
        return ae;
        }                                     
        if (sz_avail <= 2 * sz_stream + sizeof(T)) {
        return INSUFFICIENT_MAIN_MEMORY;
        }
        sz_avail -= 2 * sz_stream;

    
        // number of elements that will fit in memory (M) -R
        sz_orig_substr = sz_avail / sizeof(T);

        // Round the original substream length off to an integral number of
        // chunks.  This is for systems like HP-UX that cannot map in
        // overlapping regions.  It is also required for BTE's that are
        // capable of freeing chunks as they are read.
        {
        TPIE_OS_OFFSET sz_chunk_size = instream->chunk_size();
    
        sz_orig_substr = sz_chunk_size *
            ((sz_orig_substr + sz_chunk_size - 1) /sz_chunk_size);
        // WARNING sz_orig_substr now may not fit in memory!!! -R
        }

        // number of memoryloads in input ceil(N/M) -R
        nb_orig_substr = static_cast<arity_t>((len + sz_orig_substr - 1) / 
                          sz_orig_substr);
  
        // Account for the space that a merge object will use.
        {
        TPIE_OS_SIZE_T sz_avail_during_merge = sz_avail - m_obj->space_usage_overhead();
        TPIE_OS_SIZE_T sz_stream_during_merge = sz_stream +m_obj->space_usage_per_stream();
    
        merge_arity = static_cast<arity_t>((sz_avail_during_merge +
                            sz_stream_during_merge - 1) / 
                           sz_stream_during_merge);
        }

        // Make sure that the AMI is willing to provide us with the number
        // of substreams we want.  It may not be able to due to operating
        // system restrictions, such as on the number of regions that can be
        // mmap()ed in.
        {
        int ami_available_streams = instream->available_streams();
    
        if (ami_available_streams != -1) {
            if (ami_available_streams <= 4) {
            return INSUFFICIENT_AVAILABLE_STREAMS;
            }
            //  safe to cast, since ami_available_streams > 4
            if (merge_arity > static_cast<arity_t>(ami_available_streams) - 2) {
            merge_arity = ami_available_streams - 2;
            TP_LOG_DEBUG_ID("Reduced merge arity due to AMI restrictions.");
            }
        }
        }
        TP_LOG_DEBUG_ID("partition_and_merge(): merge arity = "
                << static_cast<TPIE_OS_OUTPUT_SIZE_T>(merge_arity));
        if (merge_arity < 2) {
        return INSUFFICIENT_MAIN_MEMORY;
        }
  
  
        //#define MINIMIZE_INITIAL_SUBSTREAM_LENGTH
#ifdef MINIMIZE_INITIAL_SUBSTREAM_LENGTH
  
        // Make the substreams as small as possible without increasing the
        // height of the merge tree.
        {
        // The tree height is the ceiling of the log base merge_arity
        // of the number of original substreams.
    
        double tree_height = log((double)nb_orig_substr)/ log((double)merge_arity);
        tp_assert(tree_height > 0, "Negative or zero tree height!");
    
        tree_height = ceil(tree_height);
    
        // See how many substreams we could possibly fit in the tree
        // without increasing the height.
        double max_original_substreams = pow((double)merge_arity, tree_height);
        tp_assert(max_original_substreams >= nb_orig_substr,
              "Number of permitted substreams was reduced.");

        // How big will such substreams be?
        double new_sz_original_substream = ceil((double)len /
                            max_original_substreams);
        tp_assert(new_sz_original_substream <= sz_orig_substr,
              "Size of original streams increased.");
    
        sz_orig_substr = (size_t)new_sz_original_substream;
        TP_LOG_DEBUG_ID("Memory constraints set original substreams = " << nb_orig_substr);
    
        nb_orig_substr = (len + sz_orig_substr - 1) / sz_orig_substr;
        TP_LOG_DEBUG_ID("Tree height constraints set original substreams = " << nb_orig_substr);
        }                
#endif // MINIMIZE_INITIAL_SUBSTREAM_LENGTH

    
        // Create a temporary stream, then iterate through the substreams,
        // processing each one and writing it to the corresponding substream
        // of the temporary stream.
        initial_tmp_stream = tpie_new<stream<T> >();
        mm_stream = tpie_new_array<T>(static_cast<TPIE_OS_SIZE_T>(sz_orig_substr));
  
        instream->seek(0);
        assert(ae == NO_ERROR);

        tp_assert(static_cast<TPIE_OS_OFFSET>(nb_orig_substr * sz_orig_substr - len) < sz_orig_substr,
              "Total substream length too long or too many.");
        tp_assert(len - static_cast<TPIE_OS_OFFSET>(nb_orig_substr - 1) * sz_orig_substr <= sz_orig_substr,
              "Total substream length too short or too few.");        
  
        for (ii = 0; ii++ < nb_orig_substr; ) {
    
        TPIE_OS_OFFSET mm_len;
        if (ii == nb_orig_substr) {
            mm_len = len % sz_orig_substr;
            // If it is an exact multiple, then the mod will come out 0,
            // which is wrong.
            if (!mm_len) {
            mm_len = sz_orig_substr;
            }
        } else {
            mm_len = sz_orig_substr;
        }
#ifndef TPIE_NDEBUG
        TPIE_OS_OFFSET mm_len_bak = mm_len;
#endif
    
        // Read a memory load out of the input stream.
        ae = instream->read_array(mm_stream, &mm_len);
        if (ae != NO_ERROR) {
            return ae;
        }
        tp_assert(mm_len == mm_len_bak,
              "Did not read the requested number of objects." <<
              "\n\tmm_len = " << mm_len <<
              "\n\tmm_len_bak = " << mm_len_bak << '.');
    
        // Solve in main memory. We know it fits, so cast to TPIE_OS_SIZE_T
        m_obj->main_mem_operate(mm_stream, static_cast<TPIE_OS_SIZE_T>(mm_len));
    
        // Write the result out to the temporary stream.
        ae = initial_tmp_stream->write_array(mm_stream, static_cast<TPIE_OS_SIZE_T>(mm_len));
        if (ae != NO_ERROR) {
            return ae;
        }            
        } //for
        tpie_delete_array(mm_stream, static_cast<TPIE_OS_SIZE_T>(sz_orig_substr));
  
        // Make sure the total length of the temporary stream is the same as
        // the total length of the original input stream.
        tp_assert(instream->stream_len() == initial_tmp_stream->stream_len(),
              "Stream lengths do not match:" <<
              "\n\tinstream->stream_len() = " << instream->stream_len() <<
              "\n\tinitial_tmp_stream->stream_len() = " <<
              initial_tmp_stream->stream_len() << ".\n");
  
        // Set up the loop invariants for the first iteration of hte main
        // loop.
        current_input = initial_tmp_stream;
        current_substream_len = sz_orig_substr;
  
        // Pointers to the substreams that will be merged.
        stream<T>* *the_substreams = tpie_new_array<stream<T>* >(merge_arity);
  
        //Monitoring prints.
        TP_LOG_DEBUG_ID("Number of runs from run formation is "
                << static_cast<TPIE_OS_OUTPUT_SIZE_T>(nb_orig_substr));
        TP_LOG_DEBUG_ID("Merge arity is " 
                << static_cast<TPIE_OS_OUTPUT_SIZE_T>(merge_arity));
  
  
        k = 0;
        // The main loop.  At the outermost level we are looping over levels
        // of the merge tree.  Typically this will be very small, e.g. 1-3.
        // CAVEAT: is the cast o.k.??
        for( ; current_substream_len < len;
         current_substream_len *= merge_arity) {
    
        // The number of substreams to be processed at this level.
        arity_t substream_count;
    
        // Set up to process a given level.
        tp_assert(len == current_input->stream_len(),
              "Current level stream not same length as input." <<
              "\n\tlen = " << len <<
              "\n\tcurrent_input->stream_len() = " <<
              current_input->stream_len() << ".\n");
    
        // Do we have enough main memory to merge all the substreams on
        // the current level into the output stream?  If so, then we will
        // do so, if not then we need an additional level of iteration to
        // process the substreams in groups.
        substream_count = static_cast<arity_t>((len + current_substream_len - 1) /
                               current_substream_len);
    
        if (substream_count <= merge_arity) {
      
            TP_LOG_DEBUG_ID("Merging substreams directly to the output stream.");
      
            // Create all the substreams
            for (sub_start = 0, ii = 0 ;
             ii < substream_count;
             sub_start += current_substream_len, ii++) {
    
            sub_end = sub_start + current_substream_len - 1;
            if (sub_end >= len) {
                sub_end = len - 1;
            }
            current_input->new_substream(READ_STREAM, sub_start, sub_end, the_substreams + ii);
            // The substreams are read-once.
            the_substreams[ii]->persist(PERSIST_READ_ONCE);
            }               
      
            tp_assert((sub_start >= len) &&
                  (sub_start < len + current_substream_len),
                  "Loop ended in wrong location.");
      
            // Fool the OS into unmapping the current block of the input
            // stream so that blocks of the substreams can be mapped in
            // without overlapping it.  This is needed for correct execution
            // on HP-UX.
            //this needs to be cleaned up..Laura
            current_input->seek(0);
            assert(ae == NO_ERROR);

            // Merge them into the output stream.
            ae = single_merge(the_substreams, substream_count, outstream, m_obj);
            if (ae != NO_ERROR) {
            return ae;
            }
            // Delete the substreams.
            for (ii = 0; ii < substream_count; ii++) {
                tpie_delete(the_substreams[ii]);
            }
            // And the current input, which is an intermediate stream of
            // some kind.
            tpie_delete(current_input);
        } else {
      
            //substream_count  is >  merge_arity
            TP_LOG_DEBUG_ID("Merging substreams to an intermediate stream.");
      
            // Create the next intermediate stream.
            intermediate_tmp_stream = new stream<T>;
      
            // Fool the OS into unmapping the current block of the input
            // stream so that blocks of the substreams can be mapped in
            // without overlapping it.  This is needed for correct execution
            // on HU-UX.
            //this needs to be cleaned up..Laura
            current_input->seek(0);
            assert(ae == NO_ERROR);

            // Loop through the substreams of the current stream, merging as
            // many as we can at a time until all are done with.
            for (sub_start = 0, ii = 0, jj = 0;
             ii < substream_count;
             sub_start += current_substream_len, ii++, jj++) {
    
            sub_end = sub_start + current_substream_len - 1;
            if (sub_end >= len) {
                sub_end = len - 1;
            }
            current_input->new_substream(READ_STREAM, sub_start, sub_end, the_substreams + jj); 
            // The substreams are read-once.
            the_substreams[jj]->persist(PERSIST_READ_ONCE);
                    
            // If we've got all we can handle or we've seen them all, then
            // merge them.
            if ((jj >= static_cast<int>(merge_arity) - 1) || 
                (ii == substream_count - 1)) {
      
                tp_assert(jj <= static_cast<int>(merge_arity) - 1,
                      "Index got too large.");
#if DEBUG_ASSERTIONS
                // Check the lengths before the merge.
                TPIE_OS_OFFSET sz_output, sz_output_after_merge;
                TPIE_OS_OFFSET sz_substream_total;
      
                {
                unsigned int kk;
        
                sz_output = intermediate_tmp_stream->stream_len();
                sz_substream_total = 0;
        
                for (kk = jj+1; kk--; ) {
                    sz_substream_total += the_substreams[kk]->stream_len();
                }                          
        
                }
#endif 
      
                // This should append to the stream, since
                // single_merge() does not rewind the output before
                // merging.
                ae = single_merge(the_substreams, jj+1,
                          intermediate_tmp_stream, m_obj);
                if (ae != NO_ERROR) {
                return ae;
                }
      
#if DEBUG_ASSERTIONS
                // Verify the total lengths after the merge.
                sz_output_after_merge = intermediate_tmp_stream->stream_len();
                tp_assert(sz_output_after_merge - sz_output ==
                      sz_substream_total,
                      "Stream lengths do not add up: " <<
                      sz_output_after_merge - sz_output <<
                      " written when " <<
                      sz_substream_total <<
                      " were to have been read.");
                                  
#endif 
      
                // Delete the substreams.  jj is currently the index of the
                // largest, so we want to bump it up before the idiomatic
                // loop.
                for (jj++; jj--; )
                    tpie_delete(the_substreams[jj]);
      
                // Now jj should be -1 so that it gets bumped back up to 0
                // before the next iteration of the outer loop.
                tp_assert((jj == -1), "Index not reduced to -1.");
      
            } // if                
            } //for
      
            // Get rid of the current input stream and use the next one.
            tpie_delete(current_input);
            current_input = intermediate_tmp_stream;
        }
    
        k++;
    
        }

        //Monitoring prints.
        TP_LOG_DEBUG_ID("Number of passes incl run formation is " << k+1);
  
        tpie_delete_array(the_substreams, merge_arity);
        return NO_ERROR;

    }

template<class T , class CMPR >

void tpie::ami::ptr_merge_sorted	(	typename tpie::array< tpie::auto_ptr< file_stream< T > > >::iterator	start,
		typename tpie::array< tpie::auto_ptr< file_stream< T > > >::iterator	end,
		file_stream< T > *	outStream,
		CMPR *	cmp
	)

Merging with a heap that keeps a pointer to the records rather than the records themselves: CMPR is the class of the comparison object, and must contain the method compare() which is called from within the merge.

This is one of the merge entry points for merging without the merge_management_object used by TPIE's merge. These routines perform the special case of merging when the the required output is the original records interleaved according to a comparison operator or function.

Definition at line 181 of file merge_sorted_runs.h.

References tpie::ami::merge_heap_ptr_op< REC, CMPR >::allocate(), and merge_sorted_runs().

                                        {

            // make a merge heap of pointers which uses the user's comparison
            // object and initialize it
            merge_heap_ptr_obj<T,CMPR> mrgheap (cmp);
            mrgheap.allocate(end-start);
        
            // Rewind all the input streams
            for (typename tpie::array<std::auto_ptr<file_stream<T> > >::iteratorarity_t i=start; 
                 i != end; ++i)
                i->seek(0); 
        
            merge_sorted_runs(start, end, outStream, mrgheap);
        }

template<class T , class M >

err tpie::ami::single_merge	(	stream< T > **	instreams,
		arity_t	arity,
		stream< T > *	outstream,
		M *	m_obj
	)

Merges arity streams in memory using a merge management object and write result into outstream.

Definition at line 298 of file merge.h.

References END_OF_STREAM, MERGE_CONTINUE, MERGE_DONE, MERGE_OUTPUT, MERGE_READ_MULTIPLE, NO_ERROR, OBJECT_INITIALIZATION, tpie::ami::stream< T >::read_item(), tp_assert, tpie::tpie_delete_array(), and tpie::ami::stream< T >::write_item().

Referenced by merge(), and partition_and_merge().

                                             {

        TPIE_OS_SIZE_T ii;
        err ami_err;
  
        // Create an array of pointers for the input.
        T* *in_objects = tpie_new_array<T*>(arity);
  
        // Create an array of flags the merge object can use to ask for more
        // input from specific streams.
        merge_flag* taken_flags  = tpie_new_array<merge_flag>(arity);
  
        // An index to speed things up when the merge object takes only from
        // one index.
        int taken_index;
  
        //Output of the merge object.
        T merge_out;
  
#if DEBUG_PERFECT_MERGE
        unsigned int input_count = 0, output_count = 0;
#endif    
  
        // Rewind and read the first item from every stream; count the
        // number of non-null items read
        for (ii = arity; ii--; ) {
        if ((ami_err = instreams[ii]->seek(0)) != NO_ERROR) {
            tpie_delete_array(in_objects, arity);
            tpie_delete_array(taken_flags, arity);
            return ami_err;
        }
        if ((ami_err = instreams[ii]->read_item(&(in_objects[ii]))) !=
            NO_ERROR) {
            //error on read
            if (ami_err == END_OF_STREAM) {
            in_objects[ii] = NULL;
            } else {
                tpie_delete_array(in_objects, arity);
                tpie_delete_array(taken_flags, arity);
            return ami_err;
            }
            // Set the taken flags to 0 before we call intialize()
            taken_flags[ii] = 0;
        } else {
            //item read succesfully
#if DEBUG_PERFECT_MERGE
            input_count++;
#endif                
        }
        }
  
        // Initialize the merge object.
        if (((ami_err = m_obj->initialize(arity, in_objects, taken_flags, 
                          taken_index)) != NO_ERROR) &&
        (ami_err != MERGE_READ_MULTIPLE)) {
        return OBJECT_INITIALIZATION;
        }      
  
  
        // Now simply call the merge object repeatedly until it claims to
        // be done or generates an error.
        while (1) {
        if (ami_err == MERGE_READ_MULTIPLE) {
            for (ii = arity; ii--; ) {
            if (taken_flags[ii]) {
                ami_err = instreams[ii]->read_item(&(in_objects[ii]));
                if (ami_err != NO_ERROR) {
                if (ami_err == END_OF_STREAM) {
                    in_objects[ii] = NULL;
                } else {
                    tpie_delete_array(in_objects, arity);
                    tpie_delete_array(taken_flags, arity);
                    return ami_err;
                }
                } else {
#if DEBUG_PERFECT_MERGE                    
                input_count++;
#endif
                }
            }
            // Clear all flags before operate is called.
            taken_flags[ii] = 0;
            }
        } else {
            // The last call took at most one item.
            if (taken_index >= 0) {
            ami_err = instreams[taken_index]->
                read_item(&(in_objects[taken_index]));
            if (ami_err != NO_ERROR) {
                if (ami_err == END_OF_STREAM) {
                in_objects[taken_index] = NULL;
                } else {
                    tpie_delete_array(in_objects, arity);
                    tpie_delete_array(taken_flags, arity);
                return ami_err;
                }
            } else {
#if DEBUG_PERFECT_MERGE                    
                input_count++;
#endif
            }
            taken_flags[taken_index] = 0;
            }
        }
        ami_err = m_obj->operate(in_objects, taken_flags, taken_index,
                     &merge_out);
        if (ami_err == MERGE_DONE) {
            break;
        } else if (ami_err == MERGE_OUTPUT) {
#if DEBUG_PERFECT_MERGE
            output_count++;
#endif                    
            if ((ami_err = outstream->write_item(merge_out)) !=
            NO_ERROR) {
                tpie_delete_array(in_objects, arity);
                tpie_delete_array(taken_flags, arity);
            return ami_err;
            }            
        } else if ((ami_err != MERGE_CONTINUE) &&
               (ami_err != MERGE_READ_MULTIPLE)) {
            tpie_delete_array(in_objects, arity);
            tpie_delete_array(taken_flags, arity);
            return ami_err;
        }
        }
  
#if DEBUG_PERFECT_MERGE
        tp_assert(input_count == output_count,
              "Merge done, input_count = " << input_count <<
              ", output_count = " << output_count << '.');
#endif

        tpie_delete_array(in_objects, arity);
        tpie_delete_array(taken_flags, arity);

        return NO_ERROR;
    };