diff --git a/core/src/main/scala/ox/abandonOnInterrupt.scala b/core/src/main/scala/ox/abandonOnInterrupt.scala new file mode 100644 index 00000000..13d90460 --- /dev/null +++ b/core/src/main/scala/ox/abandonOnInterrupt.scala @@ -0,0 +1,278 @@ +package ox + +import ox.channels.Channel +import ox.channels.ChannelClosed + +import java.io.IOException +import java.io.InputStream +import java.io.OutputStream +import java.util.concurrent.CompletableFuture +import java.util.concurrent.ExecutionException +import java.util.concurrent.atomic.AtomicReference + +/** Runs `op` on a newly-started detached (unmanaged, never-joined) virtual thread, and awaits its result interruptibly. If the calling + * thread is interrupted, the wait is abandoned: an [[InterruptedException]] is thrown, and `op`'s eventual result or exception is + * discarded — note that `op` keeps running, side effects included. If `op` throws while the caller is still waiting, the exception is + * re-thrown to the caller. + * + * As `op` runs outside of any concurrency scope, it must not use scope capabilities (such as forking), and [[ForkLocal]] values are not + * propagated. + * + * Use for blocking operations which don't respond to interruption, and can't be unblocked by closing an underlying resource (see the + * variant with an `onAbandon` parameter otherwise). + */ +def abandonOnInterrupt[T](op: => T): T = abandonOnInterrupt(op)(()) + +/** As [[abandonOnInterrupt]]; additionally, when the wait is abandoned, `onAbandon` is started fire-and-forget on another detached thread — + * never on the interrupted caller, as cleanup (e.g. closing a resource) might itself block. Exceptions thrown by `onAbandon` are + * discarded. + * + * Use `onAbandon` to release the underlying resource, e.g. `abandonOnInterrupt(statement.execute())(connection.close())`: for resources + * which support asynchronous close, this actively unblocks the abandoned operation as well. + */ +def abandonOnInterrupt[T](op: => T)(onAbandon: => Unit): T = + val result = new CompletableFuture[T]() + startDetachedThread("run") { + try result.complete(op).discard + catch case t: Throwable => result.completeExceptionally(t).discard + } + try result.get() + catch + // a direct InterruptedException means the caller was interrupted while waiting: the wait is abandoned; an + // ExecutionException-wrapped one is a failure of `op` itself, re-thrown below like any other op exception + case e: InterruptedException => + fireAndForget("on-abandon")(onAbandon) + throw e + case e: ExecutionException => + val cause = e.getCause + cause.addSuppressed(e) // so that no context is lost + throw cause + end try +end abandonOnInterrupt + +private[ox] def startDetachedThread(name: String)(body: => Unit): Unit = + Thread.ofVirtual().name(s"ox-detached-$name").start(() => body).discard + +/** Runs `body` on a detached thread, discarding any exceptions it throws: used for cleanup which must never block or fail the initiating + * (typically: just-interrupted) thread — e.g. closing a resource, which might itself block. + */ +private[ox] def fireAndForget(name: String)(body: => Unit): Unit = + startDetachedThread(name) { + try body + catch case _: Throwable => () + } + +private def asIOException(t: Throwable): IOException = t match + case io: IOException => io + case _ => new IOException(t) + +/** Wraps `is` so that reads become interruptible: a detached virtual thread reads chunks of up to `chunkSize` bytes from `is`, passing them + * to the returned stream via a rendezvous channel (no read-ahead beyond the single in-flight chunk). An interrupted read abandons the + * wait: the in-flight chunk is not lost, and is returned by the next read. + * + * With `closeOnAbandon = true`, an interrupted read additionally closes `is` (fire-and-forget, on a detached thread), and permanently + * closes the returned stream. + * + * Closing the returned stream closes `is`; note that the detached thread might not terminate immediately: it may stay blocked in its + * current read of `is` until that read returns (closing e.g. a [[java.io.FileInputStream]] does not unblock a pending read), or remain + * parked holding an already-read chunk which won't be received anymore. + * + * The returned stream is not thread-safe (single reader). For stdin, create a single process-wide wrapper (multiple wrappers would compete + * for input). + */ +def abandonOnInterruptReads(is: InputStream, chunkSize: Int = 8192, closeOnAbandon: Boolean = false): InputStream = + require(chunkSize > 0, "chunkSize must be positive") + val chan = Channel.rendezvous[Chunk[Byte]] + + startDetachedThread("reads") { + try + var running = true + while running do + val buf = new Array[Byte](chunkSize) + val n = is.read(buf) + if n == -1 then + chan.doneOrClosed().discard + running = false + // the n > 0 guard (instead of a plain else) prevents sending empty chunks, should a contract-violating + // stream return 0 for a non-empty buffer + else if n > 0 then + // avoiding a copy when the buffer came back full (a fresh buffer is allocated each iteration) + val chunk = Chunk.fromArray(if n == chunkSize then buf else buf.take(n)) + if chan.sendOrClosed(chunk).isInstanceOf[ChannelClosed] then running = false + end if + end while + catch case t: Throwable => chan.errorOrClosed(t).discard + } + + new InputStream: + private var current: Chunk[Byte] = Chunk.empty + private var pos = 0 + private var eof = false + private var closed = false + + // returns false on EOF; after returning true, at least one byte is available in `current` + private def ensureAvailable(): Boolean = + if closed then throw new IOException("Stream closed") + else if eof then false + else if pos < current.length then true + else + try + chan.receiveOrClosed() match + case ChannelClosed.Done => eof = true; false + case ChannelClosed.Error(t) => throw asIOException(t) + case c => current = c.asInstanceOf[Chunk[Byte]]; pos = 0; true + catch + case e: InterruptedException => + // the interrupt arrived before the rendezvous completed, so the detached thread still holds any in-flight + // chunk; unless we're closing, it will be received by the next read + if closeOnAbandon then + closed = true + chan.doneOrClosed().discard + fireAndForget("reads-close")(is.close()) + throw e + + override def read(): Int = + if !ensureAvailable() then -1 + else + val b = current(pos) & 0xff + pos += 1 + b + + override def read(b: Array[Byte], off: Int, len: Int): Int = + java.util.Objects.checkFromIndexSize(off, len, b.length) + if len == 0 then 0 + else if !ensureAvailable() then -1 + else + val n = math.min(len, current.length - pos) + var i = 0 + while i < n do + b(off + i) = current(pos + i) + i += 1 + pos += n + n + end if + end read + + override def available(): Int = if closed || eof then 0 else current.length - pos + + override def close(): Unit = + if !closed then + closed = true + chan.doneOrClosed().discard // signals the detached thread to stop at its next send + is.close() + end new +end abandonOnInterruptReads + +private enum WriteCommand: + case Write(chunk: Chunk[Byte]) + case Flush(ack: CompletableFuture[Unit]) + case Close(ack: CompletableFuture[Unit]) + +/** Wraps `os` so that writes and flushes become interruptible: a detached virtual thread performs the actual writes. Writes are passed + * through a rendezvous channel, providing backpressure: if a write to `os` blocks, the wrapper's operations eventually block as well — but + * interruptibly. `flush()` and `close()` await completion by the detached thread, interruptibly; `close()` closes `os` after completing + * the writes queued before it. If an operation is abandoned before its command is handed over to the detached thread (and `closeOnAbandon` + * is not set), `os` remains open, and `close()` can be retried. + * + * An error thrown by an (asynchronously performed) underlying write surfaces, as an [[IOException]], on the next write/flush/close call; + * the wrapper is then permanently broken. + * + * An interrupted operation abandons the wait. With `closeOnAbandon = true`, it additionally closes `os` (fire-and-forget, on a detached + * thread), and permanently closes the returned stream. + * + * The returned stream is not thread-safe (single writer). + */ +def abandonOnInterruptWrites(os: OutputStream, closeOnAbandon: Boolean = false): OutputStream = + val chan = Channel.rendezvous[WriteCommand] + val pendingError = new AtomicReference[Throwable]() + + startDetachedThread("writes") { + var running = true + while running do + chan.receiveOrClosed() match + case WriteCommand.Write(chunk) => + if pendingError.get() == null then + try os.write(chunk.toArray) + catch case t: Throwable => pendingError.set(t) + case WriteCommand.Flush(ack) => + pendingError.get() match + case null => + try + os.flush(); ack.complete(()).discard + catch + case t: Throwable => + pendingError.set(t) + ack.completeExceptionally(t).discard + case t => ack.completeExceptionally(t).discard + case WriteCommand.Close(ack) => + // a pending write error surfaces on close as well, so that a successful close guarantees all writes landed + val closeError = try + os.close(); null + catch case t: Throwable => t + val error = if pendingError.get() != null then pendingError.get() else closeError + if error == null then ack.complete(()).discard else ack.completeExceptionally(error).discard + running = false + case _: ChannelClosed => running = false + end while + } + + new OutputStream: + private var closed = false + + private def checkUsable(): Unit = + if closed then throw new IOException("Stream closed") + pendingError.get() match + case null => () + case t => throw asIOException(t) + + // runs op; if the wait is abandoned (interrupted), performs the close-on-abandon cleanup & rethrows + private def abandoning[T](op: => T, alreadyClosing: Boolean = false): T = + try op + catch + case e: InterruptedException => + if closeOnAbandon then + closed = true + chan.doneOrClosed().discard + // when abandoning close()'s await, the detached thread is already executing the delivered Close command — + // closing again would run two concurrent os.close() calls + if !alreadyClosing then fireAndForget("writes-close")(os.close()) + throw e + + private def sendCommand(cmd: WriteCommand): Unit = + abandoning { + if chan.sendOrClosed(cmd).isInstanceOf[ChannelClosed] then throw new IOException("Stream closed") + } + + private def await(ack: CompletableFuture[Unit], alreadyClosing: Boolean = false): Unit = + abandoning( + { + try unwrapExecutionException(ack.get()) + catch + case e: IOException => throw e + case e: InterruptedException => throw e + case t: Throwable => throw new IOException(t) + }, + alreadyClosing + ) + + override def write(b: Int): Unit = write(Array[Byte](b.toByte), 0, 1) + + override def write(b: Array[Byte], off: Int, len: Int): Unit = + java.util.Objects.checkFromIndexSize(off, len, b.length) + checkUsable() + if len > 0 then sendCommand(WriteCommand.Write(Chunk.fromArray(b.slice(off, off + len)))) + + override def flush(): Unit = + checkUsable() + val ack = new CompletableFuture[Unit]() + sendCommand(WriteCommand.Flush(ack)) + await(ack) + + override def close(): Unit = + if !closed then + val ack = new CompletableFuture[Unit]() + sendCommand(WriteCommand.Close(ack)) + closed = true + await(ack, alreadyClosing = true) + end new +end abandonOnInterruptWrites diff --git a/core/src/test/scala/ox/AbandonOnInterruptTest.scala b/core/src/test/scala/ox/AbandonOnInterruptTest.scala new file mode 100644 index 00000000..eebe4cb1 --- /dev/null +++ b/core/src/test/scala/ox/AbandonOnInterruptTest.scala @@ -0,0 +1,288 @@ +package ox + +import org.scalatest.concurrent.Eventually +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.scalatest.time.{Millis, Seconds, Span} +import ox.* +import ox.util.Trail + +import java.io.* +import java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger} +import scala.concurrent.duration.* + +class AbandonOnInterruptTest extends AnyFlatSpec with Matchers with Eventually: + override implicit val patienceConfig: PatienceConfig = PatienceConfig(timeout = Span(5, Seconds), interval = Span(10, Millis)) + + "abandonOnInterrupt" should "return the operation's result" in { + abandonOnInterrupt(42) shouldBe 42 + } + + it should "propagate the operation's exception" in { + the[RuntimeException] thrownBy abandonOnInterrupt(throw new RuntimeException("boom")) should have message "boom" + } + + it should "abandon the operation on interrupt, letting it complete on the detached thread" in { + val release = CountDownLatch(1) + val opCompleted = CountDownLatch(1) + + supervised { + timeoutOption(100.millis) { + abandonOnInterrupt { + uninterruptible(release.await()) + opCompleted.countDown() + } + } shouldBe None + } + opCompleted.getCount shouldBe 1 // op still blocked, even though the scope completed + release.countDown() + opCompleted.await() // op completes on the detached thread after being released + } + + it should "run onAbandon when interrupted, on a separate thread" in { + val release = CountDownLatch(1) + val abandoned = CountDownLatch(1) + + supervised { + timeoutOption(100.millis) { + abandonOnInterrupt(uninterruptible(release.await()))(abandoned.countDown()) + } shouldBe None + } + abandoned.await() // onAbandon ran + release.countDown() + } + + it should "not run onAbandon when the operation succeeds or fails" in { + val abandoned = AtomicBoolean(false) + + abandonOnInterrupt(42)(abandoned.set(true)) shouldBe 42 + the[RuntimeException] thrownBy abandonOnInterrupt(throw new RuntimeException("boom"))(abandoned.set(true)) should + have message "boom" + abandoned.get() shouldBe false + } + + it should "treat an InterruptedException thrown by the operation as a failure, not abandonment" in { + val abandoned = AtomicBoolean(false) + an[InterruptedException] shouldBe thrownBy(abandonOnInterrupt(throw new InterruptedException("from op"))(abandoned.set(true))) + Thread.interrupted() shouldBe false // the caller's interrupt flag wasn't set + abandoned.get() shouldBe false + } + + // read() blocks (uninterruptibly) until released, then serves `data` bytes, then EOF + private class BlockingInputStream(release: CountDownLatch, data: Array[Byte]) extends InputStream: + val closed = AtomicBoolean(false) + private var pos = 0 + override def read(): Int = + uninterruptible(release.await()) + if pos < data.length then + val b = data(pos) & 0xff; pos += 1; b + else -1 + override def close(): Unit = closed.set(true) + end BlockingInputStream + + "abandonOnInterruptReads" should "read the content through, in chunks" in { + val data = Array.tabulate[Byte](100)(_.toByte) + val wrapped = abandonOnInterruptReads(ByteArrayInputStream(data), chunkSize = 7) + wrapped.readAllBytes() shouldBe data + wrapped.read() shouldBe -1 + } + + it should "propagate an underlying exception" in { + val failing = new InputStream: + override def read(): Int = throw new IOException("disk on fire") + val wrapped = abandonOnInterruptReads(failing) + the[IOException] thrownBy wrapped.read() should have message "disk on fire" + the[IOException] thrownBy wrapped.read() should have message "disk on fire" // error is persistent + } + + it should "abandon an interrupted read without losing the in-flight chunk" in { + val release = CountDownLatch(1) + val wrapped = abandonOnInterruptReads(BlockingInputStream(release, Array[Byte](42)), chunkSize = 1) + + supervised { + timeoutOption(100.millis)(wrapped.read()) shouldBe None // interrupted: abandoned + release.countDown() + wrapped.read() shouldBe 42 // the chunk read after abandonment is served to the next read + wrapped.read() shouldBe -1 + } + } + + it should "close the underlying stream on abandonment when closeOnAbandon is set" in { + val release = CountDownLatch(1) + val underlying = BlockingInputStream(release, Array[Byte](42)) + val wrapped = abandonOnInterruptReads(underlying, closeOnAbandon = true) + + supervised { + timeoutOption(100.millis)(wrapped.read()) shouldBe None + } + release.countDown() + eventually(underlying.closed.get() shouldBe true) // the close is fire-and-forget on a detached thread + an[IOException] shouldBe thrownBy(wrapped.read()) // the wrapper is permanently closed + } + + it should "close the underlying stream when the wrapper is closed" in { + val underlying = ByteArrayInputStream(Array[Byte](1, 2, 3)) + val closed = AtomicBoolean(false) + val tracking = new FilterInputStream(underlying): + override def close(): Unit = + closed.set(true); super.close() + val wrapped = abandonOnInterruptReads(tracking) + wrapped.read() shouldBe 1 + wrapped.close() + closed.get() shouldBe true + an[IOException] shouldBe thrownBy(wrapped.read()) + } + + it should "report available bytes from the current chunk" in { + val wrapped = abandonOnInterruptReads(ByteArrayInputStream(Array[Byte](1, 2, 3)), chunkSize = 3) + wrapped.available() shouldBe 0 // no chunk received yet + wrapped.read() shouldBe 1 + wrapped.available() shouldBe 2 + wrapped.close() + wrapped.available() shouldBe 0 + } + + // write() blocks (uninterruptibly) until released; records writes and close + private class BlockingOutputStream(release: CountDownLatch) extends OutputStream: + val closed = AtomicBoolean(false) + val written = java.util.concurrent.ConcurrentLinkedQueue[Int]() + override def write(b: Int): Unit = + uninterruptible(release.await()) + written.add(b).discard + override def close(): Unit = closed.set(true) + + "abandonOnInterruptWrites" should "write the content through" in { + val underlying = ByteArrayOutputStream() + val wrapped = abandonOnInterruptWrites(underlying) + wrapped.write(Array[Byte](1, 2, 3)) + wrapped.write(4) + wrapped.flush() + underlying.toByteArray shouldBe Array[Byte](1, 2, 3, 4) + wrapped.close() + } + + it should "close the underlying stream after completing queued writes" in { + val underlying = ByteArrayOutputStream() + val closed = AtomicBoolean(false) + val tracking = new FilterOutputStream(underlying): + override def close(): Unit = + closed.set(true); super.close() + override def write(b: Array[Byte], off: Int, len: Int): Unit = + // rejecting out-of-order writes, so that a close happening before draining queued writes is observable + if closed.get() then throw new IOException("write after close") + underlying.write(b, off, len) + val wrapped = abandonOnInterruptWrites(tracking) + wrapped.write(Array[Byte](1, 2, 3)) + wrapped.close() + closed.get() shouldBe true + underlying.toByteArray shouldBe Array[Byte](1, 2, 3) + an[IOException] shouldBe thrownBy(wrapped.write(4)) + } + + it should "surface a pending write error on close" in { + val failing = new OutputStream: + override def write(b: Int): Unit = throw new IOException("pipe broken") + val wrapped = abandonOnInterruptWrites(failing) + wrapped.write(42) // enqueued; fails asynchronously + an[IOException] shouldBe thrownBy(wrapped.close()) + } + + it should "abandon a flush blocked awaiting completion on interrupt" in { + val release = CountDownLatch(1) + val flushed = AtomicBoolean(false) + val underlying = new OutputStream: + override def write(b: Int): Unit = () + override def flush(): Unit = + uninterruptible(release.await()); flushed.set(true) + val wrapped = abandonOnInterruptWrites(underlying) + + supervised { + // the Flush command is delivered promptly (the detached thread is idle), so the caller blocks awaiting the + // ack, while the detached thread blocks in the underlying flush — this exercises the ack-await abandon path + timeoutOption(100.millis)(wrapped.flush()) shouldBe None + } + release.countDown() + eventually(flushed.get() shouldBe true) + } + + it should "close the underlying stream once when close is abandoned while awaiting completion (closeOnAbandon)" in { + val release = CountDownLatch(1) + val closeCount = AtomicInteger(0) + val underlying = new OutputStream: + override def write(b: Int): Unit = () + override def close(): Unit = + uninterruptible(release.await()); closeCount.incrementAndGet().discard + val wrapped = abandonOnInterruptWrites(underlying, closeOnAbandon = true) + + supervised { + // the Close command is delivered; the caller is interrupted while awaiting the ack — the abandonment must not + // start a second, concurrent close + timeoutOption(100.millis)(wrapped.close()) shouldBe None + } + release.countDown() + // both the delivered Close and any (buggy) second close would complete shortly after the release + Thread.sleep(200) + closeCount.get() shouldBe 1 + } + + it should "surface an asynchronous write error on the next operation" in { + val failing = new OutputStream: + override def write(b: Int): Unit = throw new IOException("pipe broken") + val wrapped = abandonOnInterruptWrites(failing) + wrapped.write(42) // enqueued; the failure happens asynchronously + an[IOException] shouldBe thrownBy(wrapped.flush()) + an[IOException] shouldBe thrownBy(wrapped.write(43)) // the error is persistent + } + + it should "abandon a blocked write on interrupt" in { + val release = CountDownLatch(1) + val underlying = BlockingOutputStream(release) + val wrapped = abandonOnInterruptWrites(underlying) + + supervised { + wrapped.write(1) // handed to the detached thread, which blocks in the underlying write + // the rendezvous channel now backpressures: this write blocks, but interruptibly + timeoutOption(100.millis)(wrapped.write(2)) shouldBe None + } + release.countDown() + eventually(underlying.written.contains(1) shouldBe true) + } + + it should "close the underlying stream on abandonment when closeOnAbandon is set" in { + val release = CountDownLatch(1) + val underlying = BlockingOutputStream(release) + val wrapped = abandonOnInterruptWrites(underlying, closeOnAbandon = true) + + supervised { + wrapped.write(1) + timeoutOption(100.millis)(wrapped.write(2)) shouldBe None + } + release.countDown() + eventually(underlying.closed.get() shouldBe true) + an[IOException] shouldBe thrownBy(wrapped.write(3)) + } + + "abandonOnInterruptReads" should "let a scope cancel a fork blocked on an uninterruptible read" in { + val release = CountDownLatch(1) + val underlying = BlockingInputStream(release, Array[Byte](42)) + val wrapped = abandonOnInterruptReads(underlying, closeOnAbandon = true) + val trail = Trail() + + supervised { + forkDiscard { + try wrapped.read().discard + catch + case e: InterruptedException => + trail.add("fork interrupted") + throw e + } + sleep(100.millis) // let the fork block in the (wrapped, interruptible) read + trail.add("body done") + } // scope end cancels the fork promptly — this must not hang + trail.get shouldBe Vector("body done", "fork interrupted") + release.countDown() + eventually(underlying.closed.get() shouldBe true) + } + +end AbandonOnInterruptTest diff --git a/doc/other/best-practices.md b/doc/other/best-practices.md index 3b361f33..e5c6235e 100644 --- a/doc/other/best-practices.md +++ b/doc/other/best-practices.md @@ -52,50 +52,3 @@ blocking methods within the forks. Virtual threads are normally not visible when using tools such as `jstack` or IntelliJ's debugger. To inspect their stack traces, you'll need to create a thread dump to a file using `jcmd [pid] Thread.dump_to_file [file]`, or use Intellij's thread dump utility, when paused in the debugger. - -## Dealing with uninterruptible stdin - -Some I/O operations, like reading from stdin, block the thread on the `read` syscall and only unblock when data becomes -available or the stream is closed; such a call is uninterruptible. The problem with stdin specifically is that it can't -be easily closed, making it impossible to interrupt such operations directly. This pattern extends to other similar -blocking operations that behave like stdin. - -The solution is to delegate the blocking operation to a separate thread and use a [channel](../streaming/channels.md) -for communication. This thread cannot be managed by Ox, as Ox always attempts to run cleanup on application shutdown and -that means interrupting all forks. Some blocking I/O can't, however, be interrupted on the JVM and the advised way of -dealing with that is to just close the resource which in turn makes read/write methods throw an `IOException`. In the -case of stdin closing it is usually not what you want to do. - -To work around that you can sacrifice a thread and since receiving from a channel is interruptible, this makes the -overall operation interruptible as well: - -```scala -import ox.*, channels.* -import scala.io.StdIn - -object stdinSupport: - private lazy val chan: Channel[String] = - val rvChan = Channel.rendezvous[String] - - Thread - .ofVirtual() - .start(() => forever(rvChan.sendOrClosed(StdIn.readLine()).discard)) - - rvChan - - def readLineInterruptibly: String = - try chan.receive() - catch - case iex: InterruptedException => - // Handle interruption appropriately - throw iex -``` - -This pattern allows you to use stdin (or similar blocking operations) with Ox's timeout and interruption mechanisms, -such as `timeoutOption` or scope cancellation. - -Note that for better stdin performance, you can use `Channel.buffered` instead of a rendezvous channel, or even use -`java.lang.System.in` directly and proxy raw data through the channel. Keep in mind that this solution leaks a thread -that will remain blocked on stdin for the lifetime of the application. It's possible to avoid this trade-off by using -libraries that employ JNI/JNA to access stdin, such as [JLine 3](https://jline.org/docs/intro), which can use raw mode -with non-blocking or timeout-based reads, allowing the thread to be properly interrupted and cleaned up. \ No newline at end of file diff --git a/doc/other/dictionary.md b/doc/other/dictionary.md index e5bbc0fc..370b110a 100644 --- a/doc/other/dictionary.md +++ b/doc/other/dictionary.md @@ -37,6 +37,11 @@ Errors: Other: * **computation combinator**: a method which takes user-provided functions and manages their execution, e.g. using concurrency, interruption, and appropriately handling errors; examples include `par`, `race`, `retry`, `timeout` +* **detached thread**: an unmanaged virtual thread, running outside of any concurrency scope, which is never joined; + used to perform uninterruptible blocking operations on behalf of forks (see `abandonOnInterrupt`) +* **abandoning** an operation: when a fork waiting for the result of a blocking, uninterruptible operation is + interrupted, the operation's result is given up: the operation keeps running on its detached thread (as it can't be + interrupted), and its eventual result, or exception, is discarded Channels: * **values** can be **sent** to a channel, or **received** from a channel diff --git a/doc/structured-concurrency/interruptions.md b/doc/structured-concurrency/interruptions.md index 0fe59179..3c63e82e 100644 --- a/doc/structured-concurrency/interruptions.md +++ b/doc/structured-concurrency/interruptions.md @@ -1,7 +1,23 @@ # Interruptions -When catching exceptions, care must be taken not to catch & fail to propagate an `InterruptedException`. Doing so will -prevent the scope cleanup mechanisms to make appropriate progress, as the scope won't finish until all started threads +Ox implements cancellation using Java's interruptions. When a concurrency scope +ends (or when using operations such as `timeout` or `race`), any forks that are +still running are interrupted by calling `Thread.interrupt()` on their backing +(virtual) threads. This mechanism is cooperative: an interruption is only "seen" +by a fork when it is blocked in an interruptible operation (which then throws an +`InterruptedException`), or when the code explicitly checks the thread's +interrupt flag. Hence, two things can go wrong: + +* the fork catches, but doesn't propagate the `InterruptedException`, or +* the fork is blocked in an operation which is not interruptible at all. + +Both cases are covered below. + +## Propagating `InterruptedException`s + +When catching exceptions, care must be taken not to catch & fail to propagate an +`InterruptedException`. Doing so will prevent the scope cleanup mechanisms to +make appropriate progress, as the scope won't finish until all started threads complete. A good solution is to catch only non-fatal exception using `NonFatal`, e.g.: @@ -26,3 +42,165 @@ supervised { // do something else } ``` + +## Which operations are interruptible? + +Most blocking operations from the JDK's concurrency toolbox are interruptible, +and respond to interruption by throwing an `InterruptedException`: +[`Object.wait`, `Thread.sleep`, +`Thread.join`](https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/lang/Thread.html#interrupt()), +`LockSupport.park`, `ReentrantLock.lockInterruptibly`, blocking queue +operations, and all blocking operations provided by Ox itself (such as receiving +from a [channel](../streaming/channels.md)). + +For I/O, the situation is more nuanced: + +* **NIO channels** implementing + [`InterruptibleChannel`](https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/nio/channels/InterruptibleChannel.html) + (`FileChannel`, `SocketChannel`, `ServerSocketChannel`, `DatagramChannel`, + `Pipe` channels) are interruptible, but destructively so: the interrupt + **closes the channel**, and the blocked thread receives a + `ClosedByInterruptException`. Beware that for `FileChannel`, interruption + closes the channel but doesn't abort an in-flight kernel read/write, and an + interrupted read [has been observed to return normally while silently closing + the channel](https://bugs.openjdk.org/browse/JDK-6979009) (see also [this + analysis](https://gamlor.info/posts-output/2019-11-13-file-channel-closes-when-interrupted/en/)). +* **Blocking socket I/O** (`java.net.Socket`, `ServerSocket`, `DatagramSocket`) + is interruptible **on virtual threads** — which is what Ox's forks run on. + Since [JEP 444](https://openjdk.org/jeps/444) (Java 21), these blocking + methods are specified to be interruptible when invoked on a virtual thread: + the interrupt closes the socket, and the blocked thread receives a + `SocketException`. Note that this only applies to virtual threads — see below + for platform threads. +* **`Process.waitFor`** is interruptible, and throws an `InterruptedException`. + +As with channels, socket interruption is destructive: after a cancelled read the +socket is closed and can't be reused. That's usually fine for cancellation +purposes, but it does mean an interrupt can't be used to merely "abandon" a +single read attempt. + +## Uninterruptible operations + +Some blocking operations on the JVM don't respond to interruption at all: the +interrupt flag is set, but the thread stays blocked. There is [no general +technique to unblock such a +thread](https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/lang/doc-files/threadPrimitiveDeprecation.html) +— the JDK's official advice is to close the underlying resource from another +thread, which causes the blocked operation to fail with an exception (e.g. an +`IOException`). Note that this only works for resources supporting asynchronous +close — sockets, NIO channels, subprocesses (via destroy) — but not for e.g. +`FileInputStream`, where closing does not unblock an in-progress read. + +When a fork is blocked in such an operation, ending a scope (or cancelling via +`timeout`/`race`) will stall until the operation completes on its own. Known +uninterruptible operations include: + +* **Classic `java.io` stream reads & writes**: `InputStream.read` / + `OutputStream.write` and their subclasses & wrappers — `System.in`, + `FileInputStream`, `FileOutputStream`, `RandomAccessFile`, `Reader`s/`Writer`s + over them. This is intended, permanent behavior: the request to make such + reads interruptible was [closed as "Won't + Fix"](https://bugs.openjdk.org/browse/JDK-4514257). This applies also on + virtual threads, for non-socket sources. +* **`System.out.println` (and other console writes), when blocked**: printing + blocks if the console, pipe or terminal on the other end isn't consuming data + (e.g. a full pipe buffer). The write is a classic stream write, so it's + [uninterruptible](https://bugs.openjdk.org/browse/JDK-4514257); moreover, + since `PrintStream` methods are `synchronized`, other threads calling + `println` will then block uninterruptibly on the monitor as well. +* **Socket I/O on platform threads**: unlike on virtual threads, blocking + `java.net.Socket` reads/writes/accepts on a platform thread [do not respond to + interruption](https://wiki.sei.cmu.edu/confluence/display/java/THI04-J.+Ensure+that+threads+performing+blocking+operations+can+be+terminated) + — only closing the socket unblocks them. This matters for code running outside + of Ox scopes, or in libraries that manage their own (platform) thread pools. +* **DNS resolution**: `InetAddress.getByName` delegates to the OS resolver, with + [no way to set a timeout or interrupt the + lookup](https://bbossola.wordpress.com/2015/10/23/java-no-timeout-on-dns-resolution/) + from Java. +* **JDBC calls**: e.g. `Statement.execute` [ignores + interrupts](https://bugs.openjdk.org/browse/JDK-6393812). The sanctioned + cancellation mechanism, `Statement.cancel`, is a cooperative, server-side + protocol: the blocked thread resumes only once the database sends back an + error reply — so [it won't help if the server is hung or the network is + broken](https://docs.oracle.com/en/database/oracle/oracle-database/18/jjdbc/JDBC-troubleshooting.html). + Use `setQueryTimeout` / `setNetworkTimeout` defensively. +* **Acquiring locks**: entering a `synchronized` block/method, as well as + [`ReentrantLock.lock()`](https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/locks/ReentrantLock.html#lock()), + block ignoring interrupts. If lock acquisition should be cancellable, use + [`lockInterruptibly()`](https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/locks/ReentrantLock.html#lockInterruptibly()). +* **Reading a subprocess's output/error streams**: these are classic streams + (see above); the reliable way to unblock a reader is to [destroy the + process](https://www.dontpanicblog.co.uk/2023/05/07/handling-blocked-process-output-stream/). +* **Memory-mapped file access**: touching a `MappedByteBuffer` page that must be + faulted in from disk blocks inside the kernel, below any Java-level + interruption mechanism; such [page faults can take hundreds of + milliseconds](https://groups.google.com/g/mechanical-sympathy/c/yL4Yaedgqg4) + (or longer, e.g. on network filesystems). + +To make cancellation work with such operations, run them through the +abandon-on-interrupt utilities described below, which delegate the blocking +operation to a detached thread — so that waiting becomes interruptible. Note +that regular [scope-managed resources](../utils/resources.md) (`useInScope`, +`releaseAfterScope`) are released only after all forks complete, so they won't +unblock a fork stuck in uninterruptible I/O. + +## Abandoning uninterruptible operations + +Since an uninterruptible operation can't be cancelled, the way out is to not +wait for it uninterruptibly: the operation is delegated to a **detached thread** +— an unmanaged virtual thread, running outside of any concurrency scope, which +is never joined — while the calling fork awaits the result interruptibly. On +interruption, the operation is **abandoned**: its result is given up (the fork +proceeds with an `InterruptedException`), while the detached thread keeps +running until the operation completes — its result is then discarded. Ox +provides three utilities implementing this pattern: + +* `abandonOnInterrupt(op)` — runs a one-off blocking operation on a detached + thread, awaiting its result interruptibly. An optional second parameter list + provides cleanup, started (on another detached thread) when the wait is + abandoned: `abandonOnInterrupt(statement.execute())(connection.close())`. For + resources supporting asynchronous close, the cleanup actively unblocks the + abandoned operation as well — nothing is leaked. +* `abandonOnInterruptReads(inputStream)` — wraps an `InputStream`, so that reads + become interruptible; chunks are read by a detached thread and passed through + a rendezvous channel. An interrupted read doesn't lose the in-flight chunk — + it's returned by the next read. With `closeOnAbandon = true`, an interrupted + read additionally closes the underlying stream, and the wrapper becomes + permanently closed. +* `abandonOnInterruptWrites(outputStream)` — the same for writes: the actual + writing happens on a detached thread, and the wrapper's + `write`/`flush`/`close` block interruptibly. This also covers writes blocked + on a full pipe — the case of a stuck `println`. + +For example, reading from stdin (which can neither be interrupted nor +meaningfully closed) in a way that works with Ox's timeouts and scope +cancellation: + +```scala mdoc:compile-only +import ox.* +import scala.concurrent.duration.* + +// one process-wide wrapper: multiple wrappers over System.in would compete for input +lazy val stdin = abandonOnInterruptReads(System.in) + +supervised { + val firstByte: Option[Int] = timeoutOption(1.second)(stdin.read()) + println(s"Read: $firstByte") +} +``` + +If the operation is abandoned, the detached thread remains blocked until the +underlying operation completes — for stdin, possibly for the application's +lifetime. This is the deliberate trade-off of the pattern: a (cheap, virtual) +thread is potentially left behind, in exchange for interruptibility. When the +underlying resource supports asynchronous close (sockets, subprocesses via +destroy), prefer passing `onAbandon` cleanup / `closeOnAbandon = true`, which +unblocks the detached thread too. Also note that abandoned operations keep +running — including their side effects (e.g. an abandoned JDBC query keeps +executing on the server; use `Statement.cancel`-based cleanup in `onAbandon` +where relevant). + +Alternatively, for terminal I/O specifically, libraries that access stdin via +JNI/JNA, such as [JLine 3](https://jline.org/docs/intro), can use raw mode with +non-blocking or timeout-based reads, avoiding the detached-thread trade-off +altogether.