Repository navigation
[System.Text.Json] DeserializeAsyncEnumerable is delayed emitting objects when writing asynchronously to stream #70107
Description
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.
Activity
eiriktsarpalis commentedon Jun 7, 2022
The root cause of this behavior is the current implementation of the internal
ReadFromStreamAsyncmethod. 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 ofDeserializeAsyncEnumerable.We could try tweaking this method slightly specifically for
DeserializeAsyncMethod, possibly trading off deserialization performance for decreased delays in element reads.