resolve review comments and restructure blockmanasger code

Author: zhzhan

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -249,8 +249,10 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
 
   // Generate the random name for a temp folder in OffHeap
   // Add a timestamp as the suffix here to make it more safe
-  val offHeapFolderName = "spark-" + randomUUID.toString()
-  conf.set("spark.offHeapStore.folderName", offHeapFolderName)
+  val extBlkFolderName = "spark-" + randomUUID.toString()
+  @deprecated("Use extBlkFolderName instead.", "1.4.0")
+  val tachyonFolderName = extBlkFolderName
+  conf.set("spark.extBlkStore.folderName", extBlkFolderName)
 
   def isLocal: Boolean = (master == "local" || master.startsWith("local["))
 

core/src/main/scala/org/apache/spark/executor/ExecutorExitCode.scala

@@ -33,11 +33,11 @@ object ExecutorExitCode {
   /** DiskStore failed to create a local temporary directory after many attempts. */
   val DISK_STORE_FAILED_TO_CREATE_DIR = 53
 
-  /** OffHeapStore failed to initialize after many attempts. */
-  val OFFHEAP_STORE_FAILED_TO_INITIALIZE = 54
+  /** ExtBlkStore failed to initialize after many attempts. */
+  val ExtBlk_STORE_FAILED_TO_INITIALIZE = 54
 
-  /** OffHeapStore failed to create a local temporary directory after many attempts. */
-  val OFFHEAP_STORE_FAILED_TO_CREATE_DIR = 55
+  /** ExtBlkStore failed to create a local temporary directory after many attempts. */
+  val ExtBlk_STORE_FAILED_TO_CREATE_DIR = 55
 
   def explainExitCode(exitCode: Int): String = {
     exitCode match {
@@ -46,9 +46,11 @@ object ExecutorExitCode {
       case OOM => "OutOfMemoryError"
       case DISK_STORE_FAILED_TO_CREATE_DIR =>
         "Failed to create local directory (bad spark.local.dir?)"
-      case OFFHEAP_STORE_FAILED_TO_INITIALIZE => "OffHeap Store failed to initialize."
-      case OFFHEAP_STORE_FAILED_TO_CREATE_DIR =>
-        "OffHeap Store failed to create a local temporary directory."
+      // TODO: replace external block store with concreate implementation desc
+      case ExtBlk_STORE_FAILED_TO_INITIALIZE => "External Block Store failed to initialize."
+      // TODO: replace external block store with concreate implementation desc
+      case ExtBlk_STORE_FAILED_TO_CREATE_DIR =>
+        "External Block Store failed to create a local temporary directory."
       case _ =>
         "Unknown executor exit code (" + exitCode + ")" + (
           if (exitCode > 128) {

core/src/main/scala/org/apache/spark/rdd/RDD.scala

@@ -1475,9 +1475,9 @@ abstract class RDD[T: ClassTag](
 
       val persistence = if (storageLevel != StorageLevel.NONE) storageLevel.description else ""
       val storageInfo = rdd.context.getRDDStorageInfo.filter(_.id == rdd.id).map(info =>
-        "    CachedPartitions: %d; MemorySize: %s; OffHeap: %s; DiskSize: %s".format(
+        "    CachedPartitions: %d; MemorySize: %s; ExtBlkStoreSize: %s; DiskSize: %s".format(
           info.numCachedPartitions, bytesToString(info.memSize),
-          bytesToString(info.offHeapSize), bytesToString(info.diskSize)))
+          bytesToString(info.extBlkStoreSize), bytesToString(info.diskSize)))
 
       s"$rdd [$persistence]" +: storageInfo
     }

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -78,10 +78,10 @@ private[spark] class BlockManager(
   private val blockInfo = new TimeStampedHashMap[BlockId, BlockInfo]
 
   // Actual storage of where blocks are kept
-  private var offHeapInitialized = false
+  private var extBlkStoreInitialized = false
   private[spark] val memoryStore = new MemoryStore(this, maxMemory)
   private[spark] val diskStore = new DiskStore(this, diskBlockManager)
-  private[spark] lazy val offHeapStore: OffHeapStore = new OffHeapStore(this, executorId)
+  private[spark] lazy val extBlkStore: ExtBlockStore = new ExtBlockStore(this, executorId)
 
   private[spark]
   val externalShuffleServiceEnabled = conf.getBoolean("spark.shuffle.service.enabled", false)
@@ -367,10 +367,10 @@ private[spark] class BlockManager(
     if (info.tellMaster) {
       val storageLevel = status.storageLevel
       val inMemSize = Math.max(status.memSize, droppedMemorySize)
-      val inOffHeapSize = status.offHeapSize
+      val inExtBlkStoreSize = status.extBlkStoreSize
       val onDiskSize = status.diskSize
       master.updateBlockInfo(
-        blockManagerId, blockId, storageLevel, inMemSize, onDiskSize, inOffHeapSize)
+        blockManagerId, blockId, storageLevel, inMemSize, onDiskSize, inExtBlkStoreSize)
     } else {
       true
     }
@@ -388,15 +388,15 @@ private[spark] class BlockManager(
           BlockStatus(StorageLevel.NONE, 0L, 0L, 0L)
         case level =>
           val inMem = level.useMemory && memoryStore.contains(blockId)
-          val inOffHeap = level.useOffHeap && offHeapStore.contains(blockId)
+          val inExtBlkStore = level.useOffHeap && extBlkStore.contains(blockId)
           val onDisk = level.useDisk && diskStore.contains(blockId)
           val deserialized = if (inMem) level.deserialized else false
-          val replication = if (inMem || inOffHeap || onDisk) level.replication else 1
-          val storageLevel = StorageLevel(onDisk, inMem, inOffHeap, deserialized, replication)
+          val replication = if (inMem || inExtBlkStore || onDisk) level.replication else 1
+          val storageLevel = StorageLevel(onDisk, inMem, inExtBlkStore, deserialized, replication)
           val memSize = if (inMem) memoryStore.getSize(blockId) else 0L
-          val offHeapSize = if (inOffHeap) offHeapStore.getSize(blockId) else 0L
+          val extBlkStoreSize = if (inExtBlkStore) extBlkStore.getSize(blockId) else 0L
           val diskSize = if (onDisk) diskStore.getSize(blockId) else 0L
-          BlockStatus(storageLevel, memSize, diskSize, offHeapSize)
+          BlockStatus(storageLevel, memSize, diskSize, extBlkStoreSize)
       }
     }
   }
@@ -476,11 +476,11 @@ private[spark] class BlockManager(
           }
         }
 
-        // Look for the block in offheap
+        // Look for the block in external blk store
         if (level.useOffHeap) {
-          logDebug(s"Getting block $blockId from offheap")
-          if (offHeapStore.contains(blockId)) {
-            offHeapStore.getBytes(blockId) match {
+          logDebug(s"Getting block $blockId from ExtBlkStore")
+          if (extBlkStore.contains(blockId)) {
+            extBlkStore.getBytes(blockId) match {
               case Some(bytes) =>
                 if (!asBlockResult) {
                   return Some(bytes)
@@ -489,7 +489,7 @@ private[spark] class BlockManager(
                     dataDeserialize(blockId, bytes), DataReadMethod.Memory, info.size))
                 }
               case None =>
-                logDebug(s"Block $blockId not found in offHeap")
+                logDebug(s"Block $blockId not found in extBlkStore")
             }
           }
         }
@@ -757,8 +757,8 @@ private[spark] class BlockManager(
             // We will drop it to disk later if the memory store can't hold it.
             (true, memoryStore)
           } else if (putLevel.useOffHeap) {
-            // Use off-heap storage
-            (false, offHeapStore)
+            // Use external blk storage
+            (false, extBlkStore)
           } else if (putLevel.useDisk) {
             // Don't get back the bytes from put unless we replicate them
             (putLevel.replication > 1, diskStore)
@@ -793,7 +793,7 @@ private[spark] class BlockManager(
 
         val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
         if (putBlockStatus.storageLevel != StorageLevel.NONE) {
-          // Now that the block is in either the memory, offheap, or disk store,
+          // Now that the block is in either the memory, extBlkStore, or disk store,
           // let other threads read it, and tell the master about it.
           marked = true
           putBlockInfo.markReady(size)
@@ -1090,10 +1090,11 @@ private[spark] class BlockManager(
         // Removals are idempotent in disk store and memory store. At worst, we get a warning.
         val removedFromMemory = memoryStore.remove(blockId)
         val removedFromDisk = diskStore.remove(blockId)
-        val removedFromOffHeap = if (offHeapInitialized) offHeapStore.remove(blockId) else false
-        if (!removedFromMemory && !removedFromDisk && !removedFromOffHeap) {
+        val removedFromExtBlkStore =
+          if (extBlkStoreInitialized) extBlkStore.remove(blockId) else false
+        if (!removedFromMemory && !removedFromDisk && !removedFromExtBlkStore) {
           logWarning(s"Block $blockId could not be removed as it was not found in either " +
-            "the disk, memory, or offheap store")
+            "the disk, memory, or external blk store")
         }
         blockInfo.remove(blockId)
         if (tellMaster && info.tellMaster) {
@@ -1127,7 +1128,7 @@ private[spark] class BlockManager(
           val level = info.level
           if (level.useMemory) { memoryStore.remove(id) }
           if (level.useDisk) { diskStore.remove(id) }
-          if (level.useOffHeap) { offHeapStore.remove(id) }
+          if (level.useOffHeap) { extBlkStore.remove(id) }
           iterator.remove()
           logInfo(s"Dropped block $id")
         }
@@ -1207,8 +1208,8 @@ private[spark] class BlockManager(
     blockInfo.clear()
     memoryStore.clear()
     diskStore.clear()
-    if (offHeapInitialized) {
-      offHeapStore.clear()
+    if (extBlkStoreInitialized) {
+      extBlkStore.clear()
     }
     metadataCleaner.cancel()
     broadcastCleaner.cancel()

core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala

@@ -60,9 +60,9 @@ class BlockManagerMasterEndpoint(
       context.reply(true)
 
     case UpdateBlockInfo(
-      blockManagerId, blockId, storageLevel, deserializedSize, size, offHeapSize) =>
+      blockManagerId, blockId, storageLevel, deserializedSize, size, extBlkStoreSize) =>
       context.reply(updateBlockInfo(
-        blockManagerId, blockId, storageLevel, deserializedSize, size, offHeapSize))
+        blockManagerId, blockId, storageLevel, deserializedSize, size, extBlkStoreSize))
 
     case GetLocations(blockId) =>
       context.reply(getLocations(blockId))
@@ -314,7 +314,7 @@ class BlockManagerMasterEndpoint(
       storageLevel: StorageLevel,
       memSize: Long,
       diskSize: Long,
-      offHeapSize: Long): Boolean = {
+      extBlkStoreSize: Long): Boolean = {
 
     if (!blockManagerInfo.contains(blockManagerId)) {
       if (blockManagerId.isDriver && !isLocal) {
@@ -332,7 +332,7 @@ class BlockManagerMasterEndpoint(
     }
 
     blockManagerInfo(blockManagerId).updateBlockInfo(
-      blockId, storageLevel, memSize, diskSize, offHeapSize)
+      blockId, storageLevel, memSize, diskSize, extBlkStoreSize)
 
     var locations: mutable.HashSet[BlockManagerId] = null
     if (blockLocations.containsKey(blockId)) {
@@ -396,8 +396,8 @@ case class BlockStatus(
     storageLevel: StorageLevel,
     memSize: Long,
     diskSize: Long,
-    offHeapSize: Long) {
-  def isCached: Boolean = memSize + diskSize + offHeapSize > 0
+    extBlkStoreSize: Long) {
+  def isCached: Boolean = memSize + diskSize + extBlkStoreSize > 0
 }
 
 @DeveloperApi
@@ -429,7 +429,7 @@ private[spark] class BlockManagerInfo(
       storageLevel: StorageLevel,
       memSize: Long,
       diskSize: Long,
-      offHeapSize: Long) {
+      extBlkStoreSize: Long) {
 
     updateLastSeenMs()
 
@@ -445,9 +445,9 @@ private[spark] class BlockManagerInfo(
     }
 
     if (storageLevel.isValid) {
-      /* isValid means it is either stored in-memory, on-disk or on-OffHeap.
+      /* isValid means it is either stored in-memory, on-disk or on-extBlkStore.
        * The memSize here indicates the data size in or dropped from memory,
-       * offHeapSize here indicates the data size in or dropped from offHeap,
+       * extBlkStoreSize here indicates the data size in or dropped from extBlkStore,
        * and the diskSize here indicates the data size in or dropped to disk.
        * They can be both larger than 0, when a block is dropped from memory to disk.
        * Therefore, a safe way to set BlockStatus is to set its info in accurate modes. */
@@ -464,9 +464,9 @@ private[spark] class BlockManagerInfo(
           blockId, blockManagerId.hostPort, Utils.bytesToString(diskSize)))
       }
       if (storageLevel.useOffHeap) {
-        _blocks.put(blockId, BlockStatus(storageLevel, 0, 0, offHeapSize))
-        logInfo("Added %s on offHeap on %s (size: %s)".format(
-          blockId, blockManagerId.hostPort, Utils.bytesToString(offHeapSize)))
+        _blocks.put(blockId, BlockStatus(storageLevel, 0, 0, extBlkStoreSize))
+        logInfo("Added %s on ExtBlkStore on %s (size: %s)".format(
+          blockId, blockManagerId.hostPort, Utils.bytesToString(extBlkStoreSize)))
       }
     } else if (_blocks.containsKey(blockId)) {
       // If isValid is not true, drop the block.
@@ -482,8 +482,8 @@ private[spark] class BlockManagerInfo(
           blockId, blockManagerId.hostPort, Utils.bytesToString(blockStatus.diskSize)))
       }
       if (blockStatus.storageLevel.useOffHeap) {
-        logInfo("Removed %s on %s on offheap (size: %s)".format(
-          blockId, blockManagerId.hostPort, Utils.bytesToString(blockStatus.offHeapSize)))
+        logInfo("Removed %s on %s on extBlkStore (size: %s)".format(
+          blockId, blockManagerId.hostPort, Utils.bytesToString(blockStatus.extBlkStoreSize)))
       }
     }
   }

core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala

@@ -60,7 +60,7 @@ private[spark] object BlockManagerMessages {
       var storageLevel: StorageLevel,
       var memSize: Long,
       var diskSize: Long,
-      var offHeapSize: Long)
+      var extBlkStoreSize: Long)
     extends ToBlockManagerMaster
     with Externalizable {
 
@@ -72,7 +72,7 @@ private[spark] object BlockManagerMessages {
       storageLevel.writeExternal(out)
       out.writeLong(memSize)
       out.writeLong(diskSize)
-      out.writeLong(offHeapSize)
+      out.writeLong(extBlkStoreSize)
     }
 
     override def readExternal(in: ObjectInput): Unit = Utils.tryOrIOException {
@@ -81,7 +81,7 @@ private[spark] object BlockManagerMessages {
       storageLevel = StorageLevel(in)
       memSize = in.readLong()
       diskSize = in.readLong()
-      offHeapSize = in.readLong()
+      extBlkStoreSize = in.readLong()
     }
   }
 

core/src/main/scala/org/apache/spark/storage/OffHeapBlockManager.scala renamed to core/src/main/scala/org/apache/spark/storage/ExtBlockManager.scala

@@ -23,13 +23,13 @@ import org.apache.spark.Logging
 import scala.util.control.NonFatal
 
 
-trait OffHeapBlockManager {
+private[spark] abstract class ExtBlockManager {
 
   /**
    * desc for the implementation.
    *
    */
-  def desc(): String = {"OffHeap"}
+  def desc(): String = {"External Block Store"}
 
   /**
    * initialize a concrete block manager implementation.
@@ -39,7 +39,7 @@ trait OffHeapBlockManager {
   def init(blockManager: BlockManager, executorId: String): Unit
 
   /**
-   * remove the cache from offheap
+   * remove the cache from ExtBlkStore
    *
    * @throws java.io.IOException when FS failure in removing file.
    */
@@ -53,14 +53,14 @@ trait OffHeapBlockManager {
   def fileExists(blockId: BlockId): Boolean
 
   /**
-   * save the cache to the offheap.
+   * save the cache to the ExtBlkStore.
    *
    * @throws java.io.IOException when FS failure in put blocks.
    */
   def putBytes(blockId: BlockId, bytes: ByteBuffer)
 
   /**
-   * retrieve the cache from offheap
+   * retrieve the cache from ExtBlkStore
    *
    * @throws java.io.IOException when FS failure in get blocks.
    */
@@ -78,32 +78,4 @@ trait OffHeapBlockManager {
    *
    */
   def addShutdownHook()
-
-  final def setup(blockManager: BlockManager, executorId: String): Unit = {
-    init(blockManager, executorId)
-    addShutdownHook()
-  }
-}
-
-object OffHeapBlockManager extends Logging{
-  val MAX_DIR_CREATION_ATTEMPTS = 10
-  val SUB_DIRS_PER_DIR = "64"
-  def create(blockManager: BlockManager, executorId: String): Option[OffHeapBlockManager] = {
-    val sNames = blockManager.conf.getOption("spark.offHeapStore.blockManager")
-    sNames match {
-      case Some(name) =>
-        try {
-          val instance = Class.forName(name)
-            .newInstance()
-            .asInstanceOf[OffHeapBlockManager]
-          instance.setup(blockManager, executorId)
-          Some(instance)
-        } catch {
-          case NonFatal(t) =>
-            logError("Cannot initialize offHeap store")
-            None
-        }
-      case None => None
-    }
-  }
 }