opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.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.behaviors;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.ActorSelection;
  13 import akka.actor.Cancellable;
  14 import com.google.common.annotations.VisibleForTesting;
  15 import com.google.common.base.Optional;
  16 import com.google.common.base.Preconditions;
  17 import com.google.common.collect.ImmutableMap;
  18 import com.google.common.collect.ImmutableMap.Builder;
  19 import com.google.protobuf.ByteString;
  20 import java.io.IOException;
  21 import java.util.ArrayList;
  22 import java.util.Collection;
  23 import java.util.Collections;
  24 import java.util.HashMap;
  25 import java.util.List;
  26 import java.util.Map;
  27 import java.util.Map.Entry;
  28 import java.util.concurrent.TimeUnit;
  29 import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
  30 import org.opendaylight.controller.cluster.raft.ClientRequestTrackerImpl;
  31 import org.opendaylight.controller.cluster.raft.FollowerLogInformation;
  32 import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
  33 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  34 import org.opendaylight.controller.cluster.raft.RaftState;
  35 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  36 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
  37 import org.opendaylight.controller.cluster.raft.base.messages.InitiateInstallSnapshot;
  38 import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
  39 import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
  40 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
  41 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  42 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  43 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
  44 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
  45 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
  46 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  47 import scala.concurrent.duration.FiniteDuration;
  48
  49 /**
  50  * The behavior of a RaftActor when it is in the Leader state
  51  * <p/>
  52  * Leaders:
  53  * <ul>
  54  * <li> Upon election: send initial empty AppendEntries RPCs
  55  * (heartbeat) to each server; repeat during idle periods to
  56  * prevent election timeouts (§5.2)
  57  * <li> If command received from client: append entry to local log,
  58  * respond after entry applied to state machine (§5.3)
  59  * <li> If last log index ≥ nextIndex for a follower: send
  60  * AppendEntries RPC with log entries starting at nextIndex
  61  * <ul>
  62  * <li> If successful: update nextIndex and matchIndex for
  63  * follower (§5.3)
  64  * <li> If AppendEntries fails because of log inconsistency:
  65  * decrement nextIndex and retry (§5.3)
  66  * </ul>
  67  * <li> If there exists an N such that N > commitIndex, a majority
  68  * of matchIndex[i] ≥ N, and log[N].term == currentTerm:
  69  * set commitIndex = N (§5.3, §5.4).
  70  */
  71 public abstract class AbstractLeader extends AbstractRaftActorBehavior {
  72
  73     // The index of the first chunk that is sent when installing a snapshot
  74     public static final int FIRST_CHUNK_INDEX = 1;
  75
  76     // The index that the follower should respond with if it needs the install snapshot to be reset
  77     public static final int INVALID_CHUNK_INDEX = -1;
  78
  79     // This would be passed as the hash code of the last chunk when sending the first chunk
  80     public static final int INITIAL_LAST_CHUNK_HASH_CODE = -1;
  81
  82     private final Map<String, FollowerLogInformation> followerToLog;
  83     private final Map<String, FollowerToSnapshot> mapFollowerToSnapshot = new HashMap<>();
  84
  85     private Cancellable heartbeatSchedule = null;
  86
  87     private List<ClientRequestTracker> trackerList = new ArrayList<>();
  88
  89     protected final int minReplicationCount;
  90
  91     protected final int minIsolatedLeaderPeerCount;
  92
  93     private Optional<ByteString> snapshot;
  94
  95     public AbstractLeader(RaftActorContext context) {
  96         super(context);
  97
  98         final Builder<String, FollowerLogInformation> ftlBuilder = ImmutableMap.builder();
  99         for (String followerId : context.getPeerAddresses().keySet()) {
 100             FollowerLogInformation followerLogInformation =
 101                 new FollowerLogInformationImpl(followerId,
 102                     context.getCommitIndex(), -1,
 103                     context.getConfigParams().getElectionTimeOutInterval());
 104
 105             ftlBuilder.put(followerId, followerLogInformation);
 106         }
 107         followerToLog = ftlBuilder.build();
 108
 109         leaderId = context.getId();
 110
 111         LOG.debug("Election:Leader has following peers: {}", getFollowerIds());
 112
 113         minReplicationCount = getMajorityVoteCount(getFollowerIds().size());
 114
 115         // the isolated Leader peer count will be 1 less than the majority vote count.
 116         // this is because the vote count has the self vote counted in it
 117         // for e.g
 118         // 0 peers = 1 votesRequired , minIsolatedLeaderPeerCount = 0
 119         // 2 peers = 2 votesRequired , minIsolatedLeaderPeerCount = 1
 120         // 4 peers = 3 votesRequired, minIsolatedLeaderPeerCount = 2
 121         minIsolatedLeaderPeerCount = minReplicationCount > 0 ? (minReplicationCount - 1) : 0;
 122
 123         snapshot = Optional.absent();
 124
 125         // Immediately schedule a heartbeat
 126         // Upon election: send initial empty AppendEntries RPCs
 127         // (heartbeat) to each server; repeat during idle periods to
 128         // prevent election timeouts (§5.2)
 129         scheduleHeartBeat(new FiniteDuration(0, TimeUnit.SECONDS));
 130     }
 131
 132     /**
 133      * Return an immutable collection of follower identifiers.
 134      *
 135      * @return Collection of follower IDs
 136      */
 137     protected final Collection<String> getFollowerIds() {
 138         return followerToLog.keySet();
 139     }
 140
 141     private Optional<ByteString> getSnapshot() {
 142         return snapshot;
 143     }
 144
 145     @VisibleForTesting
 146     void setSnapshot(Optional<ByteString> snapshot) {
 147         this.snapshot = snapshot;
 148     }
 149
 150     @Override
 151     protected RaftActorBehavior handleAppendEntries(ActorRef sender,
 152         AppendEntries appendEntries) {
 153
 154         if(LOG.isDebugEnabled()) {
 155             LOG.debug(appendEntries.toString());
 156         }
 157
 158         return this;
 159     }
 160
 161     @Override
 162     protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
 163         AppendEntriesReply appendEntriesReply) {
 164
 165         if(! appendEntriesReply.isSuccess()) {
 166             if(LOG.isDebugEnabled()) {
 167                 LOG.debug(appendEntriesReply.toString());
 168             }
 169         }
 170
 171         // Update the FollowerLogInformation
 172         String followerId = appendEntriesReply.getFollowerId();
 173         FollowerLogInformation followerLogInformation =
 174             followerToLog.get(followerId);
 175
 176         if(followerLogInformation == null){
 177             LOG.error("Unknown follower {}", followerId);
 178             return this;
 179         }
 180
 181         followerLogInformation.markFollowerActive();
 182
 183         if (appendEntriesReply.isSuccess()) {
 184             followerLogInformation
 185                 .setMatchIndex(appendEntriesReply.getLogLastIndex());
 186             followerLogInformation
 187                 .setNextIndex(appendEntriesReply.getLogLastIndex() + 1);
 188         } else {
 189
 190             // TODO: When we find that the follower is out of sync with the
 191             // Leader we simply decrement that followers next index by 1.
 192             // Would it be possible to do better than this? The RAFT spec
 193             // does not explicitly deal with it but may be something for us to
 194             // think about
 195
 196             followerLogInformation.decrNextIndex();
 197         }
 198
 199         // Now figure out if this reply warrants a change in the commitIndex
 200         // If there exists an N such that N > commitIndex, a majority
 201         // of matchIndex[i] ≥ N, and log[N].term == currentTerm:
 202         // set commitIndex = N (§5.3, §5.4).
 203         for (long N = context.getCommitIndex() + 1; ; N++) {
 204             int replicatedCount = 1;
 205
 206             for (FollowerLogInformation info : followerToLog.values()) {
 207                 if (info.getMatchIndex().get() >= N) {
 208                     replicatedCount++;
 209                 }
 210             }
 211
 212             if (replicatedCount >= minReplicationCount) {
 213                 ReplicatedLogEntry replicatedLogEntry = context.getReplicatedLog().get(N);
 214                 if (replicatedLogEntry != null &&
 215                     replicatedLogEntry.getTerm() == currentTerm()) {
 216                     context.setCommitIndex(N);
 217                 }
 218             } else {
 219                 break;
 220             }
 221         }
 222
 223         // Apply the change to the state machine
 224         if (context.getCommitIndex() > context.getLastApplied()) {
 225             applyLogToStateMachine(context.getCommitIndex());
 226         }
 227
 228         return this;
 229     }
 230
 231     protected ClientRequestTracker removeClientRequestTracker(long logIndex) {
 232
 233         ClientRequestTracker toRemove = findClientRequestTracker(logIndex);
 234         if(toRemove != null) {
 235             trackerList.remove(toRemove);
 236         }
 237
 238         return toRemove;
 239     }
 240
 241     protected ClientRequestTracker findClientRequestTracker(long logIndex) {
 242         for (ClientRequestTracker tracker : trackerList) {
 243             if (tracker.getIndex() == logIndex) {
 244                 return tracker;
 245             }
 246         }
 247         return null;
 248     }
 249
 250     @Override
 251     protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
 252         RequestVoteReply requestVoteReply) {
 253         return this;
 254     }
 255
 256     @Override
 257     public RaftState state() {
 258         return RaftState.Leader;
 259     }
 260
 261     @Override
 262     public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
 263         Preconditions.checkNotNull(sender, "sender should not be null");
 264
 265         Object message = fromSerializableMessage(originalMessage);
 266
 267         if (message instanceof RaftRPC) {
 268             RaftRPC rpc = (RaftRPC) message;
 269             // If RPC request or response contains term T > currentTerm:
 270             // set currentTerm = T, convert to follower (§5.1)
 271             // This applies to all RPC messages and responses
 272             if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
 273                 context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
 274
 275                 return switchBehavior(new Follower(context));
 276             }
 277         }
 278
 279         try {
 280             if (message instanceof SendHeartBeat) {
 281                 sendHeartBeat();
 282                 return this;
 283
 284             } else if(message instanceof InitiateInstallSnapshot) {
 285                 installSnapshotIfNeeded();
 286
 287             } else if(message instanceof SendInstallSnapshot) {
 288                 // received from RaftActor
 289                 setSnapshot(Optional.of(((SendInstallSnapshot) message).getSnapshot()));
 290                 sendInstallSnapshot();
 291
 292             } else if (message instanceof Replicate) {
 293                 replicate((Replicate) message);
 294
 295             } else if (message instanceof InstallSnapshotReply){
 296                 handleInstallSnapshotReply((InstallSnapshotReply) message);
 297
 298             }
 299         } finally {
 300             scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval());
 301         }
 302
 303         return super.handleMessage(sender, message);
 304     }
 305
 306     private void handleInstallSnapshotReply(InstallSnapshotReply reply) {
 307         String followerId = reply.getFollowerId();
 308         FollowerToSnapshot followerToSnapshot = mapFollowerToSnapshot.get(followerId);
 309         FollowerLogInformation followerLogInformation = followerToLog.get(followerId);
 310         followerLogInformation.markFollowerActive();
 311
 312         if (followerToSnapshot != null &&
 313             followerToSnapshot.getChunkIndex() == reply.getChunkIndex()) {
 314
 315             if (reply.isSuccess()) {
 316                 if(followerToSnapshot.isLastChunk(reply.getChunkIndex())) {
 317                     //this was the last chunk reply
 318                     if(LOG.isDebugEnabled()) {
 319                         LOG.debug("InstallSnapshotReply received, " +
 320                                 "last chunk received, Chunk:{}. Follower:{} Setting nextIndex:{}",
 321                             reply.getChunkIndex(), followerId,
 322                             context.getReplicatedLog().getSnapshotIndex() + 1
 323                         );
 324                     }
 325
 326                     followerLogInformation.setMatchIndex(
 327                         context.getReplicatedLog().getSnapshotIndex());
 328                     followerLogInformation.setNextIndex(
 329                         context.getReplicatedLog().getSnapshotIndex() + 1);
 330                     mapFollowerToSnapshot.remove(followerId);
 331
 332                     if(LOG.isDebugEnabled()) {
 333                         LOG.debug("followerToLog.get(followerId).getNextIndex().get()=" +
 334                             followerToLog.get(followerId).getNextIndex().get());
 335                     }
 336
 337                     if (mapFollowerToSnapshot.isEmpty()) {
 338                         // once there are no pending followers receiving snapshots
 339                         // we can remove snapshot from the memory
 340                         setSnapshot(Optional.<ByteString>absent());
 341                     }
 342
 343                 } else {
 344                     followerToSnapshot.markSendStatus(true);
 345                 }
 346             } else {
 347                 LOG.info("InstallSnapshotReply received, " +
 348                         "sending snapshot chunk failed, Will retry, Chunk:{}",
 349                     reply.getChunkIndex()
 350                 );
 351
 352                 followerToSnapshot.markSendStatus(false);
 353             }
 354
 355         } else {
 356             LOG.error("ERROR!!" +
 357                     "FollowerId in InstallSnapshotReply not known to Leader" +
 358                     " or Chunk Index in InstallSnapshotReply not matching {} != {}",
 359                 followerToSnapshot.getChunkIndex(), reply.getChunkIndex()
 360             );
 361
 362             if(reply.getChunkIndex() == INVALID_CHUNK_INDEX){
 363                 // Since the Follower did not find this index to be valid we should reset the follower snapshot
 364                 // so that Installing the snapshot can resume from the beginning
 365                 followerToSnapshot.reset();
 366             }
 367         }
 368     }
 369
 370     private void replicate(Replicate replicate) {
 371         long logIndex = replicate.getReplicatedLogEntry().getIndex();
 372
 373         if(LOG.isDebugEnabled()) {
 374             LOG.debug("Replicate message {}", logIndex);
 375         }
 376
 377         // Create a tracker entry we will use this later to notify the
 378         // client actor
 379         trackerList.add(
 380             new ClientRequestTrackerImpl(replicate.getClientActor(),
 381                 replicate.getIdentifier(),
 382                 logIndex)
 383         );
 384
 385         if (followerToLog.isEmpty()) {
 386             context.setCommitIndex(logIndex);
 387             applyLogToStateMachine(logIndex);
 388         } else {
 389             sendAppendEntries();
 390         }
 391     }
 392
 393     private void sendAppendEntries() {
 394         // Send an AppendEntries to all followers
 395         for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
 396             final String followerId = e.getKey();
 397             ActorSelection followerActor = context.getPeerActorSelection(followerId);
 398
 399             if (followerActor != null) {
 400                 FollowerLogInformation followerLogInformation = followerToLog.get(followerId);
 401                 long followerNextIndex = followerLogInformation.getNextIndex().get();
 402                 boolean isFollowerActive = followerLogInformation.isFollowerActive();
 403                 List<ReplicatedLogEntry> entries = null;
 404
 405                 if (mapFollowerToSnapshot.get(followerId) != null) {
 406                     // if install snapshot is in process , then sent next chunk if possible
 407                     if (isFollowerActive && mapFollowerToSnapshot.get(followerId).canSendNextChunk()) {
 408                         sendSnapshotChunk(followerActor, followerId);
 409                     } else {
 410                         // we send a heartbeat even if we have not received a reply for the last chunk
 411                         sendAppendEntriesToFollower(followerActor, followerNextIndex,
 412                             Collections.<ReplicatedLogEntry>emptyList());
 413                     }
 414
 415                 } else {
 416                     long leaderLastIndex = context.getReplicatedLog().lastIndex();
 417                     long leaderSnapShotIndex = context.getReplicatedLog().getSnapshotIndex();
 418
 419                     if (isFollowerActive &&
 420                         context.getReplicatedLog().isPresent(followerNextIndex)) {
 421                         // FIXME : Sending one entry at a time
 422                         entries = context.getReplicatedLog().getFrom(followerNextIndex, 1);
 423
 424                     } else if (isFollowerActive && followerNextIndex >= 0 &&
 425                         leaderLastIndex >= followerNextIndex ) {
 426                         // if the followers next index is not present in the leaders log, and
 427                         // if the follower is just not starting and if leader's index is more than followers index
 428                         // then snapshot should be sent
 429
 430                         if(LOG.isDebugEnabled()) {
 431                             LOG.debug("InitiateInstallSnapshot to follower:{}," +
 432                                     "follower-nextIndex:{}, leader-snapshot-index:{},  " +
 433                                     "leader-last-index:{}", followerId,
 434                                 followerNextIndex, leaderSnapShotIndex, leaderLastIndex
 435                             );
 436                         }
 437                         actor().tell(new InitiateInstallSnapshot(), actor());
 438
 439                         // we would want to sent AE as the capture snapshot might take time
 440                         entries =  Collections.<ReplicatedLogEntry>emptyList();
 441
 442                     } else {
 443                         //we send an AppendEntries, even if the follower is inactive
 444                         // in-order to update the followers timestamp, in case it becomes active again
 445                         entries =  Collections.<ReplicatedLogEntry>emptyList();
 446                     }
 447
 448                     sendAppendEntriesToFollower(followerActor, followerNextIndex, entries);
 449
 450                 }
 451             }
 452         }
 453     }
 454
 455     private void sendAppendEntriesToFollower(ActorSelection followerActor, long followerNextIndex,
 456         List<ReplicatedLogEntry> entries) {
 457         followerActor.tell(
 458             new AppendEntries(currentTerm(), context.getId(),
 459                 prevLogIndex(followerNextIndex),
 460                 prevLogTerm(followerNextIndex), entries,
 461                 context.getCommitIndex()).toSerializable(),
 462             actor()
 463         );
 464     }
 465
 466     /**
 467      * An installSnapshot is scheduled at a interval that is a multiple of
 468      * a HEARTBEAT_INTERVAL. This is to avoid the need to check for installing
 469      * snapshots at every heartbeat.
 470      *
 471      * Install Snapshot works as follows
 472      * 1. Leader sends a InitiateInstallSnapshot message to self
 473      * 2. Leader then initiates the capture snapshot by sending a CaptureSnapshot message to actor
 474      * 3. RaftActor on receipt of the CaptureSnapshotReply (from Shard), stores the received snapshot in the replicated log
 475      * and makes a call to Leader's handleMessage , with SendInstallSnapshot message.
 476      * 4. Leader , picks the snapshot from im-mem ReplicatedLog and sends it in chunks to the Follower
 477      * 5. On complete, Follower sends back a InstallSnapshotReply.
 478      * 6. On receipt of the InstallSnapshotReply for the last chunk, Leader marks the install complete for that follower
 479      * and replenishes the memory by deleting the snapshot in Replicated log.
 480      *
 481      */
 482     private void installSnapshotIfNeeded() {
 483         for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
 484             final ActorSelection followerActor = context.getPeerActorSelection(e.getKey());
 485
 486             if (followerActor != null) {
 487                 long nextIndex = e.getValue().getNextIndex().get();
 488
 489                 if (!context.getReplicatedLog().isPresent(nextIndex) &&
 490                     context.getReplicatedLog().isInSnapshot(nextIndex)) {
 491                     LOG.info("{} follower needs a snapshot install", e.getKey());
 492                     if (snapshot.isPresent()) {
 493                         // if a snapshot is present in the memory, most likely another install is in progress
 494                         // no need to capture snapshot
 495                         sendSnapshotChunk(followerActor, e.getKey());
 496
 497                     } else {
 498                         initiateCaptureSnapshot();
 499                         //we just need 1 follower who would need snapshot to be installed.
 500                         // when we have the snapshot captured, we would again check (in SendInstallSnapshot)
 501                         // who needs an install and send to all who need
 502                         break;
 503                     }
 504
 505                 }
 506             }
 507         }
 508     }
 509
 510     // on every install snapshot, we try to capture the snapshot.
 511     // Once a capture is going on, another one issued will get ignored by RaftActor.
 512     private void initiateCaptureSnapshot() {
 513         LOG.info("Initiating Snapshot Capture to Install Snapshot, Leader:{}", getLeaderId());
 514         ReplicatedLogEntry lastAppliedEntry = context.getReplicatedLog().get(context.getLastApplied());
 515         long lastAppliedIndex = -1;
 516         long lastAppliedTerm = -1;
 517
 518         if (lastAppliedEntry != null) {
 519             lastAppliedIndex = lastAppliedEntry.getIndex();
 520             lastAppliedTerm = lastAppliedEntry.getTerm();
 521         } else if (context.getReplicatedLog().getSnapshotIndex() > -1)  {
 522             lastAppliedIndex = context.getReplicatedLog().getSnapshotIndex();
 523             lastAppliedTerm = context.getReplicatedLog().getSnapshotTerm();
 524         }
 525
 526         boolean isInstallSnapshotInitiated = true;
 527         actor().tell(new CaptureSnapshot(lastIndex(), lastTerm(),
 528                 lastAppliedIndex, lastAppliedTerm, isInstallSnapshotInitiated),
 529             actor());
 530     }
 531
 532
 533     private void sendInstallSnapshot() {
 534         for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
 535             ActorSelection followerActor = context.getPeerActorSelection(e.getKey());
 536
 537             if (followerActor != null) {
 538                 long nextIndex = e.getValue().getNextIndex().get();
 539
 540                 if (!context.getReplicatedLog().isPresent(nextIndex) &&
 541                     context.getReplicatedLog().isInSnapshot(nextIndex)) {
 542                     sendSnapshotChunk(followerActor, e.getKey());
 543                 }
 544             }
 545         }
 546     }
 547
 548     /**
 549      *  Sends a snapshot chunk to a given follower
 550      *  InstallSnapshot should qualify as a heartbeat too.
 551      */
 552     private void sendSnapshotChunk(ActorSelection followerActor, String followerId) {
 553         try {
 554             if (snapshot.isPresent()) {
 555                 followerActor.tell(
 556                     new InstallSnapshot(currentTerm(), context.getId(),
 557                         context.getReplicatedLog().getSnapshotIndex(),
 558                         context.getReplicatedLog().getSnapshotTerm(),
 559                         getNextSnapshotChunk(followerId,snapshot.get()),
 560                         mapFollowerToSnapshot.get(followerId).incrementChunkIndex(),
 561                         mapFollowerToSnapshot.get(followerId).getTotalChunks(),
 562                         Optional.of(mapFollowerToSnapshot.get(followerId).getLastChunkHashCode())
 563                     ).toSerializable(),
 564                     actor()
 565                 );
 566                 LOG.info("InstallSnapshot sent to follower {}, Chunk: {}/{}",
 567                     followerActor.path(), mapFollowerToSnapshot.get(followerId).getChunkIndex(),
 568                     mapFollowerToSnapshot.get(followerId).getTotalChunks());
 569             }
 570         } catch (IOException e) {
 571             LOG.error(e, "InstallSnapshot failed for Leader.");
 572         }
 573     }
 574
 575     /**
 576      * Acccepts snaphot as ByteString, enters into map for future chunks
 577      * creates and return a ByteString chunk
 578      */
 579     private ByteString getNextSnapshotChunk(String followerId, ByteString snapshotBytes) throws IOException {
 580         FollowerToSnapshot followerToSnapshot = mapFollowerToSnapshot.get(followerId);
 581         if (followerToSnapshot == null) {
 582             followerToSnapshot = new FollowerToSnapshot(snapshotBytes);
 583             mapFollowerToSnapshot.put(followerId, followerToSnapshot);
 584         }
 585         ByteString nextChunk = followerToSnapshot.getNextChunk();
 586         if (LOG.isDebugEnabled()) {
 587             LOG.debug("Leader's snapshot nextChunk size:{}", nextChunk.size());
 588         }
 589         return nextChunk;
 590     }
 591
 592     private void sendHeartBeat() {
 593         if (!followerToLog.isEmpty()) {
 594             sendAppendEntries();
 595         }
 596     }
 597
 598     private void stopHeartBeat() {
 599         if (heartbeatSchedule != null && !heartbeatSchedule.isCancelled()) {
 600             heartbeatSchedule.cancel();
 601         }
 602     }
 603
 604     private void scheduleHeartBeat(FiniteDuration interval) {
 605         if (followerToLog.isEmpty()) {
 606             // Optimization - do not bother scheduling a heartbeat as there are
 607             // no followers
 608             return;
 609         }
 610
 611         stopHeartBeat();
 612
 613         // Schedule a heartbeat. When the scheduler triggers a SendHeartbeat
 614         // message is sent to itself.
 615         // Scheduling the heartbeat only once here because heartbeats do not
 616         // need to be sent if there are other messages being sent to the remote
 617         // actor.
 618         heartbeatSchedule = context.getActorSystem().scheduler().scheduleOnce(
 619             interval, context.getActor(), new SendHeartBeat(),
 620             context.getActorSystem().dispatcher(), context.getActor());
 621     }
 622
 623     @Override
 624     public void close() throws Exception {
 625         stopHeartBeat();
 626     }
 627
 628     @Override
 629     public String getLeaderId() {
 630         return context.getId();
 631     }
 632
 633     protected boolean isLeaderIsolated() {
 634         int minPresent = minIsolatedLeaderPeerCount;
 635         for (FollowerLogInformation followerLogInformation : followerToLog.values()) {
 636             if (followerLogInformation.isFollowerActive()) {
 637                 --minPresent;
 638                 if (minPresent == 0) {
 639                     break;
 640                 }
 641             }
 642         }
 643         return (minPresent != 0);
 644     }
 645
 646     /**
 647      * Encapsulates the snapshot bytestring and handles the logic of sending
 648      * snapshot chunks
 649      */
 650     protected class FollowerToSnapshot {
 651         private ByteString snapshotBytes;
 652         private int offset = 0;
 653         // the next snapshot chunk is sent only if the replyReceivedForOffset matches offset
 654         private int replyReceivedForOffset;
 655         // if replyStatus is false, the previous chunk is attempted
 656         private boolean replyStatus = false;
 657         private int chunkIndex;
 658         private int totalChunks;
 659         private int lastChunkHashCode = AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE;
 660         private int nextChunkHashCode = AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE;
 661
 662         public FollowerToSnapshot(ByteString snapshotBytes) {
 663             this.snapshotBytes = snapshotBytes;
 664             int size = snapshotBytes.size();
 665             totalChunks = ( size / context.getConfigParams().getSnapshotChunkSize()) +
 666                 ((size % context.getConfigParams().getSnapshotChunkSize()) > 0 ? 1 : 0);
 667             if(LOG.isDebugEnabled()) {
 668                 LOG.debug("Snapshot {} bytes, total chunks to send:{}",
 669                     size, totalChunks);
 670             }
 671             replyReceivedForOffset = -1;
 672             chunkIndex = AbstractLeader.FIRST_CHUNK_INDEX;
 673         }
 674
 675         public ByteString getSnapshotBytes() {
 676             return snapshotBytes;
 677         }
 678
 679         public int incrementOffset() {
 680             if(replyStatus) {
 681                 // if prev chunk failed, we would want to sent the same chunk again
 682                 offset = offset + context.getConfigParams().getSnapshotChunkSize();
 683             }
 684             return offset;
 685         }
 686
 687         public int incrementChunkIndex() {
 688             if (replyStatus) {
 689                 // if prev chunk failed, we would want to sent the same chunk again
 690                 chunkIndex =  chunkIndex + 1;
 691             }
 692             return chunkIndex;
 693         }
 694
 695         public int getChunkIndex() {
 696             return chunkIndex;
 697         }
 698
 699         public int getTotalChunks() {
 700             return totalChunks;
 701         }
 702
 703         public boolean canSendNextChunk() {
 704             // we only send a false if a chunk is sent but we have not received a reply yet
 705             return replyReceivedForOffset == offset;
 706         }
 707
 708         public boolean isLastChunk(int chunkIndex) {
 709             return totalChunks == chunkIndex;
 710         }
 711
 712         public void markSendStatus(boolean success) {
 713             if (success) {
 714                 // if the chunk sent was successful
 715                 replyReceivedForOffset = offset;
 716                 replyStatus = true;
 717                 lastChunkHashCode = nextChunkHashCode;
 718             } else {
 719                 // if the chunk sent was failure
 720                 replyReceivedForOffset = offset;
 721                 replyStatus = false;
 722             }
 723         }
 724
 725         public ByteString getNextChunk() {
 726             int snapshotLength = getSnapshotBytes().size();
 727             int start = incrementOffset();
 728             int size = context.getConfigParams().getSnapshotChunkSize();
 729             if (context.getConfigParams().getSnapshotChunkSize() > snapshotLength) {
 730                 size = snapshotLength;
 731             } else {
 732                 if ((start + context.getConfigParams().getSnapshotChunkSize()) > snapshotLength) {
 733                     size = snapshotLength - start;
 734                 }
 735             }
 736
 737             if(LOG.isDebugEnabled()) {
 738                 LOG.debug("length={}, offset={},size={}",
 739                     snapshotLength, start, size);
 740             }
 741             ByteString substring = getSnapshotBytes().substring(start, start + size);
 742             nextChunkHashCode = substring.hashCode();
 743             return substring;
 744         }
 745
 746         /**
 747          * reset should be called when the Follower needs to be sent the snapshot from the beginning
 748          */
 749         public void reset(){
 750             offset = 0;
 751             replyStatus = false;
 752             replyReceivedForOffset = offset;
 753             chunkIndex = AbstractLeader.FIRST_CHUNK_INDEX;
 754             lastChunkHashCode = AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE;
 755         }
 756
 757         public int getLastChunkHashCode() {
 758             return lastChunkHashCode;
 759         }
 760     }
 761
 762     // called from example-actor for printing the follower-states
 763     public String printFollowerStates() {
 764         final StringBuilder sb = new StringBuilder();
 765
 766         sb.append('[');
 767         for (FollowerLogInformation followerLogInformation : followerToLog.values()) {
 768             sb.append('{');
 769             sb.append(followerLogInformation.getId());
 770             sb.append(" state:");
 771             sb.append(followerLogInformation.isFollowerActive());
 772             sb.append("},");
 773         }
 774         sb.append(']');
 775
 776         return sb.toString();
 777     }
 778
 779     @VisibleForTesting
 780     public FollowerLogInformation getFollower(String followerId) {
 781         return followerToLog.get(followerId);
 782     }
 783
 784     @VisibleForTesting
 785     protected void setFollowerSnapshot(String followerId, FollowerToSnapshot snapshot) {
 786         mapFollowerToSnapshot.put(followerId, snapshot);
 787     }
 788
 789     @VisibleForTesting
 790     public int followerSnapshotSize() {
 791         return mapFollowerToSnapshot.size();
 792     }
 793
 794     @VisibleForTesting
 795     public int followerLogSize() {
 796         return followerToLog.size();
 797     }
 798 }