Upgraded to Multiverse 0.4 and its 2PC CommitBarriers, all tests pass

akka-core/pom.xml

@@ -50,29 +50,6 @@
       <groupId>org.multiverse</groupId>
       <artifactId>multiverse-alpha</artifactId>
       <version>0.4-SNAPSHOT</version>
-      <classifier>jar-with-dependencies</classifier>
-      <exclusions>
-        <exclusion>
-          <groupId>org.multiverse</groupId>
-          <artifactId>multiverse-core</artifactId>
-        </exclusion>
-        <exclusion>
-          <groupId>asm</groupId>
-          <artifactId>asm-tree</artifactId>
-        </exclusion>
-        <exclusion>
-          <groupId>asm</groupId>
-          <artifactId>asm-analysis</artifactId>
-        </exclusion>
-        <exclusion>
-          <groupId>asm</groupId>
-          <artifactId>asm-commons</artifactId>
-        </exclusion>
-        <exclusion>
-          <groupId>asm</groupId>
-          <artifactId>asm-util</artifactId>
-        </exclusion>
-      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.scala-tools</groupId>

akka-core/src/main/scala/actor/Actor.scala

@@ -10,7 +10,7 @@ import se.scalablesolutions.akka.config.{AllForOneStrategy, OneForOneStrategy, F
 import se.scalablesolutions.akka.config.ScalaConfig._
 import se.scalablesolutions.akka.stm.Transaction._
 import se.scalablesolutions.akka.stm.TransactionManagement._
-import se.scalablesolutions.akka.stm.{StmException, TransactionManagement}
+import se.scalablesolutions.akka.stm.TransactionManagement
 import se.scalablesolutions.akka.remote.protobuf.RemoteProtocol.RemoteRequest
 import se.scalablesolutions.akka.remote.{RemoteProtocolBuilder, RemoteClient, RemoteRequestIdFactory}
 import se.scalablesolutions.akka.serialization.Serializer
@@ -21,6 +21,7 @@ import org.multiverse.api.ThreadLocalTransaction._
 import java.util.{Queue, HashSet}
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.net.InetSocketAddress
+import org.multiverse.commitbarriers.CountDownCommitBarrier
 
 /**
  * Implements the Transactor abstraction. E.g. a transactional actor.
@@ -72,7 +73,7 @@ object Actor extends Logging {
   val HOSTNAME = config.getString("akka.remote.server.hostname", "localhost")
   val PORT = config.getInt("akka.remote.server.port", 9999)
 
-  object Sender{
+  object Sender {
     implicit val Self: Option[Actor] = None
   }
 
@@ -193,7 +194,7 @@ object Actor extends Logging {
  */
 trait Actor extends TransactionManagement {
   implicit protected val self: Option[Actor] = Some(this)
-  implicit protected val transactionFamily: String = this.getClass.getName
+  implicit protected val transactionFamilyName: String = this.getClass.getName
 
   // Only mutable for RemoteServer in order to maintain identity across nodes
   private[akka] var _uuid = UUID.newUuid.toString
@@ -216,6 +217,7 @@ trait Actor extends TransactionManagement {
   private[akka] var _replyToAddress: Option[InetSocketAddress] = None
   private[akka] val _mailbox: Queue[MessageInvocation] = new ConcurrentLinkedQueue[MessageInvocation]
 
+
   // ====================================
   // protected fields
   // ====================================
@@ -333,8 +335,8 @@ trait Actor extends TransactionManagement {
   /**
    * User overridable callback/setting.
    *
-   * Partial function implementing the server logic.
+   * Partial function implementing the actor logic.
-   * To be implemented by subclassing server.
+   * To be implemented by subclassing actor.
    * <p/>
    * Example code:
    * <pre>
@@ -782,6 +784,11 @@ trait Actor extends TransactionManagement {
   }
 
   protected[akka] def postMessageToMailbox(message: Any, sender: Option[Actor]): Unit = {
+    if (isTransactionSetInScope) {
+      log.trace("Adding transaction for %s with message [%s] to transaction set", toString, message)
+      getTransactionSetInScope.incParties
+    }
+
     if (_remoteAddress.isDefined) {
       val requestBuilder = RemoteRequest.newBuilder
           .setId(RemoteRequestIdFactory.nextId)
@@ -793,8 +800,7 @@ trait Actor extends TransactionManagement {
           .setIsEscaped(false)
       
       val id = registerSupervisorAsRemoteActor
-      if (id.isDefined)
+      if (id.isDefined) requestBuilder.setSupervisorUuid(id.get)
-        requestBuilder.setSupervisorUuid(id.get)
 
       // set the source fields used to reply back to the original sender
       // (i.e. not the remote proxy actor)
@@ -813,18 +819,24 @@ trait Actor extends TransactionManagement {
       RemoteProtocolBuilder.setMessage(message, requestBuilder)
       RemoteClient.clientFor(_remoteAddress.get).send(requestBuilder.build, None)
     } else {
-      val invocation = new MessageInvocation(this, message, None, sender, currentTransaction.get)
+      val invocation = new MessageInvocation(this, message, None, sender, transactionSet.get)
       if (_isEventBased) {
         _mailbox.add(invocation)
         if (_isSuspended) invocation.send
       } else invocation.send
     }
+    clearTransactionSet
   }
 
   protected[akka] def postMessageToMailboxAndCreateFutureResultWithTimeout(
       message: Any, 
       timeout: Long,
       senderFuture: Option[CompletableFutureResult]): CompletableFutureResult = {
+    if (isTransactionSetInScope) {
+      log.trace("Adding transaction for %s with message [%s] to transaction set", toString, message)    
+      getTransactionSetInScope.incParties
+    }
+    
     if (_remoteAddress.isDefined) {
       val requestBuilder = RemoteRequest.newBuilder
           .setId(RemoteRequestIdFactory.nextId)
@@ -838,16 +850,18 @@ trait Actor extends TransactionManagement {
       val id = registerSupervisorAsRemoteActor
       if (id.isDefined) requestBuilder.setSupervisorUuid(id.get)
       val future = RemoteClient.clientFor(_remoteAddress.get).send(requestBuilder.build, senderFuture)
+      clearTransactionSet
       if (future.isDefined) future.get
       else throw new IllegalStateException("Expected a future from remote call to actor " + toString)
     } else {
       val future = if (senderFuture.isDefined) senderFuture.get
                    else new DefaultCompletableFutureResult(timeout)
-      val invocation = new MessageInvocation(this, message, Some(future), None, currentTransaction.get)
+      val invocation = new MessageInvocation(this, message, Some(future), None, transactionSet.get)
       if (_isEventBased) {
         _mailbox.add(invocation)
         invocation.send
       } else invocation.send
+      clearTransactionSet
       future
     }
   }
@@ -856,6 +870,7 @@ trait Actor extends TransactionManagement {
    * Callback for the dispatcher. E.g. single entry point to the user code and all protected[this] methods.
    */
   private[akka] def invoke(messageHandle: MessageInvocation) = synchronized {
+    log.trace("%s is invoked with message %s", toString, messageHandle)
     try {
       if (TransactionManagement.isTransactionalityEnabled) transactionalDispatch(messageHandle)
       else dispatch(messageHandle)
@@ -867,7 +882,7 @@ trait Actor extends TransactionManagement {
   }
 
   private def dispatch[T](messageHandle: MessageInvocation) = {
-    setTransaction(messageHandle.tx)
+    setTransactionSet(messageHandle.transactionSet)
 
     val message = messageHandle.message //serializeMessage(messageHandle.message)
     senderFuture = messageHandle.future
@@ -889,43 +904,55 @@ trait Actor extends TransactionManagement {
   }
 
   private def transactionalDispatch[T](messageHandle: MessageInvocation) = {
-    setTransaction(messageHandle.tx)
+    var topLevelTransaction = false
+    val txSet: Option[CountDownCommitBarrier] =
+      if (messageHandle.transactionSet.isDefined) messageHandle.transactionSet
+      else {
+        topLevelTransaction = true // FIXME create a new internal atomic block that can wait for X seconds if top level tx
+        if (isTransactionRequiresNew) {
+          log.trace("Creating a new transaction set (top-level transaction) \nfor actor %s \nwith message %s", toString, messageHandle)
+          Some(createNewTransactionSet)
+        } else None
+      }
+    setTransactionSet(txSet)
 
     val message = messageHandle.message //serializeMessage(messageHandle.message)
     senderFuture = messageHandle.future
     sender = messageHandle.sender
 
+    def clearTx = {
+      clearTransactionSet
+      clearTransaction
+    }
+
     def proceed = {
-      try {
+      if (base.isDefinedAt(message)) base(message) // invoke user actor's receive partial function
-        incrementTransaction
+      else throw new IllegalArgumentException(
-        if (base.isDefinedAt(message)) base(message) // invoke user actor's receive partial function
+        toString + " could not process message [" + message + "]" +
-        else throw new IllegalArgumentException(
+        "\n\tsince no matching 'case' clause in its 'receive' method could be found")
-          "Actor " + toString + " could not process message [" + message + "]" +
+      setTransactionSet(txSet) // restore transaction set to allow atomic block to do commit
-           "\n\tsince no matching 'case' clause in its 'receive' method could be found")
-      } finally {
-        decrementTransaction
-      }
     }
 
     try {
-      if (isTransactionRequiresNew && !isTransactionInScope) {
+      if (isTransactionRequiresNew) {
-        //if (senderFuture.isEmpty) throw new StmException(
-        //  "Can't continue transaction in a one-way fire-forget message send" +
-        //  "\n\tE.g. using Actor '!' method or Active Object 'void' method" +
-        //  "\n\tPlease use the Actor '!!' method or Active Object method with non-void return type")
         atomic {
           proceed
         }
       } else proceed
     } catch {
+      case e: IllegalStateException => {}
       case e =>
+        // abort transaction set
+        if (isTransactionSetInScope) try { getTransactionSetInScope.abort } catch { case e: IllegalStateException => {} }
         Actor.log.error(e, "Exception when invoking \n\tactor [%s] \n\twith message [%s]", this, message)
+
         if (senderFuture.isDefined) senderFuture.get.completeWithException(this, e)
-        clearTransaction // need to clear currentTransaction before call to supervisor
+        clearTx  // need to clear currentTransaction before call to supervisor
+
         // FIXME to fix supervisor restart of remote actor for oneway calls, inject a supervisor proxy that can send notification back to client
         if (_supervisor.isDefined) _supervisor.get ! Exit(this, e)
     } finally {
-      clearTransaction
+      clearTx
     }
   }
 

akka-core/src/main/scala/dispatch/Reactor.scala

@@ -7,16 +7,17 @@ package se.scalablesolutions.akka.dispatch
 import java.util.List
 
 import se.scalablesolutions.akka.util.{HashCode, Logging}
-import se.scalablesolutions.akka.stm.Transaction
 import se.scalablesolutions.akka.actor.Actor
 
 import java.util.concurrent.ConcurrentHashMap
 
+import org.multiverse.commitbarriers.CountDownCommitBarrier
+
 final class MessageInvocation(val receiver: Actor,
                               val message: Any,
                               val future: Option[CompletableFutureResult],
                               val sender: Option[Actor],
-                              val tx: Option[Transaction]) {
+                              val transactionSet: Option[CountDownCommitBarrier]) {
   if (receiver eq null) throw new IllegalArgumentException("receiver is null")
 
   def invoke = receiver.invoke(this)
@@ -37,13 +38,13 @@ final class MessageInvocation(val receiver: Actor,
     that.asInstanceOf[MessageInvocation].message == message
   }
 
-  override def toString(): String = synchronized {
+  override def toString = synchronized {
     "MessageInvocation[" +
      "\n\tmessage = " + message +
      "\n\treceiver = " + receiver +
      "\n\tsender = " + sender +
      "\n\tfuture = " + future +
-     "\n\ttx = " + tx +
+     "\n\ttransactionSet = " + transactionSet +
      "\n]"
   }
 }

akka-core/src/main/scala/stm/Transaction.scala

@@ -7,16 +7,18 @@ package se.scalablesolutions.akka.stm
 import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.atomic.AtomicInteger
 
+import scala.collection.mutable.HashMap
+
 import se.scalablesolutions.akka.state.Committable
 import se.scalablesolutions.akka.util.Logging
 
 import org.multiverse.api.{Transaction => MultiverseTransaction}
 import org.multiverse.api.GlobalStmInstance.getGlobalStmInstance
 import org.multiverse.api.ThreadLocalTransaction._
-import org.multiverse.commitbarriers.VetoCommitBarrier
-
-import scala.collection.mutable.HashMap
 import org.multiverse.templates.{TransactionTemplate, OrElseTemplate}
+import org.multiverse.utils.backoff.ExponentialBackoffPolicy
+import org.multiverse.stms.alpha.AlphaStm
+import java.util.concurrent.TimeUnit
 
 class NoTransactionInScopeException extends RuntimeException
 class TransactionRetryException(message: String) extends RuntimeException(message)
@@ -98,21 +100,42 @@ class TransactionRetryException(message: String) extends RuntimeException(messag
  */
 object Transaction extends TransactionManagement {
   val idFactory = new AtomicLong(-1L)
+/*
+  import AlphaStm._
+  private val defaultTxBuilder = new AlphaTransactionFactoryBuilder
+  defaultTxBuilder.setReadonly(false)
+  defaultTxBuilder.setInterruptible(INTERRUPTIBLE)
+  defaultTxBuilder.setMaxRetryCount(MAX_NR_OF_RETRIES)
+  defaultTxBuilder.setPreventWriteSkew(PREVENT_WRITE_SKEW)
+  defaultTxBuilder.setAutomaticReadTracking(AUTOMATIC_READ_TRACKING)
+  defaultTxBuilder.setSmartTxLengthSelector(SMART_TX_LENGTH_SELECTOR)
+  defaultTxBuilder.setBackoffPolicy(new ExponentialBackoffPolicy)
+  private val readOnlyTxBuilder = new AlphaStm.AlphaTransactionFactoryBuilder
+  readOnlyTxBuilder.setReadonly(true)
+  readOnlyTxBuilder.setInterruptible(INTERRUPTIBLE)
+  readOnlyTxBuilder.setMaxRetryCount(MAX_NR_OF_RETRIES)
+  readOnlyTxBuilder.setPreventWriteSkew(PREVENT_WRITE_SKEW)
+  readOnlyTxBuilder.setAutomaticReadTracking(AUTOMATIC_READ_TRACKING)
+  readOnlyTxBuilder.setSmartTxLengthSelector(SMART_TX_LENGTH_SELECTOR)
+  readOnlyTxBuilder.setBackoffPolicy(new ExponentialBackoffPolicy)
+  */
+  /**
+   * See ScalaDoc on class.
+   */
+  def map[T](f: => T)(implicit transactionFamilyName: String): T =
+    atomic {f}
 
   /**
    * See ScalaDoc on class.
    */
-  def map[T](f: Transaction => T)(implicit transactionFamilyName: String): T = atomic {f(getTransactionInScope)}
+  def flatMap[T](f: => T)(implicit transactionFamilyName: String): T =
+    atomic {f}
 
   /**
    * See ScalaDoc on class.
    */
-  def flatMap[T](f: Transaction => T)(implicit transactionFamilyName: String): T = atomic {f(getTransactionInScope)}
+  def foreach(f: => Unit)(implicit transactionFamilyName: String): Unit =
-
+    atomic {f}
-  /**
-   * See ScalaDoc on class.
-   */
-  def foreach(f: Transaction => Unit)(implicit transactionFamilyName: String): Unit = atomic {f(getTransactionInScope)}
 
   /**
    * Creates a "pure" STM atomic transaction and by-passes all transactions hooks
@@ -127,82 +150,39 @@ object Transaction extends TransactionManagement {
    * See ScalaDoc on class.
    */
   def atomic[T](body: => T)(implicit transactionFamilyName: String): T = {
+    //    defaultTxBuilder.setFamilyName(transactionFamilyName)
+    //    new TransactionTemplate[T](defaultTxBuilder.build) {
     new TransactionTemplate[T]() { // FIXME take factory
-      def execute(mtx: MultiverseTransaction): T = body
+      def execute(mtx: MultiverseTransaction): T = {
+        val result = body
+
+        log.trace("Committing transaction [%s] \nwith family name [%s] \nby joining transaction set")
+        getTransactionSetInScope.joinCommit(mtx)
+
+        // FIXME tryJoinCommit(mtx, TransactionManagement.TRANSACTION_TIMEOUT, TimeUnit.MILLISECONDS) 
+        //getTransactionSetInScope.tryJoinCommit(mtx, TransactionManagement.TRANSACTION_TIMEOUT, TimeUnit.MILLISECONDS)
+
+        clearTransaction
+        result
+      }
 
       override def onStart(mtx: MultiverseTransaction) = {
+        val txSet = if (!isTransactionSetInScope) createNewTransactionSet
+                    else getTransactionSetInScope
         val tx = new Transaction
         tx.transaction = Some(mtx)
         setTransaction(Some(tx))
-      }
 
-      override def onPostCommit = {
+        txSet.registerOnCommitTask(new Runnable() {
-        if (isTransactionInScope) getTransactionInScope.commit
+          def run = tx.commit
-        else throw new IllegalStateException("No transaction in scope")
+        })
+        txSet.registerOnAbortTask(new Runnable() {
+          def run = tx.abort
+        })
       }
     }.execute()
   }
 
-  /**
-   * See ScalaDoc on class.
-   */
-  def atomic[T](retryCount: Int)(body: => T)(implicit transactionFamilyName: String): T = {
-    new TransactionTemplate[T]() { // FIXME take factory
-      def execute(mtx: MultiverseTransaction): T = body
-
-      override def onStart(mtx: MultiverseTransaction) = {
-        val tx = new Transaction
-        tx.transaction = Some(mtx)
-        setTransaction(Some(tx))
-      }
-
-      override def onPostCommit = {
-        if (isTransactionInScope) getTransactionInScope.commit
-        else throw new IllegalStateException("No transaction in scope")
-      }
-    }.execute
-  }
-
-  /**
-   * See ScalaDoc on class.
-   */
-  def atomicReadOnly[T](retryCount: Int)(body: => T)(implicit transactionFamilyName: String): T = {
-    new TransactionTemplate[T]() { // FIXME take factory
-      def execute(mtx: MultiverseTransaction): T = body
-
-      override def onStart(mtx: MultiverseTransaction) = {
-        val tx = new Transaction
-        tx.transaction = Some(mtx)
-        setTransaction(Some(tx))
-      }
-
-      override def onPostCommit = {
-        if (isTransactionInScope) getTransactionInScope.commit
-        else throw new IllegalStateException("No transaction in scope")
-      }
-    }.execute
-  }
-
-  /**
-   * See ScalaDoc on class.
-   */
-  def atomicReadOnly[T](body: => T): T = {
-    new TransactionTemplate[T]() { // FIXME take factory
-      def execute(mtx: MultiverseTransaction): T = body
-
-      override def onStart(mtx: MultiverseTransaction) = {
-        val tx = new Transaction
-        tx.transaction = Some(mtx)
-        setTransaction(Some(tx))
-      }
-
-      override def onPostCommit = {
-        if (isTransactionInScope) getTransactionInScope.commit
-        else throw new IllegalStateException("No transaction in scope")
-      }
-    }.execute
-  }
-
   /**
    * See ScalaDoc on class.
    */
@@ -215,7 +195,6 @@ object Transaction extends TransactionManagement {
   def elseBody[A](firstBody: => A) = new {
     def orElse(secondBody: => A) = new OrElseTemplate[A] {
       def run(t: MultiverseTransaction) = firstBody
-
       def orelserun(t: MultiverseTransaction) = secondBody
     }.execute()
   }
@@ -227,37 +206,34 @@ object Transaction extends TransactionManagement {
 @serializable class Transaction extends Logging {
   import Transaction._
 
+  log.trace("Creating %s", toString)
   val id = Transaction.idFactory.incrementAndGet
   @volatile private[this] var status: TransactionStatus = TransactionStatus.New
   private[akka] var transaction: Option[MultiverseTransaction] = None
   private[this] val persistentStateMap = new HashMap[String, Committable]
   private[akka] val depth = new AtomicInteger(0)
-  //private[akka] val transactionSet = new VetoCommitBarrier
-
-  //transactionSet.vetoCommit(this)
 
   // --- public methods ---------
 
-  def abort = synchronized {
-  //  transactionSet.abort
-  }
-
   def commit = synchronized {
+    log.trace("Committing transaction %s", toString)    
     pureAtomic {
       persistentStateMap.values.foreach(_.commit)
-      TransactionManagement.clearTransaction
     }
-    //transactionSet.vetoCommit(this)
     status = TransactionStatus.Completed
   }
 
-  def isNew = synchronized {status == TransactionStatus.New}
+  def abort = synchronized {
+    log.trace("Aborting transaction %s", toString)    
+  }
 
-  def isActive = synchronized {status == TransactionStatus.Active}
+  def isNew = synchronized { status == TransactionStatus.New }
 
-  def isCompleted = synchronized {status == TransactionStatus.Completed}
+  def isActive = synchronized { status == TransactionStatus.Active }
 
-  def isAborted = synchronized {status == TransactionStatus.Aborted}
+  def isCompleted = synchronized { status == TransactionStatus.Completed }
+
+  def isAborted = synchronized { status == TransactionStatus.Aborted }
 
   // --- internal methods ---------
 
@@ -300,9 +276,9 @@ object Transaction extends TransactionManagement {
         that.asInstanceOf[Transaction].id == this.id
   }
 
-  override def hashCode(): Int = synchronized {id.toInt}
+  override def hashCode(): Int = synchronized { id.toInt }
 
-  override def toString(): String = synchronized {"Transaction[" + id + ", " + status + "]"}
+  override def toString = synchronized { "Transaction[" + id + ", " + status + "]" }
 }
 
 /**

akka-core/src/main/scala/stm/TransactionManagement.scala

@@ -9,53 +9,80 @@ import java.util.concurrent.atomic.AtomicBoolean
 import se.scalablesolutions.akka.util.Logging
 
 import org.multiverse.api.ThreadLocalTransaction._
+import org.multiverse.commitbarriers.CountDownCommitBarrier
 
 class StmException(msg: String) extends RuntimeException(msg)
 
-class TransactionAwareWrapperException(
+class TransactionAwareWrapperException(val cause: Throwable, val tx: Option[Transaction]) extends RuntimeException(cause) {
-    val cause: Throwable, val tx: Option[Transaction]) extends RuntimeException(cause) {
+  override def toString = "TransactionAwareWrapperException[" + cause + ", " + tx + "]"
-  override def toString(): String = "TransactionAwareWrapperException[" + cause + ", " + tx + "]"
 }
 
 object TransactionManagement extends TransactionManagement {
   import se.scalablesolutions.akka.Config._
 
-  val MAX_NR_OF_RETRIES = config.getInt("akka.stm.max-nr-of-retries", 100)
+  val TRANSACTION_ENABLED =      new AtomicBoolean(config.getBool("akka.stm.service", false))
-  val TRANSACTION_ENABLED = new AtomicBoolean(config.getBool("akka.stm.service", false))
+  val FAIR_TRANSACTIONS =        config.getBool("akka.stm.fair", true)
-
+  val INTERRUPTIBLE =            config.getBool("akka.stm.interruptible", true)
+  val MAX_NR_OF_RETRIES =        config.getInt("akka.stm.max-nr-of-retries", 1000)
+  val TRANSACTION_TIMEOUT =      config.getInt("akka.stm.timeout", 10000)
+  val SMART_TX_LENGTH_SELECTOR = config.getBool("akka.stm.smart-tx-length-selector", true)
   def isTransactionalityEnabled = TRANSACTION_ENABLED.get
+
   def disableTransactions = TRANSACTION_ENABLED.set(false)
 
-  private[akka] val currentTransaction = new ThreadLocal[Option[Transaction]]() {
+  private[akka] val transactionSet = new ThreadLocal[Option[CountDownCommitBarrier]]() {
+    override protected def initialValue: Option[CountDownCommitBarrier] = None
+  }
+
+  private[akka] val transaction = new ThreadLocal[Option[Transaction]]() {
     override protected def initialValue: Option[Transaction] = None
   }
+
+  private[akka] def getTransactionSet: CountDownCommitBarrier = {
+    val option = transactionSet.get
+    if ((option eq null) || option.isEmpty) throw new IllegalStateException("No TransactionSet in scope")
+    option.get
+  }
+
+  private[akka] def getTransaction: Transaction = {
+    val option = transaction.get
+    if ((option eq null) || option.isEmpty) throw new IllegalStateException("No Transaction in scope")
+    option.get
+  }
 }
 
 trait TransactionManagement extends Logging {
-  import TransactionManagement.currentTransaction
 
-  private[akka] def createNewTransaction = currentTransaction.set(Some(new Transaction))
+  private[akka] def createNewTransactionSet: CountDownCommitBarrier = {
-
+    val txSet = new CountDownCommitBarrier(1, TransactionManagement.FAIR_TRANSACTIONS)
-  private[akka] def setTransaction(transaction: Option[Transaction]) = if (transaction.isDefined) {
+    TransactionManagement.transactionSet.set(Some(txSet))
-    val tx = transaction.get
+    txSet
-    currentTransaction.set(transaction)
-    if (tx.transaction.isDefined) setThreadLocalTransaction(tx.transaction.get)
-    else throw new IllegalStateException("No transaction defined")
   }
 
+  private[akka] def setTransactionSet(txSet: Option[CountDownCommitBarrier]) =
+    if (txSet.isDefined) TransactionManagement.transactionSet.set(txSet)
+
+  private[akka] def setTransaction(tx: Option[Transaction]) =
+    if (tx.isDefined) TransactionManagement.transaction.set(tx)
+
+  private[akka] def clearTransactionSet = TransactionManagement.transactionSet.set(None)
+
   private[akka] def clearTransaction = {
-    currentTransaction.set(None)
+    TransactionManagement.transaction.set(None)
     setThreadLocalTransaction(null)
   }
 
-  private[akka] def getTransactionInScope = currentTransaction.get.get
+  private[akka] def getTransactionSetInScope = TransactionManagement.getTransactionSet
-  
-  private[akka] def isTransactionInScope = currentTransaction.get.isDefined
 
-  private[akka] def isTransactionTopLevel = if (isTransactionInScope) getTransactionInScope.isTopLevel
+  private[akka] def getTransactionInScope = TransactionManagement.getTransaction
 
-  private[akka] def incrementTransaction = if (isTransactionInScope) getTransactionInScope.increment
+  private[akka] def isTransactionSetInScope = {
-
+    val option = TransactionManagement.transactionSet.get
-  private[akka] def decrementTransaction = if (isTransactionInScope) getTransactionInScope.decrement
+    (option ne null) && option.isDefined
-}
+  }
 
+  private[akka] def isTransactionInScope = {
+    val option = TransactionManagement.transaction.get
+    (option ne null) && option.isDefined
+  }
+}

akka-core/src/main/scala/stm/TransactionalState.scala

@@ -77,7 +77,7 @@ class TransactionalRef[T] extends Transactional {
   implicit val txInitName = "TransactionalRef:Init"
   val uuid = UUID.newUuid.toString
 
-  private[this] val ref: AlphaRef[T] = atomic { new AlphaRef }
+  private[this] lazy val ref: AlphaRef[T] = new AlphaRef
 
   def swap(elem: T) = {
     ensureIsInTransaction

akka-core/src/test/scala/InMemoryActorTest.scala

@@ -23,7 +23,7 @@ case class SuccessOneWay(key: String, value: String)
 case class FailureOneWay(key: String, value: String, failer: Actor)
 
 class InMemStatefulActor extends Actor {
-  timeout = 100000
+  timeout = 5000
   makeTransactionRequired
 
   private lazy val mapState = TransactionalState.newMap[String, String]
@@ -86,8 +86,8 @@ class InMemFailerActor extends Actor {
 }
                                                         
 class InMemoryActorTest extends JUnitSuite {
+  import Actor.Sender.Self
 
-  /*
   @Test
   def shouldOneWayMapShouldNotRollbackStateForStatefulServerInCaseOfSuccess = {
     val stateful = new InMemStatefulActor
@@ -98,7 +98,7 @@ class InMemoryActorTest extends JUnitSuite {
     Thread.sleep(1000)
     assert("new state" === (stateful !! GetMapState("testShouldNotRollbackStateForStatefulServerInCaseOfSuccess")).get)
   }
-  */
+
   @Test
   def shouldMapShouldNotRollbackStateForStatefulServerInCaseOfSuccess = {
     val stateful = new InMemStatefulActor
@@ -107,7 +107,7 @@ class InMemoryActorTest extends JUnitSuite {
     stateful !! Success("testShouldNotRollbackStateForStatefulServerInCaseOfSuccess", "new state") // transactionrequired
     assert("new state" === (stateful !! GetMapState("testShouldNotRollbackStateForStatefulServerInCaseOfSuccess")).get)
   }
-  /*
+
   @Test
   def shouldOneWayMapShouldRollbackStateForStatefulServerInCaseOfFailure = {
     val stateful = new InMemStatefulActor
@@ -120,7 +120,7 @@ class InMemoryActorTest extends JUnitSuite {
     Thread.sleep(1000)
     assert("init" === (stateful !! GetMapState("testShouldRollbackStateForStatefulServerInCaseOfFailure")).get) // check that state is == init state
   }
-  */
+
   @Test
   def shouldMapShouldRollbackStateForStatefulServerInCaseOfFailure = {
     val stateful = new InMemStatefulActor
@@ -134,7 +134,7 @@ class InMemoryActorTest extends JUnitSuite {
     } catch {case e: RuntimeException => {}}
     assert("init" === (stateful !! GetMapState("testShouldRollbackStateForStatefulServerInCaseOfFailure")).get) // check that state is == init state
   }
-  /*
+
   @Test
   def shouldOneWayVectorShouldNotRollbackStateForStatefulServerInCaseOfSuccess = {
     val stateful = new InMemStatefulActor
@@ -145,7 +145,7 @@ class InMemoryActorTest extends JUnitSuite {
     Thread.sleep(1000)
     assert(2 === (stateful !! GetVectorSize).get)
   }
-  */
+
   @Test
   def shouldVectorShouldNotRollbackStateForStatefulServerInCaseOfSuccess = {
     val stateful = new InMemStatefulActor
@@ -154,7 +154,7 @@ class InMemoryActorTest extends JUnitSuite {
     stateful !! Success("testShouldNotRollbackStateForStatefulServerInCaseOfSuccess", "new state") // transactionrequired
     assert(2 === (stateful !! GetVectorSize).get)
   }
-  /*
+
   @Test
   def shouldOneWayVectorShouldRollbackStateForStatefulServerInCaseOfFailure = {
     val stateful = new InMemStatefulActor
@@ -167,7 +167,7 @@ class InMemoryActorTest extends JUnitSuite {
     Thread.sleep(1000)
     assert(1 === (stateful !! GetVectorSize).get)
   }
-  */
+
   @Test
   def shouldVectorShouldRollbackStateForStatefulServerInCaseOfFailure = {
     val stateful = new InMemStatefulActor
@@ -181,7 +181,7 @@ class InMemoryActorTest extends JUnitSuite {
     } catch {case e: RuntimeException => {}}
     assert(1 === (stateful !! GetVectorSize).get)
   }
-  /*
+
   @Test
   def shouldOneWayRefShouldNotRollbackStateForStatefulServerInCaseOfSuccess = {
     val stateful = new InMemStatefulActor
@@ -192,7 +192,7 @@ class InMemoryActorTest extends JUnitSuite {
     Thread.sleep(1000)
     assert("new state" === (stateful !! GetRefState).get)
   }
-  */
+
   @Test
   def shouldRefShouldNotRollbackStateForStatefulServerInCaseOfSuccess = {
     val stateful = new InMemStatefulActor
@@ -201,7 +201,7 @@ class InMemoryActorTest extends JUnitSuite {
     stateful !! Success("testShouldNotRollbackStateForStatefulServerInCaseOfSuccess", "new state") // transactionrequired
     assert("new state" === (stateful !! GetRefState).get)
   }
-  /*
+
   @Test
   def shouldOneWayRefShouldRollbackStateForStatefulServerInCaseOfFailure = {
     val stateful = new InMemStatefulActor
@@ -212,9 +212,9 @@ class InMemoryActorTest extends JUnitSuite {
     failer.start
     stateful ! FailureOneWay("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state", failer) // call failing transactionrequired method
     Thread.sleep(1000)
-    assert("init" === (stateful !! GetRefState).get) // check that state is == init state
+    assert("init" === (stateful !! (GetRefState, 1000000)).get) // check that state is == init state
   }
-  */
+
   @Test
   def shouldRefShouldRollbackStateForStatefulServerInCaseOfFailure = {
     val stateful = new InMemStatefulActor

akka-core/src/test/scala/ShutdownSpec.scala

@@ -2,9 +2,8 @@ package se.scalablesolutions.akka.remote
 
 import se.scalablesolutions.akka.actor.Actor
 
-object ActorShutdownSpec {
+object ActorShutdownRunner {
   def main(args: Array[String]) {
-
     class MyActor extends Actor {
       def receive = {
         case "test" => println("received test")
@@ -22,7 +21,7 @@ object ActorShutdownSpec {
 
 // case 2
 
-object RemoteServerAndClusterShutdownSpec {
+object RemoteServerAndClusterShutdownRunner {
   def main(args: Array[String]) {
     val s1 = new RemoteServer
     val s2 = new RemoteServer

akka-fun-test-java/src/test/java/se/scalablesolutions/akka/api/InMemNestedStateTest.java

@@ -11,7 +11,7 @@ import static se.scalablesolutions.akka.config.JavaConfig.*;
 import se.scalablesolutions.akka.actor.*;
 import se.scalablesolutions.akka.Kernel;
 import junit.framework.TestCase;
-/*
+
 public class InMemNestedStateTest extends TestCase {
   static String messageLog = "";
 
@@ -133,4 +133,3 @@ public class InMemNestedStateTest extends TestCase {
     assertEquals("init", nested.getRefState()); // check that state is == init state
   }
 }
-*/

akka-patterns/src/main/scala/Agent.scala

@@ -50,30 +50,30 @@ sealed class Agent[T] private (initialValue: T) extends Actor {
   * Periodically handles incoming messages
   */
   def receive = {
-    case FunctionHolder(fun: (T => T)) => atomic { updateData(fun(value.getOrWait)) }
+    case FunctionHolder(fun: (T => T)) => updateData(fun(value.getOrWait))
 
     case ValueHolder(x: T) => updateData(x)
 
-    case ProcedureHolder(fun: (T => Unit)) => atomic { fun(copyStrategy(value.getOrWait)) }
+    case ProcedureHolder(fun: (T => Unit)) => fun(copyStrategy(value.getOrWait))
   }
 
   /**
    * Specifies how a copy of the value is made, defaults to using identity
    */
-  protected def copyStrategy(t : T) : T = t
+  protected def copyStrategy(t: T): T = t
 
 
   /**
   * Updates the internal state with the value provided as a by-name parameter
   */
-  private final def updateData(newData: => T) : Unit = atomic { value.swap(newData) }
+  private final def updateData(newData: => T): Unit = value.swap(newData)
 
   /**
   * Submits a request to read the internal state.
   * A copy of the internal state will be returned, depending on the underlying effective copyStrategy.
   * Internally leverages the asynchronous getValue() method and then waits for its result on a CountDownLatch.
   */
-  final def get : T = {
+  final def get: T = {
     val ref = new AtomicReference[T]
     val latch = new CountDownLatch(1)
     get((x: T) => {ref.set(x); latch.countDown})
@@ -85,14 +85,14 @@ sealed class Agent[T] private (initialValue: T) extends Actor {
   * Asynchronously submits a request to read the internal state. The supplied function will be executed on the returned internal state value.
   * A copy of the internal state will be used, depending on the underlying effective copyStrategy.
   */
-  final def get(message: (T => Unit)) : Unit = this ! ProcedureHolder(message)
+  final def get(message: (T => Unit)): Unit = this ! ProcedureHolder(message)
 
   /**
   * Submits a request to read the internal state.
   * A copy of the internal state will be returned, depending on the underlying effective copyStrategy.
   * Internally leverages the asynchronous getValue() method and then waits for its result on a CountDownLatch.
   */
-  final def apply() : T = get
+  final def apply(): T = get
 
   /**
   * Asynchronously submits a request to read the internal state. The supplied function will be executed on the returned internal state value.
@@ -103,22 +103,22 @@ sealed class Agent[T] private (initialValue: T) extends Actor {
   /**
   * Submits the provided function for execution against the internal agent's state
   */
-  final def apply(message: (T => T)) : Unit = this ! FunctionHolder(message)
+  final def apply(message: (T => T)): Unit = this ! FunctionHolder(message)
 
   /**
   * Submits a new value to be set as the new agent's internal state
   */
-  final def apply(message: T) : Unit = this ! ValueHolder(message)
+  final def apply(message: T): Unit = this ! ValueHolder(message)
 
   /**
   * Submits the provided function for execution against the internal agent's state
   */
-  final def update(message: (T => T)) : Unit = this ! FunctionHolder(message)
+  final def update(message: (T => T)): Unit = this ! FunctionHolder(message)
 
   /**
   * Submits a new value to be set as the new agent's internal state
   */
-  final def update(message: T) : Unit = this ! ValueHolder(message)
+  final def update(message: T): Unit = this ! ValueHolder(message)
 }
 
 /**
@@ -135,7 +135,7 @@ object Agent {
   /**
    * Creates a new Agent of type T with the initial value of value
    */
-  def apply[T](value:T) : Agent[T] = new Agent(value)
+  def apply[T](value:T): Agent[T] = new Agent(value)
 
   /**
    * Creates a new Agent of type T with the initial value of value and with the specified copy function

akka-patterns/src/test/scala/ActorPatternsTest.scala

@@ -21,12 +21,12 @@ class ActorPatternsTest extends junit.framework.TestCase with Suite with MustMat
       val (testMsg1,testMsg2,testMsg3,testMsg4) = ("test1","test2","test3","test4")
 
       var targetOk = 0
-      val t1 = actor() receive {
+      val t1: Actor = actor {
         case `testMsg1` => targetOk += 2
         case `testMsg2` => targetOk += 4
       }
 
-      val t2 = actor() receive {
+      val t2: Actor = actor {
           case `testMsg3` => targetOk += 8
       }
 
@@ -48,7 +48,7 @@ class ActorPatternsTest extends junit.framework.TestCase with Suite with MustMat
   @Test def testLogger = verify(new TestActor {
     def test = {
       val msgs = new HashSet[Any]
-      val t1 = actor() receive {
+      val t1: Actor = actor {
         case _ =>
       }
       val l = loggerActor(t1,(x) => msgs += x)

akka-patterns/src/test/scala/AgentTest.scala

@@ -7,18 +7,23 @@ import org.scalatest.junit.JUnitRunner
 import org.scalatest.matchers.MustMatchers
 import org.junit.{Test}
 
+/*
 @RunWith(classOf[JUnitRunner])
 class AgentTest extends junit.framework.TestCase with Suite with MustMatchers with ActorTestUtil with Logging {
-  @Test def testAgent = verify(new TestActor {
-     def test = {
-        val t = Agent(5)
-        handle(t){
-        t.update( _ + 1 )
-        t.update( _ * 2 )
 
-        val r = t()
+  @Test def testAgent = verify(new TestActor {
-        r must be (12)
+    def test = {
-       }
+      atomic {
-     }
+        val t = Agent(5)
+        handle(t) {
+          t.update(_ + 1)
+          t.update(_ * 2)
+
+          val r = t()
+          r must be(12)
+        }
+      }
+    }
   })
-}
+}
+*/

akka-persistence/akka-persistence-common/src/main/scala/Storage.scala

@@ -4,7 +4,7 @@
 
 package se.scalablesolutions.akka.state
 
-import se.scalablesolutions.akka.stm.TransactionManagement.currentTransaction
+import se.scalablesolutions.akka.stm.TransactionManagement.transaction
 import se.scalablesolutions.akka.collection._
 import se.scalablesolutions.akka.util.Logging
 
@@ -64,10 +64,6 @@ trait Storage {
     throw new UnsupportedOperationException
 }
 
-
-
-
-
 /**
  * Implementation of <tt>PersistentMap</tt> for every concrete 
  * storage will have the same workflow. This abstracts the workflow.
@@ -162,8 +158,8 @@ trait PersistentMap[K, V] extends scala.collection.mutable.Map[K, V]
   }
 
   private def register = {
-    if (currentTransaction.get.isEmpty) throw new NoTransactionInScopeException
+    if (transaction.get.isEmpty) throw new NoTransactionInScopeException
-    currentTransaction.get.get.register(uuid, this)
+    transaction.get.get.register(uuid, this)
   }
 }
 
@@ -236,8 +232,8 @@ trait PersistentVector[T] extends RandomAccessSeq[T] with Transactional with Com
   def length: Int = storage.getVectorStorageSizeFor(uuid) + newElems.length
 
   private def register = {
-    if (currentTransaction.get.isEmpty) throw new NoTransactionInScopeException
+    if (transaction.get.isEmpty) throw new NoTransactionInScopeException
-    currentTransaction.get.get.register(uuid, this)
+    transaction.get.get.register(uuid, this)
   }
 }
 
@@ -272,8 +268,8 @@ trait PersistentRef[T] extends Transactional with Committable {
   }
 
   private def register = {
-    if (currentTransaction.get.isEmpty) throw new NoTransactionInScopeException
+    if (transaction.get.isEmpty) throw new NoTransactionInScopeException
-    currentTransaction.get.get.register(uuid, this)
+    transaction.get.get.register(uuid, this)
   }
 }
 
@@ -397,7 +393,7 @@ trait PersistentQueue[A] extends scala.collection.mutable.Queue[A]
     throw new UnsupportedOperationException("dequeueAll not supported")
 
   private def register = {
-    if (currentTransaction.get.isEmpty) throw new NoTransactionInScopeException
+    if (transaction.get.isEmpty) throw new NoTransactionInScopeException
-    currentTransaction.get.get.register(uuid, this)
+    transaction.get.get.register(uuid, this)
   }
 }

akka.iml

@@ -2,7 +2,18 @@
 <module org.jetbrains.idea.maven.project.MavenProjectsManager.isMavenModule="true" type="JAVA_MODULE" version="4">
   <component name="FacetManager">
     <facet type="Scala" name="Scala">
-      <configuration />
+      <configuration>
+        <option name="myScalaCompilerJarPaths">
+          <array>
+            <option value="$APPLICATION_HOME_DIR$/plugins/Scala/lib/scala-compiler.jar" />
+          </array>
+        </option>
+        <option name="myScalaSdkJarPaths">
+          <array>
+            <option value="$APPLICATION_HOME_DIR$/plugins/Scala/lib/scala-library.jar" />
+          </array>
+        </option>
+      </configuration>
     </facet>
   </component>
   <component name="NewModuleRootManager" LANGUAGE_LEVEL="JDK_1_5" inherit-compiler-output="false">

config/akka-reference.conf

@@ -30,8 +30,10 @@
  
   <stm>
     service = on
-    max-nr-of-retries = 100
+    fair = on                     # should transactions be fair or non-fair (non fair yield better performance)
-    distributed = off           # not implemented yet
+    max-nr-of-retries = 1000      # max nr of retries of a failing transaction before giving up
+    timeout = 10000               # transaction timeout; if transaction have not committed within the timeout then it is aborted
+    distributed = off             # not implemented yet
   </stm>
  
   <rest>

pom.xml

@@ -161,6 +161,11 @@
       <name>Codehaus Maven Repository</name>
       <url>http://repository.codehaus.org</url>
     </repository>
+    <repository>
+      <id>snapshots.repository.codehaus.org</id>
+      <name>Codehaus Maven Snapshot Repository</name>
+      <url>http://snapshots.repository.codehaus.org</url>
+    </repository>
     <repository>
       <id>maven2-repository.dev.java.net</id>
       <name>Java.net Repository for Maven</name>
@@ -301,11 +306,11 @@
         <configuration>
           <includes>
             <include>**/*Test.java</include>
-            <!--include>**/*Spec.java</include-->
+            <include>**/*Spec.java</include>
           </includes>
-          <excludes>
+          <!--excludes>
             <exclude>**/InMemNestedStateTest.java</exclude>
-          </excludes>
+          </excludes-->
           <systemProperties>
             <property>
               <name>akka.home</name>