Merge pull request #3499 from durban/wipTimers

.cirrus.yml

@@ -7,7 +7,7 @@ jvm_highcore_task:
     - name: JVM high-core-count 2.12
       script: sbt '++ 2.12' testsJVM/test
     - name: JVM high-core-count 2.13
-      script: sbt '++ 2.13' testsJVM/test
+      script: sbt '++ 2.13' testsJVM/test stressTests/Jcstress/run
     - name: JVM high-core-count 3
       script: sbt '++ 3' testsJVM/test
 
@@ -20,7 +20,7 @@ jvm_arm_highcore_task:
     - name: JVM ARM high-core-count 2.12
       script: sbt '++ 2.12' testsJVM/test
     - name: JVM ARM high-core-count 2.13
-      script: sbt '++ 2.13' testsJVM/test
+      script: sbt '++ 2.13' testsJVM/test stressTests/Jcstress/run
     - name: JVM ARM high-core-count 3
       script: sbt '++ 3' testsJVM/test
 

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

@@ -70,9 +70,38 @@ class SleepBenchmark {
       .unsafeRunSync()
   }
 
+  def sleepRaceBenchmark(implicit runtime: IORuntime): Int = {
+    def fiber(i: Int): IO[Int] = {
+      val sleepRace = IO.race(IO.sleep(1.nanosecond), IO.sleep(1000.nanoseconds))
+      sleepRace.flatMap { _ =>
+        IO(i).flatMap { j =>
+          sleepRace.flatMap { _ =>
+            if (j > 1000)
+              sleepRace.flatMap(_ => IO.pure(j))
+            else
+              sleepRace.flatMap(_ => fiber(j + 1))
+          }
+        }
+      }
+    }
+
+    List
+      .range(0, size)
+      .traverse(fiber(_).start)
+      .flatMap(_.traverse(_.joinWithNever))
+      .map(_.sum)
+      .unsafeRunSync()
+  }
+
   @Benchmark
   def sleep(): Int = {
     import cats.effect.unsafe.implicits.global
     sleepBenchmark
   }
+
+  @Benchmark
+  def sleepRace(): Int = {
+    import cats.effect.unsafe.implicits.global
+    sleepRaceBenchmark
+  }
 }

build.sbt

@@ -331,6 +331,7 @@ val nativeProjects: Seq[ProjectReference] =
 val undocumentedRefs =
   jsProjects ++ nativeProjects ++ Seq[ProjectReference](
     benchmarks,
+    stressTests,
     example.jvm,
     graalVMExample,
     tests.jvm,
@@ -359,7 +360,8 @@ lazy val rootJVM = project
     std.jvm,
     example.jvm,
     graalVMExample,
-    benchmarks)
+    benchmarks,
+    stressTests)
   .enablePlugins(NoPublishPlugin)
 
 lazy val rootJS = project.aggregate(jsProjects: _*).enablePlugins(NoPublishPlugin)
@@ -383,6 +385,7 @@ lazy val kernel = crossProject(JSPlatform, JVMPlatform, NativePlatform)
       ProblemFilters.exclude[Problem]("cats.effect.kernel.GenConcurrent#Memoize*")
     )
   )
+  .disablePlugins(JCStressPlugin)
   .jsSettings(
     libraryDependencies += "org.scala-js" %%% "scala-js-macrotask-executor" % MacrotaskExecutorVersion % Test
   )
@@ -418,6 +421,7 @@ lazy val kernelTestkit = crossProject(JSPlatform, JVMPlatform, NativePlatform)
         "cats.effect.kernel.testkit.TestContext#Task.copy")
     )
   )
+  .disablePlugins(JCStressPlugin)
 
 /**
  * The laws which constrain the abstractions. This is split from kernel to avoid jar file and
@@ -433,6 +437,7 @@ lazy val laws = crossProject(JSPlatform, JVMPlatform, NativePlatform)
       "org.typelevel" %%% "cats-laws" % CatsVersion,
       "org.typelevel" %%% "discipline-specs2" % DisciplineVersion % Test)
   )
+  .disablePlugins(JCStressPlugin)
 
 /**
  * Concrete, production-grade implementations of the abstractions. Or, more simply-put: IO. Also
@@ -806,6 +811,7 @@ lazy val core = crossProject(JSPlatform, JVMPlatform, NativePlatform)
       } else Seq()
     }
   )
+  .disablePlugins(JCStressPlugin)
 
 /**
  * Test support for the core project, providing various helpful instances like ScalaCheck
@@ -821,6 +827,7 @@ lazy val testkit = crossProject(JSPlatform, JVMPlatform, NativePlatform)
       "org.specs2" %%% "specs2-core" % Specs2Version % Test
     )
   )
+  .disablePlugins(JCStressPlugin)
 
 /**
  * Unit tests for the core project, utilizing the support provided by testkit.
@@ -938,6 +945,7 @@ lazy val std = crossProject(JSPlatform, JVMPlatform, NativePlatform)
       ProblemFilters.exclude[MissingClassProblem]("cats.effect.std.JavaSecureRandom$")
     )
   )
+  .disablePlugins(JCStressPlugin)
 
 /**
  * A trivial pair of trivial example apps primarily used to show that IOApp works as a practical
@@ -976,4 +984,13 @@ lazy val benchmarks = project
       "-Dcats.effect.tracing.exceptions.enhanced=false"))
   .enablePlugins(NoPublishPlugin, JmhPlugin)
 
+lazy val stressTests = project
+  .in(file("stress-tests"))
+  .dependsOn(core.jvm, std.jvm)
+  .settings(
+    name := "cats-effect-stress-tests",
+    Jcstress / version := "0.16",
+  )
+  .enablePlugins(NoPublishPlugin, JCStressPlugin)
+
 lazy val docs = project.in(file("site-docs")).dependsOn(core.jvm).enablePlugins(MdocPlugin)

core/jvm/src/main/java/cats/effect/unsafe/TimerSkipListNodeBase.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2020-2023 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 java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+/**
+ * Base class for `TimerSkipList#Node`, because we can't use `AtomicReferenceFieldUpdater` from
+ * Scala.
+ */
+@SuppressWarnings("serial") // do not serialize this!
+abstract class TimerSkipListNodeBase<C, N extends TimerSkipListNodeBase<C, N>>
+    extends AtomicReference<N> {
+
+  private volatile C callback;
+
+  @SuppressWarnings("rawtypes")
+  private static final AtomicReferenceFieldUpdater<TimerSkipListNodeBase, Object> CALLBACK =
+      AtomicReferenceFieldUpdater.newUpdater(TimerSkipListNodeBase.class, Object.class, "callback");
+
+  protected TimerSkipListNodeBase(C cb, N next) {
+    super(next);
+    this.callback = cb;
+  }
+
+  public final N getNext() {
+    return this.get(); // could be `getAcquire`
+  }
+
+  public final boolean casNext(N ov, N nv) {
+    return this.compareAndSet(ov, nv);
+  }
+
+  public final C getCb() {
+    return this.callback; // could be `getAcquire`
+  }
+
+  public final boolean casCb(C ov, C nv) {
+    return CALLBACK.compareAndSet(this, ov, nv);
+  }
+}