One of the blogs I follow is Rudiger's blog - in particular
this article with a light benchmark I find interesting as it shows the clear performance advantage of the Disruptor, and also just how slow the out of the box services can be for particular use cases.
In the article, Rudiger tests out a number of different implementations of a pi calculator, and finds quickly that 1million small tasks is a nice way to put enough pressure on the concurrency frameworks to split out a winner. The out of the box java Executor service is actually relatively slow, coming in last position.
One of the items I have on my to-do list is actually to build a fast ExecutorService implementation, and fortunately after a busy few months I've had a chance to sit down and write it. For those new to concurrency on Java, the ExecutorService interface and the Executors factory methods provide an easy way to get into launching threads and handling off composable tasks to other threads for asynchronous work.
The work below borrows heavily from Rudiger's work but also the work at
1024cores.net, and of course I can't neglect to mention
Nitsan Wakart's blog and
Martin Thompson's numerous talks on the topic.
The target: Get a working ExecutorService that can be used in the Pi calculator and get a result that is better than out-of-the-box.
Getting a baseline
Of course, since my target is to beat the out-of-the-box frameworks, I spent some time downloading the gists and hooking up the libs in maven. Only thing is I didn't hook up the Abstraktor tests as I am not too familiar with the project (happy to take a pull request).
Interestingly, the out of the box services perform pretty much as described in Rudiger's post, with the caveat that disruptor services have a much sharper degradation in performance on this run. Worth mentioning is that this is only a 2-socket 8-core E5410's (the old Harpertown series), so not nearly as new as Rudiger's kit.
First cut
I've heard good things about the LinkedTransferQueue but not had a need to test it out. As a first cut, I decided to write up a simple Executor implementation that as part of the submit() call.
private static class LtqExecutor<T> implements ExecutorService {
private volatile boolean running;
private volatile boolean stopped;
private final int threadCount;
private final BlockingQueue<Runnable> queue;
private final Thread[] threads;
public LtqExecutor(int threadCount) {
this.threadCount = threadCount;
this.queue = new LinkedTransferQueue();
running = true;
stopped = false;
threads = new Thread[threadCount];
for(int i = 0; i < threadCount; i++) {
threads[i] = new Thread(new Worker());
threads[i].start();
}
}
public void execute(Runnable runnable) {
try {
queue.put(runnable);
} catch (InterruptedException ie) {
throw new RuntimeException(ie);
}
}
private class Worker implements Runnable {
public void run() {
while(running) {
Runnable runnable = null;
try {
runnable = queue.take();
} catch (InterruptedException ie) {
// was interrupted - just go round the loop again,
// if running = false will cause it to exit
}
try {
if (runnable != null) {
runnable.run();
}
} catch (Exception e) {
System.out.println("failed because of: " + e.toString());
e.printStackTrace();
}
}
}
}
I also removed the test for getQueue.size(). Actually, performance of this was rather impressive on its own, achieving results in 1/3rd the time of the ootb services, while of course it comes nowhere near Disruptor.
| threads |
disruptor2 |
threadpi |
custom / LTQ |
| 1 |
954 |
1455 |
1111 |
| 2 |
498 |
1473 |
1318 |
| 3 |
327 |
2108 |
802 |
| 4 |
251 |
2135 |
762 |
| 5 |
203 |
2195 |
643 |
| 6 |
170 |
2167 |
581 |
| 7 |
7502 |
2259 |
653 |
| 8 |
9469 |
2444 |
5238 |
Trying some things
ArrayBlockingQueue and SynchronousQueue
I also tried substituting ABQ and SQ instead of LinkedTransferQueue, since my understanding is LTQ has some additional unnecessary overheads. Perhaps ABQ and SQ would be a little more cache friendly since ABQ is possibly better laid out in memory, and SQ as I've used it implementations before.
| threads |
disruptor2 |
threadpi |
custom / LTQ |
custom / SQ |
custom / ABQ 1000 |
| 1 |
954 |
1455 |
1111 |
3738 |
2570 |
| 2 |
498 |
1473 |
1318 |
2638 |
2570 |
| 3 |
327 |
2108 |
802 |
3925 |
1875 |
| 4 |
251 |
2135 |
762 |
4035 |
1792 |
| 5 |
203 |
2195 |
643 |
4740 |
1692 |
| 6 |
170 |
2167 |
581 |
4635 |
1684 |
| 7 |
7502 |
2259 |
653 |
4362 |
1668 |
| 8 |
9469 |
2444 |
5238 |
4566 |
1677 |
Ouch. No need to mention these again.
Striping the LTQ
Clearly Disruptor's approach of limiting contended writes has a benefit. Would the LinkedTransferQueue get a little faster by striping the input queues - one per thread?
In this model, instead of having one queue into the Executor, each thread has its own thread, and the put() method round-robins across the threads.
Actually, this was a bit better for some workloads but with much faster degradation.
| threads |
disruptor2 |
threadpi |
custom / LTQ |
Custom / striped LTQ |
| 1 |
954 |
1455 |
1111 |
1237 |
| 2 |
498 |
1473 |
1318 |
1127 |
| 3 |
327 |
2108 |
802 |
713 |
| 4 |
251 |
2135 |
762 |
570 |
| 5 |
203 |
2195 |
643 |
498 |
| 6 |
170 |
2167 |
581 |
386 |
| 7 |
7502 |
2259 |
653 |
5264 |
| 8 |
9469 |
2444 |
5238 |
5262 |
Customising a queue
A ringbuffer based queue
With a baseline set, I knew from reading Nitsan's many posts and the 1024cores.net site, I new there were some optimisations I could get from the problem space. What about using a queue with a ringbuffer?
By introducing a custom lock-free queue and substituting it in my Executor logic, using a put().
public void put(Runnable r) throws InterruptedException {
while (true) {
long toWrite = nextSlotToWrite.get();
if (toWrite < (lastConsumed.get() + capacity)) {
buffer[(int) (toWrite % capacity)] = r;
nextSlotToWrite.incrementAndGet();
break;
} else {
// no space in queue, sleep
LockSupport.parkNanos(1);
}
}
}
While the take() looks like this.
public Runnable take() throws InterruptedException {
while(!Thread.interrupted()) {
long lastFilled = nextSlotToWrite.get() - 1;
long lastConsumedCache = lastConsumed.get();
if (lastConsumedCache < lastFilled) {
// fetch the value from buffer first, to avoid a wrap race
int bufferslot = (int) ((lastConsumedCache + 1) % capacity);
Runnable r = buffer[bufferslot];
// attempt to cas
boolean casWon = lastConsumed.compareAndSet(
lastConsumedCache, lastConsumedCache + 1);
if (casWon) {
return r;
} else {
}
} else {
// cas failed, do not sleep
}
}
throw new InterruptedException();
}
Can we go faster by SPMC?
In particular, for our case, there is only ever a single writer thread issuing put()s to the queue. Can we take advantage of this by tweaking making the queue an SPMC queue (single producer many consumer)?
public void put(Runnable r) throws InterruptedException {
while (true) {
if (toWrite < (lastConsumed.get() + capacity)) {
buffer[(int) (toWrite & REMAINDER_MASK)] = r;
toWrite += 1;
nextSlotToWrite.lazySet(toWrite);
break;
} else {
// no space in queue, sleep
LockSupport.parkNanos(1);
}
}
}
Note that I use a lazySet() to write, but this doesn't help, I'm speculating that it's really the .get()'s in the below that are causing the issues.
public Runnable take() throws InterruptedException {
while(!Thread.interrupted()) {
long lastFilled = nextSlotToWrite.get() - 1;
long lastConsumedCache = lastConsumed.get();
if (lastConsumedCache < lastFilled) {
// fetch the value from buffer first, to avoid a wrap race
int bufferslot = (int) ((lastConsumedCache + 1) & REMAINDER_MASK);
Runnable r = buffer[bufferslot];
// attempt to cas
boolean casWon = lastConsumed.compareAndSet(
lastConsumedCache, lastConsumedCache + 1);
if (casWon) {
return r;
} else {
// if cas failed, then we are not the owner so try again
}
} else {
// no values available to take, wait a while
LockSupport.parkNanos(1);
}
}
throw new InterruptedException();
}
By this point I've also started to use 2^ sized queues (following the disruptor guidance on fast modulo). Actually, this didn't make a lot of difference but I've left it in.
Caching the get()s
How can I reduce contention when things are busy? Having a look at the take() method we can see that there are two .get() calls to atomic fields, this will be causing volatile reads from memory. On CAS failure, it's not always the case that we need to re-read the variable. So what happens if we move that to only happen during the "things are too busy" loop?
public Runnable take() throws InterruptedException {
long lastFilled = nextSlotToWrite.get() - 1;
long lastConsumedCache = lastConsumed.get();
while(!Thread.interrupted()) {
if (lastConsumedCache < lastFilled) {
// fetch the value from buffer first, to avoid a wrap race
int bufferslot = (int) ((lastConsumedCache + 1) & REMAINDER_MASK);
Runnable r = buffer[bufferslot];
// attempt to cas
boolean casWon = lastConsumed.compareAndSet(
lastConsumedCache, lastConsumedCache + 1);
if (casWon) {
return r;
} else {
// if cas failed, then we are not the owner so try again
// LockSupport.parkNanos(1);
// note that the lastFilled does not need to change just yet
// but probably the lastConsumed did
lastConsumedCache = lastConsumed.get();
}
} else {
// no values available to take, wait a while
LockSupport.parkNanos(1);
lastFilled = nextSlotToWrite.get() - 1;
// lastConsumedCache = lastConsumed.get();
}
}
throw new InterruptedException();
}
}
| threads |
akkapi |
disruptor |
disruptor2 |
threadpi |
custom / LTQ |
Custom Lockfree Pp Tp Padded |
Custom with cached gets |
| 1 |
2389 |
1300 |
954 |
1455 |
1111 |
1245 |
1257 |
| 2 |
1477 |
713 |
498 |
1473 |
1318 |
992 |
1007 |
| 3 |
1266 |
492 |
327 |
2108 |
802 |
676 |
695 |
| 4 |
1629 |
423 |
251 |
2135 |
762 |
548 |
559 |
| 5 |
1528 |
400 |
203 |
2195 |
643 |
538 |
515 |
| 6 |
1541 |
398 |
170 |
2167 |
581 |
545 |
510 |
| 7 |
1482 |
1085 |
7502 |
2259 |
653 |
603 |
562 |
| 8 |
1397 |
1297 |
9469 |
2444 |
5238 |
925 |
883 |
Not a lot of difference, but it does help our under-contention numbers, which is exactly what we'd expect.
The results
Actually, the results were good, I was pretty happy with them. They're a small fraction faster than the LinkedTransferQueue with what seems like better worst case performance. Nowhere near as fast as the Disruptor but frankly that's not surprising, given the energy and talent working on it.
Of course, this is not Production ready, and YMMV but I would welcome comments and questions. Particularly if there are any issues with correctness.
Code samples on github.
