viirya commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4361879829
##########
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
Review Comment:
You're right, and the comment and the description were wrong about the kill
path: on master the task thread releases the unroll memory, its block locks and
its memory before it blocks on the `TaskRunner` lock, and now it waits before
all of that. The comment and the description now say that the task waits before
it releases its block locks and memory, so a failure listener on another
thread, e.g. a kill thread, must not wait for either.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -412,21 +566,61 @@ 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 {
+ // The nested invocation has logged each listener error.
+ case e: Throwable => listenerExceptions += e
Review Comment:
Changed to merge the suppressed exceptions of a
`TaskCompletionListenerException` and let anything else propagate through the
release in the `finally`. The left-listener test now also checks that the
suppressed exceptions are the listeners' own failures and none is a
`TaskCompletionListenerException`.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -357,27 +521,17 @@ private[spark] class TaskContextImpl(
listenerInvocationThread = Some(Thread.currentThread())
Review Comment:
The allocations in `invokeListeners` are now above the claim.
`markTaskCompleted` runs the pre-completion loop, the wait and the invocation
in a `try` whose `finally` removes a waiter or holder that this call registered
and that is still there, so `awaitListenerInvocationHandover`'s `finally` only
restores the interrupt, and `warned` is no longer captured by a closure.
Two new tests use an appender that does not ignore its exceptions: in one,
logging a listener failure fails and a listener added afterwards still runs; in
the other, the handover warning fails and the holder then runs the completion
listeners itself instead of handing the invocation to the thread that gave up.
Removing either release now fails its test. You're right that the description's
sentence did not hold for these two; it now names what is not covered.
##########
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:
Done: `markTaskCompleted` now runs on a daemon thread that is joined with a
timeout before the semaphore is released in a `finally`, and the comment says
that the completion would wait for the interrupt thread.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -412,21 +566,61 @@ 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 {
+ // The nested invocation has logged each listener error.
+ case e: Throwable => listenerExceptions += e
+ }
+ }
+
+ // 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:
Added a test for the post-status-update invocation of a completed context:
the completion listener left to it runs on that thread, where
`TaskContext.get()` is null. The scaladoc now says that post-status-update
listeners are dropped rather than left, and that listeners left to the invoking
thread run on whatever thread that is, so they must use the context passed to
them. The description says 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]
