cloud-fan commented on code in PR #52373:
URL: https://github.com/apache/spark/pull/52373#discussion_r2415548013
##########
core/src/test/scala/org/apache/spark/util/UninterruptibleThreadSuite.scala:
##########
@@ -80,130 +97,128 @@ class UninterruptibleThreadSuite extends SparkFunSuite {
t.interrupt()
interruptLatch.countDown()
t.join()
- assert(hasInterruptedException === false)
- assert(interruptStatusBeforeExit)
+ assert(!hasInterruptedException, "runUninterruptibly should not be
interrupted")
+ assert(interruptStatusBeforeExit, "interrupt flag should be set")
}
test("nested runUninterruptibly") {
val enterRunUninterruptibly = new CountDownLatch(1)
val interruptLatch = new CountDownLatch(1)
@volatile var hasInterruptedException = false
@volatile var interruptStatusBeforeExit = false
- val t = new UninterruptibleThread("test") {
+ val t = new UninterruptibleThread("runUninterruptibly") {
override def run(): Unit = {
runUninterruptibly {
enterRunUninterruptibly.countDown()
- Uninterruptibles.awaitUninterruptibly(interruptLatch, 10,
TimeUnit.SECONDS)
- hasInterruptedException = sleep(1)
- runUninterruptibly {
- if (sleep(1)) {
- hasInterruptedException = true
+ hasInterruptedException = await(interruptLatch)
+ if (!hasInterruptedException) {
+ runUninterruptibly {
+ hasInterruptedException = sleep(0)
}
- }
- if (sleep(1)) {
- hasInterruptedException = true
+ hasInterruptedException |= sleep(0)
}
}
interruptStatusBeforeExit = Thread.interrupted()
}
}
t.start()
- assert(enterRunUninterruptibly.await(10, TimeUnit.SECONDS), "await
timeout")
+ assert(!await(enterRunUninterruptibly), "await interrupted")
t.interrupt()
interruptLatch.countDown()
+ assert(!sleep(0), "sleep should not be interrupted")
+ t.interrupt()
+ assert(!sleep(0), "sleep should not be interrupted")
+ t.interrupt()
t.join()
- assert(hasInterruptedException === false)
- assert(interruptStatusBeforeExit)
+ assert(!hasInterruptedException, "runUninterruptibly should not be
interrupted")
+ assert(interruptStatusBeforeExit, "interrupt flag should be set")
}
test("no runUninterruptibly") {
@volatile var hasInterruptedException = false
- val t = new UninterruptibleThread("test") {
+ @volatile var interruptStatusBeforeExit = false
+ val t = new UninterruptibleThread("run") {
override def run(): Unit = {
- if (sleep(0)) {
- hasInterruptedException = true
- }
+ hasInterruptedException = sleep(0)
+ interruptStatusBeforeExit = Thread.interrupted()
}
}
t.interrupt()
t.start()
t.join()
- assert(hasInterruptedException === true)
+ assert(hasInterruptedException, "run should be interrupted")
+ assert(!interruptStatusBeforeExit, "interrupt flag should not be set")
}
test("SPARK-51821 uninterruptibleLock deadlock") {
- val latch = new CountDownLatch(1)
- val task = new UninterruptibleThread("task thread") {
+ val interruptLatch = new CountDownLatch(1)
+ val t = new UninterruptibleThread("run") {
override def run(): Unit = {
val channel = new AbstractInterruptibleChannel() {
override def implCloseChannel(): Unit = {
begin()
- latch.countDown()
+ interruptLatch.countDown()
try {
Thread.sleep(Long.MaxValue)
} catch {
- case _: InterruptedException =>
Thread.currentThread().interrupt()
+ case e: InterruptedException =>
+ log.info("sleep interrupted", e)
+ Thread.currentThread().interrupt()
}
}
}
channel.close()
}
}
- task.start()
- assert(latch.await(10, TimeUnit.SECONDS), "await timeout")
- task.interrupt()
- task.join()
+ t.start()
+ assert(!await(interruptLatch), "await interrupted")
+ t.interrupt()
+ t.join()
}
test("stress test") {
Review Comment:
how long does this test run now?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
