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

merged sbt branch with master

Browse source
This commit is contained in:
Jonas Bonér 2010-03-07 08:14:48 +01:00
commit b7ed47ee73
31 changed files with 525 additions and 740 deletions
Download patch file Download diff file Expand all files Collapse all files

3
akka-cluster/akka-cluster-jgroups/src/main/scala/JGroupsClusterActor.scala
View file

@ -44,7 +44,8 @@ class JGroupsClusterActor extends BasicClusterActor {
log debug "UNSUPPORTED: JGroupsClusterActor::unblock" //TODO HotSwap back and flush the buffer
})
})
channel.map(_.connect(name))
channel.foreach(_.connect(name))
}
protected def toOneNode(dest : Address, msg: Array[Byte]): Unit =

9
akka-cluster/akka-cluster-shoal/src/main/scala/ShoalClusterActor.scala
View file

@ -85,6 +85,7 @@ class ShoalClusterActor extends BasicClusterActor {
*/
protected def createCallback : CallBack = {
import org.scala_tools.javautils.Imports._
import ClusterActor._
val me = this
new CallBack {
def processNotification(signal : Signal) {
@ -92,10 +93,10 @@ class ShoalClusterActor extends BasicClusterActor {
signal.acquire()
if(isActive) {
signal match {
case ms : MessageSignal => me send Message(ms.getMemberToken,ms.getMessage)
case jns : JoinNotificationSignal => me send View(Set[ADDR_T]() ++ jns.getCurrentCoreMembers.asScala - serverName)
case fss : FailureSuspectedSignal => me send Zombie(fss.getMemberToken)
case fns : FailureNotificationSignal => me send Zombie(fns.getMemberToken)
case ms : MessageSignal => me send Message[ADDR_T](ms.getMemberToken,ms.getMessage)
case jns : JoinNotificationSignal => me send View[ADDR_T](Set[ADDR_T]() ++ jns.getCurrentCoreMembers.asScala - serverName)
case fss : FailureSuspectedSignal => me send Zombie[ADDR_T](fss.getMemberToken)
case fns : FailureNotificationSignal => me send Zombie[ADDR_T](fns.getMemberToken)
case _ => log.debug("Unhandled signal: [%s]",signal)
}
}

96
akka-core/src/main/scala/actor/Actor.scala
View file

@ -10,13 +10,14 @@ 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
import se.scalablesolutions.akka.util.{HashCode, Logging, UUID}
import org.multiverse.api.ThreadLocalTransaction._
import org.multiverse.commitbarriers.CountDownCommitBarrier
import java.util.{Queue, HashSet}
import java.util.concurrent.ConcurrentLinkedQueue
@ -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
}
@ -98,9 +99,7 @@ object Actor extends Logging {
* The actor is started when created.
* Example:
* <pre>
* import Actor._
*
* val a = actor {
* val a = Actor.init {
* ... // init stuff
* } receive {
* case msg => ... // handle message
@ -108,8 +107,8 @@ object Actor extends Logging {
* </pre>
*
*/
def actor(body: => Unit) = {
def handler(body: => Unit) = new {
def init[A](body: => Unit) = {
def handler[A](body: => Unit) = new {
def receive(handler: PartialFunction[Any, Unit]) = new Actor() {
start
body
@ -198,7 +197,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
@ -219,6 +218,7 @@ trait Actor extends TransactionManagement {
private[akka] var _replyToAddress: Option[InetSocketAddress] = None
private[akka] val _mailbox: Queue[MessageInvocation] = new ConcurrentLinkedQueue[MessageInvocation]
// ====================================
// protected fields
// ====================================
@ -309,9 +309,9 @@ trait Actor extends TransactionManagement {
* If 'trapExit' is set for the actor to act as supervisor, then a faultHandler must be defined.
* Can be one of:
* <pre/>
* AllForOneStrategy(maxNrOfRetries: Int, withinTimeRange: Int)
* faultHandler = Some(AllForOneStrategy(maxNrOfRetries, withinTimeRange))
*
* OneForOneStrategy(maxNrOfRetries: Int, withinTimeRange: Int)
* faultHandler = Some(OneForOneStrategy(maxNrOfRetries, withinTimeRange))
* </pre>
*/
protected var faultHandler: Option[FaultHandlingStrategy] = None
@ -334,8 +334,8 @@ trait Actor extends TransactionManagement {
/**
* User overridable callback/setting.
*
* Partial function implementing the server logic.
* To be implemented by subclassing server.
* Partial function implementing the actor logic.
* To be implemented by subclassing actor.
* <p/>
* Example code:
* <pre>
@ -785,6 +785,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)
@ -796,8 +801,7 @@ trait Actor extends TransactionManagement {
.setIsEscaped(false)
val id = registerSupervisorAsRemoteActor
if(id.isDefined)
requestBuilder.setSupervisorUuid(id.get)
if (id.isDefined) requestBuilder.setSupervisorUuid(id.get)
// set the source fields used to reply back to the original sender
// (i.e. not the remote proxy actor)
@ -816,7 +820,7 @@ 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
@ -824,12 +828,18 @@ trait Actor extends TransactionManagement {
else
invocation.send
}
clearTransactionSet
}
protected[akka] def postMessageToMailboxAndCreateFutureResultWithTimeout(
message: Any,
timeout: Long,
senderFuture: Option[CompletableFuture]): CompletableFuture = {
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)
@ -843,16 +853,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 DefaultCompletableFuture(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
}
}
@ -861,6 +873,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)
@ -872,7 +885,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
@ -894,43 +907,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 {
incrementTransaction
if (base.isDefinedAt(message)) base(message) // invoke user actor's receive partial function
else throw new IllegalArgumentException(
"Actor " + toString + " could not process message [" + message + "]" +
"\n\tsince no matching 'case' clause in its 'receive' method could be found")
} finally {
decrementTransaction
}
if (base.isDefinedAt(message)) base(message) // invoke user actor's receive partial function
else throw new IllegalArgumentException(
toString + " could not process message [" + message + "]" +
"\n\tsince no matching 'case' clause in its 'receive' method could be found")
setTransactionSet(txSet) // restore transaction set to allow atomic block to do commit
}
try {
if (isTransactionRequiresNew && !isTransactionInScope) {
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")
if (isTransactionRequiresNew) {
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
}
}
@ -1042,6 +1067,5 @@ trait Actor extends TransactionManagement {
that.asInstanceOf[Actor]._uuid == _uuid
}
override def toString(): String = "Actor[" + id + ":" + uuid + "]"
override def toString = "Actor[" + id + ":" + uuid + "]"
}

4
akka-core/src/main/scala/actor/BootableActorLoaderService.scala
View file

@ -31,10 +31,10 @@ trait BootableActorLoaderService extends Bootable with Logging {
val toDeploy = for (f <- DEPLOY_DIR.listFiles().toArray.toList.asInstanceOf[List[File]]) yield f.toURL
log.info("Deploying applications from [%s]: [%s]", DEPLOY, toDeploy.toArray.toList)
new URLClassLoader(toDeploy.toArray, ClassLoader.getSystemClassLoader)
} else if (getClass.getClassLoader.getResourceAsStream("akka.conf") ne null) {
} else if (getClass.getClassLoader.getResourceAsStream("aop.xml") ne null) {
getClass.getClassLoader
} else throw new IllegalStateException(
"AKKA_HOME is not defined and no 'akka.conf' can be found on the classpath, aborting")
"AKKA_HOME is not defined and akka-<version>.jar can not be found on the classpath; aborting...")
)
}

1
akka-core/src/main/scala/dispatch/Future.scala
View file

@ -13,6 +13,7 @@ class FutureTimeoutException(message: String) extends RuntimeException(message)
object Futures {
/**
* FIXME document
* <pre>
* val future = Futures.future(1000) {
* ... // do stuff

9
akka-core/src/main/scala/dispatch/Reactor.scala
View file

@ -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[CompletableFuture],
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]"
}
}

1
akka-core/src/main/scala/remote/BootableRemoteActorService.scala
View file

@ -24,7 +24,6 @@ trait BootableRemoteActorService extends Bootable with Logging {
abstract override def onLoad = {
if(config.getBool("akka.remote.server.service", true)){
log.info("Starting up Cluster Service")
Cluster.start
super.onLoad //Initialize BootableActorLoaderService before remote service
log.info("Initializing Remote Actors Service...")

142
akka-core/src/main/scala/remote/Cluster.scala
View file

@ -26,8 +26,8 @@ trait Cluster {
def relayMessage(to: Class[_ <: Actor], msg: AnyRef): Unit
def lookup[T](pf: PartialFunction[RemoteAddress, T]): Option[T]
def foreach(f : (RemoteAddress) => Unit) : Unit
def foreach(f: (RemoteAddress) => Unit): Unit
}
/**
@ -48,7 +48,15 @@ private[remote] object ClusterActor {
sealed trait ClusterMessage
private[remote] case class RelayedMessage(actorClassFQN: String, msg: AnyRef) extends ClusterMessage
private[remote] case class Message[ADDR_T](sender: ADDR_T, msg: Array[Byte])
private[remote] case object PapersPlease extends ClusterMessage
private[remote] case class Papers(addresses: List[RemoteAddress]) extends ClusterMessage
private[remote] case object Block extends ClusterMessage
private[remote] case object Unblock extends ClusterMessage
private[remote] case class View[ADDR_T](othersPresent: Set[ADDR_T]) extends ClusterMessage
private[remote] case class Zombie[ADDR_T](address: ADDR_T) extends ClusterMessage
private[remote] case class RegisterLocalNode(server: RemoteAddress) extends ClusterMessage
private[remote] case class DeregisterLocalNode(server: RemoteAddress) extends ClusterMessage
private[remote] case class Node(endpoints: List[RemoteAddress])
}
@ -59,76 +67,65 @@ private[remote] object ClusterActor {
*/
abstract class BasicClusterActor extends ClusterActor {
import ClusterActor._
case class Message(sender : ADDR_T,msg : Array[Byte])
case object PapersPlease extends ClusterMessage
case class Papers(addresses: List[RemoteAddress]) extends ClusterMessage
case object Block extends ClusterMessage
case object Unblock extends ClusterMessage
case class View(othersPresent : Set[ADDR_T]) extends ClusterMessage
case class Zombie(address: ADDR_T) extends ClusterMessage
case class RegisterLocalNode(server: RemoteAddress) extends ClusterMessage
case class DeregisterLocalNode(server: RemoteAddress) extends ClusterMessage
type ADDR_T
@volatile private var local: Node = Node(Nil)
@volatile private var remotes: Map[ADDR_T, Node] = Map()
override def init = {
remotes = new HashMap[ADDR_T, Node]
remotes = new HashMap[ADDR_T, Node]
}
override def shutdown = {
remotes = Map()
remotes = Map()
}
def receive = {
case v @ View(members) => {
case v: View[ADDR_T] => {
// Not present in the cluster anymore = presumably zombies
// Nodes we have no prior knowledge existed = unknowns
val zombies = Set[ADDR_T]() ++ remotes.keySet -- members
val unknown = members -- remotes.keySet
val zombies = Set[ADDR_T]() ++ remotes.keySet -- v.othersPresent
val unknown = v.othersPresent -- remotes.keySet
log debug ("Updating view")
log debug ("Other memebers: [%s]",members)
log debug ("Zombies: [%s]",zombies)
log debug ("Unknowns: [%s]",unknown)
log debug ("Other memebers: [%s]", v.othersPresent)
log debug ("Zombies: [%s]", zombies)
log debug ("Unknowns: [%s]", unknown)
// Tell the zombies and unknowns to provide papers and prematurely treat the zombies as dead
broadcast(zombies ++ unknown, PapersPlease)
remotes = remotes -- zombies
}
case Zombie(x) => { //Ask the presumed zombie for papers and prematurely treat it as dead
log debug ("Killing Zombie Node: %s", x)
broadcast(x :: Nil, PapersPlease)
remotes = remotes - x
case z: Zombie[ADDR_T] => { //Ask the presumed zombie for papers and prematurely treat it as dead
log debug ("Killing Zombie Node: %s", z.address)
broadcast(z.address :: Nil, PapersPlease)
remotes = remotes - z.address
}
case rm @ RelayedMessage(_, _) => {
case rm@RelayedMessage(_, _) => {
log debug ("Relaying message: %s", rm)
broadcast(rm)
}
case m @ Message(src,msg) => {
(Cluster.serializer in (msg, None)) match {
case m: Message[ADDR_T] => {
val (src, msg) = (m.sender, m.msg)
(Cluster.serializer in (msg, None)) match {
case PapersPlease => {
log debug ("Asked for papers by %s", src)
broadcast(src :: Nil, Papers(local.endpoints))
case PapersPlease => {
log debug ("Asked for papers by %s", src)
broadcast(src :: Nil, Papers(local.endpoints))
if (remotes.get(src).isEmpty) // If we were asked for papers from someone we don't know, ask them!
broadcast(src :: Nil, PapersPlease)
}
case Papers(x) => remotes = remotes + (src -> Node(x))
case RelayedMessage(c, m) => ActorRegistry.actorsFor(c).foreach(_ send m)
case unknown => log debug ("Unknown message: %s", unknown.toString)
if (remotes.get(src).isEmpty) // If we were asked for papers from someone we don't know, ask them!
broadcast(src :: Nil, PapersPlease)
}
case Papers(x) => remotes = remotes + (src -> Node(x))
case RelayedMessage(c, m) => ActorRegistry.actorsFor(c).foreach(_ send m)
case unknown => log debug ("Unknown message: %s", unknown.toString)
}
}
case RegisterLocalNode(s) => {
@ -147,12 +144,12 @@ abstract class BasicClusterActor extends ClusterActor {
/**
* Implement this in a subclass to add node-to-node messaging
*/
protected def toOneNode(dest : ADDR_T, msg : Array[Byte]) : Unit
protected def toOneNode(dest: ADDR_T, msg: Array[Byte]): Unit
/**
* Implement this in a subclass to add node-to-many-nodes messaging
*/
protected def toAllNodes(msg : Array[Byte]) : Unit
protected def toAllNodes(msg: Array[Byte]): Unit
/**
* Sends the specified message to the given recipients using the serializer
@ -160,7 +157,7 @@ abstract class BasicClusterActor extends ClusterActor {
*/
protected def broadcast[T <: AnyRef](recipients: Iterable[ADDR_T], msg: T): Unit = {
lazy val m = Cluster.serializer out msg
for (r <- recipients) toOneNode(r,m)
for (r <- recipients) toOneNode(r, m)
}
/**
@ -175,11 +172,11 @@ abstract class BasicClusterActor extends ClusterActor {
*/
def lookup[T](handleRemoteAddress: PartialFunction[RemoteAddress, T]): Option[T] =
remotes.values.toList.flatMap(_.endpoints).find(handleRemoteAddress isDefinedAt _).map(handleRemoteAddress)
/**
* Applies the given function to all remote addresses known
*/
def foreach(f : (RemoteAddress) => Unit) : Unit = remotes.values.toList.flatMap(_.endpoints).foreach(f)
def foreach(f: (RemoteAddress) => Unit): Unit = remotes.values.toList.flatMap(_.endpoints).foreach(f)
/**
* Registers a local endpoint
@ -206,36 +203,38 @@ abstract class BasicClusterActor extends ClusterActor {
* Loads a specified ClusterActor and delegates to that instance.
*/
object Cluster extends Cluster with Logging {
lazy val DEFAULT_SERIALIZER_CLASS_NAME = Serializer.Java.getClass.getName
@volatile private[remote] var clusterActor: Option[ClusterActor] = None
// FIXME Use the supervisor member field
@volatile private[remote] var supervisor: Option[Supervisor] = None
private[remote] lazy val serializer: Serializer = {
val className = config.getString("akka.remote.cluster.serializer", Serializer.Java.getClass.getName)
Class.forName(className).newInstance.asInstanceOf[Serializer]
}
private[remote] def createClusterActor : Option[ClusterActor] = {
private[remote] lazy val serializer: Serializer =
Class.forName(config.getString("akka.remote.cluster.serializer", DEFAULT_SERIALIZER_CLASS_NAME))
.newInstance.asInstanceOf[Serializer]
private[remote] def createClusterActor: Option[ClusterActor] = {
val name = config.getString("akka.remote.cluster.actor")
if (name.isEmpty) throw new IllegalArgumentException(
"Can't start cluster since the 'akka.remote.cluster.actor' configuration option is not defined")
try {
name map { fqn =>
val a = Class.forName(fqn).newInstance.asInstanceOf[ClusterActor]
a.start
a
name map {
fqn =>
Class.forName(fqn).newInstance.asInstanceOf[ClusterActor]
}
}
catch {
case e => log.error(e,"Couldn't load Cluster provider: [%s]",name.getOrElse("Not specified")); None
case e => log.error(e, "Couldn't load Cluster provider: [%s]", name.getOrElse("Not specified")); None
}
}
private[remote] def createSupervisor(actor : ClusterActor) : Option[Supervisor] = {
private[remote] def createSupervisor(actor: ClusterActor): Option[Supervisor] = {
val sup = SupervisorFactory(
SupervisorConfig(
RestartStrategy(OneForOne, 5, 1000, List(classOf[Exception])),
Supervise(actor, LifeCycle(Permanent)) :: Nil)
).newInstance
sup.start
Some(sup)
}
@ -249,20 +248,23 @@ object Cluster extends Cluster with Logging {
def deregisterLocalNode(hostname: String, port: Int): Unit = clusterActor.foreach(_.deregisterLocalNode(hostname, port))
def relayMessage(to: Class[_ <: Actor], msg: AnyRef): Unit = clusterActor.foreach(_.relayMessage(to, msg))
def foreach(f : (RemoteAddress) => Unit) : Unit = clusterActor.foreach(_.foreach(f))
def start : Unit = synchronized {
if(supervisor.isEmpty) {
for(actor <- createClusterActor;
sup <- createSupervisor(actor)) {
clusterActor = Some(actor)
supervisor = Some(sup)
def foreach(f: (RemoteAddress) => Unit): Unit = clusterActor.foreach(_.foreach(f))
def start: Unit = synchronized {
log.info("Starting up Cluster Service...")
if (supervisor.isEmpty) {
for (actor <- createClusterActor;
sup <- createSupervisor(actor)) {
clusterActor = Some(actor)
supervisor = Some(sup)
sup.start
}
}
}
def shutdown : Unit = synchronized {
def shutdown: Unit = synchronized {
log.info("Shutting down Cluster Service...")
supervisor.foreach(_.stop)
supervisor = None
clusterActor = None

2
akka-core/src/main/scala/remote/RemoteServer.scala
View file

@ -58,7 +58,7 @@ object RemoteNode extends RemoteServer
*/
object RemoteServer {
val HOSTNAME = config.getString("akka.remote.server.hostname", "localhost")
val PORT = config.getInt("akka.remote.server.port", 9999)
val PORT = config.getInt("akka.remote.server.port", 9966)
val CONNECTION_TIMEOUT_MILLIS = config.getInt("akka.remote.server.connection-timeout", 1000)

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

@ -7,15 +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.templates.OrElseTemplate
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)
@ -30,8 +33,8 @@ class TransactionRetryException(message: String) extends RuntimeException(messag
* Here are some examples (assuming implicit transaction family name in scope):
* <pre>
* import se.scalablesolutions.akka.stm.Transaction._
*
* atomic {
*
* atomic {
* .. // do something within a transaction
* }
* </pre>
@ -39,8 +42,8 @@ class TransactionRetryException(message: String) extends RuntimeException(messag
* Example of atomic transaction management using atomic block with retry count:
* <pre>
* import se.scalablesolutions.akka.stm.Transaction._
*
* atomic(maxNrOfRetries) {
*
* atomic(maxNrOfRetries) {
* .. // do something within a transaction
* }
* </pre>
@ -49,10 +52,10 @@ class TransactionRetryException(message: String) extends RuntimeException(messag
* Which is a good way to reduce contention and transaction collisions.
* <pre>
* import se.scalablesolutions.akka.stm.Transaction._
*
* atomically {
*
* atomically {
* .. // try to do something
* } orElse {
* } orElse {
* .. // if transaction clashes try do do something else to minimize contention
* }
* </pre>
@ -61,11 +64,11 @@ class TransactionRetryException(message: String) extends RuntimeException(messag
*
* <pre>
* import se.scalablesolutions.akka.stm.Transaction._
* for (tx <- Transaction) {
* for (tx <- Transaction) {
* ... // do transactional stuff
* }
*
* val result = for (tx <- Transaction) yield {
* val result = for (tx <- Transaction) yield {
* ... // do transactional stuff yielding a result
* }
* </pre>
@ -78,17 +81,17 @@ class TransactionRetryException(message: String) extends RuntimeException(messag
*
* // You can use them together with Transaction in a for comprehension since
* // TransactionalRef is also monadic
* for {
* for {
* tx <- Transaction
* ref <- refs
* } {
* ... // use the ref inside a transaction
* }
*
* val result = for {
* val result = for {
* tx <- Transaction
* ref <- refs
* } yield {
* } yield {
* ... // use the ref inside a transaction, yield a result
* }
* </pre>
@ -97,101 +100,87 @@ 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) }
/**
* See ScalaDoc on class.
*/
def foreach(f: Transaction => Unit)(implicit transactionFamilyName: String): Unit = atomic { f(getTransactionInScope) }
def foreach(f: => Unit)(implicit transactionFamilyName: String): Unit =
atomic {f}
/**
* Creates a "pure" STM atomic transaction and by-passes all transactions hooks
* such as persistence etc.
* Only for internal usage.
*/
private[akka] def pureAtomic[T](body: => T): T = new AtomicTemplate[T](
getGlobalStmInstance, "internal", false, false, TransactionManagement.MAX_NR_OF_RETRIES) {
private[akka] def pureAtomic[T](body: => T): T = new TransactionTemplate[T]() {
def execute(mtx: MultiverseTransaction): T = body
}.execute()
/**
* See ScalaDoc on class.
*/
def atomic[T](body: => T)(implicit transactionFamilyName: String): T = new AtomicTemplate[T](
getGlobalStmInstance, transactionFamilyName, false, false, TransactionManagement.MAX_NR_OF_RETRIES) {
def execute(mtx: MultiverseTransaction): T = body
override def postStart(mtx: MultiverseTransaction) = {
val tx = new Transaction
tx.transaction = Some(mtx)
setTransaction(Some(tx))
}
override def postCommit = {
if (isTransactionInScope) getTransactionInScope.commit
else throw new IllegalStateException("No transaction in scope")
}
}.execute()
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 = {
val result = body
/**
* See ScalaDoc on class.
*/
def atomic[T](retryCount: Int)(body: => T)(implicit transactionFamilyName: String): T = {
new AtomicTemplate[T](getGlobalStmInstance, transactionFamilyName, false, false, retryCount) {
def execute(mtx: MultiverseTransaction): T = body
override def postStart(mtx: MultiverseTransaction) = {
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 postCommit = {
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 AtomicTemplate[T](getGlobalStmInstance, transactionFamilyName, false, true, retryCount) {
def execute(mtx: MultiverseTransaction): T = body
override def postStart(mtx: MultiverseTransaction) = {
val tx = new Transaction
tx.transaction = Some(mtx)
setTransaction(Some(tx))
txSet.registerOnCommitTask(new Runnable() {
def run = tx.commit
})
txSet.registerOnAbortTask(new Runnable() {
def run = tx.abort
})
}
override def postCommit = {
if (isTransactionInScope) getTransactionInScope.commit
else throw new IllegalStateException("No transaction in scope")
}
}.execute
}
/**
* See ScalaDoc on class.
*/
def atomicReadOnly[T](body: => T): T = {
new AtomicTemplate[T](true) {
def execute(mtx: MultiverseTransaction): T = body
override def postStart(mtx: MultiverseTransaction) = {
val tx = new Transaction
tx.transaction = Some(mtx)
setTransaction(Some(tx))
}
override def postCommit = {
if (isTransactionInScope) getTransactionInScope.commit
else throw new IllegalStateException("No transaction in scope")
}
}.execute
}.execute()
}
/**
@ -216,23 +205,28 @@ 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)
// --- public methods ---------
def commit = synchronized {
log.trace("Committing transaction %s", toString)
pureAtomic {
persistentStateMap.values.foreach(_.commit)
TransactionManagement.clearTransaction
}
status = TransactionStatus.Completed
}
def abort = synchronized {
log.trace("Aborting transaction %s", toString)
}
def isNew = synchronized { status == TransactionStatus.New }
def isActive = synchronized { status == TransactionStatus.Active }
@ -259,13 +253,13 @@ object Transaction extends TransactionManagement {
private def ensureIsActiveOrAborted =
if (!(status == TransactionStatus.Active || status == TransactionStatus.Aborted))
throw new IllegalStateException(
"Expected ACTIVE or ABORTED transaction - current status [" + status + "]: " + toString)
throw new IllegalStateException(
"Expected ACTIVE or ABORTED transaction - current status [" + status + "]: " + toString)
private def ensureIsActiveOrNew =
if (!(status == TransactionStatus.Active || status == TransactionStatus.New))
throw new IllegalStateException(
"Expected ACTIVE or NEW transaction - current status [" + status + "]: " + toString)
throw new IllegalStateException(
"Expected ACTIVE or NEW transaction - current status [" + status + "]: " + toString)
// For reinitialize transaction after sending it over the wire
private[akka] def reinit = synchronized {
@ -277,14 +271,14 @@ object Transaction extends TransactionManagement {
}
override def equals(that: Any): Boolean = synchronized {
that != null &&
that.isInstanceOf[Transaction] &&
that.asInstanceOf[Transaction].id == this.id
that != null &&
that.isInstanceOf[Transaction] &&
that.asInstanceOf[Transaction].id == this.id
}
override def hashCode(): Int = synchronized { id.toInt }
override def toString(): String = synchronized { "Transaction[" + id + ", " + status + "]" }
override def toString = synchronized { "Transaction[" + id + ", " + status + "]" }
}
/**

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

@ -9,51 +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(
val cause: Throwable, val tx: Option[Transaction]) extends RuntimeException(cause) {
override def toString(): String = "TransactionAwareWrapperException[" + cause + ", " + tx + "]"
class TransactionAwareWrapperException(val cause: Throwable, val tx: Option[Transaction]) extends RuntimeException(cause) {
override def toString = "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: ThreadLocal[Option[Transaction]] = 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 setTransaction(transaction: Option[Transaction]) = if (transaction.isDefined) {
val tx = transaction.get
currentTransaction.set(transaction)
if (tx.transaction.isDefined) setThreadLocalTransaction(tx.transaction.get)
else throw new IllegalStateException("No transaction defined")
private[akka] def createNewTransactionSet: CountDownCommitBarrier = {
val txSet = new CountDownCommitBarrier(1, TransactionManagement.FAIR_TRANSACTIONS)
TransactionManagement.transactionSet.set(Some(txSet))
txSet
}
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 isTransactionInScope = currentTransaction.get.isDefined
private[akka] def getTransactionSetInScope = TransactionManagement.getTransactionSet
private[akka] def incrementTransaction = if (isTransactionInScope) getTransactionInScope.increment
private[akka] def getTransactionInScope = TransactionManagement.getTransaction
private[akka] def decrementTransaction = if (isTransactionInScope) getTransactionInScope.decrement
}
private[akka] def isTransactionSetInScope = {
val option = TransactionManagement.transactionSet.get
(option ne null) && option.isDefined
}
private[akka] def isTransactionInScope = {
val option = TransactionManagement.transaction.get
(option ne null) && option.isDefined
}
}

5
akka-core/src/main/scala/stm/TransactionalState.scala
View file

@ -8,8 +8,7 @@ import se.scalablesolutions.akka.stm.Transaction.atomic
import se.scalablesolutions.akka.stm.NoTransactionInScopeException
import se.scalablesolutions.akka.collection._
import se.scalablesolutions.akka.util.UUID
import org.multiverse.datastructures.refs.manual.Ref;
import org.multiverse.stms.alpha.AlphaRef
/**
* Example Scala usage:
@ -78,7 +77,7 @@ class TransactionalRef[T] extends Transactional {
implicit val txInitName = "TransactionalRef:Init"
val uuid = UUID.newUuid.toString
private[this] val ref: Ref[T] = atomic { new Ref }
private[this] lazy val ref: AlphaRef[T] = new AlphaRef
def swap(elem: T) = {
ensureIsInTransaction

28
akka-core/src/test/scala/InMemoryActorTest.scala
View file

@ -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

2
akka-core/src/test/scala/MemoryTest.scala
View file

@ -22,9 +22,11 @@ class MemoryFootprintTest extends JUnitSuite {
// Actors are put in AspectRegistry when created so they won't be GCd here
val totalMem = Runtime.getRuntime.totalMemory - Runtime.getRuntime.freeMemory
println("Memory before " + totalMem)
(1 until NR_OF_ACTORS).foreach(i => new Mem)
val newTotalMem = Runtime.getRuntime.totalMemory - Runtime.getRuntime.freeMemory
println("Memory aftor " + newTotalMem)
val memPerActor = (newTotalMem - totalMem) / NR_OF_ACTORS
println("Memory footprint per actor is : " + memPerActor)

4
akka-core/src/test/scala/PerformanceTest.scala
View file

@ -1,4 +1,4 @@
package test
package se.scalablesolutions.akka
import org.scalatest.junit.JUnitSuite
import org.junit.Test
@ -279,7 +279,7 @@ class PerformanceTest extends JUnitSuite {
var nrOfMessages = 2000000
var nrOfActors = 4
var akkaTime = stressTestAkkaActors(nrOfMessages, nrOfActors, 1000 * 20)
var akkaTime = stressTestAkkaActors(nrOfMessages, nrOfActors, 1000 * 30)
var scalaTime = stressTestScalaActors(nrOfMessages, nrOfActors, 1000 * 40)
var ratio: Double = scalaTime.toDouble / akkaTime.toDouble

5
akka-core/src/test/scala/ShutdownSpec.scala
View file

@ -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

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

@ -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
}
}
*/

2
akka-kernel/src/main/scala/Kernel.scala
View file

@ -79,7 +79,7 @@ object Kernel extends Logging {
(____ /__|_ \__|_ \(____ /
\/ \/ \/ \/
""")
log.info(" Running version %s", Config.VERSION)
log.info(" Running version %s", Config.VERSION)
log.info("==============================")
}
}

24
akka-patterns/src/main/scala/Agent.scala
View file

@ -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

62
akka-patterns/src/main/scala/Patterns.scala
View file

@ -3,14 +3,14 @@ package se.scalablesolutions.akka.actor.patterns
import se.scalablesolutions.akka.actor.Actor
object Patterns {
type PF[A,B] = PartialFunction[A,B]
type PF[A, B] = PartialFunction[A, B]
/**
* Creates a new PartialFunction whose isDefinedAt is a combination
* of the two parameters, and whose apply is first to call filter.apply and then filtered.apply
*/
def filter[A,B](filter : PF[A,Unit],filtered : PF[A,B]) : PF[A,B] = {
case a : A if filtered.isDefinedAt(a) && filter.isDefinedAt(a) =>
def filter[A, B](filter: PF[A, Unit], filtered: PF[A, B]): PF[A, B] = {
case a: A if filtered.isDefinedAt(a) && filter.isDefinedAt(a) =>
filter(a)
filtered(a)
}
@ -18,39 +18,42 @@ object Patterns {
/**
* Interceptor is a filter(x,y) where x.isDefinedAt is considered to be always true
*/
def intercept[A,B](interceptor : (A) => Unit, interceptee : PF[A,B]) : PF[A,B] = filter(
{ case a if a.isInstanceOf[A] => interceptor(a) },
interceptee
def intercept[A, B](interceptor: (A) => Unit, interceptee: PF[A, B]): PF[A, B] = filter(
{case a if a.isInstanceOf[A] => interceptor(a)},
interceptee
)
//FIXME 2.8, use default params with CyclicIterator
def loadBalancerActor(actors : => InfiniteIterator[Actor]) : Actor = new Actor with LoadBalancer {
def loadBalancerActor(actors: => InfiniteIterator[Actor]): Actor = new Actor with LoadBalancer {
val seq = actors
}
def dispatcherActor(routing : PF[Any,Actor], msgTransformer : (Any) => Any) : Actor = new Actor with Dispatcher {
override def transform(msg : Any) = msgTransformer(msg)
def dispatcherActor(routing: PF[Any, Actor], msgTransformer: (Any) => Any): Actor = new Actor with Dispatcher {
override def transform(msg: Any) = msgTransformer(msg)
def routes = routing
}
def dispatcherActor(routing : PF[Any,Actor]) : Actor = new Actor with Dispatcher {
def routes = routing
def dispatcherActor(routing: PF[Any, Actor]): Actor = new Actor with Dispatcher {
def routes = routing
}
def loggerActor(actorToLog : Actor, logger : (Any) => Unit) : Actor = dispatcherActor (
{ case _ => actorToLog },
def loggerActor(actorToLog: Actor, logger: (Any) => Unit): Actor = dispatcherActor(
{case _ => actorToLog},
logger
)
)
}
trait Dispatcher { self : Actor =>
trait Dispatcher {
self: Actor =>
protected def transform(msg : Any) : Any = msg
protected def routes : PartialFunction[Any,Actor]
protected def dispatch : PartialFunction[Any,Unit] = {
protected def transform(msg: Any): Any = msg
protected def routes: PartialFunction[Any, Actor]
protected def dispatch: PartialFunction[Any, Unit] = {
case a if routes.isDefinedAt(a) => {
if(self.sender.isDefined)
if (self.sender.isDefined)
routes(a) forward transform(a)
else
routes(a) send transform(a)
@ -60,19 +63,22 @@ trait Dispatcher { self : Actor =>
def receive = dispatch
}
trait LoadBalancer extends Dispatcher { self : Actor =>
protected def seq : InfiniteIterator[Actor]
trait LoadBalancer extends Dispatcher {
self: Actor =>
protected def seq: InfiniteIterator[Actor]
protected def routes = { case x if seq.hasNext => seq.next }
protected def routes = {case x if seq.hasNext => seq.next}
}
trait InfiniteIterator[T] extends Iterator[T]
class CyclicIterator[T](items : List[T]) extends InfiniteIterator[T] {
@volatile private[this] var current : List[T] = items
class CyclicIterator[T](items: List[T]) extends InfiniteIterator[T] {
@volatile private[this] var current: List[T] = items
def hasNext = items != Nil
def next = {
val nc = if(current == Nil) items else current
val nc = if (current == Nil) items else current
current = nc.tail
nc.head
}

6
akka-patterns/src/test/scala/ActorPatternsTest.scala
View file

@ -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)

25
akka-patterns/src/test/scala/AgentTest.scala
View file

@ -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()
r must be (12)
}
}
@Test def testAgent = verify(new TestActor {
def test = {
atomic {
val t = Agent(5)
handle(t) {
t.update(_ + 1)
t.update(_ * 2)
val r = t()
r must be(12)
}
}
}
})
}
*/

16
akka-persistence/akka-persistence-cassandra/src/test/scala/CassandraPersistentActorSpec.scala
View file

@ -1,13 +1,9 @@
package se.scalablesolutions.akka.state
import se.scalablesolutions.akka.actor.Actor
import junit.framework.TestCase
import se.scalablesolutions.akka.actor.{Actor, Transactor}
import org.junit.Test
import org.junit.Assert._
import org.apache.cassandra.service.CassandraDaemon
import org.junit.BeforeClass
import org.junit.Before
import org.scalatest.junit.JUnitSuite
@ -28,9 +24,8 @@ case class SetRefStateOneWay(key: String)
case class SuccessOneWay(key: String, value: String)
case class FailureOneWay(key: String, value: String, failer: Actor)
class CassandraPersistentActor extends Actor {
class CassandraPersistentActor extends Transactor {
timeout = 100000
makeTransactionRequired
private lazy val mapState = CassandraStorage.newMap
private lazy val vectorState = CassandraStorage.newVector
@ -66,8 +61,7 @@ class CassandraPersistentActor extends Actor {
}
}
@serializable class PersistentFailerActor extends Actor {
makeTransactionRequired
@serializable class PersistentFailerActor extends Transactor {
def receive = {
case "Failure" =>
throw new RuntimeException("expected")
@ -76,8 +70,8 @@ class CassandraPersistentActor extends Actor {
class CassandraPersistentActorSpec extends JUnitSuite {
@Before
def startCassandra = EmbeddedCassandraService.start
//@Before
//def startCassandra = EmbeddedCassandraService.start
@Test
def testMapShouldNotRollbackStateForStatefulServerInCaseOfSuccess = {

27
akka-persistence/akka-persistence-common/src/main/scala/Storage.scala
View file

@ -4,14 +4,15 @@
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
import org.codehaus.aspectwerkz.proxy.Uuid
// FIXME move to 'stm' package + add message with more info
class NoTransactionInScopeException extends RuntimeException
class StorageException(message: String) extends RuntimeException(message)
/**
* Example Scala usage.
* <p/>
@ -64,10 +65,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 +159,8 @@ trait PersistentMap[K, V] extends scala.collection.mutable.Map[K, V]
}
private def register = {
if (currentTransaction.get.isEmpty) throw new NoTransactionInScopeException
currentTransaction.get.get.register(uuid, this)
if (transaction.get.isEmpty) throw new NoTransactionInScopeException
transaction.get.get.register(uuid, this)
}
}
@ -236,8 +233,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
currentTransaction.get.get.register(uuid, this)
if (transaction.get.isEmpty) throw new NoTransactionInScopeException
transaction.get.get.register(uuid, this)
}
}
@ -272,8 +269,8 @@ trait PersistentRef[T] extends Transactional with Committable {
}
private def register = {
if (currentTransaction.get.isEmpty) throw new NoTransactionInScopeException
currentTransaction.get.get.register(uuid, this)
if (transaction.get.isEmpty) throw new NoTransactionInScopeException
transaction.get.get.register(uuid, this)
}
}
@ -397,7 +394,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
currentTransaction.get.get.register(uuid, this)
if (transaction.get.isEmpty) throw new NoTransactionInScopeException
transaction.get.get.register(uuid, this)
}
}

13
akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala
View file

@ -8,7 +8,7 @@ import org.junit.Assert._
import _root_.dispatch.json.{JsNumber, JsValue}
import _root_.dispatch.json.Js._
import se.scalablesolutions.akka.actor.Actor
import se.scalablesolutions.akka.actor.{Transactor, Actor}
/**
* A persistent actor based on MongoDB storage.
@ -29,10 +29,10 @@ case class MultiDebit(accountNo: String, amounts: List[BigInt], failer: Actor)
case class Credit(accountNo: String, amount: BigInt)
case object LogSize
class BankAccountActor extends Actor {
makeTransactionRequired
private val accountState = MongoStorage.newMap
private val txnLog = MongoStorage.newVector
class BankAccountActor extends Transactor {
private lazy val accountState = MongoStorage.newMap
private lazy val txnLog = MongoStorage.newVector
def receive: PartialFunction[Any, Unit] = {
// check balance
@ -91,8 +91,7 @@ class BankAccountActor extends Actor {
}
}
@serializable class PersistentFailerActor extends Actor {
makeTransactionRequired
@serializable class PersistentFailerActor extends Transactor {
def receive = {
case "Failure" =>
throw new RuntimeException("expected")

103
akka-persistence/akka-persistence-redis/src/main/scala/RedisStorageBackend.scala
View file

@ -72,11 +72,11 @@ private [akka] object RedisStorageBackend extends
* base64(T1):base64("debasish.programming_language") -> "scala"
* </i>
*/
def insertMapStorageEntryFor(name: String, key: Array[Byte], value: Array[Byte]) {
def insertMapStorageEntryFor(name: String, key: Array[Byte], value: Array[Byte]): Unit = withErrorHandling {
insertMapStorageEntriesFor(name, List((key, value)))
}
def insertMapStorageEntriesFor(name: String, entries: List[Tuple2[Array[Byte], Array[Byte]]]) {
def insertMapStorageEntriesFor(name: String, entries: List[Tuple2[Array[Byte], Array[Byte]]]): Unit = withErrorHandling {
mset(entries.map(e =>
(makeRedisKey(name, e._1), new String(e._2))))
}
@ -89,22 +89,22 @@ private [akka] object RedisStorageBackend extends
* <li>: is chosen since it cannot appear in base64 encoding charset</li>
* <li>both parts of the key need to be based64 encoded since there can be spaces within each of them</li>
*/
private [this] def makeRedisKey(name: String, key: Array[Byte]): String = {
private [this] def makeRedisKey(name: String, key: Array[Byte]): String = withErrorHandling {
"%s:%s".format(new String(encode(name.getBytes)), new String(encode(key)))
}
private [this] def makeKeyFromRedisKey(redisKey: String) = {
private [this] def makeKeyFromRedisKey(redisKey: String) = withErrorHandling {
val nk = redisKey.split(':').map{e: String => decode(e.getBytes)}
(nk(0), nk(1))
}
private [this] def mset(entries: List[(String, String)]) {
private [this] def mset(entries: List[(String, String)]): Unit = withErrorHandling {
entries.foreach {e: (String, String) =>
db.set(e._1, e._2)
}
}
def removeMapStorageFor(name: String): Unit = {
def removeMapStorageFor(name: String): Unit = withErrorHandling {
db.keys("%s:*".format(encode(name.getBytes))) match {
case None =>
throw new Predef.NoSuchElementException(name + " not present")
@ -113,18 +113,19 @@ private [akka] object RedisStorageBackend extends
}
}
def removeMapStorageFor(name: String, key: Array[Byte]): Unit = {
def removeMapStorageFor(name: String, key: Array[Byte]): Unit = withErrorHandling {
db.delete(makeRedisKey(name, key))
}
def getMapStorageEntryFor(name: String, key: Array[Byte]): Option[Array[Byte]] =
def getMapStorageEntryFor(name: String, key: Array[Byte]): Option[Array[Byte]] = withErrorHandling {
db.get(makeRedisKey(name, key)) match {
case None =>
throw new Predef.NoSuchElementException(new String(key) + " not present")
case Some(s) => Some(s.getBytes)
}
}
def getMapStorageSizeFor(name: String): Int = {
def getMapStorageSizeFor(name: String): Int = withErrorHandling {
db.keys("%s:*".format(new String(encode(name.getBytes)))) match {
case None => 0
case Some(keys) =>
@ -132,7 +133,7 @@ private [akka] object RedisStorageBackend extends
}
}
def getMapStorageFor(name: String): List[(Array[Byte], Array[Byte])] = {
def getMapStorageFor(name: String): List[(Array[Byte], Array[Byte])] = withErrorHandling {
db.keys("%s:*".format(new String(encode(name.getBytes)))) match {
case None =>
throw new Predef.NoSuchElementException(name + " not present")
@ -143,7 +144,7 @@ private [akka] object RedisStorageBackend extends
def getMapStorageRangeFor(name: String, start: Option[Array[Byte]],
finish: Option[Array[Byte]],
count: Int): List[(Array[Byte], Array[Byte])] = {
count: Int): List[(Array[Byte], Array[Byte])] = withErrorHandling {
import scala.collection.immutable.TreeMap
val wholeSorted =
@ -188,19 +189,19 @@ private [akka] object RedisStorageBackend extends
}
}
def insertVectorStorageEntryFor(name: String, element: Array[Byte]) {
def insertVectorStorageEntryFor(name: String, element: Array[Byte]): Unit = withErrorHandling {
db.lpush(new String(encode(name.getBytes)), new String(element))
}
def insertVectorStorageEntriesFor(name: String, elements: List[Array[Byte]]) {
def insertVectorStorageEntriesFor(name: String, elements: List[Array[Byte]]): Unit = withErrorHandling {
elements.foreach(insertVectorStorageEntryFor(name, _))
}
def updateVectorStorageEntryFor(name: String, index: Int, elem: Array[Byte]) {
def updateVectorStorageEntryFor(name: String, index: Int, elem: Array[Byte]): Unit = withErrorHandling {
db.lset(new String(encode(name.getBytes)), index, new String(elem))
}
def getVectorStorageEntryFor(name: String, index: Int): Array[Byte] = {
def getVectorStorageEntryFor(name: String, index: Int): Array[Byte] = withErrorHandling {
db.lindex(new String(encode(name.getBytes)), index) match {
case None =>
throw new Predef.NoSuchElementException(name + " does not have element at " + index)
@ -208,7 +209,7 @@ private [akka] object RedisStorageBackend extends
}
}
def getVectorStorageRangeFor(name: String, start: Option[Int], finish: Option[Int], count: Int): List[Array[Byte]] = {
def getVectorStorageRangeFor(name: String, start: Option[Int], finish: Option[Int], count: Int): List[Array[Byte]] = withErrorHandling {
/**
* <tt>count</tt> is the max number of results to return. Start with
* <tt>start</tt> or 0 (if <tt>start</tt> is not defined) and go until
@ -237,11 +238,11 @@ private [akka] object RedisStorageBackend extends
}
}
def insertRefStorageFor(name: String, element: Array[Byte]) {
def insertRefStorageFor(name: String, element: Array[Byte]): Unit = withErrorHandling {
db.set(new String(encode(name.getBytes)), new String(element))
}
def getRefStorageFor(name: String): Option[Array[Byte]] = {
def getRefStorageFor(name: String): Option[Array[Byte]] = withErrorHandling {
db.get(new String(encode(name.getBytes))) match {
case None =>
throw new Predef.NoSuchElementException(name + " not present")
@ -250,12 +251,13 @@ private [akka] object RedisStorageBackend extends
}
// add to the end of the queue
def enqueue(name: String, item: Array[Byte]): Boolean = {
def enqueue(name: String, item: Array[Byte]): Boolean = withErrorHandling {
db.rpush(new String(encode(name.getBytes)), new String(item))
}
// pop from the front of the queue
def dequeue(name: String): Option[Array[Byte]] = {
def dequeue(name: String): Option[Array[Byte]] = withErrorHandling {
db.lpop(new String(encode(name.getBytes))) match {
case None =>
throw new Predef.NoSuchElementException(name + " not present")
@ -265,7 +267,7 @@ private [akka] object RedisStorageBackend extends
}
// get the size of the queue
def size(name: String): Int = {
def size(name: String): Int = withErrorHandling {
db.llen(new String(encode(name.getBytes))) match {
case None =>
throw new Predef.NoSuchElementException(name + " not present")
@ -275,26 +277,28 @@ private [akka] object RedisStorageBackend extends
// return an array of items currently stored in the queue
// start is the item to begin, count is how many items to return
def peek(name: String, start: Int, count: Int): List[Array[Byte]] = count match {
case 1 =>
db.lindex(new String(encode(name.getBytes)), start) match {
case None =>
throw new Predef.NoSuchElementException("No element at " + start)
case Some(s) =>
List(s.getBytes)
}
case n =>
db.lrange(new String(encode(name.getBytes)), start, start + count - 1) match {
case None =>
throw new Predef.NoSuchElementException(
"No element found between " + start + " and " + (start + count - 1))
case Some(es) =>
es.map(_.get.getBytes)
}
def peek(name: String, start: Int, count: Int): List[Array[Byte]] = withErrorHandling {
count match {
case 1 =>
db.lindex(new String(encode(name.getBytes)), start) match {
case None =>
throw new Predef.NoSuchElementException("No element at " + start)
case Some(s) =>
List(s.getBytes)
}
case n =>
db.lrange(new String(encode(name.getBytes)), start, start + count - 1) match {
case None =>
throw new Predef.NoSuchElementException(
"No element found between " + start + " and " + (start + count - 1))
case Some(es) =>
es.map(_.get.getBytes)
}
}
}
// completely delete the queue
def remove(name: String): Boolean = {
def remove(name: String): Boolean = withErrorHandling {
db.delete(new String(encode(name.getBytes))) match {
case Some(1) => true
case _ => false
@ -302,7 +306,7 @@ private [akka] object RedisStorageBackend extends
}
// add item to sorted set identified by name
def zadd(name: String, zscore: String, item: Array[Byte]): Boolean = {
def zadd(name: String, zscore: String, item: Array[Byte]): Boolean = withErrorHandling {
db.zadd(new String(encode(name.getBytes)), zscore, new String(item)) match {
case Some(1) => true
case _ => false
@ -310,7 +314,7 @@ private [akka] object RedisStorageBackend extends
}
// remove item from sorted set identified by name
def zrem(name: String, item: Array[Byte]): Boolean = {
def zrem(name: String, item: Array[Byte]): Boolean = withErrorHandling {
db.zrem(new String(encode(name.getBytes)), new String(item)) match {
case Some(1) => true
case _ => false
@ -318,7 +322,7 @@ private [akka] object RedisStorageBackend extends
}
// cardinality of the set identified by name
def zcard(name: String): Int = {
def zcard(name: String): Int = withErrorHandling {
db.zcard(new String(encode(name.getBytes))) match {
case None =>
throw new Predef.NoSuchElementException(name + " not present")
@ -326,7 +330,7 @@ private [akka] object RedisStorageBackend extends
}
}
def zscore(name: String, item: Array[Byte]): String = {
def zscore(name: String, item: Array[Byte]): String = withErrorHandling {
db.zscore(new String(encode(name.getBytes)), new String(item)) match {
case None =>
throw new Predef.NoSuchElementException(new String(item) + " not present")
@ -334,7 +338,7 @@ private [akka] object RedisStorageBackend extends
}
}
def zrange(name: String, start: Int, end: Int): List[Array[Byte]] = {
def zrange(name: String, start: Int, end: Int): List[Array[Byte]] = withErrorHandling {
db.zrange(new String(encode(name.getBytes)), start.toString, end.toString, RedisClient.ASC, false) match {
case None =>
throw new Predef.NoSuchElementException(name + " not present")
@ -343,5 +347,16 @@ private [akka] object RedisStorageBackend extends
}
}
def flushDB = db.flushDb
def flushDB = withErrorHandling(db.flushDb)
private def withErrorHandling[T](body: => T): T = {
try {
body
} catch {
case e: java.lang.NullPointerException =>
throw new StorageException("Could not connect to Redis server")
case e =>
throw new StorageException("Error in Redis: " + e.getMessage)
}
}
}

6
akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala
View file

@ -29,6 +29,7 @@ case object LogSize
class AccountActor extends Transactor {
private lazy val accountState = RedisStorage.newMap
private lazy val txnLog = RedisStorage.newVector
//timeout = 5000
def receive = {
// check balance
@ -86,6 +87,7 @@ class AccountActor extends Transactor {
}
@serializable class PersistentFailerActor extends Transactor {
//timeout = 5000
def receive = {
case "Failure" =>
throw new RuntimeException("expected")
@ -138,7 +140,7 @@ class RedisPersistentActorSpec extends TestCase {
bactor.start
bactor !! Credit("a-123", 5000)
assertEquals(BigInt(5000), (bactor !! Balance("a-123")).get)
assertEquals(BigInt(5000), (bactor !! (Balance("a-123"), 5000)).get)
val failer = new PersistentFailerActor
failer.start
@ -147,7 +149,7 @@ class RedisPersistentActorSpec extends TestCase {
fail("should throw exception")
} catch { case e: RuntimeException => {}}
assertEquals(BigInt(5000), (bactor !! Balance("a-123")).get)
assertEquals(BigInt(5000), (bactor !! (Balance("a-123"), 5000)).get)
// should not count the failed one
assertEquals(3, (bactor !! LogSize).get)

341
akka-util-java/src/main/java/se/scalablesolutions/akka/stm/AtomicTemplate.java
View file

@ -1,341 +0,0 @@
package se.scalablesolutions.akka.stm;
import static org.multiverse.api.GlobalStmInstance.getGlobalStmInstance;
import org.multiverse.api.Stm;
import static org.multiverse.api.ThreadLocalTransaction.getThreadLocalTransaction;
import static org.multiverse.api.ThreadLocalTransaction.setThreadLocalTransaction;
import org.multiverse.api.Transaction;
import org.multiverse.api.TransactionStatus;
import org.multiverse.api.exceptions.CommitFailureException;
import org.multiverse.api.exceptions.LoadException;
import org.multiverse.api.exceptions.RetryError;
import org.multiverse.api.exceptions.TooManyRetriesException;
import org.multiverse.templates.AbortedException;
import org.multiverse.utils.latches.CheapLatch;
import org.multiverse.utils.latches.Latch;
import static java.lang.String.format;
import java.util.logging.Logger;
/**
* A Template that handles the boilerplate code for transactions. A transaction will be placed if none is available
* around a section and if all goes right, commits at the end.
* <p/>
* example:
* <pre>
* new AtomicTemplate(){
* Object execute(Transaction t){
* queue.push(1);
* return null;
* }
* }.execute();
* </pre>
* <p/>
* It could also be that the transaction is retried (e.g. caused by optimistic locking failures). This is also a task
* for template. In the future this retry behavior will be customizable.
* <p/>
* If a transaction already is available on the TransactionThreadLocal, no new transaction is started and essentially
* the whole AtomicTemplate is ignored.
* <p/>
* If no transaction is available on the TransactionThreadLocal, a new one will be created and used during the execution
* of the AtomicTemplate and will be removed once the AtomicTemplate finishes.
* <p/>
* All uncaught throwable's lead to a rollback of the transaction.
* <p/>
* AtomicTemplates are not thread-safe to use.
* <p/>
* AtomicTemplates can completely work without threadlocals. See the {@link AtomicTemplate#AtomicTemplate(org.multiverse.api.Stm
* ,String, boolean, boolean, int)} for more information.
*
* @author Peter Veentjer
*/
public abstract class AtomicTemplate<E> {
private final static Logger logger = Logger.getLogger(AtomicTemplate.class.getName());
private final Stm stm;
private final boolean ignoreThreadLocalTransaction;
private final int retryCount;
private final boolean readonly;
private int attemptCount;
private final String familyName;
/**
* Creates a new AtomicTemplate that uses the STM stored in the GlobalStm and works the the {@link
* org.multiverse.utils.ThreadLocalTransaction}.
*/
public AtomicTemplate() {
this(getGlobalStmInstance());
}
public AtomicTemplate(boolean readonly) {
this(getGlobalStmInstance(), null, false, readonly, Integer.MAX_VALUE);
}
/**
* Creates a new AtomicTemplate using the provided stm. The transaction used is stores/retrieved from the {@link
* org.multiverse.utils.ThreadLocalTransaction}.
*
* @param stm the stm to use for transactions.
* @throws NullPointerException if stm is null.
*/
public AtomicTemplate(Stm stm) {
this(stm, null, false, false, Integer.MAX_VALUE);
}
public AtomicTemplate(String familyName, boolean readonly, int retryCount) {
this(getGlobalStmInstance(), familyName, false, readonly, retryCount);
}
/**
* Creates a new AtomicTemplate that uses the provided STM. This method is provided to make Multiverse easy to
* integrate with environment that don't want to depend on threadlocals.
*
* @param stm the stm to use for transactions.
* @param ignoreThreadLocalTransaction true if this Template should completely ignore the ThreadLocalTransaction.
* This is useful for using the AtomicTemplate in other environments that don't
* want to depend on threadlocals but do want to use the AtomicTemplate.
* @throws NullPointerException if stm is null.
*/
public AtomicTemplate(Stm stm, String familyName, boolean ignoreThreadLocalTransaction, boolean readonly,
int retryCount) {
if (stm == null) {
throw new NullPointerException();
}
if (retryCount < 0) {
throw new IllegalArgumentException();
}
this.stm = stm;
this.ignoreThreadLocalTransaction = ignoreThreadLocalTransaction;
this.readonly = readonly;
this.retryCount = retryCount;
this.familyName = familyName;
}
public String getFamilyName() {
return familyName;
}
/**
* Returns the current attempt. Value will always be larger than zero and increases everytime the transaction needs
* to be retried.
*
* @return the current attempt count.
*/
public final int getAttemptCount() {
return attemptCount;
}
/**
* Returns the number of retries that this AtomicTemplate is allowed to do. The returned value will always be equal
* or larger than 0.
*
* @return the number of retries.
*/
public final int getRetryCount() {
return retryCount;
}
/**
* Returns the {@link Stm} used by this AtomicTemplate to execute transactions on.
*
* @return the Stm used by this AtomicTemplate.
*/
public final Stm getStm() {
return stm;
}
/**
* Check if this AtomicTemplate ignores the ThreadLocalTransaction.
*
* @return true if this AtomicTemplate ignores the ThreadLocalTransaction, false otherwise.
*/
public final boolean isIgnoreThreadLocalTransaction() {
return ignoreThreadLocalTransaction;
}
/**
* Checks if this AtomicTemplate executes readonly transactions.
*
* @return true if it executes readonly transactions, false otherwise.
*/
public final boolean isReadonly() {
return readonly;
}
/**
* This is the method can be overridden to do pre-start tasks.
*/
public void preStart() {
}
/**
* This is the method can be overridden to do post-start tasks.
*
* @param t the transaction used for this execution.
*/
public void postStart(Transaction t) {
}
/**
* This is the method can be overridden to do pre-commit tasks.
*/
public void preCommit() {
}
/**
* This is the method can be overridden to do post-commit tasks.
*/
public void postCommit() {
}
/**
* This is the method that needs to be implemented.
*
* @param t the transaction used for this execution.
* @return the result of the execution.
*
* @throws Exception the Exception thrown
*/
public abstract E execute(Transaction t) throws Exception;
/**
* Executes the template.
*
* @return the result of the {@link #execute(org.multiverse.api.Transaction)} method.
*
* @throws InvisibleCheckedException if a checked exception was thrown while executing the {@link
* #execute(org.multiverse.api.Transaction)} method.
* @throws AbortedException if the exception was explicitly aborted.
* @throws TooManyRetriesException if the template retried the transaction too many times. The cause of the last
* failure (also an exception) is included as cause. So you have some idea where
* to look for problems
*/
public final E execute() {
try {
return executeChecked();
} catch (Exception ex) {
if (ex instanceof RuntimeException) {
throw (RuntimeException) ex;
} else {
throw new AtomicTemplate.InvisibleCheckedException(ex);
}
}
}
/**
* Executes the Template and rethrows the checked exception instead of wrapping it in a InvisibleCheckedException.
*
* @return the result
*
* @throws Exception the Exception thrown inside the {@link #execute(org.multiverse.api.Transaction)}
* method.
* @throws AbortedException if the exception was explicitly aborted.
* @throws TooManyRetriesException if the template retried the transaction too many times. The cause of the last
* failure (also an exception) is included as cause. So you have some idea where to
* look for problems
*/
public final E executeChecked() throws Exception {
preStart();
Transaction t = getTransaction();
if (noUsableTransaction(t)) {
t = startTransaction();
setTransaction(t);
postStart(t);
try {
attemptCount = 1;
Exception lastRetryCause = null;
while (attemptCount - 1 <= retryCount) {
boolean abort = true;
boolean reset = false;
try {
E result = execute(t);
if (t.getStatus().equals(TransactionStatus.aborted)) {
String msg = format("Transaction with familyname %s is aborted", t.getFamilyName());
throw new AbortedException(msg);
}
preCommit();
t.commit();
abort = false;
reset = false;
postCommit();
return result;
} catch (RetryError e) {
Latch latch = new CheapLatch();
t.abortAndRegisterRetryLatch(latch);
latch.awaitUninterruptible();
//since the abort is already done, no need to do it again.
abort = false;
} catch (CommitFailureException ex) {
lastRetryCause = ex;
reset = true;
//ignore, just retry the transaction
} catch (LoadException ex) {
lastRetryCause = ex;
reset = true;
//ignore, just retry the transaction
} finally {
if (abort) {
t.abort();
if (reset) {
t = t.abortAndReturnRestarted();
setTransaction(t);
}
}
}
attemptCount++;
}
throw new TooManyRetriesException("Too many retries", lastRetryCause);
} finally {
setTransaction(null);
}
} else {
return execute(t);
}
}
private Transaction startTransaction() {
return readonly ? stm.startReadOnlyTransaction(familyName) : stm.startUpdateTransaction(familyName);
}
private boolean noUsableTransaction(Transaction t) {
return t == null || t.getStatus() != TransactionStatus.active;
}
/**
* Gets the current Transaction stored in the TransactionThreadLocal.
* <p/>
* If the ignoreThreadLocalTransaction is set, the threadlocal stuff is completeley ignored.
*
* @return the found transaction, or null if none is found.
*/
private Transaction getTransaction() {
return ignoreThreadLocalTransaction ? null : getThreadLocalTransaction();
}
/**
* Stores the transaction in the TransactionThreadLocal.
* <p/>
* This call is ignored if the ignoreThreadLocalTransaction is true.
*
* @param t the transaction to set (is allowed to be null).
*/
private void setTransaction(Transaction t) {
if (!ignoreThreadLocalTransaction) {
setThreadLocalTransaction(t);
}
}
public static class InvisibleCheckedException extends RuntimeException {
public InvisibleCheckedException(Exception cause) {
super(cause);
}
@Override
public Exception getCause() {
return (Exception) super.getCause();
}
}
}

17
akka.iml
View file

@ -2,6 +2,23 @@
<module org.jetbrains.idea.maven.project.MavenProjectsManager.isMavenModule="true" type="JAVA_MODULE" version="4">
<component name="FacetManager">
<facet type="Scala" name="Scala">
<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>
<facet type="Spring" name="Spring">
<configuration />
</facet>
<facet type="WebBeans" name="Web Beans">
<configuration />
</facet>
</component>

7
config/akka-reference.conf
View file

@ -30,8 +30,10 @@
<stm>
service = on
max-nr-of-retries = 100
distributed = off # not implemented yet
fair = on # should transactions be fair or non-fair (non fair yield better performance)
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>
@ -47,6 +49,7 @@
zlib-compression-level = 6 # Options: 0-9 (1 being fastest and 9 being the most compressed), default is 6
<cluster>
service = on # FIXME add 'service = on' for <cluster>
name = "default" # The name of the cluster
#actor = "se.scalablesolutions.akka.remote.JGroupsClusterActor" # FQN of an implementation of ClusterActor
serializer = "se.scalablesolutions.akka.serialization.Serializer$Java" # FQN of the serializer class

41
project/build/AkkaProject.scala
View file

@ -1,3 +1,39 @@
/*-------------------------------------------------------------------------------
Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
----------------------------------------------------
-------- sbt buildfile for the Akka project --------
----------------------------------------------------
Akka implements a unique hybrid of:
* Actors , which gives you:
* Simple and high-level abstractions for concurrency and parallelism.
* Asynchronous, non-blocking and highly performant event-driven programming
model.
* Very lightweight event-driven processes (create ~6.5 million actors on
4 G RAM).
* Supervision hierarchies with let-it-crash semantics. For writing highly
fault-tolerant systems that never stop, systems that self-heal.
* Software Transactional Memory (STM). (Distributed transactions coming soon).
* Transactors: combine actors and STM into transactional actors. Allows you to
compose atomic message flows with automatic rollback and retry.
* Remoting: highly performant distributed actors with remote supervision and
error management.
* Cluster membership management.
Akka also has a set of add-on modules:
* Persistence: A set of pluggable back-end storage modules that works in sync
with the STM.
* Cassandra distributed and highly scalable database.
* MongoDB document database.
* Redis data structures database (upcoming)
* REST (JAX-RS): Expose actors as REST services.
* Comet: Expose actors as Comet services.
* Security: Digest and Kerberos based security.
* Microkernel: Run Akka as a stand-alone kernel.
-------------------------------------------------------------------------------*/
import sbt._
class AkkaParent(info: ProjectInfo) extends ParentProject(info) {
@ -5,7 +41,8 @@ class AkkaParent(info: ProjectInfo) extends ParentProject(info) {
val sunjdmk = "sunjdmk" at "http://wp5.e-taxonomy.eu/cdmlib/mavenrepo"
val databinder = "DataBinder" at "http://databinder.net/repo"
val configgy = "Configgy" at "http://www.lag.net/repo"
val multiverse = "Multiverse" at "http://multiverse.googlecode.com/svn/maven-repository/releases"
val codehaus = "Codehaus" at "http://repository.codehaus.org"
val codehaus_snapshots = "Codehaus Snapshots" at "http://snapshots.repository.codehaus.org"
val jboss = "jBoss" at "http://repository.jboss.org/maven2"
val guiceyfruit = "GuiceyFruit" at "http://guiceyfruit.googlecode.com/svn/repo/releases/"
val embeddedrepo = "embedded repo" at "http://guice-maven.googlecode.com/svn/trunk"
@ -63,7 +100,7 @@ class AkkaParent(info: ProjectInfo) extends ParentProject(info) {
class AkkaJavaUtilProject(info: ProjectInfo) extends DefaultProject(info) {
val guicey = "org.guiceyfruit" % "guice-core" % "2.0-beta-4" % "compile"
val protobuf = "com.google.protobuf" % "protobuf-java" % "2.2.0" % "compile"
val multiverse = "org.multiverse" % "multiverse-alpha" % "0.3" % "compile"
val multiverse = "org.multiverse" % "multiverse-alpha" % "0.4-SNAPSHOT" % "compile"
}
class AkkaAMQPProject(info: ProjectInfo) extends DefaultProject(info) {