StoreCopyServer now report hasData correctly depending on file size

Add fileSize() to PagedFile interface
neo4j · Sep 16, 2016 · 5f00d4e · 5f00d4e
1 parent 667a9b4
commit 5f00d4e
Show file tree

Hide file tree

Showing 7 changed files with 64 additions and 11 deletions.
diff --git a/community/io/src/main/java/org/neo4j/io/pagecache/PagedFile.java b/community/io/src/main/java/org/neo4j/io/pagecache/PagedFile.java
@@ -130,6 +130,11 @@ public interface PagedFile extends AutoCloseable
      */
     int pageSize();
 
+    /**
+     * Size of file, in bytes.
+     */
+    long fileSize() throws IOException;
+
     /**
      * Flush all dirty pages into the file channel, and force the file channel to disk.
      */

diff --git a/community/io/src/main/java/org/neo4j/io/pagecache/impl/muninn/MuninnPagedFile.java b/community/io/src/main/java/org/neo4j/io/pagecache/impl/muninn/MuninnPagedFile.java
@@ -172,6 +172,17 @@ public int pageSize()
         return filePageSize;
     }
 
+    @Override
+    public long fileSize()
+    {
+        final long lastPageId = getLastPageId();
+        if ( lastPageId < 0 )
+        {
+            return 0L;
+        }
+        return (lastPageId + 1) * pageSize();
+    }
+
     File file()
     {
         return swapper.file();

diff --git a/community/io/src/test/java/org/neo4j/adversaries/pagecache/AdversarialPagedFile.java b/community/io/src/test/java/org/neo4j/adversaries/pagecache/AdversarialPagedFile.java
@@ -69,6 +69,13 @@ public int pageSize()
         return delegate.pageSize();
     }
 
+    @Override
+    public long fileSize() throws IOException
+    {
+        adversary.injectFailure( IllegalStateException.class );
+        return delegate.fileSize();
+    }
+
     @Override
     public void flushAndForce() throws IOException
     {

diff --git a/community/io/src/test/java/org/neo4j/io/pagecache/DelegatingPagedFile.java b/community/io/src/test/java/org/neo4j/io/pagecache/DelegatingPagedFile.java
@@ -52,6 +52,12 @@ public int pageSize()
         return delegate.pageSize();
     }
 
+    @Override
+    public long fileSize() throws IOException
+    {
+        return delegate.fileSize();
+    }
+
     public void close() throws IOException
     {
         delegate.close();

diff --git a/community/io/src/test/java/org/neo4j/io/pagecache/StubPagedFile.java b/community/io/src/test/java/org/neo4j/io/pagecache/StubPagedFile.java
@@ -53,6 +53,17 @@ public int pageSize()
         return exposedPageSize;
     }
 
+    @Override
+    public long fileSize() throws IOException
+    {
+        final long lastPageId = getLastPageId();
+        if ( lastPageId < 0 )
+        {
+            return 0L;
+        }
+        return (lastPageId + 1) * pageSize();
+    }
+
     @Override
     public void flushAndForce() throws IOException
     {

diff --git a/enterprise/com/src/main/java/org/neo4j/com/storecopy/StoreCopyServer.java b/enterprise/com/src/main/java/org/neo4j/com/storecopy/StoreCopyServer.java
@@ -165,20 +165,25 @@ public RequestContext flushStoresAndStreamStoreFiles( String triggerName, StoreW
 
                     // Read from paged file if mapping exists. Otherwise read through file system.
                     final Optional<PagedFile> optionalPagedFile = pageCache.getExistingMapping( file );
-                    try ( ReadableByteChannel fileChannel = optionalPagedFile.isPresent() ?
-                                                            optionalPagedFile.get().openReadableByteChannel() :
-                                                            fileSystem.open( file, "r" ) )
+                    if ( optionalPagedFile.isPresent() )
                     {
-                        monitor.startStreamingStoreFile( file );
-                        writer.write( relativePath( storeDirectory, file ), fileChannel,
-                                temporaryBuffer, file.length() > 0, recordSize );
-                        monitor.finishStreamingStoreFile( file );
+                        PagedFile pagedFile = optionalPagedFile.get();
+                        long fileSize = pagedFile.fileSize();
+                        try ( ReadableByteChannel fileChannel = pagedFile.openReadableByteChannel() )
+                        {
+                            doWrite( writer, temporaryBuffer, file, recordSize, fileChannel, fileSize );
+                        }
+                        finally
+                        {
+                            pagedFile.close();
+                        }
                     }
-                    finally
+                    else
                     {
-                        if ( optionalPagedFile.isPresent() )
+                        try ( ReadableByteChannel fileChannel = fileSystem.open( file, "r" ) )
                         {
-                            optionalPagedFile.get().close();
+                            long fileSize = file.length();
+                            doWrite( writer, temporaryBuffer, file, recordSize, fileChannel, fileSize );
                         }
                     }
                 }
@@ -195,4 +200,13 @@ public RequestContext flushStoresAndStreamStoreFiles( String triggerName, StoreW
             throw new ServerFailureException( e );
         }
     }
+
+    private void doWrite( StoreWriter writer, ByteBuffer temporaryBuffer, File file, int recordSize,
+            ReadableByteChannel fileChannel, long fileSize ) throws IOException
+    {
+        monitor.startStreamingStoreFile( file );
+        writer.write( relativePath( storeDirectory, file ), fileChannel,
+                temporaryBuffer, fileSize > 0, recordSize );
+        monitor.finishStreamingStoreFile( file );
+    }
 }
diff --git a/enterprise/com/src/main/java/org/neo4j/com/storecopy/ToNetworkStoreWriter.java b/enterprise/com/src/main/java/org/neo4j/com/storecopy/ToNetworkStoreWriter.java
@@ -60,7 +60,6 @@ public long write( String path, ReadableByteChannel data, ByteBuffer temporaryBu
             totalWritten += 4;
             totalWritten += buffer.write( data );
             buffer.close();
-
         }
         return totalWritten;
     }