complete OOM UT and fix bugs

Author: Pengzna

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/MemoryManager.java

@@ -80,7 +80,7 @@ public void gcQueryMemoryPool(MemoryPool pool) {
         LOG.info("Manager gc query memory pool {}", pool);
         queryMemoryPools.remove(pool);
         long reclaimedMemory = pool.getAllocatedBytes();
-        pool.releaseSelf(String.format("GC query memory pool %s", pool));
+        pool.releaseSelf(String.format("GC query memory pool %s", pool), false);
         currentAvailableMemoryInBytes.addAndGet(reclaimedMemory);
     }
 

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/pool/AbstractMemoryPool.java

@@ -96,35 +96,41 @@ public long tryToReclaimLocalMemory(long neededBytes) {
      * called when one layer pool is successfully executed and exited.
      */
     @Override
-    public void releaseSelf(String reason) {
-        this.memoryActionLock.lock();
+    public void releaseSelf(String reason, boolean isTriggeredInternal) {
         try {
-            if (this.isBeingArbitrated.get()) {
-                this.condition.await();
+            if (!isTriggeredInternal) {
+                this.memoryActionLock.lock();
+                if (this.isBeingArbitrated.get()) {
+                    this.condition.await();
+                }
             }
             LOG.info("[{}] starts to releaseSelf because of {}", this, reason);
             this.isClosed = true;
-            // update father
-            Optional.ofNullable(this.parent).ifPresent(parent -> parent.gcChildPool(this, false));
+            // gc self from father
+            Optional.ofNullable(this.parent).ifPresent(parent -> parent.gcChildPool(this, false,
+                                                                                    isTriggeredInternal));
+            // gc all children
             for (MemoryPool child : this.children) {
-                gcChildPool(child, true);
+                gcChildPool(child, true, isTriggeredInternal);
             }
             LOG.info("[{}] finishes to releaseSelf", this);
         } catch (InterruptedException e) {
             LOG.error("Failed to release self because ", e);
             Thread.currentThread().interrupt();
         } finally {
-            this.memoryActionLock.unlock();
+            if (!isTriggeredInternal) {
+                this.memoryActionLock.unlock();
+            }
             // Make these objs be GCed by JVM quickly.
             this.parent = null;
             this.children.clear();
         }
     }
 
     @Override
-    public void gcChildPool(MemoryPool child, boolean force) {
+    public void gcChildPool(MemoryPool child, boolean force, boolean isTriggeredInternal) {
         if (force) {
-            child.releaseSelf(String.format("[%s] releaseChildPool", this));
+            child.releaseSelf(String.format("[%s] releaseChildPool", this), isTriggeredInternal);
         }
         // reclaim child's memory and update stats
         this.stats.setAllocatedBytes(

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/pool/MemoryPool.java

@@ -34,9 +34,22 @@ public interface MemoryPool {
 
     Object tryToAcquireMemoryInternal(long bytes);
 
-    void releaseSelf(String reason);
-
-    void gcChildPool(MemoryPool child, boolean force);
+    /**
+     * Release all self's resources. Called by user or called automatically by itself when OOM.
+     *
+     * @param reason:              release reason, for logging.
+     * @param isTriggeredInternal: if true, it is called automatically. if false, called by user.
+     */
+    void releaseSelf(String reason, boolean isTriggeredInternal);
+
+    /**
+     * Called by `releaseSelf` to release children's resource.
+     *
+     * @param child:               child pool
+     * @param force:               if false, called to gc self from father
+     * @param isTriggeredInternal: passed from upper caller `releaseSelf`
+     */
+    void gcChildPool(MemoryPool child, boolean force, boolean isTriggeredInternal);
 
     long getAllocatedBytes();
 

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/pool/impl/OperatorMemoryPool.java

@@ -54,8 +54,8 @@ public void bindMemoryConsumer(MemoryConsumer memoryConsumer) {
     }
 
     @Override
-    public void releaseSelf(String reason) {
-        super.releaseSelf(reason);
+    public void releaseSelf(String reason, boolean isTriggeredInternal) {
+        super.releaseSelf(reason, isTriggeredInternal);
         // since it is already closed, its stats will not be updated. so here we can use its
         // stats out of memoryActionLock.
         this.memoryAllocator.returnMemoryToManager(getAllocatedBytes());
@@ -125,7 +125,7 @@ public Object requireMemory(long bytes) {
             LOG.warn("[{}] detected an OOM exception when request memory, will ABORT this " +
                      "query and release corresponding memory...",
                      this);
-            findRootQueryPool().releaseSelf(String.format(e.getMessage()));
+            findRootQueryPool().releaseSelf(String.format(e.getMessage()), true);
             return null;
         } finally {
             this.memoryActionLock.unlock();

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/pool/impl/TaskMemoryPool.java

@@ -37,9 +37,9 @@ public TaskMemoryPool(MemoryPool parent, String poolName, MemoryManager memoryMa
     }
 
     @Override
-    public synchronized void releaseSelf(String reason) {
+    public void releaseSelf(String reason, boolean isTriggeredInternal) {
+        super.releaseSelf(reason, isTriggeredInternal);
         this.memoryManager.removeCorrespondingTaskMemoryPool(Thread.currentThread().getName());
-        super.releaseSelf(reason);
     }
 
     @Override

hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/memory/MemoryManageTest.java

@@ -114,4 +114,19 @@ public void testOOM() {
         ByteBuf memoryBlock = (ByteBuf) query2Task1Operator1MemoryPool.requireMemory(requireBytes);
         Assert.assertNull(memoryBlock);
     }
+
+    @Test
+    public void testReleaseMemory() {
+
+    }
+
+    @Test
+    public void testLocalArbitration() {
+
+    }
+
+    @Test
+    public void testGlobalArbitration() {
+
+    }
 }