[Forked] [SPARK-53394][CORE] UninterruptibleLock.isInterruptible should avoid duplicated interrupt

[shkhrgpt]

Original Description:

What changes were proposed in this pull request?

This PR fixes UninterruptibleLock.isInterruptible to avoid duplicated interruption when the thread is already interrupted.

Why are the changes needed?

The "uninterruptible" semantic of UninterruptibleThreadis broken (i.e., UninterruptibleThread is interruptible even if it's under runUninterruptibly) after the fix apache#50594. The probelm is that the state of
shouldInterruptThread becomes unsafe when there are multiple interrupts concurrently.

For example, thread A could interrupt UninterruptibleThread ut first before UninterruptibleThread enters runUninterruptibly. Right after that, another thread B starts to invoke ut.interrupt() and pass through uninterruptibleLock.isInterruptible (becasue at this point, shouldInterruptThread = uninterruptible = false). Before thread B invokes super.interrupt(), UninterruptibleThread ut enters runUninterruptibly and pass through uninterruptibleLock.getAndSetUninterruptible and set uninterruptible = true. Then, thread ut continues the check uninterruptibleLock.isInterruptPending. However, uninterruptibleLock.isInterruptPending return false at this point (due to shouldInterruptThread = Thread.interrupted = true) even though thread B is actully interrupting. As a result, the state of shouldInterruptThread becomes inconsistent between thread B and thread ut. Then, as uninterruptibleLock.isInterruptPending returns false, ut to continute to execute f. At the same time, thread B invokes super.interrupt(), and f could be interrupted

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Manually tested. The issue can be easily reproduced if we run UninterruptibleThreadSuite.stress test for 100 times in a row:

[info]   true did not equal false (UninterruptibleThreadSuite.scala:208)
[info]   org.scalatest.exceptions.TestFailedException:
[info]   at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
[info]   at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
[info]   at org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1231)
[info]   at org.scalatest.Assertions$AssertionsHelper.macroAssert(Assertions.scala:1295)
[info]   at org.apache.spark.util.UninterruptibleThreadSuite.$anonfun$new$7(UninterruptibleThreadSuite.scala:208)
...

And the issue is gone after the fix.

Was this patch authored or co-authored using generative AI tooling?

No

Original Author: Ngone51