Iterant, version 2

benchmarks/shared/src/main/scala/monix/benchmarks/IterantMapFoldBenchmark.scala

@@ -0,0 +1,77 @@
+/*
+ * Copyright (c) 2014-2018 by The Monix Project Developers.
+ * See the project homepage at: https://monix.io
+ *
+ * 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 monix.benchmarks
+
+import java.util.concurrent.TimeUnit
+import monix.eval.Coeval
+import monix.tail.Iterant
+import org.openjdk.jmh.annotations._
+
+/** To do comparative benchmarks between versions:
+  *
+  *     benchmarks/run-benchmark IterantMapFoldBenchmark
+  *
+  * This will generate results in `benchmarks/results`.
+  *
+  * Or to run the benchmark from within SBT:
+  *
+  *     jmh:run -i 10 -wi 10 -f 2 -t 1 monix.benchmarks.IterantMapFoldBenchmark
+  *
+  * Which means "10 iterations", "10 warm-up iterations", "2 forks", "1 thread".
+  * Please note that benchmarks should be usually executed at least in
+  * 10 iterations (as a rule of thumb), but more is better.
+  */
+@State(Scope.Thread)
+@BenchmarkMode(Array(Mode.Throughput))
+@OutputTimeUnit(TimeUnit.SECONDS)
+class IterantMapFoldBenchmark {
+  import IterantMapFoldBenchmark._
+
+  @Benchmark
+  def granular(): Long =
+    granularRef.map(_ + 1).foldLeftL(0L)(_ + _).value
+
+  @Benchmark
+  def batched(): Long =
+    batchedRef.map(_ + 1).foldLeftL(0L)(_ + _).value
+}
+
+object IterantMapFoldBenchmark {
+  val size = 1000
+
+  val granularRef: Iterant[Coeval, Int] = {
+    var stream = Iterant[Coeval].empty[Int]
+    var idx = 0
+    while (idx < size) {
+      stream = idx +: stream
+      idx += 1
+    }
+    stream
+  }
+
+  val batchedRef: Iterant[Coeval, Int] = {
+    var stream = Iterant[Coeval].empty[Int]
+    var idx = 0
+    while (idx < size) {
+      val rest = stream
+      stream = rest ++ Iterant[Coeval].range(0, 10)
+      idx += 1
+    }
+    stream
+  }
+}

build.sbt

@@ -12,7 +12,9 @@ addCommandAlias("ci-js",      ";clean ;coreJS/test:compile  ;coreJS/test")
 addCommandAlias("release",    ";project monix ;+clean ;+package ;+publishSigned ;sonatypeReleaseAll")
 
 val catsVersion = "1.1.0"
-val catsEffectVersion = "1.0.0-RC2"
+// Hash version is safe, containing a laws fix over 1.0.0-RC2:
+// https://github.com/typelevel/cats-effect/pull/277
+val catsEffectVersion = "1.0.0-RC2-93ac33d"
 val jcToolsVersion = "2.1.1"
 val reactiveStreamsVersion = "1.0.2"
 val scalaTestVersion = "3.0.4"
@@ -44,8 +46,8 @@ lazy val warnUnusedImport = Seq(
 
 lazy val sharedSettings = warnUnusedImport ++ Seq(
   organization := "io.monix",
-  scalaVersion := "2.12.4",
-  crossScalaVersions := Seq("2.11.12", "2.12.4"),
+  scalaVersion := "2.12.6",
+  crossScalaVersions := Seq("2.11.12", "2.12.6"),
 
   scalacOptions ++= Seq(
     // warnings
@@ -57,7 +59,9 @@ lazy val sharedSettings = warnUnusedImport ++ Seq(
     "-language:implicitConversions",
     "-language:experimental.macros",
     // possibly deprecated options
-    "-Ywarn-inaccessible"
+    "-Ywarn-inaccessible",
+    // absolutely necessary for Iterant
+    "-Ypartial-unification"
   ),
 
   // Force building with Java 8
@@ -297,6 +301,8 @@ def mimaSettings(projectName: String) = Seq(
     exclude[IncompatibleResultTypeProblem]("monix.eval.Task.foreach"),
     // Breakage - changed type
     exclude[IncompatibleResultTypeProblem]("monix.execution.Cancelable.empty"),
+    // Breackage — made CompositeException final
+    exclude[FinalClassProblem]("monix.execution.exceptions.CompositeException"),
     // Breakage — extra implicit param
     exclude[DirectMissingMethodProblem]("monix.eval.TaskInstancesLevel0.catsEffect"),
     exclude[DirectMissingMethodProblem]("monix.eval.instances.CatsConcurrentEffectForTask.this"),
@@ -359,7 +365,9 @@ def mimaSettings(projectName: String) = Seq(
     exclude[IncompatibleResultTypeProblem]("monix.execution.internal.collection.ArrayStack.clone"),
     exclude[MissingTypesProblem]("monix.execution.internal.collection.ArrayStack"),
     exclude[DirectMissingMethodProblem]("monix.eval.internal.TaskCancellation#RaiseCancelable.this"),
-    exclude[MissingClassProblem]("monix.eval.internal.TaskBracket$ReleaseRecover")
+    exclude[MissingClassProblem]("monix.eval.internal.TaskBracket$ReleaseRecover"),
+    exclude[MissingClassProblem]("monix.eval.instances.ParallelApplicative$"),
+    exclude[MissingClassProblem]("monix.eval.instances.ParallelApplicative")
   )
 )
 
@@ -442,14 +450,14 @@ lazy val tailCommon =
 
 lazy val tailJVM = project.in(file("monix-tail/jvm"))
   .configure(profile)
-  .dependsOn(evalJVM % "compile->compile; test->test")
+  .dependsOn(evalJVM % "test->test")
   .dependsOn(executionJVM)
   .settings(tailCommon)
 
 lazy val tailJS = project.in(file("monix-tail/js"))
   .enablePlugins(ScalaJSPlugin)
   .configure(profile)
-  .dependsOn(evalJS % "compile->compile; test->test")
+  .dependsOn(evalJS % "test->test")
   .dependsOn(executionJS)
   .settings(scalaJSSettings)
   .settings(tailCommon)

monix-eval/shared/src/main/scala/monix/eval/internal/ForkedRegister.scala

@@ -17,7 +17,7 @@
 
 package monix.eval.internal
 
-import monix.eval.Task.{Async, Context, FlatMap, Map}
+import monix.eval.Task.{Async, Context, ContextSwitch, FlatMap, Map}
 import monix.eval.{Callback, Task}
 
 import scala.annotation.tailrec
@@ -48,6 +48,7 @@ private[eval] object ForkedRegister {
       case Async(_: ForkedRegister[_], _, _, _) => true
       case FlatMap(other, _) => detect(other, limit - 1)
       case Map(other, _, _) => detect(other, limit - 1)
+      case ContextSwitch(other, _, _) => detect(other, limit - 1)
       case _ => false
     } else {
       false

monix-eval/shared/src/test/scala/monix/eval/BaseLawsSuite.scala

@@ -195,7 +195,7 @@ trait ArbitraryInstancesBase extends monix.execution.ArbitraryInstances {
   implicit def arbitraryPfExToA[A](implicit A: Arbitrary[A]): Arbitrary[PartialFunction[Throwable, A]] =
     Arbitrary {
       val fun = implicitly[Arbitrary[Int => A]]
-      for (f <- fun.arbitrary) yield PartialFunction((t: Throwable) => f(t.hashCode()))
+      for (f <- fun.arbitrary) yield { case (t: Throwable) => f(t.hashCode()) }
     }
 
   implicit def arbitraryCoevalToLong[A, B](implicit A: Arbitrary[A], B: Arbitrary[B]): Arbitrary[Coeval[A] => B] =
@@ -216,7 +216,9 @@ trait ArbitraryInstancesBase extends monix.execution.ArbitraryInstances {
   implicit def equalityCoeval[A](implicit A: Eq[A]): Eq[Coeval[A]] =
     new Eq[Coeval[A]] {
       def eqv(lh: Coeval[A], rh: Coeval[A]): Boolean = {
-        Eq[Try[A]].eqv(lh.runTry(), rh.runTry())
+        val lht = lh.runTry()
+        val rht = rh.runTry()
+        Eq[Try[A]].eqv(lht, rht)
       }
     }
 

monix-eval/shared/src/test/scala/monix/eval/TestUtils.scala

@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2014-2018 by The Monix Project Developers.
+ * See the project homepage at: https://monix.io
+ *
+ * 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 monix.eval
+
+import java.io.{ByteArrayOutputStream, PrintStream}
+import scala.util.control.NonFatal
+
+/**
+  * INTERNAL API — test utilities.
+  */
+trait TestUtils {
+
+  /**
+    * Silences `System.err`, only printing the output in case exceptions are
+    * thrown by the executed `thunk`.
+    */
+  def silenceSystemErr[A](thunk: => A): A = synchronized {
+    // Silencing System.err
+    val oldErr = System.err
+    val outStream = new ByteArrayOutputStream()
+    val fakeErr = new PrintStream(outStream)
+    System.setErr(fakeErr)
+    try {
+      val result = thunk
+      System.setErr(oldErr)
+      result
+    } catch {
+      case NonFatal(e) =>
+        System.setErr(oldErr)
+        // In case of errors, print whatever was caught
+        fakeErr.close()
+        val out = outStream.toString("utf-8")
+        if (out.nonEmpty) oldErr.println(out)
+        throw e
+    }
+  }
+
+  /**
+    * Catches `System.err` output, for testing purposes.
+    */
+  def catchSystemErr(thunk: => Unit): String = synchronized {
+    val oldErr = System.err
+    val outStream = new ByteArrayOutputStream()
+    val fakeErr = new PrintStream(outStream)
+    System.setErr(fakeErr)
+    try {
+      thunk
+    } finally {
+      System.setErr(oldErr)
+      fakeErr.close()
+    }
+    outStream.toString("utf-8")
+  }
+}

monix-eval/shared/src/test/scala/monix/eval/TypeClassLawsForParallelApplicativeSuite.scala

@@ -19,7 +19,8 @@ package monix.eval
 
 import cats.Applicative
 import cats.laws.discipline.ApplicativeTests
-import monix.eval.instances.{CatsParallelForTask, ParallelApplicative}
+import monix.eval.instances.CatsParallelForTask
+import monix.execution.internal.ParallelApplicative
 
 object TypeClassLawsForParallelApplicativeSuite extends BaseLawsSuite {
   implicit val ap: Applicative[Task] =

monix-execution/js/src/main/scala/monix/execution/internal/DefaultUncaughtExceptionReporter.scala

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2014-2018 by The Monix Project Developers.
+ * See the project homepage at: https://monix.io
+ *
+ * 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 monix.execution
+package internal
+
+import scala.concurrent.ExecutionContext
+
+/**
+  * INTERNAL API — implements [[UncaughtExceptionReporter.default]].
+  */
+private[execution] object DefaultUncaughtExceptionReporter
+  extends UncaughtExceptionReporter {
+
+  def reportFailure(e: Throwable): Unit =
+    logger(e)
+
+  private[this] lazy val logger =
+    ExecutionContext.defaultReporter
+}

monix-execution/js/src/main/scala/monix/execution/internal/Platform.scala

@@ -19,7 +19,6 @@ package monix.execution.internal
 
 import monix.execution.exceptions.CompositeException
 import monix.execution.schedulers.CanBlock
-
 import scala.concurrent.Awaitable
 import scala.concurrent.duration.Duration
 
@@ -96,17 +95,22 @@ private[monix] object Platform {
     rest.filter(_ ne first).toList match {
       case Nil => first
       case nonEmpty =>
-        CompositeException(first :: nonEmpty)
+        first match {
+          case CompositeException(errors) =>
+            val list = errors.toList
+            CompositeException(list ::: nonEmpty)
+          case _ =>
+            CompositeException(first :: nonEmpty)
+        }
     }
 
-  /** Useful utility that combines an `Either` result, which is what
+  /**
+    * Useful utility that combines an `Either` result, which is what
     * `MonadError#attempt` returns.
     */
   def composeErrors(first: Throwable, second: Either[Throwable, _]): Throwable =
     second match {
-      case Left(e2) if e2 ne first =>
-        CompositeException(List(first, e2))
-      case _ =>
-        first
+      case Left(e2) => composeErrors(first, e2)
+      case _ => first
     }
 }

monix-execution/js/src/main/scala/monix/execution/schedulers/StandardContext.scala

@@ -44,4 +44,4 @@ private[execution] class StandardContext(reporter: UncaughtExceptionReporter)
 }
 
 private[execution] object StandardContext
-  extends StandardContext(UncaughtExceptionReporter.LogExceptionsToStandardErr)
+  extends StandardContext(UncaughtExceptionReporter.default)

monix-execution/jvm/src/main/scala/monix/execution/internal/DefaultUncaughtExceptionReporter.scala

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2014-2018 by The Monix Project Developers.
+ * See the project homepage at: https://monix.io
+ *
+ * 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 monix.execution
+package internal
+
+/**
+  * INTERNAL API — implements [[UncaughtExceptionReporter.default]].
+  */
+private[execution] object DefaultUncaughtExceptionReporter
+  extends UncaughtExceptionReporter {
+
+  def reportFailure(e: Throwable): Unit =
+    Thread.getDefaultUncaughtExceptionHandler match {
+      case null => e.printStackTrace()
+      case h => h.uncaughtException(Thread.currentThread(), e)
+    }
+}

monix-execution/jvm/src/main/scala/monix/execution/schedulers/Defaults.scala

@@ -18,12 +18,14 @@
 package monix.execution.schedulers
 
 import java.util.concurrent.{Executors, ScheduledExecutorService}
-import monix.execution.UncaughtExceptionReporter.LogExceptionsToStandardErr
+import monix.execution.UncaughtExceptionReporter
 
 private[schedulers] object Defaults {
-  /** Internal. Provides the `Scheduler.DefaultScheduledExecutor` instance. */
+  /**
+    * Internal. Provides the `Scheduler.DefaultScheduledExecutor` instance.
+    */
   lazy val scheduledExecutor: ScheduledExecutorService =
     Executors.newSingleThreadScheduledExecutor(
-      ThreadFactoryBuilder("monix-scheduler", LogExceptionsToStandardErr, daemonic = true)
+      ThreadFactoryBuilder("monix-scheduler", UncaughtExceptionReporter.default, daemonic = true)
     )
 }