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=fc2f137e886085b9f8de644a2a1ae2b660281fa3;hp=075b2873e45332364c09aee83c49e6b23e40780c;hb=13ba9adfa24716a7b27bc4cfef198b3fa5c577b0;hpb=c68b9f38d84bd45c85b8133c2054ecdd4c413a8f 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 075b2873e4..fc2f137e88 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 @@ -14,9 +14,10 @@ import java.util.Random; import java.util.concurrent.TimeUnit; 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.SerializationUtils; -import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries; +import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries; import org.opendaylight.controller.cluster.raft.base.messages.ApplyState; import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; @@ -38,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 */ @@ -58,10 +61,39 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { */ protected String leaderId = null; + private short leaderPayloadVersion = -1; + + private long replicatedToAllIndex = -1; + + private final String logName; + + private final RaftState state; - protected AbstractRaftActorBehavior(RaftActorContext context) { + protected AbstractRaftActorBehavior(RaftActorContext context, RaftState state) { this.context = context; + this.state = state; this.LOG = context.getLogger(); + + logName = String.format("%s (%s)", context.getId(), state); + } + + @Override + public RaftState state() { + return state; + } + + public String logName() { + return logName; + } + + @Override + public void setReplicatedToAllIndex(long replicatedToAllIndex) { + this.replicatedToAllIndex = replicatedToAllIndex; + } + + @Override + public long getReplicatedToAllIndex() { + return replicatedToAllIndex; } /** @@ -95,12 +127,12 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { if (appendEntries.getTerm() < currentTerm()) { if(LOG.isDebugEnabled()) { LOG.debug("{}: Cannot append entries because sender term {} is less than {}", - context.getId(), appendEntries.getTerm(), currentTerm()); + logName(), appendEntries.getTerm(), currentTerm()); } sender.tell( new AppendEntriesReply(context.getId(), currentTerm(), false, - lastIndex(), lastTerm()), actor() + lastIndex(), lastTerm(), context.getPayloadVersion()), actor() ); return this; } @@ -132,12 +164,9 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param requestVote * @return */ - protected RaftActorBehavior requestVote(ActorRef sender, - RequestVote requestVote) { + protected RaftActorBehavior requestVote(ActorRef sender, RequestVote requestVote) { - if(LOG.isDebugEnabled()) { - LOG.debug("{}: Received {}", context.getId(), requestVote); - } + LOG.debug("{}: In requestVote: {}", logName(), requestVote); boolean grantVote = false; @@ -173,7 +202,11 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { } } - sender.tell(new RequestVoteReply(currentTerm(), grantVote), actor()); + RequestVoteReply reply = new RequestVoteReply(currentTerm(), grantVote); + + LOG.debug("{}: requestVote returning: {}", logName(), reply); + + sender.tell(reply, actor()); return this; } @@ -225,7 +258,7 @@ 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()); } @@ -351,12 +384,12 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { // around as the rest wont be present either LOG.warn( "{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}", - context.getId(), i, i, index); + logName(), i, i, index); break; } } if(LOG.isDebugEnabled()) { - LOG.debug("{}: Setting last applied to {}", context.getId(), newLastApplied); + LOG.debug("{}: Setting last applied to {}", logName(), newLastApplied); } context.setLastApplied(newLastApplied); @@ -364,7 +397,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { // will be used during recovery //in case if the above code throws an error and this message is not sent, it would be fine // as the append entries received later would initiate add this message to the journal - actor().tell(new ApplyLogEntries((int) context.getLastApplied()), actor()); + actor().tell(new ApplyJournalEntries(context.getLastApplied()), actor()); } protected Object fromSerializableMessage(Object serializable){ @@ -389,12 +422,21 @@ 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 {}", context.getId(), this.state(), behavior.state()); + LOG.info("{} :- Switching from behavior {} to {}", logName(), this.state(), behavior.state()); try { close(); } catch (Exception e) { - LOG.error("{}: Failed to close behavior : {}", context.getId(), this.state(), e); + LOG.error("{}: Failed to close behavior : {}", logName(), this.state(), e); } return behavior; @@ -423,17 +465,23 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { } - protected long fakeSnapshot(final long minReplicatedToAllIndex, final long currentReplicatedIndex) { - // we would want to keep the lastApplied as its used while capturing snapshots - long tempMin = Math.min(minReplicatedToAllIndex, - (context.getLastApplied() > -1 ? context.getLastApplied() - 1 : -1)); + /** + * Performs a snapshot with no capture on the replicated log. + * It clears the log from the supplied index or last-applied-1 which ever is minimum. + * + * @param snapshotCapturedIndex + */ + protected void performSnapshotWithoutCapture(final long snapshotCapturedIndex) { + long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex, this); - if (tempMin > -1 && context.getReplicatedLog().isPresent(tempMin)) { - context.getReplicatedLog().snapshotPreCommit(tempMin, context.getTermInformation().getCurrentTerm()); - context.getReplicatedLog().snapshotCommit(); - return tempMin; + if(actualIndex != -1){ + setReplicatedToAllIndex(actualIndex); } - return currentReplicatedIndex; } + + protected String getId(){ + return context.getId(); + } + }