minor bug fix

core/src/main/scala/org/apache/spark/storage/memory/MemoryEntryManager.scala

@@ -59,34 +59,14 @@ class FIFOMemoryEntryManager[K, V <: MemoryEntry[_]] extends MemoryEntryManager[
     }
   }
 
-  def foo(freedMemory: Long, space: Long, blockIsEvictable: (K) => Boolean,
-          hasWriteLock: (K, Boolean) => Boolean): (ArrayBuffer[K], Long) = {
-    val selectedBlocks = new ArrayBuffer[K]
-    var freed = freedMemory
-    entries.synchronized {
-      val iterator = entries.entrySet().iterator()
-      while (freedMemory < space && iterator.hasNext) {
-        val pair = iterator.next()
-        val blockId = pair.getKey
-        if (blockIsEvictable(blockId)) {
-          if (hasWriteLock(blockId, false)) {
-            selectedBlocks += blockId
-            freed += pair.getValue.size
-          }
-        }
-      }
-    }
-    (selectedBlocks, freed)
-  }
-
   override def containsEntry(key: K): Boolean = {
     entries.synchronized {
       entries.containsKey(key)
     }
   }
 }
 
-class LRUMemoryEntryManager[K, V] extends MemoryEntryManager[K, V] {
+class LRUMemoryEntryManager[K, V <: MemoryEntry[_]] extends MemoryEntryManager[K, V] {
   val entries = new util.LinkedHashMap[K, V](32, 0.75f, true)
 
   override def getEntry(key: K): V = {
@@ -118,4 +98,25 @@ class LRUMemoryEntryManager[K, V] extends MemoryEntryManager[K, V] {
       entries.containsKey(key)
     }
   }
+
+
+  def foo(freedMemory: Long, space: Long, blockIsEvictable: (K) => Boolean,
+          hasWriteLock: (K) => Boolean): (ArrayBuffer[K], Long) = {
+    val selectedBlocks = new ArrayBuffer[K]
+    var freed = freedMemory
+    entries.synchronized {
+      val iterator = entries.entrySet().iterator()
+      while (freedMemory < space && iterator.hasNext) {
+        val pair = iterator.next()
+        val blockId = pair.getKey
+        if (blockIsEvictable(blockId)) {
+          if (hasWriteLock(blockId)) {
+            selectedBlocks += blockId
+            freed += pair.getValue.size
+          }
+        }
+      }
+    }
+    (selectedBlocks, freed)
+  }
 }

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

@@ -461,10 +461,10 @@ private[spark] class MemoryStore(
 //          }
 //        }
 //      }
-      val foo = new FIFOMemoryEntryManager[BlockId, MemoryEntry[_]]
-      (selectedBlocks, freedMemory) = foo.foo(freedMemory, space,
+//      val foo = new FIFOMemoryEntryManager[BlockId, MemoryEntry[_]]
+      (selectedBlocks, freedMemory) = entries.foo(freedMemory, space,
         (blockId: BlockId) => blockIsEvictable(blockId),
-        (blockId: BlockId, bar) => blockInfoManager.lockForWriting(blockId, bar).isDefined)
+        (blockId: BlockId) => blockInfoManager.lockForWriting(blockId, true).isDefined)
 
       def dropBlock[T](blockId: BlockId, entry: MemoryEntry[T]): Unit = {
         val data = entry match {