Skip to content

Commit fec7a59

Browse files
ViveliDuChViveliDuCh
committed
Address PR feedback. MemoryStream base change. Spillover buffer for Convert edge case. Flush strategy update.
1 parent bd5cf98 commit fec7a59

16 files changed

Lines changed: 341 additions & 144 deletions

File tree

‎src/libraries/Common/src/System/IO/ReadOnlyMemoryStream.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
11
// Licensed to the .NET Foundation under one or more agreements.
22
// The .NET Foundation licenses this file to you under the MIT license.
33

4+
// On net11.0+, the public ReadOnlyMemoryStream in System.Runtime (CoreLib) supersedes this internal copy.
5+
#if !NET11_0_OR_GREATER
6+
47
using System.Threading;
58
using System.Threading.Tasks;
69

@@ -213,3 +216,5 @@ private static void ValidateBufferArguments(byte[] buffer, int offset, int count
213216
#endif
214217
}
215218
}
219+
220+
#endif // !NET11_0_OR_GREATER

‎src/libraries/System.Memory/ref/System.Memory.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -171,11 +171,17 @@ public ReadOnlySequenceStream(System.Buffers.ReadOnlySequence<byte> sequence) {
171171
public override long Length { get { throw null; } }
172172
public override long Position { get { throw null; } set { } }
173173
public override void Flush() { }
174+
public override System.Threading.Tasks.Task FlushAsync(System.Threading.CancellationToken cancellationToken) { throw null; }
174175
public override int Read(byte[] buffer, int offset, int count) { throw null; }
175176
public override int Read(System.Span<byte> buffer) { throw null; }
177+
public override System.Threading.Tasks.Task<int> ReadAsync(byte[] buffer, int offset, int count, System.Threading.CancellationToken cancellationToken) { throw null; }
178+
public override System.Threading.Tasks.ValueTask<int> ReadAsync(System.Memory<byte> buffer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
176179
public override long Seek(long offset, System.IO.SeekOrigin origin) { throw null; }
177180
public override void SetLength(long value) { }
178181
public override void Write(byte[] buffer, int offset, int count) { }
182+
public override void Write(System.ReadOnlySpan<byte> buffer) { }
183+
public override System.Threading.Tasks.Task WriteAsync(byte[] buffer, int offset, int count, System.Threading.CancellationToken cancellationToken) { throw null; }
184+
public override System.Threading.Tasks.ValueTask WriteAsync(System.ReadOnlyMemory<byte> buffer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
179185
}
180186
}
181187
namespace System.Runtime.InteropServices

‎src/libraries/System.Memory/src/System.Memory.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
<Project Sdk="Microsoft.NET.Sdk">
1+
<Project Sdk="Microsoft.NET.Sdk">
22

33
<PropertyGroup>
44
<TargetFramework>$(NetCoreAppCurrent)</TargetFramework>

‎src/libraries/System.Memory/src/System/Buffers/ReadOnlySequenceStream.cs‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ public sealed class ReadOnlySequenceStream : Stream
1919
{
2020
private ReadOnlySequence<byte> _sequence;
2121
private SequencePosition _position;
22-
private long _positionPastEnd; // -1 if within bounds, or the actual position if past end
22+
private long _absolutePosition;
2323
private bool _isDisposed;
2424

2525
/// <summary>
@@ -30,7 +30,7 @@ public ReadOnlySequenceStream(ReadOnlySequence<byte> sequence)
3030
{
3131
_sequence = sequence;
3232
_position = sequence.Start;
33-
_positionPastEnd = -1;
33+
_absolutePosition = 0;
3434
_isDisposed = false;
3535
}
3636

@@ -61,24 +61,23 @@ public override long Position
6161
get
6262
{
6363
EnsureNotDisposed();
64-
return _positionPastEnd >= 0 ? _positionPastEnd : _sequence.Slice(_sequence.Start, _position).Length;
64+
return _absolutePosition;
6565
}
6666
set
6767
{
6868
EnsureNotDisposed();
6969
ArgumentOutOfRangeException.ThrowIfNegative(value);
7070

71-
// Allow seeking past the end
72-
if (value >= Length)
71+
if (value >= _sequence.Length)
7372
{
7473
_position = _sequence.End;
75-
_positionPastEnd = value;
7674
}
7775
else
7876
{
7977
_position = _sequence.GetPosition(value, _sequence.Start);
80-
_positionPastEnd = -1;
8178
}
79+
80+
_absolutePosition = value;
8281
}
8382
}
8483

@@ -94,7 +93,7 @@ public override int Read(Span<byte> buffer)
9493
{
9594
EnsureNotDisposed();
9695

97-
if (_positionPastEnd >= 0)
96+
if (_absolutePosition >= _sequence.Length)
9897
{
9998
return 0;
10099
}
@@ -108,6 +107,7 @@ public override int Read(Span<byte> buffer)
108107

109108
remaining.Slice(0, n).CopyTo(buffer);
110109
_position = _sequence.GetPosition(n, _position);
110+
_absolutePosition += n;
111111
return n;
112112
}
113113

@@ -116,9 +116,10 @@ public override Task<int> ReadAsync(byte[] buffer, int offset, int count, Cancel
116116
{
117117
ValidateBufferArguments(buffer, offset, count);
118118

119-
// If cancellation was requested, bail early
120119
if (cancellationToken.IsCancellationRequested)
120+
{
121121
return Task.FromCanceled<int>(cancellationToken);
122+
}
122123

123124
int n = Read(buffer, offset, count);
124125
return Task.FromResult(n);
@@ -161,8 +162,8 @@ public override long Seek(long offset, SeekOrigin origin)
161162
long absolutePosition = origin switch
162163
{
163164
SeekOrigin.Begin => offset,
164-
SeekOrigin.Current => (_positionPastEnd >= 0 ? _positionPastEnd : _sequence.Slice(_sequence.Start, _position).Length) + offset,
165-
SeekOrigin.End => Length + offset,
165+
SeekOrigin.Current => _absolutePosition + offset,
166+
SeekOrigin.End => _sequence.Length + offset,
166167
_ => throw new ArgumentException(SR.Argument_InvalidSeekOrigin, nameof(origin))
167168
};
168169

@@ -173,23 +174,26 @@ public override long Seek(long offset, SeekOrigin origin)
173174
}
174175

175176
// Update position - seeking past end is allowed
176-
if (absolutePosition >= Length)
177+
if (absolutePosition >= _sequence.Length)
177178
{
178179
_position = _sequence.End;
179-
_positionPastEnd = absolutePosition;
180180
}
181181
else
182182
{
183183
_position = _sequence.GetPosition(absolutePosition, _sequence.Start);
184-
_positionPastEnd = -1;
185184
}
186185

186+
_absolutePosition = absolutePosition;
187187
return absolutePosition;
188188
}
189189

190190
/// <inheritdoc />
191191
public override void Flush() { }
192192

193+
/// <inheritdoc />
194+
public override Task FlushAsync(CancellationToken cancellationToken) =>
195+
cancellationToken.IsCancellationRequested ? Task.FromCanceled(cancellationToken) : Task.CompletedTask;
196+
193197
/// <inheritdoc />
194198
public override void SetLength(long value) => throw new NotSupportedException(SR.NotSupported_UnwritableStream);
195199

‎src/libraries/System.Memory/tests/System.Memory.Tests.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
<Project Sdk="Microsoft.NET.Sdk">
1+
<Project Sdk="Microsoft.NET.Sdk">
22
<PropertyGroup>
33
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
44
<TestRuntime>true</TestRuntime>

‎src/libraries/System.Private.CoreLib/src/Resources/Strings.resx‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3068,9 +3068,6 @@
30683068
<data name="InvalidOperation_SpanOverlappedOperation" xml:space="preserve">
30693069
<value>This operation is invalid on overlapping buffers.</value>
30703070
</data>
3071-
<data name="InvalidOperation_StreamResyncExceededMaxIterations" xml:space="preserve">
3072-
<value>Stream resynchronization exceeded maximum iterations.</value>
3073-
</data>
30743071
<data name="InvalidOperation_TimeProviderNullLocalTimeZone" xml:space="preserve">
30753072
<value>The operation cannot be performed when TimeProvider.LocalTimeZone is null.</value>
30763073
</data>

‎src/libraries/System.Private.CoreLib/src/System/IO/ReadOnlyMemoryStream.cs‎

Lines changed: 75 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,14 @@
77
namespace System.IO;
88

99
/// <summary>
10-
/// Provides a seekable, read-only <see cref="Stream"/> over a <see cref="ReadOnlyMemory{Byte}"/>.
10+
/// Provides a seekable, read-only <see cref="MemoryStream"/> over a <see cref="ReadOnlyMemory{Byte}"/>.
1111
/// </summary>
1212
/// <remarks>
1313
/// <para>This type is not thread-safe. Synchronize access if the stream is used concurrently.</para>
1414
/// <para>The stream cannot be written to. <see cref="CanWrite"/> always returns <see langword="false"/>.</para>
15+
/// <para><see cref="GetBuffer"/> throws and <see cref="TryGetBuffer"/> returns <see langword="false"/>.</para>
1516
/// </remarks>
16-
public sealed class ReadOnlyMemoryStream : Stream
17+
public sealed class ReadOnlyMemoryStream : MemoryStream
1718
{
1819
private ReadOnlyMemory<byte> _buffer;
1920
private int _position;
@@ -23,7 +24,7 @@ public sealed class ReadOnlyMemoryStream : Stream
2324
/// Initializes a new instance of the <see cref="ReadOnlyMemoryStream"/> class over the specified <see cref="ReadOnlyMemory{Byte}"/>.
2425
/// </summary>
2526
/// <param name="source">The <see cref="ReadOnlyMemory{Byte}"/> to wrap.</param>
26-
public ReadOnlyMemoryStream(ReadOnlyMemory<byte> source)
27+
public ReadOnlyMemoryStream(ReadOnlyMemory<byte> source) : base()
2728
{
2829
_buffer = source;
2930
_isOpen = true;
@@ -38,6 +39,17 @@ public ReadOnlyMemoryStream(ReadOnlyMemory<byte> source)
3839
/// <inheritdoc/>
3940
public override bool CanWrite => false;
4041

42+
/// <inheritdoc/>
43+
public override int Capacity
44+
{
45+
get
46+
{
47+
EnsureNotClosed();
48+
return _buffer.Length;
49+
}
50+
set => throw new NotSupportedException(SR.NotSupported_MemStreamNotExpandable);
51+
}
52+
4153
/// <inheritdoc/>
4254
public override long Length
4355
{
@@ -72,10 +84,16 @@ public override int ReadByte()
7284
{
7385
EnsureNotClosed();
7486

75-
if (_position >= _buffer.Length)
76-
return -1;
87+
ReadOnlySpan<byte> span = _buffer.Span;
88+
int position = _position;
89+
90+
if ((uint)position < (uint)span.Length)
91+
{
92+
_position++;
93+
return span[position];
94+
}
7795

78-
return _buffer.Span[_position++];
96+
return -1;
7997
}
8098

8199
/// <inheritdoc/>
@@ -93,7 +111,9 @@ public override int Read(Span<byte> buffer)
93111

94112
int remaining = _buffer.Length - _position;
95113
if (remaining <= 0 || buffer.Length == 0)
114+
{
96115
return 0;
116+
}
97117

98118
int bytesToRead = Math.Min(remaining, buffer.Length);
99119
_buffer.Span.Slice(_position, bytesToRead).CopyTo(buffer);
@@ -109,7 +129,9 @@ public override Task<int> ReadAsync(byte[] buffer, int offset, int count, Cancel
109129
EnsureNotClosed();
110130

111131
if (cancellationToken.IsCancellationRequested)
132+
{
112133
return Task.FromCanceled<int>(cancellationToken);
134+
}
113135

114136
return Task.FromResult(Read(buffer, offset, count));
115137
}
@@ -120,7 +142,9 @@ public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken
120142
EnsureNotClosed();
121143

122144
if (cancellationToken.IsCancellationRequested)
145+
{
123146
return ValueTask.FromCanceled<int>(cancellationToken);
147+
}
124148

125149
return new ValueTask<int>(Read(buffer.Span));
126150
}
@@ -169,7 +193,9 @@ public override long Seek(long offset, SeekOrigin origin)
169193
};
170194

171195
if (newPosition < 0)
196+
{
172197
throw new IOException(SR.IO_SeekBeforeBegin);
198+
}
173199

174200
ArgumentOutOfRangeException.ThrowIfGreaterThan(newPosition, int.MaxValue, nameof(offset));
175201

@@ -187,6 +213,49 @@ public override long Seek(long offset, SeekOrigin origin)
187213
/// <inheritdoc/>
188214
public override void Write(ReadOnlySpan<byte> buffer) => throw new NotSupportedException(SR.NotSupported_UnwritableStream);
189215

216+
/// <inheritdoc/>
217+
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw new NotSupportedException(SR.NotSupported_UnwritableStream);
218+
219+
/// <inheritdoc/>
220+
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default) => throw new NotSupportedException(SR.NotSupported_UnwritableStream);
221+
222+
/// <inheritdoc/>
223+
public override byte[] GetBuffer() =>
224+
throw new UnauthorizedAccessException(SR.UnauthorizedAccess_MemStreamBuffer);
225+
226+
/// <inheritdoc/>
227+
public override bool TryGetBuffer(out ArraySegment<byte> buffer)
228+
{
229+
buffer = default;
230+
return false;
231+
}
232+
233+
/// <inheritdoc/>
234+
public override byte[] ToArray()
235+
{
236+
EnsureNotClosed();
237+
if (_buffer.Length == 0)
238+
{
239+
return Array.Empty<byte>();
240+
}
241+
242+
byte[] copy = GC.AllocateUninitializedArray<byte>(_buffer.Length);
243+
_buffer.Span.CopyTo(copy);
244+
return copy;
245+
}
246+
247+
/// <inheritdoc/>
248+
public override void WriteTo(Stream stream)
249+
{
250+
ArgumentNullException.ThrowIfNull(stream);
251+
EnsureNotClosed();
252+
253+
if (_buffer.Length > 0)
254+
{
255+
stream.Write(_buffer.Span);
256+
}
257+
}
258+
190259
/// <inheritdoc/>
191260
public override void Flush() { }
192261

0 commit comments

Comments
 (0)