[System.Text.Json] DeserializeAsyncEnumerable is delayed emitting objects when writing asynchronously to stream

[ohadvano]

Description

When invoking DeserializeAsyncEnumerable on a utf8Json content stream, and the stream writer has not finished yet, the IAsyncEnumerable is emitted with the first element only after the second element is written to the stream. The second will be emitted after the third is written, and this goes on. Consequently, the last message in the array will not be emitted unless the stream is closed, although it contains the necessary data to deserialize the object.

Reproduction Steps

Csproj:

<Project Sdk="Microsoft.NET.Sdk">
    <PropertyGroup>
        <OutputType>Exe</OutputType>
        <TargetFramework>net6.0</TargetFramework>
        <ImplicitUsings>enable</ImplicitUsings>
        <Nullable>enable</Nullable>
    </PropertyGroup>
    <ItemGroup>
        <PackageReference Include="Nerdbank.Streams" Version="[2.8.57]" />
    </ItemGroup>
</Project>

Code:

namespace DeserializeAsyncEnumerable.Test;

using Nerdbank.Streams;
using System;
using System.Diagnostics;
using System.Text.Json;
using System.Threading.Tasks;

public class Program
{
    public static async Task Main(string[] args)
    {
        var testStream = new SimplexStream();
        _ = WriteJsonElementsPeriodicallyToStreamAsync(testStream);

        var deserializationOptions = new JsonSerializerOptions { DefaultBufferSize = 64 };
        var testElements = JsonSerializer.DeserializeAsyncEnumerable<Test>(testStream, deserializationOptions);

        var actualCount = 1;
        var stopwatch = Stopwatch.StartNew();
        await foreach (var testElement in testElements)
        {
            Console.WriteLine($"[{actualCount}][Time elapsed: " + stopwatch.Elapsed.TotalSeconds + "; Data: " + testElement?.Data + "]");
            actualCount += 1;
        }
    }

    private static async Task WriteJsonElementsPeriodicallyToStreamAsync(SimplexStream testStream)
    {
        var streamWriter = new StreamWriter(testStream) { AutoFlush = true };

        var count = 1;
        var maxElements = 10;
        var delayInSeconds = 2;

        await streamWriter.WriteAsync($"[{{\"Data\":\"This will be printed after ~{delayInSeconds * count} seconds; Expected: {delayInSeconds * (count - 1)}\"}}");
        count += 1;

        while (count <= maxElements)
        {
            await Task.Delay(TimeSpan.FromSeconds(delayInSeconds));

            await streamWriter.WriteAsync($",{{\"Data\":\"This will be printed after ~{delayInSeconds * count} seconds; Expected: {delayInSeconds * (count - 1)}\"}}");
            count += 1;
        }

        await streamWriter.WriteAsync($"]");

        var preventFromLastMessageToBeEmitted = true;
        if (!preventFromLastMessageToBeEmitted)
        {
            testStream.CompleteWriting();
        }
    }

    private class Test
    {
        public string? Data { get; init; }
    }
}

Expected behavior

When writing the first JSON element (as string) to the stream, the IAsyncEnumerable will receive its first deserialized element in the enumeration, unblocking the enumeration.

Actual behavior

When writing the first JSON element (as string) to the stream, the IAsyncEnumerable is yet to receive the first deserialized element. The first deserialized element will only be received after the second JSON element (as string) will be written to the stream. Consequently, the last message in the array will not be emitted unless the stream is closed, although it contains the necessary data to deserialize the object.

Regression?

No response

Known Workarounds

As a workaround in my particular case, I am using Newtonsoft.Json's JsonTextReader.ReadAsync to asynchronously read JSON tokens from the stream, and deserializing manually to the required object. This way the code works as expected.

Configuration

I am using .NET 6, Windows 11, x64

Other information

The real flow that produced this behavior is not using SimplexStream from Nerdbank.Streams, but using NetworkStream.

Tagging subscribers to this area: @dotnet/area-system-text-json, @gregsdennis
See info in area-owners.md if you want to be subscribed.

Issue Details

---

Description

When invoking DeserializeAsyncEnumerable on a utf8Json content stream, and the stream writer has not finished yet, the IAsyncEnumerable is emitted with the first element only after the second element is written to the stream. The second will be emitted after the third is written, and this goes on. Consequently, the last message in the array will not be emitted unless the stream is closed, although it contains the necessary data to deserialize the object.

Reproduction Steps

Csproj:

<Project Sdk="Microsoft.NET.Sdk">
    <PropertyGroup>
        <OutputType>Exe</OutputType>
        <TargetFramework>net6.0</TargetFramework>
        <ImplicitUsings>enable</ImplicitUsings>
        <Nullable>enable</Nullable>
    </PropertyGroup>
    <ItemGroup>
        <PackageReference Include="Nerdbank.Streams" Version="[2.8.57]" />
    </ItemGroup>
</Project>

Code:

namespace DeserializeAsyncEnumerable.Test;

using Nerdbank.Streams;
using System;
using System.Diagnostics;
using System.Text.Json;
using System.Threading.Tasks;

public class Program
{
    public static async Task Main(string[] args)
    {
        var testStream = new SimplexStream();
        _ = WriteJsonElementsPeriodicallyToStreamAsync(testStream);

        var deserializationOptions = new JsonSerializerOptions { DefaultBufferSize = 64 };
        var testElements = JsonSerializer.DeserializeAsyncEnumerable<Test>(testStream, deserializationOptions);

        var actualCount = 1;
        var stopwatch = Stopwatch.StartNew();
        await foreach (var testElement in testElements)
        {
            Console.WriteLine($"[{actualCount}][Time elapsed: " + stopwatch.Elapsed.TotalSeconds + "; Data: " + testElement?.Data + "]");
            actualCount += 1;
        }
    }

    private static async Task WriteJsonElementsPeriodicallyToStreamAsync(SimplexStream testStream)
    {
        var streamWriter = new StreamWriter(testStream) { AutoFlush = true };

        var count = 1;
        var maxElements = 10;
        var delayInSeconds = 2;

        await streamWriter.WriteAsync($"[{{\"Data\":\"This will be printed after ~{delayInSeconds * count} seconds; Expected: {delayInSeconds * (count - 1)}\"}}");
        count += 1;

        while (count <= maxElements)
        {
            await Task.Delay(TimeSpan.FromSeconds(delayInSeconds));

            await streamWriter.WriteAsync($",{{\"Data\":\"This will be printed after ~{delayInSeconds * count} seconds; Expected: {delayInSeconds * (count - 1)}\"}}");
            count += 1;
        }

        await streamWriter.WriteAsync($"]");

        var preventFromLastMessageToBeEmitted = true;
        if (!preventFromLastMessageToBeEmitted)
        {
            testStream.CompleteWriting();
        }
    }

    private class Test
    {
        public string? Data { get; init; }
    }
}

Expected behavior

When writing the first JSON element (as string) to the stream, the IAsyncEnumerable will receive its first deserialized element in the enumeration, unblocking the enumeration.

Actual behavior

When writing the first JSON element (as string) to the stream, the IAsyncEnumerable is yet to receive the first deserialized element. The first deserialized element will only be received after the second JSON element (as string) will be written to the stream. Consequently, the last message in the array will not be emitted unless the stream is closed, although it contains the necessary data to deserialize the object.

Regression?

No response

Known Workarounds

As a workaround in my particular case, I am using Newtonsoft.Json's JsonTextReader.ReadAsync to asynchronously read JSON tokens from the stream, and deserializing manually to the required object. This way the code works as expected.

Configuration

I am using .NET 6, Windows 11, x64

Other information

No response

Author:	ohadvano
Assignees:	-
Labels:	area-System.Text.Json, untriaged
Milestone:	-

[eiriktsarpalis]

The root cause of this behavior is the current implementation of the internal ReadFromStreamAsync method. The method will keep reading from the stream until either the internal buffer is filled or the stream EOF's: this was intended as a performance optimization to avoid redundant partial deserialization passes, but it is clear that it could impact liveness specifically for the case of DeserializeAsyncEnumerable.

We could try tweaking this method slightly specifically for DeserializeAsyncMethod, possibly trading off deserialization performance for decreased delays in element reads.