viirya commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4404163533
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -412,21 +596,62 @@ private[spark] class TaskContextImpl(
// [[markTaskFailed]] here will have no effect and we simply
resume running the
// remaining completion handlers.
try {
- listenerInvocationThread = None
- markTaskFailed(e)
+ markTaskFailed(e, Nested)
} catch {
case t: Throwable => e.addSuppressed(t)
- } finally {
- synchronized {
- if (listenerInvocationThread.isEmpty) {
- listenerInvocationThread = Some(Thread.currentThread())
- }
- }
}
listenerExceptions += e
logError(log"Error in ${MDC(LISTENER, name)}", e)
}
}
+
+ def runLeftListeners(invoke: () => Unit): Unit = {
+ try {
+ invoke()
+ } 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 (!invokesFailureListeners && failureCauseOpt.isDefined &&
!onFailureCallbacks.empty()) {
+ val cause = failureCauseOpt.get
+ Some(() => runLeftListeners(() => invokeTaskFailureListeners(cause,
Nested)))
+ } else if (!listeners.empty()) {
+ val listener = listeners.pop()
+ Some(() => runListener(listener))
+ } else if (mode != Nested && nextListenerInvocationThread.isEmpty &&
completed &&
Review Comment:
Thanks, the `NON_TASK_WRITER` path makes this worse than the NPE. I don't
want to set the `TaskContext` on a thread that does not own the task either, so
the leftover branch is gone: as on master, only the thread that runs the
listeners' invocation decides what it runs, and a completion listener added
after `markTaskCompleted()` has run the others, while another thread holds the
invocation, stays in the stack. The broadcast lock is then released by
`releaseAllLocksForTask` for the right task. The only listeners an invocation
now runs on behalf of other threads are the failure listeners left to the
completion listener invocation of `markTaskCompleted()`, on the completing
thread, before teardown. The scaladoc about listeners having to use the context
passed to them is removed with it, and "listeners added after completion are
not run by another thread invoking listeners" checks both kinds of listener.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -280,27 +358,115 @@ 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. If the task
+ * body succeeded, it lets a fatal error reach the executor, which then
exits instead of waiting
+ * for that thread; if the task body failed, `runTaskWithListeners` adds the
error to the task's
+ * error as a suppressed exception. And if the other thread handed the
invocation over just
+ * before the failure, `markTaskCompleted()` releases it, so the completion
listeners run only
+ * once another listener is added. `markTaskCompleted()` also removes this
thread as the next
+ * invocation thread when the wait ends this way.
+ */
+ 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
+ }
+ remainingNs = deadlineNs - System.nanoTime()
+ }
+ handedOver = listenerInvocationThread.contains(currentThread)
+ if (handedOver) None else listenerInvocationThread
+ }
+ // Outside the lock: walking the stack of another thread stops it at a
safepoint, and
+ // logging may block, which would otherwise hold up every thread using
this TaskContext.
+ // The holder does not change during the wait, since it hands the
invocation to this
+ // thread, so its stack trace is attached to the first warning only.
+ holder match {
+ case Some(holder) =>
+ val holderStack = if (warned) None else Some(holder.getStackTrace)
+ // If the holder has handed over meanwhile, its stack shows
whatever it does next.
+ if (synchronized(listenerInvocationThread.contains(holder))) {
+ val waitedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNs)
+ val message = log"Task ${MDC(TASK_ATTEMPT_ID, taskAttemptId)}
has been waiting " +
+ log"for ${MDC(TOTAL_TIME, waitedMs)} ms to run its completion
listeners until " +
+ log"thread ${MDC(THREAD_NAME, holder.getName)} finishes
running its task listeners."
+ holderStack match {
+ case Some(trace) =>
+ val stack = new Exception(s"Stack trace of thread
${holder.getName}")
+ stack.setStackTrace(trace)
+ logWarning(message, stack)
Review Comment:
Done: a failure to warn no longer ends the wait. Like
`PipelinedWriterTask.awaitExit`, the wait keeps the first failure and goes on
(the wait itself is not covered), and `markTaskCompleted()` throws it once it
has run the completion listeners, with a failure of those listeners as a
suppressed exception. "a failing handover warning does not end the wait"
replaces the stale-waiter test: the completing thread runs the pending listener
after the handover and then throws the failure.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -412,21 +596,62 @@ private[spark] class TaskContextImpl(
// [[markTaskFailed]] here will have no effect and we simply
resume running the
// remaining completion handlers.
try {
- listenerInvocationThread = None
- markTaskFailed(e)
+ markTaskFailed(e, Nested)
Review Comment:
Both catches (`runListener` and `invokeTaskPreCompletionListener`) now catch
only `TaskCompletionListenerException`. I kept the nested invocation rather
than a plain `markTaskFailed(e)`: with item 1, only the completion listener
invocation of `markTaskCompleted()` runs failure listeners left to it, so in
any other invocation, e.g. a completion listener added after completion that
fails on the adding thread, a plain `markTaskFailed(e)` would record the
failure and never run the failure listeners, which master runs. "an error of a
failure listener invocation is not hidden by the failure of the listener that
started it" covers both catches.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -217,30 +240,85 @@ 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 run them only after its
own listeners, after
+ // the task has released its memory and reported its result. 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. 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. If such a task is killed, `spark.task.reaper.enabled` with a
positive
Review Comment:
Reworded the comment and the description along those lines. Having
`markInterrupted` only record the failure and leave the failure listeners to
the task thread would change where failure listeners run for killed tasks whose
interrupt listeners fail, which they have done on the kill thread since
SPARK-56330, so I'd leave it out of this PR.
--
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]
