akka: simpler scalability, fault-tolerance, concurrency & remoting through actors

Jonas Bonér

Akka:Simpler Scalability, Fault-tolerance,

Concurrency & Remoting through Actors

http://akkasource.org

Scalable Solutions AB

Wednesday, December 30, 2009

The devil is in the state

Wrong, let me rephrase

The devil is in

the mutable state

Definitions&

Philosophy

What is a Value?A Value is something that

does not change

Discussion based onhttp://clojure.org/state

by Rich Hickey

What is an Identity?A stable logical entity

associated with a series of different Values

over time

What is State?The Value

an entity with a specific Identity

has at a particular point in time

How do we know if something has State?

If a function is invoked with the same arguments at

two different points in time and returns different values...

...then it has state

The ProblemThe unification Of

Identity & Value

They are not the same

We need to separate Identity & Value

...add a level of indirection

Software Transactional Memory Managed References

Message-Passing ConcurrencyActors/Active Objects

Dataflow ConcurrencyDataflow (Single-Assignment) Variables

The problems with Shared State Concurrency

Shared-State Concurrency>Concurrent access to shared, mutable state. >Protect mutable state with locks >The JavaC#C/C++RubyPythonetc.

Shared-State Concurrency is incredibly hard

>Inherently very hard to use reliably>Even the experts get it wrong

Example of Shared-State Concurrency

Transfer funds between bank accounts

AccountpublicclassAccount{privateintbalance;publicvoidwithdraw(intamount){balance‐=amount;}publicvoiddeposit(intamount){balance+=amount;}}

Not thread-safeWednesday, December 30, 2009

Let’s make it thread-safepublicclassAccount{privateintbalance;publicsynchronizedvoidwithdraw(intamount){balance‐=amount;}publicsynchronizedvoiddeposit(intamount){balance+=amount;}}

Thread-safe right?Wednesday, December 30, 2009

It’s still brokenTransfers are not atomic

Let’s write an atomic transfer method

publicclassAccount{...

publicsynchronizedvoidtransferTo(Accountto,doubleamount){this.withdraw(amount);to.deposit(amount);}...}

This will work right?Wednesday, December 30, 2009

Let’s transfer fundsAccountalice=...Accountbob=...//inonethreadalice.transferTo(bob,10.0D);//inanotherthreadbob.transferTo(alice,3.0D);

Might lead to

DEADLOCKDarn, this is really hard!!!

We need to enforce lock ordering>How? >Java won’t help us >Need to use code convention (names etc.) >Requires knowledge about the internal state and implementation of Account

>…runs counter to the principles of encapsulation in OOP

>Opens up a Can of Worms

The problem with locksLocks do not composeTaking too few locksTaking too many locksTaking the wrong locksTaking locks in the wrong orderError recovery is hard

It’s justtoo hard

Java bet on the wrong horse?

Perhaps, but we’re not

completely screwed There are alternatives

We need better and more high-level

abstractions

Alternative Paradigms>Software Transactional Memory (STM) >Message-Passing Concurrency (Actors) >Dataflow Concurrency

Actors

•Originates in a 1973 paper by Carl Hewitt

• Implemented in Erlang, Occam, Oz•Encapsulates state and behavior•Closer to the definition of OO than classes

Actors

“OOP to me means only messaging, local retention and protection and

hiding of state-process, and extreme late-binding of all things.”

“Actually I made up the term “object-oriented”, and I can tell you

I did not have C++ in mind.”

Replace C++ with Java or C#

Alan Kay (father of SmallTalk and OOP)

Actors• Implements Message-Passing Concurrency• Share NOTHING• Isolated lightweight processes• Communicates through messages• Asynchronous and non-blocking

Actor Model of Concurrency

• No shared state … hence, nothing to synchronize.• Each actor has a mailbox (message queue)

• Non-blocking send• Blocking receive• Messages are immutable• Highly performant and scalable

• SEDA-style (Staged Event-Driven Architecture)

• Easier to reason about• Raised abstraction level• Easier to avoid

–Race conditions–Deadlocks–Starvation–Live locks

Akka Actors• Asynchronous

–Fire-and-forget–Futures (Send Receive Reply Eventually)

• Synchronous• Message loop with pattern (message) matching

• Erlang-style

Two different models• Thread-based• Event-based

–Very lightweight–Can easily create millions on a single workstation (6.5 million on 4 G RAM)

> Akka (Java/Scala)> Kilim (Java)> Jetlang (Java)> Actor’s Guild (Java)> ActorFoundry (Java)> Actorom (Java)> FunctionalJava (Java)> GParallelizer (Groovy)> Fan Actors (Fan)

Actor libs for the JVM

Akka TransactorsSupervisor hierarchies

STMPersistent

Fault tolerance & Scalablility

Distributed

RESTful Comet

Secure

caseobjectTick

classCounterextendsActor{privatevarcounter=0

defreceive={caseTick=>counter+=1println(counter)}}

Actors

valworker=actor{caseWork(fn)=>fn()}

Actorsanonymous

valworker=actor{...//init}receive{caseWork(fn)=>fn()}

Actorsanonymous

//fire‐forgetcounter!Tick

Send: !

classSomeActorextendsActor{defreceive={caseUser(name)=>//useimplicitsendersender.get!(“Hi”+name)}}

classSomeActorextendsActor{defreceive={caseUser(name)=>//usereplyreply(“Hi”+name)}}

//usesFuturewithdefaulttimeoutvalresultOption=actor!!Messagevalresult=resultOptiongetOrElsedefaultResult

//usesFuturewithexplicittimeout(actor!!(Message,1000)).getOrElse(thrownewException(“Timedout”))

Send: !!

classSomeActorextendsActor{defreceive={caseUser(name)=>//usereplyreply(“Hi”+name)}}

classSomeActorextendsActor{defreceive={caseUser(name)=>//storeawaythesenderfuture//toresolvelateror//somewhereelse...=senderFuture}}

actor.startactor.stop

spawn(classOf[MyActor])

//callbackoverridedefshutdown={...//cleanupbeforeshutdown}

Start / Stop

publicclassCounter{privateintcounter=0;publicvoidcount(){counter++;System.out.println(counter);}}

Active Objects: Java

Countercounter=(Counter)ActiveObject.newInstance(Counter.class,1000);

Create: POJO

Countercounter=(Counter)ActiveObject.newInstance(Counter.class,CounterImpl.class,1000);

Create: Interface & Implementation

classCounter{privatevarcounter=0defcount={counter+=1println(counter)}}

Active Objects

valcounter=ActiveObject.newInstance(classOf[Counter],1000)

Create: POSO

counter.count

@onewayclassCounter{privatevarcounter=0@onewaydefcount={counter+=1println(counter)}}

//definethecaseclasscaseclassRegister(user:User)

//createandsendanewcaseclassmessageactor!Register(user)

//tuplesactor!(username,password)

//listsactor!List(“bill”,“bob”,“alice”)

Immutable messages

<akka>version="0.6"<actor>timeout=5000serialize‐messages=off</actor></akka>

Actors: config

Akka Dispatchers

classDispatchers{defnewThreadBasedDispatcher(actor:Actor)

defnewExecutorBasedEventDrivenDispatcher(name:String)...}

Dispatchers

classMyActorextendsActor{dispatcher=Dispatchers.newThreadBasedDispatcher(this)...}

actor.dispatcher=dispatcher//beforestarted

Set dispatcher

valdispatcher=Dispatchers.newExecutorBasedEventDrivenDispatcher.withNewThreadPoolWithBoundedBlockingQueue(100).setCorePoolSize(16).setMaxPoolSize(128).setKeepAliveTimeInMillis(60000).setRejectionPolicy(newCallerRunsPolicy).buildThreadPool

EventBasedDispatcherFluent DSL

When to use which dispatcher?

Thread-based actors• One thread per Actor• Good: • Threads (actors) don’t block each other• Good for long-running tasks

• Bad:• Poor scalability• Bad for short running tasks

• Use for a limited number of Actors• Use for low frequency of messages

Event-based actors

• Backed by thread pool • Lightweight: • Can create millions of Actors • 6.5 million on 4 G RAM

•Best scalability and performance•2 million messages in 8 seconds

Single threaded event-based actors

•Best performance• Millions of Actors•Bad: •one actor can block all other actors

•Does not take advantage of multicore

MessageQueues

• Unbounded LinkedBlockingQueue• Bounded LinkedBlockingQueue• Bounded ArrayBlockingQueue

Bounded SynchronousQueue

Plus different options per queue

Akka Supervision

Stolen from

ErlangWednesday, December 30, 2009

9 nines

Supervisor hierarchiesOneForOne

Supervisor hierarchiesAllForOne

AllForOneStrategy(maxNrOfRetries,withinTimeRange)

OneForOneStrategy(maxNrOfRetries,withinTimeRange)

Fault handlers

link(actor)unlink(actor)

startLink(actor)spawnLink(classOf[MyActor])

Linking

trapExit=List(classOf[ServiceException],classOf[PersistenceException])

trapExit

classSupervisorextendsActor{trapExit=List(classOf[Throwable])faultHandler=Some(OneForOneStrategy(5,5000))

defreceive={caseRegister(actor)=>link(actor)}}

Supervision

classFaultTolerantextendsActor{...overridedefpreRestart(reason:AnyRef)={...//cleanupbeforerestart}overridedefpostRestart(reason:AnyRef)={...//initafterrestart}}

Manage failure

RestartStrategy(AllForOne,//restartpolicy10,//max#ofrestartretries5000)//withintimeinmillis

LifeCycle(//Permanent:alwaysberestarted//Temporary:restartedifexitedthroughERRPermanent)

Declarative config

objectfactoryextendsSupervisorFactory(SupervisorConfig(RestartStrategy(AllForOne,3,10000),Supervise(actor1,LifeCycle(Permanent))::Supervise(actor2,LifeCycle(Temporary))::Nil))

factory.newSupervisor.start

Declarative config

valactors=ActorRegistry.actorsFor(FQN)

valactors=ActorRegistry.actorsFor(classOf[..])

ActorRegistry

overridedefinit={...//inittheactor}

Initialize actor

init is called on startup

Akka Remote Actors

//usehost&portinconfigRemoteNode.startRemoteNode.start(classLoader)

RemoteNode.start("localhost",9999)RemoteNode.start("localhost",9999,classLoader)

Remote Server

//usehost&portinconfigvalserver=newRemoteServer

server.start("localhost",9999)

Remote Server

spawnRemote(classOf[MyActor],host,port)

startLinkRemote(actor,host,port)

spawnLinkRemote(classOf[MyActor],host,port)

Remote actors

<akka><remote>service=onhostname="localhost"port=9999connection‐timeout=1000</remote></akka>

Remote config

Akka STM

Software Transactional Memory (STM)

What is STM?

STM: overview>See the memory (heap and stack) as a transactional dataset

>Similar to a databasebegin commit abort/rollback

>Transactions are retried automatically upon collision

>Rolls back the memory on abort

> Transactions can nest> Transactions compose (yipee!!)atomic{..atomic{..}}

STM: overview

>All operations in scope of a transaction:Need to be idempotentCan’t have side-effects

STM: restrictions

Akka STMis based on the ideas of

Clojure STM

2 things: 1. Managed References2. Persistent Datastructures

• Typical OO: direct access to mutable objects

Managed ReferencesTypical OO - Direct

references to Mutable Objects

• Unifies identity and value

• Anything can change at any time

• Consistency is a user problem

• Encapsulation doesn’t solve concurrency problems

• Managed Reference: separates Identity & ValueClojure - Indirect references

to Immutable Objects

"ethel"

"fred"

• Separates identity and value

• Obtaining value requires explicit dereference

• Values can never change

• Never an inconsistent value

• Encapsulation is orthogonal

Copyright Rich Hickey 2009

• Separates Identity from Value- Values are immutable- Identity (Ref) holds Values

• Change is a function• Compare-and-swap (CAS)• Abstraction of time• Can only be altered within a transaction

Managed References

valref=TransactionalState.newRef(HashTrie[String,User]())

valusers=ref.getvalnewUsers=users+//createsnewHashTrie(“bill”‐>newUser(“bill”,“secret”)

ref.swap(newUsers)

Managed References

valusersRef=TransactionalState.newRef(HashTrie[String,User]())

for(users<‐usersRef){users+(name‐>user)}

valuser=for(users<‐usersRef)yield{users(name)}

Managed References

for{users<‐usersRefuser<‐usersroles<‐rolesRefrole<‐rolesifuser.hasRole(role)}{...//dostuff}

Managed References

//wrapsaRefwithaHashTrievalusers=TransactionalState.newMap[String,User]

//wrapsaRefwithaVectorvalusers=TransactionalState.newVector[User]

Managed Referencesconvenience classes

Persistent datastructures

• Immutable• Change is a function• Old version still available after change• Very fast with performance guarantees (near constant time)

• Thread safe• Iteration safe

Bit-partitioned hash trieBit-partitioned hash tries

Structural sharing with path copying

Path Copyingint count 15

INode root

HashMapint count 16

INode root

HashMap

Approach

• Programming with values is critical

• By eschewing morphing in place, we just need to manage the succession of values (states) of an identity

• A timeline coordination problem

• Several semantics possible

• Managed references

• Variable-like cells with coordination semantics

importse.scalablesolutions.akka.collection._

valhashTrie=newHashTrie[K,V]

//API(+extendsMap)defget(key:K):Vdef+[A>:V](pair:(K,A)):HashTrie[K,A]def‐(key:K):HashTrie[K,A]defempty[A]:HashTrie[K,A]

Persistent datastructuresHashTrie

Approach

importse.scalablesolutions.akka.collection._

valvector=newVector[T]

//API(+extendsRandomAccessSeq)defapply(i:Int):Tdef+[A>:T](obj:A):Vector[A]defpop:HashTrie[K,A]//removetaildefupdate[A>:T](i:Int,obj:A):Vector[A]

Persistent datastructuresVector

Approach

• Transactional Memory- Atomic, Consistent, Isolated (ACI)- MVCC- Rolls back in memory and retries automatically on clash

• Works together with Managed References• Map, Vector and Ref abstraction

Akka STM

classUserRegistryextendsTransactor{privatelazyvalstorage=TransactionalState.newMap[String,User]

defreceive={caseNewUser(user)=>storage+(user.name‐>user)...}}

STM: declarative API

classUserRegistryextendsActor{makeTransactionRequiredprivatelazyvalstorage=TransactionalState.newMap[String,User]

defreceive={caseNewUser(user)=>storage+(user.name‐>user)...}}

STM: declarative API

@transactionrequiredclassUserRegistry{}

STM: declarative Java API

importse.scalablesolutions.akka.stm.Transaction._

atomic{..//dosomethingwithinatransaction}

atomic(maxNrOfRetries){..//dosomethingwithinatransaction}

atomicReadOnly{..//dosomethingwithinatransaction}

STM: high-order fun API

importse.scalablesolutions.akka.stm.Transaction._

atomically{..//trytodosomething}orElse{..//iftxclash;trydodosomethingelse}

STM: high-order fun API

valuserStorage=TransactionalState.newMap[String,User]

for(tx<‐Transaction()){userStorage.put(user.name,user)}

STM: monadic API

valuserStorage=TransactionalState.newMap[String,User]

valusers=for{tx<‐Transaction()name<‐userNamesifuserStorage.contains(name)}yielduserStorage.get(name)//transactional

STM: monadic API

<akka><stm>service=ondistributed=off</stm></akka>

STM: config

TransactionManagement.disableTransactions

STM: disable

Akka Serialization

SerializersScalaJSONJavaJSONProtobufSBinary

JavaWednesday, December 30, 2009

valfoo=newFoovaljson=Serializer.ScalaJSON.out(foo)valfooCopy=Serializer.ScalaJSON.in(json).asInstanceOf[Foo]

SerializersScala 2 JSON & JSON 2 Scala

importsbinary.DefaultProtocol._valusers=("user1","passwd1")::("user2","passwd2")::("user3","passwd3")::Nilvalbytes=Serializer.SBinary.out(users)

valusersCopy:List[Tuple2[String,String]]]=Serializer.SBinary.in(bytes)

SerializersScala 2 Binary & Binary 2 Scala

valpojo=ProtobufPOJO.getDefaultInstance.toBuilder.setId(1).setName("protobuf").setStatus(true).buildvalbytes=pojo.toByteArray

valpojoCopy=Serializer.Protobuf.in(bytes,classOf[ProtobufPOJO])

SerializersProtobuf

caseclassMyMessage(id:String,value:Tuple2[String,Int])extendsSerializable.ScalaJSON

valmessage=MyMessage("id",("hello",34))valjson=message.toJSON

Serializable

Akka Persistence

• Pluggable storage backend- Cassandra - MongoDB- Redis

• Map, Vector and Ref abstraction• MVCC• Works together with STM

Persistence

Akka Persistence API

//transactionalCassandra‐backedMapvalmap=CassandraStorage.newMap

//transactionalRedis‐backedVectorvalvector=RedisStorage.newVector

//transactionalMongo‐backedRefvalref=MongoStorage.newRef

<akka><storage><cassandra>hostname="127.0.0.1"port=9160storage‐format="protobuf"consistency‐level=quorum</cassandra>

<mongodb>hostname="127.0.0.1"port=27017dbname="mydb"</mongodb></storage></akka>

Persistence: config

Akka’s Cassandra API

valsessions=newCassandraSessionPool(keyspace,StackPool(SocketProvider(host,port)),Protocol.Binary,consistencyLevel)

Create a session pool

Akka Cassandra API//getacolumnvalcolumn=sessions.withSession{session=>session|(key,newColumnPath(columnFamily,superColumn,serializer.out(name))}

valvalue=if(column.isDefined)Some(serializer.in(column.get.value,None))elseNone

Automatic connection management

Akka Cassandra API//addacolumnsessions.withSession{session=>session++|(key,newColumnPath(cf,null,serializer.out(name)),serializer.out(value),System.currentTimeMillis,consistencyLevel)}

Akka REST

@Path("/count")classCounterextendsActor{privatevarcounter=0

@GET@Produces(Array("text/html"))defcount=(this!!Tick).getOrElse(<h1>Errorincounter</h1>)

defreceive={caseTick=>counter+=1reply(<h1>Tick:{counter}</h1>)}}}

RESTful actors

<akka><rest>service=onhostname=“localhost”port=9998</rest></akka>

REST: config

classBoot{objectfactoryextendsSupervisorFactory(SupervisorConfig(RestartStrategy(OneForOne,3,100),Supervise(newCounter,LifeCycle(Permanent))::Supervise(newChat,LifeCycle(Permanent))::Nil)))valsupervisor=factory.newInstancesupervisor.start}

Boot classes

<akka>boot=["sample.rest.Boot","sample.comet.Boot"]...</akka>

Boot config

Akka Comet

Comet actors• Based on Atmosphere project• Portable• Supports natively:

• Tomcat 4, 5, 6• Jetty 5, 6, 7• GlassFish 1, 2, 3• Weblogic 9.x, 10.x• Grizzly 1.9.x• JBossWeb 2.x

• Annotation based

@Path("/chat")classChatextendsActor{caseclassChat(who:String,what:String,message:String)

@Suspend//receivingendpoint@GET@Produces(Array("text/html"))defsuspend=<!‐‐suspend‐‐>

//sendingendpoint@Broadcast(Array(classOf[XSSHtmlFilter],classOf[JsonFilter]))@Consumes(Array("application/x‐www‐form‐urlencoded"))@Produces(Array("text/html"))@POSTdefpublishMessage(form:MultivaluedMap[String,String])=(this!!Chat(form.getFirst("name"),form.getFirst("action"),form.getFirst("message"))).getOrElse("Systemerror")

defreceive={caseChat(..)=>..}}

Comet actors

Akka Security

classSampleAuthenticationServiceextendsDigestAuthenticationActor{

//Useanin‐memorynonce‐mapasdefaultoverridedefmkNonceMap=newHashMap[String,Long]

//Changethistowhateveryouwantoverridedefrealm=“sample”

//Username,passwordandrolesforausernameoverridedefuserInfo(uname:String):Option[UserInfo]={...//getuserwithpasswordandrolesSome(UserInfo(uname,password,roles))}}

Security: service

classSecuredActorextendsActor{@RolesAllowed(Array(“admin”))defresetSystem=(this!!Reset).getOrElse(<error>Couldnotresetsystem</error>)

defreceive={caseReset=>...}}

Security: usage

<akka><rest>service=onhostname=“localhost”port=9998filters=[“AkkaSecurityFilterFactory”]authenticator=“SimpleAuthenticationService”</rest></akka>

Security: config

Akka Lift

classBoot{//Lift’sbootstrapclassdefboot{LiftRules.httpAuthProtectedResource.prepend{case(ParsePath("akka‐lift"::Nil,_,_,_))=>Full(AuthRole("admin"))}LiftRules.authentication=HttpBasicAuthentication("lift"){case("guest","guest",req)=>userRoles(AuthRole("admin"))}objectfactoryextendsSupervisorFactory(SupervisorConfig(RestartStrategy(OneForOne,3,100),List(Supervise(newAkkaService,LifeCycle(Permanent))))factory.newInstance.start}}

Lift integration

<web‐app><servlet><servlet‐name>AkkaServlet</servlet‐name><servlet‐class>se.ss.akka.rest.AkkaServlet</servlet‐class></servlet><servlet‐mapping><servlet‐name>AkkaServlet</servlet‐name><url‐pattern>/*</url‐pattern></servlet‐mapping></web‐app>

Lift integration

Akka HotSwap

HotSwapactor!HotSwap(Some({//newbodycasePing=>...casePong=>...}))

Akka AMQP

valproducer=AMQP.newProducer(config,hostname,port,exchangeName,serializer,None,None,//listeners100)

producer!Message(“Hithere”,routingId)

AMQP: producer

valconsumer=AMQP.newConsumer(config,hostname,port,exchangeName,ExchangeType.Direct,serializer,None,100,passive,durable,Map[String,AnyRef())

consumer!MessageConsumerListener(queueName,routingId,actor{caseMessage(payload,_,_,_,_)=>...//processmessage})

AMQP: consumer

Akka Kernel

java‐jarakka‐0.6.jar\‐Dakka.config=<path>/akka.conf

Start Kernel

exportAKKA_HOME=<pathtoakkadist>java‐jar$AKKA_HOME/dist/akka‐0.6.jar

Akka Deployment

Akka as a library

Using the Actor moduleWednesday, December 30, 2009

Akka as a library

Add STM moduleWednesday, December 30, 2009

Akka as a library

Add Persistence module as a CacheWednesday, December 30, 2009

Akka as a library

Add Persistence module as primary SoRWednesday, December 30, 2009

Akka as a library

Add REST and Comet modulesWednesday, December 30, 2009

Akka as stand-alone Kernel

<log>console=onfilename="./logs/akka.log"roll="daily"level="debug"syslog_host=".."syslog_server_name=".."</log>

Logging

Monitoring & Management Provisioningand more...

Part of commercial license

Learn morehttp://akkasource.org

Professional help

http://scalablesolutions.sejonas@jonasboner.com

Consulting Training Mentoring

EOFWednesday, December 30, 2009

akka: simpler scalability, fault-tolerance, concurrency & remoting through actors

Technology

remoting custom sinks

scalable solutions - jfokus€¦ · akka is the platform...

net remoting architecture. slide 2 cite 4420.net remoting...

simpler concurrency, scalability & fault-tolerance through...

scalable...

simpler concurrency, scalability & fault-tolerance through...

akka documentation · 2019-03-28 · akka documentation,...

group akka

distributed app remoting

gpars remoting

powershell 2.0 remoting - ravichaganti.com · a layman’s...

secretos de powershell remoting

microsoft .net remoting .net remoting · pdf file ·...

powershell remoting

akka knolx

powershell 2 remoting

scalable solutions · akka: simpler concurrency,...

net remoting

akka patterns

net remoting