X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FCandidate.java;h=4d51922bc22f7c2bc63886989e354cfb0e912288;hb=b8c6400766f7324dd57d059bd48e435569fe1a27;hp=8baf0d8e815d96957c1d4e6cee3ad55b1bd21f07;hpb=5aa58404a8ee1ad053742780439823309360a3a1;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java index 8baf0d8e81..4d51922bc2 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java @@ -10,7 +10,9 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; import akka.actor.ActorSelection; +import java.util.ArrayList; import java.util.Collection; +import org.opendaylight.controller.cluster.raft.PeerInfo; import org.opendaylight.controller.cluster.raft.RaftActorContext; import org.opendaylight.controller.cluster.raft.RaftState; import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; @@ -44,31 +46,44 @@ public class Candidate extends AbstractRaftActorBehavior { private final int votesRequired; - private final Collection peers; + private final Collection votingPeers = new ArrayList<>(); public Candidate(RaftActorContext context) { super(context, RaftState.Candidate); - peers = context.getPeerIds(); + for(PeerInfo peer: context.getPeers()) { + if(peer.isVoting()) { + votingPeers.add(peer.getId()); + } + } if(LOG.isDebugEnabled()) { - LOG.debug("{}: Election: Candidate has following peers: {}", logName(), peers); + LOG.debug("{}: Election: Candidate has following voting peers: {}", logName(), votingPeers); } - votesRequired = getMajorityVoteCount(peers.size()); + votesRequired = getMajorityVoteCount(votingPeers.size()); startNewTerm(); - if(peers.isEmpty()){ - actor().tell(ELECTION_TIMEOUT, actor()); + if(votingPeers.isEmpty()){ + actor().tell(ElectionTimeout.INSTANCE, actor()); } else { scheduleElection(electionDuration()); } + } + @Override + public final String getLeaderId() { + return null; + } + @Override + public final short getLeaderPayloadVersion() { + return -1; } - @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender, + @Override + protected RaftActorBehavior handleAppendEntries(ActorRef sender, AppendEntries appendEntries) { if(LOG.isDebugEnabled()) { @@ -86,17 +101,14 @@ public class Candidate extends AbstractRaftActorBehavior { return this; } - @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender, - AppendEntriesReply appendEntriesReply) { - + @Override + protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender, AppendEntriesReply appendEntriesReply) { return this; } - @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender, - RequestVoteReply requestVoteReply) { - - LOG.debug("{}: handleRequestVoteReply: {}, current voteCount: {}", logName(), requestVoteReply, - voteCount); + @Override + protected RaftActorBehavior handleRequestVoteReply(ActorRef sender, RequestVoteReply requestVoteReply) { + LOG.debug("{}: handleRequestVoteReply: {}, current voteCount: {}", logName(), requestVoteReply, voteCount); if (requestVoteReply.isVoteGranted()) { voteCount++; @@ -111,9 +123,25 @@ public class Candidate extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) { + if (originalMessage instanceof ElectionTimeout) { + LOG.debug("{}: Received ElectionTimeout", logName()); - Object message = fromSerializableMessage(originalMessage); + if (votesRequired == 0) { + // If there are no peers then we should be a Leader + // We wait for the election timeout to occur before declare + // ourselves the leader. This gives enough time for a leader + // who we do not know about (as a peer) + // to send a message to the candidate + return internalSwitchBehavior(RaftState.Leader); + } + + startNewTerm(); + scheduleElection(electionDuration()); + return this; + } + + final Object message = fromSerializableMessage(originalMessage); if (message instanceof RaftRPC) { RaftRPC rpc = (RaftRPC) message; @@ -133,23 +161,6 @@ public class Candidate extends AbstractRaftActorBehavior { } } - if (message instanceof ElectionTimeout) { - LOG.debug("{}: Received ElectionTimeout", logName()); - - if (votesRequired == 0) { - // If there are no peers then we should be a Leader - // We wait for the election timeout to occur before declare - // ourselves the leader. This gives enough time for a leader - // who we do not know about (as a peer) - // to send a message to the candidate - - return internalSwitchBehavior(RaftState.Leader); - } - startNewTerm(); - scheduleElection(electionDuration()); - return this; - } - return super.handleMessage(sender, message); } @@ -170,7 +181,7 @@ public class Candidate extends AbstractRaftActorBehavior { // Request for a vote // TODO: Retry request for vote if replies do not arrive in a reasonable // amount of time TBD - for (String peerId : peers) { + for (String peerId : votingPeers) { ActorSelection peerActor = context.getPeerActorSelection(peerId); if(peerActor != null) { RequestVote requestVote = new RequestVote( @@ -186,7 +197,8 @@ public class Candidate extends AbstractRaftActorBehavior { } } - @Override public void close() throws Exception { + @Override + public void close() { stopElection(); } }