AHeise commented on a change in pull request #11507: [FLINK-16587] Add basic CheckpointBarrierHandler for unaligned checkpoint URL: https://github.com/apache/flink/pull/11507#discussion_r405054807
########## File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java ########## @@ -0,0 +1,417 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.streaming.runtime.io; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.checkpoint.CheckpointException; +import org.apache.flink.runtime.checkpoint.CheckpointFailureReason; +import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter; +import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; +import org.apache.flink.runtime.concurrent.FutureUtils; +import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker; +import org.apache.flink.runtime.io.network.api.CheckpointBarrier; +import org.apache.flink.runtime.io.network.buffer.Buffer; +import org.apache.flink.runtime.io.network.buffer.BufferReceivedListener; +import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Arrays; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.function.Function; +import java.util.stream.IntStream; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * {@link CheckpointBarrierUnaligner} is used for triggering checkpoint while reading the first barrier + * and keeping track of the number of received barriers and consumed barriers. + */ +@Internal +public class CheckpointBarrierUnaligner extends CheckpointBarrierHandler { + + private static final Logger LOG = LoggerFactory.getLogger(CheckpointBarrierUnaligner.class); + + private final String taskName; + + /** + * Tag the state of which input channel has read the barrier. If one channel has read the barrier by task, + * the respective in-flight input buffers should be empty when triggering unaligned checkpoint. + */ + private final boolean[] barrierConsumedChannels; + + private int numBarrierConsumed; + + /** + * Contains the offsets of the channel indices for each gate when flattening the channels of all gates. + * + * <p>For example, consider 3 gates with 4 channels, {@code gateChannelOffsets = [0, 4, 8]}. + */ + private final int[] gateChannelOffsets; + + private final InputChannelInfo[] channelInfos; + + /** + * The checkpoint id to guarantee that we would trigger only one checkpoint when reading the same barrier from + * different channels. + * + * <p>Note: this checkpoint is valid in respect to <b>consumed</b> barriers in contrast to + * {@link ThreadSafeUnaligner#currentCheckpointId}. + */ + private long currentCheckpointId = -1L; + + /** Encapsulates state that is shared between netty threads and task thread. */ + private final ThreadSafeUnaligner threadSafeUnaligner; + + CheckpointBarrierUnaligner( + int[] numberOfInputChannelsPerGate, + ChannelStateWriter channelStateWriter, + String taskName, + AbstractInvokable toNotifyOnCheckpoint) { + super(toNotifyOnCheckpoint); + + this.taskName = taskName; + + final int numGates = numberOfInputChannelsPerGate.length; + + gateChannelOffsets = new int[numGates]; + for (int index = 1; index < numGates; index++) { + gateChannelOffsets[index] = gateChannelOffsets[index - 1] + numberOfInputChannelsPerGate[index - 1]; + } + + final int totalNumChannels = gateChannelOffsets[numGates - 1] + numberOfInputChannelsPerGate[numGates - 1]; + barrierConsumedChannels = new boolean[totalNumChannels]; + Arrays.fill(barrierConsumedChannels, true); + + channelInfos = IntStream.range(0, numGates) + .mapToObj(gateIndex -> IntStream.range(0, numberOfInputChannelsPerGate[gateIndex]) + .mapToObj(channelIndex -> new InputChannelInfo(gateIndex, channelIndex))) + .flatMap(Function.identity()) + .toArray(InputChannelInfo[]::new); + + threadSafeUnaligner = new ThreadSafeUnaligner(totalNumChannels, checkNotNull(channelStateWriter), this); + } + + @Override + public void releaseBlocksAndResetBarriers() { + // make sure no additional data is persisted + Arrays.fill(barrierConsumedChannels, true); + // the next barrier that comes must assume it is the first + numBarrierConsumed = 0; + threadSafeUnaligner.releaseBlocksAndResetBarriers(); + } + + /** + * For unaligned checkpoint, it never blocks processing from the task aspect. + * + * <p>For PoC, we do not consider the possibility that the unaligned checkpoint would + * not perform due to the max configured unaligned checkpoint size. + */ + @Override + public boolean isBlocked(int channelIndex) { + return false; + } + + /** + * We still need to trigger checkpoint while reading the first barrier from one channel, because this might happen + * earlier than the previous async trigger via mailbox by netty thread. And the {@link AbstractInvokable} has the + * deduplication logic to guarantee trigger checkpoint only once finally. + * + * <p>Note this is also suitable for the trigger case of local input channel. + */ + @Override + public boolean processBarrier( + CheckpointBarrier receivedBarrier, + int channelIndex, + long bufferedBytes) throws IOException { + long barrierId = receivedBarrier.getId(); + if (currentCheckpointId < barrierId) { + currentCheckpointId = barrierId; + numBarrierConsumed = 0; + Arrays.fill(barrierConsumedChannels, false); + } + if (currentCheckpointId == barrierId) { + barrierConsumedChannels[channelIndex] = true; + numBarrierConsumed++; + } + // processBarrier is called from task thread and can actually happen before notifyBarrierReceived on empty + // buffer queues + // to avoid replicating any logic, we simply call notifyBarrierReceived here as well + threadSafeUnaligner.notifyBarrierReceived(receivedBarrier, channelInfos[channelIndex]); + return false; + } + + @Override + public boolean processCancellationBarrier(CancelCheckpointMarker cancelBarrier) throws Exception { + final long barrierId = cancelBarrier.getCheckpointId(); + + if (numBarrierConsumed > 0) { + // this is only true if some alignment is in progress and nothing was canceled + + if (barrierId == currentCheckpointId) { + // cancel this alignment + if (LOG.isDebugEnabled()) { + LOG.debug("{}: Checkpoint {} canceled, aborting alignment.", taskName, barrierId); + } + + releaseBlocksAndResetBarriers(); + notifyAbortOnCancellationBarrier(barrierId); + } + else if (barrierId > currentCheckpointId) { + // we canceled the next which also cancels the current + LOG.warn("{}: Received cancellation barrier for checkpoint {} before completing current checkpoint {}. " + + "Skipping current checkpoint.", + taskName, + barrierId, + currentCheckpointId); + + // this stops the current alignment + releaseBlocksAndResetBarriers(); + + // the next checkpoint starts as canceled + currentCheckpointId = barrierId; + threadSafeUnaligner.setCurrentCheckpointId(currentCheckpointId); + notifyAbortOnCancellationBarrier(barrierId); + } + + // else: ignore trailing (cancellation) barrier from an earlier checkpoint (obsolete now) + + } + else if (barrierId > currentCheckpointId) { + // first barrier of a new checkpoint is directly a cancellation + + // by setting the currentCheckpointId to this checkpoint while keeping the numBarriers + // at zero means that no checkpoint barrier can start a new alignment + currentCheckpointId = barrierId; + threadSafeUnaligner.setCurrentCheckpointId(currentCheckpointId); + + if (LOG.isDebugEnabled()) { + LOG.debug("{}: Checkpoint {} canceled, skipping alignment.", taskName, barrierId); + } + + notifyAbortOnCancellationBarrier(barrierId); + } + + // else: trailing barrier from either + // - a previous (subsumed) checkpoint + // - the current checkpoint if it was already canceled + return false; + } + + @Override + public boolean processEndOfPartition() throws Exception { + threadSafeUnaligner.processEndOfPartition(); + + if (numBarrierConsumed > 0) { + // let the task know we skip a checkpoint + notifyAbort( + currentCheckpointId, + new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED_INPUT_END_OF_STREAM)); + // no chance to complete this checkpoint + releaseBlocksAndResetBarriers(); + return true; + } + return false; + } + + @Override + public long getLatestCheckpointId() { + return currentCheckpointId; + } + + @Override + public long getAlignmentDurationNanos() { + return 0; + } + + @Override + public String toString() { + return String.format("%s: last checkpoint: %d", taskName, currentCheckpointId); + } + + @Override + public void checkpointSizeLimitExceeded(long maxBufferedBytes) { + } + + @Override + public void close() throws IOException { + super.close(); + threadSafeUnaligner.close(); + } + + @Override + protected boolean isBarrierConsumed(long checkpointId, int channelIndex) { + if (checkpointId < currentCheckpointId) { + return true; + } + if (checkpointId > currentCheckpointId) { + return false; + } + return barrierConsumedChannels[channelIndex]; + } + + public CompletableFuture<Void> getAllBarriersReceivedFuture(long checkpointId) { + return threadSafeUnaligner.getAllBarriersReceivedFuture(checkpointId); + } + + @Override + public Optional<BufferReceivedListener> getBufferReceivedListener() { + return Optional.of(threadSafeUnaligner); + } + + private int getFlattenedChannelIndex(InputChannelInfo channelInfo) { + return gateChannelOffsets[channelInfo.getGateIdx()] + channelInfo.getInputChannelIdx(); + } + + private static class ThreadSafeUnaligner implements BufferReceivedListener, Closeable { Review comment: Annotated and I renamed all members with the same name to make it more clear. I'd like to avoid moving it out, since it's basically two sides of the same coin for me. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
