(Quick Reference)

5.1 Actors Principles - Reference Documentation

Authors: The Whole GPars Gang

Version: 1.1.0

5.1 Actors Principles

Actors share a pool of threads, which are dynamically assigned to actors when the actors need to react to messages sent to them. The threads are returned back to the pool once a message has been processed and the actor is idle waiting for some more messages to arrive. Actors become detached from the underlying threads and so a relatively small thread pool can serve potentially unlimited number of actors. Virtually unlimited scalability in number of actors is the main advantage of event-based actors , which are detached from the underlying physical threads.

Here are some examples of how to use actors. This is how you create an actor that prints out all messages that it receives.

import static groovyx.gpars.actor.Actors.actor

def console = actor { loop { react { println it } }

Notice the loop() method call, which ensures that the actor doesn't stop after having processed the first message.

As an alternative you can extend the DefaultActor class and override the act() method. Once you instantiate the actor, you need to start it so that it attaches itself to the thread pool and can start accepting messages. The actor() factory method will take care of starting the actor.

class CustomActor extends DefaultActor {
    @Override
    protected void act() {
        loop {
            react {
                println it
            }
        }
    }
}

def console=new CustomActor() console.start()

Messages can be sent to the actor using multiple methods

console.send('Message')
console 'Message'
console.sendAndWait 'Message'                                                     //Wait for a reply
console.sendAndContinue 'Message', {reply -> println "I received reply: $reply"}  //Forward the reply to a function

Creating an asynchronous service

import static groovyx.gpars.actor.Actors.actor

final def decryptor = actor { loop { react {String message-> reply message.reverse() } } }

def console = actor { decryptor.send 'lellarap si yvoorG' react { println 'Decrypted message: ' + it } }

console.join()

As you can see, you create new actors with the actor() method passing in the actor's body as a closure parameter. Inside the actor's body you can use loop() to iterate, react() to receive messages and reply() to send a message to the actor, which has sent the currently processed message. The sender of the current message is also available through the actor's sender property. When the decryptor actor doesn't find a message in its message queue at the time when react() is called, the react() method gives up the thread and returns it back to the thread pool for other actors to pick it up. Only after a new message arrives to the actor's message queue, the closure of the react() method gets scheduled for processing with the pool. Event-based actors internally simulate continuations - actor's work is split into sequentially run chunks, which get invoked once a message is available in the inbox. Each chunk for a single actor can be performed by a different thread from the thread pool.

Groovy flexible syntax with closures allows our library to offer multiple ways to define actors. For instance, here's an example of an actor that waits for up to 30 seconds to receive a reply to its message. Actors allow time DSL defined by org.codehaus.groovy.runtime.TimeCategory class to be used for timeout specification to the react() method, provided the user wraps the call within a TimeCategory use block.

def friend = Actors.actor {
    react {
        //this doesn't reply -> caller won't receive any answer in time
        println it
        //reply 'Hello' //uncomment this to answer conversation
        react {
            println it
        }
    }
}

def me = Actors.actor { friend.send('Hi') //wait for answer 1sec react(1000) {msg -> if (msg == Actor.TIMEOUT) { friend.send('I see, busy as usual. Never mind.') stop() } else { //continue conversation println "Thank you for $msg" } } }

me.join()

When a timeout expires when waiting for a message, the Actor.TIMEOUT message arrives instead. Also the onTimeout() handler is invoked, if present on the actor:

def friend = Actors.actor {
    react {
        //this doesn't reply -> caller won't receive any answer in time
        println it
        //reply 'Hello' //uncomment this to answer conversation
        react {
            println it
        }
    }
}

def me = Actors.actor { friend.send('Hi')

delegate.metaClass.onTimeout = {-> friend.send('I see, busy as usual. Never mind.') stop() }

//wait for answer 1sec react(1000) {msg -> if (msg != Actor.TIMEOUT) { //continue conversation println "Thank you for $msg" } } }

me.join()

Notice the possibility to use Groovy meta-programming to define actor's lifecycle notification methods (e.g. onTimeout() ) dynamically. Obviously, the lifecycle methods can be defined the usual way when you decide to define a new class for your actor.

class MyActor extends DefaultActor {
    public void onTimeout() {
        …
    }

protected void act() { … } }

Actors guarantee thread-safety for non-thread-safe code

Actors guarantee that always at most one thread processes the actor's body at a time and also under the covers the memory gets synchronized each time a thread gets assigned to an actor so the actor's state can be safely modified by code in the body without any other extra (synchronization or locking) effort .

class MyCounterActor extends DefaultActor {
    private Integer counter = 0

protected void act() { loop { react { counter++ } } } }

Ideally actor's code should never be invoked directly from outside so all the code of the actor class can only be executed by the thread handling the last received message and so all the actor's code is implicitly thread-safe . If any of the actor's methods is allowed to be called by other objects directly, the thread-safety guarantee for the actor's code and state are no longer valid .

Simple calculator

A little bit more realistic example of an event-driven actor that receives two numeric messages, sums them up and sends the result to the console actor.

import groovyx.gpars.group.DefaultPGroup

//not necessary, just showing that a single-threaded pool can still handle multiple actors def group = new DefaultPGroup(1);

final def console = group.actor { loop { react { println 'Result: ' + it } } }

final def calculator = group.actor { react {a -> react {b -> console.send(a + b) } } }

calculator.send 2 calculator.send 3

calculator.join() group.shutdown()

Notice that event-driven actors require special care regarding the react() method. Since event_driven actors need to split the code into independent chunks assignable to different threads sequentially and continuations are not natively supported on JVM, the chunks are created artificially. The react() method creates the next message handler. As soon as the current message handler finishes, the next message handler (continuation) gets scheduled.

Concurrent Merge Sort Example

For comparison I'm also including a more involved example performing a concurrent merge sort of a list of integers using actors. You can see that thanks to flexibility of Groovy we came pretty close to the Scala model, although I still miss Scala pattern matching for message handling.

import groovyx.gpars.group.DefaultPGroup
import static groovyx.gpars.actor.Actors.actor

Closure createMessageHandler(def parentActor) { return { react {List<Integer> message -> assert message != null switch (message.size()) { case 0..1: parentActor.send(message) break case 2: if (message[0] <= message[1]) parentActor.send(message) else parentActor.send(message[-1..0]) break default: def splitList = split(message)

def child1 = actor(createMessageHandler(delegate)) def child2 = actor(createMessageHandler(delegate)) child1.send(splitList[0]) child2.send(splitList[1])

react {message1 -> react {message2 -> parentActor.send merge(message1, message2) } } } } } }

def console = new DefaultPGroup(1).actor { react { println "Sorted array:t${it}" System.exit 0 } }

def sorter = actor(createMessageHandler(console)) sorter.send([1, 5, 2, 4, 3, 8, 6, 7, 3, 9, 5, 3]) console.join()

def split(List<Integer> list) { int listSize = list.size() int middleIndex = listSize / 2 def list1 = list[0..<middleIndex] def list2 = list[middleIndex..listSize - 1] return [list1, list2] }

List<Integer> merge(List<Integer> a, List<Integer> b) { int i = 0, j = 0 final int newSize = a.size() + b.size() List<Integer> result = new ArrayList<Integer>(newSize)

while ((i < a.size()) && (j < b.size())) { if (a[i] <= b[j]) result << a[i++] else result << b[j++] }

if (i < a.size()) result.addAll(a[i..-1]) else result.addAll(b[j..-1]) return result }

Since actors reuse threads from a pool, the script will work with virtually any size of a thread pool, no matter how many actors are created along the way.

Actor lifecycle methods

Each Actor can define lifecycle observing methods, which will be called whenever a certain lifecycle event occurs.
  • afterStart() - called right after the actor has been started.
  • afterStop(List undeliveredMessages) - called right after the actor is stopped, passing in all the unprocessed messages from the queue.
  • onInterrupt(InterruptedException e) - called when the actor's thread gets interrupted. Thread interruption will result in the stopping the actor in any case.
  • onTimeout() - called when no messages are sent to the actor within the timeout specified for the currently blocking react method.
  • onException(Throwable e) - called when an exception occurs in the actor's event handler. Actor will stop after return from this method.

You can either define the methods statically in your Actor class or add them dynamically to the actor's metaclass:

class MyActor extends DefaultActor {
    public void afterStart() {
        …
    }
    public void onTimeout() {
        …
    }

protected void act() { … } }

def myActor = actor {
    delegate.metaClass.onException = {
        log.error('Exception occurred', it)
    }

… }

To help performance, you may consider using the silentStart() method instead of start() when starting a DynamicDispatchActor or a ReactiveActor . Calling silentStart() will by-pass some of the start-up machinery and as a result will also avoid calling the afterStart() method. Due to its stateful nature, DefaultActor cannot be started silently.

Pool management

Actors can be organized into groups and as a default there's always an application-wide pooled actor group available. And just like the Actors abstract factory can be used to create actors in the default group, custom groups can be used as abstract factories to create new actors instances belonging to these groups.

def myGroup = new DefaultPGroup()

def actor1 = myGroup.actor { … }

def actor2 = myGroup.actor { … }

The parallelGroup property of an actor points to the group it belongs to. It by default points to the default actor group, which is Actors.defaultActorPGroup , and can only be changed before the actor is started.

class MyActor extends StaticDispatchActor<Integer> {
    private static PGroup group = new DefaultPGroup(100)

MyActor(...) { this.parallelGroup = group … } }

The actors belonging to the same group share the underlying thread pool of that group. The pool by default contains n + 1 threads, where n stands for the number of CPUs detected by the JVM. The pool size can be set explicitly either by setting the gpars.poolsize system property or individually for each actor group by specifying the appropriate constructor parameter.

def myGroup = new DefaultPGroup(10)  //the pool will contain 10 threads

The thread pool can be manipulated through the appropriate DefaultPGroup class, which delegates to the Pool interface of the thread pool. For example, the resize() method allows you to change the pool size any time and the resetDefaultSize() sets it back to the default value. The shutdown() method can be called when you need to safely finish all tasks, destroy the pool and stop all the threads in order to exit JVM in an organized manner.

… (n+1 threads in the default pool after startup)

Actors.defaultActorPGroup.resize 1 //use one-thread pool

… (1 thread in the pool)

Actors.defaultActorPGroup.resetDefaultSize()

… (n+1 threads in the pool)

Actors.defaultActorPGroup.shutdown()

As an alternative to the DefaultPGroup , which creates a pool of daemon threads, the NonDaemonPGroup class can be used when non-daemon threads are required.

def daemonGroup = new DefaultPGroup()

def actor1 = daemonGroup.actor { … }

def nonDaemonGroup = new NonDaemonPGroup()

def actor2 = nonDaemonGroup.actor { … }

class MyActor { def MyActor() { this.parallelGroup = nonDaemonGroup }

void act() {...} }

Actors belonging to the same group share the underlying thread pool. With pooled actor groups you can split your actors to leverage multiple thread pools of different sizes and so assign resources to different components of your system and tune their performance.

def coreActors = new NonDaemonPGroup(5)  //5 non-daemon threads pool
def helperActors = new DefaultPGroup(1)  //1 daemon thread pool

def priceCalculator = coreActors.actor { … }

def paymentProcessor = coreActors.actor { … }

def emailNotifier = helperActors.actor { … }

def cleanupActor = helperActors.actor { … }

//increase size of the core actor group coreActors.resize 6

//shutdown the group's pool once you no longer need the group to release resources helperActors.shutdown()

Do not forget to shutdown custom pooled actor groups, once you no longer need them and their actors, to preserve system resources.

The default actor group

Actors that didn't have their parallelGroup property changed or that were created through any of the factory methods on the Actors class share a common group Actors.defaultActorPGroup . This group uses a resizeable thread pool with an upper limit of 1000 threads . This gives you the comfort of having the pool automatically adjust to the demand of the actors. On the other hand, with a growing number of actors the pool may become too big an inefficient. It is advisable to group your actors into your own PGroups with fixed size thread pools for all but trivial applications.

Common trap: App terminates while actors do not receive messages

Most likely you're using daemon threads and pools, which is the default setting, and your main thread finishes. Calling actor.join() on any, some or all of your actors would block the main thread until the actor terminates and thus keep all your actors running. Alternatively use instances of NonDaemonPGroup and assign some of your actors to these groups.

def nonDaemonGroup = new NonDaemonPGroup()
def myActor = nonDaemonGroup.actor {...}

alternatively

def nonDaemonGroup = new NonDaemonPGroup()

class MyActor extends DefaultActor { def MyActor() { this.parallelGroup = nonDaemonGroup }

void act() {...} }

def myActor = new MyActor()

Blocking Actors

Instead of event-driven continuation-styled actors, you may in some scenarios prefer using blocking actors. Blocking actors hold a single pooled thread for their whole life-time including the time when waiting for messages. They avoid some of the thread management overhead, since they never fight for threads after start, and also they let you write straight code without the necessity of continuation style, since they only do blocking message reads via the receive method. Obviously the number of blocking actors running concurrently is limited by the number of threads available in the shared pool. On the other hand, blocking actors typically provide better performance compared to continuation-style actors, especially when the actor's message queue rarely gets empty.

def decryptor = blockingActor {
    while (true) {
        receive {message ->
            if (message instanceof String) reply message.reverse()
            else stop()
        }
    }
}

def console = blockingActor { decryptor.send 'lellarap si yvoorG' println 'Decrypted message: ' + receive() decryptor.send false }

[decryptor, console]*.join()

Blocking actors increase the number of options to tune performance of your applications. They may in particular be good candidates for high-traffic positions in your actor network.