cacheMemoryMap -> unrollMemoryMap

xiliu82 · Jul 18, 2014 · e9c3cb0 · e9c3cb0
1 parent 198e374
commit e9c3cb0
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 14 deletions.
diff --git a/core/src/main/scala/org/apache/spark/CacheManager.scala b/core/src/main/scala/org/apache/spark/CacheManager.scala
@@ -20,10 +20,8 @@ package org.apache.spark
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 
-import org.apache.spark.executor.InputMetrics
 import org.apache.spark.rdd.RDD
 import org.apache.spark.storage._
-import org.apache.spark.util.collection.SizeTrackingAppendOnlyBuffer
 
 /**
  * Spark class responsible for passing RDDs partition contents to the BlockManager and making

diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -67,13 +67,13 @@ class SparkEnv (
     val metricsSystem: MetricsSystem,
     val conf: SparkConf) extends Logging {
 
-  // A mapping of thread ID to amount of memory, in bytes, used for shuffle
+  // A mapping of thread ID to amount of memory, in bytes, used for shuffle aggregations
   // All accesses should be manually synchronized
   val shuffleMemoryMap = mutable.HashMap[Long, Long]()
 
-  // A mapping of thread ID to amount of memory, in bytes, used for unrolling an RDD partition
+  // A mapping of thread ID to amount of memory, in bytes, used for unrolling a block
   // All accesses should be manually synchronized
-  val cacheMemoryMap = mutable.HashMap[Long, Long]()
+  val unrollMemoryMap = mutable.HashMap[Long, Long]()
 
   private val pythonWorkers = mutable.HashMap[(String, Map[String, String]), PythonWorkerFactory]()
 

diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -268,9 +268,9 @@ private[spark] class Executor(
       } finally {
         val threadId = Thread.currentThread().getId
         val shuffleMemoryMap = env.shuffleMemoryMap
-        val cacheMemoryMap = env.cacheMemoryMap
+        val unrollMemoryMap = env.unrollMemoryMap
         shuffleMemoryMap.synchronized { shuffleMemoryMap.remove(threadId) }
-        cacheMemoryMap.synchronized { cacheMemoryMap.remove(threadId) }
+        unrollMemoryMap.synchronized { unrollMemoryMap.remove(threadId) }
         runningTasks.remove(taskId)
       }
     }

diff --git a/core/src/main/scala/org/apache/spark/storage/MemoryStore.scala b/core/src/main/scala/org/apache/spark/storage/MemoryStore.scala
@@ -197,7 +197,7 @@ private[spark] class MemoryStore(blockManager: BlockManager, maxMemory: Long)
     val memoryRequestPeriod = 1000  // How frequently we request for more memory for our buffer
 
     val threadId = Thread.currentThread().getId
-    val cacheMemoryMap = SparkEnv.get.cacheMemoryMap
+    val unrollMemoryMap = SparkEnv.get.unrollMemoryMap
     var buffer = new SizeTrackingAppendOnlyBuffer[Any]
 
     try {
@@ -212,15 +212,15 @@ private[spark] class MemoryStore(blockManager: BlockManager, maxMemory: Long)
           previousSize = currentSize
 
           // Atomically check whether there is sufficient memory in the global pool to continue
-          cacheMemoryMap.synchronized {
-            val previouslyOccupiedMemory = cacheMemoryMap.get(threadId).getOrElse(0L)
-            val otherThreadsMemory = cacheMemoryMap.values.sum - previouslyOccupiedMemory
+          unrollMemoryMap.synchronized {
+            val previouslyOccupiedMemory = unrollMemoryMap.get(threadId).getOrElse(0L)
+            val otherThreadsMemory = unrollMemoryMap.values.sum - previouslyOccupiedMemory
 
             // Request for memory for the local buffer, and return whether request is granted
             def requestForMemory(): Boolean = {
               val availableMemory = freeMemory - otherThreadsMemory
               val granted = availableMemory > memoryToRequest
-              if (granted) { cacheMemoryMap(threadId) = memoryToRequest }
+              if (granted) { unrollMemoryMap(threadId) = memoryToRequest }
               granted
             }
 
@@ -247,8 +247,8 @@ private[spark] class MemoryStore(blockManager: BlockManager, maxMemory: Long)
       // Unless we return an iterator that depends on the buffer, free up space for other threads
       if (enoughMemory) {
         buffer = null
-        cacheMemoryMap.synchronized {
-          cacheMemoryMap(threadId) = 0
+        unrollMemoryMap.synchronized {
+          unrollMemoryMap(threadId) = 0
         }
       }
     }