Resolve race condition in interruptibleMany after interruption

typelevel · Jul 4, 2022 · 89a85e1 · 89a85e1
1 parent 7887b34
commit 89a85e1
Showing 1 changed file with 10 additions and 0 deletions.
diff --git a/core/jvm/src/main/scala/cats/effect/IOFiberPlatform.scala b/core/jvm/src/main/scala/cats/effect/IOFiberPlatform.scala
@@ -49,6 +49,7 @@ private[effect] abstract class IOFiberPlatform[A] extends AtomicBoolean(false) {
         cb <- IO(new AtomicReference[Either[Throwable, Unit] => Unit](null))
 
         canInterrupt <- IO(new juc.Semaphore(0))
+        manyDone <- IO(new AtomicBoolean(false))
 
         target <- IO uncancelable { _ =>
           IO.async[Thread] { initCb =>
@@ -79,6 +80,14 @@ private[effect] abstract class IOFiberPlatform[A] extends AtomicBoolean(false) {
                     if (cb0 != null) {
                       cb0(RightUnit)
                     }
+                  } else {
+                    // wait for the hot loop to finish
+                    // we can probably also do this with canInterrupt, but that seems confusing
+                    // this needs to be a busy-wait otherwise it will be interrupted
+                    while (!manyDone.get()) {}
+                    Thread.interrupted() // clear the status
+
+                    ()
                   }
                 }
 
@@ -123,6 +132,7 @@ private[effect] abstract class IOFiberPlatform[A] extends AtomicBoolean(false) {
                       }
                     } finally {
                       canInterrupt.release()
+                      manyDone.set(true) // signal that we're done looping
                     }
                   }
                 }