raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Migrating Agents to greener pastures

Browse source
This commit is contained in:
Viktor Klang 2013-01-23 02:06:49 +01:00
parent 13b1324509
commit 9522add9b7
9 changed files with 396 additions and 444 deletions
Download patch file Download diff file Expand all files Collapse all files

121
akka-docs/rst/java/agents.rst
View file

@ -15,77 +15,35 @@ functions that are asynchronously applied to the Agent's state and whose return
value becomes the Agent's new state. The state of an Agent should be immutable.
While updates to Agents are asynchronous, the state of an Agent is always
immediately available for reading by any thread (using ``get``) without any
messages.
immediately available for reading by any thread (using ``get``) without any messages.
Agents are reactive. The update actions of all Agents get interleaved amongst
threads in a thread pool. At any point in time, at most one ``send`` action for
threads in an ``ExecutionContext``. At any point in time, at most one ``send`` action for
each Agent is being executed. Actions dispatched to an agent from another thread
will occur in the order they were sent, potentially interleaved with actions
dispatched to the same agent from other sources.
dispatched to the same agent from other threads.
If an Agent is used within an enclosing transaction, then it will participate in
that transaction. Agents are integrated with the STM - any dispatches made in
that transaction. Agents are integrated with Scala STM - any dispatches made in
a transaction are held until that transaction commits, and are discarded if it
is retried or aborted.
Creating and stopping Agents
Creating Agents
============================
Agents are created by invoking ``new Agent(value, system)`` passing in the
Agent's initial value and a reference to the ``ActorSystem`` for your
application. An ``ActorSystem`` is required to create the underlying Actors. See
:ref:`actor-systems` for more information about actor systems.
Here is an example of creating an Agent:
Agents are created by invoking ``new Agent<ValueType>(value, executionContext)`` – passing in the Agent's initial
value and providing an ``ExecutionContext`` to be used for it:
.. includecode:: code/docs/agent/AgentDocTest.java
:include: import-system,import-agent
:include: import-agent,create
:language: java
.. includecode:: code/docs/agent/AgentDocTest.java#create
:language: java
An Agent will be running until you invoke ``close`` on it. Then it will be
eligible for garbage collection (unless you hold on to it in some way).
.. includecode:: code/docs/agent/AgentDocTest.java#close
:language: java
Updating Agents
===============
You update an Agent by sending a function that transforms the current value or
by sending just a new value. The Agent will apply the new value or function
atomically and asynchronously. The update is done in a fire-forget manner and
you are only guaranteed that it will be applied. There is no guarantee of when
the update will be applied but dispatches to an Agent from a single thread will
occur in order. You apply a value or a function by invoking the ``send``
function.
.. includecode:: code/docs/agent/AgentDocTest.java#import-function
:language: java
.. includecode:: code/docs/agent/AgentDocTest.java#send
:language: java
You can also dispatch a function to update the internal state but on its own
thread. This does not use the reactive thread pool and can be used for
long-running or blocking operations. You do this with the ``sendOff``
method. Dispatches using either ``sendOff`` or ``send`` will still be executed
in order.
.. includecode:: code/docs/agent/AgentDocTest.java#send-off
:language: java
Reading an Agent's value
========================
Agents can be dereferenced (you can get an Agent's value) by calling the get
method:
Agents can be dereferenced (you can get an Agent's value) by invoking the Agent
with ``get()`` like this:
.. includecode:: code/docs/agent/AgentDocTest.java#read-get
:language: java
@ -94,15 +52,58 @@ Reading an Agent's current value does not involve any message passing and
happens immediately. So while updates to an Agent are asynchronous, reading the
state of an Agent is synchronous.
You can also get a ``Future`` to the Agents value, that will be completed after the
currently queued updates have completed:
Awaiting an Agent's value
=========================
It is also possible to read the value after all currently queued sends have
completed. You can do this with ``await``:
.. includecode:: code/docs/agent/AgentDocTest.java#import-timeout
.. includecode:: code/docs/agent/AgentDocTest.java
:include: import-future,read-future
:language: java
.. includecode:: code/docs/agent/AgentDocTest.java#read-await
See :ref:`futures-java` for more information on ``Futures``.
Updating Agents (send & alter)
==============================
You update an Agent by sending a function (``akka.dispatch.Mapper``) that transforms the current value or
by sending just a new value. The Agent will apply the new value or function
atomically and asynchronously. The update is done in a fire-forget manner and
you are only guaranteed that it will be applied. There is no guarantee of when
the update will be applied but dispatches to an Agent from a single thread will
occur in order. You apply a value or a function by invoking the ``send``
function.
.. includecode:: code/docs/agent/AgentDocTest.java
:include: import-function,send
:language: java
You can also dispatch a function to update the internal state but on its own
thread. This does not use the reactive thread pool and can be used for
long-running or blocking operations. You do this with the ``sendOff``
method. Dispatches using either ``sendOff`` or ``send`` will still be executed
in order.
.. includecode:: code/docs/agent/AgentDocTest.java
:include: import-function,send-off
:language: java
All ``send`` methods also have a corresponding ``alter`` method that returns a ``Future``.
See :ref:`futures-java` for more information on ``Futures``.
.. includecode:: code/docs/agent/AgentDocTest.java
:include: import-future,import-function,alter
:language: java
.. includecode:: code/docs/agent/AgentDocTest.java
:include: import-future,import-function,alter-off
:language: java
Transactional Agents
====================
If an Agent is used within an enclosing transaction, then it will participate in
that transaction. If you send to an Agent within a transaction then the dispatch
to the Agent will be held until that transaction commits, and discarded if the
transaction is aborted. Here's an example:
.. includecode:: code/docs/agent/AgentDocTest.java#transfer-example
:language: java

113
akka-docs/rst/java/code/docs/agent/AgentDocTest.java
View file

@ -5,107 +5,112 @@ package docs.agent;
import static org.junit.Assert.*;
import scala.concurrent.ExecutionContext;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import akka.testkit.AkkaSpec;
//#import-system
import akka.actor.ActorSystem;
//#import-system
import scala.concurrent.Await;
import scala.concurrent.duration.Duration;
//#import-agent
import akka.agent.Agent;
import scala.concurrent.ExecutionContext;
import akka.agent.Agent;
import akka.dispatch.ExecutionContexts;
//#import-agent
//#import-function
import akka.japi.Function;
import akka.dispatch.Mapper;
//#import-function
//#import-timeout
import akka.util.Timeout;
import static java.util.concurrent.TimeUnit.SECONDS;
//#import-timeout
//#import-future
import scala.concurrent.Future;
//#import-future
public class AgentDocTest {
private static ActorSystem testSystem;
private static ExecutionContext ec;
@BeforeClass
public static void beforeAll() {
testSystem = ActorSystem.create("AgentDocTest", AkkaSpec.testConf());
ec = testSystem.dispatcher();
}
@AfterClass
public static void afterAll() {
testSystem.shutdown();
testSystem = null;
}
private static ExecutionContext ec = ExecutionContexts.global();
@Test
public void createAndClose() {
//#create
ActorSystem system = ActorSystem.create("app");
Agent<Integer> agent = new Agent<Integer>(5, system);
public void createAndRead() throws Exception {
//#create
ExecutionContext ec = ExecutionContexts.global();
Agent<Integer> agent = new Agent<Integer>(5, ec);
//#create
//#close
agent.close();
//#close
//#read-get
Integer result = agent.get();
//#read-get
system.shutdown();
//#read-future
Future<Integer> future = agent.future();
//#read-future
assertEquals(result, new Integer(5));
assertEquals(Await.result(future, Duration.create(5,"s")), new Integer(5));
}
@Test
public void sendAndSendOffAndReadAwait() {
Agent<Integer> agent = new Agent<Integer>(5, testSystem);
public void sendAndSendOffAndReadAwait() throws Exception {
Agent<Integer> agent = new Agent<Integer>(5, ec);
//#send
// send a value
agent.send(7);
// send a function
agent.send(new Function<Integer, Integer>() {
agent.send(new Mapper<Integer, Integer>() {
public Integer apply(Integer i) {
return i * 2;
}
});
//#send
Function<Integer, Integer> longRunningOrBlockingFunction = new Function<Integer, Integer>() {
Mapper<Integer, Integer> longRunningOrBlockingFunction = new Mapper<Integer, Integer>() {
public Integer apply(Integer i) {
return i * 1;
}
};
ExecutionContext theExecutionContextToExecuteItIn = ec;
//#send-off
// sendOff a function
agent.sendOff(longRunningOrBlockingFunction, ec);
agent.sendOff(longRunningOrBlockingFunction,
theExecutionContextToExecuteItIn);
//#send-off
//#read-await
Integer result = agent.await(new Timeout(5, SECONDS));
//#read-await
assertEquals(result, new Integer(14));
agent.close();
assertEquals(Await.result(agent.future(), Duration.create(5,"s")), new Integer(14));
}
@Test
public void readWithGet() {
Agent<Integer> agent = new Agent<Integer>(5, testSystem);
@Test
public void alterAndAlterOff() throws Exception {
Agent<Integer> agent = new Agent<Integer>(5, ec);
//#read-get
Integer result = agent.get();
//#read-get
//#alter
// alter a value
Future<Integer> f1 = agent.alter(7);
assertEquals(result, new Integer(5));
// alter a function (Mapper)
Future<Integer> f2 = agent.alter(new Mapper<Integer, Integer>() {
public Integer apply(Integer i) {
return i * 2;
}
});
//#alter
agent.close();
}
Mapper<Integer, Integer> longRunningOrBlockingFunction = new Mapper<Integer, Integer>() {
public Integer apply(Integer i) {
return i * 1;
}
};
ExecutionContext theExecutionContextToExecuteItIn = ec;
//#alter-off
// alterOff a function (Mapper)
Future<Integer> f3 = agent.alterOff(longRunningOrBlockingFunction,
theExecutionContextToExecuteItIn);
//#alter-off
assertEquals(Await.result(f3, Duration.create(5,"s")), new Integer(14));
}
}

60
akka-docs/rst/scala/agents.rst
View file

@ -19,10 +19,10 @@ immediately available for reading by any thread (using ``get`` or ``apply``)
without any messages.
Agents are reactive. The update actions of all Agents get interleaved amongst
threads in a thread pool. At any point in time, at most one ``send`` action for
threads in an ``ExecutionContext``. At any point in time, at most one ``send`` action for
each Agent is being executed. Actions dispatched to an agent from another thread
will occur in the order they were sent, potentially interleaved with actions
dispatched to the same agent from other sources.
dispatched to the same agent from other threads.
If an Agent is used within an enclosing transaction, then it will participate in
that transaction. Agents are integrated with Scala STM - any dispatches made in
@ -30,32 +30,33 @@ a transaction are held until that transaction commits, and are discarded if it
is retried or aborted.
Creating and stopping Agents
Creating Agents
============================
Agents are created by invoking ``Agent(value)`` passing in the Agent's initial
value:
value and providing an implicit ``ExecutionContext`` to be used for it, for these
examples we're going to use the default global one, but YMMV:
.. includecode:: code/docs/agent/AgentDocSpec.scala#create
Note that creating an Agent requires an implicit ``ActorSystem`` (for creating
the underlying actors). See :ref:`actor-systems` for more information about
actor systems. An ActorSystem can be in implicit scope when creating an Agent:
Reading an Agent's value
========================
.. includecode:: code/docs/agent/AgentDocSpec.scala#create-implicit-system
Agents can be dereferenced (you can get an Agent's value) by invoking the Agent
with parentheses like this:
Or the ActorSystem can be passed explicitly when creating an Agent:
.. includecode:: code/docs/agent/AgentDocSpec.scala#read-apply
.. includecode:: code/docs/agent/AgentDocSpec.scala#create-explicit-system
Or by using the get method:
An Agent will be running until you invoke ``close`` on it. Then it will be
eligible for garbage collection (unless you hold on to it in some way).
.. includecode:: code/docs/agent/AgentDocSpec.scala#read-get
.. includecode:: code/docs/agent/AgentDocSpec.scala#close
Reading an Agent's current value does not involve any message passing and
happens immediately. So while updates to an Agent are asynchronous, reading the
state of an Agent is synchronous.
Updating Agents
===============
Updating Agents (send & alter)
======================
You update an Agent by sending a function that transforms the current value or
by sending just a new value. The Agent will apply the new value or function
@ -75,37 +76,22 @@ in order.
.. includecode:: code/docs/agent/AgentDocSpec.scala#send-off
All ``send`` methods also have a corresponding ``alter`` method that returns a ``Future``.
See :ref:`futures-scala` for more information on ``Futures``.
Reading an Agent's value
========================
Agents can be dereferenced (you can get an Agent's value) by invoking the Agent
with parentheses like this:
.. includecode:: code/docs/agent/AgentDocSpec.scala#read-apply
Or by using the get method:
.. includecode:: code/docs/agent/AgentDocSpec.scala#read-get
Reading an Agent's current value does not involve any message passing and
happens immediately. So while updates to an Agent are asynchronous, reading the
state of an Agent is synchronous.
.. includecode:: code/docs/agent/AgentDocSpec.scala#alter
.. includecode:: code/docs/agent/AgentDocSpec.scala#alter-off
Awaiting an Agent's value
=========================
It is also possible to read the value after all currently queued sends have
completed. You can do this with ``await``:
.. includecode:: code/docs/agent/AgentDocSpec.scala#read-await
You can also get a ``Future`` to this value, that will be completed after the
You can also get a ``Future`` to the Agents value, that will be completed after the
currently queued updates have completed:
.. includecode:: code/docs/agent/AgentDocSpec.scala#read-future
See :ref:`futures-scala` for more information on ``Futures``.
Transactional Agents
====================

158
akka-docs/rst/scala/code/docs/agent/AgentDocSpec.scala
View file

@ -7,54 +7,46 @@ import language.postfixOps
import akka.agent.Agent
import scala.concurrent.duration._
import akka.util.Timeout
import scala.concurrent.{ Await, ExecutionContext }
import akka.testkit._
import scala.concurrent.Future
class AgentDocSpec extends AkkaSpec {
"create and close" in {
"create" in {
//#create
import scala.concurrent.ExecutionContext.Implicits.global
import akka.agent.Agent
val agent = Agent(5)
//#create
//#close
agent.close()
//#close
}
"create with implicit system" in {
//#create-implicit-system
import akka.actor.ActorSystem
import akka.agent.Agent
"read value" in {
import scala.concurrent.ExecutionContext.Implicits.global
val agent = Agent(0)
implicit val system = ActorSystem("app")
{
//#read-apply
val result = agent()
//#read-apply
result must be === 0
}
{
//#read-get
val result = agent.get
//#read-get
result must be === 0
}
val agent = Agent(5)
//#create-implicit-system
agent.close()
system.shutdown()
}
"create with explicit system" in {
//#create-explicit-system
import akka.actor.ActorSystem
import akka.agent.Agent
val system = ActorSystem("app")
val agent = Agent(5)(system)
//#create-explicit-system
agent.close()
system.shutdown()
{
//#read-future
val future = agent.future
//#read-future
Await.result(future, 5 seconds) must be === 0
}
}
"send and sendOff" in {
val agent = Agent(0)
import system.dispatcher
val agent = Agent(0)(ExecutionContext.global)
//#send
// send a value
agent send 7
@ -64,70 +56,47 @@ class AgentDocSpec extends AkkaSpec {
agent send (_ * 2)
//#send
def longRunningOrBlockingFunction = (i: Int) ⇒ i * 1
def longRunningOrBlockingFunction = (i: Int) ⇒ i * 1 // Just for the example code
def someExecutionContext() = scala.concurrent.ExecutionContext.Implicits.global // Just for the example code
//#send-off
// the ExecutionContext you want to run the function on
implicit val ec = someExecutionContext()
// sendOff a function
agent sendOff (longRunningOrBlockingFunction)
agent sendOff longRunningOrBlockingFunction
//#send-off
val result = agent.await(Timeout(5 seconds))
result must be === 16
Await.result(agent.future, 5 seconds) must be === 16
}
"read with apply" in {
val agent = Agent(0)
"alter and alterOff" in {
val agent = Agent(0)(ExecutionContext.global)
//#alter
// alter a value
val f1: Future[Int] = agent alter 7
//#read-apply
val result = agent()
//#read-apply
// alter a function
val f2: Future[Int] = agent alter (_ + 1)
val f3: Future[Int] = agent alter (_ * 2)
//#alter
result must be === 0
}
def longRunningOrBlockingFunction = (i: Int) ⇒ i * 1 // Just for the example code
def someExecutionContext() = ExecutionContext.global // Just for the example code
"read with get" in {
val agent = Agent(0)
//#alter-off
// the ExecutionContext you want to run the function on
implicit val ec = someExecutionContext()
// alterOff a function
val f4: Future[Int] = agent alterOff longRunningOrBlockingFunction
//#alter-off
//#read-get
val result = agent.get
//#read-get
result must be === 0
}
"read with await" in {
val agent = Agent(0)
//#read-await
import scala.concurrent.duration._
import akka.util.Timeout
implicit val timeout = Timeout(5 seconds)
val result = agent.await
//#read-await
result must be === 0
}
"read with future" in {
val agent = Agent(0)
//#read-future
import scala.concurrent.Await
implicit val timeout = Timeout(5 seconds)
val future = agent.future
val result = Await.result(future, timeout.duration)
//#read-future
result must be === 0
Await.result(f4, 5 seconds) must be === 16
}
"transfer example" in {
//#transfer-example
import scala.concurrent.ExecutionContext.Implicits.global
import akka.agent.Agent
import scala.concurrent.duration._
import akka.util.Timeout
import scala.concurrent.stm._
def transfer(from: Agent[Int], to: Agent[Int], amount: Int): Boolean = {
@ -145,25 +114,25 @@ class AgentDocSpec extends AkkaSpec {
val to = Agent(20)
val ok = transfer(from, to, 50)
implicit val timeout = Timeout(5 seconds)
val fromValue = from.await // -> 50
val toValue = to.await // -> 70
val fromValue = from.future // -> 50
val toValue = to.future // -> 70
//#transfer-example
fromValue must be === 50
toValue must be === 70
Await.result(fromValue, 5 seconds) must be === 50
Await.result(toValue, 5 seconds) must be === 70
ok must be === true
}
"monadic example" in {
def println(a: Any) = ()
//#monadic-example
import scala.concurrent.ExecutionContext.Implicits.global
val agent1 = Agent(3)
val agent2 = Agent(5)
// uses foreach
var result = 0
for (value ← agent1) {
result = value + 1
}
for (value ← agent1)
println(value)
// uses map
val agent3 = for (value ← agent1) yield value + 1
@ -178,15 +147,8 @@ class AgentDocSpec extends AkkaSpec {
} yield value1 + value2
//#monadic-example
result must be === 4
agent3() must be === 4
agent4() must be === 4
agent5() must be === 8
agent1.close()
agent2.close()
agent3.close()
agent4.close()
agent5.close()
}
}