viirya commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4421488562
##########
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:
Done, as you suggested. I tried to cover it with an appender that throws an
`InterruptedException`, but log4j wraps whatever an appender throws in an
`AppenderLoggingException`, so it does not reach this catch through log4j; the
description lists it as not covered.
##########
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:
Updated the three comments: a completion listener added while another thread
invokes listeners runs there only if that thread runs completion listeners, and
otherwise not until a completion listener is added while no thread invokes
listeners; the `invokeListeners()` scaladoc now says that the listeners added
to `listeners` run there like the others, and that any other invocation does
not run the listeners of another kind left to it. The test is renamed to "...
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:
I'd leave this one. With `Nested` here, a post-status-update listener that
calls `markTaskFailed` would run the failure listeners on the task thread after
`Task.run` has unset its `TaskContext` and the executor has released the task's
block locks and memory, and a completion listener added after completion would
run them on the thread adding it, e.g. a kill thread. Those are the cases the
`invokeListeners()` scaladoc says they must not run in, and on master they do
not run there either. In the completion listener invocation of
`markTaskCompleted()`, they run once that listener has returned, before the
next completion listener.
##########
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:
Thanks, brought both back with a failure listener whose exception's
`getMessage` throws a `LinkageError`: "an error that escapes the listener loop
still releases the invocation", and "a fatal error of a failure listener
invocation is not hidden by the failure of the listener that started it", for
both a completion and a pre-completion listener. The description now lists only
the handover after an escape as not covered.
##########
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:
Done.
##########
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:
Done: `warned` is set before the call. The test "a failing handover warning
does not end the wait" now checks that only the first failing warning carries
the stack trace. I left `PipelinedWriterTask.awaitExit` as it is, since it is
not part of this PR.
##########
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:
I'd keep them separate. What they share is taking the stack outside the lock
and wrapping it in an `Exception`, which is a few lines; the check that the
thread is still the one waited for reads different state, and the wait loops
differ on purpose in how they handle interrupts, which both comments explain. A
helper would also touch `PythonRunner`, which this PR otherwise does not change.
##########
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:
Done, with `fail(s"${thread.getName} failed", e)`, which keeps `e` as the
cause; scalastyle does not allow `throw new AssertionError` here.
--
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]
