You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/10/10 09:32:48 UTC
[incubator-celeborn] branch branch-0.3 updated: [CELEBORN-1032][WORKER] Use scheduleWithFixedDelay instead of scheduleAtFixedRate in worker timeout check threads pool
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.3 by this push:
new a1ae4b44e [CELEBORN-1032][WORKER] Use scheduleWithFixedDelay instead of scheduleAtFixedRate in worker timeout check threads pool
a1ae4b44e is described below
commit a1ae4b44e2a87b5eb3c28896396514253215ba3b
Author: zwangsheng <bi...@apache.org>
AuthorDate: Tue Oct 10 17:32:20 2023 +0800
[CELEBORN-1032][WORKER] Use scheduleWithFixedDelay instead of scheduleAtFixedRate in worker timeout check threads pool
### What changes were proposed in this pull request?
Use `scheduleWithFixedDelay` instead of `scheduleAtFixedRate` when worker submit push/fetch timeout check task.
### Why are the changes needed?
`scheduleAtFixedRate` means
> Creates and executes a periodic action that becomes enabled first after the given initial delay, and subsequently with the given period; that is executions will commence after initialDelay then initialDelay+period, then initialDelay + 2 * period, and so on.
`scheduleWithFixedDelay` means
> Creates and executes a periodic action that becomes enabled first after the given initial delay, and subsequently with the given delay between the termination of one execution and the commencement of the next.
So in `scheduleAtFixedRate` case, we experience task stacking in DelayedWorkQueue, due to the long execution time of a single task.
Too much task remain in `DelayedWorkQueue` will cost lot memory and CPU to put or pop.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
GA
Before:
```patch
[arthas6]$ getstatic org.apache.celeborn.common.network.client.TransportResponseHandler pushTimeoutChecker
field: pushTimeoutChecker
ScheduledThreadPoolExecutor[
continueExistingPeriodicTasksAfterShutdown=Boolean[false],
executeExistingDelayedTasksAfterShutdown=Boolean[true],
removeOnCancel=Boolean[true],
sequencer=AtomicLong[912892],
ctl=AtomicInteger[-536870908],
COUNT_BITS=Integer[29],
CAPACITY=Integer[536870911],
RUNNING=Integer[-536870912],
SHUTDOWN=Integer[0],
STOP=Integer[536870912],
TIDYING=Integer[1073741824],
TERMINATED=Integer[1610612736],
+ workQueue=DelayedWorkQueue[isEmpty=false;size=157377],
mainLock=ReentrantLock[java.util.concurrent.locks.ReentrantLock3e51a6ff[Unlocked]],
workers=HashSet[isEmpty=false;size=4],
termination=ConditionObject[java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject4944ebf],
largestPoolSize=Integer[4],
completedTaskCount=Long[0],
threadFactory=[com.google.common.util.concurrent.ThreadFactoryBuilder$172260dff],
handler=AbortPolicy[java.util.concurrent.ThreadPoolExecutor$AbortPolicy50f54af5],
keepAliveTime=Long[0],
allowCoreThreadTimeOut=Boolean[false],
corePoolSize=Integer[4],
maximumPoolSize=Integer[2147483647],
defaultHandler=AbortPolicy[java.util.concurrent.ThreadPoolExecutor$AbortPolicy50f54af5],
shutdownPerm=RuntimePermission[("java.lang.RuntimePermission" "modifyThread")],
acc=null,
ONLY_ONE=Boolean[true],
$assertionsDisabled=Boolean[true],
]
```
```log
// from arthas threads command
ID NAME GROUP PRIORITY STATE %CPU DELTA_TIME TIME INTERRUPTED DAEMON
61 push-timeout-checker-1 main 5 WAITING 9.24 0.018 820:55.660 false true
57 push-timeout-checker-0 main 5 WAITING 9.09 0.018 818:6.882 false true
67 push-timeout-checker-3 main 5 WAITING 8.44 0.017 816:47.813 false true
68 fetch-timeout-checker-3 main 5 WAITING 8.42 0.017 825:41.115 false true
66 fetch-timeout-checker-2 main 5 WAITING 8.18 0.016 824:17.086 false true
58 fetch-timeout-checker-0 main 5 WAITING 8.16 0.016 799:45.428 false true
65 push-timeout-checker-2 main 5 WAITING 8.15 0.016 835:18.827 false true
62 fetch-timeout-checker-1 main 5 WAITING 8.14 0.016 804:47.836 false true
```
After:
```patch
[arthas6]$ getstatic org.apache.celeborn.common.network.client.TransportResponseHandler pushTimeoutChecker
field: pushTimeoutChecker
ScheduledThreadPoolExecutor[
continueExistingPeriodicTasksAfterShutdown=Boolean[false],
executeExistingDelayedTasksAfterShutdown=Boolean[true],
removeOnCancel=Boolean[true],
sequencer=AtomicLong[151380],
ctl=AtomicInteger[-536870908],
COUNT_BITS=Integer[29],
CAPACITY=Integer[536870911],
RUNNING=Integer[-536870912],
SHUTDOWN=Integer[0],
STOP=Integer[536870912],
TIDYING=Integer[1073741824],
TERMINATED=Integer[1610612736],
+ workQueue=DelayedWorkQueue[isEmpty=false;size=447],
mainLock=ReentrantLock[java.util.concurrent.locks.ReentrantLock7319376f[Unlocked]],
workers=HashSet[isEmpty=false;size=4],
termination=ConditionObject[java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject16b22de2],
largestPoolSize=Integer[4],
completedTaskCount=Long[0],
threadFactory=[com.google.common.util.concurrent.ThreadFactoryBuilder$11108fdba],
handler=AbortPolicy[java.util.concurrent.ThreadPoolExecutor$AbortPolicy358470bb],
keepAliveTime=Long[0],
allowCoreThreadTimeOut=Boolean[false],
corePoolSize=Integer[4],
maximumPoolSize=Integer[2147483647],
defaultHandler=AbortPolicy[java.util.concurrent.ThreadPoolExecutor$AbortPolicy358470bb],
shutdownPerm=RuntimePermission[("java.lang.RuntimePermission" "modifyThread")],
acc=null,
ONLY_ONE=Boolean[true],
$assertionsDisabled=Boolean[true],
]
```
Closes #1970 from zwangsheng/CELEBORN-1032.
Authored-by: zwangsheng <bi...@apache.org>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
(cherry picked from commit 50843c23e804da8852339dc6856e5e2276927637)
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../celeborn/common/network/client/TransportResponseHandler.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
index de1bd7379..09f6fe893 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
@@ -109,7 +109,7 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
if (checkPushTimeout) {
pushCheckerScheduleFuture =
- pushTimeoutChecker.scheduleAtFixedRate(
+ pushTimeoutChecker.scheduleWithFixedDelay(
() -> failExpiredPushRequest(),
pushTimeoutCheckerInterval,
pushTimeoutCheckerInterval,
@@ -118,7 +118,7 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
if (checkFetchTimeout) {
fetchCheckerScheduleFuture =
- fetchTimeoutChecker.scheduleAtFixedRate(
+ fetchTimeoutChecker.scheduleWithFixedDelay(
() -> failExpiredFetchRequest(),
fetchTimeoutCheckerInterval,
fetchTimeoutCheckerInterval,