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/12 06:59:36 UTC
[incubator-celeborn] branch main updated: [CELEBORN-502][REFACTOR] Merge GetBlacklistResponse to HeartbeatFromApplication (#1408)
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 e5722126e [CELEBORN-502][REFACTOR] Merge GetBlacklistResponse to HeartbeatFromApplication (#1408)
e5722126e is described below
commit e5722126e9412b70175e7b9377704b1581105c33
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Wed Apr 12 14:59:32 2023 +0800
[CELEBORN-502][REFACTOR] Merge GetBlacklistResponse to HeartbeatFromApplication (#1408)
* [CELEBORN-502][REFACTOR] Merge GetBlacklistResponse to HeartbeatFromApplication
---
.../celeborn/client/ApplicationHeartbeater.scala | 38 ++++-
.../celeborn/client/ChangePartitionManager.scala | 4 +-
.../org/apache/celeborn/client/CommitManager.scala | 7 +-
.../apache/celeborn/client/LifecycleManager.scala | 170 ++-------------------
.../celeborn/client/WorkerStatusTracker.scala | 158 +++++++++++++++++++
common/src/main/proto/TransportMessages.proto | 10 +-
.../org/apache/celeborn/common/CelebornConf.scala | 10 --
.../celeborn/common/metrics/source/RPCSource.scala | 4 -
.../common/protocol/message/ControlMessages.scala | 46 +++++-
docs/configuration/client.md | 1 -
.../celeborn/service/deploy/master/Master.scala | 20 ++-
11 files changed, 269 insertions(+), 199 deletions(-)
diff --git a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
index a00dc8920..13383d30c 100644
--- a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
@@ -19,19 +19,22 @@ package org.apache.celeborn.client
import java.util.concurrent.{ScheduledFuture, TimeUnit}
+import scala.collection.JavaConverters._
import scala.concurrent.duration.DurationInt
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.haclient.RssHARetryClient
import org.apache.celeborn.common.internal.Logging
-import org.apache.celeborn.common.protocol.message.ControlMessages.{HeartbeatFromApplication, ZERO_UUID}
+import org.apache.celeborn.common.protocol.message.ControlMessages.{HeartbeatFromApplication, HeartbeatFromApplicationResponse, ZERO_UUID}
+import org.apache.celeborn.common.protocol.message.StatusCode
import org.apache.celeborn.common.util.ThreadUtils
class ApplicationHeartbeater(
appId: String,
conf: CelebornConf,
rssHARetryClient: RssHARetryClient,
- shuffleMetrics: () => (Long, Long)) extends Logging {
+ shuffleMetrics: () => (Long, Long),
+ workerStatusTracker: WorkerStatusTracker) extends Logging {
// Use independent app heartbeat threads to avoid being blocked by other operations.
private val appHeartbeatIntervalMs = conf.appHeartbeatIntervalMs
@@ -48,9 +51,17 @@ class ApplicationHeartbeater(
val (tmpTotalWritten, tmpFileCount) = shuffleMetrics()
logDebug(s"Send app heartbeat with $tmpTotalWritten $tmpFileCount")
val appHeartbeat =
- HeartbeatFromApplication(appId, tmpTotalWritten, tmpFileCount, ZERO_UUID)
- rssHARetryClient.send(appHeartbeat)
- logDebug("Successfully send app heartbeat.")
+ HeartbeatFromApplication(
+ appId,
+ tmpTotalWritten,
+ tmpFileCount,
+ workerStatusTracker.blacklist.asScala.keys.toList.asJava,
+ ZERO_UUID)
+ val response = requestHeartbeat(appHeartbeat)
+ if (response.statusCode == StatusCode.SUCCESS) {
+ logDebug("Successfully send app heartbeat.")
+ workerStatusTracker.handleHeartbeatResponse(response)
+ }
} catch {
case it: InterruptedException =>
logWarning("Interrupted while sending app heartbeat.")
@@ -66,6 +77,23 @@ class ApplicationHeartbeater(
TimeUnit.MILLISECONDS)
}
+ private def requestHeartbeat(message: HeartbeatFromApplication)
+ : HeartbeatFromApplicationResponse = {
+ try {
+ rssHARetryClient.askSync[HeartbeatFromApplicationResponse](
+ message,
+ classOf[HeartbeatFromApplicationResponse])
+ } catch {
+ case e: Exception =>
+ logError(s"AskSync HeartbeatFromApplication failed.", e)
+ HeartbeatFromApplicationResponse(
+ StatusCode.REQUEST_FAILED,
+ List.empty.asJava,
+ List.empty.asJava,
+ List.empty.asJava)
+ }
+ }
+
def stop(): Unit = {
appHeartbeat.cancel(true)
ThreadUtils.shutdown(appHeartbeatHandlerThread, 800.millis)
diff --git a/client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala b/client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala
index eb55d22ba..daf058b08 100644
--- a/client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala
@@ -207,7 +207,7 @@ class ChangePartitionManager(
// Blacklist all failed workers
if (changePartitions.exists(_.causes.isDefined)) {
changePartitions.filter(_.causes.isDefined).foreach { changePartition =>
- lifecycleManager.blacklistPartition(
+ lifecycleManager.workerStatusTracker.blacklistWorkerFromPartition(
shuffleId,
changePartition.oldPartition,
changePartition.causes.get)
@@ -252,7 +252,7 @@ class ChangePartitionManager(
.workerSnapshots(shuffleId)
.keySet()
.asScala
- .filter(w => !lifecycleManager.blacklist.keySet().contains(w))
+ .filter(w => !lifecycleManager.workerStatusTracker.blacklist.keySet().contains(w))
.toList
if (candidates.size < 1 || (pushReplicateEnabled && candidates.size < 2)) {
logError("[Update partition] failed for not enough candidates for revive.")
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 9ebc3d312..634bfdda8 100644
--- a/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
@@ -146,7 +146,8 @@ class CommitManager(appId: String, val conf: CelebornConf, lifecycleManager: Lif
slaveIds,
commitFilesFailedWorkers)
}
- lifecycleManager.recordWorkerFailure(commitFilesFailedWorkers)
+ lifecycleManager.workerStatusTracker.recordWorkerFailure(
+ commitFilesFailedWorkers)
} finally {
// when batch commit thread ends, we need decrementInFlightNum
commitHandler.decrementInFlightNum(shuffleCommittedInfo, workerToRequests)
@@ -210,7 +211,7 @@ class CommitManager(appId: String, val conf: CelebornConf, lifecycleManager: Lif
numMappers,
partitionId,
r =>
- lifecycleManager.recordWorkerFailure(r))
+ lifecycleManager.workerStatusTracker.recordWorkerFailure(r))
}
def releasePartitionResource(shuffleId: Int, partitionId: Int): Unit = {
@@ -237,7 +238,7 @@ class CommitManager(appId: String, val conf: CelebornConf, lifecycleManager: Lif
def tryFinalCommit(shuffleId: Int): Boolean = {
getCommitHandler(shuffleId).tryFinalCommit(
shuffleId,
- r => lifecycleManager.recordWorkerFailure(r))
+ r => lifecycleManager.workerStatusTracker.recordWorkerFailure(r))
}
def isStageEnd(shuffleId: Int): Boolean = {
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 f76a5158e..6d73a2e3b 100644
--- a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -56,8 +56,6 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
private val lifecycleHost = Utils.localHostName
private val shuffleExpiredCheckIntervalMs = conf.shuffleExpiredCheckIntervalMs
- private val workerExcludedCheckIntervalMs = conf.workerExcludedCheckIntervalMs
- private val workerExcludedExpireTimeout = conf.workerExcludedExpireTimeout
private val pushReplicateEnabled = conf.pushReplicateEnabled
private val partitionSplitThreshold = conf.partitionSplitThreshold
private val partitionSplitMode = conf.partitionSplitMode
@@ -72,7 +70,6 @@ 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
@@ -103,14 +100,10 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
private val registeringShuffleRequest =
JavaUtils.newConcurrentHashMap[Int, util.Set[RegisterCallContext]]()
- // blacklist
- val blacklist = new ShuffleFailedWorkers()
-
// Threads
private val forwardMessageThread =
ThreadUtils.newDaemonSingleThreadScheduledExecutor("master-forward-message-thread")
private var checkForShuffleRemoval: ScheduledFuture[_] = _
- private var getBlacklist: ScheduledFuture[_] = _
// init driver rss meta rpc service
override val rpcEnv: RpcEnv = RpcEnv.create(
@@ -124,8 +117,14 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
private val rssHARetryClient = new RssHARetryClient(rpcEnv, conf)
val commitManager = new CommitManager(appId, conf, this)
+ val workerStatusTracker = new WorkerStatusTracker(conf, this, commitManager)
private val heartbeater =
- new ApplicationHeartbeater(appId, conf, rssHARetryClient, () => commitManager.commitMetrics())
+ new ApplicationHeartbeater(
+ appId,
+ conf,
+ rssHARetryClient,
+ () => commitManager.commitMetrics(),
+ workerStatusTracker)
private val changePartitionManager = new ChangePartitionManager(conf, this)
private val releasePartitionManager = new ReleasePartitionManager(appId, conf, this)
@@ -153,24 +152,12 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
shuffleExpiredCheckIntervalMs,
shuffleExpiredCheckIntervalMs,
TimeUnit.MILLISECONDS)
-
- // noinspection ConvertExpressionToSAM
- getBlacklist = forwardMessageThread.scheduleAtFixedRate(
- new Runnable {
- override def run(): Unit = Utils.tryLogNonFatalError {
- self.send(GetBlacklist(blacklist.asScala.keys.toList.asJava))
- }
- },
- workerExcludedCheckIntervalMs,
- workerExcludedCheckIntervalMs,
- TimeUnit.MILLISECONDS)
}
override def onStop(): Unit = {
import scala.concurrent.duration._
checkForShuffleRemoval.cancel(true)
- getBlacklist.cancel(true)
ThreadUtils.shutdown(forwardMessageThread, 800.millis)
commitManager.stop()
@@ -204,8 +191,6 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
override def receive: PartialFunction[Any, Unit] = {
case RemoveExpiredShuffle =>
removeExpiredShuffle()
- case msg: GetBlacklist =>
- handleGetBlacklist(msg)
case StageEnd(applicationId, shuffleId) =>
logInfo(s"Received StageEnd request, ${Utils.makeShuffleKey(applicationId, shuffleId)}.")
handleStageEnd(applicationId, shuffleId)
@@ -455,7 +440,7 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
}
candidatesWorkers.removeAll(connectFailedWorkers.asScala.keys.toList.asJava)
- recordWorkerFailure(connectFailedWorkers)
+ workerStatusTracker.recordWorkerFailure(connectFailedWorkers)
// Third, for each slot, LifecycleManager should ask Worker to reserve the slot
// and prepare the pushing data env.
@@ -499,58 +484,6 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
}
}
- def blacklistPartition(
- shuffleId: Int,
- oldPartition: PartitionLocation,
- cause: StatusCode): Unit = {
- val failedWorker = new ShuffleFailedWorkers()
-
- def blacklistPartitionWorker(
- partition: PartitionLocation,
- statusCode: StatusCode): Unit = {
- val tmpWorker = partition.getWorker
- val worker = workerSnapshots(shuffleId).keySet().asScala.find(_.equals(tmpWorker))
- if (worker.isDefined) {
- failedWorker.put(worker.get, (statusCode, System.currentTimeMillis()))
- }
- }
-
- if (oldPartition != null) {
- cause match {
- case StatusCode.PUSH_DATA_WRITE_FAIL_MASTER =>
- blacklistPartitionWorker(oldPartition, StatusCode.PUSH_DATA_WRITE_FAIL_MASTER)
- case StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE
- if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
- blacklistPartitionWorker(oldPartition.getPeer, StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE)
- case StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER =>
- blacklistPartitionWorker(oldPartition, StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER)
- case StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE
- if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
- blacklistPartitionWorker(
- oldPartition.getPeer,
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE)
- case StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER =>
- blacklistPartitionWorker(oldPartition, StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER)
- case StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE
- if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
- blacklistPartitionWorker(
- oldPartition.getPeer,
- StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE)
- case StatusCode.PUSH_DATA_TIMEOUT_MASTER =>
- blacklistPartitionWorker(oldPartition, StatusCode.PUSH_DATA_TIMEOUT_MASTER)
- case StatusCode.PUSH_DATA_TIMEOUT_SLAVE
- if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
- blacklistPartitionWorker(
- oldPartition.getPeer,
- StatusCode.PUSH_DATA_TIMEOUT_SLAVE)
- case _ =>
- }
- }
- if (!failedWorker.isEmpty) {
- recordWorkerFailure(failedWorker)
- }
- }
-
private def handleRevive(
context: RpcCallContext,
applicationId: String,
@@ -700,44 +633,6 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
logInfo(s"Unregister for $shuffleId success.")
}
- private def handleGetBlacklist(msg: GetBlacklist): Unit = {
- val res = requestGetBlacklist(rssHARetryClient, msg)
- if (res.statusCode == StatusCode.SUCCESS) {
- logInfo(s"Received Blacklist from Master, blacklist: ${res.blacklist} " +
- s"unknown workers: ${res.unknownWorkers}, shutdown workers: ${res.shutdownWorkers}")
- resolveShutdownWorkers(res.shutdownWorkers)
- val current = System.currentTimeMillis()
- val reserved = blacklist.asScala
- .filter { case (_, entry) =>
- val (statusCode, registerTime) = entry
- statusCode match {
- case StatusCode.WORKER_SHUTDOWN |
- StatusCode.NO_AVAILABLE_WORKING_DIR |
- StatusCode.RESERVE_SLOTS_FAILED |
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER |
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE |
- StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER |
- StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE |
- StatusCode.PUSH_DATA_TIMEOUT_MASTER |
- StatusCode.PUSH_DATA_TIMEOUT_SLAVE
- if current - registerTime < workerExcludedExpireTimeout =>
- true
- case StatusCode.UNKNOWN_WORKER => true
- case _ => false
- }
- }.asJava
- val reservedBlackList = new ShuffleFailedWorkers()
- reservedBlackList.putAll(reserved)
- blacklist.clear()
- blacklist.putAll(
- res.blacklist.asScala.map(_ -> (StatusCode.WORKER_IN_BLACKLIST -> current)).toMap.asJava)
- blacklist.putAll(
- res.unknownWorkers.asScala.map(_ -> (StatusCode.UNKNOWN_WORKER -> current)).toMap.asJava)
- // put reserved blacklist at last to cover blacklist's local status.
- blacklist.putAll(reservedBlackList)
- }
- }
-
/* ========================================================== *
| END OF EVENT HANDLER |
* ========================================================== */
@@ -788,7 +683,7 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
logError(s"Aggregated error of reserveSlots failure:${failureInfos.asScala.foldLeft("")(
(x, y) => s"$x \n $y")}")
}
- recordWorkerFailure(reserveSlotFailedWorkers)
+ workerStatusTracker.recordWorkerFailure(reserveSlotFailedWorkers)
new util.ArrayList[WorkerInfo](reserveSlotFailedWorkers.asScala.keys.toList.asJava)
}
@@ -949,7 +844,7 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
// add candidates to avoid revive action passed in slots only 2 worker
retryCandidates.addAll(candidates)
// remove blacklist from retryCandidates
- retryCandidates.removeAll(blacklist.keys().asScala.toList.asJava)
+ retryCandidates.removeAll(workerStatusTracker.blacklist.keys().asScala.toList.asJava)
if (retryCandidates.size < 1 || (pushReplicateEnabled && retryCandidates.size < 2)) {
logError("Retry reserve slots failed caused by not enough slots.")
noAvailableSlots = true
@@ -1188,41 +1083,6 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
}
}
- private def requestGetBlacklist(
- rssHARetryClient: RssHARetryClient,
- message: GetBlacklist): GetBlacklistResponse = {
- try {
- rssHARetryClient.askSync[GetBlacklistResponse](message, classOf[GetBlacklistResponse])
- } catch {
- case e: Exception =>
- logError(s"AskSync GetBlacklist failed.", e)
- GetBlacklistResponse(
- StatusCode.REQUEST_FAILED,
- List.empty.asJava,
- List.empty.asJava,
- List.empty.asJava)
- }
- }
-
- def recordWorkerFailure(failures: ShuffleFailedWorkers): Unit = {
- val failedWorker = new ShuffleFailedWorkers(failures)
- logInfo(s"Report Worker Failure: ${failedWorker.asScala}, current blacklist $blacklist")
- failedWorker.asScala.foreach { case (worker, (statusCode, registerTime)) =>
- if (!blacklist.containsKey(worker)) {
- blacklist.put(worker, (statusCode, registerTime))
- } else {
- statusCode match {
- case StatusCode.WORKER_SHUTDOWN |
- StatusCode.NO_AVAILABLE_WORKING_DIR |
- StatusCode.RESERVE_SLOTS_FAILED |
- StatusCode.UNKNOWN_WORKER =>
- blacklist.put(worker, (statusCode, blacklist.get(worker)._2))
- case _ => // Not cover
- }
- }
- }
- }
-
def checkQuota(): CheckQuotaResponse = {
try {
rssHARetryClient.askSync[CheckQuotaResponse](
@@ -1236,16 +1096,6 @@ class LifecycleManager(appId: String, val conf: CelebornConf) extends RpcEndpoin
}
}
- private def resolveShutdownWorkers(newShutdownWorkers: JList[WorkerInfo]): Unit = {
- // 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/client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala b/client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala
new file mode 100644
index 000000000..4d8b2eecd
--- /dev/null
+++ b/client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.client
+
+import java.util.{HashSet => JHashSet, Set => JSet}
+
+import scala.collection.JavaConverters._
+
+import org.apache.celeborn.client.LifecycleManager.ShuffleFailedWorkers
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.meta.WorkerInfo
+import org.apache.celeborn.common.protocol.PartitionLocation
+import org.apache.celeborn.common.protocol.message.ControlMessages.HeartbeatFromApplicationResponse
+import org.apache.celeborn.common.protocol.message.StatusCode
+
+class WorkerStatusTracker(
+ conf: CelebornConf,
+ lifecycleManager: LifecycleManager,
+ commitManager: CommitManager) extends Logging {
+ private val workerExcludedExpireTimeout = conf.workerExcludedExpireTimeout
+
+ // blacklist
+ val blacklist = new ShuffleFailedWorkers()
+ private val shuttingWorkers: JSet[WorkerInfo] = new JHashSet[WorkerInfo]()
+
+ def blacklistWorkerFromPartition(
+ shuffleId: Int,
+ oldPartition: PartitionLocation,
+ cause: StatusCode): Unit = {
+ val failedWorker = new ShuffleFailedWorkers()
+
+ def blacklistWorker(partition: PartitionLocation, statusCode: StatusCode): Unit = {
+ val tmpWorker = partition.getWorker
+ val worker =
+ lifecycleManager.workerSnapshots(shuffleId).keySet().asScala.find(_.equals(tmpWorker))
+ if (worker.isDefined) {
+ failedWorker.put(worker.get, (statusCode, System.currentTimeMillis()))
+ }
+ }
+
+ if (oldPartition != null) {
+ cause match {
+ case StatusCode.PUSH_DATA_WRITE_FAIL_MASTER =>
+ blacklistWorker(oldPartition, StatusCode.PUSH_DATA_WRITE_FAIL_MASTER)
+ case StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE
+ if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
+ blacklistWorker(oldPartition.getPeer, StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE)
+ case StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER =>
+ blacklistWorker(oldPartition, StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER)
+ case StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE
+ if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
+ blacklistWorker(
+ oldPartition.getPeer,
+ StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE)
+ case StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER =>
+ blacklistWorker(oldPartition, StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER)
+ case StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE
+ if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
+ blacklistWorker(
+ oldPartition.getPeer,
+ StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE)
+ case StatusCode.PUSH_DATA_TIMEOUT_MASTER =>
+ blacklistWorker(oldPartition, StatusCode.PUSH_DATA_TIMEOUT_MASTER)
+ case StatusCode.PUSH_DATA_TIMEOUT_SLAVE
+ if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
+ blacklistWorker(
+ oldPartition.getPeer,
+ StatusCode.PUSH_DATA_TIMEOUT_SLAVE)
+ case _ =>
+ }
+ }
+ if (!failedWorker.isEmpty) {
+ recordWorkerFailure(failedWorker)
+ }
+ }
+
+ def recordWorkerFailure(failures: ShuffleFailedWorkers): Unit = {
+ val failedWorker = new ShuffleFailedWorkers(failures)
+ logInfo(s"Report Worker Failure: ${failedWorker.asScala}, current blacklist $blacklist")
+ failedWorker.asScala.foreach { case (worker, (statusCode, registerTime)) =>
+ if (!blacklist.containsKey(worker)) {
+ blacklist.put(worker, (statusCode, registerTime))
+ } else {
+ statusCode match {
+ case StatusCode.WORKER_SHUTDOWN |
+ StatusCode.NO_AVAILABLE_WORKING_DIR |
+ StatusCode.RESERVE_SLOTS_FAILED |
+ StatusCode.UNKNOWN_WORKER =>
+ blacklist.put(worker, (statusCode, blacklist.get(worker)._2))
+ case _ => // Not cover
+ }
+ }
+ }
+ }
+
+ def handleHeartbeatResponse(res: HeartbeatFromApplicationResponse): Unit = {
+ if (res.statusCode == StatusCode.SUCCESS) {
+ logInfo(s"Received Blacklist from Master, blacklist: ${res.blacklist} " +
+ s"unknown workers: ${res.unknownWorkers}, shutdown workers: ${res.shuttingWorkers}")
+ resolveShutdownWorkers(res.shuttingWorkers)
+ val current = System.currentTimeMillis()
+ val reserved = blacklist.asScala
+ .filter { case (_, entry) =>
+ val (statusCode, registerTime) = entry
+ statusCode match {
+ case StatusCode.WORKER_SHUTDOWN |
+ StatusCode.NO_AVAILABLE_WORKING_DIR |
+ StatusCode.RESERVE_SLOTS_FAILED |
+ StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER |
+ StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE |
+ StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER |
+ StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE |
+ StatusCode.PUSH_DATA_TIMEOUT_MASTER |
+ StatusCode.PUSH_DATA_TIMEOUT_SLAVE
+ if current - registerTime < workerExcludedExpireTimeout =>
+ true
+ case StatusCode.UNKNOWN_WORKER => true
+ case _ => false
+ }
+ }.asJava
+ val reservedBlackList = new ShuffleFailedWorkers()
+ reservedBlackList.putAll(reserved)
+ blacklist.clear()
+ blacklist.putAll(
+ res.blacklist.asScala.map(_ -> (StatusCode.WORKER_IN_BLACKLIST -> current)).toMap.asJava)
+ blacklist.putAll(
+ res.unknownWorkers.asScala.map(_ -> (StatusCode.UNKNOWN_WORKER -> current)).toMap.asJava)
+ // put reserved blacklist at last to cover blacklist's local status.
+ blacklist.putAll(reservedBlackList)
+ }
+ }
+
+ private def resolveShutdownWorkers(newShutdownWorkers: java.util.List[WorkerInfo]): Unit = {
+ // shutdownWorkers only retain workers appeared in response.
+ shuttingWorkers.retainAll(newShutdownWorkers)
+ newShutdownWorkers.asScala.filterNot(shuttingWorkers.asScala.contains)
+ .foreach { workerInfo =>
+ commitManager.handleShutdownWorker(workerInfo)
+ }
+ shuttingWorkers.addAll(newShutdownWorkers)
+ }
+}
diff --git a/common/src/main/proto/TransportMessages.proto b/common/src/main/proto/TransportMessages.proto
index 638e23e15..8e533a8d1 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -67,6 +67,7 @@ enum MessageType {
STAGE_END_RESPONSE = 46;
PARTITION_SPLIT = 47;
REGISTER_MAP_PARTITION_TASK = 48;
+ HEARTBEAT_FROM_APPLICATION_RESPONSE = 49;
}
message PbStorageInfo {
@@ -282,6 +283,14 @@ message PbHeartbeatFromApplication {
int64 totalWritten = 2;
int64 fileCount = 3 ;
string requestId = 4;
+ repeated PbWorkerInfo localBlackList = 5;
+}
+
+message PbHeartbeatFromApplicationResponse {
+ int32 status = 1;
+ repeated PbWorkerInfo blacklist = 2;
+ repeated PbWorkerInfo unknownWorkers = 3;
+ repeated PbWorkerInfo shuttingWorkers = 4;
}
message PbGetBlacklist {
@@ -292,7 +301,6 @@ message PbGetBlacklistResponse {
int32 status = 1;
repeated PbWorkerInfo blacklist = 2;
repeated PbWorkerInfo unknownWorkers = 3;
- repeated PbWorkerInfo shutdownWorkers = 4;
}
message PbCheckQuota {
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 fdc9c3eb3..e5d3bb589 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -535,7 +535,6 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def appHeartbeatTimeoutMs: Long = get(APPLICATION_HEARTBEAT_TIMEOUT)
def appHeartbeatIntervalMs: Long = get(APPLICATION_HEARTBEAT_INTERVAL)
def shuffleExpiredCheckIntervalMs: Long = get(SHUFFLE_EXPIRED_CHECK_INTERVAL)
- def workerExcludedCheckIntervalMs: Long = get(WORKER_EXCLUDED_INTERVAL)
def workerExcludedExpireTimeout: Long = get(WORKER_EXCLUDED_EXPIRE_TIMEOUT)
def blacklistSlaveEnabled: Boolean = get(BLACKLIST_SLAVE_ENABLED)
def shuffleRangeReadFilterEnabled: Boolean = get(SHUFFLE_RANGE_READ_FILTER_ENABLED)
@@ -1416,15 +1415,6 @@ object CelebornConf extends Logging {
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("60s")
- val WORKER_EXCLUDED_INTERVAL: ConfigEntry[Long] =
- buildConf("celeborn.worker.excluded.checkInterval")
- .withAlternative("rss.get.blacklist.delay")
- .categories("client")
- .version("0.2.0")
- .doc("Interval for client to refresh excluded worker list.")
- .timeConf(TimeUnit.MILLISECONDS)
- .createWithDefaultString("30s")
-
val WORKER_EXCLUDED_EXPIRE_TIMEOUT: ConfigEntry[Long] =
buildConf("celeborn.worker.excluded.expireTimeout")
.categories("client")
diff --git a/common/src/main/scala/org/apache/celeborn/common/metrics/source/RPCSource.scala b/common/src/main/scala/org/apache/celeborn/common/metrics/source/RPCSource.scala
index 946d6b7de..52ac8e2e3 100644
--- a/common/src/main/scala/org/apache/celeborn/common/metrics/source/RPCSource.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/metrics/source/RPCSource.scala
@@ -49,7 +49,6 @@ class RPCSource(conf: CelebornConf, role: String) extends AbstractSource(conf, r
addCounter(RPCReleaseSlotsNum)
addCounter(RPCReleaseSlotsSize)
addCounter(RPCUnregisterShuffleNum)
- addCounter(RPCGetBlacklistNum)
addCounter(RPCReportWorkerUnavailableNum)
addCounter(RPCReportWorkerUnavailableSize)
addCounter(RPCCheckQuotaNum)
@@ -88,8 +87,6 @@ class RPCSource(conf: CelebornConf, role: String) extends AbstractSource(conf, r
incCounter(RPCReleaseSlotsSize, messageLen)
case _: PbUnregisterShuffle =>
incCounter(RPCUnregisterShuffleNum)
- case _: GetBlacklist =>
- incCounter(RPCGetBlacklistNum)
case _: ReportWorkerUnavailable =>
incCounter(RPCReportWorkerUnavailableNum)
incCounter(RPCReportWorkerUnavailableSize, messageLen)
@@ -123,7 +120,6 @@ object RPCSource {
val RPCReleaseSlotsNum = "RPCReleaseSlotsNum"
val RPCReleaseSlotsSize = "RPCReleaseSlotsSize"
val RPCUnregisterShuffleNum = "RPCUnregisterShuffleNum"
- val RPCGetBlacklistNum = "RPCGetBlacklistNum"
val RPCReportWorkerUnavailableNum = "RPCReportWorkerUnavailableNum"
val RPCReportWorkerUnavailableSize = "RPCReportWorkerUnavailableSize"
val RPCCheckQuotaNum = "RPCCheckQuotaNum"
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 c6176ce5c..34390ac2b 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
@@ -317,15 +317,21 @@ object ControlMessages extends Logging {
appId: String,
totalWritten: Long,
fileCount: Long,
+ localBlacklist: util.List[WorkerInfo],
override var requestId: String = ZERO_UUID) extends MasterRequestMessage
+ case class HeartbeatFromApplicationResponse(
+ statusCode: StatusCode,
+ blacklist: util.List[WorkerInfo],
+ unknownWorkers: util.List[WorkerInfo],
+ shuttingWorkers: util.List[WorkerInfo]) extends Message
+
case class GetBlacklist(localBlacklist: util.List[WorkerInfo]) extends MasterMessage
case class GetBlacklistResponse(
statusCode: StatusCode,
blacklist: util.List[WorkerInfo],
- unknownWorkers: util.List[WorkerInfo],
- shutdownWorkers: util.List[WorkerInfo]) extends Message
+ unknownWorkers: util.List[WorkerInfo]) extends Message
case class CheckQuota(userIdentifier: UserIdentifier) extends Message
@@ -603,15 +609,29 @@ object ControlMessages extends Logging {
.setStatus(status.getValue).build().toByteArray
new TransportMessage(MessageType.APPLICATION_LOST_RESPONSE, payload)
- case HeartbeatFromApplication(appId, totalWritten, fileCount, requestId) =>
+ case HeartbeatFromApplication(appId, totalWritten, fileCount, localBlacklist, requestId) =>
val payload = PbHeartbeatFromApplication.newBuilder()
.setAppId(appId)
.setRequestId(requestId)
.setTotalWritten(totalWritten)
.setFileCount(fileCount)
+ .addAllLocalBlackList(localBlacklist.asScala.map(
+ PbSerDeUtils.toPbWorkerInfo(_, true)).toList.asJava)
.build().toByteArray
new TransportMessage(MessageType.HEARTBEAT_FROM_APPLICATION, payload)
+ case HeartbeatFromApplicationResponse(statusCode, blacklist, unknownWorkers, shuttingWorkers) =>
+ val payload = PbHeartbeatFromApplicationResponse.newBuilder()
+ .setStatus(statusCode.getValue)
+ .addAllBlacklist(
+ blacklist.asScala.map(PbSerDeUtils.toPbWorkerInfo(_, true)).toList.asJava)
+ .addAllUnknownWorkers(
+ unknownWorkers.asScala.map(PbSerDeUtils.toPbWorkerInfo(_, true)).toList.asJava)
+ .addAllShuttingWorkers(
+ shuttingWorkers.asScala.map(PbSerDeUtils.toPbWorkerInfo(_, true)).toList.asJava)
+ .build().toByteArray
+ new TransportMessage(MessageType.HEARTBEAT_FROM_APPLICATION_RESPONSE, payload)
+
case GetBlacklist(localBlacklist) =>
val payload = PbGetBlacklist.newBuilder()
.addAllLocalBlackList(localBlacklist.asScala.map { workerInfo =>
@@ -621,15 +641,13 @@ object ControlMessages extends Logging {
.build().toByteArray
new TransportMessage(MessageType.GET_BLACKLIST, payload)
- case GetBlacklistResponse(statusCode, blacklist, unknownWorkers, shutdownWorkers) =>
+ case GetBlacklistResponse(statusCode, blacklist, unknownWorkers) =>
val builder = PbGetBlacklistResponse.newBuilder()
.setStatus(statusCode.getValue)
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)
@@ -942,8 +960,22 @@ object ControlMessages extends Logging {
pbHeartbeatFromApplication.getAppId,
pbHeartbeatFromApplication.getTotalWritten,
pbHeartbeatFromApplication.getFileCount,
+ new util.ArrayList[WorkerInfo](pbHeartbeatFromApplication.getLocalBlackListList.asScala
+ .map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava),
pbHeartbeatFromApplication.getRequestId)
+ case HEARTBEAT_FROM_APPLICATION_RESPONSE =>
+ val pbHeartbeatFromApplicationResponse =
+ PbHeartbeatFromApplicationResponse.parseFrom(message.getPayload)
+ HeartbeatFromApplicationResponse(
+ Utils.toStatusCode(pbHeartbeatFromApplicationResponse.getStatus),
+ pbHeartbeatFromApplicationResponse.getBlacklistList.asScala
+ .map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava,
+ pbHeartbeatFromApplicationResponse.getUnknownWorkersList.asScala
+ .map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava,
+ pbHeartbeatFromApplicationResponse.getShuttingWorkersList.asScala
+ .map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava)
+
case GET_BLACKLIST =>
val pbGetBlacklist = PbGetBlacklist.parseFrom(message.getPayload)
GetBlacklist(new util.ArrayList[WorkerInfo](pbGetBlacklist.getLocalBlackListList.asScala
@@ -956,8 +988,6 @@ 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/docs/configuration/client.md b/docs/configuration/client.md
index 064eda1c7..fc9ceac4c 100644
--- a/docs/configuration/client.md
+++ b/docs/configuration/client.md
@@ -84,6 +84,5 @@ license: |
| celeborn.test.fetchFailure | false | Whether to test fetch chunk failure | 0.2.0 |
| celeborn.test.retryCommitFiles | false | Fail commitFile request for test | 0.2.0 |
| celeborn.test.retryRevive | false | Fail push data and request for test | 0.2.0 |
-| celeborn.worker.excluded.checkInterval | 30s | Interval for client to refresh excluded worker list. | 0.2.0 |
| celeborn.worker.excluded.expireTimeout | 600s | Timeout time for LifecycleManager to clear reserved excluded worker. | 0.2.0 |
<!--end-include-->
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 e2ab23144..3cf4df280 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
@@ -223,11 +223,17 @@ private[celeborn] class Master(
}
override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
- case HeartbeatFromApplication(appId, totalWritten, fileCount, requestId) =>
+ case HeartbeatFromApplication(appId, totalWritten, fileCount, localBlacklist, requestId) =>
logDebug(s"Received heartbeat from app $appId")
executeWithLeaderChecker(
context,
- handleHeartbeatFromApplication(context, appId, totalWritten, fileCount, requestId))
+ handleHeartbeatFromApplication(
+ context,
+ appId,
+ totalWritten,
+ fileCount,
+ localBlacklist,
+ requestId))
case pbRegisterWorker: PbRegisterWorker =>
val requestId = pbRegisterWorker.getRequestId
@@ -598,8 +604,7 @@ private[celeborn] class Master(
GetBlacklistResponse(
StatusCode.SUCCESS,
new util.ArrayList(statusSystem.blacklist),
- msg.localBlacklist,
- shutdownWorkerSnapshot))
+ msg.localBlacklist))
}
private def handleGetWorkerInfos(context: RpcCallContext): Unit = {
@@ -631,6 +636,7 @@ private[celeborn] class Master(
appId: String,
totalWritten: Long,
fileCount: Long,
+ localBlacklist: util.List[WorkerInfo],
requestId: String): Unit = {
statusSystem.handleAppHeartbeat(
appId,
@@ -638,7 +644,11 @@ private[celeborn] class Master(
fileCount,
System.currentTimeMillis(),
requestId)
- context.reply(OneWayMessageResponse)
+ context.reply(HeartbeatFromApplicationResponse(
+ StatusCode.SUCCESS,
+ new util.ArrayList(statusSystem.blacklist),
+ localBlacklist,
+ shutdownWorkerSnapshot))
}
private def computeUserResourceConsumption(userIdentifier: UserIdentifier)