Repository navigation
Conversation
| def modify[B](f: A => (A, B)): F[B] = | ||
| state.flatModify { s => // uncancellable to avoid losing wake-up signals | ||
| val (value, result) = f(s.value) | ||
| if (!s.waiters.exists(_.accepts(value))) |
There was a problem hiding this comment.
Have you tried partitioning first to avoid two traversals of the list?
There was a problem hiding this comment.
We're dealing with a single consumer within groupWithin and the predicate is cheap vs. allocating new lists each time in partition where in practice the first if-clause should execute more often.
Arguably, this could be premature optimization (and I can run quick bechmarks to confirm) or impl should reflect spsc. Open to suggestions, but I'm starting to lean towards a scoped impl of this class for spsc rather than a more generic mpmc. Probably separate impls, ConditiedRef.mpmc, ConditiedRef.spsc (only this would be needed for PR), etc. will avoid confusion for future maintenance too.
There was a problem hiding this comment.
So there's a small hit when buffer size is small, but overall not very noticable. I'm fine with changing it to call partition first and have a single traversal.
There was a problem hiding this comment.
Ok, ran some more benchmarks and there's a drop in perf in the scenarios where there is no chunking (single elements), so I would prefer to keep it for my use case.
|
Benchmarks show that the new impl is improvement in all benchmarked situations, and especially when using a larger buffer size and/or chunks from upstream producer. Based on this, we should swap the impl of groupWithin so that there is only one method. Should we commit these to a text file under the benchmarks module? |
No. I don't see any other benchmarks with results committed. |
…timeouts to existing api
…timeouts to existing api
mmienko
changed the title
…o groupChunkWithin
This branch has not been deployed
2 participants
Add a performant groupWithin impl that works on chunks.
I experimented without other concurrency primitives like a Ref and Synchronous Queue, or a Ref of 2 Deferred, but they all obfuscated the main logic of this method and made it difficult to verify it's correct. SignallingRef had the perfect api, but was not performant in benchmarks. So, I created a trimmed down version called ConditionedRef which only pushes updates if they pass a predicate.