Merge pull request #179 from jboner/migrate-transactor

Migrate transactor to scala-stm

akka-docs/java/code/akka/docs/agent/AgentDocTest.java

@@ -23,7 +23,6 @@ import akka.japi.Function;
 //#import-function
 
 //#import-timeout
-import akka.util.Duration;
 import akka.util.Timeout;
 import static java.util.concurrent.TimeUnit.SECONDS;
 //#import-timeout
@@ -86,7 +85,7 @@ public class AgentDocTest {
     //#send-off
 
     //#read-await
-    Integer result = agent.await(new Timeout(Duration.create(5, SECONDS)));
+    Integer result = agent.await(new Timeout(5, SECONDS));
     //#read-await
 
     assertEquals(result, new Integer(14));

akka-docs/java/code/akka/docs/transactor/CoordinatedCounter.java

@@ -0,0 +1,40 @@
+/**
+ * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.docs.transactor;
+
+//#class
+import akka.actor.*;
+import akka.transactor.*;
+import scala.concurrent.stm.*;
+
+public class CoordinatedCounter extends UntypedActor {
+    private Ref<Integer> count = Stm.ref(0);
+
+    private void increment(InTxn txn) {
+        Integer newValue = count.get(txn) + 1;
+        count.set(newValue, txn);
+    }
+
+    public void onReceive(Object incoming) throws Exception {
+        if (incoming instanceof Coordinated) {
+            Coordinated coordinated = (Coordinated) incoming;
+            Object message = coordinated.getMessage();
+            if (message instanceof Increment) {
+                Increment increment = (Increment) message;
+                if (increment.hasFriend()) {
+                    increment.getFriend().tell(coordinated.coordinate(new Increment()));
+                }
+                coordinated.atomic(new Atomically() {
+                    public void atomically(InTxn txn) {
+                        increment(txn);
+                    }
+                });
+            }
+        } else if ("GetCount".equals(incoming)) {
+            getSender().tell(count.single().get());
+        }
+    }
+}
+//#class

akka-docs/java/code/akka/docs/transactor/Coordinator.java

@@ -0,0 +1,27 @@
+/**
+ * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.docs.transactor;
+
+import akka.actor.*;
+import akka.transactor.*;
+import scala.concurrent.stm.*;
+
+public class Coordinator extends UntypedActor {
+    public void onReceive(Object incoming) throws Exception {
+        if (incoming instanceof Coordinated) {
+            Coordinated coordinated = (Coordinated) incoming;
+            Object message = coordinated.getMessage();
+            if (message instanceof Message) {
+                //#coordinated-atomic
+                coordinated.atomic(new Atomically() {
+                    public void atomically(InTxn txn) {
+                        // do something in the coordinated transaction ...
+                    }
+                });
+                //#coordinated-atomic
+            }
+        }
+    }
+}

akka-docs/java/code/akka/docs/transactor/Counter.java

@@ -0,0 +1,28 @@
+/**
+ * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.docs.transactor;
+
+//#class
+import akka.transactor.*;
+import scala.concurrent.stm.*;
+
+public class Counter extends UntypedTransactor {
+    Ref<Integer> count = Stm.ref(0);
+
+    public void atomically(InTxn txn, Object message) {
+        if (message instanceof Increment) {
+            Integer newValue = count.get(txn) + 1;
+            count.set(newValue, txn);
+        }
+    }
+
+    @Override public boolean normally(Object message) {
+        if ("GetCount".equals(message)) {
+            getSender().tell(count.single().get());
+            return true;
+        } else return false;
+    }
+}
+//#class

akka-docs/java/code/akka/docs/transactor/FriendlyCounter.java

@@ -0,0 +1,39 @@
+/**
+ * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.docs.transactor;
+
+//#class
+import akka.actor.*;
+import akka.transactor.*;
+import java.util.Set;
+import scala.concurrent.stm.*;
+
+public class FriendlyCounter extends UntypedTransactor {
+    Ref<Integer> count = Stm.ref(0);
+
+    @Override public Set<SendTo> coordinate(Object message) {
+        if (message instanceof Increment) {
+            Increment increment = (Increment) message;
+            if (increment.hasFriend())
+                return include(increment.getFriend(), new Increment());
+        }
+        return nobody();
+    }
+
+    public void atomically(InTxn txn, Object message) {
+        if (message instanceof Increment) {
+            Integer newValue = count.get(txn) + 1;
+            count.set(newValue, txn);
+        }
+    }
+
+    @Override public boolean normally(Object message) {
+        if ("GetCount".equals(message)) {
+            getSender().tell(count.single().get());
+            return true;
+        } else return false;
+    }
+}
+//#class

akka-docs/java/code/akka/docs/transactor/Increment.java

@@ -0,0 +1,27 @@
+/**
+ * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.docs.transactor;
+
+//#class
+import akka.actor.ActorRef;
+
+public class Increment {
+    private ActorRef friend = null;
+
+    public Increment() {}
+
+    public Increment(ActorRef friend) {
+        this.friend = friend;
+    }
+
+    public boolean hasFriend() {
+        return friend != null;
+    }
+
+    public ActorRef getFriend() {
+        return friend;
+    }
+}
+//#class

akka-docs/java/code/akka/docs/transactor/Message.java

@@ -0,0 +1,7 @@
+/**
+ * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.docs.transactor;
+
+public class Message {}

akka-docs/java/code/akka/docs/transactor/TransactorDocJavaSpec.scala

@@ -0,0 +1,11 @@
+/**
+ * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.docs.transactor
+
+import org.scalatest.junit.JUnitWrapperSuite
+
+class TransactorDocJavaSpec extends JUnitWrapperSuite(
+  "akka.docs.transactor.TransactorDocTest",
+  Thread.currentThread.getContextClassLoader)

akka-docs/java/code/akka/docs/transactor/TransactorDocTest.java

@@ -0,0 +1,99 @@
+/**
+ * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.docs.transactor;
+
+import static org.junit.Assert.*;
+import org.junit.Test;
+
+//#imports
+import akka.actor.*;
+import akka.dispatch.Await;
+import akka.transactor.Coordinated;
+import akka.util.Duration;
+import akka.util.Timeout;
+import static java.util.concurrent.TimeUnit.SECONDS;
+//#imports
+
+public class TransactorDocTest {
+
+    @Test
+    public void coordinatedExample() {
+        //#coordinated-example
+        ActorSystem system = ActorSystem.create("CoordinatedExample");
+
+        ActorRef counter1 = system.actorOf(new Props(CoordinatedCounter.class));
+        ActorRef counter2 = system.actorOf(new Props(CoordinatedCounter.class));
+
+        Timeout timeout = new Timeout(5, SECONDS);
+
+        counter1.tell(new Coordinated(new Increment(counter2), timeout));
+
+        Integer count = (Integer) Await.result(counter1.ask("GetCount", timeout), timeout.duration());
+        //#coordinated-example
+
+        assertEquals(count, new Integer(1));
+
+        system.shutdown();
+    }
+
+    @Test
+    public void coordinatedApi() {
+        //#create-coordinated
+        Timeout timeout = new Timeout(5, SECONDS);
+        Coordinated coordinated = new Coordinated(timeout);
+        //#create-coordinated
+
+        ActorSystem system = ActorSystem.create("CoordinatedApi");
+        ActorRef actor = system.actorOf(new Props(Coordinator.class));
+
+        //#send-coordinated
+        actor.tell(new Coordinated(new Message(), timeout));
+        //#send-coordinated
+
+        //#include-coordinated
+        actor.tell(coordinated.coordinate(new Message()));
+        //#include-coordinated
+
+        coordinated.await();
+
+        system.shutdown();
+    }
+
+    @Test
+    public void counterTransactor() {
+        ActorSystem system = ActorSystem.create("CounterTransactor");
+        ActorRef counter = system.actorOf(new Props(Counter.class));
+
+        Timeout timeout = new Timeout(5, SECONDS);
+        Coordinated coordinated = new Coordinated(timeout);
+        counter.tell(coordinated.coordinate(new Increment()));
+        coordinated.await();
+
+        Integer count = (Integer) Await.result(counter.ask("GetCount", timeout), timeout.duration());
+        assertEquals(count, new Integer(1));
+
+        system.shutdown();
+    }
+
+    @Test
+    public void friendlyCounterTransactor() {
+        ActorSystem system = ActorSystem.create("FriendlyCounterTransactor");
+        ActorRef friend = system.actorOf(new Props(Counter.class));
+        ActorRef friendlyCounter = system.actorOf(new Props(FriendlyCounter.class));
+
+        Timeout timeout = new Timeout(5, SECONDS);
+        Coordinated coordinated = new Coordinated(timeout);
+        friendlyCounter.tell(coordinated.coordinate(new Increment(friend)));
+        coordinated.await();
+
+        Integer count1 = (Integer) Await.result(friendlyCounter.ask("GetCount", timeout), timeout.duration());
+        assertEquals(count1, new Integer(1));
+
+        Integer count2 = (Integer) Await.result(friend.ask("GetCount", timeout), timeout.duration());
+        assertEquals(count2, new Integer(1));
+
+        system.shutdown();
+    }
+}

akka-docs/java/index.rst

@@ -17,6 +17,7 @@ Java API
    routing
    remoting
    serialization
+   stm
    agents
-   extending-akka
    transactors
+   extending-akka

akka-docs/java/stm.rst

@@ -0,0 +1,60 @@
+
+.. _stm-java:
+
+#####################################
+ Software Transactional Memory (Java)
+#####################################
+
+
+Overview of STM
+===============
+
+An `STM <http://en.wikipedia.org/wiki/Software_transactional_memory>`_ turns the
+Java heap into a transactional data set with begin/commit/rollback
+semantics. Very much like a regular database. It implements the first three
+letters in `ACID`_; ACI:
+
+* Atomic
+* Consistent
+* Isolated
+
+.. _ACID: http://en.wikipedia.org/wiki/ACID
+
+Generally, the STM is not needed very often when working with Akka. Some
+use-cases (that we can think of) are:
+
+- When you really need composable message flows across many actors updating
+  their **internal local** state but need them to do that atomically in one big
+  transaction. Might not be often, but when you do need this then you are
+  screwed without it.
+- When you want to share a datastructure across actors.
+
+The use of STM in Akka is inspired by the concepts and views in `Clojure`_\'s
+STM. Please take the time to read `this excellent document`_ about state in
+clojure and view `this presentation`_ by Rich Hickey (the genius behind
+Clojure).
+
+.. _Clojure: http://clojure.org/
+.. _this excellent document: http://clojure.org/state
+.. _this presentation: http://www.infoq.com/presentations/Value-Identity-State-Rich-Hickey
+
+
+Scala STM
+=========
+
+The STM supported in Akka is `ScalaSTM`_ which will be soon included in the
+Scala standard library.
+
+.. _ScalaSTM: http://nbronson.github.com/scala-stm/
+
+The STM is based on Transactional References (referred to as Refs). Refs are
+memory cells, holding an (arbitrary) immutable value, that implement CAS
+(Compare-And-Swap) semantics and are managed and enforced by the STM for
+coordinated changes across many Refs.
+
+
+Integration with Actors
+=======================
+
+In Akka we've also integrated Actors and STM in :ref:`agents-java` and
+:ref:`transactors-java`.

akka-docs/java/transactors.rst

@@ -1,6 +1,149 @@
 .. _transactors-java:
 
-Transactors (Java)
-==================
+####################
+ Transactors (Java)
+####################
 
-The Akka Transactors module has not been migrated to Akka 2.0-SNAPSHOT yet.
+.. sidebar:: Contents
+
+   .. contents:: :local:
+
+
+Why Transactors?
+================
+
+Actors are excellent for solving problems where you have many independent
+processes that can work in isolation and only interact with other Actors through
+message passing. This model fits many problems. But the actor model is
+unfortunately a terrible model for implementing truly shared state. E.g. when
+you need to have consensus and a stable view of state across many
+components. The classic example is the bank account where clients can deposit
+and withdraw, in which each operation needs to be atomic. For detailed
+discussion on the topic see `this JavaOne presentation
+<http://www.slideshare.net/jboner/state-youre-doing-it-wrong-javaone-2009>`_.
+
+STM on the other hand is excellent for problems where you need consensus and a
+stable view of the state by providing compositional transactional shared
+state. Some of the really nice traits of STM are that transactions compose, and
+it raises the abstraction level from lock-based concurrency.
+
+Akka's Transactors combine Actors and STM to provide the best of the Actor model
+(concurrency and asynchronous event-based programming) and STM (compositional
+transactional shared state) by providing transactional, compositional,
+asynchronous, event-based message flows.
+
+Generally, the STM is not needed very often when working with Akka. Some
+use-cases (that we can think of) are:
+
+- When you really need composable message flows across many actors updating
+  their **internal local** state but need them to do that atomically in one big
+  transaction. Might not be often but when you do need this then you are
+  screwed without it.
+
+- When you want to share a datastructure across actors.
+
+
+Actors and STM
+==============
+
+You can combine Actors and STM in several ways. An Actor may use STM internally
+so that particular changes are guaranteed to be atomic. Actors may also share
+transactional datastructures as the STM provides safe shared state across
+threads.
+
+It's also possible to coordinate transactions across Actors or threads so that
+either the transactions in a set all commit successfully or they all fail. This
+is the focus of Transactors and the explicit support for coordinated
+transactions in this section.
+
+
+Coordinated transactions
+========================
+
+Akka provides an explicit mechanism for coordinating transactions across
+actors. Under the hood it uses a ``CommitBarrier``, similar to a CountDownLatch.
+
+Here is an example of coordinating two simple counter UntypedActors so that they
+both increment together in coordinated transactions. If one of them was to fail
+to increment, the other would also fail.
+
+.. includecode:: code/akka/docs/transactor/Increment.java#class
+   :language: java
+
+.. includecode:: code/akka/docs/transactor/CoordinatedCounter.java#class
+   :language: java
+
+.. includecode:: code/akka/docs/transactor/TransactorDocTest.java#imports
+   :language: java
+
+.. includecode:: code/akka/docs/transactor/TransactorDocTest.java#coordinated-example
+   :language: java
+
+To start a new coordinated transaction that you will also participate in, create
+a ``Coordinated`` object, passing in a ``Timeout``:
+
+.. includecode:: code/akka/docs/transactor/TransactorDocTest.java#create-coordinated
+   :language: java
+
+To start a coordinated transaction that you won't participate in yourself you
+can create a ``Coordinated`` object with a message and send it directly to an
+actor. The recipient of the message will be the first member of the coordination
+set:
+
+.. includecode:: code/akka/docs/transactor/TransactorDocTest.java#send-coordinated
+   :language: java
+
+To include another actor in the same coordinated transaction that you've created
+or received, use the ``coordinate`` method on that object. This will increment
+the number of parties involved by one and create a new ``Coordinated`` object to
+be sent.
+
+.. includecode:: code/akka/docs/transactor/TransactorDocTest.java#include-coordinated
+   :language: java
+
+To enter the coordinated transaction use the atomic method of the coordinated
+object, passing in an ``akka.transactor.Atomically`` object.
+
+.. includecode:: code/akka/docs/transactor/Coordinator.java#coordinated-atomic
+   :language: java
+
+The coordinated transaction will wait for the other transactions before
+committing. If any of the coordinated transactions fail then they all fail.
+
+
+UntypedTransactor
+=================
+
+UntypedTransactors are untyped actors that provide a general pattern for
+coordinating transactions, using the explicit coordination described above.
+
+Here's an example of a simple untyped transactor that will join a coordinated
+transaction:
+
+.. includecode:: code/akka/docs/transactor/Counter.java#class
+   :language: java
+
+You could send this Counter transactor a ``Coordinated(Increment)`` message. If
+you were to send it just an ``Increment`` message it will create its own
+``Coordinated`` (but in this particular case wouldn't be coordinating
+transactions with any other transactors).
+
+To coordinate with other transactors override the ``coordinate`` method. The
+``coordinate`` method maps a message to a set of ``SendTo`` objects, pairs of
+``ActorRef`` and a message. You can use the ``include`` and ``sendTo`` methods
+to easily coordinate with other transactors.
+
+Here's an example of coordinating an increment, using an untyped transactor,
+similar to the explicitly coordinated example above.
+
+.. includecode:: code/akka/docs/transactor/FriendlyCounter.java#class
+   :language: java
+
+To execute directly before or after the coordinated transaction, override the
+``before`` and ``after`` methods. They do not execute within the transaction.
+
+To completely bypass coordinated transactions override the ``normally``
+method. Any message matched by ``normally`` will not be matched by the other
+methods, and will not be involved in coordinated transactions. In this method
+you can implement normal actor behavior, or use the normal STM atomic for local
+transactions.