You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/02/28 16:17:49 UTC
[3/7] flink git commit: [FLINK-8760][runtime] Correctly propagate
moreAvailable flag through SingleInputGate
[FLINK-8760][runtime] Correctly propagate moreAvailable flag through SingleInputGate
Previously if we SingleInputGate was re-eqnqueuing an input channel, isMoreAvailable
might incorrectly return false. This might caused some dead locks.
(cherry picked from commit 6c9e267)
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/651462e6
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/651462e6
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/651462e6
Branch: refs/heads/release-1.5
Commit: 651462e6b22c51ce14bd9ea6db389ef6a1f38e55
Parents: 6b7a448
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Fri Feb 23 11:20:21 2018 +0100
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Wed Feb 28 17:16:55 2018 +0100
----------------------------------------------------------------------
.../partition/consumer/BufferOrEvent.java | 6 +-
.../partition/consumer/SingleInputGate.java | 1 +
.../partition/consumer/UnionInputGate.java | 9 ++-
.../partition/consumer/SingleInputGateTest.java | 62 ++++++++++++++++----
.../partition/consumer/TestInputChannel.java | 14 ++++-
.../partition/consumer/UnionInputGateTest.java | 33 ++++++-----
6 files changed, 92 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/651462e6/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java
index 3e93ae6..d1da438 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java
@@ -39,7 +39,7 @@ public class BufferOrEvent {
* This is not needed outside of the input gate unioning logic and cannot
* be set outside of the consumer package.
*/
- private final boolean moreAvailable;
+ private boolean moreAvailable;
private int channelIndex;
@@ -99,4 +99,8 @@ public class BufferOrEvent {
return String.format("BufferOrEvent [%s, channelIndex = %d]",
isBuffer() ? buffer : event, channelIndex);
}
+
+ public void setMoreAvailable(boolean moreAvailable) {
+ this.moreAvailable = moreAvailable;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/651462e6/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
index a1f3cdc..be4035c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
@@ -540,6 +540,7 @@ public class SingleInputGate implements InputGate {
// will come for that channel
if (result.get().moreAvailable()) {
queueChannel(currentChannel);
+ moreAvailable = true;
}
final Buffer buffer = result.get().buffer();
http://git-wip-us.apache.org/repos/asf/flink/blob/651462e6/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
index 481599c..393e087 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
@@ -181,6 +181,7 @@ public class UnionInputGate implements InputGate, InputGateListener {
final int channelIndexOffset = inputGateToIndexOffsetMap.get(inputGate);
bufferOrEvent.setChannelIndex(channelIndexOffset + bufferOrEvent.getChannelIndex());
+ bufferOrEvent.setMoreAvailable(bufferOrEvent.moreAvailable() || inputGateWithData.moreInputGatesAvailable);
return Optional.ofNullable(bufferOrEvent);
}
@@ -193,18 +194,20 @@ public class UnionInputGate implements InputGate, InputGateListener {
private InputGateWithData waitAndGetNextInputGate() throws IOException, InterruptedException {
while (true) {
InputGate inputGate;
+ boolean moreInputGatesAvailable;
synchronized (inputGatesWithData) {
while (inputGatesWithData.size() == 0) {
inputGatesWithData.wait();
}
inputGate = inputGatesWithData.remove();
enqueuedInputGatesWithData.remove(inputGate);
+ moreInputGatesAvailable = enqueuedInputGatesWithData.size() > 0;
}
// In case of inputGatesWithData being inaccurate do not block on an empty inputGate, but just poll the data.
Optional<BufferOrEvent> bufferOrEvent = inputGate.pollNextBufferOrEvent();
if (bufferOrEvent.isPresent()) {
- return new InputGateWithData(inputGate, bufferOrEvent.get());
+ return new InputGateWithData(inputGate, bufferOrEvent.get(), moreInputGatesAvailable);
}
}
}
@@ -212,10 +215,12 @@ public class UnionInputGate implements InputGate, InputGateListener {
private static class InputGateWithData {
private final InputGate inputGate;
private final BufferOrEvent bufferOrEvent;
+ private final boolean moreInputGatesAvailable;
- public InputGateWithData(InputGate inputGate, BufferOrEvent bufferOrEvent) {
+ public InputGateWithData(InputGate inputGate, BufferOrEvent bufferOrEvent, boolean moreInputGatesAvailable) {
this.inputGate = checkNotNull(inputGate);
this.bufferOrEvent = checkNotNull(bufferOrEvent);
+ this.moreInputGatesAvailable = moreInputGatesAvailable;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/651462e6/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
index e94411d..8c54c1f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
@@ -100,15 +100,40 @@ public class SingleInputGateTest {
inputGate.notifyChannelNonEmpty(inputChannels[0].getInputChannel());
inputGate.notifyChannelNonEmpty(inputChannels[1].getInputChannel());
- verifyBufferOrEvent(inputGate, true, 0);
- verifyBufferOrEvent(inputGate, true, 1);
- verifyBufferOrEvent(inputGate, true, 0);
- verifyBufferOrEvent(inputGate, false, 1);
- verifyBufferOrEvent(inputGate, false, 0);
+ verifyBufferOrEvent(inputGate, true, 0, true);
+ verifyBufferOrEvent(inputGate, true, 1, true);
+ verifyBufferOrEvent(inputGate, true, 0, true);
+ verifyBufferOrEvent(inputGate, false, 1, true);
+ verifyBufferOrEvent(inputGate, false, 0, false);
// Return null when the input gate has received all end-of-partition events
assertTrue(inputGate.isFinished());
- assertFalse(inputGate.getNextBufferOrEvent().isPresent());
+ }
+
+ @Test(timeout = 120 * 1000)
+ public void testIsMoreAvailableReadingFromSingleInputChannel() throws Exception {
+ // Setup
+ final SingleInputGate inputGate = createInputGate();
+
+ final TestInputChannel[] inputChannels = new TestInputChannel[]{
+ new TestInputChannel(inputGate, 0),
+ new TestInputChannel(inputGate, 1)
+ };
+
+ inputGate.setInputChannel(
+ new IntermediateResultPartitionID(), inputChannels[0].getInputChannel());
+
+ inputGate.setInputChannel(
+ new IntermediateResultPartitionID(), inputChannels[1].getInputChannel());
+
+ // Test
+ inputChannels[0].readBuffer();
+ inputChannels[0].readBuffer(false);
+
+ inputGate.notifyChannelNonEmpty(inputChannels[0].getInputChannel());
+
+ verifyBufferOrEvent(inputGate, true, 0, true);
+ verifyBufferOrEvent(inputGate, true, 0, false);
}
@Test
@@ -428,13 +453,28 @@ public class SingleInputGateTest {
}
static void verifyBufferOrEvent(
- InputGate inputGate,
- boolean isBuffer,
- int channelIndex) throws IOException, InterruptedException {
+ InputGate inputGate,
+ boolean expectedIsBuffer,
+ int expectedChannelIndex,
+ boolean expectedMoreAvailable) throws IOException, InterruptedException {
final Optional<BufferOrEvent> bufferOrEvent = inputGate.getNextBufferOrEvent();
assertTrue(bufferOrEvent.isPresent());
- assertEquals(isBuffer, bufferOrEvent.get().isBuffer());
- assertEquals(channelIndex, bufferOrEvent.get().getChannelIndex());
+ assertEquals(expectedIsBuffer, bufferOrEvent.get().isBuffer());
+ assertEquals(expectedChannelIndex, bufferOrEvent.get().getChannelIndex());
+ assertEquals(expectedMoreAvailable, bufferOrEvent.get().moreAvailable());
+ if (!expectedMoreAvailable) {
+ try {
+ assertFalse(inputGate.pollNextBufferOrEvent().isPresent());
+ }
+ catch (UnsupportedOperationException ex) {
+ /**
+ * {@link UnionInputGate#pollNextBufferOrEvent()} is unsupported at the moment.
+ */
+ if (!(inputGate instanceof UnionInputGate)) {
+ throw ex;
+ }
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/651462e6/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java
index f9060f3..3ae3a8a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java
@@ -56,20 +56,28 @@ public class TestInputChannel {
}
public TestInputChannel read(Buffer buffer) throws IOException, InterruptedException {
+ return read(buffer, true);
+ }
+
+ public TestInputChannel read(Buffer buffer, boolean moreAvailable) throws IOException, InterruptedException {
if (stubbing == null) {
- stubbing = when(mock.getNextBuffer()).thenReturn(Optional.of(new BufferAndAvailability(buffer, true, 0)));
+ stubbing = when(mock.getNextBuffer()).thenReturn(Optional.of(new BufferAndAvailability(buffer, moreAvailable, 0)));
} else {
- stubbing = stubbing.thenReturn(Optional.of(new BufferAndAvailability(buffer, true, 0)));
+ stubbing = stubbing.thenReturn(Optional.of(new BufferAndAvailability(buffer, moreAvailable, 0)));
}
return this;
}
public TestInputChannel readBuffer() throws IOException, InterruptedException {
+ return readBuffer(true);
+ }
+
+ public TestInputChannel readBuffer(boolean moreAvailable) throws IOException, InterruptedException {
final Buffer buffer = mock(Buffer.class);
when(buffer.isBuffer()).thenReturn(true);
- return read(buffer);
+ return read(buffer, moreAvailable);
}
public TestInputChannel readEndOfPartitionEvent() throws IOException, InterruptedException {
http://git-wip-us.apache.org/repos/asf/flink/blob/651462e6/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
index 9b16471..912cd5b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.runtime.taskmanager.TaskActions;
import org.junit.Test;
+import static org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateTest.verifyBufferOrEvent;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -93,22 +94,22 @@ public class UnionInputGateTest {
ig2.notifyChannelNonEmpty(inputChannels[1][3].getInputChannel());
ig2.notifyChannelNonEmpty(inputChannels[1][4].getInputChannel());
- SingleInputGateTest.verifyBufferOrEvent(union, true, 0); // gate 1, channel 0
- SingleInputGateTest.verifyBufferOrEvent(union, true, 3); // gate 2, channel 0
- SingleInputGateTest.verifyBufferOrEvent(union, true, 1); // gate 1, channel 1
- SingleInputGateTest.verifyBufferOrEvent(union, true, 4); // gate 2, channel 1
- SingleInputGateTest.verifyBufferOrEvent(union, true, 2); // gate 1, channel 2
- SingleInputGateTest.verifyBufferOrEvent(union, true, 5); // gate 2, channel 1
- SingleInputGateTest.verifyBufferOrEvent(union, false, 0); // gate 1, channel 0
- SingleInputGateTest.verifyBufferOrEvent(union, true, 6); // gate 2, channel 1
- SingleInputGateTest.verifyBufferOrEvent(union, false, 1); // gate 1, channel 1
- SingleInputGateTest.verifyBufferOrEvent(union, true, 7); // gate 2, channel 1
- SingleInputGateTest.verifyBufferOrEvent(union, false, 2); // gate 1, channel 2
- SingleInputGateTest.verifyBufferOrEvent(union, false, 3); // gate 2, channel 0
- SingleInputGateTest.verifyBufferOrEvent(union, false, 4); // gate 2, channel 1
- SingleInputGateTest.verifyBufferOrEvent(union, false, 5); // gate 2, channel 2
- SingleInputGateTest.verifyBufferOrEvent(union, false, 6); // gate 2, channel 3
- SingleInputGateTest.verifyBufferOrEvent(union, false, 7); // gate 2, channel 4
+ verifyBufferOrEvent(union, true, 0, true); // gate 1, channel 0
+ verifyBufferOrEvent(union, true, 3, true); // gate 2, channel 0
+ verifyBufferOrEvent(union, true, 1, true); // gate 1, channel 1
+ verifyBufferOrEvent(union, true, 4, true); // gate 2, channel 1
+ verifyBufferOrEvent(union, true, 2, true); // gate 1, channel 2
+ verifyBufferOrEvent(union, true, 5, true); // gate 2, channel 1
+ verifyBufferOrEvent(union, false, 0, true); // gate 1, channel 0
+ verifyBufferOrEvent(union, true, 6, true); // gate 2, channel 1
+ verifyBufferOrEvent(union, false, 1, true); // gate 1, channel 1
+ verifyBufferOrEvent(union, true, 7, true); // gate 2, channel 1
+ verifyBufferOrEvent(union, false, 2, true); // gate 1, channel 2
+ verifyBufferOrEvent(union, false, 3, true); // gate 2, channel 0
+ verifyBufferOrEvent(union, false, 4, true); // gate 2, channel 1
+ verifyBufferOrEvent(union, false, 5, true); // gate 2, channel 2
+ verifyBufferOrEvent(union, false, 6, true); // gate 2, channel 3
+ verifyBufferOrEvent(union, false, 7, false); // gate 2, channel 4
// Return null when the input gate has received all end-of-partition events
assertTrue(union.isFinished());