feat(server): support MemoryManagement for graph query framework

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

@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.memory;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.hugegraph.memory.arbitrator.MemoryArbitrator;
+import org.apache.hugegraph.memory.arbitrator.MemoryArbitratorImpl;
+import org.apache.hugegraph.memory.pool.MemoryPool;
+import org.apache.hugegraph.memory.pool.impl.QueryMemoryPool;
+import org.apache.hugegraph.util.Bytes;
+import org.apache.hugegraph.util.ExecutorUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MemoryManager {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(MemoryManager.class);
+    private static final String QUERY_MEMORY_POOL_NAME_PREFIX = "QueryMemoryPool";
+    private static final String ARBITRATE_MEMORY_POOL_NAME = "ArbitrateMemoryPool";
+    private static final String DELIMINATOR = "_";
+    private static final int ARBITRATE_MEMORY_THREAD_NUM = 12;
+    // TODO: read it from conf, current 1G
+    public static final long MAX_MEMORY_CAPACITY_IN_BYTES = Bytes.GB;
+    private final AtomicLong currentMemoryCapacityInBytes =
+            new AtomicLong(MAX_MEMORY_CAPACITY_IN_BYTES);
+    private final Set<MemoryPool> queryMemoryPools = new CopyOnWriteArraySet<>();
+    private final MemoryArbitrator memoryArbitrator;
+    private final ExecutorService arbitrateExecutor;
+    // TODO: integrated with mingzhen's monitor thread
+    // private final Runnable queryGCThread;
+
+    private MemoryManager() {
+        this.memoryArbitrator = new MemoryArbitratorImpl();
+        this.arbitrateExecutor = ExecutorUtil.newFixedThreadPool(ARBITRATE_MEMORY_THREAD_NUM,
+                                                                 ARBITRATE_MEMORY_POOL_NAME);
+    }
+
+    public MemoryPool addQueryMemoryPool() {
+        int count = queryMemoryPools.size();
+        String poolName =
+                QUERY_MEMORY_POOL_NAME_PREFIX + DELIMINATOR + count + DELIMINATOR +
+                System.currentTimeMillis();
+        MemoryPool queryPool = new QueryMemoryPool(poolName, this);
+        queryMemoryPools.add(queryPool);
+        return queryPool;
+    }
+
+    public void gcQueryMemoryPool(MemoryPool pool) {
+        queryMemoryPools.remove(pool);
+        long reclaimedMemory = pool.getAllocatedBytes();
+        pool.releaseSelf();
+        currentMemoryCapacityInBytes.addAndGet(reclaimedMemory);
+    }
+
+    public long triggerLocalArbitration(MemoryPool targetPool, long neededBytes) {
+        Future<Long> future =
+                arbitrateExecutor.submit(
+                        () -> memoryArbitrator.reclaimLocally(targetPool, neededBytes));
+        try {
+            return future.get(MemoryArbitrator.MAX_WAIT_TIME_FOR_LOCAL_RECLAIM,
+                              TimeUnit.MILLISECONDS);
+        } catch (TimeoutException e) {
+            LOGGER.warn("MemoryManager: arbitration locally for {} timed out", targetPool, e);
+        } catch (InterruptedException | ExecutionException e) {
+            LOGGER.error("MemoryManager: arbitration locally for {} interrupted or failed",
+                         targetPool,
+                         e);
+        }
+        return 0;
+    }
+
+    public long triggerGlobalArbitration(MemoryPool requestPool, long neededBytes) {
+        Future<Long> future =
+                arbitrateExecutor.submit(
+                        () -> memoryArbitrator.reclaimGlobally(requestPool, neededBytes));
+        try {
+            return future.get(MemoryArbitrator.MAX_WAIT_TIME_FOR_GLOBAL_RECLAIM,
+                              TimeUnit.MILLISECONDS);
+        } catch (TimeoutException e) {
+            LOGGER.warn("MemoryManager: arbitration globally for {} timed out", requestPool, e);
+        } catch (InterruptedException | ExecutionException e) {
+            LOGGER.error("MemoryManager: arbitration globally for {} interrupted or failed",
+                         requestPool, e);
+        }
+        return 0;
+    }
+
+    private static class MemoryManagerHolder {
+
+        private static final MemoryManager INSTANCE = new MemoryManager();
+
+        private MemoryManagerHolder() {
+            // empty constructor
+        }
+    }
+
+    public static MemoryManager getInstance() {
+        return MemoryManagerHolder.INSTANCE;
+    }
+}

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

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.memory.allocator;
+
+// TODO(pjz): implement different memory allocate strategy.
+public interface MemoryAllocator {
+
+    Object tryToAllocateOffHeap(long size);
+
+    Object forceAllocateOffHeap(long size);
+}

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

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.memory.allocator;
+
+import org.apache.hugegraph.memory.MemoryManager;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+
+/**
+ * This class makes fully use of Netty's efficient memory management strategy.
+ */
+public class NettyMemoryAllocator implements MemoryAllocator {
+
+    private final PooledByteBufAllocator offHeapAllocator = PooledByteBufAllocator.DEFAULT;
+
+    @Override
+    public ByteBuf forceAllocateOffHeap(long size) {
+        return offHeapAllocator.directBuffer((int) size);
+    }
+
+    @Override
+    public ByteBuf tryToAllocateOffHeap(long size) {
+        if (offHeapAllocator.metric().usedDirectMemory() + size <
+            MemoryManager.MAX_MEMORY_CAPACITY_IN_BYTES) {
+            return offHeapAllocator.directBuffer((int) size);
+        }
+        return null;
+    }
+}

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

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.memory.arbitrator;
+
+import org.apache.hugegraph.memory.pool.MemoryPool;
+
+public interface MemoryArbitrator {
+
+    long MAX_WAIT_TIME_FOR_LOCAL_RECLAIM = 1000;
+
+    long MAX_WAIT_TIME_FOR_GLOBAL_RECLAIM = 5000;
+
+    long reclaimLocally(MemoryPool queryPool, long neededBytes);
+
+    long reclaimGlobally(MemoryPool queryPool, long neededBytes);
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.memory.arbitrator;
+
+import org.apache.hugegraph.memory.pool.MemoryPool;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MemoryArbitratorImpl implements MemoryArbitrator {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(MemoryArbitratorImpl.class);
+
+    @Override
+    public long reclaimLocally(MemoryPool queryPool, long neededBytes) {
+        long startTime = System.currentTimeMillis();
+        long res = queryPool.tryToReclaimLocalMemory(neededBytes);
+        LOGGER.info("[{}] reclaim local memory: {} bytes, took {} ms",
+                    Thread.currentThread().getName(),
+                    res,
+                    System.currentTimeMillis() - startTime);
+        return res;
+    }
+
+    @Override
+    public long reclaimGlobally(MemoryPool queryPool, long neededBytes) {
+        long startTime = System.currentTimeMillis();
+        // TODO
+        // 1. select the query task that uses the most memory
+        // 2. suspend that task
+        // 3. apply disk spill to that task
+        LOGGER.info("[{}] reclaim global memory: {} bytes, took {} ms",
+                    Thread.currentThread().getName(),
+                    0,
+                    System.currentTimeMillis() - startTime);
+        return 0;
+    }
+}

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

@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.memory.consumer;
+
+// TODO(pjz): integrated it with HG objects such as edges and vertex.
+public interface MemoryConsumer {
+
+}