core: Queue/Channel/Meter improvements and fixes

getkyo · Oct 18, 2024 · 5addfb5 · 5addfb5
1 parent 3e5682b
commit 5addfb5
Show file tree

Hide file tree

Showing 31 changed files with 1,418 additions and 806 deletions.
diff --git a/kyo-bench/src/main/scala/kyo/bench/Bench.scala b/kyo-bench/src/main/scala/kyo/bench/Bench.scala
@@ -48,7 +48,7 @@ object Bench:
 
     abstract class Base[A](expectedResult: A) extends Bench[A](expectedResult):
         def zioBench(): zio.UIO[A]
-        def kyoBenchFiber(): kyo.<[A, kyo.Async] = kyoBench()
+        def kyoBenchFiber(): kyo.<[A, kyo.Async & kyo.Abort[Throwable]] = kyoBench()
         def kyoBench(): kyo.<[A, kyo.IO]
         def catsBench(): cats.effect.IO[A]
     end Base

diff --git a/kyo-bench/src/main/scala/kyo/bench/CollectParBench.scala b/kyo-bench/src/main/scala/kyo/bench/CollectParBench.scala
@@ -10,7 +10,9 @@ class CollectParBench extends Bench.ForkOnly(Seq.fill(1000)(1)):
     override def kyoBenchFiber() =
         import kyo.*
 
-        Async.parallel(kyoTasks)
+        // TODO inference issue
+        val x = Async.parallel(kyoTasks)
+        x
     end kyoBenchFiber
 
     def catsBench() =

diff --git a/kyo-bench/src/main/scala/kyo/bench/EnqueueDequeueBench.scala b/kyo-bench/src/main/scala/kyo/bench/EnqueueDequeueBench.scala
@@ -22,7 +22,7 @@ class EnqueueDequeueBench extends Bench.ForkOnly(()):
 
         import kyo.Access
 
-        def loop(c: Channel[Unit], i: Int): Unit < Async =
+        def loop(c: Channel[Unit], i: Int): Unit < (Async & Abort[Closed]) =
             if i >= depth then
                 IO.unit
             else

diff --git a/kyo-bench/src/main/scala/kyo/bench/HttpClientBench.scala b/kyo-bench/src/main/scala/kyo/bench/HttpClientBench.scala
@@ -35,7 +35,7 @@ class HttpClientBench extends Bench.ForkOnly("pong"):
     override def kyoBenchFiber() =
         import kyo.*
 
-        Abort.run(Requests(_.get(kyoUrl))).map(_.getOrThrow)
+        Requests(_.get(kyoUrl))
     end kyoBenchFiber
 
     val zioUrl =

diff --git a/kyo-bench/src/main/scala/kyo/bench/HttpClientContentionBench.scala b/kyo-bench/src/main/scala/kyo/bench/HttpClientContentionBench.scala
@@ -37,7 +37,9 @@ class HttpClientContentionBench
     override def kyoBenchFiber() =
         import kyo.*
 
-        Async.parallel(Seq.fill(concurrency)(Abort.run(Requests(_.get(kyoUrl))).map(_.getOrThrow)))
+        // TODO inference issue
+        val x = Async.parallel(Seq.fill(concurrency)(Requests(_.get(kyoUrl))))
+        x
     end kyoBenchFiber
 
     val zioUrl =

diff --git a/kyo-bench/src/main/scala/kyo/bench/HttpClientRaceContentionBench.scala b/kyo-bench/src/main/scala/kyo/bench/HttpClientRaceContentionBench.scala
@@ -58,7 +58,9 @@ class HttpClientRaceContentionBench
     override def kyoBenchFiber() =
         import kyo.*
 
-        Async.race(Seq.fill(concurrency)(Requests.let(kyoClient)(Abort.run(Requests(_.get(kyoUrl)))).map(_.getOrThrow)))
+        // TODO inference issue
+        val x = Async.race(Seq.fill(concurrency)(Requests.let(kyoClient)(Requests(_.get(kyoUrl)))))
+        x
     end kyoBenchFiber
 
     val zioUrl =

diff --git a/kyo-bench/src/main/scala/kyo/bench/MtlBench.scala b/kyo-bench/src/main/scala/kyo/bench/MtlBench.scala
@@ -21,15 +21,13 @@ class MtlBench extends Bench(()):
                     _    <- Var.update((state: State) => state.copy(value = state.value + 1))
                 yield ()
             )
-        Abort.run[Throwable](
-            Var.run(State(2))(
-                Emit.run(
-                    Env.run(EnvValue("config"))(
-                        testKyo.andThen(Var.get[State])
-                    )
+        Var.run(State(2))(
+            Emit.run(
+                Env.run(EnvValue("config"))(
+                    testKyo.andThen(Var.get[State])
                 )
             )
-        ).eval
+        )
     end syncKyo
 
     @Benchmark

diff --git a/kyo-bench/src/main/scala/kyo/bench/PingPongBench.scala b/kyo-bench/src/main/scala/kyo/bench/PingPongBench.scala
@@ -37,11 +37,11 @@ class PingPongBench extends Bench.ForkOnly(()):
     override def kyoBenchFiber() =
         import kyo.*
 
-        def repeat[A](n: Int)(io: A < Async): A < Async =
+        def repeat[A](n: Int)(io: A < (Async & Abort[Closed])): A < (Async & Abort[Closed]) =
             if n <= 1 then io
             else io.flatMap(_ => repeat(n - 1)(io))
 
-        def iterate(promise: Promise[Nothing, Unit], n: Int): Unit < Async =
+        def iterate(promise: Promise[Nothing, Unit], n: Int): Unit < (Async & Abort[Closed]) =
             for
                 ref  <- AtomicInt.init(n)
                 chan <- Channel.init[Unit](1)
@@ -52,7 +52,7 @@ class PingPongBench extends Bench.ForkOnly(()):
                         n <- ref.decrementAndGet
                         _ <- if n == 0 then promise.complete(Result.unit).unit else IO.unit
                     yield ()
-                _ <- repeat(depth)(Async.run[Nothing, Unit, Any](effect))
+                _ <- repeat(depth)(Async.run[Closed, Unit, Any](effect))
             yield ()
 
         for

diff --git a/kyo-bench/src/main/scala/kyo/bench/ProducerConsumerBench.scala b/kyo-bench/src/main/scala/kyo/bench/ProducerConsumerBench.scala
@@ -30,7 +30,7 @@ class ProducerConsumerBench extends Bench.ForkOnly(()):
 
         import kyo.Access
 
-        def repeat[A](n: Int)(io: A < Async): A < Async =
+        def repeat[A](n: Int)(io: A < (Async & Abort[Closed])): A < (Async & Abort[Closed]) =
             if n <= 1 then io
             else io.flatMap(_ => repeat(n - 1)(io))
 

diff --git a/kyo-bench/src/main/scala/kyo/bench/SemaphoreBench.scala b/kyo-bench/src/main/scala/kyo/bench/SemaphoreBench.scala
@@ -22,7 +22,7 @@ class SemaphoreBench extends Bench.ForkOnly(()):
     override def kyoBenchFiber() =
         import kyo.*
 
-        def loop(s: Meter, i: Int): Unit < Async =
+        def loop(s: Meter, i: Int): Unit < (Async & Abort[Closed]) =
             if i >= depth then
                 IO.unit
             else

diff --git a/kyo-bench/src/main/scala/kyo/bench/SemaphoreContentionBench.scala b/kyo-bench/src/main/scala/kyo/bench/SemaphoreContentionBench.scala
@@ -40,7 +40,7 @@ class SemaphoreContentionBench extends Bench.ForkOnly(()):
             if n <= 1 then io
             else io.flatMap(_ => repeat(n - 1)(io))
 
-        def loop(sem: Meter, cdl: Latch, i: Int = 0): Unit < Async =
+        def loop(sem: Meter, cdl: Latch, i: Int = 0): Unit < (Async & Abort[Closed]) =
             if i >= depth then
                 cdl.release
             else

diff --git a/kyo-core/js/src/main/scala/kyo/queuesStubs.scala b/kyo-core/js/src/main/scala/kyo/queuesStubs.scala
@@ -1,9 +1,21 @@
 package org.jctools.queues
 
 import java.util.ArrayDeque
+import scala.annotation.tailrec
 
 class StubQueue[A](capacity: Int) extends ArrayDeque[A]:
     def isFull = size() >= capacity
+    def drain(f: A => Unit): Unit =
+        given [B]: CanEqual[B, B] = CanEqual.derived
+        @tailrec def loop(): Unit =
+            super.poll() match
+                case null =>
+                case value =>
+                    f(value)
+                    loop()
+        end loop
+        loop()
+    end drain
     override def offer(e: A): Boolean =
         !isFull && super.offer(e)
 end StubQueue
@@ -16,8 +28,8 @@ case class SpmcArrayQueue[A](capacity: Int) extends StubQueue[A](capacity)
 
 case class SpscArrayQueue[A](capacity: Int) extends StubQueue[A](capacity)
 
-case class MpmcUnboundedXaddArrayQueue[A](chunkSize: Int) extends ArrayDeque[A] {}
+case class MpmcUnboundedXaddArrayQueue[A](chunkSize: Int) extends StubQueue[A](Int.MaxValue)
 
-case class MpscUnboundedArrayQueue[A](chunkSize: Int) extends ArrayDeque[A] {}
+case class MpscUnboundedArrayQueue[A](chunkSize: Int) extends StubQueue[A](Int.MaxValue)
 
-case class SpscUnboundedArrayQueue[A](chunkSize: Int) extends ArrayDeque[A] {}
+case class SpscUnboundedArrayQueue[A](chunkSize: Int) extends StubQueue[A](Int.MaxValue)