How to kill all threads caused by Scala .par when one throws and exception?

Question

This code blocks, the exceptions that are thrown don't kill the loop

List(1, 2, 3, 4, 5).par.foreach { i =>
        println("i = " + i)

        if (i == 5) {
          println("Sleeping forever")
          java.lang.Thread.sleep(Long.MaxValue)
        }

        throw new IllegalArgumentException("foo")
      } must throwA[IllegalArgumentException]

Is there a way to use .par but make it blow up properly?

I would just use plain `Futures` and `traverse` instead of `par` — Luis Miguel Mejía Suárez, Oct 11 '21 at 12:46
@LuisMiguelMejíaSuárez Worked like a charm. Post as an answer so I can accept it. — samthebest, Oct 16 '21 at 16:06

score 0 · Answer 1 · answered Oct 11 '21 at 10:32

I think that when using the par libs you should expect the exception to occour once all the threads actually finish and are are joined back into the current thread, I suspect this because by looking at the implementation of the foreach method (cmd clicking on foreach) a method named executeAndWaitResult is used

Here are some other q/s that seem somewhat similar perhaps it helps interrupt scala parallel collection How to cancel Future in Scala?

https://immutables.pl/2016/10/08/parallel-futures-and-exceptions/

samthebest · Accepted Answer · 2021-10-21T11:59:48.077

This seems to work, but verbose though

implicit class PimpedListScala[T](l: List[T]) {
    def parForeachThrowOnException(f: T => Unit, sleepMillis: Long): Unit = {
      var exception: Option[Throwable] = None

      val future = Future(
        l.par.foreach(e =>
          try {
            f(e)
          } catch {
            case t: Throwable =>
              if (exception.isEmpty) exception = Some(t)
          }
        )
      )

      while (exception.isEmpty && !future.isCompleted) {
        java.lang.Thread.sleep(sleepMillis)
      }

      exception.foreach(throw _)
    }
  }

I've also tried this

def parForeachThrowOnException(f: T => Unit): Unit =
      Await.result(Future.traverse(l)(a => Future(f(a))), Duration.Inf)

but this works unpredictably. For a live experiment it took a full 2 hours for the first exception thrown to propagate up and kill the application.

How to kill all threads caused by Scala .par when one throws and exception?

2 Answers2