From 47dd04639688797df668f0213daaae125e0b3f79 Mon Sep 17 00:00:00 2001 From: koi2000 Date: Sun, 24 Aug 2025 16:22:57 +0800 Subject: [PATCH] refactor(pd): refactor common module --- .../hugegraph/pd/client/ClientCache.java | 4 +- hugegraph-pd/hg-pd-common/pom.xml | 5 + .../org/apache/hugegraph/pd/common/Cache.java | 109 ++++++++++++++ .../apache/hugegraph/pd/common/Consts.java | 28 ++++ .../pd/common/DefaultThreadFactory.java | 39 +++++ .../hugegraph/pd/common/GraphCache.java | 137 ++++++++++++------ .../hugegraph/pd/common/PDException.java | 3 +- .../hugegraph/pd/common/PartitionCache.java | 23 +-- .../pd/util/DefaultThreadFactory.java | 44 ++++++ .../hugegraph/pd/util/ExecutorUtil.java | 63 ++++++++ 10 files changed, 400 insertions(+), 55 deletions(-) create mode 100644 hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/Cache.java create mode 100644 hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/Consts.java create mode 100644 hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/DefaultThreadFactory.java create mode 100644 hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/util/DefaultThreadFactory.java create mode 100644 hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/util/ExecutorUtil.java diff --git a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/ClientCache.java b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/ClientCache.java index 9e584583a9..973843863f 100644 --- a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/ClientCache.java +++ b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/ClientCache.java @@ -61,7 +61,9 @@ private GraphCache getGraphCache(String graphName) { if ((graph = caches.get(graphName)) == null) { synchronized (caches) { if ((graph = caches.get(graphName)) == null) { - graph = new GraphCache(); + Metapb.Graph.Builder builder = Metapb.Graph.newBuilder().setGraphName(graphName); + Metapb.Graph g = builder.build(); + graph = new GraphCache(g); caches.put(graphName, graph); } } diff --git a/hugegraph-pd/hg-pd-common/pom.xml b/hugegraph-pd/hg-pd-common/pom.xml index 79cfbe4112..465fb13b92 100644 --- a/hugegraph-pd/hg-pd-common/pom.xml +++ b/hugegraph-pd/hg-pd-common/pom.xml @@ -44,5 +44,10 @@ commons-collections4 4.4 + + org.apache.logging.log4j + log4j-slf4j-impl + ${log4j2.version} + diff --git a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/Cache.java b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/Cache.java new file mode 100644 index 0000000000..a7c5e376e4 --- /dev/null +++ b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/Cache.java @@ -0,0 +1,109 @@ +/* + * 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.pd.common; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +public class Cache implements Closeable { + + ScheduledExecutorService ex = + Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("hg-cache")); + private ConcurrentMap map = new ConcurrentHashMap(); + private ScheduledFuture future; + private Runnable checker = () -> { + for (Map.Entry e : map.entrySet()) { + if (e.getValue().getValue() == null) { + map.remove(e.getKey()); + } + } + }; + + public Cache() { + future = ex.scheduleWithFixedDelay(checker, 1, 1, TimeUnit.SECONDS); + } + + public CacheValue put(String key, T value, long ttl) { + return map.put(key, new CacheValue(value, ttl)); + } + + public T get(String key) { + CacheValue value = map.get(key); + if (value == null) { + return null; + } + T t = value.getValue(); + if (t == null) { + map.remove(key); + } + return t; + } + + public boolean keepAlive(String key, long ttl) { + CacheValue value = map.get(key); + if (value == null) { + return false; + } + value.keepAlive(ttl); + return true; + } + + @Override + public void close() throws IOException { + try { + future.cancel(true); + ex.shutdownNow(); + } catch (Exception e) { + try { + ex.shutdownNow(); + } catch (Exception ex) { + + } + } + } + + private class CacheValue { + + private final T value; + long outTime; + + protected CacheValue(T value, long ttl) { + this.value = value; + this.outTime = System.currentTimeMillis() + ttl; + } + + protected T getValue() { + if (System.currentTimeMillis() >= outTime) { + return null; + } + return value; + } + + protected void keepAlive(long ttl) { + this.outTime = System.currentTimeMillis() + ttl; + } + + } +} diff --git a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/Consts.java b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/Consts.java new file mode 100644 index 0000000000..a113cfa84a --- /dev/null +++ b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/Consts.java @@ -0,0 +1,28 @@ +/* + * 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.pd.common; + +import io.grpc.Metadata; + +public class Consts { + + public static final Metadata.Key CREDENTIAL_KEY = Metadata.Key.of("credential", + Metadata.ASCII_STRING_MARSHALLER); + public static final Metadata.Key TOKEN_KEY = Metadata.Key.of("Pd-Token", + Metadata.ASCII_STRING_MARSHALLER); +} diff --git a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/DefaultThreadFactory.java b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/DefaultThreadFactory.java new file mode 100644 index 0000000000..d5f716dcfc --- /dev/null +++ b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/DefaultThreadFactory.java @@ -0,0 +1,39 @@ +/* + * 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.pd.common; + +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; + +public class DefaultThreadFactory implements ThreadFactory { + + private final AtomicInteger number = new AtomicInteger(1); + private final String prefix; + + public DefaultThreadFactory(String prefix) { + this.prefix = prefix + "-"; + } + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r, prefix + number.getAndIncrement()); + t.setDaemon(true); + t.setPriority(Thread.NORM_PRIORITY); + return t; + } +} diff --git a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/GraphCache.java b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/GraphCache.java index 8a576e1b6b..053cc02645 100644 --- a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/GraphCache.java +++ b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/GraphCache.java @@ -17,22 +17,27 @@ package org.apache.hugegraph.pd.common; +import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock; -import com.google.common.collect.Range; - +import org.apache.commons.collections4.CollectionUtils; import org.apache.hugegraph.pd.grpc.Metapb.Graph; import org.apache.hugegraph.pd.grpc.Metapb.Partition; +import com.google.common.collect.Range; import com.google.common.collect.RangeMap; import com.google.common.collect.TreeRangeMap; import lombok.Data; +import lombok.extern.slf4j.Slf4j; @Data +@Slf4j public class GraphCache { private Graph graph; @@ -41,13 +46,30 @@ public class GraphCache { private ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); private Map state = new ConcurrentHashMap<>(); private Map partitions = new ConcurrentHashMap<>(); - private RangeMap range = new SynchronizedRangeMap().rangeMap; + private volatile RangeMap range = TreeRangeMap.create(); public GraphCache(Graph graph) { this.graph = graph; } - public GraphCache() { + public void init(List ps) { + Map gps = new ConcurrentHashMap<>(ps.size(), 1); + if (!CollectionUtils.isEmpty(ps)) { + WriteLock lock = getLock().writeLock(); + try { + lock.lock(); + for (Partition p : ps) { + gps.put(p.getId(), p); + range.put(Range.closedOpen(p.getStartKey(), p.getEndKey()), p.getId()); + } + } catch (Exception e) { + log.warn("init graph with error:", e); + } finally { + lock.unlock(); + } + } + setPartitions(gps); + } public Partition getPartition(Integer id) { @@ -59,58 +81,87 @@ public Partition addPartition(Integer id, Partition p) { } public Partition removePartition(Integer id) { + Partition p = partitions.get(id); + if (p != null) { + RangeMap range = getRange(); + if (Objects.equals(p.getId(), range.get(p.getStartKey())) && + Objects.equals(p.getId(), range.get(p.getEndKey() - 1))) { + WriteLock lock = getLock().writeLock(); + lock.lock(); + try { + range.remove(range.getEntry(p.getStartKey()).getKey()); + } catch (Exception e) { + log.warn("remove partition with error:", e); + } finally { + lock.unlock(); + } + } + } return partitions.remove(id); } - public class SynchronizedRangeMap, V> { - - private final RangeMap rangeMap = TreeRangeMap.create(); - private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); - - public void put(Range range, V value) { - lock.writeLock().lock(); - try { - rangeMap.put(range, value); - } finally { - lock.writeLock().unlock(); + public void removePartitions() { + getState().clear(); + RangeMap range = getRange(); + WriteLock lock = getLock().writeLock(); + try { + lock.lock(); + if (range != null) { + range.clear(); } + } catch (Exception e) { + log.warn("remove partition with error:", e); + } finally { + lock.unlock(); } + getPartitions().clear(); + getInitialized().set(false); + } - public V get(K key) { - lock.readLock().lock(); - try { - return rangeMap.get(key); - } finally { - lock.readLock().unlock(); - } - } + /* + * 需要外部加写锁 + * */ + public void reset() { + partitions.clear(); + try { + range.clear(); + } catch (Exception e) { - public void remove(Range range) { - lock.writeLock().lock(); - try { - rangeMap.remove(range); - } finally { - lock.writeLock().unlock(); - } } + } - public Map.Entry, V> getEntry(K key) { - lock.readLock().lock(); - try { - return rangeMap.getEntry(key); - } finally { - lock.readLock().unlock(); - } + public boolean updatePartition(Partition partition) { + int partId = partition.getId(); + Partition p = getPartition(partId); + if (p != null && p.equals(partition)) { + return false; } - - public void clear() { - lock.writeLock().lock(); + WriteLock lock = getLock().writeLock(); + try { + lock.lock(); + RangeMap range = getRange(); + addPartition(partId, partition); try { - rangeMap.clear(); - } finally { - lock.writeLock().unlock(); + if (p != null) { + // The old [1-3) is overwritten by [2-3). When [1-3) becomes [1-2), the + // original [1-3) should not be deleted. + // Only when it is confirmed that the old start and end are both your own can + // the old be deleted (i.e., before it is overwritten). + if (Objects.equals(partId, range.get(partition.getStartKey())) && + Objects.equals(partId, range.get(partition.getEndKey() - 1))) { + range.remove(range.getEntry(partition.getStartKey()).getKey()); + } + } + range.put(Range.closedOpen(partition.getStartKey(), partition.getEndKey()), partId); + } catch (Exception e) { + log.warn("update partition with error:", e); } + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + lock.unlock(); } + return true; } } diff --git a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/PDException.java b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/PDException.java index b398137e82..5f60fa30cb 100644 --- a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/PDException.java +++ b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/PDException.java @@ -18,8 +18,7 @@ package org.apache.hugegraph.pd.common; public class PDException extends Exception { - - private final int errorCode; + private int errorCode = 0; public PDException(int error) { super(String.format("Error code = %d", error)); diff --git a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/PartitionCache.java b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/PartitionCache.java index 31cc29deed..5674db36ae 100644 --- a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/PartitionCache.java +++ b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/common/PartitionCache.java @@ -41,8 +41,8 @@ */ public class PartitionCache { - private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); - private final Map locks = new HashMap<>(); + private ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); + private volatile Map locks = new ConcurrentHashMap<>(); Lock writeLock = readWriteLock.writeLock(); // One cache per graph private volatile Map> keyToPartIdCache; @@ -53,8 +53,8 @@ public class PartitionCache { private volatile Map graphCache; public PartitionCache() { - keyToPartIdCache = new HashMap<>(); - partitionCache = new HashMap<>(); + keyToPartIdCache = new ConcurrentHashMap<>(); + partitionCache = new ConcurrentHashMap<>(); shardGroupCache = new ConcurrentHashMap<>(); storeCache = new ConcurrentHashMap<>(); graphCache = new ConcurrentHashMap<>(); @@ -214,7 +214,8 @@ public void updatePartition(String graphName, int partId, Metapb.Partition parti } } - partitionCache.computeIfAbsent(graphName, k -> new HashMap<>()).put(partId, partition); + partitionCache.computeIfAbsent(graphName, k -> new ConcurrentHashMap<>()) + .put(partId, partition); keyToPartIdCache.computeIfAbsent(graphName, k -> TreeRangeMap.create()) .put(Range.closedOpen(partition.getStartKey(), partition.getEndKey()), partId); @@ -270,8 +271,8 @@ public void remove(String graphName, int id) { public void removePartitions() { writeLock.lock(); try { - partitionCache = new HashMap<>(); - keyToPartIdCache = new HashMap<>(); + partitionCache = new ConcurrentHashMap<>(); + keyToPartIdCache = new ConcurrentHashMap<>(); locks.clear(); } finally { writeLock.unlock(); @@ -315,6 +316,10 @@ public Metapb.ShardGroup getShardGroup(int groupId) { return shardGroupCache.get(groupId); } + public Map getShardGroups() { + return this.shardGroupCache; + } + public boolean addStore(Long storeId, Metapb.Store store) { Metapb.Store oldStore = storeCache.get(storeId); if (oldStore != null && oldStore.equals(store)) { @@ -358,8 +363,8 @@ public List getGraphs() { public void reset() { writeLock.lock(); try { - partitionCache = new HashMap<>(); - keyToPartIdCache = new HashMap<>(); + partitionCache = new ConcurrentHashMap<>(); + keyToPartIdCache = new ConcurrentHashMap<>(); shardGroupCache = new ConcurrentHashMap<>(); storeCache = new ConcurrentHashMap<>(); graphCache = new ConcurrentHashMap<>(); diff --git a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/util/DefaultThreadFactory.java b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/util/DefaultThreadFactory.java new file mode 100644 index 0000000000..09230e2376 --- /dev/null +++ b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/util/DefaultThreadFactory.java @@ -0,0 +1,44 @@ +/* + * 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.pd.util; + +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; +public class DefaultThreadFactory implements ThreadFactory { + + private final AtomicInteger number = new AtomicInteger(1); + private final String namePrefix; + private boolean daemon; + + public DefaultThreadFactory(String prefix, boolean daemon) { + this.namePrefix = prefix + "-"; + this.daemon = daemon; + } + + public DefaultThreadFactory(String prefix) { + this(prefix, true); + } + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(null, r, namePrefix + number.getAndIncrement(), 0); + t.setDaemon(daemon); + t.setPriority(Thread.NORM_PRIORITY); + return t; + } +} diff --git a/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/util/ExecutorUtil.java b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/util/ExecutorUtil.java new file mode 100644 index 0000000000..04cbd96f79 --- /dev/null +++ b/hugegraph-pd/hg-pd-common/src/main/java/org/apache/hugegraph/pd/util/ExecutorUtil.java @@ -0,0 +1,63 @@ +/* + * 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.pd.util; + +import java.util.Map; +import java.util.concurrent.*; + +public final class ExecutorUtil { + + private static Map pools = new ConcurrentHashMap<>(); + + public static ThreadPoolExecutor getThreadPoolExecutor(String name) { + if (name == null) { + return null; + } + return pools.get(name); + } + + public static ThreadPoolExecutor createExecutor(String name, int coreThreads, int maxThreads, + int queueSize) { + + return createExecutor(name, coreThreads, maxThreads, queueSize, true); + } + + public static ThreadPoolExecutor createExecutor(String name, int coreThreads, int maxThreads, + int queueSize, boolean daemon) { + ThreadPoolExecutor res = pools.get(name); + if (res != null) { + return res; + } + synchronized (pools) { + res = pools.get(name); + if (res != null) { + return res; + } + BlockingQueue queue; + if (queueSize <= 0) { + queue = new SynchronousQueue(); + } else { + queue = new LinkedBlockingQueue<>(queueSize); + } + res = new ThreadPoolExecutor(coreThreads, maxThreads, 60L, TimeUnit.SECONDS, queue, + new DefaultThreadFactory(name, daemon)); + pools.put(name, res); + } + return res; + } +}