Updated comment in MemoryStore.dropFromMemory
Keep TorrentBroadcast piece blocks until unpersist is called

…ey fall out of scope. Also replaced TimeStampedHashMap to BoundedHashMaps and TimeStampedWeakValueHashMap for the necessary hashmap behavior.

Conflicts:
	core/src/main/scala/org/apache/spark/MapOutputTracker.scala
	core/src/main/scala/org/apache/spark/SparkContext.scala
	core/src/main/scala/org/apache/spark/scheduler/ResultTask.scala
	core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala
	core/src/main/scala/org/apache/spark/storage/BlockManager.scala
	core/src/main/scala/org/apache/spark/util/TimeStampedHashMap.scala
	core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala

…tches in finalize() methods, and replaced ArrayBlockingQueue to LinkedBlockingQueue to avoid blocking in Java's finalizing thread.

…mpedWeakValueHashMap.

…p shuffle metadata in MapOutputTrackerWorker.

Conflicts:
	core/src/main/scala/org/apache/spark/Dependency.scala
	core/src/main/scala/org/apache/spark/MapOutputTracker.scala
	core/src/main/scala/org/apache/spark/SparkContext.scala
	core/src/main/scala/org/apache/spark/SparkEnv.scala
	core/src/main/scala/org/apache/spark/rdd/RDD.scala
	core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
	core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala
	core/src/main/scala/org/apache/spark/storage/BlockManager.scala
	core/src/main/scala/org/apache/spark/storage/ThreadingTest.scala
	core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala

…ark into cleanup

Conflicts:
	core/src/main/scala/org/apache/spark/broadcast/BroadcastFactory.scala
	core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
	core/src/main/scala/org/apache/spark/storage/MemoryStore.scala

As of this commit, Spark does not clean up broadcast blocks.
This will be done in the next commit.

There is not currently a way to query the blocks on the executors,
an operation that is deceptively simple to accomplish. This commit
adds this mechanism in order to verify that blocks are in fact
persisted/unpersisted on the executors in the tests.

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

Rather than having a special purpose BroadcastHelperBlockId just for TorrentBroadcast,
we now have a single BroadcastBlockId that has a possibly empty field. This simplifies
broadcast clean-up because now we only have to look for one type of block.

This commit also simplifies BlockId JSON de/serialization in general by parsing the
name through regex with apply.

Not all blocks are reported to the master. In HttpBroadcast and
TorrentBroadcast, for instance, most blocks are not reported to master.
The lack of a mechanism to get local block statuses on each executor
makes it difficult to test the correctness of un/persisting a broadcast.

This new functionality, though only used for testing at the moment, is
general enough to be used for other things in the future.

This allows us to get rid of WrappedJavaHashMap without much duplicate code.

This largely accounts for the cases when WeakReference becomes no longer strongly
reachable, in which case the map should return None for all get() operations, and
should skip the entry for all listing operations.

I am merging this. I will take one more detailed look in the context of my original changes in the main PR.

Conflicts:
	core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
	core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

Conflicts:
	core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
	core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

Patrick's comments

…emoveShuffle and removeBroadcast in BlockManager*

…Cleaner.

…h blocks are told about to the driver. Also, fixed Broadcast API to hide destroy functionality.

Conflicts:
	core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
	core/src/main/scala/org/apache/spark/storage/BlockManager.scala
	core/src/main/scala/org/apache/spark/storage/BlockManagerMasterActor.scala
	core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala