Batch tx applications when notified of new committed entries in raft

Instead of applying one request at the time now we try to groip requests in batches and apply them together if possible. Note that the code will make sure to not mix application of tx with other requests since it could be unsafe.
neo4j · Jun 7, 2016 · 70a7a7f · 70a7a7f
1 parent bf9db07
commit 70a7a7f
Show file tree

Hide file tree

Showing 24 changed files with 523 additions and 216 deletions.
diff --git a/community/kernel/src/main/java/org/neo4j/kernel/impl/api/TransactionQueue.java b/community/kernel/src/main/java/org/neo4j/kernel/impl/api/TransactionQueue.java
@@ -43,16 +43,16 @@ public TransactionQueue( int maxSize, Applier applier )
         this.applier = applier;
     }
 
-    public void queue( TransactionToApply transacion ) throws Exception
+    public void queue( TransactionToApply transaction ) throws Exception
     {
         if ( size == 0 )
         {
-            first = last = transacion;
+            first = last = transaction;
         }
         else
         {
-            last.next( transacion );
-            last = transacion;
+            last.next( transaction );
+            last = transaction;
         }
 
         if ( ++size == maxSize )

diff --git a/community/kernel/src/main/java/org/neo4j/kernel/impl/api/TransactionToApply.java b/community/kernel/src/main/java/org/neo4j/kernel/impl/api/TransactionToApply.java
@@ -20,6 +20,7 @@
 package org.neo4j.kernel.impl.api;
 
 import java.io.IOException;
+import java.util.function.Consumer;
 
 import org.neo4j.helpers.collection.Visitor;
 import org.neo4j.kernel.impl.transaction.TransactionRepresentation;
@@ -66,6 +67,8 @@ public class TransactionToApply implements CommandsToApply
     // These fields are provided by commit process/storage engine
     private Commitment commitment;
 
+    private Consumer<Long> callback;
+
     /**
      * Used when committing a transaction that hasn't already gotten a transaction id assigned.
      */
@@ -123,11 +126,20 @@ public void commitment( Commitment commitment, long transactionId )
     {
         this.commitment = commitment;
         this.transactionId = transactionId;
+        if ( callback != null )
+        {
+            commitment.onClosed( callback );
+        }
     }
 
     @Override
     public TransactionToApply next()
     {
         return nextTransactionInBatch;
     }
+
+    public void onClose( Consumer<Long> callback )
+    {
+        this.callback = callback;
+    }
 }
diff --git a/community/kernel/src/main/java/org/neo4j/kernel/impl/transaction/log/Commitment.java b/community/kernel/src/main/java/org/neo4j/kernel/impl/transaction/log/Commitment.java
@@ -19,6 +19,8 @@
  */
 package org.neo4j.kernel.impl.transaction.log;
 
+import java.util.function.Consumer;
+
 import org.neo4j.kernel.impl.api.TransactionCommitProcess;
 import org.neo4j.kernel.impl.api.TransactionToApply;
 import org.neo4j.kernel.impl.transaction.tracing.LogAppendEvent;
@@ -54,6 +56,12 @@ public boolean hasLegacyIndexChanges()
         {
             return false;
         }
+
+        @Override
+        public void onClosed( Consumer<Long> callback )
+        {
+            throw new UnsupportedOperationException();
+        }
     };
 
     /**
@@ -75,4 +83,10 @@ public boolean hasLegacyIndexChanges()
      * @return whether or not this transaction contains legacy index changes.
      */
     boolean hasLegacyIndexChanges();
+
+    /**
+     * Register a callback to be triggered when the commitment has been published as closed
+     * @param callback The callback to be invoked when the commitment is published as closed
+     */
+    void onClosed( Consumer<Long> callback );
 }
diff --git a/...ity/kernel/src/main/java/org/neo4j/kernel/impl/transaction/log/TransactionCommitment.java b/...ity/kernel/src/main/java/org/neo4j/kernel/impl/transaction/log/TransactionCommitment.java
@@ -19,6 +19,8 @@
  */
 package org.neo4j.kernel.impl.transaction.log;
 
+import java.util.function.Consumer;
+
 class TransactionCommitment implements Commitment
 {
     private final boolean hasLegacyIndexChanges;
@@ -27,6 +29,7 @@ class TransactionCommitment implements Commitment
     private final LogPosition logPosition;
     private final TransactionIdStore transactionIdStore;
     private boolean markedAsCommitted;
+    private Consumer<Long> callback;
 
     TransactionCommitment( boolean hasLegacyIndexChanges, long transactionId, long transactionChecksum,
             LogPosition logPosition, TransactionIdStore transactionIdStore )
@@ -50,6 +53,10 @@ public void publishAsClosed()
     {
         transactionIdStore.transactionClosed( transactionId,
                 logPosition.getLogVersion(), logPosition.getByteOffset() );
+        if ( callback != null )
+        {
+            callback.accept( transactionId );
+        }
     }
 
     @Override
@@ -63,4 +70,10 @@ public boolean hasLegacyIndexChanges()
     {
         return hasLegacyIndexChanges;
     }
+
+    @Override
+    public void onClosed( Consumer<Long> callback )
+    {
+        this.callback = callback;
+    }
 }
diff --git a/community/kernel/src/test/java/org/neo4j/graphdb/SchemaIndexWaitingAcceptanceTest.java b/community/kernel/src/test/java/org/neo4j/graphdb/SchemaIndexWaitingAcceptanceTest.java
@@ -50,8 +50,7 @@ public class SchemaIndexWaitingAcceptanceTest
         protected void configure( GraphDatabaseFactory databaseFactory )
         {
             List<KernelExtensionFactory<?>> extensions;
-            extensions = Collections.<KernelExtensionFactory<?>>singletonList( singleInstanceSchemaIndexProviderFactory(
-                    "test", provider ) );
+            extensions = Collections.<KernelExtensionFactory<?>>singletonList( singleInstanceSchemaIndexProviderFactory( "test", provider ) );
             ((TestGraphDatabaseFactory) databaseFactory).addKernelExtensions( extensions );
         }
     };

diff --git a/community/kernel/src/test/java/org/neo4j/kernel/impl/transaction/log/FakeCommitment.java b/community/kernel/src/test/java/org/neo4j/kernel/impl/transaction/log/FakeCommitment.java
@@ -19,13 +19,16 @@
  */
 package org.neo4j.kernel.impl.transaction.log;
 
+import java.util.function.Consumer;
+
 public class FakeCommitment implements Commitment
 {
     public static final int CHECKSUM = 3;
     private final long id;
     private final TransactionIdStore transactionIdStore;
     private boolean committed;
     private boolean hasLegacyIndexChanges = false;
+    private Consumer<Long> callback;
 
     public FakeCommitment( long id, TransactionIdStore transactionIdStore )
     {
@@ -50,6 +53,10 @@ public void publishAsCommitted()
     public void publishAsClosed()
     {
         transactionIdStore.transactionClosed( id, 1, 2 );
+        if ( callback != null )
+        {
+            callback.accept( id );
+        }
     }
 
     @Override
@@ -68,4 +75,10 @@ public boolean hasLegacyIndexChanges()
     {
         return hasLegacyIndexChanges;
     }
+
+    @Override
+    public void onClosed( Consumer<Long> callback )
+    {
+        this.callback = callback;
+    }
 }
diff --git a/community/kernel/src/test/java/org/neo4j/kernel/impl/transaction/log/StubbedCommitment.java b/community/kernel/src/test/java/org/neo4j/kernel/impl/transaction/log/StubbedCommitment.java
diff --git a/...ise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/DistributedOperation.java b/...ise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/DistributedOperation.java
@@ -34,7 +34,7 @@
 /**
  * A uniquely identifiable operation.
  */
-public class DistributedOperation implements ReplicatedContent
+public class  DistributedOperation implements ReplicatedContent
 {
     private final ReplicatedContent content;
     private final GlobalSession<CoreMember> globalSession;

diff --git a/...e/src/main/java/org/neo4j/coreedge/raft/replication/id/ReplicatedIdAllocationRequest.java b/...e/src/main/java/org/neo4j/coreedge/raft/replication/id/ReplicatedIdAllocationRequest.java
@@ -19,14 +19,12 @@
  */
 package org.neo4j.coreedge.raft.replication.id;
 
-import java.util.Optional;
 import java.util.function.Consumer;
 
 import org.neo4j.coreedge.raft.replication.tx.CoreReplicatedContent;
-import org.neo4j.coreedge.raft.state.CoreStateMachines;
+import org.neo4j.coreedge.raft.state.CommandDispatcher;
 import org.neo4j.coreedge.raft.state.Result;
 import org.neo4j.coreedge.server.CoreMember;
-import org.neo4j.coreedge.raft.replication.ReplicatedContent;
 import org.neo4j.kernel.impl.store.id.IdType;
 
 import static java.lang.String.format;
@@ -105,8 +103,8 @@ public String toString()
     }
 
     @Override
-    public void dispatch( CoreStateMachines coreStateMachines, long commandIndex, Consumer<Result> callback )
+    public void dispatch( CommandDispatcher commandDispatcher, long commandIndex, Consumer<Result> callback )
     {
-        coreStateMachines.dispatch( this, commandIndex, callback );
+        commandDispatcher.dispatch( this, commandIndex, callback );
     }
 }
diff --git a/...-edge/src/main/java/org/neo4j/coreedge/raft/replication/token/ReplicatedTokenRequest.java b/...-edge/src/main/java/org/neo4j/coreedge/raft/replication/token/ReplicatedTokenRequest.java
@@ -20,12 +20,10 @@
 package org.neo4j.coreedge.raft.replication.token;
 
 import java.util.Arrays;
-import java.util.Optional;
 import java.util.function.Consumer;
 
-import org.neo4j.coreedge.raft.replication.ReplicatedContent;
 import org.neo4j.coreedge.raft.replication.tx.CoreReplicatedContent;
-import org.neo4j.coreedge.raft.state.CoreStateMachines;
+import org.neo4j.coreedge.raft.state.CommandDispatcher;
 import org.neo4j.coreedge.raft.state.Result;
 
 public class ReplicatedTokenRequest implements CoreReplicatedContent
@@ -91,8 +89,8 @@ public String toString()
     }
 
     @Override
-    public void dispatch( CoreStateMachines coreStateMachines, long commandIndex, Consumer<Result> callback )
+    public void dispatch( CommandDispatcher commandDispatcher, long commandIndex, Consumer<Result> callback )
     {
-        coreStateMachines.dispatch( this, commandIndex, callback );
+        commandDispatcher.dispatch( this, commandIndex, callback );
     }
 }
diff --git a/...core-edge/src/main/java/org/neo4j/coreedge/raft/replication/tx/CoreReplicatedContent.java b/...core-edge/src/main/java/org/neo4j/coreedge/raft/replication/tx/CoreReplicatedContent.java
@@ -22,10 +22,10 @@
 import java.util.function.Consumer;
 
 import org.neo4j.coreedge.raft.replication.ReplicatedContent;
-import org.neo4j.coreedge.raft.state.CoreStateMachines;
+import org.neo4j.coreedge.raft.state.CommandDispatcher;
 import org.neo4j.coreedge.raft.state.Result;
 
 public interface CoreReplicatedContent extends ReplicatedContent
 {
-    void dispatch( CoreStateMachines coreStateMachines, long commandIndex, Consumer<Result> callback );
+    void dispatch( CommandDispatcher commandDispatcher, long commandIndex, Consumer<Result> callback );
 }
diff --git a/...core-edge/src/main/java/org/neo4j/coreedge/raft/replication/tx/ReplicatedTransaction.java b/...core-edge/src/main/java/org/neo4j/coreedge/raft/replication/tx/ReplicatedTransaction.java
@@ -20,10 +20,9 @@
 package org.neo4j.coreedge.raft.replication.tx;
 
 import java.util.Arrays;
-import java.util.Optional;
 import java.util.function.Consumer;
 
-import org.neo4j.coreedge.raft.state.CoreStateMachines;
+import org.neo4j.coreedge.raft.state.CommandDispatcher;
 import org.neo4j.coreedge.raft.state.Result;
 
 public class ReplicatedTransaction implements CoreReplicatedContent
@@ -41,9 +40,9 @@ public byte[] getTxBytes()
     }
 
     @Override
-    public void dispatch( CoreStateMachines coreStateMachines, long commandIndex, Consumer<Result> callback )
+    public void dispatch( CommandDispatcher commandDispatcher, long commandIndex, Consumer<Result> callback )
     {
-        coreStateMachines.dispatch( this, commandIndex, callback );
+        commandDispatcher.dispatch( this, commandIndex, callback );
     }
 
     @Override