holdenk commented on a change in pull request #28708:
URL: https://github.com/apache/spark/pull/28708#discussion_r442604350
##########
File path: core/src/main/scala/org/apache/spark/MapOutputTracker.scala
##########
@@ -775,7 +802,12 @@ private[spark] class MapOutputTrackerMaster(
override def stop(): Unit = {
mapOutputRequests.offer(PoisonPill)
threadpool.shutdown()
- sendTracker(StopMapOutputTracker)
+ try {
Review comment:
So this is only because we call shutdown during the tests that this is
needed. It doesn't throw exceptions normally.
##########
File path: core/src/main/scala/org/apache/spark/internal/config/package.scala
##########
@@ -420,6 +420,30 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val STORAGE_SHUFFLE_DECOMMISSION_ENABLED =
+ ConfigBuilder("spark.storage.decommission.shuffleBlocks.enabled")
+ .doc("Whether to transfer shuffle blocks during block manager
decommissioning. Requires " +
+ "a migratable shuffle resolver (like sort based shuffe)")
+ .version("3.1.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ private[spark] val STORAGE_SHUFFLE_DECOMMISSION_MAX_THREADS =
+ ConfigBuilder("spark.storage.decommission.shuffleBlocks.maxThreads")
+ .doc("Maximum number of threads to use in migrating shuffle files.")
+ .version("3.1.0")
+ .intConf
+ .checkValue(_ > 0, "The maximum number of threads should be positive")
+ .createWithDefault(10)
Review comment:
Sure, why not. I'll change this to default to
`SHUFFLE_MAPOUTPUT_DISPATCHER_NUM_THREADS` if it's not configured. Sound good?
##########
File path: core/src/main/scala/org/apache/spark/internal/config/package.scala
##########
@@ -420,6 +420,30 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val STORAGE_SHUFFLE_DECOMMISSION_ENABLED =
Review comment:
Sure sounds good, I'll rename it.
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
##########
@@ -650,6 +658,23 @@ private[spark] class BlockManager(
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[_]): StreamCallbackWithID = {
+
+ if (decommissioner.isDefined) {
+ throw new BlockSavedOnDecommissionedBlockManagerException(blockId)
+ }
+
+ if (blockId.isShuffle || blockId.isInternalShuffle) {
+ logInfo(s"Putting shuffle block ${blockId}")
+ try {
+ return migratableResolver.putShuffleBlockAsStream(blockId,
serializerManager)
+ } catch {
+ case e: ClassCastException => throw new SparkException(
Review comment:
Yeah were logging which kind of block we received. We only really care
about the type but it's easier to just print the blockId its self.
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val 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) =>
+ logInfo(s"Trying to migrate shuffle ${shuffleBlockInfo} to
${peer}")
+ val blocks =
+ bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
+ logInfo(s"Got migration sub-blocks ${blocks}")
+ blocks.foreach { case (blockId, buffer) =>
+ logInfo(s"Migrating sub-block ${blockId}")
+ bm.blockTransferService.uploadBlockSync(
+ peer.host,
+ peer.port,
+ peer.executorId,
+ blockId,
+ buffer,
+ StorageLevel.DISK_ONLY,
+ null)// class tag, we don't need for shuffle
+ logDebug(s"Migrated sub block ${blockId}")
+ }
+ logInfo(s"Migrated ${shuffleBlockInfo} to ${peer}")
+ }
+ }
+ // This catch is intentionally outside of the while running block.
+ // if we encounter errors migrating to an executor we want to stop.
+ } catch {
+ case e: Exception =>
+ migrating match {
+ case Some(shuffleMap) =>
+ logError(s"Error ${e} during migration, adding ${shuffleMap}
back to migration queue")
Review comment:
Sure.
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val 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
Review comment:
Let me think about that, I'm not sure it's worth the complexity increase.
##########
File path:
core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala
##########
@@ -55,6 +58,25 @@ private[spark] class IndexShuffleBlockResolver(
def getDataFile(shuffleId: Int, mapId: Long): File = getDataFile(shuffleId,
mapId, None)
+ /**
+ * Get the shuffle files that are stored locally. Used for block migrations.
+ */
+ override def getStoredShuffles(): Set[ShuffleBlockInfo] = {
+ // Matches ShuffleIndexBlockId name
+ val pattern = "shuffle_(\\d+)_(\\d+)_.+\\.index".r
+ val rootDirs = blockManager.diskBlockManager.localDirs
+ // ExecutorDiskUtil puts things inside one level hashed sub directories
+ val searchDirs = rootDirs.flatMap(_.listFiles()).filter(_.isDirectory())
++ rootDirs
+ val filenames = searchDirs.flatMap(_.list())
+ logDebug(s"Got block files ${filenames.toList}")
+ filenames.flatMap { fname =>
+ pattern.findAllIn(fname).matchData.map {
+ matched => ShuffleBlockInfo(matched.group(1).toInt,
matched.group(2).toLong)
+ }
+ }.toSet
Review comment:
If the list of shuffle blocks doesn't fit in memory we've got some other
problems (namely in the map status tracker etc.)
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockId.scala
##########
@@ -40,6 +40,9 @@ sealed abstract class BlockId {
def isRDD: Boolean = isInstanceOf[RDDBlockId]
def isShuffle: Boolean = isInstanceOf[ShuffleBlockId] ||
isInstanceOf[ShuffleBlockBatchId]
def isBroadcast: Boolean = isInstanceOf[BroadcastBlockId]
+ def isInternalShuffle: Boolean = {
Review comment:
It's internal in that it's internal to the shuffle block implementation.
Changing `isShuffle` might be ok, I'd have to grep through the code base to see
where it's used first though, I'll take a look.
##########
File path: core/src/main/scala/org/apache/spark/internal/config/package.scala
##########
@@ -420,6 +420,30 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val STORAGE_SHUFFLE_DECOMMISSION_ENABLED =
+ ConfigBuilder("spark.storage.decommission.shuffleBlocks.enabled")
+ .doc("Whether to transfer shuffle blocks during block manager
decommissioning. Requires " +
+ "a migratable shuffle resolver (like sort based shuffe)")
+ .version("3.1.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ private[spark] val STORAGE_SHUFFLE_DECOMMISSION_MAX_THREADS =
+ ConfigBuilder("spark.storage.decommission.shuffleBlocks.maxThreads")
+ .doc("Maximum number of threads to use in migrating shuffle files.")
+ .version("3.1.0")
+ .intConf
+ .checkValue(_ > 0, "The maximum number of threads should be positive")
+ .createWithDefault(10)
+
+
+ private[spark] val STORAGE_RDD_DECOMMISSION_ENABLED =
Review comment:
Sounds good, I'll rename.
##########
File path:
core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala
##########
@@ -168,7 +168,8 @@ private[spark] class NettyBlockTransferService(
// Everything else is encoded using our binary protocol.
val metadata =
JavaUtils.bufferToArray(serializer.newInstance().serialize((level, classTag)))
- val asStream = blockData.size() >
conf.get(config.MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM)
+ val asStream = (blockData.size() >
conf.get(config.MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM) ||
+ blockId.isInternalShuffle || blockId.isShuffle)
Review comment:
It's for simplicity on the implementation side since we always write it
to disk anyways.
##########
File path:
core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala
##########
@@ -148,6 +170,82 @@ private[spark] class IndexShuffleBlockResolver(
}
}
+ /**
+ * Write a provided shuffle block as a stream. Used for block migrations.
+ * ShuffleBlockBatchIds must contain the full range represented in the
ShuffleIndexBlock.
+ * Requires the caller to delete any shuffle index blocks where the shuffle
block fails to
+ * put.
+ */
+ override def putShuffleBlockAsStream(blockId: BlockId, serializerManager:
SerializerManager):
+ StreamCallbackWithID = {
+ val file = blockId match {
+ case ShuffleIndexBlockId(shuffleId, mapId, _) =>
+ getIndexFile(shuffleId, mapId)
+ case ShuffleDataBlockId(shuffleId, mapId, _) =>
+ getDataFile(shuffleId, mapId)
+ case _ =>
+ throw new Exception(s"Unexpected shuffle block transfer ${blockId} as
" +
+ s"${blockId.getClass().getSimpleName()}")
+ }
+ val fileTmp = Utils.tempFileWith(file)
+ val channel = Channels.newChannel(
+ serializerManager.wrapStream(blockId,
+ new FileOutputStream(fileTmp)))
+
+ new StreamCallbackWithID {
+
+ override def getID: String = blockId.name
+
+ override def onData(streamId: String, buf: ByteBuffer): Unit = {
+ while (buf.hasRemaining) {
+ channel.write(buf)
+ }
+ }
+
+ override def onComplete(streamId: String): Unit = {
+ logTrace(s"Done receiving shuffle block $blockId, now storing on local
disk.")
+ channel.close()
+ val diskSize = fileTmp.length()
+ this.synchronized {
Review comment:
The corresponding logic of finalizing is also synchronized. I think it's
fairly remote but if (somehow) we've got some other process renaming the same
file (say a speculative task), we could end up with junk data if we don't
synchronize here.
##########
File path: core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala
##########
@@ -126,6 +128,10 @@ private[spark] class CompressedMapStatus(
override def location: BlockManagerId = loc
+ override def updateLocation(bm: BlockManagerId): Unit = {
Review comment:
Sure, that's a better name.
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
##########
@@ -650,6 +658,23 @@ private[spark] class BlockManager(
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[_]): StreamCallbackWithID = {
+
+ if (decommissioner.isDefined) {
Review comment:
sure, it should get inlined anyways.
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
##########
@@ -242,8 +244,7 @@ private[spark] class BlockManager(
private var blockReplicationPolicy: BlockReplicationPolicy = _
- private var blockManagerDecommissioning: Boolean = false
- private var decommissionManager: Option[BlockManagerDecommissionManager] =
None
+ @volatile private var decommissioner: Option[BlockManagerDecommissioner] =
None
Review comment:
If we receive two decommissioning messages at the same time. we don't
want to launch two of them.
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,281 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val migrating = Option(shufflesToMigrate.poll())
+ migrating match {
+ case None =>
+ logInfo("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) =>
+ logInfo(s"Trying to migrate shuffle ${shuffleBlockInfo} to
${peer}")
+ val blocks =
+ bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
+ logInfo(s"Got migration sub-blocks ${blocks}")
+ blocks.foreach { case (blockId, buffer) =>
+ logInfo(s"Migrating sub-block ${blockId}")
+ bm.blockTransferService.uploadBlockSync(
+ peer.host,
+ peer.port,
+ peer.executorId,
+ blockId,
+ buffer,
+ StorageLevel.DISK_ONLY,
+ null)// class tag, we don't need for shuffle
+ logInfo(s"Migrated sub block ${blockId}")
+ }
+ logInfo(s"Migrated ${shuffleBlockInfo} to ${peer}")
+ }
+ }
+ // This catch is intentionally outside of the while running block.
+ // if we encounter errors migrating to an executor we want to stop.
+ } catch {
+ case e: Exception =>
+ migrating match {
+ case Some(shuffleMap) =>
+ logError(s"Error ${e} during migration, adding ${shuffleMap}
back to migration queue")
+ shufflesToMigrate.add(shuffleMap)
+ case None =>
+ logError(s"Error ${e} while waiting for block to migrate")
Review comment:
Sure.
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val 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) =>
+ logInfo(s"Trying to migrate shuffle ${shuffleBlockInfo} to
${peer}")
+ val blocks =
+ bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
+ logInfo(s"Got migration sub-blocks ${blocks}")
+ blocks.foreach { case (blockId, buffer) =>
+ logInfo(s"Migrating sub-block ${blockId}")
+ bm.blockTransferService.uploadBlockSync(
+ peer.host,
+ peer.port,
+ peer.executorId,
+ blockId,
+ buffer,
+ StorageLevel.DISK_ONLY,
+ null)// class tag, we don't need for shuffle
+ logDebug(s"Migrated sub block ${blockId}")
+ }
+ logInfo(s"Migrated ${shuffleBlockInfo} to ${peer}")
+ }
+ }
+ // This catch is intentionally outside of the while running block.
+ // if we encounter errors migrating to an executor we want to stop.
+ } catch {
+ case e: Exception =>
+ migrating match {
+ case Some(shuffleMap) =>
+ logError(s"Error ${e} during migration, adding ${shuffleMap}
back to migration queue")
+ shufflesToMigrate.add(shuffleMap)
+ case None =>
+ logError(s"Error ${e} while waiting for block to migrate")
+ }
+ }
+ }
+ }
+
+ // Shuffles which are either in queue for migrations or migrated
+ private val migratingShuffles = mutable.HashSet[ShuffleBlockInfo]()
+
+ // Shuffles which are queued for migration
+ private[storage] val shufflesToMigrate =
+ new java.util.concurrent.ConcurrentLinkedQueue[ShuffleBlockInfo]()
+
+ @volatile private var stopped = false
+
+ private val migrationPeers =
+ mutable.HashMap[BlockManagerId, ShuffleMigrationRunnable]()
+
+ private lazy val blockMigrationExecutor =
+ ThreadUtils.newDaemonSingleThreadExecutor("block-manager-decommission")
+
+ private val blockMigrationRunnable = new Runnable {
+ val sleepInterval =
conf.get(config.STORAGE_DECOMMISSION_REPLICATION_REATTEMPT_INTERVAL)
+
+ override def run(): Unit = {
+ if (!conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED) &&
+ !conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logWarning("Decommissioning, but no task configured set one or
both:\n" +
+ s"${config.STORAGE_RDD_DECOMMISSION_ENABLED.key}\n" +
+ s"${config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED.key}")
+ stopped = true
+ }
+ while (!stopped && !Thread.interrupted()) {
+ logInfo("Iterating on migrating from the block manager.")
+ try {
+ // If enabled we migrate shuffle blocks first as they are more
expensive.
+ if (conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all shuffle blocks")
+ offloadShuffleBlocks()
Review comment:
sure.
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
##########
@@ -489,6 +491,24 @@ class BlockManagerMasterEndpoint(
storageLevel: StorageLevel,
memSize: Long,
diskSize: Long): Boolean = {
+ logInfo(s"Updating block info on master ${blockId} for ${blockManagerId}")
+
+ if (blockId.isInternalShuffle) {
+ blockId match {
+ case ShuffleIndexBlockId(shuffleId, mapId, _) =>
+ // Don't update the map output on just the index block
Review comment:
Yeah sub blocks are migrated in order (we don't parallel map them).
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val 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) =>
+ logInfo(s"Trying to migrate shuffle ${shuffleBlockInfo} to
${peer}")
+ val blocks =
+ bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
+ logInfo(s"Got migration sub-blocks ${blocks}")
+ blocks.foreach { case (blockId, buffer) =>
+ logInfo(s"Migrating sub-block ${blockId}")
+ bm.blockTransferService.uploadBlockSync(
+ peer.host,
+ peer.port,
+ peer.executorId,
+ blockId,
+ buffer,
+ StorageLevel.DISK_ONLY,
+ null)// class tag, we don't need for shuffle
+ logDebug(s"Migrated sub block ${blockId}")
+ }
+ logInfo(s"Migrated ${shuffleBlockInfo} to ${peer}")
+ }
+ }
+ // This catch is intentionally outside of the while running block.
+ // if we encounter errors migrating to an executor we want to stop.
+ } catch {
+ case e: Exception =>
+ migrating match {
+ case Some(shuffleMap) =>
+ logError(s"Error ${e} during migration, adding ${shuffleMap}
back to migration queue")
+ shufflesToMigrate.add(shuffleMap)
+ case None =>
+ logError(s"Error ${e} while waiting for block to migrate")
+ }
+ }
+ }
+ }
+
+ // Shuffles which are either in queue for migrations or migrated
+ private val migratingShuffles = mutable.HashSet[ShuffleBlockInfo]()
+
+ // Shuffles which are queued for migration
+ private[storage] val shufflesToMigrate =
+ new java.util.concurrent.ConcurrentLinkedQueue[ShuffleBlockInfo]()
+
+ @volatile private var stopped = false
+
+ private val migrationPeers =
+ mutable.HashMap[BlockManagerId, ShuffleMigrationRunnable]()
+
+ private lazy val blockMigrationExecutor =
+ ThreadUtils.newDaemonSingleThreadExecutor("block-manager-decommission")
+
+ private val blockMigrationRunnable = new Runnable {
+ val sleepInterval =
conf.get(config.STORAGE_DECOMMISSION_REPLICATION_REATTEMPT_INTERVAL)
+
+ override def run(): Unit = {
+ if (!conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED) &&
+ !conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logWarning("Decommissioning, but no task configured set one or
both:\n" +
+ s"${config.STORAGE_RDD_DECOMMISSION_ENABLED.key}\n" +
+ s"${config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED.key}")
+ stopped = true
+ }
+ while (!stopped && !Thread.interrupted()) {
+ logInfo("Iterating on migrating from the block manager.")
+ try {
+ // If enabled we migrate shuffle blocks first as they are more
expensive.
+ if (conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all shuffle blocks")
+ offloadShuffleBlocks()
+ logInfo("Done starting workers to migrate shuffle blocks")
+ }
+ if (conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all cached RDD blocks")
+ decommissionRddCacheBlocks()
+ logInfo("Attempt to replicate all cached blocks done")
+ }
+ logInfo(s"Waiting for ${sleepInterval} before refreshing
migrations.")
+ Thread.sleep(sleepInterval)
+ } catch {
+ case e: InterruptedException =>
+ logInfo("Interrupted during migration, will not refresh
migrations.")
+ stopped = true
+ case NonFatal(e) =>
+ logError("Error occurred while trying to replicate for block
manager decommissioning.",
+ e)
+ stopped = true
+ }
+ }
+ }
+ }
+
+ lazy val shuffleMigrationPool = ThreadUtils.newDaemonCachedThreadPool(
+ "migrate-shuffles",
+ conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_MAX_THREADS))
+ /**
Review comment:
sure
##########
File path: core/src/main/scala/org/apache/spark/SparkEnv.scala
##########
@@ -367,7 +367,8 @@ object SparkEnv extends Logging {
externalShuffleClient
} else {
None
- }, blockManagerInfo)),
+ }, blockManagerInfo,
+ mapOutputTracker.asInstanceOf[MapOutputTrackerMaster])),
Review comment:
`registerOrLookupEndpoint` does the trick, if the
BlockManagerMasterEndpoint is being constructed the `mapOutputTracker` is the
`mapOutputTrackerMaster`.
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val 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) =>
+ logInfo(s"Trying to migrate shuffle ${shuffleBlockInfo} to
${peer}")
+ val blocks =
+ bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
+ logInfo(s"Got migration sub-blocks ${blocks}")
+ blocks.foreach { case (blockId, buffer) =>
+ logInfo(s"Migrating sub-block ${blockId}")
+ bm.blockTransferService.uploadBlockSync(
+ peer.host,
+ peer.port,
+ peer.executorId,
+ blockId,
+ buffer,
+ StorageLevel.DISK_ONLY,
+ null)// class tag, we don't need for shuffle
+ logDebug(s"Migrated sub block ${blockId}")
+ }
+ logInfo(s"Migrated ${shuffleBlockInfo} to ${peer}")
+ }
+ }
+ // This catch is intentionally outside of the while running block.
+ // if we encounter errors migrating to an executor we want to stop.
+ } catch {
+ case e: Exception =>
+ migrating match {
+ case Some(shuffleMap) =>
+ logError(s"Error ${e} during migration, adding ${shuffleMap}
back to migration queue")
+ shufflesToMigrate.add(shuffleMap)
+ case None =>
+ logError(s"Error ${e} while waiting for block to migrate")
+ }
+ }
+ }
+ }
+
+ // Shuffles which are either in queue for migrations or migrated
+ private val migratingShuffles = mutable.HashSet[ShuffleBlockInfo]()
+
+ // Shuffles which are queued for migration
+ private[storage] val shufflesToMigrate =
+ new java.util.concurrent.ConcurrentLinkedQueue[ShuffleBlockInfo]()
+
+ @volatile private var stopped = false
+
+ private val migrationPeers =
+ mutable.HashMap[BlockManagerId, ShuffleMigrationRunnable]()
+
+ private lazy val blockMigrationExecutor =
+ ThreadUtils.newDaemonSingleThreadExecutor("block-manager-decommission")
+
+ private val blockMigrationRunnable = new Runnable {
+ val sleepInterval =
conf.get(config.STORAGE_DECOMMISSION_REPLICATION_REATTEMPT_INTERVAL)
+
+ override def run(): Unit = {
+ if (!conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED) &&
+ !conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logWarning("Decommissioning, but no task configured set one or
both:\n" +
+ s"${config.STORAGE_RDD_DECOMMISSION_ENABLED.key}\n" +
+ s"${config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED.key}")
+ stopped = true
+ }
+ while (!stopped && !Thread.interrupted()) {
+ logInfo("Iterating on migrating from the block manager.")
+ try {
+ // If enabled we migrate shuffle blocks first as they are more
expensive.
+ if (conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all shuffle blocks")
+ offloadShuffleBlocks()
+ logInfo("Done starting workers to migrate shuffle blocks")
+ }
+ if (conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all cached RDD blocks")
+ decommissionRddCacheBlocks()
+ logInfo("Attempt to replicate all cached blocks done")
+ }
+ logInfo(s"Waiting for ${sleepInterval} before refreshing
migrations.")
+ Thread.sleep(sleepInterval)
+ } catch {
+ case e: InterruptedException =>
+ logInfo("Interrupted during migration, will not refresh
migrations.")
+ stopped = true
+ case NonFatal(e) =>
+ logError("Error occurred while trying to replicate for block
manager decommissioning.",
+ e)
+ stopped = true
+ }
+ }
+ }
+ }
+
+ lazy val shuffleMigrationPool = ThreadUtils.newDaemonCachedThreadPool(
+ "migrate-shuffles",
+ conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_MAX_THREADS))
+ /**
+ * Tries to offload all shuffle blocks that are registered with the shuffle
service locally.
+ * Note: this does not delete the shuffle files in-case there is an
in-progress fetch
+ * but rather shadows them.
+ * Requires an Indexed based shuffle resolver.
+ * Note: if called in testing please call stopOffloadingShuffleBlocks to
avoid thread leakage.
+ */
+ private[storage] def offloadShuffleBlocks(): Unit = {
+ // Update the queue of shuffles to be migrated
+ logInfo("Offloading shuffle blocks")
+ val localShuffles = bm.migratableResolver.getStoredShuffles()
+ val newShufflesToMigrate = localShuffles.&~(migratingShuffles).toSeq
+ shufflesToMigrate.addAll(newShufflesToMigrate.asJava)
+ migratingShuffles ++= newShufflesToMigrate
+
+ // Update the threads doing migrations
+ // TODO: Sort & only start as many threads as min(||blocks||, ||targets||)
using location pref
Review comment:
Ah rocking, I'll drop it.
##########
File path: core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala
##########
@@ -178,6 +184,10 @@ private[spark] class HighlyCompressedMapStatus private (
override def location: BlockManagerId = loc
+ override def updateLocation(bm: BlockManagerId): Unit = {
Review comment:
Since location defined inside of each mapstatus we'd have to change
that. I've tried to minimize any changes which would impact code paths with
this feature turned off.
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
##########
@@ -489,6 +491,24 @@ class BlockManagerMasterEndpoint(
storageLevel: StorageLevel,
memSize: Long,
diskSize: Long): Boolean = {
+ logInfo(s"Updating block info on master ${blockId} for ${blockManagerId}")
Review comment:
Yeah I'll drop this down to Debug.
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val 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) =>
+ logInfo(s"Trying to migrate shuffle ${shuffleBlockInfo} to
${peer}")
+ val blocks =
+ bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
+ logInfo(s"Got migration sub-blocks ${blocks}")
+ blocks.foreach { case (blockId, buffer) =>
+ logInfo(s"Migrating sub-block ${blockId}")
+ bm.blockTransferService.uploadBlockSync(
+ peer.host,
+ peer.port,
+ peer.executorId,
+ blockId,
+ buffer,
+ StorageLevel.DISK_ONLY,
+ null)// class tag, we don't need for shuffle
+ logDebug(s"Migrated sub block ${blockId}")
+ }
+ logInfo(s"Migrated ${shuffleBlockInfo} to ${peer}")
+ }
+ }
+ // This catch is intentionally outside of the while running block.
+ // if we encounter errors migrating to an executor we want to stop.
+ } catch {
+ case e: Exception =>
+ migrating match {
+ case Some(shuffleMap) =>
+ logError(s"Error ${e} during migration, adding ${shuffleMap}
back to migration queue")
+ shufflesToMigrate.add(shuffleMap)
+ case None =>
+ logError(s"Error ${e} while waiting for block to migrate")
+ }
+ }
+ }
+ }
+
+ // Shuffles which are either in queue for migrations or migrated
+ private val migratingShuffles = mutable.HashSet[ShuffleBlockInfo]()
+
+ // Shuffles which are queued for migration
+ private[storage] val shufflesToMigrate =
+ new java.util.concurrent.ConcurrentLinkedQueue[ShuffleBlockInfo]()
+
+ @volatile private var stopped = false
+
+ private val migrationPeers =
+ mutable.HashMap[BlockManagerId, ShuffleMigrationRunnable]()
+
+ private lazy val blockMigrationExecutor =
+ ThreadUtils.newDaemonSingleThreadExecutor("block-manager-decommission")
+
+ private val blockMigrationRunnable = new Runnable {
+ val sleepInterval =
conf.get(config.STORAGE_DECOMMISSION_REPLICATION_REATTEMPT_INTERVAL)
+
+ override def run(): Unit = {
+ if (!conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED) &&
+ !conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logWarning("Decommissioning, but no task configured set one or
both:\n" +
+ s"${config.STORAGE_RDD_DECOMMISSION_ENABLED.key}\n" +
+ s"${config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED.key}")
+ stopped = true
+ }
+ while (!stopped && !Thread.interrupted()) {
+ logInfo("Iterating on migrating from the block manager.")
+ try {
+ // If enabled we migrate shuffle blocks first as they are more
expensive.
Review comment:
How about "If enabled we start migrating shuffle blocks as they are
non-blocking"?
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val 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) =>
+ logInfo(s"Trying to migrate shuffle ${shuffleBlockInfo} to
${peer}")
+ val blocks =
+ bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
+ logInfo(s"Got migration sub-blocks ${blocks}")
+ blocks.foreach { case (blockId, buffer) =>
+ logInfo(s"Migrating sub-block ${blockId}")
+ bm.blockTransferService.uploadBlockSync(
+ peer.host,
+ peer.port,
+ peer.executorId,
+ blockId,
+ buffer,
+ StorageLevel.DISK_ONLY,
+ null)// class tag, we don't need for shuffle
+ logDebug(s"Migrated sub block ${blockId}")
+ }
+ logInfo(s"Migrated ${shuffleBlockInfo} to ${peer}")
+ }
+ }
+ // This catch is intentionally outside of the while running block.
+ // if we encounter errors migrating to an executor we want to stop.
+ } catch {
+ case e: Exception =>
+ migrating match {
+ case Some(shuffleMap) =>
+ logError(s"Error ${e} during migration, adding ${shuffleMap}
back to migration queue")
+ shufflesToMigrate.add(shuffleMap)
+ case None =>
+ logError(s"Error ${e} while waiting for block to migrate")
+ }
+ }
+ }
+ }
+
+ // Shuffles which are either in queue for migrations or migrated
+ private val migratingShuffles = mutable.HashSet[ShuffleBlockInfo]()
+
+ // Shuffles which are queued for migration
+ private[storage] val shufflesToMigrate =
+ new java.util.concurrent.ConcurrentLinkedQueue[ShuffleBlockInfo]()
+
+ @volatile private var stopped = false
+
+ private val migrationPeers =
+ mutable.HashMap[BlockManagerId, ShuffleMigrationRunnable]()
+
+ private lazy val blockMigrationExecutor =
+ ThreadUtils.newDaemonSingleThreadExecutor("block-manager-decommission")
+
+ private val blockMigrationRunnable = new Runnable {
+ val sleepInterval =
conf.get(config.STORAGE_DECOMMISSION_REPLICATION_REATTEMPT_INTERVAL)
+
+ override def run(): Unit = {
+ if (!conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED) &&
+ !conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logWarning("Decommissioning, but no task configured set one or
both:\n" +
+ s"${config.STORAGE_RDD_DECOMMISSION_ENABLED.key}\n" +
+ s"${config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED.key}")
+ stopped = true
+ }
+ while (!stopped && !Thread.interrupted()) {
+ logInfo("Iterating on migrating from the block manager.")
+ try {
+ // If enabled we migrate shuffle blocks first as they are more
expensive.
+ if (conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all shuffle blocks")
+ offloadShuffleBlocks()
+ logInfo("Done starting workers to migrate shuffle blocks")
+ }
+ if (conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all cached RDD blocks")
+ decommissionRddCacheBlocks()
+ logInfo("Attempt to replicate all cached blocks done")
+ }
+ logInfo(s"Waiting for ${sleepInterval} before refreshing
migrations.")
+ Thread.sleep(sleepInterval)
+ } catch {
+ case e: InterruptedException =>
+ logInfo("Interrupted during migration, will not refresh
migrations.")
+ stopped = true
+ case NonFatal(e) =>
+ logError("Error occurred while trying to replicate for block
manager decommissioning.",
+ e)
+ stopped = true
+ }
+ }
+ }
+ }
+
+ lazy val shuffleMigrationPool = ThreadUtils.newDaemonCachedThreadPool(
+ "migrate-shuffles",
+ conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_MAX_THREADS))
+ /**
+ * Tries to offload all shuffle blocks that are registered with the shuffle
service locally.
+ * Note: this does not delete the shuffle files in-case there is an
in-progress fetch
+ * but rather shadows them.
+ * Requires an Indexed based shuffle resolver.
+ * Note: if called in testing please call stopOffloadingShuffleBlocks to
avoid thread leakage.
+ */
+ private[storage] def offloadShuffleBlocks(): Unit = {
+ // Update the queue of shuffles to be migrated
+ logInfo("Offloading shuffle blocks")
+ val localShuffles = bm.migratableResolver.getStoredShuffles()
+ val newShufflesToMigrate = localShuffles.&~(migratingShuffles).toSeq
+ shufflesToMigrate.addAll(newShufflesToMigrate.asJava)
+ migratingShuffles ++= newShufflesToMigrate
+
+ // Update the threads doing migrations
+ // TODO: Sort & only start as many threads as min(||blocks||, ||targets||)
using location pref
+ val livePeerSet = bm.getPeers(false).toSet
+ val currentPeerSet = migrationPeers.keys.toSet
+ val deadPeers = currentPeerSet.&~(livePeerSet)
Review comment:
sure
##########
File path:
core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ExecutorService
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.shuffle.{MigratableResolver, ShuffleBlockInfo}
+import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Class to handle block manager decommissioning retries.
+ * It creates a Thread to retry offloading all RDD cache and Shuffle blocks
+ */
+private[storage] class BlockManagerDecommissioner(
+ conf: SparkConf,
+ bm: BlockManager) extends Logging {
+
+ private val maxReplicationFailuresForDecommission =
+ conf.get(config.STORAGE_DECOMMISSION_MAX_REPLICATION_FAILURE_PER_BLOCK)
+
+ /**
+ * This runnable consumes any shuffle blocks in the queue for migration.
This part of a
+ * producer/consumer where the main migration loop updates the queue of
blocks to be migrated
+ * periodically. On migration failure, the current thread will reinsert the
block for another
+ * thread to consume. Each thread migrates blocks to a different particular
executor to avoid
+ * distribute the blocks as quickly as possible without overwhelming any
particular executor.
+ *
+ * There is no preference for which peer a given block is migrated to.
+ * This is notable different than the RDD cache block migration (further
down in this file)
+ * which uses the existing priority mechanism for determining where to
replicate blocks to.
+ * Generally speaking cache blocks are less impactful as they normally
represent narrow
+ * transformations and we normally have less cache present than shuffle data.
+ *
+ * The producer/consumer model is chosen for shuffle block migration to
maximize
+ * 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
+ override def run(): Unit = {
+ var migrating: Option[ShuffleBlockInfo] = None
+ logInfo(s"Starting migration thread for ${peer}")
+ // Once a block fails to transfer to an executor stop trying to transfer
more blocks
+ try {
+ while (running && !Thread.interrupted()) {
+ val 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) =>
+ logInfo(s"Trying to migrate shuffle ${shuffleBlockInfo} to
${peer}")
+ val blocks =
+ bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo)
+ logInfo(s"Got migration sub-blocks ${blocks}")
+ blocks.foreach { case (blockId, buffer) =>
+ logInfo(s"Migrating sub-block ${blockId}")
+ bm.blockTransferService.uploadBlockSync(
+ peer.host,
+ peer.port,
+ peer.executorId,
+ blockId,
+ buffer,
+ StorageLevel.DISK_ONLY,
+ null)// class tag, we don't need for shuffle
+ logDebug(s"Migrated sub block ${blockId}")
+ }
+ logInfo(s"Migrated ${shuffleBlockInfo} to ${peer}")
+ }
+ }
+ // This catch is intentionally outside of the while running block.
+ // if we encounter errors migrating to an executor we want to stop.
+ } catch {
+ case e: Exception =>
+ migrating match {
+ case Some(shuffleMap) =>
+ logError(s"Error ${e} during migration, adding ${shuffleMap}
back to migration queue")
+ shufflesToMigrate.add(shuffleMap)
+ case None =>
+ logError(s"Error ${e} while waiting for block to migrate")
+ }
+ }
+ }
+ }
+
+ // Shuffles which are either in queue for migrations or migrated
+ private val migratingShuffles = mutable.HashSet[ShuffleBlockInfo]()
+
+ // Shuffles which are queued for migration
+ private[storage] val shufflesToMigrate =
+ new java.util.concurrent.ConcurrentLinkedQueue[ShuffleBlockInfo]()
+
+ @volatile private var stopped = false
+
+ private val migrationPeers =
+ mutable.HashMap[BlockManagerId, ShuffleMigrationRunnable]()
+
+ private lazy val blockMigrationExecutor =
+ ThreadUtils.newDaemonSingleThreadExecutor("block-manager-decommission")
+
+ private val blockMigrationRunnable = new Runnable {
+ val sleepInterval =
conf.get(config.STORAGE_DECOMMISSION_REPLICATION_REATTEMPT_INTERVAL)
+
+ override def run(): Unit = {
+ if (!conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED) &&
+ !conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logWarning("Decommissioning, but no task configured set one or
both:\n" +
+ s"${config.STORAGE_RDD_DECOMMISSION_ENABLED.key}\n" +
+ s"${config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED.key}")
+ stopped = true
+ }
+ while (!stopped && !Thread.interrupted()) {
+ logInfo("Iterating on migrating from the block manager.")
+ try {
+ // If enabled we migrate shuffle blocks first as they are more
expensive.
+ if (conf.get(config.STORAGE_SHUFFLE_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all shuffle blocks")
+ offloadShuffleBlocks()
+ logInfo("Done starting workers to migrate shuffle blocks")
+ }
+ if (conf.get(config.STORAGE_RDD_DECOMMISSION_ENABLED)) {
+ logDebug("Attempting to replicate all cached RDD blocks")
+ decommissionRddCacheBlocks()
+ logInfo("Attempt to replicate all cached blocks done")
+ }
+ logInfo(s"Waiting for ${sleepInterval} before refreshing
migrations.")
+ Thread.sleep(sleepInterval)
Review comment:
Yeah we can split this into two threads to simplify matters.
----------------------------------------------------------------
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:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
