7
\$\begingroup\$

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.

\$\endgroup\$
  • 1
    \$\begingroup\$ Have you considered to use SynchronizedStream via Stream.Synchronized? \$\endgroup\$ – Peter Csala Jun 22 at 14:28
  • 1
    \$\begingroup\$ @PeterCsala No, but good point. Thank you very much! Will learn something about it. \$\endgroup\$ – aepot Jun 22 at 14:43
  • 1
    \$\begingroup\$ @aepot Minor but important observation: You are using lock statements on those objects on which you are working on. Your _inputStream is locked even though it is received as input. Please use separate lock objects for these. Here you can find some valuable resources about this topic: 1, 2, 3 \$\endgroup\$ – Peter Csala Jun 24 at 9:35
  • 1
    \$\begingroup\$ @PeterCsala You're right but I'm aware of it and was cereful in using locks. I know that it's violates some programming culture and will consider fixing it. Thank you again! \$\endgroup\$ – aepot Jun 24 at 10:10
  • 1
    \$\begingroup\$ @aepot I've jut forgot to ask you about memory optimization. Is it a concern for you? Because fortunately there are plenty of good techniques to reduce the memory footprint. Like instead of directly allocating a byte[] you could use ArrayPool. Instead of directly allocating a MemoryStream you could use RecycableMemoryStream. You could also take advantage of the PipeReader \ SequenceReader 1 \$\endgroup\$ – Peter Csala Jun 30 at 6:37
3
\$\begingroup\$

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.

| improve this answer | |
\$\endgroup\$
  • 1
    \$\begingroup\$ Thank you! Unfortunately progress reporting is required. \$\endgroup\$ – aepot Jun 23 at 14:23
  • 1
    \$\begingroup\$ Great thanks for the suggestions! I have no more questions related to my code. \$\endgroup\$ – aepot Jun 27 at 13:14

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.