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/04/15 14:36:53 UTC
[incubator-celeborn] branch main updated: [CELEBORN-523][REFACTOR] Remove unnecessary code in WorkerPartitionLocationInfo (#1427)
This is an automated email from the ASF dual-hosted git repository.
zhouky 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 932ccd084 [CELEBORN-523][REFACTOR] Remove unnecessary code in WorkerPartitionLocationInfo (#1427)
932ccd084 is described below
commit 932ccd0841004eb56698b32e5df94809fa46246e
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Sat Apr 15 22:36:48 2023 +0800
[CELEBORN-523][REFACTOR] Remove unnecessary code in WorkerPartitionLocationInfo (#1427)
---
.../common/meta/WorkerPartitionLocationInfo.scala | 65 ----------------------
.../celeborn/service/deploy/worker/Worker.scala | 2 -
2 files changed, 67 deletions(-)
diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerPartitionLocationInfo.scala b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerPartitionLocationInfo.scala
index 11aef2181..623592079 100644
--- a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerPartitionLocationInfo.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerPartitionLocationInfo.scala
@@ -71,54 +71,6 @@ class WorkerPartitionLocationInfo extends Logging {
slavePartitionLocations.remove(shuffleKey)
}
- def getAllMasterLocations(shuffleKey: String): util.List[PartitionLocation] = {
- getMasterLocations(shuffleKey)
- }
-
- def getAllSlaveLocations(shuffleKey: String): util.List[PartitionLocation] = {
- getSlaveLocations(shuffleKey)
- }
-
- def getMasterLocations(
- shuffleKey: String,
- partitionIdOpt: Option[Int] = None): util.List[PartitionLocation] = {
- getLocations(shuffleKey, masterPartitionLocations, partitionIdOpt)
- }
-
- def getSlaveLocations(
- shuffleKey: String,
- partitionIdOpt: Option[Int] = None): util.List[PartitionLocation] = {
- getLocations(shuffleKey, slavePartitionLocations, partitionIdOpt)
- }
-
- def getLocations(
- shuffleKey: String,
- partitionInfo: PartitionInfo,
- partitionIdOpt: Option[Int] = None): util.List[PartitionLocation] = {
- val partitionMap = partitionInfo.get(shuffleKey)
- if (partitionMap != null) {
- partitionIdOpt match {
- case Some(partitionId) =>
- partitionMap.values().asScala.filter(_.getId == partitionId)
- .toList.asJava
- case None =>
- new util.ArrayList(partitionMap.values())
- }
- } else {
- new util.ArrayList[PartitionLocation]()
- }
- }
-
- def removeMasterPartitions(shuffleKey: String): (util.Map[String, Integer], Integer) = {
- val uniqueIds = getAllMasterIds(shuffleKey)
- removeMasterPartitions(shuffleKey, uniqueIds)
- }
-
- def removeSlavePartitions(shuffleKey: String): (util.Map[String, Integer], Integer) = {
- val uniqueIds = getAllSlaveIds(shuffleKey)
- removeSlavePartitions(shuffleKey, uniqueIds)
- }
-
def removeMasterPartitions(
shuffleKey: String,
uniqueIds: util.Collection[String]): (util.Map[String, Integer], Integer) = {
@@ -190,23 +142,6 @@ class WorkerPartitionLocationInfo extends Logging {
} else null
}
- private def getAllIds(
- shuffleKey: String,
- partitionInfo: PartitionInfo): util.List[String] = {
- val partitionMap = partitionInfo.get(shuffleKey)
- if (partitionMap != null) {
- partitionMap.values().asScala.map(_.getUniqueId).toList.asJava
- } else null
- }
-
- private def getAllMasterIds(shuffleKey: String): util.List[String] = {
- getAllIds(shuffleKey, masterPartitionLocations)
- }
-
- private def getAllSlaveIds(shuffleKey: String): util.List[String] = {
- getAllIds(shuffleKey, slavePartitionLocations)
- }
-
def isEmpty: Boolean = {
masterPartitionLocations.isEmpty && slavePartitionLocations.isEmpty
}
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
index 032296174..2fcb62f0b 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -427,8 +427,6 @@ private[celeborn] class Worker(
@VisibleForTesting
def cleanup(expiredShuffleKeys: JHashSet[String]): Unit = synchronized {
expiredShuffleKeys.asScala.foreach { shuffleKey =>
- partitionLocationInfo.removeMasterPartitions(shuffleKey)
- partitionLocationInfo.removeSlavePartitions(shuffleKey)
partitionLocationInfo.removeShuffle(shuffleKey)
shufflePartitionType.remove(shuffleKey)
shufflePushDataTimeout.remove(shuffleKey)