rewrite MongoCache with Scaffeine
This commit is contained in:
parent
591b74f26d
commit
3af9dd4e2b
|
@ -200,7 +200,7 @@ object User extends LilaController {
|
||||||
def topWeek = Open { implicit ctx =>
|
def topWeek = Open { implicit ctx =>
|
||||||
negotiate(
|
negotiate(
|
||||||
html = notFound,
|
html = notFound,
|
||||||
api = _ => env.cached.topWeek(true).map { users =>
|
api = _ => env.cached.topWeek(()).map { users =>
|
||||||
Ok(Json toJson users.map(env.jsonView.lightPerfIsOnline))
|
Ok(Json toJson users.map(env.jsonView.lightPerfIsOnline))
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
|
@ -16,7 +16,7 @@ import play.api.libs.json._
|
||||||
|
|
||||||
final class Preload(
|
final class Preload(
|
||||||
tv: Tv,
|
tv: Tv,
|
||||||
leaderboard: Boolean => Fu[List[User.LightPerf]],
|
leaderboard: Unit => Fu[List[User.LightPerf]],
|
||||||
tourneyWinners: Fu[List[Winner]],
|
tourneyWinners: Fu[List[Winner]],
|
||||||
timelineEntries: String => Fu[List[Entry]],
|
timelineEntries: String => Fu[List[Entry]],
|
||||||
streamsOnAir: () => Fu[List[StreamOnAir]],
|
streamsOnAir: () => Fu[List[StreamOnAir]],
|
||||||
|
@ -40,7 +40,7 @@ final class Preload(
|
||||||
simuls zip
|
simuls zip
|
||||||
tv.getBestGame zip
|
tv.getBestGame zip
|
||||||
(ctx.userId ?? timelineEntries) zip
|
(ctx.userId ?? timelineEntries) zip
|
||||||
leaderboard(true) zip
|
leaderboard(()) zip
|
||||||
tourneyWinners zip
|
tourneyWinners zip
|
||||||
dailyPuzzle() zip
|
dailyPuzzle() zip
|
||||||
streamsOnAir() zip
|
streamsOnAir() zip
|
||||||
|
|
|
@ -11,12 +11,12 @@ final class Cached(
|
||||||
mongoCache: MongoCache.Builder,
|
mongoCache: MongoCache.Builder,
|
||||||
defaultTtl: FiniteDuration) {
|
defaultTtl: FiniteDuration) {
|
||||||
|
|
||||||
def nbImportedBy(userId: String): Fu[Int] = count(Query imported userId)
|
def nbImportedBy(userId: String): Fu[Int] = countCache(Query imported userId)
|
||||||
def clearNbImportedByCache(userId: String) = count.remove(Query imported userId)
|
def clearNbImportedByCache(userId: String) = countCache.remove(Query imported userId)
|
||||||
|
|
||||||
def nbPlaying(userId: String): Fu[Int] = countShortTtl(Query nowPlaying userId)
|
def nbPlaying(userId: String): Fu[Int] = countShortTtl(Query nowPlaying userId)
|
||||||
|
|
||||||
def nbTotal: Fu[Int] = count($empty)
|
def nbTotal: Fu[Int] = countCache($empty)
|
||||||
|
|
||||||
private implicit val userHandler = User.userBSONHandler
|
private implicit val userHandler = User.userBSONHandler
|
||||||
|
|
||||||
|
@ -29,7 +29,7 @@ final class Cached(
|
||||||
f = (o: Bdoc) => coll countSel o,
|
f = (o: Bdoc) => coll countSel o,
|
||||||
timeToLive = 5.seconds)
|
timeToLive = 5.seconds)
|
||||||
|
|
||||||
private val count = mongoCache(
|
private val countCache = mongoCache(
|
||||||
prefix = "game:count",
|
prefix = "game:count",
|
||||||
f = (o: Bdoc) => coll countSel o,
|
f = (o: Bdoc) => coll countSel o,
|
||||||
timeToLive = defaultTtl,
|
timeToLive = defaultTtl,
|
||||||
|
|
|
@ -1,40 +1,44 @@
|
||||||
package lila.memo
|
package lila.memo
|
||||||
|
|
||||||
|
import com.github.blemale.scaffeine.{ Scaffeine, Cache }
|
||||||
import org.joda.time.DateTime
|
import org.joda.time.DateTime
|
||||||
import reactivemongo.bson._
|
import reactivemongo.bson._
|
||||||
import reactivemongo.bson.Macros
|
import reactivemongo.bson.Macros
|
||||||
import scala.concurrent.duration._
|
import scala.concurrent.duration._
|
||||||
import spray.caching.{ LruCache, Cache }
|
|
||||||
|
|
||||||
import lila.db.BSON.BSONJodaDateTimeHandler
|
import lila.db.BSON.BSONJodaDateTimeHandler
|
||||||
import lila.db.dsl._
|
import lila.db.dsl._
|
||||||
|
|
||||||
final class MongoCache[K, V: MongoCache.Handler] private (
|
final class MongoCache[K, V: MongoCache.Handler] private (
|
||||||
prefix: String,
|
prefix: String,
|
||||||
expiresAt: () => DateTime,
|
cache: Cache[K, Fu[V]],
|
||||||
cache: Cache[V],
|
mongoExpiresAt: () => DateTime,
|
||||||
coll: Coll,
|
coll: Coll,
|
||||||
f: K => Fu[V],
|
f: K => Fu[V],
|
||||||
keyToString: K => String) {
|
keyToString: K => String) {
|
||||||
|
|
||||||
def apply(k: K): Fu[V] = cache(k) {
|
def apply(k: K): Fu[V] = cache.get(k, k =>
|
||||||
coll.find(select(k)).uno[Entry] flatMap {
|
coll.find(select(k)).uno[Entry] flatMap {
|
||||||
case None => f(k) flatMap { v =>
|
case None => f(k) flatMap { v =>
|
||||||
coll.insert(makeEntry(k, v)) recover
|
persist(k, v) inject v
|
||||||
lila.db.recoverDuplicateKey(_ => ()) inject v
|
|
||||||
}
|
}
|
||||||
case Some(entry) => fuccess(entry.v)
|
case Some(entry) => fuccess(entry.v)
|
||||||
|
})
|
||||||
|
|
||||||
|
def remove(k: K): Funit = {
|
||||||
|
val fut = f(k)
|
||||||
|
cache.put(k, fut)
|
||||||
|
fut flatMap { v =>
|
||||||
|
persist(k, v).void
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
def remove(k: K): Funit =
|
|
||||||
coll.remove(select(k)).void >>- (cache remove k)
|
|
||||||
|
|
||||||
private case class Entry(_id: String, v: V, e: DateTime)
|
private case class Entry(_id: String, v: V, e: DateTime)
|
||||||
|
|
||||||
private implicit val entryBSONHandler = Macros.handler[Entry]
|
private implicit val entryBSONHandler = Macros.handler[Entry]
|
||||||
|
|
||||||
private def makeEntry(k: K, v: V) = Entry(makeKey(k), v, expiresAt())
|
private def persist(k: K, v: V): Funit =
|
||||||
|
coll.insert(Entry(makeKey(k), v, mongoExpiresAt())).void recover lila.db.recoverDuplicateKey(_ => ())
|
||||||
|
|
||||||
private def makeKey(k: K) = s"$prefix:${keyToString(k)}"
|
private def makeKey(k: K) = s"$prefix:${keyToString(k)}"
|
||||||
|
|
||||||
|
@ -45,22 +49,28 @@ object MongoCache {
|
||||||
|
|
||||||
private type Handler[T] = BSONHandler[_ <: BSONValue, T]
|
private type Handler[T] = BSONHandler[_ <: BSONValue, T]
|
||||||
|
|
||||||
private def expiresAt(ttl: Duration)(): DateTime =
|
// expire in mongo 3 seconds before in heap,
|
||||||
DateTime.now plusSeconds ttl.toSeconds.toInt
|
// to make sure the mongo cache is cleared
|
||||||
|
// when the heap value expires
|
||||||
|
private def mongoExpiresAt(ttl: FiniteDuration): () => DateTime = {
|
||||||
|
val seconds = ttl.toSeconds.toInt - 3
|
||||||
|
() => DateTime.now plusSeconds seconds
|
||||||
|
}
|
||||||
|
|
||||||
final class Builder(coll: Coll) {
|
final class Builder(coll: Coll) {
|
||||||
|
|
||||||
def apply[K, V: Handler](
|
def apply[K, V: Handler](
|
||||||
prefix: String,
|
prefix: String,
|
||||||
f: K => Fu[V],
|
f: K => Fu[V],
|
||||||
maxCapacity: Int = 512,
|
maxCapacity: Int = 1024,
|
||||||
initialCapacity: Int = 64,
|
|
||||||
timeToLive: FiniteDuration,
|
timeToLive: FiniteDuration,
|
||||||
timeToLiveMongo: Option[FiniteDuration] = None,
|
|
||||||
keyToString: K => String): MongoCache[K, V] = new MongoCache[K, V](
|
keyToString: K => String): MongoCache[K, V] = new MongoCache[K, V](
|
||||||
prefix = prefix,
|
prefix = prefix,
|
||||||
expiresAt = expiresAt(timeToLiveMongo | timeToLive),
|
cache = Scaffeine()
|
||||||
cache = LruCache(maxCapacity, initialCapacity, timeToLive),
|
.expireAfterWrite(timeToLive)
|
||||||
|
.maximumSize(maxCapacity)
|
||||||
|
.build[K, Fu[V]],
|
||||||
|
mongoExpiresAt = mongoExpiresAt(timeToLive),
|
||||||
coll = coll,
|
coll = coll,
|
||||||
f = f,
|
f = f,
|
||||||
keyToString = keyToString)
|
keyToString = keyToString)
|
||||||
|
@ -68,11 +78,13 @@ object MongoCache {
|
||||||
def single[V: Handler](
|
def single[V: Handler](
|
||||||
prefix: String,
|
prefix: String,
|
||||||
f: => Fu[V],
|
f: => Fu[V],
|
||||||
timeToLive: FiniteDuration,
|
timeToLive: FiniteDuration) = new MongoCache[Unit, V](
|
||||||
timeToLiveMongo: Option[FiniteDuration] = None) = new MongoCache[Boolean, V](
|
|
||||||
prefix = prefix,
|
prefix = prefix,
|
||||||
expiresAt = expiresAt(timeToLiveMongo | timeToLive),
|
cache = Scaffeine()
|
||||||
cache = LruCache(timeToLive = timeToLive),
|
.expireAfterWrite(timeToLive)
|
||||||
|
.maximumSize(1)
|
||||||
|
.build[Unit, Fu[V]],
|
||||||
|
mongoExpiresAt = mongoExpiresAt(timeToLive),
|
||||||
coll = coll,
|
coll = coll,
|
||||||
f = _ => f,
|
f = _ => f,
|
||||||
keyToString = _.toString)
|
keyToString = _.toString)
|
||||||
|
|
|
@ -18,8 +18,7 @@ final class TournamentStatsApi(mongoCache: lila.memo.MongoCache.Builder) {
|
||||||
prefix = "tournament:stats",
|
prefix = "tournament:stats",
|
||||||
keyToString = identity,
|
keyToString = identity,
|
||||||
f = fetch,
|
f = fetch,
|
||||||
timeToLive = 10 minutes,
|
timeToLive = 10 minutes)
|
||||||
timeToLiveMongo = 90.days.some)
|
|
||||||
|
|
||||||
private def fetch(tournamentId: Tournament.ID): Fu[TournamentStats] = for {
|
private def fetch(tournamentId: Tournament.ID): Fu[TournamentStats] = for {
|
||||||
rating <- PlayerRepo.averageRating(tournamentId)
|
rating <- PlayerRepo.averageRating(tournamentId)
|
||||||
|
|
|
@ -116,12 +116,12 @@ final class WinnersApi(
|
||||||
f = fetchAll,
|
f = fetchAll,
|
||||||
timeToLive = ttl)
|
timeToLive = ttl)
|
||||||
|
|
||||||
def all: Fu[AllWinners] = allCache(true)
|
def all: Fu[AllWinners] = allCache(())
|
||||||
|
|
||||||
// because we read on secondaries, delay cache clear
|
// because we read on secondaries, delay cache clear
|
||||||
def clearCache(tour: Tournament) =
|
def clearCache(tour: Tournament) =
|
||||||
if (tour.schedule.exists(_.freq.isDailyOrBetter))
|
if (tour.schedule.exists(_.freq.isDailyOrBetter))
|
||||||
scheduler.once(5.seconds) { allCache.remove(true) }
|
scheduler.once(5.seconds) { allCache.remove(()) }
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue