Improve JsonSerializer.ReadAsync(Stream) throughput

[stephentoub]

Some low hanging fruit:

• The method was using the Stream.ReadAsync overload that returns a Task<int>. Changed it to take a Memory<byte> so that it instead returns a ValueTask<int>.
• The method was clearing the full rented buffer upon returning it to the pool, even when only using a small portion of it. Changed it to only clear what was used.
• The method was resizing the buffer unnecessarily due to faulty logic around how much data remained. Fixed it to only resize when more than half the buffer is full, which was the original intention.
• The ReadCore method is a bit chunky to call, initializing a new Utf8JsonReader each time, copying large structs, etc. Since we need to read all of the data from the stream anyway, I changed it to try to fill the buffer, which then minimizes the number of times we need to call ReadCore, in particular avoiding the extra empty call at the end. We can tweak this further in the future as needed, e.g. only making a second attempt to fill the buffer rather than doing so aggressively.
• Also fixed the exception to contain the full amount of data read from the stream, not just from the most recent call.

Before:

                            Method |     Mean |     Error |    StdDev |  Gen 0 |  Gen 1 | Allocated |
---------------------------------- |---------:|----------:|----------:|-------:|-------:|----------:|
       ReadSimpleClassMemoryStream | 4.678 us | 0.0380 us | 0.0317 us | 0.1526 |      - |     328 B |
 ReadSimpleClassPipeReaderAsStream | 5.411 us | 0.0542 us | 0.0453 us | 0.0916 | 0.0076 |     256 B |

After:

                            Method |     Mean |     Error |    StdDev |  Gen 0 |  Gen 1 | Allocated |
---------------------------------- |---------:|----------:|----------:|-------:|-------:|----------:|
       ReadSimpleClassMemoryStream | 3.265 us | 0.0997 us | 0.0884 us | 0.0801 | 0.0038 |     184 B |
 ReadSimpleClassPipeReaderAsStream | 3.922 us | 0.0754 us | 0.0668 us | 0.0534 | 0.0076 |     184 B |

Benchmark (example SimpleTestClass borrowed from @ahsonkhan):

using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Attributes.Jobs;
using BenchmarkDotNet.Running;
using System;
using System.IO;
using System.IO.Pipelines;
using System.Text;
using System.Text.Json.Serialization;
using System.Threading.Tasks;

[MemoryDiagnoser]
[InProcess]
public class Benchmark
{
    private static void Main() => BenchmarkRunner.Run<Benchmark>();

    [Benchmark]
    public async Task ReadSimpleClassMemoryStream()
    {
        _memoryStream.SetLength(0);
        await _memoryStream.WriteAsync(_dataUtf8, 0, _dataUtf8.Length);
        _memoryStream.Position = 0;

        await JsonSerializer.ReadAsync<SimpleTestClass>(_memoryStream);
    }

    [Benchmark]
    public async Task ReadSimpleClassPipeReaderAsStream()
    {
        await _pipe.Writer.WriteAsync(_dataUtf8);
        _pipe.Writer.Complete();

        await JsonSerializer.ReadAsync<SimpleTestClass>(_pipe.Reader.AsStream());

        _pipe.Reader.Complete();
        _pipe.Reset();
    }

    public enum SampleEnum
    {
        One = 1,
        Two = 2
    }

    public class SimpleTestClass
    {
        public short MyInt16 { get; set; }
        public int MyInt32 { get; set; }
        public long MyInt64 { get; set; }
        public ushort MyUInt16 { get; set; }
        public uint MyUInt32 { get; set; }
        public ulong MyUInt64 { get; set; }
        public byte MyByte { get; set; }
        public char MyChar { get; set; }
        public string MyString { get; set; }
        public decimal MyDecimal { get; set; }
        public bool MyBooleanTrue { get; set; }
        public bool MyBooleanFalse { get; set; }
        public float MySingle { get; set; }
        public double MyDouble { get; set; }
        public DateTime MyDateTime { get; set; }
        public SampleEnum MyEnum { get; set; }
    }

    private readonly byte[] _dataUtf8 = Encoding.UTF8.GetBytes(
        @"{" +
        @"""MyInt16"" : 1," +
        @"""MyInt32"" : 2," +
        @"""MyInt64"" : 3," +
        @"""MyUInt16"" : 4," +
        @"""MyUInt32"" : 5," +
        @"""MyUInt64"" : 6," +
        @"""MyByte"" : 7," +
        @"""MyChar"" : ""a""," +
        @"""MyString"" : ""Hello""," +
        @"""MyBooleanTrue"" : true," +
        @"""MyBooleanFalse"" : false," +
        @"""MySingle"" : 1.1," +
        @"""MyDouble"" : 2.2," +
        @"""MyDecimal"" : 3.3," +
        @"""MyDateTime"" : ""2019-01-30T12:01:02.0000000Z""," +
        @"""MyEnum"" : 2" + // int by default
        @"}");
    private readonly MemoryStream _memoryStream = new MemoryStream();
    private readonly Pipe _pipe = new Pipe();
}

cc: @ahsonkhan, @steveharter, @bartonjs, @davidfowl, @jkotas

[Tornhoof]

If you're optimizing for MemoryStream cases, you could also look at TryGetBuffer. It needs to be allowed specifically during creation of the memorystream, but then something like

if (stream is MemoryStream ms && ms.TryGetBuffer(out var buffer))
{
// use buffer directly
 var span = new ReadOnlySpan<byte>(buffer.Array, buffer.Offset, buffer.Count);
// use span based utf8json overload.
}

works.

[ahsonkhan]

Otherwise, LGTM.

[ahsonkhan]

I asked this before (#35609 (comment)), but

Renting int.MaxValue would throw OOM. Should we cap it to something smaller (either 2GB, or maybe int.MaxValue - 56)?

Is there guidelines or common pattern to follow in such scenarios?

[davidfowl]

Nice!

[stephentoub]

If you're optimizing for MemoryStream cases, you could also look at TryGetBuffer. It needs to be allowed specifically during creation of the memorystream, but then something like

Yup, that might be reasonable. There's a cost associated with special-casing like that, and a caller could just do it themselves, but if we think MemoryStream would commonly be used here, it'd be reasonable (I don't know if it would be). I was using it here just as a data point to compare against.