Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
51 commits
Select commit Hold shift + click to select a range
1e752f1
Added unpersist method to Broadcast.
Feb 5, 2014
80dd977
Fix for Broadcast unpersist patch.
Feb 6, 2014
e427a9e
Added ContextCleaner to automatically clean RDDs and shuffles when th…
tdas Feb 14, 2014
8512612
Changed TimeStampedHashMap to use WrappedJavaHashMap.
tdas Feb 14, 2014
a24fefc
Merge remote-tracking branch 'apache/master' into state-cleanup
tdas Mar 11, 2014
cb0a5a6
Fixed docs and styles.
tdas Mar 11, 2014
ae9da88
Removed unncessary TimeStampedHashMap from DAGScheduler, added try-ca…
tdas Mar 12, 2014
e61daa0
Modifications based on the comments on PR 126.
tdas Mar 13, 2014
a7260d3
Added try-catch in context cleaner and null value cleaning in TimeSta…
tdas Mar 17, 2014
892b952
Removed use of BoundedHashMap, and made BlockManagerSlaveActor cleanu…
tdas Mar 18, 2014
e1fba5f
Style fix
tdas Mar 19, 2014
f2881fd
Changed ContextCleaner to use ReferenceQueue instead of finalizer
tdas Mar 25, 2014
620eca3
Changes based on PR comments.
tdas Mar 25, 2014
a007307
Merge remote-tracking branch 'apache/master' into state-cleanup
tdas Mar 25, 2014
d2f8b97
Removed duplicate unpersistRDD.
tdas Mar 25, 2014
6c9dcf6
Added missing Apache license
tdas Mar 25, 2014
c7ccef1
Merge branch 'bc-unpersist-merge' of github.com:ignatich/incubator-sp…
andrewor14 Mar 26, 2014
ba52e00
Refactor broadcast classes
andrewor14 Mar 26, 2014
d0edef3
Add framework for broadcast cleanup
andrewor14 Mar 26, 2014
544ac86
Clean up broadcast blocks through BlockManager*
andrewor14 Mar 26, 2014
e95479c
Add tests for unpersisting broadcast
andrewor14 Mar 27, 2014
f201a8d
Test broadcast cleanup in ContextCleanerSuite + remove BoundedHashMap
andrewor14 Mar 27, 2014
c92e4d9
Merge github.com:apache/spark into cleanup
andrewor14 Mar 27, 2014
0d17060
Import, comments, and style fixes (minor)
andrewor14 Mar 28, 2014
34f436f
Generalize BroadcastBlockId to remove BroadcastHelperBlockId
andrewor14 Mar 28, 2014
fbfeec8
Add functionality to query executors for their local BlockStatuses
andrewor14 Mar 29, 2014
88904a3
Make TimeStampedWeakValueHashMap a wrapper of TimeStampedHashMap
andrewor14 Mar 29, 2014
e442246
Merge github.com:apache/spark into cleanup
andrewor14 Mar 29, 2014
8557c12
Merge github.com:apache/spark into cleanup
andrewor14 Mar 30, 2014
7edbc98
Merge remote-tracking branch 'apache-github/master' into state-cleanup
tdas Mar 31, 2014
634a097
Merge branch 'state-cleanup' of github.com:tdas/spark into cleanup
andrewor14 Mar 31, 2014
7ed72fb
Fix style test fail + remove verbose test message regarding broadcast
andrewor14 Mar 31, 2014
5016375
Address TD's comments
andrewor14 Apr 1, 2014
f0aabb1
Correct semantics for TimeStampedWeakValueHashMap + add tests
andrewor14 Apr 2, 2014
762a4d8
Merge pull request #1 from andrewor14/cleanup
tdas Apr 2, 2014
a6460d4
Merge github.com:apache/spark into cleanup
andrewor14 Apr 4, 2014
c5b1d98
Address Patrick's comments
andrewor14 Apr 4, 2014
a2cc8bc
Merge remote-tracking branch 'apache/master' into state-cleanup
tdas Apr 4, 2014
ada45f0
Merge branch 'state-cleanup' of github.com:tdas/spark into cleanup
andrewor14 Apr 4, 2014
cd72d19
Make automatic cleanup configurable (not documented)
andrewor14 Apr 4, 2014
b27f8e8
Merge pull request #3 from andrewor14/cleanup
tdas Apr 4, 2014
a430f06
Fixed compilation errors.
tdas Apr 4, 2014
104a89a
Fixed failing BroadcastSuite unit tests by introducing blocking for r…
tdas Apr 4, 2014
6222697
Fixed bug and adding unit test for removeBroadcast in BlockManagerSuite.
tdas Apr 4, 2014
41c9ece
Added more unit tests for BlockManager, DiskBlockManager, and Context…
tdas Apr 7, 2014
2b95b5e
Added more documentation on Broadcast implementations, specially whic…
tdas Apr 7, 2014
4d05314
Scala style fix.
tdas Apr 7, 2014
cff023c
Fixed issues based on Andrew's comments.
tdas Apr 7, 2014
d25a86e
Fixed stupid typo.
tdas Apr 7, 2014
f489fdc
Merge remote-tracking branch 'apache/master' into state-cleanup
tdas Apr 8, 2014
61b8d6e
Fixed issue with Tachyon + new BlockManager methods.
tdas Apr 8, 2014
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Changes based on PR comments.
  • Loading branch information
tdas committed Mar 25, 2014
commit 620eca349808befa6c339bc5acc351c484495557
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/MapOutputTracker.scala
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ private[spark] abstract class MapOutputTracker(conf: SparkConf) extends Logging
* their cache of map output locations if this happens.
*/
protected var epoch: Long = 0
protected val epochLock = new java.lang.Object
protected val epochLock = new AnyRef

/** Remembers which map output locations are currently being fetched on a worker */
private val fetching = new HashSet[Int]
Expand Down Expand Up @@ -305,7 +305,7 @@ private[spark] class MapOutputTrackerMaster(conf: SparkConf)
cachedSerializedStatuses.clear()
}

protected def cleanup(cleanupTime: Long) {
private def cleanup(cleanupTime: Long) {
mapStatuses.clearOldValues(cleanupTime)
cachedSerializedStatuses.clearOldValues(cleanupTime)
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this should be private

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ class BlockManagerMasterActor(val isLocal: Boolean, conf: SparkConf) extends Act
private def removeShuffle(shuffleId: Int) {
// Nothing to do in the BlockManagerMasterActor data structures
val removeMsg = RemoveShuffle(shuffleId)
blockManagerInfo.values.map { bm =>
blockManagerInfo.values.foreach { bm =>
bm.slaveActor ! removeMsg
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this can be foreach

}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,14 +45,14 @@ import scala.reflect.ClassTag
private[spark] class BoundedHashMap[A, B](bound: Int, useLRU: Boolean)
extends WrappedJavaHashMap[A, B, A, B] with SynchronizedMap[A, B] {

protected[util] val internalJavaMap = Collections.synchronizedMap(new LinkedHashMap[A, B](
private[util] val internalJavaMap = Collections.synchronizedMap(new LinkedHashMap[A, B](
bound / 8, (0.75).toFloat, useLRU) {
override protected def removeEldestEntry(eldest: JMapEntry[A, B]): Boolean = {
size() > bound
}
})

protected[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _] = {
private[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _] = {
new BoundedHashMap[K1, V1](bound, useLRU)
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,9 @@ private[util] case class TimeStampedValue[T](timestamp: Long, value: T)
private[spark] class TimeStampedHashMap[A, B](updateTimeStampOnGet: Boolean = false)
extends WrappedJavaHashMap[A, B, A, TimeStampedValue[B]] with Logging {

protected[util] val internalJavaMap = new ConcurrentHashMap[A, TimeStampedValue[B]]()
private[util] val internalJavaMap = new ConcurrentHashMap[A, TimeStampedValue[B]]()

protected[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _] = {
private[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _] = {
new TimeStampedHashMap[K1, V1]()
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,11 +51,11 @@ private[spark] class TimeStampedWeakValueHashMap[A, B]()
/** Counter for counting the number of inserts */
private val insertCounts = new AtomicInteger(0)

protected[util] val internalJavaMap: util.Map[A, TimeStampedWeakValue[B]] = {
private[util] val internalJavaMap: util.Map[A, TimeStampedWeakValue[B]] = {
new ConcurrentHashMap[A, TimeStampedWeakValue[B]]()
}

protected[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _] = {
private[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _] = {
new TimeStampedWeakValueHashMap[K1, V1]()
}

Expand All @@ -68,15 +68,12 @@ private[spark] class TimeStampedWeakValueHashMap[A, B]()
}

override def get(key: A): Option[B] = {
Option(internalJavaMap.get(key)) match {
case Some(weakValue) =>
val value = weakValue.weakValue.get
if (value == null) {
internalJavaMap.remove(key)
}
Option(value)
case None =>
None
Option(internalJavaMap.get(key)).flatMap { weakValue =>
val value = weakValue.weakValue.get
if (value == null) {
internalJavaMap.remove(key)
}
Option(value)
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about

Option(internalJavaMap.get(key)).map { weakValue =>
    val value = weakValue.weakValue.get
    if (value == null) {
        internalJavaMap.remove(key)
    }
    value
}

?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not the same logic. This when value is null this makes the function return Some(null) instead of None. Changing map to flatMap is the solution.

}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -46,11 +46,15 @@ private[spark] abstract class WrappedJavaHashMap[K, V, IK, IV] extends Map[K, V]

/* Methods that must be defined. */

/** Internal Java HashMap that is being wrapped. */
protected[util] val internalJavaMap: JMap[IK, IV]
/**
* Internal Java HashMap that is being wrapped.
* Scoped private[util] so that rest of Spark code cannot
* directly access the internal map.
*/
private[util] val internalJavaMap: JMap[IK, IV]

/** Method to get a new instance of the internal Java HashMap. */
protected[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _]
private[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _]

/*
Methods that convert between internal and external types. These implementations
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,12 @@

package org.apache.spark.util

import scala.collection.mutable.{ArrayBuffer, HashMap, Map}
import scala.util.Random

import java.util
import java.lang.ref.WeakReference

import scala.collection.mutable.{ArrayBuffer, HashMap, Map}
import scala.util.Random

import org.scalatest.FunSuite

class WrappedJavaHashMapSuite extends FunSuite {
Expand Down Expand Up @@ -203,9 +203,9 @@ class WrappedJavaHashMapSuite extends FunSuite {
}

class TestMap[A, B] extends WrappedJavaHashMap[A, B, A, B] {
protected[util] val internalJavaMap: util.Map[A, B] = new util.HashMap[A, B]()
private[util] val internalJavaMap: util.Map[A, B] = new util.HashMap[A, B]()

protected[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _] = {
private[util] def newInstance[K1, V1](): WrappedJavaHashMap[K1, V1, _, _] = {
new TestMap[K1, V1]
}
}