Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Apply commands to state machine in dedicated executor.
This means that the main raft thread is no longer respsonsible for applying commands, so its speed should not be limited by expensive application.
- Loading branch information
1 parent
f75c8d5
commit 9f2998e
Showing
27 changed files
with
638 additions
and
380 deletions.
There are no files selected for viewing
32 changes: 32 additions & 0 deletions
32
enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/ConsensusListener.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| * Copyright (c) 2002-2016 "Neo Technology," | ||
| * Network Engine for Objects in Lund AB [http://neotechnology.com] | ||
| * | ||
| * This file is part of Neo4j. | ||
| * | ||
| * Neo4j is free software: you can redistribute it and/or modify | ||
| * it under the terms of the GNU Affero General Public License as | ||
| * published by the Free Software Foundation, either version 3 of the | ||
| * License, or (at your option) any later version. | ||
| * | ||
| * This program is distributed in the hope that it will be useful, | ||
| * but WITHOUT ANY WARRANTY; without even the implied warranty of | ||
| * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||
| * GNU Affero General Public License for more details. | ||
| * | ||
| * You should have received a copy of the GNU Affero General Public License | ||
| * along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
| */ | ||
| package org.neo4j.coreedge.raft; | ||
|
|
||
| /** | ||
| * A consensus listener is notified when a particular index in the consensus log | ||
| * is considered committed. | ||
| */ | ||
| public interface ConsensusListener | ||
| { | ||
| /** | ||
| * Called when the highest committed index increases. | ||
| */ | ||
| void notifyCommitted(); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
75 changes: 75 additions & 0 deletions
75
enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/state/LastAppliedState.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| /* | ||
| * Copyright (c) 2002-2016 "Neo Technology," | ||
| * Network Engine for Objects in Lund AB [http://neotechnology.com] | ||
| * | ||
| * This file is part of Neo4j. | ||
| * | ||
| * Neo4j is free software: you can redistribute it and/or modify | ||
| * it under the terms of the GNU Affero General Public License as | ||
| * published by the Free Software Foundation, either version 3 of the | ||
| * License, or (at your option) any later version. | ||
| * | ||
| * This program is distributed in the hope that it will be useful, | ||
| * but WITHOUT ANY WARRANTY; without even the implied warranty of | ||
| * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||
| * GNU Affero General Public License for more details. | ||
| * | ||
| * You should have received a copy of the GNU Affero General Public License | ||
| * along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
| */ | ||
| package org.neo4j.coreedge.raft.state; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| import org.neo4j.storageengine.api.ReadPastEndException; | ||
| import org.neo4j.storageengine.api.ReadableChannel; | ||
| import org.neo4j.storageengine.api.WritableChannel; | ||
|
|
||
| public class LastAppliedState | ||
| { | ||
| private final long lastApplied; | ||
|
|
||
| public LastAppliedState( long lastApplied ) | ||
| { | ||
| this.lastApplied = lastApplied; | ||
| } | ||
|
|
||
| public long get() | ||
| { | ||
| return lastApplied; | ||
| } | ||
|
|
||
| public static class Marshal implements StateMarshal<LastAppliedState> | ||
| { | ||
| @Override | ||
| public LastAppliedState startState() | ||
| { | ||
| return new LastAppliedState( -1 ); | ||
| } | ||
|
|
||
| @Override | ||
| public long ordinal( LastAppliedState lastAppliedState ) | ||
| { | ||
| return lastAppliedState.get(); | ||
| } | ||
|
|
||
| @Override | ||
| public void marshal( LastAppliedState lastAppliedState, WritableChannel channel ) throws IOException | ||
| { | ||
| channel.putLong( lastAppliedState.get() ); | ||
| } | ||
|
|
||
| @Override | ||
| public LastAppliedState unmarshal( ReadableChannel source ) throws IOException | ||
| { | ||
| try | ||
| { | ||
| return new LastAppliedState( source.getLong() ); | ||
| } | ||
| catch( ReadPastEndException e ) | ||
| { | ||
| return null; | ||
| } | ||
| } | ||
| } | ||
| } |
124 changes: 124 additions & 0 deletions
124
enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/state/StateMachineApplier.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| /* | ||
| * Copyright (c) 2002-2016 "Neo Technology," | ||
| * Network Engine for Objects in Lund AB [http://neotechnology.com] | ||
| * | ||
| * This file is part of Neo4j. | ||
| * | ||
| * Neo4j is free software: you can redistribute it and/or modify | ||
| * it under the terms of the GNU Affero General Public License as | ||
| * published by the Free Software Foundation, either version 3 of the | ||
| * License, or (at your option) any later version. | ||
| * | ||
| * This program is distributed in the hope that it will be useful, | ||
| * but WITHOUT ANY WARRANTY; without even the implied warranty of | ||
| * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||
| * GNU Affero General Public License for more details. | ||
| * | ||
| * You should have received a copy of the GNU Affero General Public License | ||
| * along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
| */ | ||
| package org.neo4j.coreedge.raft.state; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.concurrent.Executor; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.function.Supplier; | ||
|
|
||
| import org.neo4j.coreedge.raft.ConsensusListener; | ||
| import org.neo4j.coreedge.raft.log.RaftLogEntry; | ||
| import org.neo4j.coreedge.raft.log.RaftStorageException; | ||
| import org.neo4j.coreedge.raft.log.ReadableRaftLog; | ||
| import org.neo4j.kernel.internal.DatabaseHealth; | ||
| import org.neo4j.kernel.lifecycle.LifecycleAdapter; | ||
| import org.neo4j.logging.Log; | ||
| import org.neo4j.logging.LogProvider; | ||
|
|
||
| import static java.lang.System.currentTimeMillis; | ||
|
|
||
| public class StateMachineApplier extends LifecycleAdapter implements ConsensusListener | ||
| { | ||
| public static final long NOTHING_APPLIED = -1; | ||
|
|
||
| private final StateMachine stateMachine; | ||
| private final ReadableRaftLog raftLog; | ||
| private final StateStorage<LastAppliedState> lastAppliedStorage; | ||
| private final int flushEvery; | ||
| private final Supplier<DatabaseHealth> dbHealth; | ||
| private final Log log; | ||
| private long lastApplied = NOTHING_APPLIED; | ||
|
|
||
| private Executor executor; | ||
|
|
||
| private long commitIndex = NOTHING_APPLIED; | ||
|
|
||
| public StateMachineApplier( | ||
| StateMachine stateMachine, | ||
| ReadableRaftLog raftLog, | ||
| StateStorage<LastAppliedState> lastAppliedStorage, | ||
| Executor executor, | ||
| int flushEvery, | ||
| Supplier<DatabaseHealth> dbHealth, | ||
| LogProvider logProvider ) | ||
| { | ||
| this.stateMachine = stateMachine; | ||
| this.raftLog = raftLog; | ||
| this.lastAppliedStorage = lastAppliedStorage; | ||
| this.flushEvery = flushEvery; | ||
| this.log = logProvider.getLog( getClass() ); | ||
| this.dbHealth = dbHealth; | ||
| this.executor = executor; | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void notifyCommitted() | ||
| { | ||
| long commitIndex = raftLog.commitIndex(); | ||
| if ( this.commitIndex != commitIndex ) | ||
| { | ||
| this.commitIndex = commitIndex; | ||
| executor.execute( () -> { | ||
|
|
||
| try | ||
| { | ||
| applyUpTo( commitIndex ); | ||
| } | ||
| catch ( Exception e ) | ||
| { | ||
| log.error( "Failed to apply up to index " + commitIndex, e ); | ||
| dbHealth.get().panic( e ); | ||
| } | ||
| } ); | ||
| } | ||
| } | ||
|
|
||
| private void applyUpTo( long commitIndex ) throws IOException, RaftStorageException | ||
| { | ||
|
|
||
| while ( lastApplied < commitIndex ) | ||
| { | ||
| long indexToApply = lastApplied + 1; | ||
|
|
||
| RaftLogEntry logEntry = raftLog.readLogEntry( indexToApply ); | ||
| stateMachine.applyCommand( logEntry.content(), indexToApply ); | ||
|
|
||
| lastApplied = indexToApply; | ||
|
|
||
| if ( indexToApply % this.flushEvery == 0 ) | ||
| { | ||
| stateMachine.flush(); | ||
| lastAppliedStorage.persistStoreData( new LastAppliedState( lastApplied ) ); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void start() throws IOException, RaftStorageException | ||
| { | ||
| lastApplied = lastAppliedStorage.getInitialState().get(); | ||
| log.info( "Replaying commands from index %d to index %d", lastApplied, raftLog.commitIndex() ); | ||
|
|
||
| long start = currentTimeMillis(); | ||
| applyUpTo( raftLog.commitIndex() ); | ||
| log.info( "Replay done, took %d ms", currentTimeMillis() - start ); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.