In-memory version of tracker+blocked shuffle checked in.

conf/java-opts

--Dspark.shuffle.class=spark.CustomParallelLocalFileShuffle
+-Dspark.shuffle.class=spark.TrackedCustomBlockedInMemoryShuffle
 -Dspark.shuffle.masterHostAddress=127.0.0.1
 -Dspark.shuffle.masterTrackerPort=22222 
--Dspark.shuffle.trackerStrategy=spark.BalanceRemainingShuffleTrackerStrategy 
+-Dspark.shuffle.trackerStrategy=spark.BalanceConnectionsShuffleTrackerStrategy 
 -Dspark.shuffle.maxRxConnections=40
 -Dspark.shuffle.maxTxConnections=120 
 -Dspark.shuffle.blockSize=4096

src/scala/spark/CustomBlockedInMemoryShuffle.scala

@@ -83,7 +83,7 @@ extends Shuffle[K, V, C] with Logging {
         var writeStartTime: Long = 0
         
         buckets(i).foreach(pair => {
-          // Open a new file if necessary
+          // Open a new stream if necessary
           if (!isDirty) {
             splitName = CustomBlockedInMemoryShuffle.getSplitName(shuffleId, 
               myIndex, i, blockNum)
@@ -98,7 +98,7 @@ extends Shuffle[K, V, C] with Logging {
           oos.writeObject(pair)
           isDirty = true
           
-          // Close the old file if has crossed the blockSize limit
+          // Close the old stream if has crossed the blockSize limit
           if (baos.size > Shuffle.BlockSize) {
             CustomBlockedInMemoryShuffle.splitsCache(splitName) = 
               baos.toByteArray
@@ -477,30 +477,9 @@ object CustomBlockedInMemoryShuffle extends Logging {
     nextShuffleId.getAndIncrement()
   }
   
-  // Returns a standard ThreadFactory except all threads are daemons
-  private def newDaemonThreadFactory: ThreadFactory = {
-    new ThreadFactory {
-      def newThread(r: Runnable): Thread = {
-        var t = Executors.defaultThreadFactory.newThread(r)
-        t.setDaemon(true)
-        return t
-      }
-    }
-  }
-
-  // Wrapper over newFixedThreadPool
-  def newDaemonFixedThreadPool(nThreads: Int): ThreadPoolExecutor = {
-    var threadPool =
-      Executors.newFixedThreadPool(nThreads).asInstanceOf[ThreadPoolExecutor]
-
-    threadPool.setThreadFactory(newDaemonThreadFactory)
-    
-    return threadPool
-  }
-  
   class ShuffleServer
   extends Thread with Logging {
-    var threadPool = newDaemonFixedThreadPool(Shuffle.MaxTxConnections)
+    var threadPool = Shuffle.newDaemonFixedThreadPool(Shuffle.MaxTxConnections)
 
     var serverSocket: ServerSocket = null
 

src/scala/spark/TrackedCustomBlockedLocalFileShuffle.scala

@@ -836,6 +836,7 @@ object TrackedCustomBlockedLocalFileShuffle extends Logging {
           blockNumIn.close()
           
           oos.writeObject(BLOCKNUM)
+          oos.flush()
           
           val startTime = System.currentTimeMillis
           var curTime = startTime