viirya commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4421487928
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -266,7 +352,10 @@ private[spark] class TaskContextImpl(
try {
markTaskFailed(e)
} catch {
- case t: Throwable => e.addSuppressed(t)
+ // Failures of the failure listeners. Any other error comes from the
invocation itself
+ // (e.g. an OutOfMemoryError) and propagates, so that the executor
still sees a fatal
+ // one.
+ case t: TaskCompletionListenerException => e.addSuppressed(t)
Review Comment:
Good catch, `getMessage` is listener code. All three catches now take
`NonFatal` errors, as you suggested, so an `OutOfMemoryError` still propagates.
The same applied to the catch in `markTaskCompleted` after the completion
listener invocation: a non-fatal error that escapes it would have dropped the
pre-completion listeners' failures, so it now adds such an error to them too.
"an exception whose getMessage throws does not end the listener loops" covers
the pre-completion, completion and left failure listener loops.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -218,30 +243,91 @@ 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, and the
task would go on to
+ // release its memory with them pending. 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 that another thread leaves to this
invocation run before the
+ // next completion listener (see `invokeListeners()`). A completion
listener added after this
+ // method has run the others runs on the thread adding it, unless another
thread is invoking
+ // listeners at that moment: that thread does not run it (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. The task waits
+ // here before it releases its block locks and memory (the executor does
that once `Task.run`
+ // returns), so a failure listener on another thread, e.g. a kill thread,
must not wait for
+ // either. Nothing bounds a wait for a kill thread: with
`spark.task.reaper.enabled`, every
+ // kill runs on a reaper thread, which checks
`spark.task.reaper.killTimeout` only after its
+ // kill call returns. If another thread holds the invocation and the task
is killed, the reaper
+ // with a positive kill timeout bounds the wait by killing the executor
JVM, except in local
+ // mode, where it only logs an error.
+ val currentThread = Thread.currentThread()
+ var mode: InvocationMode = Claimed
+ var awaitHandover = false
+ // Whether this call registered this thread as the listener invocation
thread or as the next
+ // one, so that it must make sure not to leave it registered.
+ var registered = false
val preCompletionExceptions = new ArrayBuffer[Throwable](2)
def nextPreCompletionListener(): Option[TaskCompletionListener] =
synchronized {
if (onPreCompleteCallbacks.empty()) {
// Set together with finding no more pre-completion listeners, so that
one added later
- // is handed to `addTaskCompletionListener` instead of being left in
the stack.
+ // is handed to `addTaskCompletionListener` instead of being left in
the stack. The
+ // listener invocation is claimed in the same step (see above).
completed = true
+ if (listenerInvocationThread.isEmpty) {
+ listenerInvocationThread = Some(currentThread)
+ registered = true
+ } else if (listenerInvocationThread.contains(currentThread)) {
+ // Called from a listener on this thread: run the completion
listeners as part of that
+ // invocation.
+ mode = Nested
+ } else {
+ nextListenerInvocationThread = Some(currentThread)
+ awaitHandover = true
+ registered = true
+ }
None
} else {
Some(onPreCompleteCallbacks.pop())
}
}
var listener: Option[TaskCompletionListener] = None
- while ({listener = nextPreCompletionListener(); listener.nonEmpty}) {
- invokeTaskPreCompletionListener(listener.get, preCompletionExceptions)
- }
-
try {
- invokeTaskCompletionListeners(error)
- } catch {
- case e: TaskCompletionListenerException if
preCompletionExceptions.nonEmpty =>
- // Merge the completion listeners' own failures, not the exception
that wraps them. Any
- // other error comes from invokeListeners itself (e.g. an
OutOfMemoryError) and is left
- // to propagate, so that the executor still sees a fatal one.
- preCompletionExceptions ++= e.getSuppressed.toSeq
+ while ({listener = nextPreCompletionListener(); listener.nonEmpty}) {
+ invokeTaskPreCompletionListener(listener.get, preCompletionExceptions)
+ }
+ try {
+ val warningFailure =
+ if (awaitHandover) awaitListenerInvocationHandover(currentThread)
else None
+ try {
+ invokeTaskCompletionListeners(error, mode)
+ } catch {
+ // The failure to warn came first, so it is the one thrown.
+ case e: Throwable if warningFailure.exists(_ ne e) =>
warningFailure.get.addSuppressed(e)
+ }
+ warningFailure.foreach(throw _)
Review Comment:
Done: if a pre-completion listener failed, a non-fatal failure to warn is
added to `preCompletionExceptions` after their failures, so the
`TaskCompletionListenerException` reports both, with the completion listeners'
failures as suppressed exceptions of the failure to warn. The comment at L308
now says that the failure to warn comes after the pre-completion listeners'
failures and before those of the completion listeners. "a failing handover
warning is reported with the failures of the pre-completion listeners" covers
it.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -449,6 +641,51 @@ private[spark] class TaskContextImpl(
}
}
}
+
+ def runLeftFailureListeners(cause: Throwable): Unit = {
+ try {
+ invokeTaskFailureListeners(cause, Nested)
+ } catch {
+ // Merge the listeners' own failures, which the nested invocation has
logged, not the
+ // exception that wraps them. Any other error comes from that
invocation itself (e.g. an
+ // OutOfMemoryError) and propagates, through the release below.
+ case e: TaskCompletionListenerException => listenerExceptions ++=
e.getSuppressed.toSeq
+ }
+ }
+
+ // Chooses the next thing to do under the TaskContext lock, to be done
outside of it.
+ def nextStep(): Option[() => Unit] = synchronized {
+ if (completesTask && failureCauseOpt.isDefined &&
!onFailureCallbacks.empty()) {
Review Comment:
Right, it runs before the completion listeners that have not run yet. I
reworded the scaladoc and the description that way. This is also the order in
which a failing completion listener runs the failure listeners, on master as
well (case 3 in `runListener`): the completion listeners before it have run,
and the failures of the failure listeners are reported in what
`markTaskCompleted` throws, with `error` as the previous error. The left
failure listeners now follow the same rule, so I'd keep it that way rather than
special-case them; the scaladoc says that they are reported like those.
--
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]
