private int currentRecoveryBatchCount;
+
+
public RaftActor(String id, Map<String, String> peerAddresses) {
this(id, peerAddresses, Optional.<ConfigParams>absent());
}
* @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();
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 appendAndPersist(
final ReplicatedLogEntry replicatedLogEntry) {
- appendAndPersist(null, null, replicatedLogEntry);
+ appendAndPersist(replicatedLogEntry, null);
}
@Override
return dataSize;
}
- public void appendAndPersist(final ActorRef clientActor,
- final String identifier,
- final ReplicatedLogEntry replicatedLogEntry) {
+ 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 {
- dataSize += replicatedLogEntry.size();
+ 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 ||
- dataSize > dataThreshold)) {
+ ( 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 / commitdiff