diff --git a/library-js/src/scala/concurrent/ExecutionContext.scala b/library-js/src/scala/concurrent/ExecutionContext.scala index 017b302a8169..31b74ea15b34 100644 --- a/library-js/src/scala/concurrent/ExecutionContext.scala +++ b/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,7 +203,7 @@ 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`. * @@ -205,7 +211,7 @@ object ExecutionContext { * @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]]. @@ -213,11 +219,11 @@ object ExecutionContext { * @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() } diff --git a/library/src/scala/caps/package.scala b/library/src/scala/caps/package.scala index 05da4482bfa7..d8f5389b6792 100644 --- a/library/src/scala/caps/package.scala +++ b/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. diff --git a/library/src/scala/concurrent/Awaitable.scala b/library/src/scala/concurrent/Awaitable.scala index bc6bb7244cd9..ac3dd39907b8 100644 --- a/library/src/scala/concurrent/Awaitable.scala +++ b/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 diff --git a/library/src/scala/concurrent/BatchingExecutor.scala b/library/src/scala/concurrent/BatchingExecutor.scala index 1f86859a5658..5956861a65ab 100644 --- a/library/src/scala/concurrent/BatchingExecutor.scala +++ b/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)` diff --git a/library/src/scala/concurrent/BlockContext.scala b/library/src/scala/concurrent/BlockContext.scala index 1ba884408ee7..eec183e413a7 100644 --- a/library/src/scala/concurrent/BlockContext.scala +++ b/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 { diff --git a/library/src/scala/concurrent/ExecutionContext.scala b/library/src/scala/concurrent/ExecutionContext.scala index 5ad8ef448a90..2c156660d53c 100644 --- a/library/src/scala/concurrent/ExecutionContext.scala +++ b/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,7 +271,7 @@ 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`. * @@ -272,7 +279,7 @@ object ExecutionContext { * @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]]. @@ -280,11 +287,11 @@ object ExecutionContext { * @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() } diff --git a/library/src/scala/concurrent/Future.scala b/library/src/scala/concurrent/Future.scala index 9900232f0f7c..c9cd5e9eacca 100644 --- a/library/src/scala/concurrent/Future.scala +++ b/library/src/scala/concurrent/Future.scala @@ -12,6 +12,9 @@ package scala.concurrent +import language.experimental.captureChecking +import caps.* + import scala.language.`2.13` import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.LockSupport @@ -110,7 +113,7 @@ import scala.concurrent.impl.Promise.DefaultPromise * * @tparam T the type of the value contained in this `Future` */ -trait Future[+T] extends Awaitable[T] { +trait Future[+T] extends Awaitable[T] uses ExecutionContext { this: Future[T]^{any.except[ThreadLocal]} => /* Callbacks */ @@ -131,7 +134,7 @@ trait Future[+T] extends Awaitable[T] { * @param executor the `ExecutionContext` on which the callback will be executed * @group Callbacks */ - def onComplete[U](f: Try[T] => U)(implicit executor: ExecutionContext): Unit + def onComplete[U](f: Try[T] ->{any.except[ThreadLocal]} U)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Unit /* Miscellaneous */ @@ -172,7 +175,7 @@ trait Future[+T] extends Awaitable[T] { * @return a failed projection of this `Future`. * @group Transformations */ - def failed: Future[Throwable] = transform(Future.failedFun)(using parasitic) + def failed: Future[Throwable]^{this, ExecutionContext} = transform(Future.failedFun)(using parasitic) /* Monadic operations */ @@ -189,7 +192,7 @@ trait Future[+T] extends Awaitable[T] { * @param executor the `ExecutionContext` on which the callback will be executed * @group Callbacks */ - def foreach[U](f: T => U)(implicit executor: ExecutionContext): Unit = onComplete { _ foreach f } + def foreach[U](f: T ->{any.except[ThreadLocal]} U)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Unit = onComplete { _ foreach f } /** Creates a new future by applying the 's' function to the successful result of * this future, or the 'f' function to the failed result. If there is any non-fatal @@ -203,7 +206,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` that will be completed with the transformed value * @group Transformations */ - def transform[S](s: T => S, f: Throwable => Throwable)(implicit executor: ExecutionContext): Future[S] = + def transform[S](s: T ->{any.except[ThreadLocal]} S, f: Throwable ->{any.except[ThreadLocal]} Throwable)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, s, f, executor, ExecutionContext} = transform { t => if (t.isInstanceOf[Success[T]]) t map s @@ -220,7 +223,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` that will be completed with the transformed value * @group Transformations */ - def transform[S](f: Try[T] => Try[S])(implicit executor: ExecutionContext): Future[S] + def transform[S](f: Try[T] ->{any.except[ThreadLocal]} Try[S])(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, f, executor, ExecutionContext} /** Creates a new Future by applying the specified function, which produces a Future, to the result * of this Future. If there is any non-fatal exception thrown when 'f' @@ -232,7 +235,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` that will be completed with the transformed value * @group Transformations */ - def transformWith[S](f: Try[T] => Future[S])(implicit executor: ExecutionContext): Future[S] + def transformWith[S](f: Try[T] ->{any.except[ThreadLocal]} Future[S]^)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, f, executor, ExecutionContext} /** Creates a new future by applying a function to the successful result of @@ -258,7 +261,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` which will be completed with the result of the application of the function * @group Transformations */ - def map[S](f: T => S)(implicit executor: ExecutionContext): Future[S] = transform(_ map f) + def map[S](f: T ->{any.except[ThreadLocal]} S)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, f, executor, ExecutionContext} = transform(_ map f) /** Creates a new future by applying a function to the successful result of * this future, and returns the result of the function as the new future. @@ -273,7 +276,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` which will be completed with the result of the application of the function * @group Transformations */ - def flatMap[S](f: T => Future[S])(implicit executor: ExecutionContext): Future[S] = transformWith { + def flatMap[S](f: T ->{any.except[ThreadLocal]} Future[S]^)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, f, executor, ExecutionContext} = transformWith { t => if(t.isInstanceOf[Success[T]]) f(t.asInstanceOf[Success[T]].value) else this.asInstanceOf[Future[S]] // Safe cast @@ -287,7 +290,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` with the result of the inner `Future` * @group Transformations */ - def flatten[S](implicit ev: T <:< Future[S]): Future[S] = flatMap(ev)(using parasitic) + def flatten[S](implicit ev: T <:< Future[S]): Future[S]^{this, ExecutionContext} = flatMap(ev)(using parasitic) /** Creates a new future by filtering the value of the current future with a predicate. * @@ -312,7 +315,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` which will hold the successful result of this `Future` if it matches the predicate or a `NoSuchElementException` * @group Transformations */ - def filter(p: T => Boolean)(implicit executor: ExecutionContext): Future[T] = + def filter(p: T ->{any.except[ThreadLocal]} Boolean)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[T]^{this, p, executor, ExecutionContext} = transform { t => if (t.isInstanceOf[Success[T]]) { @@ -328,7 +331,7 @@ trait Future[+T] extends Awaitable[T] { * @param executor the `ExecutionContext` on which the predicate will be executed * @return a `Future` which will hold the successful result of this `Future` if it matches the predicate, else a failure holding `NoSuchElementException` */ - final def withFilter(p: T => Boolean)(implicit executor: ExecutionContext): Future[T] = filter(p)(using executor) + final def withFilter(p: T ->{any.except[ThreadLocal]} Boolean)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[T]^{this, p, executor, ExecutionContext} = filter(p)(using executor) /** Creates a new future by mapping the value of the current future, if the given partial function is defined at that value. * @@ -358,7 +361,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` holding the result of application of the `PartialFunction` or a `NoSuchElementException` * @group Transformations */ - def collect[S](pf: PartialFunction[T, S])(implicit executor: ExecutionContext): Future[S] = + def collect[S](pf: PartialFunction[T, S]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, pf, executor, ExecutionContext} = transform { t => if (t.isInstanceOf[Success[T]]) @@ -385,7 +388,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` with the successful value of this `Future` or the result of the `PartialFunction` * @group Transformations */ - def recover[U >: T](pf: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Future[U] = + def recover[U >: T](pf: PartialFunction[Throwable, U]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[U]^{this, pf, executor, ExecutionContext} = transform { _ recover pf } /** Creates a new future that will handle any matching throwable that this @@ -408,7 +411,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` with the successful value of this `Future` or the outcome of the `Future` returned by the `PartialFunction` * @group Transformations */ - def recoverWith[U >: T](pf: PartialFunction[Throwable, Future[U]])(implicit executor: ExecutionContext): Future[U] = + def recoverWith[U >: T](pf: PartialFunction[Throwable, Future[U]]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[U]^{this, pf, executor, ExecutionContext} = transformWith { t => if (t.isInstanceOf[Failure[T]]) { @@ -432,7 +435,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` with the results of both futures or the failure of the first of them that failed * @group Transformations */ - def zip[U](that: Future[U]): Future[(T, U)] = + def zip[U](that: Future[U]^{any.except[ThreadLocal]}): Future[(T, U)]^{this, that, ExecutionContext} = zipWith(that)(Future.zipWithTuple2Fun)(using parasitic) /** Zips the values of `this` and `that` future using a function `f`, @@ -452,7 +455,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` with the result of the application of `f` to the results of `this` and `that` * @group Transformations */ - def zipWith[U, R](that: Future[U])(f: (T, U) => R)(implicit executor: ExecutionContext): Future[R] = { + def zipWith[U, R](that: Future[U]^{any.except[ThreadLocal]})(f: (T, U) ->{any.except[ThreadLocal]} R)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[R]^{this, that, f, executor, ExecutionContext} = { // This is typically overridden by the implementation in DefaultPromise, which provides // symmetric fail-fast behavior regardless of which future fails first. // @@ -481,7 +484,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` with the successful result of this or that `Future` or the failure of this `Future` if both fail * @group Transformations */ - def fallbackTo[U >: T](that: Future[U]): Future[U] = + def fallbackTo[U >: T](that: Future[U]^{any.except[ThreadLocal]}): Future[U]^{this, that, ExecutionContext} = if (this eq that) this else { implicit val ec = parasitic @@ -500,7 +503,7 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` holding the casted result of this `Future` or a `ClassCastException` otherwise * @group Transformations */ - def mapTo[S](implicit tag: ClassTag[S]): Future[S] = { + def mapTo[S](implicit tag: ClassTag[S]): Future[S]^{this, ExecutionContext} = { implicit val ec = parasitic val boxedClass = { val c = tag.runtimeClass @@ -543,10 +546,10 @@ trait Future[+T] extends Awaitable[T] { * @return a `Future` which will be completed with the exact same outcome as this `Future` but after the `PartialFunction` has been executed. * @group Callbacks */ - def andThen[U](pf: PartialFunction[Try[T], U])(implicit executor: ExecutionContext): Future[T] = + def andThen[U](pf: PartialFunction[Try[T], U]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[T]^{this, pf, executor, ExecutionContext} = transform { result => - try pf.applyOrElse[Try[T], Any](result, Future.id[Try[T]]) + try pf.applyOrElse[Try[T], Any](result, identity[Try[T]]) catch { case t if NonFatal(t) => executor.reportFailure(t) } // TODO: use `finally`? result @@ -560,7 +563,7 @@ trait Future[+T] extends Awaitable[T] { * @define nonDeterministic * Note: using this method yields nondeterministic dataflow programs. */ -object Future { +object Future uses ExecutionContext { /** Utilities, hoisted functions, etc. */ @@ -576,11 +579,11 @@ object Future { classOf[Unit] -> classOf[scala.runtime.BoxedUnit] ) - private final val _cachedId: AnyRef => AnyRef = Predef.identity + private final val _cachedId: AnyRef -> AnyRef = Predef.identity - private[concurrent] final def id[T]: T => T = _cachedId.asInstanceOf[T => T] + private[concurrent] final def id[T]: T -> T = _cachedId.asInstanceOf[T -> T] - private[concurrent] final val collectFailed = + private[concurrent] final val collectFailed: Any -> Nothing = (t: Any) => throw new NoSuchElementException("Future.collect partial function is not defined at: " + t) with NoStackTrace private[concurrent] final val filterFailure = @@ -592,20 +595,20 @@ object Future { private[concurrent] final val failedFailureFuture: Future[Nothing] = scala.concurrent.Future.fromTry(failedFailure) - private final val _failedFun: Try[Any] => Try[Throwable] = + private final val _failedFun: Try[Any] -> Try[Throwable] = v => if (v.isInstanceOf[Failure[Any]]) Success(v.asInstanceOf[Failure[Any]].exception) else failedFailure - private[concurrent] final def failedFun[T]: Try[T] => Try[Throwable] = _failedFun.asInstanceOf[Try[T] => Try[Throwable]] + private[concurrent] final def failedFun[T]: Try[T] -> Try[Throwable] = _failedFun.asInstanceOf[Try[T] -> Try[Throwable]] private[concurrent] final val recoverWithFailedMarker: Future[Nothing] = scala.concurrent.Future.failed(new Throwable with NoStackTrace) private[concurrent] final val recoverWithFailed = (t: Throwable) => recoverWithFailedMarker - private final val _zipWithTuple2: (Any, Any) => (Any, Any) = Tuple2.apply - private[concurrent] final def zipWithTuple2Fun[T,U] = _zipWithTuple2.asInstanceOf[(T,U) => (T,U)] + private final val _zipWithTuple2: (Any, Any) -> (Any, Any) = Tuple2.apply + private[concurrent] final def zipWithTuple2Fun[T,U] = _zipWithTuple2.asInstanceOf[(T,U) -> (T,U)] - private final val _addToBuilderFun: (Builder[Any, Nothing], Any) => Builder[Any, Nothing] = (b: Builder[Any, Nothing], e: Any) => b += e + private final val _addToBuilderFun: (Builder[Any, Nothing], Any) -> Builder[Any, Nothing] = (b: Builder[Any, Nothing], e: Any) => b += e private[concurrent] final def addToBuilderFun[A, M] = _addToBuilderFun.asInstanceOf[Function2[Builder[A, M], A, Builder[A, M]]] private[concurrent] def waitUndefinedError(): Nothing = @@ -615,7 +618,8 @@ object Future { throw new TimeoutException(s"Future timed out after [$delay]") /** A Future which is never completed. */ - object never extends Future[Nothing] { + val never: Future[Nothing] = caps.unsafe.unsafeAssumePure(neverImpl) + private object neverImpl extends Future[Nothing] uses ExecutionContext { @throws[TimeoutException] @throws[InterruptedException] @@ -652,26 +656,26 @@ object Future { timeoutError(atMost) } - override final def onComplete[U](f: Try[Nothing] => U)(implicit executor: ExecutionContext): Unit = () + override final def onComplete[U](f: Try[Nothing] ->{any.except[ThreadLocal]} U)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Unit = () override final def isCompleted: Boolean = false override final def value: Option[Try[Nothing]] = None - override final def failed: Future[Throwable] = this - override final def foreach[U](f: Nothing => U)(implicit executor: ExecutionContext): Unit = () - override final def transform[S](s: Nothing => S, f: Throwable => Throwable)(implicit executor: ExecutionContext): Future[S] = this - override final def transform[S](f: Try[Nothing] => Try[S])(implicit executor: ExecutionContext): Future[S] = this - override final def transformWith[S](f: Try[Nothing] => Future[S])(implicit executor: ExecutionContext): Future[S] = this - override final def map[S](f: Nothing => S)(implicit executor: ExecutionContext): Future[S] = this - override final def flatMap[S](f: Nothing => Future[S])(implicit executor: ExecutionContext): Future[S] = this - override final def flatten[S](implicit ev: Nothing <:< Future[S]): Future[S] = this - override final def filter(p: Nothing => Boolean)(implicit executor: ExecutionContext): Future[Nothing] = this - override final def collect[S](pf: PartialFunction[Nothing, S])(implicit executor: ExecutionContext): Future[S] = this - override final def recover[U >: Nothing](pf: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Future[U] = this - override final def recoverWith[U >: Nothing](pf: PartialFunction[Throwable, Future[U]])(implicit executor: ExecutionContext): Future[U] = this - override final def zip[U](that: Future[U]): Future[(Nothing, U)] = this - override final def zipWith[U, R](that: Future[U])(f: (Nothing, U) => R)(implicit executor: ExecutionContext): Future[R] = this - override final def fallbackTo[U >: Nothing](that: Future[U]): Future[U] = this - override final def mapTo[S](implicit tag: ClassTag[S]): Future[S] = this - override final def andThen[U](pf: PartialFunction[Try[Nothing], U])(implicit executor: ExecutionContext): Future[Nothing] = this + override final def failed: Future[Throwable]^{this} = this + override final def foreach[U](f: Nothing ->{any.except[ThreadLocal]} U)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Unit = () + override final def transform[S](s: Nothing ->{any.except[ThreadLocal]} S, f: Throwable ->{any.except[ThreadLocal]} Throwable)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this} = this + override final def transform[S](f: Try[Nothing] ->{any.except[ThreadLocal]} Try[S])(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this} = this + override final def transformWith[S](f: Try[Nothing] ->{any.except[ThreadLocal]} Future[S]^)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this} = this + override final def map[S](f: Nothing ->{any.except[ThreadLocal]} S)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this} = this + override final def flatMap[S](f: Nothing ->{any.except[ThreadLocal]} Future[S]^)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this} = this + override final def flatten[S](implicit ev: Nothing <:< Future[S]): Future[S]^{this} = this + override final def filter(p: Nothing ->{any.except[ThreadLocal]} Boolean)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[Nothing]^{this} = this + override final def collect[S](pf: PartialFunction[Nothing, S]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this} = this + override final def recover[U >: Nothing](pf: PartialFunction[Throwable, U]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[U]^{this} = this + override final def recoverWith[U >: Nothing](pf: PartialFunction[Throwable, Future[U]]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[U]^{this} = this + override final def zip[U](that: Future[U]^{any.except[ThreadLocal]}): Future[(Nothing, U)]^{this} = this + override final def zipWith[U, R](that: Future[U]^{any.except[ThreadLocal]})(f: (Nothing, U) ->{any.except[ThreadLocal]} R)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[R]^{this} = this + override final def fallbackTo[U >: Nothing](that: Future[U]^{any.except[ThreadLocal]}): Future[U]^{this} = this + override final def mapTo[S](implicit tag: ClassTag[S]): Future[S]^{this} = this + override final def andThen[U](pf: PartialFunction[Try[Nothing], U]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[Nothing]^{this} = this override final def toString(): String = "Future()" } @@ -721,7 +725,7 @@ object Future { * @param executor the execution context on which the future is run * @return the `Future` holding the result of the computation */ - final def apply[T](body: => T)(implicit executor: ExecutionContext): Future[T] = + final def apply[T](body: ->{any.except[ThreadLocal]} T)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[T]^{body, executor, ExecutionContext} = unit.map(_ => body) /** Starts an asynchronous computation and returns a `Future` instance with the result of that computation once it completes. @@ -742,7 +746,7 @@ object Future { * @param executor the execution context on which the `body` is evaluated in * @return the `Future` holding the result of the computation */ - final def delegate[T](body: => Future[T])(implicit executor: ExecutionContext): Future[T] = + final def delegate[T](body: ->{any.except[ThreadLocal]} Future[T]^)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[T]^{body, executor, ExecutionContext} = unit.flatMap(_ => body) /** Simple version of `Future.traverse`. Asynchronously and non-blockingly transforms, in essence, a `IterableOnce[Future[A]]` @@ -756,10 +760,16 @@ object Future { * @param executor the `ExecutionContext` on which the sequencing will be executed * @return the `Future` of the resulting collection */ - final def sequence[A, CC[X] <: IterableOnce[X], To](in: CC[Future[A]])(implicit bf: BuildFrom[CC[Future[A]], A, To], executor: ExecutionContext): Future[To] = - in.iterator.foldLeft(successful(bf.newBuilder(in))) { - (fr, fa) => fr.zipWith(fa)(Future.addToBuilderFun) - }.map(_.result())(using if (executor.isInstanceOf[BatchingExecutor]) executor else parasitic) + final def sequence[A, CC[X] <: IterableOnce[X], To](in: CC[Future[A]^]^)(implicit bf: BuildFrom[CC[Future[A]^], A, To], executor: ExecutionContext^{any.except[ThreadLocal]}): Future[To]^{in, executor, ExecutionContext} = + def sequenceCorrect[A, CC[X] <: IterableOnce[X], To, In^, C^ <: {any.except[ThreadLocal]}] + (in: CC[Future[A]^{C}]^{In}) + (using bf: BuildFrom[CC[Future[A]^{C}]^{In}, A, To], executor: ExecutionContext^{any.except[ThreadLocal]}): Future[To]^{In, C, executor, ExecutionContext} = + in.iterator.foldLeft[Future[Builder[A, To]]^{C, executor, ExecutionContext}](successful(bf.newBuilder(in))) { + (fr, fa) => fr.zipWith(fa)(Future.addToBuilderFun) + }.map(_.result())(using if (executor.isInstanceOf[BatchingExecutor]) executor else parasitic) + + // SAFETY: the sequence method is supposed to be typed like above ^ + sequenceCorrect[A, CC, To, {}, {}](in.asInstanceOf[CC[Future[A]]])(using bf.asInstanceOf[BuildFrom[CC[Future[A]^{}], A, To]], executor) /** Asynchronously and non-blockingly returns a new `Future` to the result of the first future * in the list that is completed. This means no matter if it is completed as a success or as a failure. @@ -769,35 +779,40 @@ object Future { * @param executor the `ExecutionContext` on which the futures' completion handlers will be executed * @return the `Future` holding the result of the future that is first to be completed */ - final def firstCompletedOf[T](futures: IterableOnce[Future[T]])(implicit executor: ExecutionContext): Future[T] = { - val i = futures.iterator - if (!i.hasNext) Future.never - else { - val p = Promise[T]() - val firstCompleteHandler = new AtomicReference(List.empty[() => Unit]) with (Try[T] => Unit) { - final def apply(res: Try[T]): Unit = { - val deregs = getAndSet(null) - if (deregs != null) { - p.tryComplete(res) // tryComplete is likely to be cheaper than complete - deregs.foreach(_.apply()) + final def firstCompletedOf[T](futures: IterableOnce[Future[T]^]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[T]^{futures} = { + def impl[T, C^ <: {any.except[ThreadLocal]}](futures: IterableOnce[Future[T]^{C}]^)(using executor: ExecutionContext^{any.except[ThreadLocal]}): Future[T]^{futures, C} = { + val i = futures.iterator + if (!i.hasNext) Future.never + else { + val p = Promise[T]() + object firstCompleteHandler extends AtomicReference[List[() ->{C, executor, ExecutionContext} Unit]](List.empty) { + final def apply(res: Try[T]): Unit = { + val deregs: List[() ->{C, executor, ExecutionContext} Unit] = getAndSet(null).asInstanceOf + if (deregs != null) { + p.tryComplete(res) // tryComplete is likely to be cheaper than complete + deregs.foreach(v => v()) + } } } - } - var completed = false - while (i.hasNext && !completed) { - val deregs = firstCompleteHandler.get - if (deregs == null) completed = true - else i.next() match { - case dp: DefaultPromise[T @unchecked] => - val d = dp.onCompleteWithUnregister(firstCompleteHandler) - if (!firstCompleteHandler.compareAndSet(deregs, d :: deregs)) - d.apply() - case f => - f.onComplete(firstCompleteHandler) + var completed = false + while (i.hasNext && !completed) { + val deregs = firstCompleteHandler.get + if (deregs == null) completed = true + else i.next() match { + case dp: DefaultPromise[T @unchecked] => + val d = dp.onCompleteWithUnregister(firstCompleteHandler.apply) + if (!firstCompleteHandler.compareAndSet(deregs, d :: deregs)) + d.apply() + case f => + f.onComplete(firstCompleteHandler.apply) + } } + p.future } - p.future } + + // SAFETY: rely on monotonicity (?). TODO In the future: needs capture set variables + impl(futures.asInstanceOf[IterableOnce[Future[T]]^{futures}])(using executor) } /** Asynchronously and non-blockingly returns a `Future` that will hold the optional result @@ -809,15 +824,16 @@ object Future { * @param executor the `ExecutionContext` on which the futures' completion handlers will be executed * @return the `Future` holding the optional result of the search */ - final def find[T](futures: scala.collection.immutable.Iterable[Future[T]])(p: T => Boolean)(implicit executor: ExecutionContext): Future[Option[T]] = { - def searchNext(i: Iterator[Future[T]]): Future[Option[T]] = + final def find[T](futures: scala.collection.immutable.Iterable[Future[T]^]^{any.except[ThreadLocal]})(p: T ->{any.except[ThreadLocal]} Boolean)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[Option[T]]^{p, futures, executor, ExecutionContext} = { + def searchNext[C^ <: {any.except[ThreadLocal]}](i: Iterator[Future[T]^{C}]^{any.except[ThreadLocal]}): Future[Option[T]]^{p, i, C, executor, ExecutionContext} = if (!i.hasNext) successful(None) else i.next().transformWith { case Success(r) if p(r) => successful(Some(r)) case _ => searchNext(i) } - searchNext(futures.iterator) + // SAFETY: rely on monotonicity (?). TODO In the future: needs capture set variables + searchNext(futures.asInstanceOf[scala.collection.immutable.Iterable[Future[T]]^{futures}].iterator) } /** A non-blocking, asynchronous left fold over the specified futures, @@ -842,10 +858,11 @@ object Future { * @param executor the `ExecutionContext` on which the fold operation will be executed * @return the `Future` holding the result of the fold */ - final def foldLeft[T, R](futures: scala.collection.immutable.Iterable[Future[T]])(zero: R)(op: (R, T) => R)(implicit executor: ExecutionContext): Future[R] = - foldNext(futures.iterator, zero, op) + final def foldLeft[T, R](futures: scala.collection.immutable.Iterable[Future[T]^]^{any.except[ThreadLocal]})(zero: R)(op: (R, T) ->{any.except[ThreadLocal]} R)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[R]^{futures, op, executor, ExecutionContext} = + // SAFETY: rely on monotonicity (?). TODO In the future: needs capture set variables + foldNext(futures.asInstanceOf[scala.collection.immutable.Iterable[Future[T]]^{futures}].iterator, zero, op) - private final def foldNext[T, R](i: Iterator[Future[T]], prevValue: R, op: (R, T) => R)(implicit executor: ExecutionContext): Future[R] = + private final def foldNext[T, R, C^ <: {any.except[ThreadLocal]}](i: Iterator[Future[T]^{C}]^{any.except[ThreadLocal]}, prevValue: R, op: (R, T) ->{any.except[ThreadLocal]} R)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[R]^{i, C, op, executor, ExecutionContext} = if (!i.hasNext) successful(prevValue) else i.next().flatMap { value => foldNext(i, op(prevValue, value), op) } @@ -870,9 +887,11 @@ object Future { */ @deprecated("use Future.foldLeft instead", "2.12.0") // not removed in 2.13, to facilitate 2.11/2.12/2.13 cross-building; remove further down the line (see scala/scala#6319) - def fold[T, R](futures: IterableOnce[Future[T]])(zero: R)(@deprecatedName("foldFun") op: (R, T) => R)(implicit executor: ExecutionContext): Future[R] = - if (futures.isEmpty) successful(zero) - else sequence(futures)(using ArrayBuffer, executor).map(_.foldLeft(zero)(op)) + def fold[T, R](futures: IterableOnce[Future[T]^]^{any.except[ThreadLocal]})(zero: R)(@deprecatedName("foldFun") op: (R, T) ->{any.except[ThreadLocal]} R)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[R]^{futures, op, executor, ExecutionContext} = + // SAFETY: rely on monotonicity (?). TODO In the future: needs capture set variables + val futs = futures.asInstanceOf[IterableOnce[Future[T]]^{futures}] + if (futs.iterator.isEmpty) successful(zero) + else sequence(futs)(using ArrayBuffer, executor).map(_.foldLeft(zero)(op)) /** Initiates a non-blocking, asynchronous, fold over the supplied futures * where the fold-zero is the result value of the first `Future` in the collection. @@ -891,9 +910,11 @@ object Future { */ @deprecated("use Future.reduceLeft instead", "2.12.0") // not removed in 2.13, to facilitate 2.11/2.12/2.13 cross-building; remove further down the line (see scala/scala#6319) - final def reduce[T, R >: T](futures: IterableOnce[Future[T]])(op: (R, T) => R)(implicit executor: ExecutionContext): Future[R] = - if (futures.isEmpty) failed(new NoSuchElementException("reduce attempted on empty collection")) - else sequence(futures)(using ArrayBuffer, executor).map(_ reduceLeft op) + final def reduce[T, R >: T](futures: IterableOnce[Future[T]^]^{any.except[ThreadLocal]})(op: (R, T) ->{any.except[ThreadLocal]} R)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[R]^{futures, op, executor, ExecutionContext} = + // SAFETY: rely on monotonicity (?). TODO In the future: needs capture set variables + val futs = futures.asInstanceOf[IterableOnce[Future[T]]^{futures}] + if (futs.iterator.isEmpty) failed(new NoSuchElementException("reduce attempted on empty collection")) + else sequence(futs)(using ArrayBuffer, executor).map(_ reduceLeft op) /** Initiates a non-blocking, asynchronous, left reduction over the supplied futures * where the zero is the result value of the first `Future`. @@ -912,8 +933,9 @@ object Future { * @param executor the `ExecutionContext` on which the reduce operation will be executed * @return the `Future` holding the result of the reduce */ - final def reduceLeft[T, R >: T](futures: scala.collection.immutable.Iterable[Future[T]])(op: (R, T) => R)(implicit executor: ExecutionContext): Future[R] = { - val i = futures.iterator + final def reduceLeft[T, R >: T](futures: scala.collection.immutable.Iterable[Future[T]^]^{any.except[ThreadLocal]})(op: (R, T) ->{any.except[ThreadLocal]} R)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[R]^{futures, op, executor, ExecutionContext} = { + // SAFETY: rely on monotonicity (?). TODO In the future: needs capture set variables + val i = futures.iterator.asInstanceOf[Iterator[Future[T]]^{futures}] if (!i.hasNext) failed(new NoSuchElementException("reduceLeft attempted on empty collection")) else i.next() flatMap { v => foldNext(i, v, op) } } @@ -939,10 +961,13 @@ object Future { * @param executor the `ExecutionContext` on which `fn` and the resulting Futures will be executed * @return the `Future` of the collection of results */ - final def traverse[A, B, M[X] <: IterableOnce[X]](in: M[A])(fn: A => Future[B])(implicit bf: BuildFrom[M[A], B, M[B]], executor: ExecutionContext): Future[M[B]] = - in.iterator.foldLeft(successful(bf.newBuilder(in))) { - (fr, a) => fr.zipWith(fn(a))(Future.addToBuilderFun) - }.map(_.result())(using if (executor.isInstanceOf[BatchingExecutor]) executor else parasitic) + final def traverse[A, B, M[X] <: IterableOnce[X]](in: M[A]^{any.except[ThreadLocal]})(fn: A ->{any.except[ThreadLocal]} Future[B]^)(implicit bf: BuildFrom[M[A]^{in}, B, M[B]], executor: ExecutionContext^{any.except[ThreadLocal]}): Future[M[B]]^{in, fn, executor, ExecutionContext} = + def traverseCorrect[A, B, M[X] <: IterableOnce[X], C^ <: {any.except[ThreadLocal]}](in: M[A]^{any.except[ThreadLocal]})(fn: A ->{any.except[ThreadLocal]} Future[B]^{C})(using bf: BuildFrom[M[A]^{in}, B, M[B]], executor: ExecutionContext^{any.except[ThreadLocal]}): Future[M[B]]^{in, fn, C, executor, ExecutionContext} = + in.iterator.foldLeft[Future[Builder[B, M[B]]]^{in, fn, C, executor, ExecutionContext}](successful(bf.newBuilder(in))) { + (fr, a) => fr.zipWith(fn(a))(Future.addToBuilderFun) + }.map(_.result())(using if (executor.isInstanceOf[BatchingExecutor]) executor else parasitic) + + traverseCorrect(in)(fn.asInstanceOf[A ->{fn} Future[B]])(using bf, executor) } @deprecated("Superseded by `scala.concurrent.Batchable`", "2.13.0") diff --git a/library/src/scala/concurrent/Promise.scala b/library/src/scala/concurrent/Promise.scala index f50f3fa2936f..8fa18e254aa9 100644 --- a/library/src/scala/concurrent/Promise.scala +++ b/library/src/scala/concurrent/Promise.scala @@ -12,6 +12,9 @@ package scala.concurrent +import language.experimental.captureChecking +import caps.* + import scala.language.`2.13` import scala.util.{ Try, Success, Failure } @@ -36,9 +39,9 @@ import scala.util.{ Try, Success, Failure } * * @tparam T the type of the value held by this promise and its associated future */ -trait Promise[T] { +trait Promise[T] uses ExecutionContext { this: Promise[T]^{any.except[ThreadLocal]} => /** Future containing the value of this promise. */ - def future: Future[T] + def future: Future[T]^{this} /** Returns whether the promise has already been completed with * a value or an exception. @@ -73,7 +76,7 @@ trait Promise[T] { * @param other the future whose result will be used to complete this promise * @return This promise */ - def completeWith(other: Future[T]): this.type = { + def completeWith(other: Future[T]^): this.type = { if (other ne this.future) // this tryCompleteWith this doesn't make much sense other.onComplete(this tryComplete _)(using ExecutionContext.parasitic) @@ -85,7 +88,7 @@ trait Promise[T] { * @return This promise */ @deprecated("Since this method is semantically equivalent to `completeWith`, use that instead.", "2.13.0") - final def tryCompleteWith(other: Future[T]): this.type = completeWith(other) + final def tryCompleteWith(other: Future[T]^): this.type = completeWith(other) /** Completes the promise with a value. * @@ -132,7 +135,7 @@ object Promise { * @tparam T the type of the value in the promise * @return the newly created `Promise` instance */ - final def apply[T](): Promise[T] = new impl.Promise.DefaultPromise[T]() + final def apply[T](): Promise[T] = impl.Promise.DefaultPromise[T]() /** Creates an already completed Promise with the specified exception. * @@ -156,5 +159,5 @@ object Promise { * @param result the `Try` value (success or failure) to complete the promise with * @return the newly created `Promise` instance */ - final def fromTry[T](result: Try[T]): Promise[T] = new impl.Promise.DefaultPromise[T](result) + final def fromTry[T](result: Try[T]): Promise[T] = impl.Promise.DefaultPromise[T](result) } diff --git a/library/src/scala/concurrent/duration/Deadline.scala b/library/src/scala/concurrent/duration/Deadline.scala index b9b210519884..24ed39690e50 100644 --- a/library/src/scala/concurrent/duration/Deadline.scala +++ b/library/src/scala/concurrent/duration/Deadline.scala @@ -14,6 +14,8 @@ package scala.concurrent.duration import scala.language.`2.13` +import language.experimental.captureChecking + /** This class stores a deadline, as obtained via `Deadline.now` or the * duration DSL: * diff --git a/library/src/scala/concurrent/duration/Duration.scala b/library/src/scala/concurrent/duration/Duration.scala index b7e73d1a14f6..3f877c6a4f26 100644 --- a/library/src/scala/concurrent/duration/Duration.scala +++ b/library/src/scala/concurrent/duration/Duration.scala @@ -12,6 +12,8 @@ package scala.concurrent.duration +import language.experimental.captureChecking + import scala.language.`2.13` import java.lang.{ Double => JDouble } import scala.collection.StringParsers diff --git a/library/src/scala/concurrent/duration/DurationConversions.scala b/library/src/scala/concurrent/duration/DurationConversions.scala index b90d93e02ffb..27db67445951 100644 --- a/library/src/scala/concurrent/duration/DurationConversions.scala +++ b/library/src/scala/concurrent/duration/DurationConversions.scala @@ -12,6 +12,8 @@ package scala.concurrent.duration +import language.experimental.captureChecking + import scala.language.`2.13` import DurationConversions._ diff --git a/library/src/scala/concurrent/duration/package.scala b/library/src/scala/concurrent/duration/package.scala index 4111462c77f0..e0afef498c9b 100644 --- a/library/src/scala/concurrent/duration/package.scala +++ b/library/src/scala/concurrent/duration/package.scala @@ -12,6 +12,8 @@ package scala.concurrent +import language.experimental.captureChecking + import scala.language.`2.13` import scala.language.implicitConversions diff --git a/library/src/scala/concurrent/impl/ExecutionContextImpl.scala b/library/src/scala/concurrent/impl/ExecutionContextImpl.scala index b97baaad32ef..67c1ae1714b5 100644 --- a/library/src/scala/concurrent/impl/ExecutionContextImpl.scala +++ b/library/src/scala/concurrent/impl/ExecutionContextImpl.scala @@ -12,12 +12,15 @@ package scala.concurrent.impl +import language.experimental.captureChecking +import caps.* + import scala.language.`2.13` import java.util.concurrent.{ Semaphore, ForkJoinPool, ForkJoinWorkerThread, Callable, Executor, ExecutorService, ThreadFactory, TimeUnit } import java.util.Collection import scala.concurrent.{ BlockContext, ExecutionContext, CanAwait, ExecutionContextExecutor, ExecutionContextExecutorService } -private[scala] class ExecutionContextImpl private[impl] (final val executor: Executor, final val reporter: Throwable => Unit) extends ExecutionContextExecutor { +private[scala] class ExecutionContextImpl private[impl] (final val executor: Executor^{any.except[ThreadLocal]}, final val reporter: Throwable ->{any.except[ThreadLocal]} Unit) extends ExecutionContextExecutor { require(executor ne null, "Executor must not be null") override final def execute(runnable: Runnable): Unit = executor.execute(runnable) override final def reportFailure(t: Throwable): Unit = reporter(t) @@ -29,7 +32,7 @@ private[concurrent] object ExecutionContextImpl { final val daemonic: Boolean, final val maxBlockers: Int, final val prefix: String, - final val uncaught: Thread.UncaughtExceptionHandler) extends ThreadFactory with ForkJoinPool.ForkJoinWorkerThreadFactory { + final val uncaught: Thread.UncaughtExceptionHandler^{any.except[ThreadLocal]}) extends ThreadFactory with ForkJoinPool.ForkJoinWorkerThreadFactory { require(prefix ne null, "DefaultThreadFactory.prefix must be non null") require(maxBlockers >= 0, "DefaultThreadFactory.maxBlockers must be greater-or-equal-to 0") @@ -37,7 +40,7 @@ private[concurrent] object ExecutionContextImpl { private final val blockerPermits = new Semaphore(maxBlockers) @annotation.nowarn("cat=deprecation") - def wire[T <: Thread](thread: T): T = { + def wire[C^, T <: Thread^{C}](thread: T): T = { thread.setDaemon(daemonic) thread.setUncaughtExceptionHandler(uncaught) thread.setName(prefix + "-" + thread.getId()) @@ -46,14 +49,14 @@ private[concurrent] object ExecutionContextImpl { def newThread(runnable: Runnable): Thread = wire(new Thread(runnable)) - def newThread(fjp: ForkJoinPool): ForkJoinWorkerThread = - wire(new ForkJoinWorkerThread(fjp) with BlockContext { + def newThread(fjp: ForkJoinPool^): ForkJoinWorkerThread^{fjp} = + wire[{fjp}, ForkJoinWorkerThread^{fjp}](new ForkJoinWorkerThread(fjp) with BlockContext { private final var isBlocked: Boolean = false // This is only ever read & written if this thread is the current thread final override def blockOn[T](thunk: => T)(implicit permission: CanAwait): T = if ((Thread.currentThread eq this) && !isBlocked && blockerPermits.tryAcquire()) { try { - val b: (ForkJoinPool.ManagedBlocker & (() => T)) = - new ForkJoinPool.ManagedBlocker with (() => T) { + val b: (ForkJoinPool.ManagedBlocker & (() -> T))^{thunk} = + new ForkJoinPool.ManagedBlocker with (() -> T) { private final var result: T = null.asInstanceOf[T] private final var done: Boolean = false final override def block(): Boolean = { @@ -79,7 +82,7 @@ private[concurrent] object ExecutionContextImpl { }) } - def createDefaultExecutorService(reporter: Throwable => Unit): ExecutionContextExecutorService = { + def createDefaultExecutorService(reporter: Throwable ->{any.except[ThreadLocal]} Unit): ExecutionContextExecutorService^{reporter} = { def getInt(name: String, default: String) = (try System.getProperty(name, default) catch { case e: SecurityException => default }) match { @@ -109,14 +112,14 @@ private[concurrent] object ExecutionContextImpl { } } - def fromExecutor(e: Executor | Null, reporter: Throwable => Unit = ExecutionContext.defaultReporter): ExecutionContextExecutor = + def fromExecutor(e: (Executor^{any.except[ThreadLocal]}) | Null, reporter: Throwable ->{any.except[ThreadLocal]} Unit = ExecutionContext.defaultReporter): ExecutionContextExecutor^{e, reporter} = e match { case null => createDefaultExecutorService(reporter) case some => new ExecutionContextImpl(some, reporter) } - def fromExecutorService(es: ExecutorService | Null, reporter: Throwable => Unit = ExecutionContext.defaultReporter): - ExecutionContextExecutorService = es match { + def fromExecutorService(es: (ExecutorService^{any.except[ThreadLocal]}) | Null, reporter: Throwable ->{any.except[ThreadLocal]} Unit = ExecutionContext.defaultReporter): + ExecutionContextExecutorService^{es, reporter} = es match { case null => createDefaultExecutorService(reporter) case some => // This is a anonymous class extending a Java class, so we left inferred flexible types in the signatures. diff --git a/library/src/scala/concurrent/impl/Promise.scala b/library/src/scala/concurrent/impl/Promise.scala index 1551592923f1..99f8fef61d7b 100644 --- a/library/src/scala/concurrent/impl/Promise.scala +++ b/library/src/scala/concurrent/impl/Promise.scala @@ -12,6 +12,9 @@ package scala.concurrent.impl +import language.experimental.captureChecking +import caps.* + import scala.language.`2.13` import scala.concurrent.{Batchable, CanAwait, ExecutionContext, ExecutionException, Future, TimeoutException} import scala.concurrent.duration.Duration @@ -49,7 +52,7 @@ private[impl] final class CompletionLatch[T] extends AbstractQueuedSynchronizer } } -private[concurrent] object Promise { +private[concurrent] object Promise uses ExecutionContext, ExecutionContext initially { /** Link represents a completion dependency between 2 DefaultPromises. * As the DefaultPromise referred to by a Link can itself be linked to another promise * `relink` traverses such chains and compresses them so that the link always points @@ -64,17 +67,20 @@ private[concurrent] object Promise { * @tparam T the type of the promised value * @param to the target `DefaultPromise` that this link initially points to */ - private[concurrent] final class Link[T](to: DefaultPromise[T]) extends AtomicReference[DefaultPromise[T]](to) { + private[concurrent] final class Link[T](to: DefaultPromise[T]^) extends AtomicReference[DefaultPromise[T]^{to}](to) { /** Compresses this chain and returns the currently known root of this chain of Links. * * @param owner the `DefaultPromise` that owns this link, used to complete it if the root is already resolved * @return the root `DefaultPromise` of the dependency chain, or `owner` (after completing it with the resolved value) if the chain has already been completed */ - final def promise(owner: DefaultPromise[T]): DefaultPromise[T] = { - val c = get() + final def promise(owner: DefaultPromise[T]^): DefaultPromise[T]^{owner, to} = { + val c = getPromise compressed(current = c, target = c, owner = owner) } + // See https://github.com/scala/scala3/issues/26556 + private[impl] def getPromise: DefaultPromise[T]^{to} = get().asInstanceOf[DefaultPromise[T]^{to}] + /** The combination of traversing and possibly unlinking of a given `target` DefaultPromise. * * @param current the `DefaultPromise` most recently read from this link's atomic reference @@ -82,16 +88,18 @@ private[concurrent] object Promise { * @param owner the `DefaultPromise` that owns this link, used to complete it if the root is already resolved * @return the root `DefaultPromise` reachable from `target`, or `owner` if the root was already completed */ - @inline @tailrec private final def compressed(current: DefaultPromise[T], target: DefaultPromise[T], owner: DefaultPromise[T]): DefaultPromise[T] = { - val value = target.get() - if (value.isInstanceOf[Callbacks[?]]) { - if (compareAndSet(current, target)) target // Link - else compressed(current = get(), target = target, owner = owner) // Retry - } else if (value.isInstanceOf[Link[?]]) compressed(current = current, target = value.asInstanceOf[Link[T]].get(), owner = owner) // Compress - else /*if (value.isInstanceOf[Try[T]])*/ { - owner.unlink(value.asInstanceOf[Try[T]]) // Discard links - owner - } + @inline @tailrec private final def compressed(current: DefaultPromise[T]^, target: DefaultPromise[T]^, owner: DefaultPromise[T]^): DefaultPromise[T]^{target, owner} = { + val value = target.getRef + value match + case v: Callbacks[T] => + if (compareAndSet(current, target)) target // Link + else compressed(current = getPromise, target = target, owner = owner) // Retry + case v: Link[T] => + val v1: DefaultPromise[T]^{v} = v.getPromise + compressed(current = current, target = v1, owner = owner) // Compress + case v: Try[T] => + owner.unlink(/* We know a Try here is safe */ caps.unsafe.unsafeAssumePure(v)) // Discard links + owner } } @@ -116,7 +124,7 @@ private[concurrent] object Promise { } // Left non-final to enable addition of extra fields by Java/Scala converters in scala-java8-compat. - class DefaultPromise[T] private (initial: AnyRef) extends AtomicReference[AnyRef](initial) with scala.concurrent.Promise[T] with scala.concurrent.Future[T] with (Try[T] => Unit) { + class DefaultPromise[T] private (initial: AnyRef^{any.except[ThreadLocal]}) extends AtomicReference[AnyRef^{initial}](initial) with scala.concurrent.Promise[T] with scala.concurrent.Future[T] with (Try[T] => Unit) uses ExecutionContext { this: DefaultPromise[T]^{any.except[ThreadLocal]} => /** Constructs a new, completed, Promise. * * @param result the completed result value to initialize this promise with @@ -124,7 +132,7 @@ private[concurrent] object Promise { final def this(result: Try[T]) = this(resolve(result): AnyRef) /** Constructs a new, un-completed, Promise. */ - final def this() = this(Noop: AnyRef) + final def this() = this(Noop: AnyRef^{ExecutionContext}) /** WARNING: the `resolved` value needs to have been pre-resolved using `resolve()` * INTERNAL API @@ -132,30 +140,30 @@ private[concurrent] object Promise { * @param resolved the pre-resolved `Try` value to complete this promise with */ override final def apply(resolved: Try[T]): Unit = - tryComplete0(get(), resolved) + tryComplete0(getRef, resolved) /** Returns the associated `Future` with this `Promise` */ - override final def future: Future[T] = this + override final def future: Future[T]^{this} = this - override final def transform[S](f: Try[T] => Try[S])(implicit executor: ExecutionContext): Future[S] = - dispatchOrAddCallbacks(get(), new Transformation[T, S](Xform_transform, f, executor)) + override final def transform[S](f: Try[T] ->{any.except[ThreadLocal]} Try[S])(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, f, executor, ExecutionContext} = + dispatchOrAddCallbacks(getRef, Transformation[T, S](Xform_transform, f, executor)) - override final def transformWith[S](f: Try[T] => Future[S])(implicit executor: ExecutionContext): Future[S] = - dispatchOrAddCallbacks(get(), new Transformation[T, S](Xform_transformWith, f, executor)) + override final def transformWith[S](f: Try[T] ->{any.except[ThreadLocal]} Future[S]^)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, f, executor, ExecutionContext} = + dispatchOrAddCallbacks(getRef, Transformation[T, S](Xform_transformWith, f, executor)) - override final def zipWith[U, R](that: Future[U])(f: (T, U) => R)(implicit executor: ExecutionContext): Future[R] = { - val state = get() + override final def zipWith[U, R](that: Future[U]^{any.except[ThreadLocal]})(f: (T, U) ->{any.except[ThreadLocal]} R)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[R]^{this, that, f, executor, ExecutionContext} = { + val state = getRef if (state.isInstanceOf[Try[?]]) { - if (state.asInstanceOf[Try[T]].isFailure) this.asInstanceOf[Future[R]] + if (state.asInstanceOf[Try[T]].isFailure) this.asInstanceOf[Future[R]^{this}] else { val l = state.asInstanceOf[Success[T]].get that.map(r => f(l, r)) } } else { val buffer = new AtomicReference[Success[Any]]() - val zipped = new DefaultPromise[R]() + val zipped: DefaultPromise[R]^{this, ExecutionContext} = DefaultPromise[R]() - val thisF: Try[T] => Unit = { + val thisF: Try[T] ->{this, zipped, f} Unit = { case left: Success[?] => val right = buffer.getAndSet(left).asInstanceOf[Success[U]] if (right ne null) @@ -164,7 +172,7 @@ private[concurrent] object Promise { zipped.tryComplete(f.asInstanceOf[Failure[R]]) } - val thatF: Try[U] => Unit = { + val thatF: Try[U] ->{this, zipped, f} Unit = { case right: Success[?] => val left = buffer.getAndSet(right).asInstanceOf[Success[T]] if (left ne null) @@ -173,60 +181,60 @@ private[concurrent] object Promise { zipped.tryComplete(f.asInstanceOf[Failure[R]]) } // Cheaper than this.onComplete since we already polled the state - this.dispatchOrAddCallbacks(state, new Transformation[T, Unit](Xform_onComplete, thisF, executor)) + this.dispatchOrAddCallbacks(state, Transformation[T, Unit](Xform_onComplete, thisF, executor)) that.onComplete(thatF) zipped.future } } - override final def foreach[U](f: T => U)(implicit executor: ExecutionContext): Unit = { - val state = get() - if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, new Transformation[T, Unit](Xform_foreach, f, executor)) + override final def foreach[U](f: T ->{any.except[ThreadLocal]} U)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Unit = { + val state = getRef + if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, Transformation[T, Unit](Xform_foreach, f, executor)) } - override final def flatMap[S](f: T => Future[S])(implicit executor: ExecutionContext): Future[S] = { - val state = get() - if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, new Transformation[T, S](Xform_flatMap, f, executor)) + override final def flatMap[S](f: T ->{any.except[ThreadLocal]} Future[S]^)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, f, executor, ExecutionContext} = { + val state = getRef + if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, Transformation[T, S](Xform_flatMap, f, executor)) else this.asInstanceOf[Future[S]] } - override final def map[S](f: T => S)(implicit executor: ExecutionContext): Future[S] = { - val state = get() - if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, new Transformation[T, S](Xform_map, f, executor)) + override final def map[S](f: T ->{any.except[ThreadLocal]} S)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, f, executor, ExecutionContext} = { + val state = getRef + if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, Transformation[T, S](Xform_map, f, executor)) else this.asInstanceOf[Future[S]] } - override final def filter(p: T => Boolean)(implicit executor: ExecutionContext): Future[T] = { - val state = get() - if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, new Transformation[T, T](Xform_filter, p, executor)) // Short-circuit if we get a Success + override final def filter(p: T ->{any.except[ThreadLocal]} Boolean)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[T]^{this, p, executor, ExecutionContext} = { + val state = getRef + if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, Transformation[T, T](Xform_filter, p, executor)) // Short-circuit if we get a Success else this } - override final def collect[S](pf: PartialFunction[T, S])(implicit executor: ExecutionContext): Future[S] = { - val state = get() - if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, new Transformation[T, S](Xform_collect, pf, executor)) // Short-circuit if we get a Success + override final def collect[S](pf: PartialFunction[T, S]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[S]^{this, pf, executor, ExecutionContext} = { + val state = getRef + if (!state.isInstanceOf[Failure[?]]) dispatchOrAddCallbacks(state, Transformation[T, S](Xform_collect, pf, executor)) // Short-circuit if we get a Success else this.asInstanceOf[Future[S]] } - override final def recoverWith[U >: T](pf: PartialFunction[Throwable, Future[U]])(implicit executor: ExecutionContext): Future[U] = { - val state = get() - if (!state.isInstanceOf[Success[?]]) dispatchOrAddCallbacks(state, new Transformation[T, U](Xform_recoverWith, pf, executor)) // Short-circuit if we get a Failure + override final def recoverWith[U >: T](pf: PartialFunction[Throwable, Future[U]]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[U]^{this, pf, executor, ExecutionContext} = { + val state = getRef + if (!state.isInstanceOf[Success[?]]) dispatchOrAddCallbacks(state, Transformation[T, U](Xform_recoverWith, pf, executor)) // Short-circuit if we get a Failure else this.asInstanceOf[Future[U]] } - override final def recover[U >: T](pf: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Future[U] = { - val state = get() - if (!state.isInstanceOf[Success[?]]) dispatchOrAddCallbacks(state, new Transformation[T, U](Xform_recover, pf, executor)) // Short-circuit if we get a Failure + override final def recover[U >: T](pf: PartialFunction[Throwable, U]^{any.except[ThreadLocal]})(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Future[U]^{this, pf, executor, ExecutionContext} = { + val state = getRef + if (!state.isInstanceOf[Success[?]]) dispatchOrAddCallbacks(state, Transformation[T, U](Xform_recover, pf, executor)) // Short-circuit if we get a Failure else this.asInstanceOf[Future[U]] } - override final def mapTo[S](implicit tag: scala.reflect.ClassTag[S]): Future[S] = + override final def mapTo[S](implicit tag: scala.reflect.ClassTag[S]): Future[S]^{this, ExecutionContext} = if (!get().isInstanceOf[Failure[?]]) super[Future].mapTo[S](using tag) // Short-circuit if we get a Success else this.asInstanceOf[Future[S]] - override final def onComplete[U](func: Try[T] => U)(implicit executor: ExecutionContext): Unit = - dispatchOrAddCallbacks(get(), new Transformation[T, Unit](Xform_onComplete, func, executor)) + override final def onComplete[U](func: Try[T] ->{any.except[ThreadLocal]} U)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): Unit = + dispatchOrAddCallbacks(getRef, Transformation[T, Unit](Xform_onComplete, func, executor)) /** The same as [[onComplete]], but additionally returns a function which can be * invoked to unregister the callback function. Removing a callback from a long-lived @@ -237,9 +245,9 @@ private[concurrent] object Promise { * @param executor the `ExecutionContext` used to run the callback * @return a function which, when invoked, unregisters the callback so it will not be called when this future completes */ - private[concurrent] final def onCompleteWithUnregister[U](func: Try[T] => U)(implicit executor: ExecutionContext): () => Unit = { - val t = new Transformation[T, Unit](Xform_onComplete, func, executor) - dispatchOrAddCallbacks(get(), t) + private[concurrent] final def onCompleteWithUnregister[U](func: Try[T] ->{any.except[ThreadLocal]} U)(implicit executor: ExecutionContext^{any.except[ThreadLocal]}): () ->{this, func, executor, ExecutionContext} Unit = { + val t = Transformation[T, Unit](Xform_onComplete, func, executor) + dispatchOrAddCallbacks(getRef, t) () => unregisterCallback(t) } @@ -248,7 +256,7 @@ private[concurrent] object Promise { else Future.failedFailureFuture // Cached instance in case of already known success @tailrec override final def toString(): String = { - val state = get() + val state = getRef if (state.isInstanceOf[Try[?]]) "Future("+state+")" else if (state.isInstanceOf[Link[?]]) state.asInstanceOf[Link[T]].promise(this).toString else /*if (state.isInstanceOf[Callbacks[T]]) */ "Future()" @@ -294,33 +302,33 @@ private[concurrent] object Promise { @tailrec // returns null if not completed private final def value0: Try[T] | Null = { - val state = get() + val state = getRef if (state.isInstanceOf[Try[?]]) state.asInstanceOf[Try[T]] else if (state.isInstanceOf[Link[?]]) state.asInstanceOf[Link[T]].promise(this).value0 else /*if (state.isInstanceOf[Callbacks[T]])*/ null } override final def tryComplete(value: Try[T]): Boolean = { - val state = get() + val state = getRef if (state.isInstanceOf[Try[?]]) false else tryComplete0(state, resolve(value)) } @tailrec // WARNING: important that the supplied Try really is resolve():d - private[Promise] final def tryComplete0(state: AnyRef, resolved: Try[T]): Boolean = + private[Promise] final def tryComplete0(state: AnyRef^, resolved: Try[T]): Boolean = if (state.isInstanceOf[Callbacks[?]]) { if (compareAndSet(state, resolved)) { if (state ne Noop) submitWithValue(state.asInstanceOf[Callbacks[T]], resolved) true - } else tryComplete0(get(), resolved) + } else tryComplete0(getRef, resolved) } else if (state.isInstanceOf[Link[?]]) { - val p = state.asInstanceOf[Link[T]].promise(this) // If this returns owner/this, we are in a completed link - (p ne this) && p.tryComplete0(p.get(), resolved) // Use this to get tailcall optimization and avoid re-resolution + val p = state.asInstanceOf[Link[T]^{state}].promise(this) // If this returns owner/this, we are in a completed link + (p ne this) && p.tryComplete0(p.getRef, resolved) // Use this to get tailcall optimization and avoid re-resolution } else /* if(state.isInstanceOf[Try[T]]) */ false - override final def completeWith(other: Future[T]): this.type = { + override final def completeWith(other: Future[T]^): this.type = { if (other ne this) { - val state = get() + val state = getRef if (!state.isInstanceOf[Try[?]]) { val resolved = if (other.isInstanceOf[DefaultPromise[?]]) other.asInstanceOf[DefaultPromise[T]].value0 else other.value.orNull if (resolved ne null) tryComplete0(state, resolved) @@ -340,20 +348,21 @@ private[concurrent] object Promise { * @param callbacks the callbacks to dispatch immediately if this promise is already completed, or to add to the pending list otherwise * @return the `callbacks` argument, returned unchanged to allow chaining */ - @tailrec private final def dispatchOrAddCallbacks[C <: Callbacks[T]](state: AnyRef, callbacks: C): C = + @tailrec private final def dispatchOrAddCallbacks[Cap^, C <: Callbacks[T]](state: AnyRef^, callbacks: C^{Cap}): C^{this, state, Cap} = if (state.isInstanceOf[Try[?]]) { submitWithValue(callbacks, state.asInstanceOf[Try[T]]) // invariant: callbacks should never be Noop here callbacks } else if (state.isInstanceOf[Callbacks[?]]) { - if(compareAndSet(state, if (state ne Noop) concatCallbacks(callbacks, state.asInstanceOf[Callbacks[T]]) else callbacks)) callbacks - else dispatchOrAddCallbacks(get(), callbacks) + val newRef = if (state ne Noop) concatCallbacks(callbacks, state.asInstanceOf[Callbacks[T]^{state}]) else callbacks + if(compareAndSet(state, newRef)) callbacks + else dispatchOrAddCallbacks(getRef, callbacks) } else /*if (state.isInstanceOf[Link[T]])*/ { - val p = state.asInstanceOf[Link[T]].promise(this) - p.dispatchOrAddCallbacks(p.get(), callbacks) + val p = state.asInstanceOf[Link[T]^{state}].promise(this) + p.dispatchOrAddCallbacks(p.getRef, callbacks) } - @tailrec private def unregisterCallback(t: Transformation[?, ?]): Unit = { - val state = get() + @tailrec private def unregisterCallback(t: Transformation[?, ?]^): Unit = { + val state = getRef if (state eq t) { if (!compareAndSet(state, Noop)) unregisterCallback(t) } else if (state.isInstanceOf[ManyCallbacks[?]]) { @@ -362,20 +371,20 @@ private[concurrent] object Promise { } // IMPORTANT: Noop should never be passed in here, neither as left OR as right - @tailrec private final def concatCallbacks(left: Callbacks[T], right: Callbacks[T]): Callbacks[T] = + @tailrec private final def concatCallbacks(left: Callbacks[T]^, right: Callbacks[T]^): Callbacks[T]^{left, right} = if (left.isInstanceOf[Transformation[T, ?]]) new ManyCallbacks[T](left.asInstanceOf[Transformation[T, ?]], right) else /*if (left.isInstanceOf[ManyCallbacks[T]) */ { // This should only happen when linking - val m = left.asInstanceOf[ManyCallbacks[T]] + val m = left.asInstanceOf[ManyCallbacks[T]^{left}] concatCallbacks(m.rest, new ManyCallbacks(m.first, right)) } - @tailrec private final def removeCallback(cs: Callbacks[T], t: Transformation[?, ?], result: Callbacks[T] | Null = null): AnyRef = + @tailrec private final def removeCallback(cs: Callbacks[T]^, t: Transformation[?, ?]^, result: (Callbacks[T]^) | Null = null): AnyRef^{result, cs} = if (cs eq t) { if (result == null) Noop else result } else if (cs.isInstanceOf[ManyCallbacks[?]]) { - val m = cs.asInstanceOf[ManyCallbacks[T]] + val m = cs.asInstanceOf[ManyCallbacks[T]^{cs}] if (m.first eq t) { if (result == null) m.rest else concatCallbacks(m.rest, result) @@ -385,7 +394,7 @@ private[concurrent] object Promise { // IMPORTANT: Noop should not be passed in here, `callbacks` cannot be null @tailrec - private final def submitWithValue(callbacks: Callbacks[T], resolved: Try[T]): Unit = + private final def submitWithValue(callbacks: Callbacks[T]^, resolved: Try[T]): Unit = if(callbacks.isInstanceOf[ManyCallbacks[T]]) { val m: ManyCallbacks[T] = callbacks.asInstanceOf[ManyCallbacks[T]] m.first.submitWithValue(resolved) @@ -399,17 +408,17 @@ private[concurrent] object Promise { * @param target the `DefaultPromise` to link this promise's root to * @param link a reusable `Link` instance for the connection, or null to create a new one */ - @tailrec private[concurrent] final def linkRootOf(target: DefaultPromise[T], link: Link[T] | Null): Unit = + @tailrec private[concurrent] final def linkRootOf(target: DefaultPromise[T]^, link: (Link[T]^) | Null): Unit = if (this ne target) { - val state = get() + val state = getRef if (state.isInstanceOf[Try[?]]) { - if(!target.tryComplete0(target.get(), state.asInstanceOf[Try[T]])) + if(!target.tryComplete0(target.getRef, state.asInstanceOf[Try[T]])) throw new IllegalStateException("Cannot link completed promises together") } else if (state.isInstanceOf[Callbacks[?]]) { val l = if (link ne null) link else new Link(target) val p = l.promise(this) if ((this ne p) && compareAndSet(state, l)) { - if (state ne Noop) p.dispatchOrAddCallbacks(p.get(), state.asInstanceOf[Callbacks[T]]) // Noop-check is important here + if (state ne Noop) p.dispatchOrAddCallbacks(p.getRef, state.asInstanceOf[Callbacks[T]]) // Noop-check is important here } else linkRootOf(p, l) } else /* if (state.isInstanceOf[Link[T]]) */ state.asInstanceOf[Link[T]].promise(this).linkRootOf(target, link) @@ -421,11 +430,12 @@ private[concurrent] object Promise { * @param resolved the already-resolved result to complete all promises in the link chain with */ @tailrec private[concurrent] final def unlink(resolved: Try[T]): Unit = { - val state = get() + val state = getRef if (state.isInstanceOf[Link[?]]) { - val next = if (compareAndSet(state, resolved)) state.asInstanceOf[Link[T]].get() else this + val next = if (compareAndSet(state, resolved)) state.asInstanceOf[Link[T]^{state}].getPromise else this next.unlink(resolved) - } else tryComplete0(state, resolved) + } else + tryComplete0(state, resolved) } @throws[IOException] @@ -436,8 +446,17 @@ private[concurrent] object Promise { @throws[ClassNotFoundException] private def readObject(in: ObjectInputStream): Unit = throw new NotSerializableException("Promises and Futures cannot be deserialized") + + private[impl] def getRef: Try[T] | (Link[T]^{this}) | (Callbacks[T]^{this}) = get().asInstanceOf[Try[T] | (Link[T]^{this}) | (Callbacks[T]^{this})] } + // SAFETY: already completed promises need not any captures + object DefaultPromise: + private[concurrent] def apply[T](): DefaultPromise[T] = + caps.unsafe.unsafeAssumePure(new DefaultPromise[T](Noop: AnyRef)) + private[concurrent] def apply[T](t: Try[T]): DefaultPromise[T] = + caps.unsafe.unsafeAssumePure(new DefaultPromise[T](t: AnyRef)) + // Constant byte tags for unpacking transformation function inputs or outputs // These need to be Ints to get compiled into constants. final val Xform_noop = 0 @@ -456,11 +475,16 @@ private[concurrent] object Promise { */ sealed trait Callbacks[-T] - final class ManyCallbacks[-T](final val first: Transformation[T, ?], final val rest: Callbacks[T]) extends Callbacks[T] { + final class ManyCallbacks[-T](_first: Transformation[T, ?]^, _rest: Callbacks[T]^) extends Callbacks[T] { + final val first: Transformation[T, ?]^{this} = _first + final val rest: Callbacks[T]^{this} = _rest override final def toString(): String = "ManyCallbacks" } - private final val Noop = new Transformation[Nothing, Nothing](Xform_noop, null: (Any => Any) | Null, ExecutionContext.parasitic) + private final val Noop: Transformation[Nothing, Nothing] = + caps.unsafe.unsafeAssumePure: + Transformation[Nothing, Nothing](Xform_noop, null: (Any -> Any) | Null, ExecutionContext.parasitic) + /** A Transformation[F, T] receives an F (it is a Callback[F]) and applies a transformation function to that F, * Producing a value of type T (it is a Promise[T]). @@ -471,13 +495,13 @@ private[concurrent] object Promise { * @tparam T the output type (the value type produced by the transformation) */ final class Transformation[-F, T] private ( - @annotation.stableNull private final var _fun: (Any => Any) | Null, - @annotation.stableNull private final var _ec: ExecutionContext | Null, + @annotation.stableNull private final var _fun: (Any ->{any.except[ThreadLocal]} Any) | Null, + @annotation.stableNull private final var _ec: (ExecutionContext^{any.except[ThreadLocal]}) | Null, @annotation.stableNull private final var _arg: Try[F @uncheckedVariance] | Null, private final val _xform: Int - ) extends DefaultPromise[T]() with Callbacks[F] with Runnable with Batchable { - final def this(xform: Int, f: (? => ?) | Null, ec: ExecutionContext) = - this(f.asInstanceOf[(Any => Any) | Null], ec.prepare(): @nowarn("cat=deprecation"), null, xform) + ) extends DefaultPromise[T]() with Callbacks[F] with Runnable with Batchable uses ExecutionContext { + final def this(xform: Int, f: (? ->{any.except[ThreadLocal]} ?) | Null, ec: ExecutionContext^{any.except[ThreadLocal]}) = + this(f.asInstanceOf[(Any ->{f} Any) | Null], ec.prepare(): @nowarn("cat=deprecation"), null, xform) final def benefitsFromBatching: Boolean = _xform != Xform_onComplete && _xform != Xform_foreach @@ -488,7 +512,7 @@ private[concurrent] object Promise { final def submitWithValue(resolved: Try[F]): this.type = { _arg = resolved val e = _ec - try e.nn.execute(this) /* Safe publication of _arg, _fun, _ec */ + try e.nn.execute(caps.unsafe.unsafeAssumePure(this)) /* Safe publication of _arg, _fun, _ec */ catch { case t: Throwable => _fun = null // allow to GC @@ -503,7 +527,7 @@ private[concurrent] object Promise { private final def handleFailure(t: Throwable, e: ExecutionContext): Unit = { val wasInterrupted = t.isInstanceOf[InterruptedException] if (wasInterrupted || NonFatal(t)) { - val completed = tryComplete0(get(), resolve(Failure(t))) + val completed = tryComplete0(getRef, resolve(Failure(t))) if (completed && wasInterrupted) Thread.currentThread.interrupt() // Report or rethrow failures which are unlikely to otherwise be noticed @@ -563,10 +587,14 @@ private[concurrent] object Promise { Failure(new IllegalStateException("BUG: encountered transformation promise with illegal type: " + _xform)) // Safe not to `resolve` } if (resolvedResult ne null) - tryComplete0(get(), resolvedResult.asInstanceOf[Try[T]]) // T is erased anyway so we won't have any use for it above + tryComplete0(getRef, resolvedResult.asInstanceOf[Try[T]]) // T is erased anyway so we won't have any use for it above } catch { case t: Throwable => handleFailure(t, ec) } } } + + object Transformation: + private[impl] def apply[F, T](xform: Int, f: (? ->{any.except[ThreadLocal]} ?) | Null, ec: ExecutionContext^{any.except[ThreadLocal]}): Transformation[F, T]^{f, ec, ExecutionContext} = + new Transformation[F, T](f.asInstanceOf[(Any ->{f} Any) | Null], ec.prepare(): @nowarn("cat=deprecation"), null, xform) } diff --git a/project/Build.scala b/project/Build.scala index e440f6207742..9c29bb2eaa3c 100644 --- a/project/Build.scala +++ b/project/Build.scala @@ -42,6 +42,7 @@ object Build { import ScaladocConfigs._ import Versions._ + // Run tests with filter through vulpix test suite val testCompilation = inputKey[Unit]("runs integration test with the supplied filter") @@ -69,6 +70,8 @@ object Build { organizationName := "LAMP/EPFL", organizationHomepage := Some(url("http://lamp.epfl.ch")), + resolvers += Resolver.scalaNightlyRepository, + // Note: bench/profiles/projects.yml should be updated accordingly. scalacOptions ++= Seq( "-feature", @@ -983,6 +986,7 @@ object Build { Compile / compile / scalacOptions ++= Seq( // Needed so that the library sources are visible when `dotty.tools.dotc.core.Definitions#init` is called "-sourcepath", (Compile / sourceDirectories).value.map(_.getCanonicalPath).distinct.mkString(File.pathSeparator), + "-explain", ), // Packaging configuration of the stdlib Compile / publishArtifact := true, diff --git a/project/Versions.scala b/project/Versions.scala index 1088afdebd20..bb8c11994548 100644 --- a/project/Versions.scala +++ b/project/Versions.scala @@ -12,7 +12,7 @@ object Versions { * * Warning: Change of this variable needs to be consulted with `expectedTastyVersion` */ - val referenceVersion = "3.9.0-RC1" + val referenceVersion = "3.10.0-RC1-bin-20260716-61591f7-NIGHTLY" /** Version of the Scala compiler targeted in the current release cycle * Contains a version without RC/SNAPSHOT/NIGHTLY specific suffixes