Each BatchingPagedFile has one read, one write cursor

to be able to paralellize some more moving forward.
neo4j · Mar 3, 2015 · f0a9109 · f0a9109
1 parent b098e20
commit f0a9109
Show file tree

Hide file tree

Showing 8 changed files with 159 additions and 129 deletions.
diff --git a/...ency-check/src/test/java/org/neo4j/unsafe/impl/batchimport/ParallelBatchImporterTest.java b/...ency-check/src/test/java/org/neo4j/unsafe/impl/batchimport/ParallelBatchImporterTest.java
@@ -75,6 +75,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
+
 import static org.neo4j.function.Functions.constant;
 import static org.neo4j.helpers.collection.IteratorUtil.count;
 import static org.neo4j.unsafe.impl.batchimport.AdditionalInitialIds.EMPTY;
@@ -314,7 +315,6 @@ public void write( ByteBuffer data, long position, Pool<ByteBuffer> pool )
                         LockSupport.parkNanos( random.nextInt( 500 ) * 1_000_000 ); // slowness comes from here
                     }
                     delegate.write( data, position, pool );
-
                 }
             };
         }

diff --git a/...ernel/src/main/java/org/neo4j/unsafe/impl/batchimport/NodeFirstRelationshipProcessor.java b/...ernel/src/main/java/org/neo4j/unsafe/impl/batchimport/NodeFirstRelationshipProcessor.java
@@ -52,15 +52,13 @@ public boolean process( NodeRecord node )
     {
         long nodeId = node.getId();
         long firstRel = nodeRelationshipLink.getFirstRel( nodeId, this );
-        if ( firstRel == -1 )
+        if ( firstRel != -1 )
         {
-            return false;
+            node.setNextRel( firstRel );
-        }
+            if ( nodeRelationshipLink.isDense( nodeId ) )
-
+            {
-        node.setNextRel( firstRel );
+                node.setDense( true );
-        if ( nodeRelationshipLink.isDense( nodeId ) )
+            }
-        {
-            node.setDense( true );
         }
         return true;
     }

diff --git a/community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/ParallelBatchImporter.java b/community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/ParallelBatchImporter.java
@@ -170,7 +170,7 @@ public void doImport( Input input ) throws IOException
 
             // Switch to reverse updating mode and release references that are no longer used so they can be collected
             writerFactory.awaitEverythingWritten();
-            neoStore.switchToUpdateMode();
+            neoStore.flush();
 
             // Remaining node processors
             nodeLabelsCache = new NodeLabelsCache( AUTO, neoStore.getLabelRepository().getHighId() );

diff --git a/community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/store/BatchingNeoStore.java b/community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/store/BatchingNeoStore.java
@@ -51,8 +51,6 @@
 import static org.neo4j.graphdb.factory.GraphDatabaseSettings.dense_node_threshold;
 import static org.neo4j.helpers.collection.MapUtil.stringMap;
 import static org.neo4j.kernel.impl.store.StoreFactory.configForStoreDir;
-import static org.neo4j.unsafe.impl.batchimport.store.BatchingPageCache.Mode.APPEND_ONLY;
-import static org.neo4j.unsafe.impl.batchimport.store.BatchingPageCache.Mode.UPDATE;
 
 /**
  * Creator and accessor of {@link NeoStore} with some logic to provide very batch friendly services to the
@@ -67,7 +65,7 @@ public class BatchingNeoStore implements AutoCloseable
     private final BatchingRelationshipTypeTokenRepository relationshipTypeRepository;
     private final StringLogger logger;
     private final Config neo4jConfig;
-    private final BatchingPageCache pageCacheFactory;
+    private final BatchingPageCache pageCache;
     private final NeoStore neoStore;
     private final WriterFactory writerFactory;
 
@@ -84,9 +82,9 @@ public BatchingNeoStore( FileSystemAbstraction fileSystem, String storeDir,
                         GraphDatabaseSettings.class ),
                 new File( storeDir ) );
 
-        this.pageCacheFactory = new BatchingPageCache( fileSystem, config.fileChannelBufferSize(),
+        this.pageCache = new BatchingPageCache( fileSystem, config.fileChannelBufferSize(),
-                config.bigFileChannelBufferSizeMultiplier(), writerFactory, writeMonitor, APPEND_ONLY );
+                config.bigFileChannelBufferSizeMultiplier(), writerFactory, writeMonitor );
-        this.neoStore = newNeoStore( pageCacheFactory );
+        this.neoStore = newNeoStore( pageCache );
         flushNeoStoreAndAwaitEverythingWritten();
         if ( alreadyContainsData( neoStore ) )
         {
@@ -126,7 +124,7 @@ public static void createStore( FileSystemAbstraction fileSystem, String storeDi
     {
         PageCache pageCache = new BatchingPageCache( fileSystem, Configuration.DEFAULT.fileChannelBufferSize(),
                 Configuration.DEFAULT.bigFileChannelBufferSizeMultiplier(),
-                BatchingPageCache.SYNCHRONOUS, Monitor.NO_MONITOR, APPEND_ONLY );
+                BatchingPageCache.SYNCHRONOUS, Monitor.NO_MONITOR );
         StoreFactory storeFactory = new StoreFactory(
                 fileSystem, new File( storeDir ), pageCache, StringLogger.DEV_NULL, new Monitors() );
         storeFactory.createNeoStore().close();
@@ -180,11 +178,6 @@ public CountsTracker getCountsStore()
         return neoStore.getCounts();
     }
 
-    public void switchToUpdateMode()
-    {
-        pageCacheFactory.setMode( UPDATE );
-    }
-
     @Override
     public void close()
     {
@@ -205,4 +198,9 @@ private void flushNeoStoreAndAwaitEverythingWritten()
         // That's why we have to wait for any pending I/O jobs to be written after the flush.
         writerFactory.awaitEverythingWritten();
     }
+
+    public void flush() throws IOException
+    {
+        pageCache.flush();
+    }
 }