X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FAbstractRaftActorBehavior.java;h=4651237de7407e65d127cc67067ac1c273b221c4;hp=fc2f137e886085b9f8de644a2a1ae2b660281fa3;hb=a57559cab8f0dd9204fe4848c85d96659115b63b;hpb=a89f83014714207a4ccb704eb75050d758266d71 diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java index fc2f137e88..4651237de7 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java @@ -10,6 +10,7 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; import akka.actor.Cancellable; +import com.google.common.base.Preconditions; import java.util.Random; import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.raft.ClientRequestTracker; @@ -38,9 +39,6 @@ import scala.concurrent.duration.FiniteDuration; * set currentTerm = T, convert to follower (§5.1) */ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { - - protected static final ElectionTimeout ELECTION_TIMEOUT = new ElectionTimeout(); - /** * Information about the RaftActor whose behavior this class represents */ @@ -56,33 +54,41 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { */ private Cancellable electionCancel = null; - /** - * - */ - protected String leaderId = null; - - private short leaderPayloadVersion = -1; - private long replicatedToAllIndex = -1; private final String logName; private final RaftState state; - protected AbstractRaftActorBehavior(RaftActorContext context, RaftState state) { - this.context = context; - this.state = state; + AbstractRaftActorBehavior(final RaftActorContext context, final RaftState state) { + this.context = Preconditions.checkNotNull(context); + this.state = Preconditions.checkNotNull(state); this.LOG = context.getLogger(); logName = String.format("%s (%s)", context.getId(), state); } + public static RaftActorBehavior createBehavior(final RaftActorContext context, final RaftState state) { + switch (state) { + case Candidate: + return new Candidate(context); + case Follower: + return new Follower(context); + case IsolatedLeader: + return new IsolatedLeader(context); + case Leader: + return new Leader(context); + default: + throw new IllegalArgumentException("Unhandled state " + state); + } + } + @Override - public RaftState state() { + public final RaftState state() { return state; } - public String logName() { + protected final String logName() { return logName; } @@ -106,7 +112,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param appendEntries The AppendEntries message - * @return + * @return a new behavior if it was changed or the current behavior */ protected abstract RaftActorBehavior handleAppendEntries(ActorRef sender, AppendEntries appendEntries); @@ -118,7 +124,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender * @param appendEntries - * @return + * @return a new behavior if it was changed or the current behavior */ protected RaftActorBehavior appendEntries(ActorRef sender, AppendEntries appendEntries) { @@ -151,7 +157,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param appendEntriesReply The AppendEntriesReply message - * @return + * @return a new behavior if it was changed or the current behavior */ protected abstract RaftActorBehavior handleAppendEntriesReply(ActorRef sender, AppendEntriesReply appendEntriesReply); @@ -162,12 +168,28 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender * @param requestVote - * @return + * @return a new behavior if it was changed or the current behavior */ protected RaftActorBehavior requestVote(ActorRef sender, RequestVote requestVote) { LOG.debug("{}: In requestVote: {}", logName(), requestVote); + boolean grantVote = canGrantVote(requestVote); + + if(grantVote) { + context.getTermInformation().updateAndPersist(requestVote.getTerm(), requestVote.getCandidateId()); + } + + RequestVoteReply reply = new RequestVoteReply(currentTerm(), grantVote); + + LOG.debug("{}: requestVote returning: {}", logName(), reply); + + sender.tell(reply, actor()); + + return this; + } + + protected boolean canGrantVote(RequestVote requestVote){ boolean grantVote = false; // Reply false if term < currentTerm (§5.1) @@ -177,7 +199,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { // If votedFor is null or candidateId, and candidate’s log is at // least as up-to-date as receiver’s log, grant vote (§5.2, §5.4) } else if (votedFor() == null || votedFor() - .equals(requestVote.getCandidateId())) { + .equals(requestVote.getCandidateId())) { boolean candidateLatest = false; @@ -191,24 +213,15 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { if (requestVote.getLastLogTerm() > lastTerm()) { candidateLatest = true; } else if ((requestVote.getLastLogTerm() == lastTerm()) - && requestVote.getLastLogIndex() >= lastIndex()) { + && requestVote.getLastLogIndex() >= lastIndex()) { candidateLatest = true; } if (candidateLatest) { grantVote = true; - context.getTermInformation().updateAndPersist(requestVote.getTerm(), - requestVote.getCandidateId()); } } - - RequestVoteReply reply = new RequestVoteReply(currentTerm(), grantVote); - - LOG.debug("{}: requestVote returning: {}", logName(), reply); - - sender.tell(reply, actor()); - - return this; + return grantVote; } /** @@ -221,15 +234,14 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param requestVoteReply The RequestVoteReply message - * @return + * @return a new behavior if it was changed or the current behavior */ protected abstract RaftActorBehavior handleRequestVoteReply(ActorRef sender, RequestVoteReply requestVoteReply); /** - * Creates a random election duration * - * @return + * @return a random election duration */ protected FiniteDuration electionDuration() { long variance = new Random().nextInt(context.getConfigParams().getElectionTimeVariance()); @@ -246,89 +258,70 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { } } + protected boolean canStartElection() { + return context.getRaftPolicy().automaticElectionsEnabled() && context.isVotingMember(); + } + /** * schedule a new election * - * @param interval + * @param interval the duration after which we should trigger a new election */ protected void scheduleElection(FiniteDuration interval) { stopElection(); - // Schedule an election. When the scheduler triggers an ElectionTimeout - // message is sent to itself - electionCancel = - context.getActorSystem().scheduler().scheduleOnce(interval, - context.getActor(), ELECTION_TIMEOUT, - context.getActorSystem().dispatcher(), context.getActor()); + // Schedule an election. When the scheduler triggers an ElectionTimeout message is sent to itself + electionCancel = context.getActorSystem().scheduler().scheduleOnce(interval, context.getActor(), + ElectionTimeout.INSTANCE, context.getActorSystem().dispatcher(), context.getActor()); } /** - * Get the current term - * @return + * @return the current term */ protected long currentTerm() { return context.getTermInformation().getCurrentTerm(); } /** - * Get the candidate for whom we voted in the current term - * @return + * @return the candidate for whom we voted in the current term */ protected String votedFor() { return context.getTermInformation().getVotedFor(); } /** - * Get the actor associated with this behavior - * @return + * @return the actor associated with this behavior */ protected ActorRef actor() { return context.getActor(); } /** - * Get the term from the last entry in the log * - * @return + * @return the term from the last entry in the log */ protected long lastTerm() { return context.getReplicatedLog().lastTerm(); } /** - * Get the index from the last entry in the log - * - * @return + * @return the index from the last entry in the log */ protected long lastIndex() { return context.getReplicatedLog().lastIndex(); } /** - * Find the client request tracker for a specific logIndex - * - * @param logIndex - * @return - */ - protected ClientRequestTracker findClientRequestTracker(long logIndex) { - return null; - } - - /** - * Find the client request tracker for a specific logIndex - * * @param logIndex - * @return + * @return the client request tracker for the specified logIndex */ protected ClientRequestTracker removeClientRequestTracker(long logIndex) { return null; } - /** - * Find the log index from the previous to last entry in the log * - * @return + * @return log index from the previous to last entry in the log */ protected long prevLogIndex(long index){ ReplicatedLogEntry prevEntry = @@ -340,8 +333,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { } /** - * Find the log term from the previous to last entry in the log - * @return + * @return log term from the previous to last entry in the log */ protected long prevLogTerm(long index){ ReplicatedLogEntry prevEntry = @@ -360,30 +352,27 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { protected void applyLogToStateMachine(final long index) { long newLastApplied = context.getLastApplied(); // Now maybe we apply to the state machine - for (long i = context.getLastApplied() + 1; - i < index + 1; i++) { - ActorRef clientActor = null; - String identifier = null; - ClientRequestTracker tracker = removeClientRequestTracker(i); - - if (tracker != null) { - clientActor = tracker.getClientActor(); - identifier = tracker.getIdentifier(); - } - ReplicatedLogEntry replicatedLogEntry = - context.getReplicatedLog().get(i); + for (long i = context.getLastApplied() + 1; i < index + 1; i++) { + ReplicatedLogEntry replicatedLogEntry = context.getReplicatedLog().get(i); if (replicatedLogEntry != null) { // Send a local message to the local RaftActor (it's derived class to be // specific to apply the log to it's index) - actor().tell(new ApplyState(clientActor, identifier, - replicatedLogEntry), actor()); + + final ApplyState msg; + final ClientRequestTracker tracker = removeClientRequestTracker(i); + if (tracker != null) { + msg = new ApplyState(tracker.getClientActor(), tracker.getIdentifier(), replicatedLogEntry); + } else { + msg = new ApplyState(null, null, replicatedLogEntry); + } + + actor().tell(msg, actor()); newLastApplied = i; } else { //if one index is not present in the log, no point in looping // around as the rest wont be present either - LOG.warn( - "{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}", + LOG.warn("{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}", logName(), i, i, index); break; } @@ -414,34 +403,34 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { return requestVote(sender, (RequestVote) message); } else if (message instanceof RequestVoteReply) { return handleRequestVoteReply(sender, (RequestVoteReply) message); + } else { + return null; } - return this; - } - - @Override public String getLeaderId() { - return leaderId; } @Override - public short getLeaderPayloadVersion() { - return leaderPayloadVersion; + public RaftActorBehavior switchBehavior(RaftActorBehavior behavior) { + return internalSwitchBehavior(behavior); } - public void setLeaderPayloadVersion(short leaderPayloadVersion) { - this.leaderPayloadVersion = leaderPayloadVersion; + protected RaftActorBehavior internalSwitchBehavior(RaftState newState) { + if(context.getRaftPolicy().automaticElectionsEnabled()){ + return internalSwitchBehavior(createBehavior(context, newState)); + } + return this; } - protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) { - LOG.info("{} :- Switching from behavior {} to {}", logName(), this.state(), behavior.state()); + private RaftActorBehavior internalSwitchBehavior(RaftActorBehavior newBehavior) { + LOG.info("{} :- Switching from behavior {} to {}", logName(), this.state(), newBehavior.state()); try { close(); } catch (Exception e) { LOG.error("{}: Failed to close behavior : {}", logName(), this.state(), e); } - - return behavior; + return newBehavior; } + protected int getMajorityVoteCount(int numPeers) { // Votes are required from a majority of the peers including self. // The numMajority field therefore stores a calculated value @@ -473,7 +462,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param snapshotCapturedIndex */ protected void performSnapshotWithoutCapture(final long snapshotCapturedIndex) { - long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex, this); + long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex); if(actualIndex != -1){ setReplicatedToAllIndex(actualIndex); @@ -483,5 +472,4 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { protected String getId(){ return context.getId(); } - }