feat(server): adoption for off-heap memory management in kout module

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java

@@ -26,6 +26,7 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 
 import org.apache.hugegraph.HugeGraph;
@@ -34,6 +35,9 @@
 import org.apache.hugegraph.backend.id.Id;
 import org.apache.hugegraph.backend.query.QueryResults;
 import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.memory.MemoryManager;
+import org.apache.hugegraph.memory.pool.MemoryPool;
+import org.apache.hugegraph.memory.pool.impl.TaskMemoryPool;
 import org.apache.hugegraph.structure.HugeVertex;
 import org.apache.hugegraph.traversal.algorithm.HugeTraverser;
 import org.apache.hugegraph.traversal.algorithm.KoutTraverser;
@@ -93,27 +97,40 @@
                   "'{}', max degree '{}', capacity '{}' and limit '{}'",
                   graph, source, direction, edgeLabel, depth,
                   nearest, maxDegree, capacity, limit);
+        MemoryPool queryPool = MemoryManager.getInstance().addQueryMemoryPool();
+        Optional.ofNullable(queryPool).ifPresent(pool -> {
+            MemoryPool currentTaskPool = pool.addChildPool("kout-main-task");
+            MemoryManager.getInstance()
+                         .bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(),
+                                                          (TaskMemoryPool) currentTaskPool);
+            MemoryPool currentOperationPool = currentTaskPool.addChildPool("kout-main-operation");
+        });
 
-        ApiMeasurer measure = new ApiMeasurer();
+        try {
+            ApiMeasurer measure = new ApiMeasurer();
 
-        Id sourceId = VertexAPI.checkAndParseVertexId(source);
-        Directions dir = Directions.convert(EdgeAPI.parseDirection(direction));
+            Id sourceId = VertexAPI.checkAndParseVertexId(source);
+            Directions dir = Directions.convert(EdgeAPI.parseDirection(direction));
 
-        HugeGraph g = graph(manager, graph);
+            HugeGraph g = graph(manager, graph);
 
-        Set<Id> ids;
-        try (KoutTraverser traverser = new KoutTraverser(g)) {
-            ids = traverser.kout(sourceId, dir, edgeLabel, depth,
-                                 nearest, maxDegree, capacity, limit);
-            measure.addIterCount(traverser.vertexIterCounter.get(),
-                                 traverser.edgeIterCounter.get());
-        }
+            Set<Id> ids;
+            try (KoutTraverser traverser = new KoutTraverser(g)) {
+                ids = traverser.kout(sourceId, dir, edgeLabel, depth,
+                                     nearest, maxDegree, capacity, limit);
+                measure.addIterCount(traverser.vertexIterCounter.get(),
+                                     traverser.edgeIterCounter.get());
+            }
 
-        if (count_only) {
-            return manager.serializer(g, measure.measures())
-                          .writeMap(ImmutableMap.of("vertices_size", ids.size()));
+            if (count_only) {
+                return manager.serializer(g, measure.measures())
+                              .writeMap(ImmutableMap.of("vertices_size", ids.size()));
+            }
+            return manager.serializer(g, measure.measures()).writeList("vertices", ids);
+        } finally {
+            Optional.ofNullable(queryPool)
+                    .ifPresent(pool -> MemoryManager.getInstance().gcQueryMemoryPool(pool));
         }
-        return manager.serializer(g, measure.measures()).writeList("vertices", ids);
     }
 
     @POST

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java

@@ -260,6 +260,7 @@ private Iterator<HugeVertex> queryVerticesByIds(IdQuery query) {
                 return QueryResults.emptyIterator();
             }
             if (needCacheVertex(vertex)) {
+                vertex.convertIdToOnHeapIfNeeded();
                 this.verticesCache.update(vertex.id(), vertex);
             }
             return QueryResults.iterator(vertex);
@@ -295,6 +296,7 @@ private Iterator<HugeVertex> queryVerticesByIds(IdQuery query) {
             for (HugeVertex vertex : listIterator.list()) {
                 // Skip large vertex
                 if (needCacheVertex(vertex)) {
+                    vertex.convertIdToOnHeapIfNeeded();
                     this.verticesCache.update(vertex.id(), vertex);
                 }
             }
@@ -353,6 +355,7 @@ protected Iterator<HugeEdge> queryEdgesFromBackend(Query query) {
         if (edges.isEmpty()) {
             this.edgesCache.update(cacheKey, Collections.emptyList());
         } else if (edges.size() <= MAX_CACHE_EDGES_PER_QUERY) {
+            edges.forEach(HugeEdge::convertIdToOnHeapIfNeeded);
             this.edgesCache.update(cacheKey, edges);
         }
 
@@ -378,6 +381,7 @@ protected void commitMutation2Backend(BackendMutation... mutations) {
                     vertexIds[vertexOffset++] = vertex.id();
                     if (needCacheVertex(vertex)) {
                         // Update cache
+                        vertex.convertIdToOnHeapIfNeeded();
                         this.verticesCache.updateIfPresent(vertex.id(), vertex);
                     } else {
                         // Skip large vertex

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransaction.java

@@ -176,6 +176,9 @@ private void clearCache(boolean notify) {
     private void updateCache(SchemaElement schema) {
         this.resetCachedAllIfReachedCapacity();
 
+        // convert schema.id to on heap if needed.
+        schema.convertIdToOnHeapIfNeeded();
+
         // update id cache
         Id prefixedId = generateId(schema.type(), schema.id());
         this.idCache.update(prefixedId, schema);
@@ -204,14 +207,20 @@ private void invalidateCache(HugeType type, Id id) {
         this.arrayCaches.remove(type, id);
     }
 
+    /**
+     * Ids used in cache must be on-heap object
+     */
     private static Id generateId(HugeType type, Id id) {
         // NOTE: it's slower performance to use:
         // String.format("%x-%s", type.code(), name)
-        return IdGenerator.of(type.string() + "-" + id.asString());
+        return new IdGenerator.StringId(type.string() + "-" + id.asString());
     }
 
+    /**
+     * Ids used in cache must be on-heap object
+     */
     private static Id generateId(HugeType type, String name) {
-        return IdGenerator.of(type.string() + "-" + name);
+        return new IdGenerator.StringId(type.string() + "-" + name);
     }
 
     @Override

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java

@@ -42,6 +42,7 @@
 import com.google.common.collect.ImmutableSet;
 
 public class CachedSchemaTransactionV2 extends SchemaTransactionV2 {
+
     private final Cache<Id, Object> idCache;
     private final Cache<Id, Object> nameCache;
 
@@ -51,8 +52,8 @@
     private EventListener cacheEventListener;
 
     public CachedSchemaTransactionV2(MetaDriver metaDriver,
-                                   String cluster,
-                                   HugeGraphParams graphParams) {
+                                     String cluster,
+                                     HugeGraphParams graphParams) {
         super(metaDriver, cluster, graphParams);
 
         final long capacity = graphParams.configuration()
@@ -223,6 +224,9 @@
     private void updateCache(SchemaElement schema) {
         this.resetCachedAllIfReachedCapacity();
 
+        // convert schema.id to on heap if needed.
+        schema.convertIdToOnHeapIfNeeded();
+
         // update id cache
         Id prefixedId = generateId(schema.type(), schema.id());
         this.idCache.update(prefixedId, schema);
@@ -268,10 +272,12 @@
             value = super.getSchema(type, id);
             if (value != null) {
                 this.resetCachedAllIfReachedCapacity();
+                // convert schema.id to on heap if needed.
+                SchemaElement schema = (SchemaElement) value;
+                schema.convertIdToOnHeapIfNeeded();
 
-                this.idCache.update(prefixedId, value);
+                this.idCache.update(prefixedId, schema);
 
-                SchemaElement schema = (SchemaElement) value;
                 Id prefixedName = generateId(schema.type(), schema.name());
                 this.nameCache.update(prefixedName, schema);
             }
@@ -321,6 +327,9 @@
             if (results.size() <= free) {
                 // Update cache
                 for (T schema : results) {
+                    // convert schema.id to on heap if needed.
+                    schema.convertIdToOnHeapIfNeeded();
+
                     Id prefixedId = generateId(schema.type(), schema.id());
                     this.idCache.update(prefixedId, schema);
 
@@ -481,7 +490,7 @@
     }
 
     private static class CachedTypes
-        extends ConcurrentHashMap<HugeType, Boolean> {
+            extends ConcurrentHashMap<HugeType, Boolean> {
 
         private static final long serialVersionUID = -2215549791679355996L;
     }

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/id/IdGenerator.java

@@ -22,6 +22,7 @@
 
 import org.apache.hugegraph.backend.id.Id.IdType;
 import org.apache.hugegraph.backend.serializer.BytesBuffer;
+import org.apache.hugegraph.memory.consumer.factory.IdFactory;
 import org.apache.hugegraph.structure.HugeVertex;
 import org.apache.hugegraph.util.E;
 import org.apache.hugegraph.util.LongEncoding;
@@ -35,19 +36,19 @@
     public abstract Id generate(HugeVertex vertex);
 
     public static Id of(String id) {
-        return new StringId(id);
+        return IdFactory.getInstance().newStringId(id);
     }
 
     public static Id of(UUID id) {
-        return new UuidId(id);
+        return IdFactory.getInstance().newUuidId(id);
     }
 
     public static Id of(String id, boolean uuid) {
-        return uuid ? new UuidId(id) : new StringId(id);
+        return uuid ? IdFactory.getInstance().newUuidId(id) : IdFactory.getInstance().newStringId(id);
     }
 
     public static Id of(long id) {
-        return new LongId(id);
+        return IdFactory.getInstance().newLongId(id);
     }
 
     public static Id of(Object id) {
@@ -66,11 +67,11 @@
     public static Id of(byte[] bytes, IdType type) {
         switch (type) {
             case LONG:
-                return new LongId(bytes);
+                return IdFactory.getInstance().newLongId(bytes);
             case UUID:
-                return new UuidId(bytes);
+                return IdFactory.getInstance().newUuidId(bytes);
             case STRING:
-                return new StringId(bytes);
+                return IdFactory.getInstance().newStringId(bytes);
             default:
                 throw new AssertionError("Invalid id type " + type);
         }

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java

@@ -22,11 +22,13 @@
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 
 import org.apache.commons.lang3.NotImplementedException;
 import org.apache.hugegraph.backend.query.ConditionQuery;
 import org.apache.hugegraph.backend.query.Query;
 import org.apache.hugegraph.backend.serializer.BytesBuffer;
+import org.apache.hugegraph.memory.MemoryManager;
 import org.apache.hugegraph.type.HugeType;
 import org.apache.hugegraph.type.define.Directions;
 import org.apache.hugegraph.type.define.HugeKeys;
@@ -135,7 +137,16 @@
 
     public abstract void clear(Session session);
 
-    public abstract Iterator<BackendEntry> query(Session session, Query query);
+    public Iterator<BackendEntry> query(Session session, Query query) {
+        Optional.ofNullable(MemoryManager.getInstance()
+                                         .getCorrespondingTaskMemoryPool(
+                                                 Thread.currentThread().getName()))
+                .ifPresent(currentTaskPool -> {
+                    // Some system-query(not requested by user) don't have memory pool.
+                    currentTaskPool.addChildPool("BackendTable-Iterator");
+                });
+        return null;
+    }
 
     public Iterator<BackendEntry> queryOlap(Session session, Query query) {
         throw new NotImplementedException();

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java

@@ -687,7 +687,7 @@ public static synchronized CoreOptions instance() {
             "memory.mode",
             "The memory mode used for query in HugeGraph.",
             disallowEmpty(),
-            "off-heap"
+            "disable"
     );
 
     public static final ConfigOption<Long> MAX_MEMORY_CAPACITY = new ConfigOption<>(

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

@@ -77,7 +77,7 @@ public class MemoryManager {
     private final MemoryArbitrator memoryArbitrator;
     private final ExecutorService arbitrateExecutor;
 
-    private static MemoryMode MEMORY_MODE = MemoryMode.ENABLE_OFF_HEAP_MANAGEMENT;
+    private static MemoryMode MEMORY_MODE = MemoryMode.DISABLE_MEMORY_MANAGEMENT;
 
     private MemoryManager() {
         this.memoryArbitrator = new MemoryArbitratorImpl(this);
@@ -89,6 +89,10 @@ private MemoryManager() {
     }
 
     public MemoryPool addQueryMemoryPool() {
+        if (MEMORY_MODE == MemoryMode.DISABLE_MEMORY_MANAGEMENT) {
+            return null;
+        }
+
         int count = queryMemoryPools.size();
         String poolName =
                 QUERY_MEMORY_POOL_NAME_PREFIX + DELIMINATOR + count + DELIMINATOR +