private int currentRecoveryBatchCount;
+
+
public RaftActor(String id, Map<String, String> peerAddresses) {
this(id, peerAddresses, Optional.<ConfigParams>absent());
}
onRecoveryComplete();
- RaftActorBehavior oldBehavior = currentBehavior;
- currentBehavior = new Follower(context);
- handleBehaviorChange(oldBehavior, currentBehavior);
+ initializeBehavior();
}
}
}
timer.start();
// Apply the snapshot to the actors state
- applyRecoverySnapshot(ByteString.copyFrom(snapshot.getState()));
+ applyRecoverySnapshot(snapshot.getState());
timer.stop();
LOG.info("Recovery snapshot applied for {} in {}: snapshotIndex={}, snapshotTerm={}, journal-size=" +
replicatedLog.size(), persistenceId(), timer.toString(),
- replicatedLog.snapshotIndex, replicatedLog.snapshotTerm);
+ replicatedLog.getSnapshotIndex(), replicatedLog.getSnapshotTerm());
}
private void onRecoveredJournalLogEntry(ReplicatedLogEntry logEntry) {
"Persistence Id = " + persistenceId() +
" Last index in log={}, snapshotIndex={}, snapshotTerm={}, " +
"journal-size={}",
- replicatedLog.lastIndex(), replicatedLog.snapshotIndex,
- replicatedLog.snapshotTerm, replicatedLog.size());
+ replicatedLog.lastIndex(), replicatedLog.getSnapshotIndex(),
+ replicatedLog.getSnapshotTerm(), replicatedLog.size());
+
+ initializeBehavior();
+ }
+
+ protected void initializeBehavior(){
+ changeCurrentBehavior(new Follower(context));
+ }
+ protected void changeCurrentBehavior(RaftActorBehavior newBehavior){
RaftActorBehavior oldBehavior = currentBehavior;
- currentBehavior = new Follower(context);
+ currentBehavior = newBehavior;
handleBehaviorChange(oldBehavior, currentBehavior);
}
snapshot.getLastAppliedTerm()
);
}
- applySnapshot(ByteString.copyFrom(snapshot.getState()));
+
+ applySnapshot(snapshot.getState());
//clears the followers log, sets the snapshot index to ensure adjusted-index works
replicatedLog = new ReplicatedLogImpl(snapshot);
} else if (message instanceof CaptureSnapshot) {
LOG.info("CaptureSnapshot received by actor");
- CaptureSnapshot cs = (CaptureSnapshot)message;
- captureSnapshot = cs;
- createSnapshot();
- } else if (message instanceof CaptureSnapshotReply){
- LOG.info("CaptureSnapshotReply received by actor");
- CaptureSnapshotReply csr = (CaptureSnapshotReply) message;
+ if(captureSnapshot == null) {
+ captureSnapshot = (CaptureSnapshot)message;
+ createSnapshot();
+ }
- ByteString stateInBytes = csr.getSnapshot();
- LOG.info("CaptureSnapshotReply stateInBytes size:{}", stateInBytes.size());
- handleCaptureSnapshotReply(stateInBytes);
+ } else if (message instanceof CaptureSnapshotReply){
+ handleCaptureSnapshotReply(((CaptureSnapshotReply) message).getSnapshot());
} else {
if (!(message instanceof AppendEntriesMessages.AppendEntries)
if (oldBehavior != currentBehavior){
onStateChanged();
}
- if (oldBehavior != null) {
- // it can happen that the state has not changed but the leader has changed.
- onLeaderChanged(oldBehavior.getLeaderId(), currentBehavior.getLeaderId());
- if (getRoleChangeNotifier().isPresent() && oldBehavior.state() != currentBehavior.state()) {
- // we do not want to notify when the behavior/role is set for the first time (i.e follower)
- getRoleChangeNotifier().get().tell(new RoleChanged(getId(), oldBehavior.state().name(),
- currentBehavior.state().name()), getSelf());
- }
+ String oldBehaviorLeaderId = oldBehavior == null? null : oldBehavior.getLeaderId();
+ String oldBehaviorState = oldBehavior == null? null : oldBehavior.state().name();
+
+ // it can happen that the state has not changed but the leader has changed.
+ onLeaderChanged(oldBehaviorLeaderId, currentBehavior.getLeaderId());
+
+ if (getRoleChangeNotifier().isPresent() &&
+ (oldBehavior == null || (oldBehavior.state() != currentBehavior.state()))) {
+ getRoleChangeNotifier().get().tell(
+ new RoleChanged(getId(), oldBehaviorState , currentBehavior.state().name()),
+ getSelf());
}
}
* @param identifier
* @param data
*/
- protected void persistData(ActorRef clientActor, String identifier,
- Payload data) {
+ protected void persistData(final ActorRef clientActor, final String identifier,
+ final Payload data) {
ReplicatedLogEntry replicatedLogEntry = new ReplicatedLogImplEntry(
context.getReplicatedLog().lastIndex() + 1,
LOG.debug("Persist data {}", replicatedLogEntry);
}
+ final RaftActorContext raftContext = getRaftActorContext();
+
replicatedLog
- .appendAndPersist(clientActor, identifier, replicatedLogEntry);
- }
+ .appendAndPersist(replicatedLogEntry, new Procedure<ReplicatedLogEntry>() {
+ @Override
+ public void apply(ReplicatedLogEntry replicatedLogEntry) throws Exception {
+ if(!hasFollowers()){
+ // Increment the Commit Index and the Last Applied values
+ raftContext.setCommitIndex(replicatedLogEntry.getIndex());
+ raftContext.setLastApplied(replicatedLogEntry.getIndex());
+
+ // Apply the state immediately
+ applyState(clientActor, identifier, data);
+
+ // Send a ApplyLogEntries message so that we write the fact that we applied
+ // the state to durable storage
+ self().tell(new ApplyLogEntries((int) replicatedLogEntry.getIndex()), self());
+
+ // Check if the "real" snapshot capture has been initiated. If no then do the fake snapshot
+ if(!hasSnapshotCaptureInitiated){
+ raftContext.getReplicatedLog().snapshotPreCommit(raftContext.getLastApplied(),
+ raftContext.getTermInformation().getCurrentTerm());
+ raftContext.getReplicatedLog().snapshotCommit();
+ } else {
+ LOG.debug("Skipping fake snapshotting for {} because real snapshotting is in progress", getId());
+ }
+ } else if (clientActor != null) {
+ // Send message for replication
+ currentBehavior.handleMessage(getSelf(),
+ new Replicate(clientActor, identifier,
+ replicatedLogEntry)
+ );
+ }
+
+ }
+ }); }
protected String getId() {
return context.getId();
*
* @param snapshot A snapshot of the state of the actor
*/
- protected abstract void applyRecoverySnapshot(ByteString snapshot);
+ protected abstract void applyRecoverySnapshot(byte[] snapshotBytes);
/**
* This method is called during recovery at the end of a batch to apply the current batched
* operations when the derived actor is out of sync with it's peers
* and the only way to bring it in sync is by applying a snapshot
*
- * @param snapshot A snapshot of the state of the actor
+ * @param snapshotBytes A snapshot of the state of the actor
*/
- protected abstract void applySnapshot(ByteString snapshot);
+ protected abstract void applySnapshot(byte[] snapshotBytes);
/**
* This method will be called by the RaftActor when the state of the
return peerAddress;
}
- private void handleCaptureSnapshotReply(ByteString stateInBytes) {
+ private void handleCaptureSnapshotReply(byte[] snapshotBytes) {
+ LOG.info("CaptureSnapshotReply received by actor: snapshot size {}", snapshotBytes.length);
+
// create a snapshot object from the state provided and save it
// when snapshot is saved async, SaveSnapshotSuccess is raised.
- Snapshot sn = Snapshot.create(stateInBytes.toByteArray(),
+ Snapshot sn = Snapshot.create(snapshotBytes,
context.getReplicatedLog().getFrom(captureSnapshot.getLastAppliedIndex() + 1),
captureSnapshot.getLastIndex(), captureSnapshot.getLastTerm(),
captureSnapshot.getLastAppliedIndex(), captureSnapshot.getLastAppliedTerm());
if (isLeader() && captureSnapshot.isInstallSnapshotInitiated()) {
// this would be call straight to the leader and won't initiate in serialization
- currentBehavior.handleMessage(getSelf(), new SendInstallSnapshot(stateInBytes));
+ currentBehavior.handleMessage(getSelf(), new SendInstallSnapshot(
+ ByteString.copyFrom(snapshotBytes)));
}
captureSnapshot = null;
hasSnapshotCaptureInitiated = false;
}
+ protected boolean hasFollowers(){
+ return getRaftActorContext().getPeerAddresses().keySet().size() > 0;
+ }
+
private class ReplicatedLogImpl extends AbstractReplicatedLogImpl {
+ private static final int DATA_SIZE_DIVIDER = 5;
+ private long dataSizeSinceLastSnapshot = 0;
+
public ReplicatedLogImpl(Snapshot snapshot) {
super(snapshot.getLastAppliedIndex(), snapshot.getLastAppliedTerm(),
snapshot.getUnAppliedEntries());
@Override public void apply(DeleteEntries param)
throws Exception {
//FIXME : Doing nothing for now
+ dataSize = 0;
+ for(ReplicatedLogEntry entry : journal){
+ dataSize += entry.size();
+ }
}
});
}
@Override public void appendAndPersist(
final ReplicatedLogEntry replicatedLogEntry) {
- appendAndPersist(null, null, replicatedLogEntry);
+ appendAndPersist(replicatedLogEntry, null);
}
- public void appendAndPersist(final ActorRef clientActor,
- final String identifier,
- final ReplicatedLogEntry replicatedLogEntry) {
+ @Override
+ public int dataSize() {
+ return dataSize;
+ }
+
+ public void appendAndPersist(
+ final ReplicatedLogEntry replicatedLogEntry,
+ final Procedure<ReplicatedLogEntry> callback) {
if(LOG.isDebugEnabled()) {
LOG.debug("Append log entry and persist {} ", replicatedLogEntry);
new Procedure<ReplicatedLogEntry>() {
@Override
public void apply(ReplicatedLogEntry evt) throws Exception {
+ int logEntrySize = replicatedLogEntry.size();
+
+ dataSize += logEntrySize;
+ long dataSizeForCheck = dataSize;
+
+ dataSizeSinceLastSnapshot += logEntrySize;
+ long journalSize = lastIndex()+1;
+
+ if(!hasFollowers()) {
+ // When we do not have followers we do not maintain an in-memory log
+ // due to this the journalSize will never become anything close to the
+ // snapshot batch count. In fact will mostly be 1.
+ // Similarly since the journal's dataSize depends on the entries in the
+ // journal the journal's dataSize will never reach a value close to the
+ // memory threshold.
+ // By maintaining the dataSize outside the journal we are tracking essentially
+ // what we have written to the disk however since we no longer are in
+ // need of doing a snapshot just for the sake of freeing up memory we adjust
+ // the real size of data by the DATA_SIZE_DIVIDER so that we do not snapshot as often
+ // as if we were maintaining a real snapshot
+ dataSizeForCheck = dataSizeSinceLastSnapshot / DATA_SIZE_DIVIDER;
+ }
+
+ long dataThreshold = Runtime.getRuntime().totalMemory() *
+ getRaftActorContext().getConfigParams().getSnapshotDataThresholdPercentage() / 100;
+
// when a snaphsot is being taken, captureSnapshot != null
if (hasSnapshotCaptureInitiated == false &&
- journal.size() % context.getConfigParams().getSnapshotBatchCount() == 0) {
+ ( journalSize % context.getConfigParams().getSnapshotBatchCount() == 0 ||
+ dataSizeForCheck > dataThreshold)) {
+
+ dataSizeSinceLastSnapshot = 0;
LOG.info("Initiating Snapshot Capture..");
long lastAppliedIndex = -1;
long lastAppliedTerm = -1;
ReplicatedLogEntry lastAppliedEntry = get(context.getLastApplied());
- if (lastAppliedEntry != null) {
+ if (!hasFollowers()) {
+ lastAppliedIndex = replicatedLogEntry.getIndex();
+ lastAppliedTerm = replicatedLogEntry.getTerm();
+ } else if (lastAppliedEntry != null) {
lastAppliedIndex = lastAppliedEntry.getIndex();
lastAppliedTerm = lastAppliedEntry.getTerm();
}
null);
hasSnapshotCaptureInitiated = true;
}
- // Send message for replication
- if (clientActor != null) {
- currentBehavior.handleMessage(getSelf(),
- new Replicate(clientActor, identifier,
- replicatedLogEntry)
- );
+ if(callback != null){
+ callback.apply(replicatedLogEntry);
}
}
}
Code Review / controller.git / blobdiff