[FLINK-18991] Optimize reading InputChannel state

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Invalid
• Fix Version/s: 1.12.0
• Component/s: Runtime / Task
• Assignee: Roman Khachatryan
• Reporter: Roman Khachatryan
• Created: 18/Aug/20 17:04
• Updated: 26/Aug/20 07:40
• Resolved: 26/Aug/20 07:38

Description

Curently, channel state is read not sequentially.

Inverting control would make it more efficient.

Current call chain: 

 StreamTask.readRecoveredChannelState ->  strictly after reading output; when state is *consumed*, requests partitions
    SingleInputGate.readRecoveredState -> submits a task to an executor from main task thread; loop through special channels
        RecoveredInputChannel.readRecoveredState  -> loop while has data; buffer = bufferManager.requestBufferBlocking();
                where bufferManager = new BufferManager(inputGate.getMemorySegmentProvider(), this, 0);
            ChannelStateReader.readInputData  

Proposed call chain:

StreamTask.readRecoveredChannelState ->  strictly after reading output; when state is *consumed*, requests partitions
    ChannelStateReader.readInputData (another thread, maybe same executor, maybe change injection)
        loop through state handles ordererd by handle, offset
        request buffer from channel
        pass buffer to RecoveredInputChannel.onRecoveredStateBuffer

  Depends on FLINK-18989

Issue Links

• is blocked by: FLINK-18989 Write input and output channel state to separate files