diff --git a/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala index 8c949f8776..46bf609c7a 100644 --- a/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala @@ -50,6 +50,8 @@ object ActorModelSpec { case object Restart extends ActorModelMessage + case object DoubleStop extends ActorModelMessage + case class ThrowException(e: Throwable) extends ActorModelMessage val Ping = "Ping" @@ -86,6 +88,7 @@ object ActorModelSpec { case Restart ⇒ ack; busy.switchOff(); throw new Exception("Restart requested") case Interrupt ⇒ ack; sender ! Status.Failure(new ActorInterruptedException(new InterruptedException("Ping!"))); busy.switchOff(); throw new InterruptedException("Ping!") case ThrowException(e: Throwable) ⇒ ack; busy.switchOff(); throw e + case DoubleStop ⇒ ack; context.stop(self); context.stop(self); busy.switchOff } } @@ -190,13 +193,13 @@ object ActorModelSpec { } def assertRef(actorRef: ActorRef, dispatcher: MessageDispatcher = null)( - suspensions: Long = statsFor(actorRef).suspensions.get(), - resumes: Long = statsFor(actorRef).resumes.get(), - registers: Long = statsFor(actorRef).registers.get(), - unregisters: Long = statsFor(actorRef).unregisters.get(), - msgsReceived: Long = statsFor(actorRef).msgsReceived.get(), - msgsProcessed: Long = statsFor(actorRef).msgsProcessed.get(), - restarts: Long = statsFor(actorRef).restarts.get())(implicit system: ActorSystem) { + suspensions: Long = statsFor(actorRef, dispatcher).suspensions.get(), + resumes: Long = statsFor(actorRef, dispatcher).resumes.get(), + registers: Long = statsFor(actorRef, dispatcher).registers.get(), + unregisters: Long = statsFor(actorRef, dispatcher).unregisters.get(), + msgsReceived: Long = statsFor(actorRef, dispatcher).msgsReceived.get(), + msgsProcessed: Long = statsFor(actorRef, dispatcher).msgsProcessed.get(), + restarts: Long = statsFor(actorRef, dispatcher).restarts.get())(implicit system: ActorSystem) { val stats = statsFor(actorRef, Option(dispatcher).getOrElse(actorRef.asInstanceOf[LocalActorRef].underlying.dispatcher)) val deadline = System.currentTimeMillis + 1000 try { @@ -426,6 +429,14 @@ abstract class ActorModelSpec(config: String) extends AkkaSpec(config) with Defa assert(f5.value.isEmpty) } } + + "not double-deregister" in { + implicit val dispatcher = interceptedDispatcher() + val a = newTestActor(dispatcher.id) + a ! DoubleStop + awaitCond(statsFor(a, dispatcher).registers.get == 1) + awaitCond(statsFor(a, dispatcher).unregisters.get == 1) + } } } diff --git a/akka-actor-tests/src/test/scala/akka/routing/ResizerSpec.scala b/akka-actor-tests/src/test/scala/akka/routing/ResizerSpec.scala index 863922491b..2130afe107 100644 --- a/akka-actor-tests/src/test/scala/akka/routing/ResizerSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/routing/ResizerSpec.scala @@ -24,6 +24,9 @@ object ResizerSpec { } } } + bal-disp { + type = BalancingDispatcher + } """ class TestActor extends Actor { @@ -133,7 +136,7 @@ class ResizerSpec extends AkkaSpec(ResizerSpec.config) with DefaultTimeout with pressureThreshold = 0, messagesPerResize = 1) - val router = system.actorOf(Props[BusyActor].withRouter(RoundRobinRouter(resizer = Some(resizer)))) + val router = system.actorOf(Props[BusyActor].withRouter(RoundRobinRouter(resizer = Some(resizer))).withDispatcher("bal-disp")) val latch1 = new TestLatch(1) router ! (latch1, busy) @@ -179,10 +182,10 @@ class ResizerSpec extends AkkaSpec(ResizerSpec.config) with DefaultTimeout with Await.result(router ? CurrentRoutees, 5 seconds).asInstanceOf[RouterRoutees].routees.size must be(2) def loop(loops: Int, t: Int, latch: TestLatch, count: AtomicInteger) = { - (10 millis).dilated.sleep + (100 millis).dilated.sleep for (m ← 0 until loops) { router.!((t, latch, count)) - (10 millis).dilated.sleep + (100 millis).dilated.sleep } } @@ -198,7 +201,7 @@ class ResizerSpec extends AkkaSpec(ResizerSpec.config) with DefaultTimeout with // a whole bunch should max it out val count2 = new AtomicInteger val latch2 = TestLatch(10) - loop(10, 200, latch2, count2) + loop(10, 500, latch2, count2) Await.ready(latch2, TestLatch.DefaultTimeout) count2.get must be(10) @@ -238,7 +241,7 @@ class ResizerSpec extends AkkaSpec(ResizerSpec.config) with DefaultTimeout with // let it cool down for (m ← 0 to 3) { router ! 1 - (200 millis).dilated.sleep + (500 millis).dilated.sleep } Await.result(router ? CurrentRoutees, 5 seconds).asInstanceOf[RouterRoutees].routees.size must be < (z) diff --git a/akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala b/akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala index 9de51bdabf..ad3702d556 100644 --- a/akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala @@ -15,6 +15,7 @@ import com.typesafe.config.ConfigFactory import akka.pattern.ask import java.util.concurrent.ConcurrentHashMap import com.typesafe.config.Config +import akka.dispatch.Dispatchers object RoutingSpec { @@ -51,6 +52,7 @@ object RoutingSpec { case (sender, message) ⇒ Nil } } + def routerDispatcher: String = Dispatchers.DefaultDispatcherId } } @@ -539,6 +541,8 @@ class RoutingSpec extends AkkaSpec(RoutingSpec.config) with DefaultTimeout with //#crRouter case class VoteCountRouter() extends RouterConfig { + def routerDispatcher: String = Dispatchers.DefaultDispatcherId + //#crRoute def createRoute(routeeProps: Props, routeeProvider: RouteeProvider): Route = { val democratActor = routeeProvider.context.actorOf(Props(new DemocratActor()), "d") diff --git a/akka-actor/src/main/scala/akka/dispatch/Future.scala b/akka-actor/src/main/scala/akka/dispatch/Future.scala index 39e9f27d26..d6f4751d6e 100644 --- a/akka-actor/src/main/scala/akka/dispatch/Future.scala +++ b/akka-actor/src/main/scala/akka/dispatch/Future.scala @@ -20,6 +20,7 @@ import akka.event.Logging.Debug import java.util.concurrent.TimeUnit.NANOSECONDS import java.util.concurrent.{ ExecutionException, Callable, TimeoutException } import java.util.concurrent.atomic.{ AtomicInteger, AtomicReferenceFieldUpdater } +import akka.pattern.AskTimeoutException object Await { @@ -823,8 +824,9 @@ class DefaultPromise[T](implicit val executor: ExecutionContext) extends Abstrac def result(atMost: Duration)(implicit permit: CanAwait): T = ready(atMost).value.get match { - case Left(e) ⇒ throw e - case Right(r) ⇒ r + case Left(e: AskTimeoutException) ⇒ throw new AskTimeoutException(e.getMessage, e) // to get meaningful stack trace + case Left(e) ⇒ throw e + case Right(r) ⇒ r } def value: Option[Either[Throwable, T]] = getState match { diff --git a/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala b/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala index 27853b49db..cc15ae2173 100644 --- a/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala +++ b/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala @@ -189,7 +189,7 @@ private[akka] abstract class Mailbox(val actor: ActorCell) extends MessageQueue final def processAllSystemMessages() { var nextMessage = systemDrain() try { - while (nextMessage ne null) { + while ((nextMessage ne null) && !isClosed) { if (debug) println(actor.self + " processing system message " + nextMessage + " with children " + actor.childrenRefs) actor systemInvoke nextMessage nextMessage = nextMessage.next diff --git a/akka-actor/src/main/scala/akka/routing/Routing.scala b/akka-actor/src/main/scala/akka/routing/Routing.scala index c200be0871..f83bca2db9 100644 --- a/akka-actor/src/main/scala/akka/routing/Routing.scala +++ b/akka-actor/src/main/scala/akka/routing/Routing.scala @@ -9,15 +9,14 @@ import akka.util.duration._ import akka.config.ConfigurationException import akka.pattern.pipe import akka.pattern.AskSupport - import com.typesafe.config.Config - import scala.collection.JavaConversions.iterableAsScalaIterable - import java.util.concurrent.atomic.{ AtomicLong, AtomicBoolean } import java.util.concurrent.TimeUnit - +import java.util.concurrent.locks.ReentrantLock import akka.jsr166y.ThreadLocalRandom +import akka.util.Unsafe +import akka.dispatch.Dispatchers /** * A RoutedActorRef is an ActorRef that has a set of connected ActorRef and it uses a Router to @@ -26,25 +25,88 @@ import akka.jsr166y.ThreadLocalRandom private[akka] class RoutedActorRef(_system: ActorSystemImpl, _props: Props, _supervisor: InternalActorRef, _path: ActorPath) extends LocalActorRef( _system, - _props.copy(creator = () ⇒ _props.routerConfig.createActor()), + _props.copy(creator = () ⇒ _props.routerConfig.createActor(), dispatcher = _props.routerConfig.routerDispatcher), _supervisor, _path) { - private val routeeProps = _props.copy(routerConfig = NoRouter) - private val resizeProgress = new AtomicBoolean + /* + * CAUTION: RoutedActorRef is PROBLEMATIC + * ====================================== + * + * We are constructing/assembling the children outside of the scope of the + * Router actor, inserting them in its childrenRef list, which is not at all + * synchronized. This is done exactly once at start-up, all other accesses + * are done from the Router actor. This means that the only thing which is + * really hairy is making sure that the Router does not touch its childrenRefs + * before we are done with them: lock the monitor of the actor cell (hence the + * override of newActorCell) and use that to block the Router constructor for + * as long as it takes to setup the RoutedActorRef itself. + */ + override def newActorCell( + system: ActorSystemImpl, + ref: InternalActorRef, + props: Props, + supervisor: InternalActorRef, + receiveTimeout: Option[Duration]): ActorCell = + { + val cell = super.newActorCell(system, ref, props, supervisor, receiveTimeout) + Unsafe.instance.monitorEnter(cell) + cell + } + + private[akka] val routerConfig = _props.routerConfig + private[akka] val routeeProps = _props.copy(routerConfig = NoRouter) + private[akka] val resizeInProgress = new AtomicBoolean private val resizeCounter = new AtomicLong @volatile private var _routees: IndexedSeq[ActorRef] = IndexedSeq.empty[ActorRef] // this MUST be initialized during createRoute def routees = _routees + @volatile + private var _routeeProvider: RouteeProvider = _ + def routeeProvider = _routeeProvider + + val route = + try { + _routeeProvider = routerConfig.createRouteeProvider(actorContext) + val r = routerConfig.createRoute(routeeProps, routeeProvider) + // initial resize, before message send + routerConfig.resizer foreach { r ⇒ + if (r.isTimeForResize(resizeCounter.getAndIncrement())) + r.resize(routeeProps, routeeProvider) + } + r + } finally { + assert(Thread.holdsLock(actorContext)) + Unsafe.instance.monitorExit(actorContext) // unblock Router’s constructor + } + + if (routerConfig.resizer.isEmpty && _routees.isEmpty) + throw new ActorInitializationException("router " + routerConfig + " did not register routees!") + + /* + * end of construction + */ + + def applyRoute(sender: ActorRef, message: Any): Iterable[Destination] = message match { + case _: AutoReceivedMessage ⇒ Destination(this, this) :: Nil + case Terminated(_) ⇒ Destination(this, this) :: Nil + case CurrentRoutees ⇒ + sender ! RouterRoutees(_routees) + Nil + case _ ⇒ + if (route.isDefinedAt(sender, message)) route(sender, message) + else Nil + } + /** * Adds the routees to existing routees. * Adds death watch of the routees so that they are removed when terminated. * Not thread safe, but intended to be called from protected points, such as * `RouterConfig.createRoute` and `Resizer.resize` */ - private[akka] def addRoutees(newRoutees: IndexedSeq[ActorRef]) { + private[akka] def addRoutees(newRoutees: IndexedSeq[ActorRef]): Unit = { _routees = _routees ++ newRoutees // subscribe to Terminated messages for all route destinations, to be handled by Router actor newRoutees foreach underlying.watch @@ -56,34 +118,11 @@ private[akka] class RoutedActorRef(_system: ActorSystemImpl, _props: Props, _sup * Not thread safe, but intended to be called from protected points, such as * `Resizer.resize` */ - private[akka] def removeRoutees(abandonedRoutees: IndexedSeq[ActorRef]) { + private[akka] def removeRoutees(abandonedRoutees: IndexedSeq[ActorRef]): Unit = { _routees = _routees diff abandonedRoutees abandonedRoutees foreach underlying.unwatch } - private val routeeProvider = _props.routerConfig.createRouteeProvider(actorContext) - val route = _props.routerConfig.createRoute(routeeProps, routeeProvider) - // initial resize, before message send - resize() - - def applyRoute(sender: ActorRef, message: Any): Iterable[Destination] = message match { - case _: AutoReceivedMessage ⇒ Nil - case Terminated(_) ⇒ Nil - case CurrentRoutees ⇒ - sender ! RouterRoutees(_routees) - Nil - case _ ⇒ - if (route.isDefinedAt(sender, message)) route(sender, message) - else Nil - } - - if (_props.routerConfig.resizer.isEmpty && _routees.isEmpty) - throw new ActorInitializationException("router " + _props.routerConfig + " did not register routees!") - - _routees match { - case x ⇒ _routees = x // volatile write to publish the route before sending messages - } - override def !(message: Any)(implicit sender: ActorRef = null): Unit = { resize() @@ -95,20 +134,15 @@ private[akka] class RoutedActorRef(_system: ActorSystemImpl, _props: Props, _sup } applyRoute(s, message) match { - case Nil ⇒ super.!(message)(s) - case refs ⇒ refs foreach (p ⇒ p.recipient.!(msg)(p.sender)) + case Destination(_, x) :: Nil if x eq this ⇒ super.!(message)(s) + case refs ⇒ refs foreach (p ⇒ p.recipient.!(msg)(p.sender)) } } - def resize() { - for (r ← _props.routerConfig.resizer) { - if (r.isTimeForResize(resizeCounter.getAndIncrement()) && resizeProgress.compareAndSet(false, true)) { - try { - r.resize(routeeProps, routeeProvider) - } finally { - resizeProgress.set(false) - } - } + def resize(): Unit = { + for (r ← routerConfig.resizer) { + if (r.isTimeForResize(resizeCounter.getAndIncrement()) && resizeInProgress.compareAndSet(false, true)) + super.!(Router.Resize) } } } @@ -139,6 +173,11 @@ trait RouterConfig { def createActor(): Router = new Router {} + /** + * Dispatcher ID to use for running the “head” actor, i.e. the [[akka.routing.Router]]. + */ + def routerDispatcher: String + /** * Overridable merge strategy, by default completely prefers “this” (i.e. no merge). */ @@ -246,13 +285,20 @@ trait CustomRoute { */ trait Router extends Actor { - val ref = self match { - case x: RoutedActorRef ⇒ x - case _ ⇒ throw new ActorInitializationException("Router actor can only be used in RoutedActorRef") + // make sure that we synchronize properly to get the childrenRefs into our CPU cache + val ref = context.synchronized { + self match { + case x: RoutedActorRef ⇒ x + case _ ⇒ throw new ActorInitializationException("Router actor can only be used in RoutedActorRef") + } } final def receive = ({ + case Router.Resize ⇒ + try ref.routerConfig.resizer foreach (_.resize(ref.routeeProps, ref.routeeProvider)) + finally assert(ref.resizeInProgress.getAndSet(false)) + case Terminated(child) ⇒ ref.removeRoutees(IndexedSeq(child)) if (ref.routees.isEmpty) context.stop(self) @@ -264,6 +310,10 @@ trait Router extends Actor { } } +private object Router { + case object Resize +} + /** * Used to broadcast a message to all connections in a router; only the * contained message will be forwarded, i.e. the `Broadcast(...)` @@ -302,6 +352,7 @@ case class Destination(sender: ActorRef, recipient: ActorRef) //TODO add @SerialVersionUID(1L) when SI-4804 is fixed case object NoRouter extends RouterConfig { def createRoute(props: Props, routeeProvider: RouteeProvider): Route = null + def routerDispatcher: String = "" override def withFallback(other: RouterConfig): RouterConfig = other } @@ -311,13 +362,17 @@ case object NoRouter extends RouterConfig { case object FromConfig extends RouterConfig { def createRoute(props: Props, routeeProvider: RouteeProvider): Route = throw new ConfigurationException("router " + routeeProvider.context.self + " needs external configuration from file (e.g. application.conf)") + def routerDispatcher: String = Dispatchers.DefaultDispatcherId } /** * Java API: Router configuration which has no default, i.e. external configuration is required. + * + * This can be used when the dispatcher to be used for the head Router needs to be configured + * (defaults to default-dispatcher). */ //TODO add @SerialVersionUID(1L) when SI-4804 is fixed -case class FromConfig() extends RouterConfig { +case class FromConfig(val routerDispatcher: String = Dispatchers.DefaultDispatcherId) extends RouterConfig { def createRoute(props: Props, routeeProvider: RouteeProvider): Route = throw new ConfigurationException("router " + routeeProvider.context.self + " needs external configuration from file (e.g. application.conf)") } @@ -348,7 +403,8 @@ object RoundRobinRouter { * using `actorFor` in [[akka.actor.ActorRefProvider]] */ //TODO add @SerialVersionUID(1L) when SI-4804 is fixed -case class RoundRobinRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, override val resizer: Option[Resizer] = None) +case class RoundRobinRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, override val resizer: Option[Resizer] = None, + val routerDispatcher: String = Dispatchers.DefaultDispatcherId) extends RouterConfig with RoundRobinLike { /** @@ -374,6 +430,11 @@ case class RoundRobinRouter(nrOfInstances: Int = 0, routees: Iterable[String] = * Java API */ def this(resizer: Resizer) = this(resizer = Some(resizer)) + + /** + * Java API for setting routerDispatcher + */ + def withDispatcher(dispatcherId: String) = copy(routerDispatcher = dispatcherId) } trait RoundRobinLike { this: RouterConfig ⇒ @@ -428,7 +489,8 @@ object RandomRouter { * using `actorFor` in [[akka.actor.ActorRefProvider]] */ //TODO add @SerialVersionUID(1L) when SI-4804 is fixed -case class RandomRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, override val resizer: Option[Resizer] = None) +case class RandomRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, override val resizer: Option[Resizer] = None, + val routerDispatcher: String = Dispatchers.DefaultDispatcherId) extends RouterConfig with RandomLike { /** @@ -454,6 +516,11 @@ case class RandomRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, * Java API */ def this(resizer: Resizer) = this(resizer = Some(resizer)) + + /** + * Java API for setting routerDispatcher + */ + def withDispatcher(dispatcherId: String) = copy(routerDispatcher = dispatcherId) } trait RandomLike { this: RouterConfig ⇒ @@ -514,7 +581,8 @@ object SmallestMailboxRouter { * using `actorFor` in [[akka.actor.ActorRefProvider]] */ //TODO add @SerialVersionUID(1L) when SI-4804 is fixed -case class SmallestMailboxRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, override val resizer: Option[Resizer] = None) +case class SmallestMailboxRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, override val resizer: Option[Resizer] = None, + val routerDispatcher: String = Dispatchers.DefaultDispatcherId) extends RouterConfig with SmallestMailboxLike { /** @@ -540,6 +608,11 @@ case class SmallestMailboxRouter(nrOfInstances: Int = 0, routees: Iterable[Strin * Java API */ def this(resizer: Resizer) = this(resizer = Some(resizer)) + + /** + * Java API for setting routerDispatcher + */ + def withDispatcher(dispatcherId: String) = copy(routerDispatcher = dispatcherId) } trait SmallestMailboxLike { this: RouterConfig ⇒ @@ -659,7 +732,8 @@ object BroadcastRouter { * using `actorFor` in [[akka.actor.ActorRefProvider]] */ //TODO add @SerialVersionUID(1L) when SI-4804 is fixed -case class BroadcastRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, override val resizer: Option[Resizer] = None) +case class BroadcastRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, override val resizer: Option[Resizer] = None, + val routerDispatcher: String = Dispatchers.DefaultDispatcherId) extends RouterConfig with BroadcastLike { /** @@ -686,6 +760,10 @@ case class BroadcastRouter(nrOfInstances: Int = 0, routees: Iterable[String] = N */ def this(resizer: Resizer) = this(resizer = Some(resizer)) + /** + * Java API for setting routerDispatcher + */ + def withDispatcher(dispatcherId: String) = copy(routerDispatcher = dispatcherId) } trait BroadcastLike { this: RouterConfig ⇒ @@ -732,7 +810,8 @@ object ScatterGatherFirstCompletedRouter { */ //TODO add @SerialVersionUID(1L) when SI-4804 is fixed case class ScatterGatherFirstCompletedRouter(nrOfInstances: Int = 0, routees: Iterable[String] = Nil, within: Duration, - override val resizer: Option[Resizer] = None) + override val resizer: Option[Resizer] = None, + val routerDispatcher: String = Dispatchers.DefaultDispatcherId) extends RouterConfig with ScatterGatherFirstCompletedLike { if (within <= Duration.Zero) throw new IllegalArgumentException( @@ -761,6 +840,11 @@ case class ScatterGatherFirstCompletedRouter(nrOfInstances: Int = 0, routees: It * Java API */ def this(resizer: Resizer, w: Duration) = this(resizer = Some(resizer), within = w) + + /** + * Java API for setting routerDispatcher + */ + def withDispatcher(dispatcherId: String) = copy(routerDispatcher = dispatcherId) } trait ScatterGatherFirstCompletedLike { this: RouterConfig ⇒ @@ -795,14 +879,22 @@ trait Resizer { * for the initial resize and continues with 1 for the first message. Make sure to perform * initial resize before first message (messageCounter == 0), because there is no guarantee * that resize will be done when concurrent messages are in play. + * + * CAUTION: this method is invoked from the thread which tries to send a + * message to the pool, i.e. the ActorRef.!() method, hence it may be called + * concurrently. */ def isTimeForResize(messageCounter: Long): Boolean + /** * Decide if the capacity of the router need to be changed. Will be invoked when `isTimeForResize` * returns true and no other resize is in progress. * Create and register more routees with `routeeProvider.registerRoutees(newRoutees) * or remove routees with `routeeProvider.unregisterRoutees(abandonedRoutees)` and * sending [[akka.actor.PoisonPill]] to them. + * + * This method is invoked only in the context of the Router actor in order to safely + * create/stop children. */ def resize(props: Props, routeeProvider: RouteeProvider) } diff --git a/akka-docs/java/code/akka/docs/jrouting/CustomRouterDocTestBase.java b/akka-docs/java/code/akka/docs/jrouting/CustomRouterDocTestBase.java index 2a2e5c7f22..a20a351f06 100644 --- a/akka-docs/java/code/akka/docs/jrouting/CustomRouterDocTestBase.java +++ b/akka-docs/java/code/akka/docs/jrouting/CustomRouterDocTestBase.java @@ -17,6 +17,7 @@ import akka.util.Duration; import akka.util.Timeout; import akka.dispatch.Await; import akka.dispatch.Future; +import akka.dispatch.Dispatchers; import akka.testkit.AkkaSpec; import com.typesafe.config.ConfigFactory; import static akka.pattern.Patterns.ask; @@ -38,6 +39,19 @@ public class CustomRouterDocTestBase { public void tearDown() { system.shutdown(); } + + public static class MyActor extends UntypedActor { + @Override public void onReceive(Object o) {} + } + + @Test + public void demonstrateDispatchers() { + //#dispatchers + final ActorRef router = system.actorOf(new Props(MyActor.class) + .withRouter(new RoundRobinRouter(5).withDispatcher("head")) // “head” router runs on "head" dispatcher + .withDispatcher("workers")); // MyActor “workers” run on "workers" dispatcher + //#dispatchers + } //#crTest @Test @@ -105,6 +119,10 @@ public class CustomRouterDocTestBase { //#crRouter public static class VoteCountRouter extends CustomRouterConfig { + + @Override public String routerDispatcher() { + return Dispatchers.DefaultDispatcherId(); + } //#crRoute @Override diff --git a/akka-docs/java/routing.rst b/akka-docs/java/routing.rst index a422900440..265e31a984 100644 --- a/akka-docs/java/routing.rst +++ b/akka-docs/java/routing.rst @@ -8,11 +8,6 @@ Routing (Java) .. contents:: :local: -Akka-core includes some building blocks to build more complex message flow handlers, they are listed and explained below: - -Router ------- - A Router is an actor that routes incoming messages to outbound actors. The router routes the messages sent to it to its underlying actors called 'routees'. @@ -249,6 +244,16 @@ This is an example of how to programatically create a resizable router: *It is also worth pointing out that if you define the ``router`` in the configuration file then this value will be used instead of any programmatically sent parameters.* +.. note:: + + Resizing is triggered by sending messages to the actor pool, but it is not + completed synchronously; instead a message is sent to the “head” + :class:`Router` to perform the size change. Thus you cannot rely on resizing + to instantaneously create new workers when all others are busy, because the + message just sent will be queued to the mailbox of a busy actor. To remedy + this, configure the pool to use a balancing dispatcher, see `Configuring + Dispatchers`_ for more information. + Custom Router ^^^^^^^^^^^^^ @@ -312,3 +317,23 @@ A router with dynamically resizable number of routees is implemented by providin in ``resizer`` method of the ``RouterConfig``. See ``akka.routing.DefaultResizer`` for inspiration of how to write your own resize strategy. +Configuring Dispatchers +^^^^^^^^^^^^^^^^^^^^^^^ + +The dispatcher for created children of the router will be taken from +:class:`Props` as described in :ref:`dispatchers-java`. For a dynamic pool it +makes sense to configure the :class:`BalancingDispatcher` if the precise +routing is not so important (i.e. no consistent hashing or round-robin is +required); this enables newly created routees to pick up work immediately by +stealing it from their siblings. + +The “head” router, of couse, cannot run on the same balancing dispatcher, +because it does not process the same messages, hence this special actor does +not use the dispatcher configured in :class:`Props`, but takes the +``routerDispatcher`` from the :class:`RouterConfig` instead, which defaults to +the actor system’s default dispatcher. All standard routers allow setting this +property in their constructor or factory method, custom routers have to +implement the method in a suitable way. + +.. includecode:: code/akka/docs/jrouting/CustomRouterDocTestBase.java#dispatchers + diff --git a/akka-docs/scala/code/akka/docs/routing/RouterDocSpec.scala b/akka-docs/scala/code/akka/docs/routing/RouterDocSpec.scala new file mode 100644 index 0000000000..229c66f13e --- /dev/null +++ b/akka-docs/scala/code/akka/docs/routing/RouterDocSpec.scala @@ -0,0 +1,29 @@ +/** + * Copyright (C) 2009-2012 Typesafe Inc. + */ +package akka.docs.routing + +import RouterDocSpec.MyActor +import akka.actor.{ Props, Actor } +import akka.testkit.AkkaSpec +import akka.routing.RoundRobinRouter + +object RouterDocSpec { + class MyActor extends Actor { + def receive = { + case _ ⇒ + } + } +} + +class RouterDocSpec extends AkkaSpec { + + import RouterDocSpec._ + + //#dispatchers + val router = system.actorOf(Props[MyActor] + .withRouter(RoundRobinRouter(5, routerDispatcher = "router")) // “head” will run on "router" dispatcher + .withDispatcher("workers")) // MyActor workers will run on "workers" dispatcher + //#dispatchers + +} \ No newline at end of file diff --git a/akka-docs/scala/routing.rst b/akka-docs/scala/routing.rst index f67841df2c..161ab88db9 100644 --- a/akka-docs/scala/routing.rst +++ b/akka-docs/scala/routing.rst @@ -8,11 +8,6 @@ Routing (Scala) .. contents:: :local: -Akka-core includes some building blocks to build more complex message flow handlers, they are listed and explained below: - -Router ------- - A Router is an actor that routes incoming messages to outbound actors. The router routes the messages sent to it to its underlying actors called 'routees'. @@ -250,6 +245,16 @@ This is an example of how to programatically create a resizable router: *It is also worth pointing out that if you define the ``router`` in the configuration file then this value will be used instead of any programmatically sent parameters.* +.. note:: + + Resizing is triggered by sending messages to the actor pool, but it is not + completed synchronously; instead a message is sent to the “head” + :class:`Router` to perform the size change. Thus you cannot rely on resizing + to instantaneously create new workers when all others are busy, because the + message just sent will be queued to the mailbox of a busy actor. To remedy + this, configure the pool to use a balancing dispatcher, see `Configuring + Dispatchers`_ for more information. + Custom Router ^^^^^^^^^^^^^ @@ -311,3 +316,23 @@ A router with dynamically resizable number of routees is implemented by providin in ``resizer`` method of the ``RouterConfig``. See ``akka.routing.DefaultResizer`` for inspiration of how to write your own resize strategy. +Configuring Dispatchers +^^^^^^^^^^^^^^^^^^^^^^^ + +The dispatcher for created children of the router will be taken from +:class:`Props` as described in :ref:`dispatchers-scala`. For a dynamic pool it +makes sense to configure the :class:`BalancingDispatcher` if the precise +routing is not so important (i.e. no consistent hashing or round-robin is +required); this enables newly created routees to pick up work immediately by +stealing it from their siblings. + +The “head” router, of couse, cannot run on the same balancing dispatcher, +because it does not process the same messages, hence this special actor does +not use the dispatcher configured in :class:`Props`, but takes the +``routerDispatcher`` from the :class:`RouterConfig` instead, which defaults to +the actor system’s default dispatcher. All standard routers allow setting this +property in their constructor or factory method, custom routers have to +implement the method in a suitable way. + +.. includecode:: code/akka/docs/routing/RouterDocSpec.scala#dispatchers + diff --git a/akka-remote/src/main/scala/akka/routing/RemoteRouterConfig.scala b/akka-remote/src/main/scala/akka/routing/RemoteRouterConfig.scala index 21f5c400b0..3b1791db8e 100644 --- a/akka-remote/src/main/scala/akka/routing/RemoteRouterConfig.scala +++ b/akka-remote/src/main/scala/akka/routing/RemoteRouterConfig.scala @@ -30,6 +30,8 @@ case class RemoteRouterConfig(local: RouterConfig, nodes: Iterable[String]) exte override def createActor(): Router = local.createActor() + override def routerDispatcher: String = local.routerDispatcher + override def resizer: Option[Resizer] = local.resizer override def withFallback(other: RouterConfig): RouterConfig = other match {