Merge pull request #3081 from durban/issue3075

Try to fix #3075
typelevel · Jul 11, 2022 · badc924 · badc924
2 parents eacb835 + eafb05a
commit badc924
Showing 1 changed file with 29 additions and 20 deletions.
diff --git a/core/jvm/src/main/scala/cats/effect/IOFiberPlatform.scala b/core/jvm/src/main/scala/cats/effect/IOFiberPlatform.scala
@@ -112,47 +112,56 @@ private[effect] abstract class IOFiberPlatform[A] extends AtomicBoolean(false) {
       } yield {
         Some {
           IO async { finCb =>
-            val trigger = IO {
+            val trigger = IO.defer {
               if (!many) {
                 cb.set(finCb)
               }
 
               // if done is false, and we can't get the semaphore, it means
               // that the action hasn't *yet* started, so we busy-wait for it
-              var break = true
-              while (break && !done.get()) {
-                if (canInterrupt.tryAcquire()) {
+
+              def busyWait(): IO[Unit] = { // NB: side-effecting, see `defer` above
+                if (done.get()) {
+                  IO.unit // ok, we're done
+                } else if (canInterrupt.tryAcquire()) {
                   try {
                     target.interrupt()
                   } finally {
-                    break = false
                     canInterrupt.release()
                   }
+                  IO.unit // ok, we've interrupted it
+                } else {
+                  IO.defer(busyWait()) // retry
                 }
               }
+
+              busyWait()
             }
 
             val repeat = if (many) {
-              IO {
-                // we need the outer try because we may be done *before* we enter
-                try {
-                  while (!done.get()) {
-                    if (canInterrupt.tryAcquire()) {
-                      try {
-                        while (!done.get()) {
-                          target.interrupt() // it's hammer time!
-                        }
-                      } finally {
-                        canInterrupt.release()
-                      }
+
+              def reallyTryToInterrupt(): Boolean = {
+                if (canInterrupt.tryAcquire()) {
+                  try {
+                    while (!done.get()) {
+                      target.interrupt() // it's hammer time!
                     }
+                  } finally {
+                    canInterrupt.release()
                   }
-                } finally {
-                  manyDone.set(true) // signal that we're done looping
+                  true // ok
+                } else {
+                  false // retry
                 }
+              }
 
-                finCb(RightUnit)
+              def loop: IO[Unit] = IO.defer {
+                if (done.get() || reallyTryToInterrupt()) IO.unit
+                else loop
               }
+
+              loop.guarantee(IO { manyDone.set(true) }) *> IO { finCb(RightUnit) }
+
             } else {
               IO {
                 if (done.get() && cb.get() != null) {