org.apache.flink.runtime.io.network.partition.consumer

Class SingleInputGate

java.lang.Object

org.apache.flink.runtime.io.network.partition.consumer.InputGate

org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate

org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate

All Implemented Interfaces:

AutoCloseable, AvailabilityProvider, ChannelStateHolder, CheckpointableInput, PullingAsyncDataInput<BufferOrEvent>

public class SingleInputGate
extends IndexedInputGate

An input gate consumes one or more partitions of a single produced intermediate result.

Each intermediate result is partitioned over its producing parallel subtasks; each of these partitions is furthermore partitioned into one or more subpartitions.

As an example, consider a map-reduce program, where the map operator produces data and the reduce operator consumes the produced data.

 +-----+              +---------------------+              +--------+
 | Map | = produce => | Intermediate Result | <= consume = | Reduce |
 +-----+              +---------------------+              +--------+
 

When deploying such a program in parallel, the intermediate result will be partitioned over its producing parallel subtasks; each of these partitions is furthermore partitioned into one or more subpartitions.

                            Intermediate result
               +-----------------------------------------+
               |                      +----------------+ |              +-----------------------+
 +-------+     | +-------------+  +=> | Subpartition 1 | | <=======+=== | Input Gate | Reduce 1 |
 | Map 1 | ==> | | Partition 1 | =|   +----------------+ |         |    +-----------------------+
 +-------+     | +-------------+  +=> | Subpartition 2 | | <==+    |
               |                      +----------------+ |    |    | Subpartition request
               |                                         |    |    |
               |                      +----------------+ |    |    |
 +-------+     | +-------------+  +=> | Subpartition 1 | | <==+====+
 | Map 2 | ==> | | Partition 2 | =|   +----------------+ |    |         +-----------------------+
 +-------+     | +-------------+  +=> | Subpartition 2 | | <==+======== | Input Gate | Reduce 2 |
               |                      +----------------+ |              +-----------------------+
               +-----------------------------------------+
 

In the above example, two map subtasks produce the intermediate result in parallel, resulting in two partitions (Partition 1 and 2). Each of these partitions is further partitioned into two subpartitions -- one for each parallel reduce subtask.

Nested Class Summary

Nested classes/interfaces inherited from class org.apache.flink.runtime.io.network.partition.consumer.InputGate

InputGate.InputWithData<INPUT,DATA>

Nested classes/interfaces inherited from interface org.apache.flink.runtime.io.PullingAsyncDataInput

PullingAsyncDataInput.EndOfDataStatus

Nested classes/interfaces inherited from interface org.apache.flink.runtime.io.AvailabilityProvider

AvailabilityProvider.AvailabilityHelper

Field Summary

Fields inherited from class org.apache.flink.runtime.io.network.partition.consumer.InputGate

availabilityHelper, priorityAvailabilityHelper

Fields inherited from interface org.apache.flink.runtime.io.AvailabilityProvider

AVAILABLE

Constructor Summary

Constructors

Constructor and Description

SingleInputGate(String owningTaskName, int gateIndex, IntermediateDataSetID consumedResultId, ResultPartitionType consumedPartitionType, IndexRange subpartitionIndexRange, int numberOfInputChannels, PartitionProducerStateProvider partitionProducerStateProvider, SupplierWithException<BufferPool,IOException> bufferPoolFactory, BufferDecompressor bufferDecompressor, MemorySegmentProvider memorySegmentProvider, int segmentSize, ThroughputCalculator throughputCalculator, BufferDebloater bufferDebloater, TieredStorageConsumerClient tieredStorageConsumerClient, TieredStorageNettyServiceImpl nettyService, List<TieredStorageConsumerSpec> tieredStorageConsumerSpecs)

Method Summary

Modifier and Type	Method and Description
void	acknowledgeAllRecordsProcessed(InputChannelInfo channelInfo)
void	announceBufferSize(int newBufferSize)
void	close()
void	convertRecoveredInputChannels()
void	finishReadRecoveredState()
BufferPool	getBufferPool()
InputChannel	getChannel(int channelIndex) Returns the channel of this gate.
CompletableFuture<Void>	getCloseFuture()
ResultPartitionType	getConsumedPartitionType() Returns the type of this input channel's consumed result partition.
int	getGateIndex() Returns the index of this input gate.
Map<org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate.SubpartitionInfo,InputChannel>	getInputChannels()
protected PrioritizedDeque<InputChannel>	getInputChannelsWithData()
Duration	getLastEstimatedTimeToConsume()
Optional<BufferOrEvent>	getNext() Blocking call waiting for next BufferOrEvent.
int	getNumberOfInputChannels()
int	getNumberOfQueuedBuffers()
String	getOwningTaskName()
long	getSizeOfQueuedBuffers()
CompletableFuture<Void>	getStateConsumedFuture()
List<InputChannelInfo>	getUnfinishedChannels() Returns the list of channels that have not received EndOfPartitionEvent.
PullingAsyncDataInput.EndOfDataStatus	hasReceivedEndOfData() Tells if we consumed all available data.
boolean	isFinished()
Optional<BufferOrEvent>	pollNext() Poll the BufferOrEvent.
void	requestPartitions()
void	resumeConsumption(InputChannelInfo channelInfo)
void	retriggerPartitionRequest(IntermediateResultPartitionID partitionId, int subpartitionIndex) Retriggers a partition request.
void	sendTaskEvent(TaskEvent event)
void	setBufferPool(BufferPool bufferPool)
void	setInputChannels(InputChannel... channels)
void	setup() Setup gate, potentially heavy-weight, blocking operation comparing to just creation.
void	setupChannels() Assign the exclusive buffers to all remote input channels directly for credit-based mode.
String	toString()
void	triggerDebloating()
void	updateInputChannel(ResourceID localLocation, NettyShuffleDescriptor shuffleDescriptor)

Methods inherited from class org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate

blockConsumption, checkpointStarted, checkpointStopped, convertToPriorityEvent, getInputGateIndex

Methods inherited from class org.apache.flink.runtime.io.network.partition.consumer.InputGate

getAvailableFuture, getChannelInfos, getPriorityEventAvailableFuture, setChannelStateWriter

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.flink.runtime.io.network.partition.consumer.CheckpointableInput

getChannelInfos

Methods inherited from interface org.apache.flink.runtime.io.AvailabilityProvider

and, isApproximatelyAvailable, isAvailable, or

Constructor Detail

SingleInputGate

public SingleInputGate(String owningTaskName,
                       int gateIndex,
                       IntermediateDataSetID consumedResultId,
                       ResultPartitionType consumedPartitionType,
                       IndexRange subpartitionIndexRange,
                       int numberOfInputChannels,
                       PartitionProducerStateProvider partitionProducerStateProvider,
                       SupplierWithException<BufferPool,IOException> bufferPoolFactory,
                       @Nullable
                       BufferDecompressor bufferDecompressor,
                       MemorySegmentProvider memorySegmentProvider,
                       int segmentSize,
                       ThroughputCalculator throughputCalculator,
                       @Nullable
                       BufferDebloater bufferDebloater,
                       @Nullable
                       TieredStorageConsumerClient tieredStorageConsumerClient,
                       @Nullable
                       TieredStorageNettyServiceImpl nettyService,
                       @Nullable
                       List<TieredStorageConsumerSpec> tieredStorageConsumerSpecs)

Method Detail

getInputChannelsWithData

protected PrioritizedDeque<InputChannel> getInputChannelsWithData()

setup

public void setup()
           throws IOException

Description copied from class: InputGate

Setup gate, potentially heavy-weight, blocking operation comparing to just creation.

Specified by:

setup in class InputGate

Throws:

IOException

getStateConsumedFuture

public CompletableFuture<Void> getStateConsumedFuture()

Specified by:

getStateConsumedFuture in class InputGate

requestPartitions

public void requestPartitions()

Specified by:

requestPartitions in class InputGate

convertRecoveredInputChannels

@VisibleForTesting
public void convertRecoveredInputChannels()

finishReadRecoveredState

public void finishReadRecoveredState()
                              throws IOException

Specified by:

finishReadRecoveredState in class InputGate

Throws:

IOException

getNumberOfInputChannels

public int getNumberOfInputChannels()

Specified by:

getNumberOfInputChannels in interface CheckpointableInput

Specified by:

getNumberOfInputChannels in class InputGate

getGateIndex

public int getGateIndex()

Description copied from class: IndexedInputGate

Returns the index of this input gate. Only supported on

Specified by:

getGateIndex in class IndexedInputGate

getUnfinishedChannels

public List<InputChannelInfo> getUnfinishedChannels()

Description copied from class: IndexedInputGate

Returns the list of channels that have not received EndOfPartitionEvent.

Specified by:

getUnfinishedChannels in class IndexedInputGate

announceBufferSize

@VisibleForTesting
public void announceBufferSize(int newBufferSize)

triggerDebloating

public void triggerDebloating()

Specified by:

triggerDebloating in class IndexedInputGate

getLastEstimatedTimeToConsume

public Duration getLastEstimatedTimeToConsume()

getConsumedPartitionType

public ResultPartitionType getConsumedPartitionType()

Returns the type of this input channel's consumed result partition.

Returns:

consumed result partition type

getBufferPool

public BufferPool getBufferPool()

getOwningTaskName

public String getOwningTaskName()

getNumberOfQueuedBuffers

public int getNumberOfQueuedBuffers()

getSizeOfQueuedBuffers

public long getSizeOfQueuedBuffers()

getCloseFuture

public CompletableFuture<Void> getCloseFuture()

getChannel

public InputChannel getChannel(int channelIndex)

Description copied from class: InputGate

Returns the channel of this gate.

Specified by:

getChannel in class InputGate

setBufferPool

public void setBufferPool(BufferPool bufferPool)

setupChannels

@VisibleForTesting
public void setupChannels()
                                      throws IOException

Assign the exclusive buffers to all remote input channels directly for credit-based mode.

Throws:

IOException

setInputChannels

public void setInputChannels(InputChannel... channels)

updateInputChannel

public void updateInputChannel(ResourceID localLocation,
                               NettyShuffleDescriptor shuffleDescriptor)
                        throws IOException,
                               InterruptedException

Throws:

IOException

InterruptedException

retriggerPartitionRequest

public void retriggerPartitionRequest(IntermediateResultPartitionID partitionId,
                                      int subpartitionIndex)
                               throws IOException

Retriggers a partition request.

Throws:

IOException

close

public void close()
           throws IOException

Throws:

IOException

isFinished

public boolean isFinished()

Specified by:

isFinished in interface PullingAsyncDataInput<BufferOrEvent>

Specified by:

isFinished in class InputGate

Returns:

true if is finished and for example end of input was reached, false otherwise.

hasReceivedEndOfData

public PullingAsyncDataInput.EndOfDataStatus hasReceivedEndOfData()

Description copied from interface: PullingAsyncDataInput

Tells if we consumed all available data.

Moreover it tells us the reason why there is no more data incoming. If any of the upstream subtasks finished because of the stop-with-savepoint --no-drain, we should not drain the input. See also StopMode.

toString

public String toString()

Overrides:

toString in class Object

getNext

public Optional<BufferOrEvent> getNext()
                                throws IOException,
                                       InterruptedException

Description copied from class: InputGate

Blocking call waiting for next BufferOrEvent.

Note: It should be guaranteed that the previous returned buffer has been recycled before getting next one.

Specified by:

getNext in class InputGate

Returns:

Optional.empty() if InputGate.isFinished() returns true.

Throws:

IOException

InterruptedException

pollNext

public Optional<BufferOrEvent> pollNext()
                                 throws IOException,
                                        InterruptedException

Description copied from class: InputGate

Poll the BufferOrEvent.

Note: It should be guaranteed that the previous returned buffer has been recycled before polling next one.

Specified by:

pollNext in interface PullingAsyncDataInput<BufferOrEvent>

Specified by:

pollNext in class InputGate

Returns:

Optional.empty() if there is no data to return or if InputGate.isFinished() returns true.

Throws:

IOException

InterruptedException

sendTaskEvent

public void sendTaskEvent(TaskEvent event)
                   throws IOException

Specified by:

sendTaskEvent in class InputGate

Throws:

IOException

resumeConsumption

public void resumeConsumption(InputChannelInfo channelInfo)
                       throws IOException

Specified by:

resumeConsumption in interface CheckpointableInput

Specified by:

resumeConsumption in class InputGate

Throws:

IOException

acknowledgeAllRecordsProcessed

public void acknowledgeAllRecordsProcessed(InputChannelInfo channelInfo)
                                    throws IOException

Specified by:

acknowledgeAllRecordsProcessed in class InputGate

Throws:

IOException

getInputChannels

public Map<org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate.SubpartitionInfo,InputChannel> getInputChannels()