[SPARK-46895][CORE] Replace Timer with single thread scheduled executor #44718

beliefer · 2024-01-13T11:55:53Z

What changes were proposed in this pull request?

This PR propose to replace Timer with single thread scheduled executor.

Why are the changes needed?

The javadoc recommends ScheduledThreadPoolExecutor instead of Timer.

This change based on the following two points.
System time sensitivity

Timer scheduling is based on the absolute time of the operating system and is sensitive to the operating system's time. Once the operating system's time changes, Timer scheduling is no longer precise.
The scheduled Thread Pool Executor scheduling is based on relative time and is not affected by changes in operating system time.

Are anomalies captured

Timer does not capture exceptions thrown by Timer Tasks, and in addition, Timer is single threaded. Once a scheduling task encounters an exception, the entire thread will terminate and other tasks that need to be scheduled will no longer be executed.
The scheduled Thread Pool Executor implements scheduling functions based on a thread pool. After a task throws an exception, other tasks can still execute normally.

Does this PR introduce any user-facing change?

'No'.

How was this patch tested?

GA tests.

Was this patch authored or co-authored using generative AI tooling?

'No'.

beliefer · 2024-01-15T06:20:32Z

@dongjoon-hyun @srowen @LuciferYang Please help me to take a review.

srowen · 2024-01-15T12:25:13Z

core/src/main/scala/org/apache/spark/BarrierCoordinator.scala

@@ -168,7 +170,7 @@ private[spark] class BarrierCoordinator(
        // we may timeout for the sync.
        if (requesters.isEmpty) {
          initTimerTask(this)
-          timer.schedule(timerTask, timeoutInSecs * 1000)
+          timer.schedule(timerTask, timeoutInSecs * 1000, TimeUnit.MILLISECONDS)


You can just remove "* 1000" and specify units as SECONDS

beliefer · 2024-01-19T09:29:01Z

cc @dongjoon-hyun @LuciferYang .

srowen · 2024-01-19T12:12:06Z

core/src/main/scala/org/apache/spark/BarrierTaskContext.scala

@@ -71,7 +72,7 @@ class BarrierTaskContext private[spark] (
      }
    }
    // Log the update of global sync every 60 seconds.
-    timer.schedule(timerTask, 60000, 60000)
+    timer.scheduleAtFixedRate(timerTask, 60000, 60000, TimeUnit.MILLISECONDS)


Same idea here, this is just 1 minute

core/src/main/scala/org/apache/spark/BarrierCoordinator.scala

LuciferYang · 2024-01-19T14:08:49Z

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

-    starvationTimer.cancel()
-    abortTimer.cancel()
+    starvationTimer.shutdown()
+    abortTimer.shutdown()


It's fine, but would it be more appropriate to use ThreadUtils.shutdown? I am not sure.

Sounds good.

LuciferYang · 2024-01-19T14:12:57Z

launcher/src/main/java/org/apache/spark/launcher/LauncherServer.java

@@ -128,7 +125,8 @@ private LauncherServer() throws IOException {
      this.threadIds = new AtomicLong();
      this.factory = new NamedThreadFactory(THREAD_NAME_FMT);
      this.secretToPendingApps = new ConcurrentHashMap<>();
-      this.timeoutTimer = new Timer("LauncherServer-TimeoutTimer", true);
+      this.timeoutTimer = new ScheduledThreadPoolExecutor(
+              1, new NamedThreadFactory("LauncherServer-TimeoutTimer"));


nit: Indentation

LuciferYang · 2024-01-19T14:14:57Z

launcher/src/main/java/org/apache/spark/launcher/LauncherServer.java

-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.*;


Suggested change

import java.util.concurrent.*;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ConcurrentMap;

import java.util.concurrent.ScheduledThreadPoolExecutor;

import java.util.concurrent.ThreadFactory;

import java.util.concurrent.TimeUnit;

I tend to avoid using import *

beliefer · 2024-01-21T08:52:53Z

The GA failure is unrelated.

LuciferYang

Fine to me.

But I hope @srowen could approve before merge. Thanks ~

dongjoon-hyun

@beliefer . Although it may share the same target, this does look like a follow-up for the main PR about ConsoleProgressBar.scala as of now. Given that the change affects many other places, please make a new JIRA ID.

[SPARK-46698][CORE] Replace Timer with single thread scheduled executor for ConsoleProgressBar. #44701

beliefer · 2024-01-29T01:31:29Z

@dongjoon-hyun new jira ID created.

dongjoon-hyun · 2024-01-29T05:36:47Z

Thank you. Could you re-trigger the failed test pipelines?

beliefer · 2024-01-29T12:42:07Z

@dongjoon-hyun The GA failure is unrelated.

Stale

dongjoon-hyun · 2024-02-03T22:09:18Z

Thank you for updating. Could you make CI happy?

[info] *** 23 TESTS FAILED ***
[error] Failed tests:
[error] 	org.apache.spark.deploy.yarn.YarnShuffleAlternateNameConfigWithLevelDBBackendSuite
[error] 	org.apache.spark.deploy.yarn.YarnShuffleIntegrationWithLevelDBBackendSuite
[error] 	org.apache.spark.deploy.yarn.YarnClusterSuite
[error] 	org.apache.spark.deploy.yarn.YarnShuffleAuthWithLevelDBBackendSuite
[error] 	org.apache.spark.deploy.yarn.YarnShuffleAlternateNameConfigWithRocksDBBackendSuite
[error] 	org.apache.spark.deploy.yarn.YarnShuffleIntegrationWithRocksDBBackendSuite
[error] 	org.apache.spark.deploy.yarn.YarnShuffleAuthWithRocksDBBackendSuite
[error] (yarn / Test / test) sbt.TestsFailedException: Tests unsuccessful
[error] Total time: 662 s (11:02), completed Feb 1, 2024, 3:01:15 AM

beliefer · 2024-02-04T01:58:56Z

Thank you for updating. Could you make CI happy?

I will try again.

dongjoon-hyun · 2024-02-04T02:40:41Z

Thank you, @beliefer .

beliefer · 2024-02-05T08:01:02Z

@dongjoon-hyun All the tests passed.

srowen

I think it's OK

srowen · 2024-02-05T12:13:14Z

core/src/main/scala/org/apache/spark/BarrierTaskContext.scala

@@ -283,6 +284,7 @@ object BarrierTaskContext {
  @Since("2.4.0")
  def get(): BarrierTaskContext = TaskContext.get().asInstanceOf[BarrierTaskContext]

-  private val timer = new Timer("Barrier task timer for barrier() calls.")
+  private val timer = ThreadUtils.newSingleThreadScheduledExecutor(


Hm I guess there is no place to shut this down. It's not critical and not something we have to change here. I wonder if it should be a daemon thread though? seems weird to spawn a thread that nothing in the code can terminate

The origin Timer is not a daemon thread, so I followed the same behavior.
There are some code used to cancel the timer task.

spark/core/src/main/scala/org/apache/spark/BarrierTaskContext.scala

Line 120 in db8c305

timerTask.cancel()

That doesn't shut down the executor though, and the other code paths do. I don't see how the non-daemon thread ever terminates. It may not matter as the Spark context lives as long as the app. It'd be tidy to shut it down, or make it a daemon anyway, but not strictly related to the change you're making

Got it. Let's change it to daemon thread.

LuciferYang · 2024-02-06T09:23:47Z

All test passed. Merged into master for Spark 4.0. Thanks @beliefer @srowen and @dongjoon-hyun ~

beliefer · 2024-02-06T09:34:33Z

@LuciferYang @srowen @dongjoon-hyun Thank you!

### What changes were proposed in this pull request? This PR propose to replace `Timer` with single thread scheduled executor. ### Why are the changes needed? The javadoc recommends `ScheduledThreadPoolExecutor` instead of `Timer`. ![屏幕快照 2024-01-12 下午12 47 57](https://github.com/apache/spark/assets/8486025/4fc5ed61-6bb9-4768-915a-ad919a067d04) This change based on the following two points. **System time sensitivity** Timer scheduling is based on the absolute time of the operating system and is sensitive to the operating system's time. Once the operating system's time changes, Timer scheduling is no longer precise. The scheduled Thread Pool Executor scheduling is based on relative time and is not affected by changes in operating system time. **Are anomalies captured** Timer does not capture exceptions thrown by Timer Tasks, and in addition, Timer is single threaded. Once a scheduling task encounters an exception, the entire thread will terminate and other tasks that need to be scheduled will no longer be executed. The scheduled Thread Pool Executor implements scheduling functions based on a thread pool. After a task throws an exception, other tasks can still execute normally. ### Does this PR introduce _any_ user-facing change? 'No'. ### How was this patch tested? GA tests. ### Was this patch authored or co-authored using generative AI tooling? 'No'. Closes apache#44718 from beliefer/replace-timer-with-threadpool. Authored-by: beliefer <beliefer@163.com> Signed-off-by: yangjie01 <yangjie01@baidu.com> (cherry picked from commit 5d5b3a5)

github-actions bot added WEB UI CORE labels Jan 13, 2024

beliefer force-pushed the replace-timer-with-threadpool branch 2 times, most recently from 15f80b4 to a1cade9 Compare January 14, 2024 03:16

beliefer changed the title ~~[WIP][CORE] Replace Timer with single thread scheduled executor~~ [SPARK-46698][CORE][FOLLOWUP] Replace Timer with single thread scheduled executor Jan 14, 2024

srowen reviewed Jan 15, 2024

View reviewed changes

srowen reviewed Jan 19, 2024

View reviewed changes

beliefer force-pushed the replace-timer-with-threadpool branch from 7c6530a to db8f03c Compare January 19, 2024 13:43

LuciferYang reviewed Jan 19, 2024

View reviewed changes

beliefer force-pushed the replace-timer-with-threadpool branch 3 times, most recently from 8d4176b to 562c725 Compare January 21, 2024 05:17

beliefer force-pushed the replace-timer-with-threadpool branch 3 times, most recently from 86060fd to fdbf3b6 Compare January 23, 2024 03:53

LuciferYang approved these changes Jan 24, 2024

View reviewed changes

dongjoon-hyun previously requested changes Jan 28, 2024

View reviewed changes

beliefer changed the title ~~[SPARK-46698][CORE][FOLLOWUP] Replace Timer with single thread scheduled executor~~ [SPARK-46895][CORE] Replace Timer with single thread scheduled executor Jan 29, 2024

beliefer force-pushed the replace-timer-with-threadpool branch 2 times, most recently from 9de4cb5 to 197c164 Compare February 1, 2024 01:45

beliefer added 2 commits February 4, 2024 09:57

[WIP][CORE] Replace Timer with single thread scheduled executor

a5ea103

Update code

735df21

beliefer force-pushed the replace-timer-with-threadpool branch from 197c164 to 735df21 Compare February 4, 2024 01:58

Revert LauncherServer

db8c305

beliefer force-pushed the replace-timer-with-threadpool branch from b1dc93c to db8c305 Compare February 5, 2024 03:29

srowen reviewed Feb 5, 2024

View reviewed changes

Update code

f7d641f

LuciferYang approved these changes Feb 6, 2024

View reviewed changes

LuciferYang closed this in 5d5b3a5 Feb 6, 2024

jshmchenxi mentioned this pull request Aug 31, 2024

[SPARK-46895][CORE][3.5] Replace Timer with single thread scheduled executor #47945

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-46895][CORE] Replace Timer with single thread scheduled executor #44718

[SPARK-46895][CORE] Replace Timer with single thread scheduled executor #44718

beliefer commented Jan 13, 2024

beliefer commented Jan 15, 2024 •

edited

Loading

srowen Jan 15, 2024

beliefer commented Jan 19, 2024

srowen Jan 19, 2024

LuciferYang Jan 19, 2024

beliefer Jan 20, 2024

LuciferYang Jan 19, 2024

LuciferYang Jan 19, 2024

LuciferYang Jan 19, 2024

beliefer commented Jan 21, 2024

LuciferYang left a comment

dongjoon-hyun left a comment •

edited

Loading

beliefer commented Jan 29, 2024

dongjoon-hyun commented Jan 29, 2024

beliefer commented Jan 29, 2024

dongjoon-hyun commented Feb 3, 2024 •

edited

Loading

beliefer commented Feb 4, 2024

dongjoon-hyun commented Feb 4, 2024

beliefer commented Feb 5, 2024

srowen left a comment

srowen Feb 5, 2024

beliefer Feb 5, 2024

srowen Feb 5, 2024

beliefer Feb 5, 2024

LuciferYang commented Feb 6, 2024

beliefer commented Feb 6, 2024

[SPARK-46895][CORE] Replace Timer with single thread scheduled executor #44718

[SPARK-46895][CORE] Replace Timer with single thread scheduled executor #44718

Conversation

beliefer commented Jan 13, 2024

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

beliefer commented Jan 15, 2024 • edited Loading

Choose a reason for hiding this comment

beliefer commented Jan 19, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

beliefer commented Jan 21, 2024

LuciferYang left a comment

Choose a reason for hiding this comment

dongjoon-hyun left a comment • edited Loading

Choose a reason for hiding this comment

beliefer commented Jan 29, 2024

dongjoon-hyun commented Jan 29, 2024

beliefer commented Jan 29, 2024

dongjoon-hyun commented Feb 3, 2024 • edited Loading

beliefer commented Feb 4, 2024

dongjoon-hyun commented Feb 4, 2024

beliefer commented Feb 5, 2024

srowen left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuciferYang commented Feb 6, 2024

beliefer commented Feb 6, 2024

beliefer commented Jan 15, 2024 •

edited

Loading

dongjoon-hyun left a comment •

edited

Loading

dongjoon-hyun commented Feb 3, 2024 •

edited

Loading