Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/126#discussion_r10948794
--- Diff:
core/src/main/scala/org/apache/spark/util/TimeStampedHashMap.scala ---
@@ -19,108 +19,66 @@ package org.apache.spark.util
import java.util.concurrent.ConcurrentHashMap
-import scala.collection.JavaConversions
-import scala.collection.immutable
-import scala.collection.mutable.Map
-
import org.apache.spark.Logging
+private[util] case class TimeStampedValue[T](timestamp: Long, value: T)
+
/**
- * This is a custom implementation of scala.collection.mutable.Map which
stores the insertion
- * timestamp along with each key-value pair. If specified, the timestamp
of each pair can be
- * updated every time it is accessed. Key-value pairs whose timestamp are
older than a particular
- * threshold time can then be removed using the clearOldValues method.
This is intended to
- * be a drop-in replacement of scala.collection.mutable.HashMap.
+ * A map that stores the timestamp of when a key was inserted along with
the value. If specified,
+ * the timestamp of each pair can be updated every time it is accessed.
+ * Key-value pairs whose timestamps are older than a particular
+ * threshold time can then be removed using the clearOldValues method. It
exposes a
+ * scala.collection.mutable.Map interface to allow it to be a drop-in
replacement for Scala
+ * HashMaps.
+ *
+ * Internally, it uses a Java ConcurrentHashMap, so all operations on this
HashMap are thread-safe.
+ *
* @param updateTimeStampOnGet When enabled, the timestamp of a pair will
be
* updated when it is accessed
*/
-class TimeStampedHashMap[A, B](updateTimeStampOnGet: Boolean = false)
- extends Map[A, B]() with Logging {
- val internalMap = new ConcurrentHashMap[A, (B, Long)]()
-
- def get(key: A): Option[B] = {
- val value = internalMap.get(key)
- if (value != null && updateTimeStampOnGet) {
- internalMap.replace(key, value, (value._1, currentTime))
- }
- Option(value).map(_._1)
- }
-
- def iterator: Iterator[(A, B)] = {
- val jIterator = internalMap.entrySet().iterator()
- JavaConversions.asScalaIterator(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] = {
- val newMap = new TimeStampedHashMap[A, B]
- newMap.internalMap.putAll(this.internalMap)
- newMap.internalMap.remove(key)
- newMap
- }
+private[spark] class TimeStampedHashMap[A, B](updateTimeStampOnGet:
Boolean = false)
+ extends WrappedJavaHashMap[A, B, A, TimeStampedValue[B]] with Logging {
- override def += (kv: (A, B)): this.type = {
- internalMap.put(kv._1, (kv._2, currentTime))
- this
- }
+ protected[util] val internalJavaMap = new ConcurrentHashMap[A,
TimeStampedValue[B]]()
--- End diff --
Changed. Less confusing, that protected[util]
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
