diff options
6 files changed, 48 insertions, 19 deletions
diff --git a/src/library/scala/concurrent/ExecutionContext.scala b/src/library/scala/concurrent/ExecutionContext.scala index 078b05c517..2650022e1e 100644 --- a/src/library/scala/concurrent/ExecutionContext.scala +++ b/src/library/scala/concurrent/ExecutionContext.scala @@ -30,8 +30,9 @@ trait ExecutionContext { def future[T](body: => T): Future[T] - /** Only callable from the tasks running on the same execution context. */ - def blockingCall[T](body: Awaitable[T]): T + def blocking[T](atMost: Duration)(body: =>T): T + + def blocking[T](atMost: Duration)(awaitable: Awaitable[T]): T } diff --git a/src/library/scala/concurrent/akka/ExecutionContextImpl.scala b/src/library/scala/concurrent/akka/ExecutionContextImpl.scala index 28638d1247..1cff7211f3 100644 --- a/src/library/scala/concurrent/akka/ExecutionContextImpl.scala +++ b/src/library/scala/concurrent/akka/ExecutionContextImpl.scala @@ -44,8 +44,17 @@ class ExecutionContextImpl(executorService: ExecutorService) extends ExecutionCo p.future } + def blocking[T](atMost: Duration)(body: =>T): T = blocking(body2awaitable(body), atMost) + + def blocking[T](atMost: Duration)(awaitable: Awaitable[T]): T = { + currentExecutionContext.get match { + case null => awaitable.await(atMost)(null) // outside - TODO - fix timeout case + case x => x.blockingCall(awaitable) // inside an execution context thread + } + } + /** Only callable from the tasks running on the same execution context. */ - def blockingCall[T](body: Awaitable[T]): T = { + private def blockingCall[T](body: Awaitable[T]): T = { releaseStack() // TODO see what to do with timeout diff --git a/src/library/scala/concurrent/akka/Future.scala b/src/library/scala/concurrent/akka/Future.scala index c48009554c..5036768d36 100644 --- a/src/library/scala/concurrent/akka/Future.scala +++ b/src/library/scala/concurrent/akka/Future.scala @@ -11,7 +11,7 @@ package scala.concurrent.akka import scala.concurrent.{Awaitable, ExecutionContext} -import scala.util.continuations._ +//import scala.util.continuations._ @@ -24,7 +24,7 @@ trait Future[+T] extends scala.concurrent.Future[T] with Awaitable[T] { * Returns the result of this Future without blocking, by suspending execution and storing it as a * continuation until the result is available. */ - def apply(): T @cps[Future[Any]] = shift(this flatMap (_: T => Future[Any])) + //def apply(): T @cps[Future[Any]] = shift(this flatMap (_: T => Future[Any])) /** Tests whether this Future has been completed. */ diff --git a/src/library/scala/concurrent/akka/Promise.scala b/src/library/scala/concurrent/akka/Promise.scala index 3b77f14f70..e7b6b50aeb 100644 --- a/src/library/scala/concurrent/akka/Promise.scala +++ b/src/library/scala/concurrent/akka/Promise.scala @@ -13,7 +13,7 @@ package scala.concurrent.akka import java.util.concurrent.TimeUnit.{ NANOSECONDS, MILLISECONDS } import java.util.concurrent.atomic.AtomicReferenceFieldUpdater import scala.concurrent.{Awaitable, ExecutionContext, resolve, resolver, blocking, CanAwait, TimeoutException} -import scala.util.continuations._ +//import scala.util.continuations._ import scala.util.Duration import scala.annotation.tailrec @@ -25,7 +25,7 @@ trait Promise[T] extends scala.concurrent.Promise[T] with Future[T] { // TODO refine answer and return types here from Any to type parameters // then move this up in the hierarchy - + /* final def <<(value: T): Future[T] @cps[Future[Any]] = shift { cont: (Future[T] => Future[Any]) => cont(complete(Right(value))) @@ -47,7 +47,7 @@ trait Promise[T] extends scala.concurrent.Promise[T] with Future[T] { p.future } - + */ // TODO finish this once we introduce something like dataflow streams /* diff --git a/src/library/scala/concurrent/default/TaskImpl.scala b/src/library/scala/concurrent/default/TaskImpl.scala index a38541df5d..4bddb740ef 100644 --- a/src/library/scala/concurrent/default/TaskImpl.scala +++ b/src/library/scala/concurrent/default/TaskImpl.scala @@ -257,7 +257,16 @@ private[concurrent] final class ExecutionContextImpl extends ExecutionContext { def promise[T]: Promise[T] = new PromiseImpl[T](this) - def blockingCall[T](b: Awaitable[T]): T = b match { + def blocking[T](atMost: Duration)(body: =>T): T = blocking(body2awaitable(body), atMost) + + def blocking[T](atMost: Duration)(awaitable: Awaitable[T]): T = { + currentExecutionContext.get match { + case null => awaitable.await(atMost)(null) // outside - TODO - fix timeout case + case x => x.blockingCall(awaitable) // inside an execution context thread + } + } + + private def blockingCall[T](b: Awaitable[T]): T = b match { case fj: TaskImpl[_] if fj.executor.pool eq pool => fj.await(Duration.fromNanos(0)) case _ => diff --git a/src/library/scala/concurrent/package.scala b/src/library/scala/concurrent/package.scala index 7552100af2..d40eb4e2a1 100644 --- a/src/library/scala/concurrent/package.scala +++ b/src/library/scala/concurrent/package.scala @@ -76,6 +76,11 @@ package object concurrent { def promise[T]()(implicit execCtx: ExecutionContext = executionContext): Promise[T] = execCtx promise + /** Wraps a block of code into an awaitable object. */ + def body2awaitable[T](body: =>T) = new Awaitable[T] { + def await(atMost: Duration)(implicit cb: CanAwait) = body + } + /** Used to block on a piece of code which potentially blocks. * * @param body A piece of code which contains potentially blocking or long running calls. @@ -85,14 +90,10 @@ package object concurrent { * - InterruptedException - in the case that a wait within the blockable object was interrupted * - TimeoutException - in the case that the blockable object timed out */ - def blocking[T](atMost: Duration)(body: =>T): T = blocking(body2awaitable(body), atMost) + def blocking[T](atMost: Duration)(body: =>T)(implicit execCtx: ExecutionContext = executionContext): T = + executionContext.blocking(atMost)(body) - /** Wraps a block of code into an awaitable object. */ - def body2awaitable[T](body: =>T) = new Awaitable[T] { - def await(atMost: Duration)(implicit cb: CanAwait) = body - } - - /** Blocks on a blockable object. + /** Blocks on an awaitable object. * * @param awaitable An object with a `block` method which runs potentially blocking or long running calls. * @@ -101,23 +102,32 @@ package object concurrent { * - InterruptedException - in the case that a wait within the blockable object was interrupted * - TimeoutException - in the case that the blockable object timed out */ + def blocking[T](atMost: Duration)(awaitable: Awaitable[T])(implicit execCtx: ExecutionContext = executionContext): T = + executionContext.blocking(atMost)(awaitable) + + /* + def blocking[T](atMost: Duration)(body: =>T): T = blocking(body2awaitable(body), atMost) + def blocking[T](awaitable: Awaitable[T], atMost: Duration): T = { currentExecutionContext.get match { case null => awaitable.await(atMost)(null) // outside - TODO - fix timeout case case x => x.blockingCall(awaitable) // inside an execution context thread } } + */ object await { - def ready[T](awaitable: Awaitable[T], atMost: Duration): Awaitable[T] = { + def ready[T](atMost: Duration)(awaitable: Awaitable[T])(implicit execCtx: ExecutionContext = executionContext): Awaitable[T] = { try blocking(awaitable, atMost) catch { case _ => } awaitable } - def result[T](awaitable: Awaitable[T], atMost: Duration): T = blocking(awaitable, atMost) + def result[T](atMost: Duration)(awaitable: Awaitable[T])(implicit execCtx: ExecutionContext = executionContext): T = { + blocking(awaitable, atMost) + } } - + } |