[FLINK-12490][network] Introduce Input and NetworkInput interfaces #8476

pnowojski · 2019-05-17T13:16:24Z

This PR depends on #8467

It introduces Input interface, it's implementation NetworkInput (extracted from StreamInputProcessor) and use them in StreamInputProcessor.

Local benchmark run shows increase in records throughput of SumLongsBenchmark from 8600 records/ms to 9200-9400 records/ms.

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)

flinkbot · 2019-05-17T13:17:39Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

sunhaibotb · 2019-05-20T09:19:38Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Input.java

+ * Basic interface for inputs of stream operators.
+ */
+@Internal
+public interface Input extends AsyncDataInput<StreamElement>, Closeable {


Can we change it to implement AutoCloseable ?

Generally speaking Closeable is better and preferred - it extends Autocloseable and provide stronger/easier to use contract of close() being idempotent.

Is there a reason why Input#close() can not be idempotent?

Generally speaking Closeable is better and preferred - it extends Autocloseable and provide stronger/easier to use contract of close() being idempotent.

You are right.

Is there a reason why Input#close() can not be idempotent?

No.

sunhaibotb

About currentChannel in StreamInputProcessor, I left some comments. @pnowojski

sunhaibotb · 2019-05-20T09:45:22Z

...streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java

+	private void processElement(StreamElement recordOrMark) throws Exception {
+		if (recordOrMark.isWatermark()) {
+			// handle watermark
+			statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);


It seems missing some logic about assigning value to currentChannel , and currently it is always - 1.

I think that the input channel index should be hidden in NetworkInput.

True, there was a bug. However I'm not sure if we should squeeze more functionality to the NetworkInput. As it is now, it's a pretty small and nice class, that has a single responsibility - deserializing records, it would be nice to keep it as that.

I was actually thinking about maybe latter adding another wrapper class, that would wrap Input and encapsulate watermark handling?

In fact, I care about that the input channel index is exposed to input processors. As long as it's not exposed to the input processor (for example, encapsulating a class as you said), it's fine to me.

sunhaibotb · 2019-05-20T09:47:15Z

...streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java

+			statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);
+		} else if (recordOrMark.isStreamStatus()) {
+			// handle stream status
+			statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);


Same as the previous comment.

sunhaibotb · 2019-05-20T10:22:20Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/NetworkInput.java

+				}
+			}
+
+			Optional<BufferOrEvent> bufferOrEvent = barrierHandler.pollNext();


In StreamTwoInputSeletableProcessor, input1 and input2 each correspond to a NetworkInput, and each NetworkInput should contain its own InputGate (inputGate1 or inputGate2). But I guess your BarrierHandler should contain one UnionInputGate unioned inputGate1 and inputGate2 in case of two inputs (because BarrierHandler must see all input channels), so I think the current implementation of BarrierHandler may not meet the requirements of StreamTwoInputSeletableProcessor.

True, that's the next thing that I want to address. In this commit I just want to make sure that the basic abstraction of NetworkInput is not introducing performance regressions, which was one of our concern in the offline discussions.

sunhaibotb · 2019-05-20T11:21:31Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Input.java

+ */
+@Internal
+public interface Input extends AsyncDataInput<StreamElement>, Closeable {
+}


Similar to InputChannel#getChannelIndex(), adding int getInputIndex() to Input can simplify some processing in StreamTwoInputSelectableProcessor. So I suggest to add it.

I agree, but I didn't want to add a dead code now - we can extend this interface in a follow up commit.

sunhaibotb · 2019-05-20T11:27:27Z

...streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java

+		StreamElement recordOrMark = input.pollNext();
+		if (recordOrMark == null) {
+			input.isAvailable().get();
+			return input.isFinished();


Should here bereturn !input.isFinished() ?

😳

This might explain deadlocks that were found by our tests :)

StefanRRichter

Had some more comments on top of what @sunhaibotb already suggested. From the desciption, I also assume that the two input cases is not included in the PR on purpose, right?

StefanRRichter · 2019-05-20T10:02:40Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Input.java

+ * Basic interface for inputs of stream operators.
+ */
+@Internal
+public interface Input extends AsyncDataInput<StreamElement>, Closeable {


If I look at what interfaces are extended, the name of this interface looks a bit strange/over simplified from that perspective. So does the comment btw.

I've also noticed that (I was struggling with finding a good name for AsyncDataInput ), but what would you suggest?

Maybe something along the line of StreamTask[Async][Data]Input. This should make role of this interface a lot clearer than the super generic name that it has now.

StreamTaskInput is not that bad. And the concrete class StreamTaskNetworkInput?

👍 That sounds likw two names that create a mental model in my head about what this does.

@sunhaibotb, are you fine with renaming those renames? (and fyi, since renaming will affect a wip StreamSelectableTwoInputStreamProcessor PR.

@pnowojski @sunhaibotb ping so that we can address this if we aggree upon my suggestion.

@sunhaibotb, are you fine with renaming those renames? (and fyi, since renaming will affect a wip StreamSelectableTwoInputStreamProcessor PR.

It's fine to me. @pnowojski @StefanRRichter

I've renamed the classes.

StefanRRichter · 2019-05-20T10:30:27Z

...aming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java

@@ -76,4 +62,6 @@
 	 * @return The duration in nanoseconds
 	 */
 	long getAlignmentDurationNanos();
+
+	int getNumberOfInputChannels();


getNumberOfInputChannels // @return number of input channels
:(

getAlignmentDurationNanos // @return The duration in nanoseconds? But it is really a nit.

Yes I know and I'm not a big fan of it as well :)

(maybe you didn't notice, but I've already added the comment you requested yesterday.)

Upsi, no I first replied to your anwers and now checked out the updated branch :D

StefanRRichter · 2019-05-20T10:31:55Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/NetworkInput.java

+	 * The channel from which a buffer came, tracked so that we can appropriately map
+	 * the watermarks and watermark statuses to the correct channel index of the correct valve.
+	 */
+	private int currentChannel = -1;


Nit/Personal taste: I would do all field initialization in constructors only.

Also would use a static final int UNSPECIFIED = -1;

This field could currenly also just be a local variable.

StefanRRichter · 2019-05-20T14:30:49Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/NetworkInput.java

+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Implementation of taking {@link InputGate} as {@link Input}.


I know in some sense that is true, but it actually takes a CheckpointBarrierHandler.

StefanRRichter · 2019-05-20T15:07:43Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/NetworkInput.java

+@Internal
+public final class NetworkInput implements Input {
+
+	private final int numberOfInputChannels;


At least currently, this field has no real use.

pnowojski

Thanks for the reviews @sunhaibotb and @StefanRRichter. Sorry for pushing the PR with such obvious bugs (travis was not passing because of them).

There was also one more bug that we have missed (found by another dead locks on travis) and it's about already merged InputGate#isAvailable() changes. I have fixed it in this PR in a new commit:

[hotfix][network] Fix InputGate#isAvailable not completed while InputGate#isFinished is true

Please check it out.

pnowojski · 2019-05-21T14:45:30Z

...aming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java

@@ -76,4 +62,6 @@
 	 * @return The duration in nanoseconds
 	 */
 	long getAlignmentDurationNanos();
+
+	int getNumberOfInputChannels();


getNumberOfInputChannels // @return number of input channels
:(

pnowojski · 2019-05-21T14:49:12Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Input.java

+ * Basic interface for inputs of stream operators.
+ */
+@Internal
+public interface Input extends AsyncDataInput<StreamElement>, Closeable {


Generally speaking Closeable is better and preferred - it extends Autocloseable and provide stronger/easier to use contract of close() being idempotent.

Is there a reason why Input#close() can not be idempotent?

pnowojski · 2019-05-21T14:50:06Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Input.java

+ */
+@Internal
+public interface Input extends AsyncDataInput<StreamElement>, Closeable {
+}


I agree, but I didn't want to add a dead code now - we can extend this interface in a follow up commit.

pnowojski · 2019-05-21T14:51:26Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Input.java

+ * Basic interface for inputs of stream operators.
+ */
+@Internal
+public interface Input extends AsyncDataInput<StreamElement>, Closeable {


I've also noticed that (I was struggling with finding a good name for AsyncDataInput ), but what would you suggest?

pnowojski · 2019-05-21T14:55:45Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/NetworkInput.java

+				}
+			}
+
+			Optional<BufferOrEvent> bufferOrEvent = barrierHandler.pollNext();


True, that's the next thing that I want to address. In this commit I just want to make sure that the basic abstraction of NetworkInput is not introducing performance regressions, which was one of our concern in the offline discussions.

pnowojski · 2019-05-21T14:57:39Z

...streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java

+		StreamElement recordOrMark = input.pollNext();
+		if (recordOrMark == null) {
+			input.isAvailable().get();
+			return input.isFinished();


😳

This might explain deadlocks that were found by our tests :)

pnowojski · 2019-05-22T13:07:23Z

...streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java

+	private void processElement(StreamElement recordOrMark) throws Exception {
+		if (recordOrMark.isWatermark()) {
+			// handle watermark
+			statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);


True, there was a bug. However I'm not sure if we should squeeze more functionality to the NetworkInput. As it is now, it's a pretty small and nice class, that has a single responsibility - deserializing records, it would be nice to keep it as that.

I was actually thinking about maybe latter adding another wrapper class, that would wrap Input and encapsulate watermark handling?

StefanRRichter · 2019-05-24T08:27:50Z

General comment first, I noticed that Travis indicates some related test failures.

StefanRRichter · 2019-05-27T08:28:40Z

@pnowojski I cannot check the Travis result because the build is not going through.

StefanRRichter

For most part the PR looks good, had a couple more comments and we also require a working build.

StefanRRichter · 2019-05-23T13:57:34Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/NetworkInput.java

+
+	@Override
+	public int getLastChannel() {
+		return currentChannel;


Consistent naming between field and method?

Hmm, you are right. Somehow I felt that the the lastChannel is a better name for the user of the class while currentChannel better reflects the logic inside, but consistency trumps this.

StefanRRichter · 2019-05-27T08:56:44Z

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

@@ -254,7 +248,19 @@ private void handleEndOfPartitionEvent(BufferOrEvent bufferOrEvent, InputGate in
 				throw new IllegalStateException("Couldn't find input gate in set of remaining " +
 					"input gates.");
 			}
+			if (isFinished()) {
+				markAvailable();


The naming now is a bit misleading, if I read this as a text: this is finished and no event will become available anymore, so I mark it as available? Maybe this shifted the whole concept from indicating availability of data to indicating that some thing in the status has changed, either data is available or the input noticed that is it finished.

Check my other response, actually I was never thinking about this in terms of "data availability" but in terms of "input being available for processing/handling". If input is in isFinished state, it has to be isAvailable as well from this perspective.

StefanRRichter · 2019-05-27T08:58:40Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/AsyncDataInput.java

+	 * not completed futures should become completed once there is more input available or if
+	 * the input {@link #isFinished()}.
+	 */
+	CompletableFuture<?> isAvailable();


So should this be isAvailableOrFinished or maybe awaitStatusChange?

Originally I was proposing isBlocked() name for this method - whether the input/source/thingy is blocked and waiting for more input. Someone else (Stephan?) preferred isAvailable, since isBlocked returning a feature was confusing for him and Becket. IMO isAvailable is good enough, as long as you think about it in the similar terms: is the input available for further processing - in other words, the opposite of isBlocked(). Not if has it more data available.

Either way I would vote isBlocked > isAvailable > available > awaitStatusChange > isAvailableOrFinished.

Can we postpone the name voting until the source interface refactoring, because first and foremost I want this naming convention be in sync in those two places? We can rename it quite easily, but I don't want to spend too much time discussing it twice with a different audience.

StefanRRichter · 2019-05-27T09:00:22Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Input.java

+ * Basic interface for inputs of stream operators.
+ */
+@Internal
+public interface Input extends AsyncDataInput<StreamElement>, Closeable {


@pnowojski @sunhaibotb ping so that we can address this if we aggree upon my suggestion.

pnowojski

I have addressed/responded to your comments @StefanRRichter. Tests were not passing because of some indirect conflict when rebasing the code on the latest master that happened almost at the same time I was pushing the code. Rebased again, hopefully travis will be finally green.

pnowojski · 2019-05-27T09:58:34Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/AsyncDataInput.java

+	 * not completed futures should become completed once there is more input available or if
+	 * the input {@link #isFinished()}.
+	 */
+	CompletableFuture<?> isAvailable();


Originally I was proposing isBlocked() name for this method - whether the input/source/thingy is blocked and waiting for more input. Someone else (Stephan?) preferred isAvailable, since isBlocked returning a feature was confusing for him and Becket. IMO isAvailable is good enough, as long as you think about it in the similar terms: is the input available for further processing - in other words, the opposite of isBlocked(). Not if has it more data available.

Either way I would vote isBlocked > isAvailable > available > awaitStatusChange > isAvailableOrFinished.

Can we postpone the name voting until the source interface refactoring, because first and foremost I want this naming convention be in sync in those two places? We can rename it quite easily, but I don't want to spend too much time discussing it twice with a different audience.

pnowojski · 2019-05-27T10:00:26Z

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

@@ -254,7 +248,19 @@ private void handleEndOfPartitionEvent(BufferOrEvent bufferOrEvent, InputGate in
 				throw new IllegalStateException("Couldn't find input gate in set of remaining " +
 					"input gates.");
 			}
+			if (isFinished()) {
+				markAvailable();


Check my other response, actually I was never thinking about this in terms of "data availability" but in terms of "input being available for processing/handling". If input is in isFinished state, it has to be isAvailable as well from this perspective.

pnowojski · 2019-05-27T10:03:50Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/NetworkInput.java

+
+	@Override
+	public int getLastChannel() {
+		return currentChannel;


Hmm, you are right. Somehow I felt that the the lastChannel is a better name for the user of the class while currentChannel better reflects the logic inside, but consistency trumps this.

StefanRRichter · 2019-05-27T10:12:32Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/Input.java

+ */
+@Internal
+public interface Input extends NullableAsyncDataInput<StreamElement>, Closeable {
+	int UNSPECIFED = -1;


BTW UNSPECIFED -> UNSPECIFIED

StefanRRichter

Looks good so far, found one more type. I think we can postpone the discussion about the name of the available-method. However, I would like to have the discussion about the name of Input resolved before merging.

...streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java

…Gate#isFinished is true Finished InputGates should be available for reading, otherwise reader can deadlock waiting on a future that won't be completed ever.

…utProcessor

Make the most commonly expected condition the first one.

pnowojski requested a review from StefanRRichter May 17, 2019 13:16

rmetzger added the review=description? label May 17, 2019

pnowojski force-pushed the f12490 branch from a740bc6 to aa5a82a Compare May 17, 2019 14:17

rmetzger added the component=Runtime/Network label May 17, 2019

sunhaibotb reviewed May 20, 2019

View reviewed changes

StefanRRichter requested changes May 21, 2019

View reviewed changes

pnowojski force-pushed the f12490 branch from aa5a82a to b586c2a Compare May 22, 2019 13:09

pnowojski commented May 22, 2019

View reviewed changes

pnowojski force-pushed the f12490 branch from b586c2a to 633da8c Compare May 23, 2019 08:24

1u0 mentioned this pull request May 24, 2019

[FLINK-12535][network] Make CheckpointBarrierHandler non-blocking #8467

Merged

pnowojski force-pushed the f12490 branch 2 times, most recently from 3b85ebe to 6f25edf Compare May 24, 2019 14:22

StefanRRichter requested changes May 27, 2019

View reviewed changes

pnowojski commented May 27, 2019

View reviewed changes

pnowojski force-pushed the f12490 branch from 6f25edf to 6eed02f Compare May 27, 2019 10:07

StefanRRichter reviewed May 27, 2019

View reviewed changes

pnowojski force-pushed the f12490 branch 2 times, most recently from 202c539 to 9b8cd5d Compare May 27, 2019 11:57

sunhaibotb reviewed May 29, 2019

View reviewed changes

...streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java Outdated Show resolved Hide resolved

sunhaibotb mentioned this pull request May 30, 2019

[FLINK-11877] Implement the runtime handling of the InputSelectable interface #8124

Merged

pnowojski force-pushed the f12490 branch 2 times, most recently from 5d59353 to 5342482 Compare June 4, 2019 11:34

pnowojski force-pushed the f12490 branch 2 times, most recently from 552310f to 42175da Compare June 5, 2019 12:57

pnowojski and others added 6 commits June 5, 2019 15:00

[hotfix][network] Simplify UnionInputGate#isFinished

966d3a0

[hotfix][network] Fix InputGate#isAvailable not completed while Input…

313b0e9

…Gate#isFinished is true Finished InputGates should be available for reading, otherwise reader can deadlock waiting on a future that won't be completed ever.

[FLINK-12490][network] Introduce Input interface

5e6c20c

[FLINK-12490][network] Implement NetworkInput and use it in StreamInp…

bf2922f

…utProcessor

[hotfix][operator] Reorder if conditions in StreamInputProcessor

a472350

Make the most commonly expected condition the first one.

[hotfix][operator] Remove unused field in StreamInputProcessor

d9ef528

pnowojski force-pushed the f12490 branch from 42175da to d9ef528 Compare June 5, 2019 13:00

pnowojski merged commit 0bd3cf6 into apache:master Jun 5, 2019

pnowojski deleted the f12490 branch June 5, 2019 13:00

[FLINK-12490][network] Introduce Input and NetworkInput interfaces #8476

[FLINK-12490][network] Introduce Input and NetworkInput interfaces #8476

Conversation

pnowojski commented May 17, 2019

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

Documentation

flinkbot commented May 17, 2019

Review Progress

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sunhaibotb May 23, 2019 • edited

Choose a reason for hiding this comment

sunhaibotb left a comment

Choose a reason for hiding this comment

sunhaibotb May 20, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sunhaibotb May 20, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

StefanRRichter left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

StefanRRichter commented May 24, 2019

StefanRRichter commented May 27, 2019

StefanRRichter left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

StefanRRichter left a comment

Choose a reason for hiding this comment

sunhaibotb May 23, 2019 •

edited

sunhaibotb May 20, 2019 •

edited

sunhaibotb May 20, 2019 •

edited