Java Concurrency with Jetlang Actors

Last week, I wrote about my experiments with Kilim, a Java actor framework. This week, I take my same toy example and rewrite it using Jetlang, another Java actor framework. Jetlang is a Java port of the Retlang actor framework for C#/.NET, which is inspired by Erlang actors. From what I can see from reading about Retlang, Jetlang is still missing some functionality, but its still in its early stages (version 0.1.6 at the time of writing this).

Like Kilim, concurrency in Jetlang is made safer using a shared-nothing approach. Unlike Kilim, message communication happens using publish-subscribe. Actor threads are modeled as Fibers and Mailboxes are modeled as Channels. Actors sign up to receive events from Channels using predefined callbacks such as onReceive(), onStop() etc, and they perform an action defined by act() in these callbacks. An actor would publish to a Channel within its act() method.

If you have read my previous post, and the previous paragraph, there is really not that much to the code. Like my Kilim example, a lot of it is adapted from the PingPong example in either distribution. There is an ActorManager which is responsible for setting everything up, which you can see below:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74

// Source: src/main/java/com/mycompany/myapp/concurrent/jetlang/ActorManager.java package com.mycompany.myapp.concurrent.jetlang; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.jetlang.channels.Channel; import org.jetlang.channels.MemoryChannel; import org.jetlang.core.Disposable; import org.jetlang.fibers.Fiber; import org.jetlang.fibers.PoolFiberFactory; import com.mycompany.myapp.concurrent.kilim.Message; public class ActorManager { public final static int NUM_ACTORS = 3; public final static String STOP = "__STOP__"; public static void main(String[] args) { ExecutorService exec = Executors.newCachedThreadPool(); PoolFiberFactory factory = new PoolFiberFactory(exec); // when the poison pill is received, the fiber.dispose() call will // call this and decrement the countdown latch. The onstop.await() // will block until the latch is zero, so that way the manager waits // for all the actors to complete before exiting final CountDownLatch onstop = new CountDownLatch(NUM_ACTORS); Disposable dispose = new Disposable() { public void dispose() { onstop.countDown(); } }; Fiber downloadFiber = factory.create(); downloadFiber.add(dispose); DownloadActor downloadActor = new DownloadActor(Channels.downloadChannel, Channels.indexChannel, downloadFiber); Fiber indexFiber = factory.create(); indexFiber.add(dispose); IndexActor indexActor = new IndexActor(Channels.indexChannel, Channels.writeChannel, indexFiber); Fiber writeFiber = factory.create(); writeFiber.add(dispose); WriteActor writeActor = new WriteActor(Channels.writeChannel, (Channel<Message>) null, writeFiber); downloadActor.start(); indexActor.start(); writeActor.start(); // seed the incoming channel with 10,000 requests for (int i = 0; i < 10000; i++) { String payload = "Requested " + i; System.out.println(payload); Channels.downloadChannel.publish(new Message(i, payload)); } // send the poison pill to stop processing Channels.downloadChannel.publish(new Message(0, ActorManager.STOP)); try { onstop.await(); } catch (InterruptedException e) { throw new RuntimeException(e); } exec.shutdown(); } }

To stop the ActorManager, we rely on a similar poison pill approach that I used for the Kilim example. Each actor gets the poison pill, places it on its outbox and then terminates its Fiber by calling dispose(). This triggers off an onDispose() event which is decrements the count on the onstop CountDownLatch. The ActorManager waits on the CountDownLatch to terminate.

The Channels class is simply a holder class that holds final static instances of the channels. There is no difference in my case in having it here from instantiating them within the ActorManager.main(), but it looked like a good way to keep them by themselves since they are our "shared" resources here.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

// Source: src/main/java/com/mycompany/myapp/concurrent/jetlang/Channels.java package com.mycompany.myapp.concurrent.jetlang; import org.jetlang.channels.Channel; import org.jetlang.channels.MemoryChannel; import com.mycompany.myapp.concurrent.kilim.Message; public class Channels { public static final Channel<Message> downloadChannel = new MemoryChannel<Message>(); public static final Channel<Message> indexChannel = new MemoryChannel<Message>(); public static final Channel<Message> writeChannel = new MemoryChannel<Message>(); }

Next up is an abstract Actor class built out of a Fiber and Channels representing the Actors inbox and outbox. There is an abstract act() method that application specific Actor subclasses will implement.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

// Source: src/main/java/com/mycompany/myapp/concurrent/jetlang/Actor.java package com.mycompany.myapp.concurrent.jetlang; import org.jetlang.channels.Channel; import org.jetlang.core.Callback; import org.jetlang.fibers.Fiber; import com.mycompany.myapp.concurrent.kilim.Message; public abstract class Actor { private Channel<Message> inChannel; private Channel<Message> outChannel; private Fiber fiber; public Actor(Channel<Message> inChannel, Channel<Message> outChannel, Fiber fiber) { this.inChannel = inChannel; this.outChannel = outChannel; this.fiber = fiber; } public void start() { // set up subscription listener Callback<Message> onRecieve = new Callback<Message>() { public void onMessage(Message message) { act(message); if (outChannel != null) { outChannel.publish(message); } // process poison pill, dispose current actor and pass the message // on to the next actor in the chain (above) if (message.payload instanceof String && ActorManager.STOP.equals(message.payload)) { fiber.dispose(); } } }; // subscribe to incoming channel inChannel.subscribe(fiber, onRecieve); // start the fiber fiber.start(); } public abstract void act(Message message); }

As before, the three actors are pretty trivial and pointless, so I just put them in here without much explanation.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

// Source: src/main/java/com/mycompany/myapp/concurrent/jetlang/DownloadActor.java package com.mycompany.myapp.concurrent.jetlang; import org.jetlang.channels.Channel; import org.jetlang.fibers.Fiber; import com.mycompany.myapp.concurrent.kilim.Message; public class DownloadActor extends Actor { public DownloadActor(Channel<Message> inChannel, Channel<Message> outChannel, Fiber fiber) { super(inChannel, outChannel, fiber); } @Override public void act(Message message) { String payload = (String) message.payload; payload = payload.replaceFirst("Requested ", "Downloaded "); System.out.println(payload); message.payload = payload; } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

// Source: src/main/java/com/mycompany/myapp/concurrent/jetlang/IndexActor.java package com.mycompany.myapp.concurrent.jetlang; import org.jetlang.channels.Channel; import org.jetlang.fibers.Fiber; import com.mycompany.myapp.concurrent.kilim.Message; public class IndexActor extends Actor { public IndexActor(Channel<Message> inChannel, Channel<Message> outChannel, Fiber fiber) { super(inChannel, outChannel, fiber); } @Override public void act(Message message) { String payload = (String) message.payload; payload = payload.replaceFirst("Downloaded ", "Indexed "); System.out.println(payload); message.payload = payload; } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

// Source: src/main/java/com/mycompany/myapp/concurrent/jetlang/WriteActor.java package com.mycompany.myapp.concurrent.jetlang; import org.jetlang.channels.Channel; import org.jetlang.fibers.Fiber; import com.mycompany.myapp.concurrent.kilim.Message; public class WriteActor extends Actor { public WriteActor(Channel<Message> inChannel, Channel<Message> outChannel, Fiber fiber) { super(inChannel, null, fiber); } @Override public void act(Message message) { String payload = (String) message.payload; payload = payload.replaceFirst("Indexed ", "Wrote "); System.out.println(payload); message.payload = payload; } }

Running the ActorManager produces the expected interleaved output, showing that the Actors are acting asynchronously.

I was curious about the assertion that Events are a bad idea for high-concurrency servers (PDF), so I figured that now I had two trivial versions of my application that did nothing substantial, it would be interesting to see if Kilim's wait-loop approach performed better than Jetlang's pub/sub approach. Here are the numbers - note that the times are just wall-clock times taken under very uncontrolled situations. The red line is for Kilim numbers and the green line is Jetlang's. So it looks like (at least in my test) the message passing overhead for Jetlang is slightly higher than Kilim, and increases with higher loads. However, what is an acceptable overhead for message passing would depend a lot on the application - it would probably be less important as the running times of the actors increases.

#-Tasks Kilim Jetlang 1 9 10 10 26 49 100 71 105 1000 605 630 10000 4648 5997 100000 37076 46426

Update - 2009-01-02

The performance numbers shown above are skewed due to the presence of Console IO in the code. While it is likely that real-world actors will do something more meaningful than replacing one string with another, possibly resulting in blocking similar to the System.out.println() calls above, we want to compare the frameworks themselves. Numbers with println() calls removed are provided in my post here.

Personally, now that I understand Jetlang a bit more than I did last week, here is some feedback. I found Jetlang initially slightly harder to understand than Kilim, but using it was simpler, mainly because there is no bytecode manipulation. I haven't used Jetlang enough to praise/blame it from a programming perspective, but here is a review of Retlang from a C#/Python developer that may be useful. Mike Rettig (the (original) author of Retlang and Jetlang) also posts interesting information on Jetlang in his blog.

Merry Christmas, everyone! Have a great holiday season.