[SPARK-51821][CORE] Call interrupt() without holding uninterruptibleLock to avoid possible deadlock

[vrozov]

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.base@17.0.14/AbstractInterruptibleChannel.java:157)
        - waiting to lock <0x000000070f3019d0> (a java.lang.Object)
        at java.lang.Thread.interrupt(java.base@17.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.base@17.0.14/FutureTask.java:264)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.14/ThreadPoolExecutor.java:1136)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.14/ThreadPoolExecutor.java:635)
        at java.lang.Thread.run(java.base@17.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.base@17.0.14/AbstractInterruptibleChannel.java:112)
        - locked <0x000000070f3019d0> (a java.lang.Object)
        at org.apache.spark.util.UninterruptibleThreadSuite$InterruptibleChannel.<init>(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

[HyukjinKwon]

cc @Ngone51 FYI

[wangyum]

Thank you @vrozov, I'll verify it later.

[Ngone51]

It doesn't seem to be a right use case of UninterruptibleThread. Shouldn't the whole block be executed within runUninterruptibly()? e.g.,

override def run(): Unit = {
  this.runUninterruptibly {
    ...
  }
}

The task thread can be correctly interrupted if the whole block run inside runUninterruptibly() with the limited Thread.sleep().

[Ngone51]

I see where the problem is. Spark task always uses the UninterruptibleThread but this.runUninterruptibly() is only called for the Spark task that run with KafkaConsumer (#17761 is the original PR that introduced UninterruptibleThread).

[vrozov]

UninterruptibleThread was introduced as part of SPARK-14169 and by design can be used to run and runUninterruptibly.

[mridulm]

Streaming also appears to use runUninterruptibly.
Are we proposing to adapt it to not need this construct @Ngone51 ?

[Ngone51]

Yes, I think we should avoid using UninterruptibleThread when it is not really needed. The problem is that executor can't distinguish the tasks for different workloads. So we have to compromise with UninterruptibleThread as the default task thread and call runUninterruptibly when it is necessary.

[vrozov]

@mridulm @Ngone51 Streaming requires UninterruptibleThread, please see SPARK-21248. I also don't think that it is necessary to revisit usage of UninterruptibleThread. The run() method is not affected at all. The only affected method (overridden) is interrupt() and with the fix it also won't be impacted. The only difference with Thread.interrupt() is acquiring uninterruptibleLock that is a low cost operation when there is no contention (multiple threads calling interrupt() concurrently) and as Thread.interrupt() acquires blockerLock as well, there is pretty much no difference at all.

[mridulm]

The test would fail even if we replace UninterruptibleThread with regular Thread right ?
I am probably missing some context here.

[vrozov]

@mridulm Do you refer to "SPARK-51821 uninterruptibleLock deadlock" test? No, it will not fail. Why do you think it would fail?

[cloud-fan]

code style nit: shall we move the if condition into a method like def shouldDoThreadInterruption?

[vrozov]

@cloud-fan is this covered by code style guidelines? If not, I'll make the change in case other changes are necessary, otherwise, let's keep it this way.

[cloud-fan]

It's not a code style guideline thing, it's like the Java best practice that we should not have super long methods, the if condition shouldn't be very long either.

[mridulm]

Agree, refactoring into appropriate method(s) improves clarity and readability

[vrozov]

Neither synchronized blocks are very long. For example

spark/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala

Line 773 in ef336ad

allocationManager.synchronized {

is much longer.

I'll make the change if another revision is necessary.

[vrozov]

OK

[cloud-fan]

Let's not get blocked by a code style nit. @vrozov If you don't mind I can help to fix the code style via the Github UI and then we can move forward.

[cloud-fan]

Or I can merge it as it is and open a follow-up PR to fix the code style. Which way do you prefer?

[vrozov]

@cloud-fan Please wait. I was working on a different PR that I am now ready to open. I will fix the code style on Monday.

[vrozov]

@cloud-fan @mridulm Introduced UninterruptibleLock class with 2 helper methods.

[cloud-fan]

is Thread.interrupted() a cheap function? Previously we only call it once, but now it's many times.

[cloud-fan]

shall we do it separately after this while loop?

[mridulm]

It would effectively be the same @cloud-fan - interrupted is same is isInterrupted when thread has not been interrupted.
And when it is interrupted, we are going to break out of this loop anyway (so only once).

[vrozov]

is Thread.interrupted() a cheap function

Yes, it is. It is as cheap as getting volatile boolean.

Previously we only call it once, but now it's many times.

In the best and most common case scenarios it is called once.

shall we do it separately after this while loop?

No, we don't want to enter sleep if interrupted() was already called.

[mridulm]

@mridulm Do you refer to "SPARK-51821 uninterruptibleLock deadlock" test? No, it will not fail. Why do you think it would fail?

My bad, there was some other local changes which messed up the test

[mridulm]

Ok, I see the problem.
The expectation is for begin and end to be within try/finally - but usually coding pattern would result in the try being used to catch InterruptedException and handle it there (via interrupt).

So instead of :

try {
  begin();
  try {
    expensiveOp();
  } finally {
    end();
  }
} catch (InterruptedException ex) {
  Thread.currentThread.interrupt();
}

it is common to do:

begin();
try {
  expensiveOp();
} catch (InterruptedException ex) {
  Thread.currentThread.interrupt();
} finally {
  end();
}

While former is how it should be perhaps done, it is not realistic to expect everyone to do so.
Thanks for the PR, let me take a look at the proposed change

[mridulm]

Actually, I think my formulation above will still have deadlock - though for a different reason, sigh.

[mridulm]

Interesting work @vrozov !
Thanks for contributing this.

[mridulm]

Let us move this into a private method, instead of inlining it within while.
Same for if in interrupt

[vrozov]

Same as #50594 (comment)

[wangyum]

@mridulm @vrozov Is it possible just move super.interrupt() out of synchronized block?

[mridulm]

I have to think through it @wangyum, but that does look like a very promising idea !
I am also wondering if the interrupt in runUninterruptibly needs to be within the synchronized block as well (just set a local bool within finally to decide whether to interrupt or not).

To reviewers, for ref, see here

[cloud-fan]

@wangyum I don't think this works. After we determine needInterrupt and release the lock, def runUninterruptibly may proceed and start to run the function f. Then def interrupt proceeds and calls super.interrupt, the function f gets interrupted which is not expected.

That's why we need an extra wait in def runUninterruptibly, to wait for super.interrupt to complete, and then proceed to reset the interrupted stats and run the function f.

[vrozov]

@mridulm

Actually, I think my formulation above will still have deadlock - though for a different reason, sigh.

I don't think that begin()/end() are supposed to be used as

try {
  begin();
  try {
    expensiveOp();
  } finally {
    end();
  }
} catch (InterruptedException ex) {
  Thread.currentThread.interrupt();
}

as end() would throw ClosedByInterruptException in finally and hide InterruptedException.

[mridulm]

Thanks @cloud-fan - that is exactly the scenario I missed with a quick look !

[vrozov]

Thank you @vrozov, I'll verify it later.

@wangyum did you have a chance to verify the fix?

[wangyum]

@wangyum did you have a chance to verify the fix?
Yes, it is works as expected.

[cloud-fan]

It's not very intuitive to know the relationship between the boolean return value and the function name. So returning true means keep waiting? shall we name the function shouldAwaitInterrupt?

[vrozov]

@cloud-fan did more refactoring, please take a look

[cloud-fan]

thanks, merging to master/4.0!

[vrozov]

Should it be also backported to 3.5 branch?

[cloud-fan]

we can, @vrozov can you open a backport PR for 3.5?

[vrozov]

It looks that Java 8 behaves differently when interrupt() is called on not started thread.

[vrozov]

@cloud-fan #50594