Skip to content

Parallel performance improvement #2159

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vasilmkd merged 15 commits into from
Jul 30, 2021
Merged

Parallel performance improvement #2159

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
0cd2ceb
Add GenSpawn.mapBoth & instances
manufacturist Jul 26, 2021
7859ba3
Implement CommutativeApplicative.map2 in terms of mapBoth
manufacturist Jul 26, 2021
f33d6bc
Override GenConcurrent.mapBoth
manufacturist Jul 26, 2021
09e53c2
Drop Deferred usage from mapBoth override in GenConcurrent
manufacturist Jul 27, 2021
2413e27
Remove mapBoth override from GenSpawn instances
manufacturist Jul 27, 2021
6d68a0c
Move mapBoth to GenSpawn & remove overrides in instances
manufacturist Jul 27, 2021
9fd83cf
Fix 'safety: lhs error during rhs interruptible region' ResourceSpec
manufacturist Jul 27, 2021
006a0e4
Add onCancel finalizer
manufacturist Jul 27, 2021
ae121bd
Drop mapBoth; override map2, map2Eval in commutativeApplicativeForPar…
manufacturist Jul 29, 2021
d5f4f69
Merge branch 'series/3.x' into 2155-map-both-implementation
manufacturist Jul 29, 2021
2b8a670
Nitpick variable name (from old mapBoth implementation)
manufacturist Jul 29, 2021
f068de8
Add a Kleisli parTraverse stack overflow unit test
vasilmkd Jul 25, 2021
aac0738
Use Ref instead of capturing side effects
vasilmkd Jul 25, 2021
0f3400f
Make map2 & map2Eval cancellation safe
manufacturist Jul 30, 2021
8b19b6b
Poll after fiber cancel
manufacturist Jul 30, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 49 additions & 3 deletions kernel/shared/src/main/scala/cats/effect/kernel/instances/GenSpawnInstances.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,10 @@

package cats.effect.kernel.instances

import cats.{~>, Align, Applicative, CommutativeApplicative, Functor, Monad, Parallel}
import cats.{~>, Align, Applicative, CommutativeApplicative, Eval, Functor, Monad, Parallel}
import cats.data.Ior
import cats.implicits._
import cats.effect.kernel.{GenSpawn, ParallelF}
import cats.effect.kernel.{GenSpawn, Outcome, ParallelF}

trait GenSpawnInstances {

Expand Down Expand Up @@ -53,7 +53,53 @@ trait GenSpawnInstances {
final override def map2[A, B, Z](fa: ParallelF[F, A], fb: ParallelF[F, B])(

@johnynek johnynek Jul 30, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this code is exactly the same as below, except for the Eval wrapping, no?

Why not have one private method that returns Now[ParallelF[F, Z]] and in the map2 case pass fa, Now(fb)? I think the extra now wrap unwrap is likely trivial.

@djspiewak djspiewak Jul 31, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the extra now wrap unwrap is likely trivial.

The unwrap is pretty trivial but the allocation is… less so. Particularly since you pay that penalty on a per-item basis.

f: (A, B) => Z): ParallelF[F, Z] =
ParallelF(
F.both(ParallelF.value(fa), ParallelF.value(fb)).map { case (a, b) => f(a, b) }
F.uncancelable { poll =>
for {
fiberA <- F.start(ParallelF.value(fa))
fiberB <- F.start(ParallelF.value(fb))

a <- F
.onCancel(poll(fiberA.join), F.both(fiberA.cancel, fiberB.cancel).void)
.flatMap[A] {
Comment thread
vasilmkd marked this conversation as resolved.
case Outcome.Succeeded(fa) => fa
case Outcome.Errored(e) => fiberB.cancel *> F.raiseError(e)
case Outcome.Canceled() => fiberB.cancel *> poll(F.canceled *> F.never)
}

z <- F.onCancel(poll(fiberB.join), fiberB.cancel).flatMap[Z] {
case Outcome.Succeeded(fb) => fb.map(b => f(a, b))
case Outcome.Errored(e) => F.raiseError(e)
case Outcome.Canceled() => poll(F.canceled *> F.never)
}
} yield z
}
)

final override def map2Eval[A, B, Z](fa: ParallelF[F, A], fb: Eval[ParallelF[F, B]])(
f: (A, B) => Z): Eval[ParallelF[F, Z]] =
Eval.now(
ParallelF(
F.uncancelable { poll =>
for {
fiberA <- F.start(ParallelF.value(fa))
fiberB <- F.start(ParallelF.value(fb.value))

a <- F
.onCancel(poll(fiberA.join), F.both(fiberA.cancel, fiberB.cancel).void)
.flatMap[A] {
case Outcome.Succeeded(fa) => fa
case Outcome.Errored(e) => fiberB.cancel *> F.raiseError(e)
case Outcome.Canceled() => fiberB.cancel *> poll(F.canceled *> F.never)
}

z <- F.onCancel(poll(fiberB.join), fiberB.cancel).flatMap[Z] {
case Outcome.Succeeded(fb) => fb.map(b => f(a, b))
case Outcome.Errored(e) => F.raiseError(e)
case Outcome.Canceled() => poll(F.canceled *> F.never)
}
} yield z
}
)
)

final override def ap[A, B](ff: ParallelF[F, A => B])(
Expand Down
22 changes: 19 additions & 3 deletions tests/shared/src/test/scala/cats/effect/KleisliIOSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =>
Expand Down