diff --git a/src/LightProto/PooledSegmentBufferWriter.cs b/src/LightProto/PooledSegmentBufferWriter.cs new file mode 100644 index 0000000..d252c8c --- /dev/null +++ b/src/LightProto/PooledSegmentBufferWriter.cs @@ -0,0 +1,193 @@ +using System.Buffers; +using System.Collections.Concurrent; + +namespace LightProto +{ + /// + /// A pooled, segmented buffer used to bridge synchronous codecs with asynchronous stream I/O. + /// + internal sealed class PooledSegmentBufferWriter : IBufferWriter + { + private const int MinimumSegmentSize = 4096; + + private static readonly ConcurrentBag WriterPool = new(); + private static readonly ConcurrentBag SegmentPool = new(); + + private Segment? first; + private Segment? current; + + private PooledSegmentBufferWriter() { } + + public static PooledSegmentBufferWriter Rent() + { + if (WriterPool.TryTake(out var writer)) + { + return writer; + } + + return new PooledSegmentBufferWriter(); + } + + public static void Return(PooledSegmentBufferWriter writer) + { + writer.Reset(); + WriterPool.Add(writer); + } + + public void Advance(int count) + { + if (current is null) + { + if (count == 0) + { + return; + } + + throw new InvalidOperationException("No buffer has been requested."); + } + + var segment = current; + if ((uint)count > (uint)(segment.Buffer.Length - segment.Written)) + { + throw new ArgumentOutOfRangeException(nameof(count)); + } + + segment.Written += count; + } + + public Memory GetMemory(int sizeHint = 0) + { + var segment = GetWritableSegment(sizeHint); + return segment.Buffer.AsMemory(segment.Written); + } + + public Span GetSpan(int sizeHint = 0) + { + var segment = GetWritableSegment(sizeHint); + return segment.Buffer.AsSpan(segment.Written); + } + + public async Task ReadToEndAsync(Stream source, CancellationToken cancellationToken) + { + while (true) + { + var segment = GetWritableSegment(MinimumSegmentSize); + var read = await source + .ReadAsync(segment.Buffer, segment.Written, segment.Buffer.Length - segment.Written, cancellationToken) + .ConfigureAwait(false); + if (read == 0) + { + return; + } + + segment.Written += read; + } + } + + public async Task WriteToAsync(Stream destination, CancellationToken cancellationToken) + { + for (var segment = first; segment is not null; segment = segment.NextSegment) + { + if (segment.Written == 0) + { + continue; + } + + await destination.WriteAsync(segment.Buffer, 0, segment.Written, cancellationToken).ConfigureAwait(false); + } + } + + public ReadOnlySequence GetReadOnlySequence() + { + if (first is null || current is null) + { + return ReadOnlySequence.Empty; + } + + for (var segment = first; segment is not null; segment = segment.NextSegment) + { + segment.SetMemory(); + } + + return new ReadOnlySequence(first, 0, current, current.Written); + } + + private Segment GetWritableSegment(int sizeHint) + { + if (sizeHint < 0) + { + throw new ArgumentOutOfRangeException(nameof(sizeHint)); + } + + if (current is null || current.Buffer.Length - current.Written < Math.Max(sizeHint, 1)) + { + var segment = RentSegment(Math.Max(sizeHint, MinimumSegmentSize)); + if (current is null) + { + first = segment; + } + else + { + segment.SetRunningIndex(current.RunningIndex + current.Written); + current.NextSegment = segment; + } + + current = segment; + } + + return current; + } + + private void Reset() + { + var segment = first; + first = null; + current = null; + + while (segment is not null) + { + var next = segment.NextSegment; + segment.Reset(); + SegmentPool.Add(segment); + segment = next; + } + } + + private static Segment RentSegment(int minimumSize) + { + if (!SegmentPool.TryTake(out var segment)) + { + segment = new Segment(); + } + + segment.Buffer = ArrayPool.Shared.Rent(minimumSize); + return segment; + } + + private sealed class Segment : ReadOnlySequenceSegment + { + public byte[] Buffer = null!; + public int Written; + + public Segment? NextSegment + { + get => (Segment?)Next; + set => Next = value; + } + + public void SetMemory() => Memory = Buffer.AsMemory(0, Written); + + public void SetRunningIndex(long runningIndex) => RunningIndex = runningIndex; + + public void Reset() + { + ArrayPool.Shared.Return(Buffer); + Buffer = null!; + Written = 0; + Memory = default; + Next = null; + RunningIndex = 0; + } + } + } +} diff --git a/src/LightProto/PublicAPI/net/PublicAPI.Unshipped.txt b/src/LightProto/PublicAPI/net/PublicAPI.Unshipped.txt index 2904c4b..14a2d0f 100644 --- a/src/LightProto/PublicAPI/net/PublicAPI.Unshipped.txt +++ b/src/LightProto/PublicAPI/net/PublicAPI.Unshipped.txt @@ -1,4 +1,15 @@ #nullable enable +static LightProto.Serializer.DeserializeAsync(System.IO.Stream! source, LightProto.IProtoReader! reader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.DeserializeAsync(System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.DeserializeDynamicallyAsync(System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.DeserializeNonGenericAsync(System.IO.Stream! source, LightProto.IProtoReader! reader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.DeserializeNonGenericAsync(System.Type! type, System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeAsync(System.IO.Stream! destination, T instance, LightProto.IProtoWriter! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeAsync(System.IO.Stream! destination, T instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeDynamicallyAsync(System.IO.Stream! destination, T instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeNonGenericAsync(System.IO.Stream! destination, object? instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeNonGenericAsync(System.IO.Stream! destination, object? instance, LightProto.IProtoWriter! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeToAsync(this T instance, System.IO.Stream! destination, LightProto.IProtoWriter! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! LightProto.IProtoWriter.CalculateLongSize(object! value) -> long LightProto.IProtoWriter.CalculateLongSize(T value) -> long LightProto.PackedRepeatedOptimizer diff --git a/src/LightProto/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt b/src/LightProto/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt index cc7beb4..d902fdb 100644 --- a/src/LightProto/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt +++ b/src/LightProto/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt @@ -1,4 +1,13 @@ #nullable enable +static LightProto.Serializer.DeserializeAsync(System.IO.Stream! source, LightProto.IProtoReader! reader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.DeserializeDynamicallyAsync(System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.DeserializeNonGenericAsync(System.IO.Stream! source, LightProto.IProtoReader! reader, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.DeserializeNonGenericAsync(System.Type! type, System.IO.Stream! source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeAsync(System.IO.Stream! destination, T instance, LightProto.IProtoWriter! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeDynamicallyAsync(System.IO.Stream! destination, T instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeNonGenericAsync(System.IO.Stream! destination, object? instance, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeNonGenericAsync(System.IO.Stream! destination, object? instance, LightProto.IProtoWriter! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +static LightProto.Serializer.SerializeToAsync(this T instance, System.IO.Stream! destination, LightProto.IProtoWriter! writer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! LightProto.IProtoWriter.CalculateLongSize(object! value) -> long LightProto.IProtoWriter.CalculateLongSize(T value) -> long LightProto.PackedRepeatedOptimizer diff --git a/src/LightProto/Serializer.Async.cs b/src/LightProto/Serializer.Async.cs new file mode 100644 index 0000000..21ddaec --- /dev/null +++ b/src/LightProto/Serializer.Async.cs @@ -0,0 +1,90 @@ +using System.Runtime.CompilerServices; + +namespace LightProto +{ +#pragma warning disable RS0026 // CancellationToken is intentionally optional for the asynchronous API surface. + public static partial class Serializer + { + /// + /// Asynchronously writes a fully serialized protocol-buffer message to the supplied stream. + /// The message is encoded synchronously into a pooled buffer before asynchronous I/O begins. + /// + public static async Task SerializeAsync( + Stream destination, + T instance, + IProtoWriter writer, + CancellationToken cancellationToken = default + ) + { + if (destination is null) + { + throw new ArgumentNullException(nameof(destination)); + } + if (writer is null) + { + throw new ArgumentNullException(nameof(writer)); + } + + var buffer = PooledSegmentBufferWriter.Rent(); + try + { + Serialize(buffer, instance, writer); + await buffer.WriteToAsync(destination, cancellationToken).ConfigureAwait(false); + } + finally + { + PooledSegmentBufferWriter.Return(buffer); + } + } + + /// + /// Asynchronously reads a protocol-buffer message from the supplied stream until the end of the stream. + /// The accumulated message is decoded synchronously after all input has been received. + /// + public static async Task DeserializeAsync( + Stream source, + IProtoReader reader, + CancellationToken cancellationToken = default + ) + { + if (source is null) + { + throw new ArgumentNullException(nameof(source)); + } + if (reader is null) + { + throw new ArgumentNullException(nameof(reader)); + } + + var buffer = PooledSegmentBufferWriter.Rent(); + try + { + await buffer.ReadToEndAsync(source, cancellationToken).ConfigureAwait(false); + return Deserialize(buffer.GetReadOnlySequence(), reader); + } + finally + { + PooledSegmentBufferWriter.Return(buffer); + } + } + +#if NET7_0_OR_GREATER + /// + /// Asynchronously writes a fully serialized protocol-buffer message to the supplied stream. + /// The message is encoded synchronously into a pooled buffer before asynchronous I/O begins. + /// + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static Task SerializeAsync(Stream destination, T instance, CancellationToken cancellationToken = default) + where T : IProtoParser => SerializeAsync(destination, instance, T.ProtoWriter, cancellationToken); + + /// + /// Asynchronously reads a protocol-buffer message from the supplied stream until the end of the stream. + /// The accumulated message is decoded synchronously after all input has been received. + /// + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static Task DeserializeAsync(Stream source, CancellationToken cancellationToken = default) + where T : IProtoParser => DeserializeAsync(source, T.ProtoReader, cancellationToken); +#endif + } +#pragma warning restore RS0026 +} diff --git a/src/LightProto/Serializer.Dynamically.cs b/src/LightProto/Serializer.Dynamically.cs index 8735923..b7b9abb 100644 --- a/src/LightProto/Serializer.Dynamically.cs +++ b/src/LightProto/Serializer.Dynamically.cs @@ -43,6 +43,20 @@ public static void SerializeDynamically< #endif T>(Stream destination, T instance) => Serialize(destination, instance, GetProtoWriter()); + /// + /// Asynchronously writes a fully serialized protocol-buffer message to the supplied stream. + /// The message is encoded synchronously into a pooled buffer before asynchronous I/O begins. + /// +#if NET7_0_OR_GREATER + [RequiresDynamicCode(AOTWarning)] +#endif + public static Task SerializeDynamicallyAsync< +#if NET7_0_OR_GREATER + [DynamicallyAccessedMembers(LightProtoRequiredMembers)] +#endif + T>(Stream destination, T instance, CancellationToken cancellationToken = default) => + SerializeAsync(destination, instance, GetProtoWriter(), cancellationToken); + /// /// Creates a new instance from a protocol-buffer stream /// @@ -58,6 +72,20 @@ public static T DeserializeDynamically< #endif T>(Stream source) => Deserialize(source, GetProtoReader()); + /// + /// Asynchronously reads a protocol-buffer message from the supplied stream until the end of the stream. + /// The accumulated message is decoded synchronously after all input has been received. + /// +#if NET7_0_OR_GREATER + [RequiresDynamicCode(AOTWarning)] +#endif + public static Task DeserializeDynamicallyAsync< +#if NET7_0_OR_GREATER + [DynamicallyAccessedMembers(LightProtoRequiredMembers)] +#endif + T>(Stream source, CancellationToken cancellationToken = default) => + DeserializeAsync(source, GetProtoReader(), cancellationToken); + /// /// Serializes the given message to a byte array. /// diff --git a/src/LightProto/Serializer.Extensions.cs b/src/LightProto/Serializer.Extensions.cs index 2eea418..e472c07 100644 --- a/src/LightProto/Serializer.Extensions.cs +++ b/src/LightProto/Serializer.Extensions.cs @@ -178,6 +178,17 @@ public static byte[] ToByteArray(this T message, IProtoWriter writer) public static void SerializeTo(this T instance, Stream destination, IProtoWriter writer) => Serialize(destination, instance, writer); + /// + /// Asynchronously serializes the instance to the given destination stream. + /// The instance is encoded synchronously into a pooled buffer before asynchronous I/O begins. + /// + public static Task SerializeToAsync( + this T instance, + Stream destination, + IProtoWriter writer, + CancellationToken cancellationToken = default + ) => SerializeAsync(destination, instance, writer, cancellationToken); + /// /// Serializes the instance to the given destination buffer. /// diff --git a/src/LightProto/Serializer.NonGeneric.cs b/src/LightProto/Serializer.NonGeneric.cs index 00c6c56..7fad0c7 100644 --- a/src/LightProto/Serializer.NonGeneric.cs +++ b/src/LightProto/Serializer.NonGeneric.cs @@ -4,6 +4,7 @@ namespace LightProto { +#pragma warning disable RS0026 // CancellationToken is intentionally optional for the asynchronous API surface. public static partial class Serializer { /// @@ -42,6 +43,60 @@ public static void SerializeNonGeneric(Stream destination, object? instance, IPr ctx.Flush(); } + /// + /// Asynchronously writes a fully serialized protocol-buffer message to the supplied stream. + /// The message is encoded synchronously into a pooled buffer before asynchronous I/O begins. + /// +#if NET7_0_OR_GREATER + [RequiresDynamicCode(AOTWarning)] + [RequiresUnreferencedCode(AOTWarning)] +#endif + public static Task SerializeNonGenericAsync(Stream destination, object? instance, CancellationToken cancellationToken = default) + { + if (instance is null) + { + return Task.CompletedTask; + } + + return SerializeNonGenericAsync(destination, instance, GetProtoWriter(instance.GetType()), cancellationToken); + } + + /// + /// Asynchronously writes a fully serialized protocol-buffer message to the supplied stream. + /// The message is encoded synchronously into a pooled buffer before asynchronous I/O begins. + /// + public static async Task SerializeNonGenericAsync( + Stream destination, + object? instance, + IProtoWriter writer, + CancellationToken cancellationToken = default + ) + { + if (destination is null) + { + throw new ArgumentNullException(nameof(destination)); + } + if (writer is null) + { + throw new ArgumentNullException(nameof(writer)); + } + if (instance is null) + { + return; + } + + var buffer = PooledSegmentBufferWriter.Rent(); + try + { + SerializeNonGeneric(buffer, instance, writer); + await buffer.WriteToAsync(destination, cancellationToken).ConfigureAwait(false); + } + finally + { + PooledSegmentBufferWriter.Return(buffer); + } + } + /// /// Writes a protocol-buffer representation of the given instance to the supplied writer. /// @@ -148,6 +203,53 @@ public static object DeserializeNonGeneric(Stream source, IProtoReader reader) return reader.ParseFrom(ref ctx); } + /// + /// Asynchronously reads a protocol-buffer message from the supplied stream until the end of the stream. + /// The accumulated message is decoded synchronously after all input has been received. + /// +#if NET7_0_OR_GREATER + [RequiresDynamicCode(AOTWarning)] +#endif + public static Task DeserializeNonGenericAsync( +#if NET7_0_OR_GREATER + [DynamicallyAccessedMembers(LightProtoRequiredMembers)] +#endif + Type type, + Stream source, + CancellationToken cancellationToken = default + ) => DeserializeNonGenericAsync(source, GetProtoReader(type), cancellationToken); + + /// + /// Asynchronously reads a protocol-buffer message from the supplied stream until the end of the stream. + /// The accumulated message is decoded synchronously after all input has been received. + /// + public static async Task DeserializeNonGenericAsync( + Stream source, + IProtoReader reader, + CancellationToken cancellationToken = default + ) + { + if (source is null) + { + throw new ArgumentNullException(nameof(source)); + } + if (reader is null) + { + throw new ArgumentNullException(nameof(reader)); + } + + var buffer = PooledSegmentBufferWriter.Rent(); + try + { + await buffer.ReadToEndAsync(source, cancellationToken).ConfigureAwait(false); + return DeserializeNonGeneric(buffer.GetReadOnlySequence(), reader); + } + finally + { + PooledSegmentBufferWriter.Return(buffer); + } + } + /// /// Creates a new instance from a protocol-buffer stream /// @@ -242,4 +344,5 @@ public static IProtoWriter GetProtoWriter( #endif Type type) => (IProtoWriter)GetProtoParser(type, isReader: false); } +#pragma warning restore RS0026 } diff --git a/tests/LightProto.Tests/SerializerTests.cs b/tests/LightProto.Tests/SerializerTests.cs index ae9bb4d..4bf7a0b 100644 --- a/tests/LightProto.Tests/SerializerTests.cs +++ b/tests/LightProto.Tests/SerializerTests.cs @@ -71,6 +71,90 @@ public async Task TestStream() await Assert.That(parsed.Name).IsEquivalentTo(obj.Name); } + [Test] + public async Task AsyncStreamMethods_ShouldUseAsyncIoAndRoundTrip() + { + var original = CreateTestContract(); + using var destination = new AsyncOnlyStream(); + +#if NET6_0_OR_GREATER + await Serializer.SerializeAsync(destination, original); +#else + await Serializer.SerializeAsync(destination, original, TestContract.ProtoWriter); +#endif + + await Assert.That(destination.AsyncWriteCount).IsGreaterThan(0); + + using var source = new AsyncOnlyStream(destination.ToArray()); +#if NET6_0_OR_GREATER + var parsed = await Serializer.DeserializeAsync(source); +#else + var parsed = await Serializer.DeserializeAsync(source, TestContract.ProtoReader); +#endif + + await Assert.That(source.AsyncReadCount).IsGreaterThan(0); + await Assert.That(parsed).IsEquivalentTo(original); + } + + sealed class AsyncOnlyStream : Stream + { + readonly MemoryStream inner; + + public AsyncOnlyStream() + { + inner = new MemoryStream(); + } + + public AsyncOnlyStream(byte[] bytes) + { + inner = new MemoryStream(bytes, writable: true); + } + + public int AsyncReadCount { get; private set; } + + public int AsyncWriteCount { get; private set; } + + public byte[] ToArray() => inner.ToArray(); + + public override bool CanRead => true; + + public override bool CanSeek => inner.CanSeek; + + public override bool CanWrite => true; + + public override long Length => inner.Length; + + public override long Position + { + get => inner.Position; + set => inner.Position = value; + } + + public override void Flush() => throw new NotSupportedException("Synchronous I/O is not supported."); + + public override int Read(byte[] buffer, int offset, int count) => + throw new NotSupportedException("Synchronous I/O is not supported."); + + public override long Seek(long offset, SeekOrigin origin) => inner.Seek(offset, origin); + + public override void SetLength(long value) => inner.SetLength(value); + + public override void Write(byte[] buffer, int offset, int count) => + throw new NotSupportedException("Synchronous I/O is not supported."); + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + AsyncReadCount++; + return inner.ReadAsync(buffer, offset, count, cancellationToken); + } + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + AsyncWriteCount++; + return inner.WriteAsync(buffer, offset, count, cancellationToken); + } + } + class BufferSegment : ReadOnlySequenceSegment { public BufferSegment(byte[] memory)