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?


---

Reply via email to