Pluggable caches in FinalFlatMap #396
Conversation
…dd options for changing them. Both caches dump based on time
| private lazy val runtime = Runtime.getRuntime | ||
| private def memoryWaterMark = { | ||
| val used = (runtime.totalMemory - runtime.freeMemory).toDouble / runtime.maxMemory | ||
| used > 0.8 |
There was a problem hiding this comment.
seems like a magic number. Config? val?
High memory water mark when to flush and when to crush down a value
| private def keySpaceTooBig = keyMap.size > cacheSize | ||
|
|
||
| override def cleanup = { | ||
| Future { |
There was a problem hiding this comment.
Shouldn't this use the pool?
There was a problem hiding this comment.
I'm not sure how the pool could use the pool to shutdown? it would be recursive right?
There was a problem hiding this comment.
Is it not safe to call shutdown from a thread that is created by the pool? If not, how do we ensure that?
Not a big deal, but if you don't want to tackle, add an issue, a link to it here and a comment explaining the concern about correctness and using the pool to shut itself down.
There was a problem hiding this comment.
You could call shutdown, but this currently waits to ensure its shutdown before returning -- if we drop that part (awaitShutdown). Then we could do it inside the pool?
| @@ -49,4 +49,9 @@ object Constants { | |||
| val DEFAULT_MAX_WAITING_FUTURES = MaxWaitingFutures(10) | |||
| val DEFAULT_MAX_FUTURE_WAIT_TIME = MaxFutureWaitTime(Duration.fromSeconds(60)) | |||
| val DEFAULT_FM_PREFER_LOCAL_DEPENDENCY = PreferLocalDependency(false) | |||
| val DEFAULT_FLUSH_FREQUENCY = FlushFrequency(Duration.fromSeconds(10)) | |||
| val DEFAULT_USE_ASYNC_CACHE = UseAsyncCache(false) | |||
| val DEFAULT_ASYNCPOOLSIZE = AsyncPoolSize(10) | |||
There was a problem hiding this comment.
What about: Runtime.getRuntime().availableProcessors() as the default pool size?
http://stackoverflow.com/questions/4759570/finding-number-of-cores-in-java
|
@johnynek poke |
| merge(key, List(monoid.sumOption(dataCP).get)) | ||
| } | ||
| } else { | ||
| if(keyMap.putIfAbsent(key, oldQueue) != null) { // Not there |
There was a problem hiding this comment.
more comments here. So this branch happens when there IS a value for this key, right? Not there is confusing to me.
| import scala.collection.mutable.SynchronizedQueue | ||
| import com.twitter.summingbird.online.option.{ValueCombinerCacheSize, AsyncPoolSize, FlushFrequency, SoftMemoryFlushPercent} | ||
| import java.util.concurrent.{Executors, ConcurrentHashMap, TimeUnit} | ||
| import scala.collection.JavaConversions._ |
There was a problem hiding this comment.
I don't like these. Can we replace with JavaConverters._ and use the .asJava .asScala syntax. Implicit conversions of this kind bite us.
| lastDump = System.currentTimeMillis | ||
| startKeyset.flatMap{case k => | ||
| Option(keyMap.remove(k)).map(_.toSeq).flatMap(monoid.sumOption(_)).map((k, _)) | ||
| }.toMap |
There was a problem hiding this comment.
(breakOut) magic is supposedly faster here:
import scala.collection.breakOut
flatMap { k => ... }(breakOut)
http://stackoverflow.com/questions/1715681/scala-2-8-breakout
Pluggable caches in FinalFlatMap
|
Thanks for pushing this through. Next up, see if we can get it upping Ashu's throughput? |
Added an synchronous cache, pulled cache logic out thats used by the FFM. Allow ticks to trigger flushing.