From 5d11fe96c446300add6a34d6e3f2a109301cfc74 Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Fri, 9 Sep 2022 18:57:35 +0000 Subject: [PATCH 1/8] Backport Scala Native support --- .github/workflows/ci.yml | 33 ++- build.sbt | 82 +++--- .../scala/cats/effect/IOFiberConstants.scala | 0 .../scala/cats/effect/IOFiberPlatform.scala | 0 .../scala/cats/effect/SyncIOConstants.scala | 0 .../unsafe/WorkStealingThreadPool.scala | 10 + .../cats/effect/SyncIOCompanionPlatform.scala | 0 .../cats/effect/unsafe/FiberMonitor.scala | 7 +- .../cats/effect/unsafe/ref/package.scala | 0 .../cats/effect/tracing/TracingPlatform.scala | 2 +- .../src/main/scala/cats/effect/IOApp.scala | 244 ++++++++++++++++++ .../cats/effect/IOCompanionPlatform.scala | 43 +++ .../main/scala/cats/effect/IOPlatform.scala | 19 ++ .../effect/tracing/TracingConstants.scala | 27 ++ .../cats/effect/tracing/TracingPlatform.scala | 34 +++ .../unsafe/IORuntimeCompanionPlatform.scala | 60 +++++ .../IORuntimeConfigCompanionPlatform.scala | 55 ++++ .../unsafe/PollingExecutorScheduler.scala | 163 ++++++++++++ .../unsafe/QueueExecutorScheduler.scala | 31 +++ .../unsafe/SchedulerCompanionPlatform.scala | 23 ++ .../scala/cats/effect/example/Example.scala | 0 example/test-native.sh | 47 ++++ .../cats/effect/kernel/ClockPlatform.scala | 0 .../cats/effect/kernel/AsyncPlatform.scala | 19 ++ .../cats/effect/kernel/ResourcePlatform.scala | 19 ++ .../laws/ReaderWriterStateTFreeSyncSpec.scala | 7 + project/CI.scala | 12 +- project/Common.scala | 11 +- project/plugins.sbt | 8 +- .../effect/std/RandomCompanionPlatform.scala | 0 .../effect/std/UUIDGenCompanionPlatform.scala | 0 .../cats/effect/std/DispatcherPlatform.scala | 19 ++ .../cats/effect/std/JavaSecureRandom.scala | 63 +++++ .../test/scala/cats/effect/IOFiberSpec.scala | 0 .../scala/cats/effect/tracing/TraceSpec.scala | 0 .../cats/effect/tracing/TracingSpec.scala | 0 .../cats/effect/ContSpecBasePlatform.scala | 0 .../scala/cats/effect/DetectPlatform.scala | 1 + .../scala/catseffect/examplesplatform.scala | 2 +- .../effect/SyncIOPlatformSpecification.scala | 0 .../scala/cats/effect/DetectPlatform.scala | 1 + .../scala/cats/effect/DetectPlatform.scala | 23 ++ .../cats/effect/IOPlatformSpecification.scala | 34 +++ .../scala/cats/effect/RunnersPlatform.scala | 23 ++ .../scala/cats/effect/std/RandomSpec.scala | 12 +- 45 files changed, 1080 insertions(+), 54 deletions(-) rename core/{js => js-native}/src/main/scala/cats/effect/IOFiberConstants.scala (100%) rename core/{js => js-native}/src/main/scala/cats/effect/IOFiberPlatform.scala (100%) rename core/{js => js-native}/src/main/scala/cats/effect/SyncIOConstants.scala (100%) rename core/{js => js-native}/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala (75%) rename core/{jvm => jvm-native}/src/main/scala/cats/effect/SyncIOCompanionPlatform.scala (100%) rename core/{jvm => jvm-native}/src/main/scala/cats/effect/unsafe/FiberMonitor.scala (99%) rename core/{jvm => jvm-native}/src/main/scala/cats/effect/unsafe/ref/package.scala (100%) create mode 100644 core/native/src/main/scala/cats/effect/IOApp.scala create mode 100644 core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala create mode 100644 core/native/src/main/scala/cats/effect/IOPlatform.scala create mode 100644 core/native/src/main/scala/cats/effect/tracing/TracingConstants.scala create mode 100644 core/native/src/main/scala/cats/effect/tracing/TracingPlatform.scala create mode 100644 core/native/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala create mode 100644 core/native/src/main/scala/cats/effect/unsafe/IORuntimeConfigCompanionPlatform.scala create mode 100644 core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala create mode 100644 core/native/src/main/scala/cats/effect/unsafe/QueueExecutorScheduler.scala create mode 100644 core/native/src/main/scala/cats/effect/unsafe/SchedulerCompanionPlatform.scala rename example/{js => js-native}/src/main/scala/cats/effect/example/Example.scala (100%) create mode 100755 example/test-native.sh rename kernel/{jvm => jvm-native}/src/main/scala/cats/effect/kernel/ClockPlatform.scala (100%) create mode 100644 kernel/native/src/main/scala/cats/effect/kernel/AsyncPlatform.scala create mode 100644 kernel/native/src/main/scala/cats/effect/kernel/ResourcePlatform.scala rename std/{js => js-native}/src/main/scala/cats/effect/std/RandomCompanionPlatform.scala (100%) rename std/{js => js-native}/src/main/scala/cats/effect/std/UUIDGenCompanionPlatform.scala (100%) create mode 100644 std/native/src/main/scala/cats/effect/std/DispatcherPlatform.scala create mode 100644 std/native/src/main/scala/cats/effect/std/JavaSecureRandom.scala rename tests/{shared => js-jvm}/src/test/scala/cats/effect/IOFiberSpec.scala (100%) rename tests/{shared => js-jvm}/src/test/scala/cats/effect/tracing/TraceSpec.scala (100%) rename tests/{shared => js-jvm}/src/test/scala/cats/effect/tracing/TracingSpec.scala (100%) rename tests/{js => js-native}/src/test/scala/cats/effect/ContSpecBasePlatform.scala (100%) rename tests/{jvm => jvm-native}/src/test/scala/cats/effect/SyncIOPlatformSpecification.scala (100%) create mode 100644 tests/native/src/test/scala/cats/effect/DetectPlatform.scala create mode 100644 tests/native/src/test/scala/cats/effect/IOPlatformSpecification.scala create mode 100644 tests/native/src/test/scala/cats/effect/RunnersPlatform.scala diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 571efe32b0..9696e23740 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,13 +23,13 @@ jobs: strategy: matrix: os: [ubuntu-latest, windows-latest, macos-latest] - scala: [3.0.2, 2.12.15, 2.13.7] + scala: [3.1.2, 2.12.15, 2.13.7] java: [temurin@8, temurin@11, temurin@17, graalvm@11] - ci: [ciJVM, ciJS, ciFirefox, ciChrome] + ci: [ciJVM, ciNative, ciJS, ciFirefox, ciChrome] exclude: - - scala: 3.0.2 + - scala: 3.1.2 java: temurin@11 - - scala: 3.0.2 + - scala: 3.1.2 java: graalvm@11 - scala: 2.12.15 java: temurin@11 @@ -38,21 +38,31 @@ jobs: - scala: 2.12.15 java: graalvm@11 - os: windows-latest - scala: 3.0.2 + scala: 3.1.2 - os: macos-latest - scala: 3.0.2 + scala: 3.1.2 - os: windows-latest scala: 2.12.15 - os: macos-latest scala: 2.12.15 - ci: ciFirefox - scala: 3.0.2 + scala: 3.1.2 - ci: ciChrome - scala: 3.0.2 + scala: 3.1.2 - ci: ciFirefox scala: 2.12.15 - ci: ciChrome scala: 2.12.15 + - ci: ciNative + java: temurin@11 + - ci: ciNative + java: temurin@17 + - ci: ciNative + java: graalvm@11 + - os: windows-latest + ci: ciNative + - os: macos-latest + ci: ciNative - ci: ciJS java: temurin@11 - ci: ciJS @@ -191,7 +201,7 @@ jobs: - shell: bash run: sbt '++${{ matrix.scala }}' '${{ matrix.ci }}' - - if: (matrix.scala == '2.13.7' || matrix.scala == '3.0.2') && matrix.ci == 'ciJVM' + - if: (matrix.scala == '2.13.7' || matrix.scala == '3.1.2') && matrix.ci == 'ciJVM' shell: bash run: sbt '++${{ matrix.scala }}' docs/mdoc @@ -205,6 +215,11 @@ jobs: shell: bash run: example/test-js.sh ${{ matrix.scala }} + - name: Test Example Native App Using Binary + if: matrix.ci == 'ciNative' && matrix.os == 'ubuntu-latest' + shell: bash + run: example/test-native.sh ${{ matrix.scala }} + - name: Scalafix tests if: matrix.scala == '2.13.7' && matrix.ci == 'ciJVM' && matrix.os == 'ubuntu-latest' shell: bash diff --git a/build.sbt b/build.sbt index 8f37123d24..d897d17fdd 100644 --- a/build.sbt +++ b/build.sbt @@ -113,7 +113,7 @@ val Windows = "windows-latest" val MacOS = "macos-latest" val Scala213 = "2.13.7" -val Scala3 = "3.0.2" +val Scala3 = "3.1.2" ThisBuild / crossScalaVersions := Seq(Scala3, "2.12.15", Scala213) ThisBuild / tlVersionIntroduced := Map("3" -> "3.1.1") @@ -162,6 +162,11 @@ ThisBuild / githubWorkflowBuild := Seq( name = Some("Test Example JavaScript App Using Node"), cond = Some(s"matrix.ci == 'ciJS' && matrix.os == '$PrimaryOS'") ), + WorkflowStep.Run( + List("example/test-native.sh ${{ matrix.scala }}"), + name = Some("Test Example Native App Using Binary"), + cond = Some(s"matrix.ci == 'ciNative' && matrix.os == '$PrimaryOS'") + ), WorkflowStep.Run( List("cd scalafix", "sbt test"), name = Some("Scalafix tests"), @@ -193,7 +198,7 @@ ThisBuild / githubWorkflowBuildMatrixExclusions := { ci <- jsCiVariants.tail } yield MatrixExclude(Map("ci" -> ci, "scala" -> scala)) - val jsJavaAndOSFilters = jsCiVariants.flatMap { ci => + val nativeJsJavaAndOSFilters = (CI.Native.command :: jsCiVariants).flatMap { ci => val javaFilters = (ThisBuild / githubWorkflowJavaVersions).value.filterNot(Set(ScalaJSJava)).map { java => MatrixExclude(Map("ci" -> ci, "java" -> java.render)) @@ -209,7 +214,7 @@ ThisBuild / githubWorkflowBuildMatrixExclusions := { MatrixExclude(Map("os" -> Windows, "java" -> GraalVM.render)) ) - scalaJavaFilters ++ windowsAndMacScalaFilters ++ jsScalaFilters ++ jsJavaAndOSFilters ++ flakyFilters + scalaJavaFilters ++ windowsAndMacScalaFilters ++ jsScalaFilters ++ nativeJsJavaAndOSFilters ++ flakyFilters } lazy val useJSEnv = @@ -245,11 +250,11 @@ ThisBuild / apiURL := Some(url("https://typelevel.org/cats-effect/api/3.x/")) ThisBuild / autoAPIMappings := true -val CatsVersion = "2.7.0" -val Specs2Version = "4.13.1" -val ScalaCheckVersion = "1.15.4" -val DisciplineVersion = "1.2.5" -val CoopVersion = "1.1.1" +val CatsVersion = "2.8.0" +val Specs2Version = "4.16.0" +val ScalaCheckVersion = "1.16.0" +val DisciplineVersion = "1.4.0" +val CoopVersion = "1.2.0" val MacrotaskExecutorVersion = "1.0.0" @@ -257,21 +262,37 @@ tlReplaceCommandAlias("ci", CI.AllCIs.map(_.toString).mkString) addCommandAlias("release", "tlRelease") addCommandAlias(CI.JVM.command, CI.JVM.toString) +addCommandAlias(CI.Native.command, CI.Native.toString) addCommandAlias(CI.JS.command, CI.JS.toString) addCommandAlias(CI.Firefox.command, CI.Firefox.toString) addCommandAlias(CI.Chrome.command, CI.Chrome.toString) addCommandAlias("prePR", "; root/clean; scalafmtSbt; +root/scalafmtAll; +root/headerCreate") +val nativeProjects: Seq[ProjectReference] = + Seq( + kernel.native, + kernelTestkit.native, + laws.native, + core.native, + testkit.native, + tests.native, + std.native, + example.native) + val jsProjects: Seq[ProjectReference] = Seq(kernel.js, kernelTestkit.js, laws.js, core.js, testkit.js, testsJS, std.js, example.js) val undocumentedRefs = - jsProjects ++ Seq[ProjectReference](benchmarks, example.jvm, tests.jvm, tests.js) + jsProjects ++ nativeProjects ++ Seq[ProjectReference]( + benchmarks, + example.jvm, + tests.jvm, + tests.js) lazy val root = project .in(file(".")) - .aggregate(rootJVM, rootJS) + .aggregate(rootJVM, rootJS, rootNative) .enablePlugins(NoPublishPlugin) .enablePlugins(ScalaUnidocPlugin) .settings( @@ -295,29 +316,32 @@ lazy val rootJVM = project lazy val rootJS = project.aggregate(jsProjects: _*).enablePlugins(NoPublishPlugin) +lazy val rootNative = project.aggregate(nativeProjects: _*).enablePlugins(NoPublishPlugin) + /** * The core abstractions and syntax. This is the most general definition of Cats Effect, without * any concrete implementations. This is the "batteries not included" dependency. */ -lazy val kernel = crossProject(JSPlatform, JVMPlatform) +lazy val kernel = crossProject(JSPlatform, JVMPlatform, NativePlatform) .in(file("kernel")) .settings( name := "cats-effect-kernel", libraryDependencies += "org.typelevel" %%% "cats-core" % CatsVersion) .settings( - libraryDependencies += ("org.specs2" %%% "specs2-core" % Specs2Version % Test) - .cross(CrossVersion.for3Use2_13) - .exclude("org.scala-js", "scala-js-macrotask-executor_sjs1_2.13") + libraryDependencies += "org.specs2" %%% "specs2-core" % Specs2Version % Test ) .jsSettings( libraryDependencies += "org.scala-js" %%% "scala-js-macrotask-executor" % MacrotaskExecutorVersion % Test ) + .nativeSettings( + libraryDependencies += "io.github.cquiroz" %%% "scala-java-time" % "2.4.0" + ) /** * Reference implementations (including a pure ConcurrentBracket), generic ScalaCheck * generators, and useful tools for testing code written against Cats Effect. */ -lazy val kernelTestkit = crossProject(JSPlatform, JVMPlatform) +lazy val kernelTestkit = crossProject(JSPlatform, JVMPlatform, NativePlatform) .in(file("kernel-testkit")) .dependsOn(kernel) .settings( @@ -336,7 +360,7 @@ lazy val kernelTestkit = crossProject(JSPlatform, JVMPlatform) * dependency issues. As a consequence of this split, some things which are defined in * kernelTestkit are *tested* in the Test scope of this project. */ -lazy val laws = crossProject(JSPlatform, JVMPlatform) +lazy val laws = crossProject(JSPlatform, JVMPlatform, NativePlatform) .in(file("laws")) .dependsOn(kernel, kernelTestkit % Test) .settings( @@ -351,7 +375,7 @@ lazy val laws = crossProject(JSPlatform, JVMPlatform) * contains some general datatypes built on top of IO which are useful in their own right, as * well as some utilities (such as IOApp). This is the "batteries included" dependency. */ -lazy val core = crossProject(JSPlatform, JVMPlatform) +lazy val core = crossProject(JSPlatform, JVMPlatform, NativePlatform) .in(file("core")) .dependsOn(kernel, std) .settings( @@ -621,23 +645,21 @@ lazy val core = crossProject(JSPlatform, JVMPlatform) * Test support for the core project, providing various helpful instances like ScalaCheck * generators for IO and SyncIO. */ -lazy val testkit = crossProject(JSPlatform, JVMPlatform) +lazy val testkit = crossProject(JSPlatform, JVMPlatform, NativePlatform) .in(file("testkit")) .dependsOn(core, kernelTestkit) .settings( name := "cats-effect-testkit", libraryDependencies ++= Seq( "org.scalacheck" %%% "scalacheck" % ScalaCheckVersion, - ("org.specs2" %%% "specs2-core" % Specs2Version % Test) - .cross(CrossVersion.for3Use2_13) - .exclude("org.scala-js", "scala-js-macrotask-executor_sjs1_2.13") + "org.specs2" %%% "specs2-core" % Specs2Version % Test ) ) /** * Unit tests for the core project, utilizing the support provided by testkit. */ -lazy val tests: CrossProject = crossProject(JSPlatform, JVMPlatform) +lazy val tests: CrossProject = crossProject(JSPlatform, JVMPlatform, NativePlatform) .in(file("tests")) .dependsOn(core, laws % Test, kernelTestkit % Test, testkit % Test) .enablePlugins(BuildInfoPlugin, NoPublishPlugin) @@ -645,11 +667,7 @@ lazy val tests: CrossProject = crossProject(JSPlatform, JVMPlatform) name := "cats-effect-tests", libraryDependencies ++= Seq( "org.scalacheck" %%% "scalacheck" % ScalaCheckVersion, - ("org.specs2" %%% "specs2-scalacheck" % Specs2Version % Test) - .cross(CrossVersion.for3Use2_13) - .exclude("org.scala-js", "scala-js-macrotask-executor_sjs1_2.13") - .exclude("org.scalacheck", "scalacheck_2.13") - .exclude("org.scalacheck", "scalacheck_sjs1_2.13"), + "org.specs2" %%% "specs2-scalacheck" % Specs2Version % Test, "org.typelevel" %%% "discipline-specs2" % DisciplineVersion % Test, "org.typelevel" %%% "cats-kernel-laws" % CatsVersion % Test ), @@ -689,18 +707,14 @@ lazy val testsJVM = tests * implementations will require IO, and thus those tests will be located within the core * project. */ -lazy val std = crossProject(JSPlatform, JVMPlatform) +lazy val std = crossProject(JSPlatform, JVMPlatform, NativePlatform) .in(file("std")) .dependsOn(kernel) .settings( name := "cats-effect-std", libraryDependencies ++= Seq( "org.scalacheck" %%% "scalacheck" % ScalaCheckVersion % Test, - ("org.specs2" %%% "specs2-scalacheck" % Specs2Version % Test) - .cross(CrossVersion.for3Use2_13) - .exclude("org.scala-js", "scala-js-macrotask-executor_sjs1_2.13") - .exclude("org.scalacheck", "scalacheck_2.13") - .exclude("org.scalacheck", "scalacheck_sjs1_2.13") + "org.specs2" %%% "specs2-scalacheck" % Specs2Version % Test ) ) .jsSettings( @@ -711,7 +725,7 @@ lazy val std = crossProject(JSPlatform, JVMPlatform) * A trivial pair of trivial example apps primarily used to show that IOApp works as a practical * runtime on both target platforms. */ -lazy val example = crossProject(JSPlatform, JVMPlatform) +lazy val example = crossProject(JSPlatform, JVMPlatform, NativePlatform) .in(file("example")) .dependsOn(core) .enablePlugins(NoPublishPlugin) diff --git a/core/js/src/main/scala/cats/effect/IOFiberConstants.scala b/core/js-native/src/main/scala/cats/effect/IOFiberConstants.scala similarity index 100% rename from core/js/src/main/scala/cats/effect/IOFiberConstants.scala rename to core/js-native/src/main/scala/cats/effect/IOFiberConstants.scala diff --git a/core/js/src/main/scala/cats/effect/IOFiberPlatform.scala b/core/js-native/src/main/scala/cats/effect/IOFiberPlatform.scala similarity index 100% rename from core/js/src/main/scala/cats/effect/IOFiberPlatform.scala rename to core/js-native/src/main/scala/cats/effect/IOFiberPlatform.scala diff --git a/core/js/src/main/scala/cats/effect/SyncIOConstants.scala b/core/js-native/src/main/scala/cats/effect/SyncIOConstants.scala similarity index 100% rename from core/js/src/main/scala/cats/effect/SyncIOConstants.scala rename to core/js-native/src/main/scala/cats/effect/SyncIOConstants.scala diff --git a/core/js/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala b/core/js-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala similarity index 75% rename from core/js/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala rename to core/js-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala index a1cc68082a..03e0bcaac6 100644 --- a/core/js/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala +++ b/core/js-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala @@ -29,4 +29,14 @@ private[effect] sealed abstract class WorkStealingThreadPool private () private[effect] def rescheduleFiber(fiber: IOFiber[_]): Unit private[effect] def scheduleFiber(fiber: IOFiber[_]): Unit private[effect] def canExecuteBlockingCode(): Boolean + private[unsafe] def liveFibers(): ( + Set[IOFiber[_]], + Map[WorkerThread, (Option[IOFiber[_]], Set[IOFiber[_]])], + Set[IOFiber[_]]) = ??? +} + +private[unsafe] sealed abstract class WorkerThread private () extends Thread { + private[unsafe] def isOwnedBy(threadPool: WorkStealingThreadPool): Boolean + private[unsafe] def monitor(fiber: IOFiber[_]): WeakBag.Handle + private[unsafe] def index: Int } diff --git a/core/jvm/src/main/scala/cats/effect/SyncIOCompanionPlatform.scala b/core/jvm-native/src/main/scala/cats/effect/SyncIOCompanionPlatform.scala similarity index 100% rename from core/jvm/src/main/scala/cats/effect/SyncIOCompanionPlatform.scala rename to core/jvm-native/src/main/scala/cats/effect/SyncIOCompanionPlatform.scala diff --git a/core/jvm/src/main/scala/cats/effect/unsafe/FiberMonitor.scala b/core/jvm-native/src/main/scala/cats/effect/unsafe/FiberMonitor.scala similarity index 99% rename from core/jvm/src/main/scala/cats/effect/unsafe/FiberMonitor.scala rename to core/jvm-native/src/main/scala/cats/effect/unsafe/FiberMonitor.scala index 00abd59a64..70ec17ab53 100644 --- a/core/jvm/src/main/scala/cats/effect/unsafe/FiberMonitor.scala +++ b/core/jvm-native/src/main/scala/cats/effect/unsafe/FiberMonitor.scala @@ -169,6 +169,10 @@ private[effect] object FiberMonitor { } } + private[FiberMonitor] final val BagReferences + : ConcurrentLinkedQueue[WeakReference[WeakBag[IOFiber[_]]]] = + new ConcurrentLinkedQueue() + private[FiberMonitor] final val Bags: ThreadLocal[WeakBag[IOFiber[_]]] = ThreadLocal.withInitial { () => val bag = new WeakBag[IOFiber[_]]() @@ -176,7 +180,4 @@ private[effect] object FiberMonitor { bag } - private[FiberMonitor] final val BagReferences - : ConcurrentLinkedQueue[WeakReference[WeakBag[IOFiber[_]]]] = - new ConcurrentLinkedQueue() } diff --git a/core/jvm/src/main/scala/cats/effect/unsafe/ref/package.scala b/core/jvm-native/src/main/scala/cats/effect/unsafe/ref/package.scala similarity index 100% rename from core/jvm/src/main/scala/cats/effect/unsafe/ref/package.scala rename to core/jvm-native/src/main/scala/cats/effect/unsafe/ref/package.scala diff --git a/core/jvm/src/main/scala/cats/effect/tracing/TracingPlatform.scala b/core/jvm/src/main/scala/cats/effect/tracing/TracingPlatform.scala index 5629e37e6c..1d6f03ec08 100644 --- a/core/jvm/src/main/scala/cats/effect/tracing/TracingPlatform.scala +++ b/core/jvm/src/main/scala/cats/effect/tracing/TracingPlatform.scala @@ -39,7 +39,7 @@ private[tracing] abstract class TracingPlatform extends ClassValue[TracingEvent] } } - @nowarn("cat=unused") + @nowarn("msg=never used") private[tracing] def applyStackTraceFilter( callSiteClassName: String, callSiteMethodName: String, diff --git a/core/native/src/main/scala/cats/effect/IOApp.scala b/core/native/src/main/scala/cats/effect/IOApp.scala new file mode 100644 index 0000000000..8d5f0a5bf0 --- /dev/null +++ b/core/native/src/main/scala/cats/effect/IOApp.scala @@ -0,0 +1,244 @@ +/* + * 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 + +import scala.concurrent.CancellationException +import scala.concurrent.duration._ + +/** + * The primary entry point to a Cats Effect application. Extend this trait rather than defining + * your own `main` method. This avoids the need to run [[IO.unsafeRunAsync]] (or similar) on + * your own. + * + * `IOApp` takes care of the messy details of properly setting up (and tearing down) the + * [[unsafe.IORuntime]] needed to run the [[IO]] which represents your application. All of the + * associated thread pools (if relevant) will be configured with the assumption that your + * application is fully contained within the `IO` produced by the [[run]] method. Note that the + * exact details of how the runtime will be configured are very platform-specific. Part of the + * point of `IOApp` is to insulate users from the details of the underlying runtime (whether JVM + * or JavaScript). + * + * {{{ + * object MyApplication extends IOApp { + * def run(args: List[String]) = + * for { + * _ <- IO.print("Enter your name: ") + * name <- IO.readln + * _ <- IO.println("Hello, " + name) + * } yield ExitCode.Success + * } + * }}} + * + * In the above example, `MyApplication` will be a runnable class with a `main` method, visible + * to Sbt, IntelliJ, or plain-old `java`. When run externally, it will print, read, and print in + * the obvious way, producing a final process exit code of 0. Any exceptions thrown within the + * `IO` will be printed to standard error and the exit code will be set to 1. In the event that + * the main [[Fiber]] (represented by the `IO` returned by `run`) is canceled, the runtime will + * produce an exit code of 1. + * + * Note that exit codes are an implementation-specific feature of the underlying runtime, as are + * process arguments. Naturally, all JVMs support these functions, as does NodeJS, but some + * JavaScript execution environments will be unable to replicate these features (or they simply + * may not make sense). In such cases, exit codes may be ignored and/or argument lists may be + * empty. + * + * Note that in the case of the above example, we would actually be better off using + * [[IOApp.Simple]] rather than `IOApp` directly, since we are neither using `args` nor are we + * explicitly producing a custom [[ExitCode]]: + * + * {{{ + * object MyApplication extends IOApp.Simple { + * val run = + * for { + * _ <- IO.print("Enter your name: ") + * name <- IO.readln + * _ <- IO.println(s"Hello, " + name) + * } yield () + * } + * }}} + * + * It is valid to define `val run` rather than `def run` because `IO`'s evaluation is lazy: it + * will only run when the `main` method is invoked by the runtime. + * + * In the event that the process receives an interrupt signal (`SIGINT`) due to Ctrl-C (or any + * other mechanism), it will immediately `cancel` the main fiber. Assuming this fiber is not + * within an `uncancelable` region, this will result in interrupting any current activities and + * immediately invoking any finalizers (see: [[IO.onCancel]] and [[IO.bracket]]). The process + * will not shut down until the finalizers have completed. For example: + * + * {{{ + * object InterruptExample extends IOApp.Simple { + * val run = + * IO.bracket(startServer)( + * _ => IO.never)( + * server => IO.println("shutting down") *> server.close) + * } + * }}} + * + * If we assume the `startServer` function has type `IO[Server]` (or similar), this kind of + * pattern is very common. When this process receives a `SIGINT`, it will immediately print + * "shutting down" and run the `server.close` effect. + * + * One consequence of this design is it is possible to build applications which will ignore + * process interrupts. For example, if `server.close` runs forever, the process will ignore + * interrupts and will need to be cleaned up using `SIGKILL` (i.e. `kill -9`). This same + * phenomenon can be demonstrated by using [[IO.uncancelable]] to suppress all interruption + * within the application itself: + * + * {{{ + * object Zombie extends IOApp.Simple { + * val run = IO.never.uncancelable + * } + * }}} + * + * The above process will run forever and ignore all interrupts. The only way it will shut down + * is if it receives `SIGKILL`. + * + * It is possible (though not necessary) to override various platform-specific runtime + * configuration options, such as `computeWorkerThreadCount` (which only exists on the JVM). + * Please note that the default configurations have been extensively benchmarked and are optimal + * (or close to it) in most conventional scenarios. + * + * However, with that said, there really is no substitute to benchmarking your own application. + * Every application and scenario is unique, and you will always get the absolute best results + * by performing your own tuning rather than trusting someone else's defaults. `IOApp`'s + * defaults are very ''good'', but they are not perfect in all cases. One common example of this + * is applications which maintain network or file I/O worker threads which are under heavy load + * in steady-state operations. In such a performance profile, it is usually better to reduce the + * number of compute worker threads to "make room" for the I/O workers, such that they all sum + * to the number of physical threads exposed by the kernel. + * + * @see + * [[IO]] + * @see + * [[run]] + * @see + * [[ResourceApp]] + * @see + * [[IOApp.Simple]] + */ +trait IOApp { + + private[this] var _runtime: unsafe.IORuntime = null + + /** + * The runtime which will be used by `IOApp` to evaluate the [[IO]] produced by the `run` + * method. This may be overridden by `IOApp` implementations which have extremely specialized + * needs, but this is highly unlikely to ever be truly needed. As an example, if an + * application wishes to make use of an alternative compute thread pool (such as + * `Executors.fixedThreadPool`), it is almost always better to leverage [[IO.evalOn]] on the + * value produced by the `run` method, rather than directly overriding `runtime`. + * + * In other words, this method is made available to users, but its use is strongly discouraged + * in favor of other, more precise solutions to specific use-cases. + * + * This value is guaranteed to be equal to [[unsafe.IORuntime.global]]. + */ + protected def runtime: unsafe.IORuntime = _runtime + + /** + * The configuration used to initialize the [[runtime]] which will evaluate the [[IO]] + * produced by `run`. It is very unlikely that users will need to override this method. + */ + protected def runtimeConfig: unsafe.IORuntimeConfig = unsafe.IORuntimeConfig() + + /** + * The entry point for your application. Will be called by the runtime when the process is + * started. If the underlying runtime supports it, any arguments passed to the process will be + * made available in the `args` parameter. The numeric value within the resulting [[ExitCode]] + * will be used as the exit code when the process terminates unless terminated exceptionally + * or by interrupt. + * + * @param args + * The arguments passed to the process, if supported by the underlying runtime. For example, + * `java com.company.MyApp --foo --bar baz` or `node com-mycompany-fastopt.js --foo --bar + * baz` would each result in `List("--foo", "--bar", "baz")`. + * @see + * [[IOApp.Simple!.run:cats\.effect\.IO[Unit]*]] + */ + def run(args: List[String]): IO[ExitCode] + + final def main(args: Array[String]): Unit = { + if (runtime == null) { + import unsafe.IORuntime + + val installed = IORuntime installGlobal { + IORuntime( + IORuntime.defaultComputeExecutionContext, + IORuntime.defaultComputeExecutionContext, + IORuntime.defaultScheduler, + () => (), + runtimeConfig) + } + + if (!installed) { + System + .err + .println( + "WARNING: Cats Effect global runtime already initialized; custom configurations will be ignored") + } + + _runtime = IORuntime.global + } + + // An infinite heartbeat to keep main alive. This is similar to + // `IO.never`, except `IO.never` doesn't schedule any tasks and is + // insufficient to keep main alive. The tick is fast enough that + // it isn't silently discarded, as longer ticks are, but slow + // enough that we don't interrupt often. 1 hour was chosen + // empirically. + lazy val keepAlive: IO[Nothing] = + IO.sleep(1.hour) >> keepAlive + + Spawn[IO] + .raceOutcome[ExitCode, Nothing](run(args.toList), keepAlive) + .flatMap { + case Left(Outcome.Canceled()) => + IO.raiseError(new CancellationException("IOApp main fiber was canceled")) + case Left(Outcome.Errored(t)) => IO.raiseError(t) + case Left(Outcome.Succeeded(code)) => code + case Right(Outcome.Errored(t)) => IO.raiseError(t) + case Right(_) => sys.error("impossible") + } + .unsafeRunFiber( + System.exit(0), + t => { + t.printStackTrace() + System.exit(1) + }, + c => System.exit(c.code) + )(runtime) + + () + } + +} + +object IOApp { + + /** + * A simplified version of [[IOApp]] for applications which ignore their process arguments and + * always produces [[ExitCode.Success]] (unless terminated exceptionally or interrupted). + * + * @see + * [[IOApp]] + */ + trait Simple extends IOApp { + def run: IO[Unit] + final def run(args: List[String]): IO[ExitCode] = run.as(ExitCode.Success) + } +} diff --git a/core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala b/core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala new file mode 100644 index 0000000000..6025313575 --- /dev/null +++ b/core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala @@ -0,0 +1,43 @@ +/* + * 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 + +import java.time.Instant + +private[effect] abstract class IOCompanionPlatform { this: IO.type => + + def blocking[A](thunk: => A): IO[A] = + // do our best to mitigate blocking + IO.cede *> apply(thunk).guarantee(IO.cede) + + private[effect] def interruptible[A](many: Boolean, thunk: => A): IO[A] = { + val _ = many + blocking(thunk) + } + + def interruptible[A](thunk: => A): IO[A] = interruptible(false, thunk) + + def interruptibleMany[A](thunk: => A): IO[A] = interruptible(true, thunk) + + def suspend[A](hint: Sync.Type)(thunk: => A): IO[A] = { + val _ = hint + apply(thunk) + } + + def realTimeInstant: IO[Instant] = asyncForIO.realTimeInstant + +} diff --git a/core/native/src/main/scala/cats/effect/IOPlatform.scala b/core/native/src/main/scala/cats/effect/IOPlatform.scala new file mode 100644 index 0000000000..50edeb36d6 --- /dev/null +++ b/core/native/src/main/scala/cats/effect/IOPlatform.scala @@ -0,0 +1,19 @@ +/* + * 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 + +abstract private[effect] class IOPlatform[+A] diff --git a/core/native/src/main/scala/cats/effect/tracing/TracingConstants.scala b/core/native/src/main/scala/cats/effect/tracing/TracingConstants.scala new file mode 100644 index 0000000000..3c57fdcd3e --- /dev/null +++ b/core/native/src/main/scala/cats/effect/tracing/TracingConstants.scala @@ -0,0 +1,27 @@ +/* + * 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 tracing + +private[effect] object TracingConstants { + + final val isCachedStackTracing = false + + final val isFullStackTracing = false + + final val isStackTracing = isFullStackTracing || isCachedStackTracing +} diff --git a/core/native/src/main/scala/cats/effect/tracing/TracingPlatform.scala b/core/native/src/main/scala/cats/effect/tracing/TracingPlatform.scala new file mode 100644 index 0000000000..f214f0a832 --- /dev/null +++ b/core/native/src/main/scala/cats/effect/tracing/TracingPlatform.scala @@ -0,0 +1,34 @@ +/* + * 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.tracing + +import scala.annotation.nowarn + +private[tracing] abstract class TracingPlatform { self: Tracing.type => + + @nowarn("msg=never used") + def calculateTracingEvent(key: Any): TracingEvent = null + + @nowarn("msg=never used") + private[tracing] def applyStackTraceFilter( + callSiteClassName: String, + callSiteMethodName: String, + callSiteFileName: String): Boolean = false + + private[tracing] def decodeMethodName(name: String): String = name + +} diff --git a/core/native/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala b/core/native/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala new file mode 100644 index 0000000000..4b5104e751 --- /dev/null +++ b/core/native/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala @@ -0,0 +1,60 @@ +/* + * 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.unsafe + +import scala.concurrent.ExecutionContext + +private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type => + + def defaultComputeExecutionContext: ExecutionContext = QueueExecutorScheduler + + def defaultScheduler: Scheduler = QueueExecutorScheduler + + private[this] var _global: IORuntime = null + + def installGlobal(global: => IORuntime): Boolean = { + if (_global == null) { + _global = global + true + } else { + false + } + } + + private[effect] def resetGlobal(): Unit = + _global = null + + lazy val global: IORuntime = { + if (_global == null) { + installGlobal { + IORuntime( + defaultComputeExecutionContext, + defaultComputeExecutionContext, + defaultScheduler, + () => (), + IORuntimeConfig()) + } + } + + _global + } + + private[effect] def registerFiberMonitorMBean(fiberMonitor: FiberMonitor): () => Unit = { + val _ = fiberMonitor + () => () + } +} diff --git a/core/native/src/main/scala/cats/effect/unsafe/IORuntimeConfigCompanionPlatform.scala b/core/native/src/main/scala/cats/effect/unsafe/IORuntimeConfigCompanionPlatform.scala new file mode 100644 index 0000000000..d363aada8b --- /dev/null +++ b/core/native/src/main/scala/cats/effect/unsafe/IORuntimeConfigCompanionPlatform.scala @@ -0,0 +1,55 @@ +/* + * 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.duration.Duration +import scala.util.Try + +private[unsafe] abstract class IORuntimeConfigCompanionPlatform { this: IORuntimeConfig.type => + // TODO make the cancelation and auto-yield properties have saner names + protected final val Default: IORuntimeConfig = { + val cancelationCheckThreshold = + Option(System.getenv("CATS_EFFECT_CANCELATION_CHECK_THRESHOLD")) + .flatMap(x => Try(x.toInt).toOption) + .getOrElse(512) + + val autoYieldThreshold = + Option(System.getenv("CATS_EFFECT_AUTO_YIELD_THRESHOLD_MULTIPLIER")) + .flatMap(x => Try(x.toInt).toOption) + .getOrElse(2) * cancelationCheckThreshold + + val enhancedExceptions = Option(System.getenv("CATS_EFFECT_TRACING_EXCEPTIONS_ENHANCED")) + .flatMap(x => Try(x.toBoolean).toOption) + .getOrElse(DefaultEnhancedExceptions) + + val traceBufferSize = Option(System.getenv("CATS_EFFECT_TRACING_BUFFER_SIZE")) + .flatMap(x => Try(x.toInt).toOption) + .getOrElse(DefaultTraceBufferSize) + + val shutdownHookTimeout = Option(System.getenv("CATS_EFFECT_SHUTDOWN_HOOK_TIMEOUT")) + .flatMap(x => Try(Duration(x)).toOption) + .getOrElse(DefaultShutdownHookTimeout) + + apply( + cancelationCheckThreshold, + autoYieldThreshold, + enhancedExceptions, + traceBufferSize, + shutdownHookTimeout) + } +} diff --git a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala new file mode 100644 index 0000000000..1057d776f7 --- /dev/null +++ b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala @@ -0,0 +1,163 @@ +/* + * 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.unsafe + +import scala.concurrent.{ExecutionContext, ExecutionContextExecutor} +import scala.concurrent.duration._ +import scala.util.control.NonFatal + +import java.util.{ArrayDeque, PriorityQueue} + +abstract class PollingExecutorScheduler extends ExecutionContextExecutor with Scheduler { + + import PollingExecutorScheduler._ + + private[this] var needsReschedule: Boolean = true + private[this] var inLoop: Boolean = false + private[this] var cachedNow: Long = _ + + private[this] var executeQueue: ArrayDeque[Runnable] = new ArrayDeque + private[this] var cachedExecuteQueue: ArrayDeque[Runnable] = new ArrayDeque + private[this] val sleepQueue: PriorityQueue[ScheduledTask] = new PriorityQueue + + private[this] val noop: Runnable = () => () + + private[this] def scheduleIfNeeded(): Unit = if (needsReschedule) { + ExecutionContext.global.execute(() => loop()) + needsReschedule = false + } + + final def execute(runnable: Runnable): Unit = { + scheduleIfNeeded() + executeQueue.addLast(runnable) + } + + final def sleep(delay: FiniteDuration, task: Runnable): Runnable = + if (delay == Duration.Zero) { + execute(task) + noop + } else { + scheduleIfNeeded() + val now = if (inLoop) cachedNow else monotonicNanos() + val scheduledTask = new ScheduledTask(now + delay.toNanos, task) + sleepQueue.offer(scheduledTask) + scheduledTask + } + + def reportFailure(t: Throwable): Unit = t.printStackTrace() + + def nowMillis() = System.currentTimeMillis() + + def monotonicNanos() = System.nanoTime() + + /** + * @param timeout + * the maximum duration for which to block. ''However'', if `timeout == Inf` and there are + * no remaining events to poll for, this method should return `false` immediately. This is + * unfortunate but necessary so that this `ExecutionContext` can yield to the Scala Native + * global `ExecutionContext` which is currently hard-coded into every test framework, + * including JUnit, MUnit, and specs2. + * + * @return + * whether poll should be called again (i.e., there are more events to be polled) + */ + protected def poll(timeout: Duration): Boolean + + private[this] def loop(): Unit = { + needsReschedule = false + inLoop = true + + var continue = true + + while (continue) { + // cache the timestamp for this tick + cachedNow = monotonicNanos() + + // execute the timers + while (!sleepQueue.isEmpty() && sleepQueue.peek().canceled) { + sleepQueue.poll() + } + + while (!sleepQueue.isEmpty() && sleepQueue.peek().at <= cachedNow) { + val task = sleepQueue.poll() + try { + task.runnable.run() + } catch { + case NonFatal(t) => + reportFailure(t) + } + } + + // swap the task queues + val todo = executeQueue + executeQueue = cachedExecuteQueue + cachedExecuteQueue = todo + + // do all the tasks + while (!todo.isEmpty()) { + val runnable = todo.poll() + try { + runnable.run() + } catch { + case NonFatal(t) => + reportFailure(t) + } + } + + // cleanup canceled timers + val sleepIter = sleepQueue.iterator() + while (sleepIter.hasNext()) { + if (sleepIter.next().canceled) sleepIter.remove() + } + + // finally we poll + val timeout = + if (!executeQueue.isEmpty()) + Duration.Zero + else if (!sleepQueue.isEmpty()) + (sleepQueue.peek().at - cachedNow).nanos + else + Duration.Inf + + val needsPoll = poll(timeout) + + continue = needsPoll || !executeQueue.isEmpty() || !sleepQueue.isEmpty() + } + + needsReschedule = true + inLoop = false + } + +} + +object PollingExecutorScheduler { + + private final class ScheduledTask( + val at: Long, + val runnable: Runnable, + var canceled: Boolean = false + ) extends Runnable + with Comparable[ScheduledTask] { + + def run(): Unit = canceled = true + + def compareTo(that: ScheduledTask): Int = + java.lang.Long.compare(this.at, that.at) + + } + +} diff --git a/core/native/src/main/scala/cats/effect/unsafe/QueueExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/QueueExecutorScheduler.scala new file mode 100644 index 0000000000..99595b79b3 --- /dev/null +++ b/core/native/src/main/scala/cats/effect/unsafe/QueueExecutorScheduler.scala @@ -0,0 +1,31 @@ +/* + * 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.unsafe + +import scala.concurrent.duration._ + +private[effect] object QueueExecutorScheduler extends PollingExecutorScheduler { + + def poll(timeout: Duration): Boolean = { + if (timeout != Duration.Zero && timeout.isFinite) { + val nanos = timeout.toNanos + Thread.sleep(nanos / 1000000, (nanos % 1000000).toInt) + } + false + } + +} diff --git a/core/native/src/main/scala/cats/effect/unsafe/SchedulerCompanionPlatform.scala b/core/native/src/main/scala/cats/effect/unsafe/SchedulerCompanionPlatform.scala new file mode 100644 index 0000000000..f6e4964808 --- /dev/null +++ b/core/native/src/main/scala/cats/effect/unsafe/SchedulerCompanionPlatform.scala @@ -0,0 +1,23 @@ +/* + * 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.unsafe + +private[unsafe] abstract class SchedulerCompanionPlatform { this: Scheduler.type => + + def createDefaultScheduler(): (Scheduler, () => Unit) = (QueueExecutorScheduler, () => ()) + +} diff --git a/example/js/src/main/scala/cats/effect/example/Example.scala b/example/js-native/src/main/scala/cats/effect/example/Example.scala similarity index 100% rename from example/js/src/main/scala/cats/effect/example/Example.scala rename to example/js-native/src/main/scala/cats/effect/example/Example.scala diff --git a/example/test-native.sh b/example/test-native.sh new file mode 100755 index 0000000000..79b6318fa5 --- /dev/null +++ b/example/test-native.sh @@ -0,0 +1,47 @@ +#!/usr/bin/env bash + +# This script mostly just ensures that we can use native to run an example application. + +set -euo pipefail # STRICT MODE +IFS=$'\n\t' # http://redsymbol.net/articles/unofficial-bash-strict-mode/ + +cd $(dirname $0)/.. + +sbt ++$1 exampleNative/nativeLink + +output=$(mktemp) +expected=$(mktemp) + +cd example/native/target/scala-$(echo $1 | sed -E 's/^(2\.[0-9]+)\.[0-9]+$/\1/')/ + +set +e +./cats-effect-example-out left right > $output +result=$? +set -e + +if [[ $result -ne 2 ]]; then + exit 1 +fi + +echo $'left +left +left +left +left +right +right +right +right +right +left +left +left +left +left +right +right +right +right +right' > $expected + +exec diff $output $expected diff --git a/kernel/jvm/src/main/scala/cats/effect/kernel/ClockPlatform.scala b/kernel/jvm-native/src/main/scala/cats/effect/kernel/ClockPlatform.scala similarity index 100% rename from kernel/jvm/src/main/scala/cats/effect/kernel/ClockPlatform.scala rename to kernel/jvm-native/src/main/scala/cats/effect/kernel/ClockPlatform.scala diff --git a/kernel/native/src/main/scala/cats/effect/kernel/AsyncPlatform.scala b/kernel/native/src/main/scala/cats/effect/kernel/AsyncPlatform.scala new file mode 100644 index 0000000000..d05bcc3700 --- /dev/null +++ b/kernel/native/src/main/scala/cats/effect/kernel/AsyncPlatform.scala @@ -0,0 +1,19 @@ +/* + * 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.kernel + +private[kernel] trait AsyncPlatform[F[_]] diff --git a/kernel/native/src/main/scala/cats/effect/kernel/ResourcePlatform.scala b/kernel/native/src/main/scala/cats/effect/kernel/ResourcePlatform.scala new file mode 100644 index 0000000000..8ea619eed5 --- /dev/null +++ b/kernel/native/src/main/scala/cats/effect/kernel/ResourcePlatform.scala @@ -0,0 +1,19 @@ +/* + * 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.kernel + +private[effect] trait ResourcePlatform extends Serializable diff --git a/laws/shared/src/test/scala/cats/effect/laws/ReaderWriterStateTFreeSyncSpec.scala b/laws/shared/src/test/scala/cats/effect/laws/ReaderWriterStateTFreeSyncSpec.scala index 2cff66d431..5ce5ca1596 100644 --- a/laws/shared/src/test/scala/cats/effect/laws/ReaderWriterStateTFreeSyncSpec.scala +++ b/laws/shared/src/test/scala/cats/effect/laws/ReaderWriterStateTFreeSyncSpec.scala @@ -24,6 +24,7 @@ import cats.free.FreeT import cats.laws.discipline.{arbitrary, MiniInt}, arbitrary._ import freeEval.{syncForFreeT, FreeEitherSync} import org.specs2.mutable._ +import org.specs2.scalacheck._ import org.typelevel.discipline.specs2.mutable.Discipline class ReaderWriterStateTFreeSyncSpec @@ -34,6 +35,12 @@ class ReaderWriterStateTFreeSyncSpec import FreeSyncGenerators._ import SyncTypeGenerators._ + implicit val params: Parameters = + if (cats.platform.Platform.isNative) + Parameters(minTestsOk = 5) + else + Parameters(minTestsOk = 100) + implicit val scala_2_12_is_buggy : Eq[FreeT[Eval, Either[Throwable, *], Either[Int, Either[Throwable, Int]]]] = eqFreeSync[Either[Throwable, *], Either[Int, Either[Throwable, Int]]] diff --git a/project/CI.scala b/project/CI.scala index 2bb32ae71e..59471be878 100644 --- a/project/CI.scala +++ b/project/CI.scala @@ -64,6 +64,16 @@ object CI { mimaReport = true, suffixCommands = List("exampleJS/compile")) + case object Native + extends CI( + command = "ciNative", + rootProject = "rootNative", + jsEnv = None, + testCommands = List("test"), + mimaReport = true, + suffixCommands = List("exampleNative/compile") + ) + case object Firefox extends CI( command = "ciFirefox", @@ -83,5 +93,5 @@ object CI { suffixCommands = List()) val AllJSCIs: List[CI] = List(JS, Firefox, Chrome) - val AllCIs: List[CI] = JVM :: AllJSCIs + val AllCIs: List[CI] = JVM :: Native :: AllJSCIs } diff --git a/project/Common.scala b/project/Common.scala index a6270b5e60..5764767128 100644 --- a/project/Common.scala +++ b/project/Common.scala @@ -17,7 +17,10 @@ import sbt._, Keys._ import de.heikoseeberger.sbtheader.HeaderPlugin.autoImport._ +import org.typelevel.sbt.TypelevelMimaPlugin.autoImport._ import org.typelevel.sbt.TypelevelPlugin +import sbtcrossproject.CrossPlugin.autoImport._ +import scalanativecrossproject.NativePlatform object Common extends AutoPlugin { @@ -28,6 +31,12 @@ object Common extends AutoPlugin { Seq( headerLicense := Some( HeaderLicense.ALv2(s"${startYear.value.get}-2022", organizationName.value) - ) + ), + tlVersionIntroduced ++= { + if (crossProjectPlatform.?.value.contains(NativePlatform)) + List("2.12", "2.13", "3").map(_ -> "3.4.0").toMap + else + Map.empty + } ) } diff --git a/project/plugins.sbt b/project/plugins.sbt index b691eab7ab..cd57e0e82e 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -2,11 +2,13 @@ libraryDependencies += "org.scala-js" %% "scalajs-env-selenium" % "1.1.1" addSbtPlugin("org.typelevel" % "sbt-typelevel" % "0.5.0-M1") -addSbtPlugin("org.scala-js" % "sbt-scalajs" % "1.7.1") -addSbtPlugin("org.portable-scala" % "sbt-scalajs-crossproject" % "1.1.0") +addSbtPlugin("org.scala-js" % "sbt-scalajs" % "1.10.1") +addSbtPlugin("org.portable-scala" % "sbt-scalajs-crossproject" % "1.2.0") +addSbtPlugin("org.scala-native" % "sbt-scala-native" % "0.4.7") +addSbtPlugin("org.portable-scala" % "sbt-scala-native-crossproject" % "1.2.0") addSbtPlugin("pl.project13.scala" % "sbt-jmh" % "0.4.3") addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.5") -addSbtPlugin("org.scalameta" % "sbt-mdoc" % "2.2.23") +addSbtPlugin("org.scalameta" % "sbt-mdoc" % "2.3.3") addSbtPlugin("com.github.sbt" % "sbt-unidoc" % "0.5.0") addSbtPlugin("com.lightbend.sbt" % "sbt-java-formatter" % "0.7.0") addSbtPlugin("com.eed3si9n" % "sbt-buildinfo" % "0.10.0") diff --git a/std/js/src/main/scala/cats/effect/std/RandomCompanionPlatform.scala b/std/js-native/src/main/scala/cats/effect/std/RandomCompanionPlatform.scala similarity index 100% rename from std/js/src/main/scala/cats/effect/std/RandomCompanionPlatform.scala rename to std/js-native/src/main/scala/cats/effect/std/RandomCompanionPlatform.scala diff --git a/std/js/src/main/scala/cats/effect/std/UUIDGenCompanionPlatform.scala b/std/js-native/src/main/scala/cats/effect/std/UUIDGenCompanionPlatform.scala similarity index 100% rename from std/js/src/main/scala/cats/effect/std/UUIDGenCompanionPlatform.scala rename to std/js-native/src/main/scala/cats/effect/std/UUIDGenCompanionPlatform.scala diff --git a/std/native/src/main/scala/cats/effect/std/DispatcherPlatform.scala b/std/native/src/main/scala/cats/effect/std/DispatcherPlatform.scala new file mode 100644 index 0000000000..b0bbdbfbfe --- /dev/null +++ b/std/native/src/main/scala/cats/effect/std/DispatcherPlatform.scala @@ -0,0 +1,19 @@ +/* + * 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.std + +private[std] trait DispatcherPlatform[F[_]] diff --git a/std/native/src/main/scala/cats/effect/std/JavaSecureRandom.scala b/std/native/src/main/scala/cats/effect/std/JavaSecureRandom.scala new file mode 100644 index 0000000000..77f70493af --- /dev/null +++ b/std/native/src/main/scala/cats/effect/std/JavaSecureRandom.scala @@ -0,0 +1,63 @@ +/* + * 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.std + +import scala.annotation.nowarn +import scala.scalanative.libc.errno +import scala.scalanative.unsafe._ +import scala.scalanative.unsigned._ + +private[std] class JavaSecureRandom() extends java.util.Random(0L) { + + override def setSeed(x: Long): Unit = () + + override def nextBytes(bytes: Array[Byte]): Unit = { + val len = bytes.length + val buffer = stackalloc[Byte](256) + var i = 0 + while (i < len) { + val n = Math.min(256, len - i) + if (sysrandom.getentropy(buffer, n.toULong) < 0) + throw new RuntimeException(s"getentropy: ${errno.errno}") + + var j = 0L + while (j < n) { + bytes(i) = buffer(j) + i += 1 + j += 1 + } + } + } + + override protected final def next(numBits: Int): Int = { + if (numBits <= 0) { + 0 // special case because the formula on the last line is incorrect for numBits == 0 + } else { + val bytes = stackalloc[CInt]() + sysrandom.getentropy(bytes.asInstanceOf[Ptr[Byte]], sizeof[CInt]) + val rand32: Int = !bytes + rand32 & (-1 >>> (32 - numBits)) // Clear the (32 - numBits) higher order bits + } + } + +} + +@extern +@nowarn +private[std] object sysrandom { + def getentropy(buf: Ptr[Byte], buflen: CSize): Int = extern +} diff --git a/tests/shared/src/test/scala/cats/effect/IOFiberSpec.scala b/tests/js-jvm/src/test/scala/cats/effect/IOFiberSpec.scala similarity index 100% rename from tests/shared/src/test/scala/cats/effect/IOFiberSpec.scala rename to tests/js-jvm/src/test/scala/cats/effect/IOFiberSpec.scala diff --git a/tests/shared/src/test/scala/cats/effect/tracing/TraceSpec.scala b/tests/js-jvm/src/test/scala/cats/effect/tracing/TraceSpec.scala similarity index 100% rename from tests/shared/src/test/scala/cats/effect/tracing/TraceSpec.scala rename to tests/js-jvm/src/test/scala/cats/effect/tracing/TraceSpec.scala diff --git a/tests/shared/src/test/scala/cats/effect/tracing/TracingSpec.scala b/tests/js-jvm/src/test/scala/cats/effect/tracing/TracingSpec.scala similarity index 100% rename from tests/shared/src/test/scala/cats/effect/tracing/TracingSpec.scala rename to tests/js-jvm/src/test/scala/cats/effect/tracing/TracingSpec.scala diff --git a/tests/js/src/test/scala/cats/effect/ContSpecBasePlatform.scala b/tests/js-native/src/test/scala/cats/effect/ContSpecBasePlatform.scala similarity index 100% rename from tests/js/src/test/scala/cats/effect/ContSpecBasePlatform.scala rename to tests/js-native/src/test/scala/cats/effect/ContSpecBasePlatform.scala diff --git a/tests/js/src/main/scala/cats/effect/DetectPlatform.scala b/tests/js/src/main/scala/cats/effect/DetectPlatform.scala index ebec3eaefd..5d7da3968c 100644 --- a/tests/js/src/main/scala/cats/effect/DetectPlatform.scala +++ b/tests/js/src/main/scala/cats/effect/DetectPlatform.scala @@ -36,4 +36,5 @@ trait DetectPlatform { } def isJS: Boolean = true + def isNative: Boolean = false } diff --git a/tests/js/src/main/scala/catseffect/examplesplatform.scala b/tests/js/src/main/scala/catseffect/examplesplatform.scala index b9405510ab..8656f5ad01 100644 --- a/tests/js/src/main/scala/catseffect/examplesplatform.scala +++ b/tests/js/src/main/scala/catseffect/examplesplatform.scala @@ -49,7 +49,7 @@ package examples { register(LeakedFiber) register(UndefinedProcessExit) - @nowarn("cat=unused") + @nowarn("msg=never used") def main(paperweight: Array[String]): Unit = { val args = js.Dynamic.global.process.argv.asInstanceOf[js.Array[String]] val app = args(2) diff --git a/tests/jvm/src/test/scala/cats/effect/SyncIOPlatformSpecification.scala b/tests/jvm-native/src/test/scala/cats/effect/SyncIOPlatformSpecification.scala similarity index 100% rename from tests/jvm/src/test/scala/cats/effect/SyncIOPlatformSpecification.scala rename to tests/jvm-native/src/test/scala/cats/effect/SyncIOPlatformSpecification.scala diff --git a/tests/jvm/src/test/scala/cats/effect/DetectPlatform.scala b/tests/jvm/src/test/scala/cats/effect/DetectPlatform.scala index 321fe8ac34..8131a2c7ec 100644 --- a/tests/jvm/src/test/scala/cats/effect/DetectPlatform.scala +++ b/tests/jvm/src/test/scala/cats/effect/DetectPlatform.scala @@ -19,4 +19,5 @@ package cats.effect trait DetectPlatform { def isWSL: Boolean = System.getProperty("os.version").contains("-WSL") def isJS: Boolean = false + def isNative: Boolean = false } diff --git a/tests/native/src/test/scala/cats/effect/DetectPlatform.scala b/tests/native/src/test/scala/cats/effect/DetectPlatform.scala new file mode 100644 index 0000000000..252b674256 --- /dev/null +++ b/tests/native/src/test/scala/cats/effect/DetectPlatform.scala @@ -0,0 +1,23 @@ +/* + * 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 + +trait DetectPlatform { + def isWSL: Boolean = System.getProperty("os.version").contains("-WSL") + def isJS: Boolean = false + def isNative: Boolean = true +} diff --git a/tests/native/src/test/scala/cats/effect/IOPlatformSpecification.scala b/tests/native/src/test/scala/cats/effect/IOPlatformSpecification.scala new file mode 100644 index 0000000000..2aecef7f76 --- /dev/null +++ b/tests/native/src/test/scala/cats/effect/IOPlatformSpecification.scala @@ -0,0 +1,34 @@ +/* + * 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 + +import org.specs2.ScalaCheck + +trait IOPlatformSpecification { self: BaseSpec with ScalaCheck => + + def platformSpecs = "platform" should { + "realTimeInstant should return an Instant constructed from realTime" in ticked { + implicit ticker => + val op = for { + now <- IO.realTimeInstant + realTime <- IO.realTime + } yield now.toEpochMilli == realTime.toMillis + + op must completeAs(true) + } + } +} diff --git a/tests/native/src/test/scala/cats/effect/RunnersPlatform.scala b/tests/native/src/test/scala/cats/effect/RunnersPlatform.scala new file mode 100644 index 0000000000..5aaa1d4418 --- /dev/null +++ b/tests/native/src/test/scala/cats/effect/RunnersPlatform.scala @@ -0,0 +1,23 @@ +/* + * 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 + +import cats.effect.unsafe._ + +trait RunnersPlatform { + protected def runtime(): IORuntime = IORuntime.global +} diff --git a/tests/shared/src/test/scala/cats/effect/std/RandomSpec.scala b/tests/shared/src/test/scala/cats/effect/std/RandomSpec.scala index 47257fb840..00b08eb915 100644 --- a/tests/shared/src/test/scala/cats/effect/std/RandomSpec.scala +++ b/tests/shared/src/test/scala/cats/effect/std/RandomSpec.scala @@ -26,7 +26,17 @@ class RandomSpec extends BaseSpec { bytes1 <- random1.nextBytes(128) random2 <- Random.javaSecuritySecureRandom[IO](2) bytes2 <- random2.nextBytes(256) - } yield bytes1.length == 128 && bytes2.length == 256 + bytes3 <- random2.nextBytes(1024) + } yield bytes1.length == 128 && bytes2.length == 256 && bytes3.length == 1024 + } + + "overrides SecureRandom#next" in real { + for { + secureRandom <- Random.javaSecuritySecureRandom[IO] + secureInts <- secureRandom.nextInt.replicateA(3) + insecureRandom <- Random.scalaUtilRandomSeedInt[IO](0) + insecureInts <- insecureRandom.nextInt.replicateA(3) + } yield secureInts != insecureInts } "prevent array reference from leaking in ThreadLocalRandom.nextBytes impl" in real { From 097252152ddb4c33ac923106936416d092960dfc Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Sun, 11 Sep 2022 11:16:22 -0700 Subject: [PATCH 2/8] Gracefully handle negative sleeps --- .../scala/cats/effect/unsafe/PollingExecutorScheduler.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala index 1057d776f7..c825c8ccba 100644 --- a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala +++ b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala @@ -47,7 +47,7 @@ abstract class PollingExecutorScheduler extends ExecutionContextExecutor with Sc } final def sleep(delay: FiniteDuration, task: Runnable): Runnable = - if (delay == Duration.Zero) { + if (delay <= Duration.Zero) { execute(task) noop } else { From 0832bdfbb746b262acdb4138f5e9ec1ffbc1ca35 Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Tue, 13 Sep 2022 17:19:35 +0000 Subject: [PATCH 3/8] Add `pollEvery` parameter for fairness, default 64 --- .../unsafe/PollingExecutorScheduler.scala | 20 +++++++++---------- .../unsafe/QueueExecutorScheduler.scala | 3 ++- 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala index c825c8ccba..e8f54d3a51 100644 --- a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala +++ b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala @@ -22,7 +22,9 @@ import scala.util.control.NonFatal import java.util.{ArrayDeque, PriorityQueue} -abstract class PollingExecutorScheduler extends ExecutionContextExecutor with Scheduler { +abstract class PollingExecutorScheduler(pollEvery: Int) + extends ExecutionContextExecutor + with Scheduler { import PollingExecutorScheduler._ @@ -30,8 +32,7 @@ abstract class PollingExecutorScheduler extends ExecutionContextExecutor with Sc private[this] var inLoop: Boolean = false private[this] var cachedNow: Long = _ - private[this] var executeQueue: ArrayDeque[Runnable] = new ArrayDeque - private[this] var cachedExecuteQueue: ArrayDeque[Runnable] = new ArrayDeque + private[this] val executeQueue: ArrayDeque[Runnable] = new ArrayDeque private[this] val sleepQueue: PriorityQueue[ScheduledTask] = new PriorityQueue private[this] val noop: Runnable = () => () @@ -102,20 +103,17 @@ abstract class PollingExecutorScheduler extends ExecutionContextExecutor with Sc } } - // swap the task queues - val todo = executeQueue - executeQueue = cachedExecuteQueue - cachedExecuteQueue = todo - - // do all the tasks - while (!todo.isEmpty()) { - val runnable = todo.poll() + // do up to pollEvery tasks + var i = 0 + while (i < pollEvery && !executeQueue.isEmpty()) { + val runnable = executeQueue.poll() try { runnable.run() } catch { case NonFatal(t) => reportFailure(t) } + i += 1 } // cleanup canceled timers diff --git a/core/native/src/main/scala/cats/effect/unsafe/QueueExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/QueueExecutorScheduler.scala index 99595b79b3..c53036b5dc 100644 --- a/core/native/src/main/scala/cats/effect/unsafe/QueueExecutorScheduler.scala +++ b/core/native/src/main/scala/cats/effect/unsafe/QueueExecutorScheduler.scala @@ -18,7 +18,8 @@ package cats.effect.unsafe import scala.concurrent.duration._ -private[effect] object QueueExecutorScheduler extends PollingExecutorScheduler { +// JVM WSTP sets ExternalQueueTicks = 64 so we steal it here +private[effect] object QueueExecutorScheduler extends PollingExecutorScheduler(64) { def poll(timeout: Duration): Boolean = { if (timeout != Duration.Zero && timeout.isFinite) { From 67726e7174b3bea724029fba1407c0e73e85a10a Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Tue, 13 Sep 2022 17:37:48 +0000 Subject: [PATCH 4/8] `ScheduledTask` -> `SleepTask` --- .../effect/unsafe/PollingExecutorScheduler.scala | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala index e8f54d3a51..734bf236f6 100644 --- a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala +++ b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala @@ -33,7 +33,7 @@ abstract class PollingExecutorScheduler(pollEvery: Int) private[this] var cachedNow: Long = _ private[this] val executeQueue: ArrayDeque[Runnable] = new ArrayDeque - private[this] val sleepQueue: PriorityQueue[ScheduledTask] = new PriorityQueue + private[this] val sleepQueue: PriorityQueue[SleepTask] = new PriorityQueue private[this] val noop: Runnable = () => () @@ -54,9 +54,9 @@ abstract class PollingExecutorScheduler(pollEvery: Int) } else { scheduleIfNeeded() val now = if (inLoop) cachedNow else monotonicNanos() - val scheduledTask = new ScheduledTask(now + delay.toNanos, task) - sleepQueue.offer(scheduledTask) - scheduledTask + val SleepTask = new SleepTask(now + delay.toNanos, task) + sleepQueue.offer(SleepTask) + SleepTask } def reportFailure(t: Throwable): Unit = t.printStackTrace() @@ -144,16 +144,16 @@ abstract class PollingExecutorScheduler(pollEvery: Int) object PollingExecutorScheduler { - private final class ScheduledTask( + private final class SleepTask( val at: Long, val runnable: Runnable, var canceled: Boolean = false ) extends Runnable - with Comparable[ScheduledTask] { + with Comparable[SleepTask] { def run(): Unit = canceled = true - def compareTo(that: ScheduledTask): Int = + def compareTo(that: SleepTask): Int = java.lang.Long.compare(this.at, that.at) } From 459136429d9503d94012c11becc6fa8ad6fc1a5b Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Tue, 13 Sep 2022 18:02:04 +0000 Subject: [PATCH 5/8] Refactor `sleep` cancelation Now behaves like `ScheduledExecutorService#setRemoveOnCancelPolicy(true)` used on JVM --- .../unsafe/PollingExecutorScheduler.scala | 33 +++++-------------- 1 file changed, 9 insertions(+), 24 deletions(-) diff --git a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala index 734bf236f6..5cc9ae13e2 100644 --- a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala +++ b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala @@ -26,8 +26,6 @@ abstract class PollingExecutorScheduler(pollEvery: Int) extends ExecutionContextExecutor with Scheduler { - import PollingExecutorScheduler._ - private[this] var needsReschedule: Boolean = true private[this] var inLoop: Boolean = false private[this] var cachedNow: Long = _ @@ -54,9 +52,9 @@ abstract class PollingExecutorScheduler(pollEvery: Int) } else { scheduleIfNeeded() val now = if (inLoop) cachedNow else monotonicNanos() - val SleepTask = new SleepTask(now + delay.toNanos, task) - sleepQueue.offer(SleepTask) - SleepTask + val sleepTask = new SleepTask(now + delay.toNanos, task) + sleepQueue.offer(sleepTask) + sleepTask } def reportFailure(t: Throwable): Unit = t.printStackTrace() @@ -89,10 +87,6 @@ abstract class PollingExecutorScheduler(pollEvery: Int) cachedNow = monotonicNanos() // execute the timers - while (!sleepQueue.isEmpty() && sleepQueue.peek().canceled) { - sleepQueue.poll() - } - while (!sleepQueue.isEmpty() && sleepQueue.peek().at <= cachedNow) { val task = sleepQueue.poll() try { @@ -116,12 +110,6 @@ abstract class PollingExecutorScheduler(pollEvery: Int) i += 1 } - // cleanup canceled timers - val sleepIter = sleepQueue.iterator() - while (sleepIter.hasNext()) { - if (sleepIter.next().canceled) sleepIter.remove() - } - // finally we poll val timeout = if (!executeQueue.isEmpty()) @@ -140,22 +128,19 @@ abstract class PollingExecutorScheduler(pollEvery: Int) inLoop = false } -} - -object PollingExecutorScheduler { - - private final class SleepTask( + private[this] final class SleepTask( val at: Long, - val runnable: Runnable, - var canceled: Boolean = false + val runnable: Runnable ) extends Runnable with Comparable[SleepTask] { - def run(): Unit = canceled = true + def run(): Unit = { + sleepQueue.remove(this) + () + } def compareTo(that: SleepTask): Int = java.lang.Long.compare(this.at, that.at) - } } From d8a0441a98dae7c4892b456049da33e85f3a9969 Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Wed, 14 Sep 2022 00:20:46 +0000 Subject: [PATCH 6/8] Rip out `cachedNow` stuff --- .../effect/unsafe/PollingExecutorScheduler.scala | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala index 5cc9ae13e2..05f8c56c10 100644 --- a/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala +++ b/core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala @@ -27,8 +27,6 @@ abstract class PollingExecutorScheduler(pollEvery: Int) with Scheduler { private[this] var needsReschedule: Boolean = true - private[this] var inLoop: Boolean = false - private[this] var cachedNow: Long = _ private[this] val executeQueue: ArrayDeque[Runnable] = new ArrayDeque private[this] val sleepQueue: PriorityQueue[SleepTask] = new PriorityQueue @@ -51,7 +49,7 @@ abstract class PollingExecutorScheduler(pollEvery: Int) noop } else { scheduleIfNeeded() - val now = if (inLoop) cachedNow else monotonicNanos() + val now = monotonicNanos() val sleepTask = new SleepTask(now + delay.toNanos, task) sleepQueue.offer(sleepTask) sleepTask @@ -78,16 +76,13 @@ abstract class PollingExecutorScheduler(pollEvery: Int) private[this] def loop(): Unit = { needsReschedule = false - inLoop = true var continue = true while (continue) { - // cache the timestamp for this tick - cachedNow = monotonicNanos() - // execute the timers - while (!sleepQueue.isEmpty() && sleepQueue.peek().at <= cachedNow) { + val now = monotonicNanos() + while (!sleepQueue.isEmpty() && sleepQueue.peek().at <= now) { val task = sleepQueue.poll() try { task.runnable.run() @@ -115,7 +110,7 @@ abstract class PollingExecutorScheduler(pollEvery: Int) if (!executeQueue.isEmpty()) Duration.Zero else if (!sleepQueue.isEmpty()) - (sleepQueue.peek().at - cachedNow).nanos + Math.max(sleepQueue.peek().at - monotonicNanos(), 0).nanos else Duration.Inf @@ -125,7 +120,6 @@ abstract class PollingExecutorScheduler(pollEvery: Int) } needsReschedule = true - inLoop = false } private[this] final class SleepTask( From 2e4b27857731422426497ab4fd2c56751189867e Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Wed, 14 Sep 2022 16:40:25 +0000 Subject: [PATCH 7/8] Revert "Make `installGlobal` public" This reverts commit 7efbac0eb49239252b556db5c0b7abaa11997ea5. --- .../scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/native/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala b/core/native/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala index 4b5104e751..42c0d19b1c 100644 --- a/core/native/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala +++ b/core/native/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala @@ -26,7 +26,7 @@ private[unsafe] abstract class IORuntimeCompanionPlatform { this: IORuntime.type private[this] var _global: IORuntime = null - def installGlobal(global: => IORuntime): Boolean = { + private[effect] def installGlobal(global: => IORuntime): Boolean = { if (_global == null) { _global = global true From 5d511edac769b02eda4889d12462a779b31f33e9 Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Wed, 14 Sep 2022 16:59:08 +0000 Subject: [PATCH 8/8] Fix `suspend` impl --- .../main/scala/cats/effect/IOCompanionPlatform.scala | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala b/core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala index 6025313575..5d66f40035 100644 --- a/core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala +++ b/core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala @@ -20,6 +20,8 @@ import java.time.Instant private[effect] abstract class IOCompanionPlatform { this: IO.type => + private[this] val TypeDelay = Sync.Type.Delay + def blocking[A](thunk: => A): IO[A] = // do our best to mitigate blocking IO.cede *> apply(thunk).guarantee(IO.cede) @@ -33,10 +35,11 @@ private[effect] abstract class IOCompanionPlatform { this: IO.type => def interruptibleMany[A](thunk: => A): IO[A] = interruptible(true, thunk) - def suspend[A](hint: Sync.Type)(thunk: => A): IO[A] = { - val _ = hint - apply(thunk) - } + def suspend[A](hint: Sync.Type)(thunk: => A): IO[A] = + if (hint eq TypeDelay) + apply(thunk) + else + blocking(thunk) def realTimeInstant: IO[Instant] = asyncForIO.realTimeInstant