From 4d4cb2230a8cbab5d9b4ca50fa7a13982167ea42 Mon Sep 17 00:00:00 2001 From: Peng Junzhi Date: Mon, 9 Dec 2024 22:51:47 +0800 Subject: [PATCH 1/6] off-heap adoption for id --- .../apache/hugegraph/backend/id/IdGenerator.java | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/id/IdGenerator.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/id/IdGenerator.java index 17cc11684c..e1a3294f39 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/id/IdGenerator.java +++ b/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 class IdGenerator { 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(Object id) { 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); } From 29b78457cd77877461ac9caaff54ead26e677af5 Mon Sep 17 00:00:00 2001 From: Peng Junzhi Date: Mon, 9 Dec 2024 23:53:33 +0800 Subject: [PATCH 2/6] off-heap adoption for property & kout query pool --- .../java/org/apache/hugegraph/api/traversers/KoutAPI.java | 6 ++++++ .../main/java/org/apache/hugegraph/structure/HugeEdge.java | 5 ++++- .../org/apache/hugegraph/structure/HugeEdgeProperty.java | 7 +++++-- .../java/org/apache/hugegraph/structure/HugeVertex.java | 5 ++++- 4 files changed, 19 insertions(+), 4 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java index e15cc174ea..cc62992245 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java @@ -34,6 +34,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,6 +96,9 @@ public String get(@Context GraphManager manager, "'{}', max degree '{}', capacity '{}' and limit '{}'", graph, source, direction, edgeLabel, depth, nearest, maxDegree, capacity, limit); + MemoryPool queryPool = MemoryManager.getInstance().addQueryMemoryPool(); + MemoryPool currentTaskPool = queryPool.addChildPool("kout-main-task"); + MemoryManager.getInstance().bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), (TaskMemoryPool) currentTaskPool); ApiMeasurer measure = new ApiMeasurer(); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdge.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdge.java index 5bf7011a6f..ac23a7a1e4 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdge.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdge.java @@ -30,6 +30,7 @@ import org.apache.hugegraph.backend.query.QueryResults; import org.apache.hugegraph.backend.serializer.BytesBuffer; import org.apache.hugegraph.backend.tx.GraphTransaction; +import org.apache.hugegraph.memory.consumer.factory.PropertyFactory; import org.apache.hugegraph.perf.PerfUtil.Watched; import org.apache.hugegraph.schema.EdgeLabel; import org.apache.hugegraph.schema.PropertyKey; @@ -235,7 +236,9 @@ protected GraphTransaction tx() { @Watched(prefix = "edge") @Override protected HugeEdgeProperty newProperty(PropertyKey pkey, V val) { - return new HugeEdgeProperty<>(this, pkey, val); + Class valueType = (Class) val.getClass(); + PropertyFactory propertyFactory = PropertyFactory.getInstance(valueType); + return propertyFactory.newHugeEdgeProperty(this, pkey, val); } @Watched(prefix = "edge") diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdgeProperty.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdgeProperty.java index f89e4f0cf4..f0f99c859f 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdgeProperty.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdgeProperty.java @@ -17,6 +17,7 @@ package org.apache.hugegraph.structure; +import org.apache.hugegraph.memory.consumer.factory.PropertyFactory; import org.apache.hugegraph.schema.EdgeLabel; import org.apache.hugegraph.schema.PropertyKey; import org.apache.hugegraph.type.HugeType; @@ -67,7 +68,9 @@ public int hashCode() { public HugeEdgeProperty switchEdgeOwner() { assert this.owner instanceof HugeEdge; - return new HugeEdgeProperty<>(((HugeEdge) this.owner).switchOwner(), - this.pkey, this.value); + Class valueType = (Class) this.value.getClass(); + PropertyFactory propertyFactory = PropertyFactory.getInstance(valueType); + return propertyFactory.newHugeEdgeProperty(((HugeEdge) this.owner).switchOwner(), + this.pkey, this.value); } } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeVertex.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeVertex.java index 389b0f0e8c..29abc007e9 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeVertex.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeVertex.java @@ -40,6 +40,7 @@ import org.apache.hugegraph.backend.tx.GraphTransaction; import org.apache.hugegraph.config.CoreOptions; import org.apache.hugegraph.masterelection.StandardClusterRoleStore; +import org.apache.hugegraph.memory.consumer.factory.PropertyFactory; import org.apache.hugegraph.perf.PerfUtil.Watched; import org.apache.hugegraph.schema.EdgeLabel; import org.apache.hugegraph.schema.PropertyKey; @@ -492,7 +493,9 @@ protected GraphTransaction tx() { @Watched(prefix = "vertex") @Override protected HugeVertexProperty newProperty(PropertyKey pkey, V val) { - return new HugeVertexProperty<>(this, pkey, val); + Class valueType = (Class) val.getClass(); + PropertyFactory propertyFactory = PropertyFactory.getInstance(valueType); + return propertyFactory.newHugeVertexProperty(this, pkey, val); } @Watched(prefix = "vertex") From a402ddb82cfb9be1643b9e35b095766c62605ada Mon Sep 17 00:00:00 2001 From: Peng Junzhi <201250214@smail.nju.edu.cn> Date: Tue, 10 Dec 2024 11:26:56 +0800 Subject: [PATCH 3/6] wip --- .../hugegraph/api/traversers/KoutAPI.java | 5 ++++- .../org/apache/hugegraph/util/Consumers.java | 18 ++++++++++++++++-- 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java index cc62992245..7b09121789 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java @@ -98,7 +98,10 @@ public String get(@Context GraphManager manager, nearest, maxDegree, capacity, limit); MemoryPool queryPool = MemoryManager.getInstance().addQueryMemoryPool(); MemoryPool currentTaskPool = queryPool.addChildPool("kout-main-task"); - MemoryManager.getInstance().bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), (TaskMemoryPool) currentTaskPool); + MemoryManager.getInstance() + .bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), + (TaskMemoryPool) currentTaskPool); + MemoryPool currentOperationPool = currentTaskPool.addChildPool("kout-main-operation"); ApiMeasurer measure = new ApiMeasurer(); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java index daa54ee958..6d63ac1cf1 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java @@ -35,6 +35,9 @@ import org.apache.hugegraph.HugeException; import org.apache.hugegraph.config.CoreOptions; +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.task.TaskManager.ContextCallable; import org.slf4j.Logger; @@ -105,12 +108,23 @@ public void start(String name) { this.workers, name, this.queueSize); for (int i = 0; i < this.workers; i++) { this.runningFutures.add( - this.executor.submit(new ContextCallable<>(this::runAndDone))); + this.executor.submit( + new ContextCallable<>(() -> this.runAndDone(MemoryManager.getInstance() + .getCorrespondingTaskMemoryPool( + Thread.currentThread() + .getName()) + .findRootQueryPool())))); } } - private Void runAndDone() { + private Void runAndDone(MemoryPool queryPool) { try { + MemoryPool currentTaskPool = queryPool.addChildPool("kout-consume-task"); + MemoryManager.getInstance() + .bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), + (TaskMemoryPool) currentTaskPool); + MemoryPool currentOperationPool = + currentTaskPool.addChildPool("kout-consume-operation"); this.run(); } catch (Throwable e) { if (e instanceof StopExecution) { From 88604cf5f0dca8a1642e4e99b3da9a7500941131 Mon Sep 17 00:00:00 2001 From: Peng Junzhi <201250214@smail.nju.edu.cn> Date: Tue, 10 Dec 2024 15:55:00 +0800 Subject: [PATCH 4/6] backend entry iter --- .../hugegraph/api/traversers/KoutAPI.java | 34 +++++++++++-------- .../hugegraph/backend/store/BackendTable.java | 11 +++++- .../org/apache/hugegraph/util/Consumers.java | 13 +++---- .../backend/store/hstore/HstoreTable.java | 1 + .../backend/store/rocksdb/RocksDBTable.java | 1 + 5 files changed, 38 insertions(+), 22 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java index 7b09121789..e226e5eb00 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java @@ -103,26 +103,30 @@ public String get(@Context GraphManager manager, (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 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 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 { + queryPool.releaseSelf("Complete kout query", false); } - return manager.serializer(g, measure.measures()).writeList("vertices", ids); } @POST diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java index a9ee4b5c72..54c73ae6e3 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java @@ -27,6 +27,8 @@ 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.memory.pool.MemoryPool; import org.apache.hugegraph.type.HugeType; import org.apache.hugegraph.type.define.Directions; import org.apache.hugegraph.type.define.HugeKeys; @@ -135,7 +137,14 @@ public static final String joinTableName(String prefix, String table) { public abstract void clear(Session session); - public abstract Iterator query(Session session, Query query); + public Iterator query(Session session, Query query) { + MemoryPool currentTaskPool = MemoryManager.getInstance() + .getCorrespondingTaskMemoryPool( + Thread.currentThread().getName()); + MemoryPool currentOperationPool = + currentTaskPool.addChildPool("BackendTable-Iterator"); + return null; + } public Iterator queryOlap(Session session, Query query) { throw new NotImplementedException(); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java index 6d63ac1cf1..a21a23d71a 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java @@ -118,13 +118,13 @@ public void start(String name) { } private Void runAndDone(MemoryPool queryPool) { + MemoryPool currentTaskPool = queryPool.addChildPool("kout-consume-task"); + MemoryManager.getInstance() + .bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), + (TaskMemoryPool) currentTaskPool); + MemoryPool currentOperationPool = + currentTaskPool.addChildPool("kout-consume-operation"); try { - MemoryPool currentTaskPool = queryPool.addChildPool("kout-consume-task"); - MemoryManager.getInstance() - .bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), - (TaskMemoryPool) currentTaskPool); - MemoryPool currentOperationPool = - currentTaskPool.addChildPool("kout-consume-operation"); this.run(); } catch (Throwable e) { if (e instanceof StopExecution) { @@ -139,6 +139,7 @@ private Void runAndDone(MemoryPool queryPool) { } finally { this.done(); this.latch.countDown(); + currentTaskPool.releaseSelf("Complete kout consume task", false); } return null; } diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreTable.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreTable.java index e1830111c3..19435bbd3b 100755 --- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreTable.java +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreTable.java @@ -312,6 +312,7 @@ public Number queryNumber(Session session, Query query) { @Override public Iterator query(Session session, Query query) { + super.query(session, query); if (query.limit() == 0L && !query.noLimit()) { // LOG.debug("Return empty result(limit=0) for query {}", query); return Collections.emptyIterator(); diff --git a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBTable.java b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBTable.java index 824986d22c..22bca78008 100644 --- a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBTable.java +++ b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBTable.java @@ -143,6 +143,7 @@ public Number queryNumber(RocksDBSessions.Session session, Query query) { @Override public Iterator query(RocksDBSessions.Session session, Query query) { + super.query(session, query); if (query.limit() == 0L && !query.noLimit()) { LOG.debug("Return empty result(limit=0) for query {}", query); return Collections.emptyIterator(); From a34287098656825c57d09793cb419a7c7ec62c75 Mon Sep 17 00:00:00 2001 From: Peng Junzhi <201250214@smail.nju.edu.cn> Date: Tue, 10 Dec 2024 23:19:48 +0800 Subject: [PATCH 5/6] fix bootstrap bugs --- .../hugegraph/backend/store/BackendTable.java | 14 +++-- .../memory/consumer/factory/IdFactory.java | 61 ++++++++++++++----- .../consumer/factory/PropertyFactory.java | 14 +++-- 3 files changed, 63 insertions(+), 26 deletions(-) diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java index 54c73ae6e3..a01ea1cb96 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java @@ -22,13 +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.memory.pool.MemoryPool; import org.apache.hugegraph.type.HugeType; import org.apache.hugegraph.type.define.Directions; import org.apache.hugegraph.type.define.HugeKeys; @@ -138,11 +138,13 @@ public static final String joinTableName(String prefix, String table) { public abstract void clear(Session session); public Iterator query(Session session, Query query) { - MemoryPool currentTaskPool = MemoryManager.getInstance() - .getCorrespondingTaskMemoryPool( - Thread.currentThread().getName()); - MemoryPool currentOperationPool = - currentTaskPool.addChildPool("BackendTable-Iterator"); + 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; } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/consumer/factory/IdFactory.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/consumer/factory/IdFactory.java index 976b9cb373..3423195fb4 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/consumer/factory/IdFactory.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/consumer/factory/IdFactory.java @@ -58,7 +58,9 @@ public BinaryBackendEntry.BinaryId newBinaryId(byte[] bytes, Id id) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new BinaryIdOffHeap(bytes, null, + return taskMemoryPool == null ? + new BinaryBackendEntry.BinaryId(bytes, id) : + new BinaryIdOffHeap(bytes, null, taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), (OffHeapObject) id); case DISABLE_MEMORY_MANAGEMENT: @@ -75,7 +77,9 @@ public IdGenerator.LongId newLongId(long id) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new LongIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new IdGenerator.LongId(id) : + new LongIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), id); case DISABLE_MEMORY_MANAGEMENT: default: @@ -91,7 +95,9 @@ public IdGenerator.LongId newLongId(byte[] bytes) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new LongIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new IdGenerator.LongId(bytes) : + new LongIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), bytes); case DISABLE_MEMORY_MANAGEMENT: default: @@ -107,7 +113,9 @@ public IdGenerator.ObjectId newObjectId(Object object) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new ObjectIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new IdGenerator.ObjectId(object) : + new ObjectIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), object); case DISABLE_MEMORY_MANAGEMENT: default: @@ -123,7 +131,9 @@ public CachedBackendStore.QueryId newQueryId(Query q) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new QueryIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new CachedBackendStore.QueryId(q) : + new QueryIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), q); case DISABLE_MEMORY_MANAGEMENT: default: @@ -139,7 +149,9 @@ public IdGenerator.StringId newStringId(String id) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new StringIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new IdGenerator.StringId(id) : + new StringIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), id); case DISABLE_MEMORY_MANAGEMENT: default: @@ -155,7 +167,9 @@ public IdGenerator.StringId newStringId(byte[] bytes) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new StringIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new IdGenerator.StringId(bytes) : + new StringIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), bytes); case DISABLE_MEMORY_MANAGEMENT: default: @@ -171,7 +185,9 @@ public IdGenerator.UuidId newUuidId(String id) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new UuidIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new IdGenerator.UuidId(id) : + new UuidIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), id); case DISABLE_MEMORY_MANAGEMENT: default: @@ -187,7 +203,9 @@ public IdGenerator.UuidId newUuidId(byte[] bytes) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new UuidIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new IdGenerator.UuidId(bytes) : + new UuidIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), bytes); case DISABLE_MEMORY_MANAGEMENT: default: @@ -203,7 +221,9 @@ public IdGenerator.UuidId newUuidId(UUID id) { .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new UuidIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), + return taskMemoryPool == null ? + new IdGenerator.UuidId(id) : + new UuidIdOffHeap(taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), id); case DISABLE_MEMORY_MANAGEMENT: default: @@ -224,7 +244,10 @@ public EdgeId newEdgeId(HugeVertex ownerVertex, .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new EdgeIdOffHeap(ownerVertex, + return taskMemoryPool == null ? + new EdgeId(ownerVertex, direction, edgeLabelId, subLabelId, sortValues, + otherVertex) : + new EdgeIdOffHeap(ownerVertex, direction, null, null, @@ -235,7 +258,7 @@ public EdgeId newEdgeId(HugeVertex ownerVertex, (OffHeapObject) edgeLabelId, (OffHeapObject) subLabelId, (OffHeapObject) otherVertex.id() - ); + ); case DISABLE_MEMORY_MANAGEMENT: default: return new EdgeId(ownerVertex, direction, edgeLabelId, subLabelId, sortValues, @@ -256,7 +279,10 @@ public EdgeId newEdgeId(Id ownerVertexId, .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new EdgeIdOffHeap((Id) null, + return taskMemoryPool == null ? + new EdgeId(ownerVertexId, direction, edgeLabelId, subLabelId, sortValues, + otherVertexId) : + new EdgeIdOffHeap((Id) null, direction, null, null, @@ -267,7 +293,7 @@ public EdgeId newEdgeId(Id ownerVertexId, (OffHeapObject) edgeLabelId, (OffHeapObject) subLabelId, (OffHeapObject) otherVertexId - ); + ); case DISABLE_MEMORY_MANAGEMENT: default: return new EdgeId(ownerVertexId, direction, edgeLabelId, subLabelId, sortValues, @@ -289,7 +315,10 @@ public EdgeId newEdgeId(Id ownerVertexId, .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new EdgeIdOffHeap(null, + return taskMemoryPool == null ? + new EdgeId(ownerVertexId, direction, edgeLabelId, subLabelId, sortValues, + otherVertexId, directed) : + new EdgeIdOffHeap(null, direction, null, null, @@ -301,7 +330,7 @@ public EdgeId newEdgeId(Id ownerVertexId, (OffHeapObject) edgeLabelId, (OffHeapObject) subLabelId, (OffHeapObject) otherVertexId - ); + ); case DISABLE_MEMORY_MANAGEMENT: default: return new EdgeId(ownerVertexId, direction, edgeLabelId, subLabelId, sortValues, diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/consumer/factory/PropertyFactory.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/consumer/factory/PropertyFactory.java index c628155eb1..112236fd94 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/consumer/factory/PropertyFactory.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/consumer/factory/PropertyFactory.java @@ -47,8 +47,11 @@ public HugeEdgeProperty newHugeEdgeProperty(HugeElement owner, PropertyKey ke .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new HugeEdgePropertyOffHeap<>( - taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), owner, key, value); + return taskMemoryPool == null ? + new HugeEdgeProperty<>(owner, key, value) : + new HugeEdgePropertyOffHeap<>( + taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), owner, key, + value); case DISABLE_MEMORY_MANAGEMENT: default: return new HugeEdgeProperty<>(owner, key, value); @@ -64,8 +67,11 @@ public HugeVertexProperty newHugeVertexProperty(HugeElement owner, PropertyKe .getCorrespondingTaskMemoryPool( Thread.currentThread() .getName()); - return new HugeVertexPropertyOffHeap<>( - taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), owner, key, value); + return taskMemoryPool == null ? + new HugeVertexProperty<>(owner, key, value) : + new HugeVertexPropertyOffHeap<>( + taskMemoryPool.getCurrentWorkingOperatorMemoryPool(), owner, key, + value); case DISABLE_MEMORY_MANAGEMENT: default: return new HugeVertexProperty<>(owner, key, value); From db3ee80a38619e3568e334800b4b80139b1b1cad Mon Sep 17 00:00:00 2001 From: Peng Junzhi <201250214@smail.nju.edu.cn> Date: Wed, 11 Dec 2024 17:13:10 +0800 Subject: [PATCH 6/6] cache use on-heap id && adopt for disabled mode --- .../hugegraph/api/traversers/KoutAPI.java | 16 ++++++----- .../backend/cache/CachedGraphTransaction.java | 4 +++ .../cache/CachedSchemaTransaction.java | 13 +++++++-- .../cache/CachedSchemaTransactionV2.java | 19 +++++++++---- .../apache/hugegraph/config/CoreOptions.java | 2 +- .../hugegraph/memory/MemoryManager.java | 6 ++++- .../hugegraph/schema/SchemaElement.java | 9 ++++++- .../apache/hugegraph/structure/HugeEdge.java | 7 +++++ .../hugegraph/structure/HugeVertex.java | 8 ++++++ .../traversal/algorithm/HugeTraverser.java | 2 +- .../org/apache/hugegraph/util/Consumers.java | 27 ++++++++++++------- 11 files changed, 86 insertions(+), 27 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java index e226e5eb00..65d59d8f8d 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java +++ b/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; @@ -97,11 +98,13 @@ public String get(@Context GraphManager manager, graph, source, direction, edgeLabel, depth, nearest, maxDegree, capacity, limit); MemoryPool queryPool = MemoryManager.getInstance().addQueryMemoryPool(); - MemoryPool currentTaskPool = queryPool.addChildPool("kout-main-task"); - MemoryManager.getInstance() - .bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), - (TaskMemoryPool) currentTaskPool); - MemoryPool currentOperationPool = currentTaskPool.addChildPool("kout-main-operation"); + 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"); + }); try { ApiMeasurer measure = new ApiMeasurer(); @@ -125,7 +128,8 @@ public String get(@Context GraphManager manager, } return manager.serializer(g, measure.measures()).writeList("vertices", ids); } finally { - queryPool.releaseSelf("Complete kout query", false); + Optional.ofNullable(queryPool) + .ifPresent(pool -> MemoryManager.getInstance().gcQueryMemoryPool(pool)); } } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java index 83ab7f51ad..39cdce8fd1 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java @@ -260,6 +260,7 @@ private Iterator 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 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 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 diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransaction.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransaction.java index 4f9e5f5937..41502789b9 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransaction.java +++ b/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 diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java index e6a5e78533..31bcf42382 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java +++ b/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 idCache; private final Cache nameCache; @@ -51,8 +52,8 @@ public class CachedSchemaTransactionV2 extends SchemaTransactionV2 { 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 @@ protected void addSchema(SchemaElement schema) { 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 @@ protected T getSchema(HugeType type, Id id) { 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 @@ protected List getAllSchema(HugeType type) { 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 @@ public CachedTypes cachedTypes() { } private static class CachedTypes - extends ConcurrentHashMap { + extends ConcurrentHashMap { private static final long serialVersionUID = -2215549791679355996L; } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java index 2bfbedd2ae..9ee7abe049 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java +++ b/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 MAX_MEMORY_CAPACITY = new ConfigOption<>( diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/MemoryManager.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/MemoryManager.java index b2ac195e3e..46b1311257 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/MemoryManager.java +++ b/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 + diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/schema/SchemaElement.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/schema/SchemaElement.java index 966d3eed8a..250515a6e3 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/schema/SchemaElement.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/schema/SchemaElement.java @@ -24,6 +24,7 @@ import org.apache.hugegraph.HugeGraph; import org.apache.hugegraph.backend.id.Id; import org.apache.hugegraph.backend.id.IdGenerator; +import org.apache.hugegraph.memory.consumer.OffHeapObject; import org.apache.hugegraph.type.Nameable; import org.apache.hugegraph.type.Typeable; import org.apache.hugegraph.type.define.SchemaStatus; @@ -58,9 +59,9 @@ public abstract class SchemaElement implements Nameable, Typeable, protected final HugeGraph graph; - private final Id id; private final String name; private final Userdata userdata; + private Id id; private SchemaStatus status; public SchemaElement(final HugeGraph graph, Id id, String name) { @@ -83,6 +84,12 @@ public Id id() { return this.id; } + public void convertIdToOnHeapIfNeeded() { + if (this.id instanceof OffHeapObject) { + this.id = (Id) ((OffHeapObject) this.id).zeroCopyReadFromByteBuf(); + } + } + public long longId() { return this.id.asLong(); } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdge.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdge.java index ac23a7a1e4..041232e63b 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdge.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeEdge.java @@ -30,6 +30,7 @@ import org.apache.hugegraph.backend.query.QueryResults; import org.apache.hugegraph.backend.serializer.BytesBuffer; import org.apache.hugegraph.backend.tx.GraphTransaction; +import org.apache.hugegraph.memory.consumer.OffHeapObject; import org.apache.hugegraph.memory.consumer.factory.PropertyFactory; import org.apache.hugegraph.perf.PerfUtil.Watched; import org.apache.hugegraph.schema.EdgeLabel; @@ -80,6 +81,12 @@ public HugeEdge(final HugeGraph graph, Id id, EdgeLabel label) { this.isOutEdge = true; } + public void convertIdToOnHeapIfNeeded() { + if (this.id instanceof OffHeapObject) { + this.id = (Id) ((OffHeapObject) this.id).zeroCopyReadFromByteBuf(); + } + } + @Override public HugeType type() { // NOTE: we optimize the edge type that let it include direction diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeVertex.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeVertex.java index 29abc007e9..1772f32c33 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeVertex.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/structure/HugeVertex.java @@ -40,6 +40,7 @@ import org.apache.hugegraph.backend.tx.GraphTransaction; import org.apache.hugegraph.config.CoreOptions; import org.apache.hugegraph.masterelection.StandardClusterRoleStore; +import org.apache.hugegraph.memory.consumer.OffHeapObject; import org.apache.hugegraph.memory.consumer.factory.PropertyFactory; import org.apache.hugegraph.perf.PerfUtil.Watched; import org.apache.hugegraph.schema.EdgeLabel; @@ -93,6 +94,13 @@ public HugeVertex(final HugeGraph graph, Id id, VertexLabel label) { } } + public void convertIdToOnHeapIfNeeded() { + if (this.id instanceof OffHeapObject) { + this.id = (Id) ((OffHeapObject) this.id).zeroCopyReadFromByteBuf(); + } + } + + @Override public HugeType type() { if (label != null && diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java index 8122c79080..2608de95d8 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java @@ -639,7 +639,7 @@ protected void checkVertexExist(Id vertexId, String name) { this.graph.vertex(vertexId); } catch (NotFoundException e) { throw new IllegalArgumentException(String.format( - "The %s with id '%s' does not exist", name, vertexId), e); + "The %s with id '%s' does not exist: %s", name, vertexId, e.getMessage()), e); } } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java index a21a23d71a..820eab74b6 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java @@ -20,6 +20,8 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Objects; +import java.util.Optional; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; @@ -112,18 +114,21 @@ public void start(String name) { new ContextCallable<>(() -> this.runAndDone(MemoryManager.getInstance() .getCorrespondingTaskMemoryPool( Thread.currentThread() - .getName()) - .findRootQueryPool())))); + .getName()))))); } } - private Void runAndDone(MemoryPool queryPool) { - MemoryPool currentTaskPool = queryPool.addChildPool("kout-consume-task"); - MemoryManager.getInstance() - .bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), - (TaskMemoryPool) currentTaskPool); - MemoryPool currentOperationPool = - currentTaskPool.addChildPool("kout-consume-operation"); + private Void runAndDone(MemoryPool taskPool) { + MemoryPool currentTaskPool = null; + if (Objects.nonNull(taskPool)) { + currentTaskPool = taskPool.findRootQueryPool().addChildPool("kout-consume-task"); + MemoryManager.getInstance() + .bindCorrespondingTaskMemoryPool(Thread.currentThread().getName(), + (TaskMemoryPool) currentTaskPool); + MemoryPool currentOperationPool = + currentTaskPool.addChildPool("kout-consume-operation"); + } + try { this.run(); } catch (Throwable e) { @@ -139,7 +144,9 @@ private Void runAndDone(MemoryPool queryPool) { } finally { this.done(); this.latch.countDown(); - currentTaskPool.releaseSelf("Complete kout consume task", false); + Optional.ofNullable(currentTaskPool) + .ifPresent(pool -> pool.releaseSelf("Complete kout consume task", + false)); } return null; }