runtime: Console.ReadLine() is blocked in Linux even though the StandardInput of the process has been disposed

I’m working on a program that chains the standard output and standard input of a collection of processes, similar to the piping functionality in a shell. I use a worker thread to relay bytes between the output of upstream process and the input of the downstream process. When _upstream.StandardOutput.BaseStream.Read return 0, the worker thread dispose the StandardInput of the downstream process, and in that case, Console.ReadLine() in the downstream process is supposed to return null, so that the downstream process knows there is no more input and starts to wrap up and exit.

This works fine in Linux when 2 processes are chained, however, when more than 2 processes are chained, Console.ReadLine() in the downstream process continues to block even though the StandardInput of the process has been disposed. However, the same code works fine on windows when chaining any numbers of processes.

Repro Code

worker-thread-pipe: work-thread.exe

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Runtime.InteropServices;
using System.Threading;

namespace ConsoleApplication
{
    public class Program
    {
        /// <summary>
        /// Number of child processes
        /// </summary>
        private static int childCount = 0;

        /// <summary>
        /// Wait for all processes to finish
        /// </summary>
        private static ManualResetEventSlim eventSlim = new ManualResetEventSlim();

        /// <summary>
        /// Chain native commands.
        /// e.g. copy-async "C:\WINDOWS\system32\ipconfig.exe" "C:\Program Files (x86)\Microsoft VS Code\bin\cat.exe"
        /// </summary>
        /// <remarks>
        /// Expect executables and their arguments to be represented in this way:
        ///   [executable]|[arguments]
        /// </remarks>
        public static void Main(string[] args)
        {
            if (args.Length == 0)
            {
                // Use default command chain
                if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
                {
                    Console.WriteLine("Demo piping behavior: 'ipconfig.exe | cat.exe'");
                    args = new string[] { @"C:\WINDOWS\system32\ipconfig.exe",
                                          @"C:\Program Files (x86)\Microsoft VS Code\bin\cat.exe" };
                }
                else if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
                {
                    Console.WriteLine("Demo pipeing behavior: 'ifconfig | cat'");
                    args = new string[] { "/sbin/ifconfig",
                                          "/bin/cat" };
                }
            }

            List<Process> processes = new List<Process>();
            for (int i = 0; i < args.Length; i++)
            {
                ProcessStartInfo startInfo = null;
                // Expect executables and their arguments to be represented in this way: [executable]|[arguments]
                int colonIndex = args[i].IndexOf('|');
                if (colonIndex == -1 && !string.IsNullOrWhiteSpace(args[i]))
                {
                    startInfo = new ProcessStartInfo(args[i]);
                }
                else if (colonIndex > 0)
                {
                    startInfo = new ProcessStartInfo(
                                    args[i].Substring(0, colonIndex),
                                    args[i].Substring(colonIndex + 1));
                }
                else
                {
                    throw new ArgumentException("args are incorrect.");
                }

                if (i > 0) { startInfo.RedirectStandardInput = true; }
                if (i < args.Length - 1) { startInfo.RedirectStandardOutput = true; }

                Process proc = new Process() { StartInfo = startInfo, EnableRaisingEvents = true };
                proc.Exited += new EventHandler(ProcessExitHandler);

                processes.Add(proc);
            }

            childCount = processes.Count;
            List<ProcessPipe> pipes = new List<ProcessPipe>();

            Process upstream = null;
            for (int i = 0; i < processes.Count; i++)
            {
                Process proc = processes[i];
                proc.Start();

                if (upstream != null)
                {
                    var pipe = new ProcessPipe(upstream, proc);
                    pipe.Start();
                    pipes.Add(pipe);
                }
                upstream = proc;
            }

            foreach (var pipe in pipes)
            {
                pipe.Wait();
            }

            // Wait for all child process to finish
            eventSlim.Wait();
        }

        internal static void ProcessExitHandler(object sender, System.EventArgs e)
        {
            // Set the event if it's the last child process
            if (Interlocked.Decrement(ref childCount) == 0)
                eventSlim.Set();
        }
    }

    internal class ProcessPipe
    {
        private byte[] _buffer;
        private Process _upstream, _downstream;
        private Thread _thread;

        internal ProcessPipe(Process upstream, Process downstream)
        {
            _buffer = new byte[1024];
            _upstream = upstream;
            _downstream = downstream;
        }

        private void StartPiping()
        {
            int count = 0;
            try {
                while ((count = _upstream.StandardOutput.BaseStream.Read(_buffer, 0, 1024)) > 0)
                {
                    _downstream.StandardInput.BaseStream.Write(_buffer, 0, count);
                    _downstream.StandardInput.BaseStream.Flush();
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("Exception when reading/writing from worker-thread: {0}", ex.Message);
            }
            finally
            {
                Console.WriteLine("{0} exiting, disposing StandardInput of process '{1} {2}'", _thread.Name, _downstream.StartInfo.FileName, _downstream.StartInfo.Arguments);
                try {
                    _downstream.StandardInput.Dispose();
                } catch (Exception ex) {
                    Console.WriteLine("Disposeing StandardInput failed: {0}", ex.Message);
                }
            }
        }

        internal void Start()
        {
            _thread = new Thread(new ThreadStart(StartPiping));
            _thread.Name = string.Format("{0} -> {1} :Piping Thread", _upstream.StartInfo.FileName, _downstream.StartInfo.FileName);
            _thread.Start();
        }

        internal void Wait()
        {
            _thread.Join();
        }
    }
}

slow-outputer: slow.exe

using System;
using System.Threading;

namespace NativePipe
{
    public class Slow
    {
        private static string[] outputs = new string[] {
            "This is a console app to write output in slow manner",
            "Hello world",
            "C# across platforms is awesome!",
            "We cannot go back time",
            "Unless we have a time machine~",
            "One last output :)"
        };

        public static void Main(string[] args)
        {

            foreach (string str in outputs)
            {
                Console.WriteLine(str);
                Console.Out.Flush();
                Thread.Sleep(2000);
            }
        }
    }
}

mimic cat.exe with logging: mycat.exe

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication
{
    public class Program
    {
        public static void Main(string[] args)
        {
            if (args.Length != 1)
            {
                Console.Error.WriteLine("Wrong Usage!");
                return;
            }

            bool first = true;
            string fileName = args[0];
            string envVarName = RuntimeInformation.IsOSPlatform(OSPlatform.Windows) ? "USERPROFILE" : "HOME";
            string homeDir = Environment.GetEnvironmentVariable(envVarName);
            string tempDir = Path.Combine(homeDir, "temp");
            if (!Directory.Exists(tempDir)) { Directory.CreateDirectory(tempDir); }

            using (FileStream fs = new FileStream(Path.Combine(tempDir, fileName), FileMode.Create))
            using (StreamWriter sw = new StreamWriter(fs, Encoding.UTF8){AutoFlush = true})
            {
                sw.WriteLine("IsInputRedirected? {0}", Console.IsInputRedirected);
                sw.WriteLine("IsOutputRedirected? {0}", Console.IsOutputRedirected);
                sw.WriteLine("IsErrorRedirected? {0}", Console.IsErrorRedirected);

                while (true)
                {
                    if (first) {
                        first = false;
                        sw.WriteLine("-- About to read from stdin --");
                    }
                    string line = Console.ReadLine(); // <== Block here in Linux even though StandardInput of the process has been disposed
                    if (line == null)
                    {
                        sw.WriteLine("Console.In closed.");
                        break;
                    }
                    else
                    {
                        sw.WriteLine(line);
                        Console.WriteLine(line);
                        Console.Out.Flush();
                        Thread.Sleep(1000);
                    }
                }
            }
        }
    }
}

project.json

{
    "name": <executable-name>, 
    "version": "1.0.0-*",

    "buildOptions": {
        "debugType": "portable",
        "emitEntryPoint": true
    },

    "frameworks": {
        "netcoreapp1.1": {
            "imports": [ "dnxcore50" ],
            "dependencies": {
                "Microsoft.NETCore.App": "1.1.0-preview1-001100-00"
            }
        }
    },

    "runtimes": {
        "win10-x64": { },
        "ubuntu.14.04-x64": { }
    }
}

Behavior on Windows

Chaining 3 processes runs successfully:

## First add path to slow.exe and mycat.exe to environment variable PATH
E:\arena\dotnetApp\pipe-threading\bin\Debug\netcoreapp1.1\win10-x64>work-thread.exe slow "mycat|cat1" "mycat|cat2"
This is a console app to write output in slow manner
Hello world
C# across platforms is awesome!
We cannot go back time
Unless we have a time machine~
One last output :)
slow -> mycat :Piping Thread exiting, disposing StandardInput of process 'mycat cat1'
mycat -> mycat :Piping Thread exiting, disposing StandardInput of process 'mycat cat2'

Logging from mycat.exe shows that Console.ReadLine() == null was hit in both mycat.exe processes (note the line Console.In closed. in the logs): cat1

IsInputRedirected? True
IsOutputRedirected? True
IsErrorRedirected? False
-- About to read from stdin --
This is a console app to write output in slow manner
Hello world
C# across platforms is awesome!
We cannot go back time
Unless we have a time machine~
One last output :)
Console.In closed.

cat2

IsInputRedirected? True
IsOutputRedirected? False
IsErrorRedirected? False
-- About to read from stdin --
This is a console app to write output in slow manner
Hello world
C# across platforms is awesome!
We cannot go back time
Unless we have a time machine~
One last output :)
Console.In closed.

Behavior on Linux

Chaining the same 3 processes hangs:

## First add path to slow and mycat to environment variable PATH
user@machine:~/arena/pipe-threading/bin/Debug/netcoreapp1.1/ubuntu.14.04-x64$ ./work-thread slow "mycat|cat1" "mycat|cat2"
This is a console app to write output in slow manner
Hello world
C# across platforms is awesome!
We cannot go back time
Unless we have a time machine~
One last output :)
slow -> mycat :Piping Thread exiting, disposing StandardInput of process 'mycat cat1'
<hangs here>

Logging from mycat shows that Console.ReadLine() == null was NOT hit in both mycat processes (note the line Console.In closed. is missing in the logs): cat1

user@machine:~/temp$ cat cat1
IsInputRedirected? True
IsOutputRedirected? True
IsErrorRedirected? False
-- About to read from stdin --
This is a console app to write output in slow manner
Hello world
C# across platforms is awesome!
We cannot go back time
Unless we have a time machine~
One last output :)

cat2

user@machine:~/temp$ cat cat2
IsInputRedirected? True
IsOutputRedirected? False
IsErrorRedirected? False
-- About to read from stdin --
This is a console app to write output in slow manner
Hello world
C# across platforms is awesome!
We cannot go back time
Unless we have a time machine~
One last output :)