[iotdb] branch master updated: [IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator[IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator[IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator

[ja...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new fb02f12401b [IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator[IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator[IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator
fb02f12401b is described below

commit fb02f12401be1f528044b42ed677aadb39e6c27d
Author: Liao Lanyu <14...@qq.com>
AuthorDate: Fri Jun 2 09:13:48 2023 +0800

    [IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator[IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator[IOTDB-5963] Make sure that TsBlock blocked on memory is added in queue before the next TsBlock returned by root operator
---
 .../apache/iotdb/db/mpp/execution/exchange/SharedTsBlockQueue.java  | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SharedTsBlockQueue.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SharedTsBlockQueue.java
index 905199f76e6..9c3ac0f5e74 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SharedTsBlockQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SharedTsBlockQueue.java
@@ -231,6 +231,7 @@ public class SharedTsBlockQueue {
 
     // reserve memory failed, we should wait until there is enough memory
     if (!pair.right) {
+      SettableFuture<Void> channelBlocked = SettableFuture.create();
       blockedOnMemory.addListener(
           () -> {
             synchronized (this) {
@@ -238,6 +239,7 @@ public class SharedTsBlockQueue {
               if (!blocked.isDone()) {
                 blocked.set(null);
               }
+              channelBlocked.set(null);
             }
           },
           // Use directExecutor() here could lead to deadlock. Thread A holds lock of
@@ -246,14 +248,14 @@ public class SharedTsBlockQueue {
           // Thread B holds lock of SharedTsBlockQueueB and tries to invoke the listener of
           // SharedTsBlockQueueA
           executorService);
+      return channelBlocked;
     } else { // reserve memory succeeded, add the TsBlock directly
       queue.add(tsBlock);
       if (!blocked.isDone()) {
         blocked.set(null);
       }
+      return blockedOnMemory;
     }
-
-    return blockedOnMemory;
   }
 
   /** Destroy the queue and complete the future. Should only be called in normal case */