Merge pull request #808 from paulomorgado/feature/readonlysequencestream

Added Stream over ReadOnlySequence<byte>
This commit is contained in:
Sergio Pedri
2024-11-20 21:53:33 -08:00
committed by GitHub
4 changed files with 761 additions and 1 deletions
@@ -0,0 +1,34 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using CommunityToolkit.HighPerformance.Streams;
using System;
using System.Buffers;
using System.IO;
using System.Runtime.CompilerServices;
namespace CommunityToolkit.HighPerformance;
/// <summary>
/// Helpers for working with the <see cref="ReadOnlySequence{T}"/> type.
/// </summary>
public static class ReadOnlySequenceExtensions
{
/// <summary>
/// Returns a <see cref="Stream"/> wrapping the contents of the given <see cref="Memory{T}"/> of <see cref="byte"/> instance.
/// </summary>
/// <param name="sequence">The input <see cref="ReadOnlySequence{T}"/> of <see cref="byte"/> instance.</param>
/// <returns>A <see cref="Stream"/> wrapping the data within <paramref name="sequence"/>.</returns>
/// <remarks>
/// Since this method only receives a <see cref="ReadOnlySequence{T}"/> instance, which does not track
/// the lifetime of its underlying buffer, it is responsibility of the caller to manage that.
/// In particular, the caller must ensure that the target buffer is not disposed as long
/// as the returned <see cref="Stream"/> is in use, to avoid unexpected issues.
/// </remarks>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static Stream AsStream(this ReadOnlySequence<byte> sequence)
{
return ReadOnlySequenceStream.Create(sequence);
}
}
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Diagnostics.CodeAnalysis;
using System.IO;
using System.Runtime.CompilerServices;
@@ -24,6 +25,20 @@ partial class MemoryStream
}
}
/// <summary>
/// Validates the <see cref="Stream.Position"/> argument (it needs to be in the [0, length]) range.
/// </summary>
/// <param name="position">The new <see cref="Stream.Position"/> value being set.</param>
/// <param name="length">The maximum length of the target <see cref="Stream"/>.</param>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static void ValidatePosition(long position, long length)
{
if ((ulong)position > (ulong)length)
{
ThrowArgumentOutOfRangeExceptionForPosition();
}
}
/// <summary>
/// Validates the <see cref="Stream.Read(byte[],int,int)"/> or <see cref="Stream.Write(byte[],int,int)"/> arguments.
/// </summary>
@@ -31,7 +46,7 @@ partial class MemoryStream
/// <param name="offset">The offset within <paramref name="buffer"/>.</param>
/// <param name="count">The number of elements to process within <paramref name="buffer"/>.</param>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static void ValidateBuffer(byte[]? buffer, int offset, int count)
public static void ValidateBuffer([NotNull] byte[]? buffer, int offset, int count)
{
if (buffer is null)
{
@@ -0,0 +1,304 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Buffers;
using System.IO;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
namespace CommunityToolkit.HighPerformance.Streams;
/// <summary>
/// A <see cref="Stream"/> implementation wrapping a <see cref="ReadOnlySequence{T}"/> of <see cref="byte"/> instance.
/// </summary>
internal sealed partial class ReadOnlySequenceStream : Stream
{
/// <summary>
/// The <see cref="ReadOnlySequence{T}"/> instance currently in use.
/// </summary>
private readonly ReadOnlySequence<byte> source;
/// <summary>
/// The current position within <see cref="source"/>.
/// </summary>
private long position;
/// <summary>
/// Indicates whether or not the current instance has been disposed
/// </summary>
private bool disposed;
/// <summary>
/// Initializes a new instance of the <see cref="ReadOnlySequenceStream"/> class with the specified <see cref="ReadOnlySequence{T}"/> source.
/// </summary>
/// <param name="source">The <see cref="ReadOnlySequence{T}"/> source.</param>
public ReadOnlySequenceStream(ReadOnlySequence<byte> source)
{
this.source = source;
}
/// <inheritdoc/>
public sealed override bool CanRead
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => !this.disposed;
}
/// <inheritdoc/>
public sealed override bool CanSeek
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => !this.disposed;
}
/// <inheritdoc/>
public sealed override bool CanWrite
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => false;
}
/// <inheritdoc/>
public sealed override long Length
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get
{
MemoryStream.ValidateDisposed(this.disposed);
return this.source.Length;
}
}
/// <inheritdoc/>
public sealed override long Position
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get
{
MemoryStream.ValidateDisposed(this.disposed);
return this.position;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
set
{
MemoryStream.ValidateDisposed(this.disposed);
MemoryStream.ValidatePosition(value, this.source.Length);
this.position = value;
}
}
/// <summary>
/// Creates a new <see cref="Stream"/> from the input <see cref="ReadOnlySequence{T}"/> of <see cref="byte"/> instance.
/// </summary>
/// <param name="sequence">The input <see cref="ReadOnlySequence{T}"/> instance.</param>
/// <returns>A <see cref="Stream"/> wrapping the underlying data for <paramref name="sequence"/>.</returns>
public static Stream Create(ReadOnlySequence<byte> sequence)
{
return new ReadOnlySequenceStream(sequence);
}
/// <inheritdoc/>
public sealed override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
{
if (cancellationToken.IsCancellationRequested)
{
return Task.FromCanceled(cancellationToken);
}
try
{
MemoryStream.ValidateDisposed(this.disposed);
if (this.position >= this.source.Length)
{
return Task.CompletedTask;
}
if (this.source.IsSingleSegment)
{
ReadOnlyMemory<byte> buffer = this.source.First.Slice(unchecked((int)this.position));
this.position = this.source.Length;
return destination.WriteAsync(buffer, cancellationToken).AsTask();
}
async Task CoreCopyToAsync(Stream destination, CancellationToken cancellationToken)
{
ReadOnlySequence<byte> sequence = this.source.Slice(this.position);
this.position = this.source.Length;
foreach (ReadOnlyMemory<byte> segment in sequence)
{
await destination.WriteAsync(segment, cancellationToken).ConfigureAwait(false);
}
}
return CoreCopyToAsync(destination, cancellationToken);
}
catch (OperationCanceledException e)
{
return Task.FromCanceled(e.CancellationToken);
}
catch (Exception e)
{
return Task.FromException(e);
}
}
/// <inheritdoc/>
public sealed override void Flush()
{
}
/// <inheritdoc/>
public sealed override Task FlushAsync(CancellationToken cancellationToken)
{
if (cancellationToken.IsCancellationRequested)
{
return Task.FromCanceled(cancellationToken);
}
return Task.CompletedTask;
}
/// <inheritdoc/>
public sealed override Task<int> ReadAsync(byte[]? buffer, int offset, int count, CancellationToken cancellationToken)
{
if (cancellationToken.IsCancellationRequested)
{
return Task.FromCanceled<int>(cancellationToken);
}
try
{
int result = Read(buffer, offset, count);
return Task.FromResult(result);
}
catch (OperationCanceledException e)
{
return Task.FromCanceled<int>(e.CancellationToken);
}
catch (Exception e)
{
return Task.FromException<int>(e);
}
}
public sealed override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
throw MemoryStream.GetNotSupportedException();
}
/// <inheritdoc/>
public sealed override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
{
throw MemoryStream.GetNotSupportedException();
}
/// <inheritdoc/>
public sealed override long Seek(long offset, SeekOrigin origin)
{
MemoryStream.ValidateDisposed(this.disposed);
long index = origin switch
{
SeekOrigin.Begin => offset,
SeekOrigin.Current => this.position + offset,
SeekOrigin.End => this.source.Length + offset,
_ => MemoryStream.ThrowArgumentExceptionForSeekOrigin()
};
MemoryStream.ValidatePosition(index, this.source.Length);
this.position = index;
return index;
}
/// <inheritdoc/>
public sealed override void SetLength(long value)
{
throw MemoryStream.GetNotSupportedException();
}
/// <inheritdoc/>
public sealed override int Read(byte[]? buffer, int offset, int count)
{
MemoryStream.ValidateDisposed(this.disposed);
MemoryStream.ValidateBuffer(buffer, offset, count);
if (this.position >= this.source.Length)
{
return 0;
}
ReadOnlySequence<byte> sequence = this.source.Slice(this.position);
Span<byte> destination = buffer.AsSpan(offset, count);
int bytesCopied = 0;
foreach (ReadOnlyMemory<byte> segment in sequence)
{
int bytesToCopy = Math.Min(segment.Length, destination.Length);
segment.Span.Slice(0, bytesToCopy).CopyTo(destination);
destination = destination.Slice(bytesToCopy);
bytesCopied += bytesToCopy;
this.position += bytesToCopy;
if (destination.Length == 0)
{
break;
}
}
return bytesCopied;
}
/// <inheritdoc/>
public sealed override int ReadByte()
{
MemoryStream.ValidateDisposed(this.disposed);
if (this.position == this.source.Length)
{
return -1;
}
ReadOnlySequence<byte> sequence = this.source.Slice(this.position);
this.position++;
return sequence.First.Span[0];
}
/// <inheritdoc/>
public sealed override void Write(byte[]? buffer, int offset, int count)
{
throw MemoryStream.GetNotSupportedException();
}
/// <inheritdoc/>
public sealed override void WriteByte(byte value)
{
throw MemoryStream.GetNotSupportedException();
}
/// <inheritdoc/>
protected override void Dispose(bool disposing)
{
this.disposed = true;
}
}
@@ -0,0 +1,407 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Buffers;
using System.IO;
using System.Linq;
using System.Runtime.InteropServices;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
namespace CommunityToolkit.HighPerformance.UnitTests.Streams;
[TestClass]
public partial class Test_ReadOnlySequenceStream
{
[TestMethod]
public void Test_ReadOnlySequenceStream_Lifecycle()
{
ReadOnlySequence<byte> sequence = CreateReadOnlySequence(new byte[100]);
Stream stream = sequence.AsStream();
Assert.IsTrue(stream.CanRead);
Assert.IsTrue(stream.CanSeek);
Assert.IsFalse(stream.CanWrite);
Assert.AreEqual(stream.Length, sequence.Length);
Assert.AreEqual(stream.Position, 0);
stream.Dispose();
Assert.IsFalse(stream.CanRead);
Assert.IsFalse(stream.CanSeek);
Assert.IsFalse(stream.CanWrite);
_ = Assert.ThrowsException<ObjectDisposedException>(() => stream.Length);
_ = Assert.ThrowsException<ObjectDisposedException>(() => stream.Position);
}
[TestMethod]
public void Test_ReadOnlySequenceStream_Seek()
{
Stream stream = CreateReadOnlySequence(new byte[50], new byte[50]).AsStream();
Assert.AreEqual(stream.Position, 0);
stream.Position = 42;
Assert.AreEqual(stream.Position, 42);
_ = Assert.ThrowsException<ArgumentOutOfRangeException>(() => stream.Position = -1);
_ = Assert.ThrowsException<ArgumentOutOfRangeException>(() => stream.Position = 120);
_ = stream.Seek(0, SeekOrigin.Begin);
_ = Assert.ThrowsException<ArgumentOutOfRangeException>(() => stream.Seek(-1, SeekOrigin.Begin));
_ = Assert.ThrowsException<ArgumentOutOfRangeException>(() => stream.Seek(120, SeekOrigin.Begin));
Assert.AreEqual(stream.Position, 0);
_ = stream.Seek(-1, SeekOrigin.End);
_ = Assert.ThrowsException<ArgumentOutOfRangeException>(() => stream.Seek(20, SeekOrigin.End));
_ = Assert.ThrowsException<ArgumentOutOfRangeException>(() => stream.Seek(-120, SeekOrigin.End));
Assert.AreEqual(stream.Position, stream.Length - 1);
_ = stream.Seek(42, SeekOrigin.Begin);
_ = stream.Seek(20, SeekOrigin.Current);
_ = stream.Seek(-30, SeekOrigin.Current);
_ = Assert.ThrowsException<ArgumentOutOfRangeException>(() => stream.Seek(-64, SeekOrigin.Current));
_ = Assert.ThrowsException<ArgumentOutOfRangeException>(() => stream.Seek(80, SeekOrigin.Current));
Assert.AreEqual(stream.Position, 32);
}
[TestMethod]
public void Test_ReadOnlySequenceStream_Read_Array()
{
Memory<byte> data = CreateRandomData(64);
Stream stream = CreateReadOnlySequence(data).AsStream();
stream.Position = 0;
byte[] result = new byte[data.Length];
int bytesRead = stream.Read(result, 0, result.Length);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(stream.Position, data.Length);
Assert.IsTrue(data.Span.SequenceEqual(result));
stream.Dispose();
_ = Assert.ThrowsException<ObjectDisposedException>(() => stream.Read(result, 0, result.Length));
}
[TestMethod]
public void Test_ReadOnlySequenceStream_NotFromStart_Read_Array()
{
const int offset = 8;
Memory<byte> data = CreateRandomData(64);
Stream stream = CreateReadOnlySequence(data).AsStream();
stream.Position = offset;
byte[] result = new byte[data.Length - offset];
int bytesRead = stream.Read(result, 0, result.Length);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(stream.Position, result.Length + offset);
Assert.IsTrue(data.Span.Slice(offset).SequenceEqual(result));
stream.Dispose();
_ = Assert.ThrowsException<ObjectDisposedException>(() => stream.Read(result, 0, result.Length));
}
[TestMethod]
public void Test_ReadOnlySequenceStream_ReadByte()
{
Memory<byte> data = new byte[] { 1, 128, 255, 32 };
Stream stream = CreateReadOnlySequence(data.Slice(0,2), data.Slice(2, 2)).AsStream();
Span<byte> result = stackalloc byte[4];
foreach (ref byte value in result)
{
value = checked((byte)stream.ReadByte());
}
Assert.AreEqual(stream.Position, data.Length);
Assert.IsTrue(data.Span.SequenceEqual(result));
int exitCode = stream.ReadByte();
Assert.AreEqual(exitCode, -1);
}
[TestMethod]
public void Test_ReadOnlySequenceStream_Read_Span()
{
Memory<byte> data = CreateRandomData(64);
Stream stream = CreateReadOnlySequence(data).AsStream();
stream.Position = 0;
Span<byte> result = new byte[data.Length];
int bytesRead = stream.Read(result);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(stream.Position, data.Length);
Assert.IsTrue(data.Span.SequenceEqual(result));
}
[TestMethod]
public async Task Test_ReadOnlySequenceStream_ReadAsync_Memory()
{
Memory<byte> data = CreateRandomData(64);
Stream stream = CreateReadOnlySequence(data).AsStream();
Memory<byte> result = new byte[data.Length];
int bytesRead = await stream.ReadAsync(result);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(stream.Position, data.Length);
Assert.IsTrue(data.Span.SequenceEqual(result.Span));
}
[TestMethod]
public void Test_ReadOnlySequenceStream_SigleSegment_CopyTo()
{
Memory<byte> data = CreateRandomData(64);
Stream source = CreateReadOnlySequence(data).AsStream();
Stream destination = new byte[100].AsMemory().AsStream();
source.CopyTo(destination);
Assert.AreEqual(source.Position, destination.Position);
destination.Position = 0;
Memory<byte> result = new byte[data.Length];
int bytesRead = destination.Read(result.Span);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(destination.Position, data.Length);
Assert.IsTrue(data.Span.SequenceEqual(result.Span));
}
[TestMethod]
public void Test_ReadOnlySequenceStream_CopyTo()
{
Memory<byte> data = CreateRandomData(64);
Stream source = CreateReadOnlySequence(data.Slice(0, 32), data.Slice(32)).AsStream();
Stream destination = new byte[100].AsMemory().AsStream();
source.CopyTo(destination);
Assert.AreEqual(source.Position, destination.Position);
destination.Position = 0;
Memory<byte> result = new byte[data.Length];
int bytesRead = destination.Read(result.Span);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(destination.Position, data.Length);
Assert.IsTrue(data.Span.SequenceEqual(result.Span));
}
[TestMethod]
public async Task Test_ReadOnlySequenceStream_SigleSegment_CopyToAsync()
{
Memory<byte> data = CreateRandomData(64);
Stream source = CreateReadOnlySequence(data).AsStream();
Stream destination = new byte[100].AsMemory().AsStream();
await source.CopyToAsync(destination);
Assert.AreEqual(source.Position, destination.Position);
destination.Position = 0;
Memory<byte> result = new byte[data.Length];
int bytesRead = await destination.ReadAsync(result);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(destination.Position, data.Length);
Assert.IsTrue(data.Span.SequenceEqual(result.Span));
}
[TestMethod]
public async Task Test_ReadOnlySequenceStream_SigleSegment_NotFromStart_CopyToAsync()
{
const int offset = 8;
Memory<byte> data = CreateRandomData(64);
Stream source = CreateReadOnlySequence(data).AsStream();
source.Position = offset;
Stream destination = new byte[100].AsMemory().AsStream();
await source.CopyToAsync(destination);
Assert.AreEqual(source.Position, destination.Position + offset);
destination.Position = 0;
Memory<byte> result = new byte[data.Length - offset];
int bytesRead = await destination.ReadAsync(result);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(destination.Position, data.Length - offset);
Assert.IsTrue(data.Span.Slice(offset).SequenceEqual(result.Span));
}
[TestMethod]
public async Task Test_ReadOnlySequenceStream_MultipleSegments_CopyToAsync()
{
Memory<byte> data = CreateRandomData(64);
Stream source = CreateReadOnlySequence(data.Slice(0, 16), data.Slice(16, 16), data.Slice(32, 16), data.Slice(48, 16)).AsStream();
Stream destination = new byte[100].AsMemory().AsStream();
await source.CopyToAsync(destination);
Assert.AreEqual(source.Position, destination.Position);
destination.Position = 0;
Memory<byte> result = new byte[data.Length];
int bytesRead = await destination.ReadAsync(result);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(destination.Position, data.Length);
Assert.IsTrue(data.Span.SequenceEqual(result.Span));
}
[TestMethod]
public async Task Test_ReadOnlySequenceStream_MultipleSegments_NotFromStart_CopyToAsync()
{
const int offset = 8;
Memory<byte> data = CreateRandomData(64);
Stream source = CreateReadOnlySequence(data.Slice(0, 16), data.Slice(16, 16), data.Slice(32, 16), data.Slice(48, 16)).AsStream();
source.Position = offset;
Stream destination = new byte[100].AsMemory().AsStream();
await source.CopyToAsync(destination);
Assert.AreEqual(source.Position, destination.Position + offset);
destination.Position = 0;
Memory<byte> result = new byte[data.Length - offset];
int bytesRead = await destination.ReadAsync(result);
Assert.AreEqual(bytesRead, result.Length);
Assert.AreEqual(destination.Position, data.Length - offset);
Assert.IsTrue(data.Span.Slice(offset).SequenceEqual(result.Span));
}
/// <summary>
/// Creates a random <see cref="byte"/> array filled with random data.
/// </summary>
/// <param name="count">The number of array items to create.</param>
/// <returns>The returned random array.</returns>
private static byte[] CreateRandomData(int count)
{
Random? random = new(DateTime.Now.Ticks.GetHashCode());
byte[] data = new byte[count];
foreach (ref byte n in MemoryMarshal.AsBytes(data.AsSpan()))
{
n = (byte)random.Next(0, byte.MaxValue);
}
return data;
}
/// <summary>
/// Creates a <see cref="ReadOnlySequence{T}"/> value from the input segments.
/// </summary>
/// <param name="segments">The input <see cref="ReadOnlyMemory{T}"/> segments.</param>
/// <returns>The resulting <see cref="ReadOnlySequence{T}"/> value.</returns>
private static ReadOnlySequence<byte> CreateReadOnlySequence(params ReadOnlyMemory<byte>[] segments)
{
if (segments is not { Length: > 0 })
{
return ReadOnlySequence<byte>.Empty;
}
if (segments.Length == 1)
{
return new(segments[0]);
}
ReadOnlySequenceSegmentOfByte first = new(segments[0]);
ReadOnlySequenceSegmentOfByte last = first;
long length = first.Memory.Length;
for (int i = 1; i < segments.Length; i++)
{
ReadOnlyMemory<byte> segment = segments[i];
length += segment.Length;
last = last.Append(segment);
}
return new(first, 0, last, (int)(length - last.RunningIndex));
}
/// <summary>
/// A custom <see cref="ReadOnlySequenceSegment{T}"/> that supports appending new segments.
/// </summary>
private sealed class ReadOnlySequenceSegmentOfByte : ReadOnlySequenceSegment<byte>
{
public ReadOnlySequenceSegmentOfByte(ReadOnlyMemory<byte> memory)
{
Memory = memory;
}
public ReadOnlySequenceSegmentOfByte Append(ReadOnlyMemory<byte> memory)
{
ReadOnlySequenceSegmentOfByte nextSegment = new(memory)
{
RunningIndex = RunningIndex + Memory.Length
};
Next = nextSegment;
return nextSegment;
}
}
}