() {
@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();
}
/**
* Derived actors can call the isLeader method to check if the current
* RaftActor is the Leader or not
*
* @return true it this RaftActor is a Leader false otherwise
*/
protected boolean isLeader() {
return context.getId().equals(currentBehavior.getLeaderId());
}
/**
* Derived actor can call getLeader if they need a reference to the Leader.
* This would be useful for example in forwarding a request to an actor
* which is the leader
*
* @return A reference to the leader if known, null otherwise
*/
protected ActorSelection getLeader(){
String leaderAddress = getLeaderAddress();
if(leaderAddress == null){
return null;
}
return context.actorSelection(leaderAddress);
}
/**
*
* @return the current leader's id
*/
protected String getLeaderId(){
return currentBehavior.getLeaderId();
}
protected RaftState getRaftState() {
return currentBehavior.state();
}
protected ReplicatedLogEntry getLastLogEntry() {
return replicatedLog.last();
}
protected Long getCurrentTerm(){
return context.getTermInformation().getCurrentTerm();
}
protected Long getCommitIndex(){
return context.getCommitIndex();
}
protected Long getLastApplied(){
return context.getLastApplied();
}
protected RaftActorContext getRaftActorContext() {
return context;
}
/**
* setPeerAddress sets the address of a known peer at a later time.
*
* This is to account for situations where a we know that a peer
* exists but we do not know an address up-front. This may also be used in
* situations where a known peer starts off in a different location and we
* need to change it's address
*
* Note that if the peerId does not match the list of peers passed to
* this actor during construction an IllegalStateException will be thrown.
*
* @param peerId
* @param peerAddress
*/
protected void setPeerAddress(String peerId, String peerAddress){
context.setPeerAddress(peerId, peerAddress);
}
protected void commitSnapshot(long sequenceNumber) {
context.getReplicatedLog().snapshotCommit();
// TODO: Not sure if we want to be this aggressive with trimming stuff
trimPersistentData(sequenceNumber);
}
/**
* The applyState method will be called by the RaftActor when some data
* needs to be applied to the actor's state
*
* @param clientActor A reference to the client who sent this message. This
* is the same reference that was passed to persistData
* by the derived actor. clientActor may be null when
* the RaftActor is behaving as a follower or during
* recovery.
* @param identifier The identifier of the persisted data. This is also
* the same identifier that was passed to persistData by
* the derived actor. identifier may be null when
* the RaftActor is behaving as a follower or during
* recovery
* @param data A piece of data that was persisted by the persistData call.
* This should NEVER be null.
*/
protected abstract void applyState(ActorRef clientActor, String identifier,
Object data);
/**
* This method is called during recovery at the start of a batch of state entries. Derived
* classes should perform any initialization needed to start a batch.
*/
protected abstract void startLogRecoveryBatch(int maxBatchSize);
/**
* This method is called during recovery to append state data to the current batch. This method
* is called 1 or more times after {@link #startLogRecoveryBatch}.
*
* @param data the state data
*/
protected abstract void appendRecoveredLogEntry(Payload data);
/**
* This method is called during recovery to reconstruct the state of the actor.
*
* @param snapshot A snapshot of the state of the actor
*/
protected abstract void applyRecoverySnapshot(ByteString snapshot);
/**
* This method is called during recovery at the end of a batch to apply the current batched
* log entries. This method is called after {@link #appendRecoveredLogEntry}.
*/
protected abstract void applyCurrentLogRecoveryBatch();
/**
* This method is called when recovery is complete.
*/
protected abstract void onRecoveryComplete();
/**
* This method will be called by the RaftActor when a snapshot needs to be
* created. The derived actor should respond with its current state.
*
* During recovery the state that is returned by the derived actor will
* be passed back to it by calling the applySnapshot method
*
* @return The current state of the actor
*/
protected abstract void createSnapshot();
/**
* This method can be called at any other point during normal
* 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
*/
protected abstract void applySnapshot(ByteString snapshot);
/**
* This method will be called by the RaftActor when the state of the
* RaftActor changes. The derived actor can then use methods like
* isLeader or getLeader to do something useful
*/
protected abstract void onStateChanged();
protected abstract DataPersistenceProvider persistence();
/**
* Notifier Actor for this RaftActor to notify when a role change happens
* @return ActorRef - ActorRef of the notifier or Optional.absent if none.
*/
protected abstract Optional getRoleChangeNotifier();
protected void onLeaderChanged(String oldLeader, String newLeader){};
private void trimPersistentData(long sequenceNumber) {
// Trim akka snapshots
// FIXME : Not sure how exactly the SnapshotSelectionCriteria is applied
// For now guessing that it is ANDed.
persistence().deleteSnapshots(new SnapshotSelectionCriteria(
sequenceNumber - context.getConfigParams().getSnapshotBatchCount(), 43200000));
// Trim akka journal
persistence().deleteMessages(sequenceNumber);
}
private String getLeaderAddress(){
if(isLeader()){
return getSelf().path().toString();
}
String leaderId = currentBehavior.getLeaderId();
if (leaderId == null) {
return null;
}
String peerAddress = context.getPeerAddress(leaderId);
if(LOG.isDebugEnabled()) {
LOG.debug("getLeaderAddress leaderId = {} peerAddress = {}",
leaderId, peerAddress);
}
return peerAddress;
}
private void handleCaptureSnapshotReply(ByteString stateInBytes) {
// 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(),
context.getReplicatedLog().getFrom(captureSnapshot.getLastAppliedIndex() + 1),
captureSnapshot.getLastIndex(), captureSnapshot.getLastTerm(),
captureSnapshot.getLastAppliedIndex(), captureSnapshot.getLastAppliedTerm());
persistence().saveSnapshot(sn);
LOG.info("Persisting of snapshot done:{}", sn.getLogMessage());
//be greedy and remove entries from in-mem journal which are in the snapshot
// and update snapshotIndex and snapshotTerm without waiting for the success,
context.getReplicatedLog().snapshotPreCommit(
captureSnapshot.getLastAppliedIndex(),
captureSnapshot.getLastAppliedTerm());
LOG.info("Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " +
"and term:{}", 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));
}
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());
}
public ReplicatedLogImpl() {
super();
}
@Override public void removeFromAndPersist(long logEntryIndex) {
int adjustedIndex = adjustedIndex(logEntryIndex);
if (adjustedIndex < 0) {
return;
}
// FIXME: Maybe this should be done after the command is saved
journal.subList(adjustedIndex , journal.size()).clear();
persistence().persist(new DeleteEntries(adjustedIndex), new Procedure(){
@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(replicatedLogEntry, null);
}
@Override
public int dataSize() {
return dataSize;
}
public void appendAndPersist(
final ReplicatedLogEntry replicatedLogEntry,
final Procedure callback) {
if(LOG.isDebugEnabled()) {
LOG.debug("Append log entry and persist {} ", replicatedLogEntry);
}
// FIXME : By adding the replicated log entry to the in-memory journal we are not truly ensuring durability of the logs
journal.add(replicatedLogEntry);
// When persisting events with persist it is guaranteed that the
// persistent actor will not receive further commands between the
// persist call and the execution(s) of the associated event
// handler. This also holds for multiple persist calls in context
// of a single command.
persistence().persist(replicatedLogEntry,
new Procedure() {
@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 &&
( 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 (!hasFollowers()) {
lastAppliedIndex = replicatedLogEntry.getIndex();
lastAppliedTerm = replicatedLogEntry.getTerm();
} else if (lastAppliedEntry != null) {
lastAppliedIndex = lastAppliedEntry.getIndex();
lastAppliedTerm = lastAppliedEntry.getTerm();
}
if(LOG.isDebugEnabled()) {
LOG.debug("Snapshot Capture logSize: {}", journal.size());
LOG.debug("Snapshot Capture lastApplied:{} ",
context.getLastApplied());
LOG.debug("Snapshot Capture lastAppliedIndex:{}", lastAppliedIndex);
LOG.debug("Snapshot Capture lastAppliedTerm:{}", lastAppliedTerm);
}
// send a CaptureSnapshot to self to make the expensive operation async.
getSelf().tell(new CaptureSnapshot(
lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm),
null);
hasSnapshotCaptureInitiated = true;
}
if(callback != null){
callback.apply(replicatedLogEntry);
}
}
}
);
}
}
static class DeleteEntries implements Serializable {
private static final long serialVersionUID = 1L;
private final int fromIndex;
public DeleteEntries(int fromIndex) {
this.fromIndex = fromIndex;
}
public int getFromIndex() {
return fromIndex;
}
}
private class ElectionTermImpl implements ElectionTerm {
/**
* Identifier of the actor whose election term information this is
*/
private long currentTerm = 0;
private String votedFor = null;
@Override
public long getCurrentTerm() {
return currentTerm;
}
@Override
public String getVotedFor() {
return votedFor;
}
@Override public void update(long currentTerm, String votedFor) {
if(LOG.isDebugEnabled()) {
LOG.debug("Set currentTerm={}, votedFor={}", currentTerm, votedFor);
}
this.currentTerm = currentTerm;
this.votedFor = votedFor;
}
@Override
public void updateAndPersist(long currentTerm, String votedFor){
update(currentTerm, votedFor);
// FIXME : Maybe first persist then update the state
persistence().persist(new UpdateElectionTerm(this.currentTerm, this.votedFor), new Procedure(){
@Override public void apply(UpdateElectionTerm param)
throws Exception {
}
});
}
}
static class UpdateElectionTerm implements Serializable {
private static final long serialVersionUID = 1L;
private final long currentTerm;
private final String votedFor;
public UpdateElectionTerm(long currentTerm, String votedFor) {
this.currentTerm = currentTerm;
this.votedFor = votedFor;
}
public long getCurrentTerm() {
return currentTerm;
}
public String getVotedFor() {
return votedFor;
}
}
protected class NonPersistentRaftDataProvider extends NonPersistentDataProvider {
public NonPersistentRaftDataProvider(){
}
/**
* The way snapshotting works is,
*
* - RaftActor calls createSnapshot on the Shard
*
- Shard sends a CaptureSnapshotReply and RaftActor then calls saveSnapshot
*
- When saveSnapshot is invoked on the akka-persistence API it uses the SnapshotStore to save the snapshot.
* The SnapshotStore sends SaveSnapshotSuccess or SaveSnapshotFailure. When the RaftActor gets SaveSnapshot
* success it commits the snapshot to the in-memory journal. This commitSnapshot is mimicking what is done
* in SaveSnapshotSuccess.
*
* @param o
*/
@Override
public void saveSnapshot(Object o) {
// Make saving Snapshot successful
commitSnapshot(-1L);
}
}
@VisibleForTesting
void setCurrentBehavior(AbstractRaftActorBehavior behavior) {
currentBehavior = behavior;
}
protected RaftActorBehavior getCurrentBehavior() {
return currentBehavior;
}
}