Fiber tracing: asynchronous stack tracing

benchmarks/shared/src/main/scala/cats/effect/benchmarks/MapCallsBenchmark.scala

@@ -40,7 +40,7 @@ class MapCallsBenchmark {
   import MapCallsBenchmark.test
 
   @Benchmark
-  def one(): Long = test(12000, 1)
+  def one(): Long = test(1, 1)
 
   @Benchmark
   def batch30(): Long = test(12000 / 30, 30)

build.sbt

@@ -181,7 +181,21 @@ val mimaSettings = Seq(
       exclude[IncompatibleSignatureProblem]("cats.effect.Resource.evalTap"),
       // change in encoding of value classes in generic methods https://github.com/lightbend/mima/issues/423
       exclude[IncompatibleSignatureProblem]("cats.effect.Blocker.apply"),
-      exclude[IncompatibleSignatureProblem]("cats.effect.Blocker.fromExecutorService")
+      exclude[IncompatibleSignatureProblem]("cats.effect.Blocker.fromExecutorService"),
+      // Tracing - https://github.com/typelevel/cats-effect/pull/854
+      exclude[DirectMissingMethodProblem]("cats.effect.IO#Async.apply"),
+      exclude[DirectMissingMethodProblem]("cats.effect.IO#Bind.apply"),
+      exclude[IncompatibleResultTypeProblem]("cats.effect.IO#Async.k"),
+      exclude[DirectMissingMethodProblem]("cats.effect.IO#Async.copy"),
+      exclude[IncompatibleResultTypeProblem]("cats.effect.IO#Async.copy$default$1"),
+      exclude[DirectMissingMethodProblem]("cats.effect.IO#Async.this"),
+      exclude[DirectMissingMethodProblem]("cats.effect.IO#Bind.copy"),
+      exclude[DirectMissingMethodProblem]("cats.effect.IO#Bind.this"),
+      exclude[DirectMissingMethodProblem]("cats.effect.IO#Map.index"),
+      exclude[IncompatibleMethTypeProblem]("cats.effect.IO#Map.copy"),
+      exclude[IncompatibleResultTypeProblem]("cats.effect.IO#Map.copy$default$3"),
+      exclude[IncompatibleMethTypeProblem]("cats.effect.IO#Map.this"),
+      exclude[IncompatibleMethTypeProblem]("cats.effect.IO#Map.apply")
     )
   }
 )
@@ -226,7 +240,7 @@ lazy val sharedSourcesSettings = Seq(
 lazy val root = project
   .in(file("."))
   .disablePlugins(MimaPlugin)
-  .aggregate(coreJVM, coreJS, lawsJVM, lawsJS)
+  .aggregate(coreJVM, coreJS, lawsJVM, lawsJS, tracingTests)
   .settings(skipOnPublishSettings)
 
 lazy val core = crossProject(JSPlatform, JVMPlatform)
@@ -291,6 +305,37 @@ lazy val laws = crossProject(JSPlatform, JVMPlatform)
 lazy val lawsJVM = laws.jvm
 lazy val lawsJS = laws.js
 
+lazy val FullTracingTest = config("fulltracing").extend(Test)
+
+lazy val tracingTests = project
+  .in(file("tracing-tests"))
+  .dependsOn(coreJVM)
+  .settings(commonSettings ++ skipOnPublishSettings)
+  .settings(
+    libraryDependencies ++= Seq(
+      "org.typelevel" %%% "cats-laws" % CatsVersion,
+      "org.typelevel" %%% "discipline-scalatest" % DisciplineScalatestVersion % Test
+    )
+  )
+  .configs(FullTracingTest)
+  .settings(inConfig(FullTracingTest)(Defaults.testSettings): _*)
+  .settings(
+    unmanagedSourceDirectories in FullTracingTest += {
+      baseDirectory.value.getParentFile / "src" / "fulltracing" / "scala"
+    },
+    test in Test := (test in Test).dependsOn(test in FullTracingTest).value,
+    fork in Test := true,
+    fork in FullTracingTest := true,
+    javaOptions in Test ++= Seq(
+      "-Dcats.effect.tracing=true",
+      "-Dcats.effect.stackTracingMode=cached"
+    ),
+    javaOptions in FullTracingTest ++= Seq(
+      "-Dcats.effect.tracing=true",
+      "-Dcats.effect.stackTracingMode=full"
+    )
+  )
+
 lazy val benchmarksPrev = project
   .in(file("benchmarks/vPrev"))
   .settings(commonSettings ++ skipOnPublishSettings ++ sharedSourcesSettings)

core/js/src/main/scala/cats/effect/internals/IOPlatform.scala

@@ -45,15 +45,6 @@ private[effect] object IOPlatform {
     }
   }
 
-  /**
-   * Establishes the maximum stack depth for `IO#map` operations
-   * for JavaScript.
-   *
-   * The default for JavaScript is 32, from which we subtract 1
-   * as an optimization.
-   */
-  final val fusionMaxStackDepth = 31
-
   /** Returns `true` if the underlying platform is the JVM,
    * `false` if it's JavaScript. */
   final val isJVM = false

core/js/src/main/scala/cats/effect/internals/IOTimer.scala

@@ -33,7 +33,7 @@ final private[internals] class IOTimer(ec: ExecutionContext) extends Timer[IO] {
 
   def sleep(timespan: FiniteDuration): IO[Unit] =
     IO.Async(new IOForkedStart[Unit] {
-      def apply(conn: IOConnection, cb: Either[Throwable, Unit] => Unit): Unit = {
+      def apply(conn: IOConnection, ctx: IOContext, cb: Either[Throwable, Unit] => Unit): Unit = {
         val task = setTimeout(timespan.toMillis, ec, new ScheduledTick(conn, cb))
         // On the JVM this would need a ForwardCancelable,
         // but not on top of JS as we don't have concurrency

core/js/src/main/scala/cats/effect/internals/TracingPlatform.scala

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) 2017-2019 The Typelevel Cats-effect Project Developers
+ *
+ * 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.internals
+
+object TracingPlatform {
+  final val isCachedStackTracing: Boolean = false
+
+  final val isFullStackTracing: Boolean = false
+
+  final val isStackTracing: Boolean = isFullStackTracing || isCachedStackTracing
+
+  final val traceBufferSize: Int = 32
+}

core/jvm/src/main/java/cats/effect/internals/TracingPlatform.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2017-2019 The Typelevel Cats-effect Project Developers
+ *
+ * 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.internals;
+
+import java.util.Optional;
+
+/**
+ * Holds platform-specific flags that control tracing behavior.
+ *
+ * The Scala compiler inserts a volatile bitmap access for module field accesses.
+ * Because the `tracingMode` flag is read in various IO combinators, we are opting
+ * to define it in a Java source file to avoid the volatile access.
+ *
+ * INTERNAL API.
+ */
+public final class TracingPlatform {
+
+    /**
+     * Sets stack tracing mode for a JVM process, which controls
+     * how much stack trace information is captured.
+     * Acceptable values are: NONE, CACHED, FULL.
+     */
+    private static final String stackTracingMode = Optional.ofNullable(System.getProperty("cats.effect.stackTracingMode"))
+            .filter(x -> !x.isEmpty())
+            .orElse("cached");
+
+    public static final boolean isCachedStackTracing = stackTracingMode.equalsIgnoreCase("cached");
+
+    public static final boolean isFullStackTracing = stackTracingMode.equalsIgnoreCase("full");
+
+    public static final boolean isStackTracing = isFullStackTracing || isCachedStackTracing;
+
+    /**
+     * The number of trace lines to retain during tracing. If more trace
+     * lines are produced, then the oldest trace lines will be discarded.
+     * Automatically rounded up to the nearest power of 2.
+     */
+    public static final int traceBufferSize = Optional.ofNullable(System.getProperty("cats.effect.traceBufferSize"))
+        .filter(x -> !x.isEmpty())
+        .flatMap(x -> {
+            try {
+                return Optional.of(Integer.valueOf(x));
+            } catch (Exception e) {
+                return Optional.empty();
+            }
+        })
+        .orElse(128);
+
+}

core/jvm/src/main/scala/cats/effect/internals/IOPlatform.scala

@@ -20,7 +20,7 @@ import java.util.concurrent.locks.AbstractQueuedSynchronizer
 import cats.effect.IO
 import scala.concurrent.blocking
 import scala.concurrent.duration.{Duration, FiniteDuration}
-import scala.util.{Either, Try}
+import scala.util.Either
 
 private[effect] object IOPlatform {
 
@@ -73,37 +73,6 @@ private[effect] object IOPlatform {
     }
   }
 
-  /**
-   * Establishes the maximum stack depth for `IO#map` operations.
-   *
-   * The default is `128`, from which we subtract one as an
-   * optimization. This default has been reached like this:
-   *
-   *  - according to official docs, the default stack size on 32-bits
-   *    Windows and Linux was 320 KB, whereas for 64-bits it is 1024 KB
-   *  - according to measurements chaining `Function1` references uses
-   *    approximately 32 bytes of stack space on a 64 bits system;
-   *    this could be lower if "compressed oops" is activated
-   *  - therefore a "map fusion" that goes 128 in stack depth can use
-   *    about 4 KB of stack space
-   *
-   * If this parameter becomes a problem, it can be tuned by setting
-   * the `cats.effect.fusionMaxStackDepth` system property when
-   * executing the Java VM:
-   *
-   * <pre>
-   *   java -Dcats.effect.fusionMaxStackDepth=32 \
-   *        ...
-   * </pre>
-   */
-  final val fusionMaxStackDepth =
-    Option(System.getProperty("cats.effect.fusionMaxStackDepth", ""))
-      .filter(s => s != null && s.nonEmpty)
-      .flatMap(s => Try(s.toInt).toOption)
-      .filter(_ > 0)
-      .map(_ - 1)
-      .getOrElse(127)
-
   /**
    * Returns `true` if the underlying platform is the JVM,
    * `false` if it's JavaScript.

core/jvm/src/main/scala/cats/effect/internals/IOTimer.scala

@@ -38,7 +38,7 @@ final private[internals] class IOTimer private (ec: ExecutionContext, sc: Schedu
 
   override def sleep(timespan: FiniteDuration): IO[Unit] =
     IO.Async(new IOForkedStart[Unit] {
-      def apply(conn: IOConnection, cb: T[Unit]): Unit = {
+      def apply(conn: IOConnection, ctx: IOContext, cb: T[Unit]): Unit = {
         // Doing what IO.cancelable does
         val ref = ForwardCancelable()
         conn.push(ref.cancel)