You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/06/07 09:01:56 UTC
[flink] 03/04: [FLINK-22376][runtime] RecoveredChannelStateHandler
requires the ownership of the input Buffer
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 79a7b6db86e68be4aa265e52210046d3265149ad
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Tue Jun 1 17:08:47 2021 +0300
[FLINK-22376][runtime] RecoveredChannelStateHandler requires the ownership of the input Buffer
(cherry picked from commit 2293362f0a144f47e182039c26c0d7047f87232a)
---
.../InflightDataRescalingDescriptor.java | 3 +-
.../channel/RecoveredChannelStateHandler.java | 83 ++++++++-------
.../channel/SequentialChannelStateReaderImpl.java | 18 ++--
.../channel/ChannelStateChunkReaderTest.java | 23 ++++-
.../InputChannelRecoveredStateHandlerTest.java | 111 +++++++++++++++++++++
.../channel/RecoveredChannelStateHandlerTest.java | 34 +++++++
...esultSubpartitionRecoveredStateHandlerTest.java | 107 ++++++++++++++++++++
7 files changed, 330 insertions(+), 49 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/InflightDataRescalingDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/InflightDataRescalingDescriptor.java
index 3a1ab27..53d4792 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/InflightDataRescalingDescriptor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/InflightDataRescalingDescriptor.java
@@ -128,7 +128,8 @@ public class InflightDataRescalingDescriptor implements Serializable {
private final MappingType mappingType;
- enum MappingType {
+ /** Type of mapping which should be used for this in-flight data. */
+ public enum MappingType {
IDENTITY,
RESCALING
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
index 86894c9..4a724a7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
@@ -17,7 +17,6 @@
package org.apache.flink.runtime.checkpoint.channel;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptor;
import org.apache.flink.runtime.checkpoint.RescaleMappings;
import org.apache.flink.runtime.io.network.api.SubtaskConnectionDescriptor;
@@ -51,11 +50,21 @@ interface RecoveredChannelStateHandler<Info, Context> extends AutoCloseable {
this.buffer = buffer;
this.context = context;
}
+
+ public void close() {
+ buffer.close();
+ }
}
BufferWithContext<Context> getBuffer(Info info) throws IOException, InterruptedException;
- void recover(Info info, int oldSubtaskIndex, Context context) throws IOException;
+ /**
+ * Recover the data from buffer. This method is taking over the ownership of the
+ * bufferWithContext and is fully responsible for cleaning it up both on the happy path and in
+ * case of an error.
+ */
+ void recover(Info info, int oldSubtaskIndex, BufferWithContext<Context> bufferWithContext)
+ throws IOException;
}
class InputChannelRecoveredStateHandler
@@ -84,8 +93,12 @@ class InputChannelRecoveredStateHandler
}
@Override
- public void recover(InputChannelInfo channelInfo, int oldSubtaskIndex, Buffer buffer)
+ public void recover(
+ InputChannelInfo channelInfo,
+ int oldSubtaskIndex,
+ BufferWithContext<Buffer> bufferWithContext)
throws IOException {
+ Buffer buffer = bufferWithContext.context;
try {
if (buffer.readableBytes() > 0) {
for (final RecoveredInputChannel channel : getMappedChannels(channelInfo)) {
@@ -143,8 +156,7 @@ class InputChannelRecoveredStateHandler
}
class ResultSubpartitionRecoveredStateHandler
- implements RecoveredChannelStateHandler<
- ResultSubpartitionInfo, Tuple2<BufferBuilder, BufferConsumer>> {
+ implements RecoveredChannelStateHandler<ResultSubpartitionInfo, BufferBuilder> {
private final ResultPartitionWriter[] writers;
private final boolean notifyAndBlockOnCompletion;
@@ -165,51 +177,48 @@ class ResultSubpartitionRecoveredStateHandler
}
@Override
- public BufferWithContext<Tuple2<BufferBuilder, BufferConsumer>> getBuffer(
- ResultSubpartitionInfo subpartitionInfo) throws IOException, InterruptedException {
+ public BufferWithContext<BufferBuilder> getBuffer(ResultSubpartitionInfo subpartitionInfo)
+ throws IOException, InterruptedException {
// request the buffer from any mapped subpartition as they all will receive the same buffer
final List<CheckpointedResultSubpartition> channels = getMappedChannels(subpartitionInfo);
BufferBuilder bufferBuilder = channels.get(0).requestBufferBuilderBlocking();
- return new BufferWithContext<>(
- wrap(bufferBuilder),
- Tuple2.of(bufferBuilder, bufferBuilder.createBufferConsumer()));
+ return new BufferWithContext<>(wrap(bufferBuilder), bufferBuilder);
}
@Override
public void recover(
ResultSubpartitionInfo subpartitionInfo,
int oldSubtaskIndex,
- Tuple2<BufferBuilder, BufferConsumer> bufferBuilderAndConsumer)
+ BufferWithContext<BufferBuilder> bufferWithContext)
throws IOException {
- try {
- bufferBuilderAndConsumer.f0.finish();
- if (bufferBuilderAndConsumer.f1.isDataAvailable()) {
- NetworkActionsLogger.traceRecover(
- "ResultSubpartitionRecoveredStateHandler#recover",
- bufferBuilderAndConsumer.f1,
- subpartitionInfo);
- final List<CheckpointedResultSubpartition> channels =
- getMappedChannels(subpartitionInfo);
- for (final CheckpointedResultSubpartition channel : channels) {
- // channel selector is created from the downstream's point of view: the subtask
- // of
- // downstream = subpartition index of recovered buffer
- final SubtaskConnectionDescriptor channelSelector =
- new SubtaskConnectionDescriptor(
- subpartitionInfo.getSubPartitionIdx(), oldSubtaskIndex);
- channel.add(
- EventSerializer.toBufferConsumer(channelSelector, false),
- Integer.MIN_VALUE);
- boolean added =
- channel.add(bufferBuilderAndConsumer.f1.copy(), Integer.MIN_VALUE);
- if (!added) {
- throw new IOException(
- "Buffer consumer couldn't be added to ResultSubpartition");
+ try (BufferBuilder bufferBuilder = bufferWithContext.context) {
+ try (BufferConsumer bufferConsumer =
+ bufferBuilder.createBufferConsumerFromBeginning()) {
+ bufferBuilder.finish();
+ if (bufferConsumer.isDataAvailable()) {
+ NetworkActionsLogger.traceRecover(
+ "ResultSubpartitionRecoveredStateHandler#recover",
+ bufferConsumer,
+ subpartitionInfo);
+ final List<CheckpointedResultSubpartition> channels =
+ getMappedChannels(subpartitionInfo);
+ for (final CheckpointedResultSubpartition channel : channels) {
+ // channel selector is created from the downstream's point of view: the
+ // subtask of downstream = subpartition index of recovered buffer
+ final SubtaskConnectionDescriptor channelSelector =
+ new SubtaskConnectionDescriptor(
+ subpartitionInfo.getSubPartitionIdx(), oldSubtaskIndex);
+ channel.add(
+ EventSerializer.toBufferConsumer(channelSelector, false),
+ Integer.MIN_VALUE);
+ boolean added = channel.add(bufferConsumer.copy(), Integer.MIN_VALUE);
+ if (!added) {
+ throw new IOException(
+ "Buffer consumer couldn't be added to ResultSubpartition");
+ }
}
}
}
- } finally {
- bufferBuilderAndConsumer.f1.close();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
index 8329c71..88b23c3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
@@ -196,19 +196,19 @@ class ChannelStateChunkReader {
while (length > 0) {
RecoveredChannelStateHandler.BufferWithContext<Context> bufferWithContext =
stateHandler.getBuffer(channelInfo);
- try {
- try (Closeable ignored =
- NetworkActionsLogger.measureIO(
- "ChannelStateChunkReader#readChunk", bufferWithContext.buffer)) {
- while (length > 0 && bufferWithContext.buffer.isWritable()) {
- length -= serializer.readData(source, bufferWithContext.buffer, length);
- }
+ try (Closeable ignored =
+ NetworkActionsLogger.measureIO(
+ "ChannelStateChunkReader#readChunk", bufferWithContext.buffer)) {
+ while (length > 0 && bufferWithContext.buffer.isWritable()) {
+ length -= serializer.readData(source, bufferWithContext.buffer, length);
}
- stateHandler.recover(channelInfo, oldSubtaskIndex, bufferWithContext.context);
} catch (Exception e) {
- bufferWithContext.buffer.recycle();
+ bufferWithContext.close();
throw e;
}
+
+ // Passing the ownership of buffer to inside.
+ stateHandler.recover(channelInfo, oldSubtaskIndex, bufferWithContext);
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateChunkReaderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateChunkReaderTest.java
index c163d3d..0e88489 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateChunkReaderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateChunkReaderTest.java
@@ -59,6 +59,22 @@ public class ChannelStateChunkReaderTest {
}
@Test
+ public void testBufferRecycledOnSuccess() throws IOException, InterruptedException {
+ ChannelStateSerializer serializer = new ChannelStateSerializerImpl();
+ TestRecoveredChannelStateHandler handler = new TestRecoveredChannelStateHandler();
+
+ try (FSDataInputStream stream = getStream(serializer, 10)) {
+ new ChannelStateChunkReader(serializer)
+ .readChunk(stream, serializer.getHeaderLength(), handler, "channelInfo", 0);
+ } finally {
+ checkState(!handler.requestedBuffers.isEmpty());
+ assertTrue(
+ handler.requestedBuffers.stream()
+ .allMatch(TestChannelStateByteBuffer::isRecycled));
+ }
+ }
+
+ @Test
public void testBuffersNotRequestedForEmptyStream() throws IOException, InterruptedException {
ChannelStateSerializer serializer = new ChannelStateSerializerImpl();
TestRecoveredChannelStateHandler handler = new TestRecoveredChannelStateHandler();
@@ -109,7 +125,10 @@ public class ChannelStateChunkReaderTest {
}
@Override
- public void recover(Object o, int oldSubtaskIndex, Object o2) {}
+ public void recover(
+ Object o, int oldSubtaskIndex, BufferWithContext<Object> bufferWithContext) {
+ bufferWithContext.close();
+ }
@Override
public void close() throws Exception {}
@@ -151,7 +170,7 @@ public class ChannelStateChunkReaderTest {
}
@Override
- public void recycle() {
+ public void close() {
checkArgument(!recycled);
recycled = true;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java
new file mode 100644
index 0000000..d499b90
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.runtime.checkpoint.channel;
+
+import org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptor;
+import org.apache.flink.runtime.checkpoint.RescaleMappings;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannelBuilder;
+import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateBuilder;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashSet;
+
+import static org.junit.Assert.assertEquals;
+
+/** Test of different implementation of {@link InputChannelRecoveredStateHandler}. */
+public class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandlerTest {
+ private static final int preAllocatedSegments = 3;
+ private NetworkBufferPool networkBufferPool;
+ private SingleInputGate inputGate;
+ private InputChannelRecoveredStateHandler icsHander;
+ private InputChannelInfo channelInfo;
+
+ @Before
+ public void setUp() {
+ // given: Segment provider with defined number of allocated segments.
+ networkBufferPool = new NetworkBufferPool(preAllocatedSegments, 1024);
+
+ // and: Configured input gate with recovered channel.
+ inputGate =
+ new SingleInputGateBuilder()
+ .setChannelFactory(InputChannelBuilder::buildLocalRecoveredChannel)
+ .setSegmentProvider(networkBufferPool)
+ .build();
+
+ icsHander = buildInputChannelStateHandler(inputGate);
+
+ channelInfo = new InputChannelInfo(0, 0);
+ }
+
+ private InputChannelRecoveredStateHandler buildInputChannelStateHandler(
+ SingleInputGate inputGate) {
+ return new InputChannelRecoveredStateHandler(
+ new InputGate[] {inputGate},
+ new InflightDataRescalingDescriptor(
+ new InflightDataRescalingDescriptor
+ .InflightDataGateOrPartitionRescalingDescriptor[] {
+ new InflightDataRescalingDescriptor
+ .InflightDataGateOrPartitionRescalingDescriptor(
+ new int[] {1},
+ RescaleMappings.identity(1, 1),
+ new HashSet<>(),
+ InflightDataRescalingDescriptor
+ .InflightDataGateOrPartitionRescalingDescriptor
+ .MappingType.IDENTITY)
+ }));
+ }
+
+ @Test
+ public void testRecycleBufferBeforeRecoverWasCalled() throws Exception {
+ // when: Request the buffer.
+ RecoveredChannelStateHandler.BufferWithContext<Buffer> bufferWithContext =
+ icsHander.getBuffer(channelInfo);
+
+ // and: Recycle buffer outside.
+ bufferWithContext.buffer.close();
+
+ // Close the gate for flushing the cached recycled buffers to the segment provider.
+ inputGate.close();
+
+ // then: All pre-allocated segments should be successfully recycled.
+ assertEquals(preAllocatedSegments, networkBufferPool.getNumberOfAvailableMemorySegments());
+ }
+
+ @Test
+ public void testRecycleBufferAfterRecoverWasCalled() throws Exception {
+ // when: Request the buffer.
+ RecoveredChannelStateHandler.BufferWithContext<Buffer> bufferWithContext =
+ icsHander.getBuffer(channelInfo);
+
+ // and: Recycle buffer outside.
+ icsHander.recover(channelInfo, 0, bufferWithContext);
+
+ // Close the gate for flushing the cached recycled buffers to the segment provider.
+ inputGate.close();
+
+ // then: All pre-allocated segments should be successfully recycled.
+ assertEquals(preAllocatedSegments, networkBufferPool.getNumberOfAvailableMemorySegments());
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandlerTest.java
new file mode 100644
index 0000000..a74c128
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandlerTest.java
@@ -0,0 +1,34 @@
+/*
+ * 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.runtime.checkpoint.channel;
+
+import org.junit.Test;
+
+/**
+ * Base class which contains all tests which should be implemented for every implementation of
+ * {@link InputChannelRecoveredStateHandler}.
+ */
+public abstract class RecoveredChannelStateHandlerTest {
+
+ @Test
+ public abstract void testRecycleBufferBeforeRecoverWasCalled() throws Exception;
+
+ @Test
+ public abstract void testRecycleBufferAfterRecoverWasCalled() throws Exception;
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ResultSubpartitionRecoveredStateHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ResultSubpartitionRecoveredStateHandlerTest.java
new file mode 100644
index 0000000..b188927
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ResultSubpartitionRecoveredStateHandlerTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.runtime.checkpoint.channel;
+
+import org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptor;
+import org.apache.flink.runtime.checkpoint.RescaleMappings;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
+import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.ResultPartition;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionBuilder;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashSet;
+
+import static org.junit.Assert.assertEquals;
+
+/** Test of different implementation of {@link ResultSubpartitionRecoveredStateHandler}. */
+public class ResultSubpartitionRecoveredStateHandlerTest extends RecoveredChannelStateHandlerTest {
+ private static final int preAllocatedSegments = 3;
+ private NetworkBufferPool networkBufferPool;
+ private ResultPartition partition;
+ private ResultSubpartitionRecoveredStateHandler rstHandler;
+ private ResultSubpartitionInfo channelInfo;
+
+ @Before
+ public void setUp() throws IOException {
+ // given: Segment provider with defined number of allocated segments.
+ channelInfo = new ResultSubpartitionInfo(0, 0);
+
+ networkBufferPool = new NetworkBufferPool(preAllocatedSegments, 1024);
+ partition = new ResultPartitionBuilder().setNetworkBufferPool(networkBufferPool).build();
+ partition.setup();
+
+ rstHandler = buildResultStateHandler(partition);
+ }
+
+ private ResultSubpartitionRecoveredStateHandler buildResultStateHandler(
+ ResultPartition partition) {
+ return new ResultSubpartitionRecoveredStateHandler(
+ new ResultPartitionWriter[] {partition},
+ false,
+ new InflightDataRescalingDescriptor(
+ new InflightDataRescalingDescriptor
+ .InflightDataGateOrPartitionRescalingDescriptor[] {
+ new InflightDataRescalingDescriptor
+ .InflightDataGateOrPartitionRescalingDescriptor(
+ new int[] {1},
+ RescaleMappings.identity(1, 1),
+ new HashSet<>(),
+ InflightDataRescalingDescriptor
+ .InflightDataGateOrPartitionRescalingDescriptor
+ .MappingType.IDENTITY)
+ }));
+ }
+
+ @Test
+ public void testRecycleBufferBeforeRecoverWasCalled() throws Exception {
+ // when: Request the buffer.
+ RecoveredChannelStateHandler.BufferWithContext<BufferBuilder> bufferWithContext =
+ rstHandler.getBuffer(new ResultSubpartitionInfo(0, 0));
+
+ // and: Recycle buffer outside.
+ bufferWithContext.buffer.close();
+
+ // Close the partition for flushing the cached recycled buffers to the segment provider.
+ partition.close();
+
+ // then: All pre-allocated segments should be successfully recycled.
+ assertEquals(preAllocatedSegments, networkBufferPool.getNumberOfAvailableMemorySegments());
+ }
+
+ @Test
+ public void testRecycleBufferAfterRecoverWasCalled() throws Exception {
+ // when: Request the buffer.
+ RecoveredChannelStateHandler.BufferWithContext<BufferBuilder> bufferWithContext =
+ rstHandler.getBuffer(channelInfo);
+
+ // and: Pass the buffer to recovery.
+ rstHandler.recover(channelInfo, 0, bufferWithContext);
+
+ // Close the partition for flushing the cached recycled buffers to the segment provider.
+ partition.close();
+
+ // then: All pre-allocated segments should be successfully recycled.
+ assertEquals(preAllocatedSegments, networkBufferPool.getNumberOfAvailableMemorySegments());
+ }
+}