Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Pengzna wants to merge 6 commits into from
Closed

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4d4cb22
off-heap adoption for id
Pengzna Dec 9, 2024
29b7845
off-heap adoption for property & kout query pool
Pengzna Dec 9, 2024
a402ddb
wip
Pengzna Dec 10, 2024
88604cf
backend entry iter
Pengzna Dec 10, 2024
a342870
fix bootstrap bugs
Pengzna Dec 10, 2024
db3ee80
cache use on-heap id && adopt for disabled mode
Pengzna Dec 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 32 additions & 15 deletions ...graph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/traversers/KoutAPI.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -93,27 +97,40 @@ 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();

@javeme javeme Dec 19, 2024 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we bind a MemoryPool for each graph, or bind MemoryManager to graph?

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");

@javeme javeme Aug 23, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we can let:

  1. MemoryPool.addChildPool return QueryMemoryPool
  2. QueryMemoryPool.addChildPool return TaskMemoryPool
  3. TaskMemoryPooll.addChildPool return OperatorMemoryPool

});

@javeme javeme Dec 19, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

add a method like graph.switchToMemoryPool("kout", "main")?


ApiMeasurer measure = new ApiMeasurer();
try {

@javeme javeme Dec 19, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

prefer to add a wrapper method for MemoryPool init-and-gc, then call the original method?

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)

@javeme javeme Aug 23, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

prefer if-style

.ifPresent(pool -> MemoryManager.getInstance().gcQueryMemoryPool(pool));
}
return manager.serializer(g, measure.measures()).writeList("vertices", ids);
}

@POST
Expand Down
4 changes: 4 additions & 0 deletions ...gegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,7 @@ private Iterator<HugeVertex> queryVerticesByIds(IdQuery query) {
return QueryResults.emptyIterator();
}
if (needCacheVertex(vertex)) {
vertex.convertIdToOnHeapIfNeeded();

@javeme javeme Dec 19, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's ok to just call convert in HeapCache.update(), at the same time, we avoid modifying the code everywhere

this.verticesCache.update(vertex.id(), vertex);
}
return QueryResults.iterator(vertex);
Expand Down Expand Up @@ -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);
}
}
Expand Down Expand Up @@ -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);
}

Expand All @@ -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
Expand Down
13 changes: 11 additions & 2 deletions ...egraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransaction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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());

@javeme javeme Dec 19, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

add a OnHeapIdGenerator.of() class?

}

/**
* 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
Expand Down
19 changes: 14 additions & 5 deletions ...raph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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()
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -268,10 +272,12 @@ protected <T extends SchemaElement> 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);
}
Expand Down Expand Up @@ -321,6 +327,9 @@ protected <T extends SchemaElement> List<T> 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);

Expand Down Expand Up @@ -481,7 +490,7 @@ public CachedTypes cachedTypes() {
}

private static class CachedTypes
extends ConcurrentHashMap<HugeType, Boolean> {
extends ConcurrentHashMap<HugeType, Boolean> {

private static final long serialVersionUID = -2215549791679355996L;
}
Expand Down
15 changes: 8 additions & 7 deletions ...raph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/id/IdGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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) {
Expand All @@ -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);
}
Expand Down
13 changes: 12 additions & 1 deletion ...-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -135,7 +137,16 @@ public static final String joinTableName(String prefix, String table) {

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();
Expand Down
2 changes: 1 addition & 1 deletion hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<>(
Expand Down
6 changes: 5 additions & 1 deletion hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/memory/MemoryManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

@javeme javeme Dec 19, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

MEMORY_MODE style is only for const var, and if there is only a single MemoryManager, also remove static mark: private MemoryMode memoryMode


private MemoryManager() {
this.memoryArbitrator = new MemoryArbitratorImpl(this);
Expand All @@ -89,6 +89,10 @@ private MemoryManager() {
}

public MemoryPool addQueryMemoryPool() {
if (MEMORY_MODE == MemoryMode.DISABLE_MEMORY_MANAGEMENT) {
return null;
}

int count = queryMemoryPools.size();

@javeme javeme Dec 19, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

expect this.xx stype

@javeme javeme Dec 19, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the difference between QueryMemoryPool and MemoryPool? if no difference, just naming MemoryPool is ok.
and can we make some special names for CorrespondingTaskMemoryPool and CurrentWorkingOperatorMemoryPool, such as RequestMemoryPool and RequestStageMemoryPool

String poolName =
QUERY_MEMORY_POOL_NAME_PREFIX + DELIMINATOR + count + DELIMINATOR +
Expand Down
Loading
Loading