Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Introduce a better replicator abstraction and state machine model.
This commit introduces a replicator abstraction for RAFT that internally manages tracking of operations and results together with a progress tracker. This then allows the state machines and service implementations to be greatly streamlined and simplified. All replicated content that flows through the RaftReplicator now gets associated with a unique session/operationId pair as captured by the DistributedOperation class. It also carries the command that will be dispatched to the associated state machine. All state machines now implement a simple StateMachine interface that is aligned with Lamport's consensus state machine model. They handle commands, carry their state and produce results. The individual state machines are collected in the CoreStateMachines class which is handled by CoreState. This class now owns the session tracking which lives one step above, on the replication level, and not as before at the state machine level.
- Loading branch information
1 parent
367f026
commit f5ed09b
Showing
74 changed files
with
2,105 additions
and
2,578 deletions.
There are no files selected for viewing
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
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
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
128 changes: 128 additions & 0 deletions
128
...ise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/DistributedOperation.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,128 @@ | ||
| /* | ||
| * 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.replication; | ||
|
|
||
| import io.netty.buffer.ByteBuf; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.UUID; | ||
|
|
||
| import org.neo4j.coreedge.raft.net.CoreReplicatedContentMarshal; | ||
| import org.neo4j.coreedge.raft.replication.session.GlobalSession; | ||
| import org.neo4j.coreedge.raft.replication.session.LocalOperationId; | ||
| import org.neo4j.coreedge.server.CoreMember; | ||
| import org.neo4j.storageengine.api.ReadableChannel; | ||
| import org.neo4j.storageengine.api.WritableChannel; | ||
|
|
||
| /** | ||
| * A uniquely identifiable operation. | ||
| */ | ||
| public class DistributedOperation implements ReplicatedContent | ||
| { | ||
| private final ReplicatedContent content; | ||
| private final GlobalSession<CoreMember> globalSession; | ||
| private final LocalOperationId operationId; | ||
|
|
||
| public DistributedOperation( ReplicatedContent content, GlobalSession<CoreMember> globalSession, LocalOperationId operationId ) | ||
| { | ||
| this.content = content; | ||
| this.globalSession = globalSession; | ||
| this.operationId = operationId; | ||
| } | ||
|
|
||
| public GlobalSession<CoreMember> globalSession() | ||
| { | ||
| return globalSession; | ||
| } | ||
|
|
||
| public LocalOperationId operationId() | ||
| { | ||
| return operationId; | ||
| } | ||
|
|
||
| public ReplicatedContent content() | ||
| { | ||
| return content; | ||
| } | ||
|
|
||
| public void serialize( WritableChannel channel ) throws IOException | ||
| { | ||
| channel.putLong( globalSession().sessionId().getMostSignificantBits() ); | ||
| channel.putLong( globalSession().sessionId().getLeastSignificantBits() ); | ||
| new CoreMember.CoreMemberMarshal().marshal( globalSession().owner(), channel ); | ||
|
|
||
| channel.putLong( operationId.localSessionId() ); | ||
| channel.putLong( operationId.sequenceNumber() ); | ||
|
|
||
| new CoreReplicatedContentMarshal().marshal( content, channel ); | ||
| } | ||
|
|
||
| public static DistributedOperation deserialize( ReadableChannel channel ) throws IOException | ||
| { | ||
| long mostSigBits = channel.getLong(); | ||
| long leastSigBits = channel.getLong(); | ||
| CoreMember owner = new CoreMember.CoreMemberMarshal().unmarshal( channel ); | ||
| GlobalSession<CoreMember> globalSession = new GlobalSession<>( new UUID( mostSigBits, leastSigBits ), owner ); | ||
|
|
||
| long localSessionId = channel.getLong(); | ||
| long sequenceNumber = channel.getLong(); | ||
| LocalOperationId localOperationId = new LocalOperationId( localSessionId, sequenceNumber ); | ||
|
|
||
| ReplicatedContent content = new CoreReplicatedContentMarshal().unmarshal( channel ); | ||
| return new DistributedOperation( content, globalSession, localOperationId ); | ||
| } | ||
|
|
||
| public void serialize( ByteBuf buffer ) | ||
| { | ||
| buffer.writeLong( globalSession().sessionId().getMostSignificantBits() ); | ||
| buffer.writeLong( globalSession().sessionId().getLeastSignificantBits() ); | ||
| new CoreMember.CoreMemberMarshal().marshal( (CoreMember) globalSession().owner(), buffer ); | ||
|
|
||
| buffer.writeLong( operationId.localSessionId() ); | ||
| buffer.writeLong( operationId.sequenceNumber() ); | ||
|
|
||
| new CoreReplicatedContentMarshal().marshal( content, buffer ); | ||
| } | ||
|
|
||
| public static ReplicatedContent deserialize( ByteBuf buffer ) | ||
| { | ||
| long mostSigBits = buffer.readLong(); | ||
| long leastSigBits =buffer.readLong(); | ||
| CoreMember owner = new CoreMember.CoreMemberMarshal().unmarshal( buffer ); | ||
| GlobalSession<CoreMember> globalSession = new GlobalSession<>( new UUID( mostSigBits, leastSigBits ), owner ); | ||
|
|
||
| long localSessionId = buffer.readLong(); | ||
| long sequenceNumber = buffer.readLong(); | ||
| LocalOperationId localOperationId = new LocalOperationId( localSessionId, sequenceNumber ); | ||
|
|
||
| ReplicatedContent content = new CoreReplicatedContentMarshal().unmarshal( buffer ); | ||
| return new DistributedOperation( content, globalSession, localOperationId ); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() | ||
| { | ||
| return "DistributedOperation{" + | ||
| "content=" + content + | ||
| ", globalSession=" + globalSession + | ||
| ", operationId=" + operationId + | ||
| '}'; | ||
| } | ||
| } |
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.