From daa808c0ac885f17129340ab196e8975b7bb8ccc Mon Sep 17 00:00:00 2001
From: Som Snytt <som.snytt@gmail.com>
Date: Sun, 28 Nov 2021 12:23:06 -0800
Subject: [PATCH] Do not merge throwable to itself

---
 .../scala/collection/parallel/Tasks.scala     | 10 ++--
 .../scala/collection/parallel/TaskTest.scala  | 46 ++++++++++++++++++-
 2 files changed, 50 insertions(+), 6 deletions(-)

diff --git a/core/src/main/scala/scala/collection/parallel/Tasks.scala b/core/src/main/scala/scala/collection/parallel/Tasks.scala
index ef82dbf4..fb1a278d 100644
--- a/core/src/main/scala/scala/collection/parallel/Tasks.scala
+++ b/core/src/main/scala/scala/collection/parallel/Tasks.scala
@@ -68,12 +68,12 @@ trait Task[R, +Tp] {
     mergeThrowables(that)
   }
 
-  private[parallel] def mergeThrowables(that: Task[_, _]): Unit = {
-     if (this.throwable != null && that.throwable != null)
-       this.throwable.addSuppressed(that.throwable)
-     else if (this.throwable == null && that.throwable != null)
+  private[parallel] def mergeThrowables(that: Task[_, _]): Unit =
+     if (this.throwable != null) {
+       if (that.throwable != null && (this.throwable ne that.throwable))
+         this.throwable.addSuppressed(that.throwable)
+     } else if (that.throwable != null)
        this.throwable = that.throwable
-  }
 
   // override in concrete task implementations to signal abort to other tasks
   private[parallel] def signalAbort(): Unit = {}
diff --git a/junit/src/test/scala/scala/collection/parallel/TaskTest.scala b/junit/src/test/scala/scala/collection/parallel/TaskTest.scala
index 089e5292..da2b8b19 100644
--- a/junit/src/test/scala/scala/collection/parallel/TaskTest.scala
+++ b/junit/src/test/scala/scala/collection/parallel/TaskTest.scala
@@ -29,7 +29,12 @@ class TaskTest {
         t
       }
     }
-    def mkPool(name: String) = new ForkJoinPool(1, mkFactory(name), null, false)
+    def mkPool(name: String) = {
+      val parallelism = 1
+      val handler: Thread.UncaughtExceptionHandler = null
+      val asyncMode = false
+      new ForkJoinPool(parallelism, mkFactory(name), handler, asyncMode)
+    }
 
     val one = List(1).par
     val two = List(2).par
@@ -48,4 +53,43 @@ class TaskTest {
     val r = c.filter(_ != 0).map(_ + 1)
     assertSame(myTs, r.tasksupport)
   }
+
+  // was: Wrong exception: expected scala.collection.parallel.TaskTest$SpecialControl$1 but was java.lang.IllegalArgumentException
+  @Test
+  def `t10276 exception does not suppress itself when merging`: Unit = {
+    import TestSupport._
+    import scala.util.control.ControlThrowable
+    class SpecialControl extends ControlThrowable("special")
+    val SpecialExcept = new SpecialControl
+
+    class Special {
+      def add(other: Special): Special = throw SpecialExcept
+    }
+
+    def listed(n: Int) = List.fill(n)(new Special)
+    val specials = listed(1000).par
+    assertThrows[SpecialControl](_ eq SpecialExcept)(specials.reduce(_ add _))
+  }
+}
+object TestSupport {
+  import scala.reflect.ClassTag
+  import scala.util.control.{ControlThrowable, NonFatal}
+  private val Unthrown = new ControlThrowable {}
+
+  def assertThrows[T <: Throwable: ClassTag](checker: T => Boolean)(body: => Any): Unit =
+    try {
+      body
+      throw Unthrown
+    } catch {
+      case Unthrown => fail("Expression did not throw!")
+      case e: T if checker(e) => ()
+      case failed: T =>
+        val ae = new AssertionError(s"Exception failed check: $failed")
+        ae.addSuppressed(failed)
+        throw ae
+      case NonFatal(other) =>
+        val ae = new AssertionError(s"Wrong exception: expected ${implicitly[ClassTag[T]]} but was ${other.getClass.getName}")
+        ae.addSuppressed(other)
+        throw ae
+    }
 }