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 { ///