You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by GitBox <gi...@apache.org> on 2022/12/15 08:53:10 UTC
[GitHub] [incubator-celeborn] AngersZhuuuu commented on a diff in pull request #1089: [CELEBORN-121] Refactor batchHandleCommitPartition
AngersZhuuuu commented on code in PR #1089:
URL: https://github.com/apache/incubator-celeborn/pull/1089#discussion_r1049356943
##########
client/src/main/scala/org/apache/celeborn/client/commit/CommitHandler.scala:
##########
@@ -67,9 +68,72 @@ abstract class CommitHandler(
def isPartitionInProcess(shuffleId: Int, partitionId: Int): Boolean = false
+ def batchUnCommitRequests(shuffleId: Int, shuffleCommittedInfo: ShuffleCommittedInfo)
+ : Map[WorkerInfo, collection.Set[PartitionLocation]] = {
+ // When running to here, if handleStageEnd got lock first and commitFiles,
+ // then this batch get this lock, commitPartitionRequests may contains
+ // partitions which are already committed by stageEnd process.
+ // But inProcessStageEndShuffleSet should have contain this shuffle id,
+ // can directly return empty.
+ if (this.isStageEndOrInProcess(shuffleId)) {
+ logWarning(s"Shuffle $shuffleId ended or during processing stage end.")
+ shuffleCommittedInfo.commitPartitionRequests.clear()
+ Map.empty[WorkerInfo, Set[PartitionLocation]]
+ } else {
+ val currentBatch = this.getUnCommitPartitionRequests(shuffleId, shuffleCommittedInfo)
+ shuffleCommittedInfo.commitPartitionRequests.clear()
+ currentBatch.foreach { partitionLocation =>
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .add(partitionLocation)
+ if (partitionLocation.getPeer != null) {
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .add(partitionLocation.getPeer)
+ }
+ }
+
+ if (currentBatch.nonEmpty) {
+ logWarning(s"Commit current batch HARD_SPLIT partitions for $shuffleId: " +
+ s"${currentBatch.map(_.getUniqueId).mkString("[", ",", "]")}")
+ val workerToRequests = currentBatch.flatMap { partitionLocation =>
+ if (partitionLocation.getPeer != null) {
+ Seq(partitionLocation, partitionLocation.getPeer)
+ } else {
+ Seq(partitionLocation)
+ }
+ }.groupBy(_.getWorker)
+ workerToRequests
+ } else {
+ Map.empty[WorkerInfo, Set[PartitionLocation]]
+ }
+ }
+ }
+
+ protected def getUnCommitPartitionRequests(
+ shuffleId: Int,
+ shuffleCommittedInfo: ShuffleCommittedInfo): mutable.Set[PartitionLocation]
+
+ def incrementInFlightNum(
+ shuffleCommittedInfo: ShuffleCommittedInfo,
+ workerToRequests: Map[
+ WorkerInfo,
+ collection.Set[PartitionLocation]]): Unit = {
Review Comment:
```
def incrementInFlightNum(
shuffleCommittedInfo: ShuffleCommittedInfo,
workerToRequests: Map[WorkerInfo,collection.Set[PartitionLocation]]): Unit = {
```
?
##########
client/src/main/scala/org/apache/celeborn/client/commit/CommitHandler.scala:
##########
@@ -67,9 +68,72 @@ abstract class CommitHandler(
def isPartitionInProcess(shuffleId: Int, partitionId: Int): Boolean = false
+ def batchUnCommitRequests(shuffleId: Int, shuffleCommittedInfo: ShuffleCommittedInfo)
+ : Map[WorkerInfo, collection.Set[PartitionLocation]] = {
+ // When running to here, if handleStageEnd got lock first and commitFiles,
+ // then this batch get this lock, commitPartitionRequests may contains
+ // partitions which are already committed by stageEnd process.
+ // But inProcessStageEndShuffleSet should have contain this shuffle id,
+ // can directly return empty.
+ if (this.isStageEndOrInProcess(shuffleId)) {
+ logWarning(s"Shuffle $shuffleId ended or during processing stage end.")
+ shuffleCommittedInfo.commitPartitionRequests.clear()
+ Map.empty[WorkerInfo, Set[PartitionLocation]]
+ } else {
+ val currentBatch = this.getUnCommitPartitionRequests(shuffleId, shuffleCommittedInfo)
+ shuffleCommittedInfo.commitPartitionRequests.clear()
+ currentBatch.foreach { partitionLocation =>
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .add(partitionLocation)
+ if (partitionLocation.getPeer != null) {
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .add(partitionLocation.getPeer)
+ }
+ }
+
+ if (currentBatch.nonEmpty) {
+ logWarning(s"Commit current batch HARD_SPLIT partitions for $shuffleId: " +
+ s"${currentBatch.map(_.getUniqueId).mkString("[", ",", "]")}")
+ val workerToRequests = currentBatch.flatMap { partitionLocation =>
+ if (partitionLocation.getPeer != null) {
+ Seq(partitionLocation, partitionLocation.getPeer)
+ } else {
+ Seq(partitionLocation)
+ }
+ }.groupBy(_.getWorker)
+ workerToRequests
+ } else {
+ Map.empty[WorkerInfo, Set[PartitionLocation]]
+ }
+ }
+ }
+
+ protected def getUnCommitPartitionRequests(
+ shuffleId: Int,
+ shuffleCommittedInfo: ShuffleCommittedInfo): mutable.Set[PartitionLocation]
+
+ def incrementInFlightNum(
+ shuffleCommittedInfo: ShuffleCommittedInfo,
+ workerToRequests: Map[
+ WorkerInfo,
+ collection.Set[PartitionLocation]]): Unit = {
+ shuffleCommittedInfo.allInFlightCommitRequestNum.addAndGet(
+ workerToRequests.size)
+ }
+
+ def decrementInFlightNum(
+ shuffleCommittedInfo: ShuffleCommittedInfo,
+ workerToRequests: Map[
+ WorkerInfo,
+ collection.Set[PartitionLocation]]): Unit = {
+ shuffleCommittedInfo.allInFlightCommitRequestNum.addAndGet(
Review Comment:
ditto
##########
client/src/main/scala/org/apache/celeborn/client/CommitManager.scala:
##########
@@ -91,104 +86,20 @@ class CommitManager(appId: String, val conf: CelebornConf, lifecycleManager: Lif
committedPartitionInfo.asScala.foreach { case (shuffleId, shuffleCommittedInfo) =>
batchHandleCommitPartitionExecutors.submit {
new Runnable {
- val partitionType = lifecycleManager.getPartitionType(shuffleId)
val commitHandler = getCommitHandler(shuffleId)
- def incrementInflightNum(workerToRequests: Map[
- WorkerInfo,
- collection.Set[PartitionLocation]]): Unit = {
- if (partitionType == PartitionType.MAP) {
- workerToRequests.foreach {
- case (_, partitions) =>
- partitions.groupBy(_.getId).foreach { case (id, _) =>
- val atomicInteger = shuffleCommittedInfo
- .partitionInFlightCommitRequestNum
- .computeIfAbsent(id, (k: Int) => new AtomicInteger(0))
- atomicInteger.incrementAndGet()
- }
- }
- }
- shuffleCommittedInfo.allInFlightCommitRequestNum.addAndGet(
- workerToRequests.size)
- }
-
- def decrementInflightNum(
- workerToRequests: Map[WorkerInfo, collection.Set[PartitionLocation]])
- : Unit = {
- if (partitionType == PartitionType.MAP) {
- workerToRequests.foreach {
- case (_, partitions) =>
- partitions.groupBy(_.getId).foreach { case (id, _) =>
- shuffleCommittedInfo.partitionInFlightCommitRequestNum.get(
- id).decrementAndGet()
- }
- }
- }
- shuffleCommittedInfo.allInFlightCommitRequestNum.addAndGet(
- -workerToRequests.size)
- }
-
- def getUnCommitPartitionRequests(
- commitPartitionRequests: util.Set[CommitPartitionRequest])
- : scala.collection.mutable.Set[CommitPartitionRequest] = {
- if (partitionType == PartitionType.MAP) {
- commitPartitionRequests.asScala.filterNot { request =>
- shuffleCommittedInfo.handledCommitPartitionRequests
- .contains(request.partition) && commitHandler.isPartitionInProcess(
- shuffleId,
- request.partition.getId)
- }
- } else {
- commitPartitionRequests.asScala.filterNot { request =>
- shuffleCommittedInfo.handledCommitPartitionRequests
- .contains(request.partition)
- }
- }
- }
-
override def run(): Unit = {
- val workerToRequests = shuffleCommittedInfo.synchronized {
- // When running to here, if handleStageEnd got lock first and commitFiles,
- // then this batch get this lock, commitPartitionRequests may contains
- // partitions which are already committed by stageEnd process.
- // But inProcessStageEndShuffleSet should have contain this shuffle id,
- // can directly return.
- if (commitHandler.isStageEndOrInProcess(shuffleId)) {
- logWarning(s"Shuffle $shuffleId ended or during processing stage end.")
- shuffleCommittedInfo.commitPartitionRequests.clear()
- Map.empty[WorkerInfo, Set[PartitionLocation]]
- } else {
- val currentBatch =
- getUnCommitPartitionRequests(shuffleCommittedInfo.commitPartitionRequests)
- shuffleCommittedInfo.commitPartitionRequests.clear()
- currentBatch.foreach { commitPartitionRequest =>
- shuffleCommittedInfo.handledCommitPartitionRequests
- .add(commitPartitionRequest.partition)
- if (commitPartitionRequest.partition.getPeer != null) {
- shuffleCommittedInfo.handledCommitPartitionRequests
- .add(commitPartitionRequest.partition.getPeer)
- }
- }
-
- if (currentBatch.nonEmpty) {
- logWarning(s"Commit current batch HARD_SPLIT partitions for $shuffleId: " +
- s"${currentBatch.map(_.partition.getUniqueId).mkString("[", ",", "]")}")
- val workerToRequests = currentBatch.flatMap { request =>
- if (request.partition.getPeer != null) {
- Seq(request.partition, request.partition.getPeer)
- } else {
- Seq(request.partition)
- }
- }.groupBy(_.getWorker)
- incrementInflightNum(workerToRequests)
- workerToRequests
- } else {
- Map.empty[WorkerInfo, Set[PartitionLocation]]
- }
- }
+ var workerToRequests: Map[WorkerInfo, collection.Set[PartitionLocation]] = null
+ shuffleCommittedInfo.synchronized {
+ workerToRequests =
+ commitHandler.batchUnCommitRequests(shuffleId, shuffleCommittedInfo)
+ // when batch commit thread starts to commit these requests, we should increment inFlightNum,
+ // then stage/partition end would be able to recognize all requests are over.
+ commitHandler.incrementInFlightNum(shuffleCommittedInfo, workerToRequests)
}
+
if (workerToRequests.nonEmpty) {
val commitFilesFailedWorkers = new ShuffleFailedWorkers()
- val parallelism = workerToRequests.size
+ val parallelism = Math.min(workerToRequests.size, conf.rpcMaxParallelism)
Review Comment:
Nice catch!
##########
client/src/main/scala/org/apache/celeborn/client/CommitManager.scala:
##########
@@ -52,7 +47,7 @@ case class ShuffleCommittedInfo(
committedSlaveStorageInfos: ConcurrentHashMap[String, StorageInfo],
committedMapIdBitmap: ConcurrentHashMap[String, RoaringBitmap],
currentShuffleFileCount: LongAdder,
- commitPartitionRequests: util.Set[CommitPartitionRequest],
+ commitPartitionRequests: util.Set[PartitionLocation],
Review Comment:
`needCommitPartitionLocation`?
##########
client/src/main/scala/org/apache/celeborn/client/CommitManager.scala:
##########
@@ -52,7 +47,7 @@ case class ShuffleCommittedInfo(
committedSlaveStorageInfos: ConcurrentHashMap[String, StorageInfo],
committedMapIdBitmap: ConcurrentHashMap[String, RoaringBitmap],
currentShuffleFileCount: LongAdder,
- commitPartitionRequests: util.Set[CommitPartitionRequest],
+ commitPartitionRequests: util.Set[PartitionLocation],
handledCommitPartitionRequests: util.Set[PartitionLocation],
Review Comment:
committedPartitionLocation?
##########
client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala:
##########
@@ -107,6 +109,47 @@ class MapPartitionCommitHandler(
dataCommitSuccess
}
+ override def getUnCommitPartitionRequests(
+ shuffleId: Int,
+ shuffleCommittedInfo: ShuffleCommittedInfo): mutable.Set[PartitionLocation] = {
+ shuffleCommittedInfo.commitPartitionRequests.asScala.filterNot { partitionLocation =>
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .contains(partitionLocation) && this.isPartitionInProcess(
+ shuffleId,
+ partitionLocation.getId)
+ }
+ }
+
+ override def incrementInFlightNum(
+ shuffleCommittedInfo: ShuffleCommittedInfo,
+ workerToRequests: Map[
+ WorkerInfo,
+ collection.Set[PartitionLocation]]): Unit = {
+ workerToRequests.foreach {
+ case (_, partitions) =>
+ partitions.groupBy(_.getId).foreach { case (id, _) =>
+ val atomicInteger = shuffleCommittedInfo
+ .partitionInFlightCommitRequestNum
+ .computeIfAbsent(id, (k: Int) => new AtomicInteger(0))
+ atomicInteger.incrementAndGet()
+ }
+ }
+ }
+
+ override def decrementInFlightNum(
+ shuffleCommittedInfo: ShuffleCommittedInfo,
+ workerToRequests: Map[
+ WorkerInfo,
+ collection.Set[PartitionLocation]]): Unit = {
Review Comment:
ditto
##########
client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala:
##########
@@ -107,6 +109,47 @@ class MapPartitionCommitHandler(
dataCommitSuccess
}
+ override def getUnCommitPartitionRequests(
+ shuffleId: Int,
+ shuffleCommittedInfo: ShuffleCommittedInfo): mutable.Set[PartitionLocation] = {
+ shuffleCommittedInfo.commitPartitionRequests.asScala.filterNot { partitionLocation =>
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .contains(partitionLocation) && this.isPartitionInProcess(
+ shuffleId,
+ partitionLocation.getId)
+ }
+ }
+
+ override def incrementInFlightNum(
+ shuffleCommittedInfo: ShuffleCommittedInfo,
+ workerToRequests: Map[
+ WorkerInfo,
+ collection.Set[PartitionLocation]]): Unit = {
Review Comment:
ditto
##########
client/src/main/scala/org/apache/celeborn/client/commit/CommitHandler.scala:
##########
@@ -67,9 +68,72 @@ abstract class CommitHandler(
def isPartitionInProcess(shuffleId: Int, partitionId: Int): Boolean = false
+ def batchUnCommitRequests(shuffleId: Int, shuffleCommittedInfo: ShuffleCommittedInfo)
+ : Map[WorkerInfo, collection.Set[PartitionLocation]] = {
+ // When running to here, if handleStageEnd got lock first and commitFiles,
+ // then this batch get this lock, commitPartitionRequests may contains
+ // partitions which are already committed by stageEnd process.
+ // But inProcessStageEndShuffleSet should have contain this shuffle id,
+ // can directly return empty.
+ if (this.isStageEndOrInProcess(shuffleId)) {
+ logWarning(s"Shuffle $shuffleId ended or during processing stage end.")
+ shuffleCommittedInfo.commitPartitionRequests.clear()
+ Map.empty[WorkerInfo, Set[PartitionLocation]]
+ } else {
+ val currentBatch = this.getUnCommitPartitionRequests(shuffleId, shuffleCommittedInfo)
+ shuffleCommittedInfo.commitPartitionRequests.clear()
+ currentBatch.foreach { partitionLocation =>
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .add(partitionLocation)
Review Comment:
One line seems ok?
##########
client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala:
##########
@@ -107,6 +109,47 @@ class MapPartitionCommitHandler(
dataCommitSuccess
}
+ override def getUnCommitPartitionRequests(
+ shuffleId: Int,
+ shuffleCommittedInfo: ShuffleCommittedInfo): mutable.Set[PartitionLocation] = {
+ shuffleCommittedInfo.commitPartitionRequests.asScala.filterNot { partitionLocation =>
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .contains(partitionLocation) && this.isPartitionInProcess(
+ shuffleId,
+ partitionLocation.getId)
Review Comment:
How about like this?
```
shuffleCommittedInfo.handledCommitPartitionRequests.contains(partitionLocation) &&
this.isPartitionInProcess(shuffleId, partitionLocation.getId)
```
##########
client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala:
##########
@@ -107,6 +109,47 @@ class MapPartitionCommitHandler(
dataCommitSuccess
}
+ override def getUnCommitPartitionRequests(
+ shuffleId: Int,
+ shuffleCommittedInfo: ShuffleCommittedInfo): mutable.Set[PartitionLocation] = {
+ shuffleCommittedInfo.commitPartitionRequests.asScala.filterNot { partitionLocation =>
+ shuffleCommittedInfo.handledCommitPartitionRequests
+ .contains(partitionLocation) && this.isPartitionInProcess(
+ shuffleId,
+ partitionLocation.getId)
+ }
+ }
+
+ override def incrementInFlightNum(
+ shuffleCommittedInfo: ShuffleCommittedInfo,
+ workerToRequests: Map[
+ WorkerInfo,
+ collection.Set[PartitionLocation]]): Unit = {
+ workerToRequests.foreach {
+ case (_, partitions) =>
+ partitions.groupBy(_.getId).foreach { case (id, _) =>
+ val atomicInteger = shuffleCommittedInfo
+ .partitionInFlightCommitRequestNum
+ .computeIfAbsent(id, (k: Int) => new AtomicInteger(0))
+ atomicInteger.incrementAndGet()
+ }
+ }
+ }
+
+ override def decrementInFlightNum(
+ shuffleCommittedInfo: ShuffleCommittedInfo,
+ workerToRequests: Map[
+ WorkerInfo,
+ collection.Set[PartitionLocation]]): Unit = {
+ workerToRequests.foreach {
+ case (_, partitions) =>
+ partitions.groupBy(_.getId).foreach { case (id, _) =>
+ shuffleCommittedInfo.partitionInFlightCommitRequestNum.get(
+ id).decrementAndGet()
Review Comment:
```
shuffleCommittedInfo.partitionInFlightCommitRequestNum.get(id).decrementAndGet()
```
or
```
shuffleCommittedInfo.partitionInFlightCommitRequestNum
.get(id).decrementAndGet()
```
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org