Optimize Promise #9569
Optimize Promise #9569
Conversation
| @@ -39,9 +39,12 @@ object BenchmarkUtil extends Runtime[Any] { self => | |||
| Unsafe.unsafe(implicit unsafe => rt.unsafe.run(zio).getOrThrowFiberFailure()) | |||
| } | |||
|
|
|||
| override val unsafe = super.unsafe | |||
There was a problem hiding this comment.
I noticed some benchmarks allocate new unsafe apis.
|
Will fix the CI tomorrow. |
Co-authored-by: Jules Ivanic <jules.ivanic@gmail.com>
This reverts commit 4689adf.
| ZIO.asyncInterrupt[Any, E, A]( | ||
| case Done(value) => value | ||
| case pending => | ||
| ZIO.async[Any, E, A]( // intentionally never remove the callback, interrupted fibers won't be resumed |
There was a problem hiding this comment.
Kyri and I concluded that this will likely end up retaining references to Fibers, which is problematic in a highly awaited Promise, where Fibers are interrupted.
Unfortunately, we will need to use asyncInterrupt here, though it's heavier weight. We can keep the Link structure as the removals should be rare, and it's optimized for the single waiter case.
| retry = !state.compareAndSet(oldState, newState) | ||
| } | ||
| object Promise { | ||
| private[zio] object internal { |
There was a problem hiding this comment.
From what I could tell, there isn't a great way to test this without it being package private...
|
@kyri-petrou this is ready for review. |
|
@hearnadam Your CI is failing |
|
Discussed the PR offline with Kyri today.
|
| def loop(): Boolean = | ||
| state.get match { | ||
| case pending: Pending[?, ?] => | ||
| if (state.compareAndSet(pending, Done(io))) { |
There was a problem hiding this comment.
Could the Done(io) instanciation be put outside of the loop to avoid more than 1 instanciation? 🤔
There was a problem hiding this comment.
It's possible, though passing it to the loop may have more overhead than just allocating. Since it only has a single frame, it should be okay.
Additionally, that would allocate when the promise was already completed.
| if (joiners ne null) joiners.foreach(_(io)) | ||
| def remove(waiter: IO[Nothing, Nothing] => Any): Pending[Nothing, Nothing] = self | ||
| } | ||
| private sealed abstract class Link[E, A](final val waiter: IO[E, A] => Any, final val ws: Pending[E, A]) |
There was a problem hiding this comment.
What does ws means? waiters? Can we use a more explicit name maybe?
Avoid ZIO.asyncInterrupt - it introduces a minimum of 3 suspensions + Atomic allocationNever interrupt waiters. If the callback is invoked after interruption, the fiber won't be resumedZIO.asyncInterruptAtomicRef and function allocationExitinmaketo avoid suspension (we should later addmakeWith)Complete waiters in orderMarking this as a draft as I want feedback on: 1) never removing waiters, 2) state design
The benchmarks are extremely noisy as the ZScheduler constantly waking Threads dominates. I think I've removed most of the hotspots, but I will make some changes to my measurements to make sure.