opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/FollowerLogInformation.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.raft;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Stopwatch;
  14 import java.util.concurrent.TimeUnit;
  15 import javax.annotation.Nonnull;
  16 import javax.annotation.Nullable;
  17 import org.opendaylight.controller.cluster.raft.behaviors.LeaderInstallSnapshotState;
  18
  19 /**
  20  * The state of the followers log as known by the Leader.
  21  *
  22  * @author Moiz Raja
  23  * @author Thomas Pantelis
  24  */
  25 public final class FollowerLogInformation {
  26     public static final long NO_INDEX = -1;
  27
  28     private final Stopwatch stopwatch = Stopwatch.createUnstarted();
  29
  30     private final RaftActorContext context;
  31
  32     private long nextIndex;
  33
  34     private long matchIndex;
  35
  36     private long lastReplicatedIndex = -1L;
  37
  38     private final Stopwatch lastReplicatedStopwatch = Stopwatch.createUnstarted();
  39
  40     private short payloadVersion = -1;
  41
  42     // Assume the HELIUM_VERSION version initially for backwards compatibility until we obtain the follower's
  43     // actual version via AppendEntriesReply. Although we no longer support the Helium version, a pre-Boron
  44     // follower will not have the version field in AppendEntriesReply so it will be set to 0 which is
  45     // HELIUM_VERSION.
  46     private short raftVersion = RaftVersions.HELIUM_VERSION;
  47
  48     private final PeerInfo peerInfo;
  49
  50     private LeaderInstallSnapshotState installSnapshotState;
  51
  52     private long slicedLogEntryIndex = NO_INDEX;
  53
  54     private boolean needsLeaderAddress;
  55
  56     /**
  57      * Constructs an instance.
  58      *
  59      * @param peerInfo the associated PeerInfo of the follower.
  60      * @param matchIndex the initial match index.
  61      * @param context the RaftActorContext.
  62      */
  63     @VisibleForTesting
  64     FollowerLogInformation(final PeerInfo peerInfo, final long matchIndex, final RaftActorContext context) {
  65         this.nextIndex = context.getCommitIndex();
  66         this.matchIndex = matchIndex;
  67         this.context = context;
  68         this.peerInfo = Preconditions.checkNotNull(peerInfo);
  69     }
  70
  71     /**
  72      * Constructs an instance with no matching index.
  73      *
  74      * @param peerInfo the associated PeerInfo of the follower.
  75      * @param context the RaftActorContext.
  76      */
  77     public FollowerLogInformation(final PeerInfo peerInfo, final RaftActorContext context) {
  78         this(peerInfo, NO_INDEX, context);
  79     }
  80
  81     /**
  82      * Increments the value of the follower's next index.
  83      *
  84      * @return the new value of nextIndex.
  85      */
  86     @VisibleForTesting
  87     long incrNextIndex() {
  88         return nextIndex++;
  89     }
  90
  91     /**
  92      * Decrements the value of the follower's next index.
  93      *
  94      * @return true if the next index was decremented, ie it was previously &gt;= 0, false otherwise.
  95      */
  96     public boolean decrNextIndex() {
  97         if (nextIndex < 0) {
  98             return false;
  99         }
 100
 101         nextIndex--;
 102         return true;
 103     }
 104
 105     /**
 106      * Sets the index of the follower's next log entry.
 107      *
 108      * @param nextIndex the new index.
 109      * @return true if the new index differed from the current index and the current index was updated, false
 110      *              otherwise.
 111      */
 112     @SuppressWarnings("checkstyle:hiddenField")
 113     public boolean setNextIndex(final long nextIndex) {
 114         if (this.nextIndex != nextIndex) {
 115             this.nextIndex = nextIndex;
 116             return true;
 117         }
 118
 119         return false;
 120     }
 121
 122     /**
 123      * Increments the value of the follower's match index.
 124      *
 125      * @return the new value of matchIndex.
 126      */
 127     public long incrMatchIndex() {
 128         return matchIndex++;
 129     }
 130
 131     /**
 132      * Sets the index of the follower's highest log entry.
 133      *
 134      * @param matchIndex the new index.
 135      * @return true if the new index differed from the current index and the current index was updated, false
 136      *              otherwise.
 137      */
 138     @SuppressWarnings("checkstyle:hiddenField")
 139     public boolean setMatchIndex(final long matchIndex) {
 140         // If the new match index is the index of the entry currently being sliced, then we know slicing is complete
 141         // and the follower received the entry and responded so clear the slicedLogEntryIndex
 142         if (isLogEntrySlicingInProgress() && slicedLogEntryIndex == matchIndex) {
 143             slicedLogEntryIndex = NO_INDEX;
 144         }
 145
 146         if (this.matchIndex != matchIndex) {
 147             this.matchIndex = matchIndex;
 148             return true;
 149         }
 150
 151         return false;
 152     }
 153
 154     /**
 155      * Returns the identifier of the follower.
 156      *
 157      * @return the identifier of the follower.
 158      */
 159     public String getId() {
 160         return peerInfo.getId();
 161     }
 162
 163     /**
 164      * Returns the index of the next log entry to send to the follower.
 165      *
 166      * @return index of the follower's next log entry.
 167      */
 168     public long getNextIndex() {
 169         return nextIndex;
 170     }
 171
 172     /**
 173      * Returns the index of highest log entry known to be replicated on the follower.
 174      *
 175      * @return the index of highest log entry.
 176      */
 177     public long getMatchIndex() {
 178         return matchIndex;
 179     }
 180
 181     /**
 182      * Checks if the follower is active by comparing the time of the last activity with the election time out. The
 183      * follower is active if some activity has occurred for the follower within the election time out interval.
 184      *
 185      * @return true if follower is active, false otherwise.
 186      */
 187     public boolean isFollowerActive() {
 188         if (peerInfo.getVotingState() == VotingState.VOTING_NOT_INITIALIZED) {
 189             return false;
 190         }
 191
 192         long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
 193         return stopwatch.isRunning()
 194                 && elapsed <= context.getConfigParams().getElectionTimeOutInterval().toMillis();
 195     }
 196
 197     /**
 198      * Marks the follower as active. This should be called when some activity has occurred for the follower.
 199      */
 200     public void markFollowerActive() {
 201         if (stopwatch.isRunning()) {
 202             stopwatch.reset();
 203         }
 204         stopwatch.start();
 205     }
 206
 207     /**
 208      * Marks the follower as inactive. This should only be called from unit tests.
 209      */
 210     @VisibleForTesting
 211     public void markFollowerInActive() {
 212         if (stopwatch.isRunning()) {
 213             stopwatch.stop();
 214         }
 215     }
 216
 217     /**
 218      * Returns the time since the last activity occurred for the follower.
 219      *
 220      * @return time in nanoseconds since the last activity from the follower.
 221      */
 222     public long nanosSinceLastActivity() {
 223         return stopwatch.elapsed(TimeUnit.NANOSECONDS);
 224     }
 225
 226     /**
 227      * This method checks if the next replicate message can be sent to the follower. This is an optimization to avoid
 228      * sending duplicate message too frequently if the last replicate message was sent and no reply has been received
 229      * yet within the current heart beat interval
 230      *
 231      * @return true if it is OK to replicate, false otherwise
 232      */
 233     public boolean okToReplicate() {
 234         if (peerInfo.getVotingState() == VotingState.VOTING_NOT_INITIALIZED) {
 235             return false;
 236         }
 237
 238         // Return false if we are trying to send duplicate data before the heartbeat interval
 239         if (getNextIndex() == lastReplicatedIndex && lastReplicatedStopwatch.elapsed(TimeUnit.MILLISECONDS)
 240                 < context.getConfigParams().getHeartBeatInterval().toMillis()) {
 241             return false;
 242         }
 243
 244         resetLastReplicated();
 245         return true;
 246     }
 247
 248     private void resetLastReplicated() {
 249         lastReplicatedIndex = getNextIndex();
 250         if (lastReplicatedStopwatch.isRunning()) {
 251             lastReplicatedStopwatch.reset();
 252         }
 253         lastReplicatedStopwatch.start();
 254     }
 255
 256     /**
 257      * Returns the log entry payload data version of the follower.
 258      *
 259      * @return the payload data version.
 260      */
 261     public short getPayloadVersion() {
 262         return payloadVersion;
 263     }
 264
 265     /**
 266      * Sets the payload data version of the follower.
 267      *
 268      * @param payloadVersion the payload data version.
 269      */
 270     public void setPayloadVersion(final short payloadVersion) {
 271         this.payloadVersion = payloadVersion;
 272     }
 273
 274     /**
 275      * Returns the the raft version of the follower.
 276      *
 277      * @return the raft version of the follower.
 278      */
 279     public short getRaftVersion() {
 280         return raftVersion;
 281     }
 282
 283     /**
 284      * Sets the raft version of the follower.
 285      *
 286      * @param raftVersion the raft version.
 287      */
 288     public void setRaftVersion(final short raftVersion) {
 289         this.raftVersion = raftVersion;
 290     }
 291
 292     /**
 293      * Returns the LeaderInstallSnapshotState for the in progress install snapshot.
 294      *
 295      * @return the LeaderInstallSnapshotState if a snapshot install is in progress, null otherwise.
 296      */
 297     @Nullable
 298     public LeaderInstallSnapshotState getInstallSnapshotState() {
 299         return installSnapshotState;
 300     }
 301
 302     /**
 303      * Sets the LeaderInstallSnapshotState when an install snapshot is initiated.
 304      *
 305      * @param state the LeaderInstallSnapshotState
 306      */
 307     public void setLeaderInstallSnapshotState(@Nonnull final LeaderInstallSnapshotState state) {
 308         if (this.installSnapshotState == null) {
 309             this.installSnapshotState = Preconditions.checkNotNull(state);
 310         }
 311     }
 312
 313     /**
 314      * Clears the LeaderInstallSnapshotState when an install snapshot is complete.
 315      */
 316     public void clearLeaderInstallSnapshotState() {
 317         Preconditions.checkState(installSnapshotState != null);
 318         installSnapshotState.close();
 319         installSnapshotState = null;
 320     }
 321
 322     /**
 323      * Sets the index of the log entry whose payload size exceeds the maximum size for a single message and thus
 324      * needs to be sliced into smaller chunks.
 325      *
 326      * @param index the log entry index or NO_INDEX to clear it
 327      */
 328     public void setSlicedLogEntryIndex(final long index) {
 329         slicedLogEntryIndex  = index;
 330     }
 331
 332     /**
 333      * Return whether or not log entry slicing is currently in progress.
 334      *
 335      * @return true if slicing is currently in progress, false otherwise
 336      */
 337     public boolean isLogEntrySlicingInProgress() {
 338         return slicedLogEntryIndex != NO_INDEX;
 339     }
 340
 341     public void setNeedsLeaderAddress(boolean value) {
 342         needsLeaderAddress = value;
 343     }
 344
 345     @Nullable
 346     public String needsLeaderAddress(String leaderId) {
 347         return needsLeaderAddress ? context.getPeerAddress(leaderId) : null;
 348     }
 349
 350     @Override
 351     public String toString() {
 352         return "FollowerLogInformation [id=" + getId() + ", nextIndex=" + nextIndex + ", matchIndex=" + matchIndex
 353                 + ", lastReplicatedIndex=" + lastReplicatedIndex + ", votingState=" + peerInfo.getVotingState()
 354                 + ", stopwatch=" + stopwatch.elapsed(TimeUnit.MILLISECONDS) + ", followerTimeoutMillis="
 355                 + context.getConfigParams().getElectionTimeOutInterval().toMillis() + "]";
 356     }
 357 }