Merge pull request #1548 from katzyn/async

Add AsynchronousFileChannel-based experimental FilePathAsync

h2/src/docsrc/html/advanced.html

@@ -1550,6 +1550,7 @@ The following file systems are included:
 </li><li><code>nioMapped:</code> file system that uses memory mapped files (faster in some operating systems).
     Please note that there currently is a file size limitation of 2 GB when using this file system.
     To work around this limitation, combine it with the split file system: <code>split:nioMapped:test</code>.
+</li><li><code>async:</code> experimental file system that uses <code>AsynchronousFileChannel</code> instead of <code>RandomAccessFile</code> (faster in some operating systems).
 </li><li><code>memFS:</code> in-memory file system (slower than mem; experimental; mainly used for testing the database engine itself).
 </li><li><code>memLZF:</code> compressing in-memory file system (slower than memFS but uses less memory; experimental; mainly used for testing the database engine itself).
 </li><li><code>nioMemFS:</code> stores data outside of the VM's heap - useful for large memory DBs without incurring GC costs.
@@ -1563,8 +1564,9 @@ The following file systems are included:
     The default value is 1%.
 </li></ul>
 <p>
-As an example, to use the the <code>nio</code> file system, use the following database URL:
-<code>jdbc:h2:nio:~/test</code>.
+As an example, to use the the <code>nio</code> file system with PageStore storage engine,
+use the following database URL: <code>jdbc:h2:nio:~/test;MV_STORE=FALSE</code>.
+With MVStore storage engine nio file system is used by default.
 </p>
 <p>
 To register a new file system, extend the classes <code>org.h2.store.fs.FilePath, FileBase</code>,

h2/src/main/org/h2/store/fs/FilePath.java

@@ -74,6 +74,7 @@ public abstract class FilePath {
                     "org.h2.store.fs.FilePathSplit",
                     "org.h2.store.fs.FilePathNio",
                     "org.h2.store.fs.FilePathNioMapped",
+                    "org.h2.store.fs.FilePathAsync",
                     "org.h2.store.fs.FilePathZip",
                     "org.h2.store.fs.FilePathRetryOnInterrupt"
             }) {

h2/src/main/org/h2/store/fs/FilePathAsync.java

+/*
+ * Copyright 2004-2018 H2 Group. Multiple-Licensed under the MPL 2.0,
+ * and the EPL 1.0 (http://h2database.com/html/license.html).
+ * Initial Developer: H2 Group
+ */
+package org.h2.store.fs;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousFileChannel;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.channels.NonWritableChannelException;
+import java.nio.file.OpenOption;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * This file system stores files on disk and uses
+ * java.nio.channels.AsynchronousFileChannel to access the files.
+ */
+public class FilePathAsync extends FilePathWrapper {
+
+    private static final boolean AVAILABLE;
+
+    /*
+     * Android has NIO2 only since API 26.
+     */
+    static {
+        boolean a = false;
+        try {
+            AsynchronousFileChannel.class.getName();
+            a = true;
+        } catch (Throwable e) {
+            // Nothing to do
+        }
+        AVAILABLE = a;
+    }
+
+    /**
+     * Creates new instance of FilePathAsync.
+     */
+    public FilePathAsync() {
+        if (!AVAILABLE) {
+            throw new UnsupportedOperationException("NIO2 is not available");
+        }
+    }
+
+    @Override
+    public FileChannel open(String mode) throws IOException {
+        return new FileAsync(name.substring(getScheme().length() + 1), mode);
+    }
+
+    @Override
+    public String getScheme() {
+        return "async";
+    }
+
+}
+
+/**
+ * File which uses NIO2 AsynchronousFileChannel.
+ */
+class FileAsync extends FileBase {
+
+    private static final OpenOption[] R = { StandardOpenOption.READ };
+
+    private static final OpenOption[] W = { StandardOpenOption.READ, StandardOpenOption.WRITE,
+            StandardOpenOption.CREATE };
+
+    private static final OpenOption[] RWS = { StandardOpenOption.READ, StandardOpenOption.WRITE,
+            StandardOpenOption.CREATE, StandardOpenOption.SYNC };
+
+    private static final OpenOption[] RWD = { StandardOpenOption.READ, StandardOpenOption.WRITE,
+            StandardOpenOption.CREATE, StandardOpenOption.DSYNC };
+
+    private final String name;
+
+    private final AsynchronousFileChannel channel;
+
+    private long position;
+
+    private static <T> T complete(Future<T> future) throws IOException {
+        boolean interrupted = false;
+        for (;;) {
+            try {
+                T result = future.get();
+                if (interrupted) {
+                    Thread.currentThread().interrupt();
+                }
+                return result;
+            } catch (InterruptedException e) {
+                interrupted = true;
+            } catch (ExecutionException e) {
+                throw new IOException(e.getCause());
+            }
+        }
+    }
+
+    FileAsync(String fileName, String mode) throws IOException {
+        this.name = fileName;
+        OpenOption[] options;
+        switch (mode) {
+        case "r":
+            options = R;
+            break;
+        case "rw":
+            options = W;
+            break;
+        case "rws":
+            options = RWS;
+            break;
+        case "rwd":
+            options = RWD;
+            break;
+        default:
+            throw new IllegalArgumentException(mode);
+        }
+        channel = AsynchronousFileChannel.open(Paths.get(fileName), options);
+    }
+
+    @Override
+    public void implCloseChannel() throws IOException {
+        channel.close();
+    }
+
+    @Override
+    public long position() throws IOException {
+        return position;
+    }
+
+    @Override
+    public long size() throws IOException {
+        return channel.size();
+    }
+
+    @Override
+    public int read(ByteBuffer dst) throws IOException {
+        int read = complete(channel.read(dst, position));
+        if (read > 0) {
+            position += read;
+        }
+        return read;
+    }
+
+    @Override
+    public FileChannel position(long pos) throws IOException {
+        if (pos < 0) {
+            throw new IllegalArgumentException();
+        }
+        position = pos;
+        return this;
+    }
+
+    @Override
+    public int read(ByteBuffer dst, long position) throws IOException {
+        return complete(channel.read(dst, position));
+    }
+
+    @Override
+    public int write(ByteBuffer src, long position) throws IOException {
+        try {
+            return complete(channel.write(src, position));
+        } catch (NonWritableChannelException e) {
+            throw new IOException("read only");
+        }
+    }
+
+    @Override
+    public FileChannel truncate(long newLength) throws IOException {
+        channel.truncate(newLength);
+        if (newLength < position) {
+            position = newLength;
+        }
+        return this;
+    }
+
+    @Override
+    public void force(boolean metaData) throws IOException {
+        channel.force(metaData);
+    }
+
+    @Override
+    public int write(ByteBuffer src) throws IOException {
+        int written;
+        try {
+            written = complete(channel.write(src, position));
+            position += written;
+        } catch (NonWritableChannelException e) {
+            throw new IOException("read only");
+        }
+        return written;
+    }
+
+    @Override
+    public synchronized FileLock tryLock(long position, long size, boolean shared) throws IOException {
+        return channel.tryLock(position, size, shared);
+    }
+
+    @Override
+    public String toString() {
+        return "async:" + name;
+    }
+
+}

h2/src/test/org/h2/test/store/TestConcurrent.java

@@ -47,7 +47,8 @@ public class TestConcurrent extends TestMVStore {
     @Override
     public void test() throws Exception {
         FileUtils.createDirectories(getBaseDir());
-        testInterruptReopen();
+        testInterruptReopenAsync();
+        testInterruptReopenRetryNIO();
         testConcurrentSaveCompact();
         testConcurrentDataType();
         testConcurrentAutoCommitAndChange();
@@ -64,8 +65,16 @@ public class TestConcurrent extends TestMVStore {
         testConcurrentRead();
     }
 
-    private void testInterruptReopen() {
-        String fileName = "retry:nio:" + getBaseDir() + "/" + getTestName();
+    private void testInterruptReopenAsync() {
+        testInterruptReopen("async:");
+    }
+
+    private void testInterruptReopenRetryNIO() {
+        testInterruptReopen("retry:nio:");
+    }
+
+    private void testInterruptReopen(String prefix) {
+        String fileName = prefix + getBaseDir() + "/" + getTestName();
         FileUtils.delete(fileName);
         final MVStore s = new MVStore.Builder().
                 fileName(fileName).

h2/src/test/org/h2/test/unit/TestFileSystem.java

@@ -83,6 +83,7 @@ public class TestFileSystem extends TestDb {
         String f = "split:10:" + getBaseDir() + "/fs";
         FileUtils.toRealPath(f);
         testFileSystem(getBaseDir() + "/fs");
+        testFileSystem("async:" + getBaseDir() + "/fs");
         testFileSystem("memFS:");
         testFileSystem("memLZF:");
         testFileSystem("nioMemFS:");