Flink -- Barrier

/**
 * The CheckpointBarrierHandler reacts to checkpoint barrier arriving from the input channels.
 * Different implementations may either simply track barriers, or block certain inputs on
 * barriers.
 */
public interface CheckpointBarrierHandler {

    /**
     * Returns the next {@link BufferOrEvent} that the operator may consume.
     * This call blocks until the next BufferOrEvent is available, ir until the stream
     * has been determined to be finished.
     * 
     * @return The next BufferOrEvent, or {@code null}, if the stream is finished.
     * @throws java.io.IOException Thrown, if the network or local disk I/O fails.
     * @throws java.lang.InterruptedException Thrown, if the thread is interrupted while blocking during
     *                                        waiting for the next BufferOrEvent to become available.
     */
    BufferOrEvent getNextNonBlocked() throws IOException, InterruptedException;

    /**
     * Registers the given event handler to be notified on successful checkpoints.
     * 
     * @param checkpointHandler The handler to register.
     */
    void registerCheckpointEventHandler(EventListener<CheckpointBarrier> checkpointHandler);

    /**
     * Cleans up all internally held resources.
     * 
     * @throws IOException Thrown, if the cleanup of I/O resources failed.
     */
    void cleanup() throws IOException;

    /**
     * Checks if the barrier handler has buffered any data internally.
     * @return True, if no data is buffered internally, false otherwise.
     */
    boolean isEmpty();
}

/**
 * The barrier buffer is {@link CheckpointBarrierHandler} that blocks inputs with barriers until
 * all inputs have received the barrier for a given checkpoint.
 * 
 * <p>To avoid back-pressuring the input streams (which may cause distributed deadlocks), the
 * BarrierBuffer continues receiving buffers from the blocked channels and stores them internally until 
 * the blocks are released.</p>
 */
public class BarrierBuffer implements CheckpointBarrierHandler {
    
    /** The gate that the buffer draws its input from */
    private final InputGate inputGate;   //输入

    /** Flags that indicate whether a channel is currently blocked/buffered */
    private final boolean[] blockedChannels;  //被blocked的channels
    
    /** The total number of channels that this buffer handles data from */
    private final int totalNumberOfInputChannels;
    
    /** To utility to write blocked data to a file channel */
    private final BufferSpiller bufferSpiller;  //为了不造成反压，对于被block的channl，不会真正的block，而是只是把数据放到buffer中

    /** The pending blocked buffer/event sequences. Must be consumed before requesting
     * further data from the input gate. */
    private final ArrayDeque<BufferSpiller.SpilledBufferOrEventSequence> queuedBuffered; //更多的没来得及处理的unblock buffer数据

    /** The sequence of buffers/events that has been unblocked and must now be consumed
     * before requesting further data from the input gate */
    private BufferSpiller.SpilledBufferOrEventSequence currentBuffered; //由bufferSpiller进行rollover产生的，已经unblock的buffer数据

    /** Handler that receives the checkpoint notifications */
    private EventListener<CheckpointBarrier> checkpointHandler; //创建checkpoint的逻辑

    /** The ID of the checkpoint for which we expect barriers */
    private long currentCheckpointId = -1L;

    /** The number of received barriers (= number of blocked/buffered channels) */
    private int numBarriersReceived;
    
    /** The number of already closed channels */
    private int numClosedChannels;
    
    /** Flag to indicate whether we have drawn all available input */
    private boolean endOfStream;
}

@Override
public BufferOrEvent getNextNonBlocked() throws IOException, InterruptedException {
    while (true) {
        // process buffered BufferOrEvents before grabbing new ones
        BufferOrEvent next;
        if (currentBuffered == null) { //如果currentBuffered为空，说明没有unblock的buffer数据，直接从inputGate读取
            next = inputGate.getNextBufferOrEvent();
        }
        else {
            next = currentBuffered.getNext(); //从currentBuffered读
            if (next == null) {  //如果读到的为空，说明currentBuffered已经为空
                completeBufferedSequence(); //清空当前的currentBuffered，看看queuedBuffered中还有没有需要处理的buffer
                return getNextNonBlocked();
            }
        }
        
        if (next != null) {
            if (isBlocked(next.getChannelIndex())) { //如果这个channel仍然是被block的
                // if the channel is blocked we, we just store the BufferOrEvent
                bufferSpiller.add(next); //那么我们只是把这个BufferOrEvent放到bufferSpiller里面
            }
            else if (next.isBuffer()) { //如果没有被block，就处理该条数据，如果是buffer即真实数据，那么直接返回该数据
                return next;
            }
            else if (next.getEvent().getClass() == CheckpointBarrier.class) { //如果是CheckpointBarrier
                if (!endOfStream) {
                    // process barriers only if there is a chance of the checkpoint completing
                    processBarrier((CheckpointBarrier) next.getEvent(), next.getChannelIndex()); //那么调用processBarrier，后面具体分析
                }
            }
            else {
                if (next.getEvent().getClass() == EndOfPartitionEvent.class) {
                    numClosedChannels++;
                    // no chance to complete this checkpoint
                    releaseBlocks(); //因为某个channel close了，那就永远也无法从这个channel获取barrier了，所以releaseBlocks
                }
                return next;
            }
        }
        else if (!endOfStream) {
            // end of stream. we feed the data that is still buffered
            endOfStream = true;
            releaseBlocks();//流结束了，所以也需要releaseBlocks
            return getNextNonBlocked();
        }
        else {
            return null;
        }
    }
}

private void processBarrier(CheckpointBarrier receivedBarrier, int channelIndex) throws IOException {
    final long barrierId = receivedBarrier.getId(); //取出全局barrier id

    if (numBarriersReceived > 0) { //如果之前收到过barrier
        // subsequent barrier of a checkpoint.
        if (barrierId == currentCheckpointId) {  //看下刚收到的和之前的barrierid是否一样
            // regular case
            onBarrier(channelIndex); //如果一样调用onBarrier
        }
        else if (barrierId > currentCheckpointId) { //如果大于currentCheckpointId，说明这个id已经过期了，因为在一个channel上，barrier id应该是按序发送的
            // we did not complete the current checkpoint
            LOG.warn("Received checkpoint barrier for checkpoint {} before completing current checkpoint {}. " +
                    "Skipping current checkpoint.", barrierId, currentCheckpointId);

            releaseBlocks(); //既然这个barrier已经过期，所以releaseBlocks()
            currentCheckpointId = barrierId; //设置新的barrierId
            onBarrier(channelIndex);
        }
        else { //忽略已过期的barrier
            // ignore trailing barrier from aborted checkpoint
            return;
        }
        
    }
    else if (barrierId > currentCheckpointId) { //新的barrier
        // first barrier of a new checkpoint
        currentCheckpointId = barrierId;
        onBarrier(channelIndex);
    }
    else {
        // trailing barrier from previous (skipped) checkpoint
        return;
    }

    // check if we have all barriers
    if (numBarriersReceived + numClosedChannels == totalNumberOfInputChannels) { //如果我们已经集齐所有的barrier
        if (LOG.isDebugEnabled()) {
            LOG.debug("Received all barrier, triggering checkpoint {} at {}",
                    receivedBarrier.getId(), receivedBarrier.getTimestamp());
        }

        if (checkpointHandler != null) {
            checkpointHandler.onEvent(receivedBarrier); //触发生成checkpoint
        }
        
        releaseBlocks(); 调用releaseBlocks
    }
}

protected final EventListener<CheckpointBarrier> getCheckpointBarrierListener() {
    return new EventListener<CheckpointBarrier>() {
        @Override
        public void onEvent(CheckpointBarrier barrier) {
            try {
                triggerCheckpoint(barrier.getId(), barrier.getTimestamp()); //做checkpoint
            }
            catch (Exception e) {
                throw new RuntimeException("Error triggering a checkpoint as the result of receiving checkpoint barrier", e);
            }
        }
    };
}

private void onBarrier(int channelIndex) throws IOException {
    if (!blockedChannels[channelIndex]) {
        blockedChannels[channelIndex] = true;
        numBarriersReceived++;
        
        if (LOG.isDebugEnabled()) {
            LOG.debug("Received barrier from channel " + channelIndex);
        }
    }
    else {
        throw new IOException("Stream corrupt: Repeated barrier for same checkpoint and input stream");
    }
}

/**
 * Releases the blocks on all channels. Makes sure the just written data
 * is the next to be consumed.
 */
private void releaseBlocks() throws IOException {

    for (int i = 0; i < blockedChannels.length; i++) {
        blockedChannels[i] = false;
    }
    numBarriersReceived = 0;

    if (currentBuffered == null) { //理论上，在调用releaseBlocks前，所有channel都是处于blocked状态，所以currentBuffered应该为空
        // common case: no more buffered data
        currentBuffered = bufferSpiller.rollOver(); //把block期间buffer的数据文件，设为currentBuffered
        if (currentBuffered != null) {
            currentBuffered.open();
        }
    }
    else { //不为空，是uncommon的case
        // uncommon case: buffered data pending
        // push back the pending data, if we have any
        
        // since we did not fully drain the previous sequence, we need to allocate a new buffer for this one
        BufferSpiller.SpilledBufferOrEventSequence bufferedNow = bufferSpiller.rollOverWithNewBuffer();
        if (bufferedNow != null) {
            bufferedNow.open();
            queuedBuffered.addFirst(currentBuffered); //currentBuffered不为空，所以先把当前的放到queuedBuffered里面
            currentBuffered = bufferedNow;
        }
    }
}

/**
 * Creates a new buffer spiller, spilling to one of the I/O manager's temp directories.
 * 
 * @param ioManager The I/O manager for access to teh temp directories.
 * @param pageSize The page size used to re-create spilled buffers.
 * @throws IOException Thrown if the temp files for spilling cannot be initialized.
 */
public BufferSpiller(IOManager ioManager, int pageSize) throws IOException {
    this.pageSize = pageSize;
    
    this.readBuffer = ByteBuffer.allocateDirect(READ_BUFFER_SIZE);
    this.readBuffer.order(ByteOrder.LITTLE_ENDIAN);
    
    this.headBuffer = ByteBuffer.allocateDirect(16);
    this.headBuffer.order(ByteOrder.LITTLE_ENDIAN);
    
    this.sources = new ByteBuffer[] { this.headBuffer, null }; //sources是由headBuffer和contents组成的
    
    File[] tempDirs = ioManager.getSpillingDirectories();
    this.tempDir = tempDirs[DIRECTORY_INDEX.getAndIncrement() % tempDirs.length];
    
    byte[] rndBytes = new byte[32];
    new Random().nextBytes(rndBytes);
    this.spillFilePrefix = StringUtils.byteToHexString(rndBytes) + '.';
    
    // prepare for first contents
    createSpillingChannel();
}

private void createSpillingChannel() throws IOException { //打开用于写buffer的文件
    currentSpillFile = new File(tempDir, spillFilePrefix + (fileCounter++) +".buffer");
    currentChannel = new RandomAccessFile(currentSpillFile, "rw").getChannel();
}

/**
 * Adds a buffer or event to the sequence of spilled buffers and events.
 * 
 * @param boe The buffer or event to add and spill.
 * @throws IOException Thrown, if the buffer of event could not be spilled.
 */
public void add(BufferOrEvent boe) throws IOException {
    hasWritten = true;
    try {
        ByteBuffer contents;
        if (boe.isBuffer()) { //分为buffer或event来提取contents
            Buffer buf = boe.getBuffer();
            contents = buf.getMemorySegment().wrap(0, buf.getSize());
        }
        else {
            contents = EventSerializer.toSerializedEvent(boe.getEvent());
        }
        
        headBuffer.clear(); //更新headBuffer
        headBuffer.putInt(boe.getChannelIndex());
        headBuffer.putInt(contents.remaining());
        headBuffer.put((byte) (boe.isBuffer() ? 0 : 1));
        headBuffer.flip();
        
        sources[1] = contents; //为什么加在1，因为0是headBuffer
        currentChannel.write(sources); //写入文件
    }
    finally {
        if (boe.isBuffer()) {
            boe.getBuffer().recycle();
        }
    }
}

private SpilledBufferOrEventSequence rollOverInternal(boolean newBuffer) throws IOException {
    if (!hasWritten) {
        return null;
    }
    
    ByteBuffer buf;
    if (newBuffer) { //newBuffer的区别是，是否重新创建ByteBuffer还是直接用readBuffer
        buf = ByteBuffer.allocateDirect(READ_BUFFER_SIZE);
        buf.order(ByteOrder.LITTLE_ENDIAN);
    } else {
        buf = readBuffer;
    }
    
    // create a reader for the spilled data
    currentChannel.position(0L);
    SpilledBufferOrEventSequence seq = 
            new SpilledBufferOrEventSequence(currentSpillFile, currentChannel, buf, pageSize); //把当前的spill文件封装成SpilledBufferOrEventSequence
    
    // create ourselves a new spill file
    createSpillingChannel(); //生成新的spill文件
    
    hasWritten = false;
    return seq;
}

/**
 * This class represents a sequence of spilled buffers and events, created by the
 * {@link BufferSpiller}. The sequence of buffers and events can be read back using the
 * method {@link #getNext()}.
 */

public static class SpilledBufferOrEventSequence {


    /**
     * Gets the next BufferOrEvent from the spilled sequence, or {@code null}, if the
     * sequence is exhausted.
     *         
     * @return The next BufferOrEvent from the spilled sequence, or {@code null} (end of sequence).
     * @throws IOException Thrown, if the reads failed, of if the byte stream is corrupt.
     */
    public BufferOrEvent getNext() throws IOException {
        if (buffer.remaining() < HEADER_LENGTH) {
            buffer.compact();
            
            while (buffer.position() < HEADER_LENGTH) {
                if (fileChannel.read(buffer) == -1) { //从文件channel你们把数据读到buffer中
                    if (buffer.position() == 0) {
                        // no trailing data
                        return null;
                    } else {
                        throw new IOException("Found trailing incomplete buffer or event");
                    }
                }
            }
            
            buffer.flip();
        }
        
        final int channel = buffer.getInt();
        final int length = buffer.getInt();
        final boolean isBuffer = buffer.get() == 0;
        
        
        if (isBuffer) { //如果是buffer
            // deserialize buffer

            MemorySegment seg = MemorySegmentFactory.allocateUnpooledSegment(pageSize); //创建 MemorySegment，这里是allocate unpooled的segment
            
            int segPos = 0;
            int bytesRemaining = length;
            
            while (true) {
                int toCopy = Math.min(buffer.remaining(), bytesRemaining);
                if (toCopy > 0) {
                    seg.put(segPos, buffer, toCopy); //将buffer中的数据写入MemorySegment
                    segPos += toCopy;
                    bytesRemaining -= toCopy;
                }
                
                if (bytesRemaining == 0) {
                    break;
                }
                else {
                    buffer.clear();
                    if (fileChannel.read(buffer) == -1) {
                        throw new IOException("Found trailing incomplete buffer");
                    }
                    buffer.flip();
                }
            }
            
            
            Buffer buf = new Buffer(seg, FreeingBufferRecycler.INSTANCE); //将MemorySegment封装成Buffer
            buf.setSize(length);
            
            return new BufferOrEvent(buf, channel);
        }
        else { //如果是event
            // deserialize event
            if (buffer.remaining() < length) {
                buffer.compact();

                while (buffer.position() < length) {
                    if (fileChannel.read(buffer) == -1) {
                        throw new IOException("Found trailing incomplete event");
                    }
                }

                buffer.flip();
            }

            int oldLimit = buffer.limit();
            buffer.limit(buffer.position() + length);
            AbstractEvent evt = EventSerializer.fromSerializedEvent(buffer, getClass().getClassLoader()); //将buffer封装成event
            buffer.limit(oldLimit);
            
            return new BufferOrEvent(evt, channel);
        }
    }
}

/**
 * The BarrierTracker keeps track of what checkpoint barriers have been received from
 * which input channels. Once it has observed all checkpoint barriers for a checkpoint ID,
 * it notifies its listener of a completed checkpoint.
 * 
 * <p>Unlike the {@link BarrierBuffer}, the BarrierTracker does not block the input
 * channels that have sent barriers, so it cannot be used to gain "exactly-once" processing
 * guarantees. It can, however, be used to gain "at least once" processing guarantees.</p>
 * 
 * <p>NOTE: This implementation strictly assumes that newer checkpoints have higher checkpoint IDs.</p>
 */
public class BarrierTracker implements CheckpointBarrierHandler {

    @Override
    public BufferOrEvent getNextNonBlocked() throws IOException, InterruptedException {
        while (true) {
            BufferOrEvent next = inputGate.getNextBufferOrEvent();
            if (next == null) {
                return null;
            }
            else if (next.isBuffer() || next.getEvent().getClass() != CheckpointBarrier.class) { //如果是数据就直接返回
                return next;
            }
            else {
                processBarrier((CheckpointBarrier) next.getEvent()); //如果是barrier就处理
            }
        }
    }

    private void processBarrier(CheckpointBarrier receivedBarrier) {
        
        // general path for multiple input channels
        final long barrierId = receivedBarrier.getId();

        // find the checkpoint barrier in the queue of bending barriers
        CheckpointBarrierCount cbc = null;
        int pos = 0;
        
        for (CheckpointBarrierCount next : pendingCheckpoints) { //找找看，这个barrier是否直接收到过
            if (next.checkpointId == barrierId) {
                cbc = next;
                break;
            }
            pos++;
        }
        
        if (cbc != null) { //如果收到过
            // add one to the count to that barrier and check for completion
            int numBarriersNew = cbc.incrementBarrierCount(); //计数加一
            if (numBarriersNew == totalNumberOfInputChannels) { //判断是否所有的barrier已经到全了
                // checkpoint can be triggered
                // first, remove this checkpoint and all all prior pending
                // checkpoints (which are now subsumed)
                for (int i = 0; i <= pos; i++) {
                    pendingCheckpoints.pollFirst(); //当一个checkpoint被触发时，prior的所有checkpoint就已经过期了，也一起remove掉
                }
                
                // notify the listener
                if (checkpointHandler != null) {
                    checkpointHandler.onEvent(receivedBarrier); //如果有checkpoint handler，就调用进行check point
                }
            }
        }
        else { //新的barrier
            // first barrier for that checkpoint ID
            // add it only if it is newer than the latest checkpoint.
            // if it is not newer than the latest checkpoint ID, then there cannot be a
            // successful checkpoint for that ID anyways
            if (barrierId > latestPendingCheckpointID) {
                latestPendingCheckpointID = barrierId;
                pendingCheckpoints.addLast(new CheckpointBarrierCount(barrierId));
                
                // make sure we do not track too many checkpoints
                if (pendingCheckpoints.size() > MAX_CHECKPOINTS_TO_TRACK) {
                    pendingCheckpoints.pollFirst(); //删除过多的checkpoints
                }
            }
        }
    }
}