fix(vm): thread blocking and the GVL hand-off — five defects, one subject (#695)

internal/vm/queue.go

@@ -239,23 +239,39 @@ func num2dblName(v object.Value) string {
 	return classNameOf(v)
 }
 
-// queueBlock releases the GVL and waits on ch, optionally bounded by a timer. It
-// reports whether the wait ended on the timeout rather than a wake. fn runs in
-// the calling goroutine, so timedOut is written and read without a data race.
+// queueBlock releases the GVL and waits on ch, optionally bounded by a timer, and
+// reports whether the wait ended on the timeout rather than a wake.
 func (vm *VM) queueBlock(ch chan struct{}, timeout float64, hasTimeout bool) (timedOut bool) {
+	// The wait must be interruptible: MRI's queue_do_pop sleeps through
+	// rb_thread_sleep_deadly_allow_spurious_wakeup (thread_sync.c queue_sleep),
+	// which is the general interrupt-aware thread sleep, and the surrounding
+	// rb_ensure (queue_sleep_done) unlinks the waiter on the way out. A plain
+	// channel receive has neither property, so a Thread#kill on a thread blocked
+	// in Queue#pop never landed and a join on it never returned.
 	if !hasTimeout {
-		vm.threadBlock(func() { <-ch })
+		vm.interruptibleWait(ch)
 		return false
 	}
-	vm.threadBlock(func() {
-		t := time.NewTimer(time.Duration(timeout * float64(time.Second)))
-		defer t.Stop()
-		select {
-		case <-ch:
-		case <-t.C:
-			timedOut = true
-		}
-	})
+	_, timedOut = vm.interruptibleWaitFor(ch, time.Duration(timeout*float64(time.Second)))
+	return timedOut
+}
+
+// queueWait registers a waiter on *waiters, blocks interruptibly on it, and
+// unlinks it again — the unlink through a defer, because an interrupt leaves the
+// wait by panicking and a waiter left behind would swallow the wake meant for a
+// live thread. MRI spells that defer as the rb_ensure around queue_sleep whose
+// cleanup, queue_sleep_done, is exactly this ccan_list_del.
+func (vm *VM) queueWait(waiters *[]chan struct{}, timeout float64, hasTimeout bool) (timedOut bool) {
+	ch := make(chan struct{})
+	*waiters = append(*waiters, ch)
+	defer func() { *waiters = removeChan(*waiters, ch) }()
+	timedOut = vm.queueBlock(ch, timeout, hasTimeout)
+	// sleep_forever runs RUBY_VM_CHECK_INTS_BLOCKING around its sleep (thread.c),
+	// so an interrupt that ended the wait fires HERE. Without this the caller's
+	// `while the queue is empty` loop simply went back to sleep and the interrupt
+	// was never delivered — the wait had become interruptible without becoming
+	// interrupted. The deferred unlink above still runs as this raise unwinds.
+	vm.serviceSafepointAt(vm.currentThread, true)
 	return timedOut
 }
 
@@ -271,10 +287,7 @@ func (vm *VM) queuePush(q *RQueue, args []object.Value) object.Value {
 		if nonBlock {
 			raise("ThreadError", "queue full")
 		}
-		ch := make(chan struct{})
-		q.pushq = append(q.pushq, ch)
-		timedOut := vm.queueBlock(ch, timeout, hasTimeout)
-		q.pushq = removeChan(q.pushq, ch)
+		timedOut := vm.queueWait(&q.pushq, timeout, hasTimeout)
 		if q.closed {
 			raise("ClosedQueueError", "queue closed")
 		}
@@ -298,11 +311,7 @@ func (vm *VM) queuePop(q *RQueue, nonBlock bool, timeout float64, hasTimeout boo
 		if nonBlock {
 			raise("ThreadError", "queue empty")
 		}
-		ch := make(chan struct{})
-		q.waitq = append(q.waitq, ch)
-		timedOut := vm.queueBlock(ch, timeout, hasTimeout)
-		q.waitq = removeChan(q.waitq, ch)
-		if timedOut {
+		if vm.queueWait(&q.waitq, timeout, hasTimeout) {
 			return object.NilV
 		}
 	}

internal/vm/socket.go

@@ -57,6 +57,18 @@ func (s *tcpServer) ToS() string     { return "#<TCPServer>" }
 func (s *tcpServer) Inspect() string { return "#<TCPServer>" }
 func (s *tcpServer) Truthy() bool    { return true }
 
+// ioBlock runs a blocking socket call with the GVL released, so another Ruby
+// Thread can run while this one waits on the network. MRI releases it around
+// every blocking socket call the same way (rb_io_blocking_region /
+// rb_thread_io_blocking_call, io.c, over rb_thread_call_without_gvl in thread.c),
+// and it is not a throughput optimisation: an accept or a read that keeps the
+// lock DEADLOCKS a program whose peer is another Thread of the same VM, because
+// the peer needs the lock to send the very bytes this call is waiting for.
+//
+// Every caller is a native Ruby method, which always holds the GVL, so there is
+// 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) }
+
 // 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
@@ -106,7 +118,9 @@ func (vm *VM) registerTCPSocket(tcp *RClass) {
 		}
 		host := strArg(args[0])
 		port := portString(args[1])
-		conn, err := net.Dial("tcp", net.JoinHostPort(host, port))
+		var conn net.Conn
+		var err error
+		ioBlock(vm, func() { conn, err = net.Dial("tcp", net.JoinHostPort(host, port)) })
 		if err != nil {
 			raise("SocketError", "getaddrinfo: %s", err.Error())
 		}
@@ -153,9 +167,11 @@ func (vm *VM) registerTCPServer(srv *RClass) {
 	srv.smethods["new"] = &Method{name: "new", owner: srv, native: newFn}
 	srv.smethods["open"] = &Method{name: "open", owner: srv, native: newFn}
 
-	srv.define("accept", func(_ *VM, self object.Value, _ []object.Value, _ *Proc) object.Value {
+	srv.define("accept", func(vm *VM, self object.Value, _ []object.Value, _ *Proc) object.Value {
 		s := asTCPServer(self)
-		conn, err := s.ln.Accept()
+		var conn net.Conn
+		var err error
+		ioBlock(vm, func() { conn, err = s.ln.Accept() })
 		if err != nil {
 			raise("IOError", "accept: %s", err.Error())
 		}
@@ -237,14 +253,14 @@ func (s *tcpSocket) closeConn() error      { return s.conn.Close() }
 // write/print/puts/<</flush/close/closed?/eof?/setsockopt) on cls, resolving the
 // receiver to a streamIO via get.
 func installStreamIO(cls *RClass, get func(object.Value) streamIO) {
-	cls.define("read", func(_ *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
-		return streamRead(get(self), args)
+	cls.define("read", func(vm *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
+		return streamRead(vm, get(self), args)
 	})
-	cls.define("gets", func(_ *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
-		return streamGets(get(self), args)
+	cls.define("gets", func(vm *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
+		return streamGets(vm, get(self), args)
 	})
-	cls.define("readpartial", func(_ *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
-		return streamReadpartial(get(self), args)
+	cls.define("readpartial", func(vm *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
+		return streamReadpartial(vm, get(self), args)
 	})
 	cls.define("write", func(vm *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
 		return streamWrite(vm, get(self), args)
@@ -304,8 +320,8 @@ func installStreamIO(cls *RClass, get func(object.Value) streamIO) {
 	// whatever is available (an empty ASCII-8BIT String at end of stream), the
 	// BasicSocket#recv datagram-less semantics for a connected stream. flags is
 	// accepted and ignored.
-	cls.define("recv", func(_ *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
-		return streamRecv(get(self), args)
+	cls.define("recv", func(vm *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
+		return streamRecv(vm, get(self), args)
 	})
 	// send(mesg[, flags[, dest]]) writes mesg to the stream and returns the byte
 	// count; flags and dest are accepted and ignored for a connected stream.
@@ -322,7 +338,7 @@ func installStreamIO(cls *RClass, get func(object.Value) streamIO) {
 // streamRecv implements #recv(maxlen): read up to maxlen bytes, blocking only
 // when the buffer is empty, and returning an empty ASCII-8BIT String at end of
 // stream (BasicSocket#recv semantics) rather than raising.
-func streamRecv(s streamIO, args []object.Value) object.Value {
+func streamRecv(vm *VM, s streamIO, args []object.Value) object.Value {
 	if len(args) == 0 {
 		raise("ArgumentError", "wrong number of arguments (given 0, expected 1..2)")
 	}
@@ -334,16 +350,18 @@ func streamRecv(s streamIO, args []object.Value) object.Value {
 		return object.NewStringBytesEnc(nil, "ASCII-8BIT")
 	}
 	buf := make([]byte, n)
-	k, _ := s.reader().Read(buf)
+	var k int
+	ioBlock(vm, func() { k, _ = s.reader().Read(buf) })
 	return object.NewStringBytesEnc(buf[:k], "ASCII-8BIT")
 }
 
 // streamRead implements #read: no argument reads to EOF (returning a possibly
 // empty ASCII-8BIT String); an integer length reads up to that many bytes,
 // returning nil at EOF with nothing read, exactly like MRI's IO#read.
-func streamRead(s streamIO, args []object.Value) object.Value {
+func streamRead(vm *VM, s streamIO, args []object.Value) object.Value {
 	if len(args) == 0 || object.IsNil(args[0]) {
-		data, _ := io.ReadAll(s.reader())
+		var data []byte
+		ioBlock(vm, func() { data, _ = io.ReadAll(s.reader()) })
 		return object.NewStringBytesEnc(data, "ASCII-8BIT")
 	}
 	n := int(intArg(args[0]))
@@ -354,7 +372,9 @@ func streamRead(s streamIO, args []object.Value) object.Value {
 		return object.NewStringBytesEnc(nil, "ASCII-8BIT")
 	}
 	buf := make([]byte, n)
-	k, err := io.ReadFull(s.reader(), buf)
+	var k int
+	var err error
+	ioBlock(vm, func() { k, err = io.ReadFull(s.reader(), buf) })
 	if k == 0 && err == io.EOF {
 		return object.NilV
 	}
@@ -363,12 +383,14 @@ func streamRead(s streamIO, args []object.Value) object.Value {
 
 // streamGets implements #gets: read up to and including the separator (default
 // "\n"), returning nil at EOF with nothing buffered.
-func streamGets(s streamIO, args []object.Value) object.Value {
+func streamGets(vm *VM, s streamIO, args []object.Value) object.Value {
 	sep := "\n"
 	if len(args) > 0 && !object.IsNil(args[0]) {
 		sep = strArg(args[0])
 	}
-	data, err := readUntil(s.reader(), sep)
+	var data []byte
+	var err error
+	ioBlock(vm, func() { data, err = readUntil(s.reader(), sep) })
 	if len(data) == 0 && err == io.EOF {
 		return object.NilV
 	}
@@ -378,7 +400,7 @@ func streamGets(s streamIO, args []object.Value) object.Value {
 // streamReadpartial implements #readpartial(n): return whatever is immediately
 // available (at least one byte), blocking only when the buffer is empty, and
 // raising EOFError at end of stream, matching MRI.
-func streamReadpartial(s streamIO, args []object.Value) object.Value {
+func streamReadpartial(vm *VM, s streamIO, args []object.Value) object.Value {
 	if len(args) == 0 {
 		raise("ArgumentError", "wrong number of arguments (given 0, expected 1..2)")
 	}
@@ -390,7 +412,9 @@ func streamReadpartial(s streamIO, args []object.Value) object.Value {
 		return object.NewStringBytesEnc(nil, "ASCII-8BIT")
 	}
 	buf := make([]byte, n)
-	k, err := s.reader().Read(buf)
+	var k int
+	var err error
+	ioBlock(vm, func() { k, err = s.reader().Read(buf) })
 	if k == 0 && err == io.EOF {
 		raise("EOFError", "end of file reached")
 	}

internal/vm/socket_raw.go

@@ -162,7 +162,9 @@ func (vm *VM) registerRawSocket(sock *RClass) {
 		if s.ln == nil {
 			raise("IOError", "accept: socket is not listening (call #bind first)")
 		}
-		conn, err := s.ln.Accept()
+		var conn net.Conn
+		var err error
+		ioBlock(vm, func() { conn, err = s.ln.Accept() })
 		if err != nil {
 			raise("IOError", "accept: %s", err.Error())
 		}
@@ -173,13 +175,15 @@ func (vm *VM) registerRawSocket(sock *RClass) {
 
 	// #connect(sockaddr) connects to the peer packed in sockaddr (net.Dial, for
 	// both stream and datagram sockets).
-	sock.define("connect", func(_ *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
+	sock.define("connect", func(vm *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
 		s := asRawSocket(self)
 		if len(args) < 1 {
 			raise("ArgumentError", "wrong number of arguments (given %d, expected 1)", len(args))
 		}
 		network, _ := rawSocketNetwork(s.domain, s.typ)
-		conn, err := net.Dial(network, s.resolveAddr(sockaddrBytes(args[0])))
+		var conn net.Conn
+		var err error
+		ioBlock(vm, func() { conn, err = net.Dial(network, s.resolveAddr(sockaddrBytes(args[0]))) })
 		if err != nil {
 			raise("SocketError", "connect: %s", err.Error())
 		}
@@ -222,15 +226,18 @@ func (vm *VM) registerRawSocket(sock *RClass) {
 	// #recv(maxlen [, flags]) reads up to maxlen bytes: from the connected peer
 	// for a connected socket, or one datagram for a bound datagram socket. flags
 	// is accepted and ignored.
-	sock.define("recv", func(_ *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
+	sock.define("recv", func(vm *VM, self object.Value, args []object.Value, _ *Proc) object.Value {
 		s := asRawSocket(self)
 		buf := make([]byte, rawRecvLen(args))
 		switch {
 		case s.conn != nil:
-			k, _ := s.r.Read(buf)
+			var k int
+			ioBlock(vm, func() { k, _ = s.r.Read(buf) })
 			return object.NewStringBytesEnc(buf[:k], "ASCII-8BIT")
 		case s.pconn != nil:
-			k, _, err := s.pconn.ReadFrom(buf)
+			var k int
+			var err error
+			ioBlock(vm, func() { k, _, err = s.pconn.ReadFrom(buf) })
 			if err != nil {
 				raise("SocketError", "recv: %s", err.Error())
 			}
@@ -250,13 +257,17 @@ func (vm *VM) registerRawSocket(sock *RClass) {
 		aiCls := vm.consts["Addrinfo"].(*RClass)
 		switch {
 		case s.pconn != nil:
-			k, addr, err := s.pconn.ReadFrom(buf)
+			var k int
+			var addr net.Addr
+			var err error
+			ioBlock(vm, func() { k, addr, err = s.pconn.ReadFrom(buf) })
 			if err != nil {
 				raise("SocketError", "recvfrom: %s", err.Error())
 			}
 			return object.NewArray(object.NewStringBytesEnc(buf[:k], "ASCII-8BIT"), s.addrinfoOf(aiCls, addr))
 		case s.conn != nil:
-			k, _ := s.r.Read(buf)
+			var k int
+			ioBlock(vm, func() { k, _ = s.r.Read(buf) })
 			return object.NewArray(object.NewStringBytesEnc(buf[:k], "ASCII-8BIT"), s.addrinfoOf(aiCls, s.conn.RemoteAddr()))
 		default:
 			raise("SocketError", "recvfrom: socket is neither connected nor bound")