diff --git a/Build/dlr-patches/0104-SharedIO-output-set-by-host.patch b/Build/dlr-patches/0104-SharedIO-output-set-by-host.patch new file mode 100644 index 000000000..53af2efca --- /dev/null +++ b/Build/dlr-patches/0104-SharedIO-output-set-by-host.patch @@ -0,0 +1,64 @@ +From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001 +From: Largo +Date: Sun, 8 Nov 2026 02:30:00 +0100 +Subject: SharedIO: say whether a host set the output streams + +Console.OpenStandardOutput returns Stream.Null when it cannot use the +handle (a pipe whose reader has gone), and a host that wants the output +discarded passes Stream.Null to SetOutput itself: the same object, so a +language cannot tell the two apart. IsOutputSetByHost and +IsErrorOutputSetByHost say which it is. + +Co-Authored-By: Claude Opus 5.5 +--- +diff --git a/src/core/Microsoft.Scripting/Runtime/SharedIO.cs b/src/core/Microsoft.Scripting/Runtime/SharedIO.cs +index 4ab7605..d2dfa35 100644 +--- a/src/core/Microsoft.Scripting/Runtime/SharedIO.cs ++++ b/src/core/Microsoft.Scripting/Runtime/SharedIO.cs +@@ -80,6 +80,21 @@ namespace Microsoft.Scripting.Runtime { + + private Encoding? _inputEncoding; + ++ // set by SetOutput / SetErrorOutput, cleared by RedirectToConsole ++ private bool _outputSetByHost; ++ private bool _errorSetByHost; ++ ++ /// ++ /// Whether is one a host set with , rather ++ /// than the one opened for the process's standard output. The two can be the same object: ++ /// Console.OpenStandardOutput returns Stream.Null when it cannot use the handle, and a host ++ /// that wants the output discarded passes Stream.Null itself. ++ /// ++ public bool IsOutputSetByHost => _outputSetByHost; ++ ++ /// As , for . ++ public bool IsErrorOutputSetByHost => _errorSetByHost; ++ + public Stream InputStream { get { InitializeInput(); return _inputStream!; } } + public Stream OutputStream { get { InitializeOutput(); return _outputStream!; } } + public Stream ErrorStream { get { InitializeErrorOutput(); return _errorStream!; } } +@@ -184,6 +199,7 @@ namespace Microsoft.Scripting.Runtime { + lock (_mutex) { + _outputStream = stream; + _outputWriter = writer; ++ _outputSetByHost = true; + } + } + +@@ -192,6 +208,7 @@ namespace Microsoft.Scripting.Runtime { + lock (_mutex) { + _errorStream = stream; + _errorWriter = writer; ++ _errorSetByHost = true; + } + } + +@@ -213,6 +230,8 @@ namespace Microsoft.Scripting.Runtime { + _inputReader = null; + _outputWriter = null; + _errorWriter = null; ++ _outputSetByHost = false; ++ _errorSetByHost = false; + } + } + diff --git a/Build/dlr-patches/README.md b/Build/dlr-patches/README.md index 4d10cac22..a8e540ef5 100644 --- a/Build/dlr-patches/README.md +++ b/Build/dlr-patches/README.md @@ -25,6 +25,7 @@ the apply step instead of with a confusing CS1061 deep inside IronRuby. | `0101-startup.patch` | Startup costs the DLR adds to every process: no COM type scan of every loaded assembly (nothing reads its cache); no culture-sensitive casing or comparison in option parsing and `OpenFileStream`, which made ICU open a collator; the namespace tree reads an assembly's public type names from its metadata instead of loading every exported type (a listed type that does not load, such as a forwarder to an assembly that is not there, is skipped when the namespace is enumerated, as it was before); the call-instruction shapes the interpreter builds with `MakeGenericType` are rooted so ReadyToRun precompiles them; user-defined conversion operator lookups are cached; `BackgroundCompilation.HoldBack()` lets a host hold back the interpreter's background compilations (IronRuby does while the prelude loads); `DeclaredMemberNames.MayDeclareMethod` answers from metadata whether a type declares a method name; no CoreLib collection over a DLR struct on the startup path (their code would be JIT-compiled). | | `0102-jit-on-fresh-stack.patch` | The CLR's JIT also recurses over the trees of the IL it compiles, and needs far more stack per level than the light compiler: compiling a lambda nested ~1500 Ruby levels deep (an array literal, say) overflowed a 1 MB stack - a Windows main thread's, or the thread `DeepNesting1` runs on - in `DynamicMethod.CreateDelegate`, which no managed check can catch. The light compiler now records how deep the tree went, and `LightDelegateCreator.Compile`, through which both synchronous and background compiles go, compiles a lambda deeper than 400 levels on a fresh stack (`StackGuard`, from `0101`). Hardly anything is that deep: in IronRuby.Tests, 3 of 4609 compiled lambdas, all from `DeepNesting1`. | | `0103-reflection-call-rethrow.patch` | The interpreter calls a delegate with more parameters than it has call instructions for - a Ruby method with ten or so parameters or more - through `MethodInfo.Invoke`, and rethrew what the call raised with `throw inner`, which starts the exception's stack trace over: the backtrace lost every frame between the `raise` and the caller's call site, and the frames below the rescuing one (`
`). It now rethrows through `ExceptionDispatchInfo`, which keeps them; IronRuby hides the reflection frames themselves. | +| `0104-SharedIO-output-set-by-host.patch` | `SharedIO.IsOutputSetByHost` / `IsErrorOutputSetByHost`: whether a host set the stream with `SetOutput`. On Windows .NET opens a standard output whose pipe reader has gone as `Stream.Null`, and IronRuby then writes to the handle itself so that the write reports EPIPE; a host that passes `Stream.Null` to discard the output is the same object, and has to stay a discard. | `Util/check-dlr-patches.sh DLR_DIR` checks that a DLR checkout has all of these applied: a local one is patched once and falls behind when a new patch lands here. diff --git a/Src/IronRuby.Tests/RubyTests.cs b/Src/IronRuby.Tests/RubyTests.cs index 5f111b930..52cbc8513 100644 --- a/Src/IronRuby.Tests/RubyTests.cs +++ b/Src/IronRuby.Tests/RubyTests.cs @@ -948,6 +948,8 @@ public Tests(Driver/*!*/ driver) { Thread_HostMainThread1, Thread_HostMainThread2, Windows_NotImplementedAndLinks1, + Windows_PipeBlocking1, + Windows_StdoutPipeSysWrite1, Windows_SymLinks1, Range_IntegerFastPaths1, Io_BuiltinReaders1, diff --git a/Src/IronRuby.Tests/Runtime/MiscTests.cs b/Src/IronRuby.Tests/Runtime/MiscTests.cs index a4429403c..9a7fab903 100644 --- a/Src/IronRuby.Tests/Runtime/MiscTests.cs +++ b/Src/IronRuby.Tests/Runtime/MiscTests.cs @@ -4053,6 +4053,164 @@ public void Thread_StopAlone1() { "); } + /// + /// IO.pipe on Windows is a CreatePipe pipe (WindowsPipeStream). A thread blocked reading it used to + /// be Running to the CLR, so Thread#stop? never said true and ruby/spec's `Thread.pass until + /// t.stop?` spun for ever; closing the read end from another thread did not reach the blocked read; + /// and #read_nonblock / #write_nonblock blocked on an empty / full pipe. Windows only. + /// + public void Windows_PipeBlocking1() { + if (Environment.OSVersion.Platform != PlatformID.Win32NT) { + return; + } + TestOutput(@" +require 'ruby4' +# bounded, so that a thread that never looks blocked fails the test instead of hanging it +def wait_for(seconds = 10); deadline = Time.now + seconds; Thread.pass until yield || Time.now > deadline; end +r, w = IO.pipe +t = Thread.new { begin; r.read; rescue IOError => e; e.message; end } +wait_for { t.stop? } +puts ""blocked reader: #{t.status.inspect}"" +r.close +puts ""closed underneath: #{(t.join(10) ? t.value : :still_blocked).inspect}"" +w.close +r, w = IO.pipe +puts ""empty: #{begin; r.read_nonblock(5); rescue IO::WaitReadable; :wait_readable; end.inspect}"" +puts ""empty, exception: false: #{r.read_nonblock(5, exception: false).inspect}"" +w.write ""ab"" +puts ""two bytes: #{r.read_nonblock(5).inspect}"" +n = 0 +full = begin; loop { n += w.write_nonblock('a' * 10_000) }; rescue IO::WaitWritable; :wait_writable; end +puts ""full: #{full.inspect} after #{n > 0 ? 'some' : 'no'} bytes"" +puts ""full, exception: false: #{w.write_nonblock('a', exception: false).inspect}"" +r.close +puts ""reader gone: #{begin; w.write_nonblock('a'); rescue SystemCallError => e; e.class; end}"" +w.close +r, w = IO.pipe +writer = Thread.new { w.write('a' * 100_000) } +wait_for { writer.stop? } +started = Time.now +other = begin; w.write_nonblock('b'); rescue IO::WaitWritable; :wait_writable; end +puts ""behind a blocked writer: #{other.inspect}, at once: #{Time.now - started < 2}"" +r.read(100_000) +writer.join +r.close; w.close +", @" +blocked reader: ""sleep"" +closed underneath: ""stream closed in another thread"" +empty: :wait_readable +empty, exception: false: :wait_readable +two bytes: ""ab"" +full: :wait_writable after some bytes +full, exception: false: :wait_writable +reader gone: Errno::EPIPE +behind a blocked writer: :wait_writable, at once: true +"); + } + + [System.Runtime.InteropServices.DllImport("kernel32.dll")] + private static extern IntPtr GetStdHandle(int nStdHandle); + + [System.Runtime.InteropServices.DllImport("kernel32.dll")] + private static extern bool SetStdHandle(int nStdHandle, IntPtr handle); + + [System.Runtime.InteropServices.DllImport("kernel32.dll")] + private static extern bool PeekNamedPipe(Microsoft.Win32.SafeHandles.SafePipeHandle handle, IntPtr buffer, int size, + IntPtr bytesRead, out int bytesAvailable, IntPtr bytesLeft); + + /// + /// STDOUT redirected to a pipe on Windows. The process's standard output handle is pointed at a + /// pipe for the length of the test, and SharedIO opens it as it opens standard output (A, B), or + /// is given a stream by the host (C, D). + /// A: put into non-blocking mode with io/nonblock, #syswrite on a full pipe takes what fits and + /// then raises IO::WaitWritable, as write(2) on an O_NONBLOCK descriptor 1 does. It used to + /// block, because STDOUT's stream is the console stream and #syswrite did not look behind it + /// for the pipe. + /// B: when the pipe's reader has already gone, .NET opens standard output as Stream.Null; the + /// write still has to reach the handle and report EPIPE, where it was thrown away silently. + /// C: a host that gives SharedIO Stream.Null itself wants the output discarded: nothing + /// reaches the pipe, in blocking or non-blocking mode. D: nor does it get EPIPE when the pipe's + /// reader has gone - the same Stream.Null object as B's, which is why SharedIO says which it is + /// (IsOutputSetByHost, Build/dlr-patches/0104). Windows only. + /// + public void Windows_StdoutPipeSysWrite1() { + if (Environment.OSVersion.Platform != PlatformID.Win32NT) { + return; + } + const int STD_OUTPUT_HANDLE = -11; + IntPtr saved = GetStdHandle(STD_OUTPUT_HANDLE); + Action redirect = (handle, output) => { + SetStdHandle(STD_OUTPUT_HANDLE, handle); + ConsoleStream.ResetStandardDescriptors(); + if (output == null) { + Runtime.IO.RedirectToConsole(); + } else { + Runtime.IO.SetOutput(output, System.IO.TextWriter.Null); + } + }; + // Nothing may be printed in here: standard output is the pipe under test. + try { + // A: a pipe nobody reads, filled + using (var pipe = new System.IO.Pipes.AnonymousPipeServerStream(System.IO.Pipes.PipeDirection.In)) { + redirect(pipe.ClientSafePipeHandle.DangerousGetHandle(), null); + var result = Engine.Execute(@" +require 'io/nonblock' +STDOUT.nonblock = true +begin + first = STDOUT.syswrite('a' * 2_000_000) + second = begin; STDOUT.syswrite('b' * 10); rescue IO::WaitWritable; 'wait_writable'; end + ""#{first > 0 && first < 2_000_000 ? 'short write' : first} then #{second}"" +ensure + STDOUT.nonblock = false +end +").ToString(); + Assert(result == "short write then wait_writable", result); + } + + // B: the write end of a pipe whose reader has gone + using (var pipe = new System.IO.Pipes.AnonymousPipeServerStream(System.IO.Pipes.PipeDirection.Out)) { + pipe.DisposeLocalCopyOfClientHandle(); + redirect(pipe.SafePipeHandle.DangerousGetHandle(), null); + Assert(Console.OpenStandardOutput() == System.IO.Stream.Null, "expected .NET to open a broken pipe as Stream.Null"); + var result = Engine.Execute(@" +begin; STDOUT.syswrite('x'); 'written'; rescue Errno::EPIPE; 'epipe'; end +").ToString(); + Assert(result == "epipe", result); + } + + // C: a pipe nobody reads, and a host that discards the output + using (var pipe = new System.IO.Pipes.AnonymousPipeServerStream(System.IO.Pipes.PipeDirection.In)) { + redirect(pipe.ClientSafePipeHandle.DangerousGetHandle(), System.IO.Stream.Null); + Engine.Execute(@" +require 'io/nonblock' +STDOUT.write('hidden'); STDOUT.flush +STDOUT.syswrite('x') +STDOUT.nonblock = true +STDOUT.syswrite('y') +STDOUT.nonblock = false +"); + int available; + Assert(PeekNamedPipe(pipe.SafePipeHandle, IntPtr.Zero, 0, IntPtr.Zero, out available, IntPtr.Zero)); + Assert(available == 0, available + " bytes reached the pipe"); + } + + // D: the write end of a pipe whose reader has gone, and a host that discards the output + using (var pipe = new System.IO.Pipes.AnonymousPipeServerStream(System.IO.Pipes.PipeDirection.Out)) { + pipe.DisposeLocalCopyOfClientHandle(); + redirect(pipe.SafePipeHandle.DangerousGetHandle(), System.IO.Stream.Null); + var result = Engine.Execute(@" +STDOUT.write('hidden'); STDOUT.flush +begin; STDOUT.syswrite('x').to_s; rescue Errno::EPIPE, SignalException => e; e.class.name; end +").ToString(); + Assert(result == "1", result); + } + } finally { + SetStdHandle(STD_OUTPUT_HANDLE, saved); + ConsoleStream.ResetStandardDescriptors(); + Runtime.IO.RedirectToConsole(); + } + } + /// /// What MRI does on Windows with the functions win32/Makefile.sub leaves out, and with the ones /// win32.c supplies. Dir.fchdir, Dir.for_fd, Dir#fileno, Dir.chroot, File.lutime, File.mkfifo and the diff --git a/Src/Libraries/Builtins/IoOps.cs b/Src/Libraries/Builtins/IoOps.cs index 42cd26bbd..b27d1050e 100644 --- a/Src/Libraries/Builtins/IoOps.cs +++ b/Src/Libraries/Builtins/IoOps.cs @@ -1762,14 +1762,28 @@ public static int SysWrite(BinaryOpStorage/*!*/ writeStorage, ConversionStorage< // MRI's syswrite is one write(2), so a pipe that only had room for part of the string // reports what it took rather than waiting for the rest to fit. - var pipe = DescriptorStream.Of(self.GetStream().BaseStream); + var baseStream = self.GetStream().BaseStream; + var pipe = (ISingleWriteStream)DescriptorStream.Of(baseStream) ?? baseStream as ISingleWriteStream; + // Windows has no O_NONBLOCK on the handle for the write to go by; io/nonblock's + // #nonblock= remembers the flag in the IO instead (see io/nonblock.rb). It is this IO's + // and this call's: another IO over the same pipe may be in the other mode. The pipe can + // also be the one behind STDOUT or STDERR, redirected; a blocking syswrite there keeps + // going through the standard stream as before. + object nonblock; + var windowsPipe = context.TryGetInstanceVariable(self, "@__nonblock__", out nonblock) && nonblock is bool && (bool)nonblock + ? WindowsPipeStream.Of(baseStream) : null; + bool nonBlocking = windowsPipe != null; int count = val.GetByteCount(); - if (pipe != null && count > 0) { - // the write(2) blocks on a full pipe, so IO#timeout is waited out before it - self.WaitForStream(false); + if ((pipe != null || nonBlocking) && count > 0) { + if (!nonBlocking) { + // the write(2) blocks on a full pipe, so IO#timeout is waited out before it + self.WaitForStream(false); + } int written; try { - written = pipe.WriteOnce(val.ToByteArray(), 0, count); + written = nonBlocking + ? windowsPipe.WriteNonBlocking(val.ToByteArray(), 0, count) + : pipe.WriteOnce(val.ToByteArray(), 0, count); } catch (IOException e) { throw TranslateStreamError(e); } @@ -2054,7 +2068,7 @@ private static MutableString ReadOnceWithoutWaiting(RubyIO/*!*/ io, int count, M return (queued == 0 && buffered == 0) ? null : buffer; } - var pipe = DescriptorStream.Of(stream.BaseStream); + var pipe = (INonBlockingReadStream)DescriptorStream.Of(stream.BaseStream) ?? stream.BaseStream as INonBlockingReadStream; if (pipe == null) { // A Windows pipe has no descriptor, and a ReadFile on an empty one waits for the // writer however long that is, but PeekNamedPipe says whether anything is there. diff --git a/Src/Libraries/Builtins/WindowsPipeWriteStream.cs b/Src/Libraries/Builtins/WindowsPipeWriteStream.cs deleted file mode 100644 index 9a4b38b84..000000000 --- a/Src/Libraries/Builtins/WindowsPipeWriteStream.cs +++ /dev/null @@ -1,99 +0,0 @@ -/* **************************************************************************** - * - * Copyright (c) IronRuby contributors. - * - * This source code is subject to terms and conditions of the Apache License, Version 2.0. A - * copy of the license can be found in the License.html file at the root of this distribution. If - * you cannot locate the Apache License, Version 2.0, please send an email to - * ironruby@microsoft.com. By using this source code in any fashion, you are agreeing to be bound - * by the terms of the Apache License, Version 2.0. - * - * You must not remove this notice, or any other, from this software. - * - * ***************************************************************************/ - -using System; -using System.IO; -using System.Threading; -using Microsoft.Win32.SafeHandles; - -namespace IronRuby.Builtins { - /// - /// The write end of an IO.pipe on Windows (WindowsSpawn.OsPipe). A blocking WriteFile on a - /// full pipe waits until the reader takes the bytes, however long that is, so IO#timeout - - /// which RubyIO applies to a stream that can time out as its WriteTimeout - could not bound - /// it: a write into a full pipe hung for ever. With a timeout this writes no more than the - /// pipe has room for at a time (IoReadiness.GetWindowsPipeRoom), waits for room in between, - /// and gives up with a TimeoutException, which RubyBufferedStream turns into IO::TimeoutError. - /// Without one it is the plain FileStream it was. - /// - /// A FileStream still, so what looks up the handle behind an IO (spawn redirections, - /// IO.select via IoReadiness.GetWindowsPipe) finds it. - /// - /// The same room makes #write_nonblock possible: it writes what fits and reports a full pipe - /// as EAGAIN, where a plain write would wait. - /// - internal sealed class WindowsPipeWriteStream : FileStream, INonBlockingWriteStream { - private int _writeTimeout = Timeout.Infinite; - private int _readTimeout = Timeout.Infinite; - - public WindowsPipeWriteStream(SafeFileHandle/*!*/ handle) - : base(handle, FileAccess.Write, 1, false) { - } - - public override bool CanTimeout { - get { return true; } - } - - public override int WriteTimeout { - get { return _writeTimeout; } - set { _writeTimeout = value; } - } - - // RubyIO sets both timeouts on a stream that can time out; this end is never read. - public override int ReadTimeout { - get { return _readTimeout; } - set { _readTimeout = value; } - } - - // FileStream routes Write(ReadOnlySpan) and WriteByte of a derived class through - // this overload, so it is the one place every synchronous write passes. - public override void Write(byte[]/*!*/ buffer, int offset, int count) { - int timeout = _writeTimeout; - if (timeout == Timeout.Infinite || count == 0) { - base.Write(buffer, offset, count); - return; - } - - long deadline = Environment.TickCount64 + timeout; - while (count > 0) { - int room = IoReadiness.GetWindowsPipeRoom(SafeFileHandle); - if (room == 0) { - if (Environment.TickCount64 >= deadline) { - throw new TimeoutException(); - } - Thread.Sleep(1); - continue; - } - int chunk = Math.Min(room, count); - base.Write(buffer, offset, chunk); - offset += chunk; - count -= chunk; - } - } - - /// The bytes written, or -1 when the pipe is full (EAGAIN). - public int WriteNonBlocking(byte[]/*!*/ buffer, int offset, int count) { - if (count == 0) { - return 0; - } - int room = IoReadiness.GetWindowsPipeRoom(SafeFileHandle); - if (room == 0) { - return -1; - } - int chunk = Math.Min(room, count); - base.Write(buffer, offset, chunk); - return chunk; - } - } -} diff --git a/Src/Libraries/Builtins/WindowsSpawn.cs b/Src/Libraries/Builtins/WindowsSpawn.cs index dfb963457..a67e48f6f 100644 --- a/Src/Libraries/Builtins/WindowsSpawn.cs +++ b/Src/Libraries/Builtins/WindowsSpawn.cs @@ -793,8 +793,11 @@ private static object Reap(RubyContext/*!*/ context, int pid, IntPtr handle) { SetHandleInformation(read, HANDLE_FLAG_INHERIT, 0); SetHandleInformation(write, HANDLE_FLAG_INHERIT, 0); - var reader = new FileStream(new SafeFileHandle(read, true), FileAccess.Read, 1, false); - var writer = new WindowsPipeWriteStream(new SafeFileHandle(write, true)); + // Not plain FileStreams: a blocked read has to look blocked to Thread#status and give + // way to a close from another thread, #read_nonblock / #write_nonblock must not wait, + // and IO#timeout must bound a write into a full pipe - see WindowsPipeStream. + var reader = new WindowsPipeStream(new SafeFileHandle(read, true), FileAccess.Read); + var writer = new WindowsPipeStream(new SafeFileHandle(write, true), FileAccess.Write); var reading = new RubyIO(context, reader, context.AllocateFileDescriptor(reader), IOMode.ReadOnly); var writing = new RubyIO(context, writer, context.AllocateFileDescriptor(writer), IOMode.WriteOnly); diff --git a/Src/Ruby/Builtins/ConsoleStream.cs b/Src/Ruby/Builtins/ConsoleStream.cs index 35d3c7a5b..4946e8429 100644 --- a/Src/Ruby/Builtins/ConsoleStream.cs +++ b/Src/Ruby/Builtins/ConsoleStream.cs @@ -112,7 +112,7 @@ public override void SetLength(long value) { public override void Write(byte[]/*!*/ buffer, int offset, int count) { bool output = _consoleType == ConsoleStreamType.Output; Stream stream = output ? _io.OutputStream : _io.ErrorStream; - (GetUnbufferedStream(stream, output) ?? stream).Write(buffer, offset, count); + (GetUnbufferedStream(stream, output, IsSetByHost(output)) ?? stream).Write(buffer, offset, count); } private static Stream _rawInput; @@ -155,13 +155,38 @@ private static Stream GetUnbufferedInput() { /// /// Forgets what was found out about descriptors 0, 1 and 2: a warm-start worker - /// (Hosting/WarmStart.cs) has just put different ones there. + /// (Hosting/WarmStart.cs) has just put different ones there - or, on Windows, different + /// standard handles, which decide between the console writer and a pipe stream. /// internal static void ResetStandardDescriptors() { _rawInput = null; _rawInputChecked = false; _rawOutput = null; _rawError = null; + _consoleOutput = null; + _consoleError = null; + _consoleOutputChecked = false; + _consoleErrorChecked = false; + } + + /// + /// The pipe standard output or standard error writes to on Windows when it is redirected + /// to one, or null. IO#syswrite on a non-blocking STDOUT writes to it directly, as a + /// write(2) on descriptor 1 would. + /// + internal WindowsPipeStream WindowsPipe { + get { + if (_consoleType == ConsoleStreamType.Input) { + return null; + } + bool output = _consoleType == ConsoleStreamType.Output; + return GetUnbufferedStream(output ? _io.OutputStream : _io.ErrorStream, output, IsSetByHost(output)) as WindowsPipeStream; + } + } + + // whether a host gave SharedIO the stream (SharedIO.SetOutput) rather than it opening the process's own + private bool IsSetByHost(bool output) { + return output ? _io.IsOutputSetByHost : _io.IsErrorOutputSetByHost; } /// @@ -171,7 +196,18 @@ internal static void ResetStandardDescriptors() { /// descriptor does report the error, so use one whenever the standard stream is still the one /// .NET opened (an embedder that redirected SharedIO keeps its own stream) and is redirected. /// - private static Stream GetUnbufferedStream(Stream/*!*/ stream, bool output) { + private static Stream GetUnbufferedStream(Stream/*!*/ stream, bool output, bool setByHost) { + if (Environment.OSVersion.Platform == PlatformID.Win32NT && stream == Stream.Null) { + // .NET checks the handle with a zero-byte WriteFile when it opens standard output, + // and hands out Stream.Null when that fails - which it does on a pipe whose reader + // has already gone. Everything written was then thrown away without an error, and + // a child that IO.popen(...).close had abandoned before it wrote its first line + // ran `loop { puts }` for ever. The handle itself still says EPIPE. + // A host can also give SharedIO Stream.Null on purpose, to throw the output away; + // that one stays a discard, whatever state the handle is in. + return setByHost ? null : GetWindowsConsole(output); + } + var name = stream.GetType().FullName; if (name == null || !name.EndsWith("ConsoleStream", StringComparison.Ordinal)) { return null; @@ -212,7 +248,8 @@ private static Stream GetUnbufferedStream(Stream/*!*/ stream, bool output) { /// "ÔóÇÔí"). MRI writes to a console with WriteConsoleW instead, as UTF-16, which shows /// the text whatever the code page; this does the same, decoding the bytes as UTF-8, /// Ruby's default external encoding on Windows. A stream that is redirected to a file - /// or a pipe is not a console and keeps getting the bytes unchanged. + /// or a pipe is not a console and keeps getting the bytes unchanged; a pipe through a + /// WindowsPipeStream, which reports a reader that has gone. /// private static Stream GetWindowsConsole(bool output) { ref bool isChecked = ref (output ? ref _consoleOutputChecked : ref _consoleErrorChecked); @@ -223,6 +260,12 @@ private static Stream GetWindowsConsole(bool output) { IntPtr handle = Win32Console.GetStdHandle(output ? Win32Console.STD_OUTPUT_HANDLE : Win32Console.STD_ERROR_HANDLE); if (Win32Console.GetConsoleMode(handle, out _)) { cache = new Win32ConsoleWriter(handle); + } else if (WindowsPipeStream.IsPipe(handle)) { + // .NET's stream ignores ERROR_NO_DATA, so a child writing to a pipe + // whose reader has gone - IO.popen(cmd).close - never learnt of it and + // `loop { puts }` ran for ever. This one reports EPIPE, as on Unix. + // The handle stays open: it is the process's, not this stream's. + cache = new WindowsPipeStream(new Microsoft.Win32.SafeHandles.SafeFileHandle(handle, false), FileAccess.Write); } } catch (Exception) { cache = null; diff --git a/Src/Ruby/Builtins/DescriptorStream.cs b/Src/Ruby/Builtins/DescriptorStream.cs index 20557cad2..cf52ea4fb 100644 --- a/Src/Ruby/Builtins/DescriptorStream.cs +++ b/Src/Ruby/Builtins/DescriptorStream.cs @@ -36,7 +36,7 @@ public interface INonBlockingWriteStream { int WriteNonBlocking(byte[]/*!*/ buffer, int offset, int count); } - public sealed class DescriptorStream : Stream, IDescriptorStream, INonBlockingWriteStream { + public sealed class DescriptorStream : Stream, IDescriptorStream, INonBlockingWriteStream, INonBlockingReadStream, ISingleWriteStream { [StructLayout(LayoutKind.Sequential)] private struct PollFd { public int fd; diff --git a/Src/Ruby/Builtins/WindowsPipeStream.cs b/Src/Ruby/Builtins/WindowsPipeStream.cs new file mode 100644 index 000000000..04f29a1a3 --- /dev/null +++ b/Src/Ruby/Builtins/WindowsPipeStream.cs @@ -0,0 +1,527 @@ +/* **************************************************************************** + * + * A Windows anonymous pipe end (CreatePipe), read and written the way Ruby expects + * of a pipe - what DescriptorStream is on Unix. + * + * A FileStream over the handle reads with a synchronous ReadFile, and three things + * follow from that. The thread sits inside a P/Invoke, which the CLR reports as + * Running, so Thread#status said "run" and Thread#stop? false for a thread blocked + * in a pipe read - and ruby/spec waits with `Thread.pass until thread.stop?` before + * it does anything to such a thread, for ever. Closing the IO from another thread, + * which Ruby defines as making the blocked read raise IOError, could not reach the + * read at all. And there was no way to read or write without waiting, so + * #read_nonblock on an empty pipe and #write_nonblock on a full one blocked. + * + * Here a blocking read or write tells the thread library it is blocked + * (RubyUtils.EnterNativeWait), a close from another thread cancels it with + * CancelSynchronousIo, the non-blocking read asks PeekNamedPipe how much there is + * to read, and the non-blocking write asks the pipe how much room it has. + * + * It is still a FileStream, so everything that asks for the handle behind an IO + * (spawn redirections, IO.select's PeekNamedPipe) keeps finding it. + * + * ***************************************************************************/ + +using System; +using System.Collections.Generic; +using System.IO; +using System.Runtime.InteropServices; +using System.Threading; +using Microsoft.Win32.SafeHandles; +using IronRuby.Runtime; + +namespace IronRuby.Builtins { + + /// One read(2) that does not wait: bytes read, 0 at end of file, -1 when it would block. + public interface INonBlockingReadStream { + int ReadNonBlocking(byte[]/*!*/ buffer, int offset, int count); + } + + /// + /// One write(2), as IO#syswrite makes it: on a descriptor in non-blocking mode it may take + /// only part of the bytes, and -1 means it took none. + /// + public interface ISingleWriteStream { + int WriteOnce(byte[]/*!*/ buffer, int offset, int count); + } + + public sealed class WindowsPipeStream : FileStream, INonBlockingReadStream, INonBlockingWriteStream, ISingleWriteStream { + [DllImport("kernel32.dll", SetLastError = true)] + private static extern unsafe bool ReadFile(SafeFileHandle handle, byte* buffer, int count, out int read, IntPtr overlapped); + + [DllImport("kernel32.dll", SetLastError = true)] + private static extern unsafe bool WriteFile(SafeFileHandle handle, byte* buffer, int count, out int written, IntPtr overlapped); + + [DllImport("kernel32.dll", SetLastError = true)] + private static extern bool PeekNamedPipe(SafeFileHandle handle, IntPtr buffer, int size, + IntPtr bytesRead, out int bytesAvailable, IntPtr bytesLeft); + + [DllImport("kernel32.dll", SetLastError = true)] + private static extern bool CancelSynchronousIo(IntPtr thread); + + [DllImport("kernel32.dll", SetLastError = true)] + private static extern IntPtr OpenThread(int desiredAccess, bool inheritHandle, int threadId); + + [DllImport("kernel32.dll")] + private static extern int GetCurrentThreadId(); + + [DllImport("kernel32.dll", SetLastError = true)] + private static extern bool CloseHandle(IntPtr handle); + + [DllImport("kernel32.dll", SetLastError = true)] + private static extern int GetFileType(IntPtr handle); + + private const int FILE_TYPE_UNKNOWN = 0; + private const int FILE_TYPE_PIPE = 3; + private const int THREAD_TERMINATE = 0x0001; // what CancelSynchronousIo needs + private const int ERROR_BROKEN_PIPE = 109; + private const int ERROR_NO_DATA = 232; // "the pipe is being closed": the reader is gone + private const int ERROR_OPERATION_ABORTED = 995; + + private volatile bool _closed; + + // The thread handles of the reads and writes in progress, for a close from another + // thread to cancel. + private readonly List/*!*/ _inProgress = new List(); + + // Held by a write for as long as it writes, and by a non-blocking write from measuring the + // room in the pipe to its WriteFile, so that no other write here takes that room first. + private readonly object/*!*/ _writeLock = new object(); + + // Held by a read for as long as it is in ReadFile, and by a non-blocking read from its + // PeekNamedPipe to its ReadFile. + private readonly object/*!*/ _readLock = new object(); + + /// + /// A stream over one end of a pipe. The handle is closed with the stream if the SafeFileHandle + /// owns it; standard output's does not. + /// + public WindowsPipeStream(SafeFileHandle/*!*/ handle, FileAccess access) + : base(handle, access, 1, false) { + } + + /// + /// Whether is a pipe - an anonymous one or a named one - or one + /// whose reader has already gone: GetFileType then fails and says FILE_TYPE_UNKNOWN, and + /// a child whose parent closed its end before the child wrote anything is exactly the + /// process that most needs to hear EPIPE. + /// + public static bool IsPipe(IntPtr handle) { + if (handle == IntPtr.Zero || handle == new IntPtr(-1)) { + return false; + } + int type = GetFileType(handle); + return type == FILE_TYPE_PIPE || (type == FILE_TYPE_UNKNOWN && Marshal.GetLastWin32Error() != 0); + } + + /// + /// The Windows pipe reads and writes through: itself, or the one + /// behind standard output or standard error when that is redirected to a pipe. Null if + /// there is none. + /// + public static WindowsPipeStream Of(Stream stream) { + var pipe = stream as WindowsPipeStream; + if (pipe != null) { + return pipe; + } + var console = stream as ConsoleStream; + return (console != null) ? console.WindowsPipe : null; + } + + private static Exception/*!*/ ClosedInAnotherThread() { + return RubyExceptions.CreateIOError("stream closed in another thread"); + } + + #region Reads + + /// See . + public override int Read(byte[]/*!*/ buffer, int offset, int count) { + return Read(new Span(buffer, offset, count)); + } + + /// + /// Waits for at least one byte and returns what there is, up to the buffer's length; 0 at end + /// of file. Raises IOError when the stream is closed from another thread meanwhile. + /// + public override int Read(Span buffer) { + if (buffer.Length == 0) { + return 0; + } + // One reader in ReadFile at a time, so that ReadNonBlocking can tell whether the bytes + // it has just seen are still there for it (see there). + lock (_readLock) { + return ReadLocked(buffer); + } + } + + private int ReadLocked(Span buffer) { + while (true) { + IntPtr thread = Enter(); + bool wasBlocked = RubyUtils.EnterNativeWait(); + int read, error; + bool ok; + try { + unsafe { + fixed (byte* p = buffer) { + ok = ReadFile(SafeFileHandle, p, buffer.Length, out read, IntPtr.Zero); + } + } + error = ok ? 0 : Marshal.GetLastWin32Error(); + } finally { + RubyUtils.ExitNativeWait(wasBlocked); + Leave(thread); + } + + if (ok) { + return read; + } + if (error == ERROR_BROKEN_PIPE) { + // The writer is gone and everything it wrote has been read: end of file. + return 0; + } + if (error == ERROR_OPERATION_ABORTED) { + if (_closed) { + throw ClosedInAnotherThread(); + } + // Cancelled by somebody else: let the thread take what was raised into it, and + // otherwise read on. + RubyUtils.CheckAsyncException(); + continue; + } + // The error ReadFile left, not the last one: Leave's CloseHandle has run since. + throw new IOException("read failed (error " + error + ")", unchecked((int)0x80070000) | error); + } + } + + /// One byte, waiting for it as Read does; -1 at end of file. + public override int ReadByte() { + Span one = stackalloc byte[1]; + return Read(one) == 0 ? -1 : one[0]; + } + + /// + /// A read that does not wait, as #read_nonblock makes it: the bytes read, 0 at end of file, + /// or -1 when there is nothing to read yet. + /// + public int ReadNonBlocking(byte[]/*!*/ buffer, int offset, int count) { + if (_closed) { + throw ClosedInAnotherThread(); + } + if (count == 0) { + return 0; + } + // Another thread reading - blocked in ReadFile, which takes whatever arrives first - + // means nothing here is for us now: would block. Holding the lock from the peek to the + // read keeps any other reader from taking the bytes the peek saw, which would leave + // this ReadFile waiting on an empty pipe. + if (!Monitor.TryEnter(_readLock)) { + return -1; + } + try { + int available; + if (!PeekNamedPipe(SafeFileHandle, IntPtr.Zero, 0, IntPtr.Zero, out available, IntPtr.Zero)) { + int error = Marshal.GetLastWin32Error(); + if (error == ERROR_BROKEN_PIPE) { + return 0; + } + throw new IOException("read failed (error " + error + ")", unchecked((int)0x80070000) | error); + } + if (available == 0) { + return -1; + } + // There are bytes in the pipe and no other reader, so this read returns at once. + return ReadLocked(new Span(buffer, offset, Math.Min(count, available))); + } finally { + Monitor.Exit(_readLock); + } + } + + #endregion + + #region Writes + + /// See . + public override void Write(byte[]/*!*/ buffer, int offset, int count) { + Write(new ReadOnlySpan(buffer, offset, count)); + } + + /// + /// Writes all of the buffer, waiting for room in the pipe as often as it takes. A reader that + /// has gone is an IOException carrying EPIPE. + /// + /// With a (IO#timeout, which RubyIO applies to a stream that can + /// time out) it writes no more than the pipe has room for at a time, waits for room in + /// between, and gives up with a TimeoutException at the deadline, which RubyBufferedStream + /// turns into IO::TimeoutError. A blocking WriteFile into a full pipe would wait until the + /// reader takes the bytes, however long that is. + /// + public override void Write(ReadOnlySpan buffer) { + // Held for the whole write, so that a non-blocking write cannot measure room this one + // is about to take (see WriteNonBlocking). + lock (_writeLock) { + int timeout = _writeTimeout; + long deadline = (timeout == Timeout.Infinite) ? 0 : Environment.TickCount64 + timeout; + while (buffer.Length > 0) { + int chunk = buffer.Length; + if (timeout != Timeout.Infinite) { + if (_closed) { + throw ClosedInAnotherThread(); + } + int room = WriteQuotaAvailable(); + if (room == 0 && !IsReaderGone()) { + if (Environment.TickCount64 >= deadline) { + throw new TimeoutException(); + } + Thread.Sleep(1); + continue; + } + if (room > 0) { + chunk = Math.Min(room, chunk); + } + } + int written = WriteFileOnce(buffer.Slice(0, chunk), true); + buffer = buffer.Slice(written); + } + } + } + + private int _writeTimeout = Timeout.Infinite; + private int _readTimeout = Timeout.Infinite; + + /// True: IO#timeout bounds a write (see ). + public override bool CanTimeout { + get { return true; } + } + + public override int WriteTimeout { + get { return _writeTimeout; } + set { _writeTimeout = value; } + } + + // RubyIO sets both on a stream that can time out. A read waits for data before it starts + // (RubyIO.WaitForStream, PeekNamedPipe), so the read itself needs no deadline. + public override int ReadTimeout { + get { return _readTimeout; } + set { _readTimeout = value; } + } + + /// One byte, written as Write does. + public override void WriteByte(byte value) { + ReadOnlySpan one = stackalloc byte[] { value }; + Write(one); + } + + /// + /// IO#syswrite on a pipe not in non-blocking mode: the whole buffer. Windows has no + /// O_NONBLOCK on the handle to decide it by; IO#syswrite asks the IO (io/nonblock's + /// #nonblock=) and calls WriteNonBlocking itself when it is set. + /// + public int WriteOnce(byte[]/*!*/ buffer, int offset, int count) { + Write(buffer, offset, count); + return count; + } + + /// + /// As many bytes as the pipe has room for, or -1 when it has none - a write(2) on an + /// O_NONBLOCK pipe. + /// + /// Not by switching the handle to PIPE_NOWAIT: that fails with ERROR_PIPE_BUSY as soon as + /// the pipe holds unread data, a PIPE_NOWAIT write larger than the free space writes + /// nothing at all, and the mode belongs to the pipe end, which a child shares once it has + /// been handed the handle. The pipe says how much room it has (WriteQuotaAvailable, as + /// libuv asks it), and a write of no more than that completes at once. + /// + public int WriteNonBlocking(byte[]/*!*/ buffer, int offset, int count) { + if (_closed) { + throw ClosedInAnotherThread(); + } + if (count == 0) { + return 0; + } + // Another writer in this process - a blocking write waiting for room, most likely - + // means the room is not ours to measure: would block, rather than wait for it. + // Holding the lock from the measuring to the WriteFile keeps any other write here from + // taking the room in between. A different process writing to the same pipe end (a + // child given it as its standard output) cannot be kept out like this; Windows has no + // other way to make one write not wait (see above). + if (!Monitor.TryEnter(_writeLock)) { + return -1; + } + try { + int room = WriteQuotaAvailable(); + if (room == 0) { + // Full - or the reader is gone, which the write itself reports as EPIPE. + if (!IsReaderGone()) { + return -1; + } + room = count; + } + return WriteFileOnce(new ReadOnlySpan(buffer, offset, Math.Min(count, room)), false); + } finally { + Monitor.Exit(_writeLock); + } + } + + [StructLayout(LayoutKind.Sequential)] + private struct IO_STATUS_BLOCK { + public IntPtr Status; + public UIntPtr Information; + } + + [StructLayout(LayoutKind.Sequential)] + private struct FILE_PIPE_LOCAL_INFORMATION { + public uint NamedPipeType; + public uint NamedPipeConfiguration; + public uint MaximumInstances; + public uint CurrentInstances; + public uint InboundQuota; + public uint ReadDataAvailable; + public uint OutboundQuota; + public uint WriteQuotaAvailable; + public uint NamedPipeState; + public uint NamedPipeEnd; + } + + private const int FilePipeLocalInformation = 24; + private const uint FILE_PIPE_CLOSING_STATE = 4; + + [DllImport("ntdll.dll")] + private static extern int NtQueryInformationFile(SafeFileHandle handle, out IO_STATUS_BLOCK status, + out FILE_PIPE_LOCAL_INFORMATION info, int length, int informationClass); + + private bool TryQueryPipe(out FILE_PIPE_LOCAL_INFORMATION info) { + IO_STATUS_BLOCK status; + return NtQueryInformationFile(SafeFileHandle, out status, out info, + Marshal.SizeOf(typeof(FILE_PIPE_LOCAL_INFORMATION)), FilePipeLocalInformation) >= 0; + } + + /// The room left in the pipe, in bytes; the whole request when the pipe cannot say. + private int WriteQuotaAvailable() { + FILE_PIPE_LOCAL_INFORMATION info; + if (!TryQueryPipe(out info)) { + return Int32.MaxValue; + } + return (int)Math.Min(info.WriteQuotaAvailable, (uint)Int32.MaxValue); + } + + private bool IsReaderGone() { + FILE_PIPE_LOCAL_INFORMATION info; + return TryQueryPipe(out info) && info.NamedPipeState == FILE_PIPE_CLOSING_STATE; + } + + private int WriteFileOnce(ReadOnlySpan buffer, bool mayBlock) { + while (true) { + // Registered even when it is not expected to wait: the room a non-blocking write + // measured can be taken by a blocking write here, or by a child process writing to + // the same pipe, before its WriteFile runs - and a close must still reach it then. + IntPtr thread = Enter(); + bool wasBlocked = mayBlock && RubyUtils.EnterNativeWait(); + int written, error; + bool ok; + try { + unsafe { + fixed (byte* p = buffer) { + ok = WriteFile(SafeFileHandle, p, buffer.Length, out written, IntPtr.Zero); + } + } + error = ok ? 0 : Marshal.GetLastWin32Error(); + } finally { + if (mayBlock) { + RubyUtils.ExitNativeWait(wasBlocked); + } + Leave(thread); + } + + if (ok) { + return written; + } + if (error == ERROR_NO_DATA || error == ERROR_BROKEN_PIPE) { + // Nobody will read this: EPIPE, which IoOps turns into Errno::EPIPE, or into + // SIGPIPE for the standard streams. .NET's own console stream swallows this + // error, so a `loop { puts }` child never noticed its reader had gone. + throw new IOException("write failed", DescriptorStream.EPIPE); + } + if (error == ERROR_OPERATION_ABORTED) { + if (_closed) { + throw ClosedInAnotherThread(); + } + RubyUtils.CheckAsyncException(); + continue; + } + throw new IOException("write failed (error " + error + ")", unchecked((int)0x80070000) | error); + } + } + + /// Nothing to do: this stream holds no buffer of its own. + public override void Flush() { + // Nothing is buffered here; WriteFile has handed everything to the pipe already. + } + + #endregion + + #region Cancellation + + /// + /// Registers the calling thread as about to wait on the pipe, so that a close from another + /// thread can cancel the wait. Throws if the stream is already closed. + /// + private IntPtr Enter() { + IntPtr thread = OpenThread(THREAD_TERMINATE, false, GetCurrentThreadId()); + lock (_inProgress) { + if (_closed) { + if (thread != IntPtr.Zero) { + CloseHandle(thread); + } + throw ClosedInAnotherThread(); + } + if (thread != IntPtr.Zero) { + _inProgress.Add(thread); + } + } + return thread; + } + + private void Leave(IntPtr thread) { + if (thread == IntPtr.Zero) { + return; + } + lock (_inProgress) { + _inProgress.Remove(thread); + } + CloseHandle(thread); + } + + /// + /// Cancels every read and write in progress on this stream. A thread can be between + /// registering and entering ReadFile when the cancel comes, which then finds nothing to + /// cancel, so it is repeated until each of them has left. + /// + private void CancelInProgress() { + for (int attempt = 0; attempt < 2000; attempt++) { + lock (_inProgress) { + if (_inProgress.Count == 0) { + return; + } + foreach (IntPtr thread in _inProgress) { + CancelSynchronousIo(thread); + } + } + Thread.Sleep(1); + } + } + + /// + /// Closes the stream. A read or write another thread has in progress is cancelled first and + /// raises IOError "stream closed in another thread", as Ruby's IO#close promises. + /// + protected override void Dispose(bool disposing) { + lock (_inProgress) { + _closed = true; + } + CancelInProgress(); + base.Dispose(disposing); + } + + #endregion + } +}