From 6399c4cbf178036498b648b73db9dbd87c3bc165 Mon Sep 17 00:00:00 2001 From: Jordan Brown Date: Sun, 26 Jul 2020 13:31:39 -0400 Subject: [PATCH] Process output streaming fixes --- src/Tgstation.Server.Api/ApiHeaders.cs | 4 +- .../Components/Deployment/DreamMaker.cs | 3 +- .../Components/Session/SessionController.cs | 26 +-- .../Session/SessionControllerFactory.cs | 12 +- .../Components/Session/SessionPersistor.cs | 1 + .../Components/StaticFiles/Configuration.cs | 2 +- src/Tgstation.Server.Host/System/IProcess.cs | 22 +-- .../System/IProcessExecutor.cs | 10 +- .../System/PosixProcessFeatures.cs | 5 +- src/Tgstation.Server.Host/System/Process.cs | 103 ++++++------ .../System/ProcessExecutor.cs | 150 +++++++++--------- .../System/WindowsProcessFeatures.cs | 5 + .../Instance/WatchdogTest.cs | 4 +- .../Tgstation.Server.Tests/IntegrationTest.cs | 4 +- 14 files changed, 180 insertions(+), 171 deletions(-) diff --git a/src/Tgstation.Server.Api/ApiHeaders.cs b/src/Tgstation.Server.Api/ApiHeaders.cs index 9308ad9976..cecfa88a5a 100644 --- a/src/Tgstation.Server.Api/ApiHeaders.cs +++ b/src/Tgstation.Server.Api/ApiHeaders.cs @@ -1,4 +1,4 @@ -using Microsoft.AspNetCore.Http.Headers; +using Microsoft.AspNetCore.Http.Headers; using Microsoft.Extensions.Primitives; using Microsoft.Net.Http.Headers; using System; @@ -138,7 +138,7 @@ namespace Tgstation.Server.Api void AddError(HeaderTypes headerType, string message) { if (badHeaders != HeaderTypes.None) - errorBuilder.Append(Environment.NewLine); + errorBuilder.AppendLine(); badHeaders |= headerType; errorBuilder.Append(message); } diff --git a/src/Tgstation.Server.Host/Components/Deployment/DreamMaker.cs b/src/Tgstation.Server.Host/Components/Deployment/DreamMaker.cs index 34b5896aac..3f75c34356 100644 --- a/src/Tgstation.Server.Host/Components/Deployment/DreamMaker.cs +++ b/src/Tgstation.Server.Host/Components/Deployment/DreamMaker.cs @@ -292,7 +292,8 @@ namespace Tgstation.Server.Host.Components.Deployment cancellationToken.ThrowIfCancellationRequested(); logger.LogDebug("DreamMaker exit code: {0}", exitCode); - currentDreamMakerOutput = job.Output = dm.GetCombinedOutput(); + job.Output = await dm.GetCombinedOutput(cancellationToken).ConfigureAwait(false); + currentDreamMakerOutput = job.Output; logger.LogDebug("DreamMaker output: {0}{1}", Environment.NewLine, job.Output); return exitCode; } diff --git a/src/Tgstation.Server.Host/Components/Session/SessionController.cs b/src/Tgstation.Server.Host/Components/Session/SessionController.cs index 396fb24346..c5c10b6719 100644 --- a/src/Tgstation.Server.Host/Components/Session/SessionController.cs +++ b/src/Tgstation.Server.Host/Components/Session/SessionController.cs @@ -257,21 +257,21 @@ namespace Tgstation.Server.Host.Components.Session if (disposed) return; disposed = true; - - logger.LogTrace("Disposing..."); - if (!released) - { - process.Terminate(); - byondLock.Dispose(); - } - - process.Dispose(); - bridgeRegistration?.Dispose(); - reattachInformation.Dmb?.Dispose(); // will be null when released - chatTrackingContext.Dispose(); - reattachTopicCts.Dispose(); } + logger.LogTrace("Disposing..."); + if (!released) + { + process.Terminate(); + byondLock.Dispose(); + } + + process.Dispose(); + bridgeRegistration?.Dispose(); + reattachInformation.Dmb?.Dispose(); // will be null when released + chatTrackingContext.Dispose(); + reattachTopicCts.Dispose(); + if (!released) { // finish the async callback diff --git a/src/Tgstation.Server.Host/Components/Session/SessionControllerFactory.cs b/src/Tgstation.Server.Host/Components/Session/SessionControllerFactory.cs index d1a19a4ea2..c003e78441 100644 --- a/src/Tgstation.Server.Host/Components/Session/SessionControllerFactory.cs +++ b/src/Tgstation.Server.Host/Components/Session/SessionControllerFactory.cs @@ -267,8 +267,9 @@ namespace Tgstation.Server.Host.Components.Session async Task GetDDOutput() { + // DCT x2: None available if (!platformIdentifier.IsWindows) - return process.GetCombinedOutput(); + return await process.GetCombinedOutput(default).ConfigureAwait(false); var logFilePath = ioManager.ConcatPath(dmbProvider.Directory, logFileGuid.ToString()); try @@ -347,9 +348,12 @@ namespace Tgstation.Server.Host.Components.Session } catch { - process.Terminate(); - process.Dispose(); - throw; + using (process) + { + process.Terminate(); + await process.Lifetime.ConfigureAwait(false); + throw; + } } } catch diff --git a/src/Tgstation.Server.Host/Components/Session/SessionPersistor.cs b/src/Tgstation.Server.Host/Components/Session/SessionPersistor.cs index de8cc15cab..a0513fad10 100644 --- a/src/Tgstation.Server.Host/Components/Session/SessionPersistor.cs +++ b/src/Tgstation.Server.Host/Components/Session/SessionPersistor.cs @@ -133,6 +133,7 @@ namespace Tgstation.Server.Host.Components.Session { using var process = processExecutor.GetProcess(reattachInfo.ProcessId); process.Terminate(); + await process.Lifetime.ConfigureAwait(false); } catch (Exception ex) { diff --git a/src/Tgstation.Server.Host/Components/StaticFiles/Configuration.cs b/src/Tgstation.Server.Host/Components/StaticFiles/Configuration.cs index f08ecdf70b..c18aee9d6a 100644 --- a/src/Tgstation.Server.Host/Components/StaticFiles/Configuration.cs +++ b/src/Tgstation.Server.Host/Components/StaticFiles/Configuration.cs @@ -488,7 +488,7 @@ namespace Tgstation.Server.Host.Components.StaticFiles { var exitCode = await script.Lifetime.ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); - var scriptOutput = script.GetCombinedOutput(); + var scriptOutput = await script.GetCombinedOutput(cancellationToken).ConfigureAwait(false); if (exitCode != 0) throw new JobException($"Script {I} exited with code {exitCode}:{Environment.NewLine}{scriptOutput}"); else diff --git a/src/Tgstation.Server.Host/System/IProcess.cs b/src/Tgstation.Server.Host/System/IProcess.cs index d961b65694..8392d58377 100644 --- a/src/Tgstation.Server.Host/System/IProcess.cs +++ b/src/Tgstation.Server.Host/System/IProcess.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Threading; using System.Threading.Tasks; @@ -22,24 +22,28 @@ namespace Tgstation.Server.Host.System /// /// Get the stderr output of the /// - /// The stderr output of the - string GetErrorOutput(); + /// The for the operation. + /// A resulting in the stderr output of the + Task GetErrorOutput(CancellationToken cancellationToken); /// /// Get the stdout output of the /// - /// The stdout output of the - string GetStandardOutput(); + /// The for the operation. + /// A resulting in the stdout output of the + Task GetStandardOutput(CancellationToken cancellationToken); /// /// Get the stderr and stdout output of the /// - /// The stderr and stdout output of the - string GetCombinedOutput(); + /// The for the operation. + /// A resulting in the stderr and stdout output of the + Task GetCombinedOutput(CancellationToken cancellationToken); /// - /// Terminates the process + /// Asycnhronously terminates the process. /// + /// To ensure the has ended, use the . void Terminate(); /// @@ -49,4 +53,4 @@ namespace Tgstation.Server.Host.System /// A resulting in the name of the account executing the . Task GetExecutingUsername(CancellationToken cancellationToken); } -} \ No newline at end of file +} diff --git a/src/Tgstation.Server.Host/System/IProcessExecutor.cs b/src/Tgstation.Server.Host/System/IProcessExecutor.cs index e7f0d98388..12bfb896a9 100644 --- a/src/Tgstation.Server.Host/System/IProcessExecutor.cs +++ b/src/Tgstation.Server.Host/System/IProcessExecutor.cs @@ -1,4 +1,4 @@ -namespace Tgstation.Server.Host.System +namespace Tgstation.Server.Host.System { /// /// For launching ' @@ -15,7 +15,13 @@ /// If standard error should be read /// If shell execute should not be used. Must be set if or are set. /// A new - IProcess LaunchProcess(string fileName, string workingDirectory, string arguments = null, bool readOutput = false, bool readError = false, bool noShellExecute = false); + IProcess LaunchProcess( + string fileName, + string workingDirectory, + string arguments = null, + bool readOutput = false, + bool readError = false, + bool noShellExecute = false); /// /// Get a representing the running executable. diff --git a/src/Tgstation.Server.Host/System/PosixProcessFeatures.cs b/src/Tgstation.Server.Host/System/PosixProcessFeatures.cs index fd120064b8..c06c99a812 100644 --- a/src/Tgstation.Server.Host/System/PosixProcessFeatures.cs +++ b/src/Tgstation.Server.Host/System/PosixProcessFeatures.cs @@ -73,6 +73,9 @@ namespace Tgstation.Server.Host.System if (!await ioManager.FileExists(GCorePath, cancellationToken).ConfigureAwait(false)) throw new JobException(ErrorCode.MissingGCore); + if(process.HasExited) + throw new JobException(ErrorCode.DreamDaemonOffline); + var pid = process.Id; string output; int exitCode; @@ -87,7 +90,7 @@ namespace Tgstation.Server.Host.System using (cancellationToken.Register(() => gcoreProc.Terminate())) exitCode = await gcoreProc.Lifetime.ConfigureAwait(false); - output = gcoreProc.GetCombinedOutput(); + output = await gcoreProc.GetCombinedOutput(cancellationToken).ConfigureAwait(false); logger.LogDebug("gcore output:{0}{1}", Environment.NewLine, output); } diff --git a/src/Tgstation.Server.Host/System/Process.cs b/src/Tgstation.Server.Host/System/Process.cs index ae5e5e27cb..32cca93336 100644 --- a/src/Tgstation.Server.Host/System/Process.cs +++ b/src/Tgstation.Server.Host/System/Process.cs @@ -4,17 +4,13 @@ using System.Diagnostics; using System.Text; using System.Threading; using System.Threading.Tasks; +using Tgstation.Server.Host.Extensions; namespace Tgstation.Server.Host.System { /// sealed class Process : IProcess { - /// - /// Maximum time to wait in a call to . - /// - const int MaximumWaitMilliseconds = 30000; - /// public int Id { get; } @@ -36,13 +32,8 @@ namespace Tgstation.Server.Host.System readonly global::System.Diagnostics.Process handle; - /// - /// A so that we can complete if the becomes unresponsive. - /// - readonly TaskCompletionSource emergencyLifetimeTcs; - - readonly StringBuilder outputStringBuilder; - readonly StringBuilder errorStringBuilder; + readonly Task standardOutputTask; + readonly Task standardErrorTask; readonly StringBuilder combinedStringBuilder; /// @@ -51,8 +42,8 @@ namespace Tgstation.Server.Host.System /// The value of /// The value of /// The value of - /// The value of - /// The value of + /// The value of + /// The value of /// The value of /// The value of /// If was NOT just created @@ -60,40 +51,48 @@ namespace Tgstation.Server.Host.System IProcessFeatures processFeatures, global::System.Diagnostics.Process handle, Task lifetime, - StringBuilder outputStringBuilder, - StringBuilder errorStringBuilder, + Task standardOutputTask, + Task standardErrorTask, StringBuilder combinedStringBuilder, ILogger logger, bool preExisting) { - this.processFeatures = processFeatures ?? throw new ArgumentNullException(nameof(processFeatures)); this.handle = handle ?? throw new ArgumentNullException(nameof(handle)); - this.outputStringBuilder = outputStringBuilder; - this.errorStringBuilder = errorStringBuilder; + // Do this fast because the runtime will bitch if we try to access it after it ends + Id = handle.Id; + + this.processFeatures = processFeatures ?? throw new ArgumentNullException(nameof(processFeatures)); + + this.standardOutputTask = standardOutputTask; + this.standardErrorTask = standardErrorTask; this.combinedStringBuilder = combinedStringBuilder; this.logger = logger ?? throw new ArgumentNullException(nameof(logger)); - emergencyLifetimeTcs = new TaskCompletionSource(); Lifetime = WrapLifetimeTask(lifetime ?? throw new ArgumentNullException(nameof(lifetime))); - Id = handle.Id; - if (preExisting) { Startup = Task.CompletedTask; return; } - Startup = Task.Factory.StartNew(() => - { - try + Startup = Task.Factory.StartNew( + () => { - handle.WaitForInputIdle(); - } - catch (InvalidOperationException) { } - }, default, TaskCreationOptions.LongRunning, TaskScheduler.Current); + try + { + handle.WaitForInputIdle(); + } + catch (InvalidOperationException ex) + { + logger.LogDebug(ex, "Error on WaitForInputIdle()!"); + } + }, + default, // DCT: None available + TaskCreationOptions.LongRunning, + TaskScheduler.Current); logger.LogTrace("Created process ID: {0}", Id); } @@ -103,60 +102,48 @@ namespace Tgstation.Server.Host.System async Task WrapLifetimeTask(Task lifetimeTask) { - await Task.WhenAny(lifetimeTask, emergencyLifetimeTcs.Task).ConfigureAwait(false); - if (lifetimeTask.IsCompleted) - { - var exitCode = await lifetimeTask.ConfigureAwait(false); - logger.LogTrace("PID {0} exited with code {1}", Id, exitCode); - return exitCode; - } - - logger.LogTrace("Using exit code -1 for hung PID {0}.", Id); - return -1; + // relevant: https://stackoverflow.com/a/26722542 + var exitCode = await lifetimeTask.ConfigureAwait(false); + logger.LogTrace("PID {0} exited with code {1}", Id, exitCode); + return exitCode; } /// - public string GetCombinedOutput() + public async Task GetCombinedOutput(CancellationToken cancellationToken) { if (combinedStringBuilder == null) - throw new InvalidOperationException("Output/Error reading was not enabled!"); + throw new InvalidOperationException("Output/Error stream reading was not enabled!"); + await Task.WhenAll(standardOutputTask, standardErrorTask).WithToken(cancellationToken).ConfigureAwait(false); return combinedStringBuilder.ToString().TrimStart(Environment.NewLine.ToCharArray()); } /// - public string GetErrorOutput() + public Task GetErrorOutput(CancellationToken cancellationToken) { - if (errorStringBuilder == null) - throw new InvalidOperationException("Error reading was not enabled!"); - return errorStringBuilder.ToString().TrimStart(Environment.NewLine.ToCharArray()); + if (standardErrorTask == null) + throw new InvalidOperationException("Error stream reading was not enabled!"); + return standardErrorTask.WithToken(cancellationToken); } /// - public string GetStandardOutput() + public Task GetStandardOutput(CancellationToken cancellationToken) { - if (outputStringBuilder == null) - throw new InvalidOperationException("Output reading was not enabled!"); - return outputStringBuilder.ToString().TrimStart(Environment.NewLine.ToCharArray()); + if (standardOutputTask == null) + throw new InvalidOperationException("Output stream reading was not enabled!"); + return standardOutputTask.WithToken(cancellationToken); } /// public void Terminate() { - if (handle.HasExited) + if (Lifetime.IsCompleted) return; try { logger.LogTrace("Terminating PID {0}...", Id); handle.Kill(); - if (!handle.WaitForExit(MaximumWaitMilliseconds)) - { - logger.LogError( - "PID {0} hasn't exited in {1} seconds! This may cause issues with port reuse.", - Id, - TimeSpan.FromMilliseconds(MaximumWaitMilliseconds).TotalSeconds); - emergencyLifetimeTcs.TrySetResult(null); - } + // DO NOT USE WaitForExit! https://stackoverflow.com/a/26722542 } catch (Exception e) { diff --git a/src/Tgstation.Server.Host/System/ProcessExecutor.cs b/src/Tgstation.Server.Host/System/ProcessExecutor.cs index 444e6eda9a..4c0bc5f8c8 100644 --- a/src/Tgstation.Server.Host/System/ProcessExecutor.cs +++ b/src/Tgstation.Server.Host/System/ProcessExecutor.cs @@ -1,5 +1,6 @@ using Microsoft.Extensions.Logging; using System; +using System.IO; using System.Text; using System.Threading.Tasks; @@ -34,19 +35,36 @@ namespace Tgstation.Server.Host.System var tcs = new TaskCompletionSource(); handle.Exited += (a, b) => { - int exitCode; try { - exitCode = handle.ExitCode; - } - catch (InvalidOperationException) - { - return; - } + if (tcs.Task.IsCompleted) + { + logger.LogTrace("Skipping process exit handler as the TaskCompletionSource is already set"); + return; + } - // Try because this can be invoked twice for weird reasons - if (tcs.TrySetResult(exitCode)) - logger.LogTrace("Process exit event completed"); + try + { + var exitCode = handle.ExitCode; + + // Try because this can be invoked twice for weird reasons + if (tcs.TrySetResult(exitCode)) + logger.LogTrace("Process termination event completed"); + else + logger.LogTrace("Ignoring duplicate process termination event"); + } + catch (InvalidOperationException ex) + { + if (!tcs.Task.IsCompleted) + throw; + + logger.LogTrace(ex, "Ignoring expected exception!"); + } + } + catch(Exception ex) + { + logger.LogError(ex, "Process exit handler exception!"); + } }; return tcs.Task; @@ -128,92 +146,72 @@ namespace Tgstation.Server.Host.System handle.StartInfo.UseShellExecute = !noShellExecute; - StringBuilder outputStringBuilder = null, errorStringBuilder = null, combinedStringBuilder = null; + StringBuilder combinedStringBuilder = null; - TaskCompletionSource outputReadTcs = null; - TaskCompletionSource errorReadTcs = null; + Task outputTask = null; + Task errorTask = null; + TaskCompletionSource processStartTcs = null; if (readOutput || readError) { combinedStringBuilder = new StringBuilder(); + processStartTcs = new TaskCompletionSource(); + + async Task ConsumeReader(Func readerFunc) + { + var stringBuilder = new StringBuilder(); + string text; + + await processStartTcs.Task.ConfigureAwait(false); + + var reader = readerFunc(); + while ((text = await reader.ReadLineAsync().ConfigureAwait(false)) != null) + { + combinedStringBuilder.AppendLine(); + combinedStringBuilder.Append(text); + stringBuilder.AppendLine(); + stringBuilder.Append(text); + } + + return stringBuilder.ToString(); + } + if (readOutput) { - outputStringBuilder = new StringBuilder(); + outputTask = ConsumeReader(() => handle.StandardOutput); handle.StartInfo.RedirectStandardOutput = true; - outputReadTcs = new TaskCompletionSource(); - handle.OutputDataReceived += (sender, e) => - { - if (e.Data == null) - { - outputReadTcs.SetResult(null); - return; - } - - combinedStringBuilder.Append(Environment.NewLine); - combinedStringBuilder.Append(e.Data); - outputStringBuilder.Append(Environment.NewLine); - outputStringBuilder.Append(e.Data); - }; } if (readError) { - errorStringBuilder = new StringBuilder(); + errorTask = ConsumeReader(() => handle.StandardError); handle.StartInfo.RedirectStandardError = true; - errorReadTcs = new TaskCompletionSource(); - handle.ErrorDataReceived += (sender, e) => - { - if (e.Data == null) - { - errorReadTcs.SetResult(null); - return; - } - - combinedStringBuilder.Append(Environment.NewLine); - combinedStringBuilder.Append(e.Data); - errorStringBuilder.Append(Environment.NewLine); - errorStringBuilder.Append(e.Data); - }; } } var lifetimeTask = AttachExitHandler(handle); - handle.Start(); - - static async Task AddToLifetimeTask(Task originalTask, TaskCompletionSource tcs) - { - var exitCode = await originalTask.ConfigureAwait(false); - await tcs.Task.ConfigureAwait(false); - return exitCode; - } - try { - if (readOutput) - { - handle.BeginOutputReadLine(); - lifetimeTask = AddToLifetimeTask(lifetimeTask, outputReadTcs); - } - } - catch (InvalidOperationException) { } - try - { - if (readError) - { - handle.BeginErrorReadLine(); - lifetimeTask = AddToLifetimeTask(lifetimeTask, errorReadTcs); - } - } - catch (InvalidOperationException) { } + handle.Start(); - return new Process( - processFeatures, - handle, - lifetimeTask, - outputStringBuilder, - errorStringBuilder, - combinedStringBuilder, - loggerFactory.CreateLogger(), false); + var process = new Process( + processFeatures, + handle, + lifetimeTask, + outputTask, + errorTask, + combinedStringBuilder, + loggerFactory.CreateLogger(), false); + + processStartTcs?.SetResult(null); + + return process; + } + catch (Exception ex) + { + processStartTcs?.SetException(ex); + throw; + } } catch { diff --git a/src/Tgstation.Server.Host/System/WindowsProcessFeatures.cs b/src/Tgstation.Server.Host/System/WindowsProcessFeatures.cs index 6739f12b4c..d31d7c61d4 100644 --- a/src/Tgstation.Server.Host/System/WindowsProcessFeatures.cs +++ b/src/Tgstation.Server.Host/System/WindowsProcessFeatures.cs @@ -6,6 +6,8 @@ using System.Linq; using System.Management; using System.Threading; using System.Threading.Tasks; +using Tgstation.Server.Api.Models; +using Tgstation.Server.Host.Jobs; namespace Tgstation.Server.Host.System { @@ -91,6 +93,9 @@ namespace Tgstation.Server.Host.System => Task.Factory.StartNew( () => { + if (process.HasExited) + throw new JobException(ErrorCode.DreamDaemonOffline); + using var fileStream = new FileStream(outputFile, FileMode.CreateNew); if (!NativeMethods.MiniDumpWriteDump( process.Handle, diff --git a/tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs b/tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs index 7b39618540..681c4020cd 100644 --- a/tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs +++ b/tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs @@ -87,7 +87,7 @@ namespace Tgstation.Server.Tests.Instance while (!dumpTask.IsCompleted) KillDD(false); var job = await WaitForJob(await dumpTask, 10, true, null, cancellationToken); - Assert.IsTrue(job.ErrorCode == ErrorCode.DreamDaemonOffline || job.ErrorCode == ErrorCode.GCoreFailure); + Assert.IsTrue(job.ErrorCode == ErrorCode.DreamDaemonOffline || job.ErrorCode == ErrorCode.GCoreFailure, $"{job.ErrorCode}: {job.ExceptionDetails}"); await Task.Delay(TimeSpan.FromSeconds(20), cancellationToken); var ddStatus = await instanceClient.DreamDaemon.Read(cancellationToken); @@ -209,7 +209,7 @@ namespace Tgstation.Server.Tests.Instance ourProcessHandler.Suspend(); - await Task.WhenAny(ourProcessHandler.Lifetime, Task.Delay(TimeSpan.FromSeconds(20))); + await Task.WhenAny(ourProcessHandler.Lifetime, Task.Delay(TimeSpan.FromMinutes(1))); var timeout = 20; do diff --git a/tests/Tgstation.Server.Tests/IntegrationTest.cs b/tests/Tgstation.Server.Tests/IntegrationTest.cs index 496fa12d7a..127235d59d 100644 --- a/tests/Tgstation.Server.Tests/IntegrationTest.cs +++ b/tests/Tgstation.Server.Tests/IntegrationTest.cs @@ -434,8 +434,8 @@ namespace Tgstation.Server.Tests var exitCode = await process.Lifetime.WithToken(cts.Token); Assert.AreEqual(0, exitCode); - Assert.AreEqual(String.Empty, process.GetErrorOutput().Trim()); - Assert.AreEqual("Hello World!", process.GetStandardOutput().Trim()); + Assert.AreEqual(String.Empty, (await process.GetErrorOutput(default)).Trim()); + Assert.AreEqual("Hello World!", (await process.GetStandardOutput(default)).Trim()); } } }