dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4339138527
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -217,12 +239,45 @@ private[spark] class TaskContextImpl(
// writer that adds listeners lazily while it pulls its input. And since
the completion
// listeners only run after `completed` is set, they run after every
pre-completion listener,
// even if another thread is invoking listeners at this point.
+ //
+ // The completion listeners then run on this thread too, and this method
returns only after
+ // they have run. If it returned while another thread is invoking
listeners (e.g. a kill thread
+ // running failure listeners), that thread would not run them, so cleanup
would be skipped,
+ // and the task would release its memory and report its result anyway. So
the listener
+ // invocation is claimed together with setting `completed`, which also
keeps a thread that adds
+ // a completion listener at that moment from running them; if another
thread holds it, that
+ // thread hands it over to this one when done. Failure listeners pending
at that point run
+ // first (see `invokeListeners()`). A completion listener added after this
method has run the
+ // others runs on the thread adding it, or, if another thread is invoking
listeners at that
+ // moment, on that thread before it stops (see `invokeListeners()`).
+ //
+ // The wait for the other thread has no time bound: if its listeners
block, so does this task.
+ // It logs a warning periodically, the first one with that thread's stack
trace. A kill thread
+ // runs the failure listeners holding the lock of the executor's
TaskRunner, which the task
+ // thread needs to finish anyway, so this only adds a wait for other
threads, e.g. one that
+ // failed the task from `Utils.tryWithSafeFinallyAndFailureCallbacks`. If
such a task is
+ // killed, `spark.task.reaper.enabled` with a positive
`spark.task.reaper.killTimeout` bounds
Review Comment:
**6. [Low] The reaper does not bound this wait in local mode**
In local mode the reaper only logs that it is "not killing JVM because we
are running in local mode" (`Executor.scala:1324-1327`), so with `local[*]` the
wait stays unbounded even with a positive `spark.task.reaper.killTimeout`. The
scaladoc of `startPipelinedWriter` says so for its own wait
(`PythonRunner.scala:210-212`, "except in local mode, where the reaper only
logs an error"). Could this comment and the description ("bounds it in that
case") say the same?
--
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]
