diff --git a/internal/util-logging/src/main/scala/sbt/internal/util/Terminal.scala b/internal/util-logging/src/main/scala/sbt/internal/util/Terminal.scala index 0fe75095c..bcbb63698 100644 --- a/internal/util-logging/src/main/scala/sbt/internal/util/Terminal.scala +++ b/internal/util-logging/src/main/scala/sbt/internal/util/Terminal.scala @@ -640,6 +640,7 @@ object Terminal { } override def close(): Unit = if (closed.compareAndSet(false, true)) { executor.shutdownNow() + buffer.synchronized(buffer.put(-1: Integer)) () } } diff --git a/internal/util-logging/src/test/scala/sbt/internal/util/WriteableInputStreamSpec.scala b/internal/util-logging/src/test/scala/sbt/internal/util/WriteableInputStreamSpec.scala new file mode 100644 index 000000000..79cbb59a8 --- /dev/null +++ b/internal/util-logging/src/test/scala/sbt/internal/util/WriteableInputStreamSpec.scala @@ -0,0 +1,28 @@ +/* + * sbt + * Copyright 2023, Scala center + * Copyright 2011 - 2022, Lightbend, Inc. + * Copyright 2008 - 2010, Mark Harrah + * Licensed under Apache License 2.0 (see LICENSE) + */ + +package sbt.internal.util + +import java.io.PipedInputStream +import java.util.concurrent.{ LinkedBlockingQueue, TimeUnit } +import verify.BasicTestSuite + +object WriteableInputStreamSpec extends BasicTestSuite: + test("close wakes a parked reader with EOF"): + // an underlying stream that never delivers a byte, so the reader stays parked + val stream = Terminal.WriteableInputStream(new PipedInputStream(), "test") + val result = new LinkedBlockingQueue[Integer] + val reader = new Thread(() => result.put(stream.read())) + reader.setDaemon(true) + reader.start() + // let the reader park inside the buffer take + Thread.sleep(500) + stream.close() + assert(result.poll(5, TimeUnit.SECONDS) == -1) + assert(stream.read() == -1) +end WriteableInputStreamSpec diff --git a/main/src/main/scala/sbt/Main.scala b/main/src/main/scala/sbt/Main.scala index 741fd040a..8ded3c871 100644 --- a/main/src/main/scala/sbt/Main.scala +++ b/main/src/main/scala/sbt/Main.scala @@ -939,7 +939,7 @@ object BuiltinCommands { s.log.warn("Project loading failed: (r)etry, (q)uit, (l)ast, or (i)gnore? (default: r)") val result: Int = try - ITerminal.get.withRawInput(System.in.read) match { + ITerminal.get.withRawInput(ITerminal.get.inputStream.read) match { case -1 => 'q'.toInt case b => b } diff --git a/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala b/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala index d8eb17b4b..9cc12a910 100644 --- a/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala +++ b/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala @@ -30,6 +30,7 @@ import sbt.protocol.Serialization.{ terminalSetRawMode, } import sjsonnew.support.scalajson.unsafe.Converter +import sbt.internal.util.Util import sbt.protocol.{ Attach, TerminalAttributesQuery, @@ -99,18 +100,24 @@ object VirtualTerminal { } private[sbt] def cancelRequests(name: String): Unit = { import scala.jdk.CollectionConverters.* - pendingTerminalCapabilities.asScala.foreach { - case (k @ (`name`, _), q) => - pendingTerminalCapabilities.remove(k) - q.put(TerminalCapabilitiesResponse(None, None, None)) - case _ => - } - pendingTerminalProperties.asScala.foreach { - case (k @ (`name`, _), q) => - pendingTerminalProperties.remove(k) - q.put(TerminalPropertiesResponse(0, 0, false, false, false, false)) - case _ => - } + def drain[A]( + map: ConcurrentHashMap[(String, String), ArrayBlockingQueue[A]], + default: A + ): Unit = + map.asScala.foreach { + case (k @ (`name`, _), q) => + map.remove(k) + Util.ignoreResult(q.offer(default)) + case _ => + } + drain(pendingTerminalCapabilities, TerminalCapabilitiesResponse(None, None, None)) + drain(pendingTerminalProperties, TerminalPropertiesResponse(0, 0, false, false, false, false)) + drain(pendingTerminalAttributes, TerminalAttributesResponse("", "", "", "", "")) + drain(pendingTerminalSetAttributes, ()) + drain(pendingTerminalSetSize, ()) + drain(pendingTerminalGetSize, TerminalGetSizeResponse(1, 1)) + drain(pendingTerminalSetEcho, ()) + drain(pendingTerminalSetRawMode, ()) } private[sbt] def sendTerminalAttributesQuery( channelName: String, @@ -177,7 +184,7 @@ object VirtualTerminal { ): ArrayBlockingQueue[Unit] = { val id = UUID.randomUUID.toString val queue = new ArrayBlockingQueue[Unit](1) - pendingTerminalSetEcho.put((channelName, id), queue) + pendingTerminalSetRawMode.put((channelName, id), queue) jsonRpcRequest(id, terminalSetRawMode, query) queue } diff --git a/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala b/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala index ddd8db129..440f3cc89 100644 --- a/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala +++ b/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala @@ -8,7 +8,14 @@ package sbt.internal.server -import sbt.protocol.TerminalPropertiesResponse +import sbt.protocol.{ + TerminalCapabilitiesQuery, + TerminalPropertiesResponse, + TerminalSetAttributesCommand, + TerminalSetEchoCommand, + TerminalSetRawModeCommand, + TerminalSetSizeCommand, +} import verify.BasicTestSuite object VirtualTerminalSpec extends BasicTestSuite: @@ -24,4 +31,34 @@ object VirtualTerminalSpec extends BasicTestSuite: val r = TerminalPropertiesResponse(80, 24, true, true, true, true) queue.put(r) assert(VirtualTerminal.expireTerminalPropertiesQuery("expire-race", queue) == Some(r)) + + test("cancelRequests wakes waiters on every pending terminal map"): + val name = "drain-test" + val props = VirtualTerminal.sendTerminalPropertiesQuery(name, (_, _, _) => ()) + val caps = VirtualTerminal.sendTerminalCapabilitiesQuery( + name, + (_, _, _) => (), + TerminalCapabilitiesQuery(None, None, None) + ) + val attrs = VirtualTerminal.sendTerminalAttributesQuery(name, (_, _, _) => ()) + val setAttrs = VirtualTerminal.setTerminalAttributesCommand( + name, + (_, _, _) => (), + TerminalSetAttributesCommand("", "", "", "", "") + ) + val setSize = + VirtualTerminal.setTerminalSize(name, (_, _, _) => (), TerminalSetSizeCommand(80, 24)) + val getSize = VirtualTerminal.getTerminalSize(name, (_, _, _) => ()) + val echo = VirtualTerminal.setTerminalEcho(name, (_, _, _) => (), TerminalSetEchoCommand(true)) + val raw = + VirtualTerminal.setTerminalRawMode(name, (_, _, _) => (), TerminalSetRawModeCommand(true)) + VirtualTerminal.cancelRequests(name) + List(props, caps, attrs, setAttrs, setSize, getSize, echo, raw).foreach { q => + assert(q.poll() != null) + } + + test("cancelRequests leaves other channels' waiters parked"): + val queue = VirtualTerminal.sendTerminalPropertiesQuery("other-channel", (_, _, _) => ()) + VirtualTerminal.cancelRequests("drain-test-2") + assert(queue.poll() == null) end VirtualTerminalSpec diff --git a/notes/2.0.0/terminal-client-disconnect.md b/notes/2.0.0/terminal-client-disconnect.md new file mode 100644 index 000000000..d83cb99a2 --- /dev/null +++ b/notes/2.0.0/terminal-client-disconnect.md @@ -0,0 +1,10 @@ +### Fixes + +- Don't strand server threads when a client disconnects with a terminal control query unanswered. `VirtualTerminal.cancelRequests` now wakes waiters on all pending terminal maps (set-echo, raw-mode, attributes, size), not just properties and capabilities. Fixes the server-wide freeze where one dying client wedges prompts and command dispatch for every client. ([#6841][6841], [#6840][6840]) +- Register raw-mode requests in the raw-mode map (they were registered in the set-echo map). +- Wake terminal input readers with EOF when a channel's terminal closes, so a prompt blocked on a dead client's input unwinds instead of parking forever. +- Read the failed-load prompt byte from the active terminal's input stream instead of `System.in`, which under non-virtual IO is the process's own stdin and never carries client input. +- Deliver EOF reliably on client close: `ServerSessionImpl.close()` now shuts down socket input before closing, so a client disconnect is noticed by the server even while the client's read thread is parked. + +[6841]: https://github.com/sbt/sbt/issues/6841 +[6840]: https://github.com/sbt/sbt/issues/6840 diff --git a/protocol/src/main/scala/sbt/protocol/ServerSessionImpl.scala b/protocol/src/main/scala/sbt/protocol/ServerSessionImpl.scala index 10d2c96c0..bd20bd0e1 100644 --- a/protocol/src/main/scala/sbt/protocol/ServerSessionImpl.scala +++ b/protocol/src/main/scala/sbt/protocol/ServerSessionImpl.scala @@ -135,10 +135,12 @@ private[sbt] class ServerSessionImpl( */ override def close(): Unit = if (closed.compareAndSet(false, true)) { running.set(false) - try + try { + // close() cannot deliver EOF while the read thread is parked in a native read + socket.shutdownInput() out.close() socket.close() - catch case _: IOException => () + } catch case _: IOException => () onClose() if Thread.currentThread() != readThread then try readThread.joinFor(ServerSessionImpl.ReadThreadDestroyTimeout) diff --git a/protocol/src/test/scala/sbt/protocol/ServerSessionImplSpec.scala b/protocol/src/test/scala/sbt/protocol/ServerSessionImplSpec.scala new file mode 100644 index 000000000..9f0acbc47 --- /dev/null +++ b/protocol/src/test/scala/sbt/protocol/ServerSessionImplSpec.scala @@ -0,0 +1,36 @@ +/* + * sbt + * Copyright 2023, Scala center + * Copyright 2011 - 2022, Lightbend, Inc. + * Copyright 2008 - 2010, Mark Harrah + * Licensed under Apache License 2.0 (see LICENSE) + */ + +package sbt.protocol + +import java.nio.file.Files +import java.util.concurrent.{ LinkedBlockingQueue, TimeUnit } +import org.scalasbt.ipcsocket.{ UnixDomainServerSocket, UnixDomainSocket } +import verify.BasicTestSuite + +object ServerSessionImplSpec extends BasicTestSuite: + private val isWin = System.getProperty("os.name").toLowerCase.contains("win") + test("close delivers EOF to the peer while the read thread is parked"): + // named pipes have different close semantics; the swallowed-close mechanism is unix-specific + if isWin then () + else + val path = Files.createTempDirectory("session-eof").resolve("sock") + val server = UnixDomainServerSocket(path.toString, false) + val peerResult = new LinkedBlockingQueue[Integer] + val accepted = new Thread(() => { + val conn = server.accept() + peerResult.put(conn.getInputStream.read()) + }) + accepted.setDaemon(true) + accepted.start() + val session = new ServerSessionImpl(UnixDomainSocket(path.toString, false)) + // let the session's read thread park in its native read + Thread.sleep(500) + session.close() + assert(peerResult.poll(10, TimeUnit.SECONDS) == -1) +end ServerSessionImplSpec diff --git a/server-test/src/test/scala/sbt/protocol/SilentTerminalSession.scala b/server-test/src/test/scala/sbt/protocol/SilentTerminalSession.scala new file mode 100644 index 000000000..564abf304 --- /dev/null +++ b/server-test/src/test/scala/sbt/protocol/SilentTerminalSession.scala @@ -0,0 +1,54 @@ +/* + * sbt + * Copyright 2023, Scala center + * Copyright 2011 - 2022, Lightbend, Inc. + * Copyright 2008 - 2010, Mark Harrah + * Licensed under Apache License 2.0 (see LICENSE) + */ + +package sbt.protocol + +import java.io.File +import java.net.Socket +import java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.AtomicReference +import sbt.internal.protocol.JsonRpcRequestMessage +import sbt.internal.protocol.JsonRpcNotificationMessage +import sbt.internal.util.Util +import sjsonnew.BasicJsonProtocol.* + +/** + * A client session that never answers the terminal control queries (set echo, set raw + * mode, attributes, size), for tests exercising the server's handling of a client that + * stalls or dies with such a query outstanding. + */ +final class SilentTerminalSession(socket: Socket) + extends ServerSessionImpl(socket, "silent-terminal-session-read-thread"): + import SilentTerminalSession.silentMethods + val silentQuery = new CountDownLatch(1) + val firstSilentQuery = new AtomicReference[String] + val inputRequested = new CountDownLatch(1) + override protected def onRequest(msg: JsonRpcRequestMessage): Unit = + msg.method match + case m if silentMethods(m) => + Util.ignoreResult(firstSilentQuery.compareAndSet(null, m)) + silentQuery.countDown() + case _ => Util.ignoreResult(sendJsonRpcResponse(msg.id, "bogus")) + override protected def onNotification(msg: JsonRpcNotificationMessage): Unit = + if msg.method == Serialization.readSystemIn then inputRequested.countDown() + super.onNotification(msg) +end SilentTerminalSession + +object SilentTerminalSession: + private val silentMethods = Set( + Serialization.terminalSetEcho, + Serialization.terminalSetRawMode, + Serialization.getTerminalAttributes, + Serialization.setTerminalAttributes, + Serialization.terminalGetSize, + Serialization.terminalSetSize, + ) + def connect(portfile: File): SilentTerminalSession = + val (socket, _) = ClientSocket.socket(portfile, false) + new SilentTerminalSession(socket) +end SilentTerminalSession diff --git a/server-test/src/test/scala/testpkg/FailedLoadPromptTest.scala b/server-test/src/test/scala/testpkg/FailedLoadPromptTest.scala new file mode 100644 index 000000000..02efc40b1 --- /dev/null +++ b/server-test/src/test/scala/testpkg/FailedLoadPromptTest.scala @@ -0,0 +1,92 @@ +/* + * sbt + * Copyright 2023, Scala center + * Copyright 2011 - 2022, Lightbend, Inc. + * Copyright 2008 - 2010, Mark Harrah + * Licensed under Apache License 2.0 (see LICENSE) + */ + +package testpkg + +import java.io.{ InputStream, OutputStream, PrintStream } +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.atomic.AtomicReference +import scala.concurrent.duration.* +import sbt.internal.client.NetworkClient +import sbt.internal.langserver.SbtExecParams +import sbt.internal.langserver.codec.JsonProtocol.given +import sbt.internal.util.Util +import scala.collection.mutable + +/** + * The failed-load prompt must reach the interactive client that triggered the reload, + * and its answer must get back to the server: typing 'r' retries the load. + */ +class FailedLoadPromptTest extends AbstractServerTest { + override val testDirectory: String = "client" + + private class CachingOutputStream extends OutputStream { + private val byteBuffer = new mutable.ArrayBuffer[Byte] + override def write(i: Int) = Util.ignoreResult(synchronized(byteBuffer += i.toByte)) + def text: String = new String(synchronized(byteBuffer.toArray), "UTF-8") + } + private class CachingPrintStream(val cos: CachingOutputStream = new CachingOutputStream) + extends PrintStream(cos, true) { + def text: String = cos.text + } + private class QueueInputStream extends InputStream { + private val queue = new LinkedBlockingQueue[Integer] + def push(s: String): Unit = s.getBytes("UTF-8").foreach(b => queue.put(b.toInt)) + override def read(): Int = queue.take() + } + + private def awaitUntil(deadlineSeconds: Int)(condition: => Boolean): Boolean = { + val deadline = System.nanoTime + deadlineSeconds * 1000000000L + var met = condition + while (!met && System.nanoTime < deadline) { + Thread.sleep(500) + met = condition + } + met + } + + test("an interactive client can answer the failed-load prompt") { + val buildFile = testPath.resolve("build.sbt") + val goodBuild = java.nio.file.Files.readString(buildFile) + val in = new QueueInputStream + val out = new CachingPrintStream + val err = new CachingPrintStream + val exitCode = new AtomicReference[Option[Int]](None) + val clientThread = new Thread("failed-load-prompt-test-client") { + setDaemon(true) + override def run(): Unit = { + val code = NetworkClient.client(testPath.toFile, Array.empty[String], in, out, err, false) + exitCode.set(Some(code)) + } + } + clientThread.start() + assert(awaitUntil(30)(out.text.contains("sbt:")), s"client never attached: ${out.text}") + + java.nio.file.Files.writeString(buildFile, "val = =\n") + in.push("reload\r") + // let the failed-load prompt start reading, then repair the build and answer 'r' + Thread.sleep(20000) + java.nio.file.Files.writeString(buildFile, goodBuild) + in.push("r") + // a successful retry returns the client to the command prompt; run a task to prove it. + // the server-side [success] is observed through the suite session's log notifications, + // since the in-process client does not render exec results on the provided stream. + // the prompt parks the command loop until answered: on develop (reads process stdin) + // any further exec starves; with the fix the client's 'r' resolves it and execs flow. + val id = svr.session.nextId() + svr.session.sendJsonRpc(id, "sbt/exec", SbtExecParams("show name")).get + val served = svr.session.waitForNotificationMsg(90.seconds)(_.method == "build/logMessage") + assert( + served.isSuccess, + s"server did not serve another client after the prompt: exit=${exitCode.get}\nout=${out.text}\nerr=${err.text}" + ) + + in.push("exit\r") + assert(awaitUntil(60)(exitCode.get.isDefined), s"client did not exit: ${out.text}") + } +} diff --git a/server-test/src/test/scala/testpkg/TerminalMapsDrainTest.scala b/server-test/src/test/scala/testpkg/TerminalMapsDrainTest.scala new file mode 100644 index 000000000..93b4ba9ab --- /dev/null +++ b/server-test/src/test/scala/testpkg/TerminalMapsDrainTest.scala @@ -0,0 +1,59 @@ +/* + * sbt + * Copyright 2023, Scala center + * Copyright 2011 - 2022, Lightbend, Inc. + * Copyright 2008 - 2010, Mark Harrah + * Licensed under Apache License 2.0 (see LICENSE) + */ + +package testpkg + +import java.util.concurrent.TimeUnit +import scala.concurrent.duration.* +import sbt.internal.langserver.SbtExecParams +import sbt.internal.util.Util +import sbt.protocol.{ Attach, Serialization, SilentTerminalSession } +import sbt.protocol.codec.JsonProtocol.given +import sbt.internal.langserver.codec.JsonProtocol.given + +/** + * Regression: a client that dies while the server waits on a terminal control answer + * must not strand the parked server-side thread. + */ +class TerminalMapsDrainTest extends AbstractServerTest { + override val testDirectory: String = "client" + + test("a client dying at a failed-reload prompt does not strand the server") { + val portfile = new java.io.File(testPath.toFile, "project/target/active.json") + val buildFile = testPath.resolve("build.sbt") + val goodBuild = java.nio.file.Files.readString(buildFile) + val silent = SilentTerminalSession.connect(portfile) + var parkedOn = "none" + try { + silent.initialize(10.seconds, false).get + Util.ignoreResult( + silent.sendJsonRpc(silent.nextId(), Serialization.attach, Attach(interactive = true)) + ) + java.nio.file.Files.writeString(buildFile, "val = =\n") + silent.sendJsonRpc(silent.nextId(), "sbt/exec", SbtExecParams("reload")).get + val queried = silent.silentQuery.await(60, TimeUnit.SECONDS) + val input = silent.inputRequested.await(5, TimeUnit.SECONDS) + assert(queried || input, "server never queried the terminal nor requested input") + // the command loop is now parked waiting for an answer that never comes + parkedOn = + Option(silent.firstSilentQuery.get).getOrElse(if (input) "readSystemIn" else "unknown") + } finally { + silent.close() + java.nio.file.Files.writeString(buildFile, goodBuild) + } + // EOF at the failed-load prompt maps to 'q' by design: the server must shut down + // cleanly rather than stay parked on the dead client's unanswered query. + def serverAlive: Boolean = + ProcessHandle.current.descendants.anyMatch { ph => + ph.info.command.orElse("").contains("java") + } + val deadline = 90.seconds.fromNow + while (serverAlive && deadline.hasTimeLeft()) Thread.sleep(500) + assert(!serverAlive, s"server must exit after the prompting client dies (parked on $parkedOn)") + } +}