Reactive Actors in Java

There are plenty of technologies to organize parallel computing. One of the most promising and simple ones is the model of actors. It allows to partially get rid of such major problems of parallelism as race conditions, locking waits of operations termination, endless mutual exclusions and synchronizations, and so on and so forth. This approach also simplifies the code paralleling.

We are going to review it by the example of akka framework using Java language. As for akka, it is written in Scala.

Theory and Principles

Actor is an isolated object, meaning that there are no direct references to an instance. It deals with asynchronous processing of incoming messages. A message is any invariable object implementing Serializable interface. They are stored in a queue, and when the system passes control to the actor (we’ll talk more about it soon), objects are extracted from the queue according to FIFO principle. They are processed according to the internal logic which we should set explicitly. The actors form a tree-like hierarchy. Each of them has an actor (or a system of actors) as the creator. In order to send a message to an actor, we should have a reference to it (do not mix it up with the object reference). It can be of two types: ActorRef and ActorSelection. The reference has .tell(message, sender) method that sends the message along the specified path. Sender is also a reference of ActorRef type. A receiver-actor will receive it when calling getSender() method during the message processing.

ActorRef is a reference to a certain, guaranteed to be an existing (at the moment of the reference creation) actor that is the analog of StrongReference. We can get ActorRef explicitly by calling getSender() method either when processing the messages either after having got a direct reference to the sender, or when creating children actors. To guarantee message delivery, we can use either a reply-message, or an agent. We are going to discuss it below.

ActorSelection is a weak reference. Its key distinction is the ability of creating ActorSelection, knowing just a name and a path to the actor. But the reverse side of the coin is that there are no guarantees that there is something existing on the specified path. We can send a message, but no one can guarantee its delivery. Therefore, if we need to send some data with the guarantee, but there is no direct reference to the necessary actor, we can send a random message to it and get the strong ActorRef reference from the reply.

That’s the base minimum, after which we can continue studying all of it in practice. In order to avoid manual enabling of dependencies, we should also use all libraries provided in akka (we can also utilize maven/etc).

Some Practice

The asynchronous world begins with a system of actors. Let's create it. Usually, there’s no need to have more than one actor system within the application.

ActorSystem system = ActorSystem.create("learning2hard");

The system has plenty of settings we should specify before execution. It is active right after the creation. We can either set it in the external config file, or pass by the second argument to the constructor-method. You can read more about it at the developer’s website. We don’t need it for the moment, but it’s better to know about it anyway.

Now, let’s create the kernel actor. All the future logic will be the child one to it. Technically, the actor is an object inheriting UntypedActor. Let’s do it.

public class Kernel extends UntypedActor {
    @Override
    public void onReceive(Object arg0) throws Exception {               
        unhandled(arg0);                
    }
}

The actor class is ready. unhandled(arg0) method marks the incoming object as the unprocessed one and writes about it in the log. Let’s add the kernel instance to the system of actors:

final ActorRef kernel = system.actorOf(Props.create(Kernel.class), "kernel");

We have created a child actor of the Kernel class with regard to the system and named it “kernel”. The system is the root of a hierarchic tree of actors. String names and paths are commonly used in akka. If it is necessary to pass some objects to the constructor of Kernel, we can do it with the help of Props.create(Kernel.class, obj1, obj2, obj3...).

We have made a programming standard meaning a program that does nothing useful. Let’s add some functional. For example, processing commands that are read from the console. Within the context of an actor, it is not allowed to use calls/operations that are locking the flow. Waiting for the command input to the console is one of them (Actually, there is a way to place them there. We are going to review the mysterious Future monads a bit later). Therefore, it is advisable to remove and avoid locking calls in the world of actors. Let’s apply the first method and read commands from the console outside the system of actors and then send them in the form of messages to the kernel.

Scanner sc = new Scanner(System.in);
        while (sc.hasNext()) {
            String f = sc.nextLine();
            if (f.equals("exit")) break;
            kernel.tell(f, ActorRef.noSender());
        }
        system.shutdown();

The only thing to mention is that we are sending a message from “the outer world” to the world of actors. There’s obviously no reference to the sender. That’s why we are going to use a stub.

Let’s write processing of the received string to kernel class and make a simple echo.

@Override
    public void onReceive(Object arg0) throws Exception {
        if (arg0 instanceof String) {
            String s = (String) arg0;
            System.out.println(s);
            return;
        }
        unhandled(arg0);                
    }

That’s how the logic of any actor look like and the list of locks with instanceof condition and return + unhandled(arg0) is finished, after all the locks, in case something has not been processed. We have used ActorRef.noSender() as the sender. It’s time to see how the address looks like:

The sender address: " + getSender());

We get Actor[akka://learning2hard/deadLetters]
and see that the address exists and it is not null. It points to the stub of the basket that accepts objects and does nothing with them. Such an address type is called local and absolute. It is right only within the limits of the current system of actors. We’ll discuss addresses in details below.

But one is not enough. So, lets create more actors and think up a simple distributed task and check, whether a number is prime. We are going to search them through without optimizations. Enter a number from the console, send it to the kernel. The latter will estimate the number of divisors and send work to the operating ones. Each of them will search out its own range of divisors. When at least one of them announces that a number is composite, it will write a result. If all of them have finished their work and found nothing, the number is prime.

To tell the truth, it’s easier to write than to tell about it. So, let’s create a worker with the clear logic.

public class PrimeWorker extends UntypedActor {
    @Override
    public void onReceive(Object arg0) throws Exception {
        if (arg0 instanceof Job) {
            Job task = (Job) arg0;
            for (int i = task.from; i < task.to; i++) 
                if (task.number % i == 0) { //The number is composite. 
                    getSender().tell(new JobResult(task.jobID,false), getSelf());
                    return;
                }
            getSender().tell(new JobResult(task.jobID,true), getSelf());
        }
        unhandled(arg0);                
    }
    public static class Job implements Serializable {
        private static final long serialVersionUID = 5095931000566324969L;
        public final int jobID; //The task identifier. It is used to aggregate results 
        public final int number; //The number under consideration
        public final int from; //The lower limit of the search range 
        public final int to; //The upper limit of the search range 
        public Job(int jobID, int number, int from, int to) {
            this.jobID = jobID; this.number = number; this.from = from; this.to = to;
        }
    }
    public static class JobResult implements Serializable {
        private static final long serialVersionUID = -1788069759380966076L;
        public final int jobID;
        public final boolean isPrime;
        public JobResult(int jobID, boolean isPrime) {
            this.jobID = jobID; this.isPrime = isPrime;
        }
    }
}

There should be nothing complicated about it. Now, let’s get down to the kernel code. It distributes the work and joins results.

public class Kernel extends UntypedActor {         
    //work ID -> (currentWorkerCount, isPrime)
    private TreeMap> jobs = new TreeMap>();
    private int job_id_counter = 0;
    @Override
    public void onReceive(Object arg0) throws Exception {
        if (arg0 instanceof String) {                   
            int i = Integer.valueOf((String) arg0);
            for (int j = 2; j < i; j++) 
                getContext().actorOf(Props.create(PrimeWorker.class)).tell(new PrimeWorker.Job(job_id_counter, i, j, j+1), getSelf());                  
            jobs.put(job_id_counter, Pair.get(i - 2, true));
            job_id_counter++;
            return;
        }
        if (arg0 instanceof JobResult) {
            JobResult jr = (JobResult) arg0;                    
            Pair task = jobs.get(jr.jobID);
            if (!jr.isPrime) task.second = false;
            task.first--;
            if (task.first < 1){
                System.out.println("Number " + jr.number + (task.second ? " prime" : " composite"));
                jobs.remove(jr.jobID);
            }
            getSender().tell(PoisonPill.getInstance(), getSelf()); //The actor has done his job. Send the command of deletion to it. 
        }
        unhandled(arg0);                
    }
    public static class Pair {
        public A first;
        public B second;
        public static  Pair get(C a, D b) {
            Pair p = new Pair();
            p.first = a;
            p.second = b;
            return p;
        }
    }
}

Let's not worry about the absence of checks, that was not our goal. What we can notice is some wastefulness in a way that we create actors for one operation and then delete them. Of course, we can place references to them in the array and call them from it in turns. But, fortunately, it’s all done. Let’s try to improve the code by applying off-the-shelf solutions.

Router is a specialized object that passes incoming messages to actors by using a certain strategy of choosing them. It can be of two types: a pool and a group. A group chooses actors according to the strategy in the specified path. A pool creates them itself. A router is rarely used as an independent object. As a rule, it is encapsulated to the actor. We are going to deal with such actor-routers exactly. There are plenty of choosing strategies. We can read about them at the website of the framework developers. The most multipurpose ones are SmallestMailboxRouter(SM) and BroadcastRouter. The first one chooses the least loaded actor from the set (according to the mailbox size). The second one sends out messages to all of them. Let’s declare a router with a pool of actors and SM strategies.

ActorRef router = getContext().actorOf(new SmallestMailboxPool(5).props(Props.create(PrimeWorker.class)), "workers");

In this case we assign the pool size statically. We can also create dynamic changing pools, or even write our own logic for changes. But all of that is beyond this article.

Let's modify the code of sending the kernel tasks (and remove sending the message to get rid of the actor):

if (arg0 instanceof String) {                       
            int i = Integer.valueOf((String) arg0);
            for (int j = 2; j < i; j++) 
                router.tell(new PrimeWorker.Job(job_id_counter, i, j, j+1), getSelf());                 
            jobs.put(job_id_counter, Pair.get(i - 2, true));
            job_id_counter++;
            return;
        }

It’s done. Now the router will take care of tasks distribution. We’ve reviewed the basic moments. It’s time to move on to more subtle aspects.

Execution Context. Locking Operations. State Transitions.

It has been already mentioned that we pass control to actors, but did not specify the way we do it. The execution context is in charge of it. We indicate it explicitly in the configuration file. fork-join executor is used by default. It is multi-purpose and the most efficient one for common tasks. Create some number of flows that pass control to actors by choosing them in n flows from the list. The criterion of switching to another actor is either an empty queue of messages, or processing k messages in a row, provided that nothing else is specified. It is obvious that when an actor tries to call a locking operation, the flow is locked and other actors begin being processed much slower. If the number of locked actors becomes equal to the number of flows, the system dies out.

Fortunately, there is a way out from even such an awkward situation. Its name is Future. They are not the Futures we are used to in Java, but a functional good thing from Scala. To begin with, let’s talk about Future in Java. It’s an object that can contain the result of executing an asynchronous operation. The main distinction of this future is callback, which is the possibility of performing some actions after the task termination, absolutely asynchronously. The lion’s share of the framework’s “strength” lies in futures. How does it look like?

Future f = future(new Callable() {
                @Override
                public String call() throws Exception {
                    //Some blocking stuff
                    return "hello habr";
                }
            }, getContext().dispatcher());
            f.onComplete(new OnComplete(){
                @Override
                public void onComplete(Throwable arg0, String arg1)     throws Throwable {
                    System.out.println(arg0 != null ? arg0 : arg1);                                     
                }
            }, getContext().dispatcher());

In addition to Callable (the analog of Runnable with the return value), we also pass some dispatcher during creation. That is what the actor executes. But if we make the actor context execute a locking action, will the latter freeze? The thing is in akka having two pools of flows. One of them is for actors and we can not lock it. The other one is for some trifles like Future. This pool of flows is of a variable size. Both of them are limited. But when calling a locking function, we almost (within reasonable limits) do not risk slowing down the basic set of actors.

Getting back to the mentioned above code, we can see Callback. They have left a lasting impression on the author. Those being familiar with functional languages can notice that Future behaves like a monad. Actually, it is a monad. This means that we can make compositions of them and partially apply functions. All of that is the Futures module.

Even if we do not go into details of the things that are far from the classic Java, it is a possibility of creating asynchronous operations in the code, without creating flows and thinking up the architecture. We can also input by organizing chains of “read a file, ask a user and, at the same time, process the data; save the result depending on the respond” type. All of that should be done quickly, simply and compactly.

Of course, a fly in the ointment in this situation is debugging. The code is extremely reactive. Akka executor helps you in many ways. For instance, exceptions are not derived inside a future. At all. If something has happened there, you will learn about it when noticing that OnComplete call is absent. And you will never know, what exactly happened. There’s a dirty hack for the more handy debugging — recover. Akka is famous for its failsafe policy. Futures are no exception. Therefore, they have inbuilt tools to remove an incorrect result of a future. It looks like the following:

Future f = future(new Callable() {
                @Override
                public String call() throws Exception {
                    //Some blocking stuff
                    return "hello habr";
                }
            }, getContext().dispatcher());
            f.recover(new Recover() {
                @Override
                public String recover(Throwable arg0) throws Throwable {
                    arg0.printStackTrace();
                    return null;
                }
            }, getContext().dispatcher());
            f.onComplete(new OnComplete(){
                @Override
                public void onComplete(Throwable arg0, String arg1) throws Throwable {
                    System.out.println(arg0 != null ? arg0 : arg1);                                     
                }
            }, getContext().dispatcher());

We can see that recover does not fix the error in the given example and returns null to onComplete. At least, it writes about it. As a result, it is clear that Future is an extremely powerful tool. It allows to implement asynchronous branching right amid the asynchronous code and not be afraid of locks. But it does require some caution.

There is a useful and widespread template for the world of actors. It is based on Future and its name is ask.

Future