dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4339138509
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -280,18 +338,106 @@ private[spark] class TaskContextImpl(
exception
}
- private def invokeTaskCompletionListeners(error: Option[Throwable]): Unit = {
+ /**
+ * Waits until the thread invoking listeners hands the invocation over to
`currentThread`, which
+ * has registered itself as `nextListenerInvocationThread`.
+ *
+ * Interrupts do not stop the wait, and are passed on once it is done. The
wait of
+ * `PipelinedWriterTask.awaitExit()` in `BasePythonRunner` drops them
instead, because it is a
+ * listener itself and the listeners right after it must still be able to
clean up. This wait
+ * is not a listener but part of `markTaskCompleted()`, so it leaves the
interrupt status (e.g.
+ * set by a task kill) as the completion listeners would see it without the
wait.
+ *
+ * A failure to log the wait (e.g. an OutOfMemoryError) ends it, unlike the
wait of
+ * `PipelinedWriterTask.awaitExit()`, which must go on since the writer
would otherwise use
+ * memory that the next listeners free. Ending this wait early leaves the
completion listeners
+ * to the other thread, so they may run after the task has released its
memory, but it lets a
+ * fatal error reach the executor, which then exits instead of waiting for
that thread.
+ */
+ private def awaitListenerInvocationHandover(currentThread: Thread): Unit = {
+ val startNs = System.nanoTime()
+ var interrupted = false
+ var warned = false
+ var handedOver = false
+ try {
+ while (!handedOver) {
+ val holder = synchronized {
+ val deadlineNs = System.nanoTime() +
+ TimeUnit.MILLISECONDS.toNanos(math.max(1L,
listenerHandoverWarnIntervalMs))
+ var remainingNs = deadlineNs - System.nanoTime()
+ while (!listenerInvocationThread.contains(currentThread) &&
remainingNs > 0) {
+ try {
+ wait(math.max(1L, TimeUnit.NANOSECONDS.toMillis(remainingNs)))
+ } catch {
+ case _: InterruptedException => interrupted = true
Review Comment:
**4. [Low] The SPARK-56330 test would now hang instead of failing on its
regression** (`TaskContextSuite.scala:228`, outside the diff)
"SPARK-56330: task completion during interrupt listener execution" guards
against interrupt listeners going through `invokeListeners()` again. Under that
regression, its interrupt thread would hold the invocation while its listener
waits on the semaphore that the test releases only after `verify`
(`TaskContextSuite.scala:237`). The test thread's `markTaskCompleted(None)`
(`TaskContextSuite.scala:228`) would then wait here forever, since this line
keeps the wait going on an interrupt and `SparkFunSuite`'s `failAfter` does not
interrupt anyway. So the core job would run into its time limit, as with the
`StreamingShuffleSuite` test in round 4 of #59332, whereas on master the same
regression fails `verify` right away. Its comment
(`TaskContextSuite.scala:230-233`, "would be silently skipped") also describes
the behavior that this PR removes.
Could the test call `markTaskCompleted` from `startDaemonThread`,
`joinThread` it before releasing the semaphore (released in a `finally`), and
say that the completion would wait for the interrupt thread?
--
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]
