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%2FFollower.java;h=cc2e55d51b115cc124890c3e643ed810244a7c24;hb=refs%2Fchanges%2F75%2F13675%2F1;hp=7ada8b31c54f75c9c8f8dc9ce3456260d77af80a;hpb=1222784937d4cb70662f066779133ab2bdd27cfc;p=controller.git 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 7ada8b31c5..cc2e55d51b 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 @@ -9,7 +9,9 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; +import com.google.common.annotations.VisibleForTesting; import com.google.protobuf.ByteString; +import java.util.ArrayList; import org.opendaylight.controller.cluster.raft.RaftActorContext; import org.opendaylight.controller.cluster.raft.RaftState; import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; @@ -23,8 +25,6 @@ import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply; import org.opendaylight.controller.cluster.raft.messages.RaftRPC; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; -import java.util.ArrayList; - /** * The behavior of a RaftActor in the Follower state *

@@ -36,7 +36,8 @@ import java.util.ArrayList; * */ public class Follower extends AbstractRaftActorBehavior { - private ByteString snapshotChunksCollected = ByteString.EMPTY; + + private SnapshotTracker snapshotTracker = null; public Follower(RaftActorContext context) { super(context); @@ -44,8 +45,35 @@ public class Follower extends AbstractRaftActorBehavior { scheduleElection(electionDuration()); } + private boolean isLogEntryPresent(long index){ + if(index == context.getReplicatedLog().getSnapshotIndex()){ + return true; + } + + ReplicatedLogEntry previousEntry = context.getReplicatedLog() + .get(index); + + return previousEntry != null; + + } + + private long getLogEntryTerm(long index){ + if(index == context.getReplicatedLog().getSnapshotIndex()){ + return context.getReplicatedLog().getSnapshotTerm(); + } + + ReplicatedLogEntry previousEntry = context.getReplicatedLog() + .get(index); + + if(previousEntry != null){ + return previousEntry.getTerm(); + } + + return -1; + } + @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender, - AppendEntries appendEntries) { + AppendEntries appendEntries) { if(appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) { if(LOG.isDebugEnabled()) { @@ -66,15 +94,15 @@ public class Follower extends AbstractRaftActorBehavior { // 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()); + long prevLogTerm = getLogEntryTerm(appendEntries.getPrevLogIndex()); + boolean prevEntryPresent = isLogEntryPresent(appendEntries.getPrevLogIndex()); boolean outOfSync = true; // First check if the logs are in sync or not if (lastIndex() == -1 - && appendEntries.getPrevLogIndex() != -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 @@ -82,34 +110,34 @@ public class Follower extends AbstractRaftActorBehavior { if(LOG.isDebugEnabled()) { LOG.debug("The followers log is empty and the senders prevLogIndex is {}", - appendEntries.getPrevLogIndex()); + appendEntries.getPrevLogIndex()); } } else if (lastIndex() > -1 - && appendEntries.getPrevLogIndex() != -1 - && previousEntry == null) { + && appendEntries.getPrevLogIndex() != -1 + && !prevEntryPresent) { // The follower's log is out of sync because the Leader's // prevLogIndex entry was not found in it's log if(LOG.isDebugEnabled()) { LOG.debug("The log is not empty but the prevLogIndex {} was not found in it", - appendEntries.getPrevLogIndex()); + appendEntries.getPrevLogIndex()); } } else if (lastIndex() > -1 - && previousEntry != null - && previousEntry.getTerm()!= appendEntries.getPrevLogTerm()) { + && prevEntryPresent + && prevLogTerm != 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 - if(LOG.isDebugEnabled()) { + if (LOG.isDebugEnabled()) { LOG.debug( - "Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}" - , previousEntry.getTerm() - , appendEntries.getPrevLogTerm()); + "Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}" + , prevLogTerm + , appendEntries.getPrevLogTerm()); } } else { outOfSync = false; @@ -119,9 +147,9 @@ public class Follower extends AbstractRaftActorBehavior { // We found that the log was out of sync so just send a negative // reply and return if(LOG.isDebugEnabled()) { - LOG.debug("Follower is out-of-sync, " + + LOG.debug("Follower ({}) is out-of-sync, " + "so sending negative reply, lastIndex():{}, lastTerm():{}", - lastIndex(), lastTerm() + context.getId(), lastIndex(), lastTerm() ); } sender.tell( @@ -280,48 +308,54 @@ public class Follower extends AbstractRaftActorBehavior { ); } - try { - if (installSnapshot.getChunkIndex() == installSnapshot.getTotalChunks()) { - // this is the last chunk, create a snapshot object and apply - - snapshotChunksCollected = snapshotChunksCollected.concat(installSnapshot.getData()); - if(LOG.isDebugEnabled()) { - LOG.debug("Last chunk received: snapshotChunksCollected.size:{}", - snapshotChunksCollected.size()); - } + if(snapshotTracker == null){ + snapshotTracker = new SnapshotTracker(LOG, installSnapshot.getTotalChunks()); + } - Snapshot snapshot = Snapshot.create(snapshotChunksCollected.toByteArray(), - new ArrayList(), - installSnapshot.getLastIncludedIndex(), - installSnapshot.getLastIncludedTerm(), - installSnapshot.getLastIncludedIndex(), - installSnapshot.getLastIncludedTerm()); + try { + if(snapshotTracker.addChunk(installSnapshot.getChunkIndex(), installSnapshot.getData(), + installSnapshot.getLastChunkHashCode())){ + Snapshot snapshot = Snapshot.create(snapshotTracker.getSnapshot(), + new ArrayList(), + installSnapshot.getLastIncludedIndex(), + installSnapshot.getLastIncludedTerm(), + installSnapshot.getLastIncludedIndex(), + installSnapshot.getLastIncludedTerm()); actor().tell(new ApplySnapshot(snapshot), actor()); - } else { - // we have more to go - snapshotChunksCollected = snapshotChunksCollected.concat(installSnapshot.getData()); + snapshotTracker = null; - if(LOG.isDebugEnabled()) { - LOG.debug("Chunk={},snapshotChunksCollected.size:{}", - installSnapshot.getChunkIndex(), snapshotChunksCollected.size()); - } } sender.tell(new InstallSnapshotReply( - currentTerm(), context.getId(), installSnapshot.getChunkIndex(), - true), actor()); + currentTerm(), context.getId(), installSnapshot.getChunkIndex(), + true), actor()); + + } catch (SnapshotTracker.InvalidChunkException e) { + + sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(), + -1, false), actor()); + snapshotTracker = null; + + } catch (Exception e){ - } catch (Exception e) { LOG.error(e, "Exception in InstallSnapshot of follower:"); //send reply with success as false. The chunk will be sent again on failure sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(), - installSnapshot.getChunkIndex(), false), actor()); + installSnapshot.getChunkIndex(), false), actor()); + } } @Override public void close() throws Exception { stopElection(); } + + @VisibleForTesting + ByteString getSnapshotChunksCollected(){ + return snapshotTracker != null ? snapshotTracker.getCollectedChunks() : ByteString.EMPTY; + } + + }