diff --git a/akka-actor-tests/src/test/scala/akka/dispatch/FutureSpec.scala b/akka-actor-tests/src/test/scala/akka/dispatch/FutureSpec.scala index e058218f2d..6163123632 100644 --- a/akka-actor-tests/src/test/scala/akka/dispatch/FutureSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/dispatch/FutureSpec.scala @@ -860,7 +860,7 @@ class FutureSpec extends AkkaSpec with Checkers with BeforeAndAfterAll with Defa val l1, l2 = new TestLatch val complex = Future() map { _ ⇒ - Future.blocking(system.dispatcher) + Future.blocking() val nested = Future(()) nested foreach (_ ⇒ l1.open()) Await.ready(l1, TestLatch.DefaultTimeout) // make sure nested is completed diff --git a/akka-actor/src/main/scala/akka/dispatch/Future.scala b/akka-actor/src/main/scala/akka/dispatch/Future.scala index 97ff17c075..39e9f27d26 100644 --- a/akka-actor/src/main/scala/akka/dispatch/Future.scala +++ b/akka-actor/src/main/scala/akka/dispatch/Future.scala @@ -151,6 +151,26 @@ object Futures { for (r ← fr; b ← fb) yield { r add b; r } } } + + /** + * Signals that the current thread of execution will potentially engage + * in blocking calls after the call to this method, giving the system a + * chance to spawn new threads, reuse old threads or otherwise, to prevent + * starvation and/or unfairness. + * + * Assures that any Future tasks initiated in the current thread will be + * executed asynchronously, including any tasks currently queued to be + * executed in the current thread. This is needed if the current task may + * block, causing delays in executing the remaining tasks which in some + * cases may cause a deadlock. + * + * Usage: Call this method in a callback (map, flatMap etc also count) to a Future, + * if you will be doing blocking in the callback. + * + * Note: Calling 'Await.result(future)' or 'Await.ready(future)' will automatically trigger this method. + * + */ + def blocking(): Unit = Future.blocking() } object Future { @@ -317,17 +337,22 @@ object Future { * } * */ - def blocking(implicit executor: ExecutionContext): Unit = + def blocking(): Unit = _taskStack.get match { case stack if (stack ne null) && stack.nonEmpty ⇒ + val executionContext = _executionContext.get match { + case null ⇒ throw new IllegalStateException("'blocking' needs to be invoked inside a Future callback.") + case some ⇒ some + } val tasks = stack.elems stack.clear() _taskStack.remove() - dispatchTask(() ⇒ _taskStack.get.elems = tasks, true) + dispatchTask(() ⇒ _taskStack.get.elems = tasks, true)(executionContext) case _ ⇒ _taskStack.remove() } private val _taskStack = new ThreadLocal[Stack[() ⇒ Unit]]() + private val _executionContext = new ThreadLocal[ExecutionContext]() /** * Internal API, do not call @@ -339,7 +364,7 @@ object Future { new Runnable { def run = try { - + _executionContext set executor val taskStack = Stack.empty[() ⇒ Unit] taskStack push task _taskStack set taskStack @@ -352,7 +377,10 @@ object Future { case NonFatal(e) ⇒ executor.reportFailure(e) } } - } finally { _taskStack.remove() } + } finally { + _executionContext.remove() + _taskStack.remove() + } }) } diff --git a/akka-docs/java/code/akka/docs/future/FutureDocTestBase.java b/akka-docs/java/code/akka/docs/future/FutureDocTestBase.java index b064eb803b..8fc3b29b4e 100644 --- a/akka-docs/java/code/akka/docs/future/FutureDocTestBase.java +++ b/akka-docs/java/code/akka/docs/future/FutureDocTestBase.java @@ -381,6 +381,7 @@ public class FutureDocTestBase { @Test public void useOnSuccessOnFailureAndOnComplete() { { Future future = Futures.successful("foo", system.dispatcher()); + //#onSuccess future.onSuccess(new OnSuccess() { public void onSuccess(String result) { diff --git a/akka-remote/src/main/scala/akka/remote/netty/Settings.scala b/akka-remote/src/main/scala/akka/remote/netty/Settings.scala index 0db6cabf18..daa91a3014 100644 --- a/akka-remote/src/main/scala/akka/remote/netty/Settings.scala +++ b/akka-remote/src/main/scala/akka/remote/netty/Settings.scala @@ -40,7 +40,7 @@ class NettySettings(config: Config, val systemName: String) { case value ⇒ value } - @deprecated("WARNING: This should only be used by professionals.") + @deprecated("WARNING: This should only be used by professionals.", "2.0") val PortSelector = getInt("port") val ConnectionTimeout = Duration(getMilliseconds("connection-timeout"), MILLISECONDS)