22

Lets say that I request a large json file that contains a list of many objects. I don't want them to be in memory all at once, but I would rather read and process them one by one. So I need to turn an async System.IO.Stream stream into an IAsyncEnumerable<T>. How do I use the new System.Text.Json API to do this?

private async IAsyncEnumerable<T> GetList<T>(Uri url, CancellationToken cancellationToken = default)
{
    using (var httpResponse = await httpClient.GetAsync(url, cancellationToken))
    {
        using (var stream = await httpResponse.Content.ReadAsStreamAsync())
        {
            // Probably do something with JsonSerializer.DeserializeAsync here without serializing the entire thing in one go
        }
    }
}
6
  • 1
    You'll probably need something like DeserializeAsync method Commented Oct 26, 2019 at 16:30
  • 2
    Sorry, it seems that method above loads the entire stream into the memory. You can read the data by chunks asynchonously using Utf8JsonReader, please have a look at some github samples and at existing thread as well Commented Oct 26, 2019 at 21:20
  • GetAsync by itself returns when the entire response is received. You need to use SendAsync with ` HttpCompletionOption.ResponseContentRead` instead. Once you have that you can use JSON.NET's JsonTextReader. Using System.Text.Json for this isn't that easy as this issue shows. The functionality isn't available and implementing it in a low-allocation using structs isn't trivial Commented Oct 29, 2019 at 9:30
  • The problem with deserializing in chunks is that you have to know when you have a complete chunk to deserialize. This would be difficult to cleanly accomplish for general cases. It would require parsing in advance, which could be quite a poor tradeoff in terms of performance. It would rather difficult to generalize. But if you enforce your own restrictions on your JSON, say "a single object occupies exactly 20 lines in the file", then you could essentially deserialize asynchronously by reading the file in chunks async. You would need a massive json to see benefit here though, I would imagine. Commented Oct 29, 2019 at 14:48
  • Looks like someone already answered a similar question here with full code. Commented Oct 30, 2019 at 8:13

7 Answers 7

18

TL;DR It's not trivial


Looks like someone already posted full code for a Utf8JsonStreamReader struct that reads buffers from a stream and feeds them to a Utf8JsonRreader, allowing easy deserialization with JsonSerializer.Deserialize<T>(ref newJsonReader, options);. The code isn't trivial either. The related question is here and the answer is here.

That's not enough though - HttpClient.GetAsync will return only after the entire response is received, essentially buffering everything in memory.

To avoid this, HttpClient.GetAsync(string,HttpCompletionOption ) should be used with HttpCompletionOption.ResponseHeadersRead.

The deserialization loop should check the cancellation token too, and either exit or throw if it's signalled. Otherwise the loop will go on until the entire stream is received and processed.

This code is based in the related answer's example and uses HttpCompletionOption.ResponseHeadersRead and checks the cancellation token. It can parse JSON strings that contain a proper array of items, eg :

[{"prop1":123},{"prop1":234}]

The first call to jsonStreamReader.Read() moves to the start of the array while the second moves to the start of the first object. The loop itself terminates when the end of the array (]) is detected.

private async IAsyncEnumerable<T> GetList<T>(Uri url, CancellationToken cancellationToken = default)
{
    //Don't cache the entire response
    using var httpResponse = await httpClient.GetAsync(url,                               
                                                       HttpCompletionOption.ResponseHeadersRead,  
                                                       cancellationToken);
    using var stream = await httpResponse.Content.ReadAsStreamAsync();
    using var jsonStreamReader = new Utf8JsonStreamReader(stream, 32 * 1024);

    jsonStreamReader.Read(); // move to array start
    jsonStreamReader.Read(); // move to start of the object

    while (jsonStreamReader.TokenType != JsonTokenType.EndArray)
    {
        //Gracefully return if cancellation is requested.
        //Could be cancellationToken.ThrowIfCancellationRequested()
        if(cancellationToken.IsCancellationRequested)
        {
            return;
        }

        // deserialize object
        var obj = jsonStreamReader.Deserialize<T>();
        yield return obj;

        // JsonSerializer.Deserialize ends on last token of the object parsed,
        // move to the first token of next object
        jsonStreamReader.Read();
    }
}

JSON fragments, AKA streaming JSON aka ...*

It's quite common in event streaming or logging scenarios to append individual JSON objects to a file, one element per line eg :

{"eventId":1}
{"eventId":2}
...
{"eventId":1234567}

This isn't a valid JSON document but the individual fragments are valid. This has several advantages for big data/highly concurrent scenarios. Adding a new event only requires appending a new line to the file, not parsing and rebuilding the entire file. Processing, especially parallel processing is easier for two reasons:

  • Individual elements can be retrieved one at a time, simply by reading one line from a stream.
  • The input file can be easily partitioned and split across line boundaries, feeding each part to a separate worker process, eg in a Hadoop cluster, or simply different threads in an application: Calculate the split points eg by dividing the length by the number of workers, then look for the first newline. Feed everything up to that point to a separate worker.

Using a StreamReader

The allocate-y way to do this would be to use a TextReader, read one line at a time and parse it with JsonSerializer.Deserialize :

using var reader=new StreamReader(stream);
string line;
//ReadLineAsync() doesn't accept a CancellationToken 
while((line=await reader.ReadLineAsync()) != null)
{
    var item=JsonSerializer.Deserialize<T>(line);
    yield return item;

    if(cancellationToken.IsCancellationRequested)
    {
        return;
    }
}

That's a lot simpler than the code that deserializes a proper array. There are two issues :

  • ReadLineAsync doesn't accept a cancellation token
  • Each iteration allocates a new string, one of the things we wanted to avoid by using System.Text.Json

This may be enough though as trying to produce the ReadOnlySpan<Byte> buffers needed by JsonSerializer.Deserialize isn't trivial.

Pipelines and SequenceReader

To avoid alllocations, we need to get a ReadOnlySpan<byte> from the stream. Doing this requires using System.IO.Pipeline pipes and the SequenceReader struct. Steve Gordon's An Introduction to SequenceReader explains how this class can be used to read data from a stream using delimiters.

Unfortunately, SequenceReader is a ref struct which means it can't be used in async or local methods. That's why Steve Gordon in his article creates a

private static SequencePosition ReadItems(in ReadOnlySequence<byte> sequence, bool isCompleted)

method to read items form a ReadOnlySequence and return the ending position, so the PipeReader can resume from it. Unfortunately we want to return an IEnumerable or IAsyncEnumerable, and iterator methods don't like in or out parameters either.

We could collect the deserialized items in a List or Queue and return them as a single result, but that would still allocate lists, buffers or nodes and have to wait for all items in a buffer to be deserialized before returning :

private static (SequencePosition,List<T>) ReadItems(in ReadOnlySequence<byte> sequence, bool isCompleted)

We need something that acts like an enumerable without requiring an iterator method, works with async and doesn't buffer everything the way.

Adding Channels to produce an IAsyncEnumerable

ChannelReader.ReadAllAsync returns an IAsyncEnumerable. We can return a ChannelReader from methods that couldn't work as iterators and still produce a stream of elements without caching.

Adapting Steve Gordon's code to use channels, we get the ReadItems(ChannelWriter...) and ReadLastItem methods. The first one, reads one item at a time, up to a newline using ReadOnlySpan<byte> itemBytes. This can be used by JsonSerializer.Deserialize. If ReadItems can't find the delimiter, it returns its position so the PipelineReader can pull the next chunk from the stream.

When we reach the last chunk and there's no other delimiter, ReadLastItem` reads the remaining bytes and deserializes them.

The code is almost identical to Steve Gordon's. Instead of writing to the Console, we write to the ChannelWriter.

private const byte NL=(byte)'\n';
private const int MaxStackLength = 128;

private static SequencePosition ReadItems<T>(ChannelWriter<T> writer, in ReadOnlySequence<byte> sequence, 
                          bool isCompleted, CancellationToken token)
{
    var reader = new SequenceReader<byte>(sequence);

    while (!reader.End && !token.IsCancellationRequested) // loop until we've read the entire sequence
    {
        if (reader.TryReadTo(out ReadOnlySpan<byte> itemBytes, NL, advancePastDelimiter: true)) // we have an item to handle
        {
            var item=JsonSerializer.Deserialize<T>(itemBytes);
            writer.TryWrite(item);            
        }
        else if (isCompleted) // read last item which has no final delimiter
        {
            var item = ReadLastItem<T>(sequence.Slice(reader.Position));
            writer.TryWrite(item);
            reader.Advance(sequence.Length); // advance reader to the end
        }
        else // no more items in this sequence
        {
            break;
        }
    }

    return reader.Position;
}

private static T ReadLastItem<T>(in ReadOnlySequence<byte> sequence)
{
    var length = (int)sequence.Length;

    if (length < MaxStackLength) // if the item is small enough we'll stack allocate the buffer
    {
        Span<byte> byteBuffer = stackalloc byte[length];
        sequence.CopyTo(byteBuffer);
        var item=JsonSerializer.Deserialize<T>(byteBuffer);
        return item;        
    }
    else // otherwise we'll rent an array to use as the buffer
    {
        var byteBuffer = ArrayPool<byte>.Shared.Rent(length);

        try
        {
            sequence.CopyTo(byteBuffer);
            var item=JsonSerializer.Deserialize<T>(byteBuffer);
            return item;
        }
        finally
        {
            ArrayPool<byte>.Shared.Return(byteBuffer);
        }

    }    
}

The DeserializeToChannel<T> method creates a Pipeline reader on top of the stream, creates a channel and starts a worker task that parses chunks and pushes them to the channel :

ChannelReader<T> DeserializeToChannel<T>(Stream stream, CancellationToken token)
{
    var pipeReader = PipeReader.Create(stream);    
    var channel=Channel.CreateUnbounded<T>();
    var writer=channel.Writer;
    _ = Task.Run(async ()=>{
        while (!token.IsCancellationRequested)
        {
            var result = await pipeReader.ReadAsync(token); // read from the pipe

            var buffer = result.Buffer;

            var position = ReadItems(writer,buffer, result.IsCompleted,token); // read complete items from the current buffer

            if (result.IsCompleted) 
                break; // exit if we've read everything from the pipe

            pipeReader.AdvanceTo(position, buffer.End); //advance our position in the pipe
        }

        pipeReader.Complete(); 
    },token)
    .ContinueWith(t=>{
        pipeReader.Complete();
        writer.TryComplete(t.Exception);
    });

    return channel.Reader;
}

ChannelReader.ReceiveAllAsync() can be used to consume all items through an IAsyncEnumerable<T>:

var reader=DeserializeToChannel<MyEvent>(stream,cts.Token);
await foreach(var item in reader.ReadAllAsync(cts.Token))
{
    //Do something with it 
}    
3
  • 1
    You've missed one thing unless I am mistaken. Utf8JsonStreamReader is a ref struct and thusly cannot be used within an async method. I see you've referenced this issue in other circumstances, but not regarding that initial example.
    – Shazi
    Commented Jun 13, 2020 at 20:26
  • Thanks for the detailed answer, @PanagiotisKanavos (and your deleted answer has been also helpful!). Surprising how JSON streaming is still so difficult with System.Text.Json. AFAICT, the new DeserializeAsyncEnumerable doesn't yet provide a viable solution for infinite streaming of JSON over a socked or a named pipe. It still tries to read the input stream to the end and won't produce any items until the end of the stream.
    – noseratio
    Commented Jan 17, 2022 at 5:02
  • I get "CS4012 Parameters or locals of type 'Utf8JsonStreamReader' cannot be declared in async methods or async lambda expressions." when trying to use it like you do in your first example. How did you get it to build like that?
    – fretje
    Commented Nov 10, 2023 at 12:15
7

Yes, a truly streaming JSON (de)serializer would be a nice performance improvement to have, in so many places.

Unfortunately, System.Text.Json does not do this at the time I'm writing this. I'm not sure if it will in the future - I hope so! Truly streaming deserialization of JSON turns out to be rather challenging.

You could check if the extremely fast Utf8Json supports it, perhaps.

However, there might be a custom solution for your specific situation, since your requirements seem to constrain the difficulty.

The idea is to manually read one item from the array at a time. We are making use of the fact that each item in the list is, in itself, a valid JSON object.

You can manually skip past the [ (for the first item) or the , (for each next item). Then I think your best bet is to use .NET Core's Utf8JsonReader to determine where the current object ends, and feed the scanned bytes to JsonDeserializer.

This way, you're only buffering slightly over one object at a time.

And since we're talking performance, you could get the input from a PipeReader, while you're at it. :-)

5
  • 1
    This isn't about performance at all. It's not about async deserialization, which it already does. It's about streaming access - processing JSON elements as they are parsed from the stream, the way JSON.NET's JsonTextReader does. Commented Oct 29, 2019 at 15:02
  • The relevant class in Utf8Json is JsonReader and as the author says, it's weird. JSON.NET's JsonTextReader and System.Text.Json's Utf8JsonReader share the same weirdness - you have to loop and check the current element's type as you go. Commented Oct 29, 2019 at 15:12
  • @PanagiotisKanavos Ah, yes, streaming. That's the word I was looking for! I'm updating the word "asynchronous" to "streaming". I do believe the reason to want streaming is limiting memory usage, which is a performance concern. Perhaps OP can confirm.
    – Timo
    Commented Oct 29, 2019 at 17:53
  • Performance doesn't mean speed. No matter how fast the deserializer is, if you have to process 1M items, you don't want to store them in RAM, nor wait for all of them to get deserialized before you can process the first one. Commented Oct 30, 2019 at 7:54
  • Semantics, my friend! I’m glad we’re trying to achieve the same thing after all.
    – Timo
    Commented Oct 30, 2019 at 9:36
6

I understand this is an old post, but the recently announced System.Text.Json support for IAsyncEnumerable in .Net 6 Preview 4 provides solution the problem mentioned in OP.

private async IAsyncEnumerable<T> GetList<T>(Uri url, CancellationToken cancellationToken = default)
{
    using (var httpResponse = await httpClient.GetAsync(url, cancellationToken))
    {
        using (var stream = await httpResponse.Content.ReadAsStreamAsync())
        {

            await foreach(var item in JsonSerializer.DeserializeAsyncEnumerable<T>(stream))
            {
                yield return item;
            }
        }
    }
}

This would provide an on-demand deserialization and is quite useful when working with large data. Please note that at the moment the feature is limited to root level JSON arrays.

More details on the feature could be found here

3
  • 1
    I tried using DeserializeAsyncEnumerable for streaming and ran into an issue, apparently it still tries to read the whole stream before producing any data: github.com/dotnet/runtime/issues/63864
    – noseratio
    Commented Jan 17, 2022 at 6:07
  • Same here. Is there any news?
    – Krusty
    Commented Sep 27, 2022 at 14:01
  • Streaming doesn't actually work in .NET 6 as previously mentioned, but does work correctly in .NET 7 (and assuming higher). Commented Oct 27, 2023 at 14:03
1

dotnet 8+

you can use HttpClientJsonExtensions var data = HttpClient.GetFromJsonAsAsyncEnumerable<YourResponseType>(url)

it will return IAsyncEnumerable<YourResponseType?>

which can be used by iterating over the result

await foreach (var item in response)
{
    //use it here or return
    yield return item
}
0

It feels like you need to implent your own stream reader. You have to read bytes one-by-one and stop as soon as object definition completed. It is indeed pretty low-leveled. As such you WILL NOT load entire file into RAM, but rather take the part you're dealing with. Does it seem to be an answer?

0

Instead of using multiple tasks with ChannelReader it's possible to use the System.IO.Pipelines extension package along with System.Text.Json.JsonSerializer in .NET 5 (C# 9) like this:

using System;
using System.Buffers;
using System.Collections.Generic;
using System.IO;
using System.IO.Pipelines;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;

class Program
{
    static readonly byte[] NewLineChars = {(byte)'\r', (byte)'\n'};
    static readonly byte[] WhiteSpaceChars = {(byte)'\r', (byte)'\n', (byte)' ', (byte)'\t'};

    private static async Task Main()
    {
        JsonSerializerOptions jsonOptions = new(JsonSerializerDefaults.Web);
        var json = "{\"some\":\"thing1\"}\r\n{\"some\":\"thing2\"}\r\n{\"some\":\"thing3\"}";
        var contentStream = new MemoryStream(Encoding.UTF8.GetBytes(json));
        var pipeReader = PipeReader.Create(contentStream);
        await foreach (var foo in ReadItemsAsync<Foo>(pipeReader, jsonOptions))
        {
            Console.WriteLine($"foo: {foo.Some}");
        }
    }

    static async IAsyncEnumerable<TValue> ReadItemsAsync<TValue>(PipeReader pipeReader, JsonSerializerOptions jsonOptions = null)
    {
        while (true)
        {
            var result = await pipeReader.ReadAsync();
            var buffer = result.Buffer;
            bool isCompleted = result.IsCompleted;
            SequencePosition bufferPosition = buffer.Start;
            while (true)
            {
                var(value, advanceSequence) = TryReadNextItem<TValue>(buffer, ref bufferPosition, isCompleted, jsonOptions);
                if (value != null)
                {
                    yield return value;
                }

                if (advanceSequence)
                {
                    pipeReader.AdvanceTo(bufferPosition, buffer.End); //advance our position in the pipe
                    break;
                }
            }

            if (isCompleted)
                yield break;
        }
    }

    static (TValue, bool) TryReadNextItem<TValue>(ReadOnlySequence<byte> sequence, ref SequencePosition sequencePosition, bool isCompleted, JsonSerializerOptions jsonOptions)
    {
        var reader = new SequenceReader<byte>(sequence.Slice(sequencePosition));
        while (!reader.End) // loop until we've come to the end or read an item
        {
            if (reader.TryReadToAny(out ReadOnlySpan<byte> itemBytes, NewLineChars, advancePastDelimiter: true))
            {
                sequencePosition = reader.Position;
                if (itemBytes.TrimStart(WhiteSpaceChars).IsEmpty)
                {
                    continue;
                }

                return (JsonSerializer.Deserialize<TValue>(itemBytes, jsonOptions), false);
            }
            else if (isCompleted)
            {
                // read last item
                var remainingReader = sequence.Slice(reader.Position);
                using var memoryOwner = MemoryPool<byte>.Shared.Rent((int)reader.Remaining);
                remainingReader.CopyTo(memoryOwner.Memory.Span);
                reader.Advance(remainingReader.Length); // advance reader to the end
                sequencePosition = reader.Position;
                if (!itemBytes.TrimStart(WhiteSpaceChars).IsEmpty)
                {
                    return (JsonSerializer.Deserialize<TValue>(memoryOwner.Memory.Span, jsonOptions), true);
                }
                else
                {
                    return (default, true);
                }
            }
            else
            {
                // no more items in sequence
                break;
            }
        }

        // PipeReader needs to read more
        return (default, true);
    }
}

public class Foo
{
    public string Some
    {
        get;
        set;
    }
}

Run at https://dotnetfiddle.net/6j3KGg

-3

Maybe you could use Newtonsoft.Json serializer ? https://www.newtonsoft.com/json/help/html/Performance.htm

Especially see section:

Optimize Memory Usage

Edit

You could try deserializing values from JsonTextReader, e.g.

using (var textReader = new StreamReader(stream))
using (var reader = new JsonTextReader(textReader))
{
    while (await reader.ReadAsync(cancellationToken))
    {
        yield return reader.Value;
    }
}
4
  • That doesn't answer the question. This isn't about performance at all, it's about streaming access without loading everything in memory Commented Oct 29, 2019 at 15:01
  • Have you opened the related link or just said what you think? In the link I have sent in the section I mentioned there is a code snippet of how to deserialize JSON from stream. Commented Oct 29, 2019 at 15:07
  • Read the question again please - the OP asks how to process the elements without deserializing everything in memory. Not just read from a stream, but only process what comes from the stream. I don't want them to be in memory all at once, but I would rather read and process them one by one. The relevant class in JSON.NET is JsonTextReader. Commented Oct 29, 2019 at 15:10
  • In any case, a link-only answer isn't considered a good answer, and nothing in that link answers the OP's question. A link to JsonTextReader would be better Commented Oct 29, 2019 at 15:12

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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