[
https://issues.apache.org/jira/browse/SPARK-60222?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
L. C. Hsieh updated SPARK-60222:
--------------------------------
Description:
TaskContextImpl runs task listeners on one thread at a time, tracked by
listenerInvocationThread. In the following cases, listeners that should run on
the thread that completes the task, before the task is torn down, run elsewhere
or not at all:
1. When another thread is invoking listeners as the task completes,
markTaskCompleted returns without running any completion listener. For example,
Task.kill calls markInterrupted on the kill thread, and if an interrupt
listener throws, markTaskFailed runs the failure listeners on that thread. The
completion listeners are then left until some other thread adds a listener, or
skipped entirely. The task can then reach cleanUpAllAllocatedMemory while
resources that the completion listeners would release are still in use. The
ErrorNotifier doc and a StreamingShuffleSuite test describe this race.
2. When a listener throws, invokeListeners clears listenerInvocationThread
while markTaskFailed runs the failure listeners, and reclaims it afterwards.
Another thread that adds a listener in between can take over running the
pending listeners, possibly concurrently with this thread.
3. When another thread calls markTaskFailed while the completing thread runs
the completion listeners, it returns without running the failure listeners, and
the completing thread does not run them either. Failure cleanup, such as
aborting a state store update before the completion listener releases the
store, is skipped.
The proposed fix keeps these listeners on the completing thread:
markTaskCompleted claims the invocation in the same step that sets completed,
and if another thread holds it, waits until that thread hands it over directly;
when a listener throws, the failure listeners run inside the current
invocation; and failure listeners that another thread queues while
markTaskCompleted runs the completion listeners run before its next completion
listener. This was first done in apache/spark#59332 and split out of it during
review.
Out of scope: listeners added after markTaskCompleted has finished, while
another thread holds the invocation, are still dropped, and so are
post-status-update listeners in the same situation. Running them would mean
running them on another thread, without the task's TaskContext, after the
executor has released the task's block locks and memory.
was:
TaskContextImpl runs task listeners on one thread at a time, tracked by
listenerInvocationThread. In two cases, the completion listeners do not run on
the thread that completes the task:
1. When another thread is invoking listeners as the task completes,
markTaskCompleted returns without running any completion listener. For example,
Task.kill calls markInterrupted on the kill thread, and if an interrupt
listener throws, markTaskFailed runs the failure listeners on that thread. The
completion listeners are then left until some other thread adds a listener, or
skipped entirely. The task can then reach cleanUpAllAllocatedMemory while
resources that the completion listeners would release are still in use. The
ErrorNotifier doc and a StreamingShuffleSuite test describe this race.
2. When a listener throws, invokeListeners clears listenerInvocationThread
while markTaskFailed runs the failure listeners, and reclaims it afterwards.
Another thread that adds a listener in between can take over running the
pending listeners, possibly concurrently with this thread.
Failure listeners that another thread queues while a thread runs the completion
listeners, and completion listeners queued while a thread runs failure
listeners after completion, are also dropped.
This matters for helper threads that the completion listeners stop and wait
for, such as the pipelined Python UDF writer (SPARK-60113): if the helper
thread adds a listener lazily at such a moment, it ends up running the listener
that waits for itself.
The proposed fix keeps the completion listeners on the completing thread:
markTaskCompleted waits for another thread invoking listeners and takes the
invocation over directly, failure listeners run inside the current invocation
when a listener throws, and a thread runs the listeners left to it before it
releases the invocation. This was first done in apache/spark#59332 and split
out of it during review.
> Keep task completion listeners on the completing thread when another thread
> invokes listeners
> ---------------------------------------------------------------------------------------------
>
> Key: SPARK-60222
> URL: https://issues.apache.org/jira/browse/SPARK-60222
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 4.3.0, 4.4.0
> Reporter: L. C. Hsieh
> Priority: Major
> Labels: pull-request-available
>
> TaskContextImpl runs task listeners on one thread at a time, tracked by
> listenerInvocationThread. In the following cases, listeners that should run
> on the thread that completes the task, before the task is torn down, run
> elsewhere or not at all:
> 1. When another thread is invoking listeners as the task completes,
> markTaskCompleted returns without running any completion listener. For
> example, Task.kill calls markInterrupted on the kill thread, and if an
> interrupt listener throws, markTaskFailed runs the failure listeners on that
> thread. The completion listeners are then left until some other thread adds a
> listener, or skipped entirely. The task can then reach
> cleanUpAllAllocatedMemory while resources that the completion listeners would
> release are still in use. The ErrorNotifier doc and a StreamingShuffleSuite
> test describe this race.
> 2. When a listener throws, invokeListeners clears listenerInvocationThread
> while markTaskFailed runs the failure listeners, and reclaims it afterwards.
> Another thread that adds a listener in between can take over running the
> pending listeners, possibly concurrently with this thread.
> 3. When another thread calls markTaskFailed while the completing thread runs
> the completion listeners, it returns without running the failure listeners,
> and the completing thread does not run them either. Failure cleanup, such as
> aborting a state store update before the completion listener releases the
> store, is skipped.
> The proposed fix keeps these listeners on the completing thread:
> markTaskCompleted claims the invocation in the same step that sets completed,
> and if another thread holds it, waits until that thread hands it over
> directly; when a listener throws, the failure listeners run inside the
> current invocation; and failure listeners that another thread queues while
> markTaskCompleted runs the completion listeners run before its next
> completion listener. This was first done in apache/spark#59332 and split out
> of it during review.
> Out of scope: listeners added after markTaskCompleted has finished, while
> another thread holds the invocation, are still dropped, and so are
> post-status-update listeners in the same situation. Running them would mean
> running them on another thread, without the task's TaskContext, after the
> executor has released the task's block locks and memory.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
