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/04/02 01:18:17 UTC
[incubator-celeborn] branch main updated: [CELEBORN-484][PERF] Master trigger LifecycleManager commit shutdown worker's partition location. (#1395)
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 b4f8ab19b [CELEBORN-484][PERF] Master trigger LifecycleManager commit shutdown worker's partition location. (#1395)
b4f8ab19b is described below
commit b4f8ab19bd2b7c46363dcd9293790bc6f40501b4
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Sun Apr 2 09:18:12 2023 +0800
[CELEBORN-484][PERF] Master trigger LifecycleManager commit shutdown worker's partition location. (#1395)
* [CELEBORN-484][PERF] Master trigger LifecycleManager commit shutdown worker's partition location.
---
.../org/apache/celeborn/client/CommitManager.scala | 23 +++++++++++++++++++++
.../apache/celeborn/client/LifecycleManager.scala | 24 +++++++++++++++++++---
common/src/main/proto/TransportMessages.proto | 2 ++
.../common/protocol/message/ControlMessages.scala | 21 +++++++++++--------
.../apache/celeborn/common/util/PbSerDeUtils.scala | 4 +++-
.../master/clustermeta/AbstractMetaManager.java | 15 ++++++++++++--
.../clustermeta/SingleMasterMetaManager.java | 2 +-
.../deploy/master/clustermeta/ha/MetaHandler.java | 2 +-
.../celeborn/service/deploy/master/Master.scala | 14 ++++++++++++-
.../celeborn/server/common/HttpService.scala | 2 ++
.../server/common/http/HttpRequestHandler.scala | 2 ++
11 files changed, 93 insertions(+), 18 deletions(-)
diff --git a/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala b/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
index 305873f3b..9ebc3d312 100644
--- a/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
@@ -291,4 +291,27 @@ class CommitManager(appId: String, val conf: CelebornConf, lifecycleManager: Lif
}
(totalWritten, totalFileCount)
}
+
+ def handleShutdownWorker(shuttingWorker: WorkerInfo): Unit = {
+ logError(s"Worker ${shuttingWorker.toUniqueId()} shutdown, " +
+ "commit all it's partition location.")
+ lifecycleManager.shuffleAllocatedWorkers.asScala.foreach {
+ case (shuffleId, workerToPartitionLocationInfos) =>
+ val shuffleCommittedInfo = committedPartitionInfo.get(shuffleId)
+ val partitionLocationInfo = workerToPartitionLocationInfos.get(shuttingWorker)
+ if (partitionLocationInfo != null) {
+ partitionLocationInfo.getMasterPartitions().asScala.foreach { partitionLocation =>
+ shuffleCommittedInfo.synchronized {
+ shuffleCommittedInfo.unhandledPartitionLocations.add(partitionLocation)
+ }
+ }
+
+ partitionLocationInfo.getSlavePartitions().asScala.foreach { partitionLocation =>
+ shuffleCommittedInfo.synchronized {
+ shuffleCommittedInfo.unhandledPartitionLocations.add(partitionLocation)
+ }
+ }
+ }
+ }
+ }
}
diff --git a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
index 0f92876ad..b9e3f44f1 100644
--- a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -18,7 +18,7 @@
package org.apache.celeborn.client
import java.util
-import java.util.{function, List => JList}
+import java.util.{function, HashSet => JHashSet, List => JList, Set => JSet}
import java.util.concurrent.{ConcurrentHashMap, ScheduledFuture, TimeUnit}
import scala.collection.JavaConverters._
@@ -72,6 +72,7 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
// shuffle id -> (partitionId -> newest PartitionLocation)
val latestPartitionLocation =
JavaUtils.newConcurrentHashMap[Int, ConcurrentHashMap[Int, PartitionLocation]]()
+ private val shuttingWorkers: JSet[WorkerInfo] = new JHashSet[WorkerInfo]()
private val userIdentifier: UserIdentifier = IdentityProvider.instantiate(conf).provide()
@VisibleForTesting
@@ -703,7 +704,8 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
val res = requestGetBlacklist(rssHARetryClient, msg)
if (res.statusCode == StatusCode.SUCCESS) {
logInfo(s"Received Blacklist from Master, blacklist: ${res.blacklist} " +
- s"unknown workers: ${res.unknownWorkers}")
+ s"unknown workers: ${res.unknownWorkers}, shutdown workers: ${res.shutdownWorkers}")
+ resolveShutdownWorkers(res.shutdownWorkers)
val current = System.currentTimeMillis()
val reserved = blacklist.asScala
.filter { case (_, entry) =>
@@ -1194,7 +1196,11 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
} catch {
case e: Exception =>
logError(s"AskSync GetBlacklist failed.", e)
- GetBlacklistResponse(StatusCode.REQUEST_FAILED, List.empty.asJava, List.empty.asJava)
+ GetBlacklistResponse(
+ StatusCode.REQUEST_FAILED,
+ List.empty.asJava,
+ List.empty.asJava,
+ List.empty.asJava)
}
}
@@ -1230,6 +1236,18 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
}
}
+ private def resolveShutdownWorkers(newShutdownWorkers: JList[WorkerInfo]): Unit = {
+ if (!newShutdownWorkers.isEmpty) {
+ // shutdownWorkers only retain workers appeared in response.
+ shuttingWorkers.retainAll(newShutdownWorkers)
+ newShutdownWorkers.asScala.filterNot(shuttingWorkers.asScala.contains)
+ .foreach { workerInfo =>
+ commitManager.handleShutdownWorker(workerInfo)
+ }
+ shuttingWorkers.addAll(newShutdownWorkers)
+ }
+ }
+
private def shuffleResourceExists(shuffleId: Int): Boolean = {
val workers = workerSnapshots(shuffleId)
workers != null && !workers.isEmpty
diff --git a/common/src/main/proto/TransportMessages.proto b/common/src/main/proto/TransportMessages.proto
index b7510c943..638e23e15 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -292,6 +292,7 @@ message PbGetBlacklistResponse {
int32 status = 1;
repeated PbWorkerInfo blacklist = 2;
repeated PbWorkerInfo unknownWorkers = 3;
+ repeated PbWorkerInfo shutdownWorkers = 4;
}
message PbCheckQuota {
@@ -465,4 +466,5 @@ message PbSnapshotMetaInfo {
repeated PbAppDiskUsageSnapshot appDiskUsageMetricSnapshots = 10;
optional PbAppDiskUsageSnapshot currentAppDiskUsageMetricsSnapshot = 11;
map<string, int64> lostWorkers = 12;
+ repeated PbWorkerInfo shutdownWorkers = 13;
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
index 0cb3a3a8f..c6176ce5c 100644
--- a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
@@ -324,7 +324,8 @@ object ControlMessages extends Logging {
case class GetBlacklistResponse(
statusCode: StatusCode,
blacklist: util.List[WorkerInfo],
- unknownWorkers: util.List[WorkerInfo]) extends Message
+ unknownWorkers: util.List[WorkerInfo],
+ shutdownWorkers: util.List[WorkerInfo]) extends Message
case class CheckQuota(userIdentifier: UserIdentifier) extends Message
@@ -620,16 +621,16 @@ object ControlMessages extends Logging {
.build().toByteArray
new TransportMessage(MessageType.GET_BLACKLIST, payload)
- case GetBlacklistResponse(statusCode, blacklist, unknownWorkers) =>
+ case GetBlacklistResponse(statusCode, blacklist, unknownWorkers, shutdownWorkers) =>
val builder = PbGetBlacklistResponse.newBuilder()
.setStatus(statusCode.getValue)
- builder.addAllBlacklist(blacklist.asScala.map { workerInfo =>
- PbSerDeUtils.toPbWorkerInfo(workerInfo, true)
- }
- .toList.asJava)
- builder.addAllUnknownWorkers(unknownWorkers.asScala.map { workerInfo =>
- PbSerDeUtils.toPbWorkerInfo(workerInfo, true)
- }.toList.asJava)
+ builder.addAllBlacklist(
+ blacklist.asScala.map(PbSerDeUtils.toPbWorkerInfo(_, true)).toList.asJava)
+ builder.addAllUnknownWorkers(
+ unknownWorkers.asScala.map(PbSerDeUtils.toPbWorkerInfo(_, true)).toList.asJava)
+ builder.addAllShutdownWorkers(
+ shutdownWorkers.asScala.map(PbSerDeUtils.toPbWorkerInfo(_, true)).toList.asJava)
+
val payload = builder.build().toByteArray
new TransportMessage(MessageType.GET_BLACKLIST_RESPONSE, payload)
@@ -955,6 +956,8 @@ object ControlMessages extends Logging {
pbGetBlacklistResponse.getBlacklistList.asScala
.map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava,
pbGetBlacklistResponse.getUnknownWorkersList.asScala
+ .map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava,
+ pbGetBlacklistResponse.getShutdownWorkersList.asScala
.map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava)
case CHECK_QUOTA =>
diff --git a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
index a7a4f1108..64f6d0527 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
@@ -370,7 +370,8 @@ object PbSerDeUtils {
partitionTotalFileCount: java.lang.Long,
appDiskUsageMetricSnapshots: Array[AppDiskUsageSnapShot],
currentAppDiskUsageMetricsSnapshot: AppDiskUsageSnapShot,
- lostWorkers: ConcurrentHashMap[WorkerInfo, java.lang.Long]): PbSnapshotMetaInfo = {
+ lostWorkers: ConcurrentHashMap[WorkerInfo, java.lang.Long],
+ shutdownWorkers: java.util.Set[WorkerInfo]): PbSnapshotMetaInfo = {
val builder = PbSnapshotMetaInfo.newBuilder()
.setEstimatedPartitionSize(estimatedPartitionSize)
.addAllRegisteredShuffle(registeredShuffle)
@@ -388,6 +389,7 @@ object PbSerDeUtils {
.putAllLostWorkers(lostWorkers.asScala.map {
case (worker: WorkerInfo, time: java.lang.Long) => (worker.toUniqueId(), time)
}.asJava)
+ .addAllShutdownWorkers(shutdownWorkers.asScala.map(toPbWorkerInfo(_, true)).asJava)
if (currentAppDiskUsageMetricsSnapshot != null) {
builder.setCurrentAppDiskUsageMetricsSnapshot(
toPbAppDiskUsageSnapshot(currentAppDiskUsageMetricsSnapshot))
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
index 596fffc4b..bbf0c3204 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
@@ -57,6 +57,9 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
public final ConcurrentHashMap<String, Long> appHeartbeatTime = JavaUtils.newConcurrentHashMap();
// blacklist
public final Set<WorkerInfo> blacklist = ConcurrentHashMap.newKeySet();
+ // shutdown workers
+ public final Set<WorkerInfo> shutdownWorkers = ConcurrentHashMap.newKeySet();
+
// workerLost events
public final Set<WorkerInfo> workerLostEvents = ConcurrentHashMap.newKeySet();
@@ -244,6 +247,7 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
synchronized (workers) {
if (!workers.contains(workerInfo)) {
workers.add(workerInfo);
+ shutdownWorkers.remove(workerInfo);
lostWorkers.remove(workerInfo);
}
}
@@ -269,7 +273,8 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
partitionTotalFileCount.sum(),
appDiskUsageMetric.snapShots(),
appDiskUsageMetric.currentSnapShot().get(),
- lostWorkers)
+ lostWorkers,
+ shutdownWorkers)
.toByteArray();
Files.write(file.toPath(), snapshotBytes);
}
@@ -317,6 +322,11 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
.forEach(
entry -> lostWorkers.put(WorkerInfo.fromUniqueId(entry.getKey()), entry.getValue()));
+ shutdownWorkers.addAll(
+ snapshotMetaInfo.getShutdownWorkersList().stream()
+ .map(PbSerDeUtils::fromPbWorkerInfo)
+ .collect(Collectors.toSet()));
+
partitionTotalWritten.reset();
partitionTotalWritten.add(snapshotMetaInfo.getPartitionTotalWritten());
partitionTotalFileCount.reset();
@@ -342,8 +352,9 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
registeredShuffle.forEach(shuffle -> LOG.info("RegisteredShuffle {}", shuffle));
}
- public void updateBlacklistByReportWorkerUnavailable(List<WorkerInfo> failedWorkers) {
+ public void updateMetaByReportWorkerUnavailable(List<WorkerInfo> failedWorkers) {
synchronized (this.workers) {
+ shutdownWorkers.addAll(failedWorkers);
failedWorkers.retainAll(this.workers);
this.blacklist.addAll(failedWorkers);
}
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
index 85d05ebd2..b5684be06 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
@@ -128,7 +128,7 @@ public class SingleMasterMetaManager extends AbstractMetaManager {
@Override
public void handleReportWorkerUnavailable(List<WorkerInfo> failedNodes, String requestId) {
- updateBlacklistByReportWorkerUnavailable(failedNodes);
+ updateMetaByReportWorkerUnavailable(failedNodes);
}
public void handleUpdatePartitionSize() {
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
index a890ac1ff..fc82dd3e2 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
@@ -236,7 +236,7 @@ public class MetaHandler {
request.getReportWorkerUnavailableRequest().getUnavailableList();
List<WorkerInfo> failedWorkers =
failedAddress.stream().map(MetaUtil::addrToInfo).collect(Collectors.toList());
- metaSystem.updateBlacklistByReportWorkerUnavailable(failedWorkers);
+ metaSystem.updateMetaByReportWorkerUnavailable(failedWorkers);
break;
case UpdatePartitionSize:
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
index 386da3532..223f65d68 100644
--- 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
@@ -106,6 +106,8 @@ private[celeborn] class Master(
statusSystem.workers.synchronized(new util.ArrayList[WorkerInfo](statusSystem.workers))
private def lostWorkersSnapshot: ConcurrentHashMap[WorkerInfo, java.lang.Long] =
statusSystem.workers.synchronized(JavaUtils.newConcurrentHashMap(statusSystem.lostWorkers))
+ private def shutdownWorkerSnapshot: util.List[WorkerInfo] =
+ statusSystem.workers.synchronized(new util.ArrayList[WorkerInfo](statusSystem.shutdownWorkers))
private def diskReserveSize = conf.diskReserveSize
@@ -596,7 +598,8 @@ private[celeborn] class Master(
GetBlacklistResponse(
StatusCode.SUCCESS,
new util.ArrayList(statusSystem.blacklist),
- msg.localBlacklist))
+ msg.localBlacklist,
+ shutdownWorkerSnapshot))
}
private def handleGetWorkerInfos(context: RpcCallContext): Unit = {
@@ -732,6 +735,15 @@ private[celeborn] class Master(
sb.toString()
}
+ override def getShutdownWorkers: String = {
+ val sb = new StringBuilder
+ sb.append("===================== Shutdown Workers in Master ======================\n")
+ shutdownWorkerSnapshot.asScala.foreach { worker =>
+ sb.append(s"${worker.toUniqueId()}\n")
+ }
+ sb.toString()
+ }
+
override def getBlacklistedWorkers: String = {
val sb = new StringBuilder
sb.append("==================== Blacklisted Workers in Master =====================\n")
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala b/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
index 259b59d49..0dc8c73a7 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
@@ -38,6 +38,8 @@ abstract class HttpService extends Service with Logging {
def getLostWorkers: String = throw new UnsupportedOperationException()
+ def getShutdownWorkers: String = throw new UnsupportedOperationException()
+
def getBlacklistedWorkers: String = throw new UnsupportedOperationException()
def getThreadDump: String
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
index ed2a240f8..53e4ba5f5 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
@@ -70,6 +70,8 @@ class HttpRequestHandler(
service.getLostWorkers
case "/blacklistedWorkers" if service.serviceName == Service.MASTER =>
service.getBlacklistedWorkers
+ case "/shutdownWorkers" if service.serviceName == Service.MASTER =>
+ service.getShutdownWorkers
case "/threadDump" =>
service.getThreadDump
case "/hostnames" if service.serviceName == Service.MASTER =>