From 57e6f5ec6c58da4ba97a382fb8156c49e5ebb667 Mon Sep 17 00:00:00 2001 From: Vlad Rozov Date: Wed, 7 May 2025 13:36:44 +0800 Subject: [PATCH 1/2] [SPARK-51821][CORE] Call interrupt() without holding uninterruptibleLock to avoid possible deadlock ### What changes were proposed in this pull request? Do not hold `uninterruptibleLock` monitor while calling `super.interrupt()` in `UninterruptibleThread`, instead use newly introduced `awaitInterruptThread` flag and wait for `super.interrupt()` to be called. ### Why are the changes needed? There is potential deadlock as `UninterruptibleThread` may be blocked on NIO operation and interrupting channel while holding `uninterruptibleLock` monitor may cause deadlock like in ``` Found one Java-level deadlock: ============================= "pool-1-thread-1-ScalaTest-running-UninterruptibleThreadSuite": waiting to lock monitor 0x00006000036ee3c0 (object 0x000000070f3019d0, a java.lang.Object), which is held by "task thread" "task thread": waiting to lock monitor 0x00006000036e75a0 (object 0x000000070f70fe80, a java.lang.Object), which is held by "pool-1-thread-1-ScalaTest-running-UninterruptibleThreadSuite" Java stack information for the threads listed above: =================================================== "pool-1-thread-1-ScalaTest-running-UninterruptibleThreadSuite": at java.nio.channels.spi.AbstractInterruptibleChannel$1.interrupt(java.base17.0.14/AbstractInterruptibleChannel.java:157) - waiting to lock <0x000000070f3019d0> (a java.lang.Object) at java.lang.Thread.interrupt(java.base17.0.14/Thread.java:1004) - locked <0x000000070f70fc90> (a java.lang.Object) at org.apache.spark.util.UninterruptibleThread.interrupt(UninterruptibleThread.scala:99) - locked <0x000000070f70fe80> (a java.lang.Object) at org.apache.spark.util.UninterruptibleThreadSuite.$anonfun$new$5(UninterruptibleThreadSuite.scala:159) - locked <0x000000070f70f9f8> (a java.lang.Object) at org.apache.spark.util.UninterruptibleThreadSuite$$Lambda$216/0x000000700120d6c8.apply$mcV$sp(Unknown Source) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18) at org.scalatest.enablers.Timed$$anon$1.timeoutAfter(Timed.scala:127) at org.scalatest.concurrent.TimeLimits$.failAfterImpl(TimeLimits.scala:282) at org.scalatest.concurrent.TimeLimits.failAfter(TimeLimits.scala:231) at org.scalatest.concurrent.TimeLimits.failAfter$(TimeLimits.scala:230) at org.apache.spark.SparkFunSuite.failAfter(SparkFunSuite.scala:69) at org.apache.spark.SparkFunSuite.$anonfun$test$2(SparkFunSuite.scala:155) at org.apache.spark.SparkFunSuite$$Lambda$205/0x0000007001207700.apply(Unknown Source) at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85) at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83) at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104) at org.scalatest.Transformer.apply(Transformer.scala:22) at org.scalatest.Transformer.apply(Transformer.scala:20) at org.scalatest.funsuite.AnyFunSuiteLike$$anon$1.apply(AnyFunSuiteLike.scala:226) at org.apache.spark.SparkFunSuite.withFixture(SparkFunSuite.scala:227) at org.scalatest.funsuite.AnyFunSuiteLike.invokeWithFixture$1(AnyFunSuiteLike.scala:224) at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTest$1(AnyFunSuiteLike.scala:236) at org.scalatest.funsuite.AnyFunSuiteLike$$Lambda$343/0x00000070012867b0.apply(Unknown Source) at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306) at org.scalatest.funsuite.AnyFunSuiteLike.runTest(AnyFunSuiteLike.scala:236) at org.scalatest.funsuite.AnyFunSuiteLike.runTest$(AnyFunSuiteLike.scala:218) at org.apache.spark.SparkFunSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(SparkFunSuite.scala:69) at org.scalatest.BeforeAndAfterEach.runTest(BeforeAndAfterEach.scala:234) at org.scalatest.BeforeAndAfterEach.runTest$(BeforeAndAfterEach.scala:227) at org.apache.spark.SparkFunSuite.runTest(SparkFunSuite.scala:69) at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTests$1(AnyFunSuiteLike.scala:269) at org.scalatest.funsuite.AnyFunSuiteLike$$Lambda$339/0x00000070012833e0.apply(Unknown Source) at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:413) at org.scalatest.SuperEngine$$Lambda$340/0x0000007001283998.apply(Unknown Source) at scala.collection.immutable.List.foreach(List.scala:334) at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401) at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:396) at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:475) at org.scalatest.funsuite.AnyFunSuiteLike.runTests(AnyFunSuiteLike.scala:269) at org.scalatest.funsuite.AnyFunSuiteLike.runTests$(AnyFunSuiteLike.scala:268) at org.scalatest.funsuite.AnyFunSuite.runTests(AnyFunSuite.scala:1564) at org.scalatest.Suite.run(Suite.scala:1114) at org.scalatest.Suite.run$(Suite.scala:1096) at org.scalatest.funsuite.AnyFunSuite.org$scalatest$funsuite$AnyFunSuiteLike$$super$run(AnyFunSuite.scala:1564) at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$run$1(AnyFunSuiteLike.scala:273) at org.scalatest.funsuite.AnyFunSuiteLike$$Lambda$332/0x000000700127b000.apply(Unknown Source) at org.scalatest.SuperEngine.runImpl(Engine.scala:535) at org.scalatest.funsuite.AnyFunSuiteLike.run(AnyFunSuiteLike.scala:273) at org.scalatest.funsuite.AnyFunSuiteLike.run$(AnyFunSuiteLike.scala:272) at org.apache.spark.SparkFunSuite.org$scalatest$BeforeAndAfterAll$$super$run(SparkFunSuite.scala:69) at org.scalatest.BeforeAndAfterAll.liftedTree1$1(BeforeAndAfterAll.scala:213) at org.scalatest.BeforeAndAfterAll.run(BeforeAndAfterAll.scala:210) at org.scalatest.BeforeAndAfterAll.run$(BeforeAndAfterAll.scala:208) at org.apache.spark.SparkFunSuite.run(SparkFunSuite.scala:69) at org.scalatest.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:321) at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:517) at sbt.ForkMain$Run.lambda$runTest$1(ForkMain.java:414) at sbt.ForkMain$Run$$Lambda$107/0x0000007001110000.call(Unknown Source) at java.util.concurrent.FutureTask.run(java.base17.0.14/FutureTask.java:264) at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base17.0.14/ThreadPoolExecutor.java:1136) at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base17.0.14/ThreadPoolExecutor.java:635) at java.lang.Thread.run(java.base17.0.14/Thread.java:840) "task thread": at org.apache.spark.util.UninterruptibleThread.interrupt(UninterruptibleThread.scala:96) - waiting to lock <0x000000070f70fe80> (a java.lang.Object) at org.apache.spark.util.UninterruptibleThreadSuite$InterruptibleChannel.implCloseChannel(UninterruptibleThreadSuite.scala:143) at java.nio.channels.spi.AbstractInterruptibleChannel.close(java.base17.0.14/AbstractInterruptibleChannel.java:112) - locked <0x000000070f3019d0> (a java.lang.Object) at org.apache.spark.util.UninterruptibleThreadSuite$InterruptibleChannel.(UninterruptibleThreadSuite.scala:138) at org.apache.spark.util.UninterruptibleThreadSuite$$anon$5.run(UninterruptibleThreadSuite.scala:153) Found 1 deadlock. ``` ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Added 2 new test cases to the `UninterruptibleThreadSuite` ### Was this patch authored or co-authored using generative AI tooling? No Closes #50594 from vrozov/uninterruptible. Authored-by: Vlad Rozov Signed-off-by: Wenchen Fan (cherry picked from commit bb0b2d29d37e393280253fb4cb95fee461efaace) --- .../spark/util/UninterruptibleThread.scala | 125 +++++++++++++----- .../util/UninterruptibleThreadSuite.scala | 57 +++++++- 2 files changed, 149 insertions(+), 33 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/util/UninterruptibleThread.scala b/core/src/main/scala/org/apache/spark/util/UninterruptibleThread.scala index 24788d69121b..8fba5ed944c6 100644 --- a/core/src/main/scala/org/apache/spark/util/UninterruptibleThread.scala +++ b/core/src/main/scala/org/apache/spark/util/UninterruptibleThread.scala @@ -35,21 +35,90 @@ private[spark] class UninterruptibleThread( this(null, name) } - /** A monitor to protect "uninterruptible" and "interrupted" */ - private val uninterruptibleLock = new Object + private class UninterruptibleLock { + /** + * Indicates if `this` thread are in the uninterruptible status. If so, interrupting + * "this" will be deferred until `this` enters into the interruptible status. + */ + @GuardedBy("uninterruptibleLock") + private var uninterruptible = false - /** - * Indicates if `this` thread are in the uninterruptible status. If so, interrupting - * "this" will be deferred until `this` enters into the interruptible status. - */ - @GuardedBy("uninterruptibleLock") - private var uninterruptible = false + /** + * Indicates if we should interrupt `this` when we are leaving the uninterruptible zone. + */ + @GuardedBy("uninterruptibleLock") + private var shouldInterruptThread = false - /** - * Indicates if we should interrupt `this` when we are leaving the uninterruptible zone. - */ - @GuardedBy("uninterruptibleLock") - private var shouldInterruptThread = false + /** + * Indicates that we should wait for interrupt() call before proceeding. + */ + @GuardedBy("uninterruptibleLock") + private var awaitInterruptThread = false + + /** + * Set [[uninterruptible]] to given value and returns the previous value. + */ + def getAndSetUninterruptible(value: Boolean): Boolean = synchronized { + val uninterruptible = this.uninterruptible + this.uninterruptible = value + uninterruptible + } + + def setShouldInterruptThread(value: Boolean): Unit = synchronized { + shouldInterruptThread = value + } + + def setAwaitInterruptThread(value: Boolean): Unit = synchronized { + awaitInterruptThread = value + } + + /** + * Is call to [[java.lang.Thread.interrupt()]] pending + */ + def isInterruptPending: Boolean = synchronized { + // Clear the interrupted status if it's set. + shouldInterruptThread = Thread.interrupted() || shouldInterruptThread + // wait for super.interrupt() to be called + !shouldInterruptThread && awaitInterruptThread + } + + /** + * Set [[uninterruptible]] back to false and call [[java.lang.Thread.interrupt()]] to + * recover interrupt state if necessary + */ + def recoverInterrupt(): Unit = synchronized { + uninterruptible = false + if (shouldInterruptThread) { + shouldInterruptThread = false + // Recover the interrupted status + UninterruptibleThread.super.interrupt() + } + } + + /** + * Is it safe to call [[java.lang.Thread.interrupt()]] and interrupt the current thread + * @return true when there is no concurrent [[runUninterruptibly()]] call ([[uninterruptible]] + * is true) and no concurrent [[interrupt()]] call, otherwise false + */ + def isInterruptible: Boolean = synchronized { + shouldInterruptThread = uninterruptible + // as we are releasing uninterruptibleLock before calling super.interrupt() there is a + // possibility that runUninterruptibly() would be called after lock is released but before + // super.interrupt() is called. In this case to prevent runUninterruptibly() from being + // interrupted, we use awaitInterruptThread flag. We need to set it only if + // runUninterruptibly() is not yet set uninterruptible to true (!shouldInterruptThread) and + // there is no other threads that called interrupt (awaitInterruptThread is already true) + if (!shouldInterruptThread && !awaitInterruptThread) { + awaitInterruptThread = true + true + } else { + false + } + } + } + + /** A monitor to protect "uninterruptible" and "interrupted" */ + private val uninterruptibleLock = new UninterruptibleLock /** * Run `f` uninterruptibly in `this` thread. The thread won't be interrupted before returning @@ -63,27 +132,23 @@ private[spark] class UninterruptibleThread( s"Expected: $this but was ${Thread.currentThread()}") } - if (uninterruptibleLock.synchronized { uninterruptible }) { + if (uninterruptibleLock.getAndSetUninterruptible(true)) { // We are already in the uninterruptible status. So just run "f" and return return f } - uninterruptibleLock.synchronized { - // Clear the interrupted status if it's set. - shouldInterruptThread = Thread.interrupted() || shouldInterruptThread - uninterruptible = true + while (uninterruptibleLock.isInterruptPending) { + try { + Thread.sleep(100) + } catch { + case _: InterruptedException => uninterruptibleLock.setShouldInterruptThread(true) + } } + try { f } finally { - uninterruptibleLock.synchronized { - uninterruptible = false - if (shouldInterruptThread) { - // Recover the interrupted status - super.interrupt() - shouldInterruptThread = false - } - } + uninterruptibleLock.recoverInterrupt() } } @@ -92,11 +157,11 @@ private[spark] class UninterruptibleThread( * interrupted until it enters into the interruptible status. */ override def interrupt(): Unit = { - uninterruptibleLock.synchronized { - if (uninterruptible) { - shouldInterruptThread = true - } else { + if (uninterruptibleLock.isInterruptible) { + try { super.interrupt() + } finally { + uninterruptibleLock.setAwaitInterruptThread(false) } } } diff --git a/core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala b/core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala index 9c0ee1e1303e..fbc954d05af8 100644 --- a/core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala +++ b/core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala @@ -17,6 +17,7 @@ package org.apache.spark.util +import java.nio.channels.spi.AbstractInterruptibleChannel import java.util.concurrent.{CountDownLatch, TimeUnit} import scala.util.Random @@ -115,6 +116,45 @@ class UninterruptibleThreadSuite extends SparkFunSuite { assert(interruptStatusBeforeExit) } + test("no runUninterruptibly") { + @volatile var hasInterruptedException = false + val t = new UninterruptibleThread("test") { + override def run(): Unit = { + if (sleep(0)) { + hasInterruptedException = true + } + } + } + t.interrupt() + t.start() + t.join() + assert(hasInterruptedException === true) + } + + test("SPARK-51821 uninterruptibleLock deadlock") { + val latch = new CountDownLatch(1) + val task = new UninterruptibleThread("task thread") { + override def run(): Unit = { + val channel = new AbstractInterruptibleChannel() { + override def implCloseChannel(): Unit = { + begin() + latch.countDown() + try { + Thread.sleep(Long.MaxValue) + } catch { + case _: InterruptedException => Thread.currentThread().interrupt() + } + } + } + channel.close() + } + } + task.start() + assert(latch.await(10, TimeUnit.SECONDS), "await timeout") + task.interrupt() + task.join() + } + test("stress test") { @volatile var hasInterruptedException = false val t = new UninterruptibleThread("test") { @@ -148,9 +188,20 @@ class UninterruptibleThreadSuite extends SparkFunSuite { } } t.start() - for (i <- 0 until 400) { - Thread.sleep(Random.nextInt(10)) - t.interrupt() + val threads = new Array[Thread](10) + for (j <- 0 until 10) { + threads(j) = new Thread() { + override def run(): Unit = { + for (i <- 0 until 400) { + Thread.sleep(Random.nextInt(10)) + t.interrupt() + } + } + } + threads(j).start() + } + for (j <- 0 until 10) { + threads(j).join() } t.join() assert(hasInterruptedException === false) From c2c925301825671f4e838b2098bbf65fb9db97d4 Mon Sep 17 00:00:00 2001 From: Vlad Rozov Date: Wed, 7 May 2025 08:32:47 -0700 Subject: [PATCH 2/2] fix race in "no runUninterruptibly" test --- .../apache/spark/util/UninterruptibleThreadSuite.scala | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala b/core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala index fbc954d05af8..bbfad99f25a4 100644 --- a/core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala +++ b/core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala @@ -118,15 +118,16 @@ class UninterruptibleThreadSuite extends SparkFunSuite { test("no runUninterruptibly") { @volatile var hasInterruptedException = false + val latch = new CountDownLatch(1) val t = new UninterruptibleThread("test") { override def run(): Unit = { - if (sleep(0)) { - hasInterruptedException = true - } + latch.countDown() + hasInterruptedException = sleep(1) } } - t.interrupt() t.start() + latch.await(10, TimeUnit.SECONDS) + t.interrupt() t.join() assert(hasInterruptedException === true) }