From 401c35ad0f8a75c9f1e474366f35a8823c608c09 Mon Sep 17 00:00:00 2001 From: Sergei Egorov Date: Wed, 3 Jun 2020 18:17:14 +0200 Subject: [PATCH 1/3] Use `AsynchronousFileChannel` for npipe --- .../httpclient5/NamedPipeSocket.java | 134 ++++++++++------ .../dockerjava/okhttp/NamedPipeSocket.java | 150 ++++++++++++++++++ .../okhttp/NamedPipeSocketFactory.java | 100 +----------- 3 files changed, 237 insertions(+), 147 deletions(-) create mode 100644 docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java diff --git a/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java b/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java index 130b5e16c..e2cc181e8 100644 --- a/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java +++ b/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java @@ -8,19 +8,23 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; -import java.io.RandomAccessFile; import java.net.Socket; import java.net.SocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.AsynchronousByteChannel; +import java.nio.channels.AsynchronousFileChannel; +import java.nio.channels.Channels; +import java.nio.channels.CompletionHandler; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Future; class NamedPipeSocket extends Socket { private final String socketFileName; - private RandomAccessFile file; - - private InputStream is; - - private OutputStream os; + private AsynchronousFileByteChannel channel; NamedPipeSocket(String socketFileName) { this.socketFileName = socketFileName; @@ -28,24 +32,29 @@ class NamedPipeSocket extends Socket { @Override public void close() throws IOException { - if (file != null) { - file.close(); - file = null; + if (channel != null) { + channel.close(); } } @Override - public void connect(SocketAddress endpoint) { + public void connect(SocketAddress endpoint) throws IOException { connect(endpoint, 0); } @Override - public void connect(SocketAddress endpoint, int timeout) { + public void connect(SocketAddress endpoint, int timeout) throws IOException { long startedAt = System.currentTimeMillis(); timeout = Math.max(timeout, 10_000); while (true) { try { - file = new RandomAccessFile(socketFileName, "rw"); + channel = new AsynchronousFileByteChannel( + AsynchronousFileChannel.open( + Paths.get(socketFileName), + StandardOpenOption.READ, + StandardOpenOption.WRITE + ) + ); break; } catch (FileNotFoundException e) { if (System.currentTimeMillis() - startedAt >= timeout) { @@ -55,50 +64,16 @@ public void connect(SocketAddress endpoint, int timeout) { } } } - - is = new InputStream() { - @Override - public int read(byte[] bytes, int off, int len) throws IOException { - return file.read(bytes, off, len); - } - - @Override - public int read() throws IOException { - return file.read(); - } - - @Override - public int read(byte[] bytes) throws IOException { - return file.read(bytes); - } - }; - - os = new OutputStream() { - @Override - public void write(byte[] bytes, int off, int len) throws IOException { - file.write(bytes, off, len); - } - - @Override - public void write(int value) throws IOException { - file.write(value); - } - - @Override - public void write(byte[] bytes) throws IOException { - file.write(bytes); - } - }; } @Override public InputStream getInputStream() { - return is; + return Channels.newInputStream(channel); } @Override public OutputStream getOutputStream() { - return os; + return Channels.newOutputStream(channel); } interface Kernel32 extends StdCallLibrary { @@ -108,4 +83,67 @@ interface Kernel32 extends StdCallLibrary { @SuppressWarnings("checkstyle:methodname") boolean WaitNamedPipe(String lpNamedPipeName, int nTimeOut); } + + private static class AsynchronousFileByteChannel implements AsynchronousByteChannel { + private final AsynchronousFileChannel fileChannel; + + AsynchronousFileByteChannel(AsynchronousFileChannel fileChannel) { + this.fileChannel = fileChannel; + } + + @Override + public void read(ByteBuffer dst, A attachment, CompletionHandler handler) { + fileChannel.read(dst, 0, attachment, new CompletionHandler() { + @Override + public void completed(Integer read, A attachment) { + handler.completed(read > 0 ? read : -1, attachment); + } + + @Override + public void failed(Throwable exc, A attachment) { + handler.failed(exc, attachment); + } + }); + } + + @Override + public Future read(ByteBuffer dst) { + CompletableFutureHandler future = new CompletableFutureHandler(); + fileChannel.read(dst, 0, null, future); + return future; + } + + @Override + public void write(ByteBuffer src, A attachment, CompletionHandler handler) { + fileChannel.write(src, 0, attachment, handler); + } + + @Override + public Future write(ByteBuffer src) { + return fileChannel.write(src, 0); + } + + @Override + public void close() throws IOException { + fileChannel.close(); + } + + @Override + public boolean isOpen() { + return fileChannel.isOpen(); + } + + private static class CompletableFutureHandler extends CompletableFuture implements CompletionHandler { + + @Override + public void completed(Integer read, Object attachment) { + complete(read > 0 ? read : -1); + } + + @Override + public void failed(Throwable exc, Object attachment) { + completeExceptionally(exc); + } + } + } } diff --git a/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java b/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java new file mode 100644 index 000000000..64f9ac353 --- /dev/null +++ b/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java @@ -0,0 +1,150 @@ +package com.github.dockerjava.okhttp; + +import com.sun.jna.Native; +import com.sun.jna.win32.StdCallLibrary; +import com.sun.jna.win32.W32APIOptions; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.Socket; +import java.net.SocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.AsynchronousByteChannel; +import java.nio.channels.AsynchronousFileChannel; +import java.nio.channels.Channels; +import java.nio.channels.CompletionHandler; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Future; + +class NamedPipeSocket extends Socket { + + private final String socketFileName; + + private AsynchronousFileByteChannel channel; + + NamedPipeSocket(String socketFileName) { + this.socketFileName = socketFileName; + } + + @Override + public void close() throws IOException { + if (channel != null) { + channel.close(); + } + } + + @Override + public void connect(SocketAddress endpoint) throws IOException { + connect(endpoint, 0); + } + + @Override + public void connect(SocketAddress endpoint, int timeout) throws IOException { + long startedAt = System.currentTimeMillis(); + + timeout = Math.max(timeout, 10_000); + while (true) { + try { + channel = new AsynchronousFileByteChannel( + AsynchronousFileChannel.open( + Paths.get(socketFileName), + StandardOpenOption.READ, + StandardOpenOption.WRITE + ) + ); + break; + } catch (FileNotFoundException e) { + if (System.currentTimeMillis() - startedAt >= timeout) { + throw new RuntimeException(e); + } else { + Kernel32.INSTANCE.WaitNamedPipe(socketFileName, 100); + } + } + } + } + + @Override + public InputStream getInputStream() { + return Channels.newInputStream(channel); + } + + @Override + public OutputStream getOutputStream() { + return Channels.newOutputStream(channel); + } + + interface Kernel32 extends StdCallLibrary { + + Kernel32 INSTANCE = Native.load("kernel32", Kernel32.class, W32APIOptions.DEFAULT_OPTIONS); + + @SuppressWarnings("checkstyle:methodname") + boolean WaitNamedPipe(String lpNamedPipeName, int nTimeOut); + } + + private static class AsynchronousFileByteChannel implements AsynchronousByteChannel { + private final AsynchronousFileChannel fileChannel; + + AsynchronousFileByteChannel(AsynchronousFileChannel fileChannel) { + this.fileChannel = fileChannel; + } + + @Override + public void read(ByteBuffer dst, A attachment, CompletionHandler handler) { + fileChannel.read(dst, 0, attachment, new CompletionHandler() { + @Override + public void completed(Integer read, A attachment) { + handler.completed(read > 0 ? read : -1, attachment); + } + + @Override + public void failed(Throwable exc, A attachment) { + handler.failed(exc, attachment); + } + }); + } + + @Override + public Future read(ByteBuffer dst) { + CompletableFutureHandler future = new CompletableFutureHandler(); + fileChannel.read(dst, 0, null, future); + return future; + } + + @Override + public void write(ByteBuffer src, A attachment, CompletionHandler handler) { + fileChannel.write(src, 0, attachment, handler); + } + + @Override + public Future write(ByteBuffer src) { + return fileChannel.write(src, 0); + } + + @Override + public void close() throws IOException { + fileChannel.close(); + } + + @Override + public boolean isOpen() { + return fileChannel.isOpen(); + } + + private static class CompletableFutureHandler extends CompletableFuture implements CompletionHandler { + + @Override + public void completed(Integer read, Object attachment) { + complete(read > 0 ? read : -1); + } + + @Override + public void failed(Throwable exc, Object attachment) { + completeExceptionally(exc); + } + } + } +} diff --git a/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocketFactory.java b/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocketFactory.java index d993a9c52..464910b04 100644 --- a/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocketFactory.java +++ b/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocketFactory.java @@ -1,17 +1,8 @@ package com.github.dockerjava.okhttp; -import com.github.dockerjava.okhttp.OkDockerHttpClient.OkResponse; -import com.sun.jna.platform.win32.Kernel32; - import javax.net.SocketFactory; -import java.io.FileNotFoundException; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.RandomAccessFile; import java.net.InetAddress; import java.net.Socket; -import java.net.SocketAddress; class NamedPipeSocketFactory extends SocketFactory { @@ -23,96 +14,7 @@ class NamedPipeSocketFactory extends SocketFactory { @Override public Socket createSocket() { - return new Socket() { - - RandomAccessFile file; - InputStream is; - OutputStream os; - - @Override - public void close() throws IOException { - if (file != null) { - file.close(); - file = null; - } - } - - @Override - public void connect(SocketAddress endpoint) { - connect(endpoint, 0); - } - - @Override - public void connect(SocketAddress endpoint, int timeout) { - long startedAt = System.currentTimeMillis(); - timeout = Math.max(timeout, 10_000); - while (true) { - try { - file = new RandomAccessFile(socketFileName, "rw"); - break; - } catch (FileNotFoundException e) { - if (System.currentTimeMillis() - startedAt >= timeout) { - throw new RuntimeException(e); - } else { - Kernel32.INSTANCE.WaitNamedPipe(socketFileName, 100); - } - } - } - - is = new InputStream() { - @Override - public int read(byte[] bytes, int off, int len) throws IOException { - if (OkResponse.CLOSING.get()) { - return 0; - } - return file.read(bytes, off, len); - } - - @Override - public int read() throws IOException { - if (OkResponse.CLOSING.get()) { - return 0; - } - return file.read(); - } - - @Override - public int read(byte[] bytes) throws IOException { - if (OkResponse.CLOSING.get()) { - return 0; - } - return file.read(bytes); - } - }; - - os = new OutputStream() { - @Override - public void write(byte[] bytes, int off, int len) throws IOException { - file.write(bytes, off, len); - } - - @Override - public void write(int value) throws IOException { - file.write(value); - } - - @Override - public void write(byte[] bytes) throws IOException { - file.write(bytes); - } - }; - } - - @Override - public InputStream getInputStream() { - return is; - } - - @Override - public OutputStream getOutputStream() { - return os; - } - }; + return new NamedPipeSocket(socketFileName); } @Override From 56d40fc83cc1c7aa71c5fee26994d8ab1a1de22c Mon Sep 17 00:00:00 2001 From: Sergei Egorov Date: Wed, 3 Jun 2020 19:19:20 +0200 Subject: [PATCH 2/3] ignore `AsynchronousCloseException` on read --- .../github/dockerjava/httpclient5/NamedPipeSocket.java | 9 +++++++++ .../com/github/dockerjava/okhttp/NamedPipeSocket.java | 9 +++++++++ 2 files changed, 18 insertions(+) diff --git a/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java b/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java index e2cc181e8..eb9b768bd 100644 --- a/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java +++ b/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java @@ -12,6 +12,7 @@ import java.net.SocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousByteChannel; +import java.nio.channels.AsynchronousCloseException; import java.nio.channels.AsynchronousFileChannel; import java.nio.channels.Channels; import java.nio.channels.CompletionHandler; @@ -101,6 +102,10 @@ public void completed(Integer read, A attachment) { @Override public void failed(Throwable exc, A attachment) { + if (exc instanceof AsynchronousCloseException) { + handler.completed(-1, attachment); + return; + } handler.failed(exc, attachment); } }); @@ -142,6 +147,10 @@ public void completed(Integer read, Object attachment) { @Override public void failed(Throwable exc, Object attachment) { + if (exc instanceof AsynchronousCloseException) { + complete(-1); + return; + } completeExceptionally(exc); } } diff --git a/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java b/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java index 64f9ac353..fc0c024e8 100644 --- a/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java +++ b/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java @@ -12,6 +12,7 @@ import java.net.SocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousByteChannel; +import java.nio.channels.AsynchronousCloseException; import java.nio.channels.AsynchronousFileChannel; import java.nio.channels.Channels; import java.nio.channels.CompletionHandler; @@ -102,6 +103,10 @@ public void completed(Integer read, A attachment) { @Override public void failed(Throwable exc, A attachment) { + if (exc instanceof AsynchronousCloseException) { + handler.completed(-1, attachment); + return; + } handler.failed(exc, attachment); } }); @@ -143,6 +148,10 @@ public void completed(Integer read, Object attachment) { @Override public void failed(Throwable exc, Object attachment) { + if (exc instanceof AsynchronousCloseException) { + complete(-1); + return; + } completeExceptionally(exc); } } From 218e8086bbd64186c62e73ae18c5611e6cf1c5fe Mon Sep 17 00:00:00 2001 From: Sergei Egorov Date: Thu, 4 Jun 2020 09:16:41 +0200 Subject: [PATCH 3/3] Replace `FileNotFoundException` with `NoSuchFileException` --- .../com/github/dockerjava/httpclient5/NamedPipeSocket.java | 4 ++-- .../java/com/github/dockerjava/okhttp/NamedPipeSocket.java | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java b/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java index eb9b768bd..fd080d90f 100644 --- a/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java +++ b/docker-java-transport-httpclient5/src/main/java/com/github/dockerjava/httpclient5/NamedPipeSocket.java @@ -4,7 +4,6 @@ import com.sun.jna.win32.StdCallLibrary; import com.sun.jna.win32.W32APIOptions; -import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -16,6 +15,7 @@ import java.nio.channels.AsynchronousFileChannel; import java.nio.channels.Channels; import java.nio.channels.CompletionHandler; +import java.nio.file.NoSuchFileException; import java.nio.file.Paths; import java.nio.file.StandardOpenOption; import java.util.concurrent.CompletableFuture; @@ -57,7 +57,7 @@ public void connect(SocketAddress endpoint, int timeout) throws IOException { ) ); break; - } catch (FileNotFoundException e) { + } catch (NoSuchFileException e) { if (System.currentTimeMillis() - startedAt >= timeout) { throw new RuntimeException(e); } else { diff --git a/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java b/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java index fc0c024e8..f63e92019 100644 --- a/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java +++ b/docker-java-transport-okhttp/src/main/java/com/github/dockerjava/okhttp/NamedPipeSocket.java @@ -4,7 +4,6 @@ import com.sun.jna.win32.StdCallLibrary; import com.sun.jna.win32.W32APIOptions; -import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -16,6 +15,7 @@ import java.nio.channels.AsynchronousFileChannel; import java.nio.channels.Channels; import java.nio.channels.CompletionHandler; +import java.nio.file.NoSuchFileException; import java.nio.file.Paths; import java.nio.file.StandardOpenOption; import java.util.concurrent.CompletableFuture; @@ -58,7 +58,7 @@ public void connect(SocketAddress endpoint, int timeout) throws IOException { ) ); break; - } catch (FileNotFoundException e) { + } catch (NoSuchFileException e) { if (System.currentTimeMillis() - startedAt >= timeout) { throw new RuntimeException(e); } else {