manual ci start

revert BufferedReadStream

Author: bruegth

.github/workflows/ci.yml

@@ -1,13 +1,11 @@
 name: CI
 
 on:
-  pull_request:
-    branches:
-      - main
+  workflow_dispatch:
 
 jobs:
   build:
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-latest
     services:
       # Docker without TLS (plain TCP) !DEPRECATED! with next docker release
       docker-no-tls:

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

@@ -1,27 +1,39 @@
+#if !NET45
+#endif
+
 namespace Microsoft.Net.Http.Client;
 
-internal sealed class BufferedReadStream : WriteClosableStream, IPeekableStream
+internal class BufferedReadStream : WriteClosableStream, IPeekableStream
 {
+    private const char CR = '\r';
+    private const char LF = '\n';
+
     private readonly Stream _inner;
     private readonly Socket _socket;
     private readonly byte[] _buffer;
-    private readonly ILogger _logger;
-    private int _bufferRefCount;
-    private int _bufferOffset;
-    private int _bufferCount;
+    private volatile int _bufferRefCount;
+    private int _bufferOffset = 0;
+    private int _bufferCount = 0;
+    private bool _disposed;
 
-    public BufferedReadStream(Stream inner, Socket socket, ILogger logger)
-        : this(inner, socket, 8192, logger)
-    {
-    }
+    public BufferedReadStream(Stream inner, Socket socket)
+        : this(inner, socket, 1024)
+    { }
 
-    public BufferedReadStream(Stream inner, Socket socket, int bufferLength, ILogger logger)
+    public BufferedReadStream(Stream inner, Socket socket, int bufferLength)
     {
-        _inner = inner ?? throw new ArgumentNullException(nameof(inner));
+        if (inner == null)
+        {
+            throw new ArgumentNullException(nameof(inner));
+        }
+        _inner = inner;
         _socket = socket;
-        _buffer = ArrayPool<byte>.Shared.Rent(bufferLength);
-        _logger = logger;
+#if !NET45
         _bufferRefCount = 1;
+        _buffer = ArrayPool<byte>.Shared.Rent(bufferLength);
+#else
+            _buffer = new byte[bufferLength];
+#endif
     }
 
     public override bool CanRead
@@ -55,23 +67,6 @@ public override long Position
         set { throw new NotSupportedException(); }
     }
 
-    public override bool CanCloseWrite => _socket != null || _inner is WriteClosableStream;
-
-    protected override void Dispose(bool disposing)
-    {
-        if (disposing)
-        {
-            if (Interlocked.Exchange(ref _bufferRefCount, 0) == 1)
-            {
-                ArrayPool<byte>.Shared.Return(_buffer);
-            }
-
-            _inner.Dispose();
-        }
-
-        base.Dispose(disposing);
-    }
-
     public override long Seek(long offset, SeekOrigin origin)
     {
         throw new NotSupportedException();
@@ -82,6 +77,24 @@ public override void SetLength(long value)
         throw new NotSupportedException();
     }
 
+    protected override void Dispose(bool disposing)
+    {
+        if (!_disposed)
+        {
+            _disposed = true;
+            if (disposing)
+            {
+                _inner.Dispose();
+#if !NET45
+                if (Interlocked.Decrement(ref _bufferRefCount) == 0)
+                {
+                    ArrayPool<byte>.Shared.Return(_buffer);
+                }
+#endif
+            }
+        }
+    }
+
     public override void Flush()
     {
         _inner.Flush();
@@ -124,23 +137,6 @@ public override Task<int> ReadAsync(byte[] buffer, int offset, int count, Cancel
         return _inner.ReadAsync(buffer, offset, count, cancellationToken);
     }
 
-    public override void CloseWrite()
-    {
-        if (_socket != null)
-        {
-            _socket.Shutdown(SocketShutdown.Send);
-            return;
-        }
-
-        if (_inner is WriteClosableStream writeClosableStream)
-        {
-            writeClosableStream.CloseWrite();
-            return;
-        }
-
-        throw new NotSupportedException("Cannot shutdown write on this transport");
-    }
-
     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);
@@ -157,49 +153,6 @@ public bool Peek(byte[] buffer, uint toPeek, out uint peeked, out uint available
         throw new NotSupportedException("_inner stream isn't a peekable stream");
     }
 
-    public async Task<string> ReadLineAsync(CancellationToken cancellationToken)
-    {
-        var line = new StringBuilder(_buffer.Length);
-
-        var crIndex = -1;
-
-        var lfIndex = -1;
-
-        bool crlfFound;
-
-        do
-        {
-            if (_bufferCount == 0)
-            {
-                _bufferOffset = 0;
-
-                _bufferCount = await _inner.ReadAsync(_buffer, 0, _buffer.Length, cancellationToken)
-                    .ConfigureAwait(false);
-            }
-
-            var c = (char)_buffer[_bufferOffset];
-            line.Append(c);
-
-            _bufferOffset++;
-            _bufferCount--;
-
-            switch (c)
-            {
-                case '\r':
-                    crIndex = line.Length;
-                    break;
-                case '\n':
-                    lfIndex = line.Length;
-                    break;
-            }
-
-            crlfFound = crIndex + 1 == lfIndex;
-        }
-        while (!crlfFound);
-
-        return line.ToString(0, line.Length - 2);
-    }
-
     private int ReadBuffer(byte[] buffer, int offset, int count)
     {
         if (_bufferCount > 0)
@@ -231,4 +184,101 @@ private int PeekBuffer(byte[] buffer, uint toPeek, out uint peeked, out uint ava
         remaining = 0;
         return 0;
     }
+
+    private async Task EnsureBufferedAsync(CancellationToken cancel)
+    {
+        if (_bufferCount == 0)
+        {
+            _bufferOffset = 0;
+#if !NET45
+            bool validBuffer = Interlocked.Increment(ref _bufferRefCount) > 1;
+            try
+            {
+                if (validBuffer)
+                {
+                    _bufferCount = await _inner.ReadAsync(_buffer, _bufferOffset, _buffer.Length, cancel).ConfigureAwait(false);
+                }
+            }
+            finally
+            {
+                if ((Interlocked.Decrement(ref _bufferRefCount) == 0) && validBuffer)
+                {
+                    ArrayPool<byte>.Shared.Return(_buffer);
+                }
+            }
+#else
+                _bufferCount = await _inner.ReadAsync(_buffer, _bufferOffset, _buffer.Length, cancel).ConfigureAwait(false);
+#endif
+            if (_bufferCount == 0)
+            {
+                throw new IOException("Unexpected end of stream");
+            }
+        }
+    }
+
+    // TODO: Line length limits?
+    public async Task<string> ReadLineAsync(CancellationToken cancel)
+    {
+        ThrowIfDisposed();
+        StringBuilder builder = new StringBuilder();
+        bool foundCR = false, foundCRLF = false;
+        do
+        {
+            if (_bufferCount == 0)
+            {
+                await EnsureBufferedAsync(cancel).ConfigureAwait(false);
+            }
+
+            char ch = (char)_buffer[_bufferOffset]; // TODO: Encoding enforcement
+            builder.Append(ch);
+            _bufferOffset++;
+            _bufferCount--;
+            if (ch == CR)
+            {
+                foundCR = true;
+            }
+            else if (ch == LF)
+            {
+                if (foundCR)
+                {
+                    foundCRLF = true;
+                }
+                else
+                {
+                    foundCR = false;
+                }
+            }
+        }
+        while (!foundCRLF);
+
+        return builder.ToString(0, builder.Length - 2); // Drop the CRLF
+    }
+
+    private void ThrowIfDisposed()
+    {
+        if (_disposed)
+        {
+            throw new ObjectDisposedException(nameof(BufferedReadStream));
+        }
+    }
+
+    public override bool CanCloseWrite => _socket != null || _inner is WriteClosableStream;
+
+    public override void CloseWrite()
+    {
+        if (_socket != null)
+        {
+            _socket.Shutdown(SocketShutdown.Send);
+            return;
+        }
+
+        var s = _inner as WriteClosableStream;
+        if (s != null)
+        {
+            s.CloseWrite();
+            return;
+        }
+
+        throw new NotSupportedException("Cannot shutdown write on this transport");
+    }
 }

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

@@ -177,7 +177,7 @@ private async Task<HttpResponseMessage> ProcessRequestAsync(HttpRequestMessage r
             transport = sslStream;
         }
 
-        var bufferedReadStream = new BufferedReadStream(transport, socket, _logger);
+        var bufferedReadStream = new BufferedReadStream(transport, socket);
         var connection = new HttpConnection(bufferedReadStream);
         return await connection.SendAsync(request, cancellationToken);
     }
@@ -363,7 +363,7 @@ private async Task TunnelThroughProxyAsync(HttpRequestMessage request, Stream tr
         connectRequest.SetAddressLineProperty(authority);
         connectRequest.Headers.Host = authority;
 
-        HttpConnection connection = new HttpConnection(new BufferedReadStream(transport, null, _logger));
+        HttpConnection connection = new HttpConnection(new BufferedReadStream(transport, null));
         HttpResponseMessage connectResponse;
         try
         {

test/Docker.DotNet.Tests/TestFixture.cs

@@ -1,4 +1,5 @@
 using System.IO;
+using System.Net;
 using System.Net.Security;
 using System.Security.Cryptography.X509Certificates;
 using Docker.DotNet.X509;
@@ -221,7 +222,8 @@ public static IEnumerable<object[]> GetDockerClientTypes()
                 .Select(t => new object[] { t });
         }
 
-        return allClients.Select(t => new object[] { t });
+        //return allClients.Select(t => new object[] { t });
+        return allClients.Where(t => t == TestClientsEnum.ManagedHttps).Select(t => new object[] { t });
     }
 
     public static IEnumerable<TestDaemonsEnum> GetDockerDaemonTypes()