13. Producers & Actors

Written by Filip Babić

Multi-processing communication is one of the most enticing challenges you can face when developing multi-threaded software. Multi-threading technologies have been around for years and there are a few solutions that are industry standard. In this chapter, you’ll be learning about two of those solutions, which can be really useful in your future applications.

Producing and consuming data

The first is the producer-consumer problem, which describes a two-process communication standard. One process produces data, places it in a queue, while the other picks items off of the queue, one by one, and consumes it. Hence the name. A problem arises if the consumer tries to pick data off an empty queue, or if the producer tries to overfill the queue. This is familiar to what you’ve learned so far in the book.

Furthermore, this pattern doesn’t have to describe a 1:1 relationship. One approach is for the producer to try and push as many events in the queue, and the system to consume them as fast as possible, using multiple consumers.

If you think about it, you can picture a thread pool the same way. You have one producer or worker, and threads, which are the consumers.

Producer-consumer problem

Just like with pipelines, producers face the same challenge. As previously mentioned, a full or an empty queue could cause loss of information, thread blocking or exceptions. On the other hand, creating producers with coroutines is much easier and avoids these problems. Let’s see how you’d create a basic producer using the Coroutines API.

Creating a producer

If you haven’t already, open up the starter project for this chapter with the name produce_actor. Then, open Produce.kt in the producer package in the project. Finally, to create a producer, call produce(), on a CoroutineScope, in main(), like so:

val producer = GlobalScope.produce<Int>(capacity = 10) {}

To understand what the snippet does, let’s take a look at the produce() signature, and what you can pass to it:

public fun <E> CoroutineScope.produce(
  context: CoroutineContext = EmptyCoroutineContext,
  capacity: Int = 0,
  @BuilderInference block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E>

You can pass it a context, which by default it receives from the CoroutineScope you may have passed in. You can define the capacity, which is the maximum number of items it can hold in the queue. And finally, you pass in a lambda from which you can define the behavior of the producer.

You’ll also notice something in the lambda, the BuilderInference. This annotation allows the compiler to infer the type of the outer generic function, using functions inside the lambda block. So if you use a function that the ProducerScope can run with like a String, produce() will know it’s going to create a Producer<String>.

Producing values

The producer has to produce some values. Since the return type of produce() is a ReceiveChannel, you can’t use it for sending values. You have to do it within the lambda you pass as parameter. The simplest way would be using a loop. Change the code in main(), in Produce.kt, to this:

val producer = GlobalScope.produce(capacity = 10) {
  while (isActive) {
    if (!isClosedForSend) {
      val number = Random.nextInt(0, 20)
      if (offer(number)) {
        println("$number sent")
      } else {
        println("$number discarded")
      }
    }
  }
}

In this simple example, you run a loop that produces numbers from within a coroutine. Since you do it in a coroutine, this is not really a blocking call, as the program can finish, without having to wait for the infinite loop to stop.

Here, you’re calling offer(), which attempts to queue a new element, if there’s room for the element, otherwise the element is going to be discarded. This function has a Boolean return type, which is true in the former case and false in the latter. It’s very useful, and you can use it in order to print a different message in the output. Before offering the random number, you check in the loop the value of the isClosedForSend, which becomes false as soon as the channel is closed explicitly with close() or because of an exception.

It’s very interesting to note that if you run the code now you’ll get nothing in the output. This is because produce() is not blocking and the application exits immediately.

In order to see some output you can add a simple Thread.sleep() to the end of the code like this:

val producer = GlobalScope.produce(capacity = 10) {
  while (isActive) {
    if (!isClosedForSend) {
      val number = Random.nextInt(0, 20)
      if (offer(number)) {
        println("$number sent")
      } else {
        println("$number discarded")
      }
    }
  }
}
Thread.sleep(30L)

You can wait a very short period of time like 30ms in order to have an output like this where the values will obviously be different because they are randomly generated:

3 sent
17 sent
2 sent
9 sent
15 sent
11 sent
16 sent
13 sent
10 sent
11 sent
9 discarded
8 discarded
. . .

As you can see the previous code generates 10 elements — the capacity of the channel — and then it discards the following because the channel is full and offer() returns false.

This suggests a different option for implementing the same producer. Instead of checking for the return value of the offer method you can use isFull whose meaning is pretty obvious. Its value is true when the channel is full. You can then change the previous code with this:

val producer = GlobalScope.produce(capacity = 10) {
  while (isActive) {
    val number = Random.nextInt(0, 20)
    if (!isFull) {
      if (offer(number)) {
        println("$number sent")
      }
    } else {
      println("$number discarded")
    }
  }
}
Thread.sleep(30L)

Now, you check isFull and you invoke offer() only if it’s going to be successful because the channel still has a place for the value. The discarded message is now in the else block of the test on the isFull. You can do better, though. In the previous code you’re generating the random value even if you don’t offer it to the channel and, if the channel is full, you’re basically wasting it. A better approach is the one which uses the send method instead of offer(). send() is actually suspendable and it allows you to write the following code:

val producer = GlobalScope.produce(capacity = 10) {
  while (isActive) {
    val number = Random.nextInt(0, 20)
    send(number)
    println("$number sent")
  }
}
Thread.sleep(30L)

In this case, you don’t waste any values because send() is suspendable and it won’t continue until the channel has enough space for it. You’ll then get an output like this:

11 sent
3 sent
16 sent
1 sent
11 sent
17 sent
15 sent
11 sent
3 sent
11 sent

If you run the previous code, you’ll only get the first 10 values and nothing will happen until you create a consumer for it. So let’s move onto building that!

Consuming the values

The produce function conveniently returns a ReceiveChannel. This means you can iterate through the values, transform or filter them, or run an infinite loop, reading the values.

Let’s start off with the most basic example: a while loop. Add this code to the end of main function:

while (!producer.isClosedForReceive) {
  val number = producer.poll()

  if (number != null) {
    println("$number received")
  }
}

If you build and run the application, you’ll see output like this:

12 sent
18 sent
19 sent
12 sent
17 sent
6 sent
9 sent
9 sent
1 sent
5 sent
12 received
3 sent
18 received
. . .

In the previous code you’re invoking poll() on the producer until its isClosedForReceive has a false value. This is happening if the channel has been closed on the producer side invoking the close method and all the values have been consumed. Because you use the poll method you have to check for nullability.

This is because poll is not blocking and it returns null if the channel is empty and so something else should be produced. If in the producer side you’re using send(), null values should never happen. A better approach uses coroutines.

Replace the while(true) loop with the following code:

GlobalScope.launch {
  producer.consumeEach { println("$it received") }
}

Thread.sleep(30L)

This code does the same thing as the previous loop. Because you’re launching a coroutine, you still have to halt the program so it doesn’t finish before printing out any numbers, which is why you added the Thread.sleep() invocation.

In real systems this will run normally, as long as your application is running. What consumeEach does is listen for each of the items, and processes it, ultimately removing it from the producer queue.

poll() is not blocking, and it’s returning null if nothing is available. You also have the option of using the receive method, which is suspendable. You can then replace the previous code with this:

GlobalScope.launch {
  while (isActive) {
    val value = producer.receive()
    println("$value received")
  }
}

Thread.sleep(30L)

While the producer is active you can wait for a value and then display it.

If you use a coroutine you also have another very simple way of implementing a consumer. Try to replace the previous consumer code with this:

GlobalScope.launch {
  for (value in producer) {
    println("$value received")
  }
}

Thread.sleep(30L)

In this case, you can use the classic for loop over the producer and print all its value. The ReceiveChannel implements the Iterable in a proper way so you can use what you normally use for iterating over a collection in the context of a producer/consumer pattern.

As you can see, all of these are easy to implement as opposed to having to write a complex synchronization mechanism yourself. The producer-consumer pattern is really useful when you’re trying to broadcast events from one place for other people to listen to.

A different paradigm of multi-threaded communication is when you’re trying to delegate events to others, such as work that you need to complete. This is called the actor model. Let’s see what it’s about.

Acting upon data

The actor model is a bit different from what you’ve seen so far and it exists as a possible solution for a very common problem: sharing data in a multithreading environment. When you create an instance of an object, you know you can interact with it using the operations it exposes: its interface. Most of the objects also have some state, which can change after interacting with other objects. Everything is simple if all the objects collaborate on the same thread. In a multithreading environment you have to introduce complexity in the code in order to make all the classes type safe. When multiple threads access shared data you know you can have problems like data race conditions. Locks, implicit or not, are a possible solution but are usually difficult to manage and test.

You can solve this problem by allowing the interaction with the object only to a specific component, which is also the only consumer of a queue. If you want to interact with the object in a thread safe way, you just have to send messages to the queue of the encapsulating component. The component then is only responsible to consume the message and, depending on its type, change the state of the object it encapsulates. This component is then called an actor.

When you implement a solution using actors you’re basically defining local or private states and what are the possible operations on them. Each operation maps to a message type you can send to its queue.

Usually an actor is responsible for a single operation and can delegate some other operations to other actors if needed in order to have a clear separation of concerns.

The usage of actors helps us to focus on the specific operation and ignoring all the multithreading aspects, which can be very difficult to implement.

Handling actors properly

Having many actors at your disposal, being limited only by memory, is both a great thing and a challenge. The true challenge comes when you have to clean up old actors. If you hold a reference to your actor children, in each of the actors, you’d end up with a large reference tree. And you couldn’t clean up actors as you go, since they hold references to new, fresh actors, as well.

You could create threads, in which you create new actors, so you don’t hold an implicit reference to the parent, but then again, threads are even more expensive.

Kotlin Coroutines have a more efficient way to create actors, which avoids lots of strong references and thread allocations. Let’s see how to do so.

Building actors using coroutines

Note: The Jetbrains team is currently working on both the Flow API, and complex actors. They haven’t yet decided what to do with the current Actor API, and as such it’s been marked as obsolete. However, the API still works, so it’s worth checking it out.

An actor is then a consumer bound to a specific channel that you can create invoking a simple coroutine builder called — drumroll — actor. In order to understand how it works it can be useful to have a look at its signature:

public fun <E> CoroutineScope.actor(
    context: CoroutineContext = EmptyCoroutineContext,
    capacity: Int = 0,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    onCompletion: CompletionHandler? = null,
    block: suspend ActorScope<E>.() -> Unit
): SendChannel<E>

It is very similar to produce(), but it has a few more parameters you can pass. The CoroutineStart parameter you’ve already learned about, but there’s also the CompletionHandler. You can use this handler to listen to actor completions. And, finally, there’s the block parameter, which is of the type ActorScope<E>.() -> Unit. This is another CoroutineScope, which also holds a reference to its enclosing channel, so you could poll for new values. It’s important to note how the return type is SendChannel. The returned object is the one you’re going to use in order to interact with the actor you’ve just created.

Now, since you know a bit more about actors, open up Actor.kt, in the actor package, and you should see this snippet of code:

// 1
object completionHandler : CompletionHandler {
  override fun invoke(cause: Throwable?) {
    println("Completed!")
  }
}

fun main() {
  // 2
  val actor = GlobalScope.actor<String>(
      onCompletion = completionHandler,
      capacity = 10) {
    // 3
    for (data in channel) {
      println(data)
    }
  }

  // 4
  (1..10).forEach {
    actor.offer(Random.nextInt(0, 20).toString())
  }
  // 5
  actor.close()
  // 6
  Thread.sleep(500L)
}

Here is what each part of the above code snippet is doing:

1. You create a simple implementation of the CompletionHandler, which just prints a Completed! message when the actor is complete. This is happening when close() is invoked on its SendChannel.
2. Here you create the actor passing a capacity of 10 and the reference to the CompletionHandler.
3. In the lambda of the actor you define the consumer logic. Here you’re just printing what the actor is receiving. In general here is where you can change the state of the actor depending on the received message. It’s useful to note how the reference of the channel is implicitly available in the block of the actor.
4. You implement a simple loop, which offers 10 values into the channel for the actor.
5. You sent all your values and you can close the actor.
6. In order to see the output you can use a Thread.sleep().

If you run the code, you’ll get an output like the following:

5
4
19
10
8
15
5
6
0
8
Completed!

As you can see, you’ll get 10 random values and the Completed! message as expected.

Delegating actor workload

The actor model, however, relies on delegating excess work to others. So, for example, if you’re building a robot-powered-storage system, where everything is organized by robots, you have to find a way to optimize the workload.

If a certain robot has too much to carry around, it can pass some of its work on to a different robot. And if that second robot has too much work, it can pass it to a third one, and so forth.

See how this would look in code. Select everything in Actor.kt, and replace with the following code:

fun main() {

  val items = listOf(
      Package(1, "coffee"),
      Package(2, "chair"),
      Package(3, "sugar"),
      Package(4, "t-shirts"),
      Package(5, "pillowcases"),
      Package(6, "cellphones"),
      Package(7, "skateboard"),
      Package(8, "cactus plants"),
      Package(9, "lamps"),
      Package(10, "ice cream"),
      Package(11, "rubber duckies"),
      Package(12, "blankets"),
      Package(13, "glass")
  )

  val initialRobot = WarehouseRobot(1, items)

  initialRobot.organizeItems()
  Thread.sleep(5000)
}

Here we’re using two pre-baked classes: the Package.kt and the WarehouseRobot.kt. Package.kt is pretty simple; it’s just a model holding some data. However, WarehouseRobot.kt is where the party’s at. Open up WarehouseRobot.kt. First you see the constructor with its parameters, and the companion object:

class WarehouseRobot(private val id: Int,
                     private var packages: List<Package>) {

  companion object {
    private const val ROBOT_CAPACITY = 3
  }
...
}

Each robot will have an id and a set of packages it needs to organize around. Furthermore, each robot has the same package capacity, which is three packages per robot. Of course, each robot can process the items given to it:

 private fun processItems(items: List<Package>) {
    val actor = GlobalScope.actor<Package>(
    capacity = ROBOT_CAPACITY) {

      var hasProcessedItems = false

      while (!packages.isEmpty()) {
        val currentPackage = poll()

        currentPackage?.run {
          organize(this)

          packages -= currentPackage
          hasProcessedItems = true
        }

        if (hasProcessedItems && currentPackage == null) {
          cancel()
        }
      }
    }

    items.forEach { actor.offer(it) }
  }

  private fun organize(warehousePackage: Package) =
      println("Organized package " +
        "${warehousePackage.id}:" +
        warehousePackage.name)

In this function, the robot creates a new actor, which uses the ROBOT_CAPACITY as the maximum number of items in the queue. The actor also has to use the hasProcessedItems flag, otherwise it’d close early, before there are any items processed. This happens because poll() returns a null item if the actor didn’t receive any items yet.

Once the robot processes an item, it changes its inner state, or packages, by removing the processed item from the list. Only the actor can change this private state internally.

But the key function, here, which makes this class an actor is organizeItems():

fun organizeItems() {
  val itemsToProcess = packages.take(ROBOT_CAPACITY)
  val leftoverItems = packages.drop(ROBOT_CAPACITY)

  packages = itemsToProcess
  
  val packageIds = packages.map { it.id }
        .fold("") { acc, item -> "$acc$item " }

  processItems(itemsToProcess)

  if (leftoverItems.isNotEmpty()) {
    GlobalScope.launch {
      val helperRobot = WarehouseRobot(id.inc(), leftoverItems)

      helperRobot.organizeItems()
    }
  }

  println("Robot #$id processed following packages:$packageIds")
}

Initially, the robot has to divide the items up into those it has to process and the ones that it doesn’t have the capacity for. Once it processes its items and if there are leftovers, it sends them to another robot. Finally, it returns to its station, waiting for more work. This looks a bit like recursion, since robots are creating robots within themselves. But effectively, the helper robots are created in the GlobalScope, so their parents can finish and be cleared from memory before they finish themselves.

If you build and run main() in Actor.kt, you should see the following output:

Organized package 1:coffee
Organized package 2:chair
Organized package 3:sugar
Robot #2 processed following packages:4 5 6 
Robot #1 processed following packages:1 2 3 
Organized package 7:skateboard
Organized package 8:cactus plants
Organized package 9:lamps
Robot #3 processed following packages:7 8 9 
Organized package 4:t-shirts
Organized package 5:pillowcases
Organized package 6:cellphones
Robot #4 processed following packages:10 11 12 
Organized package 10:ice cream
Organized package 11:rubber duckies
Organized package 12:blankets
Organized package 13:glass
Robot #5 processed following packages:13

The robot organizing system works!

Acting in parallel

Right now, the packages are mostly ordered, with a few exceptions to the rule. This means that, usually, once the first robot finishes its work, the helper robot starts on its packages. However, you’re building recursion-like work, which doesn’t suffer from the StackOverflowException or OutOfMemory exception, since the actors get cleaned up one by one.

To make them run in parallel, simply change the order of processItems(), with the check for creating a new actor. The order should be like this:

fun organizeItems() {
  ...
  if (leftoverItems.isNotEmpty()) {
    GlobalScope.launch {
      val helperRobot = WarehouseRobot(id.inc(), leftoverItems)

      helperRobot.organizeItems()
    }
  }

  processItems(itemsToProcess)
  ...
}

This will first create all the actors, and then run the item processing. If you run the code now, the output should be different. It should be something similar to this:

Robot #3 processed following packages:7 8 9 
Robot #1 processed following packages:1 2 3 
Robot #5 processed following packages:13 
Robot #2 processed following packages:4 5 6 
Robot #4 processed following packages:10 11 12 
Organized package 10:ice cream
Organized package 4:t-shirts
Organized package 11:rubber duckies
Organized package 5:pillowcases
Organized package 1:coffee
Organized package 13:glass
Organized package 2:chair
Organized package 6:cellphones
Organized package 12:blankets
Organized package 3:sugar
Organized package 7:skateboard
Organized package 8:cactus plants
Organized package 9:lamps

First, the actors are created and their processing prints out. Then the packages are organized depending on how fast each actor can consume the items. This can be much faster than having to run actors one-by-one, but it could also mean that if you have a large number of items left to organize, there will be a higher amount of memory allocation, and many more objects created.

For best performance, when building such systems, you should tweak the capacity of each actor, optimize their workload and data structures. You could also build tree-like structures, where you’d have both sequential and parallel actor computation.

Key points

• Produce-consumer pattern and the actor model are tried and tested mechanisms for multi-threading.
• Producer-consumer relationships are one-to-many, where you can consume the events from multiple places.
• The actor model is a way to share data in a multithread environment using a dedicated queue.
• The actor model allows you to offload large amounts of work to many smaller constructs.
• Actors have a many-to-one relationship, since you can send events from multiple places, but they all end up in one actor.
• Each actor can create new actors, delegating and offloading work.
• Building actors using threads can be expensive, which is where coroutines come in handy.
• Actors can be arranged to run in sequential order, or to run in parallel.

Where to go from here?

You now know how to build effective communication mechanisms, using produce() and actor(). You’ve got everything you need to connect multiple threads, or to fan out a large workload. In the next few chapters, you’ll see how to build a different kind of mechanism of communication like broadcasting. You’ll also see how channels’ data can be transformed and combined.

So let’s channel this positive energy and head over to the next chapter!