Actors and Threads

[yves]

Hello!

I have a very simple producer/consumer application defined with actors.
The process is synchrouneous : produce consume produce consume etc.
There are barely 3 messages in each direction.
The producer reads from Input, does some processing of it's own and passes the result to the consumer which does the actual work.

I had coded this using a different model where the producer directly fires the code from the consumer (and handles all problems...) but for many reasons, that's not satisfying.

However, I now meet a performance problem: where the latter can handle about 200000 messages per second, this drops to 20000 with the Actor model. I believe a lot of time is lost in managing the threads, and most notably in contexte switching. I have tried varied ThreadPool models, the best I have come to is by using SingleThreadedScheduler(1,1,false,false) as scheduler for both actors (a different scheduler with one thread) ; as a matter of fact, there still is a lot of fairly useless management in this case, where each message gets assigned to it's (lone) thread etc.

I was wondering if there was any way to use the Actor mecanism as a simple message passing mecanism between two well defined thread (possibly user created.)
The idea being that each thread has a run loop, with message capture and handling done directly within the thread loop and no management overhead.

Thanks!

[√iktor Ҡlang]

Hi Yves,

since you do not state what actor library you are using or what the code looks like or what configuration you have used, it is essentially impossible to answer your question.

Akka Actors can handle around 3 million messages per second, per actor, per core on a modern 3GHz-ish CPU.

Cheers,

√

--
Viktor Klang

Akka Tech Lead

Typesafe - The software stack for applications that scale

Twitter: @viktorklang

[Alec Zorab]

Just to clarify: Victor's not actually suggesting you can get 3*10^6 * # Actors * # Cores.

More like 3 * 10^6 * # Cores split between the available actors.

Unless Akka 2.1 has _incredibly_ low actor overhead, that is!

[√iktor Ҡlang]

What I meant was that of course the number of messages per actor goes down if they are contending for the same cores. (Ignoring the effects of cache coherence overhead here)

Cheers,
V

[yves]

I am using scala core Actors.
I know that they are not as good as Akka (at least, that's what I read)
I'm nowhere near the kind of throughput you indicate, and I don't think that scala vs akka is the issue.

My code is so simple that I don't believe it's the issue either.
There has to be something I am missing soemwhere.

Still, as an academic question, I wondered also if there was any way to use Actor message mecanism to communicate and synchronize between normal threads. That would be really convenient!

Yves

[yves]

Oh and by the way, I don't know if the 3 million messages per second measured correspond to a ping-pong situation which is my case.
Has any measure been done for such a scenario ?

Yves

[Paul Keeble]

I did a test comparing Erlang to Scala's core actors some time ago.
Others have contributed comparisons in github branches with Akka and
such that show the performance differences and you can find links in
the comments explaining where to see results for that. There is code
as well if you want to compare on your local machine and the github
branches taken by others contain different types of tests and
comparisons. Blog post here:
http://www.krazykoding.com/2011/07/scala-actor-v-erlang-genserver.html.

PK

[Roland Kuhn]

Hi Yves,

On 16 okt 2012, at 21:19, yves <yves....@gmail.com> wrote:

> Oh and by the way, I don't know if the 3 million messages per second measured correspond to a ping-pong situation which is my case.
> Has any measure been done for such a scenario ?
>

Yes, the number Viktor quoted was from a ping-pong test. The scala.actors may well be your performance bottleneck, especially if you are using receive{} instead of react{} (which I assume due to your talking about threads). And even react{} is a lot slower than Akka actors due to blocking synchronization. I'd love to hear about performance results when switching implementations, if you have the time.

background: waking up a thread is horribly expensive, the only cure is to never park() it in the first place, i.e. instead of having threads receive messages dispatch your actors across threads which are shared and thus mostly busy; that is what Akka does.

Regards,

Roland Kuhn
Typesafe — The software stack for applications that scale
twitter: @rolandkuhn

[yves]

I'll see if I have some time for tests tomorrow.
As for receive:react, I thought like you and used react as first choice ; result is five times longer!
Not what I expected!

Don't forget, unlike many tests around I'm not having hundreds or thousands of actors exchanging thousands or hundreds messages. I'm having but two actors exchanging a lot of messages in a synchroneous way. Don't know if that changes something. Well, from what I read, I still should have better perfs.

Yves

[√iktor Ҡlang]

On Tue, Oct 16, 2012 at 11:45 PM, yves <yves....@gmail.com> wrote:

I'll see if I have some time for tests tomorrow.
As for receive:react, I thought like you and used react as first choice ; result is five times longer!
Not what I expected!

Don't forget, unlike many tests around I'm not having hundreds or thousands of actors exchanging thousands or hundreds messages. I'm having but two actors exchanging a lot of messages in a synchroneous way. Don't know if that changes something. Well, from what I read, I still should have better perfs.

What do you mean by "synchronous"? Show code please.

Cheers,

√

 

Yves

Le mardi 16 octobre 2012 21:41:26 UTC+2, rkuhn a écrit :

Hi Yves,

On 16 okt 2012, at 21:19, yves <yves....@gmail.com> wrote:

> Oh and by the way, I don't know if the 3 million messages per second measured correspond to a ping-pong situation which is my case.
> Has any measure been done for such a scenario ?
>
Yes, the number Viktor quoted was from a ping-pong test. The scala.actors may well be your performance bottleneck, especially if you are using receive{} instead of react{} (which I assume due to your talking about threads). And even react{} is a lot slower than Akka actors due to blocking synchronization. I'd love to hear about performance results when switching implementations, if you have the time.

background: waking up a thread is horribly expensive, the only cure is to never park() it in the first place, i.e. instead of having threads receive messages dispatch your actors across threads which are shared and thus mostly busy; that is what Akka does.

Regards,

Roland Kuhn
Typesafe — The software stack for applications that scale
twitter: @rolandkuhn

--

Viktor Klang

Akka Tech Lead

Typesafe - The software stack for applications that scale

Twitter: @viktorklang

[yves]

I'll find code later, but it's just:

producer sends data and stops
consumer processes data and sends reply to producer
producer sends data and stops
consumer processes data and sends reply to producer
producer sends data and stops
consumer processes data and sends reply to producer
producer sends data and stops
consumer processes data and sends reply to producer
....

This terminates when either the producer has no more data or when the consumer tells the producer he is finished.

Yves

[pagoda_5b]

Hi Yves,

this actually only shows that communication between the actors is "sequential" in its logic.

Usually "synchronous" in this context is used to signify that an actor sends a message and blocks the thread waiting for a response before he proceeds with other tasks.

Your "sequential communication pattern" (i.e. ping-pong) doesn't need to be "synchronous", so maybe it will be more helpful if you can provide some code.

bye, Ivano

[yves]

object schx1 extends ForkJoinScheduler(1,1,false,false) {
  start
}
object schx2 extends ForkJoinScheduler(1,1,false,false) {
  start
}

final class Out(val userf: ()=>In) extends Actor {
  override val scheduler = schx1
  trapExit = true
  def act() = {
    val user = userf()
    println(user)
    link(user)
    var i=0
    loop {
      i+=1
      if (i<100*Main.m)
        user ! "info "+i
      else
        exit
      receive {
        case cont         => if (i%Main.m==0) println(s"out = ${Thread.currentThread.getName}")
        case abt          => i+=10
        case Exit(user,_) => exit
      }
    }
  }
  override def toString = "out"
}

final class In(val outf: ()=>Out) extends Actor {
  override val scheduler = schx2
  trapExit = true
  def act() = {
    var t = System.currentTimeMillis
    val out = outf()
    println(out)
    link(out)
    var i=0
    loop {
      receive {
        case s:String    => i+=1; if (i%Main.m==0) println(s"in = ${Thread.currentThread.getName} => $s"); reply((if (i%(10*Main.m-1)==0) abt else cont))
        case term        => time; exit
        case Exit(out,_) => time; exit
      }
    }   
    def time = println(s"time ${System.currentTimeMillis-t}")
  }
  override def toString = "user"
}


object Main {
  val sch1=schx1
  val sch2=schx2

  val m=10000
  def println(s: =>String) = ()
 
  def start():Unit = {
    var out:Out = null
    val in = new In(()=>out)
    out = new Out(()=>in)
    out.start
    in.start
  }
 
  def main(args:Array[String]):Unit = {
    start()
  }
}

Out send data, In processes it. Out only sends data after receiving an acknowledgement from In (because, actually in the real code, Out might have to change behaviour depending on the answer from In)
Don't be fooled by the

override val scheduler = new SingleThreadedScheduler

which are there only for additional experiements. The performance is bad with or without (the current settings gives the best performances, but it ties with other settings, such as a single scheduler limited to 2 threads)

Yves

[yves]

missing in the code above

object cont
object abt
object term


which are basic command messages.

[yves]

Some additional info:

~15s using receive and 1000000 messages (i.e. 70000 messages per second)
~10s using react and 1000000 messages (i.e. 100000 messages per second)

Intel core i5-2400 @3.10 GHz

I'm far from 3000000 messages per second! (factor 30 in the best case)

[√iktor Ҡlang]

The 3million msg/s was for Akka Actors, not Scala Actors.

Secondly, you have 2 actors and 1 thread, which means that your actors only have half a core each, so with Akka I'd expect around 1.5million per second.

However, you are using IO, and that can be dog slow.

Cheers,

√

[yves]

No.
In that exemple, I have two trheads (because I have two distinct thread pools.)
I get similar results with one threadpool limited to two threads.
The performance indeed decreases if I put only one common thread in the thread pool.

I also forgot that I'am having ping-pong, i.e. two messages for one "usefull" message.
And possibly the threads compete for the same core (due to OS choice) ; that would still be a 7.5 ratio to the expected throughput (is there anyway for a thread to tell which core it is assigned to ?). I don't believe that ratio is covered by the scala-akka difference.
I also thought about the IO. Here I now have 1 IO for 10000 messages. I did test with no IO at all and very little gain.

I'll download akka and test.

Yves

[√iktor Ҡlang]

Great, let us know what happens.

Cheers,

√

 

Yves

[yves]

I cannot create the Actor.

Seemingly I must use the system.actorOf method, but I am unable to locate it.
Is there a specific jar other than akka-actor (I'm using 2.10M7)

Yves

[yves]

Error      while compiling: src\Out.scala         during phase: typer      library version: version 2.10.0-M7     compiler version: version 2.10.0-M7   reconstructed args: -encoding Cp1252 -verbose -Yself-in-annots -Xlog-implicits -Xfuture -Xpluginsdir \eclipse\configuration\org.eclipse.osgi\bundles\404\1\.cp\lib -deprecation -Yoverride-vars -Yoverride-objects -Xexperimental -bootclasspath \jre7\lib\resources.jar;\jre7\lib\rt.jar;\jre7\lib\sunrsasign.jar;\jre7\lib\jsse.jar;\jre7\lib\jce.jar;\jre7\lib\charsets.jar;\jre7\lib\jfr.jar;\jre7\classes;\eclipse\configuration\org.eclipse.osgi\bundles\405\1\.cp\lib\scala-library.jar -explaintypes -Ywarn-dead-code -classpath \jre7\lib\resources.jar;\jre7\lib\rt.jar;\jre7\lib\jsse.jar;\jre7\lib\jce.jar;\jre7\lib\charsets.jar;\jre7\lib\jfr.jar;\jre7\lib\ext\dnsns.jar;\jre7\lib\ext\localedata.jar;\jre7\lib\ext\sunec.jar;\jre7\lib\ext\sunjce_provider.jar;\jre7\lib\ext\sunmscapi.jar;\jre7\lib\ext\sunpkcs11.jar;\jre7\lib\ext\zipfs.jar;\wk\XX\bin;\eclipse\configuration\org.eclipse.osgi\bundles\405\1\.cp\lib\scala-swing.jar;\eclipse\configuration\org.eclipse.osgi\bundles\405\1\.cp\lib\scala-actors.jar;\eclipse\configuration\org.eclipse.osgi\bundles\404\1\.cp\lib\scala-reflect.jar;\wk\XX\akka-actor_2.10.0-M7-2.1-M2.jar -Yinfer-argument-types -target:jvm-1.7    last tree to typer: Select(Select(Ident(akka), actor), ActorSystem)               symbol: object ActorSystem in package actor (flags: <module> <triedcooking>)    symbol definition: object ActorSystem                  tpe: akka.actor.ActorSystem.type        symbol owners: object ActorSystem -> package actor       context owners: value system -> method start -> object Main -> package <empty>  == Enclosing template or block ==  Block(    // 3 statements    ValDef( // val system: <?>      0      "system"      <tpt>      Apply(        "ActorSystem"        "PingPong"      )    )    ValDef( // var out: <?>      <mutable>      "out"      "Out"      null    )    ValDef( // val in: <?>      0      "in"      <tpt>      Apply(        "system"."actorOf"        Apply(          "Props"          Apply(            new In."<init>"            Function(              ()              "out"."self"            )          )        )      )    )    Assign(      "out"      Apply(        "system"."actorOf"        Apply(          "Props"          Apply(            new Out."<init>"            Function(              ()              "in"."self"            )          )        )      )    )  )  == Expanded type of tree ==  SingleType(   pre = SingleType(     pre = SingleType(pre = ThisType(package <root>), package akka)     package actor   )   object ActorSystem )  uncaught exception during compilation: java.io.PrintWriter@528aea

[yves]

Last message means I seemingly won't be able to test.
The compiler crashes somewhere and I'm at a loss to guess why.

Yves

[yves]

More to the point, I also have the following messages:

Presentation compiler crashed while type checking this file: java.lang.AssertionError: assertion failed: value config
Error in Scala compiler: assertion failed: value config
bad symbolic reference to value typesafe in package com (a classfile may be missing)

No idea where the problem lies. Is it in akka ???

[pagoda_5b]

Do you have the typesafe config library in the classpath? https://github.com/typesafehub/config

It's a dependency requirement, but If I'm correct it's been removed from the main akka distribution

[yves]

Ah ah!

That must be it.

But I cannot go further : typesafe requires install rights which I don't have and won't be given.
That means I will not be able to test akka.

Too bad.

Yves

[√iktor Ҡlang]

On Wed, Oct 17, 2012 at 4:56 PM, yves <yves....@gmail.com> wrote:

Ah ah!

That must be it.

But I cannot go further : typesafe requires install rights which I don't have and won't be given.

I have no idea what that means. Akka is just a library, just add it to your project.

http://doc.akka.io/docs/akka/2.0.3/intro/getting-started.html#Using_Akka_with_Maven

 

That means I will not be able to test akka.

I have to recommend reading the documentation (which is always a good thing in case you want to understand how something works): http://doc.akka.io/docs/akka/2.0.3/scala/actors.html

 

Too bad.

Yes, it is indeed bad not to read documentation before giving up on something.

I hope the links above clears things up,

Cheers,

√

 

Yves

[yves]

I've read the documentation.

The problem I have looks absolutely clear; my project would compile but for the fact that it seems to require a third party library (typesafe) which I can't install. If you read the error messages I have, you'll agree with me.

Why Akka requires it, I don't know. Maybe a forgotten dependency of some sort. I shouldn't need it, because typesafe is seemingly used to brideg Akka with Play (whatever that is...)

Maybe that's an artefact due to the version I used (scala 2.10M7, which isn't certainly the wistest to do...)
Anyway, it's absolutely clear that this is the only problem. The actor library is on may path, and I got normal compile errors till I should have none. And then, I get that strange messages.

Yves

[Derek Williams]

It's not a 3rd party library, Typesafe also develops Akka.

Typesafe Config is a part of Akka 2.0 that was spun off into it's own project. It is a dependency of Akka 2.1. It requires "installing" just as much as Akka does.

All dependencies for Akka 2.1 M2 are listed in it's pom.xml: 

http://repo.typesafe.com/typesafe/releases/com/typesafe/akka/akka-actor_2.10.0-M7/2.1-M2/akka-actor_2.10.0-M7-2.1-M2.pom

If you aren't using some kind of dependency management, you can get the required jar here:

http://repo.typesafe.com/typesafe/releases/com/typesafe/config/0.5.0/config-0.5.0.jar

Not having a required jar on the classpath at runtime will give you strange errors. That has nothing to do with Akka.

I hope that helps.

--

Derek Williams

[√iktor Ҡlang]

On Wed, Oct 17, 2012 at 9:54 PM, yves <yves....@gmail.com> wrote:

I've read the documentation.

The problem I have looks absolutely clear; my project would compile but for the fact that it seems to require a third party library (typesafe) which I can't install. If you read the error messages I have, you'll agree with me.

Akka is as much third party as Typesafe _Config_ which is a library just like Akka.

 

Why Akka requires it, I don't know. Maybe a forgotten dependency of some sort. I shouldn't need it, because typesafe is seemingly used to brideg Akka with Play (whatever that is...)

Why it requires it? https://oss.sonatype.org/content/groups/public/com/typesafe/akka/akka-actor_2.10.0-M7/2.1-M2/akka-actor_2.10.0-M7-2.1-M2.pom 

Maybe that's an artefact due to the version I used (scala 2.10M7, which isn't certainly the wistest to do...)

No, M7 has nothing to do with the problem here.

 

Anyway, it's absolutely clear that this is the only problem. The actor library is on may path, and I got normal compile errors till I should have none. And then, I get that strange messages.

The only problem here is that you haven't put the required dependencies on the classpath,

put the required dependency on the classpath and everything will be just fine.

Cheers,

√

 

Yves

[yves]

Thanks!

That jar link was what I was missing!
Due to security restrictions, I cannot install on the machine where I develop.
But I can of course download the jar.

I'll try tomorrow.

Yves

[Derek Williams]

On Wed, Oct 17, 2012 at 3:19 AM, yves <yves....@gmail.com> wrote:

Some additional info:

~15s using receive and 1000000 messages (i.e. 70000 messages per second)
~10s using react and 1000000 messages (i.e. 100000 messages per second)

I gave your code a run on my machine (i7-2600K @ 4.5)

each run used 1000000 messages

2 schedulers, 1 thread each

receive: 8.5s

react: 8.5s (as expected, since still 1 dedicated thread each)

1 scheduler, 2 threads

receive: 8.5s

react: 3.5s

default scheduler

receive: 10s

react: 5s

1 scheduler, 8 threads (I usually use between 4 and 8)

receive: 8.5s

react: 3.5s

Intel core i5-2400 @3.10 GHz

I'm far from 3000000 messages per second! (factor 30 in the best case)

To be fair that number was quoted for Akka, which has optimizations that will perform ops in batches (among many other optimizations).

From my own benchmarking over a year ago with Akka, after tuning Akka and the jvm, I was able to get 50,000,000 ops/sec. Not doing any useful work of course, but it does give an idea for upper bounds. I also have a (very neglected) redis client built using Akka, which is capable of reaching 1,000,000 ops/sec. This is of course including the IO with a locally running redis server, so doing a measurable amount of work. So it should be very possible to reach your goal.

I haven't done nearly as much tuning and testing of the scala core actors though, so not sure what a reasonable upper bounds is for performance.

--

Derek Williams

[yves]

I ran the Akka version for my code.

I could not get the postStop method to run so I used a crude method to estimate time: a watch...
Performances are not any better than scala Actors at this point, with about 10s for the run.
Here is the code:


import akka.actor.{Actor,ActorRef,Terminated,ActorSystem,Props}

object cont
object abt
object term

final class Out(val userf:()=>ActorRef) extends Actor {
  var user:ActorRef = null
  var i=0
  override def preStart = { user = userf(); println(user); send }
  def send = { i+=1; user ! "info "+i; if (i==100*Main.m) sys.exit }
  def receive() = {
    case `cont`        => if (i%Main.m==0) println(s"out = ${Thread.currentThread.getName}"); send
    case `abt`         => i+=10; send
    case Terminated(_) => context.stop(self)

  }
  override def toString = "out"
}

final class User(val outf:()=>ActorRef) extends Actor {
  var out:ActorRef = null
  var i = 0
  var t:Long = 0
  override def preStart = { out = outf(); println(out); t = System.currentTimeMillis }
  def receive() = {
    case s:String      => i+=1; if (i%Main.m==0) println(s"in = ${Thread.currentThread.getName} => $s"); out ! (if (i%(10*Main.m-1)==0) abt else cont)
    case `term`        => context.stop(self)
    case Terminated(_) => context.stop(self)
  }
  override def postStop = { println(s"time ${System.currentTimeMillis-t}  $i messages processed  ${i/(System.currentTimeMillis-t)/1000} messages/second") }

  override def toString = "user"
}


object Main {

  val m=10000
  def println(s: =>String) = ()
 
  def start():Unit = {

    val sys = ActorSystem.create
    var out:ActorRef = null
    var user=sys.actorOf(Props(()=>{new User(()=>out)},"user"))
    out=sys.actorOf(Props(()=>{new Out(()=>user)},"out"))

[yves]

Ok, my send method was erroneous and shut down the JVM...

  def send = { i+=1; user ! "info "+i; if (i==100*Main.m) { user ! term; context.stop(self) } }

But now I have pending threads that don't seem to gracefully terminate.
My throughput is about 2*110000 messages/second

Yves

[yves]

last Akka version.

import akka.actor.{Actor,ActorRef,Terminated,ActorSystem,Props}

object cont
object abt
object term

final class Out(val user:ActorRef) extends Actor {
  var i=0
  override def preStart = { println(user); send }

  def send = { i+=1; user ! "info "+i; if (i==100*Main.m) { user ! term; context.stop(self) } }

  def receive() = {
    case `cont`        => if (i%Main.m==0) println(s"out = ${Thread.currentThread.getName}"); send
    case `abt`         => i+=10; send
    case Terminated(_) => context.stop(self)
  }
  override def toString = "out"
}

final class User(val out:ActorRef) extends Actor {

  var i = 0
  var t:Long = 0

  override def preStart = { println(out); t = System.currentTimeMillis }
  def receive() = {
    case s:String      => i+=1; if (i%Main.m==0) println(s"user = ${Thread.currentThread.getName} => $s"); out ! (if (i%(10*Main.m-1)==0) abt else cont)

    case `term`        => context.stop(self)
    case Terminated(_) => context.stop(self)
  }

  override def postStop = { println(s"time ${System.currentTimeMillis-t}  $i messages processed  ${i/((System.currentTimeMillis-t)/1000)} messages/second") }

  override def toString = "user"
}


object Main {
  val m=10000
 

  def start():Unit = {
    val sys = ActorSystem.create
    var out:ActorRef = null

    var user=sys.actorOf(Props(()=>new User(out)))
    out=sys.actorOf(Props(()=>new Out(user)))

  }
 
  def main(args:Array[String]):Unit = {
    start()
  }
}

I still have a shutdown problem.
A dramatic performance increase stems from not calling sys.actorOf(Props(()=>new User(out),"out")), and simply removing the name sys.actorOf(Props(()=>new User(out)))

Throughput now is about 2*330000 messages per second, which is a threefold increase over the previous test!
Such a small change for such a big gain!!!
If I can still gain a x2 (why not), I'll be close to my objective!

Yves

[Johannes Rudolph]

Here's a complete standalone working sbt project with that code so
that noone checking this has to do the same work:

https://gist.github.com/3910849

--
Johannes

-----------------------------------------------
Johannes Rudolph
http://virtual-void.net

[Johannes Rudolph]

On Thu, Oct 18, 2012 at 10:47 AM, yves <yves....@gmail.com> wrote:
> But now I have pending threads that don't seem to gracefully terminate.
> My throughput is about 2*110000 messages/second

You have to call

context.system.shutdown()

in the end. This is necessary since Actors are autonomous and so the
JVM can't simply know when the program has finished.

[√iktor Ҡlang]

On Thu, Oct 18, 2012 at 11:01 AM, yves <yves....@gmail.com> wrote:

This doesn't make any sense.

val user = sys.actorOf(Props[User])

val out = sys.actorOf(Props[Out])

user ! SetOut(out)

out ! SetUser(user)

Or just use the sender...

Cheers,

√

 

  }
 
  def main(args:Array[String]):Unit = {
    start()
  }
}

I still have a shutdown problem.
A dramatic performance increase stems from not calling sys.actorOf(Props(()=>new User(out),"out")), and simply removing the name sys.actorOf(Props(()=>new User(out)))

Throughput now is about 2*330000 messages per second, which is a threefold increase over the previous test!
Such a small change for such a big gain!!!
If I can still gain a x2 (why not), I'll be close to my objective!

Yves

[Roland Kuhn]

Hi Yves,

trying your code I get 500,000 msgs/sec (as per your printout; MacBook Pro i7 quad-core 2.2GHz), no matter whether the actors are explicitly named or not. Please find additional comments inline.

18 okt 2012 kl. 11:01 skrev yves:

last Akka version.

import akka.actor.{Actor,ActorRef,Terminated,ActorSystem,Props}

object cont
object abt
object term


final class Out(val user:ActorRef) extends Actor {
  var i=0
  override def preStart = { println(user); send }
  def send = { i+=1; user ! "info "+i; if (i==100*Main.m) { user ! term; context.stop(self) } }
  def receive() = {
    case `cont`        => if (i%Main.m==0) println(s"out = ${Thread.currentThread.getName}"); send
    case `abt`         => i+=10; send
    case Terminated(_) => context.stop(self)

Since you are not using context.watch() anywhere, you won’t get Terminated() messages either.

  }
  override def toString = "out"

Actors are not printed (in the toString sense) anywhere, and passing `this` around is more than just strongly discouraged (because it breaks the actor model encapsulation; the old scala actors had the severe design flaw of not using ActorRefs). By this I mean to say that overriding toString on an Actor has no effect.

}

final class User(val out:ActorRef) extends Actor {
  var i = 0
  var t:Long = 0
  override def preStart = { println(out); t = System.currentTimeMillis }
  def receive() = {
    case s:String      => i+=1; if (i%Main.m==0) println(s"user = ${Thread.currentThread.getName} => $s"); out ! (if (i%(10*Main.m-1)==0) abt else cont)
    case `term`        => context.stop(self)
    case Terminated(_) => context.stop(self)
  }
  override def postStop = { println(s"time ${System.currentTimeMillis-t}  $i messages processed  ${i/((System.currentTimeMillis-t)/1000)} messages/second") }
  override def toString = "user"
}


object Main {
  val m=10000
 
  def start():Unit = {
    val sys = ActorSystem.create
    var out:ActorRef = null
    var user=sys.actorOf(Props(()=>new User(out)))

The actor is instantiated on a different thread, quite possibly between these two lines. I’m assuming that you do not run on an antique (i.e. single-core) system, which means that “out” may well be null. This will lead to the failure and restart of the User actor, which will break your program since the Out actor will never get the “cont” reply.

In the actor model, every communication with actors is done using messages; this also applies to creation (i.e. sending the Actor factory to the new actor “holder”). Closing over mutable state and/or sending it in messages breaks the actor model, no matter which implementation.

In an ideal world (or Scala 2.11?) your code would be rejected by the compiler because the closure you pass as actor factory is not “static”.

    out=sys.actorOf(Props(()=>new Out(user)))
  }
 
  def main(args:Array[String]):Unit = {
    start()
  }
}

I still have a shutdown problem.

Is the shutdown problem that the application sometimes does not terminate? That should be explained above.

A dramatic performance increase stems from not calling sys.actorOf(Props(()=>new User(out),"out")), and simply removing the name sys.actorOf(Props(()=>new User(out)))

I highly doubt that, since the only change is that a name will be generated (quite like a default argument). Trying it, I could not reproduce your observation.

Throughput now is about 2*330000 messages per second, which is a threefold increase over the previous test!
Such a small change for such a big gain!!!
If I can still gain a x2 (why not), I'll be close to my objective!

You are doing worst-case ping-pong with only two actors (worst case because max 1 message can be in flight). You can gain a lot by allowing things to happen in parallel. One actor is an entity which can be executing at most once at any given point in time. Having two actors for which only at most one may be active at any given time does not sound very useful to me a priori; I’d be interested in hearing more about the use case.

Regards,

Roland Kuhn
Typesafe – The software stack for applications that scale.
twitter: @rolandkuhn

[yves]

This is a pretty streamline version of my test.

import akka.actor.{Actor,ScalaActorRef,Terminated,ActorSystem,Props}

final object cont
final object abt
final object term
final case class  Start(who:ScalaActorRef)


final class Out extends Actor {
  private final var user:ScalaActorRef = null
  private final var i=0
  private final def send = { i+=1; user ! "info "+i; if (i>=100*Main.m) { user ! term; context.stop(self) } }
  final def receive() = {

    case `cont`        => if (i%Main.m==0) println(s"out = ${Thread.currentThread.getName}"); send
    case `abt`         => i+=10; send

    case Start(who)    => user=who; println(user); send
    case Terminated(_) => context.stop(self)
  }
}

final class User extends Actor {
  private final var out:ScalaActorRef = null
  private final var i = 0
  private final var t0:Long = 0
  private final var t:Long = 0
  private final def end = { t=System.currentTimeMillis-t0; println(s"time $t  $i messages processed  ${i/t} messages/millisecond"); context.system.shutdown }
  final def receive() = {

    case s:String      => i+=1; if (i%Main.m==0) println(s"user = ${Thread.currentThread.getName} => $s"); out ! (if (i%(10*Main.m-1)==0) abt else cont)

    case `term`        => context.stop(self); end
    case Start(who)    => out=who; who ! Start(self); println(out); t0=System.currentTimeMillis
    case Terminated(_) => context.stop(self); end

  }
}


object Main {
  val m=10000
 
  def start():Unit = {
    val sys = ActorSystem.create

    var user=sys.actorOf(Props[User])
    var out=sys.actorOf(Props[Out])
    user ! Start(out)

  }
 
  def main(args:Array[String]):Unit = start()
}

I'm still in the 2*300000 messages per second, which is less than I'd expect from such a simple program.

@Viktor: used your remark to improve actors classes ; still I felt the tone a bit rude.

[yves]

Oh, I forgot: why does removing the actor name increases so much the perfs ?

[√iktor Ҡlang]

That was completely unintentional, what was rude about it?

Cheers,

√

--

Viktor Klang

Akka Tech Lead

Typesafe - The software stack for applications that scale

Twitter: @viktorklang

[Alec Zorab]

Without names:

time 21976.74ms  9999910 messages processed  455022.49 messages/second

[success] Total time: 25 s, completed Oct 18, 2012 12:08:48 PM

with names:

time 20836.38ms  9999910 messages processed  479925.54 messages/second

[success] Total time: 23 s, completed Oct 18, 2012 12:12:33 PM

So essentially zero difference.

[yves]

Hi Roland,

I appreciate your very explicit explanations.
I'm having 300000, but using an i5'2400 @ 3.1GHz. That might possibly make the difference.

I'm not new in multithreading, but have little background on Actors, except that they look very easy to use. I'll remember all that you write ; tremendously usefull stuff there.

In my new code, I don't see the name making any difference. I wonder what was the artefact that caused this problem in the older versions of my code. Anyway, as you write, it was not correct according to the Actors style, and this might have had some difficult to understand consequences.

I also know that this test is not nice, with only one message flying at a time! But in real life (my life anyway), many things happen that way. Still, it's useful to put the producer on one Actor and the Consumer on another rather than having the producer directly invoke the consumer (when that is feasible, which is not always the case.) ; that's why I wrote this test, to understand how much CPU I was going to lose by using this paradigm.

Thanks!

[yves]

@Viktor: That was completely unintentional, what was rude about it?

Well, 'it makes no sense' made me feel like fool. :)

[√iktor Ҡlang]

On Thu, Oct 18, 2012 at 1:21 PM, yves <yves....@gmail.com> wrote:

@Viktor: That was completely unintentional, what was rude about it?

Well, 'it makes no sense' made me feel like fool. :)

Hey, I could be the fool for not being able to make sense out of it! :-)

--

Viktor Klang

Akka Tech Lead

Typesafe - The software stack for applications that scale

Twitter: @viktorklang

[yves]

Just a clarification: is it correct that akka actors are going to become scala official actors for 2.10 ?
If so, is it smart to move to akka now ?
How different would the final scala actors be from the current akka actors ?

Thanks!

[√iktor Ҡlang]

Scala Actors are being deprecated for 2.10 and Akka Actors is moving into the Scala Distribution in 2.10

Cheers,

√

 

Thanks!

[Roland Kuhn]

Hi Yves,

18 okt 2012 kl. 13:17 skrev yves:

Hi Roland,

I appreciate your very explicit explanations.
I'm having 300000, but using an i5'2400 @ 3.1GHz. That might possibly make the difference.

Playing with your code a little more I started to wonder why my CPU fans were coming on during the test: ping-pong between two actors which exclude each other from continuously running should not eat 100% of all my eight (hyper-threaded) cores! This sent me onto an interesting journey, where the role of the magic carpet is played by a file named application.conf:

dispatcher {

  fork-join-executor {

    parallelism-max = X

  }

}

In order to use this, I changed actor creation to use Props[whatever].withDispatcher("dispatcher").

Normally I would get 24 threads (default factor is 3, times number of cores), so I lowered that to 4, resulting in only four cores utilized, and (lo-and-behold) reduced throughput (roughly half). This is unexpected. But I was mystified by the large fraction of “system” time (and only very little “user”) time; searching for system calls being made turned up empty (prime candidate would have been querying some clock at a high rate). Then I tried

dispatcher {

  type = PinnedDispatcher

}

which gives dedicated threads to each of the two actors. Throughput reduced to 100 msgs/ms, nearly all time spent in “system”, two cores at 100% (when you would expect 2 cores at 50% each).

Now comes the important step: going back to ForkJoinPool, but with X=1, throughput increased manifold: 1784 msgs/ms (i.e. 3.6 million Akka messages per second, as we would expect).

The lesson here is the following: parking and unparking threads is EXTREMELY expensive, and your use-case does this for every single message. The reason is that one actor is still running while waking up the other actor (it will go to sleep a few nanos later). If you have many of these pairs running, so that all threads are continuously busy, you’ll see very nice throughput numbers.

I'm not new in multithreading, but have little background on Actors, except that they look very easy to use. I'll remember all that you write ; tremendously usefull stuff there.

In my new code, I don't see the name making any difference. I wonder what was the artefact that caused this problem in the older versions of my code. Anyway, as you write, it was not correct according to the Actors style, and this might have had some difficult to understand consequences.

I also know that this test is not nice, with only one message flying at a time! But in real life (my life anyway), many things happen that way. Still, it's useful to put the producer on one Actor and the Consumer on another rather than having the producer directly invoke the consumer (when that is feasible, which is not always the case.) ; that's why I wrote this test, to understand how much CPU I was going to lose by using this paradigm.

Separating code into multiple actors can be very useful for error handling (which you didn’t show, hence my question) or logical structure, I agree. And the overhead should in principle be small, unless you happen to hit a very bad corner case of thread/CPU/OS interaction.

Regards,

Roland

[yves]

@Viktor Hey, I could be the fool for not being able to make sense out of it! :-)

Anyway, I was the fool there, using Actors but not really having a clear idea of what is allowed and what is not...
My code reflected my innocence.

[yves]

Hi Roland!

I love you answer!
I had tried to play with configuration, but I'm so new that I really got nowhere, except lower my throughput (tried the PinnedDispacther for exemple.)

Coming from the classic multithread paradigm, I knew that I needed two standing threads (no thread parking because I felt this was really expensive), each speaking to the other, but did not know if that was feasible or how to get there! My best guess was PinnedDispatcher (which looked like it assigned one thread to an actor), but it failed miserably.

I'd have been unable to find that parallelism-max=1, where I now get up to about 1200000 messages per second on my system!
My target thus is reached, and I know I can safely use this for my application.

This was a very interesting journey!

I'll have to do some reading to understand what's up behind all these magic config values...

Because maybe I think I understand what is now happening, but more likely my guess is still off target.

Thanks!
Yves

[Alec Zorab]

Looking back at the original question, I"m not sure why the consumer and the producer chat to one another so much - depending on the exact scenario, I might either just send ALL THE THINGS to the consumer as fast as possible and let them sit in the mailbox (which will then take advantage of batching and save you some overhead) or I might have the consumer ask for a sized batch of stuff.

Equally, I might do a little more parallelism in the pipeline, if the problem is amenable. For example, if the preprocessing done by the producer can be parallelised

Input -> n*Preprocessors -> Gatherer (that puts things back into order) -> Consumer (possibly multiple of these too?)

Anyway, not much that can be said without knowing the actual task, but food for thought perhaps :)

[yves]

The problem is that I have on one side a parser (producer) and the other a consumer.
The parser should in most of the time ignore the tokens (i.e. pass in a very fast mode where it doesn't even form them!)
But it doesn't know what is useful for the consumer.

Hence it has to receive commands back such as: nextToken or skipTo

Which is why at the moment I require a one to one dialog.

I may later open up the possibility of having a window where more Tokens are sent, but I've yet to study whether this is really doable, and whether it's really usefull (as far as perfs are concerned!)

Yves

Le jeudi 18 octobre 2012 14:23:57 UTC+2, Alec Zorab a écrit :

Looking back at the original question, I"m not sure why the consumer and the producer chat to one another so much - depending on the exact scenario, I might either just send ALL THE THINGS to the consumer as fast as possible and let them sit in the mailbox (which will then take advantage of batching and save you some overhead) or I might have the consumer ask for a sized batch of stuff.

Equally, I might do a little more parallelism in the pipeline, if the problem is amenable. For example, if the preprocessing done by the producer can be parallelised

Input -> n*Preprocessors -> Gatherer (that puts things back into order) -> Consumer (possibly multiple of these too?)

Anyway, not much that can be said without knowing the actual task, but food for thought perhaps :)

[Alec Zorab]

If you've got forced one to one dialog like that, you're probably just paying for unnecessary overhead vs just calling some methods and doing it all on the stack. What's the actor model actually bringing to the table for you in this case?

[yves]

I know I'm paying an overhead. That's why I needed to bench before going further!

Now, what do I get:
* decoupling between producer and consumer : cleaner code
* multithreading : the producer is on one core, the consumer on the other => probably overall (much) better perfs
* consumer model : driven by its own requirements rather than those of the parser (now pull parser instead of previously push parser) => more natural to code, easier to maintain

Furthermore, this lets me hone my skills in a field where I'm still a child.

Yves

[Roland Kuhn]

The actor model allows the complete decoupling between message passing and error handling; these two are tightly woven together in traditional OO. When you call a method, you get the return value but you must also handle all possible errors. When you send a message to an actor, you only get the response back, while all failures are handled by the actor’s supervisor.

Another nice thing is the improved encapsulation: in “normal” OO you’ll typically see too many things exposed on an object, and if not you can still go in using reflection. Akka actors are completely encapsulated (unless you cheat by sending around “this” references), nobody messes with the state but the actors themselves.

It took me some time, but I’ve come to realize that the actor model is by no means only about concurrency or parallelism (although it is extremely well suited for those, of course).

Regards,

Roland

[Alec Zorab]

So the decoupling thing isn't really anything to do with actors because you could just do the same thing with plain old classes. Likewise, your consumer model isn't directly related to actors either.

You're also not getting any real benefit from multithreading here - you've described an inherently sequential system, so I'm not sure you get any upside in this particular case.

Roland's argument about error handling is a very good one, as is yours about learning through doing.
It's worthwhile understanding what you are and aren't getting from switching to actors though.

[pagoda_5b]

Maybe you could look at Iteratees?

http://jsuereth.com/scala/2012/02/29/iteratees.html

http://apocalisp.wordpress.com/2010/10/17/scalaz-tutorial-enumeration-based-io-with-iteratees/

http://www.playframework.org/documentation/2.0/Iteratees

The pattern is essentially that of a producer and a consumer working somewhat independently from each other...

On Thursday, October 18, 2012 2:54:57 PM UTC+2, yves wrote:

The problem is that I have on one side a parser (producer) and the other a consumer.
The parser should in most of the time ignore the tokens (i.e. pass in a very fast mode where it doesn't even form them!)
But it doesn't know what is useful for the consumer.

bye,

Ivano 

[yves]

I cannot really.

In the current implementation, the parser produces tokens and calls the producers methods.
But I'm happy only if I can run the producer and ask the parser the next token: however the parser doesn't work this way!
I don't see any way with standard OO, except parse all, store token into a buffer, then process the buffer; but this defeats many other things (memory requirement, needless parsing etc.); At this point I miss working ideas in this direction.

With Actors it's different: in the current implementation, the consumer can almost instantly tell "OK" (or skip), then proceed on to it's own processing. So, the parser produces a Tokenand awaits for the consumer. The consumer asks for the Token and and replies. At this point both can work in parallel; the parser prepares a new Token (skipping useless stuff if necessery) while the consumer does it's own work.

Now you see that I have decoupling, logic inversion (the consumer controls the flow rather than the other way around) and mutlithreading benefits are real.

In my case, error handling is not the issue (if either fails, both fail), but it's still an interesting point.

Yves

[yves]

I've had a look, but I fail to see how to produce the inversion of control I'm looking for.

[yves]

One last question if there is still anybody reading here!

With scala Actors, I can force an actor to wait for an answer (i.e. block) then process that answer, by opening a new receive block.

I fail to see how I can duplicate this behaviour with akka. I have seen how I can enter into another processing mode (become), but not how I can block for an answer.

Thanks!
Yves

[Roland Kuhn]

Hi Yves,

The short answer is “you don’t”; I gave the long answer earlier: blocking is BAD, don’t do it. What good would it do to block a thread instead of having it run other actors while you wait for your answer? Actors are event-driven, please forget thinking in threads.

Regards,

[yves]

I can understand the logic behind that.
But how do I do the following (which requires blocking unless I'm mistaken) ?

I have an imperative loop that produces data.
I have a consumer that wants to control the flow of data.

My idea: have the producer in an actor (but maybe standard thread is better ???); Produce one result. wait till consumer comes and take it. Produce next token and block. Repeat till end...

I'd prefer to use Actors because that gives me a high level API. Do you think it's a mistake ?

Yves

[Roland Kuhn]

AFAICT this was what we were discussing all the time on this thread, was it not? I guess the problem now lies with “I have an imperative loop”, but that is easy to fix, it is more or less just replacing “while” with “case” (in def receive). Produce your data one item at a time, keep processing state in the actor, let production be driven by getting messages from the consumer.

Regards,

Roland

[Jonas Bonér]

Read the free chapter of Derek's book. He discusses this there.

--
Jonas Bonér
CTO
Typesafe - Enterprise-Grade Scala from the Experts
Phone: +46 733 777 123
Twitter: @jboner

[yves]

@Roland: I have no mean to change the code from the producer. It's how it is and it expects an specific interface which it works with. That of course doesn't suit me.

@Jonas: I'd be interested. However, your reference is sketchy. Derek yields an awful lot of results on google!

[Johannes Rudolph]

Hey Yves,

see here:

http://www.artima.com/shop/akka_concurrency

Johannes

--
Johannes

-----------------------------------------------
Johannes Rudolph
http://virtual-void.net

[yves]

Thanks! going to read this!

[pagoda_5b]

Excerpt from the quoted articles:

"Iteratees are an immutable implementation of a Consumer/Producer pattern. A Consumer takes input and produces an output."

They are asynchronous by definition, and the Consumer can decide to ask for a produced element on demand, and what to do next.

Isn't this exactly what you're trying to achieve?

Ivano