diff --git a/src/process.tests/NullPipeConcurrencyTests.cs b/src/process.tests/NullPipeConcurrencyTests.cs new file mode 100644 index 0000000..ca3f29e --- /dev/null +++ b/src/process.tests/NullPipeConcurrencyTests.cs @@ -0,0 +1,59 @@ +// Copyright (c) The Geekeey Authors +// SPDX-License-Identifier: EUPL-1.2 + +namespace Geekeey.Process.Tests; + +internal sealed class NullPipeConcurrencyTests +{ + // A process writing enough output to fill its pipe can receive SIGPIPE if the read end is closed too early. + // Run many chatty processes at once to make premature pipe closure observable and to exercise pipe lifetime + // while processes are being started and completed concurrently. + // Test both the default targets and explicitly configured null targets, since both must keep consuming + // process output until the process exits. + + [Test] + public async Task Chatty_commands_can_run_concurrently_with_default_null_output_pipes() + { + var commands = new List(); + + for (var i = 0; i < 64; i++) + { + commands.Add(new Command(Testing.Fixture.Program.FilePath) + .WithArguments(["generate", "clob", "--length", "100000", "--lines", "2"])); + } + + var results = await Task.WhenAll(commands.Select(async command => await command.ExecuteAsync())); + + using (Assert.Multiple()) + { + foreach (var result in results) + { + await Assert.That(result.ExitCode).IsZero(); + } + } + } + + [Test] + public async Task Chatty_commands_can_run_concurrently_with_explicit_null_output_pipes() + { + var commands = new List(); + + for (var i = 0; i < 64; i++) + { + commands.Add(new Command(Testing.Fixture.Program.FilePath) + .WithArguments(["generate", "clob", "--length", "100000", "--lines", "2"]) + .WithStandardOutputPipe(PipeTarget.Null) + .WithStandardErrorPipe(PipeTarget.Null)); + } + + var results = await Task.WhenAll(commands.Select(async command => await command.ExecuteAsync())); + + using (Assert.Multiple()) + { + foreach (var result in results) + { + await Assert.That(result.ExitCode).IsZero(); + } + } + } +} diff --git a/src/process/Command.Execute.cs b/src/process/Command.Execute.cs index df140aa..80a676f 100644 --- a/src/process/Command.Execute.cs +++ b/src/process/Command.Execute.cs @@ -11,8 +11,6 @@ public sealed partial class Command return process.MainModule?.FileName; }); - private static readonly TimeSpan CancelWaitTimeout = TimeSpan.FromSeconds(5); - private static string? ProcessPath => ProcessPathLazy.Value; /// @@ -135,60 +133,90 @@ public sealed partial class Command { using var _ = process; - // timeout is triggered when the cancel timeout expires after we tried to stop the process - // -> release wait for exit and pumping tasks after that timeout - using var timeout = new CancellationTokenSource(); + // Used to trigger forceful cancellation also if an exception is thrown within this method, + // for example by the consumer-provided pipes. This ensures that the underlying process + // never outlives the execution of this method. + using var stop = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - var stdout = PipeStdOutAsync(process, timeout.Token); - var stderr = PipeStdErrAsync(process, timeout.Token); + // Used to trigger cancellation of the stdin pipe also when the process exits. The process may + // exit without fully consuming all the stdin data, so we want to avoid waiting for the + // pipe to finish if there's nothing reading from it anymore. + using var abort = CancellationTokenSource.CreateLinkedTokenSource(stop.Token); - var stdin = PipeStdInAsync(process, timeout.Token); + // Kill the process when forceful termination (via cancellation or panic) is requested + await using var watch = stop.Token.Register(process.Kill); - var pump = Task.WhenAll(stdout, stderr, stdin); + // Start piping streams in the background. In the event that any of the tasks fail, + // the corresponding standard stream(s) will be closed, so there is no risk of a deadlock. + // Output and error streams may legally outlive the process, so we don't cancel them on exit. + var stdin = PipeStdInAsync(process, abort.Token); + var stdout = PipeStdOutAsync(process, stop.Token); + var stderr = PipeStdErrAsync(process, stop.Token); - await using var registration = cancellationToken.Register(Stop, (process, timeout)); - - // wait for the process to exit or cancellation to be requested when cancellation is requested, - // we try to stop the process and then wait for it to exit with a timeout. - // When the timeout expires, we cancel the pumping tasks as well as the wait for the process exit. + // Wait for the process to exit normally or get killed + // We deliberately don't time out here because we don't want to leave a detached running process. + // If the user wants to end the execution early, they can already do so by triggering cancellation + // and killing the process. All this timeout would help us with is handle a rare edge case where + // the kill signal is sent but the process doesn't terminate for some reason. However, we don't + // really have anything we can do in that situation anyway. + var exe = process.WaitForExitAsync(CancellationToken.None); try { - await process.WaitForExitAsync(CancellationToken.None).WaitAsync(timeout.Token); - } - catch (OperationCanceledException) - { - } - - // we still wait for the pumping to complete but ignore cancellation here - try - { - await pump; - } - catch (OperationCanceledException) - { - } - - // if cancellation was requested, throw after the process was tried to stop - cancellationToken.ThrowIfCancellationRequested(); - - if (process.ExitCode is 0 || !Validation.HasFlag(ValidationMode.ZeroExitCode)) - { - return new CommandResult(process.ExitCode, process.StartTime, process.ExitTime); - } - - var message = $"Command execution failed because the underlying process ({process.FileName}#{process.Id}) " + - $"returned a non-zero exit code ({process.ExitCode})."; - throw new CommandExecutionException(this, process.ExitCode, message); - - static void Stop(object? state) - { - if (state is (Process process, CancellationTokenSource timeout)) + await foreach (var done in Task.WhenEach(stdin, stdout, stderr, exe).WithCancellation(CancellationToken.None)) { - timeout.CancelAfter(CancelWaitTimeout); - process.Kill(); + // If the process task has finished, trigger the corresponding signal to cancel the + // stdin pipe task since the process can't read any more data from it. + if (done == exe) + { + await abort.CancelAsync().ConfigureAwait(false); + } + // If a piping task failed while the process is still running, proactively terminate the process. + // It may continue running for a while even with the corresponding standard stream(s) closed, so + // we want to cut the wait short since we're going to throw an exception down the line anyway. + else if (done == stdin || done == stdout || done == stderr) + { + if (!done.IsCompletedSuccessfully && !exe.IsCompleted) + { + await stop.CancelAsync().ConfigureAwait(false); + } + } } + + // Join all tasks and propagate exceptions. If any of the tasks faulted, this will throw an aggregation + // of their exceptions. If some tasks failed while others were canceled, then the failures take + // precedence and the cancellation exceptions will be suppressed. If none of the tasks failed but + // some or all were canceled, then a single cancellation exception will be thrown. + await Task.WhenAll(stdin, stdout, stderr, exe).ConfigureAwait(false); } + catch (OperationCanceledException exception) + when (exception.CancellationToken == stop.Token || exception.CancellationToken == abort.Token) + { + // Cancellation was either requested by the consumer or triggered internally. Consumer-initiated + // cancellations will be reported separately later, while the internal ones shouldn't be reported. + } + finally + { + // The process must never outlive the execution of this method + await exe.ConfigureAwait(false); + } + + // Report forceful cancellation + if (cancellationToken.IsCancellationRequested) + { + var message = $"Command execution canceled. Underlying process ({process.FileName}#{process.Id}) was forcefully terminated."; + throw new OperationCanceledException(message, cancellationToken); + } + + // Validate the exit code if required + if (process.ExitCode is not 0 && Validation.HasFlag(ValidationMode.ZeroExitCode)) + { + var message = $"Command execution failed because the underlying process ({process.FileName}#{process.Id}) " + + $"returned a non-zero exit code ({process.ExitCode})."; + throw new CommandExecutionException(this, process.ExitCode, message); + } + + return new CommandResult(process.ExitCode, process.StartTime, process.ExitTime); } private async Task PipeStdOutAsync(Process process, CancellationToken cancellationToken = default) @@ -211,12 +239,22 @@ public sealed partial class Command { await using (process.StandardInput) { + var copyTask = StandardInputPipe.CopyToAsync(process.StandardInput, cancellationToken); + try { // Some streams do not support cancellation, so we add a fallback that drops the task and returns early. // This is important with stdin because the process might finish before the pipe has been fully // exhausted, and we don't want to wait for it. - await StandardInputPipe.CopyToAsync(process.StandardInput, cancellationToken).WaitAsync(cancellationToken); + await copyTask.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // We tried to cancel the copy task, but it may have not cooperated and may still be + // running in the background. To make sure its exception doesn't bubble up to the scheduler, + // we explicitly observe it. + _ = Catch(copyTask); + throw; } // Expect IOException: "The pipe has been ended" (Windows) or "Broken pipe" (Unix). This may happen if the // process is terminated before the pipe has been exhausted. It's not an exceptional situation because the @@ -226,6 +264,18 @@ public sealed partial class Command { // Don't catch derived exceptions, such as FileNotFoundException, to avoid false positives. } + + static async Task Catch(Task t) + { + try + { + await t; + } + catch + { + // ignored exception + } + } } } } diff --git a/src/process/PipeTarget.cs b/src/process/PipeTarget.cs index 37b6813..f9cc99f 100644 --- a/src/process/PipeTarget.cs +++ b/src/process/PipeTarget.cs @@ -119,14 +119,10 @@ public partial class PipeTarget /// Pipe target that discards all data. Functionally equivalent to a null device. /// /// - /// Using this target results in the corresponding stream (standard output or standard error) not being opened for - /// the underlying process at all. In the vast majority of cases, this behavior should be functionally equivalent to - /// piping to a null stream, but without the performance overhead of consuming and discarding unneeded data. This - /// may be undesirable in certain situations, in which case it's recommended to pipe to a null stream explicitly - /// using with . + /// The corresponding stream is consumed for the lifetime of the underlying process so that the process never writes + /// to a pipe whose read end has already been closed. /// - public static PipeTarget Null { get; } = Create((_, cancellationToken) => - !cancellationToken.IsCancellationRequested ? Task.CompletedTask : Task.FromCanceled(cancellationToken)); + public static PipeTarget Null { get; } = ToStream(Stream.Null); /// /// Creates an anonymous pipe target with the method