public ListenableFuture<Void> add(TsBlock tsBlock) {
    if (closed) {
      // queue may have been closed
      return immediateVoidFuture();
    }

    Validate.notNull(tsBlock, "TsBlock cannot be null");
    Validate.isTrue(
        blockedOnMemory == null || blockedOnMemory.isDone(), "SharedTsBlockQueue is full");
    if (!alreadyRegistered) {
      localMemoryManager
          .getQueryPool()
          .registerPlanNodeIdToQueryMemoryMap(
              localFragmentInstanceId.queryId, fullFragmentInstanceId, localPlanNodeId);
      alreadyRegistered = true;
    }
    Pair<ListenableFuture<Void>, Boolean> pair =
        localMemoryManager
            .getQueryPool()
            .reserve(
                localFragmentInstanceId.getQueryId(),
                fullFragmentInstanceId,
                localPlanNodeId,
                tsBlock.getSizeInBytes(),
                maxBytesCanReserve);
    blockedOnMemory = pair.left;
    bufferRetainedSizeInBytes += tsBlock.getSizeInBytes();

    // reserve memory failed, we should wait until there is enough memory
    if (!Boolean.TRUE.equals(pair.right)) {
      SettableFuture<Void> channelBlocked = SettableFuture.create();
      blockedOnMemory.addListener(
          () -> {
            synchronized (this) {
              queue.add(tsBlock);
              if (!blocked.isDone()) {
                blocked.set(null);
              }
              channelBlocked.set(null);
            }
          },
          // Use directExecutor() here could lead to deadlock. Thread A holds lock of
          // SharedTsBlockQueueA and tries to invoke the listener of
          // SharedTsBlockQueueB(when freeing memory to complete MemoryReservationFuture) while
          // 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;
    }
  }
Add TsBlocks to the queue. Except the first invocation, this method should be invoked only when
the returned future of last invocation completes.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/SharedTsBlockQueue.java