Response stream corruption when a request is cancelled

[anbencic-msft]

Output of dotnet --info:

.NET Core SDK (reflecting any global.json):
 Version:   2.1.503
 Commit:    4c506e0f35

Runtime Environment:
 OS Name:     Windows
 OS Version:  10.0.17763
 OS Platform: Windows
 RID:         win10-x64
 Base Path:   C:\Program Files\dotnet\sdk\2.1.503\

Host (useful for support):
  Version: 2.1.7
  Commit:  cca5d72d48

.NET Core SDKs installed:
  2.1.201 [C:\Program Files\dotnet\sdk]
  2.1.202 [C:\Program Files\dotnet\sdk]
  2.1.402 [C:\Program Files\dotnet\sdk]
  2.1.503 [C:\Program Files\dotnet\sdk]

.NET Core runtimes installed:
  Microsoft.AspNetCore.All 2.1.4 [C:\Program Files\dotnet\shared\Microsoft.AspNetCore.All]
  Microsoft.AspNetCore.All 2.1.7 [C:\Program Files\dotnet\shared\Microsoft.AspNetCore.All]
  Microsoft.AspNetCore.App 2.1.4 [C:\Program Files\dotnet\shared\Microsoft.AspNetCore.App]
  Microsoft.AspNetCore.App 2.1.7 [C:\Program Files\dotnet\shared\Microsoft.AspNetCore.App]
  Microsoft.NETCore.App 2.0.7 [C:\Program Files\dotnet\shared\Microsoft.NETCore.App]
  Microsoft.NETCore.App 2.0.9 [C:\Program Files\dotnet\shared\Microsoft.NETCore.App]
  Microsoft.NETCore.App 2.1.4 [C:\Program Files\dotnet\shared\Microsoft.NETCore.App]
  Microsoft.NETCore.App 2.1.7 [C:\Program Files\dotnet\shared\Microsoft.NETCore.App]

What version of Docker.DotNet?:

// Docker.DotNet, Version=3.125.2.0, Culture=neutral, PublicKeyToken=null
// Global type: <Module>
// Architecture: AnyCPU (64-bit preferred)
// Runtime: .NET 4.0

Steps to reproduce the issue:

using System;
using System.Diagnostics;
using System.Threading;
using Docker.DotNet.Models;
using Newtonsoft.Json;

namespace ConsoleApp13
{
    class DockerSystemEventLogger : IProgress<JSONMessage>
    {
        /// <inheritdoc />
        public void Report(JSONMessage value)
        {
            Console.WriteLine($"DockerSystemEvent: {JsonConvert.SerializeObject(value)}");
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            var rand = new Random();
            var sw = new Stopwatch();

            using (var client = new Docker.DotNet.DockerClientConfiguration(new Uri("npipe://./pipe/docker_engine")).CreateClient())
            {
                for (int i = 0; i < 10000; ++i)
                {
                    try
                    {
                        // (1) Create monitor task
                        var cts = new CancellationTokenSource();
                        var monitorTask = client.System.MonitorEventsAsync(new ContainerEventsParameters(), new DockerSystemEventLogger(), cts.Token);

                        // (2) Wait for some time to make sure we get into blocking IO call
                        Thread.Sleep(100);

                        // (3) Invoke another request that will attempt to grab the same buffer
                        var listImagesTask1 = client.Images.TagImageAsync("microsoft/windowsservercore:ltsc2016", new ImageTagParameters { RepositoryName = "microsoft/windowsservercore", Tag = $"ltsc2016c", Force = true });

                        // (4) Wait for a short bit again and cancel the monitor task - if we get lucky, we the list images call will grab the same buffer while
                        sw.Restart();
                        var iterations = rand.Next(15000000);

                        for (int j = 0; j < iterations; j++)
                        {
                            // noop
                        }
                        Console.WriteLine($"Waited for {sw.Elapsed.TotalMilliseconds} ms");

                        cts.Cancel();

                        listImagesTask1.GetAwaiter().GetResult();
                        client.Images.TagImageAsync("microsoft/windowsservercore:ltsc2016", new ImageTagParameters { RepositoryName = "microsoft/windowsservercore", Tag = $"ltsc2016c", Force = true }).GetAwaiter().GetResult();

                        monitorTask.GetAwaiter().GetResult();

                        Console.WriteLine(i);
                    }
                    catch (Exception ex)
                    {
                        try
                        {
                            Console.ForegroundColor = ConsoleColor.Red;
                            Console.WriteLine($"{i} failed: {ex}");
                            ////break;
                        }
                        finally
                        {
                            Console.ResetColor();
                        }
                    }
                }
            }

            Console.ReadLine();
        }
    }
}

What actually happened?:

DockerClient.System.MonitorEventsAsync throws ArgumentException or IOException on cancellation. This results in one of three situations that we observed:

1. "System.ArgumentException: Stream was not readable."
2. "System.IO.IOException: Invalid chunk header: HTTP/1.1 201 Created".
3. "System.IO.IOException: Unexpected end of stream"

What did you expect to happen?:

OperationCancelledException to be thrown instead.

Additional information:

The problem is with this code:

https://github.com/Microsoft/Docker.DotNet/blob/master/src/Docker.DotNet/Endpoints/StreamUtil.cs

internal static async Task MonitorStreamForMessagesAsync<T>(Task<Stream> streamTask, DockerClient client, CancellationToken cancel, IProgress<T> progress)
{
    using (var stream = await streamTask)
    {
        // ReadLineAsync must be cancelled by closing the whole stream.
        using (cancel.Register(() => stream.Dispose()))
        {
            using (var reader = new StreamReader(stream, new UTF8Encoding(false)))
            {
                string line;
                try
                {
                    while ((line = await reader.ReadLineAsync()) != null)
                    {
                        var prog = client.JsonSerializer.DeserializeObject<T>(line);
                        if (prog == null) continue;

                        progress.Report(prog);
                    }
                }
                catch (ObjectDisposedException)
                {
                    // The subsequent call to reader.ReadLineAsync() after cancellation
                    // will fail because we disposed the stream. Just ignore here.
                }
            }
        }
    }
}

Basically on cancellation, cancel.Register disposes of the underlying stream while concurrently the request is being executed - can be in any stage...writing request data, reading response, etc. This is not thread safe, and all the three exceptions stem from it.

(1) Happens when the cancellation follows almost immediately after call to DockerClient.System.MonitorEventsAsync.

If the cancellation token is signaled before the ChunkedReadStream is passed into the stream reader, it will be disposed, which will result in "System.ArgumentException: Stream was not readable." thrown out of StreamReader ctor.

(2) Happens due to corruption of the buffer owned by the BufferedReadStream.

On cancellation, the underlying streams are disposed, including the BufferedReadStream. This will cause its buffer (currently owned by the BufferedReadStream) to be released back to the ArrayPool to be reused by another request. However, the read task itself is not cancelled, so now we have two active consumers of the same buffer. The next read on the "cancelled" monitor request results in invalid data - since the buffer was in the meantime populated by headers of another response.

Here's an example of one such buffer from a dump:

HTTP/1.1 201 Created
Api-Version: 1.39
Content-Length: 0
Date: Thu, 28 Feb 2019 15:35:21 GMT
Docker-Experimental: false
Ostype: windows
Server: Docker/18.09.0 (windows)
Connection: close

7394f62a4ca490fbce5","Attributes":{"name":"root/image:42"}},"scope":"local","time":1551368121,"timeNano":1551368121566912100}

}

This was a buffer that was being used and parsed by a ChunkedReadStream that was in the middle of monitoring system events. Hence "System.IO.IOException: Invalid chunk header: HTTP/1.1 201 Created"

(3) A similar problem as (2) with buffer data corruption, just swapped roles.

1. Another request loads data into its buffer,
2. System event monitor overwrites it with its next chunk that arrived on its pipe.
3. Then another request instead of reading an empty string to terminate and not ask for more data reads the system event monitor's chunk size. Assuming there's more data, it asks for the next line, but the underlying pipe stream is already broken since there should be no more data.

Hope this helps.