HyukjinKwon commented on code in PR #59332:
URL: https://github.com/apache/spark/pull/59332#discussion_r4256691547
##########
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:
Thanks, verified in d05f9547fea. The `markTaskCompleted` comment now says
that the reaper bounds the handover wait only when the blocked listeners do not
run within its own kill call, and that otherwise only the periodic warning
shows the hang. Item 3 of the PR description now describes the single
`PipelinedWriterState` as well.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4255354335","thread_id":"inline:4255354335","verdict_sha256":"ef0b8c6e5798ce23390828b19412cb78862f22a318a0e4cc0fb82794e99a9180"}
-->
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -320,21 +430,44 @@ 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)
}
}
+ if (mode != Nested) {
+ // We have executed all listeners of `listenerStacks` that have been
added so far. Run the
+ // listeners left to this thread (see above), then deregister this
thread as the callback
+ // invocation thread, or hand the invocation over to a thread waiting
for it.
+ var released = false
+ while (!released) {
+ val pendingListeners: Option[() => Unit] = synchronized {
+ val pendingFailure = failureCauseOpt.filter(_ =>
!onFailureCallbacks.empty())
+ if (nextListenerInvocationThread.isEmpty &&
pendingFailure.isDefined) {
+ Some(() => invokeTaskFailureListeners(pendingFailure.get, Nested))
+ } else if (nextListenerInvocationThread.isEmpty && completed &&
+ (!onPreCompleteCallbacks.empty() ||
!onCompleteCallbacks.empty())) {
+ Some(() => invokeTaskCompletionListeners(None, Nested))
+ } else {
+ listenerInvocationThread = nextListenerInvocationThread
+ nextListenerInvocationThread = None
+ notifyAll()
+ released = true
+ None
+ }
+ }
+ pendingListeners.foreach { invoke =>
+ try {
+ invoke()
+ } catch {
+ case e: Throwable => listenerExceptions += e
Review Comment:
**Nit (P3):** This catch keeps the release loop reaching `released = true`
and `notifyAll()` when a deferred listener throws, and it puts that failure
into the reported exception. Both new release-loop tests ("failure listeners
queued during completion run before it releases" and "completion listeners
queued meanwhile run before failure invoker releases") use listeners that do
not throw, so nothing exercises this branch. If a later change dropped the
catch, the nested invocation's `TaskCompletionListenerException` would leave
the loop early with this thread still registered as `listenerInvocationThread`.
Later invocations, including `invokePostStatusUpdateListeners`, would return
without running anything, and a `markTaskCompleted` waiting for the handover
would never be woken. If it swallowed the exception instead, the failure would
never reach the caller. Neither change would fail a current test.
**Verification:**
- **Regression:** The test fails when the release loop lets a deferred
listener's exception escape before releasing the invocation, or swallows it,
and passes with the current implementation.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -169,37 +196,113 @@ 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
+ // it may stay in this state, by killing the executor JVM, but only if the
blocked listeners
+ // do not run within the reaper's own kill call: the reaper calls
`Task.kill` before it starts
+ // checking the timeout, and that call runs the failure listeners on the
reaper thread if an
+ // interrupt listener throws. Then only the periodic warning shows the
hang.
+ val mode = synchronized {
if (completed) return
completed = true
+ val currentThread = Thread.currentThread()
+ if (listenerInvocationThread.isEmpty) {
+ listenerInvocationThread = Some(currentThread)
+ Claimed
+ } else if (listenerInvocationThread.contains(currentThread)) {
+ // Called from a listener on this thread: run the completion listeners
as part of that
+ // invocation.
+ Nested
+ } else {
+ nextListenerInvocationThread = Some(currentThread)
+ awaitListenerInvocationHandover(currentThread)
+ Claimed
+ }
+ }
+ invokeTaskCompletionListeners(error, mode)
+ }
+
+ /**
+ * Waits until another thread invoking listeners hands the invocation over
to `currentThread`.
+ * Interrupts do not stop the wait and are restored afterwards. Must hold
the TaskContext lock.
+ */
+ private def awaitListenerInvocationHandover(currentThread: Thread): Unit = {
+ val startNs = System.nanoTime()
+ var nextWarningNs = startNs +
TimeUnit.MILLISECONDS.toNanos(listenerHandoverWarnIntervalMs)
+ var interrupted = false
+ while (!listenerInvocationThread.contains(currentThread)) {
+ try {
+ wait(math.max(1L, TimeUnit.NANOSECONDS.toMillis(nextWarningNs -
System.nanoTime())))
+ } catch {
+ case _: InterruptedException => interrupted = true
Review Comment:
**Nit (P3):** This catch keeps the handover wait going on an interrupt, and
the restore below hands the interrupt back once the invocation arrives. Both
matter on the kill path: `Task.kill` interrupts the task thread, which can land
while it waits here for a kill thread that is running failure listeners. If a
later change let the `InterruptedException` escape, `markTaskCompleted` would
throw with `completed` already set and `nextListenerInvocationThread` still
naming this thread. The holder would then hand the invocation to a thread that
has left, `listenerInvocationThread` would stay set, and none of the completion
listeners (including the pipelined writer stop and the memory cleanup) would
run. If the restore were dropped, the kill's interrupt would be silently lost.
"completion waits for another thread invoking listeners and runs them" never
interrupts the completion thread, so no current assertion fails in either case.
**Verification:**
- **Regression:** The test fails when the handover wait propagates
InterruptedException or does not restore the interrupt flag, and passes with
the current implementation.
- **Behavior:** After the interrupted wait, every pending completion
listener runs on the completing thread.
--
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]
