Source.queue vs Source.actorRef vs custom GraphStage

[Tim Harper]

I did some benchmarking to see how well `Source.actorRef` and `Source.queue` performed. Also, I wanted to see how much better I could do if I didn't need my Source to buffer (this stream is using conflate). It turns out, `Source.queue` is slower than `Source.actorRef`. And, the custom graph stage runs quite a bit faster than `Source.actorRef`:

https://github.com/timcharper/as-actorref-repro/tree/ingestor-benchmark

`Source.queue` is probably fine enough, but in reality, there's no reason it should perform worse than `Source.actorRef`. It could be sped up significantly while retaining the same functionality by following the same pattern of using the atomicBoolean to reduce scheduling callbacks. Two internal queues could be used, one queue a java.util.concurrent.ConcurrentLinkedQueue, the other queue a mutable, non-thread-safe queue. On each run of the stage callback, the OverflowStrategy could be applied quite safely. Since the java.util.concurrent.ConcurrentLinkedQueue will be only pushed to from async threads, a current total + the current size of the non-thread-safe mutable queue could be calculated to see if overflow had occurred, and then the strategy applied accordingly.

Just wanted to post these thoughts somewhere while they were on my mind. I don't know if the optimization would be worth pursuing, just that a 2-3x improvement in speed for Source.queue should be possible.

Tim

[Viktor Klang]

Isn't CLQ.size O(N)?

--
Cheers,
√

--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> 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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[Tim Harper]

You're right. However, we could leverage the fact that the queue size only grows outside of the stage process. So, I think we could use a counter.

You received this message because you are subscribed to a topic in the Google Groups "Akka User List" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/akka-user/Qi0CMY5-pjM/unsubscribe.
To unsubscribe from this group and all its topics, send an email to akka-user+...@googlegroups.com.

[Viktor Klang]

What happens if the queue is flooded before overflow protection kicks in?

Cheers,

√

[Tim Harper]

Well, to counter this argument, I think this is a question that should be asked of Source.queue and Source.actorRef, also. Source.actorRef will buffer forever with its mailbox, and the overflow protection is applied as that mailbox is drained, when the Actor is gets an opportunity to run. Also, Source.buffer will, technically, buffer forever in the sense that every call to `offer` is queuing up a new asynchronous callback to be invoked.

If we all agree that some form of unbounded buffering is ultimately inevitable (either with an Actor mailbox or a queued up callbacks), then the question should turn to: "how do we deal optimally with it when it happens?". And, I think the answer is that the process which applies the overflow strategy should perform as few allocation as possible.

Tim

[Akka Team]

Hi Tim,

Feel free to experiment with these, I think these can be a good improvement for Source.queue. Source.actorRef is somewhat different though, it does not use a real ActorRef, it is a very special construct that makes a GraphStage look like an actor from the outside, but in fact all messages go through the actor that hosts the GraphStage (it can host more than one stage in fact). I am not sure much improvement is possible there.

Anyway, PRs are welcome :)

-Endre

Akka Team

Typesafe - Reactive apps on the JVM

Blog: letitcrash.com
Twitter: @akkateam