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%2FFollower.java;h=db62dfc2ac1cd4f10c59849c0666b9291ca946a3;hp=1cfc2e0eb9565a04a451d9301b9f6a988961983d;hb=df13d272f2f74892f2decd54ece5d8286fb1c995;hpb=5924885ac74b5fa0c729004a5b66b30654a55496 diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java index 1cfc2e0eb9..db62dfc2ac 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java @@ -12,23 +12,23 @@ import akka.actor.ActorRef; 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.ApplySnapshot; -import org.opendaylight.controller.cluster.raft.internal.messages.ElectionTimeout; +import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; +import org.opendaylight.controller.cluster.raft.base.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.InstallSnapshot; +import org.opendaylight.controller.cluster.raft.messages.RaftRPC; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; /** * The behavior of a RaftActor in the Follower state - * + *

*

- * */ public class Follower extends AbstractRaftActorBehavior { public Follower(RaftActorContext context) { @@ -38,27 +38,74 @@ public class Follower extends AbstractRaftActorBehavior { } @Override protected RaftState handleAppendEntries(ActorRef sender, - AppendEntries appendEntries, RaftState suggestedState) { + AppendEntries appendEntries) { + + if(appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) { + context.getLogger() + .info("Follower: Received {}", appendEntries.toString()); + } + + // TODO : Refactor this method into a bunch of smaller methods + // to make it easier to read. Before refactoring ensure tests + // cover the code properly + + // 1. Reply false if term < currentTerm (§5.1) + // This is handled in the appendEntries method of the base class // If we got here then we do appear to be talking to the leader leaderId = appendEntries.getLeaderId(); // 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 (lastIndex() > -1 && previousEntry != null - && previousEntry.getTerm() != appendEntries - .getPrevLogTerm()) { + boolean outOfSync = true; + + // First check if the logs are in sync or not + if (lastIndex() == -1 + && appendEntries.getPrevLogIndex() != -1) { + + // The follower's log is out of sync because the leader does have + // an entry at prevLogIndex and this follower has no entries in + // it's log. + + context.getLogger().debug( + "The followers log is empty and the senders prevLogIndex is {}", + appendEntries.getPrevLogIndex()); + + } else if (lastIndex() > -1 + && appendEntries.getPrevLogIndex() != -1 + && previousEntry == null) { + + // The follower's log is out of sync because the Leader's + // prevLogIndex entry was not found in it's log + + context.getLogger().debug( + "The log is not empty but the prevLogIndex {} was not found in it", + appendEntries.getPrevLogIndex()); + + } else if (lastIndex() > -1 + && previousEntry != null + && previousEntry.getTerm()!= appendEntries.getPrevLogTerm()) { + + // The follower's log is out of sync because the Leader's + // prevLogIndex entry does exist in the follower's log but it has + // a different term in it context.getLogger().debug( - "Cannot append entries because previous entry term " - + previousEntry.getTerm() - + " is not equal to append entries prevLogTerm " - + appendEntries.getPrevLogTerm()); + "Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}" + , previousEntry.getTerm() + , appendEntries.getPrevLogTerm()); + } else { + outOfSync = false; + } + if (outOfSync) { + // We found that the log was out of sync so just send a negative + // reply and return sender.tell( new AppendEntriesReply(context.getId(), currentTerm(), false, lastIndex(), lastTerm()), actor() @@ -70,13 +117,17 @@ public class Follower extends AbstractRaftActorBehavior { && appendEntries.getEntries().size() > 0) { context.getLogger().debug( "Number of entries to be appended = " + appendEntries - .getEntries().size()); + .getEntries().size() + ); // 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; if (context.getReplicatedLog().size() > 0) { + + // Find the entry up until which the one that is not in the + // follower's log for (int i = 0; i < appendEntries.getEntries() .size(); i++, addEntriesFrom++) { @@ -90,32 +141,37 @@ public class Follower extends AbstractRaftActorBehavior { break; } - if (newEntry != null && newEntry.getTerm() == matchEntry + if (newEntry.getTerm() == matchEntry .getTerm()) { continue; } - if (newEntry != null && newEntry.getTerm() != matchEntry - .getTerm()) { - context.getLogger().debug( - "Removing entries from log starting at " - + matchEntry.getIndex()); - context.getReplicatedLog() - .removeFrom(matchEntry.getIndex()); - break; - } + + context.getLogger().debug( + "Removing entries from log starting at " + + matchEntry.getIndex() + ); + + // Entries do not match so remove all subsequent entries + context.getReplicatedLog() + .removeFromAndPersist(matchEntry.getIndex()); + break; } } context.getLogger().debug( "After cleanup entries to be added from = " + (addEntriesFrom - + lastIndex())); + + lastIndex()) + ); // 4. Append any new entries not already in the log for (int i = addEntriesFrom; i < appendEntries.getEntries().size(); i++) { - context.getLogger().debug( - "Append entry to log " + appendEntries.getEntries().get(i) - .toString()); + + context.getLogger().info( + "Append entry to log " + appendEntries.getEntries().get( + i).getData() + .toString() + ); context.getReplicatedLog() .appendAndPersist(appendEntries.getEntries().get(i)); } @@ -147,29 +203,43 @@ public class Follower extends AbstractRaftActorBehavior { sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), true, lastIndex(), lastTerm()), actor()); - return suggestedState; + return state(); } @Override protected RaftState handleAppendEntriesReply(ActorRef sender, - AppendEntriesReply appendEntriesReply, RaftState suggestedState) { - return suggestedState; + AppendEntriesReply appendEntriesReply) { + return state(); } @Override protected RaftState handleRequestVoteReply(ActorRef sender, - RequestVoteReply requestVoteReply, RaftState suggestedState) { - return suggestedState; + RequestVoteReply requestVoteReply) { + return state(); } @Override public RaftState state() { return RaftState.Follower; } - @Override public RaftState handleMessage(ActorRef sender, Object message) { - if(message instanceof ElectionTimeout){ + @Override public RaftState handleMessage(ActorRef sender, Object originalMessage) { + + Object message = fromSerializableMessage(originalMessage); + + if (message instanceof RaftRPC) { + RaftRPC rpc = (RaftRPC) message; + // 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().updateAndPersist(rpc.getTerm(), null); + } + } + + if (message instanceof ElectionTimeout) { return RaftState.Candidate; - } else if(message instanceof InstallSnapshot){ - InstallSnapshot snapshot = (InstallSnapshot) message; - actor().tell(new ApplySnapshot(snapshot), actor()); + + } else if (message instanceof InstallSnapshot) { + InstallSnapshot installSnapshot = (InstallSnapshot) message; + actor().tell(new ApplySnapshot(installSnapshot.getData()), actor()); } scheduleElection(electionDuration());