Introduce ThreadHerd instead of StructuredConcurrencyScope, broaden Java compatibility (#319)

Moreover, replaces the `ForkLocal` implementation based on
`ScopedValue`s with a custom one, where the locals are bound to the
concurrency scope. Ox now no longer uses any preview Java features.

Author: adamw

.github/workflows/ci.yml

@@ -6,15 +6,39 @@ on:
     branches: [ master ]
     tags: [ v* ]
 jobs:
-  build:
-    uses: softwaremill/github-actions-workflows/.github/workflows/build-scala.yml@main
-    with:
-      java-version: '21'
-      compile-documentation: true
+  ci:
+    runs-on: ubuntu-24.04
+    strategy:
+      fail-fast: false
+      matrix:
+        java: [ "21", "24" ]
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+    - name: Set up JDK
+      uses: actions/setup-java@v4
+      with:
+        distribution: 'temurin'
+        java-version: ${{ matrix.java }}
+        cache: 'sbt'
+    - uses: sbt/setup-sbt@v1
+    - name: Check formatting
+      run: sbt -v scalafmtCheckAll
+    - name: Compile
+      run: sbt -v compile
+    - name: Compile documentation
+      run: sbt -v compileDocumentation
+    - name: Test
+      run: sbt -v test
+    - uses: actions/upload-artifact@v4  # upload test results
+      if: success() || failure()        # run this step even if previous step failed
+      with:
+        name: 'tests-results-java-${{ matrix.java }}'
+        path: '**/test-reports/TEST*.xml'
 
   publish:
     uses: softwaremill/github-actions-workflows/.github/workflows/publish-release.yml@main
-    needs: [build]
+    needs: [ci]
     if: github.event_name != 'pull_request' && (startsWith(github.ref, 'refs/tags/v'))
     secrets: inherit
     with:
@@ -28,5 +52,5 @@ jobs:
   auto-merge:
     # only for PRs by softwaremill-ci
     if: github.event.pull_request.user.login == 'softwaremill-ci'
-    needs: [ build, label ]
+    needs: [ ci, label ]
     uses: softwaremill/github-actions-workflows/.github/workflows/auto-merge.yml@main

.jvmopts

@@ -4,7 +4,7 @@
 [![CI](https://github.com/softwaremill/ox/workflows/CI/badge.svg)](https://github.com/softwaremill/ox/actions?query=workflow%3A%22CI%22)
 [![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.softwaremill.ox/core_3/badge.svg)](https://maven-badges.herokuapp.com/maven-central/com.softwaremill.ox/core_3)
 
-Safe direct-style concurrency and resiliency for Scala on the JVM. Requires JDK 21 & Scala 3. The areas that we'd like 
+Safe direct-style concurrency and resiliency for Scala on the JVM. Requires JDK 21+ & Scala 3. The areas that we'd like 
 to cover with Ox are:
 
 * concurrency: developer-friendly structured concurrency, high-level concurrency operators, safe low-level primitives, 

README.md

@@ -2,11 +2,11 @@ package ox
 
 import ox.channels.Actor
 import ox.channels.ActorRef
+import ox.channels.BufferCapacity
+import ox.internal.ThreadHerd
 
-import java.util.concurrent.StructuredTaskScope
 import java.util.concurrent.atomic.AtomicReference
 import scala.annotation.implicitNotFound
-import ox.channels.BufferCapacity
 
 /** Capability granted by an [[unsupervised]] concurrency scope (as well as, via subtyping, by [[supervised]] and [[supervisedError]]).
   *
@@ -22,10 +22,12 @@ import ox.channels.BufferCapacity
   "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:
-  private[ox] def scope: StructuredTaskScope[Any]
+  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.
@@ -68,16 +70,18 @@ private[ox] trait RunInScope:
   "This operation must be run within a `supervisedError` block. Alternatively, you must require that the enclosing method is run within a scope, by adding a `using OxError[E, F]` parameter list, using the desired error mode type parameters."
 )
 case class OxError[E, F[_]](
-    private[ox] val scope: StructuredTaskScope[Any],
+    private[ox] val herd: ThreadHerd,
     private[ox] val finalizers: AtomicReference[List[() => Unit]],
     private[ox] val supervisor: Supervisor[E],
-    private[ox] val errorMode: ErrorMode[E, F]
+    private[ox] val errorMode: ErrorMode[E, F],
+    private[ox] val parent: Option[OxUnsupervised],
+    private[ox] val locals: ForkLocalMap
 ) extends Ox:
   override private[ox] def asNoErrorMode: OxError[Nothing, [T] =>> T] =
     if errorMode == NoErrorMode then this.asInstanceOf[OxError[Nothing, [T] =>> T]]
-    else OxError(scope, finalizers, supervisor, NoErrorMode)
+    else OxError(herd, finalizers, supervisor, NoErrorMode, parent, locals)
 end OxError
 
 object OxError:
-  def apply[E, F[_]](s: Supervisor[E], em: ErrorMode[E, F]): OxError[E, F] =
-    OxError(DoNothingScope[Any](oxThreadFactory), new AtomicReference(Nil), s, em)
+  def apply[E, F[_]](s: Supervisor[E], em: ErrorMode[E, F], parent: Option[OxUnsupervised], locals: ForkLocalMap): OxError[E, F] =
+    OxError(ThreadHerd(oxThreadFactory), new AtomicReference(Nil), s, em, parent, locals)

core/src/main/scala/ox/Ox.scala

@@ -1,7 +1,10 @@
 package ox
 
+import ox.internal.currentScope
+
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.Semaphore
 import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.{CompletableFuture, Semaphore}
 import scala.concurrent.ExecutionException
 import scala.util.control.NonFatal
 
@@ -34,15 +37,17 @@ def forkError[E, F[_], T](using OxError[E, F])(f: => F[T]): Fork[T] =
   val oxError = summon[OxError[E, F]]
   // the separate result future is needed to wait for the result, as there's no .join on individual tasks (only whole scopes can be joined)
   val result = new CompletableFuture[T]()
-  oxError.scope.fork { () =>
+
+  oxError.herd.startThread {
+    currentScope.set(oxError) // propagating the current scope
     val supervisor = oxError.supervisor
     try
       val resultOrError = f
       val errorMode = oxError.errorMode
       if errorMode.isError(resultOrError) then
         // result is never completed, the supervisor should end the scope
         supervisor.forkAppError(errorMode.getError(resultOrError))
-      else result.complete(errorMode.getT(resultOrError))
+      else result.complete(errorMode.getT(resultOrError)).discard
     catch
       case e: Throwable =>
         // we notify the supervisor first, so that if this is the first failing fork in the scope, the supervisor will
@@ -52,6 +57,7 @@ def forkError[E, F[_], T](using OxError[E, F])(f: => F[T]): Fork[T] =
         if !supervisor.forkException(e) then result.completeExceptionally(e).discard
     end try
   }
+
   new ForkUsingResult(result) {}
 end forkError
 
@@ -81,7 +87,9 @@ def forkUserError[E, F[_], T](using OxError[E, F])(f: => F[T]): Fork[T] =
   val oxError = summon[OxError[E, F]]
   val result = new CompletableFuture[T]()
   oxError.supervisor.forkStarts()
-  oxError.scope.fork { () =>
+
+  oxError.herd.startThread:
+    currentScope.set(oxError) // propagating the current scope
     val supervisor = oxError.supervisor.asInstanceOf[DefaultSupervisor[E]]
     try
       val resultOrError = f
@@ -96,7 +104,7 @@ def forkUserError[E, F[_], T](using OxError[E, F])(f: => F[T]): Fork[T] =
       case e: Throwable =>
         if !supervisor.forkException(e) then result.completeExceptionally(e).discard
     end try
-  }
+
   new ForkUsingResult(result) {}
 end forkUserError
 
@@ -111,12 +119,16 @@ end forkUserError
   * For alternate behaviors, see [[fork]], [[forkUser]] and [[forkCancellable]].
   */
 def forkUnsupervised[T](f: => T)(using OxUnsupervised): UnsupervisedFork[T] =
+  val oxUnsupervised = summon[OxUnsupervised]
   val result = new CompletableFuture[T]()
-  summon[OxUnsupervised].scope.fork { () =>
-    try result.complete(f)
-    catch case e: Throwable => result.completeExceptionally(e)
-  }
+
+  oxUnsupervised.herd.startThread:
+    currentScope.set(oxUnsupervised) // propagating the current scope
+    try result.complete(f).discard
+    catch case e: Throwable => result.completeExceptionally(e).discard
+
   new ForkUsingResult(result) with UnsupervisedFork[T] {}
+end forkUnsupervised
 
 /** For each thunk in the given sequence, starts a fork using [[fork]]. All forks are guaranteed to complete before the enclosing
   * [[supervised]] or [[unsupervised]] block completes.
@@ -149,21 +161,28 @@ def forkCancellable[T](f: => T)(using OxUnsupervised): CancellableFork[T] =
   // interrupt signal
   val done = new Semaphore(0)
   val ox = summon[OxUnsupervised]
-  ox.scope.fork { () =>
-    val nestedOx = OxError(NoOpSupervisor, NoErrorMode)
-    scopedWithCapability(nestedOx) {
-      nestedOx.scope.fork { () =>
-        // "else" means that the fork is already cancelled, so doing nothing in that case
-        if !started.getAndSet(true) then
-          try result.complete(f).discard
-          catch case e: Throwable => result.completeExceptionally(e).discard
-
-        done.release() // the nested scope can now finish
+
+  ox.herd.startThread:
+    try
+      val nestedOx = OxError(NoOpSupervisor, NoErrorMode, Some(ox), ox.locals)
+      scopedWithCapability(nestedOx) {
+        nestedOx.herd.startThread {
+          currentScope.set(nestedOx) // propagating the nested scope, which has the current scope set as a parent
+          // "else" means that the fork is already cancelled, so doing nothing in that case
+          if !started.getAndSet(true) then
+            try result.complete(f).discard
+            catch case e: Throwable => result.completeExceptionally(e).discard
+
+          done.release() // the nested scope can now finish
+        }.discard
+
+        done.acquire()
       }
+    catch
+      // if this thread was interrupted, any context is already captured as part of the thread started in the nested scope
+      // hence, ignoring this exception, so that it's not logged by the uncaught exception handler
+      case e: InterruptedException =>
 
-      done.acquire()
-    }
-  }
   new ForkUsingResult(result) with CancellableFork[T]:
     override def cancel(): Either[Throwable, T] =
       cancelNow()

core/src/main/scala/ox/fork.scala

@@ -0,0 +1,13 @@
+package ox.internal
+
+import ox.OxUnsupervised
+import ox.ForkLocalMap
+
+/** Should only ever be updated when starting a new scope, for the duration of the scope's lifetime. Used to verify that forks are properly
+  * started, within a running concurrency scope, on a thread that is part of some scope in the tree.
+  */
+private[ox] val currentScope: ThreadLocal[OxUnsupervised] = new ThreadLocal[OxUnsupervised]()
+
+private[ox] def currentLocals: ForkLocalMap =
+  val scope = currentScope.get()
+  if scope == null then ForkLocalMap(Map.empty) else scope.locals

core/src/main/scala/ox/internal/ScopeContext.scala

@@ -0,0 +1,99 @@
+package ox.internal
+
+import ox.OxUnsupervised
+import ox.discard
+
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ThreadFactory
+import java.util.concurrent.atomic.AtomicBoolean
+import scala.annotation.tailrec
+
+/** Until the structured concurrency JEP is stable in an LTS release, a replacement for `StructuredTaskScope`. Downside: does not integrate
+  * with scoped values (instead, `ForkLocal` is based on a custom implementation as well). Upside: works with any Java 21+.
+  *
+  * Naming: `ThreadFlock` is part of the Structured Concurrency JEP; `ThreadGroup` is already part of the JDK. Hence, using "herd."
+  */
+private[ox] class ThreadHerd(threadFactory: ThreadFactory):
+  // capturing the owner at construction time to later check if forks are started properly
+  private val herdOwner = Thread.currentThread()
+
+  private val shutdownInProgress = new AtomicBoolean(false)
+  private val threads = ConcurrentHashMap.newKeySet[Thread]()
+
+  def startThread(t: => Unit): Unit =
+    assertNotShuttingDown()
+
+    verifyCurrentThreadInScopeTree()
+
+    val thread = threadFactory.newThread(() =>
+      try t
+      finally threads.remove(Thread.currentThread()).discard
+    )
+    threads.add(thread)
+
+    thread.start()
+  end startThread
+
+  /** Interrupts all thread in this heard, and waits for them to complete. If this process is interrupted, the exception is captured and
+    * rethrown only after all threads from the herd have completed.
+    */
+  def interruptAllAndJoinUntilCompleted(): Unit =
+    assertOnOwnerThread()
+
+    shutdownInProgress.set(true)
+
+    /*
+    If a startThread() is in progress, after the no-shutdown-assertion, the thread that it creates will be discovered
+    in the loop below, as the thread that is starting the new thread itself has to be joined; and this can only happen
+    after the new thread is added to the `threads` set.
+     */
+
+    var interruptedThreads = Set.empty[Thread] // making sure we interrupt each thread only once
+    def interruptIfNeeded(t: Thread): Unit = if !interruptedThreads.contains(t) then
+      t.interrupt()
+      interruptedThreads += t
+
+    var interruptedException: InterruptedException = null
+
+    while !threads.isEmpty do
+      threads.forEach(interruptIfNeeded)
+      threads.forEach: thread =>
+        interruptIfNeeded(thread) // double-checking, as this thread might have been added after the previous forEach
+        try thread.join()
+        catch
+          case e: InterruptedException => // if the thread is not yet done, we'll join it again in the next iteration
+            if interruptedException == null then interruptedException = e
+            else interruptedException.addSuppressed(e)
+    end while
+
+    if interruptedException != null then throw interruptedException
+  end interruptAllAndJoinUntilCompleted
+
+  def isOwnerOrHerdThread(thread: Thread): Boolean = thread == herdOwner || threads.contains(thread)
+
+  private def assertNotShuttingDown(): Unit =
+    if shutdownInProgress.get() then
+      throw new IllegalStateException("Scope is shutting down, cannot start new threads or join existing ones.")
+
+  /** Naming: verify, not assert, as this might be a user error, not a bug in Ox (which is what the other asserts assert). */
+  private def verifyCurrentThreadInScopeTree(): Unit =
+    val currentThread = Thread.currentThread()
+
+    @tailrec
+    def doAssert(scope: OxUnsupervised): Unit =
+      if scope.herd.isOwnerOrHerdThread(currentThread) then return
+      scope.parent match
+        case None              => throw new IllegalStateException("Fork cannot be started outside the tree of concurrency scopes.")
+        case Some(parentScope) => doAssert(parentScope)
+
+    currentScope.get() match
+      case null =>
+        throw new IllegalStateException("Forks can only be started inside a concurrency scope, or from other forks started in the scope.")
+      case scope => doAssert(scope)
+  end verifyCurrentThreadInScopeTree
+
+  private def assertOnOwnerThread(): Unit =
+    val current = Thread.currentThread()
+    if current != herdOwner then
+      throw new IllegalStateException("This operation can only be performed from the thread that created the scope.")
+end ThreadHerd