Optimize Mutex & AtomicCell
#3346
Conversation
BalmungSan
force-pushed
the
improve-mutex
branch
2 times, most recently
from
e38fe68
to
f0827a1
Compare
BalmungSan
force-pushed
the
improve-mutex
branch
from
f0827a1
to
1b85b44
Compare
This comment was marked as resolved.
This comment was marked as resolved.
BalmungSan
force-pushed
the
improve-mutex
branch
from
1b85b44
to
498d4af
Compare
BalmungSan
force-pushed
the
improve-mutex
branch
from
498d4af
to
d7c3234
Compare
Here's an idea, that I think could be material improvement. What if we extracted the We can use the |
Or the implementation proposed in this PR, which is good too :) Btw: I believe this PR can be targeted at 3.4.x, since it's just internal optimizations (no new APIs) |
BalmungSan
force-pushed
the
improve-mutex
branch
5 times, most recently
from
07be7bf
to
9dd1b92
Compare
BalmungSan
force-pushed
the
improve-mutex
branch
from
d4cba8f
to
9851058
Compare
| // Cancels a Fiber waiting for the Mutex. | ||
| private def cancel(thisCB: CB, thisCell: LockCell, previousCell: LockCell): F[Unit] = | ||
| F.delay { | ||
| // If we are canceled. | ||
| // First, we check if the state still contains ourselves, | ||
| // if that is the case, we swap it with the previousCell. | ||
| // This ensures any consequent attempt to acquire the Mutex | ||
| // will register its callback on the appropriate cell. | ||
| // Additionally, that confirms there is no Fiber | ||
| // currently waiting for us. | ||
| if (!state.compareAndSet(thisCell, previousCell)) { | ||
| // Otherwise, | ||
| // it means we have a Fiber waiting for us. | ||
| // Thus, we need to tell the previous cell | ||
| // to awake that Fiber instead. | ||
| var nextCB = thisCell.get() | ||
| while (nextCB eq null) { | ||
| // There is a tiny fraction of time when | ||
| // the next cell has acquired ourselves, | ||
| // but hasn't registered itself yet. | ||
| // Thus, we spin loop until that happens | ||
| nextCB = thisCell.get() | ||
| } | ||
| if (!previousCell.compareAndSet(thisCB, nextCB)) { | ||
| // However, in case the previous cell had already completed, | ||
| // then the Mutex is free and we can awake our waiting fiber. | ||
| if (nextCB ne null) nextCB.apply(RUnit) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Awaits until the Mutex is free. | ||
| private def await(thisCell: LockCell): F[Unit] = | ||
| F.asyncCheckAttempt[Unit] { thisCB => | ||
| F.delay { | ||
| val previousCell = state.getAndSet(thisCell) | ||
|
|
||
| if (previousCell eq null) { | ||
| // If the previous cell was null, | ||
| // then the Mutex is free. | ||
| RUnit.asInstanceOf[Either[Option[F[Unit]], Unit]] | ||
| } else { | ||
| // Otherwise, | ||
| // we check again that the previous cell haven't been completed yet, | ||
| // if not we tell the previous cell to awake us when they finish. | ||
| if (!previousCell.compareAndSet(null, thisCB)) { | ||
| // If it was already completed, | ||
| // then the Mutex is free. | ||
| RUnit.asInstanceOf[Either[Option[F[Unit]], Unit]] | ||
| } else { | ||
| Left(Some(cancel(thisCB, thisCell, previousCell))) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Acquires the Mutex. | ||
| private def acquire(poll: Poll[F]): F[LockCell] = | ||
| F.delay(new AtomicReference[CB]()).flatMap { thisCell => | ||
| poll(await(thisCell).map(_ => thisCell)) | ||
| } | ||
|
|
||
| // Releases the Mutex. | ||
| private def release(thisCell: LockCell): F[Unit] = | ||
| F.delay { | ||
| // If the state still contains our own cell, | ||
| // then it means nobody was waiting for the Mutex, | ||
| // and thus it can be put on a free state again. | ||
| if (!state.compareAndSet(thisCell, null)) { | ||
| // Otherwise, | ||
| // our cell is probably not empty, | ||
| // we must awake whatever Fiber is waiting for us. | ||
| val nextCB = thisCell.getAndSet(Sentinel) | ||
| if (nextCB ne null) nextCB.apply(RUnit) | ||
| } | ||
| } |
There was a problem hiding this comment.
So, the idea of this new implementation is to save the Deferred instantiation, based on the assumption that at any point in time a Fiber is waiting for only one other Fiber, and it is being waited by at most another one.
So, the idea is that we will have:
statewhich is a mutable reference to the lastCellin the chain.thisCellwhich is the chainCellof theFibertrying to access theMutex.previousCellwhich is the previousCellin the chain; maybe the one currently holding theMutex.
Acquire:
First, we create our thisCell and replace it in the state, retrieving the previousCell; atomically.
We then use F.asyncCheckAttempt to check if the Mutex is currently in use. If that is the case, we register the callback in the previousCell and wait for it. Otherwise, we can just acquire it.
Release:
We call the callback that was registered in our Cell and call it. Considering the possibility that maybe no one is actually waiting for us.
Cancelation:
In the case we were canceled while waiting for the Mutex:
- Situation A: No one is waiting for us.
Thus, we only need to ensure that any Fiber that follows awaits the previous cell; they will know how to handle the case if it is already released. - Situation B: Someone is waiting for us.
Then we need to unregister our callback from the previous cell and change it to the callback of the Fiber waiting for us; we need to double check in case theMutexwas released at that moment to rather notify whoever is waiting for us.
I believe this allocates the bare minimum and is concurrent safe.
PS: Thanks a lot to Arman (@armanbilge) for brainstorming with me on this implementation :)
BalmungSan
force-pushed
the
improve-mutex
branch
2 times, most recently
from
5b8b9d8
to
1c1b674
Compare
BalmungSan
force-pushed
the
improve-mutex
branch
from
1c1b674
to
02c301e
Compare
BalmungSan
force-pushed
the
improve-mutex
branch
from
86fd9b5
to
145833c
Compare
benchmarks/src/main/scala/cats/effect/benchmarks/AtomicCellBenchmark.scala
Outdated
Show resolved
Hide resolved
BalmungSan
force-pushed
the
improve-mutex
branch
from
145833c
to
6c070e1
Compare
BalmungSan
force-pushed
the
improve-mutex
branch
from
6c070e1
to
7a75dfd
Compare
|
@djspiewak @armanbilge before running the benchmarks I want to confirm with both of you if you think they are appropriate. |
BalmungSan
force-pushed
the
improve-mutex
branch
from
7a75dfd
to
bb2da6c
Compare
benchmarks/src/main/scala/cats/effect/benchmarks/AtomicCellBenchmark.scala
Outdated
Show resolved
Hide resolved
|
Let's see the benchmark results but overall lgtm |
benchmarks/src/main/scala/cats/effect/benchmarks/AtomicCellBenchmark.scala
Outdated
Show resolved
Hide resolved
BalmungSan
requested review from
djspiewak and
armanbilge
and removed request for
armanbilge and
djspiewak
|
@djspiewak @armanbilge benchmark results added to the description of the PR! :D |
|
Benchmarks look compelling! Thank you! |
Optimize
Mutex&AtomicCellThis PR adds two alternative implementations of
Mutex&AtomicCellbased onAsyncrather than justConcurrent, these implementations should be more efficient.This is a binary and source-compatible change. Since users don't need to do anything special to get those new implementations, rather if the underlying data type they use support
Async; e.g. likeIO, they will get the optimized versions automatically.Benchmark results
Configuration
MutexConcurrent
Async
AtomicCell(using the respectiveMutex)Concurrent
Async
Conclusions
IMHO the improvements are noticeable, being around a
20%increase in most situations.Also,
AtomicCellresults are pretty equivalent; but still slightly better, than theMutexone. Suggesting that most of the increase comes from theMutex, still I think theAsyncAtomicCellis worth it.AtomicCellBenchmarkResults.txt
MutexBenchmarkResults.txt