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=075b2873e45332364c09aee83c49e6b23e40780c;hp=7e896fed29c4889f6aec5ce39436a1970a50e03b;hb=4b207b5356775c4b4d231ae979f9f2134f617dd1;hpb=15fa131be8b16703089a6d8508546120cf15d45d 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 7e896fed29..075b2873e4 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,22 +10,22 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; import akka.actor.Cancellable; +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.ApplyState; 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.RequestVote; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; +import org.slf4j.Logger; import scala.concurrent.duration.FiniteDuration; -import java.util.Random; -import java.util.concurrent.TimeUnit; - /** * Abstract class that represents the behavior of a RaftActor *

@@ -43,6 +43,11 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { */ protected final RaftActorContext context; + /** + * + */ + protected final Logger LOG; + /** * */ @@ -56,6 +61,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { protected AbstractRaftActorBehavior(RaftActorContext context) { this.context = context; + this.LOG = context.getLogger(); } /** @@ -70,7 +76,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param appendEntries The AppendEntries message * @return */ - protected abstract RaftState handleAppendEntries(ActorRef sender, + protected abstract RaftActorBehavior handleAppendEntries(ActorRef sender, AppendEntries appendEntries); @@ -82,19 +88,21 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param appendEntries * @return */ - protected RaftState appendEntries(ActorRef sender, + protected RaftActorBehavior appendEntries(ActorRef sender, AppendEntries appendEntries) { // 1. Reply false if term < currentTerm (§5.1) if (appendEntries.getTerm() < currentTerm()) { - context.getLogger().debug( - "Cannot append entries because sender term " + appendEntries - .getTerm() + " is less than " + currentTerm()); + if(LOG.isDebugEnabled()) { + LOG.debug("{}: Cannot append entries because sender term {} is less than {}", + context.getId(), appendEntries.getTerm(), currentTerm()); + } + sender.tell( new AppendEntriesReply(context.getId(), currentTerm(), false, lastIndex(), lastTerm()), actor() ); - return state(); + return this; } @@ -113,7 +121,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param appendEntriesReply The AppendEntriesReply message * @return */ - protected abstract RaftState handleAppendEntriesReply(ActorRef sender, + protected abstract RaftActorBehavior handleAppendEntriesReply(ActorRef sender, AppendEntriesReply appendEntriesReply); /** @@ -124,11 +132,12 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param requestVote * @return */ - protected RaftState requestVote(ActorRef sender, + protected RaftActorBehavior requestVote(ActorRef sender, RequestVote requestVote) { - - context.getLogger().debug(requestVote.toString()); + if(LOG.isDebugEnabled()) { + LOG.debug("{}: Received {}", context.getId(), requestVote); + } boolean grantVote = false; @@ -166,7 +175,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { sender.tell(new RequestVoteReply(currentTerm(), grantVote), actor()); - return state(); + return this; } /** @@ -181,7 +190,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * @param requestVoteReply The RequestVoteReply message * @return */ - protected abstract RaftState handleRequestVoteReply(ActorRef sender, + protected abstract RaftActorBehavior handleRequestVoteReply(ActorRef sender, RequestVoteReply requestVoteReply); /** @@ -192,7 +201,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { protected FiniteDuration electionDuration() { long variance = new Random().nextInt(context.getConfigParams().getElectionTimeVariance()); return context.getConfigParams().getElectionTimeOutInterval().$plus( - new FiniteDuration(variance, TimeUnit.MILLISECONDS)); + new FiniteDuration(variance, TimeUnit.MILLISECONDS)); } /** @@ -272,6 +281,17 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { return null; } + /** + * Find the client request tracker for a specific logIndex + * + * @param logIndex + * @return + */ + protected ClientRequestTracker removeClientRequestTracker(long logIndex) { + return null; + } + + /** * Find the log index from the previous to last entry in the log * @@ -311,7 +331,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { i < index + 1; i++) { ActorRef clientActor = null; String identifier = null; - ClientRequestTracker tracker = findClientRequestTracker(i); + ClientRequestTracker tracker = removeClientRequestTracker(i); if (tracker != null) { clientActor = tracker.getClientActor(); @@ -321,21 +341,30 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { 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()); newLastApplied = i; } else { //if one index is not present in the log, no point in looping // around as the rest wont be present either - context.getLogger().error( - "Missing index {} from log. Cannot apply state. Ignoring {} to {}", i, i, index ); + LOG.warn( + "{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}", + context.getId(), i, i, index); break; } } - // Send a local message to the local RaftActor (it's derived class to be - // specific to apply the log to it's index) - context.getLogger().debug("Setting last applied to {}", newLastApplied); + if(LOG.isDebugEnabled()) { + LOG.debug("{}: Setting last applied to {}", context.getId(), newLastApplied); + } context.setLastApplied(newLastApplied); + + // send a message to persist a ApplyLogEntries marker message into akka's persistent journal + // 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()); } protected Object fromSerializableMessage(Object serializable){ @@ -343,7 +372,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { } @Override - public RaftState handleMessage(ActorRef sender, Object message) { + public RaftActorBehavior handleMessage(ActorRef sender, Object message) { if (message instanceof AppendEntries) { return appendEntries(sender, (AppendEntries) message); } else if (message instanceof AppendEntriesReply) { @@ -353,10 +382,58 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { } else if (message instanceof RequestVoteReply) { return handleRequestVoteReply(sender, (RequestVoteReply) message); } - return state(); + return this; } @Override public String getLeaderId() { return leaderId; } + + protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) { + LOG.info("{} :- Switching from behavior {} to {}", context.getId(), this.state(), behavior.state()); + try { + close(); + } catch (Exception e) { + LOG.error("{}: Failed to close behavior : {}", context.getId(), this.state(), e); + } + + return behavior; + } + + protected int getMajorityVoteCount(int numPeers) { + // Votes are required from a majority of the peers including self. + // The numMajority field therefore stores a calculated value + // of the number of votes required for this candidate to win an + // election based on it's known peers. + // If a peer was added during normal operation and raft replicas + // came to know about them then the new peer would also need to be + // taken into consideration when calculating this value. + // Here are some examples for what the numMajority would be for n + // peers + // 0 peers = 1 numMajority -: (0 + 1) / 2 + 1 = 1 + // 2 peers = 2 numMajority -: (2 + 1) / 2 + 1 = 2 + // 4 peers = 3 numMajority -: (4 + 1) / 2 + 1 = 3 + + int numMajority = 0; + if (numPeers > 0) { + int self = 1; + numMajority = (numPeers + self) / 2 + 1; + } + return numMajority; + + } + + 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)); + + if (tempMin > -1 && context.getReplicatedLog().isPresent(tempMin)) { + context.getReplicatedLog().snapshotPreCommit(tempMin, context.getTermInformation().getCurrentTerm()); + context.getReplicatedLog().snapshotCommit(); + return tempMin; + } + return currentReplicatedIndex; + } }