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%2FRaftActor.java;h=a5b5d2b81aa3d78f4e1bc91bd81d1abdd50eef03;hp=fdd4b2395bff7b3e9b8ef746450761a7ec41f323;hb=3676d1686706dbee6656e86a23c4bdb516d5267b;hpb=24ace09aacc620fd9768e0a7004e802f9385bcfc diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java index fdd4b2395b..a5b5d2b81a 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java @@ -14,17 +14,19 @@ import akka.actor.ActorSelection; import akka.actor.PoisonPill; import akka.japi.Procedure; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Objects; import com.google.common.base.Optional; -import com.google.common.base.Supplier; +import com.google.common.base.Preconditions; +import com.google.common.base.Verify; import com.google.common.collect.Lists; import java.io.Serializable; import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.concurrent.TimeUnit; import javax.annotation.Nonnull; +import javax.annotation.Nullable; import org.apache.commons.lang3.time.DurationFormatUtils; import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.DelegatingPersistentDataProvider; @@ -50,6 +52,7 @@ import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftS import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState; import org.opendaylight.controller.cluster.raft.client.messages.Shutdown; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; +import org.opendaylight.yangtools.concepts.Immutable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -109,14 +112,12 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { private final PersistentDataProvider persistentProvider; + private final BehaviorStateTracker behaviorStateTracker = new BehaviorStateTracker(); + private RaftActorRecoverySupport raftRecovery; private RaftActorSnapshotMessageSupport snapshotSupport; - private final BehaviorStateHolder reusableBehaviorStateHolder = new BehaviorStateHolder(); - - private final SwitchBehaviorSupplier reusableSwitchBehaviorSupplier = new SwitchBehaviorSupplier(); - private RaftActorServerConfigurationSupport serverConfigurationSupport; private RaftActorLeadershipTransferCohort leadershipTransferInProgress; @@ -189,18 +190,19 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } @VisibleForTesting - protected void changeCurrentBehavior(RaftActorBehavior newBehavior){ - if(getCurrentBehavior() != null) { + protected void changeCurrentBehavior(RaftActorBehavior newBehavior) { + final RaftActorBehavior currentBehavior = getCurrentBehavior(); + if (currentBehavior != null) { try { - getCurrentBehavior().close(); - } catch(Exception e) { - LOG.warn("{}: Error closing behavior {}", persistence(), getCurrentBehavior(), e); + currentBehavior.close(); + } catch (Exception e) { + LOG.warn("{}: Error closing behavior {}", persistence(), currentBehavior, e); } } - reusableBehaviorStateHolder.init(getCurrentBehavior()); + final BehaviorState state = behaviorStateTracker.capture(currentBehavior); setCurrentBehavior(newBehavior); - handleBehaviorChange(reusableBehaviorStateHolder, getCurrentBehavior()); + handleBehaviorChange(state, newBehavior); } @Override @@ -266,23 +268,27 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } else if(message instanceof Runnable) { ((Runnable)message).run(); } else { - switchBehavior(reusableSwitchBehaviorSupplier.handleMessage(getSender(), message)); + // Processing the message may affect the state, hence we need to capture it + final RaftActorBehavior currentBehavior = getCurrentBehavior(); + final BehaviorState state = behaviorStateTracker.capture(currentBehavior); + final RaftActorBehavior nextBehavior = currentBehavior.handleMessage(getSender(), message); + switchBehavior(state, nextBehavior); } } - void initiateLeadershipTransfer(final RaftActorLeadershipTransferCohort.OnComplete onComplete) { + private void initiateLeadershipTransfer(final RaftActorLeadershipTransferCohort.OnComplete onComplete) { LOG.debug("{}: Initiating leader transfer", persistenceId()); if(leadershipTransferInProgress == null) { - leadershipTransferInProgress = new RaftActorLeadershipTransferCohort(this, getSender()); + leadershipTransferInProgress = new RaftActorLeadershipTransferCohort(this); leadershipTransferInProgress.addOnComplete(new RaftActorLeadershipTransferCohort.OnComplete() { @Override - public void onSuccess(ActorRef raftActorRef, ActorRef replyTo) { + public void onSuccess(ActorRef raftActorRef) { leadershipTransferInProgress = null; } @Override - public void onFailure(ActorRef raftActorRef, ActorRef replyTo) { + public void onFailure(ActorRef raftActorRef) { leadershipTransferInProgress = null; } }); @@ -305,21 +311,27 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { shuttingDown = true; final RaftActorBehavior currentBehavior = context.getCurrentBehavior(); - if(currentBehavior.state() == RaftState.Leader && context.hasFollowers()) { + if (currentBehavior.state() != RaftState.Leader) { + // For non-leaders shutdown is a no-op + self().tell(PoisonPill.getInstance(), self()); + return; + } + + if (context.hasFollowers()) { initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() { @Override - public void onSuccess(ActorRef raftActorRef, ActorRef replyTo) { + public void onSuccess(ActorRef raftActorRef) { LOG.debug("{}: leader transfer succeeded - sending PoisonPill", persistenceId()); raftActorRef.tell(PoisonPill.getInstance(), raftActorRef); } @Override - public void onFailure(ActorRef raftActorRef, ActorRef replyTo) { + public void onFailure(ActorRef raftActorRef) { LOG.debug("{}: leader transfer failed - sending PoisonPill", persistenceId()); raftActorRef.tell(PoisonPill.getInstance(), raftActorRef); } }); - } else if(currentBehavior.state() == RaftState.Leader) { + } else { pauseLeader(new TimedRunnable(context.getConfigParams().getElectionTimeOutInterval(), this) { @Override protected void doRun() { @@ -331,8 +343,6 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { self().tell(PoisonPill.getInstance(), self()); } }); - } else { - self().tell(PoisonPill.getInstance(), self()); } } @@ -349,7 +359,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { if(!getRaftActorContext().getRaftPolicy().automaticElectionsEnabled()) { RaftState newState = message.getNewState(); if( newState == RaftState.Leader || newState == RaftState.Follower) { - switchBehavior(reusableSwitchBehaviorSupplier.handleMessage(getSender(), message)); + switchBehavior(behaviorStateTracker.capture(getCurrentBehavior()), + AbstractRaftActorBehavior.createBehavior(context, message.getNewState())); getRaftActorContext().getTermInformation().updateAndPersist(message.getNewTerm(), ""); } else { LOG.warn("Switching to behavior : {} - not supported", newState); @@ -357,12 +368,9 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } } - private void switchBehavior(Supplier supplier){ - reusableBehaviorStateHolder.init(getCurrentBehavior()); - - setCurrentBehavior(supplier.get()); - - handleBehaviorChange(reusableBehaviorStateHolder, getCurrentBehavior()); + private void switchBehavior(final BehaviorState oldBehaviorState, final RaftActorBehavior nextBehavior) { + setCurrentBehavior(nextBehavior); + handleBehaviorChange(oldBehaviorState, nextBehavior); } @VisibleForTesting @@ -420,7 +428,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } - private void handleBehaviorChange(BehaviorStateHolder oldBehaviorState, RaftActorBehavior currentBehavior) { + private void handleBehaviorChange(BehaviorState oldBehaviorState, RaftActorBehavior currentBehavior) { RaftActorBehavior oldBehavior = oldBehaviorState.getBehavior(); if (oldBehavior != currentBehavior){ @@ -432,7 +440,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { // it can happen that the state has not changed but the leader has changed. Optional roleChangeNotifier = getRoleChangeNotifier(); - if(!Objects.equal(lastValidLeaderId, currentBehavior.getLeaderId()) || + if(!Objects.equals(lastValidLeaderId, currentBehavior.getLeaderId()) || oldBehaviorState.getLeaderPayloadVersion() != currentBehavior.getLeaderPayloadVersion()) { if(roleChangeNotifier.isPresent()) { roleChangeNotifier.get().tell(newLeaderStateChanged(getId(), currentBehavior.getLeaderId(), @@ -601,13 +609,13 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { LOG.debug("{}: RaftPolicy used with prev.config {}, RaftPolicy used with newConfig {}", persistenceId(), oldRaftPolicy, newRaftPolicy); context.setConfigParams(configParams); - if (!Objects.equal(oldRaftPolicy, newRaftPolicy)) { + if (!Objects.equals(oldRaftPolicy, newRaftPolicy)) { // The RaftPolicy was modified. If the current behavior is Follower then re-initialize to Follower // but transfer the previous leaderId so it doesn't immediately try to schedule an election. This // avoids potential disruption. Otherwise, switch to Follower normally. RaftActorBehavior behavior = getCurrentBehavior(); - if(behavior instanceof Follower) { - String previousLeaderId = ((Follower)behavior).getLeaderId(); + if (behavior != null && behavior.state() == RaftState.Follower) { + String previousLeaderId = behavior.getLeaderId(); short previousLeaderPayloadVersion = behavior.getLeaderPayloadVersion(); LOG.debug("{}: Re-initializing to Follower with previous leaderId {}", persistenceId(), previousLeaderId); @@ -774,6 +782,36 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } } + /** + * Switch this member to non-voting status. This is a no-op for all behaviors except when we are the leader, + * in which case we need to step down. + */ + void becomeNonVoting() { + if (isLeader()) { + initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() { + @Override + public void onSuccess(ActorRef raftActorRef) { + LOG.debug("{}: leader transfer succeeded after change to non-voting", persistenceId()); + ensureFollowerState(); + } + + @Override + public void onFailure(ActorRef raftActorRef) { + LOG.debug("{}: leader transfer failed after change to non-voting", persistenceId()); + ensureFollowerState(); + } + + private void ensureFollowerState() { + // Whether or not leadership transfer succeeded, we have to step down as leader and + // switch to Follower so ensure that. + if (getRaftState() != RaftState.Follower) { + initializeBehavior(); + } + } + }); + } + } + /** * @deprecated Deprecated in favor of {@link org.opendaylight.controller.cluster.raft.base.messages.DeleteEntries} * whose type for fromIndex is long instead of int. This class was kept for backwards @@ -819,50 +857,88 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } } - private static class BehaviorStateHolder { - private RaftActorBehavior behavior; - private String lastValidLeaderId; - private short leaderPayloadVersion; + /** + * A point-in-time capture of {@link RaftActorBehavior} state critical for transitioning between behaviors. + */ + private static abstract class BehaviorState implements Immutable { + @Nullable abstract RaftActorBehavior getBehavior(); + @Nullable abstract String getLastValidLeaderId(); + @Nullable abstract short getLeaderPayloadVersion(); + } - void init(RaftActorBehavior behavior) { - this.behavior = behavior; - this.leaderPayloadVersion = behavior != null ? behavior.getLeaderPayloadVersion() : -1; + /** + * A {@link BehaviorState} corresponding to non-null {@link RaftActorBehavior} state. + */ + private static final class SimpleBehaviorState extends BehaviorState { + private final RaftActorBehavior behavior; + private final String lastValidLeaderId; + private final short leaderPayloadVersion; - String behaviorLeaderId = behavior != null ? behavior.getLeaderId() : null; - if(behaviorLeaderId != null) { - this.lastValidLeaderId = behaviorLeaderId; - } + SimpleBehaviorState(final String lastValidLeaderId, final RaftActorBehavior behavior) { + this.lastValidLeaderId = lastValidLeaderId; + this.behavior = Preconditions.checkNotNull(behavior); + this.leaderPayloadVersion = behavior.getLeaderPayloadVersion(); } + @Override RaftActorBehavior getBehavior() { return behavior; } + @Override String getLastValidLeaderId() { return lastValidLeaderId; } + @Override short getLeaderPayloadVersion() { return leaderPayloadVersion; } } - private class SwitchBehaviorSupplier implements Supplier { - private Object message; - private ActorRef sender; + /** + * Class tracking behavior-related information, which we need to keep around and pass across behavior switches. + * An instance is created for each RaftActor. It has two functions: + * - it keeps track of the last leader ID we have encountered since we have been created + * - it creates state capture needed to transition from one behavior to the next + */ + private static final class BehaviorStateTracker { + /** + * A {@link BehaviorState} corresponding to null {@link RaftActorBehavior} state. Since null behavior is only + * allowed before we receive the first message, we know the leader ID to be null. + */ + private static final BehaviorState NULL_BEHAVIOR_STATE = new BehaviorState() { + @Override + RaftActorBehavior getBehavior() { + return null; + } + + @Override + String getLastValidLeaderId() { + return null; + } - public SwitchBehaviorSupplier handleMessage(ActorRef sender, Object message){ - this.sender = sender; - this.message = message; - return this; - } + @Override + short getLeaderPayloadVersion() { + return -1; + } + }; - @Override - public RaftActorBehavior get() { - if(this.message instanceof SwitchBehavior){ - return AbstractRaftActorBehavior.createBehavior(context, ((SwitchBehavior) message).getNewState()); + private String lastValidLeaderId; + + BehaviorState capture(final RaftActorBehavior behavior) { + if (behavior == null) { + Verify.verify(lastValidLeaderId == null, "Null behavior with non-null last leader"); + return NULL_BEHAVIOR_STATE; + } + + final String leaderId = behavior.getLeaderId(); + if (leaderId != null) { + lastValidLeaderId = leaderId; } - return getCurrentBehavior().handleMessage(sender, message); + + return new SimpleBehaviorState(lastValidLeaderId, behavior); } } + }