Re: [akka-user] Implementing a bounded mailbox using lmax disruptor

[Roland Kuhn]

Hi Rajiv,

30 apr 2013 kl. 20:12 skrev Rajiv Kurian:

Also would numberOfMessages and hasMessages also be called from only one thread at a time?

numberOfMessages is AFAIK only called from the SmallestMailboxRouter, but in that case from any thread in the system concurrently. hasMessages is called upon every tell operation, i.e. concurrently.

On Tuesday, April 30, 2013 10:16:03 AM UTC-7, Rajiv Kurian wrote:

I am trying to use a ring buffer from the Disruptor framework(https://github.com/LMAX-Exchange/disruptor) to implement a bounded mailbox for an Akka actor as an experiment. The Disruptor framework uses a ring buffer as a pseudo queue structure between multiple producers (whose inserts will be serialized) and parallel consumers who could be serially processing different sections of the ring buffer.

 From reading the Akka docs it seems like the way to create a custom mailbox like this is to extend the MessageQueue trait. If my understanding is correct all the methods (enqueue, dequeue, numberOfMessages, hasMessages) etc need to support concurrent calls. I am hoping to just use disruptor's producer barrier feature to implement concurrent enqueue.

 As for the dequeue since the actor is the one calling it am I right to assume that only one thread at a time could call dequeue? The reason I am asking is that I could then implement the consumer sequence number as a simple volatile primitive. A thread could call dequeue and it would increment this sequence number. Subsequent dequeues on the same thread (in case of batch processing) would all increment the sequence number without any race. When the actor is suspended and resumed on another thread it would still see the latest sequence number without a race.

Yes, this is correct: multiple-producer single-consumer.

I apologize for the Disruptor jargon and the messed up Akka jargon too. Hoping some of you guys have tried out the Disruptor framework.

In fact I have not yet played with it myself, so I’ll be doubly curious to hear how your experiment goes.

Regards,

Dr. Roland Kuhn
Akka Tech Lead
Typesafe – Empowering professional developers to build amazing apps.
twitter: @rolandkuhn

See you at Scala Days 2013 in NYC!
June 10th - June 12th
www.scaladays.org

[kermitas]

Hello Rajiv.

I've just found your post here. It's very, very interesting!!

How your work goes? Did you make any progress with LMAX Disruptor + Akka? Do you have new benchmark? How it compares to standard akka mailboxes?

Thank you,

Artur.

W dniu środa, 1 maja 2013 00:28:30 UTC+2 użytkownik Rajiv Kurian napisał:

Thanks for the reply Roland. I haven't played with either. I am hoping to learn about it from this experiment. They have some staggering numbers for the single producer use case compared to concurrent queues, which sadly can't be used for the actor api. Their latest release benchmarks boast of 220M ops/sec for a single producer use case on an i7 Ivy Bridge @ 3.4 GHz. Multiple producers benchmark is at about 15 M ops/sec. The ring buffer though has to be of a fixed size so I can't think of a way to use it for an unbounded mailbox in a performant way. 

[Rajiv Kurian]

Hi Kermitas:

I began work on this as a weekend experiment, but my full time job has bogged me down since then. Hopefully I'll have some time to work on this in the coming month. Also I only looked into implementing bounded mailboxes and not the more common unbounded ones.

Rajiv

[Suminda Dharmasena]

Any luck with this exercise? Any FOSS implementations?

[Rajiv Kurian]

Sorry, I ended up using the disruptor for that project instead of implementing a mailbox for Akka. For my use case I had a topology with a single producer which is why disruptor made sense. Akka requires multiple producer support for its mailbox and I didn't think it would be worth the effort to get the disruptor working especially given how it really shines with single producers. AFAIK the new MPSC mailbox implementation in Akka 2.2 has a pretty high throughput, which made this port seem even less useful.

[Igor Konev]

If somebody is still interested, I've implemented a bounded mailbox using Disruptor and obtained a throughput gain of about 50% over the standard bounded mailbox. You can find the source code here https://github.com/yngui/akka-disruptor.

[Roland Kuhn]

Hi Igor,

thanks for sharing! On first (cursory) sight it looks like your mailbox might cause missed wake-ups because head and tail are read by the enqueueing thread without synchronization (i.e. disruptor cursor is moved but updates to head and tail have not been published; this might be mitigated by clever ordering of effects, but my spidey senses make me uneasy on this one).

Regards,

Roland

1 jan 2014 kl. 20:04 skrev Igor Konev <igor.m...@gmail.com>:

> If somebody is still interested, I've implemented a bounded mailbox using Disruptor and obtained a throughput gain of about 50% over the standard bounded mailbox. You can find the source code here https://github.com/yngui/akka-disruptor.
>

> --
>>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>>> Check the FAQ: http://akka.io/faq/
>>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+...@googlegroups.com.
> To post to this group, send email to akka...@googlegroups.com.
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/groups/opt_out.

Dr. Roland Kuhn
Akka Tech Lead

Typesafe – Reactive apps on the JVM.
twitter: @rolandkuhn

[Igor Konev]

Hi Roland,

Although you mentioned above that hasMessages is called upon every tell operation, it looks like it is not the case. Not taking BalancingDispatcher into account hasMessages is called from Mailbox's canBeScheduledForExecution as follows

  final def canBeScheduledForExecution(hasMessageHint: Boolean, hasSystemMessageHint: Boolean): Boolean = status match {

    case Open | Scheduled ⇒ hasMessageHint || hasSystemMessageHint || hasSystemMessages || hasMessages

    case Closed           ⇒ false

    case _                ⇒ hasSystemMessageHint || hasSystemMessages

  }

But hasMessages and hasSystemMessageHint are never false at the same time when canBeScheduledForExecution is called (indirectly) from Dispatcher:

  protected[akka] override def registerForExecution(mbox: Mailbox, hasMessageHint: Boolean, hasSystemMessageHint: Boolean): Boolean = {

    if (mbox.canBeScheduledForExecution(hasMessageHint, hasSystemMessageHint)) { //This needs to be here to ensure thread safety and no races

      ...

  protected[akka] def dispatch(receiver: ActorCell, invocation: Envelope): Unit = {

    val mbox = receiver.mailbox

    mbox.enqueue(receiver.self, invocation)

    registerForExecution(mbox, true, false)

  }

  protected[akka] def systemDispatch(receiver: ActorCell, invocation: SystemMessage): Unit = {

    val mbox = receiver.mailbox

    mbox.systemEnqueue(receiver.self, invocation)

    registerForExecution(mbox, false, true)

  }

So hasMessages is never called upon tell.

пятница, 3 января 2014 г., 10:51:33 UTC+2 пользователь rkuhn написал:

[Jonas Bonér]

Sounds interesting. In which scenarios/under which conditions does it have better throughput? 

On Wed, Jan 1, 2014 at 11:04 AM, Igor Konev <igor.m...@gmail.com> wrote:

If somebody is still interested, I've implemented a bounded mailbox using Disruptor and obtained a throughput gain of about 50% over the standard bounded mailbox. You can find the source code here https://github.com/yngui/akka-disruptor.

--
>>>>>>>>>>      Read the docs: http://akka.io/docs/
>>>>>>>>>>      Check the FAQ: http://akka.io/faq/
>>>>>>>>>>      Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+...@googlegroups.com.
To post to this group, send email to akka...@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/groups/opt_out.

--
Jonas Bonér
Phone: +46 733 777 123
Home: jonasboner.com
Twitter: @jboner

[Andrew Carroll]

Hi Igor,

I've implemented an Actor-type system (and as it stands, Akka IO) as a PoC using the Disruptor myself and was waiting for the day a good implementation becomes available for Akka!

My experience is -- and everything that the LMAX developers say -- backs up the claim that the best performance from the RingBuffer comes when batching operations, as this decreases the frequency of memory fencing instructions executed.  Essentially, use the RingBuffer for inter-core communication only.

So far I've only had a quick glance at your code and don't have really any knowledge of the internals of Akka's message passing.  Do you see any opportunity within your implementation to batch writes and reads?

This email is confidential and intended solely for the person(s) to whom it is addressed.

[Oleg Zhurakousky]

I would say it goes back to the core of the disruptor technology which essentially provides a different realization of a queue concept trough a ring buffer which is array based (not java Queue implementations). I am assuming mailboxes in Akka are Queue based and for high throughput cases Queues do introduce significant latency. I've observed similar (some times much greater) performance gains when experimented with the idea of the disruptor-based Queue channel in spring integration. There are also cache line trickery there with long padding, but that's a whole other topic. 

[Roland Kuhn]

Hi Igor,

you are right concerning the default dispatcher, but BalancingDispatcher (as you note) and some Routers also use hasMessages—as could custom dispatchers. Therefore this mailbox implementation is a specialized one, it does not implement the full interface so to speak. Could it be made to obey the full contract without losing the performance benefits?

Regards,

Roland

[Igor Konev]

Hi Jonas,

I used a modified version of the Akka throughput performance benchmark. It was necessary to configure each group of actors with different mailbox types. DisruptorMailbox is a bounded mailbox type and if every actor had used only it, this would have caused a deadlock. The destination actors were configured with DisruptorMailbox and client actors used the default configuration.

I ran the benchmark on a Core i7-3770 @ 3.40 GHz and got the following results:

вторник, 7 января 2014 г., 0:32:13 UTC+2 пользователь Jonas Bonér написал:

[√iktor Ҡlang]

Interesting!

What unbounded mailbox impl did you use?

Cheers,

√

Viktor Klang
Director of Engineering

Typesafe

Twitter: @viktorklang

[Igor Konev]

Hi Andrew,

DisruptorMessageQueue already batches reads but the implementation differs from what is done in BatchEventProcessor. BatchEventProcessor gets events one by one from the ring buffer and at the same time sends them to the event handler. But while BatchEventProcessor controls its event handler, DisruptorMessageQueue isn't aware of its recipients. So on the first dequeue call DisruptorMessageQueue gets all available events from the ring buffer and puts them into its internal buffer. On subsequent dequeue calls it just retrieves and returns the events from the internal buffer.

To implement batch writes in DisruptorMessageQueue Akka should support something like a batch tell. It can be easily emulated by sending a set of messages as an array or collection. And it will be even more efficient because Akka will construct an Envelope only for one message rather than for each message from the set.

вторник, 7 января 2014 г., 4:39:48 UTC+2 пользователь Andrew Carroll написал:

[Igor Konev]

Hi √,

понедельник, 13 января 2014 г., 18:43:39 UTC+2 пользователь √ написал:

Interesting!

What unbounded mailbox impl did you use?

akka.dispatch.UnboundedMailbox

[√iktor Ҡlang]

Hi Igor,

please try: akka.dispatch.SingleConsumerOnlyUnboundedMailbox

Cheers,

√

[Jonas Bonér]

Thanks Igor. That's some pretty interesting results. 

On Mon, Jan 13, 2014 at 5:36 PM, Igor Konev <igor.m...@gmail.com> wrote:

[Igor Konev]

Hi √,

I have replaced UnboundedMailbox with SingleConsumerOnlyUnboundedMailbox in all the tests and here is the result:

вторник, 14 января 2014 г., 16:29:57 UTC+2 пользователь √ написал:

[√iktor Ҡlang]

Hi Igor,

Looks very competetive :-)

[Guido Medina]

Hi Igor,

I have modified your initial implementation because on my tests it was blocking the Akka system at shutdown so I simplified it for SingleConsumerOnlyBoundedMailbox, look at the Akka git hub ticket and at my forked branch:

https://github.com/akka/akka/issues/17558

Best regards and many thanks for sharing it,

Guido