[CELEBORN-1032][FOLLOWUP] Use scheduleWithFixedDelay instead of sched…

…uleAtFixedRate in threads pool of master and worker ### What changes were proposed in this pull request? Use `scheduleWithFixedDelay` instead of `scheduleAtFixedRate` in thread pool of Celeborn Master and Worker. ### Why are the changes needed? Follow up apache#1970. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Internal tests. Closes apache#2048 from SteNicholas/CELEBORN-1032. Authored-by: SteNicholas <[email protected]> Signed-off-by: zky.zhoukeyong <[email protected]>
SteNicholas · Oct 27, 2023 · df40a28 · df40a28
1 parent 5d3ae31
commit df40a28
Show file tree

Hide file tree

Showing 13 changed files with 19 additions and 19 deletions.
diff --git a/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SendBufferPool.java b/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SendBufferPool.java
@@ -57,7 +57,7 @@ private SendBufferPool(int capacity, long checkInterval, long timeout) {
     pushTaskQueues = new LinkedList<>();
 
     lastAquireTime = System.currentTimeMillis();
-    cleaner.scheduleAtFixedRate(
+    cleaner.scheduleWithFixedDelay(
         () -> {
           if (System.currentTimeMillis() - lastAquireTime > timeout) {
             synchronized (this) {

diff --git a/client/src/main/java/org/apache/celeborn/client/ReviveManager.java b/client/src/main/java/org/apache/celeborn/client/ReviveManager.java
@@ -46,7 +46,7 @@ public ReviveManager(ShuffleClientImpl shuffleClient, CelebornConf conf) {
     this.interval = conf.clientPushReviveInterval();
     this.batchSize = conf.clientPushReviveBatchSize();
 
-    batchReviveRequestScheduler.scheduleAtFixedRate(
+    batchReviveRequestScheduler.scheduleWithFixedDelay(
         () -> {
           Map<Integer, Set<ReviveRequest>> shuffleMap = new HashMap<>();
           do {

diff --git a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
@@ -43,7 +43,7 @@ class ApplicationHeartbeater(
   private var appHeartbeat: ScheduledFuture[_] = _
 
   def start(): Unit = {
-    appHeartbeat = appHeartbeatHandlerThread.scheduleAtFixedRate(
+    appHeartbeat = appHeartbeatHandlerThread.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = {
           try {

diff --git a/client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala b/client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala
@@ -70,7 +70,7 @@ class ChangePartitionManager(
   def start(): Unit = {
     batchHandleChangePartition = batchHandleChangePartitionSchedulerThread.map {
       // noinspection ConvertExpressionToSAM
-      _.scheduleAtFixedRate(
+      _.scheduleWithFixedDelay(
         new Runnable {
           override def run(): Unit = {
             try {

diff --git a/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala b/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
@@ -94,7 +94,7 @@ class CommitManager(appUniqueId: String, val conf: CelebornConf, lifecycleManage
     lifecycleManager.registerWorkerStatusListener(new ShutdownWorkerListener)
 
     batchHandleCommitPartition = batchHandleCommitPartitionSchedulerThread.map {
-      _.scheduleAtFixedRate(
+      _.scheduleWithFixedDelay(
         new Runnable {
           override def run(): Unit = {
             committedPartitionInfo.asScala.foreach { case (shuffleId, shuffleCommittedInfo) =>

diff --git a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -147,7 +147,7 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
 
   override def onStart(): Unit = {
     // noinspection ConvertExpressionToSAM
-    checkForShuffleRemoval = forwardMessageThread.scheduleAtFixedRate(
+    checkForShuffleRemoval = forwardMessageThread.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = Utils.tryLogNonFatalError {
           self.send(RemoveExpiredShuffle)

diff --git a/client/src/main/scala/org/apache/celeborn/client/ReleasePartitionManager.scala b/client/src/main/scala/org/apache/celeborn/client/ReleasePartitionManager.scala
@@ -54,7 +54,7 @@ class ReleasePartitionManager(
   def start(): Unit = {
     batchHandleReleasePartition = batchHandleReleasePartitionSchedulerThread.map {
       // noinspection ConvertExpressionToSAM
-      _.scheduleAtFixedRate(
+      _.scheduleWithFixedDelay(
         new Runnable {
           override def run(): Unit = {
             try {

diff --git a/common/src/main/java/org/apache/celeborn/common/network/server/TransportChannelHandler.java b/common/src/main/java/org/apache/celeborn/common/network/server/TransportChannelHandler.java
@@ -106,7 +106,7 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
     if (enableHeartbeat) {
       heartbeatFuture =
           ctx.executor()
-              .scheduleAtFixedRate(
+              .scheduleWithFixedDelay(
                   () -> {
                     logger.debug("send heartbeat");
                     ctx.writeAndFlush(new Heartbeat());

diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala b/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala
@@ -136,7 +136,7 @@ class AppDiskUsageMetric(conf: CelebornConf) extends Logging {
     })
   }
 
-  logExecutor.scheduleAtFixedRate(
+  logExecutor.scheduleWithFixedDelay(
     new Runnable {
       override def run(): Unit = {
         if (currentSnapShot.get() != null) {

diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
@@ -136,7 +136,7 @@ private[celeborn] class Master(
     conf.estimatedPartitionSizeForEstimationUpdateInterval
   private val partitionSizeUpdateService =
     ThreadUtils.newDaemonSingleThreadScheduledExecutor("partition-size-updater")
-  partitionSizeUpdateService.scheduleAtFixedRate(
+  partitionSizeUpdateService.scheduleWithFixedDelay(
     new Runnable {
       override def run(): Unit = {
         executeWithLeaderChecker(
@@ -199,7 +199,7 @@ private[celeborn] class Master(
 
   // start threads to check timeout for workers and applications
   override def onStart(): Unit = {
-    checkForWorkerTimeOutTask = forwardMessageThread.scheduleAtFixedRate(
+    checkForWorkerTimeOutTask = forwardMessageThread.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = Utils.tryLogNonFatalError {
           self.send(ControlMessages.pbCheckForWorkerTimeout)
@@ -209,7 +209,7 @@ private[celeborn] class Master(
       workerHeartbeatTimeoutMs,
       TimeUnit.MILLISECONDS)
 
-    checkForApplicationTimeOutTask = forwardMessageThread.scheduleAtFixedRate(
+    checkForApplicationTimeOutTask = forwardMessageThread.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = Utils.tryLogNonFatalError {
           self.send(CheckForApplicationTimeOut)
@@ -219,7 +219,7 @@ private[celeborn] class Master(
       appHeartbeatTimeoutMs / 2,
       TimeUnit.MILLISECONDS)
 
-    checkForUnavailableWorkerTimeOutTask = forwardMessageThread.scheduleAtFixedRate(
+    checkForUnavailableWorkerTimeOutTask = forwardMessageThread.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = Utils.tryLogNonFatalError {
           self.send(CheckForWorkerUnavailableInfoTimeout)
@@ -230,7 +230,7 @@ private[celeborn] class Master(
       TimeUnit.MILLISECONDS)
 
     if (hasHDFSStorage) {
-      checkForHDFSRemnantDirsTimeOutTask = forwardMessageThread.scheduleAtFixedRate(
+      checkForHDFSRemnantDirsTimeOutTask = forwardMessageThread.scheduleWithFixedDelay(
         new Runnable {
           override def run(): Unit = Utils.tryLogNonFatalError {
             self.send(CheckForHDFSExpiredDirsTimeout)

diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -372,15 +372,15 @@ private[celeborn] class Worker(
     registerWithMaster()
 
     // start heartbeat
-    sendHeartbeatTask = forwardMessageScheduler.scheduleAtFixedRate(
+    sendHeartbeatTask = forwardMessageScheduler.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = Utils.tryLogNonFatalError { heartbeatToMaster() }
       },
       heartbeatInterval,
       heartbeatInterval,
       TimeUnit.MILLISECONDS)
 
-    checkFastFailTask = forwardMessageScheduler.scheduleAtFixedRate(
+    checkFastFailTask = forwardMessageScheduler.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = Utils.tryLogNonFatalError {
           unavailablePeers.entrySet().forEach { entry: JMap.Entry[WorkerInfo, Long] =>

diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitor.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitor.scala
@@ -103,7 +103,7 @@ class LocalDeviceMonitor(
   }
 
   override def startCheck(): Unit = {
-    diskChecker.scheduleAtFixedRate(
+    diskChecker.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = {
           logDebug("Device check start")

diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
@@ -214,7 +214,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
     saveCommittedFileInfosExecutor =
       ThreadUtils.newDaemonSingleThreadScheduledExecutor(
         "StorageManager-save-committed-fileinfo-thread")
-    saveCommittedFileInfosExecutor.scheduleAtFixedRate(
+    saveCommittedFileInfosExecutor.scheduleWithFixedDelay(
       new Runnable {
         override def run(): Unit = {
           if (!committedFileInfos.isEmpty) {
@@ -588,7 +588,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
   private val storageScheduler =
     ThreadUtils.newDaemonSingleThreadScheduledExecutor("storage-scheduler")
 
-  storageScheduler.scheduleAtFixedRate(
+  storageScheduler.scheduleWithFixedDelay(
     new Runnable {
       override def run(): Unit = {
         try {