Revert "ZIO Runtime System Improvements (2.1 candidate) (#8582)"

This reverts commit 0304398.

.jvmopts

@@ -1 +1 @@
--Dcats.effect.stackTracingMode=full
+-Dcats.effect.stackTracingMode=full

benchmarks/src/main/scala/zio/BroadFlatMapBenchmark.scala

@@ -10,9 +10,8 @@ import scala.concurrent.Await
 @State(JScope.Thread)
 @BenchmarkMode(Array(Mode.Throughput))
 @OutputTimeUnit(TimeUnit.SECONDS)
-@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Threads(16)
+@Warmup(iterations = 10, time = 1)
+@Measurement(iterations = 10, time = 1)
 @Fork(1)
 class BroadFlatMapBenchmark {
   @Param(Array("20"))
@@ -44,6 +43,32 @@ class BroadFlatMapBenchmark {
       .get()
   }
 
+  @Benchmark
+  def monoBroadFlatMap(): BigInt = {
+    import reactor.core.publisher.Mono
+
+    def fib(n: Int): Mono[BigInt] =
+      if (n <= 1) Mono.fromSupplier(() => n)
+      else
+        fib(n - 1).flatMap(a => fib(n - 2).flatMap(b => Mono.fromSupplier(() => a + b)))
+
+    fib(depth)
+      .block()
+  }
+
+  @Benchmark
+  def rxSingleBroadFlatMap(): BigInt = {
+    import io.reactivex.Single
+
+    def fib(n: Int): Single[BigInt] =
+      if (n <= 1) Single.fromCallable(() => n)
+      else
+        fib(n - 1).flatMap(a => fib(n - 2).flatMap(b => Single.fromCallable(() => a + b)))
+
+    fib(depth)
+      .blockingGet()
+  }
+
   @Benchmark
   def twitterBroadFlatMap(): BigInt = {
     import com.twitter.util.{Await, Future}

benchmarks/src/main/scala/zio/ForEachBenchmark.scala

@@ -11,9 +11,8 @@ import java.util.concurrent.TimeUnit
 @State(JScope.Thread)
 @BenchmarkMode(Array(Mode.Throughput))
 @OutputTimeUnit(TimeUnit.SECONDS)
-@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Threads(16)
+@Measurement(iterations = 5, timeUnit = TimeUnit.SECONDS, time = 3)
+@Warmup(iterations = 5, timeUnit = TimeUnit.SECONDS, time = 3)
 @Fork(1)
 class ForEachBenchmark {
 

benchmarks/src/main/scala/zio/ForkJoinBenchmark.scala

@@ -23,10 +23,6 @@ import java.util.concurrent.TimeUnit
 @State(JScope.Thread)
 @BenchmarkMode(Array(Mode.Throughput))
 @OutputTimeUnit(TimeUnit.SECONDS)
-@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Threads(16)
-@Fork(1)
 class ForkJoinBenchmark {
   import BenchmarkUtil.unsafeRun
 

benchmarks/src/main/scala/zio/NarrowFlatMapBenchmark.scala

@@ -10,10 +10,6 @@ import scala.concurrent.Await
 @State(JScope.Thread)
 @BenchmarkMode(Array(Mode.Throughput))
 @OutputTimeUnit(TimeUnit.SECONDS)
-@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Threads(16)
-@Fork(1)
 class NarrowFlatMapBenchmark {
   @Param(Array("1000"))
   var size: Int = _

benchmarks/src/main/scala/zio/RegionBenchmark.scala

@@ -29,10 +29,6 @@ import java.util.concurrent.TimeUnit
 @State(JScope.Thread)
 @BenchmarkMode(Array(Mode.Throughput))
 @OutputTimeUnit(TimeUnit.SECONDS)
-@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Threads(16)
-@Fork(1)
 class RegionBenchmark {
   import BenchmarkUtil.unsafeRun
 

benchmarks/src/main/scala/zio/UnsafeRunBenchmark.scala

@@ -35,10 +35,6 @@ import java.util.concurrent.TimeUnit
 @State(JScope.Thread)
 @BenchmarkMode(Array(Mode.Throughput))
 @OutputTimeUnit(TimeUnit.SECONDS)
-@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Threads(16)
-@Fork(1)
 class UnsafeRunBenchmark {
   import BenchmarkUtil.unsafeRun
 

core-tests/shared/src/test/scala/zio/RuntimeBootstrapTests.scala

@@ -620,32 +620,32 @@ object RuntimeBootstrapTests {
 
   def main(args: Array[String]): Unit = {
     val _ = ()
-    runtimeFlags()
-    helloWorld()
-    fib()
-    iteration()
-    asyncInterruption()
-    syncInterruption()
-    race()
-    autoInterruption()
-    autoInterruption2()
-    asyncInterruptionOfNever()
-    interruptRacedForks()
-    useInheritance()
-    useInheritance2()
-    asyncUninterruptible()
-    uninterruptibleClosingScope()
-    syncInterruption2()
-    acquireReleaseDisconnect()
-    disconnectedInterruption()
-    interruptibleAfterRace()
+    // runtimeFlags()
+    // helloWorld()
+    // fib()
+    // iteration()
+    // asyncInterruption()
+    // syncInterruption()
+    // race()
+    // autoInterruption()
+    // autoInterruption2()
+    // asyncInterruptionOfNever()
+    // interruptRacedForks()
+    // useInheritance()
+    // useInheritance2()
+    // asyncUninterruptible()
+    // uninterruptibleClosingScope()
+    // syncInterruption2()
+    // acquireReleaseDisconnect()
+    // disconnectedInterruption()
+    // interruptibleAfterRace()
     // uninterruptibleRace()
-    interruptionDetection()
-    interruptionRecovery()
+    // interruptionDetection()
+    // interruptionRecovery()
     // cooperativeYielding()
     // interruptionOfForkedRace()
-    stackTrace1()
-    stackTrace2()
+    // stackTrace1()
+    // stackTrace2()
     // interruptibleHole()
     // queueOfferInterruption()
     // localSupervision()

core-tests/shared/src/test/scala/zio/SupervisorSpec.scala

@@ -24,7 +24,7 @@ object SupervisorSpec extends ZIOBaseSpec {
         f           <- ZIO.sleep(10.millis).fork.supervised(onEndSup)
         _           <- (f.await).supervised(onResumeSup)
         value       <- onResumeSup.value
-      } yield assertTrue(value > 0)
+      } yield assertTrue(value == 1)
     } @@ TestAspect.nonFlaky @@ TestAspect.withLiveClock,
     suite("laws") {
       DifferSpec.diffLaws(Differ.supervisor)(genSupervisor)((left, right) =>

core-tests/shared/src/test/scala/zio/ZIOSpec.scala

@@ -341,7 +341,7 @@ object ZIOSpec extends ZIOBaseSpec {
           promise1 <- Promise.make[Nothing, Unit]
           promise2 <- Promise.make[Nothing, Unit]
           left      = promise2.await
-          right1    = (promise1.await *> ZIO.fail("fail")).uninterruptible
+          right1    = promise1.await *> ZIO.fail("fail")
           right2    = (promise1.succeed(()) *> ZIO.never).ensuring(promise2.interrupt *> ZIO.never.interruptible)
           exit     <- ZIO.collectAllPar(List(left, ZIO.collectAllPar(List(right1, right2)))).exit
         } yield assert(exit)(failsCause(containsCause(Cause.fail("fail"))))
@@ -2843,12 +2843,12 @@ object ZIOSpec extends ZIOBaseSpec {
                            .ensuring(ensuring.succeed(()) *> ZIO.interruptible(ZIO.never))
                            .mapError(_ => 42)
                        )
-                         .catchAllCause(cause => ZIO.succeed(cause))
+                         .catchAllCause(cause => ZIO.succeed(cause.defects))
                          .fork
                      }
-            cause <- ensuring.await *> fiber.interrupt.flatMap(ZIO.done(_))
-          } yield assertTrue(cause.defects.length == 1)
-        } @@ nonFlaky
+            failures <- ensuring.await *> fiber.interrupt.flatMap(ZIO.done(_))
+          } yield assertTrue(failures.length == 1)
+        }
 
     } @@ zioTag(interruption),
     suite("RTS concurrency correctness")(
@@ -4177,7 +4177,7 @@ object ZIOSpec extends ZIOBaseSpec {
           right2    = (promise1.succeed(()) *> ZIO.never).ensuring(promise2.interrupt *> ZIO.never.interruptible)
           exit     <- left.zipPar(right1.zipPar(right2)).exit
         } yield assert(exit)(failsCause(containsCause(Cause.fail("fail"))))
-      } @@ nonFlaky(100),
+      } @@ nonFlaky(1000),
       test("is interruptible") {
         for {
           promise1 <- Promise.make[Nothing, Unit]

core/js/src/main/scala/zio/internal/OneShot.scala

@@ -46,11 +46,6 @@ private[zio] final class OneShot[A] private (var value: A) {
     if (value == null) throw new Error("Cannot block for result to be set in JavaScript")
     value
   }
-
-  def get(timeout: Long): A = {
-    val _ = timeout
-    get()
-  }
 }
 
 private[zio] object OneShot {

core/js/src/main/scala/zio/internal/PlatformSpecific.scala

@@ -66,8 +66,6 @@ private[zio] trait PlatformSpecific {
    */
   final def getCurrentThreadGroup()(implicit unsafe: zio.Unsafe): String = ""
 
-  final val hasGreenThreads: Boolean = false
-
   /**
    * Returns whether the current platform is ScalaJS.
    */

core/jvm/src/main/scala/zio/RuntimePlatformSpecific.scala

@@ -16,13 +16,13 @@
 
 package zio
 
-import zio.internal.{Blocking, IsFatal, LoomSupport}
+import zio.internal.{Blocking, IsFatal}
 import zio.stacktracer.TracingImplicits.disableAutoTrace
 
 private[zio] trait RuntimePlatformSpecific {
 
   final val defaultExecutor: Executor =
-    LoomSupport.newVirtualThreadPerTaskExecutor().map(Executor.fromJavaExecutor(_)).getOrElse(Executor.makeDefault())
+    Executor.makeDefault()
 
   final val defaultBlockingExecutor: Executor =
     Blocking.blockingExecutor

core/jvm/src/main/scala/zio/ZIOAppPlatformSpecific.scala

@@ -63,7 +63,7 @@ private[zio] trait ZIOAppPlatformSpecific { self: ZIOApp =>
   private def interruptRootFibers(fiberId: FiberId)(implicit trace: Trace): UIO[Unit] =
     for {
       roots <- Fiber.roots
-      _     <- Fiber.interruptAll(roots.view.filter(fiber => fiber.isAlive() && fiber.id != fiberId))
+      _     <- Fiber.interruptAll(roots.view.filter(fiber => fiber.isAlive()(Unsafe.unsafe) && fiber.id != fiberId))
     } yield ()
 
 }

core/jvm/src/main/scala/zio/internal/DefaultExecutors.scala

@@ -22,6 +22,7 @@ import zio.stacktracer.TracingImplicits.disableAutoTrace
 import java.util.concurrent.{RejectedExecutionException, ThreadPoolExecutor}
 
 private[zio] abstract class DefaultExecutors {
+
   final def makeDefault(): zio.Executor =
     makeDefault(true)