Add support for Native multithreaded execution

.github/workflows/ci.yml

@@ -111,12 +111,37 @@ jobs:
             ci: ciChrome
           - ci: ciNative
             java: temurin@11
+            os: ubuntu-latest
+          - ci: ciNative
+            java: temurin@11
+            os: windows-latest
+          - ci: ciNative
+            java: temurin@11
+            os: macos-14
           - ci: ciNative
             java: temurin@17
+            os: ubuntu-latest
+          - ci: ciNative
+            java: temurin@17
+            os: windows-latest
+          - ci: ciNative
+            java: temurin@21
+            os: ubuntu-latest
+          - ci: ciNative
+            java: temurin@21
+            os: windows-latest
           - ci: ciNative
             java: temurin@21
+            os: macos-14
+          - ci: ciNative
+            java: graalvm@21
+            os: ubuntu-latest
+          - ci: ciNative
+            java: graalvm@21
+            os: windows-latest
           - ci: ciNative
             java: graalvm@21
+            os: macos-14
           - os: windows-latest
             ci: ciNative
           - os: macos-14
@@ -610,5 +635,5 @@ jobs:
       - name: Submit Dependencies
         uses: scalacenter/sbt-dependency-submission@v2
         with:
-          modules-ignore: cats-effect-benchmarks_3 cats-effect-benchmarks_2.12 cats-effect-benchmarks_2.13 cats-effect_3 cats-effect_2.12 cats-effect_2.13 cats-effect-example_sjs1_3 cats-effect-example_sjs1_2.12 cats-effect-example_sjs1_2.13 rootjs_3 rootjs_2.12 rootjs_2.13 ioapptestsnative_3 ioapptestsnative_2.12 ioapptestsnative_2.13 cats-effect-graalvm-example_3 cats-effect-graalvm-example_2.12 cats-effect-graalvm-example_2.13 cats-effect-tests_sjs1_3 cats-effect-tests_sjs1_2.12 cats-effect-tests_sjs1_2.13 rootjvm_3 rootjvm_2.12 rootjvm_2.13 rootnative_3 rootnative_2.12 rootnative_2.13 cats-effect-example_native0.4_3 cats-effect-example_native0.4_2.12 cats-effect-example_native0.4_2.13 cats-effect-example_3 cats-effect-example_2.12 cats-effect-example_2.13 cats-effect-tests_3 cats-effect-tests_2.12 cats-effect-tests_2.13 ioapptestsjvm_3 ioapptestsjvm_2.12 ioapptestsjvm_2.13 ioapptestsjs_3 ioapptestsjs_2.12 ioapptestsjs_2.13 cats-effect-tests_native0.4_3 cats-effect-tests_native0.4_2.12 cats-effect-tests_native0.4_2.13
+          modules-ignore: cats-effect-benchmarks_3 cats-effect-benchmarks_2.12 cats-effect-benchmarks_2.13 cats-effect_3 cats-effect_2.12 cats-effect_2.13 cats-effect-example_sjs1_3 cats-effect-example_sjs1_2.12 cats-effect-example_sjs1_2.13 rootjs_3 rootjs_2.12 rootjs_2.13 ioapptestsnative_3 ioapptestsnative_2.12 ioapptestsnative_2.13 cats-effect-graalvm-example_3 cats-effect-graalvm-example_2.12 cats-effect-graalvm-example_2.13 cats-effect-tests_sjs1_3 cats-effect-tests_sjs1_2.12 cats-effect-tests_sjs1_2.13 rootjvm_3 rootjvm_2.12 rootjvm_2.13 rootnative_3 rootnative_2.12 rootnative_2.13 cats-effect-example_native0.5_3 cats-effect-example_native0.5_2.12 cats-effect-example_native0.5_2.13 cats-effect-example_3 cats-effect-example_2.12 cats-effect-example_2.13 cats-effect-tests_3 cats-effect-tests_2.12 cats-effect-tests_2.13 ioapptestsjvm_3 ioapptestsjvm_2.12 ioapptestsjvm_2.13 ioapptestsjs_3 ioapptestsjs_2.12 ioapptestsjs_2.13 cats-effect-tests_native0.5_3 cats-effect-tests_native0.5_2.12 cats-effect-tests_native0.5_2.13
           configs-ignore: test scala-tool scala-doc-tool test-internal

build.sbt

@@ -24,6 +24,7 @@ import org.openqa.selenium.firefox.{FirefoxOptions, FirefoxProfile}
 import org.scalajs.jsenv.nodejs.NodeJSEnv
 import org.scalajs.jsenv.selenium.SeleniumJSEnv
 import sbtcrossproject.CrossProject
+import scala.scalanative.build._
 
 import JSEnv._
 
@@ -41,7 +42,7 @@ ThisBuild / git.gitUncommittedChanges := {
   }
 }
 
-ThisBuild / tlBaseVersion := "3.6"
+ThisBuild / tlBaseVersion := "3.7"
 ThisBuild / tlUntaggedAreSnapshots := false
 
 ThisBuild / organization := "org.typelevel"
@@ -254,10 +255,11 @@ ThisBuild / githubWorkflowBuildMatrixExclusions := {
   val nativeJavaAndOSFilters = {
     val ci = CI.Native.command
 
-    val javaFilters =
-      (ThisBuild / githubWorkflowJavaVersions).value.filterNot(Set(ScalaNativeJava)).map {
-        java => MatrixExclude(Map("ci" -> ci, "java" -> java.render))
-      }
+    val javaFilters = for {
+      java <- (ThisBuild / githubWorkflowJavaVersions).value.filterNot(Set(ScalaNativeJava))
+      os <- (ThisBuild / githubWorkflowOSes).value
+      if !(os == MacOS && java == LatestJava)
+    } yield MatrixExclude(Map("ci" -> ci, "java" -> java.render, "os" -> os))
 
     javaFilters ++ Seq(
       MatrixExclude(Map("os" -> Windows, "ci" -> ci)),
@@ -300,12 +302,12 @@ ThisBuild / apiURL := Some(url("https://typelevel.org/cats-effect/api/3.x/"))
 
 ThisBuild / autoAPIMappings := true
 
-val CatsVersion = "2.11.0"
-val CatsMtlVersion = "1.3.1"
-val Specs2Version = "4.20.5"
-val ScalaCheckVersion = "1.17.1"
-val DisciplineVersion = "1.4.0"
-val CoopVersion = "1.2.0"
+val CatsVersion = "2.12.0"
+val CatsMtlVersion = "1.5.0"
+val Specs2Version = "4.20.9"
+val ScalaCheckVersion = "1.18.1"
+val DisciplineVersion = "1.5.0"
+val CoopVersion = "1.3.0"
 
 val MacrotaskExecutorVersion = "1.1.1"
 
@@ -406,7 +408,7 @@ lazy val kernel = crossProject(JSPlatform, JVMPlatform, NativePlatform)
     libraryDependencies += "org.scala-js" %%% "scala-js-macrotask-executor" % MacrotaskExecutorVersion % Test
   )
   .nativeSettings(
-    libraryDependencies += "io.github.cquiroz" %%% "scala-java-time" % "2.5.0"
+    libraryDependencies += "io.github.cquiroz" %%% "scala-java-time" % "2.6.0"
   )
 
 /**
@@ -940,7 +942,13 @@ lazy val tests: CrossProject = crossProject(JSPlatform, JVMPlatform, NativePlatf
     Test / javaOptions += "-Dcats.effect.trackFiberContext=true"
   )
   .nativeSettings(
-    Compile / mainClass := Some("catseffect.examples.NativeRunner")
+    Compile / mainClass := Some("catseffect.examples.NativeRunner"),
+    nativeConfig ~= { c => // TODO: remove this when it seems to work
+      c.withSourceLevelDebuggingConfig(_.enableAll) // enable generation of debug information
+        .withOptimize(false) // disable Scala Native optimizer
+        .withMode(Mode.debug) // compile using LLVM without optimizations
+        .withCompileOptions(c.compileOptions ++ Seq("-gdwarf-4"))
+    }
   )
 
 def configureIOAppTests(p: Project): Project =
@@ -951,7 +959,7 @@ def configureIOAppTests(p: Project): Project =
       buildInfoPackage := "cats.effect",
       buildInfoKeys ++= Seq(
         "jsRunner" -> (tests.js / Compile / fastOptJS / artifactPath).value,
-        "nativeRunner" -> (tests.native / Compile / nativeLink / artifactPath).value
+        "nativeRunner" -> (tests.native / Compile / crossTarget).value / (tests.native / Compile / moduleName).value
       )
     )
 

core/js/src/main/scala/cats/effect/unsafe/BatchingMacrotaskExecutor.scala

@@ -40,8 +40,7 @@ import scala.util.control.NonFatal
 private[effect] final class BatchingMacrotaskExecutor(
     batchSize: Int,
     reportFailure0: Throwable => Unit
-) extends ExecutionContextExecutor
-    with FiberExecutor {
+) extends ExecutionContextExecutor {
 
   private[this] val queueMicrotask: js.Function1[js.Function0[Any], Any] =
     if (js.typeOf(js.Dynamic.global.queueMicrotask) == "function")

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

@@ -17,6 +17,9 @@
 package cats.effect
 package unsafe
 
+import scala.concurrent.ExecutionContext
+import scala.scalajs.{js, LinkingInfo}
+
 private[effect] sealed abstract class FiberMonitor extends FiberMonitorShared {
 
   /**
@@ -41,8 +44,8 @@ private[effect] sealed abstract class FiberMonitor extends FiberMonitorShared {
 
 private final class FiberMonitorImpl(
     // A reference to the compute pool of the `IORuntime` in which this suspended fiber bag
-    // operates. `null` if the compute pool of the `IORuntime` is not a `FiberExecutor`.
-    private[this] val compute: FiberExecutor
+    // operates. `null` if the compute pool of the `IORuntime` is not a `BatchingMacrotaskExecutor`.
+    private[this] val compute: BatchingMacrotaskExecutor
 ) extends FiberMonitor {
   private[this] val bag = new WeakBag[IOFiber[_]]()
 
@@ -92,4 +95,26 @@ private final class NoOpFiberMonitor extends FiberMonitor {
   def liveFiberSnapshot(print: String => Unit): Unit = ()
 }
 
-private[effect] object FiberMonitor extends FiberMonitorPlatform
+private[effect] object FiberMonitor {
+  def apply(compute: ExecutionContext): FiberMonitor = {
+    if (LinkingInfo.developmentMode && weakRefsAvailable) {
+      if (compute.isInstanceOf[BatchingMacrotaskExecutor]) {
+        val bmec = compute.asInstanceOf[BatchingMacrotaskExecutor]
+        new FiberMonitorImpl(bmec)
+      } else {
+        new FiberMonitorImpl(null)
+      }
+    } else {
+      new NoOpFiberMonitor()
+    }
+  }
+
+  private[this] final val Undefined = "undefined"
+
+  /**
+   * Feature-tests for all the required, well, features :)
+   */
+  private[unsafe] def weakRefsAvailable: Boolean =
+    js.typeOf(js.Dynamic.global.WeakRef) != Undefined &&
+      js.typeOf(js.Dynamic.global.FinalizationRegistry) != Undefined
+}

core/jvm/src/main/scala/cats/effect/unsafe/SchedulerCompanionPlatform.scala → core/jvm-native/src/main/scala/cats/effect/unsafe/SchedulerCompanionPlatform.scala

@@ -23,10 +23,14 @@ import java.time.temporal.ChronoField
 import java.util.concurrent.{Executors, ScheduledExecutorService}
 
 private[unsafe] abstract class SchedulerCompanionPlatform { this: Scheduler.type =>
-  def createDefaultScheduler(): (Scheduler, () => Unit) = {
+
+  def createDefaultScheduler(): (Scheduler, () => Unit) =
+    createDefaultScheduler("io-scheduler")
+
+  def createDefaultScheduler(threadPrefix: String): (Scheduler, () => Unit) = {
     val scheduler = Executors.newSingleThreadScheduledExecutor { r =>
       val t = new Thread(r)
-      t.setName("io-scheduler")
+      t.setName(threadPrefix)
       t.setDaemon(true)
       t.setPriority(Thread.MAX_PRIORITY)
       t

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

@@ -38,8 +38,6 @@ import scala.concurrent.ExecutionContextExecutor
 import scala.concurrent.duration.{Duration, FiniteDuration}
 import scala.util.control.NonFatal
 
-import java.time.Instant
-import java.time.temporal.ChronoField
 import java.util.Comparator
 import java.util.concurrent.{ConcurrentSkipListSet, ThreadLocalRandom}
 import java.util.concurrent.atomic.{
@@ -78,7 +76,8 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
     reportFailure0: Throwable => Unit
 ) extends ExecutionContextExecutor
     with Scheduler
-    with UnsealedPollingContext[P] {
+    with UnsealedPollingContext[P]
+    with WorkStealingThreadPoolPlatform[P] {
 
   import TracingConstants._
   import WorkStealingThreadPoolConstants._
@@ -628,11 +627,6 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
 
   override def nowMillis(): Long = System.currentTimeMillis()
 
-  override def nowMicros(): Long = {
-    val now = Instant.now()
-    now.getEpochSecond() * 1000000 + now.getLong(ChronoField.MICRO_OF_SECOND)
-  }
-
   /**
    * Tries to call the current worker's `sleep`, but falls back to `sleepExternal` if needed.
    */

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

@@ -24,7 +24,7 @@ import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
 import java.lang.management.ManagementFactory
-import java.util.concurrent.{Executors, ScheduledThreadPoolExecutor}
+import java.util.concurrent.Executors
 import java.util.concurrent.atomic.AtomicInteger
 
 import javax.management.ObjectName
@@ -246,19 +246,8 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type
     (ExecutionContext.fromExecutor(executor, reportFailure), { () => executor.shutdown() })
   }
 
-  def createDefaultScheduler(threadPrefix: String = "io-scheduler"): (Scheduler, () => Unit) = {
-    val scheduler = new ScheduledThreadPoolExecutor(
-      1,
-      { r =>
-        val t = new Thread(r)
-        t.setName(threadPrefix)
-        t.setDaemon(true)
-        t.setPriority(Thread.MAX_PRIORITY)
-        t
-      })
-    scheduler.setRemoveOnCancelPolicy(true)
-    (Scheduler.fromScheduledExecutor(scheduler), { () => scheduler.shutdown() })
-  }
+  def createDefaultScheduler(threadPrefix: String = "io-scheduler"): (Scheduler, () => Unit) =
+    Scheduler.createDefaultScheduler(threadPrefix)
 
   def createDefaultPollingSystem(): PollingSystem = SelectorSystem()
 

core/js-native/src/main/scala/cats/effect/unsafe/FiberExecutor.scala → core/jvm/src/main/scala/cats/effect/unsafe/WorkStealingThreadPoolPlatform.scala

@@ -17,9 +17,14 @@
 package cats.effect
 package unsafe
 
-/**
- * An introspectable executor that runs fibers. Useful for fiber dumps.
- */
-private[unsafe] trait FiberExecutor {
-  def liveTraces(): Map[IOFiber[_], Trace]
+import java.time.Instant
+import java.time.temporal.ChronoField
+
+trait WorkStealingThreadPoolPlatform[P <: AnyRef] extends Scheduler {
+  this: WorkStealingThreadPool[P] =>
+
+  override def nowMicros(): Long = {
+    val now = Instant.now()
+    now.getEpochSecond() * 1000000 + now.getLong(ChronoField.MICRO_OF_SECOND)
+  }
 }