dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4419200062
##########
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:
**1. [Low] Catching only `TaskCompletionListenerException` lets a non-fatal
error end the listener loops**
Round 2 narrowed this catch, the one in `runListener` (L631) and the one in
`runLeftFailureListeners` (L652), as [round-2 item
3](https://github.com/apache/spark/pull/59663#discussion_r4371913169) asked, so
that a fatal error from the nested invocation reaches the executor. They now
also let through non-fatal errors other than `TaskCompletionListenerException`,
and the nested invocation can throw one without any injected error: it builds
its exception with `listenerExceptions.map(_.getMessage)` (L690-691), and
`getMessage` is listener code. Logging such an exception is best effort, but
this call is not covered.
So when a failure listener throws an exception whose `getMessage` throws a
`RuntimeException`:
- Here, after a pre-completion listener failed, the error escapes before
L360 records `e` and before it is logged. It ends the loop at L299-301 before
`nextPreCompletionListener()` sets `completed` (L279), so no completion
listener runs, and one added later is only queued.
- In `runListener`, after a completion listener failed, the error escapes
the loop: the remaining completion listeners are skipped, and `e` is neither
logged nor reported.
- In `runLeftFailureListeners`, the same happens to the completion listeners
after it.
On master, `case t: Throwable => e.addSuppressed(t)` kept these loops going.
Since the comments say the point is to let fatal errors through, could these
catch `NonFatal` errors instead? That is `case NonFatal(t) =>
e.addSuppressed(t)` here and at L631, and `case NonFatal(t) =>
listenerExceptions += t` after the case at L652. An `OutOfMemoryError` would
still propagate.
##########
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:
**3. [Nit] A failure listener left to the completing thread can run after
the completion listener that released its resource**
L536-538 and the description say that this lets a failure listener that
aborts a state store update run before the completion listener that releases
the state store. That holds only if that completion listener has not run yet:
this branch runs the left failure listeners before the next completion
listener, after the ones already popped. On master they were dropped. Now they
can run on a resource that is already released, and their failures go into what
`markTaskCompleted` throws.
For example, `RocksDBStateStore` registers both listeners when it opens a
store (`RocksDBStateStoreProvider.scala:175-195`). For a read-only store, the
completion listener calls `release()`
(`RocksDBStateStoreProvider.scala:179-180`), which sets `RELEASED`. If another
thread marks the task failed after that, the failure listener left here calls
`abort()` (`RocksDBStateStoreProvider.scala:194`), and `validateState(UPDATING,
ABORTED)` (`RocksDBStateStoreProvider.scala:740`) throws
`stateStoreOperationOutOfOrder` (`RocksDBStateStoreProvider.scala:119-122`).
After a kill, the task is reported as killed anyway. After a background
`markTaskFailed`, a task whose body succeeded now fails with that error as its
reason. The failure that caused it is not in the exception, which carries
`markTaskCompleted`'s `error` (L690-691), `None` here.
Could L536-538 and the description say "before the completion listeners that
have not run yet"? It may also be worth a sentence whether such late failures
should fail the completion, or carry `failureCauseOpt` as the previous error.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -288,27 +377,120 @@ 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 warn (e.g. an OutOfMemoryError while logging) does not end
the wait either, like
+ * in `PipelinedWriterTask.awaitExit()`: the other thread does not run the
completion listeners,
+ * and the task would release its memory with them pending. The first such
failure is returned,
+ * for `markTaskCompleted()` to throw once it has run the completion
listeners. The wait itself
+ * is not covered, so that a lasting failure does not spin.
+ */
+ private def awaitListenerInvocationHandover(currentThread: Thread):
Option[Throwable] = {
+ val startNs = System.nanoTime()
+ var interrupted = false
+ var warned = false
+ var handedOver = false
+ var failure: Option[Throwable] = None
+ 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) =>
+ try {
+ warnStillWaiting(holder, startNs, withStackTrace = !warned)
+ warned = true
+ } catch {
+ case t: Throwable => failure = failure.orElse(Some(t))
Review Comment:
**4. [Nit] An `InterruptedException` from the warning is kept as the
failure, and the interrupt is lost**
This catch also takes an `InterruptedException`, e.g. from a custom appender
that blocks interruptibly. The wait then keeps it as the failure to throw at
the end, `interrupted` stays false, and L435-437 do not restore the interrupt
status, which that throw has cleared. The scaladoc says that this wait passes
interrupts on (L384), and that it handles a failure to warn like
`PipelinedWriterTask.awaitExit()` (L390-391). That method leaves
`InterruptedException` out of this catch (`PythonRunner.scala:401`) and handles
it as an interrupt (`PythonRunner.scala:419`). Maybe:
```scala
} catch {
case _: InterruptedException => interrupted = true
case t: Throwable => failure = failure.orElse(Some(t))
}
```
##########
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:
**2. [Low] A failure of the handover warning drops the pre-completion
listener failures**
Since round 2, the first failure to warn is thrown here once the completion
listeners have run, as [round-2 item
2](https://github.com/apache/spark/pull/59663#discussion_r4371913164) asked. It
is not a `TaskCompletionListenerException`, so it skips the catch at L313-318,
and the `throw` at L332-334 never runs. If a pre-completion listener failed,
`markTaskCompleted` then throws only the logging failure. The pre-completion
failures are in no exception, and the completion listeners' failures are only a
suppressed exception of the logging failure (L309). `runTaskWithListeners` then
fails the task with the appender's exception as its reason, although its
listeners failed first. That also contradicts L308: a pre-completion failure
comes before the failure to warn.
Could L311 add the failure to `preCompletionExceptions` when that is not
empty, so that L332-334 reports it together with the listener failures?
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -125,17 +140,19 @@ private[spark] class TaskContextImpl(
override def addTaskCompletionListener(listener: TaskCompletionListener):
this.type = {
val needToCallListener = synchronized {
- // If there is already a thread invoking listeners, adding the new
listener to
- // `onCompleteCallbacks` will cause that thread to execute the new
listener, and the call to
- // `invokeTaskCompletionListeners()` below will be a no-op.
+ // If there is already a thread invoking listeners, the call to
+ // `invokeTaskCompletionListeners()` below will be a no-op. If that
thread runs the completion
+ // listeners in `markTaskCompleted()`, or hands the invocation over to a
thread waiting to do
+ // so, the new listener runs there. Otherwise it is not run until a
listener is added while no
Review Comment:
**5. [Nit] Three comments say that a holder does not run the listeners added
meanwhile, but it runs those of its own kind**
The holder keeps popping its own stack, so it also runs the listeners of
that kind that other threads add while it runs:
- L146-147: if the holder is a thread that added a completion listener after
`markTaskCompleted()` returned, it runs the new listener too. "until a listener
is added" also needs to say "a completion listener": adding a failure listener
runs only the failure listeners.
- L254-256: likewise, "that thread does not run it" holds only if that
thread is not running completion listeners.
- L534-541: a failure listener invocation on a kill thread runs the failure
listeners added meanwhile, on the kill thread. "Any other invocation does not
run the listeners left to it" holds only for the other kinds.
The test name "listeners added after completion are not run by another
thread invoking listeners" (`TaskContextSuite.scala:1581`) is just as broad,
while its invoker runs post-status-update listeners. Maybe "... are not run by
a thread invoking other listeners".
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -199,11 +216,19 @@ private[spark] class TaskContextImpl(
}
private[spark] override def markTaskFailed(error: Throwable): Unit = {
+ markTaskFailed(error, Claim)
Review Comment:
**6. [Nit] `markTaskFailed` called on the holder's thread defers the failure
listeners, while a throwing listener runs them right away**
A listener that fails by throwing gets the failure listeners run nested,
before the next listener (L627). A listener that calls `markTaskFailed` itself
takes this `Claim` path instead, finds its own thread registered and returns
(L573-576). `Utils.tryWithSafeFinallyAndFailureCallbacks` does that right
before its `catchBlock` (`Utils.scala:1331-1334`). The failure listeners then
run only after that listener has returned, and only in the completion listener
invocation of `markTaskCompleted()`. In any other invocation, e.g. of
post-status-update listeners or of a completion listener added after
completion, they never run.
Could `markTaskFailed(error)` use `Nested` when this thread is the
registered one? A failure listener that calls it finds the task failed already,
so that changes only the cases above.
```scala
private[spark] override def markTaskFailed(error: Throwable): Unit = {
val mode = synchronized {
if (listenerInvocationThread.contains(Thread.currentThread())) Nested
else Claim
}
markTaskFailed(error, mode)
}
```
##########
core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleSuite.scala:
##########
@@ -1372,68 +1372,85 @@ class StreamingShuffleSuite
}
}
- // Demonstrates why background threads must not call
context.markTaskFailed() directly:
- // doing so can cause the task thread's markTaskCompleted() to silently skip
completion
- // listeners. In production, cleanupResources() is a completion listener, so
skipping it
- // leaks threads. The reader avoids this by routing background errors
through a
- // CompletableFuture so markTaskFailed/markTaskCompleted are both called
from the task thread.
+ // Shows why background threads should not call context.markTaskFailed()
directly: the
+ // failure listeners then run on that thread, and the task thread's
markTaskCompleted() blocks
+ // until they finish before it runs the completion listeners. In production,
cleanupResources()
+ // is a completion listener, so it waits for whatever the background
thread's failure listeners
+ // wait for. The reader avoids this by routing background errors through an
ErrorNotifier so
+ // that markTaskFailed/markTaskCompleted are both called from the task
thread.
test("background markTaskFailed races with task thread markTaskCompleted") {
val readerContext = createTaskContext(sparkConf, partitionId = 0)
- TaskContext.setTaskContext(readerContext)
- // Latches to force the exact interleaving that triggers the race.
+ // Latches to force the interleaving.
val failureListenerEntered = new CountDownLatch(1)
val failureListenerProceed = new CountDownLatch(1)
- @volatile var completionListenerRan = false
+ @volatile var completionListenerThread: Thread = null
- // Add a failure listener that blocks, holding the
listenerInvocationThread guard.
+ // Add a failure listener that blocks, holding the listener invocation.
readerContext.addTaskFailureListener(new
org.apache.spark.util.TaskFailureListener {
override def onTaskFailure(context: TaskContext, error: Throwable): Unit
= {
failureListenerEntered.countDown()
- failureListenerProceed.await()
+ failureListenerProceed.await(1, TimeUnit.MINUTES)
}
})
// Add a completion listener -- stands in for cleanupResources() in
production.
readerContext.addTaskCompletionListener[Unit] { _ =>
- completionListenerRan = true
+ completionListenerThread = Thread.currentThread()
}
val error = new RuntimeException("background discovery failure")
- // Background thread calls markTaskFailed -> enters invokeListeners ->
acquires
- // listenerInvocationThread guard -> blocks in the failure listener above.
+ // Background thread calls markTaskFailed -> runs the failure listeners ->
holds the listener
+ // invocation while it blocks in the failure listener above.
val bgThread = new Thread(() => {
TaskContext.setTaskContext(readerContext)
readerContext.markTaskFailed(error)
})
+ bgThread.setDaemon(true)
Review Comment:
**8. [Nit] `bgThread` does not record what it throws**
Since [round-2 item
8](https://github.com/apache/spark/pull/59663#discussion_r4371913186),
`taskThread` records its failure (L1425, L1447), but the holder does not.
Suppose its release broke after the handover, e.g. with `notifyAll()` outside
the lock. `taskThread` would still get the invocation once its `wait` times
out, and the test would pass, while the exception of `bgThread` only reaches
the default handler. Could it record its failure like `taskThread`?
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -288,27 +377,120 @@ 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 warn (e.g. an OutOfMemoryError while logging) does not end
the wait either, like
+ * in `PipelinedWriterTask.awaitExit()`: the other thread does not run the
completion listeners,
+ * and the task would release its memory with them pending. The first such
failure is returned,
+ * for `markTaskCompleted()` to throw once it has run the completion
listeners. The wait itself
+ * is not covered, so that a lasting failure does not spin.
+ */
+ private def awaitListenerInvocationHandover(currentThread: Thread):
Option[Throwable] = {
+ val startNs = System.nanoTime()
+ var interrupted = false
+ var warned = false
+ var handedOver = false
+ var failure: Option[Throwable] = None
+ 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) =>
+ try {
+ warnStillWaiting(holder, startNs, withStackTrace = !warned)
Review Comment:
**9. [Nit] While the warning keeps failing, each interval walks the holder's
stack again**
`warned` is set only when `warnStillWaiting` returns. With an appender that
keeps failing, every interval takes the holder's stack again, which stops the
holder at a safepoint, and tries to log it again, unlike what L421-422 say.
`PipelinedWriterTask.awaitExit` does the same (`PythonRunner.scala:398-399`).
Setting `warned` before the call would keep it to one stack walk:
```scala
val withStackTrace = !warned
warned = true
warnStillWaiting(holder, startNs, withStackTrace)
```
##########
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala:
##########
@@ -1195,19 +1192,428 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
completion
}
+ /** A daemon thread that records what `body` throws, for `joinThread` to
check. */
+ private class DaemonThread(name: String, body: => Unit) extends Thread(name)
{
+ setDaemon(true)
+ @volatile var error: Throwable = _
+ override def run(): Unit = {
+ try {
+ body
+ } catch {
+ case t: Throwable => error = t
+ }
+ }
+ }
+
+ private def startDaemonThread(name: String)(body: => Unit): DaemonThread = {
+ val thread = new DaemonThread(name, body)
+ thread.start()
+ thread
+ }
+
+ private def joinThread(thread: DaemonThread): Unit = {
+ thread.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!thread.isAlive, s"${thread.getName} is still running")
+ assert(thread.error == null, s"${thread.getName} failed: ${thread.error}")
Review Comment:
**11. [Nit] `joinThread` reports a thread's failure without its stack trace**
The clue formats `thread.error` with `toString`, so a failure on one of
these threads, e.g. an assertion inside a listener, shows only its message, not
where it was thrown. Maybe:
```scala
Option(thread.error).foreach { e =>
throw new AssertionError(s"${thread.getName} failed", e)
}
```
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -288,27 +377,120 @@ 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 warn (e.g. an OutOfMemoryError while logging) does not end
the wait either, like
+ * in `PipelinedWriterTask.awaitExit()`: the other thread does not run the
completion listeners,
+ * and the task would release its memory with them pending. The first such
failure is returned,
+ * for `markTaskCompleted()` to throw once it has run the completion
listeners. The wait itself
+ * is not covered, so that a lasting failure does not spin.
+ */
+ private def awaitListenerInvocationHandover(currentThread: Thread):
Option[Throwable] = {
+ val startNs = System.nanoTime()
+ var interrupted = false
+ var warned = false
+ var handedOver = false
+ var failure: Option[Throwable] = None
+ 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) =>
+ try {
+ warnStillWaiting(holder, startNs, withStackTrace = !warned)
+ warned = true
+ } catch {
+ case t: Throwable => failure = failure.orElse(Some(t))
+ }
+ case None =>
+ }
+ }
+ } finally {
+ if (interrupted) {
+ currentThread.interrupt()
+ }
+ }
+ failure
+ }
+
+ private def warnStillWaiting(holder: Thread, startNs: Long, withStackTrace:
Boolean): Unit = {
Review Comment:
**10. [Nit] `warnStillWaiting` repeats
`PipelinedWriterTask.warnStillWaiting`**
Both take the other thread's stack outside the lock, check that it is still
the thread waited for, and attach the stack to the first warning as a `new
Exception` with `setStackTrace` (`PythonRunner.scala:430-448`). The wait loops
around them follow the same pattern too. They already differ in how they handle
an `InterruptedException` (item 4). A shared helper, e.g. one in `Utils` that
turns a thread's stack into a throwable for logging, would keep them in step.
##########
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()) {
+ val cause = failureCauseOpt.get
+ Some(() => runLeftFailureListeners(cause))
+ } else if (!listeners.empty()) {
+ val listener = listeners.pop()
+ Some(() => runListener(listener))
+ } else {
+ // We have executed all listeners that have been added so far.
Deregister this thread as
+ // the callback invocation thread, or hand the invocation over to a
waiting thread.
+ if (mode != Nested) {
+ releaseListenerInvocation()
+ }
+ None
+ }
+ }
+
+ try {
+ var step: Option[() => Unit] = None
+ while ({step = nextStep(); step.nonEmpty}) {
+ step.get()
+ }
+ released = true
+ } finally {
+ if (!released) {
+ // Something escaped the loop above. Do not leave this thread
registered, which would
Review Comment:
**7. [Nit] Two of the removed escape-path tests can be kept without
injecting an error**
Your [latest
comment](https://github.com/apache/spark/pull/59663#issuecomment-6510238497)
and the description say that only an error from the invocation itself, e.g. an
`OutOfMemoryError`, reaches this release now. A listener exception whose
`getMessage` throws reaches it too (item 1). A completion listener fails first,
a failure listener in the nested invocation throws such an exception, and
L690-691 throw from the nested invocation. That brings back "an error that
escapes the listener loop still releases the invocation" and "an error of a
failure listener invocation is not hidden by the failure of the listener that
started it". With `NonFatal` (item 1), a `getMessage` that throws a
`LinkageError` still does. "an error that escapes the holder's listener loop
hands the invocation over" stays out of reach, since L627 is a no-op in a
failure listener invocation.
--
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]
