[FLINK-8694][runtime] Fix notifyDataAvailable race condition #5572

pnowojski · 2018-02-23T15:28:30Z

This fixes two bugs in network stack:
https://issues.apache.org/jira/browse/FLINK-8760
https://issues.apache.org/jira/browse/FLINK-8694

Brief change log

Please check individual commit messages.

Verifying this change

This PR adds new tests covering the previously bugged cases.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

NicoK

Overall good changes and thanks for fixing the issues.

Some smaller comments inline I'd like addressed before merging.

NicoK · 2018-02-26T11:04:48Z

...rc/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java

-		assertEquals(channelIndex, bufferOrEvent.get().getChannelIndex());
+		assertEquals(expectedIsBuffer, bufferOrEvent.get().isBuffer());
+		assertEquals(expectedChannelIndex, bufferOrEvent.get().getChannelIndex());
+		assertEquals(expectedMoreAvailable, bufferOrEvent.get().moreAvailable());


Can we also verify assertFalse(inputGate.pollNextBufferOrEvent().isPresent()); for SingleInputGate instances here? (ideally also for any other input gate, once UnionInputGate implements this.

NicoK · 2018-02-26T11:18:05Z

...ntime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java

 	/** Flag indicating whether the subpartition has been finished. */
 	private boolean isFinished;

+	private boolean flushRequested;


add @GuardedBy("buffers")?

NicoK · 2018-02-26T11:19:05Z

...ntime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java

-				readView.notifyDataAvailable();
-			}
+			flushRequested = !buffers.isEmpty();
+			notifyDataAvailable();


maybe don't even flush at all if buffers is empty

NicoK · 2018-02-26T11:29:36Z

...ntime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartition.java

 					availabilityListener);
 			}
-
+			readView.notifyDataAvailable();


I know that both read views` implementation of this function are no-ops, but both constructors do their own notification handling already, so this should probably be removed again

NicoK · 2018-02-26T12:19:24Z

...e/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java


 	@Override
 	public void notifyDataAvailable() {
-		// We do the availability listener notification one by one


Well, this is still true for the SpilledSubpartitionView and without the (potentially extended) comment here, a reader might wonder why we don't do anything here, not even forwarding it to the spilled view.

This was the remanent of a different approach to fix the problem. Reverted.

NicoK · 2018-02-26T12:23:52Z

...e/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java

+			}
+		} // else: spilled
+
+		checkState(spilledView != null, "No in-memory buffers available, but also nothing spilled.");


Maybe I see something wrong here, but isn't spilledView always non-null here? In that case, please remove the check.

NicoK · 2018-02-26T13:10:43Z

flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadTask.java

 		for (RecordWriter<?> eventualOutput : this.eventualOutputs) {
 			eventualOutput.broadcastEvent(EndOfSuperstepEvent.INSTANCE);
+			eventualOutput.clearBuffers();
+			eventualOutput.flushAll();


let's remove those two - they should be unnecessary

NicoK · 2018-02-26T13:26:18Z

...e/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java

 		assertEquals(partition.getBuffersInBacklog(), read.buffersInBacklog());
 		read.buffer().recycleBuffer();
-		assertEquals(2, listener.getNumNotifications());
+		assertEquals(1, listener.getNumNotifications());


maybe add a small comment that we don't get a new notification (anymore!) since the buffer is already available at the getNextBuffer() call?

NicoK · 2018-02-26T13:26:48Z

...e/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java

-		listener.awaitNotifications(3, 30_000);
-		assertEquals(3, listener.getNumNotifications());
+		listener.awaitNotifications(2, 30_000);
+		assertEquals(2, listener.getNumNotifications());


note that we will be notified once the spilled writer completes

NicoK · 2018-02-26T13:29:50Z

...a/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTests.java

+		env.setUp(1, 1, 0, false);
+		env.executeBenchmark(1_000_000);
+		env.tearDown();
+	}


Is this not tested via your non-IT tests now?
(we should be careful about adding integration/full stack tests because of their added time)

This is smaller scope test compared to ITCase and it covers higher load. It's a very good stress test for investigating and detecting deadlocks/race conditions in the network stack.

NicoK · 2018-02-26T14:06:25Z

...n/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java

 			"requestLock=" + requestLock +
 			", receiverId=" + receiverId +
-			", buffersAvailable=" + buffersAvailable.get() +
+			", buffersAvailable=" + hasBuffersAvailable() +


this line should now be removed (the property is not part of this object anymore and going through a synchronized block which may be nasty during debugging sessions)

NicoK · 2018-02-26T15:37:10Z

...ime/src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionView.java

+		if (nextBuffer != null) {
+			return true;
+		}
+		return fileReader.hasReachedEndOfFile();


should be !fileReader.hasReachedEndOfFile(); - I will add tests in a later PR for this

pnowojski

Addressed comments.

pnowojski · 2018-02-27T10:16:09Z

...e/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java


 	@Override
 	public void notifyDataAvailable() {
-		// We do the availability listener notification one by one


This was the remanent of a different approach to fix the problem. Reverted.

pnowojski · 2018-02-27T10:21:09Z

...a/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTests.java

+		env.setUp(1, 1, 0, false);
+		env.executeBenchmark(1_000_000);
+		env.tearDown();
+	}


This is smaller scope test compared to ITCase and it covers higher load. It's a very good stress test for investigating and detecting deadlocks/race conditions in the network stack.

NicoK

I would have expected fixup commits, but I'll trust you with the changes. After fixing the (simple) merge conflict, we're good to go.

…SingleInputGate Previously if we SingleInputGate was re-eqnqueuing an input channel, isMoreAvailable might incorrectly return false. This might caused some dead locks.

…BuffersFailureITCase

Before there was a race condition that might resulted in igonoring some notifyDataAvailable calls. This fixes the problem by moving buffersAvailable handling to Supartitions and adds stress test for flushAlways (without this fix this test is dead locking).

pnowojski · 2018-02-28T14:52:23Z

Rebased. Lets merge after green travis.

pnowojski · 2018-02-28T16:24:05Z

Thanks for merging :)

pnowojski force-pushed the f8694-proper-fix branch from e70cd04 to 57f83c7 Compare February 23, 2018 15:54

NicoK requested changes Feb 26, 2018

View reviewed changes

NicoK mentioned this pull request Feb 26, 2018

[FLINK-8694][runtime] Workaround notifyDataAvailable race condition #5557

Closed

NicoK reviewed Feb 26, 2018

View reviewed changes

NicoK mentioned this pull request Feb 26, 2018

[FLINK-8755][FLINK-8786][network] fix two bugs in spilled and spillable subpartition views #5581

Closed

pnowojski mentioned this pull request Feb 27, 2018

[FLINK-8750][runtime] Improve detection of no remaining data after EndOfPartitionEvent #5588

Closed

pnowojski force-pushed the f8694-proper-fix branch from 57f83c7 to e80460d Compare February 27, 2018 10:35

pnowojski commented Feb 27, 2018

View reviewed changes

pnowojski force-pushed the f8694-proper-fix branch from e80460d to 81d8d38 Compare February 27, 2018 10:42

NicoK approved these changes Feb 27, 2018

View reviewed changes

pnowojski added 5 commits February 28, 2018 15:49

[hotfix][tests] Deduplicate code in SingleInputGateTest

2d96723

[hotfix][runtime] Remove duplicated check

2bfdc33

[FLINK-8760][runtime] Correctly propagate moreAvailable flag through …

551fcb5

…SingleInputGate Previously if we SingleInputGate was re-eqnqueuing an input channel, isMoreAvailable might incorrectly return false. This might caused some dead locks.

[hotfix][tests] Do not hide original exception in SuccessAfterNetwork…

8fabe90

…BuffersFailureITCase

pnowojski force-pushed the f8694-proper-fix branch from 81d8d38 to 4a16fb4 Compare February 28, 2018 14:52

pnowojski closed this Feb 28, 2018

rmetzger added the component=<none> label Mar 18, 2019

[FLINK-8694][runtime] Fix notifyDataAvailable race condition #5572

[FLINK-8694][runtime] Fix notifyDataAvailable race condition #5572

Uh oh!

Conversation

pnowojski commented Feb 23, 2018

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

NicoK left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pnowojski left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

NicoK left a comment

Choose a reason for hiding this comment

Uh oh!

pnowojski commented Feb 28, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

pnowojski commented Feb 28, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

pnowojski commented Feb 28, 2018 •

edited

Loading