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

Fixed 3 bugs in Active Objects and Actor supervision + changed to use Multiverse tryJoinCommit + improved logging + added more tracing + various misc fixes

Browse source
This commit is contained in:
Jonas Bonér 2010-07-13 12:37:11 +02:00
parent c8f1a5ede1
commit b98cfd5c1f
9 changed files with 247 additions and 207 deletions
Download patch file Download diff file Expand all files Collapse all files

72
akka-core/src/main/scala/actor/ActiveObject.scala
View file

@ -474,9 +474,9 @@ object ActiveObject extends Logging {
val parent = clazz.getSuperclass
if (parent != null) injectActiveObjectContext0(activeObject, parent)
else {
log.trace(
"Can't set 'ActiveObjectContext' for ActiveObject [%s] since no field of this type could be found.",
activeObject.getClass.getName)
log.ifTrace("Can't set 'ActiveObjectContext' for ActiveObject [" +
activeObject.getClass.getName +
"] since no field of this type could be found.")
None
}
}
@ -486,7 +486,6 @@ object ActiveObject extends Logging {
private[akka] def supervise(restartStrategy: RestartStrategy, components: List[Supervise]): Supervisor =
Supervisor(SupervisorConfig(restartStrategy, components))
}
private[akka] object AspectInitRegistry extends ListenerManagement {
@ -634,11 +633,12 @@ private[akka] sealed class ActiveObjectAspect {
joinPoint: JoinPoint, isOneWay: Boolean, isVoid: Boolean, sender: AnyRef, senderFuture: CompletableFuture[Any]) {
override def toString: String = synchronized {
"Invocation [joinPoint: " + joinPoint.toString +
", isOneWay: " + isOneWay +
", isVoid: " + isVoid +
", sender: " + sender +
", senderFuture: " + senderFuture +
"Invocation [" +
"\n\t\tmethod = " + joinPoint.getRtti.asInstanceOf[MethodRtti].getMethod.getName + " @ " + joinPoint.getTarget.getClass.getName +
"\n\t\tisOneWay = " + isOneWay +
"\n\t\tisVoid = " + isVoid +
"\n\t\tsender = " + sender +
"\n\t\tsenderFuture = " + senderFuture +
"]"
}
@ -758,14 +758,14 @@ private[akka] class Dispatcher(transactionalRequired: Boolean,
}
def receive = {
case Invocation(joinPoint, isOneWay, _, sender, senderFuture) =>
case invocation @ Invocation(joinPoint, isOneWay, _, sender, senderFuture) =>
ActiveObject.log.ifTrace("Invoking active object with message:\n" + invocation)
context.foreach { ctx =>
if (sender ne null) ctx._sender = sender
if (senderFuture ne null) ctx._senderFuture = senderFuture
}
ActiveObjectContext.sender.value = joinPoint.getThis // set next sender
self.senderFuture.foreach(ActiveObjectContext.senderFuture.value = _)
if (Actor.SERIALIZE_MESSAGES) serializeArguments(joinPoint)
if (isOneWay) joinPoint.proceed
else self.reply(joinPoint.proceed)
@ -773,61 +773,53 @@ private[akka] class Dispatcher(transactionalRequired: Boolean,
// Jan Kronquist: started work on issue 121
case Link(target) => self.link(target)
case Unlink(target) => self.unlink(target)
case unexpected =>
throw new IllegalActorStateException("Unexpected message [" + unexpected + "] sent to [" + this + "]")
case unexpected => throw new IllegalActorStateException(
"Unexpected message [" + unexpected + "] sent to [" + this + "]")
}
override def preRestart(reason: Throwable) {
try {
// Since preRestart is called we know that this dispatcher
// is about to be restarted. Put the instance in a thread
// local so the new dispatcher can be initialized with the contents of the
// old.
//FIXME - This should be considered as a workaround.
crashedActorTl.set(this)
if (preRestart.isDefined) preRestart.get.invoke(target.get, ZERO_ITEM_OBJECT_ARRAY: _*)
// Since preRestart is called we know that this dispatcher
// is about to be restarted. Put the instance in a thread
// local so the new dispatcher can be initialized with the
// contents of the old.
//FIXME - This should be considered as a workaround.
crashedActorTl.set(this)
preRestart.foreach(_.invoke(target.get, ZERO_ITEM_OBJECT_ARRAY: _*))
} catch { case e: InvocationTargetException => throw e.getCause }
}
override def postRestart(reason: Throwable) {
try {
if (postRestart.isDefined) {
postRestart.get.invoke(target.get, ZERO_ITEM_OBJECT_ARRAY: _*)
}
postRestart.foreach(_.invoke(target.get, ZERO_ITEM_OBJECT_ARRAY: _*))
} catch { case e: InvocationTargetException => throw e.getCause }
}
override def init = {
// Get the crashed dispatcher from thread local and intitialize this actor with the
// contents of the old dispatcher
val oldActor = crashedActorTl.get();
if(oldActor != null) {
initialize(oldActor.targetClass,oldActor.target.get,oldActor.context)
crashedActorTl.set(null)
}
// Get the crashed dispatcher from thread local and intitialize this actor with the
// contents of the old dispatcher
val oldActor = crashedActorTl.get();
if (oldActor != null) {
initialize(oldActor.targetClass, oldActor.target.get, oldActor.context)
crashedActorTl.set(null)
}
}
override def shutdown = {
try {
if (zhutdown.isDefined) {
zhutdown.get.invoke(target.get, ZERO_ITEM_OBJECT_ARRAY: _*)
}
} catch {
case e: InvocationTargetException => throw e.getCause
} finally {
zhutdown.foreach(_.invoke(target.get, ZERO_ITEM_OBJECT_ARRAY: _*))
} catch { case e: InvocationTargetException => throw e.getCause
} finally {
AspectInitRegistry.unregister(target.get);
}
}
override def initTransactionalState = {
try {
try {
if (initTxState.isDefined && target.isDefined) initTxState.get.invoke(target.get, ZERO_ITEM_OBJECT_ARRAY: _*)
} catch { case e: InvocationTargetException => throw e.getCause }
}
private def serializeArguments(joinPoint: JoinPoint) = {
val args = joinPoint.getRtti.asInstanceOf[MethodRtti].getParameterValues
var unserializable = false

276
akka-core/src/main/scala/actor/ActorRef.scala
View file

@ -10,26 +10,27 @@ import se.scalablesolutions.akka.config.{AllForOneStrategy, OneForOneStrategy, F
import se.scalablesolutions.akka.config.ScalaConfig._
import se.scalablesolutions.akka.stm.global._
import se.scalablesolutions.akka.stm.TransactionManagement._
import se.scalablesolutions.akka.stm.TransactionManagement
import se.scalablesolutions.akka.stm.{TransactionManagement, TransactionSetAbortedException}
import se.scalablesolutions.akka.remote.protocol.RemoteProtocol._
import se.scalablesolutions.akka.remote.{RemoteNode, RemoteServer, RemoteClient, MessageSerializer, RemoteRequestProtocolIdFactory}
import se.scalablesolutions.akka.serialization.Serializer
import se.scalablesolutions.akka.util.{HashCode, Logging, UUID, ReentrantGuard}
import RemoteActorSerialization._
import org.multiverse.api.ThreadLocalTransaction._
import org.multiverse.commitbarriers.CountDownCommitBarrier
import jsr166x.{Deque, ConcurrentLinkedDeque}
import org.multiverse.api.exceptions.DeadTransactionException
import java.net.InetSocketAddress
import java.util.concurrent.locks.ReentrantLock
import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
import java.util.{Map => JMap}
import java.lang.reflect.Field
import RemoteActorSerialization._
import jsr166x.{Deque, ConcurrentLinkedDeque}
import com.google.protobuf.ByteString
import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
/**
* ActorRef is an immutable and serializable handle to an Actor.
@ -582,8 +583,22 @@ sealed class LocalActorRef private[akka](
private[this] var actorFactory: Either[Option[Class[_ <: Actor]], Option[() => Actor]] = Left(None))
extends ActorRef {
private var isDeserialized = false
private var loader: Option[ClassLoader] = None
@volatile private[akka] var _remoteAddress: Option[InetSocketAddress] = None // only mutable to maintain identity across nodes
@volatile private[akka] var _linkedActors: Option[ConcurrentHashMap[String, ActorRef]] = None
@volatile private[akka] var _supervisor: Option[ActorRef] = None
@volatile private var isInInitialization = false
@volatile private var runActorInitialization = false
@volatile private var isDeserialized = false
@volatile private var loader: Option[ClassLoader] = None
protected[akka] val _mailbox: Deque[MessageInvocation] = new ConcurrentLinkedDeque[MessageInvocation]
protected[this] val actorInstance = guard.withGuard { new AtomicReference[Actor](newActor) }
// Needed to be able to null out the 'val self: ActorRef' member variables to make the Actor
// instance elegible for garbage collection
private val actorSelfFields = findActorSelfField(actor.getClass)
if (runActorInitialization && !isDeserialized) initializeActorInstance
private[akka] def this(clazz: Class[_ <: Actor]) = this(Left(Some(clazz)))
private[akka] def this(factory: () => Actor) = this(Right(Some(factory)))
@ -629,22 +644,7 @@ sealed class LocalActorRef private[akka](
ActorRegistry.register(this)
}
// Only mutable for RemoteServer in order to maintain identity across nodes
@volatile private[akka] var _remoteAddress: Option[InetSocketAddress] = None
@volatile private[akka] var _linkedActors: Option[ConcurrentHashMap[String, ActorRef]] = None
@volatile private[akka] var _supervisor: Option[ActorRef] = None
protected[akka] val _mailbox: Deque[MessageInvocation] = new ConcurrentLinkedDeque[MessageInvocation]
protected[this] val actorInstance = guard.withGuard { new AtomicReference[Actor](newActor) }
@volatile private var isInInitialization = false
@volatile private var runActorInitialization = false
// Needed to be able to null out the 'val self: ActorRef' member variables to make the Actor
// instance elegible for garbage collection
private val actorSelfFields = findActorSelfField(actor.getClass)
if (runActorInitialization && !isDeserialized) initializeActorInstance
// ========= PUBLIC FUNCTIONS =========
/**
* Returns the mailbox.
@ -903,40 +903,10 @@ sealed class LocalActorRef private[akka](
*/
def supervisor: Option[ActorRef] = guard.withGuard { _supervisor }
// ========= AKKA PROTECTED FUNCTIONS =========
protected[akka] def supervisor_=(sup: Option[ActorRef]): Unit = guard.withGuard { _supervisor = sup }
private def spawnButDoNotStart[T <: Actor: Manifest]: ActorRef = guard.withGuard {
val actorRef = Actor.actorOf(manifest[T].erasure.asInstanceOf[Class[T]].newInstance)
if (!dispatcher.isInstanceOf[ThreadBasedDispatcher]) actorRef.dispatcher = dispatcher
actorRef
}
private[this] def newActor: Actor = {
isInInitialization = true
Actor.actorRefInCreation.value = Some(this)
val actor = actorFactory match {
case Left(Some(clazz)) =>
try {
clazz.newInstance
} catch {
case e: InstantiationException => throw new ActorInitializationException(
"Could not instantiate Actor due to:\n" + e +
"\nMake sure Actor is NOT defined inside a class/trait," +
"\nif so put it outside the class/trait, f.e. in a companion object," +
"\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.")
}
case Right(Some(factory)) =>
factory()
case _ =>
throw new ActorInitializationException(
"Can't create Actor, no Actor class or factory function in scope")
}
if (actor eq null) throw new ActorInitializationException(
"Actor instance passed to ActorRef can not be 'null'")
isInInitialization = false
actor
}
protected[akka] def postMessageToMailbox(message: Any, senderOption: Option[ActorRef]): Unit = {
joinTransaction(message)
@ -975,15 +945,6 @@ sealed class LocalActorRef private[akka](
}
}
private def joinTransaction(message: Any) = if (isTransactionSetInScope) {
import org.multiverse.api.ThreadLocalTransaction
val txSet = getTransactionSetInScope
Actor.log.trace("Joining transaction set [%s];\n\tactor %s\n\twith message [%s]", txSet, toString, message) // FIXME test to run bench without this trace call
val mtx = ThreadLocalTransaction.getThreadLocalTransaction
if ((mtx eq null) || mtx.getStatus.isDead) txSet.incParties
else txSet.incParties(mtx, 1)
}
/**
* Callback for the dispatcher. This is the ingle entry point to the user Actor implementation.
*/
@ -1003,58 +964,6 @@ sealed class LocalActorRef private[akka](
}
}
private def dispatch[T](messageHandle: MessageInvocation) = {
val message = messageHandle.message //serializeMessage(messageHandle.message)
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 (isTransactor) {
Actor.log.trace(
"Creating a new transaction set (top-level transaction)\n\tfor actor %s\n\twith message %s",
toString, messageHandle)
Some(createNewTransactionSet)
} else None
}
setTransactionSet(txSet)
try {
cancelReceiveTimeout // FIXME: leave this here?
if (isTransactor) {
val txFactory = _transactionFactory.getOrElse(DefaultGlobalTransactionFactory)
atomic(txFactory) {
actor.base(message)
setTransactionSet(txSet) // restore transaction set to allow atomic block to do commit
}
} else {
actor.base(message)
setTransactionSet(txSet) // restore transaction set to allow atomic block to do commit
}
} catch {
case e =>
_isBeingRestarted = true
// abort transaction set
if (isTransactionSetInScope) {
val txSet = getTransactionSetInScope
Actor.log.debug("Aborting transaction set [%s]", txSet)
txSet.abort
}
Actor.log.error(e, "Exception when invoking \n\tactor [%s] \n\twith message [%s]", this, message)
senderFuture.foreach(_.completeWithException(this, e))
clearTransaction
if (topLevelTransaction) clearTransactionSet
// 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
if (topLevelTransaction) clearTransactionSet
}
}
protected[akka] def handleTrapExit(dead: ActorRef, reason: Throwable): Unit = {
if (trapExit.exists(_.isAssignableFrom(reason.getClass))) {
if (faultHandler.isDefined) {
@ -1075,6 +984,7 @@ sealed class LocalActorRef private[akka](
}
protected[akka] def restart(reason: Throwable): Unit = {
_isBeingRestarted = true
val failedActor = actorInstance.get
failedActor.synchronized {
lifeCycle.get match {
@ -1117,20 +1027,6 @@ sealed class LocalActorRef private[akka](
}
}
private def shutDownTemporaryActor(temporaryActor: ActorRef) = {
Actor.log.info("Actor [%s] configured as TEMPORARY and will not be restarted.", temporaryActor.id)
temporaryActor.stop
linkedActors.remove(temporaryActor.uuid) // remove the temporary actor
// if last temporary actor is gone, then unlink me from supervisor
if (linkedActors.isEmpty) {
Actor.log.info(
"All linked actors have died permanently (they were all configured as TEMPORARY)" +
"\n\tshutting down and unlinking supervisor actor as well [%s].",
temporaryActor.id)
_supervisor.foreach(_ ! UnlinkAndStop(this))
}
}
protected[akka] def registerSupervisorAsRemoteActor: Option[String] = guard.withGuard {
if (_supervisor.isDefined) {
RemoteClient.clientFor(remoteAddress.get).registerSupervisorForActor(this)
@ -1149,6 +1045,126 @@ sealed class LocalActorRef private[akka](
protected[akka] def linkedActorsAsList: List[ActorRef] =
linkedActors.values.toArray.toList.asInstanceOf[List[ActorRef]]
// ========= PRIVATE FUNCTIONS =========
private def spawnButDoNotStart[T <: Actor: Manifest]: ActorRef = guard.withGuard {
val actorRef = Actor.actorOf(manifest[T].erasure.asInstanceOf[Class[T]].newInstance)
if (!dispatcher.isInstanceOf[ThreadBasedDispatcher]) actorRef.dispatcher = dispatcher
actorRef
}
private[this] def newActor: Actor = {
isInInitialization = true
Actor.actorRefInCreation.value = Some(this)
val actor = actorFactory match {
case Left(Some(clazz)) =>
try {
clazz.newInstance
} catch {
case e: InstantiationException => throw new ActorInitializationException(
"Could not instantiate Actor due to:\n" + e +
"\nMake sure Actor is NOT defined inside a class/trait," +
"\nif so put it outside the class/trait, f.e. in a companion object," +
"\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.")
}
case Right(Some(factory)) =>
factory()
case _ =>
throw new ActorInitializationException(
"Can't create Actor, no Actor class or factory function in scope")
}
if (actor eq null) throw new ActorInitializationException(
"Actor instance passed to ActorRef can not be 'null'")
isInInitialization = false
actor
}
private def joinTransaction(message: Any) = if (isTransactionSetInScope) {
import org.multiverse.api.ThreadLocalTransaction
val oldTxSet = getTransactionSetInScope
val currentTxSet = if (oldTxSet.isAborted || oldTxSet.isCommitted) {
clearTransactionSet
createNewTransactionSet
} else oldTxSet
Actor.log.ifTrace("Joining transaction set [" + currentTxSet + "];\n\tactor " + toString + "\n\twith message [" + message + "]")
val mtx = ThreadLocalTransaction.getThreadLocalTransaction
if ((mtx eq null) || mtx.getStatus.isDead) currentTxSet.incParties
else currentTxSet.incParties(mtx, 1)
}
private def dispatch[T](messageHandle: MessageInvocation) = {
Actor.log.ifTrace("Invoking actor with message:\n" + messageHandle)
val message = messageHandle.message //serializeMessage(messageHandle.message)
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 (isTransactor) {
Actor.log.ifTrace("Creating a new transaction set (top-level transaction)\n\tfor actor " + toString + "\n\twith message " + messageHandle)
Some(createNewTransactionSet)
} else None
}
setTransactionSet(txSet)
try {
cancelReceiveTimeout // FIXME: leave this here?
if (isTransactor) {
val txFactory = _transactionFactory.getOrElse(DefaultGlobalTransactionFactory)
atomic(txFactory) {
actor.base(message)
setTransactionSet(txSet) // restore transaction set to allow atomic block to do commit
}
} else {
actor.base(message)
setTransactionSet(txSet) // restore transaction set to allow atomic block to do commit
}
} catch {
case e: DeadTransactionException =>
handleExceptionInDispatch(
new TransactionSetAbortedException("Transaction set has been aborted by another participant"),
message, topLevelTransaction)
case e =>
handleExceptionInDispatch(e, message, topLevelTransaction)
} finally {
clearTransaction
if (topLevelTransaction) clearTransactionSet
}
}
private def shutDownTemporaryActor(temporaryActor: ActorRef) = {
Actor.log.info("Actor [%s] configured as TEMPORARY and will not be restarted.", temporaryActor.id)
temporaryActor.stop
linkedActors.remove(temporaryActor.uuid) // remove the temporary actor
// if last temporary actor is gone, then unlink me from supervisor
if (linkedActors.isEmpty) {
Actor.log.info(
"All linked actors have died permanently (they were all configured as TEMPORARY)" +
"\n\tshutting down and unlinking supervisor actor as well [%s].",
temporaryActor.id)
_supervisor.foreach(_ ! UnlinkAndStop(this))
}
}
private def handleExceptionInDispatch(e: Throwable, message: Any, topLevelTransaction: Boolean) = {
_isBeingRestarted = true
// abort transaction set
if (isTransactionSetInScope) {
val txSet = getTransactionSetInScope
Actor.log.debug("Aborting transaction set [%s]", txSet)
txSet.abort
}
Actor.log.error(e, "Exception when invoking \n\tactor [%s] \n\twith message [%s]", this, message)
senderFuture.foreach(_.completeWithException(this, e))
clearTransaction
if (topLevelTransaction) clearTransactionSet
// 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)
}
private def nullOutActorRefReferencesFor(actor: Actor) = {
actorSelfFields._1.set(actor, null)
actorSelfFields._2.set(actor, null)

2
akka-core/src/main/scala/dispatch/MessageHandling.scala
View file

@ -53,7 +53,7 @@ final class MessageInvocation(val receiver: ActorRef,
"\n\tsender = " + sender +
"\n\tsenderFuture = " + senderFuture +
"\n\ttransactionSet = " + transactionSet +
"\n]"
"]"
}
}

39
akka-core/src/main/scala/stm/JTA.scala
View file

@ -4,7 +4,9 @@
package se.scalablesolutions.akka.stm
import javax.transaction.{TransactionManager, UserTransaction, Transaction => JtaTransaction, SystemException, Status, Synchronization, TransactionSynchronizationRegistry}
import javax.transaction.{TransactionManager, UserTransaction,
Transaction => JtaTransaction, SystemException,
Status, Synchronization, TransactionSynchronizationRegistry}
import javax.naming.{InitialContext, Context, NamingException}
import se.scalablesolutions.akka.config.Config._
@ -16,7 +18,7 @@ import se.scalablesolutions.akka.util.Logging
* @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
*/
object TransactionContainer extends Logging {
val AKKA_JTA_TRANSACTION_SERVICE_CLASS = "se.scalablesolutions.akka.jta.AtomikosTransactionService"
val AKKA_JTA_TRANSACTION_SERVICE_CLASS = "se.scalablesolutions.akka.jta.AtomikosTransactionService"
val DEFAULT_USER_TRANSACTION_NAME = "java:comp/UserTransaction"
val FALLBACK_TRANSACTION_MANAGER_NAMES = "java:comp/TransactionManager" ::
"java:appserver/TransactionManager" ::
@ -119,22 +121,31 @@ class TransactionContainer private (val tm: Either[Option[UserTransaction], Opti
}
}
def begin = tm match {
case Left(Some(userTx)) => userTx.begin
case Right(Some(txMan)) => txMan.begin
case _ => throw new StmConfigurationException("Does not have a UserTransaction or TransactionManager in scope")
def begin = {
TransactionContainer.log.ifTrace("Starting JTA transaction")
tm match {
case Left(Some(userTx)) => userTx.begin
case Right(Some(txMan)) => txMan.begin
case _ => throw new StmConfigurationException("Does not have a UserTransaction or TransactionManager in scope")
}
}
def commit = tm match {
case Left(Some(userTx)) => userTx.commit
case Right(Some(txMan)) => txMan.commit
case _ => throw new StmConfigurationException("Does not have a UserTransaction or TransactionManager in scope")
def commit = {
TransactionContainer.log.ifTrace("Committing JTA transaction")
tm match {
case Left(Some(userTx)) => userTx.commit
case Right(Some(txMan)) => txMan.commit
case _ => throw new StmConfigurationException("Does not have a UserTransaction or TransactionManager in scope")
}
}
def rollback = tm match {
case Left(Some(userTx)) => userTx.rollback
case Right(Some(txMan)) => txMan.rollback
case _ => throw new StmConfigurationException("Does not have a UserTransaction or TransactionManager in scope")
def rollback = {
TransactionContainer.log.ifTrace("Aborting JTA transaction")
tm match {
case Left(Some(userTx)) => userTx.rollback
case Right(Some(txMan)) => txMan.rollback
case _ => throw new StmConfigurationException("Does not have a UserTransaction or TransactionManager in scope")
}
}
def getStatus = tm match {

7
akka-core/src/main/scala/stm/Transaction.scala
View file

@ -83,11 +83,12 @@ object Transaction {
if (JTA_AWARE) Some(TransactionContainer())
else None
log.trace("Creating %s", toString)
log.ifTrace("Creating transaction " + toString)
// --- public methods ---------
def begin = synchronized {
log.ifTrace("Starting transaction " + toString)
jta.foreach { txContainer =>
txContainer.begin
txContainer.registerSynchronization(new StmSynchronization(txContainer, this))
@ -95,14 +96,14 @@ object Transaction {
}
def commit = synchronized {
log.trace("Committing transaction %s", toString)
log.ifTrace("Committing transaction " + toString)
persistentStateMap.valuesIterator.foreach(_.commit)
status = TransactionStatus.Completed
jta.foreach(_.commit)
}
def abort = synchronized {
log.trace("Aborting transaction %s", toString)
log.ifTrace("Aborting transaction " + toString)
jta.foreach(_.rollback)
persistentStateMap.valuesIterator.foreach(_.abort)
persistentStateMap.clear

6
akka-core/src/main/scala/stm/TransactionFactory.scala
View file

@ -37,8 +37,8 @@ object TransactionConfig {
def traceLevel(level: String) = level.toLowerCase match {
case "coarse" | "course" => Transaction.TraceLevel.Coarse
case "fine" => Transaction.TraceLevel.Fine
case _ => Transaction.TraceLevel.None
case "fine" => Transaction.TraceLevel.Fine
case _ => Transaction.TraceLevel.None
}
/**
@ -126,7 +126,7 @@ object TransactionFactory {
traceLevel: TraceLevel = TransactionConfig.TRACE_LEVEL,
hooks: Boolean = TransactionConfig.HOOKS) = {
val config = new TransactionConfig(familyName, readonly, maxRetries, timeout, trackReads, writeSkew,
explicitRetries, interruptible, speculative, quickRelease, traceLevel, hooks)
explicitRetries, interruptible, speculative, quickRelease, traceLevel, hooks)
new TransactionFactory(config)
}
}

32
akka-core/src/main/scala/stm/TransactionManagement.scala
View file

@ -7,6 +7,7 @@ package se.scalablesolutions.akka.stm
import se.scalablesolutions.akka.util.Logging
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.TimeUnit
import org.multiverse.api.{StmUtils => MultiverseStmUtils}
import org.multiverse.api.ThreadLocalTransaction._
@ -14,16 +15,20 @@ import org.multiverse.api.{Transaction => MultiverseTransaction}
import org.multiverse.commitbarriers.CountDownCommitBarrier
import org.multiverse.templates.{TransactionalCallable, OrElseTemplate}
class StmException(msg: String) extends RuntimeException(msg)
class TransactionSetAbortedException(msg: String) extends RuntimeException(msg)
// TODO Should we remove TransactionAwareWrapperException? Not used anywhere yet.
class TransactionAwareWrapperException(val cause: Throwable, val tx: Option[Transaction]) extends RuntimeException(cause) {
override def toString = "TransactionAwareWrapperException[" + cause + ", " + tx + "]"
}
/**
* Internal helper methods and properties for transaction management.
*/
object TransactionManagement extends TransactionManagement {
import se.scalablesolutions.akka.config.Config._
// move to stm.global.fair?
// FIXME move to stm.global.fair?
val FAIR_TRANSACTIONS = config.getBool("akka.stm.fair", true)
private[akka] val transactionSet = new ThreadLocal[Option[CountDownCommitBarrier]]() {
@ -47,6 +52,9 @@ object TransactionManagement extends TransactionManagement {
}
}
/**
* Internal helper methods for transaction management.
*/
trait TransactionManagement {
private[akka] def createNewTransactionSet: CountDownCommitBarrier = {
@ -111,7 +119,9 @@ class LocalStm extends TransactionManagement with Logging {
factory.boilerplate.execute(new TransactionalCallable[T]() {
def call(mtx: MultiverseTransaction): T = {
factory.addHooks
body
val result = body
log.ifTrace("Committing local transaction [" + mtx + "]")
result
}
})
}
@ -145,9 +155,10 @@ class GlobalStm extends TransactionManagement with Logging {
factory.addHooks
val result = body
val txSet = getTransactionSetInScope
log.trace("Committing transaction [%s]\n\tby joining transaction set [%s]", mtx, txSet)
// FIXME ? txSet.tryJoinCommit(mtx, TransactionManagement.TRANSACTION_TIMEOUT, TimeUnit.MILLISECONDS)
try { txSet.joinCommit(mtx) } catch { case e: IllegalStateException => {} }
log.ifTrace("Committing global transaction [" + mtx + "]\n\tand joining transaction set [" + txSet + "]")
mtx.commit
// Need to catch IllegalStateException until we have fix in Multiverse, since it throws it by mistake
try { txSet.tryJoinCommit(mtx, TransactionConfig.TIMEOUT, TimeUnit.MILLISECONDS) } catch { case e: IllegalStateException => {} }
clearTransaction
result
}
@ -156,6 +167,7 @@ class GlobalStm extends TransactionManagement with Logging {
}
trait StmUtil {
/**
* Schedule a deferred task on the thread local transaction (use within an atomic).
* This is executed when the transaction commits.
@ -178,6 +190,14 @@ trait StmUtil {
/**
* Use either-orElse to combine two blocking transactions.
* Usage:
* <pre>
* either {
* ...
* } orElse {
* ...
* }
* </pre>
*/
def either[T](firstBody: => T) = new {
def orElse(secondBody: => T) = new OrElseTemplate[T] {

3
akka-core/src/test/java/se/scalablesolutions/akka/actor/TransactionalActiveObject.java
View file

@ -21,6 +21,7 @@ public class TransactionalActiveObject {
refState = new Ref();
isInitialized = true;
}
System.out.println("==========> init");
}
public String getMapState(String key) {
@ -37,6 +38,7 @@ public class TransactionalActiveObject {
public void setMapState(String key, String msg) {
mapState.put(key, msg);
System.out.println("==========> setMapState");
}
public void setVectorState(String msg) {
@ -72,6 +74,7 @@ public class TransactionalActiveObject {
mapState.put(key, msg);
vectorState.add(msg);
refState.swap(msg);
System.out.println("==========> failure");
nested.failure(key, msg, failer);
return msg;
}

17
akka-core/src/test/scala/TransactionalActiveObjectSpec.scala
View file

@ -16,18 +16,12 @@ import se.scalablesolutions.akka.config._
import se.scalablesolutions.akka.config.ActiveObjectConfigurator
import se.scalablesolutions.akka.config.JavaConfig._
import se.scalablesolutions.akka.actor._
/*
@RunWith(classOf[JUnitRunner])
class TransactionalActiveObjectSpec extends
<<<<<<< HEAD:akka-core/src/test/scala/TransactionalActiveObjectSpec.scala
Spec with
ShouldMatchers with
BeforeAndAfterAll {
=======
Spec with
ShouldMatchers with
BeforeAndAfterAll {
>>>>>>> 38e8bea3fe6a7e9fcc9c5f353124144739bdc234:akka-core/src/test/scala/TransactionalActiveObjectSpec.scala
private val conf = new ActiveObjectConfigurator
private var messageLog = ""
@ -52,7 +46,7 @@ class TransactionalActiveObjectSpec extends
}
describe("Transactional in-memory Active Object ") {
/*
it("map should not rollback state for stateful server in case of success") {
val stateful = conf.getInstance(classOf[TransactionalActiveObject])
stateful.init
@ -60,11 +54,13 @@ class TransactionalActiveObjectSpec extends
stateful.success("testShouldNotRollbackStateForStatefulServerInCaseOfSuccess", "new state")
stateful.getMapState("testShouldNotRollbackStateForStatefulServerInCaseOfSuccess") should equal("new state")
}
*/
it("map should rollback state for stateful server in case of failure") {
val stateful = conf.getInstance(classOf[TransactionalActiveObject])
stateful.init
Thread.sleep(500)
stateful.setMapState("testShouldRollbackStateForStatefulServerInCaseOfFailure", "init")
Thread.sleep(500)
val failer = conf.getInstance(classOf[ActiveObjectFailer])
try {
stateful.failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state", failer)
@ -73,6 +69,7 @@ class TransactionalActiveObjectSpec extends
stateful.getMapState("testShouldRollbackStateForStatefulServerInCaseOfFailure") should equal("init")
}
/*
it("vector should rollback state for stateful server in case of failure") {
val stateful = conf.getInstance(classOf[TransactionalActiveObject])
stateful.init
@ -112,6 +109,6 @@ class TransactionalActiveObjectSpec extends
stateful.success("testShouldNotRollbackStateForStatefulServerInCaseOfSuccess", "new state")
stateful.getRefState should equal("new state")
}
*/
}
}
*/