Add support for Native multithreaded execution

.cirrus.yml

@@ -10,3 +10,16 @@ jvm_highcore_task:
       script: sbt '++ 2.13' testsJVM/test ioAppTestsJVM/test
     - name: JVM high-core-count 3
       script: sbt '++ 3' testsJVM/test ioAppTestsJVM/test
+
+native_highcore_task:
+  only_if: $CIRRUS_TAG != '' || $CIRRUS_PR != ''
+  required_pr_labels: Cirrus Native
+  container:
+    dockerfile: .cirrus/Dockerfile
+    cpu: 8
+    memory: 16G
+  matrix:
+    - name: Native high-core-count 2.13
+      script: sbt '++ 2.13' testsNative/test ioAppTestsNative/test
+    - name: Native high-core-count 3
+      script: sbt '++ 3' testsNative/test ioAppTestsNative/test

.github/workflows/ci.yml

@@ -45,8 +45,6 @@ jobs:
             java: temurin@11
           - scala: 3.3.5
             java: temurin@21
-          - scala: 3.3.5
-            java: graalvm@21
           - scala: 2.12.20
             java: temurin@11
           - scala: 2.12.20
@@ -325,6 +323,11 @@ jobs:
         shell: bash
         run: example/test-native.sh ${{ matrix.scala }}
 
+      - name: Test GraalVM Native Image
+        if: (matrix.scala == '2.13.16' || matrix.scala == '3.3.5') && matrix.java == 'graalvm@21' && matrix.os == 'ubuntu-latest'
+        shell: bash
+        run: sbt '++ ${{ matrix.scala }}' graalVMExample/nativeImage graalVMExample/nativeImageRun
+
       - name: Scalafix tests
         if: matrix.scala == '2.13.16' && matrix.ci == 'ciJVM' && matrix.os == 'ubuntu-latest'
         shell: bash
@@ -667,5 +670,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

@@ -23,11 +23,14 @@ 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._
 
+lazy val inCI = Option(System.getenv("CI")).contains("true")
+
 // sbt-git workarounds
-ThisBuild / useConsoleForROGit := !Option(System.getenv("CI")).contains("true")
+ThisBuild / useConsoleForROGit := !inCI
 
 ThisBuild / git.gitUncommittedChanges := {
   if ((ThisBuild / githubIsWorkflowBuild).value) {
@@ -40,7 +43,7 @@ ThisBuild / git.gitUncommittedChanges := {
   }
 }
 
-ThisBuild / tlBaseVersion := "3.6"
+ThisBuild / tlBaseVersion := "3.7"
 ThisBuild / tlUntaggedAreSnapshots := false
 
 ThisBuild / organization := "org.typelevel"
@@ -201,6 +204,12 @@ ThisBuild / githubWorkflowBuild := Seq("JVM", "JS", "Native").map { platform =>
     name = Some("Test Example Native App Using Binary"),
     cond = Some(s"matrix.ci == 'ciNative' && matrix.os == '$PrimaryOS'")
   ),
+  WorkflowStep.Sbt(
+    List("graalVMExample/nativeImage", "graalVMExample/nativeImageRun"),
+    name = Some("Test GraalVM Native Image"),
+    cond = Some(
+      s"(matrix.scala == '$Scala213' || matrix.scala == '$Scala3') && matrix.java == '${GraalVM.render}' && matrix.os == '$PrimaryOS'")
+  ),
   WorkflowStep.Run(
     List("cd scalafix", "sbt test"),
     name = Some("Scalafix tests"),
@@ -224,7 +233,7 @@ ThisBuild / githubWorkflowBuildMatrixExclusions := {
   val scalaJavaFilters = for {
     scala <- (ThisBuild / githubWorkflowScalaVersions).value.filterNot(Set(Scala213))
     java <- (ThisBuild / githubWorkflowJavaVersions).value.filterNot(Set(OldGuardJava))
-    if !(scala == Scala3 && java == LatestJava)
+    if !(scala == Scala3 && (java == LatestJava || java == GraalVM))
   } yield MatrixExclude(Map("scala" -> scala, "java" -> java.render))
 
   val armFilters =
@@ -316,13 +325,13 @@ ThisBuild / autoAPIMappings := true
 
 ThisBuild / Test / testOptions += Tests.Argument("+l")
 
-val CatsVersion = "2.11.0"
-val CatsMtlVersion = "1.3.1"
-val ScalaCheckVersion = "1.17.1"
-val CoopVersion = "1.2.0"
-val MUnitVersion = "1.0.0-M11"
-val MUnitScalaCheckVersion = "1.0.0-M11"
-val DisciplineMUnitVersion = "2.0.0-M3"
+val CatsVersion = "2.12.0"
+val CatsMtlVersion = "1.5.0"
+val ScalaCheckVersion = "1.18.1"
+val CoopVersion = "1.3.0"
+val MUnitVersion = "1.1.0"
+val MUnitScalaCheckVersion = "1.1.0"
+val DisciplineMUnitVersion = "2.0.0"
 
 val MacrotaskExecutorVersion = "1.1.1"
 
@@ -352,6 +361,17 @@ Global / tlCommandAliases ++= Map(
   )
 )
 
+lazy val nativeTestSettings = Seq(
+  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"))
+  },
+  envVars ++= { if (inCI) Map("GC_MAXIMUM_HEAP_SIZE" -> "8g") else Map.empty[String, String] },
+  parallelExecution := !inCI
+)
+
 val jsProjects: Seq[ProjectReference] =
   Seq(
     kernel.js,
@@ -435,7 +455,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"
   )
 
 /**
@@ -881,7 +901,12 @@ lazy val core = crossProject(JSPlatform, JVMPlatform, NativePlatform)
         ProblemFilters.exclude[MissingClassProblem](
           "cats.effect.metrics.JsCpuStarvationMetrics"),
         ProblemFilters.exclude[MissingClassProblem](
-          "cats.effect.metrics.JsCpuStarvationMetrics$")
+          "cats.effect.metrics.JsCpuStarvationMetrics$"),
+        // all package-private classes; introduced when we made Native multithreaded
+        ProblemFilters.exclude[MissingClassProblem]("cats.effect.unsafe.FiberExecutor"),
+        ProblemFilters.exclude[IncompatibleMethTypeProblem](
+          "cats.effect.unsafe.FiberMonitorImpl.this"),
+        ProblemFilters.exclude[MissingClassProblem]("cats.effect.unsafe.FiberMonitorPlatform")
       )
     },
     mimaBinaryIssueFilters ++= {
@@ -938,6 +963,7 @@ lazy val testkit = crossProject(JSPlatform, JVMPlatform, NativePlatform)
       "org.scalacheck" %%% "scalacheck" % ScalaCheckVersion
     )
   )
+  .nativeSettings(nativeTestSettings)
 
 /**
  * Unit tests for the core project, utilizing the support provided by testkit.
@@ -969,7 +995,8 @@ 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"),
+    nativeTestSettings
   )
 
 def configureIOAppTests(p: Project): Project =
@@ -980,7 +1007,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

@@ -39,8 +39,7 @@ import scala.scalajs.{js, LinkingInfo}
 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/FiberMonitor.scala → core/jvm-native/src/main/scala/cats/effect/unsafe/FiberMonitor.scala

@@ -215,12 +215,12 @@ private[effect] final class NoOpFiberMonitor extends FiberMonitor(null) {
 }
 
 private[effect] object FiberMonitor {
-  def apply(compute: ExecutionContext): FiberMonitor = {
+  def apply(compute: ExecutionContext): FiberMonitor = if (Platform.isJvm) {
     if (TracingConstants.isStackTracing && compute.isInstanceOf[WorkStealingThreadPool[?]]) {
       val wstp = compute.asInstanceOf[WorkStealingThreadPool[?]]
       new FiberMonitor(wstp)
     } else {
       new FiberMonitor(null)
     }
-  }
+  } else new NoOpFiberMonitor
 }

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/TimerHeap.scala → core/jvm-native/src/main/scala/cats/effect/unsafe/TimerHeap.scala

@@ -34,6 +34,8 @@ import scala.annotation.tailrec
 import java.util.Arrays
 import java.util.concurrent.atomic.AtomicInteger
 
+import Platform.safePublish
+
 /**
  * A specialized heap that serves as a priority queue for timers i.e. callbacks with trigger
  * times.
@@ -486,6 +488,7 @@ private final class TimerHeap extends AtomicInteger {
 
   override def toString() = if (size > 0) "TimerHeap(...)" else "TimerHeap()"
 
+  @safePublish
   private final class Node(
       val triggerTime: Long,
       private[this] var callback: Right[Nothing, Unit] => Unit,

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

@@ -37,8 +37,6 @@ import scala.collection.mutable
 import scala.concurrent.ExecutionContextExecutor
 import scala.concurrent.duration.{Duration, FiniteDuration}
 
-import java.time.Instant
-import java.time.temporal.ChronoField
 import java.util.concurrent.{LinkedTransferQueue, ThreadLocalRandom}
 import java.util.concurrent.atomic.{
   AtomicBoolean,
@@ -77,7 +75,8 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
     private[unsafe] val uncaughtExceptionHandler: Thread.UncaughtExceptionHandler
 ) extends ExecutionContextExecutor
     with Scheduler
-    with UnsealedPollingContext[P] {
+    with UnsealedPollingContext[P]
+    with WorkStealingThreadPoolPlatform[P] {
 
   import TracingConstants._
   import WorkStealingThreadPoolConstants._
@@ -629,11 +628,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.
    */
@@ -710,6 +704,7 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
       while (i < threadCount) {
         val workerThread = workerThreads.get(i)
         if (workerThread ne currentThread) {
+          system.interrupt(workerThread, pollers(i))
           workerThread.interrupt()
         }
         i += 1

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

@@ -165,6 +165,13 @@ trait IOApp {
    */
   protected def runtimeConfig: unsafe.IORuntimeConfig = unsafe.IORuntimeConfig()
 
+  /**
+   * The [[unsafe.PollingSystem]] used by the [[runtime]] which will evaluate the [[IO]]
+   * produced by `run`. It is very unlikely that users will need to override this method.
+   *
+   * [[unsafe.PollingSystem]] implementors may provide their own flavors of [[IOApp]] that
+   * override this method.
+   */
   protected def pollingSystem: unsafe.PollingSystem =
     unsafe.IORuntime.createDefaultPollingSystem()
 
@@ -182,7 +189,8 @@ trait IOApp {
    * beyond a few percentage points, and the default value is optimal (or close to optimal) in
    * ''most'' common scenarios.
    *
-   * '''This setting is JVM-specific and will not compile on JavaScript.'''
+   * '''This setting is specific to the JVM and Scala Native, and will not compile on
+   * JavaScript.'''
    *
    * For more details on Cats Effect's runtime threading model please see
    * [[https://typelevel.org/cats-effect/docs/thread-model]].