AmandeepSingh285 commented on code in PR #3775:
URL: https://github.com/apache/celeborn/pull/3775#discussion_r3967665002
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1880,6 +1881,139 @@ class LifecycleManager(val appUniqueId: String, val
conf: CelebornConf) extends
}
}
+ private def syncEndpointReadyWorkers(
+ shuffleId: Int,
+ workersFromMaster: Set[WorkerInfo]): Unit = {
+ val currentEndpointReadyWorkers = workerStatusTracker.endpointReadyWorkers
+ val workersToRemove = currentEndpointReadyWorkers.diff(workersFromMaster)
+ val workersToConnect = workersFromMaster.diff(currentEndpointReadyWorkers)
+ val connectFailedWorkers = new ShuffleFailedWorkers()
+ setupEndpoints(workersToConnect.asJava, shuffleId, connectFailedWorkers)
+ workerStatusTracker.recordWorkerFailure(connectFailedWorkers)
+
+ val connectedWorkers =
workersToConnect.diff(connectFailedWorkers.asScala.keySet)
+ workerStatusTracker.addEndpointReadyWorkers(connectedWorkers)
+ workerStatusTracker.removeEndpointReadyWorkers(workersToRemove)
+ }
+
+ private[client] def refreshEndpointReadyWorkersFromMaster(shuffleId: Int):
Unit = {
+ val shouldRefresh = endpointReadyWorkersRefreshLock.synchronized {
+ var waitedForRefresh = false
+ val waitDeadline = System.currentTimeMillis() + rpcAskTimeoutMs
+ var remainingWaitTime = rpcAskTimeoutMs
+ while (endpointReadyWorkersRefreshInProgress && remainingWaitTime > 0) {
+ // Reuse the in-flight result instead of letting this revive observe
an incomplete pool.
+ waitedForRefresh = true
+ try {
+ endpointReadyWorkersRefreshLock.wait(remainingWaitTime)
+ } catch {
+ case _: InterruptedException =>
+ Thread.currentThread().interrupt()
+ return
+ }
+ remainingWaitTime = waitDeadline - System.currentTimeMillis()
+ }
+ if (endpointReadyWorkersRefreshInProgress) {
+ logWarning(
+ s"Timed out after ${rpcAskTimeoutMs}ms waiting for the in-flight
endpoint-ready " +
+ "workers refresh; continue using the current worker pool.")
+ }
+
+ val currentTime = System.currentTimeMillis()
+ val refreshIntervalElapsed = lastEndpointReadyWorkersRefreshAttemptTime
== 0L ||
+ currentTime - lastEndpointReadyWorkersRefreshAttemptTime >=
dynamicResourceUpdateTime
+ if (!waitedForRefresh && refreshIntervalElapsed) {
+ endpointReadyWorkersRefreshInProgress = true
+ true
+ } else {
+ false
+ }
+ }
+ if (!shouldRefresh) {
+ return
+ }
+
+ try {
+ val requestWorkersRes = requestMasterRequestWorkersWithRetry()
Review Comment:
Could this significantly increase the request load on the Master?
Previously, the Master was only queried for new workers during the revive
flow when the unavailable-worker ratio was high. With this change, every
shuffle irrespective of the ratio would request more workers from the master,
potentially resulting in a large increase in Master request call traffic at
scale. If many applications make these requests around the same time, this
could also create periodic bursts and potentially impact Master performance?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
