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=98cbd7b38179710e59733a0b08a2e53692e84f29;hb=61e85d54cfcd70053993f910092eba1ab3fcc850;hp=4e2eafcef04343f2c20aa174a23053daacd3b1af;hpb=3d4e718d0fe598b33e50b8d3bbcca3ff3bbb8b3a;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 4e2eafcef0..98cbd7b381 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,6 +10,7 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; import akka.actor.Cancellable; +import com.google.common.base.Preconditions; import java.util.Random; import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.raft.ClientRequestTracker; @@ -53,33 +54,41 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { */ private Cancellable electionCancel = null; - /** - * - */ - protected String leaderId = null; - - private short leaderPayloadVersion = -1; - private long replicatedToAllIndex = -1; private final String logName; private final RaftState state; - protected AbstractRaftActorBehavior(RaftActorContext context, RaftState state) { - this.context = context; - this.state = state; + AbstractRaftActorBehavior(final RaftActorContext context, final RaftState state) { + this.context = Preconditions.checkNotNull(context); + this.state = Preconditions.checkNotNull(state); this.LOG = context.getLogger(); logName = String.format("%s (%s)", context.getId(), state); } + public static RaftActorBehavior createBehavior(final RaftActorContext context, final RaftState state) { + switch (state) { + case Candidate: + return new Candidate(context); + case Follower: + return new Follower(context); + case IsolatedLeader: + return new IsolatedLeader(context); + case Leader: + return new Leader(context); + default: + throw new IllegalArgumentException("Unhandled state " + state); + } + } + @Override - public RaftState state() { + public final RaftState state() { return state; } - public String logName() { + protected final String logName() { return logName; } @@ -304,14 +313,6 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { return context.getReplicatedLog().lastIndex(); } - /** - * @param logIndex - * @return the client request tracker for the specified logIndex - */ - protected ClientRequestTracker findClientRequestTracker(long logIndex) { - return null; - } - /** * @param logIndex * @return the client request tracker for the specified logIndex @@ -320,7 +321,6 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { return null; } - /** * * @return log index from the previous to last entry in the log @@ -354,30 +354,27 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { protected void applyLogToStateMachine(final long index) { long newLastApplied = context.getLastApplied(); // 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 = removeClientRequestTracker(i); - - if (tracker != null) { - clientActor = tracker.getClientActor(); - identifier = tracker.getIdentifier(); - } - ReplicatedLogEntry replicatedLogEntry = - context.getReplicatedLog().get(i); + for (long i = context.getLastApplied() + 1; i < index + 1; i++) { + ReplicatedLogEntry replicatedLogEntry = context.getReplicatedLog().get(i); if (replicatedLogEntry != null) { // Send a local message to the local RaftActor (it's derived class to be // specific to apply the log to it's index) - actor().tell(new ApplyState(clientActor, identifier, - replicatedLogEntry), actor()); + + final ApplyState msg; + final ClientRequestTracker tracker = removeClientRequestTracker(i); + if (tracker != null) { + msg = new ApplyState(tracker.getClientActor(), tracker.getIdentifier(), replicatedLogEntry); + } else { + msg = new ApplyState(null, null, replicatedLogEntry); + } + + actor().tell(msg, actor()); newLastApplied = i; } else { //if one index is not present in the log, no point in looping // around as the rest wont be present either - LOG.warn( - "{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}", + LOG.warn("{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}", logName(), i, i, index); break; } @@ -408,21 +405,9 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { return requestVote(sender, (RequestVote) message); } else if (message instanceof RequestVoteReply) { return handleRequestVoteReply(sender, (RequestVoteReply) message); + } else { + return null; } - return this; - } - - @Override public String getLeaderId() { - return leaderId; - } - - @Override - public short getLeaderPayloadVersion() { - return leaderPayloadVersion; - } - - public void setLeaderPayloadVersion(short leaderPayloadVersion) { - this.leaderPayloadVersion = leaderPayloadVersion; } @Override @@ -432,7 +417,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { protected RaftActorBehavior internalSwitchBehavior(RaftState newState) { if(context.getRaftPolicy().automaticElectionsEnabled()){ - return internalSwitchBehavior(newState.createBehavior(context)); + return internalSwitchBehavior(createBehavior(context, newState)); } return this; } @@ -479,7 +464,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param snapshotCapturedIndex */ protected void performSnapshotWithoutCapture(final long snapshotCapturedIndex) { - long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex, this); + long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex); if(actualIndex != -1){ setReplicatedToAllIndex(actualIndex);