diff --git a/akka-actor-tests/src/test/scala/akka/actor/TypedActorSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/TypedActorSpec.scala index 3cc54ea6bb..b2265367c7 100644 --- a/akka-actor-tests/src/test/scala/akka/actor/TypedActorSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/actor/TypedActorSpec.scala @@ -334,7 +334,7 @@ class TypedActorSpec extends AkkaSpec with BeforeAndAfterEach with BeforeAndAfte mustStop(t) } - "be able to use work-stealing dispatcher" in { + "be able to use balancing dispatcher" in { val props = Props( timeout = Timeout(6600), dispatcher = system.dispatcherFactory.newBalancingDispatcher("pooled-dispatcher") diff --git a/akka-actor/src/main/resources/reference.conf b/akka-actor/src/main/resources/reference.conf index 3d15c76a1b..0c759f8079 100644 --- a/akka-actor/src/main/resources/reference.conf +++ b/akka-actor/src/main/resources/reference.conf @@ -116,7 +116,7 @@ akka { task-queue-type = "linked" # Specifies which type of task queue will be used, can be "array" or "linked" (default) allow-core-timeout = on # Allow core threads to time out throughput = 5 # Throughput defines the number of messages that are processed in a batch before the - # thread is returned to the pool. Set to 1 for complete fairness. + # thread is returned to the pool. Set to 1 for as fair as possible. throughput-deadline-time = 0ms # Throughput deadline for Dispatcher, set to 0 or negative for no deadline mailbox-capacity = -1 # If negative (or zero) then an unbounded mailbox is used (default) # If positive then a bounded mailbox is used and the capacity is set using the property diff --git a/akka-actor/src/main/scala/akka/dispatch/Dispatchers.scala b/akka-actor/src/main/scala/akka/dispatch/Dispatchers.scala index 315f37d124..cdcb056372 100644 --- a/akka-actor/src/main/scala/akka/dispatch/Dispatchers.scala +++ b/akka-actor/src/main/scala/akka/dispatch/Dispatchers.scala @@ -74,6 +74,7 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc throw new ConfigurationException("Wrong configuration [akka.actor.default-dispatcher]") } + // FIXME: Dispatchers registered here are are not removed, see ticket #1494 private val dispatchers = new ConcurrentHashMap[String, MessageDispatcher] /** @@ -84,7 +85,9 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc def lookup(key: String): MessageDispatcher = { dispatchers.get(key) match { case null ⇒ - // doesn't matter if we create a dispatcher that isn't used due to concurrent lookup + // It doesn't matter if we create a dispatcher that isn't used due to concurrent lookup. + // That shouldn't happen often and in case it does the actual ExecutorService isn't + // created until used, i.e. cheap. val newDispatcher = newFromConfig(key) dispatchers.putIfAbsent(key, newDispatcher) match { case null ⇒ newDispatcher @@ -161,7 +164,7 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc new Dispatcher(prerequisites, name, throughput, throughputDeadline, mailboxType, config, settings.DispatcherDefaultShutdown), ThreadPoolConfig()) /** - * Creates a executor-based event-driven dispatcher, with work-stealing, serving multiple (millions) of actors through a thread pool. + * Creates a executor-based event-driven dispatcher, with work-sharing, serving multiple (millions) of actors through a thread pool. *

* Has a fluent builder interface for configuring its semantics. */ @@ -170,7 +173,7 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc settings.DispatcherThroughputDeadlineTime, MailboxType, config, settings.DispatcherDefaultShutdown), ThreadPoolConfig()) /** - * Creates a executor-based event-driven dispatcher, with work-stealing, serving multiple (millions) of actors through a thread pool. + * Creates a executor-based event-driven dispatcher, with work-sharing, serving multiple (millions) of actors through a thread pool. *

* Has a fluent builder interface for configuring its semantics. */ @@ -180,7 +183,7 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc config, settings.DispatcherDefaultShutdown), ThreadPoolConfig()) /** - * Creates a executor-based event-driven dispatcher, with work-stealing, serving multiple (millions) of actors through a thread pool. + * Creates a executor-based event-driven dispatcher, with work-sharing, serving multiple (millions) of actors through a thread pool. *

* Has a fluent builder interface for configuring its semantics. */ @@ -190,7 +193,7 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc config, settings.DispatcherDefaultShutdown), ThreadPoolConfig()) /** - * Creates a executor-based event-driven dispatcher, with work-stealing, serving multiple (millions) of actors through a thread pool. + * Creates a executor-based event-driven dispatcher, with work-sharing, serving multiple (millions) of actors through a thread pool. *

* Has a fluent builder interface for configuring its semantics. */ diff --git a/akka-docs/general/configuration.rst b/akka-docs/general/configuration.rst index 9328e19561..9a79a0769f 100644 --- a/akka-docs/general/configuration.rst +++ b/akka-docs/general/configuration.rst @@ -28,10 +28,10 @@ configuration for each actor system, and grab the specific configuration when in :: myapp1 { - akka.logLevel = WARNING + akka.loglevel = WARNING } myapp2 { - akka.logLevel = ERROR + akka.loglevel = ERROR } .. code-block:: scala @@ -120,7 +120,7 @@ A custom ``application.conf`` might look like this:: actor { default-dispatcher { - throughput = 10 # Throughput for default Dispatcher, set to 1 for complete fairness + throughput = 10 # Throughput for default Dispatcher, set to 1 for as fair as possible } } diff --git a/akka-docs/scala/code/DispatcherDocSpec.scala b/akka-docs/scala/code/DispatcherDocSpec.scala index 13f78cbd87..f132a5e90f 100644 --- a/akka-docs/scala/code/DispatcherDocSpec.scala +++ b/akka-docs/scala/code/DispatcherDocSpec.scala @@ -14,7 +14,6 @@ import akka.actor.PoisonPill object DispatcherDocSpec { val config = """ - akka.logLevel=INFO //#my-dispatcher-config my-dispatcher { type = Dispatcher # Dispatcher is the name of the event-based dispatcher @@ -22,7 +21,7 @@ object DispatcherDocSpec { core-pool-size-factor = 2.0 # No of core threads ... ceil(available processors * factor) core-pool-size-max = 10 # maximum number of threads to cap factor-based number to throughput = 100 # Throughput defines the number of messages that are processed in a batch before the - # thread is returned to the pool. Set to 1 for complete fairness. + # thread is returned to the pool. Set to 1 for as fair as possible. } //#my-dispatcher-config diff --git a/akka-docs/scala/dispatchers.rst b/akka-docs/scala/dispatchers.rst index b4e36cf0b0..628d0a11d4 100644 --- a/akka-docs/scala/dispatchers.rst +++ b/akka-docs/scala/dispatchers.rst @@ -64,7 +64,8 @@ Let's now walk through the different dispatchers in more detail. Thread-based ^^^^^^^^^^^^ -The ``PinnedDispatcher`` binds a dedicated OS thread to each specific Actor. The messages are posted to a ``LinkedBlockingQueue`` +The ``PinnedDispatcher`` binds a dedicated OS thread to each specific Actor. The messages are posted to a +`LinkedBlockingQueue `_ which feeds the messages to the dispatcher one by one. A ``PinnedDispatcher`` cannot be shared between actors. This dispatcher has worse performance and scalability than the event-based dispatcher but works great for creating "daemon" Actors that consumes a low frequency of messages and are allowed to go off and do their own thing for a longer period of time. Another advantage with @@ -79,7 +80,8 @@ The ``PinnedDispatcher`` is configured as a event-based dispatcher with with cor Event-based ^^^^^^^^^^^ -The event-based ``Dispatcher`` binds a set of Actors to a thread pool backed up by a ``BlockingQueue``. This dispatcher is highly configurable +The event-based ``Dispatcher`` binds a set of Actors to a thread pool backed up by a +`BlockingQueue `_. This dispatcher is highly configurable and supports a fluent configuration API to configure the ``BlockingQueue`` (type of queue, max items etc.) as well as the thread pool. The event-driven dispatchers **must be shared** between multiple Actors. One best practice is to let each top-level Actor, e.g. @@ -90,21 +92,14 @@ design and implement your system in the most efficient way in regards to perform It comes with many different predefined BlockingQueue configurations: -* Bounded LinkedBlockingQueue -* Unbounded LinkedBlockingQueue -* Bounded ArrayBlockingQueue -* Unbounded ArrayBlockingQueue -* SynchronousQueue +* Bounded `LinkedBlockingQueue `_ +* Unbounded `LinkedBlockingQueue `_ +* Bounded `ArrayBlockingQueue `_ +* Unbounded `ArrayBlockingQueue `_ +* `SynchronousQueue `_ -You can also set the rejection policy that should be used, e.g. what should be done if the dispatcher (e.g. the Actor) can't keep up -and the mailbox is growing up to the limit defined. You can choose between four different rejection policies: - -* java.util.concurrent.ThreadPoolExecutor.CallerRuns - will run the message processing in the caller's thread as a way to slow him down and balance producer/consumer -* java.util.concurrent.ThreadPoolExecutor.AbortPolicy - rejected messages by throwing a ``RejectedExecutionException`` -* java.util.concurrent.ThreadPoolExecutor.DiscardPolicy - discards the message (throws it away) -* java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy - discards the oldest message in the mailbox (throws it away) - -You can read more about these policies `here `_. +When using a bounded queue and it has grown up to limit defined the message processing will run in the caller's +thread as a way to slow him down and balance producer/consumer. Here is an example of a bounded mailbox: @@ -113,7 +108,7 @@ Here is an example of a bounded mailbox: The standard :class:`Dispatcher` allows you to define the ``throughput`` it should have, as shown above. This defines the number of messages for a specific Actor the dispatcher should process in one single sweep; in other words, the -dispatcher will bunch up to ``throughput`` messages together when +dispatcher will batch process up to ``throughput`` messages together when having elected an actor to run. Setting this to a higher number will increase throughput but lower fairness, and vice versa. If you don't specify it explicitly then it uses the value (5) defined for ``default-dispatcher`` in the :ref:`configuration`. diff --git a/akka-docs/scala/typed-actors.rst b/akka-docs/scala/typed-actors.rst index 295fba6632..39c8d58ea8 100644 --- a/akka-docs/scala/typed-actors.rst +++ b/akka-docs/scala/typed-actors.rst @@ -11,7 +11,7 @@ The Typed Actors are implemented through `Typed Actors `_ then take a look at Akka's `Spring integration `_. -**WARNING:** Do not configure to use a ``WorkStealingDispatcher`` with your ``TypedActors``, it just isn't safe with how ``TypedActors`` currently are implemented. This limitation will most likely be removed in the future. +**WARNING:** Do not configure to use a ``BalancingDispatcher`` with your ``TypedActors``, it just isn't safe with how ``TypedActors`` currently are implemented. This limitation will most likely be removed in the future. Creating Typed Actors ---------------------