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