Avoid materializing Task/ValueTask in StreamReader/StreamWriter#130689
Avoid materializing Task/ValueTask in StreamReader/StreamWriter#130689jakobbotsch wants to merge 4 commits into
Task/ValueTask in StreamReader/StreamWriter#130689Conversation
`StreamReader` implements a diagnostic that throws when a user tries to start multiple simultaneous read. This is not a thread safety guarantee, but rather a best-effort diagnostic. However, it causes many StreamReader APIs to force task objects to be allocated and hurts runtime async performance. This PR switches to a try-finally inside the callees instead to implement the diagnostic and to avoid forcibly allocating the Task objects.
|
Azure Pipelines: Successfully started running 3 pipeline(s). 12 pipeline(s) were filtered out due to trigger conditions. There may be pipelines that require an authorized user to comment /azp run to run. |
There was a problem hiding this comment.
Pull request overview
This PR updates StreamReader’s “async read in progress” diagnostic to avoid forcing Task materialization in async APIs by replacing the _asyncReadTask tracking approach with a lightweight _asyncIOInProgress flag guarded via try/finally-style scopes inside async callees.
Changes:
- Replaced
_asyncReadTaskwith_asyncIOInProgressand introduced a disposable guard scope to flip the flag for the duration of async operations. - Updated async read APIs (
ReadLineAsync*,ReadToEndAsync*,ReadAsync*,ReadBlockAsync*) to use the new guard pattern instead of storing a task in a field.
| private async Task<int> ReadAsyncInternalWithGuard(Memory<char> buffer, CancellationToken cancellationToken) | ||
| { | ||
| using ThrowOnReadsScope _ = GuardAgainstReads(); | ||
| return await ReadAsyncInternal(buffer, cancellationToken).ConfigureAwait(false); | ||
| } |
There was a problem hiding this comment.
I agree this seems like an oversight, but I am not looking to change the behavior in this PR.
| // 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. | ||
| private Task _asyncReadTask = Task.CompletedTask; | ||
| private bool _asyncIOInProgress; | ||
|
|
||
| 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) |
|
Tagging subscribers to this area: @dotnet/area-system-io |
Task/ValueTask in StreamReaderTask/ValueTask in StreamReader/StreamWriter
|
@EgorBot --envvars DOTNET_JitDisasm:ReadLineAsync using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes;
public class StreamReaderReadLineTests
{
private string _text;
private byte[] _bytes;
[ParamsSource(nameof(GetLineLengthRanges))]
public Range LineLengthRange { get; set; }
public static IEnumerable<Range> GetLineLengthRanges()
{
yield return new Range() { Min = 0, Max = 0 };
yield return new Range() { Min = 1, Max = 1 };
yield return new Range() { Min = 1, Max = 8 };
yield return new Range() { Min = 9, Max = 32 };
yield return new Range() { Min = 33, Max = 128 };
yield return new Range() { Min = 129, Max = 1024 };
yield return new Range() { Min = 1025, Max = 2048 };
yield return new Range() { Min = 0, Max = 1024 };
}
public class Range
{
public int Min { get; set; }
public int Max { get; set; }
public override string ToString() => $"[{Min,4}, {Max,4}]";
}
[GlobalSetup]
public void GlobalSetup()
{
_text = GenerateLinesText(LineLengthRange, 16 * 1024);
_bytes = Encoding.UTF8.GetBytes(_text);
}
[Benchmark]
public async Task ReadLineAsync()
{
using (StreamReader reader = new StreamReader(new MemoryStream(_bytes)))
{
while (await reader.ReadLineAsync() != null) ;
}
}
private static string GenerateLinesText(Range lineLengthRange, int textTargetLength)
{
int min = lineLengthRange.Min;
int max = lineLengthRange.Max;
string newLine = Environment.NewLine;
var sb = new StringBuilder(textTargetLength + max + newLine.Length);
var random = new Random(42);
while (sb.Length < textTargetLength)
{
int charsCount = random.Next(min, max);
for (int c = 0; c < charsCount; c++)
sb.Append((char)random.Next('0', 'z'));
sb.Append(newLine);
}
return sb.ToString();
}
} |
|
PTAL @dotnet/area-system-io |
|
Azure Pipelines: Successfully started running 3 pipeline(s). 12 pipeline(s) were filtered out due to trigger conditions. There may be pipelines that require an authorized user to comment /azp run to run. |
|
Workflow state for the Holistic Review Orchestrator. {
"version": 5,
"last_dispatched_commit": "f64e6ef660a5406196ace7960851cce705ef97bc",
"last_dispatched_base_ref": "main",
"last_dispatched_base_sha": "5a5020c676517ba2496973a42edc9cf3dbee851c",
"last_reviewed_commit": "f64e6ef660a5406196ace7960851cce705ef97bc",
"last_reviewed_base_ref": "main",
"last_reviewed_base_sha": "5a5020c676517ba2496973a42edc9cf3dbee851c",
"last_recorded_worker_run_id": "29687158585",
"review_attempt_commit": "",
"review_attempt_base_ref": "",
"review_attempt_count": 0,
"max_review_attempts": 5,
"review_history_format": "holistic-review-disclosure-v1",
"review_history": [
{
"commit": "f64e6ef660a5406196ace7960851cce705ef97bc",
"review_id": 4730769278
}
]
} |
There was a problem hiding this comment.
Holistic Review
Motivation: The problem is real and well-justified. StreamReader/StreamWriter implement a best-effort same-thread concurrent-IO diagnostic by storing the returned Task in _asyncReadTask/_asyncWriteTask and checking IsCompleted. Under runtime async this forces materialization of Task/ValueTask objects that would otherwise be elided, and the linked benchmark shows measurable allocation/throughput cost.
Approach: Sound and minimal. A bool _asyncIOInProgress flag toggled by a readonly struct ... : IDisposable scope (using) inside the internal async methods replaces the stored-Task check. CheckAsyncTaskInProgress now ORs the flag with the legacy Task check, so both mechanisms coexist. The RuntimeHelpers.IsRuntimeAsync() intrinsic gates the new path only for the methods (ReadAsync(char[]), ReadBlockAsync, FlushAsync) that would otherwise need an extra wrapper state machine under async1 — keeping the old zero-overhead Task-return path for non-runtime-async callers. Methods that are already bespoke async state machines (ReadLineAsyncInternal, WriteAsyncInternal, etc.) get the using unconditionally at no extra allocation cost.
Summary: ✅ LGTM. The refactor preserves the existing diagnostic semantics while removing forced Task materialization under runtime async. I verified reentrancy safety, synchronous-completion behavior, and async1 path equivalence (details below). No blocking issues found.
Detailed Findings
✅ Correctness — Diagnostic semantics preserved
The flag is set at entry to each internal async method via the using scope and cleared on Dispose, including on exception (struct implements IDisposable, using guarantees cleanup). For synchronous completion, the async state machine runs to completion before returning, so _asyncIOInProgress is back to false by the time the (completed) task is returned — matching the old behavior where a completed _asyncReadTask also reported IsCompleted == true. For suspension, the flag stays true until the operation resumes and completes, matching the old incomplete-Task behavior.
✅ Reentrancy — No false InvalidOperation_AsyncIOInProgress
I traced the nested-call risk: ReadBlockAsyncWithGuard sets the flag, then awaits base.ReadBlockAsync → TextReader.ReadBlockAsyncInternal → StreamReader.ReadAsyncInternal. None of ReadBufferAsync, ReadAsyncInternal, or ReadBlockAsyncInternal calls CheckAsyncTaskInProgress, so a guard held on an outer frame does not cause an inner frame to throw. Only the public entry points call the check, and they set the guard after checking. This is correct.
✅ async1 path unchanged
RuntimeHelpers.IsRuntimeAsync() returns false outside runtime async, so ReadAsync(char[]), ReadBlockAsync(char[]), ReadBlockAsync(Memory), and both FlushAsync overloads fall through to the original _asyncReadTask = task / _asyncWriteTask = task code, preserving the previous allocation profile and diagnostic behavior for existing callers.
💡 Tests — No new tests, acceptable for a behavior-preserving refactor
The change is a semantics-preserving refactor rather than a behavior change, and the existing InvalidOperation_AsyncIOInProgress diagnostic tests continue to exercise both branches of CheckAsyncTaskInProgress for the non-runtime-async path. If there is a runtime-async test leg, it would be worth confirming it exercises the guarded paths, but adding dedicated tests is not required here.
💡 Observation — ReadAsync(Memory<char>) remains unguarded (pre-existing)
The ReadAsync(Memory<char>) overload returns ReadAsyncInternal(...) directly without setting any in-progress marker. This is pre-existing behavior unchanged by this PR (out of scope), noted only for completeness.
Note
This review was generated by this repository's Holistic Review agentic workflow to complement the built-in Copilot review.
Generated by Holistic Review · 99 AIC · ⌖ 10.9 AIC · ⊞ 10K
|
|
||
| return task; | ||
|
|
||
| async Task<int> ReadAsyncInternalWithGuard(Memory<char> buffer, CancellationToken cancellationToken) |
There was a problem hiding this comment.
ReadAsyncInternalWithGuard is used in only one place. Perhaps it will be cleaner to just manually "inline" the using instead of making a helper?
There was a problem hiding this comment.
Actually - maybe it is a better to avoid introducing the helpers and just switch everything to use using/_asyncIOInProgress pattern - regardless async1 or not?
The whole thing with the preexisting pattern and a task attached to a reader for unknown duration seems a bit odd.
Is there a reason to keep it (other than local function helpers)?
StreamReaderandStreamWriterimplement a diagnostic that throws when a user tries to start multiple simultaneous reads/writes. This is not a thread safety guarantee, but rather a best-effort diagnostic. However, it causes many of their APIs to force task objects to be allocated that would otherwise be avoided when called from runtime async.This PR switches to a try-finally inside the callees instead to implement the diagnostic and to avoid forcibly allocating the Task objects. To avoid regressing async1 callers we keep the
Task-based approach in the subset of cases where a wrapping continuation/state machine would otherwise be created with the try-finally approach.Benchmark: EgorBot/Benchmarks#357 (comment)