From 7d566731002b1528151c11c7c6cab9388c2016c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Mon, 17 Aug 2026 18:18:27 +0300 Subject: [PATCH 1/2] backport unit-tests with defects from #369 --- .../com/evolution/scache/LoadingCache.scala | 16 +- .../evolution/scache/CacheDefectsSpec.scala | 344 ++++++++++++++++++ 2 files changed, 354 insertions(+), 6 deletions(-) create mode 100644 scache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scala diff --git a/scache/src/main/scala/com/evolution/scache/LoadingCache.scala b/scache/src/main/scala/com/evolution/scache/LoadingCache.scala index c011fd29..8a29cfeb 100644 --- a/scache/src/main/scala/com/evolution/scache/LoadingCache.scala +++ b/scache/src/main/scala/com/evolution/scache/LoadingCache.scala @@ -11,15 +11,19 @@ import com.evolutiongaming.catshelper.ParallelHelper.* private[scache] object LoadingCache { /** - * Maximum number of CAS retry attempts before giving up. This is a safety net against infinite - * spinning under extreme contention. - */ - /** - * Maximum number of CAS retry attempts on the outer map before giving up. Inner entry-level CAS - * loops are unbounded as they always make progress. + * Maximum number of CAS retry attempts on the outer map before giving up. This is a safety net + * against infinite spinning under extreme contention. Inner entry-level CAS loops are unbounded + * as they always make progress. */ private val MaxRetries: Int = 10000 + def of[F[_]: Concurrent, K, V]: Resource[F, Cache[F, K, V]] = { + for { + ref <- Ref[F].of(EntryRefs.empty[F, K, V]).toResource + cache <- of(ref) + } yield cache + } + def of[F[_]: Concurrent, K, V]( map: EntryRefs[F, K, V], ): Resource[F, Cache[F, K, V]] = { diff --git a/scache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scala b/scache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scala new file mode 100644 index 00000000..053414f1 --- /dev/null +++ b/scache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scala @@ -0,0 +1,344 @@ +package com.evolution.scache + +import cats.data.State +import cats.effect.* +import cats.syntax.all.* +import com.evolution.scache.IOSuite.* +import com.evolution.scache.LoadingCache.EntryRefs +import org.scalatest.funsuite.AsyncFunSuite +import org.scalatest.matchers.should.Matchers + +import scala.concurrent.duration.* + +/** + * Asserts the expected behavior for four defects originally present in LoadingCache / + * ExpiringCache, fixed by rebuilding the cache on [[cats.effect.std.MapRef]] (fix in #369): + * - claim 1: loads are cancelable and cancellation cleans up the `Loading` entry; + * - claim 2: entries stuck in `Loading` state are evicted by the expiration routine; + * - claim 3: waiters on a `Loading` entry are unblocked when the load is cancelled; + * - claim 4: operations on distinct keys are independent, no shared-state CAS retries. + * + * Every stuck load is modeled with a `gate` Deferred instead of `IO.never` and released in a + * `guarantee`, so a failed assertion produces a clean test failure instead of hanging resource + * finalizers (`clear` waits on Loading entries). + */ +class CacheDefectsSpec extends AsyncFunSuite with Matchers { + + // TODO fix the cancellation of loading effect + test("claim 1: cancelled load must not block the key for subsequent calls") { + val io = for { + entryMap <- Ref.of[IO, EntryRefs[IO, Int, Int]](EntryRefs.empty) + cache = LoadingCache(entryMap) + started <- Deferred[IO, Unit] + gate <- Deferred[IO, Unit] + loader <- cache.getOrUpdate(0) { started.complete(()) *> gate.get.as(1) }.start + _ <- started.get + cancelling <- loader.cancel.start + _ = pending + result <- { + for { + cancelled <- cancelling.join.timeout(500.millis) + _ = cancelled should matchPattern { case Outcome.Succeeded(_) => } + present <- cache.get(0) + _ = present shouldEqual none + second <- cache.getOrUpdate(0)(2.pure[IO]).timeout(500.millis) + _ = second shouldEqual 2 + } yield () + }.guarantee { gate.complete(()) *> cancelling.join.void } + } yield result + io.run() + } + + // TODO introduce the `loadingTimeout` and functionality to remove entries, which are stuck at Loading state for too long + test("claim 2: expiration cleanup must evict entries stuck in Loading state") { + val config = ExpiringCache.Config[IO, Int, Int]( + expireAfterRead = 100.millis, + ) + val io = ExpiringCache.of[IO, Int, Int](config).use { cache => + for { + started <- Deferred[IO, Unit] + gate <- Deferred[IO, Unit] + // Attempted, see the unit-test for eviction of stuck `Loading` entry in `claim 3` case + loader <- cache.getOrUpdate(0) { started.complete(()) *> gate.get.as(1) }.attempt.start + _ <- started.get + result <- { + for { + _ <- cache.put(1, 1).flatten + _ <- IO.sleep(200.millis) // sleep longer than `expireAfterRead` (and `loadingTimeout`) + // Control: an ordinary value of the same age is gone, so the cleanup did run. + control <- cache.contains(1) + _ = control shouldEqual false + poisoned <- cache.contains(0) + _ = pendingUntilFixed { poisoned shouldEqual false; () } + second <- cache.getOrUpdate(0)(2.pure[IO]).timeout(500.millis) + _ = second shouldEqual 2 + } yield () + }.guarantee { gate.complete(()) *> loader.join.void } + } yield result + } + io.run() + } + + // TODO fix the cancellation of loading effect + test("claim 3: cancelling a load must unblock the fibers waiting on it") { + val io = for { + entryMap <- Ref.of[IO, EntryRefs[IO, Int, Int]](EntryRefs.empty) + cache = LoadingCache(entryMap) + started <- Deferred[IO, Unit] + gate <- Deferred[IO, Unit] + loader <- cache.getOrUpdate(0) { started.complete(()) *> gate.get.as(1) }.start + _ <- started.get + // Attempted, so that the failure this test is after is observed as a value: a fiber left to + // end in `Errored` reports the error to the runtime as unhandled the moment it finishes, + // which here races with the `join` below. + waiter <- cache.getOrUpdate(0)(99.pure[IO]).attempt.start + _ <- IO.sleep(100.millis) // simulate passage of time + cancelling <- loader.cancel.start + _ = pending // TODO the `waiter.joinWithNever` never gets joined and fails on timeout + result <- { + for { + outcome <- waiter.joinWithNever.timeout(500.millis) + _ = outcome should matchPattern { case Left(CancelledError) => } + present <- cache.get(0) + _ = present shouldEqual none + } yield () + }.guarantee { gate.complete(()) *> cancelling.join.void } + } yield result + io.run() + } + + // TODO fix the cancellation of loading effect + test("claim 3: cancelling a load removed while loading must unblock the fibers waiting on it") { + val io = for { + entryMap <- Ref.of[IO, EntryRefs[IO, Int, Int]](EntryRefs.empty) + cache = LoadingCache(entryMap) + started <- Deferred[IO, Unit] + gate <- Deferred[IO, Unit] + loader <- cache.getOrUpdate(0) { started.complete(()) *> gate.get.as(1) }.start + _ <- started.get + // Attempted, so that the failure this test is after is observed as a value: a fiber left to + // end in `Errored` reports the error to the runtime as unhandled the moment it finishes, + // which here races with the `join` below. + waiter <- cache.getOrUpdate(0)(99.pure[IO]).attempt.start + _ <- IO.sleep(100.millis) // simulate passage of time + // The entry stops being the loader's, so only the loader itself can still unblock the waiter. + _ <- cache.remove(0).flatten + cancelling <- loader.cancel.start + _ = pending // TODO `waiter.joinWithNever` never gets joined and fails on timeout + result <- { + for { + outcome <- waiter.joinWithNever.timeout(500.millis) + _ = outcome should matchPattern { case Left(CancelledError) => } + } yield () + }.guarantee { gate.complete(()) *> cancelling.join.void } + } yield result + io.run() + } + + test("claim 4: getOrUpdate must complete under sustained writes of unrelated keys") { + val io = LoadingCache.of[IO, Int, Int].use { cache => + for { + writers <- (1 to 8) + .toList + .traverse { key => + (cache.put(key, key).flatten *> cache.remove(key).flatten) + .foreverM + .start + } + _ <- IO.sleep(100.millis) + result <- { + for { + value <- cache.getOrUpdate(0)(1.pure[IO]).timeout(5.seconds) + _ = value shouldEqual 1 + } yield () + }.guarantee { writers.parTraverse_ { _.cancel } } + } yield result + } + io.run(timeout = 30.seconds) + } + + // TODO this is not possible to fix in current version - only rewrite to `MapRef` usage can resolve this! + test("claim 4 mechanism: insert of an unrelated key must not force a retry of getOrUpdate") { + val io = for { + underlying <- Ref.of[IO, EntryRefs[IO, Int, Int]](EntryRefs.empty) + attempts <- Ref[IO].of(0) + noise = insertUnrelated(underlying, 1) + cache = LoadingCache(intercepted(underlying, noise, attempts.some)) + _ = pending // TODO next line fails with `Cache CAS retry limit (10000) exceeded. This indicates extreme contention.` + value <- cache.getOrUpdate(0)(1.pure[IO]) + _ = value shouldEqual 1 + attempts <- attempts.get + _ = attempts shouldEqual 1 + keys <- cache.keys + _ = keys shouldEqual Set(0, 1) + } yield () + io.run() + } + + // TODO this is not possible to fix in current version - only rewrite to `MapRef` usage can resolve this! + test("claim 4 mechanism: parallel getOrUpdate of distinct keys causes no insert retries") { + val io = for { + underlying <- Ref.of[IO, EntryRefs[IO, Int, Int]](EntryRefs.empty) + attempts <- Ref[IO].of(0) + cache = LoadingCache(intercepted(underlying, IO.unit, attempts.some)) + _ <- (0 until 10000).toList.parTraverse { key => cache.getOrUpdate(key)(key.pure[IO]) } + size <- cache.size + _ = size shouldEqual 10000 + attempts <- attempts.get + _ = pendingUntilFixed { + attempts shouldEqual 10000 + () + } + } yield () + io.run(timeout = 30.seconds) + } + + // TODO current `EntryState.Loading` doesn't track since when element is loading, possibly could be fixed + test("evicting a stuck Loading entry unblocks fibers waiting on it") { + pending // TODO remove + val config = ExpiringCache.Config[IO, Int, Int]( + expireAfterRead = 1.minute, + ) + val io = ExpiringCache.of[IO, Int, Int](config).use { cache => + for { + started <- Deferred[IO, Unit] + gate <- Deferred[IO, Unit] + // Attempted, because the eviction makes this load fail too, and a fiber left to end in + // `Errored` state, reports the error to the runtime as unhandled the moment it finishes, before + // the `join` below gets to observe it. + loader <- cache.getOrUpdate(0) { + started.complete(()) *> gate.get.as(1) + }.attempt.start + _ <- started.get + waiter <- cache.getOrUpdate(0)(99.pure[IO]).attempt.start + result <- { + for { + // TODO `waiter.joinWithNever` never gets joined and fails on timeout + outcome <- waiter.joinWithNever.timeout(2.seconds) + _ = outcome should matchPattern { case Left(_) => } // TODO MR catch real error + _ <- gate.complete(()) + // The fiber whose load was evicted learns about it as well. + evicted <- loader.joinWithNever.timeout(2.seconds) + _ = evicted should matchPattern { case Left(_) => } // TODO MR catch real error + } yield () + }.guarantee { + gate.complete(()).attempt *> loader.join.void + } + } yield result + } + io.run() + } + + test("a new load generation does not inherit the previous generation's stuck-timer") { + val config = ExpiringCache.Config[IO, Int, Int]( + expireAfterRead = 1.minute, + ) + val io = ExpiringCache.of[IO, Int, Int](config).use { cache => + for { + started1 <- Deferred[IO, Unit] + gate1 <- Deferred[IO, Unit] + loader1 <- cache.getOrUpdate(0) { started1.complete(()) *> gate1.get.as(1) }.start + _ <- started1.get + _ <- IO.sleep(150.millis) + _ <- gate1.complete(()) + _ <- loader1.join + _ <- cache.remove(0).flatten + started2 <- Deferred[IO, Unit] + gate2 <- Deferred[IO, Unit] + loader2 <- cache.getOrUpdate(0) { started2.complete(()) *> gate2.get.as(2) }.start + _ <- started2.get + result <- { + for { + _ <- IO.sleep(150.millis) + present <- cache.contains(0) + _ = present shouldEqual true + } yield () + }.guarantee { gate2.complete(()) *> loader2.join.void } + } yield result + } + io.run() + } + + test("cancellation races neither poison the key nor leak releases") { + val io = for { + entryMap <- Ref.of[IO, EntryRefs[IO, Int, Int]](EntryRefs.empty) + cache = LoadingCache(entryMap) + balance <- Ref[IO].of(0) + _ <- (1 to 500).toList.traverse_ { i => + for { + fiber <- cache.getOrUpdate1(0) { balance.update { _ + 1 }.as((i, i, balance.update { _ - 1 }.some)) }.start + _ <- fiber.cancel.start + _ <- fiber.join + // The key must be usable right away, holding either the value of the load that made it + // in before the cancellation, or the one we compute here. + value <- cache.getOrUpdate(0)((-1).pure[IO]).timeout(1.second) + _ = value should (equal(i) or equal(-1)) + _ <- cache.remove(0).flatten + } yield () + } + // Releases of values nobody asked about are started in the background, so the balance is + // settled shortly after the last removal rather than at the moment of it. + _ <- (IO.sleep(10.millis) *> balance.get).iterateUntil { _ == 0 }.timeout(3.seconds) + } yield () + io.run(timeout = 60.seconds) + } + + private def insertUnrelated(underlying: Ref[IO, EntryRefs[IO, Int, Int]], key: Int): IO[Unit] = { + for { + entryRef <- Ref[IO].of[LoadingCache.EntryState[IO, Int]]( + LoadingCache.EntryState.Value(LoadingCache.Entry(key, none)), + ) + _ <- underlying.modify { refs => + (refs.updated(key, entryRef), ()) + } + } yield () + } + + /** + * `underlying` with every per-key `Ref` wrapped, so that each attempt of the cache to modify the + * mapping first runs `noise`, a write of some other key, and then is counted in `attempts`. + * + * That gives the deterministic version of what the `claim 4` test does with background fibers: an + * unrelated write is guaranteed to land between reading and writing the mapping, i.e. exactly + * where the shared `Ref[F, Map[K, EntryRef]]` used to lose its CAS. With a per-key `Ref` the + * attempt still succeeds, so the count stays at one attempt per insert. + * + * `noise` writes through `underlying` directly and is not counted. + */ + private def intercepted( + underlying: Ref[IO, EntryRefs[IO, Int, Int]], + noise: IO[Unit], + attempts: Option[Ref[IO, Int]], + ): Ref[IO, EntryRefs[IO, Int, Int]] = { + val observe = noise *> attempts.foldMapM { _.update { _ + 1 } } + + new Ref[IO, EntryRefs[IO, Int, Int]] { + override def access: IO[(EntryRefs[IO, Int, Int], EntryRefs[IO, Int, Int] => IO[Boolean])] = + underlying.access.map { case (a, set) => (a, a1 => observe *> set(a1)) } + + override def tryUpdate(f: EntryRefs[IO, Int, Int] => EntryRefs[IO, Int, Int]): IO[Boolean] = + observe *> underlying.tryUpdate(f) + + override def tryModify[B](f: EntryRefs[IO, Int, Int] => (EntryRefs[IO, Int, Int], B)): IO[Option[B]] = + observe *> underlying.tryModify(f) + + override def update(f: EntryRefs[IO, Int, Int] => EntryRefs[IO, Int, Int]): IO[Unit] = + observe *> underlying.update(f) + + override def modify[B](f: EntryRefs[IO, Int, Int] => (EntryRefs[IO, Int, Int], B)): IO[B] = + observe *> underlying.modify(f) + + override def tryModifyState[B](state: State[EntryRefs[IO, Int, Int], B]): IO[Option[B]] = + observe *> underlying.tryModifyState(state) + + override def modifyState[B](state: State[EntryRefs[IO, Int, Int], B]): IO[B] = + observe *> underlying.modifyState(state) + + override def set(a: EntryRefs[IO, Int, Int]): IO[Unit] = + observe *> underlying.set(a) + + override def get: IO[EntryRefs[IO, Int, Int]] = + observe *> underlying.get + } + } +} From 4592696cec5adc31ebc73954a84cb49998ac71d8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Mon, 17 Aug 2026 21:10:58 +0300 Subject: [PATCH 2/2] add benchmark for cancellation of nested caches --- .../scache/bench/CancellationBenchmark.scala | 272 ++++++++++++++++++ build.sbt | 16 +- project/plugins.sbt | 2 + 3 files changed, 289 insertions(+), 1 deletion(-) create mode 100644 benchmark/src/main/scala/com/evolution/scache/bench/CancellationBenchmark.scala diff --git a/benchmark/src/main/scala/com/evolution/scache/bench/CancellationBenchmark.scala b/benchmark/src/main/scala/com/evolution/scache/bench/CancellationBenchmark.scala new file mode 100644 index 00000000..131e03e9 --- /dev/null +++ b/benchmark/src/main/scala/com/evolution/scache/bench/CancellationBenchmark.scala @@ -0,0 +1,272 @@ +package com.evolution.scache.bench + +import CancellationBenchmark.* +import cats.data.{NonEmptyList, NonEmptyMap} +import cats.effect.unsafe.implicits.global +import cats.effect.{ExitCode, FiberIO, IO, IOApp, Resource} +import cats.syntax.all.* +import com.evolution.scache +import com.evolution.scache.ExpiringCache +import com.evolutiongaming.catshelper.ParallelHelper.* +import org.openjdk.jmh.annotations.{ + Benchmark, + BenchmarkMode, + Fork, + Level, + Measurement, + Mode, + OutputTimeUnit, + Scope, + Setup, + State, + TearDown, + Threads, + Warmup, +} + +import java.util.concurrent.TimeUnit +import scala.collection.immutable.SortedMap +import scala.concurrent.duration.* +import scala.util.Random +import scala.util.control.NoStackTrace + +/** + * To run benchmarks: {{{sbt benchmark/Jmh/run com.evolution.scache.bench.CancellationBenchmark}}} + * + * Results on Apple M3 Max using Oracle JDK 17.0.11 + * + * {{{ + * ==original== + * Benchmark Mode Cnt Score Error Units + * CancellationBenchmark.cancelSingleShot ss 10000 691.013 ± 5.735 us/op + * }}} + * + * {{{ + * ==MapRef== + * Benchmark Mode Cnt Score Error Units + * CancellationBenchmark.cancelSingleShot ss 10000 669.808 ± 6.001 us/op + * }}} + */ +@Fork(value = 1, jvmArgsAppend = Array("-Xmx4g")) +@Threads(1) +class CancellationBenchmark { + + @Benchmark + @BenchmarkMode(Array(Mode.SingleShotTime)) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + @Warmup(iterations = 200, timeUnit = TimeUnit.SECONDS) + @Measurement(iterations = 10000, timeUnit = TimeUnit.SECONDS) + def cancelSingleShot(state: CancellationState): Unit = state.subject() + +// @Benchmark +// @BenchmarkMode(Array(Mode.Throughput)) +// @OutputTimeUnit(TimeUnit.SECONDS) +// @Warmup(iterations = 1, time = 20, timeUnit = TimeUnit.SECONDS) +// @Measurement(iterations = 1, time = 180, timeUnit = TimeUnit.SECONDS) +// def cancelThroughput(state: CancellationState): Unit = state.subject() +} + +@State(Scope.Benchmark) +class CancellationState { + + private var release: IO[Unit] = IO.unit + private var cancellationFiber: FiberIO[Unit] = null + + @Setup(Level.Invocation) + def setupInvocation(): Unit = { + val consumer: Resource[IO, ConsumerOf[IO]] = ConsumerOf.make + val topicFlow: Resource[IO, TopicFlow] = TopicFlow.make(CacheOf()) + + val (flow, release) = topicFlow.allocated.unsafeRunSync() + this.release = release + + val load = consumer.use { _.poll.flatMap(flow.apply) } + this.cancellationFiber = { + val loading = for { + loadingFiber <- load.start + _ <- IO.sleep(10.milliseconds) + // _ <- loadingFiber.cancel + } yield loadingFiber + loading.unsafeRunSync() + } + } + + def subject(): Unit = + cancellationFiber.cancel.unsafeRunSync() + + @TearDown(Level.Invocation) + def tearDownInvocation(): Unit = + release.unsafeRunSync() +} + +object Test extends IOApp { + + override def run(args: List[String]): IO[ExitCode] = { + val consumer: Resource[IO, ConsumerOf[IO]] = ConsumerOf.make + val topicFlow: Resource[IO, TopicFlow] = TopicFlow.make(CacheOf()) + + for { + _ <- IO.unit + load = (consumer, topicFlow).tupled.use { case (consumer, topicFlow) => + consumer.poll.flatMap(topicFlow.apply) + .timed + .flatMap { case (duration, result) => + IO(println(s"poll done in ${ duration.toMicros }")).as(result) + } + } + .timed + .flatMap { case (duration, result) => + IO(println(s"all done in ${ duration.toMicros }")).as(result) + } + cancellation = + for { + loadingFiber <- load.start + _ <- IO.sleep(10.milliseconds) + _ <- loadingFiber.cancel + .timed + .flatMap { case (duration, result) => + IO(println(f"cancel done in ${ duration.toMicros }%08d μs")).as(result) + } + } yield () + _ <- cancellation.replicateA(3) + } yield ExitCode.Success + } +} + +object CancellationBenchmark { + type Partition = Int + val error: Throwable = new RuntimeException("ba-bam!") with NoStackTrace +} + +case class Record(key: String, value: Int) + +object Poll { + def build: NonEmptyMap[Partition, NonEmptyList[Record]] = { + val numberOfKeys = 250 // Random.nextInt(250) + 1 + val numberOfRecords = 1000 // Random.nextInt(1000) + 1 + val numberOfPartitions = 16 // Random.nextInt(16) + 1 + + val keys = (0 until numberOfKeys).map(i => f"key-$i%06d") + val records = (0 until numberOfRecords).map(Record(keys(Random.nextInt(numberOfKeys)), _)) + + NonEmptyMap.fromMapUnsafe { + SortedMap.from { + records.groupBy(_.value % numberOfPartitions).map { case (partition, records) => + partition -> NonEmptyList.fromListUnsafe(records.toList) + } + } + } + } +} + +trait ConsumerOf[F[_]] { + def poll: F[NonEmptyMap[Partition, NonEmptyList[Record]]] +} + +object ConsumerOf { + def make: Resource[IO, ConsumerOf[IO]] = { + val consumerOf = new ConsumerOf[IO] { + override def poll: IO[NonEmptyMap[Partition, NonEmptyList[Record]]] = + IO.pure(Poll.build) + } + Resource.pure(consumerOf) + } +} + +trait CacheOf[F[_]] { + def make[K, V]: Resource[F, Cache[F, K, V]] +} + +trait Cache[F[_], K, V] { + def getOrUpdate(key: K)(value: => Resource[F, V]): F[V] +} + +object CacheOf { + def apply(): CacheOf[IO] = { + class Main + new Main with CacheOf[IO] { + def make[K, V]: Resource[IO, Cache[IO, K, V]] = { + val config = ExpiringCache.Config[IO, K, V](expireAfterRead = 1.minute) + for { + cache <- scache.Cache.expiring(config) + } yield { + new Cache[IO, K, V] { + def getOrUpdate(key: K)(value: => Resource[IO, V]): IO[V] = { + cache.getOrUpdateResource(key) { value } + } + } + } + } + } + } +} + +object ReplicateRecords { + def process(record: Record): IO[NonEmptyList[Int]] = + IO.sleep(50.milliseconds) *> + // if (Random.nextInt(chanceToFailOneIn) == 0) IO.raiseError(error) + // else IO(NonEmptyList.one(1)) + IO(NonEmptyList.one(1 + record.value - record.value)) +} + +trait TopicFlow { + def apply(records: NonEmptyMap[Partition, NonEmptyList[Record]]): IO[Unit] +} + +// fill the cache +object TopicFlow { + trait PartitionFlow { + def apply(records: NonEmptyList[Record]): IO[Unit] + } + + trait KeyFlow { + def apply(records: NonEmptyList[Record]): IO[Int] + } + + def make(cacheOf: CacheOf[IO]): Resource[IO, TopicFlow] = + cacheOf.make[Partition, PartitionFlow].map { partitionCache => + new TopicFlow { + override def apply(records: NonEmptyMap[Partition, NonEmptyList[Record]]): IO[Unit] = { + for { + result <- records.parFoldMap1 { + case (partition, records) => + // val replicatePartition = + for { + partitionFlow <- partitionCache.getOrUpdate(partition) { + for { + keyCache <- cacheOf.make[String, KeyFlow] + } yield new PartitionFlow { + override def apply(records: NonEmptyList[Record]): IO[Unit] = { + records + .groupBy { _.key } + .parFoldMap1 { + case (key, records) => + // val replicateRecords = + for { + keyFlow <- keyCache.getOrUpdate(key) { + Resource.pure { + new KeyFlow { + override def apply(records: NonEmptyList[Record]): IO[Int] = { + records + .flatTraverse(ReplicateRecords.process) + .flatTap(_ => IO(println(s"$key: done"))) + .as(records.size) + } + } + } + } + _ <- keyFlow(records) + } yield () + } + } + } + } + // _ = println(s"partitionFlow: $partitionFlow ($partition, records: ${ records.size })") + result <- partitionFlow(records) + } yield result + } + } yield result + } + } + } +} diff --git a/build.sbt b/build.sbt index 59b6bf3c..7870a05c 100644 --- a/build.sbt +++ b/build.sbt @@ -67,7 +67,7 @@ lazy val root = (project in file(".")) publish / skip := true, publishArtifact := false, ) - .aggregate(`cache-adt`, scache) + .aggregate(`cache-adt`, scache, benchmark) lazy val `cache-adt` = (project in file("cache-adt")) .settings(commonSettings) @@ -93,6 +93,20 @@ lazy val scache = (project in file("scache")) ) .dependsOn(`cache-adt`) +lazy val benchmark = (project in file("benchmark")) + .enablePlugins(JmhPlugin) + .dependsOn(scache) + .settings(commonSettings) + .settings( + name := "scache-benchmark", + description := "JMH benchmarks for scache", + publish / skip := true, + publishArtifact := false, + versionPolicyCheck / skip := true, + versionPolicyReportDependencyIssues / skip := true, + coverageEnabled := false, + ) + addCommandAlias("fmt", "+scalafmtRepo") addCommandAlias("check", "+all versionPolicyCheck Compile/doc scalafmtCheckRepo") addCommandAlias("build", "all test package") diff --git a/project/plugins.sbt b/project/plugins.sbt index c4bd3e11..2d3383c1 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -9,3 +9,5 @@ addSbtPlugin("com.evolution" % "sbt-scalac-opts-plugin" % "0.2.0") addSbtPlugin("com.evolution" % "sbt-artifactory-plugin" % "0.1.2") addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.6.2") + +addSbtPlugin("pl.project13.scala" % "sbt-jmh" % "0.4.8")