Reading one source Stream by multiple consumers asynchronously

Question

The problem was using single Stream from the HTTP response in multiple consumers simultaneously. For example: while data is loading, I want write it to the some cache file and deserealize it at the same time.

This code will not work for an obvious reason. But it represents some desired things.

private readonly HttpClient _client = new HttpClient();

// BAD CODE!!!
private async Task<T> HttpAPIRequest<T>(string url, string path)
{
    using HttpResponseMessage response = await _client.GetAsync(url, HttpCompletionOption.ResponseHeadersRead).ConfigureAwait(false);
    response.EnsureSuccessStatusCode();
    using Stream responseStream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
    using FileStream fileStream = File.Create(path);
    ValueTask<T> jsonTask = JsonSerializer.DeserializeAsync<T>(responseStream);
    await responseStream.CopyToAsync(fileStream);
    return await jsonTask;
}

---

The solution

After few days of hard search for the solution I've ended up with the following one consisting of two classes:

public sealed class ConcurrentStreamReader : IDisposable
{
    private readonly Stream _inputStream;
    private readonly MemoryStream _memoryStream;
    private readonly int _bufferSize;
    private readonly IProgress<long> _progress;

    public ConcurrentStreamReader(Stream inputStream) : this(inputStream, null) { }

    public ConcurrentStreamReader(Stream inputStream, IProgress<long> progress) : this(inputStream, 81920, 0, progress) { }

    public ConcurrentStreamReader(Stream inputStream, int bufferSize, int capacity, IProgress<long> progress)
    {
        _memoryStream = new MemoryStream(capacity);
        _inputStream = inputStream;
        _bufferSize = bufferSize;
        _progress = progress;
    }

    public Stream CreateNewStream() => new BlockingStream(ReadStream().GetEnumerator(), _memoryStream);

    public async Task CopyToAsync(Stream outputStream)
    {
        foreach (ReadOnlyMemory<byte> buffer in ReadStream())
        {
            await outputStream.WriteAsync(buffer);
        }
    }

    private IEnumerable<ReadOnlyMemory<byte>> ReadStream()
    {
        long position = 0;
        byte[] buffer = new byte[_bufferSize];
        int bytesRead;
        while (true)
        {
            if (position == _memoryStream.Length)
            {
                lock (_inputStream)
                {
                    if (position < _memoryStream.Length) continue;
                    bytesRead = _inputStream.Read(buffer, 0, _bufferSize);
                    if (bytesRead == 0) break;
                    lock (_memoryStream)
                    {
                        _memoryStream.Position = _memoryStream.Length;
                        _memoryStream.Write(buffer, 0, bytesRead);
                    }
                }
                _progress?.Report(_memoryStream.Length);
            }
            else
            {
                lock (_memoryStream)
                {
                    _memoryStream.Position = position;
                    bytesRead = _memoryStream.Read(buffer, 0, _bufferSize);
                }
            }
            position += bytesRead;
            yield return new ReadOnlyMemory<byte>(buffer, 0, bytesRead);
        }
    }

    private bool _disposed;
    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }
    private void Dispose(bool disposing)
    {
        if (!_disposed)
        {
            if (disposing)
            {
                _memoryStream.Dispose();
            }
        }
        _disposed = true;
    }
    ~ConcurrentStreamReader() => Dispose(false);
}

public sealed class BlockingStream : Stream
{
    private readonly IEnumerator<ReadOnlyMemory<byte>> _sourceEnumerator;
    private readonly MemoryStream _stream;
    private long _position;

    private BlockingStream() { }

    public BlockingStream(IEnumerator<ReadOnlyMemory<byte>> sourceEnumerator, MemoryStream stream)
    {
        _stream = stream;
        _sourceEnumerator = sourceEnumerator;
        _position = 0;
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        long tail = _position + offset + count;
        while (tail >= _stream.Length && _sourceEnumerator.MoveNext()) { }
        int bytesRead;
        lock (_stream)
        {
            _stream.Position = _position;
            bytesRead = _stream.Read(buffer, offset, count);
        }
        _position += bytesRead;
        return bytesRead;
    }

    public override bool CanRead => true;
    public override bool CanSeek => false;
    public override bool CanWrite => false;
    public override long Length => throw new NotSupportedException();
    public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
    public override void Flush() => throw new NotSupportedException();
    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
    public override void SetLength(long value) => throw new NotSupportedException();
    public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
}

Additionally I've included to the solution the download Progress reporting (that I will not use in the usage example below).

And the usage

private async Task<T> HttpAPIRequest<T>(string url, string path)
{
    using HttpResponseMessage response = await _client.GetAsync(url, HttpCompletionOption.ResponseHeadersRead).ConfigureAwait(false);
    response.EnsureSuccessStatusCode();
    using Stream responseStream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
    using ConcurrentStreamReader reader = new ConcurrentStreamReader(responseStream);
    using FileStream fileStream = File.Create(path);
    using Stream deserealizerStream = reader.CreateNewStream();
    ValueTask<T> jsonTask = JsonSerializer.DeserializeAsync<T>(deserealizerStream);
    await reader.CopyToAsync(fileStream);
    return await jsonTask;
}

Environment: x64 WPF app / .NET Core 3.1

The code works correctly but it's pretty complicated for me especially near lock() statements. I'm not sure of that part and performed some code changes as random fixes while I was testing it. Suggestions are appreciated.

Answer 1

If you do not need report progress then the LoadIntoBuffer source code + CopyToAsync source code combo could be handy.

All you need to prevent the concurrent call of the LoadIntoBuffer method. (You can call it n times it does not matter, because the IsBuffered flag will short-cut the method.) This can be done by using a Monitor or a Semaphore for mutual exclusion. Or you can simply use signalling primitives as well, like AutoResetEvent, CountdownEvent.

With CopyToAsync each thread can have its own copy of the stream, so there won't be a shared resource.