Threads and Labyrinths

Between 1980 and 2003, we experienced the rise and dominance of object-oriented languages. These years were the dark ages of functional programming. Functional languages were spoken only in academic and scientific environments, barely related to industry.

An interesting problem arose with object-oriented programming: the implementation of concurrency and parallelism. These two concepts are the Achilles’ heel of structured and object-oriented programming. Imagine an implementation of threads in C ++ or Java; complexity is vast and proneness to error is very large.

Concurrency is not easy; making more than one thing with a program is related to dealing with race conditions, semaphores, mutexes, locks, shared data, and all the stuff related to multithreading. This includes basic issues to determine precisely what a program with several threads is doing, or when a variable is being accessed from multiple threads, or what its value is at a given point in time, or how to know if there are two threads in standby, and if this condition is going to release them (and when) or if it is a deadlock. Unfortunately, thread-based concurrency gives more challenges than solutions.

Today there is a lot of technical debt in the proprietary thread implementations of concurrency issues. No one wants to touch huge systems because the code is complex and the chances of a change breaking everything are very high.

In this context, functional programming experienced rebirth. With the release of Scala in 2003, F# in 2005, Clojure in 2007, and Elixir in 2012, the actor model approach was declared the winner in solving concurrency issues.

Actors 101

Actors are objects. An actor is an object that sends and receives messages. According to the actor specification, the order of the received messages is not relevant; but in Akka, there is an implementation called a mailbox, which is a stack where messages are consumed.

What the actor does with a received message depends on how you solve a specific problem. The actor could handle the message internally, it could send a message to another actor, it could create another actor, or it could take an action with the message. An Akka actor is a high-level abstraction.

The following are the main comparison points between OOP and actors:

• Unit. In OOP, the smallest processing unit is the object; in the actor model, it is the actor. We already know that in Akka, the actors are objects but an actor is a more bounded representation of reality than an object.

• Encapsulation. In both models, the smallest processing unit encapsulates state and behavior. In OOP, the state is determined by the value of the attributes in a given time, and behavior is ruled by the class methods. In the actor model, the state is determined by the messages; if there are no messages in the mailbox, the actor will wait indefinitely for messages.

• Access. In OOP, executing object methods from outside of the object is allowed (but not recommended), as well as access and modify object fields from outside of the object. In the actor model, access to the actor's methods or fields is strictly prohibited; all the communication must be done through messages.

• Globals. In OOP, for example, there is the Singleton pattern, which is a class with a single instance. Global variables and class variables exist, but they are discouraged. In the actor model, global variables don’t exist. A shared global state doesn’t exist either.

• Messages. In OOP, the messages between Objects could be mutable. In the actor model, the messages between actors are strictly immutable.

• Exceptions. In OOP exists the traditional and well-known try-catch approach, which is the most complex way to handle exceptions because you have to manage all the possible values of the variables involved. In the actor model, the “let it crash” approach exists; if something fails, let it fail. The exception scenario could affect only the actor involved, not the complete environment.

• Concurrency and parallelism. In OPP, the most used approach is the thread model, which is a complex solution to the problem. In the actor model, you don’t have to worry about concurrency and parallelism, because if everything follows the actor convention, there is no problem with parallelism.

The following are the Lightbend recommendations for the actor model:

• Think of actors as employees. Think of each actor model as a company.

• Think of the actor’s siblings as people in the same hierarchical level.

• Think of the actor’s children as the employee's subordinates.

• An actor has one (and only one) supervisor—the actor who created it.

• Actor model success is the delegation to subordinates.

The Akka implementation of the actor model has these peculiarities:

• When you create an actor, Akka provides an ActorRef.

• Actors run in real Java threads. Some actors could share the same thread.

• There are three mailbox types: Unbounded, Bounded, and Priority.

• Actors can scan their mailboxes to look for specific messages.

• There is a “dead letter” mailbox with all the actors’ terminated messages.

The Lightbend Reactive Platform ( www.lightbend.com ) is a family of five members, described as follows:

• Scala: The programming language. The Reactive Platform fully supports both Java and Scala, so you can choose what is best for you.

• Akka: Message-driven runtime. At the center of the Reactive Platform is Akka, a message-driven middleware or runtime with Scala and Java APIs.

• Spark: Apache Spark, which is written in Scala and Akka, is a fast data engine to fuel Reactive applications.

• Lagom: Reactive microservices framework. An opinionated framework for building web microservices.

• Play: Just-hit-reload web framework. The Play Framework is the just-hit-reload web development framework with Scala and Java APIs.

We have always said that Apache Spark is the Scala “killer app.”

In this book, we only cover Scala, Akka, and Spark, but if you are an enthusiastic web service developer or web developer, don’t pass on the opportunity to explore Lagom and Play, respectively, more deeply.

Actors

For our first example, we use a multilanguage greeter; that is, we enter a particular language and the program responds by replying “Good day,” in the language specified.

import akka.actor.Actor
import akka.actor.ActorSystem
import akka.actor.Props

class GreeterActor extends Actor {
  def receive = {
    case "en" => println("Good day")
    case "es" => println("Buen dia")
    case "fr" => println("Bonjour")
    case "de" => println("Guten Tag")
    case "pt" => println("Bom dia")
    case _ => println(":(")
  }
}

object Main extends App {

  // build the ActorSystem
  val actorSystem = ActorSystem("MultilangSystem")

  // instantiate the actor
  val greeter = actorSystem.actorOf(Props[GreeterActor], name = "GreeterActor")

  // send the actor some messages
  greeter ! "en"
  greeter ! "es"
  greeter ! "fr"
  greeter ! "de"
  greeter ! "pt"
  greeter ! "zh-CN"

  // shut down the actor system
  actorSystem.shutdown
}

When we run this program, the output is as follows:

$ sbt run
[info] Running Main
Good day
Buen dia
Bonjour
Guten Tag
Bom dia
:(
Process finished with exit code 0

Here is the step-by-step explanation:

• When using Akka actors, you need to import the akka.actor._ package. You at least need these classes: Actor, ActorSystem, and Props.

• You must define a class of type Actor. In this case, we called it GreeterActor . You can use any name that you want.

• The actor main performance must be defined under the actor receive() method.

• The structure of the receive()method is typical in functional programming; it is called a match expression. The lines should always go from the most specific case to the most general case. The most specific case is at the top and the more general case is at the bottom.

• The last line in the match expression should be (as good practice; not required) the default case; it says what to do if the pattern didn’t find a match.

• Recall that the underscore operator (_) in Scala means whatever. If the message doesn’t find a match, an UnhandledMessage exception is thrown, which represents poor programming practices.

• You create a Main object that extends a Scala app to give your actor a scenario that displays their histrionic abilities.

• You first need an actor system. Any name with alphanumeric characters is good; hyphens are allowed but not as the first character.

• To liven up your actor, invoke the actorOf method in the actor system. The actor will start asynchronously.

• If you want to interact, send messages with the ! operator.

• To finish the play, you must call the shutdown method in the actor system.

Actor System

Imagine the actor system as a theater company, an actors union, or a circus:

• There is a hierarchy; each actor always has a supervisor and an actor can have siblings and children.

• Actors belonging to the same system of actors share dispatchers, deployments, and addresses.

• The actor system is the meeting point where the actors are created and searched.

• Internally, the actor system is a thread controller; the actor system decides when to create threads for an application.

• If the system does not turn off the actors (with the shutdown method), the application will not end. As long as the actor system is running, the application will continue running.

Actor Reference

Imagine the actor reference as the actor agent; that is, someone who represents them and receives letters from his fans:

• The actor system’s actorOf method has two main tasks: start the actor asynchronously and return the ActorRef requested.

• The ActorRef is a handler, so you cannot directly access the actor and break the actor system encapsulation rules.

• The ActorRef follows the facade pattern over the actor; that is, it serves as a way to communicate with the actor without directly accessing the actor. Thus, you never access actor variables and methods directly, as dictated by the encapsulation principles.

• The ActorRef is immutable; you cannot change it because it is only a reference.

• An actor has one (and only one) ActorRef. An ActorRef refers to one (and only one) actor. It is a one-to-one relationship.

• To comply with the Akka actor model, the ActorRef is serializable and server independent, so you can distribute, copy, and pass references (to the actors’ fans) across the network.

Actor Communication

Actor communication is always more easily explained by example. Just remember that to send a message to an actor, you use the ! operator.

The ! operator always works with ActorRefs; never between Actor class instances.

In Akka, when you send a message to an actor with the ! operator, the actor that receives the message also receives a reference to the actor that sent the message; this reference is accessed by the sender variable. And it helps to send the actor invoker response messages. Recall that sender is a reserved word; use it wisely.

import akka.actor._

case object SendANewCat
case object LiveALife
case object BackToHeaven
case object LifeSpended {
  var remaining = 0// a default value
}

class God(indulged: ActorRef) extends Actor {

  def receive = {
    case SendANewCat =>
      println("GOD: Go!, you have seven lives")
      indulged ! LiveALife
    case LifeSpended =>
      if ( LifeSpended.remaining == 0){
        println("GOD: Time to Return!")
        indulged ! BackToHeaven
        context.stop(self)
      }
      else {
        println("GOD: one live spent, " + LifeSpended.remaining + " remaining.")
        indulged ! LiveALife
      }
    case _ => println("GOD: Sorry, I don't understand")
  }
}

class Cat extends Actor {
  var lives = 7 // All the cats born with 7 lives

  def receive = {
    case LiveALife =>
      println("CAT: Thanks God, I still have " + lives + " lives")
      lives -= 1
      LifeSpended.remaining = lives
      sender ! LifeSpended
    case BackToHeaven =>
      println("CAT: No more lives, going to Heaven")
      context.stop(self)
    case _ => println("CAT: Sorry, I don't understand")
  }
}

object CatLife extends App {
  val system = ActorSystem("CatLifeSystem")
  val sylvester = system.actorOf(Props[Cat], name = "Sylvester")
  val catsGod = system.actorOf(Props(new God(sylvester)), name = "CatsGod")

  // God sends a Cat
  catsGod ! SendANewCat

  system.terminate()
}

Running our example, we get this output:

GOD: Go!, you have seven lives
CAT: Thanks God, I still have 7 lives
GOD: one live spent, 6 remaining.
CAT: Thanks God, I still have 6 lives
GOD: one live spent, 5 remaining.
CAT: Thanks God, I still have 5 lives
GOD: one live spent, 4 remaining.
CAT: Thanks God, I still have 4 lives
GOD: one live spent, 3 remaining.
CAT: Thanks God, I still have 3 lives
GOD: one live spent, 2 remaining.
CAT: Thanks God, I still have 2 lives
GOD: one live spent, 1 remaining.
CAT: Thanks God, I still have 1 lives
GOD: Time to Return!
CAT: No more lives, going to Heaven
Process finished with exit code 0

Here is an actor communication example analysis:

• It is always advisable to model messages as classes within your application; in our example, we have four objects that we use as a message:

  • SendANewCat case object

  • LiveALife case object

  • BackToHeaven case object

  • LifeSpended case object

• CatLife is the application in which we have the main application. The first line creates the actor system and calls it CatLifeSystem.

• On the next line, we create an ActorRef for the Cat class. An ActorRef to a cat actor is loaded in the sylvester variable.

• We then create the god actor. Note that the constructor receives a reference to his indulged cat. This was used to show the relationship between the actors; we could have declared a constructor with no arguments, but this was only for demonstration purposes.

• Then, we send a message to god requesting a new cat.

• When the god actor receives the message, it starts the cat’s life, until we reach the life limit, then we stop this actor context.

• The context object is available to all actors in the actor system. It is used to stop the actors together.

• It is important to recall that cat, god, and indulged are ActorRefs and are not Actor class instances. An actor should never be accessed directly; always through messages.

• If we access the actors directly, the environment becomes unreliable in making high concurrency and parallelism. The message system and encapsulation always ensure that we are doing a parallelizable and concurrent environment because there are no shared variables or locks. All transactions are ACID.

Actor Lifecycle

In addition to the constructor, an actor has the following lifecycle methods, which are all described in Table 4-1:

• receive

• preStart

• postStop

• preRestart

• postRestart

In the following example code, The Hulk (the green superhero) is used to show the lifecycle methods:

import akka.actor._
case object GetAngry

class Hulk extends Actor {
  println("in the Hulk constructor")

  override def preStart {
    println("in the Hulk preStart")
  }

  override def postStop {
    println("in the Hulk postStop")
  }

  override def preRestart(reason: Throwable, message: Option[Any]) {
    println("in the Hulk preRestart")
    println(s" preRestart message: ${message.getOrElse("")}")
    println(s" preRestart reason: ${reason.getMessage}")
    super.preRestart(reason, message)
  }

  override def postRestart(reason: Throwable) {
    println("in the Hulk postRestart")
    println(s" postRestart reason: ${reason.getMessage}")
    super.postRestart(reason)
  }

  def receive = {
    case GetAngry => throw new Exception("ROAR!")
    case _ => println("Hulk received a message...")
  }
}

object LifecycleTest extends App {
  val system = ActorSystem("LifeCycleSystem")
  val hulk = system.actorOf(Props[Hulk], name = "TheHulk")
  println("sending Hulk a message")
  hulk ! "hello Hulk"
  Thread.sleep(5000)
  println("making Hulk get angry")
  hulk ! GetAngry
  Thread.sleep(5000)
  println("stopping Hulk")
  system.stop(hulk)
  println("shutting down Hulk system")
  system. terminate()
}

The following is the output when the program is run:

[info] Running LifecycleTest
sending Hulk a message
in the Hulk constructor
in the Hulk preStart
Hulk received a message...
making Hulk get angry
in the Hulk preRestart
[ERROR] [01/01/2015 01:01:01.964] [LifeCycleSystem-akka.actor.default-dispatcher-6] [akka://LifeCycleSystem/user/TheHulk] ROAR!
java.lang.Exception: ROAR!
        at Hulk$$anonfun$receive$1.applyOrElse(chapter04_03.scala:31)
        at akka.actor.Actor$class.aroundReceive(Actor.scala:480)
        at Hulk.aroundReceive(chapter04_03.scala:6)
        at akka.actor.ActorCell.receiveMessage(ActorCell.scala:525)
        at akka.actor.ActorCell.invoke(ActorCell.scala:494)
        at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
        at akka.dispatch.Mailbox.run(Mailbox.scala:224)
        at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
 preRestart message: GetAngry
 preRestart reason: ROAR!
in the Hulk postStop
in the Hulk constructor
in the Hulk postRestart
 postRestart reason: ROAR!
in the Hulk preStart

As an exercise, make the trace source code vs. the program output.

Starting Actors

You have already seen how to create actors from the actor system . To create actors from another actor, you must use the following context:

class GodWanabe extends Actor {
        val = context.actorOf creature (Props [Creature] name = "Creature")
        // Add the code for its creation ...
}

Let’s look at the actor lifecycle control between actors with an example based on characters from The Simpsons. Mr. Burns is the boss and has a nuclear power plant. He hires two employees, Homer Simpson and Frank Grimes, but then only fires Frank Grimes.

import akka.actor._

case class Hire(name: String)
case class Name(name: String)

class Boss extends Actor {
  def receive = {
    case Hire(name) =>
      // here the boss hire personnel
      println(s"$name is about to be hired")
      val employee = context.actorOf(Props[Employee], name = s"$name")
      employee ! Name(name)
    case _ => println(s"The Boss can't handle this message.")
  }
}

class Employee extends Actor {
  var name = "Employee name"

  override def postStop {
    println(s"I'm ($name) and Mr. Burns fired me: ${self.path}")
  }

  def receive = {
    case Name(name) => this.name = name
    case _ => println(s"The Employee $name can't handle this message.")
  }
}

object StartingActorsDemo extends App {
  val actorSystem = ActorSystem("StartingActorsSystem")
  val mrBurns = actorSystem.actorOf(Props[Boss], name = "MrBurns")

  // here the boss hires people
  mrBurns ! Hire("HomerSimpson")
  mrBurns ! Hire("FrankGrimes")

  // we wait some office cycles
  Thread.sleep(4000)

  // we look for Frank and we fire him
  println("Firing Frank Grimes ...")
  val grimes = actorSystem.actorSelection("../user/MrBurns/FrankGrimes")

  // PoisonPill, an Akka special message
  grimes ! PoisonPill
  println("now Frank Grimes is fired")
}

The following is the output when we run this program:

[info] Running StartingActorsDemo
HommerSimpson is about to be hired
FrankGrimes is about to be hired
Firing Frank Grimes ...
now Frank Grimes is fired
I'm (FrankGrimes) and Mr. Burns fired me: akka://StartingActorsSystem/user/MrBurns/FrankGrimes
Process finished with exit code -1

Let’s analyze the starting actors example code:

1. Create and use the Name and Hire utility classes to send messages between actors.

2. When the employee actor receives the Name message, assigns it to the name variable.

3. When the boss receives a Hire message, it uses the context.actorOf method to hire new employees.

4. As usual, the main program creates the actor system.

5. The main program then creates the boss actor using the actor system reference.

6. The main program sends the boss two Hire messages, with HomerSimpson and FrankGrimes as names.

7. After a pause (4 seconds), look for Frank Grimes in the actor system, then send him the PoisonPill message, which is an Akka actor system special message that asynchronously sends the stop signal to an actor. Use the postStop method to print a message after PoisonPill.

Stopping Actors

As you saw previously, there are four ways to stop an actor:

• Calling system.stop(ActorRef) from the ActorSystem level

• Calling context.stop(ActorRef) from inside an actor

• Sending an actor the PoisonPill message

• Programming a gracefulStop

Table 4-2 summarizes the ways to stop an actor.

Some aspects to consider when stopping actors:

• The stop message is asynchronous. The stop method could return before the actor is actually stopped; use it wisely.

• The shutdown process has two subprocesses. First, it suspends the actor’s mailbox. Second, it sends the stop message to all the actor children; the father actor has to wait for all its children to stop.

• When you can’t process any more messages, these messages are sent to the dead letters mailbox. You can access them with the deadLetters method in the actor system.

• When an actor is stopped, the postStop lifecycle method is invoked. Normally it is used to clean up resources.

Here is an example of code using system.stop:

import akka.actor._

class Scapegoat extends Actor {
  def receive = {
    case s:String => println("Message received: " + s)
    case _ => println("What?")
  }
}

object StopExample extends App {
  val system = ActorSystem("StopExample")
  val sg = system.actorOf(Props[Scapegoat], name = "ScapeGoat")
  sg ! "ready?"

  // stop our crash dummy
  system.stop(sg)
  system.terminate()
}

Killing Actors

The following code shows how to kill actors . It is a very violent way; discretion is advised. Normally, if you want to stop an actor gracefully, you use the methods described earlier.

import akka.actor._

class ScapeGoat extends Actor {
  def receive = {
    case s:String => println("Message received: " + s)
    case _ => println("Uh?")
  }

  override def preStart {
    println("In preStart method")
  }

  override def postStop {
    println("In postStop method")
  }

  override def preRestart(reason: Throwable, message: Option[Any]) {
    println("In preRestart method")
  }

  override def postRestart(reason: Throwable) {
    println("In postRestart method")
  }
}

object Abbatoir extends App {
  val system = ActorSystem("Abbatoir")
  val sg = system.actorOf(Props[ScapeGoat], name = "ScapeGoat")
  sg ! "say goodbye"

  // finish him!
  sg ! Kill
  system. terminate()
}

This is the code output:

In preStart method
Message received: say goodbye
In postStop method
Process finished with exit code 0

Shutting down the Actor System

As you have already seen in the examples, this is the method to shut down the actor system:

system.terminate()

Because of its importance, we dedicated this section to this method. Remember, if you don’t call the shutdown method in your application, the program will run indefinitely.

Actor Monitoring

This code shows how an actor asks to be notified when a child actor dies:

import akka.actor._

class Child extends Actor {
  def receive = {
    case _ => println("Child received a message")
  }
}

class Dad extends Actor {
  // Dad actor create a child actor
  val child = context.actorOf(Props[Child], name = "Son")
  context.watch(child)

  def receive = {
    case Terminated(child) => println("This will not end here -_-")
    case _ => println("Dad received a message")
  }
}

object ChildMonitoring extends App {

  val system = ActorSystem("ChildMonitoring")

  // we create a Dad (and it will create the Child)
  val dad = system.actorOf(Props[Dad], name = "Dad")

  // look for child, then we kill it
  val child = system.actorSelection("/user/Dad/Son")

  child ! PoisonPill
  Thread.sleep(3000)

  println("Revenge!")
  system. terminate()
}

Running this code produces the following result:

This will not end here
Revenge!
Process finished with exit code 0

Through the watch() method, an actor knows when a subordinate stops. This is very useful because it lets the supervisor handle the situation.

Note that when an exception occurs within an actor, the actor does not kill himself. In Akka, an exception makes an actor restart automatically.

Looking up Actors

In the previous example, you saw how to find a specific actor:

val child = system.actorSelection("/user/Dad/Son")

The actorSelection method is available under the actor system and within each actor instance through the context variable.

You can also look for actors with a relative path; for example, from siblings:

// From an actor brother
val bro = context.actorSelection("../myBrother")

The actorSelection method in the actor system can be used to find actors:

val child = system.actorSelection("akka://MonitoringTest/user/Dad/Son")
val child = system.actorSelection(Sec("user", "Dad", "Son"))

With the actorSelection method, you can also look for a sibling:

val child = system.actorSelection(Sec("..." "Son"))

Actor Code of Conduct

At this point, you have seen everything that you need to write actors. To achieve concurrent programming, it is important to maintain a performance style; that is, a code of ethics among actors. If you keep this code of ethics, the source code will be easy to debug and won’t have typical multithreaded programming problems, such as deadlocks and race conditions. In this section, we present the fundamental principles for a good performance.