[SPARK-51341][CORE] Cancel time task with suitable way. #50107
Conversation
beliefer
changed the title
| assert(executor.isInstanceOf[ScheduledThreadPoolExecutor]) | ||
| executor.asInstanceOf[ScheduledThreadPoolExecutor] | ||
| } | ||
| private val timer = ThreadUtils.newSingleThreadScheduledExecutor( |
There was a problem hiding this comment.
The origin code is non-daemon thread here.
Please see
5d5b3a5#diff-e9f66d336ded6632366be06dbd70e74afe9595cfc92a962f2c8554f09af53d0fL286
There was a problem hiding this comment.
This timer is never shut down. Is there a decent place to do it? otherwise, I think daemon is correct here. Non-daemon threads would stop the JVM from exiting, in theory
There was a problem hiding this comment.
Because all the BarrierTaskContext share the timer, so we can't shut down it.
It seems good to revert it with daemon thread.
| def stop(): Unit = { | ||
| timerFuture.cancel(true) | ||
| ThreadUtils.shutdown(timer) | ||
| } |
There was a problem hiding this comment.
Thanks for the change @beliefer .
Interested in knowing how this change in the ui consoleprogress bar was tested?
There was a problem hiding this comment.
To be honestly, I don't know it too.
beliefer
changed the title
beliefer
force-pushed
the
SPARK-51341
branch
from
a7010f2 to
f092f93
Compare
beliefer
changed the title
| assert(executor.isInstanceOf[ScheduledThreadPoolExecutor]) | ||
| executor.asInstanceOf[ScheduledThreadPoolExecutor] | ||
| } | ||
| private val timer = ThreadUtils.newSingleThreadScheduledExecutor( |
There was a problem hiding this comment.
This timer is never shut down. Is there a decent place to do it? otherwise, I think daemon is correct here. Non-daemon threads would stop the JVM from exiting, in theory
| @@ -169,6 +171,7 @@ private[spark] class TaskSchedulerImpl( | |||
| protected val executorIdToHost = new HashMap[String, String] | |||
|
|
|||
| private val abortTimer = ThreadUtils.newDaemonSingleThreadScheduledExecutor("task-abort-timer") | |||
| private val abortFutures = new CopyOnWriteArrayList[ScheduledFuture[_]] | |||
There was a problem hiding this comment.
Why CopyOnWriteArrayList in particular? maybe fine but would a simple synchronized list be easier?
There was a problem hiding this comment.
Yeah. less reads and more writes here.
| @@ -959,7 +965,9 @@ private[spark] class TaskSchedulerImpl( | |||
| barrierCoordinator.stop() | |||
| } | |||
| } | |||
| starvationFutures.asScala.foreach(_.cancel(false)) | |||
There was a problem hiding this comment.
My last concern is, do these Futures build up and cause a potential memory leak? I don't think they hold onto large result objects. The Future itself isn't big either. So maybe not an issue. I don't know of a clean way to delete them on completion of individual tasks.
There was a problem hiding this comment.
Good thinking. Let me revert this part. Maybe ThreadUtils.shutdown(starvationTimer) is good enough.
What changes were proposed in this pull request?
This PR proposes to cancel task with suitable way.
Why are the changes needed?
According to the discussion at
#47956 (comment)
TimerTask.cancel() doesn't work.
Does this PR introduce any user-facing change?
'No'.
How was this patch tested?
GA
Was this patch authored or co-authored using generative AI tooling?
'No'.