dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4339138602
##########
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:
**12. [Nit] `invokeTaskCompletionListeners` still has its own default**
My [round-3
comment](https://github.com/apache/spark/pull/59332#discussion_r4252547657) on
#59332 said that with a single mode parameter, "`invokeTaskCompletionListeners`
would not need its own defaults". This default, and the one on
`invokeListeners` (L501), now have one user each, L151 and L205, while
`invokeTaskFailureListeners` has none and its callers pass `Claim` (L178,
L215). Could L151 and L205 pass `Claim` too, so that every call shows how it
takes part in the 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]
