You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by an...@apache.org on 2023/06/09 06:06:58 UTC
[incubator-celeborn] branch main updated: [CELEBORN-640][WORKER] DataPushQueue should not keep waiting take tasks
This is an automated email from the ASF dual-hosted git repository.
angerszhuuuu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 6b725202a [CELEBORN-640][WORKER] DataPushQueue should not keep waiting take tasks
6b725202a is described below
commit 6b725202a2526b70e7eddbc9f9c0b15ee6ea6489
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Fri Jun 9 14:06:47 2023 +0800
[CELEBORN-640][WORKER] DataPushQueue should not keep waiting take tasks
### What changes were proposed in this pull request?
In our prod meet many times of push queue stuck caused by PushState's status was not being removed.
Caused DataPushQueue to keep waiting for taking task.
Although have resolved some bugs, here we'd better add a max wait time for taking tasks since we already have the `PUSH_DATA_TIMEOUT` check method. If the target worker is really stuck, we can retry our task finally.
### Why are the changes needed?
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
Closes #1552 from AngersZhuuuu/CELEBORN-640.
Authored-by: Angerszhuuuu <an...@gmail.com>
Signed-off-by: Angerszhuuuu <an...@gmail.com>
---
.../apache/celeborn/client/write/DataPushQueue.java | 18 +++++++++++++++---
.../org/apache/celeborn/common/CelebornConf.scala | 17 +++++++++++++----
docs/configuration/client.md | 3 ++-
3 files changed, 30 insertions(+), 8 deletions(-)
diff --git a/client/src/main/java/org/apache/celeborn/client/write/DataPushQueue.java b/client/src/main/java/org/apache/celeborn/client/write/DataPushQueue.java
index 244746dac..d6070d161 100644
--- a/client/src/main/java/org/apache/celeborn/client/write/DataPushQueue.java
+++ b/client/src/main/java/org/apache/celeborn/client/write/DataPushQueue.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,7 +53,8 @@ public class DataPushQueue {
private final int numMappers;
private final int numPartitions;
private final ShuffleClient client;
- private final long takeTaskWaitTimeMs;
+ private final long takeTaskWaitIntervalMs;
+ private final int takeTaskMaxWaitAttempts;
public DataPushQueue(
CelebornConf conf,
@@ -73,7 +75,8 @@ public class DataPushQueue {
final String mapKey = Utils.makeMapKey(shuffleId, mapId, attemptId);
this.pushState = client.getPushState(mapKey);
this.maxInFlight = conf.clientPushMaxReqsInFlight();
- this.takeTaskWaitTimeMs = conf.clientPushTakeTaskWaitTimeMs();
+ this.takeTaskWaitIntervalMs = conf.clientPushTakeTaskWaitIntervalMs();
+ this.takeTaskMaxWaitAttempts = conf.clientPushTakeTaskMaxWaitAttempts();
final int capacity = conf.clientPushQueueCapacity();
workingQueue = new LinkedBlockingQueue<>(capacity);
}
@@ -85,6 +88,7 @@ public class DataPushQueue {
public ArrayList<PushTask> takePushTasks() throws IOException, InterruptedException {
ArrayList<PushTask> tasks = new ArrayList<>();
HashMap<String, Integer> workerCapacity = new HashMap<>();
+ HashMap<String, AtomicInteger> workerWaitAttempts = new HashMap<>();
while (dataPusher.stillRunning()) {
// clear() here is necessary since inflight pushes might change after sleeping
// takeTaskWaitTimeMs
@@ -106,10 +110,17 @@ public class DataPushQueue {
oldCapacity = maxInFlight - pushState.inflightPushes(loc.hostAndPushPort());
workerCapacity.put(loc.hostAndPushPort(), oldCapacity);
}
+ workerWaitAttempts.putIfAbsent(loc.hostAndPushPort(), new AtomicInteger(0));
if (oldCapacity > 0) {
iterator.remove();
tasks.add(task);
workerCapacity.put(loc.hostAndPushPort(), oldCapacity - 1);
+ } else if (workerWaitAttempts.get(loc.hostAndPushPort()).get()
+ >= takeTaskMaxWaitAttempts) {
+ iterator.remove();
+ tasks.add(task);
+ // For such worker under high pressure, we only take one task each turn.
+ workerWaitAttempts.get(loc.hostAndPushPort()).set(0);
}
} else {
iterator.remove();
@@ -125,7 +136,8 @@ public class DataPushQueue {
}
try {
// Reaching here means no available tasks can be pushed to any worker, wait for a while
- Thread.sleep(takeTaskWaitTimeMs);
+ Thread.sleep(takeTaskWaitIntervalMs);
+ workerWaitAttempts.values().forEach(AtomicInteger::incrementAndGet);
} catch (InterruptedException ie) {
logger.info("Thread interrupted while waiting push task.");
throw ie;
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 2128a2218..a10bf4a3b 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -737,7 +737,8 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
}
def clientPushLimitInFlightSleepDeltaMs: Long = get(CLIENT_PUSH_LIMIT_IN_FLIGHT_SLEEP_INTERVAL)
def clientPushSplitPartitionThreads: Int = get(CLIENT_PUSH_SPLIT_PARTITION_THREADS)
- def clientPushTakeTaskWaitTimeMs: Long = get(CLIENT_PUSH_TAKE_TASK_WAIT_TIME)
+ def clientPushTakeTaskWaitIntervalMs: Long = get(CLIENT_PUSH_TAKE_TASK_WAIT_INTERVAL)
+ def clientPushTakeTaskMaxWaitAttempts: Int = get(CLIENT_PUSH_TAKE_TASK_MAX_WAIT_ATTEMPTS)
// //////////////////////////////////////////////////////
// Client Shuffle //
@@ -2676,14 +2677,22 @@ object CelebornConf extends Logging {
.intConf
.createWithDefault(8)
- val CLIENT_PUSH_TAKE_TASK_WAIT_TIME: ConfigEntry[Long] =
- buildConf("celeborn.client.push.takeTaskWaitTime")
+ val CLIENT_PUSH_TAKE_TASK_WAIT_INTERVAL: ConfigEntry[Long] =
+ buildConf("celeborn.client.push.takeTaskWaitInterval")
.categories("client")
- .doc("Wait time if no task available to push to worker.")
+ .doc("Wait interval if no task available to push to worker.")
.version("0.3.0")
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("50ms")
+ val CLIENT_PUSH_TAKE_TASK_MAX_WAIT_ATTEMPTS: ConfigEntry[Int] =
+ buildConf("celeborn.client.push.takeTaskMaxWaitAttempts")
+ .categories("client")
+ .doc("Max wait times if no task available to push to worker.")
+ .version("0.3.0")
+ .intConf
+ .createWithDefault(1)
+
val TEST_CLIENT_RETRY_REVIVE: ConfigEntry[Boolean] =
buildConf("celeborn.test.client.retryRevive")
.withAlternative("celeborn.test.retryRevive")
diff --git a/docs/configuration/client.md b/docs/configuration/client.md
index b60fe2243..13f6b633a 100644
--- a/docs/configuration/client.md
+++ b/docs/configuration/client.md
@@ -51,7 +51,8 @@ license: |
| celeborn.client.push.sort.randomizePartitionId.enabled | false | Whether to randomize partitionId in push sorter. If true, partitionId will be randomized when sort data to avoid skew when push to worker | 0.3.0 |
| celeborn.client.push.splitPartition.threads | 8 | Thread number to process shuffle split request in shuffle client. | 0.3.0 |
| celeborn.client.push.stageEnd.timeout | <value of celeborn.<module>.io.connectionTimeout> | Timeout for waiting StageEnd. During this process, there are `celeborn.client.requestCommitFiles.maxRetries` times for retry opportunities for committing filesand 1 times for releasing slots request. User can customize this value according to your setting. By default, the value is the max timeout value `celeborn.<module>.io.connectionTimeout`. | 0.3.0 |
-| celeborn.client.push.takeTaskWaitTime | 50ms | Wait time if no task available to push to worker. | 0.3.0 |
+| celeborn.client.push.takeTaskMaxWaitAttempts | 1 | Max wait times if no task available to push to worker. | 0.3.0 |
+| celeborn.client.push.takeTaskWaitInterval | 50ms | Wait interval if no task available to push to worker. | 0.3.0 |
| celeborn.client.registerShuffle.maxRetries | 3 | Max retry times for client to register shuffle. | 0.3.0 |
| celeborn.client.registerShuffle.retryWait | 3s | Wait time before next retry if register shuffle failed. | 0.3.0 |
| celeborn.client.requestCommitFiles.maxRetries | 2 | Max retry times for requestCommitFiles RPC. | 0.3.0 |