Kleisli parTraverse

kernel/shared/src/main/scala/cats/effect/kernel/Async.scala

@@ -18,7 +18,7 @@ package cats.effect.kernel
 
 import cats.implicits._
 import cats.data.{EitherT, Ior, IorT, Kleisli, OptionT, WriterT}
-import cats.{~>, Monoid, Semigroup}
+import cats.{~>, Eval, Monoid, Semigroup}
 
 import cats.arrow.FunctionK
 import java.util.concurrent.atomic.AtomicReference
@@ -161,6 +161,9 @@ trait Async[F[_]] extends AsyncPlatform[F] with Sync[F] with Temporal[F] {
    * Note that if you use `defaultCont` you _have_ to override `async`.
    */
   def cont[K, R](body: Cont[F, K, R]): F[R]
+
+  override def bothEval[A, B](fa: F[A], fb: Eval[F[B]]): Eval[F[(A, B)]] =
+    Eval.now(both(fa, defer(fb.value)))
 }
 
 object Async {

kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala

@@ -16,7 +16,7 @@
 
 package cats.effect.kernel
 
-import cats.{~>, Applicative}
+import cats.{~>, Applicative, Eval}
 import cats.data.{EitherT, Ior, IorT, Kleisli, OptionT, WriterT}
 import cats.{Monoid, Semigroup}
 import cats.syntax.all._
@@ -434,6 +434,9 @@ trait GenSpawn[F[_], E] extends MonadCancel[F, E] with Unique[F] {
           }
       }
     }
+
+  def bothEval[A, B](fa: F[A], fb: Eval[F[B]]): Eval[F[(A, B)]] =
+    fb.map(fb => both(fa, fb))
 }
 
 object GenSpawn {

kernel/shared/src/main/scala/cats/effect/kernel/instances/GenSpawnInstances.scala

@@ -20,6 +20,7 @@ import cats.{~>, Align, Applicative, CommutativeApplicative, Functor, Monad, Par
 import cats.data.Ior
 import cats.implicits._
 import cats.effect.kernel.{GenSpawn, ParallelF}
+import cats.Eval
 
 trait GenSpawnInstances {
 
@@ -56,6 +57,11 @@ trait GenSpawnInstances {
           F.both(ParallelF.value(fa), ParallelF.value(fb)).map { case (a, b) => f(a, b) }
         )
 
+      final override def map2Eval[A, B, Z](fa: ParallelF[F, A], fb: Eval[ParallelF[F, B]])(
+          f: (A, B) => Z): Eval[ParallelF[F, Z]] =
+        F.bothEval(ParallelF.value(fa), fb.map(ParallelF.value))
+          .map(fp => ParallelF(fp.map(f.tupled)))
+
       final override def ap[A, B](ff: ParallelF[F, A => B])(
           fa: ParallelF[F, A]): ParallelF[F, B] =
         map2(ff, fa)(_(_))

laws/shared/src/test/scala/cats/effect/laws/PureConcSpec.scala

@@ -17,11 +17,14 @@
 package cats.effect
 package laws
 
-import cats.effect.kernel.testkit.{pure, OutcomeGenerators, PureConcGenerators, TimeT}, pure._,
-TimeT._
+import cats.Eval
+import cats.effect.kernel.GenSpawn
+import cats.effect.kernel.testkit.{pure, OutcomeGenerators, PureConcGenerators, TimeT}, pure._
+import TimeT._
 import cats.laws.discipline.arbitrary._
+import cats.syntax.eq._
 
-import org.scalacheck.Prop
+import org.scalacheck.Prop, Prop.forAll
 
 import org.specs2.ScalaCheck
 import org.specs2.mutable._
@@ -34,6 +37,34 @@ class PureConcSpec extends Specification with Discipline with ScalaCheck with Ba
   import PureConcGenerators._
   import OutcomeGenerators._
 
+  "PureConc" should {
+    "respect the both and bothEval correspondence" in {
+      def remainder3(n: Int): Int = {
+        val r = n % 3
+        if (r < 0) -r else r
+      }
+
+      def wrapEval(n: Int)(fb: PureConc[Int, Int]): Eval[PureConc[Int, Int]] =
+        remainder3(n) match {
+          case 0 => Eval.always(fb)
+          case 1 => Eval.later(fb)
+          case 2 => Eval.now(fb)
+        }
+
+      forAll { (fa: PureConc[Int, Int], fb: PureConc[Int, Int], n: Int) =>
+        val left =
+          GenSpawn[PureConc[Int, *]].both(fa, fb)
+
+        val right =
+          GenSpawn[PureConc[Int, *]].bothEval(fa, wrapEval(n)(fb))
+
+        val result = pure.run(left) eqv pure.run(right.value)
+
+        result must beTrue
+      }
+    }
+  }
+
   implicit def exec(fb: TimeT[PureConc[Int, *], Boolean]): Prop =
     Prop(pure.run(TimeT.run(fb)).fold(false, _ => false, _.getOrElse(false)))
 

tests/shared/src/test/scala/cats/effect/IOSpec.scala

@@ -16,6 +16,7 @@
 
 package cats.effect
 
+import cats.Eval
 import cats.kernel.laws.discipline.MonoidTests
 import cats.laws.discipline.{AlignTests, SemigroupKTests}
 import cats.laws.discipline.arbitrary._
@@ -478,6 +479,28 @@ class IOSpec extends IOPlatformSpecification with Discipline with ScalaCheck wit
           } yield (l2 -> r2)) must completeAs(true -> true)
         }
 
+        "correspond to bothEval" in ticked { implicit ticker =>
+          def remainder3(n: Int): Int = {
+            val r = n % 3
+            if (r < 0) -r else r
+          }
+
+          def wrapEval(n: Int)(fb: IO[Int]): Eval[IO[Int]] =
+            remainder3(n) match {
+              case 0 => Eval.always(fb)
+              case 1 => Eval.later(fb)
+              case 2 => Eval.now(fb)
+            }
+
+          forAll { (fa: IO[Int], fb: IO[Int], n: Int) =>
+            val left = IO.both(fa, fb)
+
+            val right = GenSpawn[IO].bothEval(fa, wrapEval(n)(fb))
+
+            left eqv right.value
+          }
+        }
+
       }
 
       "race" should {

tests/shared/src/test/scala/cats/effect/KleisliIOSpec.scala

@@ -47,10 +47,26 @@ class KleisliIOSpec
   "Kleisli[IO, R, *]" >> {
     "should be stack safe in long traverse chains" in ticked { implicit ticker =>
       val N = 10000
-      var i = 0
 
-      List.fill(N)(0).traverse_(_ => Kleisli.liftF(IO(i += 1))).run("Go...") *>
-        IO(i) must completeAs(N)
+      val test = for {
+        ref <- Ref[IO].of(0)
+        _ <- List.fill(N)(0).traverse_(_ => Kleisli.liftF(ref.update(_ + 1))).run("Go...")
+        v <- ref.get
+      } yield v
+
+      test must completeAs(N)
+    }
+
+    "should be stack safe in long parTraverse chains" in ticked { implicit ticker =>
+      val N = 10000
+
+      val test = for {
+        ref <- Ref[IO].of(0)
+        _ <- List.fill(N)(0).parTraverse_(_ => Kleisli.liftF(ref.update(_ + 1))).run("Go...")
+        v <- ref.get
+      } yield v
+
+      test must completeAs(N)
     }
 
     "execute finalizers" in ticked { implicit ticker =>