From bf2d00c383022a52a54d046c318206f74b00d18b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 10:16:03 +0000 Subject: [PATCH] [patch] Never run the standard output and standard error delegates at once The two pipes are read concurrently, and each read loop called its delegate from its own continuation, so the two delegates ran inside each other on different thread-pool threads. A caller sharing one List or StringBuilder between them lost lines with no exception. Delivery now takes one lock per run, so the reads stay concurrent while the delegates run one at a time, and each stream keeps its own order. Fixes #95 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017UuQscGFrjteUf9Ui3jWdd --- RunCommand.Test/RunCommandTests.cs | 50 ++++++++++++++++++++++++++ RunCommand/AsyncProcessStreamReader.cs | 12 +++++-- RunCommand/OutputHandler.cs | 5 +++ 3 files changed, 65 insertions(+), 2 deletions(-) diff --git a/RunCommand.Test/RunCommandTests.cs b/RunCommand.Test/RunCommandTests.cs index 767cfbc..64562b3 100644 --- a/RunCommand.Test/RunCommandTests.cs +++ b/RunCommand.Test/RunCommandTests.cs @@ -1178,6 +1178,56 @@ public async Task ExecuteAsyncShouldDeliverOutputStillInThePipeWhenTheProcessExi } } + [TestMethod] + public async Task ExecuteAsyncShouldNeverRunTheStandardOutputAndStandardErrorDelegatesAtOnce() + { + if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows)) + { + Assert.Inconclusive("Needs a shell that interleaves many lines across both streams. The delivery this covers is in platform independent code, so the other legs cover it."); + } + + // Both delegates add to one List, which is not thread safe, exactly as a caller capturing + // a command's whole output would. Run concurrently, overlapping adds overwrite each other's + // slots and lines vanish without an exception. + const int linesPerStream = 5000; + + // Whether the delegates overlap depends on how the two reads are scheduled, so one clean + // run proves little. Repeat it. + for (int attempt = 0; attempt < 5; attempt++) + { + List lines = []; + int active = 0; + int overlaps = 0; + + void Add(string line) + { + if (Interlocked.Increment(ref active) > 1) + { + _ = Interlocked.Increment(ref overlaps); + } + + lines.Add(line); + Thread.SpinWait(50); + _ = Interlocked.Decrement(ref active); + } + + int exitCode = await RunCommand.ExecuteAsync( + "sh", + ["-c", $"i=1; while [ $i -le {linesPerStream} ]; do echo out$i; echo err$i >&2; i=$((i+1)); done"], + new LineOutputHandler(Add, Add)).ConfigureAwait(false); + + Assert.AreEqual(0, exitCode); + Assert.AreEqual(0, overlaps, $"The delegates ran inside each other {overlaps} times on attempt {attempt + 1}."); + Assert.HasCount(linesPerStream * 2, lines, $"Lines were lost on attempt {attempt + 1}."); + + // Serialising delivery must not reorder either stream. + string[] expectedOut = [.. Enumerable.Range(1, linesPerStream).Select(i => $"out{i}")]; + string[] expectedErr = [.. Enumerable.Range(1, linesPerStream).Select(i => $"err{i}")]; + Assert.AreSequenceEqual(expectedOut, [.. lines.Where(l => l.StartsWith("out", StringComparison.Ordinal))]); + Assert.AreSequenceEqual(expectedErr, [.. lines.Where(l => l.StartsWith("err", StringComparison.Ordinal))]); + } + } + [TestMethod] [DoNotParallelize] public async Task ExecuteAsyncShouldNotSpinWhileACommandThatClosedItsOutputKeepsRunning() diff --git a/RunCommand/AsyncProcessStreamReader.cs b/RunCommand/AsyncProcessStreamReader.cs index 9f0c80e..39fced5 100644 --- a/RunCommand/AsyncProcessStreamReader.cs +++ b/RunCommand/AsyncProcessStreamReader.cs @@ -15,6 +15,11 @@ internal sealed class AsyncProcessStreamReader(Process process, OutputHandler ou private bool outputHasEmitted; private bool errorHasEmitted; + // The two pipes are read concurrently, but their chunks are handed over one at a time. Without + // this the standard output and standard error delegates ran inside each other on two thread-pool + // threads, and a caller sharing one List or StringBuilder between them lost lines silently. + private readonly Lock deliveryLock = new(); + // Read the raw pipes with the caller's encoding rather than through // process.StandardOutput/StandardError. Process builds those StreamReaders with byte-order-mark // detection switched on, which replaces the encoding the caller asked for whenever the output @@ -152,9 +157,12 @@ private async Task ReadToEnd(StreamReader streamReader, char[] buffer, Action 0) + if (data.Length > 0 && onData is not null) { - onData?.Invoke(data); + lock (deliveryLock) + { + onData(data); + } } } } diff --git a/RunCommand/OutputHandler.cs b/RunCommand/OutputHandler.cs index 6d75258..647835e 100644 --- a/RunCommand/OutputHandler.cs +++ b/RunCommand/OutputHandler.cs @@ -8,6 +8,11 @@ namespace ktsu.RunCommand; /// Handles the output from a command line process, processing both standard output and standard error streams. /// It processes the data in undelimited chunks, calling the specified actions for each chunk received. /// +/// +/// The two streams are read concurrently, but the actions are never run at the same time: each one +/// returns before the other is called, so both may share state, such as one list collecting all of a +/// command's output, without locking. Each stream's chunks arrive in the order they were written. +/// public class OutputHandler { ///