[Experiment] Capture Check scala.concurrent with except

library-js/src/scala/concurrent/ExecutionContext.scala

@@ -17,6 +17,12 @@ import scala.language.`2.13`
 import java.util.concurrent.{ ExecutorService, Executor }
 import scala.annotation.implicitNotFound
 
+import language.experimental.captureChecking
+import scala.caps.*
+
+/** Marker classifier for all default execution contexts, which are not thread-local. */
+trait ConcurrentCapability extends SharedCapability, Classifier
+
 /** An `ExecutionContext` can execute program logic asynchronously,
  *  typically but not necessarily on a thread pool.
  *
@@ -101,7 +107,7 @@ trait ExecutionContext {
    */
   @deprecated("preparation of ExecutionContexts will be removed", "2.12.0")
   // This cannot be removed until there is a suitable replacement
-  def prepare(): ExecutionContext = this
+  def prepare(): ExecutionContext^{this} = this
 }
 
 /** An [[ExecutionContext]] that is also a
@@ -116,7 +122,7 @@ trait ExecutionContextExecutorService extends ExecutionContextExecutor with Exec
 
 
 /** Contains factory methods for creating execution contexts. */
-object ExecutionContext {
+object ExecutionContext extends ConcurrentCapability {
   /** The explicit global `ExecutionContext`. Invoke `global` when you want to provide the global
    *  `ExecutionContext` explicitly.
    *
@@ -135,7 +141,7 @@ object ExecutionContext {
    *
    *  @return the global `ExecutionContext`
    */
-  final lazy val global: ExecutionContextExecutor =
+  final lazy val global: ExecutionContextExecutor^{ExecutionContext} =
     scala.scalajs.concurrent.JSExecutionContext.queue
 
   /** WARNING: Only ever execute logic which will quickly return control to the caller.
@@ -156,13 +162,13 @@ object ExecutionContext {
    *
    *  Any `NonFatal` or `InterruptedException`s will be reported to the `defaultReporter`.
    */
-  object parasitic extends ExecutionContextExecutor with BatchingExecutor {
-    override final def submitForExecution(runnable: Runnable): Unit = runnable.run()
+  object parasitic extends ExecutionContextExecutor with BatchingExecutor with ConcurrentCapability {
+    override final def submitForExecution(runnable: Runnable^{ExecutionContext.parasitic}): Unit = runnable.run()
     override final def execute(runnable: Runnable): Unit = submitSyncBatched(runnable)
     override final def reportFailure(t: Throwable): Unit = defaultReporter(t)
   }
 
-  object Implicits {
+  object Implicits uses ExecutionContext {
     /** The implicit global `ExecutionContext`. Import `global` when you want to provide the global
      *  `ExecutionContext` implicitly.
      *
@@ -172,7 +178,7 @@ object ExecutionContext {
      *
      *  @return the global `ExecutionContext`
      */
-    implicit final def global: ExecutionContext = ExecutionContext.global
+    implicit final def global: ExecutionContext^{ExecutionContext} = ExecutionContext.global
   }
 
   /** Creates an `ExecutionContext` from the given `ExecutorService`.
@@ -181,7 +187,7 @@ object ExecutionContext {
    *  @param reporter  a function for error reporting
    *  @return          the `ExecutionContext` using the given `ExecutorService`
    */
-  def fromExecutorService(e: ExecutorService, reporter: Throwable => Unit): ExecutionContextExecutorService =
+  def fromExecutorService(e: ExecutorService^{any.except[ThreadLocal]}, reporter: Throwable ->{any.except[ThreadLocal]} Unit): ExecutionContextExecutorService^{e, reporter} =
     impl.ExecutionContextImpl.fromExecutorService(e, reporter)
 
   /** Creates an `ExecutionContext` from the given `ExecutorService` with the [[scala.concurrent.ExecutionContext$.defaultReporter default reporter]].
@@ -197,27 +203,27 @@ object ExecutionContext {
    *  @param e the `ExecutorService` to use. If `null`, a new `ExecutorService` is created with [[scala.concurrent.ExecutionContext$.global default configuration]].
    *  @return  the `ExecutionContext` using the given `ExecutorService`
    */
-  def fromExecutorService(e: ExecutorService): ExecutionContextExecutorService = fromExecutorService(e, defaultReporter)
+  def fromExecutorService(e: ExecutorService^{any.except[ThreadLocal]}): ExecutionContextExecutorService^{e} = fromExecutorService(e, defaultReporter)
 
   /** Creates an `ExecutionContext` from the given `Executor`.
    *
    *  @param e         the `Executor` to use. If `null`, a new `Executor` is created with [[scala.concurrent.ExecutionContext$.global default configuration]].
    *  @param reporter  a function for error reporting
    *  @return          the `ExecutionContext` using the given `Executor`
    */
-  def fromExecutor(e: Executor, reporter: Throwable => Unit): ExecutionContextExecutor =
+  def fromExecutor(e: Executor^{any.except[ThreadLocal]}, reporter: Throwable ->{any.except[ThreadLocal]} Unit): ExecutionContextExecutor^{e, reporter} =
     impl.ExecutionContextImpl.fromExecutor(e, reporter)
 
   /** Creates an `ExecutionContext` from the given `Executor` with the [[scala.concurrent.ExecutionContext$.defaultReporter default reporter]].
    *
    *  @param e the `Executor` to use. If `null`, a new `Executor` is created with [[scala.concurrent.ExecutionContext$.global default configuration]].
    *  @return  the `ExecutionContext` using the given `Executor`
    */
-  def fromExecutor(e: Executor): ExecutionContextExecutor = fromExecutor(e, defaultReporter)
+  def fromExecutor(e: Executor^{any.except[ThreadLocal]}): ExecutionContextExecutor^{e} = fromExecutor(e, defaultReporter)
 
   /** The default reporter simply prints the stack trace of the `Throwable` to [System.err](http://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/lang/System.html#err).
    *
    *  @return the function for error reporting
    */
-  final val defaultReporter: Throwable => Unit = _.printStackTrace()
+  final val defaultReporter: Throwable -> Unit = _.printStackTrace()
 }

library/src/scala/caps/package.scala

@@ -72,6 +72,15 @@ trait ExclusiveCapability extends Capability
 @experimental
 type Exclusive = ExclusiveCapability
 
+/** Marker trait for capabilities that are thread-local. These include control-flow capabilities
+ * (as they modify stacks), as well as mutex locks and other values that cannot be safety passed across stacks.
+ *
+ *  [[ThreadLocal]] has a formal meaning when
+ *  [[scala.language.experimental.captureChecking Capture Checking]]
+ *  is turned on.
+ */
+trait ThreadLocal extends SharedCapability, Classifier
+
 /** Marker trait for capabilities that capture some continuation or return point in
  *  the stack. Examples are exceptions, [[scala.util.boundary.Label labels]], [[scala.CanThrow CanThrow]]
  *  or Async contexts.
@@ -80,7 +89,7 @@ type Exclusive = ExclusiveCapability
  *  [[scala.language.experimental.captureChecking Capture Checking]]
  *  is turned on.
  */
-trait Control extends SharedCapability, Classifier
+trait Control extends ThreadLocal, Classifier
 
 /** Marker trait for classes that can consult and change the global program state.
  *  These  classes typically contain mutable variables and/or update methods.

library/src/scala/concurrent/Awaitable.scala

@@ -12,8 +12,8 @@
 
 package scala.concurrent
 
-
 import language.experimental.captureChecking
+
 import scala.language.`2.13`
 import scala.concurrent.duration.Duration
 

library/src/scala/concurrent/BatchingExecutor.scala

@@ -12,6 +12,8 @@
 
 package scala.concurrent
 
+import language.experimental.captureChecking
+
 import scala.language.`2.13`
 import java.util.concurrent.Executor
 import java.util.Objects
@@ -20,7 +22,7 @@ import scala.annotation.{switch, tailrec}
 
 /** Marker trait to indicate that a Runnable is Batchable by BatchingExecutors */
 trait Batchable {
-  self: Runnable =>
+  self: Runnable^ =>
 }
 
 private[concurrent] object BatchingExecutorStatics {
@@ -106,7 +108,7 @@ private[concurrent] trait BatchingExecutor extends Executor {
     @annotation.stableNull protected final var first: Runnable | Null,
     protected final var other: Array[Runnable | Null],
     protected final var size: Int
-  ) {
+  ) { this: AbstractBatch^{BatchingExecutor.this} /* Until separation checking, do not try to capture anything more than the executor */ =>
 
     private final def ensureCapacity(curSize: Int): Array[Runnable | Null] = {
       val curOther = this.other
@@ -152,22 +154,22 @@ private[concurrent] trait BatchingExecutor extends Executor {
           }
   }
 
-  private final class AsyncBatch private(_first: Runnable | Null, _other: Array[Runnable | Null], _size: Int) extends AbstractBatch(_first, _other, _size) with Runnable with BlockContext with (BlockContext => Throwable | Null) {
-    private final var parentBlockContext: BlockContext = BatchingExecutorStatics.MissingParentBlockContext
+  private final class AsyncBatch private(_first: Runnable | Null, _other: Array[Runnable | Null], _size: Int) extends AbstractBatch(_first, _other, _size) with Runnable with BlockContext uses BatchingExecutor.this {
+    private final var parentBlockContext: BlockContext^{this} = BatchingExecutorStatics.MissingParentBlockContext
 
     final def this(runnable: Runnable) = this(runnable, BatchingExecutorStatics.emptyBatchArray, 1)
 
     override final def run(): Unit = {
       _tasksLocal.set(this) // This is later cleared in `apply` or `runWithoutResubmit`
 
-      val f = resubmit(BlockContext.usingBlockContext(this)(this))
+      val f = resubmit(BlockContext.usingBlockContext(this)(b => this(b)))
 
       if (f != null)
         throw f
     }
 
     /* LOGIC FOR ASYNCHRONOUS BATCHES */
-    override final def apply(prevBlockContext: BlockContext): Throwable | Null = try {
+    final def apply(prevBlockContext: BlockContext^{this}): Throwable | Null = try {
       parentBlockContext = prevBlockContext
       runN(BatchingExecutorStatics.runLimit)
       null
@@ -195,7 +197,7 @@ private[concurrent] trait BatchingExecutor extends Executor {
         }
       } else cause // TODO: consider if NonFatals should simply be `reportFailure`:ed rather than rethrown
 
-    private final def cloneAndClear(): AsyncBatch = {
+    private final def cloneAndClear(): AsyncBatch^{this} = {
       val newBatch = new AsyncBatch(this.first, this.other, this.size)
       this.first = null
       this.other = BatchingExecutorStatics.emptyBatchArray
@@ -232,7 +234,8 @@ private[concurrent] trait BatchingExecutor extends Executor {
    *
    *  @param runnable the `Runnable` to submit for execution; must not be null
    */
-  protected def submitForExecution(runnable: Runnable): Unit
+  // CC notes: the runnable is allowed to capture the executor, for notifying itself.
+  protected def submitForExecution(runnable: Runnable^{this}): Unit
 
   /** Reports that an asynchronous computation failed.
    *  See `ExecutionContext.reportFailure(throwable: Throwable)`

library/src/scala/concurrent/BlockContext.scala

@@ -12,6 +12,8 @@
 
 package scala.concurrent
 
+import language.experimental.captureChecking
+
 import scala.language.`2.13`
 
 /** A context to be notified by [[scala.concurrent.blocking]] when
@@ -74,7 +76,7 @@ object BlockContext {
 
   private final val contextLocal = new ThreadLocal[BlockContext]()
 
-  private final def prefer(candidate: BlockContext): BlockContext =
+  private final def prefer(candidate: BlockContext^): BlockContext^{candidate} =
     if (candidate ne null) candidate
     else {
       val t = Thread.currentThread
@@ -94,7 +96,7 @@ object BlockContext {
    *  @param body the code to execute with the given `blockContext` installed
    *  @return the result of executing `body`
    */
-  final def withBlockContext[T](blockContext: BlockContext)(body: => T): T = {
+  final def withBlockContext[T](blockContext: BlockContext^)(body: => T): T = {
     val old = contextLocal.get // can be null
     if (old eq blockContext) body
     else {
@@ -111,7 +113,7 @@ object BlockContext {
    *  @param f the function to execute, receiving the previously installed `BlockContext` as its argument
    *  @return the value produced by applying `f`
    */
-  final def usingBlockContext[I, T](blockContext: BlockContext)(f: BlockContext => T): T = {
+  final def usingBlockContext[I, T](blockContext: BlockContext^)(f: BlockContext^{blockContext} => T): T = {
     val old = contextLocal.get // can be null
     if (old eq blockContext) f(prefer(old))
     else {

library/src/scala/concurrent/ExecutionContext.scala

@@ -16,6 +16,12 @@ import scala.language.`2.13`
 import java.util.concurrent.{ ExecutorService, Executor }
 import scala.annotation.implicitNotFound
 
+import language.experimental.captureChecking
+import scala.caps.*
+
+/** Marker classifier for all default execution contexts, which are not thread-local. */
+trait ConcurrentCapability extends SharedCapability, Classifier
+
 /** An `ExecutionContext` can execute program logic asynchronously,
  *  typically but not necessarily on a thread pool.
  *
@@ -74,6 +80,7 @@ trait ExecutionContext {
    *
    *  @param runnable  the task to execute
    */
+  // TODO: The runnable must be pure at the moment, we can allow it to be impure with separation checking.
   def execute(runnable: Runnable): Unit
 
   /** Reports that an asynchronous computation failed.
@@ -100,7 +107,7 @@ trait ExecutionContext {
    */
   @deprecated("preparation of ExecutionContexts will be removed", "2.12.0")
   // This cannot be removed until there is a suitable replacement
-  def prepare(): ExecutionContext = this
+  def prepare(): ExecutionContext^{this} = this
 }
 
 /** An [[ExecutionContext]] that is also a
@@ -115,7 +122,7 @@ trait ExecutionContextExecutorService extends ExecutionContextExecutor with Exec
 
 
 /** Contains factory methods for creating execution contexts. */
-object ExecutionContext {
+object ExecutionContext extends ConcurrentCapability {
   /** The global [[ExecutionContext]]. This default `ExecutionContext` implementation is backed by a work-stealing thread
    *  pool. It can be configured via the following system properties:
    *
@@ -196,7 +203,7 @@ object ExecutionContext {
    *
    *  @return the global [[ExecutionContext]]
    */
-  final lazy val global: ExecutionContextExecutor = impl.ExecutionContextImpl.fromExecutor(null: Executor | Null)
+  final lazy val global: ExecutionContextExecutor^{this} = impl.ExecutionContextImpl.fromExecutor(null: Executor | Null)
 
   /** WARNING: Only ever execute logic which will quickly return control to the caller.
    *
@@ -216,15 +223,15 @@ object ExecutionContext {
    *
    *  Any `NonFatal` or `InterruptedException`s will be reported to the `defaultReporter`.
    */
-  object parasitic extends ExecutionContextExecutor with BatchingExecutor {
-    override final def submitForExecution(runnable: Runnable): Unit = runnable.run()
+  object parasitic extends ExecutionContextExecutor with BatchingExecutor with ConcurrentCapability {
+    override final def submitForExecution(runnable: Runnable^{this}): Unit = runnable.run()
     override final def execute(runnable: Runnable): Unit = submitSyncBatched(runnable)
     override final def reportFailure(t: Throwable): Unit = defaultReporter(t)
   }
 
   /** See [[ExecutionContext.global]]. */
-  private[scala] lazy val opportunistic: ExecutionContextExecutor = new ExecutionContextExecutor with BatchingExecutor {
-    final override def submitForExecution(runnable: Runnable): Unit = global.execute(runnable)
+  private[scala] lazy val opportunistic: ExecutionContextExecutor^{this} = new ExecutionContextExecutor with BatchingExecutor with ConcurrentCapability {
+    final override def submitForExecution(runnable: Runnable^{this}): Unit = global.execute(runnable)
 
     final override def execute(runnable: Runnable): Unit =
       if ((!runnable.isInstanceOf[impl.Promise.Transformation[?, ?]] || runnable.asInstanceOf[impl.Promise.Transformation[?, ?]].benefitsFromBatching) && runnable.isInstanceOf[Batchable])
@@ -235,11 +242,11 @@ object ExecutionContext {
     override final def reportFailure(t: Throwable): Unit = global.reportFailure(t)
   }
 
-  object Implicits {
+  object Implicits uses ExecutionContext {
     /** An accessor that can be used to import the global `ExecutionContext` into the implicit scope,
      *  see [[ExecutionContext.global]].
      */
-    implicit final def global: ExecutionContext = ExecutionContext.global
+    implicit final def global: ExecutionContext^{ExecutionContext} = ExecutionContext.global
   }
 
   /** Creates an `ExecutionContext` from the given `ExecutorService`.
@@ -248,7 +255,7 @@ object ExecutionContext {
    *  @param reporter  a function for error reporting
    *  @return          the `ExecutionContext` using the given `ExecutorService`
    */
-  def fromExecutorService(e: ExecutorService | Null, reporter: Throwable => Unit): ExecutionContextExecutorService =
+  def fromExecutorService(e: (ExecutorService^{any.except[ThreadLocal]}) | Null, reporter: Throwable ->{any.except[ThreadLocal]} Unit): ExecutionContextExecutorService^{e, reporter} =
     impl.ExecutionContextImpl.fromExecutorService(e, reporter)
 
   /** Creates an `ExecutionContext` from the given `ExecutorService` with the [[scala.concurrent.ExecutionContext$.defaultReporter default reporter]].
@@ -264,27 +271,27 @@ object ExecutionContext {
    *  @param e the `ExecutorService` to use. If `null`, a new `ExecutorService` is created with [[scala.concurrent.ExecutionContext$.global default configuration]].
    *  @return  the `ExecutionContext` using the given `ExecutorService`
    */
-  def fromExecutorService(e: ExecutorService | Null): ExecutionContextExecutorService = fromExecutorService(e, defaultReporter)
+  def fromExecutorService(e: (ExecutorService^{any.except[ThreadLocal]}) | Null): ExecutionContextExecutorService^{e} = fromExecutorService(e, defaultReporter)
 
   /** Creates an `ExecutionContext` from the given `Executor`.
    *
    *  @param e         the `Executor` to use. If `null`, a new `Executor` is created with [[scala.concurrent.ExecutionContext$.global default configuration]].
    *  @param reporter  a function for error reporting
    *  @return          the `ExecutionContext` using the given `Executor`
    */
-  def fromExecutor(e: Executor | Null, reporter: Throwable => Unit): ExecutionContextExecutor =
+  def fromExecutor(e: (Executor^{any.except[ThreadLocal]}) | Null, reporter: Throwable ->{any.except[ThreadLocal]} Unit): ExecutionContextExecutor^{e, reporter} =
     impl.ExecutionContextImpl.fromExecutor(e, reporter)
 
   /** Creates an `ExecutionContext` from the given `Executor` with the [[scala.concurrent.ExecutionContext$.defaultReporter default reporter]].
    *
    *  @param e the `Executor` to use. If `null`, a new `Executor` is created with [[scala.concurrent.ExecutionContext$.global default configuration]].
    *  @return  the `ExecutionContext` using the given `Executor`
    */
-  def fromExecutor(e: Executor | Null): ExecutionContextExecutor = fromExecutor(e, defaultReporter)
+  def fromExecutor(e: (Executor^{any.except[ThreadLocal]}) | Null): ExecutionContextExecutor^{e} = fromExecutor(e, defaultReporter)
 
   /** The default reporter simply prints the stack trace of the `Throwable` to [[java.lang.System#err System.err]].
    *
    *  @return the function for error reporting
    */
-  final val defaultReporter: Throwable => Unit = _.printStackTrace()
+  final val defaultReporter: Throwable -> Unit = _.printStackTrace()
 }