anph
/
hugegraph-sync
forked from hugegraph/hugegraph-sync
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refact: adopt jraft to open source version 

address the
[jraft-doc](https://hugegraph.feishu.cn/wiki/JGy0wPlCsih9XMkjrNMcNfhxnBb)
This commit is contained in:
What is broken can be reforged 2023-06-30 16:45:21 +08:00 committed by V_Galaxy
parent 8c03d72744
commit 0cad1fdb45
4 changed files with 72 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hg-store-core/pom.xml
View File

@ -48,7 +48,7 @@
<dependency>
<groupId>com.alipay.sofa</groupId>
<artifactId>jraft-core</artifactId>
<version>1.3.9</version>
<version>1.3.13</version>
<exclusions>
<exclusion>
<groupId>org.rocksdb</groupId>

8
hg-store-core/src/main/java/org/apache/hugegraph/store/HeartbeatService.java
View File

@ -311,13 +311,7 @@ public class HeartbeatService implements Lifecycle<HgStoreEngineOptions>, Partit
// 统计shard状态
partition.getShardGroup().getShards().forEach(shard -> {
Metapb.ShardState state = Metapb.ShardState.SState_Normal;
if (aliveShards.containsKey(shard.getStoreId())) {
Replicator.State s = partition.getRaftNode().getReplicatorState(
aliveShards.get(shard.getStoreId()));
if (s == Replicator.State.Snapshot) {
state = Metapb.ShardState.SState_Snapshot;
}
} else {
if (! aliveShards.containsKey(shard.getStoreId())) {
state = Metapb.ShardState.SState_Offline;
}

98
hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java
View File

@ -66,6 +66,7 @@ import com.alipay.sofa.jraft.Closure;
import com.alipay.sofa.jraft.JRaftUtils;
import com.alipay.sofa.jraft.Node;
import com.alipay.sofa.jraft.RaftGroupService;
import com.alipay.sofa.jraft.ReplicatorGroup;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.conf.Configuration;
import com.alipay.sofa.jraft.core.DefaultJRaftServiceFactory;
@ -80,6 +81,7 @@ import com.alipay.sofa.jraft.storage.LogStorage;
import com.alipay.sofa.jraft.storage.impl.RocksDBLogStorage;
import com.alipay.sofa.jraft.storage.log.RocksDBSegmentLogStorage;
import com.alipay.sofa.jraft.util.Endpoint;
import com.alipay.sofa.jraft.util.ThreadId;
import com.alipay.sofa.jraft.util.Utils;
import com.alipay.sofa.jraft.util.internal.ThrowUtil;
import com.google.protobuf.CodedInputStream;
@ -117,17 +119,16 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
partitionManager = storeEngine.getPartitionManager();
stateListeners = Collections.synchronizedList(new ArrayList());
}
// public static ThreadPoolExecutor getRaftLogWriteExecutor() {
// if (raftLogWriteExecutor == null) {
// synchronized (PartitionEngine.class) {
// if (raftLogWriteExecutor == null)
// raftLogWriteExecutor = RocksDBSegmentLogStorage.createDefaultWriteExecutor();
// }
// }
// return raftLogWriteExecutor;
// }
public static ThreadPoolExecutor getRaftLogWriteExecutor() {
if (raftLogWriteExecutor == null) {
synchronized (PartitionEngine.class) {
if (raftLogWriteExecutor == null) {
raftLogWriteExecutor = RocksDBSegmentLogStorage.createDefaultWriteExecutor();
}
}
}
return raftLogWriteExecutor;
}
/**
* 记录使用本raft的分区信息
@ -209,16 +210,7 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
@Override
public LogStorage createLogStorage(final String uri, final RaftOptions raftOptions) {
if (options.getRaftOptions().isUseRocksDBSegmentLogStorage()) {
return new RocksDBSegmentLogStorage(uri, raftOptions,
RocksDBSegmentLogStorage.DEFAULT_VALUE_SIZE_THRESHOLD,
options.getRaftOptions()
.getMaxSegmentFileSize(),
options.getRaftOptions()
.getPreAllocateSegmentCount(),
options.getRaftOptions()
.getKeepInMemorySegmentCount(),
RocksDBSegmentLogStorage.DEFAULT_CHECKPOINT_INTERVAL_MS,
getRaftLogWriteExecutor());
return new RocksDBSegmentLogStorage(uri, raftOptions);
} else {
return new RocksDBLogStorage(uri, raftOptions);
}
@ -237,8 +229,6 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
nodeOptions.setRpcInstallSnapshotTimeout(
options.getRaftOptions().getRpcInstallSnapshotTimeout());
nodeOptions.setElectionTimeoutMs(options.getRaftOptions().getElectionTimeoutMs());
nodeOptions.setChannelWriteBufLowWaterMark(2 * 1024 * 1024);
nodeOptions.setChannelWriteBufHighWaterMark(4 * 1024 * 1024);
// 设置raft配置
RaftOptions raftOptions = nodeOptions.getRaftOptions();
raftOptions.setDisruptorBufferSize(options.getRaftOptions().getDisruptorBufferSize());
@ -316,7 +306,7 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
HgCmdClient rpcClient = storeEngine.getHgCmdClient();
// 生成新的Configuration对象
Configuration oldConf = raftNode.getCurrentConf();
Configuration oldConf = getCurrentConf();
Configuration conf = oldConf.copy();
if (!addPeers.isEmpty()) {
addPeers.forEach(peer -> {
@ -344,7 +334,7 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
// 3. 检查learner是否完成快照同步
boolean snapshotOk = true;
for (PeerId peerId : raftNode.listLearners()) {
Replicator.State state = raftNode.getReplicatorState(peerId);
Replicator.State state = getReplicatorState(peerId);
if (state == null || state != Replicator.State.Replicate) {
snapshotOk = false;
break;
@ -610,7 +600,7 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
// 更新shard group对象
shardGroup.changeLeader(partitionManager.getStore().getId());
onConfigurationCommitted(this.getRaftNode().getCurrentConf());
onConfigurationCommitted(getCurrentConf());
synchronized (leaderChangedEvent) {
leaderChangedEvent.notifyAll();
}
@ -619,7 +609,7 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
@Override
public void onStartFollowing(final PeerId newLeaderId, final long newTerm) {
onConfigurationCommitted(this.getRaftNode().getCurrentConf());
onConfigurationCommitted(getCurrentConf());
synchronized (leaderChangedEvent) {
leaderChangedEvent.notifyAll();
}
@ -1031,11 +1021,8 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
if (this.snapshotFlag.compareAndSet(false, true)) {
raftNode.getRaftOptions().setTruncateLog(true);
raftNode.snapshot(status -> {
log.info("Raft {} snapshot OK. ", this.getGroupId());
raftNode.getRaftOptions().setTruncateLog(false);
if (done != null) {
done.run(status);
}
@ -1222,4 +1209,57 @@ public class PartitionEngine implements Lifecycle<PartitionEngineOptions>, RaftS
return true;
}
}
public Configuration getCurrentConf() {
return new Configuration(this.raftNode.listPeers(), this.raftNode.listLearners());
}
private Replicator.State getReplicatorState(PeerId peerId) {
var replicateGroup = getReplicatorGroup();
if (replicateGroup == null) {
return null;
}
ThreadId threadId = replicateGroup.getReplicator(peerId);
if (threadId == null) {
return null;
} else {
Replicator r = (Replicator) threadId.lock();
if (r == null) {
return Replicator.State.Probe;
}
Replicator.State result = getState(r);
threadId.unlock();
return result;
}
}
private ReplicatorGroup getReplicatorGroup() {
var clz = this.raftNode.getClass();
try {
var f = clz.getDeclaredField("replicatorGroup");
f.setAccessible(true);
var group = (ReplicatorGroup) f.get(this.raftNode);
f.setAccessible(false);
return group;
} catch (NoSuchFieldException | IllegalAccessException e) {
log.info("getReplicatorGroup: error {}", e.getMessage());
return null;
}
}
private Replicator.State getState(Replicator r) {
var clz = r.getClass();
try {
var f = clz.getDeclaredField("state");
f.setAccessible(true);
var state = (Replicator.State) f.get(this.raftNode);
f.setAccessible(false);
return state;
} catch (NoSuchFieldException | IllegalAccessException e) {
log.info("getReplicatorGroup: error {}", e.getMessage());
return null;
}
}
}

2
hg-store-node/src/main/java/org/apache/hugegraph/store/node/controller/PartitionAPI.java
View File

@ -85,7 +85,7 @@ public class PartitionAPI {
raft.setGroupId(engine.getGroupId());
raft.setLeader(engine.getLeader());
raft.setRole(engine.getRaftNode().getNodeState().name());
raft.setConf(engine.getRaftNode().getCurrentConf().toString());
raft.setConf(engine.getCurrentConf().toString());
if (engine.isLeader()) {
raft.setPeers(engine.getRaftNode().listPeers());
raft.setLearners(engine.getRaftNode().listLearners());