Yet another polling system sketch

benchmarks/src/main/scala/cats/effect/benchmarks/SleepBenchmark.scala

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2020-2022 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cats.effect.benchmarks
+
+import cats.effect.IO
+import cats.effect.unsafe._
+import cats.syntax.all._
+
+import org.openjdk.jmh.annotations._
+
+import scala.concurrent.duration._
+
+import java.util.concurrent.TimeUnit
+
+/**
+ * To do comparative benchmarks between versions:
+ *
+ * benchmarks/run-benchmark SleepBenchmark
+ *
+ * This will generate results in `benchmarks/results`.
+ *
+ * Or to run the benchmark from within sbt:
+ *
+ * jmh:run -i 10 -wi 10 -f 2 -t 1 cats.effect.benchmarks.SleepBenchmark
+ *
+ * Which means "10 iterations", "10 warm-up iterations", "2 forks", "1 thread". Please note that
+ * benchmarks should be usually executed at least in 10 iterations (as a rule of thumb), but
+ * more is better.
+ */
+@State(Scope.Thread)
+@BenchmarkMode(Array(Mode.Throughput))
+@OutputTimeUnit(TimeUnit.MINUTES)
+class SleepBenchmark {
+
+  @Param(Array("10000"))
+  var size: Int = _
+
+  def sleepBenchmark(implicit runtime: IORuntime): Int = {
+    def fiber(i: Int): IO[Int] =
+      IO.sleep(1.nanosecond).flatMap { _ =>
+        IO(i).flatMap { j =>
+          IO.sleep(1.nanosecond).flatMap { _ =>
+            if (j > 1000)
+              IO.sleep(1.nanosecond).flatMap(_ => IO.pure(j))
+            else
+              IO.sleep(1.nanosecond).flatMap(_ => fiber(j + 1))
+          }
+        }
+      }
+
+    List
+      .range(0, size)
+      .traverse(fiber(_).start)
+      .flatMap(_.traverse(_.joinWithNever))
+      .map(_.sum)
+      .unsafeRunSync()
+  }
+
+  @Benchmark
+  def sleep(): Int = {
+    import cats.effect.unsafe.implicits.global
+    sleepBenchmark
+  }
+}

benchmarks/src/main/scala/cats/effect/benchmarks/SleepDrift.scala

@@ -23,6 +23,9 @@ import scala.concurrent.duration._
 
 object SleepDrift extends IOApp.Simple {
 
+  override val runtimeConfig =
+    super.runtimeConfig.copy(cpuStarvationCheckInitialDelay = Duration.Inf)
+
   val delayTwoMinutes = {
     def loop(n: Int): IO[Unit] = {
       if (n <= 0)

benchmarks/src/main/scala/cats/effect/benchmarks/WorkStealingBenchmark.scala

@@ -165,36 +165,24 @@ class WorkStealingBenchmark {
       (ExecutionContext.fromExecutor(executor), () => executor.shutdown())
     }
 
-    val (scheduler, schedDown) = {
-      val executor = Executors.newSingleThreadScheduledExecutor { r =>
-        val t = new Thread(r)
-        t.setName("io-scheduler")
-        t.setDaemon(true)
-        t.setPriority(Thread.MAX_PRIORITY)
-        t
-      }
-      (Scheduler.fromScheduledExecutor(executor), () => executor.shutdown())
-    }
-
-    val compute =
-      new WorkStealingThreadPool(
-        256,
-        "io-compute",
-        "io-blocker",
-        60.seconds,
-        _.printStackTrace())
+    val compute = new WorkStealingThreadPool(
+      256,
+      "io-compute",
+      "io-blocker",
+      60.seconds,
+      SleepSystem,
+      _.printStackTrace())
 
     val cancelationCheckThreshold =
       System.getProperty("cats.effect.cancelation.check.threshold", "512").toInt
 
     IORuntime(
       compute,
       blocking,
-      scheduler,
+      compute,
       () => {
         compute.shutdown()
         blockDown()
-        schedDown()
       },
       IORuntimeConfig(
         cancelationCheckThreshold,

build.sbt

@@ -41,7 +41,7 @@ ThisBuild / git.gitUncommittedChanges := {
   }
 }
 
-ThisBuild / tlBaseVersion := "3.4"
+ThisBuild / tlBaseVersion := "3.5"
 ThisBuild / tlUntaggedAreSnapshots := false
 
 ThisBuild / organization := "org.typelevel"

core/js-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala

@@ -18,6 +18,7 @@ package cats.effect
 package unsafe
 
 import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.FiniteDuration
 
 // Can you imagine a thread pool on JS? Have fun trying to extend or instantiate
 // this class. Unfortunately, due to the explicit branching, this type leaks
@@ -27,6 +28,13 @@ private[effect] sealed abstract class WorkStealingThreadPool private ()
   def execute(runnable: Runnable): Unit
   def reportFailure(cause: Throwable): Unit
   private[effect] def reschedule(runnable: Runnable): Unit
+  private[effect] def sleepInternal(
+      delay: FiniteDuration,
+      callback: Right[Nothing, Unit] => Unit): Runnable
+  private[effect] def sleep(
+      delay: FiniteDuration,
+      task: Runnable,
+      fallback: Scheduler): Runnable
   private[effect] def canExecuteBlockingCode(): Boolean
   private[unsafe] def liveTraces(): (
       Map[Runnable, Trace],

core/jvm/src/main/scala/cats/effect/IOApp.scala

@@ -165,6 +165,8 @@ trait IOApp {
    */
   protected def runtimeConfig: unsafe.IORuntimeConfig = unsafe.IORuntimeConfig()
 
+  protected def pollingSystem: unsafe.PollingSystem = unsafe.SleepSystem
+
   /**
    * Controls the number of worker threads which will be allocated to the compute pool in the
    * underlying runtime. In general, this should be no ''greater'' than the number of physical
@@ -317,22 +319,19 @@ trait IOApp {
         val (compute, compDown) =
           IORuntime.createWorkStealingComputeThreadPool(
             threads = computeWorkerThreadCount,
-            reportFailure = t => reportFailure(t).unsafeRunAndForgetWithoutCallback()(runtime))
+            reportFailure = t => reportFailure(t).unsafeRunAndForgetWithoutCallback()(runtime),
+            pollingSystem = pollingSystem)
 
         val (blocking, blockDown) =
           IORuntime.createDefaultBlockingExecutionContext()
 
-        val (scheduler, schedDown) =
-          IORuntime.createDefaultScheduler()
-
         IORuntime(
           compute,
           blocking,
-          scheduler,
+          compute,
           { () =>
             compDown()
             blockDown()
-            schedDown()
           },
           runtimeConfig)
       }

core/jvm/src/main/scala/cats/effect/unsafe/EventLoop.scala

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2020-2022 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cats.effect
+package unsafe
+
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
+import scala.reflect.ClassTag
+
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.atomic.AtomicBoolean
+
+trait EventLoop[+Registrar] extends ExecutionContext {
+
+  protected def registrarTag: ClassTag[_ <: Registrar]
+
+  def registrar(): Registrar
+
+}
+
+object EventLoop {
+  def unapply[R](loop: EventLoop[Any])(ct: ClassTag[R]): Option[EventLoop[R]] =
+    if (ct.runtimeClass.isAssignableFrom(loop.registrarTag.runtimeClass))
+      Some(loop.asInstanceOf[EventLoop[R]])
+    else
+      None
+
+  def fromPollingSystem(
+      name: String,
+      system: PollingSystem): (EventLoop[system.Poller], () => Unit) = {
+
+    val done = new AtomicBoolean(false)
+    val poller = system.makePoller()
+
+    val loop = new Thread(name) with EventLoop[system.Poller] with ExecutionContextExecutor {
+
+      val queue = new LinkedBlockingQueue[Runnable]
+
+      def registrarTag: ClassTag[_ <: system.Poller] = system.pollerTag
+
+      def registrar(): system.Poller = poller
+
+      def execute(command: Runnable): Unit = {
+        queue.put(command)
+        poller.interrupt(this)
+      }
+
+      def reportFailure(cause: Throwable): Unit = cause.printStackTrace()
+
+      override def run(): Unit = {
+        while (!done.get()) {
+          while (!queue.isEmpty()) queue.poll().run()
+          poller.poll(-1)
+        }
+      }
+    }
+
+    val cleanup = () => {
+      done.set(true)
+      poller.interrupt(loop)
+    }
+
+    (loop, cleanup)
+  }
+}

core/jvm/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala

@@ -33,12 +33,12 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type
 
   private[this] final val DefaultBlockerPrefix = "io-compute-blocker"
 
-  // The default compute thread pool on the JVM is now a work stealing thread pool.
   def createWorkStealingComputeThreadPool(
       threads: Int = Math.max(2, Runtime.getRuntime().availableProcessors()),
       threadPrefix: String = "io-compute",
       blockerThreadPrefix: String = DefaultBlockerPrefix,
       runtimeBlockingExpiration: Duration = 60.seconds,
+      pollingSystem: PollingSystem = SleepSystem,
       reportFailure: Throwable => Unit = _.printStackTrace())
       : (WorkStealingThreadPool, () => Unit) = {
     val threadPool =
@@ -47,6 +47,7 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type
         threadPrefix,
         blockerThreadPrefix,
         runtimeBlockingExpiration,
+        pollingSystem,
         reportFailure)
 
     val unregisterMBeans =
@@ -180,8 +181,8 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type
       installGlobal {
         val (compute, _) = createWorkStealingComputeThreadPool()
         val (blocking, _) = createDefaultBlockingExecutionContext()
-        val (scheduler, _) = createDefaultScheduler()
-        IORuntime(compute, blocking, scheduler, () => (), IORuntimeConfig())
+
+        IORuntime(compute, blocking, compute, () => (), IORuntimeConfig())
       }
     }
 

core/jvm/src/main/scala/cats/effect/unsafe/PollingSystem.scala

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2020-2022 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cats.effect
+package unsafe
+
+import scala.reflect.ClassTag
+
+abstract class PollingSystem {
+
+  type Poller <: AbstractPoller
+  def pollerTag: ClassTag[Poller]
+
+  def makePoller(): Poller
+
+  protected abstract class AbstractPoller {
+    def poll(nanos: Long): Unit
+    def interrupt(target: Thread): Unit
+  }
+}