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