add Tachyon migration code

Author: zhzhan

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

@@ -247,9 +247,10 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
   private[spark] def eventLogDir: Option[URI] = _eventLogDir
   private[spark] def eventLogCodec: Option[String] = _eventLogCodec
 
-  // Generate the random name for a temp folder in Tachyon
+  // Generate the random name for a temp folder in OffHeap
   // Add a timestamp as the suffix here to make it more safe
-  val tachyonFolderName = "spark-" + randomUUID.toString()
+  val offHeapFolderName = "spark-" + randomUUID.toString()
+  conf.set("spark.offHeapStore.folderName", offHeapFolderName)
 
   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
 
-  /** TachyonStore failed to initialize after many attempts. */
-  val TACHYON_STORE_FAILED_TO_INITIALIZE = 54
+  /** OffHeapStore failed to initialize after many attempts. */
+  val OFFHEAP_STORE_FAILED_TO_INITIALIZE = 54
 
-  /** TachyonStore failed to create a local temporary directory after many attempts. */
-  val TACHYON_STORE_FAILED_TO_CREATE_DIR = 55
+  /** OffHeapStore failed to create a local temporary directory after many attempts. */
+  val OFFHEAP_STORE_FAILED_TO_CREATE_DIR = 55
 
   def explainExitCode(exitCode: Int): String = {
     exitCode match {
@@ -46,9 +46,9 @@ object ExecutorExitCode {
       case OOM => "OutOfMemoryError"
       case DISK_STORE_FAILED_TO_CREATE_DIR =>
         "Failed to create local directory (bad spark.local.dir?)"
-      case TACHYON_STORE_FAILED_TO_INITIALIZE => "TachyonStore failed to initialize."
-      case TACHYON_STORE_FAILED_TO_CREATE_DIR =>
-        "TachyonStore failed to create a local temporary directory."
+      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."
       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; TachyonSize: %s; DiskSize: %s".format(
+        "    CachedPartitions: %d; MemorySize: %s; OffHeap: %s; DiskSize: %s".format(
           info.numCachedPartitions, bytesToString(info.memSize),
-          bytesToString(info.tachyonSize), bytesToString(info.diskSize)))
+          bytesToString(info.offHeapSize), bytesToString(info.diskSize)))
 
       s"$rdd [$persistence]" +: storageInfo
     }

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

@@ -78,19 +78,10 @@ private[spark] class BlockManager(
   private val blockInfo = new TimeStampedHashMap[BlockId, BlockInfo]
 
   // Actual storage of where blocks are kept
-  private var tachyonInitialized = false
+  private var offHeapInitialized = false
   private[spark] val memoryStore = new MemoryStore(this, maxMemory)
   private[spark] val diskStore = new DiskStore(this, diskBlockManager)
-  private[spark] lazy val tachyonStore: TachyonStore = {
-    val storeDir = conf.get("spark.tachyonStore.baseDir", "/tmp_spark_tachyon")
-    val appFolderName = conf.get("spark.tachyonStore.folderName")
-    val tachyonStorePath = s"$storeDir/$appFolderName/${this.executorId}"
-    val tachyonMaster = conf.get("spark.tachyonStore.url",  "tachyon://localhost:19998")
-    val tachyonBlockManager =
-      new TachyonBlockManager(this, tachyonStorePath, tachyonMaster)
-    tachyonInitialized = true
-    new TachyonStore(this, tachyonBlockManager)
-  }
+  private[spark] lazy val offHeapStore: OffHeapStore = new OffHeapStore(this, executorId)
 
   private[spark]
   val externalShuffleServiceEnabled = conf.getBoolean("spark.shuffle.service.enabled", false)
@@ -376,10 +367,10 @@ private[spark] class BlockManager(
     if (info.tellMaster) {
       val storageLevel = status.storageLevel
       val inMemSize = Math.max(status.memSize, droppedMemorySize)
-      val inTachyonSize = status.tachyonSize
+      val inOffHeapSize = status.offHeapSize
       val onDiskSize = status.diskSize
       master.updateBlockInfo(
-        blockManagerId, blockId, storageLevel, inMemSize, onDiskSize, inTachyonSize)
+        blockManagerId, blockId, storageLevel, inMemSize, onDiskSize, inOffHeapSize)
     } else {
       true
     }
@@ -397,15 +388,15 @@ private[spark] class BlockManager(
           BlockStatus(StorageLevel.NONE, 0L, 0L, 0L)
         case level =>
           val inMem = level.useMemory && memoryStore.contains(blockId)
-          val inTachyon = level.useOffHeap && tachyonStore.contains(blockId)
+          val inOffHeap = level.useOffHeap && offHeapStore.contains(blockId)
           val onDisk = level.useDisk && diskStore.contains(blockId)
           val deserialized = if (inMem) level.deserialized else false
-          val replication = if (inMem || inTachyon || onDisk) level.replication else 1
-          val storageLevel = StorageLevel(onDisk, inMem, inTachyon, deserialized, replication)
+          val replication = if (inMem || inOffHeap || onDisk) level.replication else 1
+          val storageLevel = StorageLevel(onDisk, inMem, inOffHeap, deserialized, replication)
           val memSize = if (inMem) memoryStore.getSize(blockId) else 0L
-          val tachyonSize = if (inTachyon) tachyonStore.getSize(blockId) else 0L
+          val offHeapSize = if (inOffHeap) offHeapStore.getSize(blockId) else 0L
           val diskSize = if (onDisk) diskStore.getSize(blockId) else 0L
-          BlockStatus(storageLevel, memSize, diskSize, tachyonSize)
+          BlockStatus(storageLevel, memSize, diskSize, offHeapSize)
       }
     }
   }
@@ -485,11 +476,11 @@ private[spark] class BlockManager(
           }
         }
 
-        // Look for the block in Tachyon
+        // Look for the block in offheap
         if (level.useOffHeap) {
-          logDebug(s"Getting block $blockId from tachyon")
-          if (tachyonStore.contains(blockId)) {
-            tachyonStore.getBytes(blockId) match {
+          logDebug(s"Getting block $blockId from offheap")
+          if (offHeapStore.contains(blockId)) {
+            offHeapStore.getBytes(blockId) match {
               case Some(bytes) =>
                 if (!asBlockResult) {
                   return Some(bytes)
@@ -498,7 +489,7 @@ private[spark] class BlockManager(
                     dataDeserialize(blockId, bytes), DataReadMethod.Memory, info.size))
                 }
               case None =>
-                logDebug(s"Block $blockId not found in tachyon")
+                logDebug(s"Block $blockId not found in offHeap")
             }
           }
         }
@@ -766,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 tachyon for off-heap storage
-            (false, tachyonStore)
+            // Use off-heap storage
+            (false, offHeapStore)
           } else if (putLevel.useDisk) {
             // Don't get back the bytes from put unless we replicate them
             (putLevel.replication > 1, diskStore)
@@ -802,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, tachyon, or disk store,
+          // Now that the block is in either the memory, offheap, or disk store,
           // let other threads read it, and tell the master about it.
           marked = true
           putBlockInfo.markReady(size)
@@ -1099,10 +1090,10 @@ 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 removedFromTachyon = if (tachyonInitialized) tachyonStore.remove(blockId) else false
-        if (!removedFromMemory && !removedFromDisk && !removedFromTachyon) {
+        val removedFromOffHeap = if (offHeapInitialized) offHeapStore.remove(blockId) else false
+        if (!removedFromMemory && !removedFromDisk && !removedFromOffHeap) {
           logWarning(s"Block $blockId could not be removed as it was not found in either " +
-            "the disk, memory, or tachyon store")
+            "the disk, memory, or offheap store")
         }
         blockInfo.remove(blockId)
         if (tellMaster && info.tellMaster) {
@@ -1136,7 +1127,7 @@ private[spark] class BlockManager(
           val level = info.level
           if (level.useMemory) { memoryStore.remove(id) }
           if (level.useDisk) { diskStore.remove(id) }
-          if (level.useOffHeap) { tachyonStore.remove(id) }
+          if (level.useOffHeap) { offHeapStore.remove(id) }
           iterator.remove()
           logInfo(s"Dropped block $id")
         }
@@ -1216,8 +1207,8 @@ private[spark] class BlockManager(
     blockInfo.clear()
     memoryStore.clear()
     diskStore.clear()
-    if (tachyonInitialized) {
-      tachyonStore.clear()
+    if (offHeapInitialized) {
+      offHeapStore.clear()
     }
     metadataCleaner.cancel()
     broadcastCleaner.cancel()

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

@@ -54,9 +54,9 @@ class BlockManagerMaster(
       storageLevel: StorageLevel,
       memSize: Long,
       diskSize: Long,
-      tachyonSize: Long): Boolean = {
+      externalBlockStoreSize: Long): Boolean = {
     val res = driverEndpoint.askWithRetry[Boolean](
-      UpdateBlockInfo(blockManagerId, blockId, storageLevel, memSize, diskSize, tachyonSize))
+      UpdateBlockInfo(blockManagerId, blockId, storageLevel, memSize, diskSize, externalBlockStoreSize))
     logDebug(s"Updated info of block $blockId")
     res
   }

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, tachyonSize) =>
+      blockManagerId, blockId, storageLevel, deserializedSize, size, offHeapSize) =>
       context.reply(updateBlockInfo(
-        blockManagerId, blockId, storageLevel, deserializedSize, size, tachyonSize))
+        blockManagerId, blockId, storageLevel, deserializedSize, size, offHeapSize))
 
     case GetLocations(blockId) =>
       context.reply(getLocations(blockId))
@@ -314,7 +314,7 @@ class BlockManagerMasterEndpoint(
       storageLevel: StorageLevel,
       memSize: Long,
       diskSize: Long,
-      tachyonSize: Long): Boolean = {
+      offHeapSize: Long): Boolean = {
 
     if (!blockManagerInfo.contains(blockManagerId)) {
       if (blockManagerId.isDriver && !isLocal) {
@@ -332,7 +332,7 @@ class BlockManagerMasterEndpoint(
     }
 
     blockManagerInfo(blockManagerId).updateBlockInfo(
-      blockId, storageLevel, memSize, diskSize, tachyonSize)
+      blockId, storageLevel, memSize, diskSize, offHeapSize)
 
     var locations: mutable.HashSet[BlockManagerId] = null
     if (blockLocations.containsKey(blockId)) {
@@ -396,8 +396,8 @@ case class BlockStatus(
     storageLevel: StorageLevel,
     memSize: Long,
     diskSize: Long,
-    tachyonSize: Long) {
-  def isCached: Boolean = memSize + diskSize + tachyonSize > 0
+    offHeapSize: Long) {
+  def isCached: Boolean = memSize + diskSize + offHeapSize > 0
 }
 
 @DeveloperApi
@@ -429,7 +429,7 @@ private[spark] class BlockManagerInfo(
       storageLevel: StorageLevel,
       memSize: Long,
       diskSize: Long,
-      tachyonSize: Long) {
+      offHeapSize: 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-Tachyon.
+      /* isValid means it is either stored in-memory, on-disk or on-OffHeap.
        * The memSize here indicates the data size in or dropped from memory,
-       * tachyonSize here indicates the data size in or dropped from Tachyon,
+       * offHeapSize here indicates the data size in or dropped from offHeap,
        * 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, tachyonSize))
-        logInfo("Added %s on tachyon on %s (size: %s)".format(
-          blockId, blockManagerId.hostPort, Utils.bytesToString(tachyonSize)))
+        _blocks.put(blockId, BlockStatus(storageLevel, 0, 0, offHeapSize))
+        logInfo("Added %s on offHeap on %s (size: %s)".format(
+          blockId, blockManagerId.hostPort, Utils.bytesToString(offHeapSize)))
       }
     } 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 tachyon (size: %s)".format(
-          blockId, blockManagerId.hostPort, Utils.bytesToString(blockStatus.tachyonSize)))
+        logInfo("Removed %s on %s on offheap (size: %s)".format(
+          blockId, blockManagerId.hostPort, Utils.bytesToString(blockStatus.offHeapSize)))
       }
     }
   }

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 tachyonSize: Long)
+      var offHeapSize: Long)
     extends ToBlockManagerMaster
     with Externalizable {
 
@@ -72,7 +72,7 @@ private[spark] object BlockManagerMessages {
       storageLevel.writeExternal(out)
       out.writeLong(memSize)
       out.writeLong(diskSize)
-      out.writeLong(tachyonSize)
+      out.writeLong(offHeapSize)
     }
 
     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()
-      tachyonSize = in.readLong()
+      offHeapSize = in.readLong()
     }
   }
 

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

@@ -36,7 +36,7 @@ trait OffHeapBlockManager {
    *
    * @throws java.io.IOException when FS init failure.
    */
-  def init(blockManager: BlockManager, executorId: String)
+  def init(blockManager: BlockManager, executorId: String): Unit
 
   /**
    * remove the cache from offheap
@@ -87,24 +87,23 @@ trait OffHeapBlockManager {
 
 object OffHeapBlockManager extends Logging{
   val MAX_DIR_CREATION_ATTEMPTS = 10
-  val subDirsPerDir = 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
-     }
+  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
+    }
   }
 }