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=1d78bb02276f94abc9847d95698910e33429d4d7;hp=167082711d82b46c61e90e8311473d051e06461c;hb=3019650e87a3fc05f80e8f6359e01ca5f1c5f197;hpb=a0c5aba42aa36337ff1c6760175918b786897c9e 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 167082711d..1d78bb0227 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,13 +10,14 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; import akka.actor.Cancellable; +import org.opendaylight.controller.cluster.raft.ClientRequestTracker; import org.opendaylight.controller.cluster.raft.RaftActorContext; import org.opendaylight.controller.cluster.raft.RaftState; import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; +import org.opendaylight.controller.cluster.raft.internal.messages.ApplyState; import org.opendaylight.controller.cluster.raft.internal.messages.ElectionTimeout; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; -import org.opendaylight.controller.cluster.raft.messages.RaftRPC; import org.opendaylight.controller.cluster.raft.messages.RequestVote; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; import scala.concurrent.duration.FiniteDuration; @@ -65,9 +66,13 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { /** * */ - private Cancellable electionCancel = null; + /** + * + */ + protected String leaderId = null; + protected AbstractRaftActorBehavior(RaftActorContext context) { this.context = context; @@ -83,77 +88,37 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param appendEntries The AppendEntries message - * @param suggestedState The state that the RaftActor should be in based - * on the base class's processing of the AppendEntries - * message * @return */ protected abstract RaftState handleAppendEntries(ActorRef sender, - AppendEntries appendEntries, RaftState suggestedState); + AppendEntries appendEntries); + /** + * appendEntries first processes the AppendEntries message and then + * delegates handling to a specific behavior + * + * @param sender + * @param appendEntries + * @return + */ protected RaftState appendEntries(ActorRef sender, - AppendEntries appendEntries, RaftState raftState){ + AppendEntries appendEntries) { // 1. Reply false if term < currentTerm (§5.1) - if(appendEntries.getTerm() < currentTerm()){ - sender.tell(new AppendEntriesReply(currentTerm(), false), actor()); - return state(); - } - - // 2. Reply false if log doesn’t contain an entry at prevLogIndex - // whose term matches prevLogTerm (§5.3) - ReplicatedLogEntry previousEntry = context.getReplicatedLog() - .get(appendEntries.getPrevLogIndex()); - - if(previousEntry == null || previousEntry.getTerm() != appendEntries.getPrevLogTerm()){ - sender.tell(new AppendEntriesReply(currentTerm(), false), actor()); + if (appendEntries.getTerm() < currentTerm()) { + context.getLogger().debug( + "Cannot append entries because sender term " + appendEntries + .getTerm() + " is less than " + currentTerm()); + sender.tell( + new AppendEntriesReply(context.getId(), currentTerm(), false, + lastIndex(), lastTerm()), actor() + ); return state(); } - if(appendEntries.getEntries() != null) { - // 3. If an existing entry conflicts with a new one (same index - // but different terms), delete the existing entry and all that - // follow it (§5.3) - int addEntriesFrom = 0; - for (int i = 0; - i < appendEntries.getEntries().size(); i++, addEntriesFrom++) { - ReplicatedLogEntry newEntry = context.getReplicatedLog() - .get(i + 1); - - if (newEntry != null && newEntry.getTerm() == appendEntries.getEntries().get(i).getTerm()){ - break; - } - if (newEntry != null && newEntry.getTerm() != appendEntries - .getEntries().get(i).getTerm()) { - context.getReplicatedLog().removeFrom(i + 1); - break; - } - } - - // 4. Append any new entries not already in the log - for (int i = addEntriesFrom; - i < appendEntries.getEntries().size(); i++) { - context.getReplicatedLog() - .append(appendEntries.getEntries().get(i)); - } - } - - // 5. If leaderCommit > commitIndex, set commitIndex = - // min(leaderCommit, index of last new entry) - context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(), - context.getReplicatedLog().last().getIndex())); - - // If commitIndex > lastApplied: increment lastApplied, apply - // log[lastApplied] to state machine (§5.3) - if (appendEntries.getLeaderCommit() > context.getLastApplied()) { - applyLogToStateMachine(appendEntries.getLeaderCommit()); - } - - sender.tell(new AppendEntriesReply(currentTerm(), true), actor()); - - return handleAppendEntries(sender, appendEntries, raftState); + return handleAppendEntries(sender, appendEntries); } /** @@ -166,17 +131,21 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param appendEntriesReply The AppendEntriesReply message - * @param suggestedState The state that the RaftActor should be in based - * on the base class's processing of the - * AppendEntriesReply message * @return */ - protected abstract RaftState handleAppendEntriesReply(ActorRef sender, - AppendEntriesReply appendEntriesReply, RaftState suggestedState); + AppendEntriesReply appendEntriesReply); + /** + * requestVote handles the RequestVote message. This logic is common + * for all behaviors + * + * @param sender + * @param requestVote + * @return + */ protected RaftState requestVote(ActorRef sender, - RequestVote requestVote, RaftState suggestedState) { + RequestVote requestVote) { boolean grantVote = false; @@ -201,20 +170,20 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { if (requestVote.getLastLogTerm() > lastTerm()) { candidateLatest = true; } else if ((requestVote.getLastLogTerm() == lastTerm()) - && requestVote.getLastLogIndex() >= lastTerm()) { + && requestVote.getLastLogIndex() >= lastIndex()) { candidateLatest = true; } if (candidateLatest) { grantVote = true; - context.getTermInformation().update(requestVote.getTerm(), + context.getTermInformation().updateAndPersist(requestVote.getTerm(), requestVote.getCandidateId()); } } sender.tell(new RequestVoteReply(currentTerm(), grantVote), actor()); - return suggestedState; + return state(); } /** @@ -227,32 +196,38 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param requestVoteReply The RequestVoteReply message - * @param suggestedState The state that the RaftActor should be in based - * on the base class's processing of the RequestVote - * message * @return */ - protected abstract RaftState handleRequestVoteReply(ActorRef sender, - RequestVoteReply requestVoteReply, RaftState suggestedState); + RequestVoteReply requestVoteReply); /** - * @return The derived class should return the state that corresponds to - * it's behavior + * Creates a random election duration + * + * @return */ - protected abstract RaftState state(); - protected FiniteDuration electionDuration() { long variance = new Random().nextInt(ELECTION_TIME_MAX_VARIANCE); return new FiniteDuration(ELECTION_TIME_INTERVAL + variance, TimeUnit.MILLISECONDS); } - protected void scheduleElection(FiniteDuration interval) { - + /** + * stop the scheduled election + */ + protected void stopElection() { if (electionCancel != null && !electionCancel.isCancelled()) { electionCancel.cancel(); } + } + + /** + * schedule a new election + * + * @param interval + */ + protected void scheduleElection(FiniteDuration interval) { + stopElection(); // Schedule an election. When the scheduler triggers an ElectionTimeout // message is sent to itself @@ -262,67 +237,133 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { context.getActorSystem().dispatcher(), context.getActor()); } + /** + * Get the current term + * @return + */ protected long currentTerm() { return context.getTermInformation().getCurrentTerm(); } + /** + * Get the candidate for whom we voted in the current term + * @return + */ protected String votedFor() { return context.getTermInformation().getVotedFor(); } + /** + * Get the actor associated with this behavior + * @return + */ protected ActorRef actor() { return context.getActor(); } + /** + * Get the term from the last entry in the log + * + * @return + */ protected long lastTerm() { - return context.getReplicatedLog().last().getTerm(); + return context.getReplicatedLog().lastTerm(); } + /** + * Get the index from the last entry in the log + * + * @return + */ protected long lastIndex() { - return context.getReplicatedLog().last().getIndex(); + return context.getReplicatedLog().lastIndex(); } + /** + * Find the client request tracker for a specific logIndex + * + * @param logIndex + * @return + */ + protected ClientRequestTracker findClientRequestTracker(long logIndex) { + return null; + } - @Override - public RaftState handleMessage(ActorRef sender, Object message) { - RaftState raftState = state(); - if (message instanceof RaftRPC) { - raftState = applyTerm((RaftRPC) message); - } - if (message instanceof AppendEntries) { - raftState = appendEntries(sender, (AppendEntries) message, - raftState); - } else if (message instanceof AppendEntriesReply) { - raftState = - handleAppendEntriesReply(sender, (AppendEntriesReply) message, - raftState); - } else if (message instanceof RequestVote) { - raftState = - requestVote(sender, (RequestVote) message, raftState); - } else if (message instanceof RequestVoteReply) { - raftState = - handleRequestVoteReply(sender, (RequestVoteReply) message, - raftState); + /** + * Find the log index from the previous to last entry in the log + * + * @return + */ + protected long prevLogIndex(long index){ + ReplicatedLogEntry prevEntry = + context.getReplicatedLog().get(index - 1); + if (prevEntry != null) { + return prevEntry.getIndex(); } - return raftState; + return -1; } - private RaftState applyTerm(RaftRPC rpc) { - // If RPC request or response contains term T > currentTerm: - // set currentTerm = T, convert to follower (§5.1) - // This applies to all RPC messages and responses - if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) { - context.getTermInformation().update(rpc.getTerm(), null); - return RaftState.Follower; + /** + * Find the log term from the previous to last entry in the log + * @return + */ + protected long prevLogTerm(long index){ + ReplicatedLogEntry prevEntry = + context.getReplicatedLog().get(index - 1); + if (prevEntry != null) { + return prevEntry.getTerm(); } - return state(); + return -1; } - private void applyLogToStateMachine(long index) { + /** + * Apply the provided index to the state machine + * + * @param index a log index that is known to be committed + */ + protected void applyLogToStateMachine(long index) { + // 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 = findClientRequestTracker(i); + + if (tracker != null) { + clientActor = tracker.getClientActor(); + identifier = tracker.getIdentifier(); + } + ReplicatedLogEntry replicatedLogEntry = + context.getReplicatedLog().get(i); + + if (replicatedLogEntry != null) { + actor().tell(new ApplyState(clientActor, identifier, + replicatedLogEntry), actor()); + } else { + context.getLogger().error( + "Missing index " + i + " from log. Cannot apply state."); + } + } // Send a local message to the local RaftActor (it's derived class to be // specific to apply the log to it's index) context.setLastApplied(index); } + @Override + public RaftState handleMessage(ActorRef sender, Object message) { + if (message instanceof AppendEntries) { + return appendEntries(sender, (AppendEntries) message); + } else if (message instanceof AppendEntriesReply) { + return handleAppendEntriesReply(sender, (AppendEntriesReply) message); + } else if (message instanceof RequestVote) { + return requestVote(sender, (RequestVote) message); + } else if (message instanceof RequestVoteReply) { + return handleRequestVoteReply(sender, (RequestVoteReply) message); + } + return state(); + } + @Override public String getLeaderId() { + return leaderId; + } }