diff --git a/direct-style-scala/skills/direct-style-scala/100-resource-management.md b/direct-style-scala/skills/direct-style-scala/100-resource-management.md index 09b7ef2..570cf93 100644 --- a/direct-style-scala/skills/direct-style-scala/100-resource-management.md +++ b/direct-style-scala/skills/direct-style-scala/100-resource-management.md @@ -2,7 +2,8 @@ ## Dependencies -- `"com.softwaremill.ox" %% "core"` — `useInScope`, `useCloseableInScope` +- `"com.softwaremill.ox" %% "core"` — `useInScope`, `useCloseableInScope`, + `resourceScope` --- @@ -16,10 +17,12 @@ when the scope ends. ## Application resource setup The `Dependencies.create` method acquires all resources within the application's -root scope: +root scope. It only registers resources — no forks — so it declares +`using ResourceScope` rather than `using Ox` (see below); the root `Ox` scope +satisfies it: ```scala -def create(using Ox): Dependencies = +def create(using ResourceScope): Dependencies = val config = Config.read.tap(Config.log) val otel = initializeOtel() val sttpBackend = useInScope( @@ -38,3 +41,33 @@ def create(using Ox): Dependencies = On scope termination (e.g., SIGTERM via `OxApp` — see [Background Processes](110-background-processes.md)), resources are released in reverse acquisition order: first the database pool, then the sttp backend. + +## Resources without concurrency: resourceScope + +When a method only needs scoped cleanup — no forks — don't open a `supervised` +scope just to get `useInScope`: that claims a concurrency capability the code +doesn't use. `resourceScope` (Ox ≥ 1.0.6) starts a dedicated, resource-only +scope, Ox's analogue of `scala.util.Using.Manager`: + +```scala +import ox.{resourceScope, useCloseableInScope} +import java.io.{FileReader, FileWriter} + +def process(): Unit = resourceScope: + val in = useCloseableInScope(FileReader("in.txt")) + val out = useCloseableInScope(FileWriter("out.txt")) + // both closed when the scope completes, out first + out.write(in.read()) +``` + +A resource scope can only be started where no concurrency scope is lexically +visible — verified at compile time, because a fork started inside a visible +resource scope could outlive it, using resources after they have been +released. Within a concurrency scope, register +resources directly instead; to use a resource scope e.g. in a fork's body, +extract it to a method that doesn't take `using Ox` — good practice in itself. + +Methods that only register resources should declare exactly that capability: +`using ResourceScope` instead of `using Ox`. Every concurrency scope is also a +`ResourceScope`, so such a method can be called from both — while its +signature no longer claims the ability to fork. diff --git a/direct-style-scala/skills/direct-style-scala/150-shared-state-across-threads.md b/direct-style-scala/skills/direct-style-scala/150-shared-state-across-threads.md index 0484c8b..c2d427b 100644 --- a/direct-style-scala/skills/direct-style-scala/150-shared-state-across-threads.md +++ b/direct-style-scala/skills/direct-style-scala/150-shared-state-across-threads.md @@ -3,7 +3,8 @@ ## Dependencies - `"com.softwaremill.ox" %% "core"` — `Flow`, `par`, `supervised`, `fork`, - `forkDiscard`, `forkUserDiscard`, channels, actors + `forkDiscard`, `forkUserDiscard`, channels, actors, `computeIntensive`, + `cede` --- @@ -26,6 +27,7 @@ when bridging a foreign API that Ox does not cover. | Mailbox or producer-consumer queue | `ox.channels.Channel[T]` | | Collection fan-out | `Flow.mapPar` or `Flow.mapParUnordered` | | Serialized access to mutable state | `Actor` | +| Long CPU-bound computation | `computeIntensive` (see below) | ## Flows @@ -287,6 +289,51 @@ stateRef.updateAndGet(state => process(state, item)).discard > **Warning:** Never use `AtomicReference` as a lifecycle, cancellation, or > shutdown flag. Model lifecycle with Ox scope structure or channel completion. +## CPU-intensive work + +Ox forks run on virtual threads, which are never preempted: a thread yields +only when it blocks. A long CPU-bound computation therefore monopolizes a +carrier thread — and since there are only as many carriers as CPU cores (by +default), a handful of such computations can starve every other virtual +thread in the process, including latency-sensitive ones like HTTP handlers. +Two remedies +(Ox ≥ 1.0.6), depending on the computation: + +- for short bursts in code you control, call `cede()` about once every + millisecond of computation (a call costs ~1µs): it yields the virtual thread + back to the scheduler and checks the interrupt flag, making the loop both + fair and cancellable; +- for long-running computations, or code that can't be instrumented with + yields (a third-party library), wrap the call in `computeIntensive`: the + computation runs on a pool of platform threads — which the OS preempts — + while the calling virtual thread blocks until the result is available. + +```scala +import ox.* + +def expensive(): BigInt = (1 to 1_000_000).map(BigInt(_)).product + +supervised: + val f1 = fork(computeIntensive(expensive())) + val f2 = fork(computeIntensive(expensive())) + f1.join() + f2.join() +``` + +Because the caller blocks, `computeIntensive` stays structured — the +computation never outlives the enclosing scope — and composes with `fork`, +`par`, `race`, and `mapPar`. On cancellation the pool task is interrupted and +the caller waits for it to complete, so computations should still call +`checkInterrupt()` (or `cede()`) periodically where possible — a +non-cooperating computation delays the scope's shutdown until it finishes, and +a `timeout` around it overshoots accordingly. + +> **Warning:** Scope context does not propagate into the computation: `fork` +> inside `computeIntensive` fails, `ForkLocal`s read their defaults, and +> thread-local integrations (MDC, OpenTelemetry context — see [OpenTelemetry +> Observability](510-opentelemetry-observability.md)) do not cross the +> boundary. Compute the value on the pool; do the forking, logging context, +> and tracing on the virtual-thread side. + ## Scope propagation Prefer local, focused scopes. If a method only needs concurrency to compute its @@ -312,8 +359,10 @@ def loadBoth(userId: UserId, accountId: AccountId): (User, Account) = ``` > **Important:** `(using Ox)` in a method signature means "I will start -> forks or register resources in your scope." If the method manages its own -> concurrency lifecycle, use a local `supervised` block instead. +> forks or register resources in your scope." If the method only registers +> resources and starts no forks, declare the narrower `(using ResourceScope)` +> instead (see [Resource Management](100-resource-management.md)); if it +> manages its own concurrency lifecycle, use a local `supervised` block. ## Choosing the right pattern @@ -326,3 +375,4 @@ Prefer the highest-level primitive that fits the shape of the problem: | **Channel** | Modeling an explicit protocol, mailbox, producer-consumer queue, or callback boundary. | | **Actor** | Multiple concurrent callers must access one mutable object serially. | | **AtomicReference** | A single shared value needs pure atomic updates; never use it for lifecycle flags. | +| **computeIntensive** | Long or non-instrumentable CPU-bound work that must not starve virtual threads. | diff --git a/direct-style-scala/skills/direct-style-scala/170-subprocesses-and-external-streams.md b/direct-style-scala/skills/direct-style-scala/170-subprocesses-and-external-streams.md new file mode 100644 index 0000000..5668c86 --- /dev/null +++ b/direct-style-scala/skills/direct-style-scala/170-subprocesses-and-external-streams.md @@ -0,0 +1,154 @@ +# Subprocesses and External Streams + +Driving a long-lived subprocess (or any blocking external connection — a raw +socket, an SSE stream) whose output you consume line by line, under structured +concurrency. Getting the ownership and teardown shape right *before* writing +the code avoids a deadlock that only shows up at shutdown. + +## Dependencies + +- `"com.softwaremill.ox" %% "core"` — `supervised`, `fork`, `forkDiscard`, + `abandonOnInterruptReads` + +--- + +## Own the work with a scope; return a result, not a live handle + +Decide the owning scope first. A reader that consumes a process's output is +concurrent work, so it belongs to a `supervised` scope whose lifetime matches +the work. The reader fork's **return value is the outcome** — read it with +`join()`; don't publish it through a shared `AtomicReference` that another +thread writes and the caller polls. There is one producer and one value. + +```scala +def runAndCollect(command: Seq[String]): Summary = + supervised: + val process = ProcessBuilder(command*).start() + try + val reader = fork: + scala.io.Source + .fromInputStream(process.getInputStream) + .getLines() + .foldLeft(Summary.empty)(_.add(_)) // the fork returns the Summary + reader.join() // its return value IS the result + finally process.destroyForcibly().discard // see the next section +``` + +> **Warning:** Never return an object that owns running forks or threads, to +> be driven by the caller afterwards — its lifetime then matches no scope, so +> cancellation, error propagation, and cleanup become manual flags and +> `try`/`finally` ladders, exactly the bookkeeping structured concurrency +> removes. Keep the work inside the scope and return a plain value; if the +> caller must interleave with the work (send input, react to events), pass a +> *consumer* into the scope (`run(...)(use: Handle => T)`) rather than handing +> a live handle out of it. + +## Teardown: a pipe read is not interruptible — destroy before the join + +A fork blocked reading a classic `java.io` stream — a subprocess's stdout +pipe, a `FileInputStream`, stdin — does **not** observe `Thread.interrupt`, +permanently and by design (see [which operations are +interruptible](https://ox.softwaremill.com/latest/structured-concurrency/interruptions.html) +in the Ox docs). Ox ends a scope by interrupting its forks and then joining +them, so interruption alone will not stop such a fork, and the join blocks +forever. For a subprocess, what unblocks the read is destroying the process: +that closes the pipe's write-end, and the read returns EOF. Streams that don't +support an asynchronous close — stdin, `FileInputStream` — can't be unblocked +this way at all; wrap those with `abandonOnInterruptReads` (below). + +Blocking *socket* reads are different: on virtual threads — which Ox forks run +on — interruption works since Java 21, destructively closing the socket. A +socket-backed reader is thus ended by scope cancellation on its own; the +deadlock here is specific to pipe and other classic stream reads. + +The destroy must be sequenced **before** the scope joins the fork — which is +why it sits in the scope body's own `finally` above. Ox releases scope-managed +resources (`useInScope`, `releaseAfterScope`) only *after* all forks complete, +so a kill registered there runs when the join has already deadlocked. On the +normal path the process has already exited and the destroy is a no-op; on +cancellation or error it is what lets the scope complete. + +> **Required:** Close or destroy a blocking external resource in the scope +> BODY's `finally`, before the scope joins the reader fork. Do NOT rely on +> `releaseAfterScope` for it — that finalizer runs *after* the join and will +> deadlock on a fork stuck in a non-interruptible read. + +## Kill the whole process tree, not just the direct child + +If the process you spawned is a launcher or wrapper that forks the real worker +(`some-launcher run the-tool …`), destroying only your direct child orphans the +worker. Destroy descendants first, then the root: + +```scala +val handle = process.toHandle +handle.descendants().forEach(_.destroyForcibly().discard) +handle.destroyForcibly().discard +``` + +> **Warning:** A forked grandchild inherits the pipe file descriptors, and a +> POSIX pipe reports EOF only once *every* write-end is closed. An orphan that +> survives the kill keeps the reader blocked forever — terminate the +> descendants, not just the process you directly spawned. + +## Reads you can't unblock by closing: `abandonOnInterruptReads` + +Destroy-and-EOF works because a subprocess supports asynchronous close. Some +resources don't: stdin can't be meaningfully closed, and closing a +`FileInputStream` does not unblock a pending read. And sometimes a single read +should be cancellable (a `timeout` around one read) without tearing the +resource down. For these, use `abandonOnInterruptReads` (Ox ≥ 1.0.6): +it wraps an `InputStream` so the actual reads run on a *detached* virtual +thread — unmanaged, never joined — while the calling fork awaits each chunk +interruptibly. On interruption the wait is abandoned: the fork proceeds with an +`InterruptedException`, and the in-flight chunk is not lost — the next read +returns it. With `closeOnAbandon = true`, an interrupted read instead closes +the underlying stream, and the wrapper becomes permanently closed. + +```scala +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()) +``` + +The trade-off: an abandoned read leaves the detached thread blocked until the +underlying read completes — for stdin, possibly for the application's lifetime. +That is a cheap virtual thread in exchange for interruptibility. For a +subprocess, keep the body-`finally` destroy as the primary teardown — it +terminates the child *and* EOFs the pipe, leaving nothing behind; wrapping the +process's stdout (`process.getInputStream`) additionally protects the reader +from a mis-sequenced teardown, but does not replace the destroy. + +For one-off uninterruptible calls (a JDBC `execute`, a DNS lookup) there is +`abandonOnInterrupt(op)(onAbandon)`, where the cleanup starts on abandonment — +e.g. `abandonOnInterrupt(statement.execute())(connection.close())`. For +resources supporting asynchronous close, the cleanup also unblocks the +abandoned operation, so nothing is leaked. `abandonOnInterruptWrites` covers +the write side — e.g. a write to the child's stdin blocked on a full pipe. + +## Don't let an unread pipe stall the process + +A subprocess blocks once an OS pipe buffer (~64 KB) fills. If you read stdout but +ignore stderr, a chatty process wedges mid-run. Either let the child inherit the +parent's stderr, or drain stderr in its own fork for the resource's lifetime — +a `forkDiscard`, torn down by the same body-`finally` destroy: + +```scala +supervised: + val process = ProcessBuilder(command*).start() + try + forkDiscard: + try scala.io.Source.fromInputStream(process.getErrorStream).getLines().foreach(log.debug) + catch case NonFatal(e) => log.debug("stderr drain ended", e) + val reader = fork(consume(process.getInputStream)) + reader.join() + finally process.destroyForcibly().discard +``` + +The drain catches `NonFatal` — logged, so a real failure stays diagnosable — +because a stray read error must not tear down the scope. It needs no stop +signal: the body-`finally` destroy EOFs its read, and the scope then joins it. diff --git a/direct-style-scala/skills/direct-style-scala/SKILL.md b/direct-style-scala/skills/direct-style-scala/SKILL.md index ad8be96..96b3614 100644 --- a/direct-style-scala/skills/direct-style-scala/SKILL.md +++ b/direct-style-scala/skills/direct-style-scala/SKILL.md @@ -85,6 +85,12 @@ def findUser(id: Id[User])(using DbTx): Either[Fail, User] = * NEVER materialize unbounded data into memory. Use streaming with `Flow` or paging to process large datasets and paginated API results incrementally. +* virtual threads are never preempted — long CPU-bound computations (a few + suffice, e.g. a `mapPar` over such work) can starve every other virtual + thread in the process. Run long or non-instrumentable compute via + `computeIntensive` (platform-thread pool; the blocking caller keeps it + structured); in CPU-bound loops you control, call `cede()` about once per + millisecond. # Direct-style Scala @@ -100,7 +106,23 @@ def findUser(id: Id[User])(using DbTx): Either[Fail, User] = job-level concurrency. Accept a parent `(using Ox)` only when a fork or resource must be tied to that parent scope's lifetime. * keep constructors plain; use factories that take `(using Ox)` and return - values that do not carry the capability. + values that do not carry the capability. If the factory only registers + resources and starts no forks, take the narrower `(using ResourceScope)`; + for scoped cleanup with no enclosing scope and no concurrency, use + `resourceScope` instead of `supervised`. +* decide the owning scope BEFORE writing concurrent code. Model a stream reader + or worker as a fork in a `supervised` scope whose lifetime matches the work; + its result is the fork's **return value** (`fork{…}.join()`), not a value + published through a shared `AtomicReference`. NEVER return an object that owns + running forks/threads to be driven later — its lifetime escapes every scope, + making cancellation and cleanup manual again. Pass a consumer into the scope + instead of handing a live handle out. +* a fork blocked reading a subprocess pipe, stdin, a file, or any other + classic `java.io` stream is NOT ended by scope cancellation — without the + right teardown shape, shutdown deadlocks. BEFORE writing code that drives a + subprocess or reads a blocking external stream (socket, SSE), read + [Subprocesses and External + Streams](170-subprocesses-and-external-streams.md). # Functional programming @@ -247,7 +269,9 @@ https://raw.githubusercontent.com/virtuslab/scala-skill/refs/heads/master/direct sbt/Scalafix boundary enforcement. - [Resource Management](100-resource-management.md) — `useInScope`, - `useCloseableInScope`, reverse-order release, scope-based cleanup. + `useCloseableInScope`, reverse-order release, scope-based cleanup; + `resourceScope` for cleanup without concurrency, `using ResourceScope` as + the narrower capability. - [Background Processes](110-background-processes.md) — `OxApp` entry point, `forkDiscard`/`forkUserDiscard` for daemon vs. user threads, @@ -264,7 +288,15 @@ https://raw.githubusercontent.com/virtuslab/scala-skill/refs/heads/master/direct - [Concurrency and Inter-Thread Communication](150-shared-state-across-threads.md) — Flows for declarative concurrent pipelines (`mapPar`, `merge`, `mapStateful`), Ox primitive selection, channels for worker mailboxes and - shutdown, actors for serialized mutable state. + shutdown, actors for serialized mutable state, `computeIntensive`/`cede` for + CPU-bound work on virtual threads. + +- [Subprocesses and External Streams](170-subprocesses-and-external-streams.md) + — driving a subprocess / socket / SSE reader as a fork whose return value is + the result; why a non-interruptible pipe read needs the resource destroyed + in the scope body's `finally` (before the join) rather than via + `releaseAfterScope`; process-tree teardown; `abandonOnInterruptReads` for + reads that can't be unblocked by closing; pipe back-pressure. ## Error Handling