Ngone51 commented on a change in pull request #31102:
URL: https://github.com/apache/spark/pull/31102#discussion_r555829938
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -66,86 +66,90 @@ private[storage] class BlockManagerDecommissioner(
* the chance of migrating all shuffle blocks before the executor is forced
to exit.
*/
private class ShuffleMigrationRunnable(peer: BlockManagerId) extends
Runnable {
- @volatile var running = true
+ @volatile var keepRunning = true
+
+ def allowRetry(shuffleBlock: ShuffleBlockInfo, failureNum: Int): Boolean =
{
+ if (failureNum < maxReplicationFailuresForDecommission) {
+ logInfo(s"Add $shuffleBlock back to migration queue for " +
+ s"retry ($failureNum / $maxReplicationFailuresForDecommission)")
+ // The block needs to retry so we should not mark it as finished
+ shufflesToMigrate.add((shuffleBlock, failureNum))
+ } else {
+ logWarning(s"Give up migrating $shuffleBlock since it's been " +
+ s"failed for $maxReplicationFailuresForDecommission times")
+ false
+ }
+ }
+
override def run(): Unit = {
- var migrating: Option[(ShuffleBlockInfo, Int)] = None
- logInfo(s"Starting migration thread for ${peer}")
+ logInfo(s"Starting shuffle block migration thread for $peer")
// Once a block fails to transfer to an executor stop trying to transfer
more blocks
- try {
- while (running && !Thread.interrupted()) {
- migrating = Option(shufflesToMigrate.poll())
- migrating match {
- case None =>
- logDebug("Nothing to migrate")
- // Nothing to do right now, but maybe a transfer will fail or a
new block
- // will finish being committed.
- val SLEEP_TIME_SECS = 1
- Thread.sleep(SLEEP_TIME_SECS * 1000L)
- case Some((shuffleBlockInfo, retryCount)) =>
- if (retryCount < maxReplicationFailuresForDecommission) {
- val blocks =
bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
- if (blocks.isEmpty) {
- logInfo(s"Ignore empty shuffle block $shuffleBlockInfo")
+ while (keepRunning) {
Review comment:
When we call `stopMigratingShuffleBlocks` explicitly, it always sets to
`keepRunning` to `false` first before shutting down the thread pool(which would
interrupt the threads). That means, `keepRunning=false` always comes first
before `Thread.isInterrupted = true`. So checking `keepRunning` only should be
enough.
For `InterruptedException` comes from the inner block of `while`, it should
be caught by `catch` block and set to `keepRunning` to `false`. And checking
`keepRunning` only also works in this case. Although, IIUC, I think we
currently don't have a way to interrupt the thread from the inner block.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
