Restructured failure case propagation

Author: djspiewak

kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala

@@ -141,6 +141,9 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] {
 
     F.deferred[Option[E]] flatMap { preempt =>
       F.ref[Set[Fiber[F, ?, ?]]](Set()) flatMap { supervision =>
+        // has to be done in parallel to avoid head of line issues
+        val cancelAllF = supervision.get.flatMap(_.toList.parTraverse_(_.cancel))
+
         MiniSemaphore[F](n) flatMap { sem =>
           val results = ta traverse { a =>
             preempt.tryGet flatMap {
@@ -154,10 +157,43 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] {
                   F.deferred[Outcome[F, E, B]] flatMap { result =>
                     val action = poll(sem.acquire) >> f(a)
                       .guaranteeCase { oc =>
-                        result.complete(oc) *> oc.fold(
-                          preempt.complete(None).void,
-                          e => preempt.complete(Some(e)).void,
-                          _ => F.unit) *> sem.release
+                        val completion = oc match {
+                          case Outcome.Succeeded(_) =>
+                            preempt.tryGet flatMap {
+                              case Some(Some(e)) =>
+                                result.complete(Outcome.Errored(e))
+
+                              case Some(None) =>
+                                result.complete(Outcome.Canceled())
+
+                              case None =>
+                                result.complete(oc)
+                            }
+
+                          case Outcome.Errored(e) =>
+                            preempt.complete(Some(e)) flatMap { won =>
+                              if (won)
+                                result.complete(oc) <* cancelAllF.start // avoid deadlock
+                              else
+                                preempt.get flatMap {
+                                  case Some(e) => result.complete(Outcome.Errored(e))
+                                  case None => result.complete(Outcome.Canceled())
+                                }
+                            }
+
+                          case Outcome.Canceled() =>
+                            preempt.complete(None) flatMap { won =>
+                              if (won)
+                                result.complete(oc) <* cancelAllF.start // avoid deadlock
+                              else
+                                preempt.get flatMap {
+                                  case Some(e) => result.complete(Outcome.Errored(e))
+                                  case None => result.complete(Outcome.Canceled())
+                                }
+                            }
+                        }
+
+                        completion *> sem.release
                       }
                       .void
                       .voidError
@@ -177,11 +213,7 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] {
             }
           }
 
-          results.flatMap(_.sequence) guaranteeCase {
-            case Outcome.Succeeded(_) => F.unit
-            // has to be done in parallel to avoid head of line issues
-            case _ => supervision.get.flatMap(_.toList.parTraverse_(_.cancel))
-          }
+          results.flatMap(_.sequence).onCancel(cancelAllF)
         }
       }
     }

tests/shared/src/test/scala/cats/effect/IOSpec.scala

@@ -1762,6 +1762,30 @@ class IOSpec extends BaseSpec with Discipline with IOPlatformSpecification {
 
         test.attempt.timeoutTo(500.millis, IO(false must beTrue)).as(ok)
       }
+
+      "run finalizers in parallel" in ticked { implicit ticker =>
+        // this test also tests to ensure that we get the errored results rather than cancels
+        // note that the first two effects will have a Canceled outcome, while the third is Errored
+        // if we just go by first wins in sequence, then Canceled is the (incorrect) result
+        // first wins *in time* is the expected semantic here
+        val test = for {
+          latch1 <- IO.deferred[Unit]
+          latch2 <- IO.deferred[Unit]
+
+          _ <- List(1, 2, 3).parTraverseN(3) {
+            case 1 =>
+              IO.never.onCancel(latch1.complete(()) *> latch2.get)
+
+            case 2 =>
+              IO.never.onCancel(latch2.complete(()) *> latch1.get)
+
+            case 3 =>
+              IO.sleep(10.millis) *> IO.raiseError(new RuntimeException)
+          }
+        } yield ()
+
+        test.attempt.void must completeAs(())
+      }
     }
 
     "parTraverseN_" should {