[FLINK-7699][core] Define the BufferListener interface to replace EventlListener in BufferProvider #4735

zhijiangW · 2017-09-27T14:47:01Z

What is the purpose of the change

Currently the EventListener is used in BufferProvider to be notified of buffer available or destroyed pool.

To be semantic clearly, we define a new BufferListener interface which can opt for a one-time only notification or to be notified repeatedly. And we can also notify the pool destroyed via explicitly method notifyBufferDestroyed.

This RP is based on #4499 whose commit is also included for passing travis. Review the third commit for this PR change.

Brief change log

The RemoteInputChannel will implement the BufferListener to wait for floating buffers from BufferProvider.

Verifying this change

This change is already covered by existing tests, such as LocalBufferPoolTest.

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

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

Documentation

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

…pe for the floating buffers

…ntlListener in BufferProvider

NicoK

Thanks for the contribution.
I really like the new interface, but we should also add a test for the new possibility to re-add the listener.

The sender backlog changes inside RemoteInputChannel probably slipped through and should go into #4509 instead. Please remove them from this PR. I'll re-evaluate there.

NicoK · 2017-09-27T15:51:40Z

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

+	 */
+	@Override
+	public boolean notifyBufferAvailable(Buffer buffer) {
+		checkState(isWaitingFloatingBuffers.get(), "This channel should be waiting for floating buffers currently.");


"This channel should be waiting for floating buffers."

NicoK · 2017-09-27T15:55:09Z

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java

 				buffer.recycle();
+				return false;


can you also add a test which verifies that requests for multiple buffers, i.e. returning true here, works?

Yes, I also consider the issue of verifying notification multi times.
And I guess you may point out it. :)
It can not work to return true for that, and I will think of proper way of verifying it.

NicoK · 2017-09-27T16:06:37Z

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

+	/**
+	 * Buffer pool notifies this channel of an available floating buffer. If the channel is released or not
+	 * needing extra buffers currently, the buffer should be recycled to buffer pool. Otherwise, the buffer
+	 * will be added into the available queue and the unannounced credit is increased by one.


The Buffer pool notifies this channel of an available floating buffer. If the channel is released or currently does not need extra buffers, the buffer should be recycled to the buffer pool. Otherwise, the buffer will be added into the <tt>availableBuffers</tt> queue and the unannounced credit is increased by one.

NicoK · 2017-09-27T16:07:47Z

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

+	private final AtomicInteger senderBacklog = new AtomicInteger(0);
+
+	/** The tag indicates whether this channel is waiting for additional floating buffers from the buffer pool. */
+	private final AtomicBoolean isWaitingFloatingBuffers = new AtomicBoolean(false);


better: isWaitingForFloatingBuffers

but in general, do we need this field? For now, I only see this being used to verify correct use and I'm not sure whether the costs verify this. Also, nothing in this PR sets it to true.

These two fields are currently used in notifyBufferAvailable() logic, so we have to define them in this PR.
In next PR #4509 , isWaitingForFloatingBuffers field will be set true by two conditions:

The number of current available buffers is less than sender backlog

There are no available floating buffers in BufferProvider
And this field also used to avoid register listener in BufferProvider multi times.

It may confuse you to only see this PR change, sorry for that.

ok, fine - so I'll see more details in the next PR and we get this in here

…ntlListener in BufferProvider

zhijiangW · 2017-09-28T07:38:06Z

@NicoK , I have submitted the updates based on your comments!

NicoK

Nice new test - I only have minor comments to address.

NicoK · 2017-09-28T15:53:19Z

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java


-		assertTrue(localBufferPool.addBufferListener(listener));
+		// Recycle the first buffer to notify both of the above listeners and the
+		// <<tt>twoTimesListener</tt> will be added into the <<tt>registeredListeners</tt>


you don't need the HTML tags here (I suggested them only for the Javadoc - some use {@code } instead), please remove them here

NicoK · 2017-09-28T15:56:54Z

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java

+		// Recycle the first buffer to notify both of the above listeners and the
+		// <<tt>twoTimesListener</tt> will be added into the <<tt>registeredListeners</tt>
+		// queue of buffer pool again
+		available1.recycle();


can you also verify the notification right after this recycle() call? By adding:
verify(oneTimeListener, times(1)).notifyBufferAvailable(any(Buffer.class));

NicoK · 2017-09-28T15:57:41Z

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java

+			@Override
+			public boolean notifyBufferAvailable(Buffer buffer) {
+				times++;
+				buffer.recycle();


nice trick with the recycle() call here leading to another invocation or the release of the buffer

…ntlListener in BufferProvider

zhijiangW · 2017-09-28T16:22:59Z

Already removed the tags and added the verify

NicoK

lgtm

zentol · 2017-10-10T12:12:46Z

merging.

…ntlListener in BufferProvider This closes apache#4735.

zhijiangW added 3 commits September 22, 2017 14:15

[FLINK-7378][core] Create a fix size (non rebalancing) buffer pool ty…

db14ccc

…pe for the floating buffers

[FLINK-7394][core] Manage exclusive buffers in RemoteInputChannel

8a71995

[FLINK-7699][core] Define the BufferListener interface to replace Eve…

75b6e39

…ntlListener in BufferProvider

NicoK requested changes Sep 27, 2017

View reviewed changes

zhijiangW force-pushed the FLINK-7699 branch from 087b6b4 to 616f81b Compare September 28, 2017 04:46

[FLINK-7699][core] Define the BufferListener interface to replace Eve…

88bb089

…ntlListener in BufferProvider

zhijiangW force-pushed the FLINK-7699 branch from 616f81b to 88bb089 Compare September 28, 2017 05:38

NicoK requested changes Sep 28, 2017

View reviewed changes

[FLINK-7699][core] Define the BufferListener interface to replace Eve…

a880b0d

…ntlListener in BufferProvider

zhijiangW mentioned this pull request Sep 29, 2017

[FLINK-7406][network] Implement Netty receiver incoming pipeline for credit-based #4509

Closed

NicoK approved these changes Sep 29, 2017

View reviewed changes

zentol pushed a commit to zentol/flink that referenced this pull request Oct 10, 2017

[FLINK-7699][core] Define the BufferListener interface to replace Eve…

f72976c

…ntlListener in BufferProvider This closes apache#4735.

zentol pushed a commit to zentol/flink that referenced this pull request Oct 10, 2017

[FLINK-7699][core] Define the BufferListener interface to replace Eve…

331b777

…ntlListener in BufferProvider This closes apache#4735.

asfgit closed this in 8706c6f Oct 10, 2017

zhijiangW deleted the FLINK-7699 branch January 9, 2018 07:57

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-7699][core] Define the BufferListener interface to replace EventlListener in BufferProvider #4735

[FLINK-7699][core] Define the BufferListener interface to replace EventlListener in BufferProvider #4735

zhijiangW commented Sep 27, 2017 •

edited

NicoK left a comment •

edited

NicoK Sep 27, 2017

NicoK Sep 27, 2017

zhijiangW Sep 27, 2017

NicoK Sep 27, 2017

NicoK Sep 27, 2017

NicoK Sep 27, 2017

zhijiangW Sep 27, 2017

NicoK Sep 28, 2017

zhijiangW commented Sep 28, 2017

NicoK left a comment

NicoK Sep 28, 2017

NicoK Sep 28, 2017

NicoK Sep 28, 2017

zhijiangW commented Sep 28, 2017

NicoK left a comment

zentol commented Oct 10, 2017

[FLINK-7699][core] Define the BufferListener interface to replace EventlListener in BufferProvider #4735

[FLINK-7699][core] Define the BufferListener interface to replace EventlListener in BufferProvider #4735

Conversation

zhijiangW commented Sep 27, 2017 • edited

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

NicoK left a comment • 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhijiangW commented Sep 28, 2017

NicoK 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

zhijiangW commented Sep 28, 2017

NicoK left a comment

Choose a reason for hiding this comment

zentol commented Oct 10, 2017

zhijiangW commented Sep 27, 2017 •

edited

NicoK left a comment •

edited