avoid notify

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java

@@ -29,6 +29,7 @@
 public class MemoryControlledWALEntryQueue {
 
   private final BlockingQueue<WALEntry> queue;
+  private static final Object nonFullCondition = new Object();
 
   public MemoryControlledWALEntryQueue() {
     queue = new LinkedBlockingQueue<>();
@@ -38,19 +39,29 @@ public WALEntry poll(long timeout, TimeUnit unit) throws InterruptedException {
     WALEntry e = queue.poll(timeout, unit);
     if (e != null) {
       SystemInfo.getInstance().getWalBufferQueueMemoryBlock().release(getElementSize(e));
+      synchronized (nonFullCondition) {
+        nonFullCondition.notifyAll();
+      }
     }
     return e;
   }
 
   public void put(WALEntry e) throws InterruptedException {
     long elementSize = getElementSize(e);
-    SystemInfo.getInstance().getWalBufferQueueMemoryBlock().allocateUntilAvailable(elementSize);
+    synchronized (nonFullCondition) {
+      while (!SystemInfo.getInstance().getWalBufferQueueMemoryBlock().allocate(elementSize)) {
+        nonFullCondition.wait();
+      }
+    }
     queue.put(e);
   }
 
   public WALEntry take() throws InterruptedException {
     WALEntry e = queue.take();
     SystemInfo.getInstance().getWalBufferQueueMemoryBlock().release(getElementSize(e));
+    synchronized (nonFullCondition) {
+      nonFullCondition.notifyAll();
+    }
     return e;
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java

@@ -277,7 +277,7 @@ public void addCompactionMemoryCost(
     }
     boolean allocateResult =
         waitUntilAcquired
-            ? compactionMemoryBlock.allocateUntilAvailable(memoryCost)
+            ? compactionMemoryBlock.allocateUntilAvailable(memoryCost, 100)
             : compactionMemoryBlock.allocate(memoryCost);
     if (!allocateResult) {
       throw new CompactionMemoryNotEnoughException(

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/IMemoryBlock.java

@@ -61,6 +61,15 @@ public abstract class IMemoryBlock implements AutoCloseable {
    */
   public abstract boolean allocateUntilAvailable(final long sizeInByte) throws InterruptedException;
 
+  /**
+   * Allocate memory managed by this memory block until the required memory is available
+   *
+   * @param sizeInByte the size of memory to be allocated, should positive
+   * @param timeInterval the time interval to wait for memory to be available
+   */
+  public abstract boolean allocateUntilAvailable(final long sizeInByte, long timeInterval)
+      throws InterruptedException;
+
   /**
    * Try to record memory managed by this memory block
    *

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryBlock.java

@@ -77,12 +77,24 @@ public boolean allocateUntilAvailable(long sizeInByte) throws InterruptedExcepti
     return true;
   }
 
+  @Override
+  public boolean allocateUntilAvailable(long sizeInByte, long timeInterval)
+      throws InterruptedException {
+    long originSize = memoryUsageInBytes.get();
+    while (true) {
+      boolean canUpdate = originSize + sizeInByte <= maxMemorySizeInByte;
+      if (canUpdate && memoryUsageInBytes.compareAndSet(originSize, originSize + sizeInByte)) {
+        break;
+      }
+      Thread.sleep(TimeUnit.MILLISECONDS.toMillis(timeInterval));
+      originSize = memoryUsageInBytes.get();
+    }
+    return true;
+  }
+
   @Override
   public void release(long sizeInByte) {
     memoryUsageInBytes.addAndGet(-sizeInByte);
-    synchronized (memoryUsageInBytes) {
-      memoryUsageInBytes.notifyAll();
-    }
   }
 
   @Override

iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/memory/MemoryManagerTest.java

@@ -10,4 +10,4 @@ public void TestGetName() {
     MemoryManager memoryManager = MemoryManager.global();
     Assert.assertEquals("GlobalMemoryManager", memoryManager.getName());
   }
-}
+}