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/19 14:08:18 UTC
[25/35] flink git commit: [hotfix][tests] Correctly set moreAvailable
flag in StreamTestSingleInputGate and handle redundant data notifications
[hotfix][tests] Correctly set moreAvailable flag in StreamTestSingleInputGate and handle redundant data notifications
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/78df079b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/78df079b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/78df079b
Branch: refs/heads/master
Commit: 78df079bec8e63361f246fc698e0787ba232f083
Parents: c6526fb
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Tue Jan 30 17:03:14 2018 +0100
Committer: Piotr Nowojski <pi...@gmail.com>
Committed: Mon Feb 19 12:21:36 2018 +0100
----------------------------------------------------------------------
.../partition/consumer/SingleInputGate.java | 17 +++++++++++++----
.../consumer/StreamTestSingleInputGate.java | 18 ++++++++++++------
2 files changed, 25 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/78df079b/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 deac251..1175c52 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
@@ -41,6 +41,7 @@ import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.runtime.taskmanager.TaskActions;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -138,6 +139,12 @@ public class SingleInputGate implements InputGate {
/** Channels, which notified this input gate about available data. */
private final ArrayDeque<InputChannel> inputChannelsWithData = new ArrayDeque<>();
+ /**
+ * Field guaranteeing uniqueness for inputChannelsWithData queue. Both of those fields should be unified
+ * onto one.
+ */
+ private final BitSet enqueuedInputChannelsWithData = new BitSet();
+
private final BitSet channelsWithEndOfPartitionEvents;
/** The partition state listener listening to failed partition requests. */
@@ -284,7 +291,7 @@ public class SingleInputGate implements InputGate {
this.networkBufferPool = checkNotNull(networkBufferPool);
this.networkBuffersPerChannel = networkBuffersPerChannel;
-
+
synchronized (requestLock) {
for (InputChannel inputChannel : inputChannels.values()) {
if (inputChannel instanceof RemoteInputChannel) {
@@ -492,7 +499,6 @@ public class SingleInputGate implements InputGate {
InputChannel currentChannel;
boolean moreAvailable;
-
synchronized (inputChannelsWithData) {
while (inputChannelsWithData.size() == 0) {
if (isReleased) {
@@ -503,17 +509,16 @@ public class SingleInputGate implements InputGate {
}
currentChannel = inputChannelsWithData.remove();
+ enqueuedInputChannelsWithData.clear(currentChannel.getChannelIndex());
moreAvailable = inputChannelsWithData.size() > 0;
}
final BufferAndAvailability result = currentChannel.getNextBuffer();
-
// Sanity check that notifications only happen when data is available
if (result == null) {
throw new IllegalStateException("Bug in input gate/channel logic: input gate got " +
"notified by channel about available data, but none was available.");
}
-
// this channel was now removed from the non-empty channels queue
// we re-add it in case it has more data, because in that case no "non-empty" notification
// will come for that channel
@@ -582,9 +587,13 @@ public class SingleInputGate implements InputGate {
int availableChannels;
synchronized (inputChannelsWithData) {
+ if (enqueuedInputChannelsWithData.get(channel.getChannelIndex())) {
+ return;
+ }
availableChannels = inputChannelsWithData.size();
inputChannelsWithData.add(channel);
+ enqueuedInputChannelsWithData.set(channel.getChannelIndex());
if (availableChannels == 0) {
inputChannelsWithData.notifyAll();
http://git-wip-us.apache.org/repos/asf/flink/blob/78df079b/flink-streaming-java/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/StreamTestSingleInputGate.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/StreamTestSingleInputGate.java b/flink-streaming-java/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/StreamTestSingleInputGate.java
index e938ac8..43f2878 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/StreamTestSingleInputGate.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/StreamTestSingleInputGate.java
@@ -96,11 +96,17 @@ public class StreamTestSingleInputGate<T> extends TestSingleInputGate {
final Answer<BufferAndAvailability> answer = new Answer<BufferAndAvailability>() {
@Override
public BufferAndAvailability answer(InvocationOnMock invocationOnMock) throws Throwable {
- InputValue<Object> input = inputQueues[channelIndex].poll();
+ ConcurrentLinkedQueue<InputValue<Object>> inputQueue = inputQueues[channelIndex];
+ InputValue<Object> input;
+ boolean moreAvailable;
+ synchronized (inputQueue) {
+ input = inputQueue.poll();
+ moreAvailable = !inputQueue.isEmpty();
+ }
if (input != null && input.isStreamEnd()) {
when(inputChannels[channelIndex].getInputChannel().isReleased()).thenReturn(
true);
- return new BufferAndAvailability(EventSerializer.toBuffer(EndOfPartitionEvent.INSTANCE), false, 0);
+ return new BufferAndAvailability(EventSerializer.toBuffer(EndOfPartitionEvent.INSTANCE), moreAvailable, 0);
} else if (input != null && input.isStreamRecord()) {
Object inputElement = input.getStreamRecord();
@@ -111,13 +117,13 @@ public class StreamTestSingleInputGate<T> extends TestSingleInputGate {
bufferBuilder.finish();
// Call getCurrentBuffer to ensure size is set
- return new BufferAndAvailability(buildSingleBuffer(bufferBuilder), false, 0);
+ return new BufferAndAvailability(buildSingleBuffer(bufferBuilder), moreAvailable, 0);
} else if (input != null && input.isEvent()) {
AbstractEvent event = input.getEvent();
- return new BufferAndAvailability(EventSerializer.toBuffer(event), false, 0);
+ return new BufferAndAvailability(EventSerializer.toBuffer(event), moreAvailable, 0);
} else {
- synchronized (inputQueues[channelIndex]) {
- inputQueues[channelIndex].wait();
+ synchronized (inputQueue) {
+ inputQueue.wait();
return answer(invocationOnMock);
}
}