viirya commented on code in PR #59332:
URL: https://github.com/apache/spark/pull/59332#discussion_r4255649563
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -169,37 +196,110 @@ private[spark] class TaskContextImpl(
}
private[spark] override def markTaskFailed(error: Throwable): Unit = {
+ markTaskFailed(error, Claim)
+ }
+
+ /**
+ * @param mode `Nested` if this thread is already invoking listeners, i.e. a
listener failed. The
+ * failure listeners then run as part of that invocation, which
this thread keeps.
+ */
+ private def markTaskFailed(error: Throwable, mode: InvocationMode): Unit = {
synchronized {
if (failureCauseOpt.isDefined) return
failureCauseOpt = Some(error)
}
- invokeTaskFailureListeners(error)
+ invokeTaskFailureListeners(error, mode)
}
private[spark] override def markTaskCompleted(error: Option[Throwable]):
Unit = {
- synchronized {
+ // The completion listeners always run on this thread, and this method
returns only after they
+ // have run. If any other thread could run them, it could be one the
listeners are meant to
+ // stop and wait for, e.g. a pipelined Python UDF writer that adds
listeners lazily while it
+ // pulls its input, which would then wait for itself. And if this method
returned before they
+ // ran, the task could release its memory while that writer still uses it.
So the listener
+ // invocation is claimed together with setting `completed`; if another
thread holds it (e.g. a
+ // kill thread running failure listeners), that thread hands it over to
this one when done.
+ //
+ // That wait has no time bound: if the other thread's listeners block, so
does this task. It
+ // logs a warning with that thread's stack trace periodically. If such a
task is killed,
+ // `spark.task.reaper.enabled` with a positive
`spark.task.reaper.killTimeout` bounds how long
Review Comment:
Good catch, confirmed: `TaskReaper.run` calls `taskRunner.kill()` before it
starts checking `killTimeout`, so when the blocked failure listeners run within
that call, the reaper never gets to the timeout. Reworded the comment in
d05f9547fea: the reaper bounds the wait only if the blocked listeners do not
run within its own kill call, and otherwise only the periodic warning shows the
hang. Also updated item 3 of the PR description to describe the single
`PipelinedWriterState` instead of the old `AtomicBoolean` claim.
--
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]
