Skip to content
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

remove RaftNode construct code to init #11

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

remove RaftNode construct code to init #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8743927
update init method
zhyzhyzhy Nov 8, 2018
fd39025
update init method
zhyzhyzhy Nov 8, 2018
2eb0662
add peerId to Peer
zhyzhyzhy Nov 8, 2018
39fa32d
fix bug
zhyzhyzhy Nov 8, 2018
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
15 changes: 15 additions & 0 deletions raft-java-core/src/main/java/com/github/wenweihu86/raft/Peer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,19 +11,30 @@
* Created by wenweihu86 on 2017/5/5.
*/
public class Peer {

private PeerId peerId;

private RaftMessage.Server server;

private RPCClient rpcClient;

private RaftConsensusService raftConsensusService;

private RaftConsensusServiceAsync raftConsensusServiceAsync;

// 需要发送给follower的下一个日志条目的索引值，只对leader有效
private long nextIndex;

// 已复制日志的最高索引值
private long matchIndex;

private volatile Boolean voteGranted;

private volatile boolean isCatchUp;

public Peer(RaftMessage.Server server) {
this.server = server;
this.peerId = new PeerId(server.getServerId());
this.rpcClient = new RPCClient(new EndPoint(
server.getEndPoint().getHost(),
server.getEndPoint().getPort()));
Expand All @@ -40,6 +51,10 @@ public RPCClient getRpcClient() {
return rpcClient;
}

public PeerId getPeerId() {
return peerId;
}

public RaftConsensusService getRaftConsensusService() {
return raftConsensusService;
}
Expand Down
13 changes: 13 additions & 0 deletions raft-java-core/src/main/java/com/github/wenweihu86/raft/PeerId.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
package com.github.wenweihu86.raft;

public class PeerId {
Copy link
Owner

Choose a reason for hiding this comment

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

PeerId作为map的key时，应该实现equals和hashcode函数。

private Integer peerId;

public PeerId(Integer peerId) {
this.peerId = peerId;
}

public Integer getPeerId() {
return peerId;
}
}
71 changes: 45 additions & 26 deletions raft-java-core/src/main/java/com/github/wenweihu86/raft/RaftNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,10 @@

import com.github.wenweihu86.raft.proto.RaftMessage;
import com.github.wenweihu86.raft.storage.SegmentedLog;
import com.github.wenweihu86.raft.util.ConfigurationUtils;
import com.google.protobuf.ByteString;
import com.github.wenweihu86.raft.storage.Snapshot;
import com.github.wenweihu86.raft.util.ConfigurationUtils;
import com.github.wenweihu86.rpc.client.RPCCallback;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
import org.apache.commons.io.FileUtils;
Expand All @@ -17,7 +17,10 @@
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* Created by wenweihu86 on 2017/5/2.
Expand All @@ -40,13 +43,13 @@ public enum NodeState {

private RaftOptions raftOptions;
private RaftMessage.Configuration configuration;
private ConcurrentMap<Integer, Peer> peerMap = new ConcurrentHashMap<>();
private ConcurrentMap<PeerId, Peer> peerMap = new ConcurrentHashMap<>();
private RaftMessage.Server localServer;
private StateMachine stateMachine;
private SegmentedLog raftLog;
private Snapshot snapshot;

private NodeState state = NodeState.STATE_FOLLOWER;
private volatile NodeState state;
// 服务器最后一次知道的任期号（初始化为 0，持续递增）
private long currentTerm;
// 在当前获得选票的候选人的Id
Expand All @@ -66,6 +69,8 @@ public enum NodeState {
private ScheduledFuture electionScheduledFuture;
private ScheduledFuture heartbeatScheduledFuture;

private AtomicBoolean hasInit = new AtomicBoolean(false);

public RaftNode(RaftOptions raftOptions,
List<RaftMessage.Server> servers,
RaftMessage.Server localServer,
Expand All @@ -79,14 +84,23 @@ public RaftNode(RaftOptions raftOptions,

this.localServer = localServer;
this.stateMachine = stateMachine;
}

public void init() {

if (!hasInit.compareAndSet(false, true)) {
throw new IllegalStateException("Raft Node has init before");
}

// load log and snapshot
raftLog = new SegmentedLog(raftOptions.getDataDir(), raftOptions.getMaxSegmentFileSize());
snapshot = new Snapshot(raftOptions.getDataDir());
snapshot.reload();

currentTerm = raftLog.getMetaData().getCurrentTerm();

votedFor = raftLog.getMetaData().getVotedFor();

commitIndex = Math.max(snapshot.getMetaData().getLastIncludedIndex(), commitIndex);
// discard old log entries
if (snapshot.getMetaData().getLastIncludedIndex() > 0
Expand All @@ -110,15 +124,16 @@ public RaftNode(RaftOptions raftOptions,
}
}
lastAppliedIndex = commitIndex;
}

public void init() {
// init state is FOLLOWER
state = NodeState.STATE_FOLLOWER;

for (RaftMessage.Server server : configuration.getServersList()) {
if (!peerMap.containsKey(server.getServerId())
if (!peerMap.containsKey(new PeerId(server.getServerId()))
&& server.getServerId() != localServer.getServerId()) {
Peer peer = new Peer(server);
peer.setNextIndex(raftLog.getLastLogIndex() + 1);
peerMap.put(server.getServerId(), peer);
peerMap.put(new PeerId(server.getServerId()), peer);
}
}

Expand Down Expand Up @@ -159,7 +174,7 @@ public boolean replicate(byte[] data, RaftMessage.EntryType entryType) {
raftLog.updateMetaData(currentTerm, null, raftLog.getFirstLogIndex());

for (RaftMessage.Server server : configuration.getServersList()) {
final Peer peer = peerMap.get(server.getServerId());
final Peer peer = peerMap.get(new PeerId(server.getServerId()));
executorService.submit(new Runnable() {
@Override
public void run() {
Expand Down Expand Up @@ -259,7 +274,7 @@ public void appendEntries(Peer peer) {
peer.getServer().getEndPoint().getHost(),
peer.getServer().getEndPoint().getPort());
if (!ConfigurationUtils.containsServer(configuration, peer.getServer().getServerId())) {
peerMap.remove(peer.getServer().getServerId());
peerMap.remove(new PeerId(peer.getServer().getServerId()));
peer.getRpcClient().stop();
}
return;
Expand Down Expand Up @@ -404,11 +419,11 @@ public void applyConfiguration(RaftMessage.LogEntry entry) {
configuration = newConfiguration;
// update peerMap
for (RaftMessage.Server server : newConfiguration.getServersList()) {
if (!peerMap.containsKey(server.getServerId())
if (!peerMap.containsKey(new PeerId(server.getServerId()))
&& server.getServerId() != localServer.getServerId()) {
Peer peer = new Peer(server);
peer.setNextIndex(raftLog.getLastLogIndex() + 1);
peerMap.put(server.getServerId(), peer);
peerMap.put(new PeerId(server.getServerId()), peer);
}
}
LOG.info("new conf is {}, leaderId={}", PRINTER.print(newConfiguration), leaderId);
Expand Down Expand Up @@ -473,7 +488,7 @@ private void startPreVote() {
if (server.getServerId() == localServer.getServerId()) {
continue;
}
final Peer peer = peerMap.get(server.getServerId());
final Peer peer = peerMap.get(new PeerId(server.getServerId()));
executorService.submit(new Runnable() {
@Override
public void run() {
Expand Down Expand Up @@ -507,7 +522,7 @@ private void startVote() {
if (server.getServerId() == localServer.getServerId()) {
continue;
}
final Peer peer = peerMap.get(server.getServerId());
final Peer peer = peerMap.get(new PeerId(server.getServerId()));
executorService.submit(new Runnable() {
@Override
public void run() {
Expand Down Expand Up @@ -597,7 +612,7 @@ public void success(RaftMessage.VoteResponse response) {
if (server.getServerId() == localServer.getServerId()) {
continue;
}
Peer peer1 = peerMap.get(server.getServerId());
Peer peer1 = peerMap.get(new PeerId(server.getServerId()));
if (peer1.isVoteGranted() != null && peer1.isVoteGranted() == true) {
voteGrantedNum += 1;
}
Expand Down Expand Up @@ -664,7 +679,7 @@ public void success(RaftMessage.VoteResponse response) {
if (server.getServerId() == localServer.getServerId()) {
continue;
}
Peer peer1 = peerMap.get(server.getServerId());
Peer peer1 = peerMap.get(new PeerId(server.getServerId()));
if (peer1.isVoteGranted() != null && peer1.isVoteGranted() == true) {
voteGrantedNum += 1;
}
Expand Down Expand Up @@ -741,7 +756,7 @@ private void advanceCommitIndex() {
int i = 0;
for (RaftMessage.Server server : configuration.getServersList()) {
if (server.getServerId() != localServer.getServerId()) {
Peer peer = peerMap.get(server.getServerId());
Peer peer = peerMap.get(new PeerId(server.getServerId()));
matchIndexes[i++] = peer.getMatchIndex();
}
}
Expand Down Expand Up @@ -928,6 +943,18 @@ private RaftMessage.InstallSnapshotRequest buildInstallSnapshotRequest(
return requestBuilder.build();
}

public void removePeer(PeerId peerId) {
peerMap.remove(peerId);
}

public boolean containsPeer(PeerId peerId) {
return peerMap.containsKey(peerId);
}

public void addPeer(Peer peer) {
peerMap.put(new PeerId(peer.getServer().getServerId()), peer);
}

public Lock getLock() {
return lock;
}
Expand Down Expand Up @@ -992,14 +1019,6 @@ public RaftMessage.Server getLocalServer() {
return localServer;
}

public NodeState getState() {
return state;
}

public ConcurrentMap<Integer, Peer> getPeerMap() {
return peerMap;
}

public ExecutorService getExecutorService() {
return executorService;
}
Expand Down
7 changes: 4 additions & 3 deletions ...ava-core/src/main/java/com/github/wenweihu86/raft/service/impl/RaftClientServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package com.github.wenweihu86.raft.service.impl;

import com.github.wenweihu86.raft.Peer;
import com.github.wenweihu86.raft.PeerId;
import com.github.wenweihu86.raft.RaftNode;
import com.github.wenweihu86.raft.proto.RaftMessage;
import com.github.wenweihu86.raft.service.RaftClientService;
Expand Down Expand Up @@ -99,7 +100,7 @@ public RaftMessage.AddPeersResponse addPeers(RaftMessage.AddPeersRequest request
return responseBuilder.build();
}
for (RaftMessage.Server server : request.getServersList()) {
if (raftNode.getPeerMap().containsKey(server.getServerId())) {
if (raftNode.containsPeer(new PeerId(server.getServerId()))) {
LOG.warn("already be added/adding to configuration");
responseBuilder.setResMsg("already be added/adding to configuration");
return responseBuilder.build();
Expand All @@ -110,7 +111,7 @@ public RaftMessage.AddPeersResponse addPeers(RaftMessage.AddPeersRequest request
final Peer peer = new Peer(server);
peer.setNextIndex(1);
requestPeers.add(peer);
raftNode.getPeerMap().putIfAbsent(server.getServerId(), peer);
raftNode.addPeer(peer);
raftNode.getExecutorService().submit(new Runnable() {
@Override
public void run() {
Expand Down Expand Up @@ -163,7 +164,7 @@ public void run() {
try {
for (Peer peer : requestPeers) {
peer.getRpcClient().stop();
raftNode.getPeerMap().remove(peer.getServer().getServerId());
raftNode.removePeer(new PeerId(peer.getServer().getServerId()));
}
} finally {
raftNode.getLock().unlock();
Expand Down