Introduce a way to close cleanly raft log segments

Before these changes the store channels, reader pools, and writers
were abruptaly closed on db shutdown.

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/log/ReadableRaftLog.java

@@ -33,11 +33,6 @@ public interface ReadableRaftLog
      */
     long prevIndex();
 
-//    /**
-//     * @return The index of the last committed entry.
-//     */
-//    long commitIndex();
-
     /**
      * Reads the term associated with the entry at the supplied index.
      *

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/log/physical/PhysicalRaftLogFile.java

@@ -95,8 +95,7 @@ public PhysicalRaftLogFile( FileSystemAbstraction fileSystem, PhysicalRaftLogFil
     public void init() throws IOException
     {
         // Make sure at least a bare bones log file is available before recovery
-        channel = openLogChannelForVersion( logFiles.getHighestLogVersion() );
-        channel.close();
+        openLogChannelForVersion( logFiles.getHighestLogVersion() ).close();
     }
 
     @Override

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/log/segmented/EntryStore.java

@@ -32,7 +32,7 @@
  * segment to the next in a transparent manner. It can thus be mainly viewed as a factory for a
  * smart segment-crossing cursor.
  */
-class EntryStore
+class EntryStore implements AutoCloseable
 {
     private Segments segments;
 
@@ -123,4 +123,10 @@ public EntryRecord get()
             }
         };
     }
+
+    @Override
+    public void close() throws DisposedException
+    {
+        segments.close();
+    }
 }

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/log/segmented/SegmentFile.java

@@ -38,6 +38,7 @@
 import org.neo4j.storageengine.api.ReadPastEndException;
 import org.neo4j.storageengine.api.WritableChannel;
 
+import static java.lang.String.format;
 import static org.neo4j.coreedge.raft.log.EntryRecord.read;
 
 /**
@@ -49,8 +50,9 @@
  *
  * Concurrent reading is thread-safe.
  */
-class SegmentFile
+class SegmentFile implements AutoCloseable
 {
+    static final String CLOSED_ERROR_MESSAGE = "segment file '%s' is closed";
     private static final SegmentHeader.Marshal headerMarshal = new SegmentHeader.Marshal();
 
     private final Log log;
@@ -65,13 +67,10 @@ class SegmentFile
     private boolean markedForDisposal;
     private Runnable onDisposal;
     private volatile boolean isDisposed;
+    private volatile boolean closed;
 
-    SegmentFile(
-            FileSystemAbstraction fileSystem,
-            File file,
-            ChannelMarshal<ReplicatedContent> contentMarshal,
-            LogProvider logProvider,
-            SegmentHeader header )
+    SegmentFile( FileSystemAbstraction fileSystem, File file, ChannelMarshal<ReplicatedContent> contentMarshal,
+            LogProvider logProvider, SegmentHeader header )
     {
         this.fileSystem = fileSystem;
         this.file = file;
@@ -82,12 +81,9 @@ class SegmentFile
         readerPool = new StoreChannelPool( fileSystem, file, "r", logProvider );
     }
 
-    static SegmentFile create(
-            FileSystemAbstraction fileSystem,
-            File file,
-            ChannelMarshal<ReplicatedContent> contentMarshal,
-            LogProvider logProvider,
-            SegmentHeader header ) throws IOException
+    static SegmentFile create( FileSystemAbstraction fileSystem, File file,
+            ChannelMarshal<ReplicatedContent> contentMarshal, LogProvider logProvider, SegmentHeader header )
+            throws IOException
     {
         SegmentFile segment = new SegmentFile( fileSystem, file, contentMarshal, logProvider, header );
 
@@ -159,6 +155,11 @@ private LogPosition findCachedStartingPosition( long offsetIndex )
 
     private PhysicalFlushableChannel getOrCreateWriter() throws IOException
     {
+        if ( closed )
+        {
+            throw new RuntimeException( format( CLOSED_ERROR_MESSAGE, file ) );
+        }
+
         if ( bufferedWriter == null )
         {
             StoreChannel channel = fileSystem.open( file, "rw" );
@@ -173,7 +174,7 @@ private PhysicalFlushableChannel getOrCreateWriter() throws IOException
      */
     WritableChannel writer() throws IOException, DisposedException
     {
-        if( markedForDisposal )
+        if ( markedForDisposal )
         {
             throw new DisposedException();
         }
@@ -223,7 +224,7 @@ void flush() throws IOException
      */
     void markForDisposal( Runnable onDisposal ) throws DisposedException
     {
-        if( markedForDisposal )
+        if ( markedForDisposal )
         {
             throw new DisposedException();
         }
@@ -236,7 +237,7 @@ void markForDisposal( Runnable onDisposal ) throws DisposedException
 
     private synchronized void checkFullDisposal()
     {
-        if ( bufferedWriter == null && readerPool.isDisposed() )
+        if ( markedForDisposal && bufferedWriter == null && readerPool.isDisposed() )
         {
             isDisposed = true;
             onDisposal.run();
@@ -262,4 +263,17 @@ public long size()
     {
         return fileSystem.getFileSize( file );
     }
+
+    @Override
+    public void close() throws DisposedException
+    {
+        if ( closed )
+        {
+            throw new RuntimeException( format( CLOSED_ERROR_MESSAGE, file ) );
+        }
+
+        closed = true;
+        closeWriter();
+        readerPool.close();
+    }
 }

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/log/segmented/SegmentedRaftLog.java

@@ -101,6 +101,12 @@ public synchronized void start() throws IOException, RaftLogCompactedException,
         entryStore = new EntryStore( state.segments );
     }
 
+    @Override
+    public void shutdown() throws DisposedException
+    {
+        entryStore.close();
+    }
+
     @Override
     public synchronized long append( RaftLogEntry entry ) throws IOException
     {

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/log/segmented/Segments.java

@@ -41,7 +41,7 @@
 /**
  * Keeps track of all the segments that the RAFT log consists of.
  */
-class Segments
+class Segments implements AutoCloseable
 {
     private final OpenEndRangeMap<Long/*minIndex*/,SegmentFile> rangeMap = new OpenEndRangeMap<>();
     private final List<SegmentFile> segmentFiles;
@@ -248,4 +248,33 @@ public ListIterator<SegmentFile> getSegmentFileIteratorAtEnd()
     {
         return segmentFiles.listIterator( segmentFiles.size() );
     }
+
+    @Override
+    public void close() throws DisposedException
+    {
+        RuntimeException error = null;
+        for ( SegmentFile segment : segmentFiles )
+        {
+            try
+            {
+                segment.close();
+            }
+            catch ( RuntimeException ex )
+            {
+                if ( error == null )
+                {
+                    error = ex;
+                }
+                else
+                {
+                    error.addSuppressed( ex );
+                }
+            }
+        }
+
+        if ( error != null )
+        {
+            throw error;
+        }
+    }
 }

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/log/segmented/StoreChannelPool.java

@@ -32,8 +32,10 @@
 /**
  * Keeps a pool of store channels available.
  */
-class StoreChannelPool
+class StoreChannelPool implements AutoCloseable
 {
+    static final String CLOSED_ERROR_MESSAGE = "store channel pool is closed";
+
     private final FileSystemAbstraction fsa;
     private final File file;
     private final String mode;
@@ -44,13 +46,13 @@ class StoreChannelPool
     private boolean markedForDisposal;
     private Runnable onDisposal;
     private boolean disposed;
+    private boolean closed;
 
     StoreChannelPool( FileSystemAbstraction fsa, File file, String mode, LogProvider logProvider )
     {
         this.fsa = fsa;
         this.file = file;
         this.mode = mode;
-
         this.log = logProvider.getLog( getClass() );
     }
 
@@ -62,7 +64,11 @@ private StoreChannel create() throws IOException
 
     synchronized StoreChannel acquire( long byteOffset ) throws IOException, DisposedException
     {
-        if( markedForDisposal )
+        if ( closed )
+        {
+            throw new RuntimeException( CLOSED_ERROR_MESSAGE );
+        }
+        if ( markedForDisposal )
         {
             throw new DisposedException();
         }
@@ -78,6 +84,10 @@ synchronized StoreChannel acquire( long byteOffset ) throws IOException, Dispose
 
     synchronized void release( StoreChannel channel )
     {
+        if ( closed )
+        {
+            throw new RuntimeException( CLOSED_ERROR_MESSAGE );
+        }
         channels.addFirst( channel );
         checkForDisposal();
     }
@@ -109,7 +119,7 @@ private void closeAllChannels()
 
     synchronized void markForDisposal( Runnable onDisposal ) throws DisposedException
     {
-        if( markedForDisposal )
+        if ( markedForDisposal )
         {
             throw new DisposedException();
         }
@@ -124,4 +134,15 @@ boolean isDisposed()
     {
         return disposed;
     }
+
+    @Override
+    public synchronized void close() throws DisposedException
+    {
+        if ( closed )
+        {
+            throw new RuntimeException( CLOSED_ERROR_MESSAGE );
+        }
+        closed = true;
+        closeAllChannels();
+    }
 }

enterprise/core-edge/src/test/java/org/neo4j/coreedge/raft/log/segmented/EntryStoreTest.java

@@ -0,0 +1,36 @@
+/*
+ * 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.log.segmented;
+
+import org.junit.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+public class EntryStoreTest
+{
+    @Test
+    public void segmentsShouldBeClosedWhenEntryStoreIsClosed() throws Exception
+    {
+        Segments segments = mock( Segments.class );
+        new EntryStore( segments ).close();
+        verify( segments ).close();
+    }
+}