Szenario 2: Actor Scheduling

Actors ziehen derzeit großes Interesse auf sich, da sie leichtgewichtiger als Threads sind und dank der ausschließlich über Nachrichten laufenden Kommunikation ein robustes Programmiermodell bieten. Hinter den Kulissen verwenden viele Actor-Frameworks jedoch Thread Pools, deren Worker Threads diejenigen Actors ausführen, die Nachrichten in ihrer Mailbox haben. Der ForkJoinPool ist demzufolge besonders für Actor-Frameworks auf der JVM relevant.

Verwendung findet eine kleine Actor-Scheduling-Implementierung, die im Kern ähnliche Mechanismen wie das Akka-Framework nutzt. Es wird an dieser Stelle nur die Definition des Tasks gezeigt, der Rest lässt sich im Code im GitHub-Repository nachvollziehen.

public class ActorForkJoinTask extends RecursiveAction {
    private final AbstractDispatcher dispatcher;
    private final Mailbox mailbox;

    public ActorForkJoinTask(AbstractDispatcher dispatcher,
        Mailbox mailbox) {
    this.dispatcher = dispatcher;
    this.mailbox = mailbox;
    }

    @Override
    protected void compute() {
     int counter = 0;
     Message message;
     while (counter++ < ActorBenchmarkConfig.MAX_CONSUME_BURST
         && (message = mailbox.pollMessage()) != null) {
         mailbox.getActor().receive(message);
     }
     mailbox.setScheduled(false);
     if (!mailbox.isEmpty()) {
         dispatcher
             .scheduleUnlessAlreadyScheduled(mailbox
                 .getActor().getId());
     }
    }
}

Der ActorForkJoinTask leitet sich von RecursiveAction ab, da er keinen Rückgabewert benötigt. Bei Ausführung des Tasks werden Nachrichten aus der Mailbox des Actors genommen und dem Actor zur Verarbeitung übergeben. Enthält die Mailbox bei Erreichen einer Obergrenze noch weitere Nachrichten, wird der Actor erneut zur Ausführung vorgemerkt. Der Dispatcher fügt hierzu einen neuen ActorForkJoinTask in die lokale Task Queue des Worker Threads ein.

Die Implementierung des Tasks für den ThreadPoolExecutor ist praktisch identisch, leitet sich aber von Runnable ab. Ist ein Actor auszuführen, plant der Dispatcher einen neuen Task in die zentrale Eingangs-Queue ein.

Sowohl beim ForkJoinPool als auch beim ThreadPoolExecutor achtet der Dispatcher intern darauf, keinen Actor doppelt einzuplanen. So ist gewährleistet, dass niemals zwei Threads gleichzeitig denselben Actor ausführen, und die Implementierung des Actors selbst muss sich nicht um Thread-Synchronisation kümmern.

Betrachtung der Ergebnisse

Der Benchmark an sich ist genauso aufgebaut wie beim MapReduce-Szenario. Im Folgenden werden Ergebnisse aus einem Szenario mit 1000 Actors gezeigt. In den Mailboxen von 200 Actors befindet sich zu Beginn eine Nachricht, von der jede in der Folge 100.000-mal an einen zufällig bestimmten Actor weitergeleitet und zum Schluss gelöscht wird. Während dem Ausführen eines Actors arbeitet dieser immer nur eine Nachricht auf einmal ab.

Beim Actor Scheduling schneidet der ForkJoinPool besser ab als der ThreadPoolExecutor (Abb. 6).

Abbildung 6 zeigt, dass der ThreadPoolExecutor für dieses Szenario überhaupt nicht geeignet ist. Die gemeinsame Eingangs-Queue führt zu einer erheblichen Menge von Konkurrenzsituationen, was Abbildung 7 durch die Thread-Zustände veranschaulicht.

Aufgrund von Konkurrenzsituationen beim Zugriff auf die gemeinsame Eingangs-Queue verbringen die Threads des ThreadPoolExecutor viel Zeit mit Warten (Abb. 7).

Der ForkJoinPool hingegen kann die Berechnung dank der lokalen Task Queues bis zur Anzahl der virtuellen Prozessoren beschleunigen. Jeder Worker Thread füllt seine eigene Task Queue fortwährend mit neuen Actor Tasks, sodass es keine Konkurrenzsituationen auf den Queues gibt.

Auffällig ist, dass der Speedup insgesamt eher gering ist, beim ThreadPoolExecutor sogar stets deutlich kleiner als 1 (das heißt die sequenzielle Berechnung ist schneller). Der Grund hierfür ist zum einen die oben beschriebene Anforderung, beim parallelen Scheduling keinen Actor gleichzeitig von mehreren Threads ausführen zu lassen. Das bringt einen spürbaren Synchronisations-Overhead mit sich, der bei der sequenziellen Berechnung nicht anfällt. Zum anderen erzeugen die Actors im beschriebenen Beispiel beim Empfang von Nachrichten kaum Rechenlast (die Nachricht wird lediglich an einen anderen Actor weitergeleitet), sodass es wenig zu parallelisieren gibt.

Was den Durchsatz betrifft, hat der ForkJoinPool beim Actor Scheduling klar die Nase vorn. Allerdings wirft ein solches Szenario auch die Frage nach der Fairness auf. Da ein Worker Thread immer den zuletzt eingefügten Task aus seiner Task Queue nimmt, kann es passieren, dass ein Actor über einen längeren Zeitraum in der Task Queue wartet, während andere Actors wiederholt ausgeführt werden. Für solche Fälle stellt der ForkJoinPool deshalb einen Parameter, asyncMode, zur Verfügung. Wird der gesetzt, arbeitet jeder Worker Thread die lokale Task-Queue in der Reihenfolge der Einplanung ab. Er riskiert damit zwar Konkurrenzsituationen mit anderen Threads, sorgt aber für mehr Gerechtigkeit. In den gezeigten Beispielen ist der Overhead durch den asyncMode relativ gering, sodass er sich zugunsten der Fairness lohnt.

Der asyncMode hilft leider nicht bei neuen Tasks, die von außerhalb eingeplant werden. Solche Tasks können in der Eingangs-Queue "verhungern", wenn die Worker Threads laufend neue Tasks für sich selbst generieren. Nicht zuletzt aufgrund dieses Problems verwenden aktuelle Weiterentwicklungen des ForkJoinPool keine zentrale Eingangs-Queue mehr, sondern verteilen auch die von außerhalb eingeplanten Tasks direkt an die Task Queues der Threads. Den aktuellen Stand der Implementierung kann man von Doug Leas Concurrency JSR-166 Interest Site herunterladen und ausprobieren. Dort finden sich zudem verschiedene Klassen, die nicht in Java 7 aufgenommen wurden oder erst für Java 8 geplant sind.