You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/08/30 09:09:37 UTC
[flink] 03/03: [hotfix][refactor] Remove
BufferListener#NotificationResult to simplify the code
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit cb92b9da820cf2e2641bd319a11311b64cd7da8a
Author: kevin.cyj <ke...@alibaba-inc.com>
AuthorDate: Fri Aug 27 20:12:24 2021 +0800
[hotfix][refactor] Remove BufferListener#NotificationResult to simplify the code
This closes #17009.
---
.../runtime/io/network/buffer/BufferListener.java | 25 ++--------------------
.../runtime/io/network/buffer/LocalBufferPool.java | 5 ++---
.../network/partition/consumer/BufferManager.java | 17 ++++++---------
.../io/network/buffer/LocalBufferPoolTest.java | 4 ++--
.../partition/consumer/RemoteInputChannelTest.java | 10 +++------
5 files changed, 16 insertions(+), 45 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferListener.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferListener.java
index 6862025..b406d7a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferListener.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferListener.java
@@ -24,27 +24,6 @@ package org.apache.flink.runtime.io.network.buffer;
*/
public interface BufferListener {
- /** Status of the notification result from the buffer listener. */
- enum NotificationResult {
- BUFFER_NOT_USED(false),
- BUFFER_USED(true);
-
- private final boolean isBufferUsed;
-
- NotificationResult(boolean isBufferUsed) {
- this.isBufferUsed = isBufferUsed;
- }
-
- /**
- * Whether the notified buffer is accepted to use by the listener.
- *
- * @return <tt>true</tt> if the notified buffer is accepted.
- */
- boolean isBufferUsed() {
- return isBufferUsed;
- }
- }
-
/**
* Notification callback if a buffer is recycled and becomes available in buffer pool.
*
@@ -59,9 +38,9 @@ public interface BufferListener {
* stack!
*
* @param buffer buffer that becomes available in buffer pool.
- * @return NotificationResult if the listener wants to be notified next time.
+ * @return true if the buffer is accepted by the listener.
*/
- NotificationResult notifyBufferAvailable(Buffer buffer);
+ boolean notifyBufferAvailable(Buffer buffer);
/** Notification callback if the buffer provider is destroyed. */
void notifyBufferDestroyed();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
index e86a090..20b89e6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.io.network.buffer;
import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.io.network.buffer.BufferListener.NotificationResult;
import org.apache.flink.util.ExceptionUtils;
import org.slf4j.Logger;
@@ -495,12 +494,12 @@ class LocalBufferPool implements BufferPool {
checkConsistentAvailability();
}
- } while (!fireBufferAvailableNotification(listener, segment).isBufferUsed());
+ } while (!fireBufferAvailableNotification(listener, segment));
mayNotifyAvailable(toNotify);
}
- private NotificationResult fireBufferAvailableNotification(
+ private boolean fireBufferAvailableNotification(
BufferListener listener, MemorySegment segment) {
// We do not know which locks have been acquired before the recycle() or are needed in the
// notification and which other threads also access them.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java
index 7b908cd..41eb12f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java
@@ -300,14 +300,10 @@ public class BufferManager implements BufferListener, BufferRecycler {
* buffer pool. Otherwise, the buffer will be added into the <tt>bufferQueue</tt>.
*
* @param buffer Buffer that becomes available in buffer pool.
- * @return NotificationResult indicates whether this channel accepts the buffer and is waiting
- * for more floating buffers.
+ * @return true if the buffer is accepted by this listener.
*/
@Override
- public BufferListener.NotificationResult notifyBufferAvailable(Buffer buffer) {
- BufferListener.NotificationResult notificationResult =
- BufferListener.NotificationResult.BUFFER_NOT_USED;
-
+ public boolean notifyBufferAvailable(Buffer buffer) {
// Assuming two remote channels with respective buffer managers as listeners inside
// LocalBufferPool.
// While canceler thread calling ch1#releaseAllResources, it might trigger
@@ -319,10 +315,11 @@ public class BufferManager implements BufferListener, BufferRecycler {
// bufferQueue lock to cause deadlock. So we check the isReleased state out of synchronized
// to resolve it.
if (inputChannel.isReleased()) {
- return notificationResult;
+ return false;
}
int numBuffers = 0;
+ boolean isBufferUsed = false;
try {
synchronized (bufferQueue) {
checkState(
@@ -339,11 +336,11 @@ public class BufferManager implements BufferListener, BufferRecycler {
// lock on bufferQueue to release buffers
if (inputChannel.isReleased()
|| bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
- return notificationResult;
+ return false;
}
bufferQueue.addFloatingBuffer(buffer);
- notificationResult = NotificationResult.BUFFER_USED;
+ isBufferUsed = true;
numBuffers += 1 + tryRequestBuffers();
bufferQueue.notifyAll();
}
@@ -353,7 +350,7 @@ public class BufferManager implements BufferListener, BufferRecycler {
inputChannel.setError(t);
}
- return notificationResult;
+ return isBufferUsed;
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
index bdc6bbe..1ea8b0c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
@@ -537,10 +537,10 @@ public class LocalBufferPoolTest extends TestLogger {
private final AtomicInteger times = new AtomicInteger(0);
@Override
- public NotificationResult notifyBufferAvailable(Buffer buffer) {
+ public boolean notifyBufferAvailable(Buffer buffer) {
times.incrementAndGet();
buffer.recycleBuffer();
- return NotificationResult.BUFFER_USED;
+ return true;
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
index 12c78c2..a098f24 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
@@ -39,7 +39,6 @@ import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.Buffer.DataType;
import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
-import org.apache.flink.runtime.io.network.buffer.BufferListener.NotificationResult;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
@@ -244,7 +243,7 @@ public class RemoteInputChannelTest {
8192,
(inputChannel, buffer, j) -> {
inputChannel.onBuffer(buffer, j, -1);
- return null;
+ return true;
});
}
@@ -270,7 +269,7 @@ public class RemoteInputChannelTest {
*/
private void testConcurrentReleaseAndSomething(
final int numberOfRepetitions,
- TriFunction<RemoteInputChannel, Buffer, Integer, Object> function)
+ TriFunction<RemoteInputChannel, Buffer, Integer, Boolean> function)
throws Exception {
// Setup
@@ -290,10 +289,7 @@ public class RemoteInputChannelTest {
for (int j = 0; j < 128; j++) {
// this is the same buffer over and over again which will be
// recycled by the RemoteInputChannel
- Object obj =
- function.apply(inputChannel, buffer.retainBuffer(), j);
- if (obj instanceof NotificationResult
- && obj == NotificationResult.BUFFER_NOT_USED) {
+ if (!function.apply(inputChannel, buffer.retainBuffer(), j)) {
buffer.recycleBuffer();
}
}