You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "waitinfuture (via GitHub)" <gi...@apache.org> on 2023/12/13 12:37:04 UTC
[PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
waitinfuture opened a new pull request, #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
As title
### Why are the changes needed?
One user reported that LifecycleManager's parmap can create huge number of threads and causes OOM.
There are four places where parmap is called:
1. When LifecycleManager commits files
2. When LifecycleManager reserves slots
3. When LifecycleManager setup connection to workers
4. When LifecycleManager call destroy slots
This PR fixes the fourth one. To be more detail, this PR eliminates `parmap` when destroying slots, and also replaces `askSync` with `ask`.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Manual test and GA.
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#issuecomment-1853875582
## [Codecov](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2156?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: `1 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`ea0fff0`)](https://app.codecov.io/gh/apache/incubator-celeborn/commit/ea0fff057f11412620c043f6353ecd52fc8e7ca3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 46.41% compared to head [(`b420e2a`)](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2156?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 46.44%.
| [Files](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2156?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...cala/org/apache/celeborn/common/CelebornConf.scala](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2156?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vY29tbW9uL0NlbGVib3JuQ29uZi5zY2FsYQ==) | 85.72% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2156?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #2156 +/- ##
==========================================
+ Coverage 46.41% 46.44% +0.04%
==========================================
Files 175 175
Lines 11377 11384 +7
Branches 1019 1019
==========================================
+ Hits 5279 5286 +7
- Misses 5745 5746 +1
+ Partials 353 352 -1
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2156?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "cxzl25 (via GitHub)" <gi...@apache.org>.
cxzl25 commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1426237748
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1306,72 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestory(status: DestroyFutureWithStatus): Unit = {
+ status.retryTimes += 1
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
+ case scala.util.Success(res) =>
+ if (res.status != StatusCode.SUCCESS && retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message return ${res.status} for $shuffleKey $retryTimes/$rpcMaxRetires, " +
+ s"will retry.")
Review Comment:
nit remove `s`
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1306,72 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestory(status: DestroyFutureWithStatus): Unit = {
Review Comment:
typo `retryDestroy`
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1306,72 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestory(status: DestroyFutureWithStatus): Unit = {
+ status.retryTimes += 1
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
+ case scala.util.Success(res) =>
+ if (res.status != StatusCode.SUCCESS && retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message return ${res.status} for $shuffleKey $retryTimes/$rpcMaxRetires, " +
+ s"will retry.")
+ retryDestory(futureWithStatus)
+ } else {
+ iter.remove()
+ }
+ case scala.util.Failure(e) =>
+ if (retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message failed $retryTimes/$rpcMaxRetires for $shuffleKey, reason: $e, " +
+ s"will retry.")
Review Comment:
ditto
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "cfmcgrady (via GitHub)" <gi...@apache.org>.
cfmcgrady commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427641188
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds, mockDestroyFailure)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
Review Comment:
```scala
futureWithStatus.future.value.foreach {
case scala.util.Success(res) =>
...
case scala.util.Failure(e) =>
...
case _ =>
...
}
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1426798479
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1306,72 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestory(status: DestroyFutureWithStatus): Unit = {
Review Comment:
done
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1306,72 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestory(status: DestroyFutureWithStatus): Unit = {
+ status.retryTimes += 1
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
+ case scala.util.Success(res) =>
+ if (res.status != StatusCode.SUCCESS && retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message return ${res.status} for $shuffleKey $retryTimes/$rpcMaxRetires, " +
+ s"will retry.")
Review Comment:
done
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1306,72 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestory(status: DestroyFutureWithStatus): Unit = {
+ status.retryTimes += 1
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
+ case scala.util.Success(res) =>
+ if (res.status != StatusCode.SUCCESS && retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message return ${res.status} for $shuffleKey $retryTimes/$rpcMaxRetires, " +
+ s"will retry.")
+ retryDestory(futureWithStatus)
+ } else {
+ iter.remove()
+ }
+ case scala.util.Failure(e) =>
+ if (retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message failed $retryTimes/$rpcMaxRetires for $shuffleKey, reason: $e, " +
+ s"will retry.")
Review Comment:
done
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#issuecomment-1857631154
Thanks, merging to main(v0.4.0)/branch-0.3(v0.3.2)
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "cfmcgrady (via GitHub)" <gi...@apache.org>.
cfmcgrady commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427638843
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds, mockDestroyFailure)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
Review Comment:
If the future was not completed the returned value will be `None`
https://github.com/scala/scala/blob/v2.12.18/src/library/scala/concurrent/Future.scala#L182-L194
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427464615
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -3712,6 +3713,14 @@ object CelebornConf extends Logging {
.intConf
.createWithDefault(16)
+ val CLIENT_RPC_MAX_RETIRES: ConfigEntry[Int] =
+ buildConf("celeborn.client.rpc.maxRetries")
+ .categories("client")
+ .version("0.3.2")
+ .doc("Max RPC retry times in LifecycleManager.")
+ .intConf
+ .createWithDefault(2)
Review Comment:
```suggestion
.createWithDefault(3)
```
The default value of most `maxRetries` in `CelebornConf` is 3.
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427803548
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds, mockDestroyFailure)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
+ case scala.util.Success(res) =>
+ if (res.status != StatusCode.SUCCESS && retryTimes < rpcMaxRetires) {
+ logError(
+ s"Request $message to ${futureWithStatus.endpoint} return ${res.status} for $shuffleKey $retryTimes/$rpcMaxRetires, " +
+ "will retry.")
+ retryDestroy(futureWithStatus, currentTime)
+ } else {
+ iter.remove()
+ }
+ case scala.util.Failure(e) =>
+ if (retryTimes < rpcMaxRetires) {
+ logError(
+ s"Request $message to ${futureWithStatus.endpoint} failed $retryTimes/$rpcMaxRetires for $shuffleKey, reason: $e, " +
+ "will retry.")
+ retryDestroy(futureWithStatus, currentTime)
+ } else {
Review Comment:
done
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
Review Comment:
removed
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "cfmcgrady (via GitHub)" <gi...@apache.org>.
cfmcgrady commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427624805
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
Review Comment:
after this change, the conf `celeborn.client.rpc.maxParallelism` will no longer be used anywhere. it safe to remove?
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#issuecomment-1857189911
Will add UT after https://github.com/apache/incubator-celeborn/pull/2159 is merged
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427803737
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
Review Comment:
done
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "cfmcgrady (via GitHub)" <gi...@apache.org>.
cfmcgrady commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427641188
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds, mockDestroyFailure)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
Review Comment:
```scala
futureWithStatus.future.value.foreach {
case scala.util.Success(res) =>
...
case scala.util.Failure(e) =>
...
case _ =>
...
}
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds, mockDestroyFailure)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
Review Comment:
If the future was not completed the returned value will be `None`
https://github.com/scala/scala/blob/v2.12.18/src/library/scala/concurrent/Future.scala#L182-L194
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "cfmcgrady (via GitHub)" <gi...@apache.org>.
cfmcgrady commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427621082
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
Review Comment:
```suggestion
slotsToDestroy.asScala.foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
```
https://github.com/databricks/scala-style-guide?tab=readme-ov-file#infix-methods
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#issuecomment-1857464292
@cfmcgrady comments addressed, please take a look again
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#issuecomment-1853841695
cc @pan3793 @cfmcgrady
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "cfmcgrady (via GitHub)" <gi...@apache.org>.
cfmcgrady commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1427647425
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1315,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds, mockDestroyFailure)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
+ case scala.util.Success(res) =>
+ if (res.status != StatusCode.SUCCESS && retryTimes < rpcMaxRetires) {
+ logError(
+ s"Request $message to ${futureWithStatus.endpoint} return ${res.status} for $shuffleKey $retryTimes/$rpcMaxRetires, " +
+ "will retry.")
+ retryDestroy(futureWithStatus, currentTime)
+ } else {
+ iter.remove()
+ }
+ case scala.util.Failure(e) =>
+ if (retryTimes < rpcMaxRetires) {
+ logError(
+ s"Request $message to ${futureWithStatus.endpoint} failed $retryTimes/$rpcMaxRetires for $shuffleKey, reason: $e, " +
+ "will retry.")
+ retryDestroy(futureWithStatus, currentTime)
+ } else {
Review Comment:
Add a log message when the retry count exceeds the maximum retry attempts.
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "cxzl25 (via GitHub)" <gi...@apache.org>.
cxzl25 commented on code in PR #2156:
URL: https://github.com/apache/incubator-celeborn/pull/2156#discussion_r1426956184
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1298,24 +1306,73 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
shuffleId: Int,
slotsToDestroy: WorkerResource): Unit = {
val shuffleKey = Utils.makeShuffleKey(appUniqueId, shuffleId)
- val parallelism = Math.min(Math.max(1, slotsToDestroy.size()), conf.clientRpcMaxParallelism)
- ThreadUtils.parmap(
- slotsToDestroy.asScala,
- "DestroySlot",
- parallelism) { case (workerInfo, (primaryLocations, replicaLocations)) =>
- val destroy = DestroyWorkerSlots(
- shuffleKey,
- primaryLocations.asScala.map(_.getUniqueId).asJava,
- replicaLocations.asScala.map(_.getUniqueId).asJava)
- var res = requestWorkerDestroySlots(workerInfo.endpoint, destroy)
- if (res.status != StatusCode.SUCCESS) {
- logDebug(s"Request $destroy return ${res.status} for $shuffleKey, " +
- s"will retry request destroy.")
- res = requestWorkerDestroySlots(
- workerInfo.endpoint,
- DestroyWorkerSlots(shuffleKey, res.failedPrimarys, res.failedReplicas))
+
+ def retryDestroy(status: DestroyFutureWithStatus, currentTime: Long): Unit = {
+ status.retryTimes += 1
+ status.startTime = currentTime
+ status.future =
+ status.endpoint.ask[DestroyWorkerSlotsResponse](status.message)
+ }
+
+ val startTime = System.currentTimeMillis()
+ val futures = new util.LinkedList[DestroyFutureWithStatus]()
+ slotsToDestroy.asScala foreach { case (workerInfo, (primaryLocations, replicaLocations)) =>
+ val primaryIds = primaryLocations.asScala.map(_.getUniqueId).asJava
+ val replicaIds = replicaLocations.asScala.map(_.getUniqueId).asJava
+ val destroy = DestroyWorkerSlots(shuffleKey, primaryIds, replicaIds)
+ val future = workerInfo.endpoint.ask[DestroyWorkerSlotsResponse](destroy)
+ futures.add(DestroyFutureWithStatus(future, destroy, workerInfo.endpoint, 1, startTime))
+ }
+
+ val timeout = conf.rpcAskTimeout.duration.toMillis
+ var remainingTime = timeout * rpcMaxRetires
+ val delta = 50
+ while (remainingTime > 0 && !futures.isEmpty) {
+ val currentTime = System.currentTimeMillis()
+ val iter = futures.iterator()
+ while (iter.hasNext) {
+ val futureWithStatus = iter.next()
+ val message = futureWithStatus.message
+ val retryTimes = futureWithStatus.retryTimes
+ if (futureWithStatus.future.isCompleted) {
+ futureWithStatus.future.value.get match {
+ case scala.util.Success(res) =>
+ if (res.status != StatusCode.SUCCESS && retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message return ${res.status} for $shuffleKey $retryTimes/$rpcMaxRetires, " +
+ "will retry.")
+ retryDestroy(futureWithStatus, currentTime)
+ } else {
+ iter.remove()
+ }
+ case scala.util.Failure(e) =>
+ if (retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message failed $retryTimes/$rpcMaxRetires for $shuffleKey, reason: $e, " +
+ "will retry.")
+ retryDestroy(futureWithStatus, currentTime)
+ } else {
+ iter.remove()
+ }
+ }
+ } else if (currentTime - futureWithStatus.startTime > timeout) {
+ if (retryTimes < rpcMaxRetires) {
+ logDebug(
+ s"Request $message failed $retryTimes/$rpcMaxRetires for $shuffleKey, reason: Timeout, " +
+ s"will retry.")
Review Comment:
```suggestion
"will retry.")
```
--
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
Re: [PR] [CELEBORN-1167] Avoid calling parmap when destroy slots [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture closed pull request #2156: [CELEBORN-1167] Avoid calling parmap when destroy slots
URL: https://github.com/apache/incubator-celeborn/pull/2156
--
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