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%2FAbstractRaftActorBehavior.java;h=fd80d66ebfbc274aa676d098afba452547d7688a;hb=4f14faf543088e101a8b3d234a30f4d6b3d68d77;hp=a1bcf8541c4c26b260be916583eb031e29687897;hpb=d8d8f731bbe6c58fcbd0e616734e2e230aaf4ab4;p=controller.git 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 a1bcf8541c..fd80d66ebf 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 @@ -39,6 +39,8 @@ import scala.concurrent.duration.FiniteDuration; */ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { + protected static final ElectionTimeout ELECTION_TIMEOUT = new ElectionTimeout(); + /** * Information about the RaftActor whose behavior this class represents */ @@ -59,6 +61,8 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { */ protected String leaderId = null; + private short leaderPayloadVersion = -1; + private long replicatedToAllIndex = -1; private final String logName; @@ -102,7 +106,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); @@ -114,7 +118,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) { @@ -128,7 +132,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { sender.tell( new AppendEntriesReply(context.getId(), currentTerm(), false, - lastIndex(), lastTerm()), actor() + lastIndex(), lastTerm(), context.getPayloadVersion()), actor() ); return this; } @@ -147,7 +151,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); @@ -158,12 +162,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) @@ -173,7 +193,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; @@ -187,24 +207,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; } /** @@ -217,15 +228,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()); @@ -245,7 +255,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { /** * schedule a new election * - * @param interval + * @param interval the duration after which we should trigger a new election */ protected void scheduleElection(FiniteDuration interval) { stopElection(); @@ -254,67 +264,57 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { // message is sent to itself electionCancel = context.getActorSystem().scheduler().scheduleOnce(interval, - context.getActor(), new ElectionTimeout(), + context.getActor(), ELECTION_TIMEOUT, 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 + * @return the client request tracker for the specified logIndex */ 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; @@ -322,9 +322,8 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { /** - * 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 = @@ -336,8 +335,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 = @@ -418,6 +416,15 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { return leaderId; } + @Override + public short getLeaderPayloadVersion() { + return leaderPayloadVersion; + } + + public void setLeaderPayloadVersion(short leaderPayloadVersion) { + this.leaderPayloadVersion = leaderPayloadVersion; + } + protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) { LOG.info("{} :- Switching from behavior {} to {}", logName(), this.state(), behavior.state()); try { @@ -460,31 +467,10 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param snapshotCapturedIndex */ protected void performSnapshotWithoutCapture(final long snapshotCapturedIndex) { - // we would want to keep the lastApplied as its used while capturing snapshots - long lastApplied = context.getLastApplied(); - long tempMin = Math.min(snapshotCapturedIndex, (lastApplied > -1 ? lastApplied - 1 : -1)); - - if(LOG.isTraceEnabled()) { - LOG.trace("{}: performSnapshotWithoutCapture: snapshotCapturedIndex: {}, lastApplied: {}, tempMin: {}", - logName, snapshotCapturedIndex, lastApplied, tempMin); - } + long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex, this); - if (tempMin > -1 && context.getReplicatedLog().isPresent(tempMin)) { - LOG.debug("{}: fakeSnapshot purging log to {} for term {}", logName(), tempMin, - context.getTermInformation().getCurrentTerm()); - - //use the term of the temp-min, since we check for isPresent, entry will not be null - ReplicatedLogEntry entry = context.getReplicatedLog().get(tempMin); - context.getReplicatedLog().snapshotPreCommit(tempMin, entry.getTerm()); - context.getReplicatedLog().snapshotCommit(); - setReplicatedToAllIndex(tempMin); - } else if(tempMin > getReplicatedToAllIndex()) { - // It's possible a follower was lagging and an install snapshot advanced its match index past - // the current replicatedToAllIndex. Since the follower is now caught up we should advance the - // replicatedToAllIndex (to tempMin). The fact that tempMin wasn't found in the log is likely - // due to a previous snapshot triggered by the memory threshold exceeded, in that case we - // trim the log to the last applied index even if previous entries weren't replicated to all followers. - setReplicatedToAllIndex(tempMin); + if(actualIndex != -1){ + setReplicatedToAllIndex(actualIndex); } }