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%2FFollowerLogInformationImpl.java;h=a8a33c30b20ef3c95c3f16003e3110e1227fad90;hp=15063cff5b4a8a9c0114dbad73709f27172fc050;hb=95d3c7975a423951dcbdecfbfa4cb6b7a23591cc;hpb=f3473ee42d45f1524dcafa6cc37e19e0393e9693 diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/FollowerLogInformationImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/FollowerLogInformationImpl.java index 15063cff5b..a8a33c30b2 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/FollowerLogInformationImpl.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/FollowerLogInformationImpl.java @@ -8,44 +8,53 @@ package org.opendaylight.controller.cluster.raft; +import com.google.common.base.Preconditions; import com.google.common.base.Stopwatch; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import org.opendaylight.controller.cluster.raft.behaviors.LeaderInstallSnapshotState; public class FollowerLogInformationImpl implements FollowerLogInformation { - private static final AtomicLongFieldUpdater NEXT_INDEX_UPDATER = AtomicLongFieldUpdater.newUpdater(FollowerLogInformationImpl.class, "nextIndex"); - private static final AtomicLongFieldUpdater MATCH_INDEX_UPDATER = AtomicLongFieldUpdater.newUpdater(FollowerLogInformationImpl.class, "matchIndex"); - - private final String id; - private final Stopwatch stopwatch = Stopwatch.createUnstarted(); private final RaftActorContext context; - private volatile long nextIndex; + private long nextIndex; - private volatile long matchIndex; + private long matchIndex; private long lastReplicatedIndex = -1L; private final Stopwatch lastReplicatedStopwatch = Stopwatch.createUnstarted(); + private short payloadVersion = -1; + + // Assume the HELIUM_VERSION version initially for backwards compatibility until we obtain the follower's + // actual version via AppendEntriesReply. Although we no longer support the Helium version, a pre-Boron + // follower will not have the version field in AppendEntriesReply so it will be set to 0 which is + // HELIUM_VERSION. + private short raftVersion = RaftVersions.HELIUM_VERSION; + + private final PeerInfo peerInfo; - public FollowerLogInformationImpl(String id, long matchIndex, RaftActorContext context) { - this.id = id; + private LeaderInstallSnapshotState installSnapshotState; + + public FollowerLogInformationImpl(PeerInfo peerInfo, long matchIndex, RaftActorContext context) { this.nextIndex = context.getCommitIndex(); this.matchIndex = matchIndex; this.context = context; + this.peerInfo = Preconditions.checkNotNull(peerInfo); } @Override - public long incrNextIndex(){ - return NEXT_INDEX_UPDATER.incrementAndGet(this); + public long incrNextIndex() { + return nextIndex++; } @Override public long decrNextIndex() { - return NEXT_INDEX_UPDATER.decrementAndGet(this); + return nextIndex--; } @Override @@ -60,7 +69,7 @@ public class FollowerLogInformationImpl implements FollowerLogInformation { @Override public long incrMatchIndex(){ - return MATCH_INDEX_UPDATER.incrementAndGet(this); + return matchIndex++; } @Override @@ -75,7 +84,7 @@ public class FollowerLogInformationImpl implements FollowerLogInformation { @Override public String getId() { - return id; + return peerInfo.getId(); } @Override @@ -90,6 +99,10 @@ public class FollowerLogInformationImpl implements FollowerLogInformation { @Override public boolean isFollowerActive() { + if(peerInfo.getVotingState() == VotingState.VOTING_NOT_INITIALIZED) { + return false; + } + long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS); return (stopwatch.isRunning()) && (elapsed <= context.getConfigParams().getElectionTimeOutInterval().toMillis()); @@ -117,6 +130,10 @@ public class FollowerLogInformationImpl implements FollowerLogInformation { @Override public boolean okToReplicate() { + if(peerInfo.getVotingState() == VotingState.VOTING_NOT_INITIALIZED) { + return false; + } + // Return false if we are trying to send duplicate data before the heartbeat interval if(getNextIndex() == lastReplicatedIndex){ if(lastReplicatedStopwatch.elapsed(TimeUnit.MILLISECONDS) < context.getConfigParams() @@ -137,14 +154,47 @@ public class FollowerLogInformationImpl implements FollowerLogInformation { lastReplicatedStopwatch.start(); } + @Override + public short getPayloadVersion() { + return payloadVersion; + } + + @Override + public void setPayloadVersion(short payloadVersion) { + this.payloadVersion = payloadVersion; + } + + @Override + public short getRaftVersion() { + return raftVersion; + } + + @Override + public void setRaftVersion(short raftVersion) { + this.raftVersion = raftVersion; + } + + @Override + @Nullable + public LeaderInstallSnapshotState getInstallSnapshotState() { + return installSnapshotState; + } + + @Override + public void setLeaderInstallSnapshotState(@Nonnull LeaderInstallSnapshotState state) { + this.installSnapshotState = Preconditions.checkNotNull(state); + } + + @Override + public void clearLeaderInstallSnapshotState() { + installSnapshotState = null; + } + @Override public String toString() { - StringBuilder builder = new StringBuilder(); - builder.append("FollowerLogInformationImpl [id=").append(id).append(", nextIndex=").append(nextIndex) - .append(", matchIndex=").append(matchIndex).append(", stopwatch=") - .append(stopwatch.elapsed(TimeUnit.MILLISECONDS)) - .append(", followerTimeoutMillis=") - .append(context.getConfigParams().getElectionTimeOutInterval().toMillis()).append("]"); - return builder.toString(); + return "FollowerLogInformationImpl [id=" + getId() + ", nextIndex=" + nextIndex + ", matchIndex=" + matchIndex + + ", lastReplicatedIndex=" + lastReplicatedIndex + ", votingState=" + peerInfo.getVotingState() + + ", stopwatch=" + stopwatch.elapsed(TimeUnit.MILLISECONDS) + ", followerTimeoutMillis=" + + context.getConfigParams().getElectionTimeOutInterval().toMillis() + "]"; } }