[GitHub] [flink] akalash commented on a change in pull request #11877: [FLINK-16641][network] Announce sender's backlog to solve the deadlock issue without exclusive buffers

[GitBox <gi...@apache.org>]

akalash commented on a change in pull request #11877:
URL: https://github.com/apache/flink/pull/11877#discussion_r662239062



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java
##########
@@ -215,9 +221,15 @@ public void recycle(MemorySegment segment) {
     }
 
     void releaseFloatingBuffers() {
+        Queue<Buffer> buffers;
         synchronized (bufferQueue) {
             numRequiredBuffers = 0;
-            bufferQueue.releaseFloatingBuffers();
+            buffers = bufferQueue.clearFloatingBuffers();
+        }
+
+        // recycle all buffers out of the synchronization block to avoid dead lock

Review comment:
       As I understand from your explanation, it relates to this BufferManager#notifyBufferAvailable. But I still don't get which threads can call the releaseFloatingBuffers simultaneously. It is not so important, but just for curiosity if you have the stacktrace of this deadlock can you share it with me(or just tell me which test)?
   Anyway, your changes look correct here.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
##########
@@ -288,9 +288,7 @@ BufferAndBacklog pollBuffer() {
 
             if (buffers.isEmpty()) {
                 flushRequested = false;
-            }
-
-            while (!buffers.isEmpty()) {
+            } else {

Review comment:
       Ok, I think I understand it now. Please, tell me I indeed understand the scenario right:
   
   - Suppose the buffers contain one unfinished Buffer.
   - The flush was requested
   - When pollBuffer() was called it collects all readable bytes from this unfinished buffer and getBuffersInBacklog returns 1(because `flushRequested` set to true)
   - One more credit is requested from downstream because getBuffersInBacklog equal to 1.
   - When pollBuffer() was called again it reads nothing because all data was read last time. But it already has one credit that should be released(it is exactly what you told about?).




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org