viirya commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4361882130
##########
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
+ // thread needs to finish anyway, so this only adds a wait for other
threads, e.g. one that
+ // failed the task from `Utils.tryWithSafeFinallyAndFailureCallbacks`. If
such a task is
+ // killed, `spark.task.reaper.enabled` with a positive
`spark.task.reaper.killTimeout` bounds
Review Comment:
Added the local-mode exception to the comment and the description, as well
as the case where the blocked listeners run on the reaper thread itself.
##########
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
Review Comment:
Reworded: the error reaches the executor only if the task body succeeded,
otherwise `runTaskWithListeners` adds it as a suppressed exception; and if the
holder has just handed over, the invocation is released, so the completion
listeners run only once another listener is added.
##########
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,
Review Comment:
Updated this one with your wording, the interrupt-listener comment (task
completion would wait for the interrupt listeners), and the four
`StreamingShuffleSuite` comments.
##########
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
+ }
+ 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.foreach { holder =>
+ val waitedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNs)
+ val message = log"Task ${MDC(TASK_ATTEMPT_ID, taskAttemptId)} has
been waiting for " +
+ log"${MDC(TOTAL_TIME, waitedMs)} ms to run its completion
listeners until thread " +
+ log"${MDC(THREAD_NAME, holder.getName)} finishes running its task
listeners."
+ if (warned) {
+ logWarning(message)
+ } else {
+ val holderStack = new Exception(s"Stack trace of thread
${holder.getName}")
+ holderStack.setStackTrace(holder.getStackTrace)
Review Comment:
It now checks under the lock, after taking the stack, that the holder still
holds the invocation, and skips the warning otherwise. A new test makes the
holder hand over and exit while its stack is taken and checks that nothing is
logged.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -518,3 +712,21 @@ private[spark] class TaskContextImpl(
}
}
}
+
+private object TaskContextImpl {
+
+ /** How an `invokeListeners()` call takes part in the listener invocation. */
+ sealed trait InvocationMode
+
+ /** Registers this thread as `listenerInvocationThread` unless another
thread is registered. */
Review Comment:
Used your wording for `Claim`, and the scaladoc now names `Task.kill` and a
pending kill in `createResourceUninterruptibly()`.
##########
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala:
##########
@@ -931,6 +934,351 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
completion
}
+ private def startDaemonThread(name: String)(body: => Unit): Thread = {
Review Comment:
Done.
##########
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
+ }
+ 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.foreach { holder =>
+ val waitedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNs)
+ val message = log"Task ${MDC(TASK_ATTEMPT_ID, taskAttemptId)} has
been waiting for " +
+ log"${MDC(TOTAL_TIME, waitedMs)} ms to run its completion
listeners until thread " +
+ log"${MDC(THREAD_NAME, holder.getName)} finishes running its task
listeners."
+ if (warned) {
+ logWarning(message)
+ } else {
+ val holderStack = new Exception(s"Stack trace of thread
${holder.getName}")
+ holderStack.setStackTrace(holder.getStackTrace)
+ logWarning(message, holderStack)
+ warned = true
+ }
+ }
+ }
+ } finally {
+ if (!handedOver) {
+ // Leaving abnormally: do not leave a stale waiter for the holder to
hand over to, or a
+ // stale holder if the invocation was handed over in the meantime.
+ synchronized {
+ if (nextListenerInvocationThread.contains(currentThread)) {
+ nextListenerInvocationThread = None
+ }
+ if (listenerInvocationThread.contains(currentThread)) {
+ releaseListenerInvocation()
+ }
+ }
+ }
+ if (interrupted) {
+ currentThread.interrupt()
+ }
+ }
+ }
+
+ /**
+ * Deregisters this thread as the listener invocation thread, or hands the
invocation over to a
+ * thread waiting for it in `markTaskCompleted()`. Must hold the TaskContext
lock.
+ */
+ private def releaseListenerInvocation(): Unit = {
+ listenerInvocationThread = nextListenerInvocationThread
+ nextListenerInvocationThread = None
+ notifyAll()
+ }
+
+ private def invokeTaskCompletionListeners(
+ error: Option[Throwable],
+ mode: InvocationMode = Claim): Unit = {
Review Comment:
Removed both defaults; every call now passes its mode.
--
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]
