benjchristensen/RxJava — pull request #9 by akarnokd (2014-12-03 16:20:25+00:00)
Title: Two-phase atomic array queue for RxRingBuffer.

Hi Ben!

I've spent some time understanding the situation and I came up with a solution that might just work: an atomic array queue which starts out with a small capacity and then switches to a large capacity in case of overfilling or due to high traffic.

The problem with the default SpscLinkedQueue and SpscArrayQueue is that they consume a lot of memory due to their fixed size and generous padding, which translates to more frequent GCs in the case of short lived queueing.

At first I thought, let the queue size double on overfill (up to a maximum), but even if I managed to figure out the correct algorithm, it ran pretty slow regardless to the size. The problem was not the resizing but the expensive CAS on every poll() to detect a potential ongoing resize operation.

The solution I came up with does this resize only once: it switches to high gear in case of overfill or a certain number of items have passed through the queue. Once in high gear, it can now use much cheaper atomic operation and thus increase throughput. Maybe it should be called RxJava's weak queue hypothesis.

The usual benchmark gives nice values: (spmc is in this case the AtomicArrayQueueUnsafe2, 1000 is actually RxRingBuffer.SIZE).
```
Benchmark Mode Samples Score Score error Units
r.i.RxRingBufferPerf.spmcCreateUseAndDestroy1 thrpt 5 27493478,930 885408,249 ops/s
r.i.RxRingBufferPerf.spmcCreateUseAndDestroy1000 thrpt 5 320608,155 4635,395 ops/s
r.i.RxRingBufferPerf.spmcRingBufferAddRemove1 thrpt 5 44589735,680 856811,156 ops/s
r.i.RxRingBufferPerf.spmcRingBufferAddRemove1000 thrpt 5 353921,131 7072,580 ops/s
r.i.RxRingBufferPerf.spscCreateUseAndDestroy1 thrpt 5 10842470,917 128615,881 ops/s
r.i.RxRingBufferPerf.spscCreateUseAndDestroy1000 thrpt 5 324563,384 2101,994 ops/s
r.i.RxRingBufferPerf.spscRingBufferAddRemove1 thrpt 5 42179656,259 723832,236 ops/s
r.i.RxRingBufferPerf.spscRingBufferAddRemove1000 thrpt 5 328218,479 29409,408 ops/s
```

If the queue stayed in its initial capacity, the ```spmcRingBufferAddRemove1``` would give around 32-36 Mops/s because of the more expensive atomic operations.

Few other notes:
- This rather feels a bit cheating the perf test.
- Since AtomicArrayQueueUnsafe2 doesn't pads its fields or the buffer, it may be vulnerable to false sharing and thus can't perform as well as an SpscArrayQueue. (This)[https://gist.github.com/akarnokd/d561a92481062c94d792] gist shows some raw offer/poll throughput.
- Maybe the high-gear could be a full SpscArrayQueue and AAQU2 just shoveling over to it during resize. However, I'm not sure if the volatile dispatch (i.e., yet another volatile field telling which queue implementation to call) in offer()/poll() makes it worth.