Re: Disruptor vs ArrayBlockingQueue tests. Disruptor is better only for one very limited case.

[Michael Barker]

Have you tried the latest, 3.0.0.beta2.

The biggest change to the way that the MultiProducerPublisher handles
contention.

WRT the 100% CPU. The only way to avoid the high CPU overhead is to
park the thread in some way, either for a short period of time waking
frequently to check for progress (SleepingWaitStrategy) or waiting on
a conditional variable for notification (BlockingWaitStrategy).

Mike.

On Sat, Jan 19, 2013 at 4:35 AM, <gobe...@msn.com> wrote:
> Hello,
> I am new to Disruptor, so apologize for my naive questions.
>
> Hello,
> I am new to Disruptor, so apologize for my naive questions.
>
> A little background: We are a financial market data vendor. We have market
> data API that processes and delivers market data messages to the clients.
> The API heavily currently uses ArrayBlockingQueue to exchange messages
> between threads. All these exchanges involve single consumer and multiple
> producers.
> I started to research Disruptor and was excited about replacing
> ArrayBlockingQueue with ArrayBlockingQueue, but have disappointing results.
> It looks to me that Disruptor is a better choice than ArrayBlockingQueue for
> a very limited case: one producer and one consumer with busy wait consumer.
> For all other cases I tried it was actually worse as far as performance and
> latency:
>
> All these tests involve pumping millions of messages between producer and
> consumer and timing completion.
> 1) I first tried testing Disruptor vs ArrayBlockingQueue using 'official'
> example: one consumer and one producer (SingleThreadedClaimStrategy/
> SleepingWaitStrategy):
> http://code.google.com/p/disruptor/wiki/CodeExampleDisruptor2x. It was 3
> times faster as ArrayBlockingQueue. So far so good. But upon further
> investigation I found that I cannot use SleepingWaitStrategy because
> "Latency spikes can occur after quiet periods.". Translation: if you do not
> have messages for a while and a message arrives, you would have a latency
> spike, that is unacceptable. I also cannot use SingleThreadedClaimStrategy
> because I have more than one producer.
> 2) I then tried MultiThreadedLowContentionClaimStrategy. And this is where I
> believe I found some sort of bug or limitation: if I have 25 threads
> producing about 200 m/s cumulative, the test runs fine for a while, but then
> the Disruptor goes into some state where all cores have 100% CPU, and
> application grinds to a halt. Bug or not, I cannot use it.
> 3) So the only acceptable combination is MultiThreadedClaimStrategy and
> BusySpinWaitStrategy. This time it meets all requirements, but was actually
> _slower_ that ArrayBlockingQueue, not to mention that one core was at 100%.
>
> Do these observation seem reasonable? I am at a loss. Any help will be
> appreciated.
>
> --
>
>

[gobe...@msn.com]

Mike, is there a migration guide for Disruptor 3? I have compile errors on almost every line. Also, it would be nice if you uploaded 3.0.0.beta2, I see only beta 1 in downloads.

I have tweaked ArrayBlockingQueue by adding takeAll and putAll bulk operation. If I use putAll (with 100 messages in each call) and takeAll, ArrayBlockingQueue outperforms all Disruptor 2 configurations with the exception of the SingleThreadedClaimStrategy/ SleepingWaitStrategy (that is useless for me).

I can send you the test if you want. These are the numbers I am getting:

Process 10000000 messages with Desruptor using SingleThreadedClaimStrategy/ SleepingWaitStrategy.

Test took: 0.296

Process 10000000 messages with Desruptor using MultiThreadedClaimStrategy/ BusySpinWaitStrategy.

Test took: 1.008

Process 10000000 messages with Desruptor using MultiThreadedClaimStrategy/ SleepingWaitStrategy.

Test took: 0.725

Process 10000000 messages with Desruptor using MultiThreadedClaimStrategy/ BlockingWaitStrategy.

Test took: 1.408

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll and put.

Test took: 0.929

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll and bulk putAll.

Test took: 0.270

[Michael Barker]

> Mike, is there a migration guide for Disruptor 3? I have compile errors on
> almost every line. Also, it would be nice if you uploaded 3.0.0.beta2, I see
> only beta 1 in downloads.

It's in Maven central. Wiki is now updated.

We haven't done a migration guide yet. The claim strategies are
replaced with a ProducerType enum. The RingBuffer.get method is
replaced with a RingBuffer.getPreallocated and RingBuffer.getPublished
for producers and consumers respectively.

In the interim you could look at the differences between the
performance test for the 2 versions.

> I can send you the test if you want. These are the numbers I am getting:

Please send it through. I'm fairly confident that the MultiProducer
changes will improve things. I'll do the port to Disruptor 3.0.

You mention running 25 producer threads. How many CPU cores do you
have available?

Mike.

[gobe...@msn.com]

Mike, I am attaching 2 files. One of them is modified JDK ArrayBlockingQueue with takeAll, putAll added and the other one is test. All tests have one consumer and one producer. My machine is 4 core windows using 64 bit Java 6.

I did not attach the the test with 25 producers using MultiThreadedLowContentionClaimStrategy (the one that melted my app) - not sure if you wanted it too or not.

If you want to port it to Disruptor 3, I would appreciate it. Thanks for the prompt replies!

[Jeff Hain]

Hi

>Mike, I am attaching 2 files.
>One of them is modified JDK ArrayBlockingQueue with takeAll,
>putAll added and the other one is test. All tests have one
>consumer and one producer. My machine is 4 core windows using
>64 bit Java 6.

I adapted your tests for Disruptor from today's GitHub, and added some stuffs
as well (see below, and attached files).


Before each test, "consumed" needs to be reset to 0, and a fresh latch needs to be created too.
That problem has no big impact on measures since we have "MESSAGES" >> "RING_SIZE",
and publishing is done in current thread, so at worse we measured time to publish and process
MESSAGES-RING_SIZE events.


I got better results after replacing, in MultiProducerSequencer,
LockSupport.parkNanos(1)
with
Thread.yield().
(This wait is done when ring buffer is full.)
This change might be less convenient for real-world cases, for it might not free as much CPU
to help workers make progress, but it helps a lot with quick-benches like this, where producers
go as fast or faster than processors.


Yet, Disruptor was still behind the ABQ in the tests, for multi-publisher case.
For testing against ABQ's "takeAll and put" case, I think it's fair (and not completely
surprising, since Disruptor is primarily latency-oriented for real-world scenarios,
and these tests measure throughput with "while true" kind of publishing).
But for testing against ABQ's "takeAll and bulk putAll" case, a fair way would be
not to put events one by one in the ring buffer, but to use lists of events as
Disruptor events, and to put 100 events at once.
Another fair way would be to use a sequence batch claiming API, but it's no longer
available in latest Disruptor, I suspect because an obvious work-around for it is to
use a list (or array) of events as just said.


I also ran your tests on an implementation of ring buffer I made (in http://code.google.com/p/jodk),
and the tests run faster with it than with the ABQ, especially when using bulk publishing.
It doesn't have the same API than Disruptor though so it might be less convenient for
some problems, but it uses the same basic idea.


-Jeff


PS: I put myself as destinatary else I suspect file attachment might not work
(already had that kind (and other kinds) of trouble with yahoo mail).

[Michael Barker]

So the interesting case is where single put, bulk take is faster than
the multi-producer. I think I know what is going on here, but I need
to do some further investigation.

For the bulk put/get case, as Jeff mentioned, if we implemented the
Disruptor using a batch stored as each element within the ring buffer
the performance is pretty good, generally over an order of magnitude
faster than the ArrayBlockingQueueEx (numbers below). However,
there's a couple of reasons why I don't like the bulk put case. In a
real system I think it is only useful for the single producer case.
In a multi-producer case it can cause head-of-line blocking issues,
where a large batch from one producer can starve the other producers
of resources. For the latency case, it is often better to allow
messages to interleave in order to get fairer distribution for
messages from all producers. Obviously with a single producer then
you don't have that problem.

The other thing with the test that I feel that 100 is an unusually
large size for a batch. For example if you are processing Ethernet
packets off the wire (and latency is important) then you'll probably
dispatching batches for each Ethernet frame. Given a maximum frame
size of 1500 bytes, then you're looking at an average message size of
15 bytes, which seems a little too small, also you'll probably have
less as there will be protocol overhead, so that that's an average
message size smaller than an Itch order delete. The original post
mentioned 25 producers running at 200 msg/sec. That would mean
potentially waiting on average 0.5 seconds to fill a batch that size
(if you wanted a Nagle's style algorithm). If you reduce the batch
size to something more realistic like 10, then the bulk put case is
only around twice as fast as the single put.

This is one of the other reasons that I removed bulk puts from version
3.0 of the Disruptor. They make benchmarks look really good, but are
probably not the best solution in real applications.

Mike.

--- Number of publishers = 1

Process 10000000 messages with Disruptor with bulk puts using
ProducerType.MULTI / BusySpin.
Test took: 0.025
Test took: 0.008
Test took: 0.008
Test took: 0.008

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 0.324
Test took: 0.268
Test took: 0.222
Test took: 0.248


--- Number of publishers = 2

Process 10000000 messages with Disruptor with bulk puts using
ProducerType.MULTI / BusySpin.
Test took: 0.016
Test took: 0.012
Test took: 0.016
Test took: 0.015

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 0.400
Test took: 0.390
Test took: 0.406
Test took: 0.353


--- Number of publishers = 4

Process 10000000 messages with Disruptor with bulk puts using
ProducerType.MULTI / BusySpin.
Test took: 0.034
Test took: 0.031
Test took: 0.031
Test took: 0.030

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 0.733
Test took: 0.710
Test took: 0.593
Test took: 0.681


--- Number of publishers = 8

Process 10000000 messages with Disruptor with bulk puts using
ProducerType.MULTI / BusySpin.
Test took: 0.067
Test took: 0.087
Test took: 0.076
Test took: 0.071

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 1.131
Test took: 1.161
Test took: 1.153
Test took: 1.143


--- Number of publishers = 16

Process 10000000 messages with Disruptor with bulk puts using
ProducerType.MULTI / BusySpin.
Test took: 0.145
Test took: 0.155
Test took: 0.176
Test took: 0.169

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 2.191
Test took: 2.176
Test took: 2.222
Test took: 2.179

> --
>
>

[gobe...@msn.com]

Jeff, Mike, 
I am going to reply to both of you. Jeff, thanks for taking look at the test and fixing my sloppy bugs.

Now a little more background on the API we develop. First of all it is very real. It is used by hundreds of clients. It reads custom binary protocol from a socket, parses it and dispatches to ArrayBlockingQueues with consumers running on different thread(s).

The typical scenarios involves the user subscribing for updates. These updates flow may be very heavy. These are processed by a single consumer. There is a 'main' producer doing heavy tcp reads, publishing, but occasionally there are may be light publications from other threads.

The number of 100 I picked for bulk put is also very real. This is what I observed when subscribing and monitoring the typical number of messages parsed/ processed. So this is the number I get from a typical TCP read. TCP buffering will result in reads much larger that 1,500. Also, UDP packets we receive in our UDP API are way larger than 1,500. This is why bulkPut is important. Using lists to implement bulkPut is possible, but not practical. If ring buffer size is large, it will be expensive to create a lot of lists. Also, you would making reusing Events difficult, (clear each list each time? GC hit...).

Starving other producers is interesting one. I would think that pulkPut would make things better, not worse by the virtue of less locks, at least for the ArrayBlockingQueue, but maybe I am wrong. ABQ also has fair/ not fair publication, but fair publication is much slower.

Jeff, would you care to explain why JODK is so much faster than Disruptor? What is the secret? :)

Thanks

[Michael Barker]

It is interesting to hear your use case. You find as you speed up
your application the size of data available starts to decrease...

I have a couple more questions. What data are you storing in the ring
buffer/queue? Is it an object or just the bytes from the network? Do
you have a size histogram for your messages? How many physical
network devices are you connected to? What are your actual
latency/throughput goals? How latency sensitive are the light
publishing threads?

For the optimal throughput and latency I would work really had to move
to a single-producer solution. If the lightly producing threads can
tolerate the latency hit I would look to round trip there events over
the network, just to avoid have to go multi-producer. In fact, at
LMAX we've designed our physical network to allow us to have only one
producer. You've seen from the benchmarks how much quicker that is.
There is also a little trick that can be played with the single
producer disruptor case that can give you publisher side batching for
no extra space cost. Also the head-of-line blocking problem (for the
concurrent structure) goes away.

For the multi-producer case I could suggest a couple of options, but
I'd need to know the questions

> important. Using lists to implement bulkPut is possible, but not practical.
> If ring buffer size is large, it will be expensive to create a lot of lists.
> Also, you would making reusing Events difficult, (clear each list each time?
> GC hit...).

True, but I was just demonstrating the principle. There are a couple
of ways that you could use the Disruptor to do this that wouldn't
involve a GC hit. The other possibility is something more like Peter
Lawrey's Chronicle library, although I think that is single producer
only.

> Starving other producers is interesting one. I would think that pulkPut
> would make things better, not worse by the virtue of less locks, at least
> for the ArrayBlockingQueue, but maybe I am wrong. ABQ also has fair/ not
> fair publication, but fair publication is much slower.

It depends on what way you want to trade off. I think there is a
strong latency v throughput trade off here. I'm in the process of
putting together a latency benchmark for multiple producers (we only
have a single producer one ATM).

> Jeff, would you care to explain why JODK is so much faster than Disruptor?
> What is the secret? :)

I would like to know too :-).

Mike.

[Jeff Hain]

>Jeff, would you care to explain why JODK is so much faster than Disruptor? What is the secret? :)

A few observations and then description of a problem I found in my code (unrelated to these tests).


One main difference is that in Disruptor, processors wait with the strategy for cursor
(i.e. max claimed sequence, and possibly under the name of "dependentSequence")
to be >= sequence to process, and then busy-spin (ensureAvailable) until they see it
published, whereas in MulticastRingBuffer workers (processors) wait "directly", with
the condilock (strategy), for sequence to process to be visibly published.


One thing that I don't understand is that the wait strategy is not signaled just after cursor
update, but later, after "availableBuffer" has been set (in publish method).
If that's to lower the risk of too-early busy-spin, then why not make processors
just wait for the sequence to be visibly published instead, without bothering with cursor?


Another difference (linked with the previous one) is that I don't use an "availableBuffer"
array, but an array of padded AtomicLong, in which the published sequence is set
(lazily in the tests, due to writeLazySets = true).
Maybe the padding helps at some point (would need to check that).
This array is also (other than in regular usage) used for shut downs in case of ring buffer
services, where CASes on sequences are used for rejections.


One other reason why MulticastRingBuffer can be faster, is that user has total control
over the wait stragegies (busy/yield/sleep/lock/etc.) through use of InterfaceCondilock,
whether it's for waiting for subscribers to make progress (readWaitCondilock), or for
publishers to publish (writeWaitCondilock).
There is no hard-coded busy-spin or parkNanos.
It can help a lot when being "in the wild" (with furious applications running around).
(I remember discussing allowing this kind of flexibility for Disruptor, with Martin, but if I recall he
was worried about people using fancy strategies and then ask for support, which I understand.)
Also, in the tests I used condilocks implementations that are good for many usages, like pure
throughput as well as quickly relieving CPU pressure if no more event (see Condilocks class).


Trying to remember (for MulticastRingBuffer), figure out (for Disruptor), and formulate how things
work, I seem to see a potential problem in MulticastRingBuffer, at least something that's not supposed
to work according to what I understand of JMM, but that has always worked in my tests (where I use
sequence as event, and always check that read event == processed sequence).
It only happens in case of multiple workers (processors), so has no impact on the tests we were
discussing.
The problem is that when a worker wants to know which is the max published sequence it can process,
instead of figuring it out by monotonically scanning the array of published sequences, it might use a
shared volatile value containing this information computed by another worker. But this other worker
might have seen the published sequence "by chance", for being on the same core than the publisher,
and the fact that another worker on another core sees the updated shared value, doesn't mean that
it can see the published event yet (unless CPU cache is always flushed into main memory all at once).
There is the same problem when a worker only checks a preceding worker's max (visibly) processed
sequence to know up to which sequence it can process.
To correct that, I just removed usage of the shared volatile info, making each worker monotonically
scan the array of published sequences, and also made workers with preceding workers also
check this array (and not only their preceding worker(s)).
I ran RingBuffersPerf again and it's now slower when more workers than core, but nothing terrible.
I'll update my Git repos with these corrections soon.


This problem doesn't arise with UnicastRingBuffer since for this one each sequence is dealt with
individually, and there is no dependency/order between workers.


-Jeff

[gobe...@msn.com]

On Monday, January 21, 2013 1:42:08 AM UTC-6, mikeb01 wrote:

It is interesting to hear your use case.  You find as you speed up
your application the size of data available starts to decrease...

Mike, we repeatedly read data from the socket and do some binary protocol parsing, generating events.

As we pump more messages/s, the size of the read only increases. So I have to disagree with you on this one.

 

I have a couple more questions.  What data are you storing in the ring
buffer/queue?  Is it an object or just the bytes from the network?  Do
you have a size histogram for your messages?  How many physical
network devices are you connected to?  What are your actual
latency/throughput goals?  How latency sensitive are the light
publishing threads?

 

The queue stores objects that wrap byte arrays with couple additional fields. Size is typically around 150 bytes.

Depending on the API, it reads data from a TCP/IP socket or listens to the UDP data stream.

The light publishing threads are mostly used for monitoring, user custom messages, etc. We can put restriction on the API to disable them for the sake of a single publisher, but it has to have strong justification.

The latency/throughput goals depend on the client that uses the API. The clients vary a lot: from GUI desktops to trading servers. So some of them are very latency sensitive  and others are not. So we try to find reasonable performance for all possible clients.

 

For the optimal throughput and latency I would work really had to move
to a single-producer solution.  If the lightly producing threads can
tolerate the latency hit I would look to round trip there events over
the network, just to avoid have to go multi-producer.  In fact, at
LMAX we've designed our physical network to allow us to have only one
producer.  You've seen from the benchmarks how much quicker that is.
There is also a little trick that can be played with the single
producer disruptor case that can give you publisher side batching for
no extra space cost.  Also the head-of-line blocking problem (for the
concurrent structure) goes away.

Yes, I agree. We may need to have a custom API config. But it will be enabled only for some latency sensitive clients that can afford busy wait. I would also be reluctant to use Disruptor until bulk put is better supported.

[Michael Barker]

> Mike, we repeatedly read data from the socket and do some binary protocol
> parsing, generating events.
> As we pump more messages/s, the size of the read only increases. So I have
> to disagree with you on this one.

Okay I must be misunderstanding how your system works.

> The queue stores objects that wrap byte arrays with couple additional
> fields. Size is typically around 150 bytes.
> Depending on the API, it reads data from a TCP/IP socket or listens to the
> UDP data stream.
> The light publishing threads are mostly used for monitoring, user custom
> messages, etc. We can put restriction on the API to disable them for the
> sake of a single publisher, but it has to have strong justification.
> The latency/throughput goals depend on the client that uses the API. The
> clients vary a lot: from GUI desktops to trading servers. So some of them
> are very latency sensitive and others are not. So we try to find reasonable
> performance for all possible clients.

However, you should still have a goal. It's difficult to recommend a
solution without one. I have implemented a very bare-bones bulk-put
API. It is in a branch "bulk-put" on Github, clone if you are
interested in testing it. It is far from final as it doesn't have
unit tests or save EventTranslator wrapper methods. The numbers are
fairly impressive (see below) even with the BlockingWaitStrategy, but
I'm worried that all I've done is optimise for the performance test
and not for your real world use case. Before you make a selection can
I recommend that you implement a test that better reflects the
situation you've outlined above. As well measuring best throughput,
pick your desired throughput e.g. 1M msg/sec and then measure the
latency through the solution and ensure that it is at tolerable
levels. Have a look at
ThrottledOnePublisherToThreeProcessorPipelineLatencyTest as an example
of how we do that for the Disruptor.

> Yes, I agree. We may need to have a custom API config. But it will be
> enabled only for some latency sensitive clients that can afford busy wait. I
> would also be reluctant to use Disruptor until bulk put is better supported.

Try out the bulk-put API and let me know if it works for you.

Performance Results:

--- Number of publishers = 1

Process 10000000 messages with Disruptor using ProducerType.MULTI /
BusySpinWaitStrategy.
Test took: 0.457
Test took: 0.395
Test took: 0.390
Test took: 0.375

Process 10000000 bulk put messages with Disruptor using
ProducerType.MULTI / BlockingWaitStrategy.
Test took: 0.121
Test took: 0.063
Test took: 0.066
Test took: 0.065

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll and put.
Test took: 1.060
Test took: 0.853
Test took: 0.885
Test took: 0.927

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 0.533
Test took: 0.677
Test took: 0.585
Test took: 0.599

--- Number of publishers = 2

Process 10000000 messages with Disruptor using ProducerType.MULTI /
BusySpinWaitStrategy.
Test took: 1.100
Test took: 1.042
Test took: 1.094
Test took: 1.100

Process 10000000 bulk put messages with Disruptor using
ProducerType.MULTI / BlockingWaitStrategy.
Test took: 0.092
Test took: 0.097
Test took: 0.101
Test took: 0.106

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll and put.
Test took: 3.514
Test took: 3.035
Test took: 3.489
Test took: 3.526

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 0.992
Test took: 1.015
Test took: 0.944
Test took: 1.009

--- Number of publishers = 4

Process 10000000 messages with Disruptor using ProducerType.MULTI /
BusySpinWaitStrategy.
Test took: 2.825
Test took: 2.840
Test took: 3.389
Test took: 3.054

Process 10000000 bulk put messages with Disruptor using
ProducerType.MULTI / BlockingWaitStrategy.
Test took: 0.199
Test took: 0.200
Test took: 0.202
Test took: 0.198

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll and put.
Test took: 5.641
Test took: 5.866
Test took: 5.395
Test took: 5.221

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 1.740
Test took: 1.717
Test took: 1.698
Test took: 1.711

--- Number of publishers = 8

Process 10000000 messages with Disruptor using ProducerType.MULTI /
BusySpinWaitStrategy.
Test took: 7.956
Test took: 8.436
Test took: 8.713
Test took: 8.125

Process 10000000 bulk put messages with Disruptor using
ProducerType.MULTI / BlockingWaitStrategy.
Test took: 0.445
Test took: 0.424
Test took: 0.453
Test took: 0.436

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll and put.
Test took: 9.520
Test took: 12.214
Test took: 9.801
Test took: 11.677

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 3.032
Test took: 3.281
Test took: 2.752
Test took: 3.114

--- Number of publishers = 16

Process 10000000 messages with Disruptor using ProducerType.MULTI /
BusySpinWaitStrategy.
Test took: 19.090
Test took: 18.517
Test took: 19.730
Test took: 17.555

Process 10000000 bulk put messages with Disruptor using
ProducerType.MULTI / BlockingWaitStrategy.
Test took: 0.862
Test took: 0.812
Test took: 0.800
Test took: 0.809

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll and put.
Test took: 18.334
Test took: 18.407
Test took: 21.337
Test took: 22.205

Process 10000000 messages with ArrayBlockingQueueEx using bulk takeAll
and bulk putAll.

Test took: 6.004
Test took: 5.604
Test took: 6.130
Test took: 5.261