1996fanrui commented on code in PR #28652:
URL: https://github.com/apache/flink/pull/28652#discussion_r3555412925
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java:
##########
@@ -113,48 +159,211 @@ public LocalInputChannel(
this.partitionManager = checkNotNull(partitionManager);
this.taskEventPublisher = checkNotNull(taskEventPublisher);
- this.channelStatePersister = new ChannelStatePersister(stateWriter,
getChannelInfo());
-
- // Migrate recovered buffers from RecoveredInputChannel if provided.
- // These buffers have been filtered but not yet consumed by the Task.
- if (!initialRecoveredBuffers.isEmpty()) {
- final int expectedCount = initialRecoveredBuffers.size();
- // Sequence number starts at Integer.MIN_VALUE, consistent with
RecoveredInputChannel.
- int seqNum = Integer.MIN_VALUE;
- while (!initialRecoveredBuffers.isEmpty()) {
- Buffer buffer = initialRecoveredBuffers.poll();
- // Determine next data type based on the next buffer in the
queue
- Buffer.DataType nextDataType =
- initialRecoveredBuffers.isEmpty()
- ? Buffer.DataType.NONE
- : initialRecoveredBuffers.peek().getDataType();
- // buffersInBacklog is set to 0 as these are recovered buffers
- BufferAndBacklog bufferAndBacklog =
- new BufferAndBacklog(buffer, 0, nextDataType,
seqNum++);
- toBeConsumedBuffers.add(bufferAndBacklog);
+ this.channelStatePersister =
+ new ChannelStatePersister(checkNotNull(stateWriter),
getChannelInfo());
+ this.inRecovery = needsRecovery;
+ this.bufferManager =
+ needsRecovery
+ ? new
BufferManager(inputGate.getMemorySegmentProvider(), this, 0, true)
+ : null;
+ this.networkBuffersPerChannel = networkBuffersPerChannel;
+ this.needsRecovery = needsRecovery;
+ if (!needsRecovery) {
+ stateConsumedFuture.complete(null);
+ }
+ }
+
+ @Override
+ void setup() throws IOException {
+ if (needsRecovery && networkBuffersPerChannel > 0) {
+ bufferManager.requestExclusiveBuffers(networkBuffersPerChannel);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // RecoverableInputChannel implementation
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void onRecoveredStateBuffer(Buffer buffer) {
+ boolean wasEmpty;
+ synchronized (recoveredBuffers) {
+ if (isReleased) {
+ buffer.recycleBuffer();
+ return;
}
- checkState(
- toBeConsumedBuffers.size() == expectedCount,
- "Buffer migration failed: expected %s buffers but got %s",
- expectedCount,
- toBeConsumedBuffers.size());
+ // Migrate recovered buffers from RecoveredInputChannel. These
buffers have been
+ // filtered but not yet consumed by the Task.
+ wasEmpty = offerRecoveredBuffer(buffer);
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ @Override
+ public void finishRecoveredBufferDelivery() throws IOException {
+ upstreamReady.join();
+ boolean wasEmpty;
+ synchronized (recoveredBuffers) {
+ checkState(inRecovery, "Recovery delivery already finished.");
+ // Append the sentinel after the last recovered buffer. The
consume path flips out of
+ // recovery only once it polls this sentinel, guaranteeing all
recovered buffers are
+ // consumed first.
+ wasEmpty =
+ offerRecoveredBuffer(
+ EventSerializer.toBuffer(
+ EndOfFetchedChannelStateEvent.INSTANCE,
false));
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ @Override
+ public Buffer requestRecoveryBufferBlocking() throws InterruptedException,
IOException {
+ checkState(
+ bufferManager != null,
+ "requestRecoveryBufferBlocking called on a Local channel
constructed with"
+ + " needsRecovery=false");
+ upstreamReady.join();
+ return bufferManager.requestBufferBlocking();
+ }
+
+ @Override
+ public void insertRecoveryCheckpointBarrierIfInRecovery(long checkpointId)
throws IOException {
+ boolean wasEmpty = false;
+ synchronized (recoveredBuffers) {
+ if (!isReleased && inRecovery) {
+ wasEmpty =
+ offerRecoveredBuffer(
+ EventSerializer.toBuffer(
+ new
RecoveryCheckpointBarrier(checkpointId), false));
+ }
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ /**
+ * Flips out of recovery the moment the consume path polls the {@code
+ * EndOfFetchedChannelStateEvent} sentinel, i.e. once all recovered
buffers have been consumed.
+ * Live upstream data may flow again afterwards.
+ */
+ @Override
+ public void onRecoveredStateConsumed() {
+ synchronized (recoveredBuffers) {
+ checkState(inRecovery, "Recovery already finished.");
+ inRecovery = false;
}
+ notifyChannelNonEmpty();
+ stateConsumedFuture.complete(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> getStateConsumedFuture() {
+ return stateConsumedFuture;
+ }
+
+ /**
+ * Appends a recovered buffer (or {@code RecoveryCheckpointBarrier} /
{@code
+ * EndOfFetchedChannelStateEvent} sentinel) to {@link #recoveredBuffers}.
+ *
+ * @return {@code true} iff {@link #recoveredBuffers} transitioned from
empty to non-empty.
+ */
+ private boolean offerRecoveredBuffer(Buffer buffer) {
+ assert Thread.holdsLock(recoveredBuffers);
+ checkState(inRecovery, "Push into recovered buffers after recovery
finished.");
+ boolean wasEmpty = recoveredBuffers.isEmpty();
+ recoveredBuffers.add(buffer);
+ return wasEmpty;
+ }
+
+ private int nextRecoverySequenceNumber() {
+ assert Thread.holdsLock(recoveredBuffers);
+ return recoverySequenceNumber++;
+ }
+
+ /**
+ * Walks {@link #recoveredBuffers} up to the {@link
RecoveryCheckpointBarrier} sentinel matching
+ * {@code checkpointId}, retaining each pre-barrier recovered data buffer
and removing the
+ * sentinel.
+ *
+ * @throws IOException if no sentinel matching {@code checkpointId} is
found (the snapshot
+ * protocol guarantees one must be present while the channel is in
recovery).
+ */
+ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws
IOException {
+ assert Thread.holdsLock(recoveredBuffers);
+ List<Buffer> retained = new ArrayList<>();
+ try {
+ Iterator<Buffer> it = recoveredBuffers.iterator();
+ while (it.hasNext()) {
+ Buffer b = it.next();
+ if (isRecoveryCheckpointBarrier(b, checkpointId)) {
+ it.remove();
+ b.recycleBuffer();
+ return retained;
+ }
+ if (b.isBuffer()) {
+ retained.add(b.retainBuffer());
Review Comment:
Address in 4451e500ab486cc1e2e962636e1b05fa5e5296dd and
6684c478dca58af11c86c088f0a6f4e60aa36c89
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
