StreamReader.ReadLineAsync() does not return before its internal buffer is full

Is there an existing issue for this?

I have searched the existing issues

Describe the bug

That bug report is about a regression between Blazor 6.0.11 and Blazor 7.0.0

Conditions:

An API endpoint on the server produces a stream of data (text) slowly: A single line of text is added to the response every second.
A method on the client side (Blazor WASM) reads that stream of data line-by-line using StreamReader.ReadLineAsync().
All prerequisites to enable response streaming are met (HttpRequestMessage.SetBrowserResponseStreamingEnabled(true), HttpCompletionOption.ResponseHeadersRead)

When using Blazor 6.0.11, each time a line-break is received in the stream of data, the method StreamReader.ReadLineAsync() returns the text that was preceding that line-break.

With Blazor 7.0.0, the method StreamReader.ReadLineAsync() will not return before the StreamReader's internal buffer has been entirely filled, which means that several lines of text can be received before StreamReader.ReadLineAsync() returns them all consecutively.

When reading the data directly from the response stream, that behavior disappear and all the data is available as soon as it is received. That seems to suggest that the issue is only due to the StreamReader.

However, that issue cannot be reproduced when the client code is executed from a console application: The behavior is then exactly the same when using .Net 6 and .Net 7: StreamReader.ReadLineAsync() returns on every line break without waiting for the buffer to be full.

Expected Behavior

The StreamReader should behave the same way when using Blazor 6.0.11 and Blazor 7.0.0: StreamReader.ReadLineAsync() should return as soon as a line-break is available from the input stream.

Steps To Reproduce

A minimal project to reproduce the issue is available here:
https://github.com/oleneveu/BlazorStreamReaderIssue

It is based on the template from Visual Studio. The code that reproduces the issue is located here:

Server side: DataStreamingController.cs
Client side: StreamData.razor

The project currently targets Blazor 7.0.0.
To compare the behavior with Blazor 6.0.11, all projects files must be modified to target .Net 6 and all packages references must be changed from 7.0.0 to 6.0.11. No other changes should be required.

Steps to reproduce:

Start the project
Open the 'Stream data' menu from the Blazor application
Blazor 7: The text 'Streamed data: Waiting for server data...' is displayed for about 17 seconds before being updated. It is then updated every 17 seconds or so.
Blazor 6: The text 'Streamed data: Waiting for server data...' is updated immediately and then every 1 second or so.

Note: Each line of text sent by the server contains 7 bytes (5 digits + CR/LF) . Within 17 seconds, 18 lines or 126 bytes are sent. When the 19th line is sent, the total size exceeds the buffer size of the StreamReader (configured to 128 bytes in the sample project).
When changing the buffer size of the StreamReader, the delay before the first update changes accordingly.

Update: I have added a console application to the solution ('ConsoleStreaming') to perform the same test that is made from the Blazor component and highlight the difference in behavior. To run this test:

Start the server (BlazorStreaming.Server)
Start the console app (ConsoleStreaming)
When a line of text is added to the response from the 'DataStreamingController.Get()' endpoint, that line is immediately displayed in the console.

Exceptions (if any)

No response

.NET Version

No response

Anything else?

No response

@lewing

@lewing FYI

I have spent some more time on this issue and concluded that it has actually nothing to do with the StreamReader.
It is related to the stream returned by HttpResponseMessage.Content.ReadAsStreamAsync().

I have updated the sample project accordingly.

The implementation on the server side has not changed.
The test now consists of displaying the number of bytes returned by responseStream.ReadAsync() in a loop.

When using Blazor 6, the call returns for each packet written by the server, without waiting for the buffer to be full.
With Blazor 7, the call only returns when enough data has been received from the server to fill the buffer.

When used from a console application, the behavior of the HttpClient is the same we had with Blazor 6.

I suspect the problem is here:

runtime/src/mono/wasm/runtime/http.ts

Line 115 in 23059de

// loop until end of browser stream or end of C# buffer

For anyone running into this issue, I created a wrapper for the blocking HTTP stream.

The usage is:

using var httpStream = await response.Content.ReadAsStreamAsync();
using var proxyStream = new ByteStream(httpStream); // Use this for streamed data operations

Source:

/// <summary>
/// Temporary fix for https://github.com/dotnet/runtime/issues/79238
/// </summary>
public class ByteStream : Stream
{
    ConcurrentQueue<byte> buffer = new ConcurrentQueue<byte>();
    private bool isDisposed = false;
    private Stream parentStream;

    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 ByteStream(Stream parentStream)
    {
        this.parentStream = parentStream;
        ReadTask(); // Run in the background
    }

    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
    {
        for (int i = 0; i < count; i++)
        {
            if (this.buffer.TryDequeue(out var value))
            {
                buffer[offset + i] = value;
            }
            else
            {
                return Task.FromResult(i);
            }
        }

        return Task.FromResult(count);
    }

    public override int Read(byte[] buffer, int offset, int count) => 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();

    protected override void Dispose(bool disposing)
    {
        this.isDisposed = true;
        base.Dispose(disposing);
    }

    private async Task ReadTask()
    {
        var buffer = new byte[1];

        while (this.isDisposed == false)
        {
            await this.parentStream.ReadAsync(buffer, 0, 1); // Never 0, blocks until that 1 byte is read
            this.buffer.Enqueue(buffer[0]);
        }
    }
}

Update: the previous workaround does not works when the stream is closed by the server. I investigated a bit more, and came up with a bypass solution, that instead of trying to solve the Blazor side, I have fixed the JS code itself. I'm no JS expert, I had to copy over some supplementary code manually because they are in a different module.

Step 1: create a JS file:

/* ================= Supplementary Types */
class MemoryView {
    constructor(pointer, length, viewType) {
        this._pointer = pointer;
        this._length = length;
        this._viewType = viewType;
    }

    _unsafe_create_view() {
        const view =
            0 == this._viewType
                ? new Uint8Array(window.Module.HEAPU8.buffer, this._pointer, this._length)
                : 1 == this._viewType
                ? new Int32Array(window.Module.HEAP32.buffer, this._pointer, this._length)
                : 2 == this._viewType
                ? new Float64Array(window.Module.HEAPF64.buffer, this._pointer, this._length)
                : null;
        if (!view)
        {
            throw new Error("NotImplementedException");
        }
    
        return view;
    }

    set(source, targetOffset) {
        if (this.isDisposed)
        {
            throw new Error("Assert failed: ObjectDisposedException");
        }

        const targetView = this._unsafe_create_view();

        if (!(source && targetView && source.constructor === targetView.constructor))
        {
            throw new Error(`Assert failed: Expected ${targetView.constructor}`);
        }
    
        targetView.set(source, targetOffset);
    }

    copyTo(target, sourceOffset) {
        if (this.isDisposed)
        {
            throw new Error("Assert failed: ObjectDisposedException");
        }

        const sourceView = this._unsafe_create_view();

        if (!(target && sourceView && target.constructor === sourceView.constructor))
        {
            throw new Error(`Assert failed: Expected ${sourceView.constructor}`);
        }
        const trimmedSource = sourceView.subarray(sourceOffset);
        target.set(trimmedSource);
    }

    slice(start, end) {
        if (this.isDisposed)
        {
            throw new Error("Assert failed: ObjectDisposedException");
        }
        const sourceView = this._unsafe_create_view();
        return sourceView.slice(start, end);
    }

    get length() {
        if (this.isDisposed)
        {
            throw new Error("Assert failed: ObjectDisposedException");
        }

        return this._length;
    }

    get byteLength() {
        if (this.isDisposed)
        {
            throw new Error("Assert failed: ObjectDisposedException");
        }
        return this._viewType == 0
            ? this._length
            : this._viewType == 1
                ? this._length << 2
                : this._viewType == 2
                    ? this._length << 3
                    : 0;
    }
}
class Span extends MemoryView {
    constructor(pointer, length, viewType) {
        super(pointer, length, viewType);
        this.is_disposed = false;
    }

    dispose() {
        this.is_disposed = true;
    }

    get isDisposed() {
        return this.is_disposed;
    }
}

/* ================= Supplementary Functions */
const Dafuq = Symbol.for("wasm promise_control");
function CreatePromiseController(e, t) {
    let promiseControl = null;
    const promise = new Promise(function (resolve, reject) {
        promiseControl = {
            isDone: false,
            promise: null,
            resolve: (data) => {
                promiseControl.isDone || ((promiseControl.isDone = true), resolve(data), e && e());
            },
            reject: (reason) => {
                promiseControl.isDone || ((promiseControl.isDone = true), reject(reason), t && t());
            },
        };
    });
    promiseControl.promise = promise;
    promise[Dafuq] = promiseControl;
    return { promise: promise, promise_control: promiseControl };
}

function WrapAsCancelablePromise(e) {
    const { promise: promise, promise_control: promiseControl } = CreatePromiseController();
    
    e()
        .then((e) => promiseControl.resolve(e))
        .catch((e) => promiseControl.reject(e));

    return promise;
}

/* ================= Bypass */
async function Bypass(res, bufferPtr, bufferLength) {
    const view = new Span(bufferPtr, bufferLength, 0);

    return WrapAsCancelablePromise(async () => {
        // If a chunk is not read yet, read it
        if (!res.__chunk && res.body)
        {
            res.__reader = res.body.getReader();
            res.__chunk = await res.__reader.read();
            res.__source_offset = 0;
        }

        // If chunk is empty, read the next
        const remainingBytesInChunk = res.__chunk.value.byteLength - res.__source_offset;
        if (remainingBytesInChunk === 0) {
            res.__chunk = await res.__reader.read(); // Read next chunk
            res.__source_offset = 0; // Reset offset
        }

        const bytesToCopy = Math.min(remainingBytesInChunk, view.byteLength);
        const sourceView = res.__chunk.value.subarray(res.__source_offset, bytesToCopy);
        view.set(sourceView, 0);
        res.__source_offset += bytesToCopy;

        return bytesToCopy;
    });
};

function InitHttpBugBypass()
{
    window.INTERNAL.http_wasm_get_streamed_response_bytes = Bypass;
}

Step 2: Add this code to any component that loads only once. It must be a Blazor component because the above code must run after the Blazor runtime has loaded. I use App.razor.

@code {
    [Inject]
    private IJSRuntime JS { get; set; }

    protected override async Task OnInitializedAsync()
    {
        await JS.InvokeVoidAsync("InitHttpBugBypass");
    }
}

Navigation Menu

Repository navigation

StreamReader.ReadLineAsync() does not return before its internal buffer is full #79238

Is there an existing issue for this?

Describe the bug

Expected Behavior

Steps To Reproduce

Exceptions (if any)

.NET Version

Anything else?

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

StreamReader.ReadLineAsync() does not return before its internal buffer is full #79238

Description

Is there an existing issue for this?

Describe the bug

Expected Behavior

Steps To Reproduce

Exceptions (if any)

.NET Version

Anything else?

Activity

mkArtakMSFT commented on Dec 5, 2022

oleneveu commented on Dec 22, 2022

svick commented on Dec 22, 2022

rabirland commented on Jan 10, 2023

rabirland commented on Jan 13, 2023

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions