Even faster async mutex #3409

armanbilge · 2023-02-09T18:09:19Z

This PR tries my idea from #3346 (comment) of re-using the UnsafeUnbounded datastructure from the async queue to implement the async mutex. /cc @BalmungSan

this PR

[info] Benchmark                           (fibers)  (iterations)   Mode  Cnt    Score   Error  Units
[info] MutexBenchmark.cancellationAsync          10          1000  thrpt   20   15.434 ± 1.058  ops/s
[info] MutexBenchmark.cancellationAsync         100          1000  thrpt   20    3.203 ± 0.464  ops/s
[info] MutexBenchmark.happyPathAsync             10          1000  thrpt   20  145.271 ± 5.933  ops/s
[info] MutexBenchmark.happyPathAsync            100          1000  thrpt   20   13.297 ± 1.656  ops/s
[info] MutexBenchmark.highContentionAsync        10          1000  thrpt   20   22.955 ± 2.932  ops/s
[info] MutexBenchmark.highContentionAsync       100          1000  thrpt   20    2.953 ± 0.479  ops/s

series/3.x

[info] Benchmark                           (fibers)  (iterations)   Mode  Cnt    Score   Error  Units
[info] MutexBenchmark.cancellationAsync          10          1000  thrpt   20   14.627 ± 1.514  ops/s
[info] MutexBenchmark.cancellationAsync         100          1000  thrpt   20    3.234 ± 0.658  ops/s
[info] MutexBenchmark.happyPathAsync             10          1000  thrpt   20  124.522 ± 7.762  ops/s
[info] MutexBenchmark.happyPathAsync            100          1000  thrpt   20   11.987 ± 0.379  ops/s
[info] MutexBenchmark.highContentionAsync        10          1000  thrpt   20   16.659 ± 2.072  ops/s
[info] MutexBenchmark.highContentionAsync       100          1000  thrpt   20    2.131 ± 0.127  ops/s

armanbilge · 2023-02-09T18:10:17Z

std/shared/src/main/scala/cats/effect/std/package.scala

+  private[std] val FailureSignal: Throwable = new RuntimeException
+    with scala.util.control.NoStackTrace


Instead of placing this in the package object we could also make a top-level object FailureSignal extends RuntimeException with NoStackTrace.

I think that either approach introduces a memory barrier unless we locally cache the value. If we locally cache the value, then they're the same. I would prefer the top-level object approach, tbh, but we really need to make sure it's pre-cached within scopes.

armanbilge · 2023-02-09T18:10:46Z

std/shared/src/main/scala/cats/effect/std/UnsafeUnbounded.scala

+
+import java.util.concurrent.atomic.AtomicReference
+
+private final class UnsafeUnbounded[A] {


Didn't make any changes here, just copy-pasta.

Oh, another advantage of pulling this out into a top-level class it is much easier to provide an specialized implementation for JS without Platform shenanigans.

I originally planned on pulling it out to the top-level but got lazy.

armanbilge · 2023-02-09T18:12:25Z

std/shared/src/main/scala/cats/effect/std/Mutex.scala

+      } catch { // no waiter found
+        case FailureSignal =>
+          locked.set(false) // release
+          try {
+            var waiter = waiters.take()
+            while (waiter eq null) waiter = waiters.take()
+            waiter(RightFalse) // waken any new waiters
+          } catch {


There's some fairness corruption here under contention, where an acquirer may cut-in-line of an acquirer that had placed itself in the queue.

Are the FIFO semantics of the current Mutex something we would like to preserve? If so, should we be louder about it on the docs?

BTW, does the Semaphore based one guarantee that as well?

Overall it's still FIFO (or should be, we should add a test if we don't have already). It's just under contention it can be slightly corrupted—but under contention, who was really "first" anyway?

I think this was the long-running debate Daniel and Fabio had for the async queue :)

BalmungSan · 2023-02-09T18:19:47Z

benchmarks/src/main/scala/cats/effect/benchmarks/MutexBenchmark.scala

-    mutex
-      .flatMap { m => m.lock.use_.replicateA_(fibers) }
-      .replicateA_(iterations)
-      .unsafeRunSync()
+    mutex.flatMap { m => m.lock.use_.replicateA_(fibers * iterations) }.unsafeRunSync()


I wonder what is the rationale for this change? And why only to the happy path?

Maybe I misunderstood the purpose of the benchmark, but we want to replicate many acquire/releases of the same mutex from the same fiber—we don't need to allocate a new mutex in each iteration, and "fibers" is not really accurate term. It's just iterations in the end.

Actually you are right, we can probably make a similar change to the other benchmarks.

armanbilge · 2023-02-10T19:35:12Z

std/shared/src/main/scala/cats/effect/std/Mutex.scala

-      F.asyncCheckAttempt[Unit] { thisCB =>
-        F.delay {
-          val previousCell = state.getAndSet(thisCell)
+    private[this] val locked = new AtomicBoolean(false)


If we swap this AtomicBoolean for an AtomicInteger then do we basically have an AsyncSemaphore?

djspiewak

Minor change

djspiewak · 2023-02-14T00:14:11Z

std/shared/src/main/scala/cats/effect/std/UnsafeUnbounded.scala

+
+import java.util.concurrent.atomic.AtomicReference
+
+private final class UnsafeUnbounded[A] {


I originally planned on pulling it out to the top-level but got lazy.

djspiewak · 2023-02-14T00:15:14Z

std/shared/src/main/scala/cats/effect/std/package.scala

+  private[std] val FailureSignal: Throwable = new RuntimeException
+    with scala.util.control.NoStackTrace


I think that either approach introduces a memory barrier unless we locally cache the value. If we locally cache the value, then they're the same. I would prefer the top-level object approach, tbh, but we really need to make sure it's pre-cached within scopes.

djspiewak · 2023-02-14T01:33:48Z

std/shared/src/main/scala/cats/effect/std/Mutex.scala

@@ -99,6 +99,7 @@ object Mutex {

    private[this] val locked = new AtomicBoolean(false)
    private[this] val waiters = new UnsafeUnbounded[Either[Throwable, Boolean] => Unit]
+    private[this] val failureSignal = FailureSignal // prefetch


Super minor thing but I'd prefer if we shadow FailureSignal here by prefetching the fully-qualified name. This avoids the need for the backticks, and also prevents people from accidentally introducing memory barriers by referring to the non-prefetched thing.

Yeah I thought about this as well. I was more afraid that I would forget to prefetch it, unless I always referred to it under a name only available as a prefetch.

Well now this way your only option is to prefetch.

If you remembered to shadow it in the first place 😆 if the shadow doesn't exist in your current scope, you have no way of knowing that you are going through the memory barrier.

I feel like we introduce new scopes less often than we introduce new code in scopes. :-P

armanbilge added 7 commits February 9, 2023 18:06

Extract UnsafeUnbounded

01ca653

Implement async mutex with UnsafeUnbounded

e1ac43f

Add test for canceled waiters

68b50b5

Fix mutex release bug

2ecd790

Add test for deadlock under contention

1e4176b

Fix release race condition

989277e

Tweak mutex happy path benchmark

95bbcba

armanbilge commented Feb 9, 2023

View reviewed changes

BalmungSan reviewed Feb 9, 2023

View reviewed changes

MiMa filters

2874d3e

armanbilge commented Feb 10, 2023

View reviewed changes

djspiewak requested changes Feb 14, 2023

View reviewed changes

Make FailureSignal an object, prefetch it

399544b

djspiewak requested changes Feb 14, 2023

View reviewed changes

Use shadowing

2ccf8be

djspiewak approved these changes Feb 14, 2023

View reviewed changes

armanbilge merged commit e9395f7 into typelevel:series/3.x Feb 14, 2023

armanbilge mentioned this pull request Apr 23, 2023

Hanging following IO.blocking(...) calls in v3.5.0 RC3 and RC4 #3548

Closed

BalmungSan mentioned this pull request Apr 25, 2023

New AsyncMutex implementation #3562

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Even faster async mutex #3409

Even faster async mutex #3409

armanbilge commented Feb 9, 2023

armanbilge Feb 9, 2023

djspiewak Feb 14, 2023

armanbilge Feb 9, 2023

armanbilge Feb 9, 2023

djspiewak Feb 14, 2023

armanbilge Feb 9, 2023

BalmungSan Feb 9, 2023

armanbilge Feb 9, 2023

BalmungSan Feb 9, 2023

armanbilge Feb 9, 2023

armanbilge Feb 10, 2023

djspiewak left a comment

djspiewak Feb 14, 2023

djspiewak Feb 14, 2023

djspiewak Feb 14, 2023

armanbilge Feb 14, 2023

djspiewak Feb 14, 2023

armanbilge Feb 14, 2023

djspiewak Feb 14, 2023

		private[std] val FailureSignal: Throwable = new RuntimeException
		with scala.util.control.NoStackTrace


		import java.util.concurrent.atomic.AtomicReference

		private final class UnsafeUnbounded[A] {

Even faster async mutex #3409

Even faster async mutex #3409

Conversation

armanbilge commented Feb 9, 2023

this PR

series/3.x

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

djspiewak left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment