Introduction

With the advent of multi-core processors concurrent programming is becoming indispensable. Scala's primary concurrency construct isactors. Actors are basically concurrent processes that communicate by exchanging messages. Actors can also be seen as a form of active objects where invoking a method corresponds to sending a message.

The Scala Actors library provides both asynchronous and synchronous message sends (the latter are implemented by exchanging several asynchronous messages). Moreover, actors may communicate usingfutures where requests are handled asynchronously, but return a representation (the future) that allows to await the reply.

This tutorial is mainly designed as a walk-through of several complete example programs that can be readily compiled and run using Scala 2.4 or newer.

First Example

Our first example consists of two actors that exchange a bunch of messages and then terminate. The first actor sends "ping" messages to the second actor, which in turn sends "pong" messages back (for each received "ping" message one "pong" message).

We start off by defining the messages that are sent and received by our actors. In this case, we can use singleton objects (in more advanced programs, messages are usually parameterized). Since we want to use pattern matching, each message is acaseobject:

case object Pingcase object Pongcase object Stop

The ping actor starts the message exchange by sending aPing message to the pong actor. ThePong message is the response from the pong actor. When the ping actor has sent a certain number ofPing messages, it sends aStop message to the pong actor.

All classes, objects and traits of the Scala actors library reside in thescala.actors package. From this package we import theActor class that we are going to extend to define our custom actors. Furthermore, we import all members of theActor object since it contains many useful actor operations:

import scala.actors.Actorimport scala.actors.Actor._

Actors are normal objects that are created by instantiating subclasses of theActor class. We define the behavior of ping actors by subclassingActor and implementing its abstractact method:

class Ping(count: int, pong: Actor) extends Actor {  def act() {    var pingsLeft = count - 1    pong ! Ping    while (true) {      receive {        case Pong =>          if (pingsLeft % 1000 == 0)            Console.println("Ping: pong")          if (pingsLeft > 0) {            pong ! Ping            pingsLeft -= 1          } else {            Console.println("Ping: stop")            pong ! Stop            exit()          }      }    }  }}

The number ofPing messages to be sent and the pong actor are passed as arguments to the constructor. The call to thereceive method inside the infinite loop suspends the actor until aPong message is sent to the actor. In that case the message is removed from the actor'smailbox and the corresponding action on the right side of the arrow is executed.

In the case wherepingsLeft is greater than zero we send aPing message topong using the send operator!, and decrement thepingsLeft counter. If thepingsLeft counter has reached zero, we send aStop message topong, and terminate the execution of the current actor by callingexit().

The class for our pong actor is defined similarly:

class Pong extends Actor {  def act() {    var pongCount = 0    while (true) {      receive {        case Ping =>          if (pongCount % 1000 == 0)            Console.println("Pong: ping "+pongCount)          sender ! Pong          pongCount = pongCount + 1        case Stop =>          Console.println("Pong: stop")          exit()      }    }  }}

There is one interesting point to notice, however. When receiving aPing message, aPong message is sent to thesender actor, which is not defined anywhere in our class! In fact, it is a method of theActor class. Usingsender, one can refer to the actor that sent the message that the current actor last received. This avoids having to explicitly pass the sender as arguments to messages.

After having defined our actor classes, we are now ready to create a Scala application that uses them:

object pingpong extends Application {  val pong = new Pong  val ping = new Ping(100000, pong)  ping.start  pong.start}

Analogous to Java threads, actors have to be started by calling theirstart method.

$ scalac pingpong.scala$ scala -cp . examples.actors.pingpongPong: ping 0Ping: pongPong: ping 1000Ping: pongPong: ping 2000...Ping: stopPong: stop

Make it Thread-less!

Actors are executed on a thread pool. Initially, there are 4 worker threads. The thread pool grows if all worker threads are blocked but there are still remaining tasks to be processed. Ideally, the size of the thread pool corresponds to the number of processor cores of the machine.

When actors call thread-blocking operations, such asreceive (or evenwait), the worker thread that is executing the current actor (self) is blocked. This means basically that the actor is represented as a blocked thread. Depending on the number of actors you want to use, you might want to avoid this, since most JVMs cannot handle more than a few thousand threads on standard hardware.

Thread-blocking operations can be avoided by usingreact to wait for new messages (the event-based pendant ofreceive). However, there is a (usually small) price to pay:react never returns. In practice, this means that at the end of a reaction to a message, one has to call some function that contains the rest of the actor's computation. Note that usingreact inside awhile loop does not work! However, since loops are common there is special library support for it in form of aloop function. It can be used like this:

loop {  react {    case A => ...    case B => ...  }}

Note thatreact calls can be nested. This allows to receive several messages in sequence, like this:

react {  case A => ...  case B => react { // if we get a B we also want a C    case C => ...  }}

To make our ping and pong actors thread-less, it suffices to simply replacewhile(true) withloop, andreceive withreact. For example, here is the modifiedact method of our pong actor:

def act() {  var pongCount = 0  loop {    react {      case Ping =>        if (pongCount % 1000 == 0)          Console.println("Pong: ping "+pongCount)        sender ! Pong        pongCount = pongCount + 1      case Stop =>        Console.println("Pong: stop")        exit()    }  }}

Second Example

In this example, we are going to write an abstraction ofproducers which provide a standard iterator interface to retrieve a sequence of produced values.

Specific producers are defined by implementing an abstractproduceValues method. Individual values are generated using theproduce method. Both methods are inherited from classProducer. For example, a producer that generates the values contained in a tree in pre-order can be defined like this:

class PreOrder(n: Tree) extends Producer[int] {  def produceValues = traverse(n)  def traverse(n: Tree) {    if (n != null) {      produce(n.elem)      traverse(n.left)      traverse(n.right)    }  }}

Producers are implemented in terms of two actors, aproducer actor, and acoordinator actor. Here is how we can implement the producer actor:

abstract class Producer[T] {  protected def produceValues: unit  protected def produce(x: T) {    coordinator ! Some(x)    receive { case Next => }  }  private val producer: Actor = actor {    receive {      case Next =>        produceValues        coordinator ! None    }  }  ...}

Note how theproducer actor is defined! This time we did not bother to create an extra subclass ofActor and implement itsact method. Instead, we simply define the actor's behaviorinline using theactor function. Arguably, this is much more concise! Moreover, actors defined usingactor are started automatically--no need to invoke theirstart method!

So, how does the producer work? When receiving aNext message, it runs the (abstract)produceValues method, which, in turn, calls theproduce method. This results in sending a sequence of values, wrapped inSome messages, to the coordinator. The sequence is terminated by aNone message.Some andNone are the two cases of Scala's standardOption class.

The coordinator synchronizes requests from clients and values coming from the producer. We can implement it like this:

private val coordinator: Actor = actor {  loop {    react {      case Next =>        producer ! Next        reply {          receive { case x: Option[_] => x }        }      case Stop => exit('stop)    }  }}

Note that inside the handler for theNext message, we usereply to return a receivedOption value to some requesting actor. We are going to explain this in the next section, so stay tuned...

def iterator = new Iterator[T] {  private var current: Any = Undefined  private def lookAhead = {    if (current == Undefined) current = coordinator !? Next    current  }  def hasNext: boolean = lookAhead match {    case Some(x) => true    case None => { coordinator ! Stop; false }  }  def next: T = lookAhead match {    case Some(x) => current = Undefined; x.asInstanceOf[T]  }}

private val Undefined = new Object