Skip to content

[SPARK-34054][CORE] BlockManagerDecommissioner code cleanup #31102

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Ngone51 wants to merge 13 commits into from
Closed

[SPARK-34054][CORE] BlockManagerDecommissioner code cleanup #31102

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
5759dc7
init
Ngone51 Jan 8, 2021
22ee11c
fix test
Ngone51 Jan 9, 2021
63ae60d
use option shuffleMigrationPool too
Ngone51 Jan 15, 2021
a93ffa2
remove redundant lines
Ngone51 Jan 15, 2021
e8efc25
revert to use ConcurrentLinkedQueue
Ngone51 Jan 15, 2021
4cdca0d
rename offload in BlockManagerSuite
Ngone51 Jan 20, 2021
4567842
rename to nextShuffleBlockToMigrate
Ngone51 Jan 20, 2021
f5647f3
sub-block
Ngone51 Jan 20, 2021
416baa7
improve the condition of delete file
Ngone51 Jan 20, 2021
16c643b
fix BlockManagerSuite
Ngone51 Jan 20, 2021
66e56b7
improve handling InterruptedException
Ngone51 Jan 20, 2021
4c8b28d
add comment
Ngone51 Mar 8, 2021
88a205e
fix test failure
Ngone51 Jun 9, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 10 additions & 3 deletions core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -296,8 +296,15 @@ private[spark] class CoarseGrainedExecutorBackend(
logInfo(msg)
try {
decommissioned = true
if (env.conf.get(STORAGE_DECOMMISSION_ENABLED)) {
val migrationEnabled = env.conf.get(STORAGE_DECOMMISSION_ENABLED) &&
(env.conf.get(STORAGE_DECOMMISSION_RDD_BLOCKS_ENABLED) ||
env.conf.get(STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED))
Comment on lines +299 to +301
Copy link
Contributor

@attilapiros attilapiros Jan 15, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Based on the above lines does STORAGE_DECOMMISSION_ENABLED has any role now?
So what about removing this config?

Copy link
Member Author

@Ngone51 Ngone51 Jan 20, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That‘s my concern too. But removal involves the API change. I intentionally try to avoid that in this refactor PR. But I still think it's worth a try in a separate PR, especially since the 3.1 hasn't released yet. cc @holdenk what's your opinion?

Copy link
Member Author

@Ngone51 Ngone51 Jan 26, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

kindly ping @holdenk

if (migrationEnabled) {
env.blockManager.decommissionBlockManager()
} else if (env.conf.get(STORAGE_DECOMMISSION_ENABLED)) {
Copy link
Contributor

@attilapiros attilapiros Jan 15, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

By removing the STORAGE_DECOMMISSION_ENABLED config this part would simply disappear.

logError(s"Storage decommissioning attempted but neither " +
s"${STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED.key} or " +
s"${STORAGE_DECOMMISSION_RDD_BLOCKS_ENABLED.key} is enabled ")
}
if (executor != null) {
executor.decommission()
Expand All @@ -324,7 +331,7 @@ private[spark] class CoarseGrainedExecutorBackend(
while (true) {
logInfo("Checking to see if we can shutdown.")
if (executor == null || executor.numRunningTasks == 0) {
if (env.conf.get(STORAGE_DECOMMISSION_ENABLED)) {
if (migrationEnabled) {
logInfo("No running tasks, checking migrations")
val (migrationTime, allBlocksMigrated) = env.blockManager.lastMigrationInfo()
// We can only trust allBlocksMigrated boolean value if there were no tasks running
Expand All @@ -340,7 +347,7 @@ private[spark] class CoarseGrainedExecutorBackend(
exitExecutor(0, ExecutorLossMessage.decommissionFinished, notifyDriver = true)
}
} else {
logInfo("Blocked from shutdown by running ${executor.numRunningtasks} tasks")
logInfo(s"Blocked from shutdown by ${executor.numRunningTasks} running tasks")
// If there is a running task it could store blocks, so make sure we wait for a
// migration loop to complete after the last task is done.
// Note: this is only advanced if there is a running task, if there
Expand Down
4 changes: 3 additions & 1 deletion core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockInfo.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,4 +25,6 @@ import org.apache.spark.annotation.Experimental
* type safe way.
*/
@Experimental
case class ShuffleBlockInfo(shuffleId: Int, mapId: Long)
case class ShuffleBlockInfo(shuffleId: Int, mapId: Long) {
override def toString: String = s"migrate_shuffle_${shuffleId}_$mapId"
}
Loading