Added peekable streams to avoid blocking read calls on pipes with no data (#390)

* added peekable streams

* set extern method to private

* don't upgrade newtonsoft

Co-authored-by: Novin Changizi <nochangi@microsoft.com>

Author: novinc

src/Docker.DotNet/DockerPipeStream.cs

@@ -8,7 +8,7 @@
 
 namespace Docker.DotNet
 {
-    internal class DockerPipeStream : WriteClosableStream
+    internal class DockerPipeStream : WriteClosableStream, IPeekableStream
     {
         private readonly PipeStream _stream;
         private readonly EventWaitHandle _event = new EventWaitHandle(false, EventResetMode.AutoReset);
@@ -44,6 +44,9 @@ public override long Position
         [DllImport("api-ms-win-core-io-l1-1-0.dll", SetLastError = true)]
         private static extern int GetOverlappedResult(SafeHandle handle, ref NativeOverlapped overlapped, out int numBytesWritten, int wait);
 
+        [DllImport("kernel32.dll", SetLastError = true)]
+        private static extern bool PeekNamedPipe(SafeHandle handle, byte[] buffer, uint nBufferSize, ref uint bytesRead, ref uint bytesAvail, ref uint BytesLeftThisMessage);
+
         public override void CloseWrite()
         {
             // The Docker daemon expects a write of zero bytes to signal the end of writes. Use native
@@ -92,6 +95,27 @@ public override Task<int> ReadAsync(byte[] buffer, int offset, int count, Cancel
             return _stream.ReadAsync(buffer, offset, count, cancellationToken);
         }
 
+        public bool Peek(byte[] buffer, uint toPeek, out uint peeked, out uint available, out uint remaining)
+        {
+            peeked = 0;
+            available = 0;
+            remaining = 0;
+
+            bool aPeekedSuccess = PeekNamedPipe(
+                _stream.SafePipeHandle,
+                buffer, toPeek,
+                ref peeked, ref available, ref remaining);
+
+            var error = Marshal.GetLastWin32Error();
+
+            if (error == 0 && aPeekedSuccess)
+            {
+                return true;
+            }
+
+            return false;
+        }
+
         public override long Seek(long offset, SeekOrigin origin)
         {
             throw new NotImplementedException();

src/Docker.DotNet/IPeekableStream.cs

@@ -0,0 +1,18 @@
+namespace Docker.DotNet
+{
+    public interface IPeekableStream
+    {
+        /// <summary>
+        /// Peek the underlying stream, can be used in order to avoid a blocking read call when no data is available
+        /// https://stackoverflow.com/questions/6846365/check-for-eof-in-namedpipeclientstream
+        /// https://msdn.microsoft.com/en-us/library/windows/desktop/aa365779(v=vs.85).aspx
+        /// </summary>
+        /// <param name="buffer">buffer to put peeked data in</param>
+        /// <param name="toPeek">max number of bytes to peek</param>
+        /// <param name="peeked">number of bytes that were peeked</param>
+        /// <param name="available">number of bytes that were available for peeking</param>
+        /// <param name="remaining">number of available bytes minus number of peeked</param>
+        /// <returns>whether peek operation succeeded</returns>
+        bool Peek(byte[] buffer, uint toPeek, out uint peeked, out uint available, out uint remaining);
+    }
+}

src/Docker.DotNet/Microsoft.Net.Http.Client/BufferedReadStream.cs

@@ -4,14 +4,15 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Docker.DotNet;
 
 #if !NET45
 using System.Buffers;
 #endif
 
 namespace Microsoft.Net.Http.Client
 {
-    internal class BufferedReadStream : WriteClosableStream
+    internal class BufferedReadStream : WriteClosableStream, IPeekableStream
     {
         private const char CR = '\r';
         private const char LF = '\n';
@@ -145,6 +146,22 @@ public override Task<int> ReadAsync(byte[] buffer, int offset, int count, Cancel
             return _inner.ReadAsync(buffer, offset, count, cancellationToken);
         }
 
+        public bool Peek(byte[] buffer, uint toPeek, out uint peeked, out uint available, out uint remaining)
+        {
+            int read = PeekBuffer(buffer, toPeek, out peeked, out available, out remaining);
+            if (read > 0)
+            {
+                return true;
+            }
+
+            if (_inner is IPeekableStream peekableStream)
+            {
+                return peekableStream.Peek(buffer, toPeek, out peeked, out available, out remaining);
+            }
+
+            throw new NotSupportedException("_inner stream isn't a peekable stream");
+        }
+
         private int ReadBuffer(byte[] buffer, int offset, int count)
         {
             if (_bufferCount > 0)
@@ -159,6 +176,24 @@ private int ReadBuffer(byte[] buffer, int offset, int count)
             return 0;
         }
 
+        private int PeekBuffer(byte[] buffer, uint toPeek, out uint peeked, out uint available, out uint remaining)
+        {
+            if (_bufferCount > 0)
+            {
+                int toCopy = Math.Min(_bufferCount, (int)toPeek);
+                Buffer.BlockCopy(_buffer, _bufferOffset, buffer, 0, toCopy);
+                peeked = (uint) toCopy;
+                available = (uint)_bufferCount;
+                remaining = available - peeked;
+                return toCopy;
+            }
+
+            peeked = 0;
+            available = 0;
+            remaining = 0;
+            return 0;
+        }
+
         private async Task EnsureBufferedAsync(CancellationToken cancel)
         {
             if (_bufferCount == 0)

src/Docker.DotNet/MultiplexedStream.cs

@@ -11,7 +11,7 @@
 
 namespace Docker.DotNet
 {
-    public class MultiplexedStream : IDisposable
+    public class MultiplexedStream : IDisposable, IPeekableStream
     {
         private readonly Stream _stream;
         private TargetStream _target;
@@ -54,6 +54,16 @@ public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken c
             return _stream.WriteAsync(buffer, offset, count, cancellationToken);
         }
 
+        public bool Peek(byte[] buffer, uint toPeek, out uint peeked, out uint available, out uint remaining)
+        {
+            if (_stream is IPeekableStream peekableStream)
+            {
+                return peekableStream.Peek(buffer, toPeek, out peeked, out available, out remaining);
+            }
+
+            throw new NotSupportedException("_stream isn't a peekable stream");
+        }
+
         public async Task<ReadResult> ReadOutputAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
             if (!_multiplexed)