From e24b6ba5ca8cc13dfc272b71b96f2ae9b62d24f5 Mon Sep 17 00:00:00 2001 From: Dameng <313880747@qq.com> Date: Sat, 11 Jul 2026 21:45:31 +0800 Subject: [PATCH] Revert "Add asynchronous Stream serializer APIs" --- src/LightProto/PooledSegmentBufferWriter.cs | 193 ------------------ .../PublicAPI/net/PublicAPI.Unshipped.txt | 11 - .../netstandard2.0/PublicAPI.Unshipped.txt | 9 - src/LightProto/Serializer.Async.cs | 90 -------- src/LightProto/Serializer.Dynamically.cs | 28 --- src/LightProto/Serializer.Extensions.cs | 11 - src/LightProto/Serializer.NonGeneric.cs | 103 ---------- tests/LightProto.Tests/SerializerTests.cs | 84 -------- 8 files changed, 529 deletions(-) delete mode 100644 src/LightProto/PooledSegmentBufferWriter.cs delete mode 100644 src/LightProto/Serializer.Async.cs diff --git a/src/LightProto/PooledSegmentBufferWriter.cs b/src/LightProto/PooledSegmentBufferWriter.cs deleted file mode 100644 index d252c8c..0000000 --- a/src/LightProto/PooledSegmentBufferWriter.cs +++ /dev/null @@ -1,193 +0,0 @@ -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 14a2d0f..2904c4b 100644 --- a/src/LightProto/PublicAPI/net/PublicAPI.Unshipped.txt +++ b/src/LightProto/PublicAPI/net/PublicAPI.Unshipped.txt @@ -1,15 +1,4 @@ #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 d902fdb..cc7beb4 100644 --- a/src/LightProto/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt +++ b/src/LightProto/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt @@ -1,13 +1,4 @@ #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 deleted file mode 100644 index 21ddaec..0000000 --- a/src/LightProto/Serializer.Async.cs +++ /dev/null @@ -1,90 +0,0 @@ -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 b7b9abb..8735923 100644 --- a/src/LightProto/Serializer.Dynamically.cs +++ b/src/LightProto/Serializer.Dynamically.cs @@ -43,20 +43,6 @@ 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 /// @@ -72,20 +58,6 @@ 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 e472c07..2eea418 100644 --- a/src/LightProto/Serializer.Extensions.cs +++ b/src/LightProto/Serializer.Extensions.cs @@ -178,17 +178,6 @@ 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 7fad0c7..00c6c56 100644 --- a/src/LightProto/Serializer.NonGeneric.cs +++ b/src/LightProto/Serializer.NonGeneric.cs @@ -4,7 +4,6 @@ namespace LightProto { -#pragma warning disable RS0026 // CancellationToken is intentionally optional for the asynchronous API surface. public static partial class Serializer { /// @@ -43,60 +42,6 @@ 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. /// @@ -203,53 +148,6 @@ 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 /// @@ -344,5 +242,4 @@ 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 4bf7a0b..ae9bb4d 100644 --- a/tests/LightProto.Tests/SerializerTests.cs +++ b/tests/LightProto.Tests/SerializerTests.cs @@ -71,90 +71,6 @@ 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)