diff --git a/internal/vm/gvl_blocking_io_test.go b/internal/vm/gvl_blocking_io_test.go index 0c9a2d78..77fa51d1 100644 --- a/internal/vm/gvl_blocking_io_test.go +++ b/internal/vm/gvl_blocking_io_test.go @@ -79,6 +79,12 @@ func TestEveryBlockingSocketCallReleasesTheGVL(t *testing.T) { // named here, which keeps this list the single place the exemption lives. releasing := map[string]bool{ "ioBlock": true, "threadBlock": true, "threadRelease": true, "ftpIO": true, + // ioBlockCancel is ioBlock plus an unblock function (#933): it releases + // the GVL exactly the same way and additionally makes the call return on + // Thread#kill. It is listed here rather than inferred, which is the whole + // point of keeping this map the single place an exemption lives — adding + // the wrapper without adding it here failed this gate, as it should. + "ioBlockCancel": true, } files, err := filepath.Glob("*.go") diff --git a/internal/vm/socket.go b/internal/vm/socket.go index 3fb0fc0e..0de8ef39 100644 --- a/internal/vm/socket.go +++ b/internal/vm/socket.go @@ -9,6 +9,7 @@ import ( "io" "net" "strconv" + "time" "github.com/go-embedded-ruby/ruby/internal/object" ) @@ -69,6 +70,72 @@ func (s *tcpServer) Truthy() bool { return true } // no nil-vm case to guard: a guard for one would be a branch no test could reach. func ioBlock(vm *VM, fn func()) { vm.threadBlock(fn) } +// socketDeadliner is the part of a net.Conn / net.Listener that can make a +// blocked call return. net.Listener does NOT declare SetDeadline — the +// interface is Accept/Close/Addr — but *net.TCPListener and *net.UnixListener +// both have it, so the capability is asked for rather than assumed. +type socketDeadliner interface{ SetDeadline(time.Time) error } + +// unblockNow makes a call blocked on v return, by putting its deadline in the +// past. It is the Go stand-in for MRI's unblock function: it does not close +// anything, so the socket is still usable afterwards. +func unblockNow(v any) { + if d, ok := v.(socketDeadliner); ok { + d.SetDeadline(time.Now()) + } +} + +// clearDeadline undoes unblockNow, so a listener that survives an interrupted +// accept still accepts. +func clearDeadline(v any) { + if d, ok := v.(socketDeadliner); ok { + d.SetDeadline(time.Time{}) + } +} + +// ioBlockCancel is ioBlock for a call that must be INTERRUPTIBLE, and reports +// whether it was interrupted. +// +// ⛔ WHY IT EXISTS. threadRelease runs fn to completion; a Thread#kill arriving +// while fn is parked in accept(2) is only noticed after fn returns, which for an +// accept nobody will ever connect to is never. Measured: MRI kills such a thread +// within 200 ms, rbgo left it alive after 3 s (#933), and that is the whole of +// the library/socket column of #916. +// +// MRI's answer is the unblock function of rb_thread_call_without_gvl, whose job +// is to make the call RETURN. rbgo already has the other half: `interruptibleWait` +// selects on the thread's interrupt channel for waits rbgo owns. A wait the +// KERNEL owns cannot be put in a select, so a watcher waits on the same channel +// and calls cancel instead. +// +// The drain happens while the GVL is still held, as drainInterrupt requires, so +// this wait cannot be woken by an edge meant for an earlier one. +// ⛔ The answer travels on a CHANNEL, not on a shared bool. A `interrupted = true` +// written by the watcher and read here is a data race that -race would fail, and +// closing `fired` before calling cancel also orders the two: by the time this +// returns true, cancel has been entered. +func ioBlockCancel(vm *VM, cancel func(), fn func()) bool { + intr := vm.currentThread.drainInterrupt() + done := make(chan struct{}) + fired := make(chan struct{}) + go func() { + select { + case <-intr: + close(fired) + cancel() + case <-done: + } + }() + vm.threadBlock(fn) + close(done) + select { + case <-fired: + return true + default: + return false + } +} + // registerSocket installs the socket transport (require "socket"): the // TCPSocket / TCPServer classes over Go net, the BasicSocket / IPSocket / Socket // ancestry, the Socket address-family / type constants, and SocketError. It also @@ -243,7 +310,18 @@ func (vm *VM) registerTCPServer(srv *RClass) { s := asTCPServer(self) var conn net.Conn var err error - ioBlock(vm, func() { conn, err = s.ln.Accept() }) + interrupted := ioBlockCancel(vm, func() { unblockNow(s.ln) }, + func() { conn, err = s.ln.Accept() }) + // Undo the deadline the cancel set, so a listener that outlives the + // interrupted thread still accepts. + clearDeadline(s.ln) + // ⛔ AN INTERRUPTED WAIT IS NOT A FAILED ONE. The error here is the + // deadline we ourselves put in the past; raising it would give a killed + // thread an IOError MRI never raises. Returning lets the thread reach its + // next safepoint and unwind on the kill, which is what it is there for. + if interrupted { + return object.NilV + } if err != nil { raise("IOError", "accept: %s", err.Error()) } diff --git a/internal/vm/socket_test.go b/internal/vm/socket_test.go index 4f6b97f5..4e50cc82 100644 --- a/internal/vm/socket_test.go +++ b/internal/vm/socket_test.go @@ -482,6 +482,54 @@ srv.close } } +// ⛔ THE UNBLOCK FUNCTION, AND THE TEST THE COVERAGE RATCHET ASKED FOR. +// threadRelease runs its fn to completion, so a Thread#kill arriving while a +// thread is parked in accept(2) used to be noticed only after fn returned — +// which, for an accept nobody will connect to, is never. Measured: MRI kills +// such a thread within 200 ms, rbgo left it alive after 3 s (#933), and that is +// the library/socket column of #916. +// +// I had proved the fix with a script run by hand. The ratchet then reported +// unblockNow at 0.0%, which says it plainly: nothing in the suite ever +// interrupts a blocking call, so nothing would notice if the cancel stopped +// working. This is that test. +// +// It waits on `status == "sleep"` rather than on a sleep of its own, which is +// what ruby/spec does here, so a slow machine makes it slower and not flaky. +func TestThreadKillInterruptsABlockedAccept(t *testing.T) { + got := runSrc(t, `require "socket" +srv = TCPServer.new("127.0.0.1", 0) +port = srv.addr[1] + +t = Thread.new { srv.accept } +Thread.pass while t.status and t.status != "sleep" +t.kill + +waited = 0.0 +while t.alive? and waited < 5.0 + sleep 0.05 + waited += 0.05 +end +p t.alive? + +# ⛔ And the listener SURVIVES it. The cancel works by putting the deadline in +# the past, so without clearing it afterwards every later accept would fail at +# once — a fix that trades a hang for a broken server. +t2 = Thread.new { srv.accept } +Thread.pass while t2.status and t2.status != "sleep" +client = TCPSocket.new("127.0.0.1", port) +peer = t2.value +p peer.class +client.close +peer.close +srv.close +`) + want := "false\nTCPSocket" + if got != want { + t.Errorf("kill through a blocked accept:\ngot %q\nwant %q", got, want) + } +} + func TestSocketFeatureRequire(t *testing.T) { if got := runSrc(t, `p require "socket"; p require "socket"`); got != "true\nfalse" { t.Fatalf("require socket = %q", got)