org.apache.flink.runtime.operators.sort

Class UnilateralSortMerger.SpillingThread

java.lang.Object

java.lang.Thread

org.apache.flink.runtime.operators.sort.UnilateralSortMerger.ThreadBase<E>

org.apache.flink.runtime.operators.sort.UnilateralSortMerger.SpillingThread

All Implemented Interfaces:

Runnable, Thread.UncaughtExceptionHandler

Direct Known Subclasses:

CombiningUnilateralSortMerger.CombiningSpillingThread

Enclosing class:

UnilateralSortMerger<E>

protected class UnilateralSortMerger.SpillingThread
extends UnilateralSortMerger.ThreadBase<E>

The thread that handles the spilling of intermediate results and sets up the merging. It also merges the channels until sufficiently few channels remain to perform the final streamed merge.

Nested Class Summary

Nested classes/interfaces inherited from class java.lang.Thread

Thread.State, Thread.UncaughtExceptionHandler

Field Summary

Fields

Modifier and Type	Field and Description
protected TypeComparator<E>	comparator
protected IOManager	ioManager
protected int	maxFanIn
protected MemoryManager	memManager
protected List<MemorySegment>	mergeReadMemory
protected int	numWriteBuffersToCluster
protected TypeSerializer<E>	serializer
protected List<MemorySegment>	writeMemory

Fields inherited from class org.apache.flink.runtime.operators.sort.UnilateralSortMerger.ThreadBase

queues

Fields inherited from class java.lang.Thread

MAX_PRIORITY, MIN_PRIORITY, NORM_PRIORITY

Constructor Summary

Constructors

Constructor and Description
SpillingThread(ExceptionHandler<IOException> exceptionHandler, UnilateralSortMerger.CircularQueues<E> queues, AbstractInvokable parentTask, MemoryManager memManager, IOManager ioManager, TypeSerializer<E> serializer, TypeComparator<E> comparator, List<MemorySegment> sortReadMemory, List<MemorySegment> writeMemory, int maxNumFileHandles) Creates the spilling thread.

Method Summary

Modifier and Type	Method and Description
protected void	disposeSortBuffers(boolean releaseMemory) Releases the memory that is registered for in-memory sorted run generation.
protected MergeIterator<E>	getMergingIterator(List<UnilateralSortMerger.ChannelWithBlockCount> channelIDs, List<List<MemorySegment>> inputSegments, List<FileIOChannel> readerList, MutableObjectIterator<E> largeRecords) Returns an iterator that iterates over the merged result from all given channels.
protected void	getSegmentsForReaders(List<List<MemorySegment>> target, List<MemorySegment> memory, int numChannels) Divides the given collection of memory buffers among numChannels sublists.
void	go() Entry point of the thread.
protected List<UnilateralSortMerger.ChannelWithBlockCount>	mergeChannelList(List<UnilateralSortMerger.ChannelWithBlockCount> channelIDs, List<MemorySegment> allReadBuffers, List<MemorySegment> writeBuffers) Merges the given sorted runs to a smaller number of sorted runs.
protected UnilateralSortMerger.ChannelWithBlockCount	mergeChannels(List<UnilateralSortMerger.ChannelWithBlockCount> channelIDs, List<List<MemorySegment>> readBuffers, List<MemorySegment> writeBuffers) Merges the sorted runs described by the given Channel IDs into a single sorted run.
protected void	registerChannelToBeRemovedAtShudown(FileIOChannel.ID channel) Adds a channel to the list of channels that are to be removed at shutdown.
protected void	registerOpenChannelToBeRemovedAtShudown(FileIOChannel channel) Adds a channel reader/writer to the list of channels that are to be removed at shutdown.
protected UnilateralSortMerger.CircularElement<E>	takeNext(BlockingQueue<UnilateralSortMerger.CircularElement<E>> queue, Queue<UnilateralSortMerger.CircularElement<E>> cache)
protected void	unregisterChannelToBeRemovedAtShudown(FileIOChannel.ID channel) Removes a channel from the list of channels that are to be removed at shutdown.
protected void	unregisterOpenChannelToBeRemovedAtShudown(FileIOChannel channel) Removes a channel reader/writer from the list of channels that are to be removed at shutdown.

Methods inherited from class org.apache.flink.runtime.operators.sort.UnilateralSortMerger.ThreadBase

internalHandleException, isRunning, run, shutdown, uncaughtException

Methods inherited from class java.lang.Thread

activeCount, checkAccess, clone, countStackFrames, currentThread, destroy, dumpStack, enumerate, getAllStackTraces, getContextClassLoader, getDefaultUncaughtExceptionHandler, getId, getName, getPriority, getStackTrace, getState, getThreadGroup, getUncaughtExceptionHandler, holdsLock, interrupt, interrupted, isAlive, isDaemon, isInterrupted, join, join, join, resume, setContextClassLoader, setDaemon, setDefaultUncaughtExceptionHandler, setName, setPriority, setUncaughtExceptionHandler, sleep, sleep, start, stop, stop, suspend, toString, yield

Methods inherited from class java.lang.Object

equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Field Detail

memManager

protected final MemoryManager memManager

ioManager

protected final IOManager ioManager

serializer

protected final TypeSerializer<E> serializer

comparator

protected final TypeComparator<E> comparator

writeMemory

protected final List<MemorySegment> writeMemory

mergeReadMemory

protected final List<MemorySegment> mergeReadMemory

maxFanIn

protected final int maxFanIn

numWriteBuffersToCluster

protected final int numWriteBuffersToCluster

Constructor Detail

SpillingThread

public SpillingThread(ExceptionHandler<IOException> exceptionHandler,
                      UnilateralSortMerger.CircularQueues<E> queues,
                      AbstractInvokable parentTask,
                      MemoryManager memManager,
                      IOManager ioManager,
                      TypeSerializer<E> serializer,
                      TypeComparator<E> comparator,
                      List<MemorySegment> sortReadMemory,
                      List<MemorySegment> writeMemory,
                      int maxNumFileHandles)

Creates the spilling thread.

Parameters:

exceptionHandler - The exception handler to call for all exceptions.

queues - The queues used to pass buffers between the threads.

parentTask - The task that started this thread. If non-null, it is used to register this thread.

memManager - The memory manager used to allocate buffers for the readers and writers.

ioManager - The I/I manager used to instantiate readers and writers from.

serializer -

comparator -

sortReadMemory -

writeMemory -

maxNumFileHandles -

Method Detail

go

public void go()
        throws IOException

Entry point of the thread.

Specified by:

go in class UnilateralSortMerger.ThreadBase<E>

Throws:

IOException - Exceptions that prohibit correct completion of the work may be thrown by the thread.

disposeSortBuffers

protected final void disposeSortBuffers(boolean releaseMemory)

Releases the memory that is registered for in-memory sorted run generation.

takeNext

protected final UnilateralSortMerger.CircularElement<E> takeNext(BlockingQueue<UnilateralSortMerger.CircularElement<E>> queue,
                                                                 Queue<UnilateralSortMerger.CircularElement<E>> cache)
                                                          throws InterruptedException

Throws:

InterruptedException

getMergingIterator

protected final MergeIterator<E> getMergingIterator(List<UnilateralSortMerger.ChannelWithBlockCount> channelIDs,
                                                    List<List<MemorySegment>> inputSegments,
                                                    List<FileIOChannel> readerList,
                                                    MutableObjectIterator<E> largeRecords)
                                             throws IOException

Returns an iterator that iterates over the merged result from all given channels.

Parameters:

channelIDs - The channels that are to be merged and returned.

inputSegments - The buffers to be used for reading. The list contains for each channel one list of input segments. The size of the inputSegments list must be equal to that of the channelIDs list.

Returns:

An iterator over the merged records of the input channels.

Throws:

IOException - Thrown, if the readers encounter an I/O problem.

mergeChannelList

protected final List<UnilateralSortMerger.ChannelWithBlockCount> mergeChannelList(List<UnilateralSortMerger.ChannelWithBlockCount> channelIDs,
                                                                                  List<MemorySegment> allReadBuffers,
                                                                                  List<MemorySegment> writeBuffers)
                                                                           throws IOException

Merges the given sorted runs to a smaller number of sorted runs.

Parameters:

channelIDs - The IDs of the sorted runs that need to be merged.

allReadBuffers -

writeBuffers - The buffers to be used by the writers.

Returns:

A list of the IDs of the merged channels.

Throws:

IOException - Thrown, if the readers or writers encountered an I/O problem.

mergeChannels

protected UnilateralSortMerger.ChannelWithBlockCount mergeChannels(List<UnilateralSortMerger.ChannelWithBlockCount> channelIDs,
                                                                   List<List<MemorySegment>> readBuffers,
                                                                   List<MemorySegment> writeBuffers)
                                                            throws IOException

Merges the sorted runs described by the given Channel IDs into a single sorted run. The merging process uses the given read and write buffers.

Parameters:

channelIDs - The IDs of the runs' channels.

readBuffers - The buffers for the readers that read the sorted runs.

writeBuffers - The buffers for the writer that writes the merged channel.

Returns:

The ID and number of blocks of the channel that describes the merged run.

Throws:

IOException

getSegmentsForReaders

protected final void getSegmentsForReaders(List<List<MemorySegment>> target,
                                           List<MemorySegment> memory,
                                           int numChannels)

Divides the given collection of memory buffers among numChannels sublists.

Parameters:

target - The list into which the lists with buffers for the channels are put.

memory - A list containing the memory buffers to be distributed. The buffers are not removed from this list.

numChannels - The number of channels for which to allocate buffers. Must not be zero.

registerChannelToBeRemovedAtShudown

protected void registerChannelToBeRemovedAtShudown(FileIOChannel.ID channel)

Adds a channel to the list of channels that are to be removed at shutdown.

Parameters:

channel - The channel id.

unregisterChannelToBeRemovedAtShudown

protected void unregisterChannelToBeRemovedAtShudown(FileIOChannel.ID channel)

Removes a channel from the list of channels that are to be removed at shutdown.

Parameters:

channel - The channel id.

registerOpenChannelToBeRemovedAtShudown

protected void registerOpenChannelToBeRemovedAtShudown(FileIOChannel channel)

Adds a channel reader/writer to the list of channels that are to be removed at shutdown.

Parameters:

channel - The channel reader/writer.

unregisterOpenChannelToBeRemovedAtShudown

protected void unregisterOpenChannelToBeRemovedAtShudown(FileIOChannel channel)

Removes a channel reader/writer from the list of channels that are to be removed at shutdown.

Parameters:

channel - The channel reader/writer.