diff --git a/src/libraries/System.Private.CoreLib/src/System/IO/StreamReader.cs b/src/libraries/System.Private.CoreLib/src/System/IO/StreamReader.cs index 319d31ddaeed55..a835e0b8d8276e 100644 --- a/src/libraries/System.Private.CoreLib/src/System/IO/StreamReader.cs +++ b/src/libraries/System.Private.CoreLib/src/System/IO/StreamReader.cs @@ -5,6 +5,7 @@ using System.Buffers.Binary; using System.Diagnostics; using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; using System.Text; using System.Threading; using System.Threading.Tasks; @@ -71,14 +72,22 @@ public class StreamReader : TextReader // We don't guarantee thread safety on StreamReader, but we should at // least prevent users from trying to read anything while an Async - // read from the same thread is in progress. + // read from the same thread is in progress. We track this with the + // following fields. + // + // Generally we prefer to use the bool _asyncIOInProgress, but in + // certain cases for async1 this would require introducing a wrapper + // state machine, and in those cases we use the Task _asyncReadTask + // instead. + // + private bool _asyncIOInProgress; private Task _asyncReadTask = Task.CompletedTask; private void CheckAsyncTaskInProgress() { - // We are not locking the access to _asyncReadTask because this is not meant to guarantee thread safety. + // We are not locking this access because this is not meant to guarantee thread safety. // We are simply trying to deter calling any Read APIs while an async Read from the same thread is in progress. - if (!_asyncReadTask.IsCompleted) + if (_asyncIOInProgress || !_asyncReadTask.IsCompleted) { ThrowAsyncIOInProgress(); } @@ -88,6 +97,24 @@ private void CheckAsyncTaskInProgress() private static void ThrowAsyncIOInProgress() => throw new InvalidOperationException(SR.InvalidOperation_AsyncIOInProgress); + private ThrowOnReadsScope GuardAgainstOtherReads() + { + return new ThrowOnReadsScope(this); + } + + private readonly struct ThrowOnReadsScope : IDisposable + { + private readonly StreamReader _reader; + + public ThrowOnReadsScope(StreamReader reader) + { + reader._asyncIOInProgress = true; + _reader = reader; + } + + public void Dispose() => _reader._asyncIOInProgress = false; + } + // StreamReader by default will ignore illegal UTF8 characters. We don't want to // throw here because we want to be able to read ill-formed data without choking. // The high level goal is to be tolerant of encoding errors when we read and very strict @@ -898,14 +925,13 @@ private int ReadBuffer(Span userBuffer, out bool readToUserBuffer) ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = ReadLineAsyncInternal(cancellationToken); - _asyncReadTask = task; - - return new ValueTask(task); + return new ValueTask(ReadLineAsyncInternal(cancellationToken)); } private async Task ReadLineAsyncInternal(CancellationToken cancellationToken) { + using ThrowOnReadsScope _ = GuardAgainstOtherReads(); + if (_charPos == _charLen && (await ReadBufferAsync(cancellationToken).ConfigureAwait(false)) == 0) { return null; @@ -1026,14 +1052,13 @@ public override Task ReadToEndAsync(CancellationToken cancellationToken) ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = ReadToEndAsyncInternal(cancellationToken); - _asyncReadTask = task; - - return task; + return ReadToEndAsyncInternal(cancellationToken); } private async Task ReadToEndAsyncInternal(CancellationToken cancellationToken) { + using ThrowOnReadsScope _ = GuardAgainstOtherReads(); + // Call ReadBuffer, then pull data out of charBuffer. StringBuilder sb = new StringBuilder(_charLen - _charPos); do @@ -1070,10 +1095,20 @@ public override Task ReadAsync(char[] buffer, int index, int count) ThrowIfDisposed(); CheckAsyncTaskInProgress(); + if (RuntimeHelpers.IsRuntimeAsync()) + { + return ReadAsyncInternalWithGuard(new Memory(buffer, index, count), CancellationToken.None); + } + Task task = ReadAsyncInternal(new Memory(buffer, index, count), CancellationToken.None).AsTask(); _asyncReadTask = task; - return task; + + async Task ReadAsyncInternalWithGuard(Memory buffer, CancellationToken cancellationToken) + { + using ThrowOnReadsScope _ = GuardAgainstOtherReads(); + return await ReadAsyncInternal(buffer, cancellationToken).ConfigureAwait(false); + } } public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) @@ -1281,10 +1316,20 @@ public override Task ReadBlockAsync(char[] buffer, int index, int count) ThrowIfDisposed(); CheckAsyncTaskInProgress(); + if (RuntimeHelpers.IsRuntimeAsync()) + { + return ReadBlockAsyncWithGuard(buffer, index, count); + } + Task task = base.ReadBlockAsync(buffer, index, count); _asyncReadTask = task; - return task; + + async Task ReadBlockAsyncWithGuard(char[] buffer, int index, int count) + { + using ThrowOnReadsScope _ = GuardAgainstOtherReads(); + return await base.ReadBlockAsync(buffer, index, count).ConfigureAwait(false); + } } public override ValueTask ReadBlockAsync(Memory buffer, CancellationToken cancellationToken = default) @@ -1304,6 +1349,11 @@ public override ValueTask ReadBlockAsync(Memory buffer, CancellationT return ValueTask.FromCanceled(cancellationToken); } + if (RuntimeHelpers.IsRuntimeAsync()) + { + return ReadBlockAsyncInternalWithGuard(buffer, cancellationToken); + } + ValueTask vt = ReadBlockAsyncInternal(buffer, cancellationToken); if (vt.IsCompletedSuccessfully) { @@ -1313,6 +1363,12 @@ public override ValueTask ReadBlockAsync(Memory buffer, CancellationT Task t = vt.AsTask(); _asyncReadTask = t; return new ValueTask(t); + + async ValueTask ReadBlockAsyncInternalWithGuard(Memory buffer, CancellationToken token) + { + using ThrowOnReadsScope _ = GuardAgainstOtherReads(); + return await ReadBlockAsyncInternal(buffer, token).ConfigureAwait(false); + } } private async ValueTask ReadBufferAsync(CancellationToken cancellationToken) diff --git a/src/libraries/System.Private.CoreLib/src/System/IO/StreamWriter.cs b/src/libraries/System.Private.CoreLib/src/System/IO/StreamWriter.cs index 9608f6bf470007..ce5dee0c97ffaa 100644 --- a/src/libraries/System.Private.CoreLib/src/System/IO/StreamWriter.cs +++ b/src/libraries/System.Private.CoreLib/src/System/IO/StreamWriter.cs @@ -43,14 +43,22 @@ public class StreamWriter : TextWriter // We don't guarantee thread safety on StreamWriter, but we should at // least prevent users from trying to write anything while an Async - // write from the same thread is in progress. + // write from the same thread is in progress. We track this with the + // following fields. + // + // Generally we prefer to use the bool _asyncIOInProgress, but in + // certain cases for async1 this would require introducing a wrapper + // state machine, and in those cases we use Task _asyncWriteTask + // instead. + // + private bool _asyncIOInProgress; private Task _asyncWriteTask = Task.CompletedTask; private void CheckAsyncTaskInProgress() { - // We are not locking the access to _asyncWriteTask because this is not meant to guarantee thread safety. + // We are not locking this access because this is not meant to guarantee thread safety. // We are simply trying to deter calling any Write APIs while an async Write from the same thread is in progress. - if (!_asyncWriteTask.IsCompleted) + if (_asyncIOInProgress || !_asyncWriteTask.IsCompleted) { ThrowAsyncIOInProgress(); } @@ -60,6 +68,24 @@ private void CheckAsyncTaskInProgress() private static void ThrowAsyncIOInProgress() => throw new InvalidOperationException(SR.InvalidOperation_AsyncIOInProgress); + private ThrowOnWritesScope GuardAgainstOtherWrites() + { + return new ThrowOnWritesScope(this); + } + + private readonly struct ThrowOnWritesScope : IDisposable + { + private readonly StreamWriter _writer; + + public ThrowOnWritesScope(StreamWriter writer) + { + writer._asyncIOInProgress = true; + _writer = writer; + } + + public void Dispose() => _writer._asyncIOInProgress = false; + } + // The high level goal is to be tolerant of encoding errors when we read and very strict // when we write. Hence, default StreamWriter encoding will throw on encoding error. // Note: when StreamWriter throws on invalid encoding chars (for ex, high surrogate character @@ -664,14 +690,13 @@ public override Task WriteAsync(char value) ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = WriteAsyncInternal(value, appendNewLine: false); - _asyncWriteTask = task; - - return task; + return WriteAsyncInternal(value, appendNewLine: false); } private async Task WriteAsyncInternal(char value, bool appendNewLine) { + using ThrowOnWritesScope _ = GuardAgainstOtherWrites(); + if (_charPos == _charLen) { await FlushAsyncInternal(flushStream: false, flushEncoder: false).ConfigureAwait(false); @@ -714,10 +739,7 @@ public override Task WriteAsync(string? value) ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = WriteAsyncInternal(value.AsMemory(), appendNewLine: false, default); - _asyncWriteTask = task; - - return task; + return WriteAsyncInternal(value.AsMemory(), appendNewLine: false, default); } else { @@ -748,10 +770,7 @@ public override Task WriteAsync(char[] buffer, int index, int count) ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = WriteAsyncInternal(new ReadOnlyMemory(buffer, index, count), appendNewLine: false, cancellationToken: default); - _asyncWriteTask = task; - - return task; + return WriteAsyncInternal(new ReadOnlyMemory(buffer, index, count), appendNewLine: false, cancellationToken: default); } public override Task WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) @@ -770,13 +789,13 @@ public override Task WriteAsync(ReadOnlyMemory buffer, CancellationToken c return Task.FromCanceled(cancellationToken); } - Task task = WriteAsyncInternal(buffer, appendNewLine: false, cancellationToken: cancellationToken); - _asyncWriteTask = task; - return task; + return WriteAsyncInternal(buffer, appendNewLine: false, cancellationToken: cancellationToken); } private async Task WriteAsyncInternal(ReadOnlyMemory source, bool appendNewLine, CancellationToken cancellationToken) { + using ThrowOnWritesScope _ = GuardAgainstOtherWrites(); + int copied = 0; while (copied < source.Length) { @@ -826,10 +845,7 @@ public override Task WriteLineAsync() ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = WriteAsyncInternal(ReadOnlyMemory.Empty, appendNewLine: true, cancellationToken: default); - _asyncWriteTask = task; - - return task; + return WriteAsyncInternal(ReadOnlyMemory.Empty, appendNewLine: true, cancellationToken: default); } public override Task WriteLineAsync(char value) @@ -846,10 +862,7 @@ public override Task WriteLineAsync(char value) ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = WriteAsyncInternal(value, appendNewLine: true); - _asyncWriteTask = task; - - return task; + return WriteAsyncInternal(value, appendNewLine: true); } public override Task WriteLineAsync(string? value) @@ -871,10 +884,7 @@ public override Task WriteLineAsync(string? value) ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = WriteAsyncInternal(value.AsMemory(), appendNewLine: true, default); - _asyncWriteTask = task; - - return task; + return WriteAsyncInternal(value.AsMemory(), appendNewLine: true, default); } public override Task WriteLineAsync(char[] buffer, int index, int count) @@ -900,10 +910,7 @@ public override Task WriteLineAsync(char[] buffer, int index, int count) ThrowIfDisposed(); CheckAsyncTaskInProgress(); - Task task = WriteAsyncInternal(new ReadOnlyMemory(buffer, index, count), appendNewLine: true, cancellationToken: default); - _asyncWriteTask = task; - - return task; + return WriteAsyncInternal(new ReadOnlyMemory(buffer, index, count), appendNewLine: true, cancellationToken: default); } public override Task WriteLineAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) @@ -921,10 +928,7 @@ public override Task WriteLineAsync(ReadOnlyMemory buffer, CancellationTok return Task.FromCanceled(cancellationToken); } - Task task = WriteAsyncInternal(buffer, appendNewLine: true, cancellationToken: cancellationToken); - _asyncWriteTask = task; - - return task; + return WriteAsyncInternal(buffer, appendNewLine: true, cancellationToken: cancellationToken); } public override Task FlushAsync() @@ -936,6 +940,12 @@ public override Task FlushAsync() ThrowIfDisposed(); CheckAsyncTaskInProgress(); + + if (RuntimeHelpers.IsRuntimeAsync()) + { + return FlushAsyncInternalWithGuard(flushStream: true, flushEncoder: true, CancellationToken.None); + } + return (_asyncWriteTask = FlushAsyncInternal(flushStream: true, flushEncoder: true, CancellationToken.None)); } @@ -956,9 +966,21 @@ public override Task FlushAsync(CancellationToken cancellationToken) ThrowIfDisposed(); CheckAsyncTaskInProgress(); + + if (RuntimeHelpers.IsRuntimeAsync()) + { + return FlushAsyncInternalWithGuard(flushStream: true, flushEncoder: true, cancellationToken); + } + return (_asyncWriteTask = FlushAsyncInternal(flushStream: true, flushEncoder: true, cancellationToken)); } + private async Task FlushAsyncInternalWithGuard(bool flushStream, bool flushEncoder, CancellationToken cancellationToken) + { + using ThrowOnWritesScope _ = GuardAgainstOtherWrites(); + await FlushAsyncInternal(flushStream, flushEncoder, cancellationToken).ConfigureAwait(false); + } + private Task FlushAsyncInternal(bool flushStream, bool flushEncoder, CancellationToken cancellationToken = default) { if (cancellationToken.IsCancellationRequested)