Use dedicated threads to read the redirected output and error streams from the child process for out-of-proc jobs (#11713)

src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs

@@ -1098,7 +1098,6 @@ namespace System.Management.Automation.Remoting.Client
                 {
                     _processCreated = false;
                 }
-                // _processInstance.Start();
             }
 
             PSEtwLog.LogAnalyticInformational(PSEventId.WSManCreateShell, PSOpcode.Connect,
@@ -1123,28 +1122,10 @@ namespace System.Management.Automation.Remoting.Client
                         _processInstance.RunspacePool.Dispose();
                     }
 
-                    stdInWriter = _processInstance.StdInWriter;
-                    // if (stdInWriter == null)
-                    {
-                        _serverProcess.OutputDataReceived += new DataReceivedEventHandler(OnOutputDataReceived);
-                        _serverProcess.ErrorDataReceived += new DataReceivedEventHandler(OnErrorDataReceived);
-                    }
-
                     _serverProcess.Exited += new EventHandler(OnExited);
-
-                    // serverProcess.Start();
                     _processInstance.Start();
 
-                    if (stdInWriter != null)
-                    {
-                        _serverProcess.CancelErrorRead();
-                        _serverProcess.CancelOutputRead();
-                    }
-
-                    // Start asynchronous reading of output/errors
-                    _serverProcess.BeginOutputReadLine();
-                    _serverProcess.BeginErrorReadLine();
-
+                    StartRedirectionReaderThreads(_serverProcess);
                     stdInWriter = new OutOfProcessTextWriter(_serverProcess.StandardInput);
                     _processInstance.StdInWriter = stdInWriter;
                 }
@@ -1172,6 +1153,86 @@ namespace System.Management.Automation.Remoting.Client
             SendOneItem();
         }
 
+        private void StartRedirectionReaderThreads(Process serverProcess)
+        {
+            Thread outputThread = new Thread(ProcessOutputData);
+            outputThread.IsBackground = true;
+            outputThread.Name = "Out-of-Proc Job Output Thread";
+
+            Thread errorThread = new Thread(ProcessErrorData);
+            errorThread.IsBackground = true;
+            errorThread.Name = "Out-of-Proc Job Error Thread";
+
+            outputThread.Start(serverProcess.StandardOutput);
+            errorThread.Start(serverProcess.StandardError);
+        }
+
+        private void ProcessOutputData(object arg)
+        {
+            if (arg is StreamReader reader)
+            {
+                try
+                {
+                    string data = reader.ReadLine();
+                    while (data != null)
+                    {
+                        HandleOutputDataReceived(data);
+                        data = reader.ReadLine();
+                    }
+                }
+                catch (IOException)
+                {
+                    // Treat this as EOF, the same as what 'Process.BeginOutputReadLine()' does.
+                }
+                catch (Exception e)
+                {
+                    _tracer.WriteMessage(
+                        "OutOfProcessClientSessionTransportManager",
+                        "ProcessOutputThread",
+                        Guid.Empty,
+                        "Transport manager output reader thread ended with error: {0}",
+                        e.Message ?? string.Empty);
+                }
+            }
+            else
+            {
+                Dbg.Assert(false, "Invalid argument. Expecting a StreamReader object.");
+            }
+        }
+
+        private void ProcessErrorData(object arg)
+        {
+            if (arg is StreamReader reader)
+            {
+                try
+                {
+                    string data = reader.ReadLine();
+                    while (data != null)
+                    {
+                        HandleErrorDataReceived(data);
+                        data = reader.ReadLine();
+                    }
+                }
+                catch (IOException)
+                {
+                    // Treat this as EOF, the same as what 'Process.BeginErrorReadLine()' does.
+                }
+                catch (Exception e)
+                {
+                    _tracer.WriteMessage(
+                        "OutOfProcessClientSessionTransportManager",
+                        "ProcessErrorThread",
+                        Guid.Empty,
+                        "Transport manager error reader thread ended with error: {0}",
+                        e.Message ?? string.Empty);
+                }
+            }
+            else
+            {
+                Dbg.Assert(false, "Invalid argument. Expecting a StreamReader object.");
+            }
+        }
+
         /// <summary>
         /// Kills the server process and disposes other resources.
         /// </summary>
@@ -1198,20 +1259,6 @@ namespace System.Management.Automation.Remoting.Client
 
         #endregion
 
-        #region Event Handlers
-
-        private void OnOutputDataReceived(object sender, DataReceivedEventArgs e)
-        {
-            HandleOutputDataReceived(e.Data);
-        }
-
-        private void OnErrorDataReceived(object sender, DataReceivedEventArgs e)
-        {
-            HandleErrorDataReceived(e.Data);
-        }
-
-        #endregion
-
         #region Helper Methods
 
         private void KillServerProcess()
@@ -1230,13 +1277,8 @@ namespace System.Management.Automation.Remoting.Client
 
                     if (_processCreated)
                     {
-                        _serverProcess.CancelOutputRead();
-                        _serverProcess.CancelErrorRead();
                         _serverProcess.Kill();
                     }
-
-                    _serverProcess.OutputDataReceived -= new DataReceivedEventHandler(OnOutputDataReceived);
-                    _serverProcess.ErrorDataReceived -= new DataReceivedEventHandler(OnErrorDataReceived);
                 }
             }
             catch (System.ComponentModel.Win32Exception)

test/powershell/engine/Job/Jobs.Tests.ps1

@@ -338,4 +338,58 @@ Describe 'Basic Job Tests' -Tags 'Feature' {
             ValidateJobInfo -job $jobToStop -state 'Stopped' -hasMoreData $false
         }
     }
+
+    Context 'Background pwsh process should terminate after job is done' {
+        It "Can clean up background pwsh process after job is done" {
+            $job = Start-Job { $pid }
+            $processId = Receive-Job $job -Wait
+
+            try {
+                $process = Get-Process -Id $processId -ErrorAction Stop
+                Wait-UntilTrue { $process.HasExited } -IntervalInMilliseconds 300 | Should -BeTrue
+            } catch {
+                $_.FullyQualifiedErrorId | Should -BeExactly 'NoProcessFoundForGivenId,Microsoft.PowerShell.Commands.GetProcessCommand'
+            }
+
+            Remove-Job $job -Force
+        }
+
+        It "Can clean up background pwsh process when job is stopped" {
+            $job = Start-Job { $pid; Start-Sleep -Second 10 }
+
+            # Wait for the pid to be received.
+            Wait-UntilTrue { [bool](Receive-Job $job -Keep) } | Should -BeTrue
+            $processId = Receive-Job $job
+
+            # Stop the job and wait for the cleanup to finish.
+            Stop-Job $job
+
+            try {
+                $process = Get-Process -Id $processId -ErrorAction Stop
+                Wait-UntilTrue { $process.HasExited } -IntervalInMilliseconds 300 | Should -BeTrue
+            } catch {
+                $_.FullyQualifiedErrorId | Should -BeExactly 'NoProcessFoundForGivenId,Microsoft.PowerShell.Commands.GetProcessCommand'
+            }
+
+            Remove-Job $job -Force
+        }
+
+        It "Can clean up background pwsh process when job is removed" {
+            $job = Start-Job { $pid; Start-Sleep -Second 10 }
+
+            # Wait for the pid to be received.
+            Wait-UntilTrue { [bool](Receive-Job $job -Keep) } | Should -BeTrue
+            $processId = Receive-Job $job
+
+            # Remove the job and wait for the cleanup to finish.
+            Remove-Job $job -Force
+
+            try {
+                $process = Get-Process -Id $processId -ErrorAction Stop
+                Wait-UntilTrue { $process.HasExited } -IntervalInMilliseconds 300 | Should -BeTrue
+            } catch {
+                $_.FullyQualifiedErrorId | Should -BeExactly 'NoProcessFoundForGivenId,Microsoft.PowerShell.Commands.GetProcessCommand'
+            }
+        }
+    }
 }