diff --git a/core/src/main/scala/ox/Ox.scala b/core/src/main/scala/ox/Ox.scala index e65a170c..1bf97b74 100644 --- a/core/src/main/scala/ox/Ox.scala +++ b/core/src/main/scala/ox/Ox.scala @@ -8,6 +8,29 @@ import ox.internal.ThreadHerd import java.util.concurrent.atomic.AtomicReference import scala.annotation.implicitNotFound +/** Capability granted by a [[resourceScope]] and, via subtyping, by any concurrency scope ([[supervised]], [[supervisedError]], + * [[unsupervised]]). + * + * Represents a capability to register resources (e.g. using [[useInScope]] or [[releaseAfterScope]]) to be released when the scope + * completes. Does not allow forking. + * + * @see + * [[OxUnsupervised]], [[Ox]] + */ +@implicitNotFound( + "This operation must be run within a `resourceScope`, or any concurrency scope (`supervised`, `supervisedError` or `unsupervised`). " + + "Alternatively, you must require that the enclosing method is run within a scope, by adding a `using ResourceScope` parameter list." +) +trait ResourceScope: + // contains null once the scope's finalizers have been run; registration then throws (see addFinalizer) + private[ox] def finalizers: AtomicReference[List[() => Unit]] + private[ox] def addFinalizer(f: () => Unit): Unit = + finalizers.updateAndGet { + case null => throw new IllegalStateException("Cannot register a resource: the scope to which it would be attached has already ended") + case fs => f :: fs + }.discard +end ResourceScope + /** Capability granted by an [[unsupervised]] concurrency scope (as well as, via subtyping, by [[supervised]] and [[supervisedError]]). * * Represents a capability to: @@ -21,16 +44,14 @@ import scala.annotation.implicitNotFound @implicitNotFound( "This operation must be run within a `supervised`, `supervisedError` or `unsupervised` block. Alternatively, you must require that the enclosing method is run within a scope, by adding a `using OxUnsupervised` parameter list." ) -trait OxUnsupervised: +trait OxUnsupervised extends ResourceScope: private[ox] def herd: ThreadHerd - private[ox] def finalizers: AtomicReference[List[() => Unit]] private[ox] def supervisor: Supervisor[Nothing] - private[ox] def addFinalizer(f: () => Unit): Unit = finalizers.updateAndGet(f :: _).discard private[ox] def parent: Option[OxUnsupervised] private[ox] def locals: ForkLocalMap end OxUnsupervised -/** Capability granted by an [[supervised]] or [[supervisedError]] concurrency scope. +/** Capability granted by a [[supervised]] or [[supervisedError]] concurrency scope. * * Represents a capability to: * - fork supervised or unsupervised, asynchronously running computations in a concurrency scope. Such forks can be created using diff --git a/core/src/main/scala/ox/resource.scala b/core/src/main/scala/ox/resource.scala index 1fe84802..705552eb 100644 --- a/core/src/main/scala/ox/resource.scala +++ b/core/src/main/scala/ox/resource.scala @@ -1,38 +1,93 @@ package ox -/** Use the given resource in the current concurrency scope. The resource is allocated using `acquire`, and released after the all forks in - * the scope complete (either successfully or with an error), using `release`. Releasing is [[uninterruptible]]. +import java.util.concurrent.atomic.AtomicReference +import scala.annotation.implicitNotFound +import scala.annotation.targetName +import scala.util.NotGiven + +@implicitNotFound( + "resourceScope cannot be started here: a concurrency scope is visible, and forks started within the resource scope could outlive it. " + + "Extract the resourceScope usage to a method which doesn't take a `using Ox` parameter." +) +opaque type NoEnclosingConcurrencyScope = Unit + +object NoEnclosingConcurrencyScope: + // in the companion, so that it's found via the implicit scope of the type, without any imports + given noEnclosingConcurrencyScope(using NotGiven[OxUnsupervised]): NoEnclosingConcurrencyScope = () + +/** Starts a new resource scope: within the given code block `f`, resources can be registered using [[useInScope]] and + * [[releaseAfterScope]]. They are released, in reverse registration order, once `f` completes (either successfully or with an exception). + * Releasing is [[uninterruptible]]. A resource scope is not a concurrency scope: no forks can be started and no [[ForkLocal]] values can + * be bound. The stdlib's analogue is `scala.util.Using.Manager`. + * + * Any concurrency scope ([[supervised]], [[supervisedError]], [[unsupervised]]) is also a resource scope, so within one you can register + * resources directly. Starting a resource scope there is disallowed (verified at compile-time), because forks started in a lexically + * visible resource scope could outlive it, using or registering resources after they've been released. For the same reason, the + * [[ResourceScope]] capability must not leak out of the scope: registration after the scope ends throws an [[IllegalStateException]]. + * + * Finalizers run with the [[ForkLocal]] values in effect where `resourceScope` was called — the same values the body sees. A finalizer + * registered through a leaked capability from a nested [[ForkLocal]] binding does not see that nested binding. + */ +def resourceScope[T](f: ResourceScope ?=> T)(using NoEnclosingConcurrencyScope): T = + val scope = new ResourceScope: + private[ox] val finalizers = new AtomicReference[List[() => Unit]](Nil) + val result = + try Right(f(using scope)) + catch case e: Throwable => Left(e) + runFinalizers(scope, result) + +/** Use the given resource in the current scope. The resource is allocated using `acquire`, and released using `release` when the scope + * completes, in reverse registration order. For concurrency scopes, release happens after all forks started within the scope have + * completed (either successfully or with an exception). Releasing is [[uninterruptible]]. + * + * If the scope has already ended (which can only happen when using a leaked, explicitly-passed capability), the resource is acquired, + * immediately released — so that cleanup is never lost — and an [[IllegalStateException]] is thrown. */ -def useInScope[T](acquire: => T)(release: T => Unit)(using OxUnsupervised): T = +def useInScope[T](acquire: => T)(release: T => Unit)(using rs: ResourceScope): T = val t = acquire - summon[OxUnsupervised].addFinalizer(() => release(t)) + try rs.addFinalizer(() => release(t)) + catch + case e: Throwable => + try uninterruptible(release(t)) + catch case e2: Throwable => e.addSuppressed(e2) + throw e t +end useInScope -/** Use the given resource, which implements [[AutoCloseable]], in the current concurrency scope. The resource is allocated using `acquire`, - * and released after the all forks in the scope complete (either successfully or with an error), using [[AutoCloseable.close()]]. - * Releasing is [[uninterruptible]]. - */ -def useCloseableInScope[T <: AutoCloseable](c: => T)(using OxUnsupervised): T = useInScope(c)(_.close()) +/** As [[useInScope]], but the resource, which implements [[AutoCloseable]], is released using [[AutoCloseable.close()]]. */ +def useCloseableInScope[T <: AutoCloseable](c: => T)(using rs: ResourceScope): T = useInScope(c)(_.close()) -/** Release the given resource, by running the `release` code block. Releasing is done after all the forks in the scope complete (either - * successfully or with an error), but before the current concurrency scope completes. Releasing is [[uninterruptible]]. - */ -def releaseAfterScope(release: => Unit)(using OxUnsupervised): Unit = useInScope(())(_ => release) +/** As [[useInScope]], but nothing is acquired — only the `release` code block is registered, to be run when the scope completes. */ +def releaseAfterScope(release: => Unit)(using rs: ResourceScope): Unit = useInScope(())(_ => release) -/** Release the given resource, which implements [[AutoCloseable]], by running its `.close()` method. Releasing is done after all the forks - * in the scope complete (either successfully or with an error), but before the current concurrency scope completes. Releasing is - * [[uninterruptible]]. - */ -def releaseCloseableAfterScope(toRelease: AutoCloseable)(using OxUnsupervised): Unit = useInScope(())(_ => toRelease.close()) +/** As [[releaseAfterScope]], but closes the given [[AutoCloseable]] resource. */ +def releaseCloseableAfterScope(toRelease: AutoCloseable)(using rs: ResourceScope): Unit = useInScope(())(_ => toRelease.close()) + +// binary-compatibility bridges for callers compiled against previous ox versions; remove in 2.0. The Scala-level +// names differ from the originals (to avoid overload ambiguity at in-package call sites), while @targetName restores +// the original JVM names, preserving linkage. + +@targetName("useInScope") +private[ox] def useInScopeCompat[T](acquire: => T)(release: T => Unit)(using ox: OxUnsupervised): T = + useInScope(acquire)(release) +@targetName("useCloseableInScope") +private[ox] def useCloseableInScopeCompat[T <: AutoCloseable](c: => T)(using ox: OxUnsupervised): T = + useCloseableInScope(c) +@targetName("releaseAfterScope") +private[ox] def releaseAfterScopeCompat(release: => Unit)(using ox: OxUnsupervised): Unit = + releaseAfterScope(release) +@targetName("releaseCloseableAfterScope") +private[ox] def releaseCloseableAfterScopeCompat(toRelease: AutoCloseable)(using ox: OxUnsupervised): Unit = + releaseCloseableAfterScope(toRelease) /** Use the given resource, acquired using `acquire` and released using `release` in the given `f` code block. Releasing is - * [[uninterruptible]]. To use multiple resources, consider creating a [[supervised]] scope and [[useInScope]] method. + * [[uninterruptible]]. To use multiple resources, consider creating a [[resourceScope]] and using the [[useInScope]] method. */ inline def use[R, T](inline acquire: R, inline release: R => Unit)(inline f: R => T): T = useInterruptible(acquire, r => uninterruptible(release(r)))(f) /** Use the given resource, acquired using `acquire` and released using `release` in the given `f` code block. Releasing might be - * interrupted. To use multiple resources, consider creating a [[supervised]] scope and [[useInScope]] method. + * interrupted. To use multiple resources, consider creating a [[resourceScope]] and using the [[useInScope]] method. * * Equivalent to a `try`-`finally` block. */ @@ -54,6 +109,6 @@ inline def useInterruptible[R, T](inline acquire: R, inline release: R => Unit)( end useInterruptible /** Use the given [[AutoCloseable]] resource, acquired using `acquire` in the given `f` code block. Releasing is [[uninterruptible]]. To use - * multiple resources, consider creating a [[supervised]] scope and [[useCloseableInScope]] method. + * multiple resources, consider creating a [[resourceScope]] and using the [[useCloseableInScope]] method. */ inline def useCloseable[R <: AutoCloseable, T](inline acquire: R)(inline f: R => T): T = use(acquire, _.close())(f) diff --git a/core/src/main/scala/ox/unsupervised.scala b/core/src/main/scala/ox/unsupervised.scala index 067004f3..b9e40aec 100644 --- a/core/src/main/scala/ox/unsupervised.scala +++ b/core/src/main/scala/ox/unsupervised.scala @@ -25,30 +25,6 @@ private[ox] def unsupervised[T](locals: ForkLocalMap, f: OxUnsupervised ?=> T): scopedWithCapability(OxError(NoOpSupervisor, NoErrorMode, Option(currentScope.get()), locals))(f) private[ox] def scopedWithCapability[T](capability: Ox)(f: Ox ?=> T): T = - def throwWithSuppressed(es: List[Throwable]): Nothing = - val e = es.head - es.tail.foreach(e.addSuppressed) - throw e - - def runFinalizers(result: Either[Throwable, T]): T = - val fs = capability.finalizers.get - if fs.isEmpty then result.fold(throw _, identity) - else - val es = uninterruptible { - fs.flatMap { f => - try - f(); None - catch case e: Throwable => Some(e) - } - } - - result match - case Left(e) => throwWithSuppressed(e :: es) - case Right(t) if es.isEmpty => t - case _ => throwWithSuppressed(es) - end if - end runFinalizers - def runWithCurrentScopeSet = val result = try @@ -59,8 +35,8 @@ private[ox] def scopedWithCapability[T](capability: Ox)(f: Ox ?=> T): T = catch case e: Throwable => Left(e) // running the finalizers only once we are sure that all child threads have been terminated, so that no new - // finalizers are added, and none are lost - runFinalizers(result) + // finalizers are added, and none are lost; registrations after the freeze (via leaked capabilities) throw + runFinalizers(capability, result) end runWithCurrentScopeSet val previousScope = currentScope.get() @@ -69,3 +45,32 @@ private[ox] def scopedWithCapability[T](capability: Ox)(f: Ox ?=> T): T = runWithCurrentScopeSet finally currentScope.set(previousScope) end scopedWithCapability + +/** Runs the scope's finalizers (in reverse registration order, uninterruptibly), first freezing the finalizer list (by setting it to + * `null`), so that later registrations fail with an exception (see [[ResourceScope.addFinalizer]]) instead of being silently lost. Must be + * called exactly once per scope. Returns the scope's result: the body exception is re-thrown with finalizer exceptions suppressed; + * finalizer exceptions alone fail the scope. + */ +private[ox] def runFinalizers[T](scope: ResourceScope, result: Either[Throwable, T]): T = + def throwWithSuppressed(es: List[Throwable]): Nothing = + val e = es.head + es.tail.foreach(e.addSuppressed) + throw e + + val fs = scope.finalizers.getAndSet(null) + val es = + if fs.isEmpty then Nil + else + uninterruptible { + fs.flatMap { f => + try + f(); None + catch case e: Throwable => Some(e) + } + } + + result match + case Left(e) => throwWithSuppressed(e :: es) + case Right(t) if es.isEmpty => t + case _ => throwWithSuppressed(es) +end runFinalizers diff --git a/core/src/test/scala/ox/ResourceTest.scala b/core/src/test/scala/ox/ResourceTest.scala index c1c30940..187c7bb0 100644 --- a/core/src/test/scala/ox/ResourceTest.scala +++ b/core/src/test/scala/ox/ResourceTest.scala @@ -6,55 +6,62 @@ import ox.* import ox.util.Trail class ResourceTest extends AnyFlatSpec with Matchers: - "useInScope" should "release resources after allocation" in { - val trail = Trail() - - unsupervised { - val r = useInScope { trail.add("allocate"); 1 }(n => trail.add(s"release $n")) - r shouldBe 1 - trail.get shouldBe Vector("allocate") + private val scopeKinds: List[(String, (ResourceScope ?=> Unit) => Unit)] = List( + "unsupervised" -> (f => unsupervised(f)), + "resourceScope" -> (f => resourceScope(f)) + ) + + for (kind, runScope) <- scopeKinds do + s"useInScope in $kind" should "release resources in reverse order" in { + val trail = Trail() + + runScope { + val r1 = useInScope { trail.add("allocate 1"); 1 }(n => trail.add(s"release $n")) + val r2 = useInScope { trail.add("allocate 2"); 2 }(n => trail.add(s"release $n")) + r1 shouldBe 1 + r2 shouldBe 2 + trail.get shouldBe Vector("allocate 1", "allocate 2") + } + trail.get shouldBe Vector("allocate 1", "allocate 2", "release 2", "release 1") } - trail.get shouldBe Vector("allocate", "release 1") - } - it should "release resources in reverse order" in { - val trail = Trail() - - unsupervised { - val r1 = useInScope { trail.add("allocate 1"); 1 }(n => trail.add(s"release $n")) - val r2 = useInScope { trail.add("allocate 2"); 2 }(n => trail.add(s"release $n")) - r1 shouldBe 1 - r2 shouldBe 2 - trail.get shouldBe Vector("allocate 1", "allocate 2") + it should "release resources when there's an exception" in { + val trail = Trail() + + try + runScope { + val r1 = useInScope { + trail.add("allocate 1"); 1 + }(n => trail.add(s"release $n")) + val r2 = useInScope { + trail.add("allocate 2"); 2 + }(n => trail.add(s"release $n")) + r1 shouldBe 1 + r2 shouldBe 2 + throw new RuntimeException + } + catch case _ => trail.add("exception") + end try + trail.get shouldBe Vector("allocate 1", "allocate 2", "release 2", "release 1", "exception") } - trail.get shouldBe Vector("allocate 1", "allocate 2", "release 2", "release 1") - } - it should "release resources when there's an exception" in { + end for + + "releaseAfterScope" should "release registered resources" in { val trail = Trail() - try - unsupervised { - val r1 = useInScope { - trail.add("allocate 1"); 1 - }(n => trail.add(s"release $n")) - val r2 = useInScope { - trail.add("allocate 2"); 2 - }(n => trail.add(s"release $n")) - r1 shouldBe 1 - r2 shouldBe 2 - throw new RuntimeException - } - catch case _ => trail.add("exception") - end try - trail.get shouldBe Vector("allocate 1", "allocate 2", "release 2", "release 1", "exception") + resourceScope { + releaseAfterScope(trail.add("release")) + trail.add("in scope") + } + trail.get shouldBe Vector("in scope", "release") } - it should "release resources when there's an exception during releasing (normal resutl)" in { + "useInScope" should "release resources when there's an exception during releasing (normal result)" in { val trail = Trail() try - unsupervised { + resourceScope { val r1 = useInScope { trail.add("allocate 1"); 1 @@ -71,18 +78,17 @@ class ResourceTest extends AnyFlatSpec with Matchers: } r1 shouldBe 1 r2 shouldBe 2 - r1 + r2 } catch case e => trail.add(s"exception ${e.getMessage}") end try trail.get shouldBe Vector("allocate 1", "allocate 2", "release 2", "release 1", "exception e2") } - it should "release resources when there's an exception during releasing (exceptional resutl)" in { + it should "release resources when there's an exception during releasing (exceptional result)" in { val trail = Trail() try - unsupervised { + resourceScope { val r1 = useInScope { trail.add("allocate 1"); 1 @@ -106,14 +112,71 @@ class ResourceTest extends AnyFlatSpec with Matchers: trail.get shouldBe Vector("allocate 1", "allocate 2", "release 2", "release 1", "exception e3") } - it should "release registered resources" in { + "resourceScope" should "attach resources to the nearest scope when nested in a concurrency scope" in { val trail = Trail() + def inner(): Unit = resourceScope { + releaseAfterScope(trail.add("inner release")) + trail.add("inner body") + } + supervised { + releaseAfterScope(trail.add("outer release")) + inner() + trail.add("outer body") + } + trail.get shouldBe Vector("inner body", "inner release", "outer body", "outer release") + } - unsupervised { - releaseAfterScope(trail.add("release")) - trail.add("in scope") + it should "attach resources to the nearest scope when a concurrency scope is nested inside" in { + val trail = Trail() + resourceScope { + releaseAfterScope(trail.add("outer release")) + supervised { + releaseAfterScope(trail.add("inner release")) + trail.add("inner body") + } + trail.add("outer body") } - trail.get shouldBe Vector("in scope", "release") + trail.get shouldBe Vector("inner body", "inner release", "outer body", "outer release") + } + + it should "throw when registering via a leaked capability after the scope ended" in { + var leaked: ResourceScope = null + resourceScope { leaked = summon[ResourceScope] } + an[IllegalStateException] shouldBe thrownBy(releaseAfterScope(())(using leaked)) + } + + it should "not compile within a visible concurrency scope, but compile via a capability-free method" in { + "supervised { resourceScope { } }" shouldNot typeCheck + "def m()(using Ox): Unit = resourceScope { }" shouldNot typeCheck + "def m(): Unit = resourceScope { }" should compile + "resourceScope { resourceScope { } }" should compile + } + + it should "not allow forking without a concurrency scope" in { + "resourceScope { forkDiscard { } }" shouldNot typeCheck + } + + it should "see fork local values of its level, in body and finalizers" in { + val fl = ForkLocal("default") + val trail = Trail() + def scoped(): Unit = resourceScope { + releaseAfterScope(trail.add(s"release ${fl.get()}")) + trail.add(s"body ${fl.get()}") + } + fl.supervisedWhere("modified") { scoped() } + trail.get shouldBe Vector("body modified", "release modified") + } + + it should "not see nested scopes' fork local values in finalizers registered with an explicit capability" in { + val fl = ForkLocal("default") + val trail = Trail() + resourceScope { + val rs = summon[ResourceScope] + fl.supervisedWhere("modified") { + releaseAfterScope(trail.add(s"release ${fl.get()}"))(using rs) + } + } + trail.get shouldBe Vector("release default") } it should "use a resource" in { @@ -162,4 +225,15 @@ class ResourceTest extends AnyFlatSpec with Matchers: trail.get shouldBe Vector("allocate", "in scope", "release", "exception e2 (e1)") } + + "a leaked scope capability" should "throw when registering after the scope ended, releasing the acquired resource" in { + val trail = Trail() + var leaked: OxUnsupervised = null + unsupervised { leaked = summon[OxUnsupervised] } + val e = the[IllegalStateException] thrownBy { + useInScope { trail.add("allocate"); 1 }(n => trail.add(s"release $n"))(using leaked) + } + e.getMessage should include("has already ended") + trail.get shouldBe Vector("allocate", "release 1") + } end ResourceTest diff --git a/doc/other/dictionary.md b/doc/other/dictionary.md index 370b110a..55003aab 100644 --- a/doc/other/dictionary.md +++ b/doc/other/dictionary.md @@ -4,7 +4,9 @@ How we use various terms throughout the codebase and the documentation (or at le Scopes: * **concurrency scope**: either `supervised` (default), `supervisedError` (permitting application errors), - or `unsupervised` + or `unsupervised`. Allow starting forks. Every concurrency scope is also a resource scope. +* **resource scope**: created using `resourceScope`, allow registering resources, which are released when the + scope completes * scope **body**: the code block passed to a concurrency scope (the `supervised`, `supervisedError` or `unsupervised` method) diff --git a/doc/utils/resources.md b/doc/utils/resources.md index 3cd6d29e..e174768b 100644 --- a/doc/utils/resources.md +++ b/doc/utils/resources.md @@ -13,10 +13,10 @@ useCloseable(new java.io.PrintWriter("test.txt")) { writer => } ``` -If a concurrency scope is available (e.g. `supervised`), or there are multiple resources to allocate, consider using the -approach described below, to avoid creating an additional syntactical scope. +If a concurrency scope is available (e.g. `supervised`), or there are multiple resources to allocate, consider using a +resource scope (described below), to avoid creating an additional syntactical scope. -Alternatively, you can use `useInterruptibly`, where the releasing might be interrupted, and which is equivalent to a +Alternatively, you can use `useInterruptible`, where the releasing might be interrupted, and which is equivalent to a `try`-`finally` block. ```{warning} @@ -24,10 +24,30 @@ To properly release resources when the entire application is interrupted, make s application's main entry point. ``` -## Within a concurrency scope +## Resource scopes -Resources can be allocated within a concurrency scope. They will be released in reverse acquisition order, after all -forks started within the scope finish (but before the scope completes). E.g.: +Resources can be attached to a **resource scope**: they are then released when the scope completes (successfully or +with an exception), in reverse registration order. Releasing is uninterruptible. Resources are registered using +`useInScope` (with acquire & release logic) or `useCloseableInScope` (for `AutoCloseable`s). A resource scope can be +obtained in two ways: + +* `resourceScope` starts a dedicated, resource-only scope - without involving concurrency. This is ox's analogue of + `scala.util.Using.Manager`: + +```scala mdoc:compile-only +import ox.{resourceScope, useCloseableInScope} +import java.io.{FileReader, FileWriter} + +def process(): Unit = resourceScope { + val in = useCloseableInScope(new FileReader("in.txt")) + val out = useCloseableInScope(new FileWriter("out.txt")) + // both closed when the scope completes, out first + out.write(in.read()) +} +``` + +* every concurrency scope (e.g. `supervised`) is also a resource scope; resources registered within one are released + after all forks started within the scope finish (but before the scope completes): ```scala mdoc:compile-only import ox.{supervised, useInScope} @@ -49,6 +69,30 @@ supervised { } ``` +Methods which only register resources can declare exactly the capability they need - `using ResourceScope` - without +claiming the ability to fork. + +A resource scope can only be started where no concurrency scope is visible - this is verified at compile-time. The +reason: forks started within a lexically visible resource scope could outlive it, using resources after they have +been released (a concurrency scope doesn't have this problem, as it waits for all forks before releasing). Within a +concurrency scope, register resources directly instead. To use a resource scope e.g. in the body of a fork, extract +it to a method which doesn't take a `using Ox` parameter - a good practice +[in itself](../other/best-practices.md#use-using-ox-sparingly): + +```scala mdoc:compile-only +import ox.{forkDiscard, resourceScope, supervised, useCloseableInScope} +import java.io.FileReader + +def handleRequest(): Unit = resourceScope { + val in = useCloseableInScope(new FileReader("in.txt")) + println(s"Processing the request using: ${in.read()}") +} + +supervised { + forkDiscard(handleRequest()) +} +``` + ### Release-only You can also register resources to be released (without acquisition logic), before the scope completes: