Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201744892
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
---
@@ -277,37 +277,17 @@ public void recycle(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.
// -> call notifyBufferAvailable() outside of the synchronized
block to avoid a deadlock (FLINK-9676)
- boolean success = false;
- boolean needMoreBuffers = false;
- try {
- needMoreBuffers = listener.notifyBufferAvailable(new
NetworkBuffer(segment, this));
- success = true;
- } catch (Throwable ignored) {
- // handled below, under the lock
- }
+ // Note that in case of any exceptions notifyBufferAvailable()
should recycle the buffer and
+ // therefore end up in this method again.
+ needMoreBuffers = listener.notifyBufferAvailable(new
NetworkBuffer(segment, this));
- if (!success || needMoreBuffers) {
+ if (needMoreBuffers) {
synchronized (availableMemorySegments) {
if (isDestroyed) {
// cleanup tasks how they would have
been done if we only had one synchronized block
- if (needMoreBuffers) {
-
listener.notifyBufferDestroyed();
- }
- if (!success) {
- returnMemorySegment(segment);
- }
+ listener.notifyBufferDestroyed();
} else {
- if (needMoreBuffers) {
-
registeredListeners.add(listener);
- }
- if (!success) {
- if
(numberOfRequestedMemorySegments > currentPoolSize) {
-
returnMemorySegment(segment);
- } else {
-
availableMemorySegments.add(segment);
-
availableMemorySegments.notify();
- }
- }
+ registeredListeners.add(listener);
--- End diff --
If there is an exception being thrown, this will not be called. Is that how
it supposed to be?
---
