Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Modified bunch HashMaps in Spark to use TimeStampedHashMap and made v…
…arious modules use CleanupTask to periodically clean up metadata.
- Loading branch information
Showing
7 changed files
with
165 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package spark.util | ||
|
|
||
| import java.util.concurrent.{TimeUnit, ScheduledFuture, Executors} | ||
| import java.util.{TimerTask, Timer} | ||
| import spark.Logging | ||
|
|
||
| class CleanupTask(name: String, cleanupFunc: (Long) => Unit) extends Logging { | ||
| val delayMins = System.getProperty("spark.cleanup.delay", "-100").toInt | ||
| val periodMins = System.getProperty("spark.cleanup.period", (delayMins / 10).toString).toInt | ||
| val timer = new Timer(name + " cleanup timer", true) | ||
| val task = new TimerTask { | ||
| def run() { | ||
| try { | ||
| if (delayMins > 0) { | ||
|
|
||
| cleanupFunc(System.currentTimeMillis() - (delayMins * 60 * 1000)) | ||
| logInfo("Ran cleanup task for " + name) | ||
| } | ||
| } catch { | ||
| case e: Exception => logError("Error running cleanup task for " + name, e) | ||
| } | ||
| } | ||
| } | ||
| if (periodMins > 0) { | ||
| timer.schedule(task, periodMins * 60 * 1000, periodMins * 60 * 1000) | ||
| } | ||
|
|
||
| def cancel() { | ||
| timer.cancel() | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| package spark.util | ||
|
|
||
| import scala.collection.JavaConversions._ | ||
| import scala.collection.mutable.{HashMap, Map} | ||
| import java.util.concurrent.ConcurrentHashMap | ||
|
|
||
| /** | ||
| * This is a custom implementation of scala.collection.mutable.Map which stores the insertion | ||
| * time stamp along with each key-value pair. Key-value pairs that are older than a particular | ||
| * threshold time can them be removed using the cleanup method. This is intended to be a drop-in | ||
| * replacement of scala.collection.mutable.HashMap. | ||
| */ | ||
| class TimeStampedHashMap[A, B] extends Map[A, B]() { | ||
| val internalMap = new ConcurrentHashMap[A, (B, Long)]() | ||
|
|
||
| def get(key: A): Option[B] = { | ||
| val value = internalMap.get(key) | ||
| if (value != null) Some(value._1) else None | ||
| } | ||
|
|
||
| def iterator: Iterator[(A, B)] = { | ||
| val jIterator = internalMap.entrySet().iterator() | ||
| jIterator.map(kv => (kv.getKey, kv.getValue._1)) | ||
| } | ||
|
|
||
| override def + [B1 >: B](kv: (A, B1)): Map[A, B1] = { | ||
| val newMap = new TimeStampedHashMap[A, B1] | ||
| newMap.internalMap.putAll(this.internalMap) | ||
| newMap.internalMap.put(kv._1, (kv._2, currentTime)) | ||
| newMap | ||
| } | ||
|
|
||
| override def - (key: A): Map[A, B] = { | ||
| internalMap.remove(key) | ||
| this | ||
| } | ||
|
|
||
| override def += (kv: (A, B)): this.type = { | ||
| internalMap.put(kv._1, (kv._2, currentTime)) | ||
| this | ||
| } | ||
|
|
||
| override def -= (key: A): this.type = { | ||
| internalMap.remove(key) | ||
| this | ||
| } | ||
|
|
||
| override def update(key: A, value: B) { | ||
| this += ((key, value)) | ||
| } | ||
|
|
||
| override def apply(key: A): B = { | ||
| val value = internalMap.get(key) | ||
| if (value == null) throw new NoSuchElementException() | ||
| value._1 | ||
| } | ||
|
|
||
| override def filter(p: ((A, B)) => Boolean): Map[A, B] = { | ||
| internalMap.map(kv => (kv._1, kv._2._1)).filter(p) | ||
| } | ||
|
|
||
| override def empty: Map[A, B] = new TimeStampedHashMap[A, B]() | ||
|
|
||
| override def size(): Int = internalMap.size() | ||
|
|
||
| override def foreach[U](f: ((A, B)) => U): Unit = { | ||
| val iterator = internalMap.entrySet().iterator() | ||
| while(iterator.hasNext) { | ||
| val entry = iterator.next() | ||
| val kv = (entry.getKey, entry.getValue._1) | ||
| f(kv) | ||
| } | ||
| } | ||
|
|
||
| def cleanup(threshTime: Long) { | ||
| val iterator = internalMap.entrySet().iterator() | ||
| while(iterator.hasNext) { | ||
| val entry = iterator.next() | ||
| if (entry.getValue._2 < threshTime) { | ||
| iterator.remove() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def currentTime: Long = System.currentTimeMillis() | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters