Detect closed pipe in redirected console output in .NET applications

Question

The .NET Console class and its default TextWriter implementation (available as Console.Out and implicitly in e.g. Console.WriteLine()) does not signal any error when the application is having its output piped to another program, and the other program terminates or closes the pipe before the application has finished. This means that the application may run for longer than necessary, writing output into a black hole.

How can I detect the closing of the other end of the redirection pipe?

A more detailed explanation follows:

Here are a pair of example programs that demonstrate the problem. Produce prints lots of integers fairly slowly, to simulate the effect of computation:

using System;
class Produce
{
    static void Main()
    {
        for (int i = 0; i < 10000; ++i)
        {
            System.Threading.Thread.Sleep(100); // added for effect
            Console.WriteLine(i);
        }
    }
}

Consume only reads the first 10 lines of input and then exits:

using System;
class Consume
{
    static void Main()
    {
        for (int i = 0; i < 10; ++i)
            Console.ReadLine();
    }
}

If these two programs are compiled, and the output of the first piped to the second, like so:

Produce | Consume

... it can be observed that Produce keeps on running long after Consume has terminated.

In reality, my Consume program is Unix-style head, and my Produce program prints data which is costly to calculate. I'd like to terminate output when the other end of the pipe has closed the connection.

How can I do this in .NET?

(I know that an obvious alternative is to pass a command-line argument to limit output, and indeed that's what I'm currently doing, but I'd still like to know how to do this since I want to be able to make more configurable judgements about when to terminate reading; e.g. piping through grep before head.)

UPDATE: It looks horribly like the System.IO.__ConsoleStream implementation in .NET is hard-coded to ignore errors 0x6D (ERROR_BROKEN_PIPE) and 0xE8 (ERROR_NO_DATA). That probably means I need to reimplement the console stream. Sigh...)

Answer 1

To solve this one, I had to write my own basic stream implementation over Win32 file handles. This wasn't terribly difficult, as I didn't need to implement asynchronous support, buffering or seeking.

Unfortunately, unsafe code needs to be used, but that generally isn't a problem for console applications that will be run locally and with full trust.

Here's the core stream:

class HandleStream : Stream
{
    SafeHandle _handle;
    FileAccess _access;
    bool _eof;

    public HandleStream(SafeHandle handle, FileAccess access)
    {
        _handle = handle;
        _access = access;
    }

    public override bool CanRead
    {
        get { return (_access & FileAccess.Read) != 0; }
    }

    public override bool CanSeek
    {
        get { return false; }
    }

    public override bool CanWrite
    {
        get { return (_access & FileAccess.Write) != 0; }
    }

    public override void Flush()
    {
        // use external buffering if you need it.
    }

    public override long Length
    {
        get { throw new NotSupportedException(); }
    }

    public override long Position
    {
        get { throw new NotSupportedException(); }
        set { throw new NotSupportedException(); }
    }

    static void CheckRange(byte[] buffer, int offset, int count)
    {
        if (offset < 0 || count < 0 || (offset + count) < 0
            || (offset + count) > buffer.Length)
            throw new ArgumentOutOfRangeException();
    }

    public bool EndOfStream
    {
        get { return _eof; }
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        CheckRange(buffer, offset, count);
        int result = ReadFileNative(_handle, buffer, offset, count);
        _eof |= result == 0;
        return result;
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        int notUsed;
        Write(buffer, offset, count, out notUsed);
    }

    public void Write(byte[] buffer, int offset, int count, out int written)
    {
        CheckRange(buffer, offset, count);
        int result = WriteFileNative(_handle, buffer, offset, count);
        _eof |= result == 0;
        written = result;
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        throw new NotSupportedException();
    }

    public override void SetLength(long value)
    {
        throw new NotSupportedException();
    }

    [return: MarshalAs(UnmanagedType.Bool)]
    [DllImport("kernel32", SetLastError=true)]
    static extern unsafe bool ReadFile(
        SafeHandle hFile, byte* lpBuffer, int nNumberOfBytesToRead,
        out int lpNumberOfBytesRead, IntPtr lpOverlapped);

    [return: MarshalAs(UnmanagedType.Bool)]
    [DllImport("kernel32.dll", SetLastError=true)]
    static extern unsafe bool WriteFile(
        SafeHandle hFile, byte* lpBuffer, int nNumberOfBytesToWrite, 
        out int lpNumberOfBytesWritten, IntPtr lpOverlapped);

    unsafe static int WriteFileNative(SafeHandle hFile, byte[] buffer, int offset, int count)
    {
        if (buffer.Length == 0)
            return 0;

        fixed (byte* bufAddr = &buffer[0])
        {
            int result;
            if (!WriteFile(hFile, bufAddr + offset, count, out result, IntPtr.Zero))
            {
                // Using Win32Exception just to get message resource from OS.
                Win32Exception ex = new Win32Exception(Marshal.GetLastWin32Error());
                int hr = ex.NativeErrorCode | unchecked((int) 0x80000000);
                throw new IOException(ex.Message, hr);
            }

            return result;
        }
    }

    unsafe static int ReadFileNative(SafeHandle hFile, byte[] buffer, int offset, int count)
    {
        if (buffer.Length == 0)
            return 0;

        fixed (byte* bufAddr = &buffer[0])
        {
            int result;
            if (!ReadFile(hFile, bufAddr + offset, count, out result, IntPtr.Zero))
            {
                Win32Exception ex = new Win32Exception(Marshal.GetLastWin32Error());
                int hr = ex.NativeErrorCode | unchecked((int) 0x80000000);
                throw new IOException(ex.Message, hr);
            }
            return result;
        }
    }
}

BufferedStream can be wrapped around it for buffering if needed, but for console output, the TextWriter will be doing character-level buffering anyway, and only flushing on newlines.

The stream abuses Win32Exception to extract an error message, rather than calling FormatMessage itself.

Building on this stream, I was able to write a simple wrapper for console I/O:

static class ConsoleStreams
{
    enum StdHandle
    {
        Input = -10,
        Output = -11,
        Error = -12,
    }

    [DllImport("kernel32.dll", SetLastError = true)]
    static extern IntPtr GetStdHandle(int nStdHandle);  

    static SafeHandle GetStdHandle(StdHandle h)
    {
        return new SafeFileHandle(GetStdHandle((int) h), true);
    }

    public static HandleStream OpenStandardInput()
    {
        return new HandleStream(GetStdHandle(StdHandle.Input), FileAccess.Read);
    }

    public static HandleStream OpenStandardOutput()
    {
        return new HandleStream(GetStdHandle(StdHandle.Output), FileAccess.Write);
    }

    public static HandleStream OpenStandardError()
    {
        return new HandleStream(GetStdHandle(StdHandle.Error), FileAccess.Write);
    }

    static TextReader _in;
    static StreamWriter _out;
    static StreamWriter _error;

    public static TextWriter Out
    {
        get
        {
            if (_out == null)
            {
                _out = new StreamWriter(OpenStandardOutput());
                _out.AutoFlush = true;
            }
            return _out;
        }
    }

    public static TextWriter Error
    {
        get
        {
            if (_error == null)
            {
                _error = new StreamWriter(OpenStandardError());
                _error.AutoFlush = true;
            }
            return _error;
        }
    }

    public static TextReader In
    {
        get
        {
            if (_in == null)
                _in = new StreamReader(OpenStandardInput());
            return _in;
        }
    }
}

The final result is that writing to the console output after the other end of the pipe has terminated the connection, results in a nice exception with the message:

The pipe is being closed

By catching and ignoring the IOException at the outermost level, it looks like I'm good to go.